Use raw sql query with a faster db + Remove SensorDataStat

Author: karimkawambwa

contrib/start.sh

@@ -13,6 +13,8 @@ touch /src/logs/gunicorn.log
 touch /src/logs/access.log
 tail -n 0 -f /src/logs/*.log &
 
+#purge all configured task queues
+celery -A sensorsafrica purge
 celery -A sensorsafrica beat -l info &> /src/logs/celery.log  &
 celery -A sensorsafrica worker --hostname=$DOKKU_APP_NAME -l info &> /src/logs/celery.log  &
 celery -A sensorsafrica flower --basic_auth=$SENSORSAFRICA_FLOWER_ADMIN_USERNAME:$SENSORSAFRICA_FLOWER_ADMIN_PASSWORD &> /src/logs/celery.log  &

sensorsafrica/admin.py

@@ -2,7 +2,7 @@
 from django.utils.html import format_html
 from django.conf.urls import include, url
 from django.template.response import TemplateResponse
-from .api.models import LastActiveNodes, SensorDataStat, City
+from .api.models import LastActiveNodes, City
 from django.db.models import Q
 
 from feinstaub.sensors.admin import (
@@ -77,57 +77,6 @@ def delete_model(self, request, obj):
     def save_related(self, request, form, formsets, change):
         pass
 
-
-@admin.register(SensorDataStat)
-class SensorDataStatAdmin(admin.ModelAdmin):
-    readonly_fields = [
-        "node",
-        "sensor",
-        "location",
-        "city_slug",
-        "value_type",
-        "average",
-        "maximum",
-        "minimum",
-        "timestamp",
-    ]
-    search_fields = ["city_slug", "value_type"]
-    list_display = [
-        "node",
-        "sensor",
-        "location",
-        "city_slug",
-        "value_type",
-        "average",
-        "maximum",
-        "minimum",
-        "timestamp",
-        "created",
-        "modified",
-    ]
-    list_filter = ["timestamp", "node", "sensor", "location"]
-
-    def get_actions(self, request):
-        actions = super(SensorDataStatAdmin, self).get_actions(request)
-        del actions["delete_selected"]
-        return actions
-
-    def has_add_permission(self, request):
-        return False
-
-    def has_delete_permission(self, request, obj=None):
-        return False
-
-    def save_model(self, request, obj, form, change):
-        pass
-
-    def delete_model(self, request, obj):
-        pass
-
-    def save_related(self, request, form, formsets, change):
-        pass
-
-
 @admin.register(City)
 class CityAdmin(admin.ModelAdmin):
     search_fields = ["slug", "name", "country"]

sensorsafrica/api/models.py

@@ -5,12 +5,18 @@
 
 
 class City(TimeStampedModel):
-    slug = models.CharField(max_length=255, db_index=True, null=False, blank=False)
-    name = models.CharField(max_length=255, db_index=True, null=False, blank=False)
-    country = models.CharField(max_length=255, db_index=True, null=False, blank=False)
-    location = models.CharField(max_length=255, db_index=True, null=False, blank=False)
-    latitude = models.DecimalField(max_digits=14, decimal_places=11, null=True, blank=True)
-    longitude = models.DecimalField(max_digits=14, decimal_places=11, null=True, blank=True)
+    slug = models.CharField(
+        max_length=255, db_index=True, null=False, blank=False)
+    name = models.CharField(
+        max_length=255, db_index=True, null=False, blank=False)
+    country = models.CharField(
+        max_length=255, db_index=True, null=False, blank=False)
+    location = models.CharField(
+        max_length=255, db_index=True, null=False, blank=False)
+    latitude = models.DecimalField(
+        max_digits=14, decimal_places=11, null=True, blank=True)
+    longitude = models.DecimalField(
+        max_digits=14, decimal_places=11, null=True, blank=True)
 
     class Meta:
         verbose_name_plural = "Cities"
@@ -20,36 +26,6 @@ def save(self, *args, **kwargs):
         return super(City, self).save(*args, **kwargs)
 
 
-class SensorDataStat(TimeStampedModel):
-    node = models.ForeignKey(Node)
-    sensor = models.ForeignKey(Sensor)
-    location = models.ForeignKey(SensorLocation)
-
-    city_slug = models.CharField(max_length=255, db_index=True, null=False, blank=False)
-    value_type = models.CharField(max_length=255, db_index=True, null=False, blank=False)
-
-    average = models.FloatField(null=False, blank=False)
-    maximum = models.FloatField(null=False, blank=False)
-    minimum = models.FloatField(null=False, blank=False)
-
-    # Number of data points averaged
-    sample_size = models.IntegerField(null=False, blank=False)
-    # Last datetime of calculated stats
-    last_datetime = models.DateTimeField()
-
-    timestamp = models.DateTimeField()
-
-    def __str__(self):
-        return "%s %s %s avg=%s min=%s max=%s" % (
-            self.timestamp,
-            self.city_slug,
-            self.value_type,
-            self.average,
-            self.minimum,
-            self.maximum,
-        )
-
-
 class LastActiveNodes(TimeStampedModel):
     node = models.ForeignKey(Node)
     location = models.ForeignKey(SensorLocation)

sensorsafrica/api/v2/router.py

@@ -5,7 +5,7 @@
 
 data_router = routers.DefaultRouter()
 
-data_router.register(r"", SensorDataStatView)
+data_router.register(r"", SensorDataStatView, basename="sensor_data_stat_view")
 
 city_router = routers.DefaultRouter()
 

sensorsafrica/api/v2/serializers.py

@@ -1,14 +1,35 @@
 from rest_framework import serializers
 
 
-class SensorDataStatSerializer(serializers.Serializer):
-    average = serializers.FloatField()
-    minimum = serializers.FloatField()
-    maximum = serializers.FloatField()
-    value_type = serializers.CharField(max_length=200)
-    start_datetime = serializers.DateTimeField()
-    end_datetime = serializers.DateTimeField()
-    city_slug = serializers.CharField(max_length=200)
+class RawSensorDataStatSerializer(serializers.Serializer):
+    average = serializers.SerializerMethodField()
+    minimum = serializers.SerializerMethodField()
+    maximum = serializers.SerializerMethodField()
+    value_type = serializers.SerializerMethodField()
+    start_datetime = serializers.SerializerMethodField()
+    end_datetime = serializers.SerializerMethodField()
+    city_slug = serializers.SerializerMethodField()
+
+    def get_city_slug(self, obj):
+        return obj[0]
+
+    def get_start_datetime(self, obj):
+        return obj[1]
+
+    def get_end_datetime(self, obj):
+        return obj[2]
+
+    def get_average(self, obj):
+        return obj[3]
+
+    def get_minimum(self, obj):
+        return obj[4]
+
+    def get_maximum(self, obj):
+        return obj[5]
+
+    def get_value_type(self, obj):
+        return obj[6]
 
 
 class CitySerializer(serializers.Serializer):

sensorsafrica/api/v2/views.py

@@ -8,11 +8,13 @@
 from django.utils import timezone
 from dateutil.relativedelta import relativedelta
 from django.db.models import ExpressionWrapper, F, FloatField, Max, Min, Sum, Avg, Q
-from django.db.models.functions import Cast, TruncDate
+from django.db.models.functions import Cast, TruncDay, TruncHour, TruncMinute, TruncMonth
 from rest_framework import mixins, pagination, viewsets
 
-from ..models import SensorDataStat, LastActiveNodes, City, Node
-from .serializers import SensorDataStatSerializer, CitySerializer
+from django.db import connection
+
+from ..models import LastActiveNodes, City, Node
+from .serializers import RawSensorDataStatSerializer, CitySerializer
 
 from feinstaub.sensors.views import StandardResultsSetPagination
 
@@ -76,7 +78,8 @@ def get_paginated_response(self, data_stats):
                 results[city_slug][value_type] = [] if from_date else {}
 
             values = results[city_slug][value_type]
-            include_result = getattr(values, "append" if from_date else "update")
+            include_result = getattr(
+                values, "append" if from_date else "update")
             include_result(
                 {
                     "average": data_stat["average"],
@@ -98,8 +101,7 @@ def get_paginated_response(self, data_stats):
 
 
 class SensorDataStatView(mixins.ListModelMixin, viewsets.GenericViewSet):
-    queryset = SensorDataStat.objects.none()
-    serializer_class = SensorDataStatSerializer
+    serializer_class = RawSensorDataStatSerializer
     pagination_class = CustomPagination
 
     @method_decorator(cache_page(3600))
@@ -112,60 +114,31 @@ def get_queryset(self):
         city_slugs = self.request.query_params.get("city", None)
         from_date = self.request.query_params.get("from", None)
         to_date = self.request.query_params.get("to", None)
+        avg = self.request.query_params.get("avg", 'day')
 
         if to_date and not from_date:
-            raise ValidationError({"from": "Must be provide along with to query"})
+            raise ValidationError(
+                {"from": "Must be provide along with to query"})
         if from_date:
-            validate_date(from_date, {"from": "Must be a date in the format Y-m-d."})
+            validate_date(
+                from_date, {"from": "Must be a date in the format Y-m-d."})
         if to_date:
-            validate_date(to_date, {"to": "Must be a date in the format Y-m-d."})
+            validate_date(
+                to_date, {"to": "Must be a date in the format Y-m-d."})
 
-        value_type_to_filter = self.request.query_params.get("value_type", None)
+        value_type_to_filter = self.request.query_params.get(
+            "value_type", None)
 
         filter_value_types = value_types[sensor_type]
         if value_type_to_filter:
-            filter_value_types = set(value_type_to_filter.upper().split(",")) & set(
+            filter_value_types = ",".join(set(value_type_to_filter.upper().split(",")) & set(
                 [x.upper() for x in value_types[sensor_type]]
-            )
+            ))
 
         if not from_date and not to_date:
-            return self._retrieve_past_24hrs(city_slugs, filter_value_types)
-
-        return self._retrieve_range(from_date, to_date, city_slugs, filter_value_types)
-
-    @staticmethod
-    def _retrieve_past_24hrs(city_slugs, filter_value_types):
-        to_date = timezone.now().replace(minute=0, second=0, microsecond=0)
-        from_date = to_date - datetime.timedelta(hours=24)
-
-        queryset = SensorDataStat.objects.filter(
-            value_type__in=filter_value_types,
-            timestamp__gte=from_date,
-            timestamp__lte=to_date,
-        )
-
-        if city_slugs:
-            queryset = queryset.filter(city_slug__in=city_slugs.split(","))
-
-        return (
-            queryset.order_by()
-            .values("value_type", "city_slug")
-            .annotate(
-                start_datetime=Min("timestamp"),
-                end_datetime=Max("timestamp"),
-                average=ExpressionWrapper(
-                    Sum(F("average") * F("sample_size")) / Sum("sample_size"),
-                    output_field=FloatField(),
-                ),
-                minimum=Min("minimum"),
-                maximum=Max("maximum"),
-            )
-            .order_by("city_slug")
-        )
-
-    @staticmethod
-    def _retrieve_range(from_date, to_date, city_slugs, filter_value_types):
-        if not to_date:
+            to_date = timezone.now().replace(minute=0, second=0, microsecond=0)
+            from_date = to_date - datetime.timedelta(hours=24)
+        elif not to_date:
             from_date = beginning_of_day(from_date)
             # Get data from_date until the end
             # of day yesterday which is the beginning of today
@@ -174,31 +147,36 @@ def _retrieve_range(from_date, to_date, city_slugs, filter_value_types):
             from_date = beginning_of_day(from_date)
             to_date = end_of_day(to_date)
 
-        queryset = SensorDataStat.objects.filter(
-            value_type__in=filter_value_types,
-            timestamp__gte=from_date,
-            timestamp__lt=to_date,
-        )
-
-        if city_slugs:
-            queryset = queryset.filter(city_slug__in=city_slugs.split(","))
-
-        return (
-            queryset.annotate(date=TruncDate("timestamp"))
-            .values("date", "value_type")
-            .annotate(
-                city_slug=F("city_slug"),
-                start_datetime=Min("timestamp"),
-                end_datetime=Max("timestamp"),
-                average=ExpressionWrapper(
-                    Sum(F("average") * F("sample_size")) / Sum("sample_size"),
-                    output_field=FloatField(),
-                ),
-                minimum=Min("minimum"),
-                maximum=Max("maximum"),
-            )
-            .order_by("-date")
-        )
+        with connection.cursor() as cursor:
+            cursor.execute(
+                """
+                    SELECT
+                        sl.city as city_slug,
+                        min(sd."timestamp") as start_datetime,
+                        max(sd."timestamp") as end_datetime,
+                        sum(CAST("value" as float)) / COUNT(*) AS average,
+                        min(CAST("value" as float)) as minimum,
+                        max(CAST("value" as float)) as maximum,
+                        v.value_type
+                    FROM
+                        sensors_sensordatavalue v
+                        INNER JOIN sensors_sensordata sd ON sd.id = sensordata_id
+                        INNER JOIN sensors_sensorlocation sl ON sl.id = location_id
+                    WHERE
+                        v.value_type IN (%s)
+                        """
+                +
+                ("AND sl.city IN (%s)" if city_slugs else "")
+                +
+                """
+                                        AND sd."timestamp" >= TIMESTAMP %s
+                                        AND sd."timestamp" <= TIMESTAMP %s
+                    GROUP BY
+                        DATE_TRUNC(%s, sd."timestamp"),
+                        v.value_type,
+                        sl.city
+                """, [filter_value_types, city_slugs, from_date, to_date, avg] if city_slugs else [filter_value_types, from_date, to_date, avg])
+            return cursor.fetchall()
 
 
 class CityView(mixins.ListModelMixin, viewsets.GenericViewSet):
@@ -225,7 +203,8 @@ def list(self, request):
             moved_to = None
             # Get data stats from 5mins before last_data_received_at
             if last_data_received_at:
-                last_5_mins = last_data_received_at - datetime.timedelta(minutes=5)
+                last_5_mins = last_data_received_at - \
+                    datetime.timedelta(minutes=5)
                 stats = (
                     SensorDataValue.objects.filter(
                         Q(sensordata__sensor__node=last_active.node.id),