Merge remote-tracking branch 'origin/bug/8343-await-all-methods-decorated-with-actions' into bug/8343-await-all-methods-decorated-with-actions

Author: briemla

OTAnalytics/application/config.py

@@ -6,7 +6,7 @@
 DEFAULT_LOG_DIR = Path("logs").absolute()
 """The log save directory."""
 
-ALLOWED_TRACK_SIZE_PARSING = 5
+ALLOWED_TRACK_SIZE_PARSING = 0
 TRACK_LENGTH_LIMIT = 12000
 GEOMETRY_CACHE_SIZE: int = 20000
 CUTTING_SECTION_MARKER: str = "#cut"

OTAnalytics/application/export_formats/event_list.py

@@ -18,6 +18,7 @@
 DIRECTION_VECTOR_Y = "vector_px_y"
 VIDEO_NAME = event.VIDEO_NAME
 INTERPOLATED_OCCURRENCE = event.INTERPOLATED_OCCURRENCE
+INTERPOLATED_EVENT_COORDINATE = event.INTERPOLATED_EVENT_COORDINATE
 INTERPOLATED_EVENT_COORDINATE_X = "interpolated_coordinate_px_x"
 INTERPOLATED_EVENT_COORDINATE_Y = "interpolated_coordinate_px_y"
 

OTAnalytics/plugin_prototypes/eventlist_exporter/eventlist_exporter.py

@@ -5,6 +5,27 @@
 from OTAnalytics.application.config import DEFAULT_EVENTLIST_FILE_TYPE
 from OTAnalytics.application.datastore import EventListParser
 from OTAnalytics.application.export_formats import event_list
+from OTAnalytics.application.export_formats.event_list import (
+    DATE_TIME_FORMAT,
+    DIRECTION_VECTOR_X,
+    DIRECTION_VECTOR_Y,
+    EVENT_COORDINATE_X,
+    EVENT_COORDINATE_Y,
+    EVENT_TYPE,
+    FRAME_NUMBER,
+    HOSTNAME,
+    INTERPOLATED_EVENT_COORDINATE_X,
+    INTERPOLATED_EVENT_COORDINATE_Y,
+    INTERPOLATED_OCCURRENCE,
+    OCCURRENCE,
+    OCCURRENCE_DATE,
+    OCCURRENCE_TIME,
+    ROAD_USER_ID,
+    ROAD_USER_TYPE,
+    SECTION_ID,
+    SECTION_NAME,
+    VIDEO_NAME,
+)
 from OTAnalytics.application.export_formats.export_mode import INITIAL_MERGE
 from OTAnalytics.application.logger import logger
 from OTAnalytics.application.use_cases.export_events import (
@@ -24,6 +45,44 @@
 OTC_CSV_FORMAT_NAME = "CSV (OpenTrafficCam)"
 OTC_OTEVENTS_FORMAT_NAME = "OTEvents (OpenTrafficCam)"
 
+OCCURRENCE_SEC = f"{OCCURRENCE}_sec"
+
+NUMBER_ROUNDED_COLUMNS = {
+    event_list.EVENT_COORDINATE_X: 1,
+    event_list.EVENT_COORDINATE_Y: 1,
+    event_list.INTERPOLATED_EVENT_COORDINATE_X: 1,
+    event_list.INTERPOLATED_EVENT_COORDINATE_Y: 1,
+    event_list.DIRECTION_VECTOR_X: 4,
+    event_list.DIRECTION_VECTOR_Y: 4,
+}
+
+DATETIME_ROUNDED_COLUMNS = {
+    event_list.OCCURRENCE: "ms",
+    event_list.INTERPOLATED_OCCURRENCE: "ms",
+}
+
+EXPORT_COLUMNS = [
+    ROAD_USER_ID,
+    ROAD_USER_TYPE,
+    HOSTNAME,
+    OCCURRENCE,
+    FRAME_NUMBER,
+    SECTION_ID,
+    EVENT_TYPE,
+    VIDEO_NAME,
+    OCCURRENCE_SEC,
+    EVENT_COORDINATE_X,
+    EVENT_COORDINATE_Y,
+    DIRECTION_VECTOR_X,
+    DIRECTION_VECTOR_Y,
+    SECTION_NAME,
+    OCCURRENCE_DATE,
+    OCCURRENCE_TIME,
+    INTERPOLATED_OCCURRENCE,
+    INTERPOLATED_EVENT_COORDINATE_X,
+    INTERPOLATED_EVENT_COORDINATE_Y,
+]
+
 
 class EventListDataFrameBuilder:
     def __init__(self, events: Iterable[Event], sections: Iterable[Section]):
@@ -39,7 +98,20 @@ def build(self) -> pd.DataFrame:
         self._split_columns_with_lists()
         self._add_section_names()
         self._add_detailed_date_time_columns()
-        return self._df
+        self._round()
+        return self._df.loc[:, EXPORT_COLUMNS]
+
+    def _round(self) -> None:
+        for column, decimals in NUMBER_ROUNDED_COLUMNS.items():
+            if column in self._df.columns:
+                self._df.loc[:, column] = self._df.loc[:, column].round(decimals)
+        for column, freq in DATETIME_ROUNDED_COLUMNS.items():
+            if column in self._df.columns:
+                self._df.loc[:, column] = (
+                    pd.to_datetime(self._df.loc[:, column])
+                    .dt.round(freq)
+                    .dt.strftime(DATE_TIME_FORMAT)
+                )
 
     def _add_detailed_date_time_columns(self) -> None:
         occurrence_column = pd.to_datetime(self._df[event_list.OCCURRENCE])
@@ -69,8 +141,21 @@ def _split_columns_with_lists(self) -> None:
                 self._df[event_list.DIRECTION_VECTOR].tolist(), index=self._df.index
             )
         )
+        self._df[
+            [
+                event_list.INTERPOLATED_EVENT_COORDINATE_X,
+                event_list.INTERPOLATED_EVENT_COORDINATE_Y,
+            ]
+        ] = pd.DataFrame(
+            self._df[event_list.INTERPOLATED_EVENT_COORDINATE].tolist(),
+            index=self._df.index,
+        )
         self._df = self._df.drop(
-            columns=[event_list.EVENT_COORDINATE, event_list.DIRECTION_VECTOR]
+            columns=[
+                event_list.EVENT_COORDINATE,
+                event_list.DIRECTION_VECTOR,
+                event_list.INTERPOLATED_EVENT_COORDINATE,
+            ]
         )
 
     def _add_section_names(self) -> None:

pyproject.toml

@@ -35,6 +35,7 @@ dependencies = [
     "openpyxl==3.1.5",
     "pandas==2.2.3",
     "pillow==11.1.0",
+    "polars==1.32.3",
     "pywin32==310; sys_platform == 'win32'",
     "requests==2.32.4",
     "seaborn==0.13.2",

tests/OTAnalytics/plugin_parser/test_streaming_parser.py

@@ -168,8 +168,8 @@ def test_parse_chunks(
         second_chunk = next(stream).as_list()
 
         assert len(first_chunk) == 4
-        assert len(second_chunk) == 2
-        assert len(expected) == 6
+        assert len(second_chunk) == 3
+        assert len(expected) == 7
         assert set(self.ids_of(expected)) == set(
             self.ids_of(first_chunk + second_chunk)
         )

tests/regression_otanalytics.py

@@ -8,7 +8,11 @@
 from OTAnalytics.application.parser.flow_parser import FlowParser
 from OTAnalytics.plugin_cli.cli_application import OtAnalyticsCliApplicationStarter
 from OTAnalytics.plugin_parser.otvision_parser import OtFlowParser
-from tests.utils.assertions import assert_two_files_equal_sorted
+from tests.utils.assertions import (
+    assert_equal_count_files,
+    assert_equal_event_files,
+    assert_two_files_equal_sorted,
+)
 from tests.utils.builders.run_configuration import create_run_config
 
 
@@ -48,7 +52,7 @@ def otflow_file(test_data_dir: Path) -> str:
 
 @pytest.fixture(scope="module")
 def cli_chunk_size() -> int:
-    return 5
+    return 1000
 
 
 @pytest.fixture
@@ -59,16 +63,11 @@ def otflow_parser() -> FlowParser:
 class TestRegressionCompleteApplication:
 
     @pytest.mark.skip
-    @pytest.mark.parametrize(
-        "cli_mode",
-        [CliMode.STREAM, CliMode.BULK],
-    )
     def test_15_min_recreate_test_data(
         self,
         otflow_file: str,
         all_track_files_test_dataset: list[Path],
         otflow_parser: FlowParser,
-        cli_mode: CliMode,
         cli_chunk_size: int,
     ) -> None:
         for test_file in tqdm(all_track_files_test_dataset, desc="test data file"):
@@ -83,7 +82,7 @@ def test_15_min_recreate_test_data(
                 test_interval=test_interval,
                 otflow_parser=otflow_parser,
                 event_formats=("csv", "otevents"),
-                cli_mode=cli_mode,
+                cli_mode=CliMode.BULK,
                 cli_chunk_size=cli_chunk_size,
             )
 
@@ -268,14 +267,6 @@ def _execute_test(
             cli_chunk_size,
         )
 
-        actual_events_file = Path(test_data_tmp_dir / save_name).with_suffix(
-            ".events.csv"
-        )
-        expected_events_file = Path(test_data_dir / save_name).with_suffix(
-            ".events.csv"
-        )
-        assert_two_files_equal_sorted(actual_events_file, expected_events_file)
-
         actual_counts_file = (
             Path(test_data_tmp_dir / save_name)
             .with_suffix(f".counts_{count_interval}min.csv")
@@ -286,7 +277,16 @@ def _execute_test(
             .with_suffix(f".counts_{count_interval}min.csv")
             .absolute()
         )
-        assert_two_files_equal_sorted(actual_counts_file, expected_counts_file)
+        assert_equal_count_files(actual_counts_file, expected_counts_file)
+
+        actual_events_file = Path(test_data_tmp_dir / save_name).with_suffix(
+            ".events.csv"
+        )
+        expected_events_file = Path(test_data_dir / save_name).with_suffix(
+            ".events.csv"
+        )
+        assert_two_files_equal_sorted(actual_events_file, expected_events_file)
+        assert_equal_event_files(actual_events_file, expected_events_file)
 
     def _run_otanalytics(
         self,

tests/utils/assertions.py

@@ -2,8 +2,12 @@
 from typing import Iterable
 from unittest.mock import Mock
 
+import polars as pl
+from polars.testing import assert_frame_equal
+
 from OTAnalytics.domain.track import Detection, Track
 from OTAnalytics.domain.track_dataset.track_dataset import TrackDataset
+from OTAnalytics.plugin_prototypes.eventlist_exporter import eventlist_exporter
 
 
 def assert_equal_detection_properties(actual: Detection, expected: Detection) -> None:
@@ -66,12 +70,10 @@ def assert_two_files_equal_sorted(
 ) -> None:
 
     with open(actual_counts_file, mode="r") as actual:
-        act_lines = sorted(actual.readlines())
-        act_lines = sorted([line.split(",", 1)[1] for line in actual.readlines()])
+        act_lines = sorted([line for line in actual.readlines()])
 
         with open(expected_counts_file, mode="r") as expected:
-            exp_lines = sorted(expected.readlines())
-            exp_lines = sorted([line.split(",", 1)[1] for line in expected.readlines()])
+            exp_lines = sorted([line for line in expected.readlines()])
 
             assert set(act_lines) == set(exp_lines), (
                 f"Sets are not equal [act len: {len(act_lines)}, "
@@ -99,3 +101,28 @@ def assert_two_files_equal_sorted(
                 f"{([a for e, a in zip(exp_lines, act_lines) if e != a] + [None])[0]}\n"
                 f"\n"
             )
+
+
+def assert_equal_event_files(
+    actual_counts_file: Path, expected_counts_file: Path
+) -> None:
+    actual = _read_event_csv(actual_counts_file)
+    expected = _read_event_csv(expected_counts_file)
+    assert_frame_equal(actual, expected)
+
+
+def assert_equal_count_files(
+    actual_counts_file: Path, expected_counts_file: Path
+) -> None:
+    actual = _read_counts_csv(actual_counts_file)
+    expected = _read_counts_csv(expected_counts_file)
+    assert_frame_equal(actual, expected)
+
+
+def _read_counts_csv(file: Path) -> pl.DataFrame:
+    return pl.read_csv(file)
+
+
+def _read_event_csv(file: Path) -> pl.DataFrame:
+    data = pl.read_csv(file)
+    return data.sort(by=eventlist_exporter.EXPORT_COLUMNS)