Merge branch 'main' into bug/9548-incomplete-filenames-in-otanalytics-if-video-name-has-multiple-dots

# Conflicts:
#	OTAnalytics/application/datastore.py
#	OTAnalytics/plugin_parser/track_export.py
#	OTAnalytics/plugin_ui/cli.py
#	tests/unit/OTAnalytics/plugin_parser/test_road_user_assignment_export.py
#	tests/unit/OTAnalytics/plugin_parser/test_track_export.py

Author: randy-seng

OTAnalytics/application/datastore.py

@@ -1,9 +1,9 @@
 from abc import ABC, abstractmethod
-from dataclasses import dataclass
 from datetime import datetime
 from pathlib import Path
 from typing import Iterable, Optional, Sequence
 
+from OTAnalytics.application.parser.track_parser import TrackParser
 from OTAnalytics.application.project import Project
 from OTAnalytics.application.use_cases.export_events import (
     EventExportSpecification,
@@ -40,49 +40,6 @@
 )
 
 
-@dataclass(frozen=True)
-class DetectionMetadata:
-    detection_classes: frozenset[str]
-
-
-@dataclass(frozen=True)
-class TrackParseResult:
-    tracks: TrackDataset
-    detection_metadata: DetectionMetadata
-    video_metadata: VideoMetadata
-
-
-@dataclass(frozen=True)
-class TracksParseResult:
-    tracks: TrackDataset
-    detections_metadata: list[DetectionMetadata]
-    videos_metadata: list[VideoMetadata]
-
-
-def combine_track_datasets(results: list[TrackParseResult]) -> TrackDataset:
-    if not results:
-        raise ValueError("No results to combine")
-    tracks = results[0].tracks
-    for result in results[1:]:
-        tracks.add_all(result.tracks)
-    return tracks
-
-
-class TrackParser(ABC):
-    def parse_files(self, files: list[Path]) -> TracksParseResult:
-        if not files:
-            raise ValueError("No files to parse")
-        results = [self.parse(file) for file in files]
-        tracks = combine_track_datasets(results)
-        detections_metadata = [result.detection_metadata for result in results]
-        videos_metadata = [result.video_metadata for result in results]
-        return TracksParseResult(tracks, detections_metadata, videos_metadata)
-
-    @abstractmethod
-    def parse(self, file: Path) -> TrackParseResult:
-        raise NotImplementedError
-
-
 class EventListParser(ABC):
     @abstractmethod
     def serialize(

OTAnalytics/application/eventlist.py

@@ -12,11 +12,15 @@
 from OTAnalytics.domain.geometry import ImageCoordinate, calculate_direction_vector
 from OTAnalytics.domain.track_dataset.track_dataset import (
     END_FRAME,
+    END_GEO_X,
+    END_GEO_Y,
     END_OCCURRENCE,
     END_VIDEO_NAME,
     END_X,
     END_Y,
     START_FRAME,
+    START_GEO_X,
+    START_GEO_Y,
     START_OCCURRENCE,
     START_VIDEO_NAME,
     START_X,
@@ -95,6 +99,8 @@ def _create_enter_scene_event(self, value: dict) -> None:
             key_occurrence=START_OCCURRENCE,
             key_frame=START_FRAME,
             key_video_name=START_VIDEO_NAME,
+            key_geo_x=START_GEO_X,
+            key_geo_y=START_GEO_Y,
         )
         self._events.append(event)
 
@@ -107,6 +113,8 @@ def _create_leave_scene_event(self, value: dict) -> None:
             key_occurrence=END_OCCURRENCE,
             key_frame=END_FRAME,
             key_video_name=END_VIDEO_NAME,
+            key_geo_x=END_GEO_X,
+            key_geo_y=END_GEO_Y,
         )
         self._events.append(event)
 
@@ -119,6 +127,8 @@ def __create_event(
         key_occurrence: str,
         key_frame: str,
         key_video_name: str,
+        key_geo_x: str,
+        key_geo_y: str,
     ) -> Event:
         image_coordinate = ImageCoordinate(value[key_x], value[key_y])
         occurrence = value[key_occurrence]
@@ -140,6 +150,8 @@ def __create_event(
             video_name=value[key_video_name],
             interpolated_occurrence=occurrence,
             interpolated_event_coordinate=image_coordinate,
+            geo_x=value.get(key_geo_x),
+            geo_y=value.get(key_geo_y),
         )
 
 

OTAnalytics/application/export_formats/event_list.py

@@ -21,6 +21,8 @@
 INTERPOLATED_EVENT_COORDINATE = event.INTERPOLATED_EVENT_COORDINATE
 INTERPOLATED_EVENT_COORDINATE_X = "interpolated_coordinate_px_x"
 INTERPOLATED_EVENT_COORDINATE_Y = "interpolated_coordinate_px_y"
+GEO_X: str = event.GEO_X
+GEO_Y: str = event.GEO_Y
 
 DATE_FORMAT = "%Y-%m-%d"
 TIME_FORMAT = "%H:%M:%S.%f"

OTAnalytics/application/export_formats/road_user_assignments.py

@@ -53,6 +53,10 @@ def _prepend_end(key: str) -> str:
 END_INTERPOLATED_EVENT_COORD_Y = _prepend_end(
     event_list.INTERPOLATED_EVENT_COORDINATE_Y
 )
+START_GEO_X = _prepend_start(event_list.GEO_X)
+START_GEO_Y = _prepend_start(event_list.GEO_Y)
+END_GEO_X = _prepend_end(event_list.GEO_X)
+END_GEO_Y = _prepend_end(event_list.GEO_Y)
 HOSTNAME = event_list.HOSTNAME
 
 DATE_FORMAT = event_list.DATE_FORMAT

OTAnalytics/application/parser/track_parser.py

@@ -0,0 +1,54 @@
+from abc import ABC, abstractmethod
+from dataclasses import dataclass
+from pathlib import Path
+
+from OTAnalytics.domain.track_dataset.track_dataset import TrackDataset
+from OTAnalytics.domain.video import VideoMetadata
+
+
+@dataclass(frozen=True)
+class DetectionMetadata:
+    detection_classes: frozenset[str]
+
+
+@dataclass(frozen=True)
+class TrackParseResult:
+    tracks: TrackDataset
+    detection_metadata: DetectionMetadata
+    video_metadata: VideoMetadata
+
+
+@dataclass(frozen=True)
+class TracksParseResult:
+    tracks: TrackDataset
+    detections_metadata: list[DetectionMetadata]
+    videos_metadata: list[VideoMetadata]
+
+
+def combine_track_datasets(results: list[TrackParseResult]) -> TrackDataset:
+    if not results:
+        raise ValueError("No results to combine")
+    tracks = results[0].tracks
+    for result in results[1:]:
+        tracks = tracks.add_all(result.tracks)
+    return tracks
+
+
+class TrackParser(ABC):
+    def parse_files(self, files: list[Path]) -> TracksParseResult:
+        if not files:
+            raise ValueError("No files to parse")
+        results = [self.parse(file) for file in files]
+        tracks = self._combine_track_datasets(results)
+        detections_metadata = [result.detection_metadata for result in results]
+        videos_metadata = [result.video_metadata for result in results]
+        return TracksParseResult(tracks, detections_metadata, videos_metadata)
+
+    def _combine_track_datasets(
+        self, parse_results: list[TrackParseResult]
+    ) -> TrackDataset:
+        return combine_track_datasets(parse_results)
+
+    @abstractmethod
+    def parse(self, file: Path) -> TrackParseResult:
+        raise NotImplementedError

OTAnalytics/application/use_cases/load_track_files.py

@@ -1,7 +1,8 @@
 from pathlib import Path
 
-from OTAnalytics.application.datastore import TrackParser, VideoParser
+from OTAnalytics.application.datastore import VideoParser
 from OTAnalytics.application.logger import logger
+from OTAnalytics.application.parser.track_parser import TrackParser
 from OTAnalytics.application.state import TracksMetadata, VideosMetadata
 from OTAnalytics.domain.progress import ProgressbarBuilder
 from OTAnalytics.domain.track_repository import TrackFileRepository, TrackRepository

OTAnalytics/application/use_cases/road_user_assignment_export.py

@@ -97,6 +97,10 @@ def __create(self, assignment: RoadUserAssignment) -> dict:
             ras.START_INTERPOLATED_EVENT_COORD_Y: start.interpolated_event_coordinate.y,
             ras.END_INTERPOLATED_EVENT_COORD_X: end.interpolated_event_coordinate.x,
             ras.END_INTERPOLATED_EVENT_COORD_Y: end.interpolated_event_coordinate.y,
+            ras.START_GEO_X: start.geo_x,
+            ras.START_GEO_Y: start.geo_y,
+            ras.END_GEO_X: end.geo_x,
+            ras.END_GEO_Y: end.geo_y,
         }
 
     def reset(self) -> None:
@@ -140,6 +144,10 @@ def reset(self) -> None:
     ras.START_INTERPOLATED_EVENT_COORD_Y,
     ras.END_INTERPOLATED_EVENT_COORD_X,
     ras.END_INTERPOLATED_EVENT_COORD_Y,
+    ras.START_GEO_X,
+    ras.START_GEO_Y,
+    ras.END_GEO_X,
+    ras.END_GEO_Y,
 ]
 
 

OTAnalytics/domain/event.py

@@ -26,6 +26,8 @@
 VIDEO_NAME = "video_name"
 INTERPOLATED_OCCURRENCE = "interpolated_occurrence"
 INTERPOLATED_EVENT_COORDINATE = "interpolated_event_coordinate"
+GEO_X: str = "geo_x"
+GEO_Y: str = "geo_y"
 
 DATE_FORMAT: str = "%Y-%m-%d %H:%M:%S.%f"
 FILE_NAME_PATTERN = r"(?P<hostname>[A-Za-z0-9]+)_.*\..*"
@@ -80,6 +82,8 @@ class Event(DataclassValidation):
             event occurred.
         interpolated_event_coordinate (ImageCoordinate): interpolated event
             coordinate between two detections.
+        geo_x (float | None): optional geo x coordinate. Defaults to None.
+        geo_y (float | None): optional geo y coordinate. Defaults to None.
     """
 
     road_user_id: str
@@ -94,6 +98,8 @@ class Event(DataclassValidation):
     video_name: str
     interpolated_occurrence: datetime
     interpolated_event_coordinate: ImageCoordinate
+    geo_x: float | None = None
+    geo_y: float | None = None
 
     def _validate(self) -> None:
         self._validate_frame_number_greater_equal_one()
@@ -129,6 +135,8 @@ def to_dict(self) -> dict:
             VIDEO_NAME: self.video_name,
             INTERPOLATED_OCCURRENCE: self.interpolated_occurrence.strftime(DATE_FORMAT),
             INTERPOLATED_EVENT_COORDINATE: self.interpolated_event_coordinate.to_list(),
+            GEO_X: self.geo_x,
+            GEO_Y: self.geo_y,
         }
 
     def to_typed_dict(self) -> dict:
@@ -151,6 +159,8 @@ def to_typed_dict(self) -> dict:
             VIDEO_NAME: self.video_name,
             INTERPOLATED_OCCURRENCE: self.interpolated_occurrence,
             INTERPOLATED_EVENT_COORDINATE: self.interpolated_event_coordinate.to_list(),
+            GEO_X: self.geo_x,
+            GEO_Y: self.geo_y,
         }
 
     def _serialized_section_id(self) -> Optional[str]:
@@ -172,6 +182,8 @@ def __init__(self) -> None:
         self.section_id: Optional[SectionId] = None
         self.interpolated_occurrence: Optional[datetime] = None
         self.interpolated_event_coordinate: Optional[ImageCoordinate] = None
+        self.geo_x: float | None = None
+        self.geo_y: float | None = None
 
     @abstractmethod
     def create_event(self, detection: Detection) -> Event:
@@ -255,6 +267,16 @@ def add_interpolated_occurrence(self, occurrence: datetime) -> None:
     def add_interpolated_event_coordinate(self, x: float, y: float) -> None:
         self.interpolated_event_coordinate = ImageCoordinate(x, y)
 
+    def add_geo_coordinate(self, geo_x: float | None, geo_y: float | None) -> None:
+        """Add geo coordinates to the event to be built.
+
+        Args:
+            geo_x: the geo x coordinate, or None if unavailable.
+            geo_y: the geo y coordinate, or None if unavailable.
+        """
+        self.geo_x = geo_x
+        self.geo_y = geo_y
+
 
 class SectionEventBuilder(EventBuilder):
     """A builder to build section events."""
@@ -320,6 +342,8 @@ def create_event(self, detection: Detection) -> Event:
             video_name=detection.video_name,
             interpolated_occurrence=self.interpolated_occurrence,
             interpolated_event_coordinate=self.interpolated_event_coordinate,
+            geo_x=self.geo_x,
+            geo_y=self.geo_y,
         )
 
 
@@ -369,6 +393,8 @@ def create_event(self, detection: Detection) -> Event:
             video_name=detection.video_name,
             interpolated_occurrence=detection.occurrence,
             interpolated_event_coordinate=self.event_coordinate,
+            geo_x=self.geo_x,
+            geo_y=self.geo_y,
         )
 
 

OTAnalytics/domain/georeference.py

@@ -0,0 +1,56 @@
+"""Domain model for georeference Birds-Eye-View coordinate metadata."""
+
+from dataclasses import dataclass
+
+
+@dataclass(frozen=True)
+class GeoreferenceMetadata:
+    """Geo-referencing metadata from a georeferenced ottrk file.
+
+    Describes the affine mapping between Birds-Eye-View pixel coordinates and UTM
+    geo coordinates for a single ottrk output file.
+
+    Attributes:
+        geo_min_x (float): West boundary in UTM easting (metres).
+        geo_min_y (float): South boundary in UTM northing (metres).
+        geo_max_x (float): East boundary in UTM easting (metres).
+        geo_max_y (float): North boundary in UTM northing (metres).
+        birds_eye_view_width (int): Width of the BEV image in pixels.
+        birds_eye_view_height (int): Height of the BEV image in pixels.
+        padding (int): Pixel padding applied to all edges of the BEV image.
+        crs (str): Coordinate reference system as a WKT or authority string.
+    """
+
+    geo_min_x: float
+    geo_min_y: float
+    geo_max_x: float
+    geo_max_y: float
+    birds_eye_view_width: int
+    birds_eye_view_height: int
+    padding: int
+    crs: str
+
+
+def pixel_to_geo(
+    x: float, y: float, metadata: GeoreferenceMetadata
+) -> tuple[float, float]:
+    """Convert a Birds-Eye-View pixel coordinate to UTM geo coordinate.
+
+    Args:
+        x (float): Pixel x coordinate (column, increases rightward).
+        y (float): Pixel y coordinate (row, increases downward).
+        metadata (GeoreferenceMetadata): Metadata containing geo bounds and image size.
+
+    Returns:
+        Tuple[float, float] in the same UTM coordinate system as the per-detection
+        geo_x/geo_y fields.
+    """
+    scale_x = (metadata.geo_max_x - metadata.geo_min_x) / (
+        metadata.birds_eye_view_width - 2 * metadata.padding
+    )
+    scale_y = (metadata.geo_max_y - metadata.geo_min_y) / (
+        metadata.birds_eye_view_height - 2 * metadata.padding
+    )
+    geo_x = metadata.geo_min_x + (x - metadata.padding) * scale_x
+    geo_y = metadata.geo_max_y - (y - metadata.padding) * scale_y
+    return geo_x, geo_y

OTAnalytics/domain/track.py

@@ -20,6 +20,8 @@
 ORIGINAL_TRACK_ID: str = "original_track_id"
 VIDEO_NAME: str = "video_name"
 INPUT_FILE: str = "input_file"
+GEO_X: str = "geo_x"
+GEO_Y: str = "geo_y"
 
 
 @dataclass(frozen=True, order=True)
@@ -114,6 +116,16 @@ def video_name(self) -> str:
     def input_file(self) -> str:
         raise NotImplementedError
 
+    @property
+    @abstractmethod
+    def geo_x(self) -> float | None:
+        raise NotImplementedError
+
+    @property
+    @abstractmethod
+    def geo_y(self) -> float | None:
+        raise NotImplementedError
+
     def to_dict(self) -> dict:
         return {
             CLASSIFICATION: self.classification,