diff --git a/dandi/pynwb_utils.py b/dandi/pynwb_utils.py index bc4a0bc5d..80fba2a20 100644 --- a/dandi/pynwb_utils.py +++ b/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. diff --git a/dandi/tests/test_pynwb_utils.py b/dandi/tests/test_pynwb_utils.py index 0de33d555..52b49367e 100644 --- a/dandi/tests/test_pynwb_utils.py +++ b/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)