import inspect
import json
import tempfile
from abc import abstractmethod
from copy import deepcopy
from datetime import datetime
from pathlib import Path
from typing import Literal
import numpy as np
import pytest
from jsonschema.validators import Draft7Validator, validate
from numpy.testing import assert_allclose, assert_array_equal
from pynwb import read_nwb
from pynwb.testing.mock.file import mock_NWBFile
from neuroconv import BaseDataInterface, NWBConverter
from neuroconv.datainterfaces.ecephys.baserecordingextractorinterface import (
BaseRecordingExtractorInterface,
)
from neuroconv.datainterfaces.ecephys.basesortingextractorinterface import (
BaseSortingExtractorInterface,
)
from neuroconv.datainterfaces.events.baseeventsinterface import _to_table_object_name
from neuroconv.datainterfaces.ophys.baseimagingextractorinterface import (
BaseImagingExtractorInterface,
)
from neuroconv.datainterfaces.ophys.basesegmentationextractorinterface import (
BaseSegmentationExtractorInterface,
)
from neuroconv.tools.fiber_photometry import get_fiber_photometry_table
from neuroconv.utils.json_schema import _NWBMetaDataEncoder
def _get_metadata_for_writing(interface) -> dict:
"""Return the interface's metadata in the format NeuroConv itself writes.
The internal fills use the dict-based format, so the tests that write a file ask for the same thing
rather than for the old list-based shape ``get_metadata()`` still hands users. Interfaces that never
exposed the old format take no such argument and are asked plainly.
"""
import inspect
signature = inspect.signature(interface.get_metadata)
if "use_new_metadata_format" in signature.parameters:
return interface.get_metadata(use_new_metadata_format=True)
return interface.get_metadata()
[docs]
class DataInterfaceTestMixin:
"""
Generic class for testing DataInterfaces.
Several of these tests are required to be run in a specific order. In this case,
there is a `test_conversion_as_lone_interface` that calls the `check` functions in
the appropriate order, after the `interface` has been created. Normally, you might
expect the `interface` to be simply created in the `setUp` method, but this class
allows you to specify multiple interface_kwargs.
Class Attributes
----------------
data_interface_cls : DataInterface
class, not instance
interface_kwargs : dict or list
When it is a dictionary, take these as arguments to the constructor of the
interface. When it is a list, each element of the list is a dictionary of
arguments to the constructor. Each dictionary will be tested one at a time.
save_directory : Path, optional
Directory where test files should be saved.
"""
data_interface_cls: type[BaseDataInterface]
interface_kwargs: dict
save_directory: Path = Path(tempfile.mkdtemp())
conversion_options: dict | None = None
maxDiff = None
# Narrowed by mixins whose `check_read_nwb` cannot read every backend
check_read_nwb_backends: tuple[str, ...] = ("hdf5", "zarr")
[docs]
@pytest.fixture
def setup_interface(self, request):
"""Add this as a fixture when you want freshly created interface in the test."""
self.test_name: str = ""
self.interface = self.data_interface_cls(**self.interface_kwargs)
return self.interface, self.test_name
[docs]
@pytest.fixture(scope="class", autouse=True)
@classmethod
def setup_default_conversion_options(cls):
cls.conversion_options = cls.conversion_options or dict()
return cls.conversion_options
[docs]
def test_source_schema_valid(self):
schema = self.data_interface_cls.get_source_schema()
Draft7Validator.check_schema(schema=schema)
[docs]
def test_conversion_options_schema_valid(self, setup_interface):
schema = self.interface.get_conversion_options_schema()
Draft7Validator.check_schema(schema=schema)
[docs]
@pytest.mark.parametrize("backend", ["hdf5", "zarr"])
def test_all_conversion_checks(self, setup_interface, tmp_path, backend):
"""Write the interface out and validate what lands on disk, once per backend.
`run_conversion` resolves the default backend configuration internally, so passing one explicitly
writes the same file; that equivalence is covered once on a mock interface in
`tests/test_minimal/test_interfaces_run_conversion.py`.
"""
metadata = self.edit_metadata(_get_metadata_for_writing(self.interface))
if "session_start_time" not in metadata["NWBFile"]:
metadata["NWBFile"].update(session_start_time=datetime.now().astimezone())
nwbfile_path = str(tmp_path / f"{self.__class__.__name__}_{self.test_name}_{backend}.nwb")
self.nwbfile_path = nwbfile_path
self.interface.run_conversion(
nwbfile_path=nwbfile_path,
overwrite=True,
metadata=metadata,
backend=backend,
**self.conversion_options,
)
if backend in self.check_read_nwb_backends:
self.check_read_nwb(nwbfile_path=nwbfile_path)
# Custom checks tend to write more files of their own, so they run against one backend only
if backend == "hdf5":
self.run_custom_checks()
[docs]
@abstractmethod
def check_read_nwb(self, nwbfile_path: str):
"""Read the produced NWB file and compare it to the interface."""
pass
[docs]
def run_custom_checks(self):
"""Override this in child classes to inject additional custom checks."""
pass
[docs]
class TemporalAlignmentMixin:
"""
Generic class for testing temporal alignment methods.
"""
data_interface_cls: type[BaseDataInterface]
interface_kwargs: dict
save_directory: Path = Path(tempfile.mkdtemp())
conversion_options: dict | None = None
maxDiff = None
[docs]
@pytest.fixture
def setup_interface(self, request):
self.test_name: str = ""
self.interface = self.data_interface_cls(**self.interface_kwargs)
return self.interface, self.test_name
[docs]
@pytest.fixture(scope="class", autouse=True)
@classmethod
def setup_default_conversion_options(cls):
cls.conversion_options = cls.conversion_options or dict()
return cls.conversion_options
[docs]
def setUpFreshInterface(self):
"""Protocol for creating a fresh instance of the interface."""
self.interface = self.data_interface_cls(**self.interface_kwargs)
[docs]
def check_interface_get_original_timestamps(self):
"""
Just to ensure each interface can call .get_original_timestamps() without an error raising.
Also, that it always returns non-empty.
"""
self.setUpFreshInterface()
original_timestamps = self.interface.get_original_timestamps()
assert len(original_timestamps) != 0
[docs]
def check_interface_get_timestamps(self):
"""
Just to ensure each interface can call .get_timestamps() without an error raising.
Also, that it always returns non-empty.
"""
self.setUpFreshInterface()
timestamps = self.interface.get_timestamps()
assert len(timestamps) != 0
[docs]
def check_interface_set_aligned_timestamps(self):
"""Ensure that internal mechanisms for the timestamps getter/setter work as expected."""
self.setUpFreshInterface()
unaligned_timestamps = self.interface.get_timestamps()
random_number_generator = np.random.default_rng(seed=0)
aligned_timestamps = (
unaligned_timestamps + 1.23 + random_number_generator.random(size=unaligned_timestamps.shape)
)
self.interface.set_aligned_timestamps(aligned_timestamps=aligned_timestamps)
retrieved_aligned_timestamps = self.interface.get_timestamps()
assert_array_equal(retrieved_aligned_timestamps, aligned_timestamps)
[docs]
def check_shift_timestamps_by_start_time(self):
"""Ensure that internal mechanisms for shifting timestamps by a starting time work as expected."""
self.setUpFreshInterface()
unaligned_timestamps = self.interface.get_timestamps()
aligned_starting_time = 1.23
self.interface.set_aligned_starting_time(aligned_starting_time=aligned_starting_time)
aligned_timestamps = self.interface.get_timestamps()
expected_timestamps = unaligned_timestamps + aligned_starting_time
assert_array_equal(aligned_timestamps, expected_timestamps)
[docs]
def check_interface_original_timestamps_inmutability(self):
"""Check aligning the timestamps for the interface does not change the value of .get_original_timestamps()."""
self.setUpFreshInterface()
pre_alignment_original_timestamps = self.interface.get_original_timestamps()
aligned_timestamps = pre_alignment_original_timestamps + 1.23
self.interface.set_aligned_timestamps(aligned_timestamps=aligned_timestamps)
post_alignment_original_timestamps = self.interface.get_original_timestamps()
assert_array_equal(post_alignment_original_timestamps, pre_alignment_original_timestamps)
[docs]
def check_nwbfile_temporal_alignment(self):
"""Check the temporally aligned timing information makes it into the NWB file."""
pass # TODO: will be easier to add when interface have 'add' methods separate from .run_conversion()
[docs]
def test_interface_alignment(self, setup_interface):
interface, test_name = setup_interface
self.check_interface_get_original_timestamps()
self.check_interface_get_timestamps()
self.check_interface_set_aligned_timestamps()
self.check_shift_timestamps_by_start_time()
self.check_interface_original_timestamps_inmutability()
self.check_nwbfile_temporal_alignment()
[docs]
class AudioInterfaceTestMixin(DataInterfaceTestMixin, TemporalAlignmentMixin):
"""
A mixin for testing Audio interfaces.
"""
# Currently asserted in the downstream testing suite; could be refactored in future PR
[docs]
def check_read_nwb(self, nwbfile_path: str):
pass
# Currently asserted in the downstream testing suite
[docs]
def test_interface_alignment(self):
pass
[docs]
class VideoInterfaceMixin(DataInterfaceTestMixin, TemporalAlignmentMixin):
"""
A mixin for testing Video interfaces.
"""
[docs]
def check_read_nwb(self, nwbfile_path: str):
nwbfile = read_nwb(nwbfile_path)
video_type = Path(self.interface_kwargs["file_paths"][0]).suffix[1:]
assert f"Video video_{video_type}" in nwbfile.acquisition
nwbfile.read_io.close()
[docs]
def check_interface_set_aligned_timestamps(self):
all_unaligned_timestamps = self.interface.get_original_timestamps()
random_number_generator = np.random.default_rng(seed=0)
aligned_timestamps = [
unaligned_timestamps + 1.23 + random_number_generator.random(size=unaligned_timestamps.shape)
for unaligned_timestamps in all_unaligned_timestamps
]
self.interface.set_aligned_timestamps(aligned_timestamps=aligned_timestamps)
retrieved_aligned_timestamps = self.interface.get_timestamps()
assert_array_equal(retrieved_aligned_timestamps, aligned_timestamps)
[docs]
def check_shift_timestamps_by_start_time(self):
self.setUpFreshInterface()
aligned_starting_time = 1.23
self.interface.set_aligned_timestamps(aligned_timestamps=self.interface.get_original_timestamps())
self.interface.set_aligned_starting_time(aligned_starting_time=aligned_starting_time)
all_aligned_timestamps = self.interface.get_timestamps()
unaligned_timestamps = self.interface.get_original_timestamps()
all_expected_timestamps = [timestamps + aligned_starting_time for timestamps in unaligned_timestamps]
[
assert_array_equal(aligned_timestamps, expected_timestamps)
for aligned_timestamps, expected_timestamps in zip(all_aligned_timestamps, all_expected_timestamps)
]
[docs]
def check_set_aligned_segment_starting_times(self):
self.setUpFreshInterface()
aligned_segment_starting_times = [
1.23 * file_path_index for file_path_index in range(len(self.interface_kwargs))
]
self.interface.set_aligned_segment_starting_times(aligned_segment_starting_times=aligned_segment_starting_times)
all_aligned_timestamps = self.interface.get_timestamps()
unaligned_timestamps = self.interface.get_original_timestamps()
all_expected_timestamps = [
timestamps + segment_starting_time
for timestamps, segment_starting_time in zip(unaligned_timestamps, aligned_segment_starting_times)
]
for aligned_timestamps, expected_timestamps in zip(all_aligned_timestamps, all_expected_timestamps):
assert_array_equal(aligned_timestamps, expected_timestamps)
[docs]
def check_interface_original_timestamps_inmutability(self):
self.setUpFreshInterface()
all_pre_alignment_original_timestamps = self.interface.get_original_timestamps()
all_aligned_timestamps = [
pre_alignment_original_timestamps + 1.23
for pre_alignment_original_timestamps in all_pre_alignment_original_timestamps
]
self.interface.set_aligned_timestamps(aligned_timestamps=all_aligned_timestamps)
all_post_alignment_original_timestamps = self.interface.get_original_timestamps()
for post_alignment_original_timestamps, pre_alignment_original_timestamps in zip(
all_post_alignment_original_timestamps, all_pre_alignment_original_timestamps
):
assert_array_equal(post_alignment_original_timestamps, pre_alignment_original_timestamps)
[docs]
class MedPCInterfaceMixin(DataInterfaceTestMixin, TemporalAlignmentMixin):
"""
A mixin for testing MedPC interfaces.
"""
[docs]
def test_conversion_options_schema_valid(self):
pass
[docs]
def check_conversion_options_schema_valid(self):
schema = self.interface.get_conversion_options_schema()
Draft7Validator.check_schema(schema=schema)
[docs]
def check_run_conversion_with_backend_configuration(
self, nwbfile_path: str, metadata: dict, backend: Literal["hdf5", "zarr"] = "hdf5"
):
nwbfile = self.interface.create_nwbfile(metadata=metadata, **self.conversion_options)
backend_configuration = self.interface.get_default_backend_configuration(nwbfile=nwbfile, backend=backend)
self.interface.run_conversion(
nwbfile_path=nwbfile_path,
overwrite=True,
metadata=metadata,
backend_configuration=backend_configuration,
**self.conversion_options,
)
[docs]
def check_conversion_in_nwbconverter(self, metadata: dict):
"""Build through an `NWBConverter` without writing; the written file is checked from the interface path."""
class TestNWBConverter(NWBConverter):
data_interface_classes = dict(Test=type(self.interface))
test_kwargs = self.test_kwargs[0] if isinstance(self.test_kwargs, list) else self.test_kwargs
converter = TestNWBConverter(source_data=dict(Test=test_kwargs))
converter.create_nwbfile(metadata=metadata, conversion_options=dict(Test=self.conversion_options))
[docs]
def test_all_conversion_checks(self, metadata: dict):
interface_kwargs = self.interface_kwargs
if isinstance(interface_kwargs, dict):
interface_kwargs = [interface_kwargs]
for num, kwargs in enumerate(interface_kwargs):
with self.subTest(str(num)):
self.case = num
self.test_kwargs = kwargs
self.interface = self.data_interface_cls(**self.test_kwargs)
self.check_metadata_schema_valid()
self.check_conversion_options_schema_valid()
self.check_metadata()
self.nwbfile_path = str(self.save_directory / f"{self.__class__.__name__}_{num}.nwb")
self.check_no_metadata_mutation(metadata=metadata)
self.check_conversion_in_nwbconverter(metadata=metadata)
self.check_run_conversion_with_backend_configuration(
nwbfile_path=self.nwbfile_path, metadata=metadata, backend="hdf5"
)
self.check_read_nwb(nwbfile_path=self.nwbfile_path)
# TODO: write and check the zarr file here too, once all H5DataIO prewraps are gone
# Any extra custom checks to run
self.run_custom_checks()
[docs]
def check_interface_get_original_timestamps(self, medpc_name_to_info_dict: dict):
"""
Just to ensure each interface can call .get_original_timestamps() without an error raising.
Also, that it always returns non-empty.
"""
self.setUpFreshInterface()
original_timestamps_dict = self.interface.get_original_timestamps(
medpc_name_to_info_dict=medpc_name_to_info_dict
)
for name in self.interface.source_data["aligned_timestamp_names"]:
original_timestamps = original_timestamps_dict[name]
assert len(original_timestamps) != 0, f"Timestamps for {name} are empty."
[docs]
def check_interface_get_timestamps(self):
"""
Just to ensure each interface can call .get_timestamps() without an error raising.
Also, that it always returns non-empty.
"""
self.setUpFreshInterface()
timestamps_dict = self.interface.get_timestamps()
for timestamps in timestamps_dict.values():
assert len(timestamps) != 0
[docs]
def check_interface_set_aligned_timestamps(self, medpc_name_to_info_dict: dict):
"""Ensure that internal mechanisms for the timestamps getter/setter work as expected."""
self.setUpFreshInterface()
unaligned_timestamps_dict = self.interface.get_original_timestamps(
medpc_name_to_info_dict=medpc_name_to_info_dict
)
random_number_generator = np.random.default_rng(seed=0)
aligned_timestamps_dict = {}
for name, unaligned_timestamps in unaligned_timestamps_dict.items():
aligned_timestamps = (
unaligned_timestamps + 1.23 + random_number_generator.random(size=unaligned_timestamps.shape)
)
aligned_timestamps_dict[name] = aligned_timestamps
self.interface.set_aligned_timestamps(aligned_timestamps_dict=aligned_timestamps_dict)
retrieved_aligned_timestamps = self.interface.get_timestamps()
for name, aligned_timestamps in aligned_timestamps_dict.items():
assert_array_equal(retrieved_aligned_timestamps[name], aligned_timestamps)
[docs]
def check_shift_timestamps_by_start_time(self, medpc_name_to_info_dict: dict):
"""Ensure that internal mechanisms for shifting timestamps by a starting time work as expected."""
self.setUpFreshInterface()
unaligned_timestamps_dict = self.interface.get_original_timestamps(
medpc_name_to_info_dict=medpc_name_to_info_dict
)
aligned_starting_time = 1.23
self.interface.set_aligned_starting_time(
aligned_starting_time=aligned_starting_time,
medpc_name_to_info_dict=medpc_name_to_info_dict,
)
aligned_timestamps = self.interface.get_timestamps()
expected_timestamps_dict = {
name: unaligned_timestamps + aligned_starting_time
for name, unaligned_timestamps in unaligned_timestamps_dict.items()
}
for name, expected_timestamps in expected_timestamps_dict.items():
assert_array_equal(aligned_timestamps[name], expected_timestamps)
[docs]
def check_interface_original_timestamps_inmutability(self, medpc_name_to_info_dict: dict):
"""Check aligning the timestamps for the interface does not change the value of .get_original_timestamps()."""
self.setUpFreshInterface()
pre_alignment_original_timestamps_dict = self.interface.get_original_timestamps(
medpc_name_to_info_dict=medpc_name_to_info_dict
)
aligned_timestamps_dict = {
name: pre_alignment_og_timestamps + 1.23
for name, pre_alignment_og_timestamps in pre_alignment_original_timestamps_dict.items()
}
self.interface.set_aligned_timestamps(aligned_timestamps_dict=aligned_timestamps_dict)
post_alignment_original_timestamps_dict = self.interface.get_original_timestamps(
medpc_name_to_info_dict=medpc_name_to_info_dict
)
for name, post_alignment_original_timestamps_dict in post_alignment_original_timestamps_dict.items():
assert_array_equal(post_alignment_original_timestamps_dict, pre_alignment_original_timestamps_dict[name])
[docs]
def test_interface_alignment(self, medpc_name_to_info_dict: dict):
interface_kwargs = self.interface_kwargs
if isinstance(interface_kwargs, dict):
interface_kwargs = [interface_kwargs]
for num, kwargs in enumerate(interface_kwargs):
with self.subTest(str(num)):
self.case = num
self.test_kwargs = kwargs
self.check_interface_get_original_timestamps(medpc_name_to_info_dict=medpc_name_to_info_dict)
self.check_interface_get_timestamps()
self.check_interface_set_aligned_timestamps(medpc_name_to_info_dict=medpc_name_to_info_dict)
self.check_shift_timestamps_by_start_time(medpc_name_to_info_dict=medpc_name_to_info_dict)
self.check_interface_original_timestamps_inmutability(medpc_name_to_info_dict=medpc_name_to_info_dict)
self.check_nwbfile_temporal_alignment()
[docs]
class MiniscopeImagingInterfaceMixin(ImagingExtractorInterfaceTestMixin):
"""
A mixin for testing Miniscope Imaging interfaces.
"""
optical_series_name = "OnePhotonSeries"
# This mixin reads the file itself instead of going through NwbImagingExtractor, so zarr is checkable
check_read_nwb_backends = ("hdf5", "zarr")
[docs]
def check_read_nwb(self, nwbfile_path: str):
from ndx_miniscope import Miniscope
nwbfile = read_nwb(nwbfile_path)
assert self.device_name in nwbfile.devices
device = nwbfile.devices[self.device_name]
assert isinstance(device, Miniscope)
imaging_plane = nwbfile.imaging_planes[self.imaging_plane_name]
assert imaging_plane.device.name == self.device_name
# Check OnePhotonSeries
assert self.photon_series_name in nwbfile.acquisition
one_photon_series = nwbfile.acquisition[self.photon_series_name]
assert one_photon_series.unit == "px"
assert one_photon_series.data.shape == (15, 752, 480)
assert one_photon_series.data.dtype == np.uint8
assert one_photon_series.rate is None
assert one_photon_series.starting_frame is None
assert one_photon_series.timestamps.shape == (15,)
interface_times = self.interface.get_original_timestamps()
assert_array_equal(one_photon_series.timestamps, interface_times)
nwbfile.read_io.close()
[docs]
class TDTFiberPhotometryInterfaceMixin(DataInterfaceTestMixin, TemporalAlignmentMixin):
"""Mixin for testing TDT Fiber Photometry interfaces."""
def test_no_metadata_mutation(self):
pass
[docs]
def test_conversion_options_schema_valid(self):
pass
[docs]
def check_conversion_options_schema_valid(self):
schema = self.interface.get_conversion_options_schema()
Draft7Validator.check_schema(schema=schema)
[docs]
def check_run_conversion_with_backend_configuration(
self, nwbfile_path: str, metadata: dict, backend: Literal["hdf5", "zarr"] = "hdf5"
):
nwbfile = self.interface.create_nwbfile(metadata=metadata, **self.conversion_options)
backend_configuration = self.interface.get_default_backend_configuration(nwbfile=nwbfile, backend=backend)
self.interface.run_conversion(
nwbfile_path=nwbfile_path,
metadata=metadata,
overwrite=True,
backend_configuration=backend_configuration,
**self.conversion_options,
)
[docs]
def check_conversion_in_nwbconverter(self, metadata: dict):
"""Build through an `NWBConverter` without writing; the written file is checked from the interface path."""
class TestNWBConverter(NWBConverter):
data_interface_classes = dict(Test=type(self.interface))
test_kwargs = self.test_kwargs[0] if isinstance(self.test_kwargs, list) else self.test_kwargs
converter = TestNWBConverter(source_data=dict(Test=test_kwargs))
converter.create_nwbfile(metadata=metadata, conversion_options=dict(Test=self.conversion_options))
[docs]
def test_all_conversion_checks(self, metadata: dict):
interface_kwargs = self.interface_kwargs
if isinstance(interface_kwargs, dict):
interface_kwargs = [interface_kwargs]
for num, kwargs in enumerate(interface_kwargs):
with self.subTest(str(num)):
self.case = num
self.test_kwargs = kwargs
self.interface = self.data_interface_cls(**self.test_kwargs)
self.check_metadata_schema_valid()
self.check_conversion_options_schema_valid()
self.check_metadata()
self.nwbfile_path = str(self.save_directory / f"{self.__class__.__name__}_{num}.nwb")
self.check_no_metadata_mutation(metadata=metadata)
self.check_conversion_in_nwbconverter(metadata=metadata)
self.check_run_conversion_with_backend_configuration(
nwbfile_path=self.nwbfile_path, metadata=metadata, backend="hdf5"
)
self.check_read_nwb(nwbfile_path=self.nwbfile_path)
# TODO: write and check the zarr file here too, once all H5DataIO prewraps are gone
# Any extra custom checks to run
self.run_custom_checks()
[docs]
def check_interface_get_original_timestamps(self):
"""
Just to ensure each interface can call .get_original_timestamps() without an error raising.
Also, that it always returns non-empty.
"""
self.setUpFreshInterface()
t1 = self.conversion_options.get("t1", 0.0)
t2 = self.conversion_options.get("t2", 0.0)
stream_name_to_timestamps = self.interface.get_original_timestamps(t1=t1, t2=t2)
for stream_name, timestamps in stream_name_to_timestamps.items():
assert len(timestamps) != 0, f"Timestamps for {stream_name} are empty."
[docs]
def check_interface_get_timestamps(self):
"""
Just to ensure each interface can call .get_timestamps() without an error raising.
Also, that it always returns non-empty.
"""
self.setUpFreshInterface()
t1 = self.conversion_options.get("t1", 0.0)
t2 = self.conversion_options.get("t2", 0.0)
stream_name_to_timestamps = self.interface.get_timestamps(t1=t1, t2=t2)
for stream_name, timestamps in stream_name_to_timestamps.items():
assert len(timestamps) != 0, f"Timestamps for {stream_name} are empty."
[docs]
def check_interface_set_aligned_timestamps(self):
"""Ensure that internal mechanisms for the timestamps getter/setter work as expected."""
t1 = self.conversion_options.get("t1", 0.0)
t2 = self.conversion_options.get("t2", 0.0)
self.setUpFreshInterface()
unaligned_stream_name_to_timestamps = self.interface.get_original_timestamps(t1=t1, t2=t2)
random_number_generator = np.random.default_rng(seed=0)
aligned_stream_name_to_timestamps = {}
for stream_name, unaligned_timestamps in unaligned_stream_name_to_timestamps.items():
aligned_timestamps = (
unaligned_timestamps + 1.23 + random_number_generator.random(size=unaligned_timestamps.shape)
)
aligned_stream_name_to_timestamps[stream_name] = aligned_timestamps
self.interface.set_aligned_timestamps(stream_name_to_aligned_timestamps=aligned_stream_name_to_timestamps)
t1 += 1.23 if t1 != 0.0 else 0.0
t2 += 2.23 if t2 != 0.0 else 0.0
retrieved_aligned_stream_name_to_timestamps = self.interface.get_timestamps(t1=t1, t2=t2)
for stream_name, aligned_timestamps in aligned_stream_name_to_timestamps.items():
retrieved_aligned_timestamps = retrieved_aligned_stream_name_to_timestamps[stream_name]
assert_array_equal(retrieved_aligned_timestamps, aligned_timestamps)
[docs]
def check_shift_timestamps_by_start_time(self):
"""Ensure that internal mechanisms for shifting timestamps by a starting time work as expected."""
t1 = self.conversion_options.get("t1", 0.0)
t2 = self.conversion_options.get("t2", 0.0)
self.setUpFreshInterface()
unaligned_stream_name_to_timestamps = self.interface.get_original_timestamps(t1=t1, t2=t2)
aligned_starting_time = 1.23
self.interface.set_aligned_starting_time(aligned_starting_time=aligned_starting_time, t1=t1, t2=t2)
t1 += aligned_starting_time if t1 != 0.0 else 0.0
t2 += aligned_starting_time if t2 != 0.0 else 0.0
aligned_stream_name_to_timestamps = self.interface.get_timestamps(t1=t1, t2=t2)
expected_timestamps_dict = {
name: unaligned_timestamps + aligned_starting_time
for name, unaligned_timestamps in unaligned_stream_name_to_timestamps.items()
}
for name, expected_timestamps in expected_timestamps_dict.items():
timestamps = aligned_stream_name_to_timestamps[name]
assert_array_equal(timestamps, expected_timestamps)
[docs]
def check_interface_original_timestamps_inmutability(self):
"""Check aligning the timestamps for the interface does not change the value of .get_original_timestamps()."""
t1 = self.conversion_options.get("t1", 0.0)
t2 = self.conversion_options.get("t2", 0.0)
self.setUpFreshInterface()
pre_alignment_stream_name_to_timestamps = self.interface.get_original_timestamps(t1=t1, t2=t2)
aligned_stream_name_to_timestamps = {
name: pre_alignment_timestamps + 1.23
for name, pre_alignment_timestamps in pre_alignment_stream_name_to_timestamps.items()
}
self.interface.set_aligned_timestamps(stream_name_to_aligned_timestamps=aligned_stream_name_to_timestamps)
post_alignment_stream_name_to_timestamps = self.interface.get_original_timestamps(t1=t1, t2=t2)
for name, post_alignment_timestamps in post_alignment_stream_name_to_timestamps.items():
pre_alignment_timestamps = pre_alignment_stream_name_to_timestamps[name]
assert_array_equal(post_alignment_timestamps, pre_alignment_timestamps)
[docs]
def test_interface_alignment(self):
interface_kwargs = self.interface_kwargs
if isinstance(interface_kwargs, dict):
interface_kwargs = [interface_kwargs]
for num, kwargs in enumerate(interface_kwargs):
with self.subTest(str(num)):
self.case = num
self.test_kwargs = kwargs
self.check_interface_get_original_timestamps()
self.check_interface_get_timestamps()
self.check_interface_set_aligned_timestamps()
self.check_shift_timestamps_by_start_time()
self.check_interface_original_timestamps_inmutability()
self.check_nwbfile_temporal_alignment()
[docs]
class PoseEstimationInterfaceTestMixin(DataInterfaceTestMixin):
"""
Generic class for testing any pose estimation interface.
Format-specific assertions belong in ``run_custom_checks``. ``TemporalAlignmentMixin`` is not a base
because the pose interfaces' alignment methods are on the way out; a child that wants them adds it.
"""
[docs]
def check_read_nwb(self, nwbfile_path: str):
"""Every container the metadata declares is in the file, named and shaped as the metadata says."""
metadata = _get_metadata_for_writing(self.interface)
nwbfile = read_nwb(nwbfile_path)
assert "behavior" in nwbfile.processing
behavior_module = nwbfile.processing["behavior"]
containers_metadata = metadata["Pose"]["PoseEstimations"]
assert len(containers_metadata) > 0, "The interface declares no PoseEstimation container."
for container_entry in containers_metadata.values():
self._check_pose_estimation_container(
nwbfile=nwbfile,
behavior_module=behavior_module,
metadata=metadata,
container_entry=container_entry,
)
nwbfile.read_io.close()
def _check_pose_estimation_container(self, nwbfile, behavior_module, metadata: dict, container_entry: dict):
"""One container entry, its cross-referenced device and skeleton, and one series per keypoint."""
from ndx_pose import PoseEstimation, PoseEstimationSeries
container_name = container_entry["name"]
assert container_name in behavior_module.data_interfaces
container = behavior_module.data_interfaces[container_name]
assert isinstance(container, PoseEstimation)
# Only fields the metadata actually carries are checked: the rest are left to ndx-pose's defaults
# by the writer, so asserting on them here would be asserting on the extension.
for field in ("description", "scorer", "source_software"):
if container_entry.get(field) is not None:
assert getattr(container, field) == container_entry[field]
device_metadata_key = container_entry.get("device_metadata_key")
if device_metadata_key is not None:
assert metadata["Devices"][device_metadata_key]["name"] in nwbfile.devices
skeleton_metadata_key = container_entry.get("skeleton_metadata_key")
if skeleton_metadata_key is not None:
skeleton_entry = metadata["Pose"]["Skeletons"][skeleton_metadata_key]
assert "Skeletons" in behavior_module.data_interfaces
assert skeleton_entry["name"] in behavior_module["Skeletons"].skeletons
assert container.skeleton.name == skeleton_entry["name"]
assert container.skeleton.nodes[:].tolist() == list(skeleton_entry["nodes"])
series_entries = container_entry["PoseEstimationSeries"]
assert len(container.pose_estimation_series) == len(series_entries)
for series_entry in series_entries.values():
series_name = series_entry["name"]
assert series_name in container.pose_estimation_series
series = container.pose_estimation_series[series_name]
assert isinstance(series, PoseEstimationSeries)
# A regularly sampled series carries a rate and a starting time instead of an explicit
# timestamps vector, so ask for the times either way.
timestamps = series.get_timestamps()
assert len(timestamps) > 0
assert series.data.ndim == 2
assert series.data.shape[0] == len(timestamps)
[docs]
class FiberPhotometryInterfaceTestMixin(DataInterfaceTestMixin, TemporalAlignmentMixin):
"""Shared tests for single-series fiber photometry interfaces.
This mixin is the contract between the *expected* values a child supplies by hand — the
response-series data and its timing, which every child determines independently for its own
format/file — and *how those values surface in the NWB file*, which is uniform across all
interfaces built on ``BaseFiberPhotometryInterface`` and therefore lives here. A child only
declares ``expected_response_series_data`` and the expected timing (``expected_rate`` +
``expected_starting_time`` for a regular series, or ``expected_timestamps`` for an irregular one);
it does not reimplement ``check_read_nwb``. The response-series expectations are deliberately *not*
derived from the interface's own reading methods (that would be circular); they are hand-supplied
literals. The ``FiberPhotometryTable`` / device / indicator assertions instead validate the
metadata → NWB mapping, and only run when the metadata actually carries a ``FiberPhotometryTable`` —
with the bare default an interface writes a lone response series and nothing else. Format
idiosyncrasies (e.g. where a session start time comes from) belong in a small dedicated override or
unit test.
"""
#: Hand-supplied expected samples of the written ``FiberPhotometryResponseSeries``. With a small
#: ``stub_samples`` in ``conversion_options`` this is a short, readable literal.
expected_response_series_data: np.ndarray
#: Expected timing: set ``expected_rate`` + ``expected_starting_time`` for a regularly sampled
#: series, or ``expected_timestamps`` for an irregular one.
expected_starting_time: float | None = None
expected_rate: float | None = None
expected_timestamps: np.ndarray | None = None
#: Expected ``unit`` of the written series. Unit is a property of the data (not editable metadata), set
#: when the series is built; uncalibrated fiber photometry defaults to "a.u.".
expected_unit: str = "a.u."
[docs]
def check_read_nwb(self, nwbfile_path: str):
metadata = self.interface.get_metadata()
fiber_photometry_metadata = metadata["FiberPhotometry"]
nwbfile = read_nwb(nwbfile_path)
self._check_response_series(nwbfile, fiber_photometry_metadata)
# The provenance chain (table, devices, indicators) is only written when the metadata supplies
# it; with the bare default a lone response series is a legal file, so only check what was asked.
if "FiberPhotometryTable" in fiber_photometry_metadata:
self._check_fiber_photometry_table(nwbfile, fiber_photometry_metadata)
self._check_devices(nwbfile, metadata)
self._check_indicators(nwbfile, fiber_photometry_metadata)
nwbfile.read_io.close()
def _check_response_series(self, nwbfile, fiber_photometry_metadata: dict):
"""The written response series must match the child's hand-supplied expected data and timing."""
series_metadata = fiber_photometry_metadata[self.interface.metadata_key]
series_name = series_metadata["name"]
assert series_name in nwbfile.acquisition, f"'{series_name}' missing from acquisition."
response_series = nwbfile.acquisition[series_name]
assert response_series.unit == self.expected_unit
assert_array_equal(response_series.data[:], self.expected_response_series_data)
# A regular series is written as rate + starting_time; an irregular one as explicit timestamps.
if self.expected_rate is not None:
assert response_series.rate == pytest.approx(self.expected_rate)
assert response_series.starting_time == pytest.approx(self.expected_starting_time)
else:
assert_array_equal(response_series.timestamps[:], self.expected_timestamps)
def _check_fiber_photometry_table(self, nwbfile, fiber_photometry_metadata: dict):
"""One table row per metadata row, with the per-row scalar fields matching the metadata."""
table = get_fiber_photometry_table(nwbfile)
assert table is not None
rows_metadata = list(fiber_photometry_metadata["FiberPhotometryTable"]["rows"].values())
assert len(table) == len(rows_metadata)
for row_index, row_metadata in enumerate(rows_metadata):
assert table["location"][row_index] == row_metadata["location"]
for wavelength_field in ("excitation_wavelength_in_nm", "emission_wavelength_in_nm"):
assert_array_equal(table[wavelength_field][row_index], row_metadata[wavelength_field])
def _check_devices(self, nwbfile, metadata: dict):
"""Every device model and device instance named in the top-level registry must be in the NWBFile."""
for device_model_metadata in metadata.get("DeviceModels", {}).values():
assert device_model_metadata["name"] in nwbfile.device_models
for device_metadata in metadata.get("Devices", {}).values():
assert device_metadata["name"] in nwbfile.devices
def _check_indicators(self, nwbfile, fiber_photometry_metadata: dict):
"""Every indicator named in the metadata must be written to the FiberPhotometry lab metadata."""
indicators = nwbfile.lab_meta_data["fiber_photometry"].fiber_photometry_indicators
for indicator_metadata in fiber_photometry_metadata["FiberPhotometryIndicators"].values():
assert indicator_metadata["name"] in indicators.indicators
[docs]
class EventsInterfaceTestMixin(DataInterfaceTestMixin):
"""Shared tests for the interfaces built on ``BaseEventsInterface``.
A subclass sets ``data_interface_cls`` and ``interface_kwargs`` and inherits the schema, metadata and
round-trip tests of ``DataInterfaceTestMixin``, with ``check_read_nwb`` asserting what is true of every
events interface whatever its source: the written tables carry the times ``get_event_times`` reports.
A subclass may also set ``event_names`` to write under the names a user would give the types, in which
case the round trip runs, and is checked, under those names. Nothing here touches ``alignment``. Nothing
here reads the source directly either, so a subclass that wants to pin the actual times of its fixture
states them in its own ``check_read_nwb``, calling ``super().check_read_nwb`` first.
"""
#: ``event_type_source_id`` to ``event_name``. Empty means the interface's own names are written.
event_names: dict[str, str] = {}
[docs]
def check_read_nwb(self, nwbfile_path: str):
"""Each type's rows in the written file carry the times ``get_event_times`` reports for it."""
events_metadata = self.edit_metadata(_get_metadata_for_writing(self.interface))["Events"]
event_types = events_metadata[self.interface.metadata_key]["event_types"]
nwbfile = read_nwb(nwbfile_path)
for event_type_source_id in self.interface.get_event_type_source_ids():
entry = event_types[event_type_source_id]
# The table this type routes into, named the way the writer names it: a declared EventTables
# entry, else the event_name of a type alone on its table, else the shared table_metadata_key.
table_metadata_key = entry.get("table_metadata_key", event_type_source_id)
declared_entry = events_metadata.get("EventTables", {}).get(table_metadata_key)
sharing_the_table = [
source_id
for source_id, other_entry in event_types.items()
if other_entry.get("table_metadata_key", source_id) == table_metadata_key
]
if declared_entry is not None:
table_name = declared_entry["table_name"]
elif len(sharing_the_table) == 1:
table_name = _to_table_object_name(entry["event_name"])
else:
table_name = _to_table_object_name(table_metadata_key)
table = nwbfile.get_events_table(table_name)
# In a shared table this type's rows are the ones labelled with its event_name.
rows = np.arange(len(table))
if "event_type" in table.colnames:
rows = np.flatnonzero(np.asarray(table["event_type"][:]) == entry["event_name"])
written_timestamps = np.asarray(table["timestamp"][:])[rows]
assert_allclose(written_timestamps, self.interface.get_event_times(event_type_source_id))