Merge pull request #791 from OpenTrafficCam/feature/8954-change-strategy-to-parse-ottrk-files-by-files-in-streamottrkparser-2

Feature/8954 change strategy to parse ottrk files by files in streamottrkparser 2

Author: randy-seng

OTAnalytics/application/parser/cli_parser.py

@@ -33,7 +33,6 @@ class CliParseError(Exception):
 class CliArguments:
     start_cli: bool
     cli_mode: CliMode
-    cli_chunk_size: int
     debug: bool
     logfile_overwrite: bool
     track_export: bool

OTAnalytics/application/run_configuration.py

@@ -70,10 +70,6 @@ def cli_bulk_mode(self) -> bool:
     def cli_stream_mode(self) -> bool:
         return self.start_cli and self._cli_args.cli_mode == CliMode.STREAM
 
-    @property
-    def cli_chunk_size(self) -> int:
-        return self._cli_args.cli_chunk_size
-
     @property
     def debug(self) -> bool:
         return self._cli_args.debug

OTAnalytics/domain/track.py

@@ -94,6 +94,11 @@ def occurrence(self) -> datetime:
     def interpolated_detection(self) -> bool:
         raise NotImplementedError
 
+    @property
+    @abstractmethod
+    def is_finished(self) -> bool:
+        raise NotImplementedError
+
     @property
     @abstractmethod
     def track_id(self) -> TrackId:

OTAnalytics/domain/track_dataset/filtered_track_dataset.py

@@ -77,6 +77,9 @@ def get_last_segments(self) -> TrackSegmentDataset:
     def get_max_confidences_for(self, track_ids: TrackIdSet) -> dict[str, float]:
         return self._filter().get_max_confidences_for(track_ids)
 
+    def split_finished(self) -> tuple[TrackDataset, TrackDataset]:
+        return self._filter().split_finished()
+
 
 class FilterByClassTrackDataset(FilteredTrackDataset):
     @property

OTAnalytics/domain/track_dataset/track_dataset.py

@@ -212,6 +212,16 @@ def clear(self) -> "TrackDataset":
         """
         raise NotImplementedError
 
+    @abstractmethod
+    def split_finished(self) -> tuple["TrackDataset", "TrackDataset"]:
+        """
+        Split this dataset into finished and remaining tracks.
+
+        Returns:
+            tuple[TrackDataset, TrackDataset]: (finished_tracks, remaining_tracks)
+        """
+        raise NotImplementedError
+
     @abstractmethod
     def as_list(self) -> list[Track]:
         raise NotImplementedError

OTAnalytics/plugin_datastore/polars_track_store.py

@@ -106,6 +106,12 @@ def occurrence(self) -> datetime:
     def interpolated_detection(self) -> bool:
         return self.__get_attribute(track.INTERPOLATED_DETECTION)
 
+    @property
+    def is_finished(self) -> bool:
+        if ottrk_dataformat.FINISHED not in self._data:
+            return False
+        return bool(self.__get_attribute(ottrk_dataformat.FINISHED))
+
     @property
     def track_id(self) -> TrackId:
         return TrackId(self._track_id)
@@ -445,6 +451,36 @@ def get_for(self, id: TrackId) -> Optional[Track]:
     def clear(self) -> "PolarsTrackDataset":
         return PolarsTrackDataset(self.track_geometry_factory)
 
+    def split_finished(self) -> tuple[TrackDataset, TrackDataset]:
+        empty = PolarsTrackDataset(
+            self.track_geometry_factory, calculator=self.calculator
+        )
+        if self._dataset.is_empty():
+            return empty, empty
+        if ottrk_dataformat.FINISHED not in self._dataset.columns:
+            return self, empty
+
+        finished_ids = (
+            self._dataset.filter(
+                pl.col(ottrk_dataformat.FINISHED).fill_null(False).eq(True)
+            )
+            .get_column(LEVEL_TRACK_ID)
+            .unique()
+            .to_list()
+        )
+        if not finished_ids:
+            return empty, self
+
+        all_ids = self._dataset.get_column(LEVEL_TRACK_ID).unique().to_list()
+        finished_id_set = set(finished_ids)
+        remaining_ids = [
+            track_id for track_id in all_ids if track_id not in finished_id_set
+        ]
+
+        finished_dataset = self._subset_by_ids(finished_ids)
+        remaining_dataset = self._subset_by_ids(remaining_ids)
+        return finished_dataset, remaining_dataset
+
     def _create_track_flyweight(self, track_id: str) -> Track:
         """Create a Track flyweight object for the given track_id."""
         track_data = self._dataset.filter(pl.col(LEVEL_TRACK_ID) == track_id)
@@ -830,6 +866,20 @@ def split(self, batches: int) -> Sequence["PolarsTrackDataset"]:
             )
         return new_batches
 
+    def _subset_by_ids(self, track_ids: list[str]) -> "PolarsTrackDataset":
+        if not track_ids:
+            return PolarsTrackDataset(
+                self.track_geometry_factory, calculator=self.calculator
+            )
+        subset = self._dataset.filter(pl.col(LEVEL_TRACK_ID).is_in(track_ids))
+        geometries = self._get_geometries_for(track_ids)
+        return PolarsTrackDataset.from_dataframe(
+            subset,
+            self.track_geometry_factory,
+            geometries,
+            calculator=self.calculator,
+        )
+
     def _get_geometries_for(
         self, track_ids: list[str]
     ) -> dict[RelativeOffsetCoordinate, PolarsTrackGeometryDataset]:
@@ -885,6 +935,8 @@ def create_empty_dataframe() -> pl.DataFrame:
     schema[track.FRAME] = pl.Int64
     schema[track.CONFIDENCE] = pl.Float64
     schema[track.INTERPOLATED_DETECTION] = pl.Boolean
+    schema[ottrk_dataformat.FIRST] = pl.Boolean
+    schema[ottrk_dataformat.FINISHED] = pl.Boolean
     schema[track.OCCURRENCE] = pl.Datetime
 
     return pl.DataFrame(schema=schema)
@@ -909,7 +961,7 @@ def _convert_tracks(tracks: Iterable[Track]) -> pl.DataFrame:
             dto = detection.to_dict()
             dto[track.ORIGINAL_TRACK_ID] = current_track.original_id.id
             dto[ottrk_dataformat.FIRST] = current_track.first_detection == detection
-            dto[ottrk_dataformat.FINISHED] = current_track.last_detection == detection
+            dto[ottrk_dataformat.FINISHED] = detection.is_finished
             prepared.append(dto)
 
     if not prepared:

OTAnalytics/plugin_datastore/python_track_store.py

@@ -93,6 +93,7 @@ class PythonDetection(Detection, DataclassValidation):
     _track_id: TrackId
     _video_name: str
     _input_file: str
+    _is_finished: bool = False
 
     @property
     def classification(self) -> str:
@@ -130,6 +131,10 @@ def occurrence(self) -> datetime:
     def interpolated_detection(self) -> bool:
         return self._interpolated_detection
 
+    @property
+    def is_finished(self) -> bool:
+        return self._is_finished
+
     @property
     def track_id(self) -> TrackId:
         return self._track_id
@@ -541,6 +546,43 @@ def _remove_from_geometry_datasets(
     def clear(self) -> "PythonTrackDataset":
         return PythonTrackDataset(self.track_geometry_factory)
 
+    def split_finished(self) -> tuple[TrackDataset, TrackDataset]:
+        empty = PythonTrackDataset(
+            self.track_geometry_factory, calculator=self.calculator
+        )
+        if not self._tracks:
+            return empty, empty
+
+        finished_tracks: dict[TrackId, Track] = {}
+        remaining_tracks: dict[TrackId, Track] = {}
+        for track_id, track in self._tracks.items():
+            if track.last_detection.is_finished:
+                finished_tracks[track_id] = track
+            else:
+                remaining_tracks[track_id] = track
+
+        finished_dataset = (
+            PythonTrackDataset(
+                self.track_geometry_factory,
+                finished_tracks,
+                self._get_geometries_for(finished_tracks.keys()),
+                calculator=self.calculator,
+            )
+            if finished_tracks
+            else empty
+        )
+        remaining_dataset = (
+            PythonTrackDataset(
+                self.track_geometry_factory,
+                remaining_tracks,
+                self._get_geometries_for(remaining_tracks.keys()),
+                calculator=self.calculator,
+            )
+            if remaining_tracks
+            else empty
+        )
+        return finished_dataset, remaining_dataset
+
     def as_list(self) -> list[Track]:
         return list(self._tracks.values())
 

OTAnalytics/plugin_datastore/track_store.py

@@ -94,6 +94,13 @@ def occurrence(self) -> datetime:
     def interpolated_detection(self) -> bool:
         return self.__get_attribute(track.INTERPOLATED_DETECTION)
 
+    @property
+    def is_finished(self) -> bool:
+        if ottrk_dataformat.FINISHED not in self._data.index:
+            return False
+        value = self.__get_attribute(ottrk_dataformat.FINISHED)
+        return bool(value)
+
     @property
     def track_id(self) -> TrackId:
         return TrackId(self._track_id)
@@ -403,6 +410,38 @@ def get_for(self, id: TrackId) -> Optional[Track]:
     def clear(self) -> "PandasTrackDataset":
         return PandasTrackDataset(self.track_geometry_factory)
 
+    def split_finished(self) -> tuple[TrackDataset, TrackDataset]:
+        empty = PandasTrackDataset(
+            self.track_geometry_factory, calculator=self.calculator
+        )
+        if self._dataset.empty:
+            return empty, empty
+        if ottrk_dataformat.FINISHED not in self._dataset.columns:
+            return self, empty
+
+        finished_mask = (
+            self._dataset[ottrk_dataformat.FINISHED].fillna(False).astype(bool).eq(True)
+        )
+        finished_ids = (
+            self._dataset.loc[finished_mask]
+            .index.get_level_values(LEVEL_TRACK_ID)
+            .unique()
+        )
+        finished_ids_list = list(finished_ids)
+        if not finished_ids_list:
+            return empty, self
+
+        index = self.get_index()
+        all_ids = list(index) if index is not None else []
+        finished_id_set = set(finished_ids_list)
+        remaining_ids_list = [
+            track_id for track_id in all_ids if track_id not in finished_id_set
+        ]
+
+        finished_dataset = self._subset_by_ids(finished_ids_list)
+        remaining_dataset = self._subset_by_ids(remaining_ids_list)
+        return finished_dataset, remaining_dataset
+
     def remove(self, track_id: TrackId) -> "PandasTrackDataset":
         remaining_tracks = self._dataset.drop(unpack(track_id), errors="ignore")
         updated_geometry_datasets = self._remove_from_geometry_dataset([track_id.id])
@@ -471,6 +510,20 @@ def split(self, batches: int) -> Sequence["PandasTrackDataset"]:
 
         return [self]
 
+    def _subset_by_ids(self, track_ids: list[str]) -> "PandasTrackDataset":
+        if not track_ids:
+            return PandasTrackDataset(
+                self.track_geometry_factory, calculator=self.calculator
+            )
+        subset = self._dataset.loc[track_ids]
+        geometries = self._get_geometries_for(track_ids)
+        return PandasTrackDataset.from_dataframe(
+            subset,
+            self.track_geometry_factory,
+            geometries,
+            calculator=self.calculator,
+        )
+
     def get_index(self) -> Index | None:
         if self._dataset.empty:
             return None
@@ -802,6 +855,8 @@ def _convert_tracks(tracks: Iterable[Track]) -> DataFrame:
         for detection in current_track.detections:
             dto = detection.to_dict()
             dto[track.ORIGINAL_TRACK_ID] = current_track.original_id.id
+            dto[ottrk_dataformat.FIRST] = current_track.first_detection == detection
+            dto[ottrk_dataformat.FINISHED] = detection.is_finished
             prepared.append(dto)
 
     if not prepared:

OTAnalytics/plugin_parser/argparse_cli_parser.py

@@ -43,13 +43,6 @@ def _setup(self) -> None:
             default=CliMode.BULK,
             required=False,
         )
-        self._parser.add_argument(
-            "--cli-chunk-size",
-            type=int,
-            help="Specify the chunk size for streaming OTAnalytics CLI.",
-            default=10,
-            required=False,
-        )
         self._parser.add_argument(
             "--show-svz",
             action="store_true",
@@ -176,7 +169,6 @@ def parse(self) -> CliArguments:
             start_cli=args.cli,
             start_webui=args.webui,
             cli_mode=args.cli_mode,
-            cli_chunk_size=args.cli_chunk_size,
             debug=args.debug,
             logfile_overwrite=args.logfile_overwrite,
             show_svz=args.show_svz,

OTAnalytics/plugin_parser/otvision_parser.py

@@ -440,6 +440,7 @@ def parse_python_detection(
         _video_name=metadata_video[ottrk_format.FILENAME]
         + metadata_video[ottrk_format.FILETYPE],
         _input_file=input_file,
+        _is_finished=det_dict.get(ottrk_format.FINISHED, False),
     )