Imprecise columns in MD hour + week

comptages/core/report.py

@@ -1,10 +1,11 @@
 import os
+from datetime import date, datetime, timedelta
+from typing import Generator
 
-from datetime import timedelta, datetime
-from openpyxl import load_workbook, Workbook
+from openpyxl import Workbook, load_workbook
 
-from comptages.datamodel import models
 from comptages.core import statistics
+from comptages.datamodel import models
 
 
 def simple_print_callback(progress):
@@ -89,7 +90,7 @@ def _prepare_yearly_report(
  workbook.save(filename=output)
 
 
-def _mondays_of_count(count: models.Count):
+def _mondays_of_count(count: models.Count) -> Generator[date, None, None]:
  """Generator that return the Mondays of the count"""
 
  start = count.start_process_date

comptages/core/statistics.py

@@ -1,10 +1,8 @@
-import pandas as pd
-
-from datetime import timedelta, datetime
+from datetime import datetime, timedelta
 
-from django.db.models import F, CharField, Value, Q
-from django.db.models import Sum
-from django.db.models.functions import ExtractHour, Trunc, Concat
+import pandas as pd
+from django.db.models import CharField, F, Q, Sum, Value
+from django.db.models.functions import Concat, ExtractHour, Trunc
 
 from comptages.core import definitions
 from comptages.datamodel import models

comptages/report/yearly_report_bike.py

@@ -1,19 +1,17 @@
 import os
-
-
-from django.db.models import Sum, Count
-from django.db.models.functions import Cast
 from django.db.models.fields import DateField
 from django.db.models.functions import (
- ExtractIsoWeekDay,
+ Cast,
  ExtractHour,
+ ExtractIsoWeekDay,
  ExtractMonth,
+ TruncDate,
 )
-
+from django.db.models import Count, Sum, Avg
 from openpyxl import load_workbook
 
 from comptages.core import definitions
-from comptages.datamodel.models import CountDetail, Section, Lane
+from comptages.datamodel.models import CountDetail, Lane, Section
 
 
 class YearlyReportBike:
@@ -24,7 +22,7 @@ def __init__(self, file_path, year, section_id):
  self.year = year
  self.section_id = section_id
 
- def values_by_direction(self):
+ def values_by_direction(self) -> "ValuesQuerySet[CountDetail, Any]":
  # Get all the count details for section and the year
  qs = CountDetail.objects.filter(
  id_lane__id_section__id=self.section_id,
@@ -34,14 +32,14 @@ def values_by_direction(self):
  )
 
  # Total by day of the week (0->monday, 7->sunday) and by direction
- result = (
+ return (
  qs.annotate(weekday=ExtractIsoWeekDay("timestamp"))
  .values("weekday")
  .annotate(total=Sum("times"))
  .values("weekday", "id_lane__direction", "total")
  )
 
- def values_by_day_and_hour(self):
+ def values_by_day_and_hour(self) -> "ValuesQuerySet[CountDetail, dict[str, Any]]":
  # Get all the count details for section and the year
  qs = CountDetail.objects.filter(
  id_lane__id_section__id=self.section_id,
@@ -54,16 +52,20 @@ def values_by_day_and_hour(self):
 
  # Total by day of the week (0->monday, 6->sunday) and by hour (0->23)
  result = (
- qs.annotate(weekday=ExtractIsoWeekDay("timestamp"))
+ qs.annotate(date=TruncDate("timestamp"))
+ .values("date")
+ .annotate(Sum("times"))
+ .annotate(weekday=ExtractIsoWeekDay("date"))
+ .values("weekday")
+ .annotate(tjm=Avg("times"))
  .annotate(hour=ExtractHour("timestamp"))
- .values("weekday", "hour")
- .annotate(tjm=Sum("times") / 51)
  .values("weekday", "hour", "tjm")
  )
-
  return result
 
- def values_by_hour_and_direction(self, direction, weekdays=[0, 1, 2, 3, 4, 5, 6]):
+ def values_by_hour_and_direction(
+ self, direction, weekdays=[0, 1, 2, 3, 4, 5, 6]
+ ) -> "ValuesQuerySet[CountDetail, Any]":
  # Get all the count details for section and the year
  qs = CountDetail.objects.filter(
  id_lane__id_section__id=self.section_id,
@@ -77,15 +79,18 @@ def values_by_hour_and_direction(self, direction, weekdays=[0, 1, 2, 3, 4, 5, 6]
 
  # Total by hour (0->23)
  result = (
- qs.annotate(hour=ExtractHour("timestamp"))
- .values("hour")
- .annotate(tjm=Sum("times") / 365)
+ qs.annotate(date=TruncDate("timestamp"))
+ .values("date")
+ .annotate(
+ tjm=Sum("times"),
+ hour=ExtractHour("timestamp"),
+ )
  .values("hour", "tjm")
  )
 
  return result
 
- def values_by_day_and_month(self):
+ def values_by_day_and_month(self) -> "ValuesQuerySet[CountDetail, Any]":
  # Get all the count details for section and the year
  qs = CountDetail.objects.filter(
  id_lane__id_section__id=self.section_id,
@@ -98,16 +103,18 @@ def values_by_day_and_month(self):
 
  # Total by day of the week (0->monday, 6->sunday) and by month (1->12)
  result = (
- qs.annotate(weekday=ExtractIsoWeekDay("timestamp"))
- .annotate(month=ExtractMonth("timestamp"))
- .values("weekday", "month")
- .annotate(tjm=Sum("times") / 12)
+ qs.annotate(date=TruncDate("timestamp"))
+ .values("date")
+ .annotate(Sum("times"))
+ .annotate(weekday=ExtractIsoWeekDay("timestamp"))
+ .values("weekday")
+ .annotate(tjm=Avg("times"), month=ExtractMonth("timestamp"))
  .values("weekday", "month", "tjm")
  )
 
  return result
 
- def values_by_day(self):
+ def values_by_day(self) -> "ValuesQuerySet[CountDetail, dict[str, Any]]":
  # Get all the count details for section and the year
  qs = CountDetail.objects.filter(
  id_lane__id_section__id=self.section_id,
@@ -125,27 +132,31 @@ def values_by_day(self):
 
  return result
 
- def values_by_day_of_week(self):
+ def values_by_day_of_week(self) -> "ValuesQuerySet[CountDetail, dict[str, Any]]":
  # Get all the count details for section and the year
  qs = CountDetail.objects.filter(
  id_lane__id_section__id=self.section_id,
  timestamp__year=self.year,
+ import_status=definitions.IMPORT_STATUS_DEFINITIVE,
  )
 
  # TODO: don't divide by 51 but actually aggregate first by the
  # real days (with sum) and then aggregate by weekday (with average)
 
  # Group by day of the week (0->monday, 7->sunday)
  result = (
- qs.annotate(weekday=ExtractIsoWeekDay("timestamp"))
+ qs.annotate(date=TruncDate("timestamp"))
+ .values("date")
+ .annotate(Sum("times"))
+ .annotate(weekday=ExtractIsoWeekDay("timestamp"))
  .values("weekday")
- .annotate(tjm=Sum("times") / 51)
+ .annotate(tjm=Avg("times"))
  .values("weekday", "tjm")
  )
 
  return result
 
- def values_by_class(self):
+ def values_by_class(self) -> "ValuesQuerySet[CountDetail, dict[str, Any]]":
  # Get all the count details for section and the year
  qs = CountDetail.objects.filter(
  id_lane__id_section__id=self.section_id,
@@ -161,7 +172,9 @@ def values_by_class(self):
  )
  return result
 
- def tjm_direction_bike(self, categories, direction, weekdays=[0, 1, 2, 3, 4, 5, 6]):
+ def tjm_direction_bike(
+ self, categories, direction, weekdays=[0, 1, 2, 3, 4, 5, 6]
+ ) -> dict:
  qs = CountDetail.objects.filter(
  id_lane__id_section__id=self.section_id,
  timestamp__year=self.year,
@@ -174,7 +187,7 @@ def tjm_direction_bike(self, categories, direction, weekdays=[0, 1, 2, 3, 4, 5,
  # TODO: avoid the division?
  return qs.aggregate(res=Sum("times"))["res"] / 365
 
- def total(self, categories=[1]):
+ def total(self, categories=[1]) -> dict:
  qs = CountDetail.objects.filter(
  timestamp__year=self.year,
  id_category__code__in=categories,
@@ -183,7 +196,7 @@ def total(self, categories=[1]):
 
  return qs.aggregate(res=Sum("times"))["res"]
 
- def max_day(self, categories=[1]):
+ def max_day(self, categories=[1]) -> tuple:
  qs = (
  CountDetail.objects.filter(
  timestamp__year=self.year,
@@ -198,7 +211,7 @@ def max_day(self, categories=[1]):
 
  return qs[0]["total"], qs[0]["date"]
 
- def max_month(self, categories=[1]):
+ def max_month(self, categories=[1]) -> tuple:
  qs = (
  CountDetail.objects.filter(
  timestamp__year=self.year,
@@ -213,7 +226,7 @@ def max_month(self, categories=[1]):
 
  return qs[0]["total"], qs[0]["month"]
 
- def min_month(self, categories=[1]):
+ def min_month(self, categories=[1]) -> tuple:
  qs = (
  CountDetail.objects.filter(
  timestamp__year=self.year,

comptages/test/__init__.py

@@ -0,0 +1,32 @@
+from datetime import datetime
+from typing import Optional
+from comptages.datamodel import models
+import pytz
+
+
+def yearly_count_for(
+ year: int,
+ installation: models.Installation,
+ class_: Optional[models.Class] = None,
+ model: Optional[models.Model] = None,
+ device: Optional[models.Device] = None,
+ sensor_type: Optional[models.SensorType] = None,
+) -> models.Count:
+ tz = pytz.timezone("Europe/Zurich")
+ model = model or models.Model.objects.all()[0]
+ device = device or models.Device.objects.all()[0]
+ sensor_type = sensor_type or models.SensorType.objects.all()[0]
+ class_ = class_ or models.Class.objects.get(name="SWISS10")
+ return models.Count.objects.create(
+ start_put_date=tz.localize(datetime(year, 1, 1)),
+ start_service_date=tz.localize(datetime(year, 1, 8)),
+ start_process_date=tz.localize(datetime(year, 1, 15)),
+ end_process_date=tz.localize(datetime(year, 12, 17)),
+ end_service_date=tz.localize(datetime(year, 12, 24)),
+ end_put_date=tz.localize(datetime(year, 12, 31)),
+ id_model=model,
+ id_device=device,
+ id_sensor_type=sensor_type,
+ id_class=class_,
+ id_installation=installation,
+ )