OP#9542: Flow distance, travel time, and avg speed in road user assignments CSV

.gitignore

@@ -1,6 +1,7 @@
 # OTAnalytics
 data/
 !tests/data
+tests/data_tmp/
 tests/data/OTCamera19_FR20_2023-05-24*
 tests/scripts/
 

OTAnalytics/application/export_formats/road_user_assignments.py

@@ -54,6 +54,9 @@ def _prepend_end(key: str) -> str:
     event_list.INTERPOLATED_EVENT_COORDINATE_Y
 )
 HOSTNAME = event_list.HOSTNAME
+FLOW_DISTANCE_M = "flow_distance_m"
+TRAVEL_TIME_S = "travel_time_s"
+AVG_SPEED_MPS = "avg_speed_mps"
 
 DATE_FORMAT = event_list.DATE_FORMAT
 TIME_FORMAT = event_list.TIME_FORMAT

OTAnalytics/application/use_cases/road_user_assignment_export.py

@@ -1,5 +1,6 @@
 from abc import ABC, abstractmethod
 from dataclasses import dataclass
+from datetime import datetime
 from pathlib import Path
 from typing import Callable, Iterable, Protocol, Self
 
@@ -24,6 +25,19 @@ class RoadUserAssignmentBuildError(Exception):
     pass
 
 
+def compute_road_user_assignment_flow_metrics(
+    flow_distance_m: float | None,
+    start_interpolated: datetime,
+    end_interpolated: datetime,
+) -> tuple[float | None, float, float | None]:
+    """Flow distance (m), travel time (s), and average speed (m/s) for CSV export."""
+    travel_time_s = (end_interpolated - start_interpolated).total_seconds()
+    avg_speed_mps: float | None = None
+    if flow_distance_m is not None and travel_time_s > 0:
+        avg_speed_mps = flow_distance_m / travel_time_s
+    return flow_distance_m, travel_time_s, avg_speed_mps
+
+
 class RoadUserAssignmentBuilder:
     def __init__(self) -> None:
         self._start_section: Section | None = None
@@ -57,6 +71,13 @@ def __create(self, assignment: RoadUserAssignment) -> dict:
         assigned_flow = assignment.assignment
         start = assignment.events.start
         end = assignment.events.end
+        flow_distance_m, travel_time_s, avg_speed_mps = (
+            compute_road_user_assignment_flow_metrics(
+                assigned_flow.distance,
+                start.interpolated_occurrence,
+                end.interpolated_occurrence,
+            )
+        )
         return {
             ras.FLOW_ID: assigned_flow.id.id,
             ras.FLOW_NAME: assigned_flow.name,
@@ -96,6 +117,9 @@ 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.FLOW_DISTANCE_M: flow_distance_m,
+            ras.TRAVEL_TIME_S: travel_time_s,
+            ras.AVG_SPEED_MPS: avg_speed_mps,
         }
 
     def reset(self) -> None:
@@ -139,6 +163,9 @@ def reset(self) -> None:
     ras.START_INTERPOLATED_EVENT_COORD_Y,
     ras.END_INTERPOLATED_EVENT_COORD_X,
     ras.END_INTERPOLATED_EVENT_COORD_Y,
+    ras.FLOW_DISTANCE_M,
+    ras.TRAVEL_TIME_S,
+    ras.AVG_SPEED_MPS,
 ]
 
 

tests/data/Testvideo_Cars-Cyclist_FR20_2020-01-01_00-00-00.otconfig

@@ -176,7 +176,7 @@
       "name": "S\u00fcd --> Nord",
       "start": "1",
       "end": "3",
-      "distance": 0.0
+      "distance": 32.5
     }
   ],
   "remark": ""

tests/data/Testvideo_Cars-Cyclist_FR20_2020-01-01_00-00-00.otflow

@@ -144,7 +144,7 @@
             "name": "S\u00fcd --> Nord",
             "start": "1",
             "end": "3",
-            "distance": 0.0
+            "distance": 32.5
         }
     ]
 }

tests/data/sections_created_test_file.otconfig

@@ -80,7 +80,7 @@
             "name": "Test-Flow",
             "start": "2",
             "end": "1",
-            "distance": null
+            "distance": 42.0
         }
     ],
     "remark": ""

tests/unit/OTAnalytics/application/use_cases/test_road_user_assignment_export.py

@@ -1,14 +1,17 @@
+from datetime import datetime, timedelta, timezone
 from unittest import mock
 from unittest.mock import Mock, call
 
 import pytest
 
 from OTAnalytics.application.analysis.road_user_assignment import (
+    EventPair,
     RoadUserAssigner,
     RoadUserAssignment,
     RoadUserAssignmentRepository,
     RoadUserAssignments,
 )
+from OTAnalytics.application.export_formats import road_user_assignments as ras
 from OTAnalytics.application.export_formats.export_mode import OVERWRITE
 from OTAnalytics.application.use_cases.assignment_repository import (
     GetRoadUserAssignments,
@@ -24,10 +27,13 @@
     RoadUserAssignmentBuildError,
     RoadUserAssignmentExporter,
     RoadUserAssignmentExporterFactory,
+    compute_road_user_assignment_flow_metrics,
 )
+from OTAnalytics.domain.flow import Flow, FlowId
 from OTAnalytics.domain.section import Section
 from OTAnalytics.domain.track_dataset.track_dataset import TrackIdSetFactory
 from OTAnalytics.domain.types import EventType
+from tests.utils.builders.event_builder import EventBuilder
 from tests.utils.builders.road_user_assignment import create_road_user_assignment
 
 
@@ -93,6 +99,133 @@ def test_build_with_max_confidence_missing(
         ):
             _builder.build(Mock())
 
+    def test_build_avg_speed_when_flow_distance_and_positive_duration(
+        self,
+        _builder: RoadUserAssignmentBuilder,
+        first_line_section: Section,
+        second_line_section: Section,
+    ) -> None:
+        t0 = datetime(2020, 1, 1, tzinfo=timezone.utc)
+        t1 = t0 + timedelta(seconds=10)
+        start_event = EventBuilder(
+            section_id=first_line_section.id.id,
+            interpolated_occurrence=t0,
+        ).build_section_event()
+        end_event = EventBuilder(
+            section_id=second_line_section.id.id,
+            interpolated_occurrence=t1,
+        ).build_section_event()
+        flow = Flow(
+            FlowId("f-speed"),
+            "f-speed",
+            first_line_section.id,
+            second_line_section.id,
+            distance=25.0,
+        )
+        assignment = RoadUserAssignment(
+            "1", "car", flow, EventPair(start_event, end_event)
+        )
+        _builder.add_start_section(first_line_section)
+        _builder.add_end_section(second_line_section)
+        _builder.add_max_confidence(0.9)
+        result = _builder.build(assignment)
+        assert result[ras.FLOW_DISTANCE_M] == 25.0
+        assert result[ras.TRAVEL_TIME_S] == 10.0
+        assert result[ras.AVG_SPEED_MPS] == 2.5
+
+    def test_build_no_avg_speed_when_zero_travel_time(
+        self,
+        _builder: RoadUserAssignmentBuilder,
+        first_line_section: Section,
+        second_line_section: Section,
+    ) -> None:
+        t0 = datetime(2020, 1, 1, tzinfo=timezone.utc)
+        start_event = EventBuilder(
+            section_id=first_line_section.id.id,
+            interpolated_occurrence=t0,
+        ).build_section_event()
+        end_event = EventBuilder(
+            section_id=second_line_section.id.id,
+            interpolated_occurrence=t0,
+        ).build_section_event()
+        flow = Flow(
+            FlowId("f-zero"),
+            "f-zero",
+            first_line_section.id,
+            second_line_section.id,
+            distance=10.0,
+        )
+        assignment = RoadUserAssignment(
+            "1", "car", flow, EventPair(start_event, end_event)
+        )
+        _builder.add_start_section(first_line_section)
+        _builder.add_end_section(second_line_section)
+        _builder.add_max_confidence(0.9)
+        result = _builder.build(assignment)
+        assert result[ras.FLOW_DISTANCE_M] == 10.0
+        assert result[ras.TRAVEL_TIME_S] == 0.0
+        assert result[ras.AVG_SPEED_MPS] is None
+
+    def test_build_avg_speed_zero_when_flow_distance_is_zero(
+        self,
+        _builder: RoadUserAssignmentBuilder,
+        first_line_section: Section,
+        second_line_section: Section,
+    ) -> None:
+        t0 = datetime(2020, 1, 1, tzinfo=timezone.utc)
+        t1 = t0 + timedelta(seconds=4)
+        start_event = EventBuilder(
+            section_id=first_line_section.id.id,
+            interpolated_occurrence=t0,
+        ).build_section_event()
+        end_event = EventBuilder(
+            section_id=second_line_section.id.id,
+            interpolated_occurrence=t1,
+        ).build_section_event()
+        flow = Flow(
+            FlowId("f-zero-dist"),
+            "f-zero-dist",
+            first_line_section.id,
+            second_line_section.id,
+            distance=0.0,
+        )
+        assignment = RoadUserAssignment(
+            "1", "car", flow, EventPair(start_event, end_event)
+        )
+        _builder.add_start_section(first_line_section)
+        _builder.add_end_section(second_line_section)
+        _builder.add_max_confidence(0.9)
+        result = _builder.build(assignment)
+        assert result[ras.FLOW_DISTANCE_M] == 0.0
+        assert result[ras.TRAVEL_TIME_S] == 4.0
+        assert result[ras.AVG_SPEED_MPS] == 0.0
+
+
+class TestComputeRoadUserAssignmentFlowMetrics:
+    def test_no_distance_leaves_speed_none(self) -> None:
+        t0 = datetime(2020, 1, 1, tzinfo=timezone.utc)
+        t1 = t0 + timedelta(seconds=2)
+        d, dt, v = compute_road_user_assignment_flow_metrics(None, t0, t1)
+        assert d is None
+        assert dt == 2.0
+        assert v is None
+
+    def test_distance_and_positive_time_gives_speed(self) -> None:
+        t0 = datetime(2020, 1, 1, tzinfo=timezone.utc)
+        t1 = t0 + timedelta(seconds=4)
+        d, dt, v = compute_road_user_assignment_flow_metrics(8.0, t0, t1)
+        assert d == 8.0
+        assert dt == 4.0
+        assert v == 2.0
+
+    def test_zero_distance_with_positive_time_gives_zero_speed(self) -> None:
+        t0 = datetime(2020, 1, 1, tzinfo=timezone.utc)
+        t1 = t0 + timedelta(seconds=2)
+        d, dt, v = compute_road_user_assignment_flow_metrics(0.0, t0, t1)
+        assert d == 0.0
+        assert dt == 2.0
+        assert v == 0.0
+
 
 class TestExportRoadUserAssignments:
     def test_export(self) -> None:

tests/unit/OTAnalytics/plugin_parser/test_road_user_assignment_export.py

@@ -1,9 +1,14 @@
+import csv
+from datetime import datetime, timedelta, timezone
+from io import StringIO
 from pathlib import Path
 from unittest.mock import Mock
 
+import pandas as pd
 from pandas import DataFrame, read_csv
 
 from OTAnalytics.application.analysis.road_user_assignment import (
+    EventPair,
     RoadUserAssignment,
     RoadUserAssignments,
 )
@@ -12,10 +17,12 @@
 from OTAnalytics.application.use_cases.road_user_assignment_export import (
     RoadUserAssignmentBuilder,
 )
+from OTAnalytics.domain.flow import Flow, FlowId
 from OTAnalytics.domain.section import Section
 from OTAnalytics.plugin_parser.road_user_assignment_export import (
     RoadUserAssignmentCsvExporter,
 )
+from tests.utils.builders.event_builder import EventBuilder
 from tests.utils.builders.road_user_assignment import create_road_user_assignment
 
 
@@ -79,4 +86,67 @@ def test_export(
         actual[ras.START_SECTION_NAME] = actual[ras.START_SECTION_NAME].astype(str)
         actual[ras.END_SECTION_NAME] = actual[ras.END_SECTION_NAME].astype(str)
 
+        for col in (ras.FLOW_DISTANCE_M, ras.TRAVEL_TIME_S, ras.AVG_SPEED_MPS):
+            expected[col] = expected[col].astype("float64")
+            actual[col] = actual[col].astype("float64")
+
         assert actual.equals(expected)
+
+    def test_export_writes_empty_cells_when_flow_distance_not_configured(
+        self,
+        test_data_tmp_dir: Path,
+        first_line_section: Section,
+        second_line_section: Section,
+    ) -> None:
+        """Unset flow.distance must yield blank CSV cells for distance and avg speed."""
+        save_path = test_data_tmp_dir / "road_user_assignments_no_distance.csv"
+        t0 = datetime(2020, 1, 1, tzinfo=timezone.utc)
+        t1 = t0 + timedelta(seconds=6)
+        start_event = EventBuilder(
+            section_id=first_line_section.id.id,
+            interpolated_occurrence=t0,
+        ).build_section_event()
+        end_event = EventBuilder(
+            section_id=second_line_section.id.id,
+            interpolated_occurrence=t1,
+        ).build_section_event()
+        flow = Flow(
+            FlowId("no-dist"),
+            "no-dist",
+            first_line_section.id,
+            second_line_section.id,
+            distance=None,
+        )
+        assignment = RoadUserAssignment(
+            "track-1", "car", flow, EventPair(start_event, end_event)
+        )
+
+        mock_factory = Mock()
+        section_repository = Mock()
+        get_all_tracks = Mock()
+        builder = RoadUserAssignmentBuilder()
+        track_dataset = Mock()
+        track_dataset.get_max_confidences_for.return_value = {"track-1": 0.85}
+        get_all_tracks.as_dataset.return_value = track_dataset
+        section_repository.get.side_effect = [first_line_section, second_line_section]
+
+        exporter = RoadUserAssignmentCsvExporter(
+            section_repository, get_all_tracks, builder, save_path
+        )
+        exporter.export(
+            RoadUserAssignments([assignment], mock_factory), OVERWRITE
+        )
+
+        text = save_path.read_text(encoding="utf-8")
+        rows = list(csv.reader(StringIO(text)))
+        assert len(rows) == 2
+        header_cells, data_cells = rows
+        idx_dist = header_cells.index(ras.FLOW_DISTANCE_M)
+        idx_speed = header_cells.index(ras.AVG_SPEED_MPS)
+        assert data_cells[idx_dist] == ""
+        assert data_cells[idx_speed] == ""
+
+        df = read_csv(save_path)
+        assert df[ras.TRAVEL_TIME_S].iloc[0] == 6.0
+        assert pd.isna(df[ras.FLOW_DISTANCE_M].iloc[0])
+        assert pd.isna(df[ras.AVG_SPEED_MPS].iloc[0])