Preserve legacy ndx-pose video references during organize

dandi/pynwb_utils.py

@@ -475,7 +475,11 @@ def _get_external_images(nwb: pynwb.NWBFile) -> list[dict]:
 
 
 def rename_nwb_external_files(metadata: list[dict], dandiset_path: str) -> None:
-    """Renames the external_file attribute in an ImageSeries datatype in an open nwb file.
+    """Rename external media references in an open NWB file.
+
+    This updates ``ImageSeries.external_file`` entries and legacy
+    ``PoseEstimation.original_videos`` references using the path mapping
+    collected in ``metadata["external_file_objects"]``.
 
     It pulls information about the ImageSeries objects from metadata:
     metadata["external_file_objects"] populated during _get_pynwb_metadata() call.
@@ -508,6 +512,7 @@ def rename_nwb_external_files(metadata: list[dict], dandiset_path: str) -> None:
                 dandiset_nwbfile_path, mode="r+", load_namespaces=True
             ) as io:
                 nwb = io.read()
+                pose_video_renames: dict[str, str] = {}
                 for ext_file_dict in image_series:
                     # retrieve nwb neurodata object of the given object id:
                     container_list = [
@@ -526,12 +531,50 @@ def rename_nwb_external_files(metadata: list[dict], dandiset_path: str) -> None:
                             ext_file_dict["external_files_renamed"],
                         )
                     ):
-                        if not is_url(str(name_old)):
+                        reference = _external_reference_key(name_old)
+                        if not is_url(reference):
                             container.external_file[no] = str(name_new)
+                            pose_video_renames[reference] = str(name_new)
+                _rename_pose_estimation_original_videos(nwb, pose_video_renames)
         if external_images:
             _rename_external_images(dandiset_nwbfile_path, external_images)
 
 
+def _external_reference_key(value: Any) -> str:
+    """Return a comparable representation of a legacy external-file reference."""
+
+    if isinstance(value, bytes):
+        value = value.decode()
+    return str(value).replace("\\", "/")
+
+
+def _rename_pose_estimation_original_videos(
+    nwb: pynwb.NWBFile, renames: dict[str, str]
+) -> None:
+    """Update legacy ndx-pose ``original_videos`` references in an NWB file.
+
+    ``original_videos`` is deprecated in ndx-pose in favor of the linked
+    ``source_video`` field.  Older files can still contain the path list, so
+    use the same old-to-new mapping as the source ``ImageSeries`` without
+    importing ndx-pose or reading waveform/video payloads.
+    """
+
+    if not renames:
+        return
+    for container in nwb.objects.values():
+        if getattr(container, "neurodata_type", None) != "PoseEstimation":
+            continue
+        original_videos = getattr(container, "original_videos", None)
+        if original_videos is None or isinstance(original_videos, (str, bytes)):
+            continue
+        for no, name_old in enumerate(original_videos):
+            reference = _external_reference_key(name_old)
+            if is_url(reference):
+                continue
+            if (name_new := renames.get(reference)) is not None:
+                original_videos[no] = name_new
+
+
 def _rename_external_images(nwbfile_path: str, external_images: list[dict]) -> None:
     """Rewrites the ``data`` of the given `ExternalImage` objects in an NWB file on disk.
 

dandi/tests/test_pynwb_utils.py

@@ -4,12 +4,21 @@
 from datetime import datetime, timezone
 from pathlib import Path
 import re
+from types import SimpleNamespace
 from typing import Any, NoReturn
 
+import h5py
 import numpy as np
+import pytest
 from pynwb import NWBHDF5IO, NWBFile, TimeSeries
+from pytest_mock import MockerFixture
 
-from ..pynwb_utils import _sanitize_nwb_version, nwb_has_external_links
+from ..pynwb_utils import (
+    _rename_pose_estimation_original_videos,
+    _sanitize_nwb_version,
+    nwb_has_external_links,
+    rename_nwb_external_files,
+)
 
 
 def test_pynwb_io(simple1_nwb: Path) -> None:
@@ -53,6 +62,122 @@ def search(v: str) -> None:
     )
 
 
+@pytest.mark.ai_generated
+def test_rename_pose_estimation_original_videos() -> None:
+    pose = SimpleNamespace(
+        neurodata_type="PoseEstimation",
+        original_videos=[b"camera\\raw.mp4", "https://example.com/remote.mp4", "other.mp4"],
+    )
+    unrelated = SimpleNamespace(
+        neurodata_type="OtherContainer", original_videos=["camera/raw.mp4"]
+    )
+    nwb = SimpleNamespace(objects={"pose": pose, "unrelated": unrelated})
+
+    _rename_pose_estimation_original_videos(
+        nwb,
+        {"camera/raw.mp4": "sub-01/session-01/source.mp4"},
+    )
+
+    assert pose.original_videos == [
+        "sub-01/session-01/source.mp4",
+        "https://example.com/remote.mp4",
+        "other.mp4",
+    ]
+    assert unrelated.original_videos == ["camera/raw.mp4"]
+
+
+@pytest.mark.ai_generated
+def test_rename_pose_estimation_original_videos_ignores_missing_values() -> None:
+    missing = SimpleNamespace(neurodata_type="PoseEstimation")
+    scalar = SimpleNamespace(
+        neurodata_type="PoseEstimation", original_videos="camera/raw.mp4"
+    )
+    scalar_bytes = SimpleNamespace(
+        neurodata_type="PoseEstimation", original_videos=b"camera/raw.mp4"
+    )
+    nwb = SimpleNamespace(objects={"missing": missing, "scalar": scalar})
+
+    _rename_pose_estimation_original_videos(nwb, {})
+    nwb.objects["scalar_bytes"] = scalar_bytes
+    _rename_pose_estimation_original_videos(
+        nwb, {"camera/raw.mp4": "sub-01/source.mp4"}
+    )
+
+    assert scalar.original_videos == "camera/raw.mp4"
+    assert scalar_bytes.original_videos == b"camera/raw.mp4"
+
+
+@pytest.mark.ai_generated
+def test_rename_pose_estimation_original_videos_persists_hdf5(
+    tmp_path: Path,
+) -> None:
+    filepath = tmp_path / "pose-videos.h5"
+    string_type = h5py.string_dtype(encoding="utf-8")
+    with h5py.File(filepath, "w") as f:
+        f.create_dataset(
+            "original_videos",
+            data=np.asarray(
+                ["camera/raw.mp4", "camera/other.mp4"], dtype=string_type
+            ),
+        )
+
+    with h5py.File(filepath, "r+") as f:
+        pose = SimpleNamespace(
+            neurodata_type="PoseEstimation",
+            original_videos=f["original_videos"],
+        )
+        nwb = SimpleNamespace(objects={"pose": pose})
+        _rename_pose_estimation_original_videos(
+            nwb,
+            {"camera/raw.mp4": "sub-01/session-01/a-much-longer-source-name.mp4"},
+        )
+
+    with h5py.File(filepath) as f:
+        assert f["original_videos"].asstr()[...].tolist() == [
+            "sub-01/session-01/a-much-longer-source-name.mp4",
+            "camera/other.mp4",
+        ]
+
+
+@pytest.mark.ai_generated
+def test_rename_nwb_external_files_updates_pose_references(
+    tmp_path: Path, mocker: MockerFixture
+) -> None:
+    image_series = SimpleNamespace(
+        object_id="image-series-id", external_file=["camera/raw.mp4"]
+    )
+    pose = SimpleNamespace(
+        neurodata_type="PoseEstimation", original_videos=["camera/raw.mp4"]
+    )
+    nwb = SimpleNamespace(
+        children=[image_series], objects={"image-series": image_series, "pose": pose}
+    )
+    io = mocker.MagicMock()
+    io.__enter__.return_value.read.return_value = nwb
+    nwb_io = mocker.patch("dandi.pynwb_utils.NWBHDF5IO", return_value=io)
+    metadata = [
+        {
+            "path": "original.nwb",
+            "dandi_path": "sub-01/sub-01.nwb",
+            "external_file_objects": [
+                {
+                    "id": "image-series-id",
+                    "external_files": ["camera/raw.mp4"],
+                    "external_files_renamed": ["sub-01/camera-renamed.mp4"],
+                }
+            ],
+        }
+    ]
+
+    rename_nwb_external_files(metadata, str(tmp_path))
+
+    assert image_series.external_file == ["sub-01/camera-renamed.mp4"]
+    assert pose.original_videos == ["sub-01/camera-renamed.mp4"]
+    nwb_io.assert_called_once()
+    assert Path(nwb_io.call_args.args[0]) == tmp_path / "sub-01" / "sub-01.nwb"
+    assert nwb_io.call_args.kwargs == {"mode": "r+", "load_namespaces": True}
+
+
 def test_nwb_has_external_links(tmp_path):
     # Create the base data
     start_time = datetime(2017, 4, 3, 11, tzinfo=timezone.utc)