Changed Database to Influx

pyobs_weather/api/views.py

@@ -10,8 +10,9 @@
 import numpy as np
 
 from pyobs_weather.weather import evaluators
-from pyobs_weather.weather.models import Station, Sensor, Value, SensorType, GoodWeather
+from pyobs_weather.weather.models import Station, Sensor, SensorType, GoodWeather
 from pyobs_weather.weather.tasks import create_evaluator
+from pyobs_weather.weather.influx import get_value, get_list_from_influx, get_current
 
 
 def stations_list(request):
@@ -30,14 +31,13 @@ def station_detail(request, station_code):
     sensors = []
     for sensor in Sensor.objects.filter(station=station):
         # get latest value
-        value = Value.objects.filter(sensor=sensor).order_by('-time').first()
-
+        value = get_value(sensor.station.code, sensor.type.code)
         # append
         sensors.append({
             'name': sensor.type.name,
             'code': sensor.type.code,
-            'value': None if value is None else value.value,
-            'time': None if value is None else value.time
+            'value': None if value is None else value[1],
+            'time': None if value is None else value[0].strftime("%Y-%m-%dT%H:%M:%S.%fZ")
         })
 
     # return all
@@ -58,23 +58,23 @@ def sensor_detail(request, station_code, sensor_code):
         return HttpResponseNotFound("Sensor not found.")
 
     # get latest value
-    value = Value.objects.filter(sensor=sensor).order_by('-time').first()
+    value = get_value(sensor.station.code, sensor.type.code)
 
     # return it
     return JsonResponse({
         'name': sensor.type.name,
         'code': sensor.type.code,
-        'value': None if value is None else value.value,
+        'value': None if value is None else value[1],
         'unit': sensor.type.unit,
-        'time': None if value is None else value.time,
+        'time': None if value is None else value[0].strftime("%Y-%m-%dT%H:%M:%S.%fZ"),
         'good': sensor.good,
         'since': sensor.since
     })
 
 
 def current(request):
     # get average station
-    station = Station.objects.get(code='current')
+    station = Station.objects.get(code='dummy')  # current -> dummy
     if station is None:
         return HttpResponseNotFound('Could not access current weather.')
 
@@ -100,11 +100,11 @@ def current(request):
         # is average sensor?
         if sensor.station == station:
             # get latest value
-            value = Value.objects.filter(sensor=sensor).order_by('-time').first()
+            value = get_current(sensor.station.code, sensor.type.code)
 
             # set it
-            sensors[sensor.type.code]['value'] = None if value is None else value.value
-            time = value.time
+            sensors[sensor.type.code]['value'] = None if value[1] is None else value[1]
+            time = value[0].strftime("%Y-%m-%dT%H:%M:%S.%fZ")
 
     # totally good?
     good = True
@@ -144,17 +144,21 @@ def history(request, sensor_type):
     else:
         end = datetime.utcnow()
         start = end - timedelta(days=1)
-
+    
     # loop all sensors of that type
     stations = []
     areas = []
     for sensor in Sensor.objects.filter(type=st, station__history=True, station__active=True):
         # get data
-        values = Value.objects.filter(sensor=sensor, time__gte=start, time__lte=end)\
-            .order_by('-time').values('time', 'value')
+        #TODO: Check here and change if necessary
+        values_temp = get_list_from_influx(sensor.station.code, sensor.type.code, int(start.timestamp()), int(end.timestamp()))
+
+        values = []
+        for value in values_temp:
+            values.append({"time": value[0].strftime("%Y-%m-%dT%H:%M:%S.%fZ"), "value": value[1]})
 
         # got average sensor?
-        if sensor.station.code == 'average':
+        if sensor.station.code == 'average':    # average -> dummy
             # loop all evaluators for this sensor to define coloured areas in plot
             for evaluator in sensor.evaluators.all():
                 # get evaluator
@@ -169,7 +173,7 @@ def history(request, sensor_type):
             'code': sensor.station.code,
             'name': sensor.station.name,
             'color': sensor.station.color,
-            'data': list(values)
+            'data': values
         })
 
     return JsonResponse({'stations': stations, 'areas': areas}, safe=False)
@@ -191,8 +195,8 @@ def sensors(request):
 
     # add latest value
     for i, sensor in enumerate(data):
-        val = Value.objects.filter(sensor=sensor).order_by('-time').first()
-        values[i]['value'] = None if val is None else val.value
+        val = get_value(sensor.station.code, sensor.type.code)
+        values[i]['value'] = None if val is None else val[1]
 
     # return all
     return JsonResponse(values, safe=False)

pyobs_weather/settings.py

@@ -150,3 +150,16 @@
     from .local_settings import *
 except ImportError:
     pass
+
+
+# InfluxDB
+INFLUXDB_URL = 'http://localhost:8086'
+INFLUXDB_TOKEN = 'Y3xVIjr_9Q6Nu_jX2fdYzRfyIOsizUWtKYA3l3AhnDyOJ509LdCHcZ2FPlBSDrBhIMtMLpolUEbqVL_lruO4qg=='
+INFLUXDB_ORG = 'Test'
+INFLUXDB_BUCKET = 'weather'
+INFLUXDB_BUCKET_5MIN = 'weather_5m_average'
+INFLUXDB_BUCKET_CURRENT = 'weather_current'
+
+# bucket: weather, weather_5min_average, weather_current
+# _measurement: <station>
+# _field: humid,press,rain,temp,winddir,windspeed

pyobs_weather/weather/influx.py

@@ -0,0 +1,63 @@
+from influxdb_client import InfluxDBClient, Point
+
+from pyobs_weather.settings import INFLUXDB_URL, INFLUXDB_ORG, INFLUXDB_TOKEN, INFLUXDB_BUCKET, INFLUXDB_BUCKET_5MIN, INFLUXDB_BUCKET_CURRENT
+
+#TODO: Bucket als function argument bei allen. Influx Werte hierher, statt settings?
+
+def get_value(station_code, sensor_code, avg=False):
+    bucket = INFLUXDB_BUCKET_5MIN if avg else INFLUXDB_BUCKET
+    client = InfluxDBClient(url=INFLUXDB_URL, token=INFLUXDB_TOKEN, org=INFLUXDB_ORG)
+    query = f"""
+        from(bucket:"{bucket}")
+            |> range(start: -10m)\
+            |> filter(fn:(r) => r._measurement == "{station_code}")
+            |> filter(fn: (r) => r["_field"] == "{sensor_code}")
+            |> last()
+        """
+    result = client.query_api().query(org=INFLUXDB_ORG, query=query)
+    value = result.to_values(columns=('_time','_value'))
+
+    return value[0] if len(value) > 0 else None                                                     #[[]] -> [], where return_value[0] = time and return_value[1] = value
+
+
+def get_list_from_influx(station_code, sensor_code, start, end):                                    #TODO geeigneten Namen finden
+    client = InfluxDBClient(url=INFLUXDB_URL, token=INFLUXDB_TOKEN, org=INFLUXDB_ORG)
+    query = f"""
+        from(bucket:"{INFLUXDB_BUCKET_5MIN}")
+            |> range(start: {start}, stop: {end})\
+            |> filter(fn:(r) => r._measurement == "{station_code}")
+            |> filter(fn: (r) => r["_field"] == "{sensor_code}")
+        """
+    result = client.query_api().query(org=INFLUXDB_ORG, query=query)
+    return result.to_values(columns=('_time','_value'))
+
+
+def write_value(station_code, sensor_code, time, value, avg=False):
+    bucket = INFLUXDB_BUCKET_5MIN if avg else INFLUXDB_BUCKET
+
+    client = InfluxDBClient(url=INFLUXDB_URL, token=INFLUXDB_TOKEN, org=INFLUXDB_ORG)
+    write_api = client.write_api()
+    p = Point(station_code).field(sensor_code, value).time(time)
+    write_api.write(bucket=bucket, record=p)
+
+
+def write_current(station_code, sensor_code, time, value, avg=False):
+    client = InfluxDBClient(url=INFLUXDB_URL, token=INFLUXDB_TOKEN, org=INFLUXDB_ORG)
+    write_api = client.write_api()
+    p = Point(station_code).field(sensor_code, value).time(time)
+    write_api.write(bucket=INFLUXDB_BUCKET_CURRENT, record=p)
+
+
+def get_current(station_code, sensor_code):
+    client = InfluxDBClient(url=INFLUXDB_URL, token=INFLUXDB_TOKEN, org=INFLUXDB_ORG)
+    query = f"""
+        from(bucket:"{INFLUXDB_BUCKET_CURRENT}")
+            |> range(start: -10m)\
+            |> filter(fn:(r) => r._measurement == "{station_code}")
+            |> filter(fn: (r) => r["_field"] == "{sensor_code}")
+            |> last()
+        """
+    result = client.query_api().query(org=INFLUXDB_ORG, query=query)
+    value = result.to_values(columns=('_time','_value'))
+
+    return value[0] if len(value) > 0 else None

pyobs_weather/weather/models.py

@@ -125,8 +125,8 @@ def save(self, *args, **kwargs):
         models.Model.save(self, *args, **kwargs)
 
         # if station doesn't want to keep history, delete old
-        if not self.sensor.station.history:
-            Value.objects.filter(time__lt=self.time, sensor=self.sensor).delete()
+        #if not self.sensor.station.history:
+        #    Value.objects.filter(time__lt=self.time, sensor=self.sensor).delete()                                   #!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!
 
     class Meta:
         unique_together = ('sensor', 'time')

pyobs_weather/weather/stations/__init__.py

@@ -5,6 +5,7 @@
 from .mcdlockearchive import McDonaldLockeArchive
 from .mcdtelnet import McDonaldTelnet
 from .monet import Monet
-from .mysql import MySQL
+#from .mysql import MySQL
 from .csv import CSV
 from .json import JSON
+from .dummy import Dummy

pyobs_weather/weather/stations/average.py

@@ -29,13 +29,12 @@ def update(self):
         This method loops all sensor types and fetches all related sensors from all stations and calculates
         5-minutes averages, which it stores in sensors of the same type.
         """
-
-        from pyobs_weather.weather.models import SensorType, Sensor, Value
+        from pyobs_weather.weather.influx import get_value, write_value
+        from pyobs_weather.weather.models import SensorType, Sensor
         log.info('Updating averages...')
 
         # get now and since
         now = datetime.utcnow().astimezone(pytz.UTC)
-        since = now - timedelta(minutes=5, seconds=30)
 
         # loop all sensor types
         for sensor_type in SensorType.objects.filter(average=True):
@@ -48,19 +47,21 @@ def update(self):
             # loop all sensors of that type
             for sensor in Sensor.objects.filter(type=sensor_type, average=True, station__active=True):
                 # get average value for this sensor for last 5:30 minutes
-                value = Value.objects.filter(sensor=sensor, time__gte=since).aggregate(models.Avg('value'))
+                value = get_value(sensor.station.code, sensor.type.code, avg=True)
 
                 # valid?
-                if value is not None and value['value__avg'] is not None:
+                if value is not None and value[1] is not None:
                     # add it
-                    values.append(value['value__avg'])
+                    values.append(value[1])
 
             # calculate average of all sensors
             avg = np.nanmean(values) if values else None
 
             # and store it
             sensor = self._add_sensor(sensor_type.code)
-            Value.objects.get_or_create(sensor=sensor, time=now, value=avg)
+            station_code = self._station.code
+            write_value(station_code, sensor_type.code, now, avg, avg=True)
+
 
 
 __all__ = ['Average']

pyobs_weather/weather/stations/current.py

@@ -31,8 +31,8 @@ def update(self):
         averages of the latest values, which it stores in sensors of the same type. Values from other stations are
         only used if they are not older than 10 minutes.
         """
-
-        from pyobs_weather.weather.models import SensorType, Sensor, Value
+        from pyobs_weather.weather.influx import get_value, write_current
+        from pyobs_weather.weather.models import SensorType, Sensor
         log.info('Updating current...')
 
         # get now
@@ -49,21 +49,19 @@ def update(self):
                     continue
 
                 # get latest value of that sensor
-                value = Value.objects.filter(sensor=sensor).order_by('-time').first()
+                value = get_value(sensor.station.code, sensor.type.code)
 
                 # valid?
-                if value is not None and value.value is not None:
-                    # and not too old?
-                    if value.time > now - timedelta(minutes=10):
-                        # add it
-                        values.append(value.value)
+                if value is not None and value[1] is not None:
+                    values.append(value[1])
 
             # calculate average
             avg = np.mean(values) if values else None
 
             # and store it
             sensor = self._add_sensor(sensor_type.code)
-            Value.objects.get_or_create(sensor=sensor, time=now, value=avg)
+            station_code = self._station.code
+            write_current(station_code, sensor_type.code, now, avg)
 
 
 __all__ = ['Current']

pyobs_weather/weather/stations/dummy.py

@@ -0,0 +1,35 @@
+import logging
+from random import random, randint
+import datetime
+
+
+from .station import WeatherStation
+
+log = logging.getLogger(__name__)
+
+
+class Dummy(WeatherStation):
+
+    def __init__(self, *args, **kwargs):
+        WeatherStation.__init__(self, *args, **kwargs)
+
+    def create_sensors(self):
+        self._add_sensor('temp')
+        self._add_sensor('humid')
+        self._add_sensor('winddir')
+        self._add_sensor('windspeed')
+        self._add_sensor('press')
+        self._add_sensor('rain')
+
+    def update(self):
+        log.info('Updating Dummy station %s...' % self._station.code)
+        time = datetime.datetime.now()
+
+        self._add_value('temp', time, round(40*random(),1))
+        self._add_value('humid', time, round(100*random(),1))
+        self._add_value('winddir', time, round(360*random(),1))
+        self._add_value('windspeed', time, round(40*random(),1))
+        self._add_value('press', time, 970.0 + round(100*random(),1))
+        self._add_value('rain', time, randint(0,1))
+
+__all__ = ['Dummy']

pyobs_weather/weather/stations/station.py

@@ -1,4 +1,7 @@
-from pyobs_weather.weather.models import Value, Station, SensorType, Sensor
+from influxdb_client import InfluxDBClient, Point
+
+from pyobs_weather.weather.models import Station, SensorType, Sensor
+from pyobs_weather.weather.influx import write_value
 
 SENSOR_TYPES = dict(
     temp=dict(code='temp', name='Temperature', unit='°C'),
@@ -52,11 +55,7 @@ def _add_value(self, sensor_code, time, value):
             value: Measured value
         """
 
-        # get sensor
-        sensor = Sensor.objects.get(station=self._station, type__code=sensor_code)
-
-        # create value
-        Value.objects.get_or_create(sensor=sensor, time=time, defaults={'value': value})
-
+        station_code = self._station.code
+        write_value(station_code, sensor_code, time, value)
 
 __all__ = ['WeatherStation']

requirements.txt

@@ -4,7 +4,6 @@ celery
 Django
 django-celery-beat
 gunicorn
-mysqlclient
 numpy
 pandas
 psycopg2-binary