Merge pull request #465 from OpenTrafficCam/feature/3987-introduce-tr…

…acksegment

feature/3987-introduce-tracksegment

OTAnalytics/application/eventlist.py

@@ -1,5 +1,140 @@
-from OTAnalytics.domain.event import Event
-from OTAnalytics.domain.track_dataset import TrackDataset
+import re
+
+from OTAnalytics.domain import track
+from OTAnalytics.domain.event import (
+    FILE_NAME_PATTERN,
+    HOSTNAME,
+    Event,
+    ImproperFormattedFilename,
+)
+from OTAnalytics.domain.geometry import ImageCoordinate, calculate_direction_vector
+from OTAnalytics.domain.track_dataset import (
+    END_FRAME,
+    END_OCCURRENCE,
+    END_VIDEO_NAME,
+    END_X,
+    END_Y,
+    START_FRAME,
+    START_OCCURRENCE,
+    START_VIDEO_NAME,
+    START_X,
+    START_Y,
+    TrackDataset,
+    TrackSegmentDataset,
+)
+from OTAnalytics.domain.types import EventType
+
+
+def extract_hostname(name: str) -> str:
+    """Extract hostname from name.
+
+    Args:
+        name (Path): name containing the hostname.
+
+    Raises:
+        ImproperFormattedFilename: if the name is not formatted as expected, an
+            exception will be raised.
+
+    Returns:
+        str: the hostname.
+    """
+    match = re.search(
+        FILE_NAME_PATTERN,
+        name,
+    )
+    if match:
+        hostname: str = match.group(HOSTNAME)
+        return hostname
+    raise ImproperFormattedFilename(f"Could not parse {name}. Hostname is missing.")
+
+
+class SceneEventListBuilder:
+    """Create enter and leave scene events for track segments."""
+
+    def __init__(self) -> None:
+        self._events: list[Event] = []
+
+    def add_enter_scene_events(
+        self, segments: TrackSegmentDataset
+    ) -> "SceneEventListBuilder":
+        """Create an enter scene event for each track segment.
+
+        Args:
+            segments (TrackSegmentDataset): segments to be used to create events for
+        """
+        segments.apply(self._create_enter_scene_event)
+        return self
+
+    def add_leave_scene_events(
+        self, segments: TrackSegmentDataset
+    ) -> "SceneEventListBuilder":
+        """Create a leave scene event for each track segment.
+
+        Args:
+            segments (TrackSegmentDataset): segments to be used to create events for
+        """
+        segments.apply(self._create_leave_scene_event)
+        return self
+
+    def build(self) -> list[Event]:
+        """Create the complete event list.
+
+        Returns:
+            list[Event]: complete event list
+        """
+        return self._events.copy()
+
+    def _create_enter_scene_event(self, value: dict) -> None:
+        event = self.__create_event(
+            value=value,
+            event_type=EventType.ENTER_SCENE,
+            key_x=START_X,
+            key_y=START_Y,
+            key_occurrence=START_OCCURRENCE,
+            key_frame=START_FRAME,
+            key_video_name=START_VIDEO_NAME,
+        )
+        self._events.append(event)
+
+    def _create_leave_scene_event(self, value: dict) -> None:
+        event = self.__create_event(
+            value=value,
+            event_type=EventType.LEAVE_SCENE,
+            key_x=END_X,
+            key_y=END_Y,
+            key_occurrence=END_OCCURRENCE,
+            key_frame=END_FRAME,
+            key_video_name=END_VIDEO_NAME,
+        )
+        self._events.append(event)
+
+    @staticmethod
+    def __create_event(
+        value: dict,
+        event_type: EventType,
+        key_x: str,
+        key_y: str,
+        key_occurrence: str,
+        key_frame: str,
+        key_video_name: str,
+    ) -> Event:
+        return Event(
+            road_user_id=value[track.TRACK_ID],
+            road_user_type=value[track.TRACK_CLASSIFICATION],
+            hostname=extract_hostname(value[key_video_name]),
+            occurrence=value[key_occurrence],
+            frame_number=value[key_frame],
+            section_id=None,
+            event_coordinate=ImageCoordinate(value[key_x], value[key_y]),
+            event_type=event_type,
+            direction_vector=calculate_direction_vector(
+                value[START_X],
+                value[START_Y],
+                value[END_X],
+                value[END_Y],
+            ),
+            video_name=value[key_video_name],
+        )
 
 
 class SceneActionDetector:
@@ -14,7 +149,7 @@ def detect(self, tracks: TrackDataset) -> list[Event]:
         Returns:
             Iterable[Event]: the scene events
         """
-        events: list[Event] = []
-        tracks.apply_to_first_segments(events.append)
-        tracks.apply_to_last_segments(events.append)
-        return events
+        builder = SceneEventListBuilder()
+        builder.add_enter_scene_events(tracks.get_first_segments())
+        builder.add_leave_scene_events(tracks.get_last_segments())
+        return builder.build()

OTAnalytics/domain/track_dataset.py

@@ -3,17 +3,38 @@
 from datetime import datetime
 from typing import Any, Callable, Iterable, Iterator, Optional, Sequence
 
-from OTAnalytics.domain.event import Event
 from OTAnalytics.domain.geometry import RelativeOffsetCoordinate
 from OTAnalytics.domain.section import Section, SectionId
 from OTAnalytics.domain.track import Track, TrackId
 
+START_X: str = "start_x"
+START_Y: str = "start_y"
+START_OCCURRENCE: str = "start_occurrence"
+START_FRAME: str = "start_frame"
+START_VIDEO_NAME: str = "start_video_name"
+END_X: str = "end_x"
+END_Y: str = "end_y"
+END_OCCURRENCE: str = "end_occurrence"
+END_FRAME: str = "end_frame"
+END_VIDEO_NAME: str = "end_video_name"
+
 
 @dataclass(frozen=True, order=True)
 class IntersectionPoint:
     index: int
 
 
+class TrackSegmentDataset(ABC):
+    """Collection of track segments. A track segment consists of a start and an end
+    point with additional information about the corresponding detection.
+    """
+
+    @abstractmethod
+    def apply(self, consumer: Callable[[dict], None]) -> None:
+        """Apply the given consumer to the track segments."""
+        raise NotImplementedError
+
+
 class TrackDataset(ABC):
     def __iter__(self) -> Iterator[Track]:
         yield from self.as_list()
@@ -149,11 +170,13 @@ def calculate_geometries_for(
         raise NotImplementedError
 
     @abstractmethod
-    def apply_to_first_segments(self, consumer: Callable[[Event], None]) -> None:
+    def get_first_segments(self) -> TrackSegmentDataset:
+        """Get first segments of each track."""
         raise NotImplementedError
 
     @abstractmethod
-    def apply_to_last_segments(self, consumer: Callable[[Event], None]) -> None:
+    def get_last_segments(self) -> TrackSegmentDataset:
+        """Get last segments of each track"""
         raise NotImplementedError
 
     @abstractmethod