add tests for reports

comptages/core/report.py

@@ -11,8 +11,9 @@
 from comptages.datamodel import models
 
 
-def simple_print_callback(progress):
-    print(f"Generating report... {progress}%")
+def simple_print_callback(progress: int):
+    if progress > 0:
+        print(f"Generating report... {progress}%")
 
 
 def prepare_reports(

comptages/core/statistics.py

@@ -531,7 +531,7 @@ def get_special_periods(first_day, last_day) -> QuerySet[models.SpecialPeriod]:
     return qs
 
 
-def get_month_data(section: models.Section, start, end) -> pd.DataFrame:
+def get_month_data(section: models.Section, start, end, direction=None) -> pd.DataFrame:
     qs = models.CountDetail.objects.filter(
         id_lane__id_section=section, timestamp__gte=start, timestamp__lt=end
     )
@@ -543,7 +543,6 @@ def get_month_data(section: models.Section, start, end) -> pd.DataFrame:
         .annotate(tm=Sum("times"))
         .values("month", "tm", "import_status")
     )
-
     if direction is not None:
         qs = qs.filter(id_lane__direction=direction)
 

comptages/report/yearly_report_bike.py

@@ -655,30 +655,32 @@ def render_section_dist(value: Union[str, Decimal, None]) -> str:
         column_offset = 1
 
         data = self.tjms_by_weekday_and_month()
-        row = row_offset
         for i in data:
-            ws.cell(row=row, column=column_offset, value=i["date"])
-            ws.cell(row=row, column=column_offset + 1, value=i["tjm"])
-            row += 1
+            ws.cell(
+                row=i["month"] + row_offset,
+                column=i["weekday"] + column_offset,
+                value=i["tjm"],
+            )
 
         """ CV_LV """
+        # Is this superseded by the `Data_yearly_stats` tab ?
 
-        ws = workbook["CV_LV"]
+        # ws = workbook["CV_LV"]
 
-        ws["F12"] = self.tjms_by_direction_bike([1], 1)
-        ws["G12"] = self.tjms_by_direction_bike([1], 2)
-        ws["H12"] = self.tjms_by_direction_bike([2, 3, 4, 5], 1)
-        ws["I12"] = self.tjms_by_direction_bike([2, 3, 4, 5], 2)
+        # ws["F12"] = self.tjms_by_direction_bike([1], 1)
+        # ws["G12"] = self.tjms_by_direction_bike([1], 2)
+        # ws["H12"] = self.tjms_by_direction_bike([2, 3, 4, 5], 1)
+        # ws["I12"] = self.tjms_by_direction_bike([2, 3, 4, 5], 2)
 
-        ws["J35"] = self.total()
-        ws["J39"] = self.max_day()[0]
-        ws["K39"] = self.max_day()[1]
+        # ws["K35"] = self.total()
+        # ws["J39"] = self.max_day()[0]
+        # ws["K39"] = self.max_day()[1]
 
-        ws["J40"] = self.max_month()[0]
-        ws["K40"] = self.max_month()[1]
+        # ws["J40"] = self.max_month()[0]
+        # ws["K40"] = self.max_month()[1]
 
-        ws["J41"] = self.min_month()[0]
-        ws["k41"] = self.min_month()[1]
+        # ws["J41"] = self.min_month()[0]
+        # ws["k41"] = self.min_month()[1]
 
         """ Data_year """
 

comptages/test/test_report.py

@@ -4,6 +4,8 @@
 from datetime import datetime
 from django.test import TransactionTestCase
 from django.core.management import call_command
+from django.db.models import Count
+from itertools import islice
 from openpyxl import load_workbook
 
 from comptages.test import utils as test_utils
@@ -13,11 +15,11 @@
 from comptages.core import report, importer
 
 
-class ImportTest(TransactionTestCase):
+class ReportTest(TransactionTestCase):
     @classmethod
     def setUpClass(cls):
         cls.test_outputs = Path("/OpenComptage/test_outputs")
-        cls.test_data = Path("/OpenComptage/test_data")
+        cls.test_data = Path("/test_data")
 
         for file in cls.test_outputs.iterdir():
             os.remove(file)
@@ -39,6 +41,51 @@ def test_report(self):
         importer.import_file(test_utils.test_data_path("00056520.V02"), count)
         report.prepare_reports("/tmp/", count)
 
+    def test_generate_yearly_reports(self):
+        call_command("importdata", "--only-count", "--limit", 100)
+        # year and section id come from --add-count`
+        year = 2021
+        section_id = "00107695"
+        installations = models.Installation.objects.filter(lane__id_section=section_id)
+        installation = installations.first()
+        assert installation
+
+        count = models.Count.objects.filter(id_installation=installation.id).first()
+        report.prepare_reports(
+            self.test_outputs, count, year, "yearly", sections_ids=[section_id]
+        )
+
+        self.assertTrue(list(Path(self.test_outputs).iterdir()))
+
+    def test_generate_yearly_reports_special_case(self):
+        call_command("importdata", "--only-count", "--limit", 100)
+
+        special_case_installations = models.Installation.objects.annotate(
+            sections=Count("lane__id_section")
+        ).filter(sections__lte=3)
+        self.assertTrue(special_case_installations.exists())
+
+        count = models.Count.objects.filter(
+            id_installation__in=special_case_installations
+        ).first()
+        assert count
+
+        year = count.start_process_date.year
+        section = (
+            models.Section.objects.filter(lane__id_installation=count.id_installation)
+            .distinct()
+            .first()
+        )
+        assert section
+
+        print(
+            f"Preparing reports for count = {count}, year = {year}, section_id = {section.id}"
+        )
+        report.prepare_reports(
+            self.test_outputs, count, year, "yearly", sections_ids=[section.id]
+        )
+        self.assertTrue(list(Path(self.test_outputs).iterdir()))
+
     def test_all_sections_default(self):
         # Test if default report features all sections for special case
         test_data_folder = "5350_1_4"
@@ -82,6 +129,102 @@ def test_all_sections_default(self):
         self.assertGreater(found_files, 0)
         self.assertEqual(found_files % n_sections, 0)
 
+    def test_all_sections_yearly(self):
+        # Test if yearly report features all sections for special case
+        test_data_folder = "ASC"
+        installation_name = "53309999"
+
+        installation = models.Installation.objects.get(name=installation_name)
+
+        model = models.Model.objects.all().first()
+        device = models.Device.objects.all().first()
+        sensor_type = models.SensorType.objects.all().first()
+        class_ = models.Class.objects.get(name="SWISS10")
+        tz = pytz.timezone("Europe/Zurich")
+
+        count = models.Count.objects.create(
+            start_put_date=tz.localize(datetime(2021, 1, 1)),
+            start_service_date=tz.localize(datetime(2021, 1, 8)),
+            end_service_date=tz.localize(datetime(2021, 12, 15)),
+            start_process_date=tz.localize(datetime(2021, 1, 15)),
+            end_process_date=tz.localize(datetime(2021, 12, 28)),
+            end_put_date=tz.localize(datetime(2021, 12, 31)),
+            id_model=model,
+            id_device=device,
+            id_sensor_type=sensor_type,
+            id_class=class_,
+            id_installation=installation,
+        )
+
+        gen = Path(test_utils.test_data_path(test_data_folder)).iterdir()
+        to_import = 100
+        imported = 0
+        for file in islice(gen, to_import):
+            importer.import_file(test_utils.test_data_path(str(file)), count)
+            imported += 1
+            print(f"Remaining: {to_import - imported}")
+
+        sections = models.Section.objects.filter(
+            lane__id_installation=installation.id, lane__countdetail__id_count=count.id
+        ).distinct()
+        self.assertTrue(sections.exists())
+
+        sections_ids = list(sections.values_list("id", flat=True))
+        report.prepare_reports(
+            file_path=self.test_outputs,
+            count=count,
+            year=count.start_process_date.year,
+            template="yearly",
+            sections_ids=sections_ids,
+        )
+        found_files = len(list(Path(self.test_outputs).iterdir()))
+        self.assertEqual(found_files, sections.count())
+
+    def test_yearly_bike_report(self):
+        # Import test data pertaining to "mobilité douce"
+        installation_name = "64540060"
+        installation = models.Installation.objects.get(name=installation_name)
+        model = models.Model.objects.all().first()
+        device = models.Device.objects.all().first()
+        sensor_type = models.SensorType.objects.all().first()
+        class_ = models.Class.objects.get(name="SPCH-MD 5C")
+        tz = pytz.timezone("Europe/Zurich")
+
+        count = models.Count.objects.create(
+            start_service_date=tz.localize(datetime(2021, 2, 1)),
+            end_service_date=tz.localize(datetime(2021, 12, 10)),
+            start_process_date=tz.localize(datetime(2021, 2, 10)),
+            end_process_date=tz.localize(datetime(2021, 12, 15)),
+            start_put_date=tz.localize(datetime(2021, 1, 1)),
+            end_put_date=tz.localize(datetime(2021, 1, 5)),
+            id_model=model,
+            id_device=device,
+            id_sensor_type=sensor_type,
+            id_class=class_,
+            id_installation=installation,
+        )
+
+        path_to_file = self.test_data.joinpath("64540060_Latenium_PS2021_ChMixte.txt")
+        importer.import_file(str(path_to_file), count)
+        print("Imported 1 count files!")
+
+        models.CountDetail.objects.update(import_status=0)
+        print("Forced import status to 'definitive' for testing purposes")
+
+        sections_ids = (
+            models.Section.objects.filter(lane__id_installation__name=installation_name)
+            .distinct()
+            .values_list("id", flat=True)
+        )
+        self.assertTrue(sections_ids.exists())
+
+        report = YearlyReportBike(
+            path_to_output_dir=self.test_outputs,
+            year=2021,
+            section_id=sections_ids.first(),
+        )
+        report.run()
+
     def test_busiest_by_season(self):
         # Import test data pertaining to "mobilité douce"
         installation = models.Installation.objects.get(name="00107695")
@@ -113,6 +256,7 @@ def test_busiest_by_season(self):
 
         # Collect count details
         details = YearlyReportBike.count_details_by_season(count)
+        assert details
 
         def inspect_leaves(d, res) -> list[int]:
             for v in d.values():
@@ -175,7 +319,7 @@ def test_busiest_by_day_month(self):
 
         # Prepare workbook
         path_to_inputs = Path("comptages/report").joinpath("template_yearly_bike.xlsx")
-        path_to_outputs = Path("/test_outputs").joinpath("yearly_bike.xlsx")
+        path_to_outputs = self.test_outputs.joinpath("yearly_bike.xlsx")
         wb = load_workbook(path_to_inputs)
 
         # Write data & save
@@ -221,7 +365,7 @@ def test_busiest_by_various_criteria(self):
 
         # Prepare workbook
         path_to_inputs = Path("comptages/report").joinpath("template_yearly_bike.xlsx")
-        path_to_outputs = Path("/test_outputs").joinpath("yearly_bike.xlsx")
+        path_to_outputs = self.test_outputs.joinpath("yearly_bike.xlsx")
         wb = load_workbook(path_to_inputs)
 
         # Write data & save

docker-compose.yml

@@ -7,7 +7,6 @@ services:
       - /tmp/.X11-unix:/tmp/.X11-unix ro
       - ${PWD}/comptages:/root/.local/share/QGIS/QGIS3/profiles/default/python/plugins/comptages/
       - ${PWD}/test_data:/test_data
-      - ${PWD}/test_outputs:/test_outputs
     command:
       - qgis
     environment:
@@ -30,7 +29,7 @@ services:
     volumes:
       - ${PWD}:/OpenComptage
       - ${PWD}/test_data:/test_data
-      - ${PWD}/test_outputs:/test_outputs
+      - ${PWD}/test_outputs:/OpenComptage/test_outputs
 
   db:
     image: postgis/postgis:${POSTGIS_VERSION:-12-3.4}