Add source_video link field to PoseEstimation

spec/ndx-pose.extensions.yaml

@@ -84,7 +84,9 @@ groups:
     shape:
     - null
     doc: Paths to the original video files. The number of files should equal the number
-      of camera devices.
+      of camera devices. Note that these string paths might be fragile unless relative
+      paths are used and care is taken to keep them consistent. Consider using the
+      'source_video' link instead for a formal reference.
     quantity: '?'
   - name: labeled_videos
     dtype: text
@@ -130,6 +132,13 @@ groups:
   - target_type: Device
     doc: Cameras used to record the videos.
     quantity: '*'
+  - name: source_video
+    target_type: ImageSeries
+    doc: Link to an ImageSeries containing the source video used for pose estimation.
+      The ImageSeries should be stored in the NWBFile (e.g., in acquisition).
+      When available, this field should be preferred over 'original_videos' as it
+      provides a formal reference rather than a file path string.
+    quantity: '?'
 - neurodata_type_def: TrainingFrame
   neurodata_type_inc: NWBDataInterface
   default_name: TrainingFrame

spec/ndx-pose.namespace.yaml

@@ -22,4 +22,4 @@ namespaces:
   schema:
   - namespace: core
   - source: ndx-pose.extensions.yaml
-  version: 0.2.0
+  version: 0.2.1

src/pynwb/ndx_pose/pose.py

@@ -3,6 +3,7 @@
 from pynwb import register_class, TimeSeries, get_class
 from pynwb.behavior import SpatialSeries
 from pynwb.core import MultiContainerInterface
+from pynwb.image import ImageSeries
 
 # TODO validate Skeleton nodes and edges correspondence, convert edges to uint
 # TODO validate that all Skeleton nodes are used in edges
@@ -120,6 +121,7 @@ class PoseEstimation(MultiContainerInterface):
         "nodes",
         "edges",
         "skeleton",  # <-- this is a link to a Skeleton object
+        "source_video",  # <-- this is a link to an ImageSeries object
     )
 
     # custom mapper in ndx_pose.io.pose maps:
@@ -149,7 +151,11 @@ class PoseEstimation(MultiContainerInterface):
             "name": "original_videos",
             "type": ("array_data", "data"),
             "shape": (None,),
-            "doc": "Paths to the original video files. The number of files should equal the number of camera devices.",
+            "doc": (
+                "Paths to the original video files. The number of files should equal the number of camera devices. "
+                "Note: these string paths might be fragile unless relative paths are used and care is taken to "
+                "keep them consistent. Consider using 'source_video' instead for a formal link to an ImageSeries."
+            ),
             "default": None,
         },
         {
@@ -203,6 +209,17 @@ class PoseEstimation(MultiContainerInterface):
             ),
             "default": None,
         },
+        {
+            "name": "source_video",
+            "type": ImageSeries,
+            "doc": (
+                "Link to an ImageSeries containing the source video used for pose estimation. "
+                "The ImageSeries should be stored in the NWBFile (e.g., in acquisition) and linked here. "
+                "When available, this field should be preferred over 'original_videos' as it provides "
+                "a formal reference rather than a file path string."
+            ),
+            "default": None,
+        },
         {
             "name": "nodes",
             "type": ("array_data", "data"),
@@ -226,7 +243,7 @@ class PoseEstimation(MultiContainerInterface):
         allow_positional=AllowPositional.ERROR,
     )
     def __init__(self, **kwargs):
-        nodes, edges, skeleton = popargs("nodes", "edges", "skeleton", kwargs)
+        nodes, edges, skeleton, source_video = popargs("nodes", "edges", "skeleton", "source_video", kwargs)
         if nodes is not None or edges is not None:
             if skeleton is not None:
                 raise ValueError("Cannot specify 'skeleton' with 'nodes' or 'edges'.")
@@ -296,6 +313,7 @@ def __init__(self, **kwargs):
         self.source_software = source_software
         self.source_software_version = source_software_version
         self.skeleton = skeleton
+        self.source_video = source_video
 
         # TODO include calibration images for 3D estimates?
         # TODO validate that the nodes correspond to the names of the pose estimation series objects

src/pynwb/ndx_pose/testing/mock/pose.py

@@ -91,6 +91,7 @@ def mock_PoseEstimation(
     scorer: Optional[str] = "DLC_resnet50_openfieldOct30shuffle1_1600",
     source_software: Optional[str] = "DeepLabCut",
     source_software_version: Optional[str] = "2.2b8",
+    source_video: Optional[ImageSeries] = None,
 ):
     """Create a mock PoseEstimation object.
 
@@ -109,6 +110,7 @@ def mock_PoseEstimation(
         source_software=source_software,
         source_software_version=source_software_version,
         skeleton=skeleton,
+        source_video=source_video,
     )
 
     if nwbfile is not None:

src/pynwb/tests/integration/hdf5/test_pose.py

@@ -2,6 +2,7 @@
 import numpy as np
 
 from pynwb import NWBHDF5IO, NWBFile
+from pynwb.image import ImageSeries
 from pynwb.testing import TestCase, remove_test_file, NWBH5IOFlexMixin
 
 from ndx_pose import (
@@ -189,6 +190,66 @@ def test_roundtrip(self):
             self.assertContainerEqual(read_pe.devices[0], self.nwbfile.devices["camera1"])
 
 
+class TestPoseEstimationRoundtripSourceVideo(TestCase):
+    """Roundtrip test for PoseEstimation with source_video link."""
+
+    def setUp(self):
+        self.nwbfile = NWBFile(
+            session_description="session_description",
+            identifier="identifier",
+            session_start_time=datetime.datetime.now(datetime.timezone.utc),
+        )
+        self.nwbfile.create_device(name="camera1")
+        self.path = "test_pose.nwb"
+
+    def tearDown(self):
+        remove_test_file(self.path)
+
+    def test_roundtrip(self):
+        """Test that source_video link survives write/read roundtrip."""
+        source_video = ImageSeries(
+            name="source_video",
+            description="Source video for pose estimation.",
+            unit="NA",
+            format="external",
+            external_file=["camera1.mp4"],
+            dimension=[640, 480],
+            starting_frame=[0],
+            rate=30.0,
+        )
+        self.nwbfile.add_acquisition(source_video)
+
+        skeleton = mock_Skeleton()
+        skeletons = Skeletons(skeletons=[skeleton])
+
+        pose_estimation_series = [mock_PoseEstimationSeries(name=name) for name in skeleton.nodes]
+        pe = PoseEstimation(
+            pose_estimation_series=pose_estimation_series,
+            description="Estimated positions of front paws using DeepLabCut.",
+            original_videos=["camera1.mp4"],
+            dimensions=np.array([[640, 480]], dtype="uint16"),
+            devices=[self.nwbfile.devices["camera1"]],
+            scorer="DLC_resnet50_openfieldOct30shuffle1_1600",
+            source_software="DeepLabCut",
+            source_software_version="2.2b8",
+            skeleton=skeleton,
+            source_video=source_video,
+        )
+
+        behavior_pm = self.nwbfile.create_processing_module(name="behavior", description="processed behavioral data")
+        behavior_pm.add(pe)
+        behavior_pm.add(skeletons)
+
+        with NWBHDF5IO(self.path, mode="w") as io:
+            io.write(self.nwbfile)
+
+        with NWBHDF5IO(self.path, mode="r", load_namespaces=True) as io:
+            read_nwbfile = io.read()
+            read_pe = read_nwbfile.processing["behavior"]["PoseEstimation"]
+            self.assertContainerEqual(read_pe.source_video, source_video)
+            self.assertEqual(read_pe.source_video.external_file[0], "camera1.mp4")
+
+
 class TestPoseEstimationRoundtripPyNWB(NWBH5IOFlexMixin, TestCase):
     """Complex, more complete roundtrip test for PoseEstimation using pynwb.testing infrastructure."""
 

src/pynwb/tests/unit/test_pose.py

@@ -6,6 +6,8 @@
 from pynwb.testing import TestCase
 from pynwb.file import Subject
 
+from pynwb.image import ImageSeries
+
 from ndx_pose import (
     PoseEstimationSeries,
     Skeleton,
@@ -227,6 +229,45 @@ def test_constructor_nodes_edges(self):
         self.assertEqual(pe.skeleton.nodes, skeleton.nodes)
         np.testing.assert_array_equal(pe.skeleton.edges, skeleton.edges)
 
+    def test_constructor_source_video(self):
+        """Test that source_video link is set correctly."""
+        source_video = ImageSeries(
+            name="source_video",
+            description="Source video for pose estimation.",
+            unit="NA",
+            format="external",
+            external_file=["camera1.mp4"],
+            dimension=[640, 480],
+            starting_frame=[0],
+            rate=30.0,
+        )
+        skeleton = mock_Skeleton()
+        pose_estimation_series = [mock_PoseEstimationSeries(name=name) for name in skeleton.nodes]
+
+        pe = PoseEstimation(
+            pose_estimation_series=pose_estimation_series,
+            description="Estimated positions of front paws using DeepLabCut.",
+            original_videos=["camera1.mp4"],
+            devices=[self.nwbfile.devices["camera1"]],
+            scorer="DLC_resnet50_openfieldOct30shuffle1_1600",
+            source_software="DeepLabCut",
+            source_software_version="2.2b8",
+            skeleton=skeleton,
+            source_video=source_video,
+        )
+        self.assertIs(pe.source_video, source_video)
+
+    def test_constructor_source_video_default_none(self):
+        """Test that source_video defaults to None."""
+        skeleton = mock_Skeleton()
+        pose_estimation_series = [mock_PoseEstimationSeries(name=name) for name in skeleton.nodes]
+
+        pe = PoseEstimation(
+            pose_estimation_series=pose_estimation_series,
+            skeleton=skeleton,
+        )
+        self.assertIsNone(pe.source_video)
+
 
 class TestSkeletonInstance(TestCase):
     def test_constructor(self):