Merge pull request #686 from OpenTrafficCam/feature/7883-add-count-plotters

Feature/7883 add count plotters

Author: randy-seng

OTAnalytics/adapter_ui/dummy_viewmodel.py

@@ -1292,6 +1292,10 @@ def create_events(self) -> None:
             initial_position=self.get_position(),
         )
         self._application.create_events()
+
+        # TODO find appropriate place to trigger update of plots
+        # self._application.update_count_plots()
+
         self.notify_flows(self.get_all_flow_ids())
         start_msg_popup.update_message(message="Creating events completed")
         sleep(1)

OTAnalytics/application/analysis/traffic_counting.py

@@ -1,9 +1,14 @@
+import re
+import unicodedata
 from abc import ABC, abstractmethod
 from collections import defaultdict
 from dataclasses import dataclass
 from datetime import datetime, timedelta, timezone
+from pathlib import Path
 from typing import Callable, Iterable, Optional
 
+from PIL.Image import Image
+
 from OTAnalytics.application.analysis.traffic_counting_specification import (
     CountingSpecificationDto,
     ExportCounts,
@@ -19,6 +24,8 @@
 from OTAnalytics.domain.section import Section, SectionId
 from OTAnalytics.domain.types import EventType
 
+DATETIME_FORMAT = "%Y-%m-%d_%H-%M-%S"
+
 LEVEL_FROM_SECTION = "from section"
 LEVEL_TO_SECTION = "to section"
 LEVEL_FLOW = "flow"
@@ -995,9 +1002,9 @@ def create_export_specification(
     return ExportSpecificationDto(counting_specification, flow_dtos)
 
 
-class ExportTrafficCounting(ExportCounts):
+class TrafficCounting:
     """
-    Use case to export traffic counting.
+    Use case to produce traffic counts.
     """
 
     def __init__(
@@ -1008,36 +1015,82 @@ def __init__(
         create_events: CreateEvents,
         assigner: RoadUserAssigner,
         tagger_factory: TaggerFactory,
-        exporter_factory: ExporterFactory,
-    ) -> None:
+    ):
         self._event_repository = event_repository
         self._flow_repository = flow_repository
         self._get_sections_by_id = get_sections_by_id
         self._create_events = create_events
         self._assigner = assigner
         self._tagger_factory = tagger_factory
-        self._exporter_factory = exporter_factory
 
-    def export(self, specification: CountingSpecificationDto) -> None:
+    def count(self, specification: CountingSpecificationDto) -> Count:
         """
-        Export the traffic countings based on the currently available events and flows.
+        Produce traffic counts based on the currently available events and flows.
 
         Args:
             specification (CountingSpecificationDto): specification of the export
         """
         if self._event_repository.is_empty():
             self._create_events()
-        events = self._event_repository.get(
-            start_date=specification.start,
-            end_date=specification.end,
-        )
-        flows = self._flow_repository.get_all()
+
+        if specification.count_all_events:
+            events = self._event_repository.get_all()
+        else:
+            events = self._event_repository.get(
+                start_date=specification.start,
+                end_date=specification.end,
+            )
+
+        flows = self.get_flows()
         assigned_flows = self._assigner.assign(events, flows)
         tagger = self._tagger_factory.create_tagger(specification)
         tagged_assignments = assigned_flows.tag(tagger)
-        counts = tagged_assignments.count(flows)
+        return tagged_assignments.count(flows)
+
+    def get_flows(self) -> list[Flow]:
+        return self._flow_repository.get_all()
+
+    def with_tagger_factory(
+        self, new_tagger_factory: TaggerFactory
+    ) -> "TrafficCounting":
+        return TrafficCounting(
+            self._event_repository,
+            self._flow_repository,
+            self._get_sections_by_id,
+            self._create_events,
+            self._assigner,
+            new_tagger_factory,
+        )
+
+    def provide_get_sections_by_id(self) -> GetSectionsById:
+        return self._get_sections_by_id
+
+
+class ExportTrafficCounting(ExportCounts):
+    """
+    Use case to export traffic counting.
+    """
+
+    def __init__(
+        self,
+        traffic_counting: TrafficCounting,
+        exporter_factory: ExporterFactory,
+    ) -> None:
+        self._traffic_counting = traffic_counting
+        self._exporter_factory = exporter_factory
+
+    def export(self, specification: CountingSpecificationDto) -> None:
+        """
+        Export the traffic countings based on the currently available events and flows.
+
+        Args:
+            specification (CountingSpecificationDto): specification of the export
+        """
+        counts = self._traffic_counting.count(specification)
+
+        flows = self._traffic_counting.get_flows()
         export_specification = create_export_specification(
-            flows, specification, self._get_sections_by_id
+            flows, specification, self._traffic_counting.provide_get_sections_by_id()
         )
         exporter = self._exporter_factory.create_exporter(export_specification)
         exporter.export(counts, specification.export_mode)
@@ -1050,3 +1103,49 @@ def get_supported_formats(self) -> Iterable[ExportFormat]:
             Iterable[ExportFormat]: supported export formats
         """
         return self._exporter_factory.get_supported_formats()
+
+
+@dataclass(frozen=True)
+class CountImage:
+    """
+    Represents an image with a counts plot.
+    """
+
+    image: Image
+    width: int
+    height: int
+    name: str
+    timestamp: datetime
+
+    def save(self, save_dir: Path) -> None:
+        time = self.timestamp.strftime(DATETIME_FORMAT)
+
+        save_dir.mkdir(parents=True, exist_ok=True)
+
+        self.image.save(save_dir / f"counts_plot_{self.safe_filename()}_{time}.png")
+
+    def safe_filename(self, max_length: int = 255, replacement: str = "_") -> str:
+        """Create a string from the image name that can be used as the filename.
+
+        Use ascii only and replace illegal characters (/:*?"<>|),
+        whitespaces and linebreaks.
+        Remove sequences of replacement char and truncate to max_length.
+
+        Args:
+            max_length (int, optional): length for truncating the resulting file name.
+                Defaults to 255.
+            replacement (str, optional): character to replace illegal characters
+                in image name. Defaults to "_".
+
+        Returns:
+            str: string safe to use as filename
+        """
+        safe = (
+            unicodedata.normalize("NFKD", self.name).encode("ascii", "ignore").decode()
+        )
+        safe = re.sub(r'[<>:"/\\|?*\n\r\t]', replacement, safe)
+        safe = re.sub(r"\s+", replacement, safe)
+        safe = re.sub(rf"{re.escape(replacement)}+", replacement, safe)
+        safe = safe.strip(replacement)
+        # Truncate to max length (preserving file extension if needed)
+        return safe[:max_length]

OTAnalytics/application/analysis/traffic_counting_specification.py

@@ -25,6 +25,7 @@ class CountingSpecificationDto:
     output_format: str
     output_file: str
     export_mode: ExportMode
+    count_all_events: bool = False
 
 
 @dataclass(frozen=True)

OTAnalytics/application/application.py

@@ -71,6 +71,7 @@
     ExportTrackStatistics,
     TrackStatisticsExportSpecification,
 )
+from OTAnalytics.application.use_cases.update_count_plots import CountPlotsUpdater
 from OTAnalytics.application.use_cases.update_project import ProjectUpdater
 from OTAnalytics.domain.date import DateRange
 from OTAnalytics.domain.filter import FilterElement, FilterElementSettingRestorer
@@ -147,6 +148,7 @@ def __init__(
         number_of_tracks_assigned_to_each_flow: NumberOfTracksAssignedToEachFlow,
         export_track_statistics: ExportTrackStatistics,
         get_current_remark: GetCurrentRemark,
+        update_count_plots: CountPlotsUpdater,
     ) -> None:
         self._datastore: Datastore = datastore
         self.track_state: TrackState = track_state
@@ -193,6 +195,7 @@ def __init__(
         )
         self._export_track_statistics = export_track_statistics
         self._get_current_remark = get_current_remark
+        self._update_count_plots = update_count_plots
 
     def connect_observers(self) -> None:
         """
@@ -708,6 +711,9 @@ def get_track_statistics_export_formats(
     ) -> Iterable[ExportFormat]:
         return self._export_track_statistics.get_supported_formats()
 
+    def update_count_plots(self) -> None:
+        self._update_count_plots.update()
+
 
 class MissingTracksError(Exception):
     pass

OTAnalytics/application/state.py

@@ -5,6 +5,7 @@
 from pathlib import Path
 from typing import Callable, Generic, Optional
 
+from OTAnalytics.application.analysis.traffic_counting import CountImage
 from OTAnalytics.application.config import DEFAULT_TRACK_OFFSET
 from OTAnalytics.application.datastore import Datastore
 from OTAnalytics.application.playback import SkipTime
@@ -206,6 +207,8 @@ def __init__(self) -> None:
         ](default=[])
         self.skip_time = ObservableProperty[SkipTime](DEFAULT_SKIP_TIME)
 
+        self.count_plots = ObservableProperty[list[CountImage]](default=[])
+
     def reset(self) -> None:
         """Reset to default settings."""
         self.selected_videos.set([])
@@ -217,6 +220,8 @@ def reset(self) -> None:
         self.track_offset.set(DEFAULT_TRACK_OFFSET)
         self.skip_time.set(DEFAULT_SKIP_TIME)
 
+        self.count_plots.set([])
+
 
 class LiveImage:
     """

OTAnalytics/application/use_cases/update_count_plots.py

@@ -0,0 +1,35 @@
+from pathlib import Path
+
+from OTAnalytics.application.analysis.traffic_counting import CountImage
+from OTAnalytics.application.state import TrackViewState
+from OTAnalytics.plugin_ui.visualization.counts.counts_plotter import CountPlotter
+
+
+class CountPlotsUpdater:
+
+    def __init__(self, state: TrackViewState, plotter: CountPlotter) -> None:
+        self._state = state
+        self._plotter = plotter
+
+    def __call__(self) -> None:
+        self.update()
+
+    def update(self) -> None:
+        width = self._state.view_width.get()
+        height = self._state.view_height.get()
+        plots = self._plotter.plot(width, height)
+
+        self._state.count_plots.set(plots)
+
+
+class CountPlotSaver:
+
+    def __init__(self, path: Path):
+        self._path = path
+
+    def __call__(self, plots: list[CountImage]) -> None:
+        self.save(plots)
+
+    def save(self, plots: list[CountImage]) -> None:
+        for plot in plots:
+            plot.save(self._path)