User story/2061 boundingboxes of detections per frame

.run/pytest-in-tests.run.xml

@@ -4,7 +4,7 @@
     <option name="INTERPRETER_OPTIONS" value="" />
     <option name="PARENT_ENVS" value="true" />
     <option name="SDK_HOME" value="" />
-    <option name="WORKING_DIRECTORY" value="$PROJECT_DIR$/tests" />
+    <option name="WORKING_DIRECTORY" value="$PROJECT_DIR$" />
     <option name="IS_MODULE_SDK" value="true" />
     <option name="ADD_CONTENT_ROOTS" value="true" />
     <option name="ADD_SOURCE_ROOTS" value="true" />

OTAnalytics/adapter_ui/view_model.py

@@ -335,3 +335,27 @@ def set_frame_track_plotting(
     @abstractmethod
     def set_analysis_frame(self, frame: AbstractFrame) -> None:
         raise NotImplementedError
+
+    @abstractmethod
+    def next_frame(self) -> None:
+        pass
+
+    @abstractmethod
+    def previous_frame(self) -> None:
+        pass
+
+    @abstractmethod
+    def update_skip_time(self, seconds: int, frames: int) -> None:
+        raise NotImplementedError
+
+    @abstractmethod
+    def get_skip_seconds(self) -> int:
+        raise NotImplementedError
+
+    @abstractmethod
+    def get_skip_frames(self) -> int:
+        raise NotImplementedError
+
+    @abstractmethod
+    def set_video_control_frame(self, frame: AbstractFrame) -> None:
+        raise NotImplementedError

OTAnalytics/application/application.py

@@ -17,6 +17,10 @@
     TrackViewState,
     VideosMetadata,
 )
+from OTAnalytics.application.ui.frame_control import (
+    SwitchToNextFrame,
+    SwitchToPreviousFrame,
+)
 from OTAnalytics.application.use_cases.config import SaveOtconfig
 from OTAnalytics.application.use_cases.create_events import (
     CreateEvents,
@@ -98,6 +102,8 @@ def __init__(
         start_new_project: StartNewProject,
         project_updater: ProjectUpdater,
         load_track_files: LoadTrackFiles,
+        previous_frame: SwitchToPreviousFrame,
+        next_frame: SwitchToNextFrame,
     ) -> None:
         self._datastore: Datastore = datastore
         self.track_state: TrackState = track_state
@@ -129,6 +135,8 @@ def __init__(
         self._track_repository_size = TrackRepositorySize(
             self._datastore._track_repository
         )
+        self._previous_frame = previous_frame
+        self._next_frame = next_frame
 
     def connect_observers(self) -> None:
         """
@@ -449,6 +457,12 @@ def get_current_track_offset(self) -> Optional[RelativeOffsetCoordinate]:
         """
         return self.track_view_state.track_offset.get()
 
+    def next_frame(self) -> None:
+        self._next_frame.set_next_frame()
+
+    def previous_frame(self) -> None:
+        self._previous_frame.set_previous_frame()
+
     def update_date_range_tracks_filter(self, date_range: DateRange) -> None:
         """Update the date range of the track filter.
 

OTAnalytics/application/datastore.py

@@ -61,6 +61,12 @@ def duration(self) -> timedelta:
             return self.expected_duration
         return timedelta(seconds=self.number_of_frames / self.recorded_fps)
 
+    @property
+    def fps(self) -> float:
+        if self.actual_fps:
+            return self.actual_fps
+        return self.recorded_fps
+
 
 @dataclass(frozen=True)
 class TrackParseResult:
@@ -120,7 +126,7 @@ def serialize(
 
 class VideoParser(ABC):
     @abstractmethod
-    def parse(self, file: Path) -> Video:
+    def parse(self, file: Path, start_date: Optional[datetime]) -> Video:
         pass
 
     @abstractmethod
@@ -294,7 +300,7 @@ def load_video_files(self, files: list[Path]) -> None:
         videos = []
         for file in files:
             try:
-                videos.append(self._video_parser.parse(file))
+                videos.append(self._video_parser.parse(file, None))
             except Exception as cause:
                 raised_exceptions.append(cause)
         if raised_exceptions:
@@ -514,7 +520,11 @@ def get_video_for(self, track_id: TrackId) -> Optional[Video]:
     def get_all_videos(self) -> list[Video]:
         return self._video_repository.get_all()
 
-    def get_image_of_track(self, track_id: TrackId) -> Optional[TrackImage]:
+    def get_image_of_track(
+        self,
+        track_id: TrackId,
+        frame: int = 0,
+    ) -> Optional[TrackImage]:
         """
         Retrieve an image for the given track.
 
@@ -525,4 +535,6 @@ def get_image_of_track(self, track_id: TrackId) -> Optional[TrackImage]:
             Optional[TrackImage]: an image of the track if the track is available and
             the image can be loaded
         """
-        return video.get_frame(0) if (video := self.get_video_for(track_id)) else None
+        return (
+            video.get_frame(frame) if (video := self.get_video_for(track_id)) else None
+        )

OTAnalytics/application/playback.py

@@ -0,0 +1,7 @@
+from dataclasses import dataclass
+
+
+@dataclass(frozen=True)
+class SkipTime:
+    seconds: int
+    frames: int

OTAnalytics/application/plotting.py

@@ -1,13 +1,17 @@
-from abc import abstractmethod
+from abc import ABC, abstractmethod
+from datetime import datetime, timedelta
+from math import floor
 from typing import Any, Callable, Generic, Iterable, Optional, Sequence, TypeVar
 
 from OTAnalytics.application.state import (
     ObservableOptionalProperty,
     ObservableProperty,
     Plotter,
     TrackViewState,
+    VideosMetadata,
 )
 from OTAnalytics.domain.track import TrackImage
+from OTAnalytics.domain.video import Video
 
 
 class Layer:
@@ -98,16 +102,31 @@ def __add(self, image: TrackImage) -> None:
             self._current_image = image
 
 
+class VisualizationTimeProvider(ABC):
+    @abstractmethod
+    def get_time(self) -> datetime:
+        raise NotImplementedError
+
+
+VideoProvider = Callable[[], list[Video]]
+
+
 class TrackBackgroundPlotter(Plotter):
     """Plot video frame as background."""
 
-    def __init__(self, track_view_state: TrackViewState) -> None:
-        self._track_view_state = track_view_state
+    def __init__(
+        self,
+        video_provider: VideoProvider,
+        visualization_time_provider: VisualizationTimeProvider,
+    ) -> None:
+        self._video_provider = video_provider
+        self._visualization_time_provider = visualization_time_provider
 
     def plot(self) -> Optional[TrackImage]:
-        if videos := self._track_view_state.selected_videos.get():
-            if len(videos) > 0:
-                return videos[0].get_frame(0)
+        if videos := self._video_provider():
+            visualization_time = self._visualization_time_provider.get_time()
+            frame_number = videos[0].get_frame_number_for(visualization_time)
+            return videos[0].get_frame(frame_number)
         return None
 
 
@@ -238,3 +257,50 @@ def _handle_remove(self, entities: Iterable[ENTITY]) -> None:
         for entity in entities:
             del self._plotter_mapping[entity]
             del self._layer_mapping[entity]
+
+
+class GetCurrentVideoPath:
+    """
+    This use case provides the currently visible video path. It uses the current filters
+    end date to retrieve the corresponding file path.
+    """
+
+    def __init__(
+        self,
+        state: TrackViewState,
+        videos_metadata: VideosMetadata,
+    ) -> None:
+        self._state = state
+        self._videos_metadata = videos_metadata
+
+    def get_video(self) -> Optional[str]:
+        if end_date := self._state.filter_element.get().date_range.end_date:
+            if metadata := self._videos_metadata.get_metadata_for(end_date):
+                return metadata.path
+        return None
+
+
+class GetCurrentFrame:
+    """
+    This use case provides the currently visible frame. It uses the current filters
+    end date to retrieve the corresponding frame.
+    """
+
+    def __init__(
+        self,
+        state: TrackViewState,
+        videos_metadata: VideosMetadata,
+    ) -> None:
+        self._state = state
+        self._videos_metadata = videos_metadata
+
+    def get_frame_number(self) -> int:
+        if end_date := self._state.filter_element.get().date_range.end_date:
+            if metadata := self._videos_metadata.get_metadata_for(end_date):
+                time_in_video = end_date - metadata.start
+                if time_in_video < timedelta(0):
+                    return 0
+                if time_in_video > metadata.duration:
+                    return metadata.number_of_frames
+                return floor(metadata.fps * time_in_video.total_seconds())
+        return 0

OTAnalytics/application/state.py

@@ -1,9 +1,11 @@
+import bisect
 from abc import ABC, abstractmethod
 from datetime import datetime
 from typing import Callable, Generic, Optional
 
 from OTAnalytics.application.config import DEFAULT_TRACK_OFFSET
 from OTAnalytics.application.datastore import Datastore, VideoMetadata
+from OTAnalytics.application.playback import SkipTime
 from OTAnalytics.application.use_cases.section_repository import GetSectionsById
 from OTAnalytics.domain.date import DateRange
 from OTAnalytics.domain.event import EventRepositoryEvent
@@ -185,6 +187,7 @@ def __init__(self) -> None:
         self.selected_videos: ObservableProperty[list[Video]] = ObservableProperty[
             list[Video]
         ](default=[])
+        self.skip_time = ObservableProperty[SkipTime](SkipTime(1, 0))
 
     def reset(self) -> None:
         """Reset to default settings."""
@@ -404,19 +407,50 @@ def _update_image(self) -> None:
 
 class VideosMetadata:
     def __init__(self) -> None:
-        self._metadata: list[VideoMetadata] = []
+        self._metadata: dict[datetime, VideoMetadata] = {}
+        self._first_video_start: Optional[datetime] = None
+        self._last_video_end: Optional[datetime] = None
 
     def update(self, metadata: VideoMetadata) -> None:
-        self._metadata.append(metadata)
-        self._metadata.sort(key=lambda current: current.start)
+        """
+        Update the stored metadata.
+        """
+        if metadata.start in self._metadata.keys():
+            raise ValueError(
+                f"metadata with start date {metadata.start} already exists."
+            )
+        self._metadata[metadata.start] = metadata
+        self._metadata = dict(sorted(self._metadata.items()))
+        self._update_start_end_by(metadata)
+
+    def _update_start_end_by(self, metadata: VideoMetadata) -> None:
+        if (not self._first_video_start) or metadata.start < self._first_video_start:
+            self._first_video_start = metadata.start
+        if (not self._last_video_end) or metadata.end > self._last_video_end:
+            self._last_video_end = metadata.end
+
+    def get_metadata_for(self, current: datetime) -> Optional[VideoMetadata]:
+        """
+        Find the metadata for the given datetime. If the datetime matches exactly a
+        start time of a video, the corresponding VideoMetadata is returned. Otherwise,
+        the metadata of the video containing the datetime will be returned.
+        """
+        if current in self._metadata:
+            return self._metadata[current]
+        keys = list(self._metadata.keys())
+        key = bisect.bisect_left(keys, current) - 1
+        metadata = self._metadata[keys[key]]
+        if metadata.start <= current <= metadata.end:
+            return metadata
+        return None
 
     @property
     def first_video_start(self) -> Optional[datetime]:
-        return self._metadata[0].recorded_start_date if self._metadata else None
+        return self._first_video_start
 
     @property
     def last_video_end(self) -> Optional[datetime]:
-        return self._metadata[-1].end if self._metadata else None
+        return self._last_video_end
 
 
 class TracksMetadata(TrackListObserver):

OTAnalytics/application/ui/frame_control.py

@@ -0,0 +1,57 @@
+from datetime import timedelta
+
+from OTAnalytics.application.state import TrackViewState, VideosMetadata
+from OTAnalytics.domain.date import DateRange
+
+
+class SwitchToNextFrame:
+    def __init__(self, state: TrackViewState, videos_metadata: VideosMetadata) -> None:
+        self._state = state
+        self._videos_metadata = videos_metadata
+
+    def set_next_frame(self) -> None:
+        if filter_element := self._state.filter_element.get():
+            current_date_range = filter_element.date_range
+            if current_date_range.start_date and current_date_range.end_date:
+                if metadata := self._videos_metadata.get_metadata_for(
+                    current_date_range.end_date
+                ):
+                    fps = metadata.fps
+                    skip_time = self._state.skip_time.get()
+                    subseconds = min(skip_time.frames, fps) / fps
+                    milliseconds = subseconds * 1000
+                    current_skip = timedelta(
+                        seconds=skip_time.seconds, milliseconds=milliseconds
+                    )
+                    next_start = current_date_range.start_date + current_skip
+                    next_end = current_date_range.end_date + current_skip
+                    next_date_range = DateRange(next_start, next_end)
+                    self._state.filter_element.set(
+                        filter_element.derive_date(next_date_range)
+                    )
+
+
+class SwitchToPreviousFrame:
+    def __init__(self, state: TrackViewState, videos_metadata: VideosMetadata) -> None:
+        self._state = state
+        self._videos_metadata = videos_metadata
+
+    def set_previous_frame(self) -> None:
+        if filter_element := self._state.filter_element.get():
+            current_date_range = filter_element.date_range
+            if current_date_range.start_date and current_date_range.end_date:
+                if metadata := self._videos_metadata.get_metadata_for(
+                    current_date_range.end_date
+                ):
+                    fps = metadata.fps
+                    skip_time = self._state.skip_time.get()
+                    subseconds = min(skip_time.frames, fps) / fps
+                    current_skip = timedelta(seconds=skip_time.seconds) + timedelta(
+                        seconds=subseconds
+                    )
+                    next_start = current_date_range.start_date - current_skip
+                    next_end = current_date_range.end_date - current_skip
+                    next_date_range = DateRange(next_start, next_end)
+                    self._state.filter_element.set(
+                        filter_element.derive_date(next_date_range)
+                    )

OTAnalytics/domain/track.py

@@ -241,6 +241,9 @@ def height(self) -> int:
         """
         pass
 
+    def save(self, name: str) -> None:
+        self.as_image().save(name)
+
 
 @dataclass(frozen=True)
 class PilImage(TrackImage):