Expose revision

src/ibl_to_nwb/_scripts/convert_brainwide_map_processed_only.py

@@ -18,12 +18,16 @@
     WheelInterface,
 )
 
+session_id = "d32876dd-8303-4720-8e7e-20678dc2fd71"
+
+# Specify the revision of the pose estimation data
+# Setting to 'None' will use whatever the latest released revision is
+revision = None
+
 base_path = Path("E:/IBL")
 nwbfiles_folder_path = base_path / "nwbfiles"
 nwbfiles_folder_path.mkdir(exist_ok=True)
 
-session_id = "d32876dd-8303-4720-8e7e-20678dc2fd71"
-
 # Initialize IBL (ONE) client to download processed data for this session
 one_cache_folder_path = base_path / "cache"
 ibl_client = ONE(
@@ -46,7 +50,9 @@
 for pose_estimation_file in pose_estimation_files:
     camera_name = pose_estimation_file.replace("alf/_ibl_", "").replace(".dlc.pqt", "")
     data_interfaces.append(
-        IblPoseEstimationInterface(one=ibl_client, session=session_id, camera_name=camera_name, include_video=False)
+        IblPoseEstimationInterface(
+            one=ibl_client, session=session_id, camera_name=camera_name, include_video=False, revision=revision
+        )
     )
 
 pupil_tracking_files = ibl_client.list_datasets(eid=session_id, filename="*features*")

src/ibl_to_nwb/datainterfaces/_pose_estimation.py

@@ -1,4 +1,5 @@
 from datetime import datetime
+from typing import Optional
 
 import numpy as np
 from ndx_pose import PoseEstimation, PoseEstimationSeries
@@ -7,32 +8,43 @@
 from one.api import ONE
 from pynwb import NWBFile
 from pynwb.image import ImageSeries
+from typing_extensions import Self
 
 
 class IblPoseEstimationInterface(BaseDataInterface):
-    def __init__(self, one: ONE, session: str, camera_name: str, include_video: bool, include_pose: bool):
+    def __init__(
+        self,
+        one: ONE,
+        session: str,
+        camera_name: str,
+        include_video: bool,
+        include_pose: bool,
+        revision: Optional[str] = None,
+    ) -> Self:
         self.one = one
         self.session = session
         self.camera_name = camera_name
         self.include_video = include_video
         self.include_pose = include_pose
 
-    def add_to_nwbfile(self, nwbfile: NWBFile, metadata: dict):
-        # Sometimes the DLC data has been revised, possibly multiple times
-        # Always use the most recent revision available
-        session_files = self.one.list_datasets(eid=self.session, filename=f"*{self.camera_name}.dlc*")
-        revision_datetime_format = "%Y-%m-%d"
-        revisions = [
-            datetime.strptime(session_file.split("#")[1], revision_datetime_format)
-            for session_file in session_files
-            if "#" in session_file
-        ]
-        revision = None
-        if any(revisions):
-            most_recent = max(revisions)
-            revision = most_recent.strftime("%Y-%m-%d")
+        self.revision = revision
+        if self.revision is None:
+            session_files = self.one.list_datasets(eid=self.session, filename=f"*{self.camera_name}.dlc*")
+            revision_datetime_format = "%Y-%m-%d"
+            revisions = [
+                datetime.strptime(session_file.split("#")[1], revision_datetime_format)
+                for session_file in session_files
+                if "#" in session_file
+            ]
 
-        camera_data = self.one.load_object(id=self.session, obj=self.camera_name, collection="alf", revision=revision)
+            if any(revisions):
+                most_recent = max(revisions)
+                self.revision = most_recent.strftime("%Y-%m-%d")
+
+    def add_to_nwbfile(self, nwbfile: NWBFile, metadata: dict) -> None:
+        camera_data = self.one.load_object(
+            id=self.session, obj=self.camera_name, collection="alf", revision=self.revision
+        )
         dlc_data = camera_data["dlc"]
         timestamps = camera_data["times"]
         number_of_frames = len(timestamps)