Feature/2485 reimplement intersection strategy for sections and flows

.gitignore

@@ -1,6 +1,7 @@
 # OTAnalytics
 data/
 !tests/data
+tests/data/OTCamera19_FR20_2023-05-24*
 tests/scripts/
 
 # Byte-compiled / optimized / DLL files

OTAnalytics/application/analysis/intersect.py

@@ -1,9 +1,16 @@
 from abc import ABC, abstractmethod
-from typing import Iterable
+from collections import defaultdict
+from typing import Iterable, Mapping
 
 from OTAnalytics.domain.event import Event
+from OTAnalytics.domain.geometry import RelativeOffsetCoordinate
 from OTAnalytics.domain.section import Section, SectionId
 from OTAnalytics.domain.track import TrackId
+from OTAnalytics.domain.types import EventType
+
+
+class IntersectionError(Exception):
+    pass
 
 
 class RunIntersect(ABC):
@@ -15,10 +22,19 @@ class RunIntersect(ABC):
     @abstractmethod
     def __call__(self, sections: Iterable[Section]) -> list[Event]:
         raise NotImplementedError
-        # bla
 
 
 class TracksIntersectingSections(ABC):
     @abstractmethod
     def __call__(self, sections: Iterable[Section]) -> dict[SectionId, set[TrackId]]:
         raise NotImplementedError
+
+
+def group_sections_by_offset(
+    sections: Iterable[Section], offset_type: EventType = EventType.SECTION_ENTER
+) -> Mapping[RelativeOffsetCoordinate, list[Section]]:
+    grouped_sections: dict[RelativeOffsetCoordinate, list[Section]] = defaultdict(list)
+    for section in sections:
+        offset = section.get_offset(offset_type)
+        grouped_sections[offset].append(section)
+    return grouped_sections

OTAnalytics/application/analysis/traffic_counting.py

@@ -16,7 +16,8 @@
 from OTAnalytics.domain.event import Event, EventRepository
 from OTAnalytics.domain.flow import Flow, FlowRepository
 from OTAnalytics.domain.section import Section, SectionId
-from OTAnalytics.domain.track import TrackId, TrackRepository
+from OTAnalytics.domain.track import TrackId
+from OTAnalytics.domain.track_repository import TrackRepository
 from OTAnalytics.domain.types import EventType
 
 LEVEL_FROM_SECTION = "from section"

OTAnalytics/application/datastore.py

@@ -22,11 +22,10 @@
     SectionListObserver,
     SectionRepository,
 )
-from OTAnalytics.domain.track import (
-    TrackDataset,
+from OTAnalytics.domain.track import TrackId, TrackImage
+from OTAnalytics.domain.track_dataset import TrackDataset
+from OTAnalytics.domain.track_repository import (
     TrackFileRepository,
-    TrackId,
-    TrackImage,
     TrackListObserver,
     TrackRepository,
 )

OTAnalytics/application/eventlist.py

@@ -1,68 +1,11 @@
-from typing import Iterable
-
-from OTAnalytics.domain.event import Event, EventType, SceneEventBuilder
-from OTAnalytics.domain.geometry import calculate_direction_vector
-from OTAnalytics.domain.track import Track
+from OTAnalytics.domain.event import Event
+from OTAnalytics.domain.track_dataset import TrackDataset
 
 
 class SceneActionDetector:
-    """Detect when a road user enters or leaves the scene.
-
-    Args:
-        scene_event_builder (SceneEventBuilder): the builder to build scene events
-    """
-
-    def __init__(self, scene_event_builder: SceneEventBuilder) -> None:
-        self._event_builder = scene_event_builder
-
-    def detect_enter_scene(self, track: Track) -> Event:
-        """Detect the first time a road user enters the scene.
-
-        Args:
-            tracks (Track): the track associated with the road user
-
-        Returns:
-            Iterable[Event]: the enter scene event
-        """
-        self._event_builder.add_event_type(EventType.ENTER_SCENE)
-        self._event_builder.add_road_user_type(track.classification)
-        first_detection = track.detections[0]
-        next_detection = track.detections[1]
-        self._event_builder.add_direction_vector(
-            calculate_direction_vector(
-                first_detection.x, first_detection.y, next_detection.x, next_detection.y
-            )
-        )
-        self._event_builder.add_event_coordinate(first_detection.x, first_detection.y)
-
-        return self._event_builder.create_event(first_detection)
-
-    def detect_leave_scene(self, track: Track) -> Event:
-        """Detect the last time a road user is seen before leaving the scene.
-
-        Args:
-            tracks (Track): the track associated with the road user
-
-        Returns:
-            Iterable[Event]: the leave scene event
-        """
-        self._event_builder.add_event_type(EventType.LEAVE_SCENE)
-        self._event_builder.add_road_user_type(track.classification)
-        last_detection = track.detections[-1]
-        second_to_last_detection = track.detections[-2]
-        self._event_builder.add_direction_vector(
-            calculate_direction_vector(
-                second_to_last_detection.x,
-                second_to_last_detection.y,
-                last_detection.x,
-                last_detection.y,
-            )
-        )
-        self._event_builder.add_event_coordinate(last_detection.x, last_detection.y)
-
-        return self._event_builder.create_event(last_detection)
+    """Detect when a road user enters or leaves the scene."""
 
-    def detect(self, tracks: Iterable[Track]) -> list[Event]:
+    def detect(self, tracks: TrackDataset) -> list[Event]:
         """Detect all enter and leave scene events.
 
         Args:
@@ -72,7 +15,6 @@ def detect(self, tracks: Iterable[Track]) -> list[Event]:
             Iterable[Event]: the scene events
         """
         events: list[Event] = []
-        for track in tracks:
-            events.append(self.detect_enter_scene(track))
-            events.append(self.detect_leave_scene(track))
+        tracks.apply_to_first_segments(events.append)
+        tracks.apply_to_last_segments(events.append)
         return events

OTAnalytics/application/geometry.py

@@ -2,36 +2,9 @@
 from functools import singledispatchmethod
 from typing import Generic, Iterable, TypeVar
 
-from OTAnalytics.domain.geometry import (
-    Coordinate,
-    Line,
-    Polygon,
-    RelativeOffsetCoordinate,
-)
-from OTAnalytics.domain.section import Section
+from OTAnalytics.domain.geometry import RelativeOffsetCoordinate
 from OTAnalytics.domain.track import Track
 
-
-class SectionGeometryBuilder:
-    def build_as_line(self, section: Section) -> Line:
-        return Line(section.get_coordinates())
-
-    def build_as_polygon(self, section: Section) -> Polygon:
-        return Polygon(section.get_coordinates())
-
-
-class TrackGeometryBuilder:
-    def build(self, track: Track, offset: RelativeOffsetCoordinate) -> Line:
-        coordinates = [
-            Coordinate(
-                detection.x + offset.x * detection.w,
-                detection.y + offset.y * detection.h,
-            )
-            for detection in track.detections
-        ]
-        return Line(coordinates)
-
-
 LINE = TypeVar("LINE")
 AREA = TypeVar("AREA")
 

OTAnalytics/application/state.py

@@ -17,10 +17,8 @@
     SectionRepositoryEvent,
     SectionType,
 )
-from OTAnalytics.domain.track import (
-    Detection,
-    TrackId,
-    TrackImage,
+from OTAnalytics.domain.track import Detection, TrackId, TrackImage
+from OTAnalytics.domain.track_repository import (
     TrackListObserver,
     TrackObserver,
     TrackRepository,

OTAnalytics/application/use_cases/create_events.py

@@ -110,7 +110,7 @@ def __init__(
 
     def __call__(self) -> None:
         """Create scene enter and leave events and save them to the event repository."""
-        tracks = self._get_tracks()
+        tracks = self._get_tracks.as_dataset()
         events = self._scene_action_detector.detect(tracks)
         self._add_events(events)