Add MVRDataset

README.md

@@ -16,8 +16,33 @@ conda activate npc_mvr
 pip install npc_mvr
 ```
 
+## Python
 ```python
-import npc_mvr
+>>> import npc_mvr
+
+>>> d = npc_mvr.MVRDataset('s3://aind-ephys-data/ecephys_670248_2023-08-03_12-04-15/behavior')
+
+# get paths
+>>> d.video_paths['behavior']
+S3Path('s3://aind-ephys-data/ecephys_670248_2023-08-03_12-04-15/behavior/Behavior_20230803T120430.mp4')
+>>> d.info_paths['behavior']
+S3Path('s3://aind-ephys-data/ecephys_670248_2023-08-03_12-04-15/behavior/Behavior_20230803T120430.json')
+>>> d.sync_path
+S3Path('s3://aind-ephys-data/ecephys_670248_2023-08-03_12-04-15/behavior/20230803T120415.h5')
+
+# get data
+>>> type(d.video_data['behavior'])
+<class 'cv2.VideoCapture'>
+>>> type(d.info_data['behavior'])
+<class 'dict'>
+>>> type(d.sync_data)
+<class 'npc_sync.sync.SyncDataset'>
+
+# get frame times for each camera on sync clock
+# - nans correspond to frames not recorded on sync
+# - first nan is metadata frame 
+>>> d.frame_times['behavior']
+array([       nan,   14.08409,   14.10075, ..., 5084.4582 , 5084.47487, 5084.49153]) 
 ```
 
 # Development

pyproject.toml

@@ -8,7 +8,7 @@ readme = "README.md"
 requires-python = ">=3.9"
 dependencies = [
     "npc-io>=0.1.20",
-    "npc-sync>=0.1.2",
+    "npc-sync>=0.1.5",
     "opencv-python-headless>=4.9.0.80",
     "typing-extensions>=4.9.0",
 ]

src/npc_mvr/mvr.py

@@ -1,4 +1,5 @@
 from __future__ import annotations
+import contextlib
 
 import json
 import logging
@@ -20,9 +21,106 @@
 """Contents of `RecordingReport` from a camera's info.json for an MVR
 recording."""
 
-
-def get_camera_name(path: str) -> Literal["eye", "face", "behavior"]:
-    names: dict[str, Literal["eye", "face", "behavior"]] = {
+CameraName: TypeAlias = Literal["eye", "face", "behavior"]
+
+class  MVRDataset:
+    """A collection of paths + data for processing the output from MVR for one
+    session.
+    
+    Expectations:
+    
+    - 3 .mp4/.avi video file paths (eye, face, behavior)
+    - 3 .json info file paths (eye, face, behavior)
+    - the associated data as Python objects for each of the above (e.g mp3 -> CV2,
+    json -> dict)
+    
+    - 1 sync file path (h5)
+    - sync data as a SyncDataset object
+    
+    Assumptions:
+    - all files live in the same directory (so we can initialize with a single path)
+    - MVR was started after sync
+    - MVR may have been stopped before sync
+    
+    >>> import npc_mvr
+
+    >>> d = npc_mvr.MVRDataset('s3://aind-ephys-data/ecephys_670248_2023-08-03_12-04-15/behavior')
+
+    # get paths
+    >>> d.video_paths['behavior']
+    S3Path('s3://aind-ephys-data/ecephys_670248_2023-08-03_12-04-15/behavior/Behavior_20230803T120430.mp4')
+    >>> d.info_paths['behavior']
+    S3Path('s3://aind-ephys-data/ecephys_670248_2023-08-03_12-04-15/behavior/Behavior_20230803T120430.json')
+    >>> d.sync_path
+    S3Path('s3://aind-ephys-data/ecephys_670248_2023-08-03_12-04-15/behavior/20230803T120415.h5')
+
+    # get data
+    >>> type(d.video_data['behavior'])
+    <class 'cv2.VideoCapture'>
+    >>> type(d.info_data['behavior'])
+    <class 'dict'>
+    >>> type(d.sync_data)
+    <class 'npc_sync.sync.SyncDataset'>
+
+    # get frame times for each camera on sync clock
+    # - nans correspond to frames not recorded on sync
+    # - first nan is metadata frame 
+    >>> d.frame_times['behavior']
+    array([       nan,   14.08409,   14.10075, ..., 5084.4582 , 5084.47487, 5084.49153])
+    
+    """
+
+    def __init__(self, session_dir: npc_io.PathLike, sync_path: npc_io.PathLike | None = None):
+        self.session_dir = npc_io.from_pathlike(session_dir)
+        if sync_path is not None:
+            self.sync_path = npc_io.from_pathlike(sync_path)
+
+    def __repr__(self) -> str:
+        return f"{self.__class__.__name__}({self.session_dir})"
+
+    @npc_io.cached_property
+    def augmented_camera_info(self) -> dict[CameraName, dict[str, int | float]]:
+        return get_augmented_camera_info(self.sync_data, *self.video_paths.values())
+
+    @npc_io.cached_property
+    def frame_times(self) -> dict[CameraName, npt.NDArray[np.float64]]:
+        """Returns frametimes (in seconds) as measured on sync clock for each
+        camera.
+        
+        - see `get_video_frame_times` for more details
+        """
+        return {get_camera_name(p.stem): times for p, times in get_video_frame_times(self.sync_data, *self.video_paths.values()).items()}
+
+    @npc_io.cached_property
+    def video_paths(self) -> dict[CameraName, upath.UPath]:
+        return {get_camera_name(p.stem): p for p in get_video_file_paths(self.session_dir)}
+
+    @npc_io.cached_property
+    def video_data(self) -> npc_io.LazyDict[str, cv2.VideoCapture]:
+        return npc_io.LazyDict(
+            (camera_name, (get_video_data, (path,), {}))
+            for camera_name, path in self.video_paths.items()
+        )
+
+    @npc_io.cached_property
+    def info_paths(self) -> dict[CameraName, upath.UPath]:
+        return {get_camera_name(p.stem): p for p in get_video_info_file_paths(self.session_dir)}
+
+    @npc_io.cached_property
+    def info_data(self) -> dict[CameraName, MVRInfoData]:
+        return {camera_name: get_video_info_data(path) for camera_name, path in self.info_paths.items()}
+
+    @npc_io.cached_property
+    def sync_path(self) -> upath.UPath:
+        return npc_sync.get_single_sync_path(self.session_dir)
+
+    @npc_io.cached_property
+    def sync_data(self) -> npc_sync.SyncDataset:
+        return npc_sync.get_sync_data(self.sync_path)
+
+
+def get_camera_name(path: str) -> CameraName:
+    names: dict[str, CameraName] = {
         "eye": "eye",
         "face": "face",
         "beh": "behavior",
@@ -239,7 +337,7 @@ def get_total_frames_in_video(
 def get_augmented_camera_info(
     sync_path_or_dataset: npc_io.PathLike | npc_sync.SyncDataset,
     *video_paths: npc_io.PathLike,
-) -> dict[Literal["eye", "face", "behavior"], dict[str, int | float]]:
+) -> dict[CameraName, dict[str, int | float]]:
     videos = get_video_file_paths(*video_paths)
     jsons = get_video_info_file_paths(*video_paths)
     camera_to_video_path = {get_camera_name(path.stem): path for path in videos}