diff --git a/CHANGELOG.md b/CHANGELOG.md index efa5ef267..02df410dc 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -11,6 +11,7 @@ * Fixed `get_json_schema_from_method_signature` to resolve PEP 563 string annotations (from `from __future__ import annotations`) before passing them to pydantic. This affected any interface defined in a module with deferred annotations (e.g. `MiniscopeConverter`, or external subclasses from SpikeInterface). [PR #1670](https://github.com/catalystneuro/neuroconv/pull/1670) ## Features +* Added dict-based metadata pipeline for imaging in `roiextractors.py`, supporting the new `MicroscopySeries`, `ImagingPlanes`, and `Devices` metadata format keyed by `metadata_key`. Old list-based functions are preserved (renamed with `_old_list_format` suffix) and dispatched automatically when `metadata_key` is not provided. [PR #1677](https://github.com/catalystneuro/neuroconv/pull/1677) ## Improvements * Added column-first fast path for writing Units tables when the table is new (no append/merge). Uses `id.extend()` + `add_column()` instead of per-row `add_unit()` calls, reducing Python overhead for large sortings. [PR #1669](https://github.com/catalystneuro/neuroconv/pull/1669) diff --git a/src/neuroconv/tools/nwb_helpers/__init__.py b/src/neuroconv/tools/nwb_helpers/__init__.py index 87a1e2081..79f682831 100644 --- a/src/neuroconv/tools/nwb_helpers/__init__.py +++ b/src/neuroconv/tools/nwb_helpers/__init__.py @@ -25,6 +25,7 @@ from ._configure_backend import configure_backend from ._dataset_configuration import get_default_dataset_io_configurations, get_existing_dataset_io_configurations from ._metadata_and_file_helpers import ( + _add_device_to_nwbfile, add_device_from_metadata, configure_and_write_nwbfile, get_default_nwbfile_metadata, diff --git a/src/neuroconv/tools/nwb_helpers/_metadata_and_file_helpers.py b/src/neuroconv/tools/nwb_helpers/_metadata_and_file_helpers.py index 6f9b30b22..5080409cd 100644 --- a/src/neuroconv/tools/nwb_helpers/_metadata_and_file_helpers.py +++ b/src/neuroconv/tools/nwb_helpers/_metadata_and_file_helpers.py @@ -189,6 +189,37 @@ def add_device_from_metadata(nwbfile: NWBFile, modality: str = "Ecephys", metada nwbfile.create_device(**dict(defaults, **device_metadata)) +def _add_device_to_nwbfile( + nwbfile: NWBFile, + device_metadata: dict, +): + """ + Add a device to an NWBFile. + + If a device with the same name already exists, the existing device is + returned without creating a duplicate. + + Parameters + ---------- + nwbfile : NWBFile + The NWB file to add the device to. + device_metadata : dict + Dictionary describing the device. Must contain at least a ``"name"`` key. + + Returns + ------- + Device + The Device object (either newly created or existing). + """ + device_name = device_metadata["name"] + + if device_name in nwbfile.devices: + return nwbfile.devices[device_name] + + device = nwbfile.create_device(**device_metadata) + return device + + def _attempt_cleanup_of_existing_nwbfile(nwbfile_path: Path) -> None: if not nwbfile_path.exists(): return diff --git a/src/neuroconv/tools/roiextractors/roiextractors.py b/src/neuroconv/tools/roiextractors/roiextractors.py index 2dc7d6537..b49415217 100644 --- a/src/neuroconv/tools/roiextractors/roiextractors.py +++ b/src/neuroconv/tools/roiextractors/roiextractors.py @@ -43,6 +43,7 @@ make_nwbfile_from_metadata, ) from ..nwb_helpers._metadata_and_file_helpers import ( + _add_device_to_nwbfile, _resolve_backend, configure_and_write_nwbfile, ) @@ -54,6 +55,130 @@ from ...utils.str_utils import human_readable_size +def _is_dict_based_metadata(metadata: dict) -> bool: + """Detect whether metadata uses the new dict-based format or old list-based format. + + Dict-based format has top-level 'Devices' key and/or 'ImagingPlanes'/'MicroscopySeries' + (plural, dict-valued) under 'Ophys'. List-based format has 'Device' (list) and + 'ImagingPlane' (list, singular) under 'Ophys'. + + Returns True for dict-based, False for list-based. + """ + if "Devices" in metadata: + return True + + ophys = metadata.get("Ophys", {}) + + if "ImagingPlanes" in ophys or "MicroscopySeries" in ophys: + return True + + if "ImagingPlane" in ophys or "Device" in ophys: + return False + + # Ambiguous or empty metadata defaults to dict-based (the new format) + return True + + +def _get_ophys_metadata_placeholders(): + """ + Returns fresh ophys metadata with centralized placeholder values. + + Placeholders are kept in one place so they are easy to identify downstream and + we make up as little metadata as possible. All fields included here are strictly + required by the NWB schema. Each call returns an independent copy. + + Until something like https://github.com/NeurodataWithoutBorders/nwb-schema/issues/672 + is accepted, we will keep this approach. + """ + metadata = get_default_nwbfile_metadata() + + default_metadata_key = "default_metadata_key" + + metadata["Devices"] = { + default_metadata_key: { + "name": "Microscope", + }, + } + + metadata["Ophys"] = { + "ImagingPlanes": { + default_metadata_key: { + "name": "ImagingPlane", + "excitation_lambda": np.nan, + "indicator": "unknown", + "location": "unknown", + "optical_channel": [ + { + "name": "OpticalChannel", + "emission_lambda": np.nan, + "description": "An optical channel of the microscope.", + } + ], + }, + }, + "MicroscopySeries": { + default_metadata_key: { + "name": "MicroscopySeries", + "unit": "n.a.", + "imaging_plane_metadata_key": default_metadata_key, + }, + }, + } + + return metadata + + +def get_full_ophys_metadata(): + """ + Returns a fully specified ophys metadata example with realistic values. + + Users can call this to get a complete example of what the metadata structure looks like, + edit only the fields they need, and discard the rest. Each call returns an independent + copy so callers can modify it freely without affecting other calls. + + # TODO: expand with segmentation metadata once we get to that PR + """ + metadata = get_default_nwbfile_metadata() + + metadata["Devices"] = { + "my_microscope": { + "name": "Microscope", + "description": "Two-photon microscope", + "manufacturer": "Thorlabs", + }, + } + + metadata["Ophys"] = { + "ImagingPlanes": { + "my_plane": { + "name": "ImagingPlane", + "description": "Imaging plane in V1", + "excitation_lambda": 920.0, + "indicator": "GCaMP6s", + "location": "V1", + "device_metadata_key": "my_microscope", + "optical_channel": [ + { + "name": "Green", + "description": "GCaMP emission", + "emission_lambda": 510.0, + } + ], + }, + }, + "MicroscopySeries": { + "my_series": { + "name": "TwoPhotonSeries", + "description": "Two-photon calcium imaging", + "unit": "n.a.", + "imaging_plane_metadata_key": "my_plane", + }, + }, + } + + return metadata + + def _get_default_ophys_metadata_old_metadata_list(): """ Returns fresh ophys default metadata dictionary. @@ -144,7 +269,7 @@ def _get_default_ophys_metadata_old_metadata_list(): def _get_default_segmentation_metadata() -> DeepDict: - """Fill default metadata for segmentation using _get_default_ophys_metadata().""" + """Fill default metadata for segmentation using _get_ophys_metadata_placeholders().""" from neuroconv.tools.nwb_helpers import get_default_nwbfile_metadata # Start with base NWB metadata @@ -215,19 +340,12 @@ def get_nwb_imaging_metadata( return metadata -def add_devices_to_nwbfile(nwbfile: NWBFile, metadata: dict | None = None) -> NWBFile: +def _add_devices_to_nwbfile_old_list_format(nwbfile: NWBFile, metadata: dict | None = None) -> NWBFile: """ - Add optical physiology devices from metadata. - - Notes - ----- - The metadata concerning the optical physiology should be stored in ``metadata['Ophys']['Device']``. + Add optical physiology devices from old list-based metadata. - Deprecation: Passing ``pynwb.device.Device`` objects directly inside - ``metadata['Ophys']['Device']`` is deprecated and will be removed on or after March 2026. - Please pass device definitions as dictionaries instead (e.g., ``{"name": "Microscope"}``). + Private implementation used internally by old list-based functions. """ - # Get device metadata from user or use defaults metadata = metadata or {} device_metadata = metadata.get("Ophys", {}).get("Device") @@ -251,6 +369,22 @@ def add_devices_to_nwbfile(nwbfile: NWBFile, metadata: dict | None = None) -> NW return nwbfile +def add_devices_to_nwbfile(nwbfile: NWBFile, metadata: dict | None = None) -> NWBFile: + """ + Add optical physiology devices from metadata. + + .. deprecated:: + ``add_devices_to_nwbfile`` is deprecated and will be removed on or after September 2026. + """ + warnings.warn( + "add_devices_to_nwbfile is deprecated and will be removed on or after September 2026. " + "Use _add_device_to_nwbfile with the new dict-based metadata format (metadata['Devices']) instead.", + FutureWarning, + stacklevel=2, + ) + return _add_devices_to_nwbfile_old_list_format(nwbfile=nwbfile, metadata=metadata) + + def _add_imaging_plane_to_nwbfile_old_list_format( nwbfile: NWBFile, metadata: dict, @@ -265,10 +399,10 @@ def _add_imaging_plane_to_nwbfile_old_list_format( nwbfile : NWBFile An previously defined -in memory- NWBFile. metadata : dict - The metadata in the neuroconv format. See `_get_default_ophys_metadata()` for an example. + The metadata in the neuroconv format. See `_get_ophys_metadata_placeholders()` for an example. imaging_plane_name: str, optional The name of the imaging plane to be added. If None, this function adds the default imaging plane - in _get_default_ophys_metadata(). + in _get_ophys_metadata_placeholders(). Returns ------- @@ -286,7 +420,7 @@ def _add_imaging_plane_to_nwbfile_old_list_format( if imaging_plane_name in nwbfile.imaging_planes: return nwbfile - add_devices_to_nwbfile(nwbfile=nwbfile, metadata=metadata) + _add_devices_to_nwbfile_old_list_format(nwbfile=nwbfile, metadata=metadata) if user_provided_a_name: # User explicitly requested a specific plane - search for it in metadata @@ -328,6 +462,76 @@ def _add_imaging_plane_to_nwbfile_old_list_format( return nwbfile +def _add_imaging_plane_to_nwbfile( + nwbfile: NWBFile, + imaging_plane_metadata: dict, + metadata: dict, +) -> ImagingPlane: + """ + Add an imaging plane to an NWBFile. + + If an imaging plane with the same name already exists, the existing one is returned. + + The device is resolved via ``device_metadata_key`` in the imaging plane metadata, + which requires the full metadata to look up the device in ``metadata["Devices"]``. + If no ``device_metadata_key`` is set, a default device is created. + + Parameters + ---------- + nwbfile : NWBFile + The NWB file to add the imaging plane to. + imaging_plane_metadata : dict + Dictionary describing the imaging plane (already extracted by the caller). + metadata : dict + The full metadata dictionary, needed to resolve ``device_metadata_key`` + references in ``metadata["Devices"]``. + + Returns + ------- + ImagingPlane + The ImagingPlane object (either newly created or existing). + """ + # Copy to avoid mutation + imaging_plane_kwargs = imaging_plane_metadata.copy() + + # Validate required fields + required_fields = ["name", "excitation_lambda", "indicator", "location", "optical_channel"] + missing_fields = [field for field in required_fields if field not in imaging_plane_kwargs] + if missing_fields: + default_imaging_plane = _get_ophys_metadata_placeholders()["Ophys"]["ImagingPlanes"]["default_metadata_key"] + placeholder_hint = "\n".join(f" {field}: {default_imaging_plane[field]!r}" for field in missing_fields) + raise ValueError( + f"Imaging plane metadata is missing required fields.\n" + f"For a complete NWB file, the following fields should be provided. " + f"If missing, a placeholder can be used instead:\n{placeholder_hint}" + ) + + # Check if already exists + imaging_plane_name = imaging_plane_kwargs["name"] + if imaging_plane_name in nwbfile.imaging_planes: + return nwbfile.imaging_planes[imaging_plane_name] + + # Resolve device + device_metadata_key = imaging_plane_kwargs.pop("device_metadata_key", None) + if device_metadata_key is not None: + device_metadata = metadata["Devices"][device_metadata_key] + else: + device_metadata = _get_ophys_metadata_placeholders()["Devices"]["default_metadata_key"] + device = _add_device_to_nwbfile(nwbfile=nwbfile, device_metadata=device_metadata) + + imaging_plane_kwargs["device"] = device + + # Convert optical channel metadata dicts to OpticalChannel objects + imaging_plane_kwargs["optical_channel"] = [ + OpticalChannel(**channel_metadata) for channel_metadata in imaging_plane_kwargs["optical_channel"] + ] + + imaging_plane = ImagingPlane(**imaging_plane_kwargs) + nwbfile.add_imaging_plane(imaging_plane) + + return imaging_plane + + def _add_image_segmentation_to_nwbfile(nwbfile: NWBFile, metadata: dict) -> NWBFile: """ Private implementation. Adds the image segmentation container to the nwb file. @@ -515,6 +719,135 @@ def _add_photon_series_to_nwbfile_old_list_format( return nwbfile +def _add_photon_series_to_nwbfile( + imaging: ImagingExtractor, + nwbfile: NWBFile, + metadata: dict, + photon_series_type: Literal["TwoPhotonSeries", "OnePhotonSeries"], + metadata_key: str, + parent_container: Literal["acquisition", "processing/ophys"] = "acquisition", + iterator_type: str | None = "v2", + iterator_options: dict | None = None, + always_write_timestamps: bool = False, +) -> NWBFile: + """ + Add a photon series using the dict-based metadata format. + + Looks up the microscopy series in ``metadata["Ophys"]["MicroscopySeries"][metadata_key]`` + and creates it in the NWBFile. Resolves the imaging plane via ``imaging_plane_metadata_key`` + in the series metadata. + + Parameters + ---------- + imaging : ImagingExtractor + The imaging extractor to get the data from. + nwbfile : NWBFile + The NWB file to add the photon series to. + metadata : dict + The full metadata dictionary with dict-based format. + photon_series_type : {'OnePhotonSeries', 'TwoPhotonSeries'} + The NWB type of photon series to create. + metadata_key : str + The key in ``metadata["Ophys"]["MicroscopySeries"]`` identifying the series. + parent_container : {'acquisition', 'processing/ophys'}, optional + The container where the photon series is added, default is nwbfile.acquisition. + iterator_type : str, default: 'v2' + The type of iterator to use when adding the photon series to the NWB file. + iterator_options : dict, optional + always_write_timestamps : bool, default: False + Set to True to always write timestamps. + + Returns + ------- + NWBFile + The NWBFile passed as an input with the photon series added. + """ + iterator_options = iterator_options or dict() + + photon_series_metadata = metadata["Ophys"]["MicroscopySeries"][metadata_key] + + # Copy to avoid mutation + photon_series_kwargs = photon_series_metadata.copy() + + # Validate required fields + required_fields = ["name", "unit"] + missing_fields = [field for field in required_fields if field not in photon_series_kwargs] + if missing_fields: + default_series = _get_ophys_metadata_placeholders()["Ophys"]["MicroscopySeries"]["default_metadata_key"] + placeholder_hint = "\n".join(f" {field}: {default_series[field]!r}" for field in missing_fields) + raise ValueError( + f"Microscopy series metadata is missing required fields.\n" + f"For a complete NWB file, the following fields should be provided. " + f"If missing, a placeholder can be used instead:\n{placeholder_hint}" + ) + + # Resolve imaging plane + imaging_plane_metadata_key = photon_series_kwargs.pop("imaging_plane_metadata_key", None) + if imaging_plane_metadata_key is not None: + imaging_plane_metadata = metadata["Ophys"]["ImagingPlanes"][imaging_plane_metadata_key] + else: + default_metadata = _get_ophys_metadata_placeholders() + imaging_plane_metadata = default_metadata["Ophys"]["ImagingPlanes"]["default_metadata_key"] + imaging_plane = _add_imaging_plane_to_nwbfile( + nwbfile=nwbfile, + imaging_plane_metadata=imaging_plane_metadata, + metadata=metadata, + ) + photon_series_kwargs["imaging_plane"] = imaging_plane + + # Add dimension if not in metadata + if "dimension" not in photon_series_kwargs: + photon_series_kwargs["dimension"] = imaging.get_sample_shape() + + # Add data iterator + imaging_extractor_iterator = _imaging_frames_to_hdmf_iterator( + imaging=imaging, + iterator_type=iterator_type, + iterator_options=iterator_options, + ) + photon_series_kwargs["data"] = imaging_extractor_iterator + + if always_write_timestamps: + timestamps = imaging.get_timestamps() + photon_series_kwargs.update(timestamps=timestamps) + else: + # Resolve timestamps: user-set > native hardware > none + timestamps_were_set = imaging.has_time_vector() + if timestamps_were_set: + timestamps = imaging.get_timestamps() + else: + timestamps = imaging.get_native_timestamps() + + timestamps_are_available = timestamps is not None + + if timestamps_are_available: + rate = calculate_regular_series_rate(series=timestamps) + timestamps_are_regular = rate is not None + starting_time = timestamps[0] + else: + rate = float(imaging.get_sampling_frequency()) + timestamps_are_regular = True + starting_time = 0.0 + + if timestamps_are_regular: + photon_series_kwargs.update(rate=rate, starting_time=starting_time) + else: + photon_series_kwargs.update(timestamps=timestamps) + + # Add the photon series to the nwbfile + photon_series_map = dict(OnePhotonSeries=OnePhotonSeries, TwoPhotonSeries=TwoPhotonSeries) + photon_series_class = photon_series_map[photon_series_type] + photon_series = photon_series_class(**photon_series_kwargs) + + if parent_container == "acquisition": + nwbfile.add_acquisition(photon_series) + elif parent_container == "processing/ophys": + ophys_module = get_module(nwbfile, name="ophys", description="contains optical physiology processed data") + ophys_module.add(photon_series) + + return nwbfile + + def _check_if_imaging_fits_into_memory(imaging: ImagingExtractor) -> None: """ Raise an error if the full traces of an imaging extractor are larger than available memory. @@ -609,10 +942,15 @@ def add_imaging_to_nwbfile( iterator_options: dict | None = None, parent_container: Literal["acquisition", "processing/ophys"] = "acquisition", always_write_timestamps: bool = False, + # TODO: move metadata_key after metadata once positional args removed (September 2026) + metadata_key: str | None = None, ) -> NWBFile: """ Add imaging data from an ImagingExtractor object to an NWBFile. + Supports both old list-based metadata (via ``photon_series_index``) and + new dict-based metadata (via ``metadata_key``). + Parameters ---------- imaging : ImagingExtractor @@ -625,6 +963,7 @@ def add_imaging_to_nwbfile( The type of photon series to be added, by default "TwoPhotonSeries". photon_series_index : int, optional The index of the photon series in the provided imaging data, by default 0. + Used with the old list-based metadata format. iterator_type : str, optional The type of iterator to use for adding the data. Commonly used to manage large datasets, by default "v2". iterator_options : dict, optional @@ -637,6 +976,9 @@ def add_imaging_to_nwbfile( By default (False), the function checks if the timestamps are uniformly sampled, and if so, stores the data using a regular sampling rate instead of explicit timestamps. If set to True, timestamps will be written explicitly, regardless of whether the sampling rate is uniform. + metadata_key : str, optional + The key in ``metadata["Ophys"]["MicroscopySeries"]`` identifying the series. + When provided, uses the new dict-based metadata format and ``photon_series_index`` is ignored. Returns ------- @@ -678,18 +1020,35 @@ def add_imaging_to_nwbfile( parent_container = positional_values.get("parent_container", parent_container) always_write_timestamps = positional_values.get("always_write_timestamps", always_write_timestamps) - add_devices_to_nwbfile(nwbfile=nwbfile, metadata=metadata) - nwbfile = _add_photon_series_to_nwbfile_old_list_format( - imaging=imaging, - nwbfile=nwbfile, - metadata=metadata, - photon_series_type=photon_series_type, - photon_series_index=photon_series_index, - iterator_type=iterator_type, - iterator_options=iterator_options, - parent_container=parent_container, - always_write_timestamps=always_write_timestamps, - ) + if metadata is None: + metadata = _get_ophys_metadata_placeholders() + + if _is_dict_based_metadata(metadata): + metadata_key = metadata_key or "default_metadata_key" + nwbfile = _add_photon_series_to_nwbfile( + imaging=imaging, + nwbfile=nwbfile, + metadata=metadata, + photon_series_type=photon_series_type, + metadata_key=metadata_key, + iterator_type=iterator_type, + iterator_options=iterator_options, + parent_container=parent_container, + always_write_timestamps=always_write_timestamps, + ) + else: + _add_devices_to_nwbfile_old_list_format(nwbfile=nwbfile, metadata=metadata) + nwbfile = _add_photon_series_to_nwbfile_old_list_format( + imaging=imaging, + nwbfile=nwbfile, + metadata=metadata, + photon_series_type=photon_series_type, + photon_series_index=photon_series_index, + iterator_type=iterator_type, + iterator_options=iterator_options, + parent_container=parent_container, + always_write_timestamps=always_write_timestamps, + ) return nwbfile @@ -1593,7 +1952,7 @@ def add_segmentation_to_nwbfile( """ # Add device: - add_devices_to_nwbfile(nwbfile=nwbfile, metadata=metadata) + _add_devices_to_nwbfile_old_list_format(nwbfile=nwbfile, metadata=metadata) # Add PlaneSegmentation: _add_plane_segmentation_to_nwbfile( diff --git a/tests/test_modalities/test_ophys/test_tools_roiextractors.py b/tests/test_modalities/test_ophys/test_tools_roiextractors.py index 8a39e1bfb..6f4127f1b 100644 --- a/tests/test_modalities/test_ophys/test_tools_roiextractors.py +++ b/tests/test_modalities/test_ophys/test_tools_roiextractors.py @@ -30,6 +30,7 @@ _check_if_imaging_fits_into_memory, add_devices_to_nwbfile, add_fluorescence_traces_to_nwbfile, + add_imaging_to_nwbfile, ) from neuroconv.tools.roiextractors.imagingextractordatachunkiterator import ( ImagingExtractorDataChunkIterator, @@ -41,6 +42,8 @@ _add_plane_segmentation_to_nwbfile, _add_summary_images_to_nwbfile, _get_default_ophys_metadata_old_metadata_list, + _get_ophys_metadata_placeholders, + get_full_ophys_metadata, ) from neuroconv.utils import dict_deep_update @@ -2264,3 +2267,460 @@ def test_add_fluorescence_traces_no_metadata_mutation(self): # Verify metadata was not mutated - compare entire dict structure assert metadata == metadata_before, "Metadata was mutated" + + +class TestAddImaging: + """Tests for the dict-based metadata imaging pipeline (add_imaging_to_nwbfile).""" + + def test_basic(self): + """Test expected values for no metadata specification.""" + nwbfile = mock_NWBFile() + num_samples = 10 + num_rows = 5 + num_columns = 5 + imaging = generate_dummy_imaging_extractor(num_samples=num_samples, num_rows=num_rows, num_columns=num_columns) + + add_imaging_to_nwbfile( + imaging=imaging, + nwbfile=nwbfile, + ) + + default_metadata = _get_ophys_metadata_placeholders() + default_key = "default_metadata_key" + default_device_metadata = default_metadata["Devices"][default_key] + default_plane_metadata = default_metadata["Ophys"]["ImagingPlanes"][default_key] + default_series_metadata = default_metadata["Ophys"]["MicroscopySeries"][default_key] + + # Default device + assert len(nwbfile.devices) == 1 + device = nwbfile.devices[default_device_metadata["name"]] + assert device.name == default_device_metadata["name"] + + # Default imaging plane + assert len(nwbfile.imaging_planes) == 1 + plane = nwbfile.imaging_planes[default_plane_metadata["name"]] + assert plane.name == default_plane_metadata["name"] + assert np.isnan(plane.excitation_lambda) + assert plane.indicator == default_plane_metadata["indicator"] + assert plane.location == default_plane_metadata["location"] + assert plane.device is device + + # Default series with correct data shape + assert len(nwbfile.acquisition) == 1 + series = nwbfile.acquisition[default_series_metadata["name"]] + assert series.name == default_series_metadata["name"] + assert series.unit == default_series_metadata["unit"] + assert series.imaging_plane is plane + assert isinstance(series.data, ImagingExtractorDataChunkIterator) + assert series.data.maxshape == (num_samples, num_rows, num_columns) + + def test_full_metadata_specification(self): + """Full metadata specification: device, imaging plane, and series are created from user metadata.""" + nwbfile = mock_NWBFile() + imaging = generate_dummy_imaging_extractor(num_samples=10, num_rows=5, num_columns=5) + + metadata = get_full_ophys_metadata() + metadata_key = "my_series" + add_imaging_to_nwbfile( + imaging=imaging, + nwbfile=nwbfile, + metadata=metadata, + metadata_key=metadata_key, + ) + + series_metadata = metadata["Ophys"]["MicroscopySeries"][metadata_key] + plane_key = series_metadata["imaging_plane_metadata_key"] + plane_metadata = metadata["Ophys"]["ImagingPlanes"][plane_key] + device_key = plane_metadata["device_metadata_key"] + device_metadata = metadata["Devices"][device_key] + + device = nwbfile.devices[device_metadata["name"]] + assert device.description == device_metadata["description"] + assert device.manufacturer == device_metadata["manufacturer"] + + plane = nwbfile.imaging_planes[plane_metadata["name"]] + assert plane.description == plane_metadata["description"] + assert plane.indicator == plane_metadata["indicator"] + assert plane.location == plane_metadata["location"] + assert plane.device is device + + series = nwbfile.acquisition[series_metadata["name"]] + assert series.description == series_metadata["description"] + assert series.imaging_plane is plane + + def test_no_imaging_plane_metadata_key(self): + """When microscopy series has no imaging_plane_metadata_key, a default imaging plane is created.""" + nwbfile = mock_NWBFile() + imaging = generate_dummy_imaging_extractor(num_samples=10, num_rows=5, num_columns=5) + + metadata = { + "Devices": {}, + "Ophys": { + "ImagingPlanes": {}, + "MicroscopySeries": { + "my_series": { + "name": "TwoPhotonSeries", + "description": "Imaging data", + "unit": "n.a.", + }, + }, + }, + } + + add_imaging_to_nwbfile( + imaging=imaging, + nwbfile=nwbfile, + metadata=metadata, + photon_series_type="TwoPhotonSeries", + metadata_key="my_series", + ) + + default_metadata = _get_ophys_metadata_placeholders() + default_key = "default_metadata_key" + default_plane_metadata = default_metadata["Ophys"]["ImagingPlanes"][default_key] + default_device_metadata = default_metadata["Devices"][default_key] + + # Default imaging plane + plane = nwbfile.imaging_planes[default_plane_metadata["name"]] + assert plane.name == default_plane_metadata["name"] + + # Default device + device = nwbfile.devices[default_device_metadata["name"]] + assert device.name == default_device_metadata["name"] + assert plane.device is device + + def test_no_device_metadata_key(self): + """When imaging plane has no device_metadata_key, a default device is created.""" + nwbfile = mock_NWBFile() + imaging = generate_dummy_imaging_extractor(num_samples=10, num_rows=5, num_columns=5) + + metadata = { + "Devices": {}, + "Ophys": { + "ImagingPlanes": { + "my_plane": { + "name": "ImagingPlane", + "description": "A plane", + "excitation_lambda": 920.0, + "indicator": "GCaMP6s", + "location": "V1", + "optical_channel": [ + { + "name": "Green", + "description": "GCaMP emission", + "emission_lambda": 510.0, + } + ], + }, + }, + "MicroscopySeries": { + "my_series": { + "name": "TwoPhotonSeries", + "description": "Imaging data", + "unit": "n.a.", + "imaging_plane_metadata_key": "my_plane", + }, + }, + }, + } + + add_imaging_to_nwbfile( + imaging=imaging, + nwbfile=nwbfile, + metadata=metadata, + photon_series_type="TwoPhotonSeries", + metadata_key="my_series", + ) + + default_metadata = _get_ophys_metadata_placeholders() + default_key = "default_metadata_key" + default_device_metadata = default_metadata["Devices"][default_key] + + device = nwbfile.devices[default_device_metadata["name"]] + assert device.name == default_device_metadata["name"] + plane = nwbfile.imaging_planes["ImagingPlane"] + assert plane.device is device + + def test_shared_imaging_plane_two_microscopy_series(self): + """Two microscopy series referencing the same imaging plane via imaging_plane_metadata_key.""" + nwbfile = mock_NWBFile() + imaging = generate_dummy_imaging_extractor(num_samples=10, num_rows=5, num_columns=5) + + shared_plane_key = "shared_plane" + first_series_key = "series_a" + second_series_key = "series_b" + metadata = { + "Devices": { + "my_device": { + "name": "Microscope", + "description": "Two-photon microscope", + }, + }, + "Ophys": { + "ImagingPlanes": { + shared_plane_key: { + "name": "SharedImagingPlane", + "description": "Shared plane", + "excitation_lambda": 920.0, + "indicator": "GCaMP6s", + "location": "V1", + "device_metadata_key": "my_device", + "optical_channel": [{"name": "Green", "description": "GCaMP", "emission_lambda": 510.0}], + }, + }, + "MicroscopySeries": { + first_series_key: { + "name": "TwoPhotonSeriesA", + "description": "First series", + "unit": "n.a.", + "imaging_plane_metadata_key": shared_plane_key, + }, + second_series_key: { + "name": "TwoPhotonSeriesB", + "description": "Second series", + "unit": "n.a.", + "imaging_plane_metadata_key": shared_plane_key, + }, + }, + }, + } + + add_imaging_to_nwbfile(imaging=imaging, nwbfile=nwbfile, metadata=metadata, metadata_key=first_series_key) + add_imaging_to_nwbfile(imaging=imaging, nwbfile=nwbfile, metadata=metadata, metadata_key=second_series_key) + + device_metadata = metadata["Devices"]["my_device"] + plane_metadata = metadata["Ophys"]["ImagingPlanes"][shared_plane_key] + first_series_metadata = metadata["Ophys"]["MicroscopySeries"][first_series_key] + second_series_metadata = metadata["Ophys"]["MicroscopySeries"][second_series_key] + + # One device, one plane, two series + assert len(nwbfile.devices) == 1 + assert len(nwbfile.imaging_planes) == 1 + assert len(nwbfile.acquisition) == 2 + + # Device + device = nwbfile.devices[device_metadata["name"]] + assert device.name == device_metadata["name"] + assert device.description == device_metadata["description"] + + # Imaging plane + plane = nwbfile.imaging_planes[plane_metadata["name"]] + assert plane.name == plane_metadata["name"] + assert plane.description == plane_metadata["description"] + assert plane.excitation_lambda == plane_metadata["excitation_lambda"] + assert plane.indicator == plane_metadata["indicator"] + assert plane.location == plane_metadata["location"] + assert plane.device is device + + # Both series share the same imaging plane + series_a = nwbfile.acquisition[first_series_metadata["name"]] + assert series_a.description == first_series_metadata["description"] + assert series_a.unit == first_series_metadata["unit"] + assert series_a.imaging_plane is plane + + series_b = nwbfile.acquisition[second_series_metadata["name"]] + assert series_b.description == second_series_metadata["description"] + assert series_b.unit == second_series_metadata["unit"] + assert series_b.imaging_plane is plane + + def test_shared_device_two_imaging_planes(self): + """Two imaging planes referencing the same device via device_metadata_key.""" + nwbfile = mock_NWBFile() + imaging = generate_dummy_imaging_extractor(num_samples=10, num_rows=5, num_columns=5) + + metadata = { + "Devices": { + "shared_device_key": { + "name": "SharedMicroscope", + "description": "Shared two-photon microscope", + "manufacturer": "Bruker", + }, + }, + "Ophys": { + "ImagingPlanes": { + "plane_v1": { + "name": "ImagingPlaneV1", + "description": "Visual cortex V1", + "excitation_lambda": 920.0, + "indicator": "GCaMP6s", + "location": "V1", + "device_metadata_key": "shared_device_key", + "optical_channel": [{"name": "Green", "description": "GCaMP", "emission_lambda": 510.0}], + }, + "plane_v2": { + "name": "ImagingPlaneV2", + "description": "Visual cortex V2", + "excitation_lambda": 920.0, + "indicator": "GCaMP6f", + "location": "V2", + "device_metadata_key": "shared_device_key", + "optical_channel": [{"name": "Green", "description": "GCaMP", "emission_lambda": 510.0}], + }, + }, + "MicroscopySeries": { + "series_v1": { + "name": "TwoPhotonSeriesV1", + "description": "V1 imaging", + "unit": "n.a.", + "imaging_plane_metadata_key": "plane_v1", + }, + "series_v2": { + "name": "TwoPhotonSeriesV2", + "description": "V2 imaging", + "unit": "n.a.", + "imaging_plane_metadata_key": "plane_v2", + }, + }, + }, + } + + add_imaging_to_nwbfile( + imaging=imaging, + nwbfile=nwbfile, + metadata=metadata, + photon_series_type="TwoPhotonSeries", + metadata_key="series_v1", + ) + add_imaging_to_nwbfile( + imaging=imaging, + nwbfile=nwbfile, + metadata=metadata, + photon_series_type="TwoPhotonSeries", + metadata_key="series_v2", + ) + + device_metadata = metadata["Devices"]["shared_device_key"] + + # One device, two planes, two series + assert len(nwbfile.devices) == 1 + assert len(nwbfile.imaging_planes) == 2 + assert len(nwbfile.acquisition) == 2 + + # Device matches metadata + unique_device = nwbfile.devices[device_metadata["name"]] + assert unique_device.name == device_metadata["name"] + assert unique_device.description == device_metadata["description"] + assert unique_device.manufacturer == device_metadata["manufacturer"] + + # Both planes share the same device + assert nwbfile.imaging_planes["ImagingPlaneV1"].device is unique_device + assert nwbfile.imaging_planes["ImagingPlaneV2"].device is unique_device + + def test_repeated_calls_reuse_default_metadata_placeholders(self): + """Repeated calls reuse the same placeholder device and imaging plane. + + Default metadata values are placeholders, not real data. When the user does not provide + metadata, neuroconv should not fabricate additional objects on each call. Instead, the + same placeholder device and placeholder so downstream tools might identify can + flag this. + """ + nwbfile = mock_NWBFile() + imaging = generate_dummy_imaging_extractor(num_samples=10, num_rows=5, num_columns=5) + + first_metadata_key = "first" + second_metadata_key = "second" + metadata = { + "Devices": {}, + "Ophys": { + "ImagingPlanes": {}, + "MicroscopySeries": { + first_metadata_key: { + "name": "TwoPhotonSeriesFirst", + "unit": "n.a.", + }, + second_metadata_key: { + "name": "TwoPhotonSeriesSecond", + "unit": "n.a.", + }, + }, + }, + } + + add_imaging_to_nwbfile(imaging=imaging, nwbfile=nwbfile, metadata=metadata, metadata_key=first_metadata_key) + add_imaging_to_nwbfile(imaging=imaging, nwbfile=nwbfile, metadata=metadata, metadata_key=second_metadata_key) + + # Placeholder device and imaging plane are reused, not duplicated + assert len(nwbfile.devices) == 1 + assert len(nwbfile.imaging_planes) == 1 + assert len(nwbfile.acquisition) == 2 + + def test_missing_required_imaging_plane_fields_raises(self): + """When an imaging plane entry is missing schema-required fields, a clear error is raised.""" + nwbfile = mock_NWBFile() + imaging = generate_dummy_imaging_extractor(num_samples=10, num_rows=5, num_columns=5) + + metadata_key = "my_series" + device_key = "my_device" + plane_key = "my_plane" + metadata = { + "Devices": {device_key: {"name": "Microscope"}}, + "Ophys": { + "ImagingPlanes": { + plane_key: { + "name": "ImagingPlane", + "device_metadata_key": device_key, + }, + }, + "MicroscopySeries": { + metadata_key: { + "name": "TwoPhotonSeries", + "unit": "n.a.", + "imaging_plane_metadata_key": plane_key, + }, + }, + }, + } + + expected_error = re.escape( + "Imaging plane metadata is missing required fields.\n" + "For a complete NWB file, the following fields should be provided. If missing, a placeholder can be used instead:\n" + " excitation_lambda: nan\n" + " indicator: 'unknown'\n" + " location: 'unknown'\n" + " optical_channel: [{'name': 'OpticalChannel', 'emission_lambda': nan, 'description': 'An optical channel of the microscope.'}]" + ) + with pytest.raises(ValueError, match=expected_error): + add_imaging_to_nwbfile(imaging=imaging, nwbfile=nwbfile, metadata=metadata, metadata_key=metadata_key) + + def test_missing_required_series_fields_raises(self): + """When a series entry is missing schema-required fields, a clear error is raised.""" + nwbfile = mock_NWBFile() + imaging = generate_dummy_imaging_extractor(num_samples=10, num_rows=5, num_columns=5) + + metadata_key = "my_series" + metadata = { + "Devices": {}, + "Ophys": { + "ImagingPlanes": {}, + "MicroscopySeries": { + metadata_key: { + "name": "TwoPhotonSeries", + }, + }, + }, + } + + expected_error = re.escape( + "Microscopy series metadata is missing required fields.\n" + "For a complete NWB file, the following fields should be provided. If missing, a placeholder can be used instead:\n" + " unit: 'n.a.'" + ) + with pytest.raises(ValueError, match=expected_error): + add_imaging_to_nwbfile(imaging=imaging, nwbfile=nwbfile, metadata=metadata, metadata_key=metadata_key) + + def test_metadata_not_mutated(self): + """Dict-based metadata is not mutated by add_imaging_to_nwbfile.""" + nwbfile = mock_NWBFile() + imaging = generate_dummy_imaging_extractor(num_samples=10, num_rows=5, num_columns=5) + + metadata = get_full_ophys_metadata() + metadata_before = deepcopy(metadata) + + add_imaging_to_nwbfile( + imaging=imaging, + nwbfile=nwbfile, + metadata=metadata, + metadata_key="my_series", + ) + + assert metadata == metadata_before, "Metadata was mutated"