diff --git a/.github/workflows/testing.yml b/.github/workflows/testing.yml index d8c5bb9fdd..644329cbdc 100644 --- a/.github/workflows/testing.yml +++ b/.github/workflows/testing.yml @@ -98,6 +98,12 @@ jobs: s3-gin-bucket: ${{ secrets.S3_GIN_BUCKET }} os: ${{ matrix.os }} + # TODO: remove this setp after this is merged https://github.com/talmolab/sleap-io/pull/143 + - name: Run Sleap Tests until sleap.io adds support for ndx-pose > 2.0 + run : | + pip install ndx-pose==0.1.1 + pytest tests/test_on_data/behavior/test_pose_estimation_interfaces.py + - name: Install full requirements run: pip install .[full] diff --git a/CHANGELOG.md b/CHANGELOG.md index 3cab36eab6..6df9fd842f 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,12 +6,14 @@ ## Features * Added `metadata` and `conversion_options` as arguments to `NWBConverter.temporally_align_data_interfaces` [PR #1162](https://github.com/catalystneuro/neuroconv/pull/1162) +* Use the latest version of ndx-pose for `DeepLabCutInterface` and `LightningPoseDataInterface` [PR #1128](https://github.com/catalystneuro/neuroconv/pull/1128) ## Improvements * Interfaces and converters now have `verbose=False` by default [PR #1153](https://github.com/catalystneuro/neuroconv/pull/1153) -# v0.6.7 (January 20, 2024) + +# v0.6.7 (January 20, 2025) ## Deprecations diff --git a/docs/conversion_examples_gallery/conftest.py b/docs/conversion_examples_gallery/conftest.py index 6618d6d529..775eb4f9d2 100644 --- a/docs/conversion_examples_gallery/conftest.py +++ b/docs/conversion_examples_gallery/conftest.py @@ -1,4 +1,5 @@ import platform +from importlib.metadata import version as importlib_version from pathlib import Path import pytest @@ -29,9 +30,15 @@ def add_data_space(doctest_namespace, tmp_path): # Hook to conditionally skip doctests in deeplabcut.rst for Python 3.9 on macOS (Darwin) def pytest_runtest_setup(item): if isinstance(item, pytest.DoctestItem): - # Check if we are running the doctest from deeplabcut.rst test_file = Path(item.fspath) + # Check if we are running the doctest from deeplabcut.rst if test_file.name == "deeplabcut.rst": # Check if Python version is 3.9 and platform is Darwin (macOS) if version.parse(python_version) < version.parse("3.10") and os == "Darwin": pytest.skip("Skipping doctests for deeplabcut.rst on Python 3.9 and macOS") + # Check if we are running the doctest from sleap.rst + # TODO: remove after this is merged https://github.com/talmolab/sleap-io/pull/143 and released + elif test_file.name in ["ecephys_pose_estimation.rst", "sleap.rst"]: + ndx_pose_version = version.parse(importlib_version("ndx-pose")) + if ndx_pose_version >= version.parse("0.2.0"): + pytest.skip("Skipping doctests because sleeps only run when ndx-pose version < 0.2.0") diff --git a/pyproject.toml b/pyproject.toml index 8a33fb9653..5915a2b544 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -118,7 +118,7 @@ sleap = [ "sleap-io>=0.0.2; python_version>='3.9'", ] deeplabcut = [ - "ndx-pose==0.1.1", + "ndx-pose>=0.2", "tables; platform_system != 'Darwin'", "tables>=3.10.1; platform_system == 'Darwin' and python_version >= '3.10'", ] @@ -128,7 +128,7 @@ video = [ "opencv-python-headless>=4.8.1.78", ] lightningpose = [ - "ndx-pose==0.1.1", + "ndx-pose>=0.2", "neuroconv[video]", ] medpc = [ diff --git a/src/neuroconv/basedatainterface.py b/src/neuroconv/basedatainterface.py index 64af908e38..95b80f6d73 100644 --- a/src/neuroconv/basedatainterface.py +++ b/src/neuroconv/basedatainterface.py @@ -212,9 +212,7 @@ def run_conversion( @staticmethod def get_default_backend_configuration( nwbfile: NWBFile, - # TODO: when all H5DataIO prewraps are gone, introduce Zarr safely - # backend: Union[Literal["hdf5", "zarr"]], - backend: Literal["hdf5"] = "hdf5", + backend: Literal["hdf5", "zarr"] = "hdf5", ) -> Union[HDF5BackendConfiguration, ZarrBackendConfiguration]: """ Fill and return a default backend configuration to serve as a starting point for further customization. diff --git a/src/neuroconv/datainterfaces/behavior/deeplabcut/_dlc_utils.py b/src/neuroconv/datainterfaces/behavior/deeplabcut/_dlc_utils.py index 14866510d2..1c64c59a94 100644 --- a/src/neuroconv/datainterfaces/behavior/deeplabcut/_dlc_utils.py +++ b/src/neuroconv/datainterfaces/behavior/deeplabcut/_dlc_utils.py @@ -1,4 +1,3 @@ -import importlib import pickle import warnings from pathlib import Path @@ -11,6 +10,8 @@ from pynwb import NWBFile from ruamel.yaml import YAML +from ....tools import get_module + def _read_config(config_file_path: FilePath) -> dict: """ @@ -93,7 +94,7 @@ def _get_cv2_timestamps(file_path: Union[Path, str]): return timestamps -def _get_movie_timestamps(movie_file, VARIABILITYBOUND=1000, infer_timestamps=True): +def _get_video_timestamps(movie_file, VARIABILITYBOUND=1000, infer_timestamps=True): """ Return numpy array of the timestamps for a video. @@ -263,13 +264,44 @@ def _write_pes_to_nwbfile( exclude_nans, pose_estimation_container_kwargs: Optional[dict] = None, ): - - from ndx_pose import PoseEstimation, PoseEstimationSeries + """ + Updated version of _write_pes_to_nwbfile to work with ndx-pose v0.2.0+ + """ + from ndx_pose import PoseEstimation, PoseEstimationSeries, Skeleton, Skeletons + from pynwb.file import Subject pose_estimation_container_kwargs = pose_estimation_container_kwargs or dict() + pose_estimation_name = pose_estimation_container_kwargs.get("name", "PoseEstimationDeepLabCut") + + # Create a subject if it doesn't exist + if nwbfile.subject is None: + subject = Subject(subject_id=animal) + nwbfile.subject = subject + else: + subject = nwbfile.subject + + # Create skeleton from the keypoints + keypoints = df_animal.columns.get_level_values("bodyparts").unique() + animal = animal if animal else "" + subject = subject if animal == subject.subject_id else None + skeleton_name = f"Skeleton{pose_estimation_name}_{animal.capitalize()}" + skeleton = Skeleton( + name=skeleton_name, + nodes=list(keypoints), + edges=np.array(paf_graph) if paf_graph else None, # Convert paf_graph to numpy array + subject=subject, + ) + + behavior_processing_module = get_module(nwbfile=nwbfile, name="behavior", description="processed behavioral data") + if "Skeletons" not in behavior_processing_module.data_interfaces: + skeletons = Skeletons(skeletons=[skeleton]) + behavior_processing_module.add(skeletons) + else: + skeletons = behavior_processing_module["Skeletons"] + skeletons.add_skeletons(skeleton) pose_estimation_series = [] - for keypoint in df_animal.columns.get_level_values("bodyparts").unique(): + for keypoint in keypoints: data = df_animal.xs(keypoint, level="bodyparts", axis=1).to_numpy() if exclude_nans: @@ -292,35 +324,31 @@ def _write_pes_to_nwbfile( ) pose_estimation_series.append(pes) - deeplabcut_version = None - is_deeplabcut_installed = importlib.util.find_spec(name="deeplabcut") is not None - if is_deeplabcut_installed: - deeplabcut_version = importlib.metadata.version(distribution_name="deeplabcut") + camera_name = pose_estimation_name + if camera_name not in nwbfile.devices: + camera = nwbfile.create_device( + name=camera_name, + description="Camera used for behavioral recording and pose estimation.", + ) + else: + camera = nwbfile.devices[camera_name] - # TODO, taken from the original implementation, improve it if the video is passed + # Create PoseEstimation container with updated arguments dimensions = [list(map(int, image_shape.split(",")))[1::2]] dimensions = np.array(dimensions, dtype="uint32") pose_estimation_default_kwargs = dict( pose_estimation_series=pose_estimation_series, description="2D keypoint coordinates estimated using DeepLabCut.", - original_videos=[video_file_path], + original_videos=[video_file_path] if video_file_path else None, dimensions=dimensions, + devices=[camera], scorer=scorer, source_software="DeepLabCut", - source_software_version=deeplabcut_version, - nodes=[pes.name for pes in pose_estimation_series], - edges=paf_graph if paf_graph else None, - **pose_estimation_container_kwargs, + skeleton=skeleton, ) pose_estimation_default_kwargs.update(pose_estimation_container_kwargs) pose_estimation_container = PoseEstimation(**pose_estimation_default_kwargs) - if "behavior" in nwbfile.processing: # TODO: replace with get_module - behavior_processing_module = nwbfile.processing["behavior"] - else: - behavior_processing_module = nwbfile.create_processing_module( - name="behavior", description="processed behavioral data" - ) behavior_processing_module.add(pose_estimation_container) return nwbfile @@ -387,7 +415,7 @@ def _add_subject_to_nwbfile( if video_file_path is None: timestamps = df.index.tolist() # setting timestamps to dummy else: - timestamps = _get_movie_timestamps(video_file_path, infer_timestamps=True) + timestamps = _get_video_timestamps(video_file_path, infer_timestamps=True) # Fetch the corresponding metadata pickle file, we extract the edges graph from here # TODO: This is the original implementation way to extract the file name but looks very brittle. Improve it diff --git a/src/neuroconv/datainterfaces/behavior/deeplabcut/deeplabcutdatainterface.py b/src/neuroconv/datainterfaces/behavior/deeplabcut/deeplabcutdatainterface.py index 8c81a0264c..c73551e596 100644 --- a/src/neuroconv/datainterfaces/behavior/deeplabcut/deeplabcutdatainterface.py +++ b/src/neuroconv/datainterfaces/behavior/deeplabcut/deeplabcutdatainterface.py @@ -12,7 +12,7 @@ class DeepLabCutInterface(BaseTemporalAlignmentInterface): """Data interface for DeepLabCut datasets.""" display_name = "DeepLabCut" - keywords = ("DLC",) + keywords = ("DLC", "DeepLabCut", "pose estimation", "behavior") associated_suffixes = (".h5", ".csv") info = "Interface for handling data from DeepLabCut." @@ -48,7 +48,18 @@ def __init__( Controls verbosity. """ # This import is to assure that the ndx_pose is in the global namespace when an pynwb.io object is created - from ndx_pose import PoseEstimation, PoseEstimationSeries # noqa: F401 + from importlib.metadata import version + + import ndx_pose # noqa: F401 + from packaging import version as version_parse + + ndx_pose_version = version("ndx-pose") + if version_parse.parse(ndx_pose_version) < version_parse.parse("0.2.0"): + raise ImportError( + "DeepLabCut interface requires ndx-pose version 0.2.0 or later. " + f"Found version {ndx_pose_version}. Please upgrade: " + "pip install 'ndx-pose>=0.2.0'" + ) from ._dlc_utils import _read_config @@ -62,6 +73,8 @@ def __init__( self.config_dict = _read_config(config_file_path=config_file_path) self.subject_name = subject_name self.verbose = verbose + self.pose_estimation_container_kwargs = dict() + super().__init__(file_path=file_path, config_file_path=config_file_path) def get_metadata(self): @@ -101,7 +114,7 @@ def add_to_nwbfile( self, nwbfile: NWBFile, metadata: Optional[dict] = None, - container_name: str = "PoseEstimation", + container_name: str = "PoseEstimationDeepLabCut", ): """ Conversion from DLC output files to nwb. Derived from dlc2nwb library. @@ -112,16 +125,19 @@ def add_to_nwbfile( nwb file to which the recording information is to be added metadata: dict metadata info for constructing the nwb file (optional). - container_name: str, default: "PoseEstimation" - Name of the container to store the pose estimation. + container_name: str, default: "PoseEstimationDeepLabCut" + name of the PoseEstimation container in the nwb + """ from ._dlc_utils import _add_subject_to_nwbfile + self.pose_estimation_container_kwargs["name"] = container_name + _add_subject_to_nwbfile( nwbfile=nwbfile, file_path=str(self.source_data["file_path"]), individual_name=self.subject_name, config_file=self.source_data["config_file_path"], timestamps=self._timestamps, - pose_estimation_container_kwargs=dict(name=container_name), + pose_estimation_container_kwargs=self.pose_estimation_container_kwargs, ) diff --git a/src/neuroconv/datainterfaces/behavior/lightningpose/lightningposedatainterface.py b/src/neuroconv/datainterfaces/behavior/lightningpose/lightningposedatainterface.py index 1211c31d15..05f569a525 100644 --- a/src/neuroconv/datainterfaces/behavior/lightningpose/lightningposedatainterface.py +++ b/src/neuroconv/datainterfaces/behavior/lightningpose/lightningposedatainterface.py @@ -40,9 +40,10 @@ def get_metadata_schema(self) -> dict: description=dict(type="string"), scorer=dict(type="string"), source_software=dict(type="string", default="LightningPose"), + camera_name=dict(type="string", default="CameraPoseEstimation"), ), patternProperties={ - "^(?!(name|description|scorer|source_software)$)[a-zA-Z0-9_]+$": dict( + "^(?!(name|description|scorer|source_software|camera_name)$)[a-zA-Z0-9_]+$": dict( title="PoseEstimationSeries", type="object", properties=dict(name=dict(type="string"), description=dict(type="string")), @@ -80,9 +81,21 @@ def __init__( verbose : bool, default: False controls verbosity. ``True`` by default. """ + # This import is to assure that the ndx_pose is in the global namespace when an pynwb.io object is created # For more detail, see https://github.com/rly/ndx-pose/issues/36 + from importlib.metadata import version + import ndx_pose # noqa: F401 + from packaging import version as version_parse + + ndx_pose_version = version("ndx-pose") + if version_parse.parse(ndx_pose_version) < version_parse.parse("0.2.0"): + raise ImportError( + "LightningPose interface requires ndx-pose version 0.2.0 or later. " + f"Found version {ndx_pose_version}. Please upgrade: " + "pip install 'ndx-pose>=0.2.0'" + ) from neuroconv.datainterfaces.behavior.video.video_utils import ( VideoCaptureContext, @@ -162,6 +175,7 @@ def get_metadata(self) -> DeepDict: description="Contains the pose estimation series for each keypoint.", scorer=self.scorer_name, source_software="LightningPose", + camera_name="CameraPoseEstimation", ) for keypoint_name in self.keypoint_names: keypoint_name_without_spaces = keypoint_name.replace(" ", "") @@ -198,7 +212,7 @@ def add_to_nwbfile( The description of how the confidence was computed, e.g., 'Softmax output of the deep neural network'. stub_test : bool, default: False """ - from ndx_pose import PoseEstimation, PoseEstimationSeries + from ndx_pose import PoseEstimation, PoseEstimationSeries, Skeleton, Skeletons metadata_copy = deepcopy(metadata) @@ -215,15 +229,14 @@ def add_to_nwbfile( original_video_name = str(self.original_video_file_path) else: original_video_name = metadata_copy["Behavior"]["Videos"][0]["name"] - - pose_estimation_kwargs = dict( - name=pose_estimation_metadata["name"], - description=pose_estimation_metadata["description"], - source_software=pose_estimation_metadata["source_software"], - scorer=pose_estimation_metadata["scorer"], - original_videos=[original_video_name], - dimensions=[self.dimension], - ) + camera_name = pose_estimation_metadata["camera_name"] + if camera_name in nwbfile.devices: + camera = nwbfile.devices[camera_name] + else: + camera = nwbfile.create_device( + name=camera_name, + description="Camera used for behavioral recording and pose estimation.", + ) pose_estimation_data = self.pose_estimation_data if not stub_test else self.pose_estimation_data.head(n=10) timestamps = self.get_timestamps(stub_test=stub_test) @@ -255,8 +268,28 @@ def add_to_nwbfile( pose_estimation_series.append(PoseEstimationSeries(**pose_estimation_series_kwargs)) - pose_estimation_kwargs.update( + # Add Skeleton(s) + nodes = [keypoint_name.replace(" ", "") for keypoint_name in self.keypoint_names] + subject = nwbfile.subject if nwbfile.subject is not None else None + name = f"Skeleton{pose_estimation_name}" + skeleton = Skeleton(name=name, nodes=nodes, subject=subject) + if "Skeletons" in behavior.data_interfaces: + skeletons = behavior.data_interfaces["Skeletons"] + skeletons.add_skeletons(skeleton) + else: + skeletons = Skeletons(skeletons=[skeleton]) + behavior.add(skeletons) + + pose_estimation_kwargs = dict( + name=pose_estimation_metadata["name"], + description=pose_estimation_metadata["description"], + source_software=pose_estimation_metadata["source_software"], + scorer=pose_estimation_metadata["scorer"], + original_videos=[original_video_name], + dimensions=[self.dimension], pose_estimation_series=pose_estimation_series, + devices=[camera], + skeleton=skeleton, ) if self.source_data["labeled_video_file_path"]: diff --git a/src/neuroconv/datainterfaces/behavior/sleap/sleapdatainterface.py b/src/neuroconv/datainterfaces/behavior/sleap/sleapdatainterface.py index 90b9b91e14..c848202705 100644 --- a/src/neuroconv/datainterfaces/behavior/sleap/sleapdatainterface.py +++ b/src/neuroconv/datainterfaces/behavior/sleap/sleapdatainterface.py @@ -49,6 +49,24 @@ def __init__( frames_per_second : float, optional The frames per second (fps) or sampling rate of the video. """ + + # This import is to assure that the ndx_pose is in the global namespace when an pynwb.io object is created + # For more detail, see https://github.com/rly/ndx-pose/issues/36 + from importlib.metadata import version + + import ndx_pose # noqa: F401 + from packaging import version as version_parse + + ndx_pose_version = version("ndx-pose") + + # TODO: remove after this is merged https://github.com/talmolab/sleap-io/pull/143 and released + if version_parse.parse(ndx_pose_version) != version_parse.parse("0.1.1"): + raise ImportError( + "SLEAP interface requires ndx-pose version 0.1.1. " + f"Found version {ndx_pose_version}. Please install the required version: " + "pip install 'ndx-pose==0.1.1'" + ) + self.file_path = Path(file_path) self.sleap_io = get_package(package_name="sleap_io") self.video_file_path = video_file_path diff --git a/tests/test_on_data/behavior/test_behavior_interfaces.py b/tests/test_on_data/behavior/test_behavior_interfaces.py index 0b5c633766..2115e6dd47 100644 --- a/tests/test_on_data/behavior/test_behavior_interfaces.py +++ b/tests/test_on_data/behavior/test_behavior_interfaces.py @@ -1,30 +1,23 @@ -import sys import unittest from datetime import datetime, timezone from pathlib import Path import numpy as np -import pandas as pd import pytest -import sleap_io from hdmf.testing import TestCase from natsort import natsorted from ndx_miniscope import Miniscope from ndx_miniscope.utils import get_timestamps from numpy.testing import assert_array_equal -from parameterized import param, parameterized from pynwb import NWBHDF5IO from pynwb.behavior import Position, SpatialSeries from neuroconv import NWBConverter from neuroconv.datainterfaces import ( - DeepLabCutInterface, FicTracDataInterface, - LightningPoseDataInterface, MedPCInterface, MiniscopeBehaviorInterface, NeuralynxNvtInterface, - SLEAPInterface, VideoInterface, ) from neuroconv.tools.testing.data_interface_mixins import ( @@ -33,7 +26,6 @@ TemporalAlignmentMixin, VideoInterfaceMixin, ) -from neuroconv.utils import DeepDict try: from ..setup_paths import BEHAVIOR_DATA_PATH, OPHYS_DATA_PATH, OUTPUT_PATH @@ -41,142 +33,6 @@ from setup_paths import BEHAVIOR_DATA_PATH, OUTPUT_PATH -class TestLightningPoseDataInterface(DataInterfaceTestMixin, TemporalAlignmentMixin): - data_interface_cls = LightningPoseDataInterface - interface_kwargs = dict( - file_path=str(BEHAVIOR_DATA_PATH / "lightningpose" / "outputs/2023-11-09/10-14-37/video_preds/test_vid.csv"), - original_video_file_path=str( - BEHAVIOR_DATA_PATH / "lightningpose" / "outputs/2023-11-09/10-14-37/video_preds/test_vid.mp4" - ), - ) - conversion_options = dict(reference_frame="(0,0) corresponds to the top left corner of the video.") - save_directory = OUTPUT_PATH - - @pytest.fixture(scope="class", autouse=True) - def setup_metadata(self, request): - - cls = request.cls - - cls.pose_estimation_name = "PoseEstimation" - cls.original_video_height = 406 - cls.original_video_width = 396 - cls.expected_keypoint_names = [ - "paw1LH_top", - "paw2LF_top", - "paw3RF_top", - "paw4RH_top", - "tailBase_top", - "tailMid_top", - "nose_top", - "obs_top", - "paw1LH_bot", - "paw2LF_bot", - "paw3RF_bot", - "paw4RH_bot", - "tailBase_bot", - "tailMid_bot", - "nose_bot", - "obsHigh_bot", - "obsLow_bot", - ] - cls.expected_metadata = DeepDict( - PoseEstimation=dict( - name=cls.pose_estimation_name, - description="Contains the pose estimation series for each keypoint.", - scorer="heatmap_tracker", - source_software="LightningPose", - ) - ) - cls.expected_metadata[cls.pose_estimation_name].update( - { - keypoint_name: dict( - name=f"PoseEstimationSeries{keypoint_name}", - description=f"The estimated position (x, y) of {keypoint_name} over time.", - ) - for keypoint_name in cls.expected_keypoint_names - } - ) - - cls.test_data = pd.read_csv(cls.interface_kwargs["file_path"], header=[0, 1, 2])["heatmap_tracker"] - - def check_extracted_metadata(self, metadata: dict): - assert metadata["NWBFile"]["session_start_time"] == datetime(2023, 11, 9, 10, 14, 37, 0) - assert self.pose_estimation_name in metadata["Behavior"] - assert metadata["Behavior"][self.pose_estimation_name] == self.expected_metadata[self.pose_estimation_name] - - def check_read_nwb(self, nwbfile_path: str): - from ndx_pose import PoseEstimation, PoseEstimationSeries - - with NWBHDF5IO(path=nwbfile_path, mode="r", load_namespaces=True) as io: - nwbfile = io.read() - - # Replacing assertIn with pytest-style assert - assert "behavior" in nwbfile.processing - assert self.pose_estimation_name in nwbfile.processing["behavior"].data_interfaces - - pose_estimation_container = nwbfile.processing["behavior"].data_interfaces[self.pose_estimation_name] - - # Replacing assertIsInstance with pytest-style assert - assert isinstance(pose_estimation_container, PoseEstimation) - - pose_estimation_metadata = self.expected_metadata[self.pose_estimation_name] - - # Replacing assertEqual with pytest-style assert - assert pose_estimation_container.description == pose_estimation_metadata["description"] - assert pose_estimation_container.scorer == pose_estimation_metadata["scorer"] - assert pose_estimation_container.source_software == pose_estimation_metadata["source_software"] - - # Using numpy's assert_array_equal - assert_array_equal( - pose_estimation_container.dimensions[:], [[self.original_video_height, self.original_video_width]] - ) - - # Replacing assertEqual with pytest-style assert - assert len(pose_estimation_container.pose_estimation_series) == len(self.expected_keypoint_names) - - for keypoint_name in self.expected_keypoint_names: - series_metadata = pose_estimation_metadata[keypoint_name] - - # Replacing assertIn with pytest-style assert - assert series_metadata["name"] in pose_estimation_container.pose_estimation_series - - pose_estimation_series = pose_estimation_container.pose_estimation_series[series_metadata["name"]] - - # Replacing assertIsInstance with pytest-style assert - assert isinstance(pose_estimation_series, PoseEstimationSeries) - - # Replacing assertEqual with pytest-style assert - assert pose_estimation_series.unit == "px" - assert pose_estimation_series.description == series_metadata["description"] - assert pose_estimation_series.reference_frame == self.conversion_options["reference_frame"] - - test_data = self.test_data[keypoint_name] - - # Using numpy's assert_array_equal - assert_array_equal(pose_estimation_series.data[:], test_data[["x", "y"]].values) - - -class TestLightningPoseDataInterfaceWithStubTest(DataInterfaceTestMixin, TemporalAlignmentMixin): - data_interface_cls = LightningPoseDataInterface - interface_kwargs = dict( - file_path=str(BEHAVIOR_DATA_PATH / "lightningpose" / "outputs/2023-11-09/10-14-37/video_preds/test_vid.csv"), - original_video_file_path=str( - BEHAVIOR_DATA_PATH / "lightningpose" / "outputs/2023-11-09/10-14-37/video_preds/test_vid.mp4" - ), - ) - - conversion_options = dict(stub_test=True) - save_directory = OUTPUT_PATH - - def check_read_nwb(self, nwbfile_path: str): - with NWBHDF5IO(path=nwbfile_path, mode="r", load_namespaces=True) as io: - nwbfile = io.read() - pose_estimation_container = nwbfile.processing["behavior"].data_interfaces["PoseEstimation"] - for pose_estimation_series in pose_estimation_container.pose_estimation_series.values(): - assert pose_estimation_series.data.shape[0] == 10 - assert pose_estimation_series.confidence.shape[0] == 10 - - class TestFicTracDataInterface(DataInterfaceTestMixin): data_interface_cls = FicTracDataInterface interface_kwargs = dict( @@ -321,268 +177,6 @@ class TestFicTracDataInterfaceTiming(TemporalAlignmentMixin): save_directory = OUTPUT_PATH -from platform import python_version - -from packaging import version - -python_version = version.parse(python_version()) -from sys import platform - - -@pytest.mark.skipif( - platform == "darwin" and python_version < version.parse("3.10"), - reason="interface not supported on macOS with Python < 3.10", -) -class TestDeepLabCutInterface(DataInterfaceTestMixin): - data_interface_cls = DeepLabCutInterface - interface_kwargs = dict( - file_path=str( - BEHAVIOR_DATA_PATH - / "DLC" - / "open_field_without_video" - / "m3v1mp4DLC_resnet50_openfieldAug20shuffle1_30000.h5" - ), - config_file_path=str(BEHAVIOR_DATA_PATH / "DLC" / "open_field_without_video" / "config.yaml"), - subject_name="ind1", - ) - save_directory = OUTPUT_PATH - - def run_custom_checks(self): - self.check_renaming_instance(nwbfile_path=self.nwbfile_path) - - def check_renaming_instance(self, nwbfile_path: str): - custom_container_name = "TestPoseEstimation" - - metadata = self.interface.get_metadata() - metadata["NWBFile"].update(session_start_time=datetime.now().astimezone()) - - self.interface.run_conversion( - nwbfile_path=nwbfile_path, overwrite=True, metadata=metadata, container_name=custom_container_name - ) - - with NWBHDF5IO(path=nwbfile_path, mode="r", load_namespaces=True) as io: - nwbfile = io.read() - assert "behavior" in nwbfile.processing - assert "PoseEstimation" not in nwbfile.processing["behavior"].data_interfaces - assert custom_container_name in nwbfile.processing["behavior"].data_interfaces - - def check_read_nwb(self, nwbfile_path: str): - with NWBHDF5IO(path=nwbfile_path, mode="r", load_namespaces=True) as io: - nwbfile = io.read() - assert "behavior" in nwbfile.processing - processing_module_interfaces = nwbfile.processing["behavior"].data_interfaces - assert "PoseEstimation" in processing_module_interfaces - - pose_estimation_series_in_nwb = processing_module_interfaces["PoseEstimation"].pose_estimation_series - expected_pose_estimation_series = ["ind1_leftear", "ind1_rightear", "ind1_snout", "ind1_tailbase"] - - expected_pose_estimation_series_are_in_nwb_file = [ - pose_estimation in pose_estimation_series_in_nwb for pose_estimation in expected_pose_estimation_series - ] - - assert all(expected_pose_estimation_series_are_in_nwb_file) - - -@pytest.fixture -def clean_pose_extension_import(): - modules_to_remove = [m for m in sys.modules if m.startswith("ndx_pose")] - for module in modules_to_remove: - del sys.modules[module] - - -@pytest.mark.skipif( - platform == "darwin" and python_version < version.parse("3.10"), - reason="interface not supported on macOS with Python < 3.10", -) -def test_deep_lab_cut_import_pose_extension_bug(clean_pose_extension_import, tmp_path): - """ - Test that the DeepLabCutInterface writes correctly without importing the ndx-pose extension. - See issues: - https://github.com/catalystneuro/neuroconv/issues/1114 - https://github.com/rly/ndx-pose/issues/36 - - """ - - interface_kwargs = dict( - file_path=str( - BEHAVIOR_DATA_PATH - / "DLC" - / "open_field_without_video" - / "m3v1mp4DLC_resnet50_openfieldAug20shuffle1_30000.h5" - ), - config_file_path=str(BEHAVIOR_DATA_PATH / "DLC" / "open_field_without_video" / "config.yaml"), - ) - - interface = DeepLabCutInterface(**interface_kwargs) - metadata = interface.get_metadata() - metadata["NWBFile"]["session_start_time"] = datetime(2023, 7, 24, 9, 30, 55, 440600, tzinfo=timezone.utc) - - nwbfile_path = tmp_path / "test.nwb" - interface.run_conversion(nwbfile_path=nwbfile_path, metadata=metadata, overwrite=True) - with NWBHDF5IO(path=nwbfile_path, mode="r") as io: - read_nwbfile = io.read() - pose_estimation_container = read_nwbfile.processing["behavior"]["PoseEstimation"] - - assert len(pose_estimation_container.fields) > 0 - - -@pytest.mark.skipif( - platform == "darwin" and python_version < version.parse("3.10"), - reason="interface not supported on macOS with Python < 3.10", -) -class TestDeepLabCutInterfaceNoConfigFile(DataInterfaceTestMixin): - data_interface_cls = DeepLabCutInterface - interface_kwargs = dict( - file_path=str( - BEHAVIOR_DATA_PATH - / "DLC" - / "open_field_without_video" - / "m3v1mp4DLC_resnet50_openfieldAug20shuffle1_30000.h5" - ), - config_file_path=None, - subject_name="ind1", - ) - save_directory = OUTPUT_PATH - - def check_read_nwb(self, nwbfile_path: str): - with NWBHDF5IO(path=nwbfile_path, mode="r", load_namespaces=True) as io: - nwbfile = io.read() - assert "behavior" in nwbfile.processing - processing_module_interfaces = nwbfile.processing["behavior"].data_interfaces - assert "PoseEstimation" in processing_module_interfaces - - pose_estimation_series_in_nwb = processing_module_interfaces["PoseEstimation"].pose_estimation_series - expected_pose_estimation_series = ["ind1_leftear", "ind1_rightear", "ind1_snout", "ind1_tailbase"] - - expected_pose_estimation_series_are_in_nwb_file = [ - pose_estimation in pose_estimation_series_in_nwb for pose_estimation in expected_pose_estimation_series - ] - - assert all(expected_pose_estimation_series_are_in_nwb_file) - - -@pytest.mark.skipif( - platform == "darwin" and python_version < version.parse("3.10"), - reason="interface not supported on macOS with Python < 3.10", -) -class TestDeepLabCutInterfaceSetTimestamps(DataInterfaceTestMixin): - data_interface_cls = DeepLabCutInterface - interface_kwargs = dict( - file_path=str( - BEHAVIOR_DATA_PATH - / "DLC" - / "open_field_without_video" - / "m3v1mp4DLC_resnet50_openfieldAug20shuffle1_30000.h5" - ), - config_file_path=str(BEHAVIOR_DATA_PATH / "DLC" / "open_field_without_video" / "config.yaml"), - subject_name="ind1", - ) - - save_directory = OUTPUT_PATH - - def run_custom_checks(self): - self.check_custom_timestamps(nwbfile_path=self.nwbfile_path) - - def check_custom_timestamps(self, nwbfile_path: str): - custom_timestamps = np.concatenate( - (np.linspace(10, 110, 1000), np.linspace(150, 250, 1000), np.linspace(300, 400, 330)) - ) - - metadata = self.interface.get_metadata() - metadata["NWBFile"].update(session_start_time=datetime.now().astimezone()) - - self.interface.set_aligned_timestamps(custom_timestamps) - assert len(self.interface._timestamps) == 2330 - - self.interface.run_conversion(nwbfile_path=nwbfile_path, metadata=metadata, overwrite=True) - - with NWBHDF5IO(path=nwbfile_path, mode="r", load_namespaces=True) as io: - nwbfile = io.read() - assert "behavior" in nwbfile.processing - processing_module_interfaces = nwbfile.processing["behavior"].data_interfaces - assert "PoseEstimation" in processing_module_interfaces - - pose_estimation_series_in_nwb = processing_module_interfaces["PoseEstimation"].pose_estimation_series - - for pose_estimation in pose_estimation_series_in_nwb.values(): - pose_timestamps = pose_estimation.timestamps - np.testing.assert_array_equal(pose_timestamps, custom_timestamps) - - # This was tested in the other test - def check_read_nwb(self, nwbfile_path: str): - pass - - -@pytest.mark.skipif( - platform == "darwin" and python_version < version.parse("3.10"), - reason="interface not supported on macOS with Python < 3.10", -) -class TestDeepLabCutInterfaceFromCSV(DataInterfaceTestMixin): - data_interface_cls = DeepLabCutInterface - interface_kwargs = dict( - file_path=str( - BEHAVIOR_DATA_PATH - / "DLC" - / "SL18_csv" - / "SL18_D19_S01_F01_BOX_SLP_20230503_112642.1DLC_resnet50_SubLearnSleepBoxRedLightJun26shuffle1_100000_stubbed.csv" - ), - config_file_path=None, - subject_name="SL18", - ) - save_directory = OUTPUT_PATH - - def check_read_nwb(self, nwbfile_path: str): - with NWBHDF5IO(path=nwbfile_path, mode="r", load_namespaces=True) as io: - nwbfile = io.read() - assert "behavior" in nwbfile.processing - processing_module_interfaces = nwbfile.processing["behavior"].data_interfaces - assert "PoseEstimation" in processing_module_interfaces - - pose_estimation_series_in_nwb = processing_module_interfaces["PoseEstimation"].pose_estimation_series - expected_pose_estimation_series = ["SL18_redled", "SL18_shoulder", "SL18_haunch", "SL18_baseoftail"] - - expected_pose_estimation_series_are_in_nwb_file = [ - pose_estimation in pose_estimation_series_in_nwb for pose_estimation in expected_pose_estimation_series - ] - - assert all(expected_pose_estimation_series_are_in_nwb_file) - - -class TestSLEAPInterface(DataInterfaceTestMixin, TemporalAlignmentMixin): - data_interface_cls = SLEAPInterface - interface_kwargs = dict( - file_path=str(BEHAVIOR_DATA_PATH / "sleap" / "predictions_1.2.7_provenance_and_tracking.slp"), - video_file_path=str(BEHAVIOR_DATA_PATH / "sleap" / "melanogaster_courtship.mp4"), - ) - save_directory = OUTPUT_PATH - - def check_read_nwb(self, nwbfile_path: str): # This is currently structured to be file-specific - with NWBHDF5IO(path=nwbfile_path, mode="r", load_namespaces=True) as io: - nwbfile = io.read() - assert "SLEAP_VIDEO_000_20190128_113421" in nwbfile.processing - processing_module_interfaces = nwbfile.processing["SLEAP_VIDEO_000_20190128_113421"].data_interfaces - assert "track=track_0" in processing_module_interfaces - - pose_estimation_series_in_nwb = processing_module_interfaces["track=track_0"].pose_estimation_series - expected_pose_estimation_series = [ - "abdomen", - "eyeL", - "eyeR", - "forelegL4", - "forelegR4", - "head", - "hindlegL4", - "hindlegR4", - "midlegL4", - "midlegR4", - "thorax", - "wingL", - "wingR", - ] - - assert set(pose_estimation_series_in_nwb) == set(expected_pose_estimation_series) - - class TestMiniscopeInterface(DataInterfaceTestMixin): data_interface_cls = MiniscopeBehaviorInterface interface_kwargs = dict(folder_path=str(OPHYS_DATA_PATH / "imaging_datasets" / "Miniscope" / "C6-J588_Disc5")) @@ -662,121 +256,6 @@ def check_metadata(self): assert metadata["NWBFile"]["session_start_time"] == datetime(2023, 5, 15, 10, 35, 29) -class CustomTestSLEAPInterface(TestCase): - savedir = OUTPUT_PATH - - @parameterized.expand( - [ - param( - data_interface=SLEAPInterface, - interface_kwargs=dict( - file_path=str(BEHAVIOR_DATA_PATH / "sleap" / "predictions_1.2.7_provenance_and_tracking.slp"), - ), - ) - ] - ) - def test_sleap_to_nwb_interface(self, data_interface, interface_kwargs): - nwbfile_path = str(self.savedir / f"{data_interface.__name__}.nwb") - - interface = SLEAPInterface(**interface_kwargs) - metadata = interface.get_metadata() - metadata["NWBFile"].update(session_start_time=datetime.now().astimezone()) - interface.run_conversion(nwbfile_path=nwbfile_path, overwrite=True, metadata=metadata) - - slp_predictions_path = interface_kwargs["file_path"] - labels = sleap_io.load_slp(slp_predictions_path) - - with NWBHDF5IO(path=nwbfile_path, mode="r", load_namespaces=True) as io: - nwbfile = io.read() - # Test matching number of processing modules - number_of_videos = len(labels.videos) - assert len(nwbfile.processing) == number_of_videos - - # Test processing module naming as video - processing_module_name = "SLEAP_VIDEO_000_20190128_113421" - assert processing_module_name in nwbfile.processing - - # For this case we have as many containers as tracks - # Each track usually represents a subject - processing_module = nwbfile.processing[processing_module_name] - processing_module_interfaces = processing_module.data_interfaces - assert len(processing_module_interfaces) == len(labels.tracks) - - # Test name of PoseEstimation containers - extracted_container_names = processing_module_interfaces.keys() - for track in labels.tracks: - expected_track_name = f"track={track.name}" - assert expected_track_name in extracted_container_names - - # Test one PoseEstimation container - container_name = f"track={track.name}" - pose_estimation_container = processing_module_interfaces[container_name] - # Test that the skeleton nodes are store as nodes in containers - expected_node_names = [node.name for node in labels.skeletons[0]] - assert expected_node_names == list(pose_estimation_container.nodes[:]) - - # Test that each PoseEstimationSeries is named as a node - for node_name in pose_estimation_container.nodes[:]: - assert node_name in pose_estimation_container.pose_estimation_series - - @parameterized.expand( - [ - param( - data_interface=SLEAPInterface, - interface_kwargs=dict( - file_path=str(BEHAVIOR_DATA_PATH / "sleap" / "melanogaster_courtship.slp"), - video_file_path=str(BEHAVIOR_DATA_PATH / "sleap" / "melanogaster_courtship.mp4"), - ), - ) - ] - ) - def test_sleap_interface_timestamps_propagation(self, data_interface, interface_kwargs): - nwbfile_path = str(self.savedir / f"{data_interface.__name__}.nwb") - - interface = SLEAPInterface(**interface_kwargs) - metadata = interface.get_metadata() - metadata["NWBFile"].update(session_start_time=datetime.now().astimezone()) - interface.run_conversion(nwbfile_path=nwbfile_path, overwrite=True, metadata=metadata) - - slp_predictions_path = interface_kwargs["file_path"] - labels = sleap_io.load_slp(slp_predictions_path) - - from neuroconv.datainterfaces.behavior.sleap.sleap_utils import ( - extract_timestamps, - ) - - expected_timestamps = set(extract_timestamps(interface_kwargs["video_file_path"])) - - with NWBHDF5IO(path=nwbfile_path, mode="r", load_namespaces=True) as io: - nwbfile = io.read() - # Test matching number of processing modules - number_of_videos = len(labels.videos) - assert len(nwbfile.processing) == number_of_videos - - # Test processing module naming as video - video_name = Path(labels.videos[0].filename).stem - processing_module_name = f"SLEAP_VIDEO_000_{video_name}" - - # For this case we have as many containers as tracks - processing_module_interfaces = nwbfile.processing[processing_module_name].data_interfaces - - extracted_container_names = processing_module_interfaces.keys() - for track in labels.tracks: - expected_track_name = f"track={track.name}" - assert expected_track_name in extracted_container_names - - container_name = f"track={track.name}" - pose_estimation_container = processing_module_interfaces[container_name] - - # Test that each PoseEstimationSeries is named as a node - for node_name in pose_estimation_container.nodes[:]: - pose_estimation_series = pose_estimation_container.pose_estimation_series[node_name] - extracted_timestamps = pose_estimation_series.timestamps[:] - - # Some frames do not have predictions associated with them, so we test for sub-set - assert set(extracted_timestamps).issubset(expected_timestamps) - - class TestVideoInterface(VideoInterfaceMixin): data_interface_cls = VideoInterface save_directory = OUTPUT_PATH diff --git a/tests/test_on_data/behavior/test_lightningpose_converter.py b/tests/test_on_data/behavior/test_lightningpose_converter.py index dd93632a48..e72a3f6871 100644 --- a/tests/test_on_data/behavior/test_lightningpose_converter.py +++ b/tests/test_on_data/behavior/test_lightningpose_converter.py @@ -64,6 +64,7 @@ def setUpClass(cls) -> None: description="Contains the pose estimation series for each keypoint.", scorer="heatmap_tracker", source_software="LightningPose", + camera_name="CameraPoseEstimation", ) cls.pose_estimation_metadata.update( diff --git a/tests/test_on_data/behavior/test_pose_estimation_interfaces.py b/tests/test_on_data/behavior/test_pose_estimation_interfaces.py new file mode 100644 index 0000000000..5dbaa4633d --- /dev/null +++ b/tests/test_on_data/behavior/test_pose_estimation_interfaces.py @@ -0,0 +1,566 @@ +import sys +from datetime import datetime, timezone +from pathlib import Path + +import numpy as np +import pandas as pd +import pytest +import sleap_io +from hdmf.testing import TestCase +from numpy.testing import assert_array_equal +from parameterized import param, parameterized +from pynwb import NWBHDF5IO + +from neuroconv.datainterfaces import ( + DeepLabCutInterface, + LightningPoseDataInterface, + SLEAPInterface, +) +from neuroconv.tools.testing.data_interface_mixins import ( + DataInterfaceTestMixin, + TemporalAlignmentMixin, +) +from neuroconv.utils import DeepDict + +try: + from ..setup_paths import BEHAVIOR_DATA_PATH, OUTPUT_PATH +except ImportError: + from setup_paths import BEHAVIOR_DATA_PATH, OUTPUT_PATH + +from importlib.metadata import version as importlib_version +from platform import python_version +from sys import platform + +from packaging import version + +python_version = version.parse(python_version()) +# TODO: remove after this is merged https://github.com/talmolab/sleap-io/pull/143 and released +ndx_pose_version = version.parse(importlib_version("ndx-pose")) + + +@pytest.mark.skipif(ndx_pose_version < version.parse("0.2.0"), reason="Interface requires ndx-pose version >= 0.2.0") +class TestLightningPoseDataInterface(DataInterfaceTestMixin, TemporalAlignmentMixin): + data_interface_cls = LightningPoseDataInterface + interface_kwargs = dict( + file_path=str(BEHAVIOR_DATA_PATH / "lightningpose" / "outputs/2023-11-09/10-14-37/video_preds/test_vid.csv"), + original_video_file_path=str( + BEHAVIOR_DATA_PATH / "lightningpose" / "outputs/2023-11-09/10-14-37/video_preds/test_vid.mp4" + ), + ) + conversion_options = dict(reference_frame="(0,0) corresponds to the top left corner of the video.") + save_directory = OUTPUT_PATH + + @pytest.fixture(scope="class", autouse=True) + def setup_metadata(self, request): + + cls = request.cls + + cls.pose_estimation_name = "PoseEstimation" + cls.original_video_height = 406 + cls.original_video_width = 396 + cls.expected_keypoint_names = [ + "paw1LH_top", + "paw2LF_top", + "paw3RF_top", + "paw4RH_top", + "tailBase_top", + "tailMid_top", + "nose_top", + "obs_top", + "paw1LH_bot", + "paw2LF_bot", + "paw3RF_bot", + "paw4RH_bot", + "tailBase_bot", + "tailMid_bot", + "nose_bot", + "obsHigh_bot", + "obsLow_bot", + ] + cls.expected_metadata = DeepDict( + PoseEstimation=dict( + name=cls.pose_estimation_name, + description="Contains the pose estimation series for each keypoint.", + scorer="heatmap_tracker", + source_software="LightningPose", + camera_name="CameraPoseEstimation", + ) + ) + cls.expected_metadata[cls.pose_estimation_name].update( + { + keypoint_name: dict( + name=f"PoseEstimationSeries{keypoint_name}", + description=f"The estimated position (x, y) of {keypoint_name} over time.", + ) + for keypoint_name in cls.expected_keypoint_names + } + ) + + cls.test_data = pd.read_csv(cls.interface_kwargs["file_path"], header=[0, 1, 2])["heatmap_tracker"] + + def check_extracted_metadata(self, metadata: dict): + assert metadata["NWBFile"]["session_start_time"] == datetime(2023, 11, 9, 10, 14, 37, 0) + assert self.pose_estimation_name in metadata["Behavior"] + assert metadata["Behavior"][self.pose_estimation_name] == self.expected_metadata[self.pose_estimation_name] + + def check_read_nwb(self, nwbfile_path: str): + from ndx_pose import PoseEstimation, PoseEstimationSeries + + with NWBHDF5IO(path=nwbfile_path, mode="r", load_namespaces=True) as io: + nwbfile = io.read() + + # Replacing assertIn with pytest-style assert + assert "behavior" in nwbfile.processing + assert self.pose_estimation_name in nwbfile.processing["behavior"].data_interfaces + assert "Skeletons" in nwbfile.processing["behavior"].data_interfaces + + pose_estimation_container = nwbfile.processing["behavior"].data_interfaces[self.pose_estimation_name] + + # Replacing assertIsInstance with pytest-style assert + assert isinstance(pose_estimation_container, PoseEstimation) + + pose_estimation_metadata = self.expected_metadata[self.pose_estimation_name] + + # Replacing assertEqual with pytest-style assert + assert pose_estimation_container.description == pose_estimation_metadata["description"] + assert pose_estimation_container.scorer == pose_estimation_metadata["scorer"] + assert pose_estimation_container.source_software == pose_estimation_metadata["source_software"] + + # Using numpy's assert_array_equal + assert_array_equal( + pose_estimation_container.dimensions[:], [[self.original_video_height, self.original_video_width]] + ) + + # Replacing assertEqual with pytest-style assert + assert len(pose_estimation_container.pose_estimation_series) == len(self.expected_keypoint_names) + + assert pose_estimation_container.skeleton.nodes[:].tolist() == self.expected_keypoint_names + + for keypoint_name in self.expected_keypoint_names: + series_metadata = pose_estimation_metadata[keypoint_name] + + # Replacing assertIn with pytest-style assert + assert series_metadata["name"] in pose_estimation_container.pose_estimation_series + + pose_estimation_series = pose_estimation_container.pose_estimation_series[series_metadata["name"]] + + # Replacing assertIsInstance with pytest-style assert + assert isinstance(pose_estimation_series, PoseEstimationSeries) + + # Replacing assertEqual with pytest-style assert + assert pose_estimation_series.unit == "px" + assert pose_estimation_series.description == series_metadata["description"] + assert pose_estimation_series.reference_frame == self.conversion_options["reference_frame"] + + test_data = self.test_data[keypoint_name] + + # Using numpy's assert_array_equal + assert_array_equal(pose_estimation_series.data[:], test_data[["x", "y"]].values) + + +@pytest.mark.skipif(ndx_pose_version < version.parse("0.2.0"), reason="Interface requires ndx-pose version >= 0.2.0") +class TestLightningPoseDataInterfaceWithStubTest(DataInterfaceTestMixin, TemporalAlignmentMixin): + data_interface_cls = LightningPoseDataInterface + interface_kwargs = dict( + file_path=str(BEHAVIOR_DATA_PATH / "lightningpose" / "outputs/2023-11-09/10-14-37/video_preds/test_vid.csv"), + original_video_file_path=str( + BEHAVIOR_DATA_PATH / "lightningpose" / "outputs/2023-11-09/10-14-37/video_preds/test_vid.mp4" + ), + ) + + conversion_options = dict(stub_test=True) + save_directory = OUTPUT_PATH + + def check_read_nwb(self, nwbfile_path: str): + with NWBHDF5IO(path=nwbfile_path, mode="r", load_namespaces=True) as io: + nwbfile = io.read() + pose_estimation_container = nwbfile.processing["behavior"].data_interfaces["PoseEstimation"] + for pose_estimation_series in pose_estimation_container.pose_estimation_series.values(): + assert pose_estimation_series.data.shape[0] == 10 + assert pose_estimation_series.confidence.shape[0] == 10 + + +@pytest.mark.skipif( + ndx_pose_version >= version.parse("0.2.0"), reason="SLEAPInterface requires ndx-pose version < 0.2.0" +) +class TestSLEAPInterface(DataInterfaceTestMixin, TemporalAlignmentMixin): + + data_interface_cls = SLEAPInterface + interface_kwargs = dict( + file_path=str(BEHAVIOR_DATA_PATH / "sleap" / "predictions_1.2.7_provenance_and_tracking.slp"), + video_file_path=str(BEHAVIOR_DATA_PATH / "sleap" / "melanogaster_courtship.mp4"), + ) + save_directory = OUTPUT_PATH + + def check_read_nwb(self, nwbfile_path: str): # This is currently structured to be file-specific + with NWBHDF5IO(path=nwbfile_path, mode="r", load_namespaces=True) as io: + nwbfile = io.read() + assert "SLEAP_VIDEO_000_20190128_113421" in nwbfile.processing + processing_module_interfaces = nwbfile.processing["SLEAP_VIDEO_000_20190128_113421"].data_interfaces + assert "track=track_0" in processing_module_interfaces + + pose_estimation_series_in_nwb = processing_module_interfaces["track=track_0"].pose_estimation_series + expected_pose_estimation_series = [ + "abdomen", + "eyeL", + "eyeR", + "forelegL4", + "forelegR4", + "head", + "hindlegL4", + "hindlegR4", + "midlegL4", + "midlegR4", + "thorax", + "wingL", + "wingR", + ] + + assert set(pose_estimation_series_in_nwb) == set(expected_pose_estimation_series) + + +@pytest.mark.skipif( + ndx_pose_version >= version.parse("0.2.0"), reason="SLEAPInterface requires ndx-pose version < 0.2.0" +) +class CustomTestSLEAPInterface(TestCase): + savedir = OUTPUT_PATH + + @parameterized.expand( + [ + param( + data_interface=SLEAPInterface, + interface_kwargs=dict( + file_path=str(BEHAVIOR_DATA_PATH / "sleap" / "predictions_1.2.7_provenance_and_tracking.slp"), + ), + ) + ] + ) + def test_sleap_to_nwb_interface(self, data_interface, interface_kwargs): + nwbfile_path = str(self.savedir / f"{data_interface.__name__}.nwb") + + interface = SLEAPInterface(**interface_kwargs) + metadata = interface.get_metadata() + metadata["NWBFile"].update(session_start_time=datetime.now().astimezone()) + interface.run_conversion(nwbfile_path=nwbfile_path, overwrite=True, metadata=metadata) + + slp_predictions_path = interface_kwargs["file_path"] + labels = sleap_io.load_slp(slp_predictions_path) + + with NWBHDF5IO(path=nwbfile_path, mode="r", load_namespaces=True) as io: + nwbfile = io.read() + # Test matching number of processing modules + number_of_videos = len(labels.videos) + assert len(nwbfile.processing) == number_of_videos + + # Test processing module naming as video + processing_module_name = "SLEAP_VIDEO_000_20190128_113421" + assert processing_module_name in nwbfile.processing + + # For this case we have as many containers as tracks + # Each track usually represents a subject + processing_module = nwbfile.processing[processing_module_name] + processing_module_interfaces = processing_module.data_interfaces + assert len(processing_module_interfaces) == len(labels.tracks) + + # Test name of PoseEstimation containers + extracted_container_names = processing_module_interfaces.keys() + for track in labels.tracks: + expected_track_name = f"track={track.name}" + assert expected_track_name in extracted_container_names + + # Test one PoseEstimation container + container_name = f"track={track.name}" + pose_estimation_container = processing_module_interfaces[container_name] + # Test that the skeleton nodes are store as nodes in containers + expected_node_names = [node.name for node in labels.skeletons[0]] + assert expected_node_names == list(pose_estimation_container.nodes[:]) + + # Test that each PoseEstimationSeries is named as a node + for node_name in pose_estimation_container.nodes[:]: + assert node_name in pose_estimation_container.pose_estimation_series + + @parameterized.expand( + [ + param( + data_interface=SLEAPInterface, + interface_kwargs=dict( + file_path=str(BEHAVIOR_DATA_PATH / "sleap" / "melanogaster_courtship.slp"), + video_file_path=str(BEHAVIOR_DATA_PATH / "sleap" / "melanogaster_courtship.mp4"), + ), + ) + ] + ) + def test_sleap_interface_timestamps_propagation(self, data_interface, interface_kwargs): + nwbfile_path = str(self.savedir / f"{data_interface.__name__}.nwb") + + interface = SLEAPInterface(**interface_kwargs) + metadata = interface.get_metadata() + metadata["NWBFile"].update(session_start_time=datetime.now().astimezone()) + interface.run_conversion(nwbfile_path=nwbfile_path, overwrite=True, metadata=metadata) + + slp_predictions_path = interface_kwargs["file_path"] + labels = sleap_io.load_slp(slp_predictions_path) + + from neuroconv.datainterfaces.behavior.sleap.sleap_utils import ( + extract_timestamps, + ) + + expected_timestamps = set(extract_timestamps(interface_kwargs["video_file_path"])) + + with NWBHDF5IO(path=nwbfile_path, mode="r", load_namespaces=True) as io: + nwbfile = io.read() + # Test matching number of processing modules + number_of_videos = len(labels.videos) + assert len(nwbfile.processing) == number_of_videos + + # Test processing module naming as video + video_name = Path(labels.videos[0].filename).stem + processing_module_name = f"SLEAP_VIDEO_000_{video_name}" + + # For this case we have as many containers as tracks + processing_module_interfaces = nwbfile.processing[processing_module_name].data_interfaces + + extracted_container_names = processing_module_interfaces.keys() + for track in labels.tracks: + expected_track_name = f"track={track.name}" + assert expected_track_name in extracted_container_names + + container_name = f"track={track.name}" + pose_estimation_container = processing_module_interfaces[container_name] + + # Test that each PoseEstimationSeries is named as a node + for node_name in pose_estimation_container.nodes[:]: + pose_estimation_series = pose_estimation_container.pose_estimation_series[node_name] + extracted_timestamps = pose_estimation_series.timestamps[:] + + # Some frames do not have predictions associated with them, so we test for sub-set + assert set(extracted_timestamps).issubset(expected_timestamps) + + +@pytest.mark.skipif( + platform == "darwin" and python_version < version.parse("3.10") or ndx_pose_version < version.parse("0.2.0"), + reason="Interface requires ndx-pose version >= 0.2.0 and not supported on macOS with Python < 3.10", +) +class TestDeepLabCutInterface(DataInterfaceTestMixin): + data_interface_cls = DeepLabCutInterface + interface_kwargs = dict( + file_path=str( + BEHAVIOR_DATA_PATH + / "DLC" + / "open_field_without_video" + / "m3v1mp4DLC_resnet50_openfieldAug20shuffle1_30000.h5" + ), + config_file_path=str(BEHAVIOR_DATA_PATH / "DLC" / "open_field_without_video" / "config.yaml"), + subject_name="ind1", + ) + save_directory = OUTPUT_PATH + + def run_custom_checks(self): + self.check_renaming_instance(nwbfile_path=self.nwbfile_path) + + def check_renaming_instance(self, nwbfile_path: str): + custom_container_name = "TestPoseEstimation" + + metadata = self.interface.get_metadata() + metadata["NWBFile"].update(session_start_time=datetime.now().astimezone()) + + self.interface.run_conversion( + nwbfile_path=nwbfile_path, overwrite=True, metadata=metadata, container_name=custom_container_name + ) + + with NWBHDF5IO(path=nwbfile_path, mode="r", load_namespaces=True) as io: + nwbfile = io.read() + assert "behavior" in nwbfile.processing + assert custom_container_name in nwbfile.processing["behavior"].data_interfaces + + def check_read_nwb(self, nwbfile_path: str): + with NWBHDF5IO(path=nwbfile_path, mode="r", load_namespaces=True) as io: + nwbfile = io.read() + assert "behavior" in nwbfile.processing + processing_module_interfaces = nwbfile.processing["behavior"].data_interfaces + assert "PoseEstimationDeepLabCut" in processing_module_interfaces + assert "Skeletons" in processing_module_interfaces + + pose_estimation_container = processing_module_interfaces["PoseEstimationDeepLabCut"] + pose_estimation_series_in_nwb = pose_estimation_container.pose_estimation_series + expected_pose_estimation_series = ["ind1_leftear", "ind1_rightear", "ind1_snout", "ind1_tailbase"] + + expected_pose_estimation_series_are_in_nwb_file = [ + pose_estimation in pose_estimation_series_in_nwb for pose_estimation in expected_pose_estimation_series + ] + + assert all(expected_pose_estimation_series_are_in_nwb_file) + + skeleton = pose_estimation_container.skeleton + assert skeleton.nodes[:].tolist() == ["snout", "leftear", "rightear", "tailbase"] + + +@pytest.mark.skipif( + platform == "darwin" and python_version < version.parse("3.10") or ndx_pose_version < version.parse("0.2.0"), + reason="Interface requires ndx-pose version >= 0.2.0 and not supported on macOS with Python < 3.10", +) +class TestDeepLabCutInterfaceNoConfigFile(DataInterfaceTestMixin): + data_interface_cls = DeepLabCutInterface + interface_kwargs = dict( + file_path=str( + BEHAVIOR_DATA_PATH + / "DLC" + / "open_field_without_video" + / "m3v1mp4DLC_resnet50_openfieldAug20shuffle1_30000.h5" + ), + config_file_path=None, + subject_name="ind1", + ) + save_directory = OUTPUT_PATH + + def check_read_nwb(self, nwbfile_path: str): + with NWBHDF5IO(path=nwbfile_path, mode="r", load_namespaces=True) as io: + nwbfile = io.read() + assert "behavior" in nwbfile.processing + processing_module_interfaces = nwbfile.processing["behavior"].data_interfaces + assert "PoseEstimationDeepLabCut" in processing_module_interfaces + + pose_estimation_series_in_nwb = processing_module_interfaces[ + "PoseEstimationDeepLabCut" + ].pose_estimation_series + expected_pose_estimation_series = ["ind1_leftear", "ind1_rightear", "ind1_snout", "ind1_tailbase"] + + expected_pose_estimation_series_are_in_nwb_file = [ + pose_estimation in pose_estimation_series_in_nwb for pose_estimation in expected_pose_estimation_series + ] + + assert all(expected_pose_estimation_series_are_in_nwb_file) + + +@pytest.mark.skipif( + platform == "darwin" and python_version < version.parse("3.10") or ndx_pose_version < version.parse("0.2.0"), + reason="Interface requires ndx-pose version >= 0.2.0 and not supported on macOS with Python < 3.10", +) +class TestDeepLabCutInterfaceSetTimestamps(DataInterfaceTestMixin): + data_interface_cls = DeepLabCutInterface + interface_kwargs = dict( + file_path=str( + BEHAVIOR_DATA_PATH + / "DLC" + / "open_field_without_video" + / "m3v1mp4DLC_resnet50_openfieldAug20shuffle1_30000.h5" + ), + config_file_path=str(BEHAVIOR_DATA_PATH / "DLC" / "open_field_without_video" / "config.yaml"), + subject_name="ind1", + ) + + save_directory = OUTPUT_PATH + + def run_custom_checks(self): + self.check_custom_timestamps(nwbfile_path=self.nwbfile_path) + + def check_custom_timestamps(self, nwbfile_path: str): + custom_timestamps = np.concatenate( + (np.linspace(10, 110, 1000), np.linspace(150, 250, 1000), np.linspace(300, 400, 330)) + ) + + metadata = self.interface.get_metadata() + metadata["NWBFile"].update(session_start_time=datetime.now().astimezone()) + + self.interface.set_aligned_timestamps(custom_timestamps) + assert len(self.interface._timestamps) == 2330 + + self.interface.run_conversion(nwbfile_path=nwbfile_path, metadata=metadata, overwrite=True) + + with NWBHDF5IO(path=nwbfile_path, mode="r", load_namespaces=True) as io: + nwbfile = io.read() + assert "behavior" in nwbfile.processing + processing_module_interfaces = nwbfile.processing["behavior"].data_interfaces + assert "PoseEstimationDeepLabCut" in processing_module_interfaces + + pose_estimation_series_in_nwb = processing_module_interfaces[ + "PoseEstimationDeepLabCut" + ].pose_estimation_series + + for pose_estimation in pose_estimation_series_in_nwb.values(): + pose_timestamps = pose_estimation.timestamps + np.testing.assert_array_equal(pose_timestamps, custom_timestamps) + + # This was tested in the other test + def check_read_nwb(self, nwbfile_path: str): + pass + + +@pytest.mark.skipif( + platform == "darwin" and python_version < version.parse("3.10") or ndx_pose_version < version.parse("0.2.0"), + reason="Interface requires ndx-pose version >= 0.2.0 and not supported on macOS with Python < 3.10", +) +class TestDeepLabCutInterfaceFromCSV(DataInterfaceTestMixin): + data_interface_cls = DeepLabCutInterface + interface_kwargs = dict( + file_path=str( + BEHAVIOR_DATA_PATH + / "DLC" + / "SL18_csv" + / "SL18_D19_S01_F01_BOX_SLP_20230503_112642.1DLC_resnet50_SubLearnSleepBoxRedLightJun26shuffle1_100000_stubbed.csv" + ), + config_file_path=None, + subject_name="SL18", + ) + save_directory = OUTPUT_PATH + + def check_read_nwb(self, nwbfile_path: str): + with NWBHDF5IO(path=nwbfile_path, mode="r", load_namespaces=True) as io: + nwbfile = io.read() + assert "behavior" in nwbfile.processing + processing_module_interfaces = nwbfile.processing["behavior"].data_interfaces + assert "PoseEstimationDeepLabCut" in processing_module_interfaces + + pose_estimation_series_in_nwb = processing_module_interfaces[ + "PoseEstimationDeepLabCut" + ].pose_estimation_series + expected_pose_estimation_series = ["SL18_redled", "SL18_shoulder", "SL18_haunch", "SL18_baseoftail"] + + expected_pose_estimation_series_are_in_nwb_file = [ + pose_estimation in pose_estimation_series_in_nwb for pose_estimation in expected_pose_estimation_series + ] + + assert all(expected_pose_estimation_series_are_in_nwb_file) + + +@pytest.fixture +def clean_pose_extension_import(): + modules_to_remove = [m for m in sys.modules if m.startswith("ndx_pose")] + for module in modules_to_remove: + del sys.modules[module] + + +@pytest.mark.skipif( + platform == "darwin" and python_version < version.parse("3.10") or ndx_pose_version < version.parse("0.2.0"), + reason="Interface requires ndx-pose version >= 0.2.0 and not supported on macOS with Python < 3.10", +) +def test_deep_lab_cut_import_pose_extension_bug(clean_pose_extension_import, tmp_path): + """ + Test that the DeepLabCutInterface writes correctly without importing the ndx-pose extension. + See issues: + https://github.com/catalystneuro/neuroconv/issues/1114 + https://github.com/rly/ndx-pose/issues/36 + + """ + + interface_kwargs = dict( + file_path=str( + BEHAVIOR_DATA_PATH + / "DLC" + / "open_field_without_video" + / "m3v1mp4DLC_resnet50_openfieldAug20shuffle1_30000.h5" + ), + config_file_path=str(BEHAVIOR_DATA_PATH / "DLC" / "open_field_without_video" / "config.yaml"), + ) + + interface = DeepLabCutInterface(**interface_kwargs) + metadata = interface.get_metadata() + metadata["NWBFile"]["session_start_time"] = datetime(2023, 7, 24, 9, 30, 55, 440600, tzinfo=timezone.utc) + + nwbfile_path = tmp_path / "test.nwb" + interface.run_conversion(nwbfile_path=nwbfile_path, metadata=metadata, overwrite=True) + with NWBHDF5IO(path=nwbfile_path, mode="r") as io: + read_nwbfile = io.read() + pose_estimation_container = read_nwbfile.processing["behavior"]["PoseEstimationDeepLabCut"] + + assert len(pose_estimation_container.fields) > 0