Merge pull request #741 from OpenTrafficCam/customer-reqs/8596-speed-up-otanalytics-by-using-vectorized-operations

Customer reqs/8596 speed up otanalytics by using vectorized operations

Author: briemla

.pre-commit-config.yaml

@@ -45,28 +45,30 @@ repos:
         entry: mypy OTAnalytics tests --config-file=pyproject.toml
         additional_dependencies:
           - av==13.0.0
-          - black==25.1.0
+          - black==24.8.0
           - customtkinter==5.2.2
-          - fastapi==0.115.13
+          - fastapi==0.116.2
           - hatch-requirements-txt==0.4.1
           - ijson==3.3.0
           - interrogate==1.7.0
-          - isort==6.0.1
+          - isort==5.13.2
           - matplotlib==3.10.0
-          - more-itertools==10.5.0
+          - more-itertools==10.7.0
           - mypy==1.16.1
-          - nicegui==2.20.0
+          - nicegui==2.24.1
           - numpy==2.1.1
           - opencv-python-headless==4.10.0.84
           - pandas-stubs==2.2.3.241126
           - pandas==2.2.3
+          - polars==1.32.3
           - pre-commit==4.1.0
+          - pyarrow==21.0.0
           - pytest-asyncio==0.24.0
           - pytest-benchmark==5.1.0
           - pytest-cov==6.0.0
           - pytest-lazy-fixtures==1.1.2
           - pytest==8.3.4
-          - starlette==0.45.3
+          - starlette==0.48.0
           - twine==6.1.0
           - types-PyYAML
           - types-flake8

OTAnalytics/adapter_ui/dummy_viewmodel.py

@@ -1330,6 +1330,7 @@ async def remove_flows(self) -> None:
             )
 
     def create_events(self) -> None:
+        logger().info("Creating events")
         start_msg_popup = self._ui_factory.minimal_info_box(
             message="Create events...",
             initial_position=self.get_position(),
@@ -1343,6 +1344,7 @@ def create_events(self) -> None:
         start_msg_popup.update_message(message="Creating events completed")
         sleep(1)
         start_msg_popup.close()
+        logger().info("Events created")
 
     def get_all_flow_ids(self) -> list[FlowId]:
         return [flow.id for flow in self.get_all_flows()]

OTAnalytics/application/analysis/intersect.py

@@ -2,31 +2,27 @@
 from collections import defaultdict
 from typing import Iterable, Mapping
 
-from OTAnalytics.domain.event import Event
+from OTAnalytics.domain.event import EventDataset
 from OTAnalytics.domain.geometry import RelativeOffsetCoordinate
 from OTAnalytics.domain.section import Section, SectionId
-from OTAnalytics.domain.track import TrackId
+from OTAnalytics.domain.track_dataset.track_dataset import TrackIdSet
 from OTAnalytics.domain.types import EventType
 
 
-class IntersectionError(Exception):
-    pass
-
-
 class RunIntersect(ABC):
     """
     Interface defining the use case to intersect the given tracks with the given
     sections.
     """
 
     @abstractmethod
-    def __call__(self, sections: Iterable[Section]) -> list[Event]:
+    def __call__(self, sections: Iterable[Section]) -> EventDataset:
         raise NotImplementedError
 
 
 class TracksIntersectingSections(ABC):
     @abstractmethod
-    def __call__(self, sections: Iterable[Section]) -> dict[SectionId, set[TrackId]]:
+    def __call__(self, sections: Iterable[Section]) -> dict[SectionId, TrackIdSet]:
         raise NotImplementedError
 
 

OTAnalytics/application/analysis/traffic_counting.py

@@ -19,9 +19,15 @@
 from OTAnalytics.application.export_formats.export_mode import ExportMode
 from OTAnalytics.application.use_cases.create_events import CreateEvents
 from OTAnalytics.application.use_cases.section_repository import GetSectionsById
-from OTAnalytics.domain.event import Event, EventRepository
+from OTAnalytics.domain.event import (
+    Event,
+    EventDataset,
+    EventRepository,
+    PythonEventDataset,
+)
 from OTAnalytics.domain.flow import Flow, FlowRepository
 from OTAnalytics.domain.section import Section, SectionId
+from OTAnalytics.domain.track_dataset.track_dataset import TrackIdSet, TrackIdSetFactory
 from OTAnalytics.domain.types import EventType
 
 DATETIME_FORMAT = "%Y-%m-%d_%H-%M-%S"
@@ -598,16 +604,23 @@ class RoadUserAssignments:
     """
 
     @property
-    def road_user_ids(self) -> list[str]:
+    def road_user_ids(self) -> TrackIdSet:
         """Returns a sorted list of all road user ids within this group of assignments.
 
         Returns:
             list[str]: the road user ids.
         """
-        return sorted([assignment.road_user for assignment in self._assignments])
+        return self._track_id_set_factory.create(
+            {assignment.road_user for assignment in self._assignments}
+        )
 
-    def __init__(self, assignments: list[RoadUserAssignment]) -> None:
+    def __init__(
+        self,
+        assignments: list[RoadUserAssignment],
+        track_id_set_factory: TrackIdSetFactory,
+    ) -> None:
         self._assignments = assignments.copy()
+        self._track_id_set_factory = track_id_set_factory
 
     def tag(self, by: Tagger) -> TaggedAssignments:
         """
@@ -688,9 +701,9 @@ class FilterBySectionEnterEvent(RoadUserAssignerDecorator):
     """Decorator to filters events by event type section-enter."""
 
     def assign(self, events: Iterable[Event], flows: list[Flow]) -> RoadUserAssignments:
-        section_enter_events: list[Event] = [
-            event for event in events if event.event_type == EventType.SECTION_ENTER
-        ]
+        section_enter_events = PythonEventDataset(
+            [event for event in events if event.event_type == EventType.SECTION_ENTER]
+        )
         return super().assign(section_enter_events, flows)
 
 
@@ -700,7 +713,9 @@ class SimpleRoadUserAssigner(RoadUserAssigner):
     """
 
     def __init__(
-        self, flow_selection: FlowSelection = MaxDurationFlowSelection()
+        self,
+        track_id_set_factory: TrackIdSetFactory,
+        flow_selection: FlowSelection = MaxDurationFlowSelection(),
     ) -> None:
         """
         Initialize the SimpleRoadUserAssigner with a flow selection strategy.
@@ -710,6 +725,7 @@ def __init__(
                 Defaults to MaxDurationFlowSelection.
         """
         self._flow_selection = flow_selection
+        self._track_id_set_factory = track_id_set_factory
 
     def assign(self, events: Iterable[Event], flows: list[Flow]) -> RoadUserAssignments:
         """
@@ -748,18 +764,18 @@ def __group_flows_by_sections(
 
     def __group_events_by_road_user(
         self, events: Iterable[Event]
-    ) -> dict[tuple[str, str], list[Event]]:
+    ) -> dict[tuple[str, str], EventDataset]:
         """
         Group events by road user.
 
         Args:
             events (Iterable[Event]): events of a road user
 
         Returns:
-            dict[tuple[RoadUserId, RoadUserType], list[Event]]: events grouped by user
+            dict[tuple[RoadUserId, RoadUserType], EventDataset]: events grouped by user
         """
-        events_by_road_user: dict[tuple[RoadUserId, RoadUserType], list[Event]] = (
-            defaultdict(list)
+        events_by_road_user: dict[tuple[RoadUserId, RoadUserType], EventDataset] = (
+            defaultdict(PythonEventDataset)
         )
         sorted_events = sorted(
             events, key=lambda _event: _event.interpolated_occurrence
@@ -774,15 +790,15 @@ def __group_events_by_road_user(
     def __assign_user_to_flow(
         self,
         flows: dict[tuple[SectionId, SectionId], list[Flow]],
-        events_by_road_user: dict[tuple[str, str], list[Event]],
+        events_by_road_user: dict[tuple[str, str], EventDataset],
     ) -> RoadUserAssignments:
         """
         Assign each user to flows based on the flow selection strategy.
 
         Args:
             flows (dict[tuple[SectionId, SectionId], list[Flow]]): flows by start and
                 end section
-            events_by_road_user (dict[str, list[Event]]): events by road user
+            events_by_road_user (dict[str, EventDataset]): events by road user
 
         Returns:
             RoadUserAssignments: group of RoadUserAssignment objects
@@ -795,35 +811,35 @@ def __assign_user_to_flow(
                     road_user_id=road_user_id, road_user_type=road_user_type
                 )
                 assignments.extend(user_assignments)
-        return RoadUserAssignments(assignments)
+        return RoadUserAssignments(assignments, self._track_id_set_factory)
 
     def __create_candidates(
         self,
         flows: dict[tuple[SectionId, SectionId], list[Flow]],
-        events: list[Event],
+        events: EventDataset,
     ) -> list[FlowCandidate]:
         """
         Create flow candidates to select one from in a later step.
 
         Args:
             flows (dict[tuple[SectionId, SectionId], list[Flow]]): flows by start and
                 end section
-            events (list[Event]): events belonging to road user
+            events (EventDataset): events belonging to road user
 
         Returns:
             list[FlowCandidate]: the flow candidates pertaining to road user
         """
         event_pairs = self.__create_event_pairs(events)
         return self.__create_candidate_flows(flows, event_pairs)
 
-    def __create_event_pairs(self, events: list[Event]) -> list[EventPair]:
+    def __create_event_pairs(self, events: EventDataset) -> list[EventPair]:
         """
         Create event pairs.
 
         Requires and assumes events to be sorted by occurrence.
 
         Args:
-            events(list[Event]): events to create the event pairs with
+            events(EventDataset): events to create the event pairs with
 
         Returns:
             list[EventPair]: event pairs

OTAnalytics/application/application.py

@@ -83,7 +83,6 @@
     SectionId,
     SectionListObserver,
 )
-from OTAnalytics.domain.track import TrackId, TrackImage
 from OTAnalytics.domain.types import EventType
 from OTAnalytics.domain.video import Video, VideoListObserver
 
@@ -415,19 +414,6 @@ def save_otflow(self, file: Path) -> None:
         """
         self._save_otflow.save(file)
 
-    def get_image_of_track(self, track_id: TrackId) -> Optional[TrackImage]:
-        """
-        Retrieve an image for the given track.
-
-        Args:
-            track_id (TrackId): identifier for the track
-
-        Returns:
-            Optional[TrackImage]: an image of the track if the track is available and
-            the image can be loaded
-        """
-        return self._datastore.get_image_of_track(track_id)
-
     def create_events(self) -> None:
         """
         Intersect all tracks with all sections and write the events into the event