Source code for neuroconv.datainterfaces.ophys.baseimagingextractorinterface
"""Author: Ben Dichter."""
import warnings
from typing import Literal
import numpy as np
from pynwb import NWBFile
from pynwb.device import Device
from pynwb.ophys import ImagingPlane, OnePhotonSeries, TwoPhotonSeries
from ._metadata_schema import _get_ophys_registry_entry_definitions, _keyed_registry
from ._metadata_template import (
_get_imaging_plane_template_entry,
_resolve_device_metadata_key,
)
from ...baseextractorinterface import BaseExtractorInterface
from ...tools.nwb_helpers._metadata_and_file_helpers import (
_get_device_model_template_entry,
_get_device_template_entry,
)
from ...utils import (
DeepDict,
dict_deep_update,
fill_defaults,
get_base_schema,
get_schema_from_hdmf_class,
)
[docs]
class BaseImagingExtractorInterface(BaseExtractorInterface):
"""Parent class for all ImagingExtractorInterfaces."""
keywords = (
"ophys",
"optical electrophysiology",
"fluorescence",
"microscopy",
"two photon",
"one photon",
"voltage imaging",
"calcium imaging",
)
def _initialize_extractor(self, interface_kwargs: dict):
"""
Initialize and return the extractor instance for imaging interfaces.
Extends the base implementation to also remove the 'photon_series_type' and
'metadata_key' parameters which are specific to the imaging interface, not the extractor.
Parameters
----------
interface_kwargs : dict
The source data parameters passed to the interface constructor.
Returns
-------
extractor_instance
An initialized imaging extractor instance.
"""
self.extractor_kwargs = interface_kwargs.copy()
self.extractor_kwargs.pop("verbose", None)
self.extractor_kwargs.pop("photon_series_type", None)
self.extractor_kwargs.pop("metadata_key", None)
extractor_class = self.get_extractor_class()
extractor_instance = extractor_class(**self.extractor_kwargs)
return extractor_instance
def __init__(
self,
verbose: bool = False,
photon_series_type: Literal["OnePhotonSeries", "TwoPhotonSeries"] = "TwoPhotonSeries",
metadata_key: str | None = None,
**source_data,
):
from roiextractors import ImagingExtractor
super().__init__(**source_data)
self.imaging_extractor: ImagingExtractor = self._extractor_instance
self.verbose = verbose
self.photon_series_type = photon_series_type
self.metadata_key = metadata_key
[docs]
def get_metadata_schema(self) -> dict:
"""
Compile the metadata schema.
The registries are objects keyed by ``metadata_key``, and the entries stay permissive: an entry is
passed to a pynwb constructor, so it may legitimately carry any field that constructor takes. What is
pinned is the shape, that an entry is an object, and the cross-reference fields
(``device_metadata_key``, ``imaging_plane_metadata_key``) that no hdmf class knows about.
Metadata in the old list-based format is validated against
``_get_metadata_schema_for_old_list_format``, and both go when that format does.
"""
from ...basedatainterface import BaseDataInterface
metadata_schema = BaseDataInterface.get_metadata_schema(self)
metadata_schema["properties"]["Ophys"] = get_base_schema(tag="Ophys")
metadata_schema["properties"]["Ophys"]["required"] = []
metadata_schema["properties"]["Ophys"]["properties"] = dict(
ImagingPlanes=_keyed_registry("#/properties/Ophys/definitions/ImagingPlaneEntry"),
MicroscopySeries=_keyed_registry("#/properties/Ophys/definitions/MicroscopySeriesEntry"),
)
metadata_schema["properties"]["Ophys"]["definitions"] = _get_ophys_registry_entry_definitions()
return metadata_schema
def _get_metadata_schema_for_old_list_format(self) -> dict:
"""
Retrieve the metadata schema for the optical physiology (Ophys) data.
Returns
-------
dict
The metadata schema dictionary containing definitions for Device, ImagingPlane,
and either OnePhotonSeries or TwoPhotonSeries based on the photon_series_type.
"""
metadata_schema = super().get_metadata_schema()
metadata_schema["required"] = ["Ophys"]
# Initiate Ophys metadata
metadata_schema["properties"]["Ophys"] = get_base_schema(tag="Ophys")
metadata_schema["properties"]["Ophys"]["required"] = ["Device", "ImagingPlane", self.photon_series_type]
metadata_schema["properties"]["Ophys"]["properties"] = dict(
Device=dict(type="array", minItems=1, items={"$ref": "#/properties/Ophys/definitions/Device"}),
ImagingPlane=dict(type="array", minItems=1, items={"$ref": "#/properties/Ophys/definitions/ImagingPlane"}),
)
metadata_schema["properties"]["Ophys"]["properties"].update(
{
self.photon_series_type: dict(
type="array",
minItems=1,
items={"$ref": f"#/properties/Ophys/definitions/{self.photon_series_type}"},
),
}
)
# Schema definition for arrays
imaging_plane_schema = get_schema_from_hdmf_class(ImagingPlane)
imaging_plane_schema["properties"]["optical_channel"].pop("maxItems")
metadata_schema["properties"]["Ophys"]["definitions"] = dict(
Device=get_schema_from_hdmf_class(Device),
ImagingPlane=imaging_plane_schema,
)
photon_series = dict(
OnePhotonSeries=OnePhotonSeries,
TwoPhotonSeries=TwoPhotonSeries,
)[self.photon_series_type]
metadata_schema["properties"]["Ophys"]["definitions"].update(
{
self.photon_series_type: get_schema_from_hdmf_class(photon_series),
}
)
fill_defaults(metadata_schema, self.get_metadata())
return metadata_schema
[docs]
def get_metadata(self, *, use_new_metadata_format: bool = True) -> DeepDict:
"""
Retrieve the metadata for the imaging data.
Parameters
----------
use_new_metadata_format : bool, default: True
When False, returns the old list-based metadata format (backward compatible).
When True, returns only NWBFile-level metadata (session_description, identifier,
etc.) without ophys keys. Ophys defaults are filled by ``add_imaging_to_nwbfile()``
internally.
Returns
-------
DeepDict
Dictionary containing metadata. When use_new_metadata_format is False, includes
device information, imaging plane details, and photon series configuration.
When True, includes only NWBFile basics.
"""
if use_new_metadata_format:
metadata = super().get_metadata()
# Mirrors ``BaseRecordingExtractorInterface``: the base states the conventional default name
# for the series it writes, and interfaces that know better overwrite it. ``MicroscopySeries``
# is the forward-looking generic, used wherever the source does not say what was imaged.
metadata["Ophys"] = {"MicroscopySeries": {self.metadata_key: dict(name="MicroscopySeries")}}
return metadata
# Old list-based path (unchanged)
from ...tools.roiextractors import get_nwb_imaging_metadata
metadata = super().get_metadata()
default_metadata = get_nwb_imaging_metadata(self.imaging_extractor, photon_series_type=self.photon_series_type)
metadata = dict_deep_update(default_metadata, metadata)
# fix troublesome data types
if "TwoPhotonSeries" in metadata["Ophys"]:
for two_photon_series in metadata["Ophys"]["TwoPhotonSeries"]:
if "dimension" in two_photon_series:
two_photon_series["dimension"] = list(two_photon_series["dimension"])
if "rate" in two_photon_series:
two_photon_series["rate"] = float(two_photon_series["rate"])
return metadata
[docs]
def get_metadata_template(self) -> DeepDict:
"""Return the imaging plane and series this interface writes, with the blanks marked.
The counterpart to :meth:`get_metadata`, which reports only what the source recorded and so
leaves a user no indication of what else the file needs. This returns those same values wrapped
in the structure the writer expects. Fill in the blanks and pass the result to ``add_to_nwbfile``
or ``run_conversion``; a blank still ``None`` at write time is an error rather than a value.
One imaging plane and one series, both under this interface's ``metadata_key`` and already
cross-referenced, hanging off one microscope. What is left ``None`` is what only the experimenter
can supply: the brain region imaged, the indicator, the excitation and emission wavelengths and
the name every object needs. The optional fields appear so that a user knows the writer accepts
them; delete the ones this recording cannot answer, since a blank left behind is refused at write
time rather than guessed at.
Rename the keys to suit the recording; they are handles, not names in the file.
"""
# What ``add_imaging_to_nwbfile`` falls back to for an interface constructed without a key.
metadata_key = self.metadata_key or "default_metadata_key"
# Prefilled through the same transitional shim the writers use, so an interface whose
# ``get_metadata`` still answers in the old list format does not leak that shape into the
# template. When the old format goes the shim goes with it, and this becomes
# ``self.get_metadata()`` with nothing else to change.
source_metadata = self._get_metadata_for_writing()
device_metadata_key = _resolve_device_metadata_key(source_metadata=source_metadata, metadata_key=metadata_key)
# Optional fields of the series being written, which differ by photon series: a one-photon
# acquisition describes its camera exposure, a two-photon one its scanner.
optional_series_fields = dict(
TwoPhotonSeries=dict(field_of_view=None, pmt_gain=None, scan_line_rate=None),
OnePhotonSeries=dict(exposure_time=None, binning=None, power=None, intensity=None),
)[self.photon_series_type]
series_entry = dict(
name=None,
description=None,
unit=None,
imaging_plane_metadata_key=metadata_key,
**optional_series_fields,
)
device_model_metadata_key = f"{device_metadata_key}_model"
template = DeepDict(
dict(
DeviceModels={device_model_metadata_key: _get_device_model_template_entry()},
Devices={
device_metadata_key: _get_device_template_entry(device_model_metadata_key=device_model_metadata_key)
},
Ophys=dict(
ImagingPlanes={
metadata_key: _get_imaging_plane_template_entry(device_metadata_key=device_metadata_key)
},
MicroscopySeries={metadata_key: series_entry},
),
)
)
# The blanks are a floor rather than an override: whatever the source recorded wins over the
# template, so a field the interface was able to read is never handed back as one to fill in.
template.deep_update(source_metadata)
return template
[docs]
def get_original_timestamps(self) -> np.ndarray:
reinitialized_extractor = self._initialize_extractor(self.source_data)
return reinitialized_extractor.get_timestamps()
[docs]
def set_aligned_timestamps(self, aligned_timestamps: np.ndarray):
self.imaging_extractor.set_times(times=aligned_timestamps)
[docs]
def add_to_nwbfile(
self,
nwbfile: NWBFile,
metadata: dict | None = None,
*args, # TODO: change to * (keyword only) on or after August 2026
photon_series_type: Literal["TwoPhotonSeries", "OnePhotonSeries"] = "TwoPhotonSeries",
photon_series_index: int = 0,
parent_container: Literal["acquisition", "processing/ophys"] = "acquisition",
stub_test: bool = False,
always_write_timestamps: bool = False,
iterator_type: str | None = "v2",
iterator_options: dict | None = None,
stub_samples: int = 100,
):
"""
Add imaging data to the NWB file
Parameters
----------
nwbfile : NWBFile
The NWB file where the imaging data will be added.
metadata : dict, optional
Metadata for the NWBFile, by default None.
photon_series_type : {"TwoPhotonSeries", "OnePhotonSeries"}, optional
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.
parent_container : {"acquisition", "processing/ophys"}, optional
Specifies the parent container to which the photon series should be added, either as part of "acquisition" or
under the "processing/ophys" module, by default "acquisition".
stub_test : bool, optional
If True, only writes a small subset of frames for testing purposes, by default False.
always_write_timestamps : bool, optional
Whether to always write timestamps, by default False.
iterator_type : {"v2", None}, default: "v2"
The type of iterator for chunked data writing.
'v2': Uses iterative write with control over chunking and progress bars.
None: Loads all data into memory before writing (not recommended for large datasets).
iterator_options : dict, optional
Options for controlling the iterative write process (buffer size, progress bars).
See the `pynwb tutorial on iterative write <https://pynwb.readthedocs.io/en/stable/tutorials/advanced_io/plot_iterative_write.html#sphx-glr-tutorials-advanced-io-plot-iterative-write-py>`_
for more information on chunked data writing.
Note: To configure chunk size and compression, use the backend configuration system
via ``get_default_backend_configuration()`` and ``configure_backend()`` after calling
this method. See the backend configuration documentation for details.
stub_samples : int, default: 100
The number of samples (frames) to use for testing.
"""
from ...tools.roiextractors import add_imaging_to_nwbfile
# TODO: Remove this block in August 2026 or after when positional arguments are no longer supported.
if args:
parameter_names = [
"photon_series_type",
"photon_series_index",
"parent_container",
"stub_test",
"always_write_timestamps",
"iterator_type",
"iterator_options",
"stub_samples",
]
num_positional_args_before_args = 2 # nwbfile, metadata
if len(args) > len(parameter_names):
raise TypeError(
f"add_to_nwbfile() takes at most {len(parameter_names) + num_positional_args_before_args} positional arguments but "
f"{len(args) + num_positional_args_before_args} were given. "
"Note: Positional arguments are deprecated and will be removed in August 2026 or after. Please use keyword arguments."
)
positional_values = dict(zip(parameter_names, args))
passed_as_positional = list(positional_values.keys())
warnings.warn(
f"Passing arguments positionally to add_to_nwbfile is deprecated "
f"and will be removed in August 2026 or after. "
f"The following arguments were passed positionally: {passed_as_positional}. "
"Please use keyword arguments instead.",
FutureWarning,
stacklevel=2,
)
photon_series_type = positional_values.get("photon_series_type", photon_series_type)
photon_series_index = positional_values.get("photon_series_index", photon_series_index)
parent_container = positional_values.get("parent_container", parent_container)
stub_test = positional_values.get("stub_test", stub_test)
always_write_timestamps = positional_values.get("always_write_timestamps", always_write_timestamps)
iterator_type = positional_values.get("iterator_type", iterator_type)
iterator_options = positional_values.get("iterator_options", iterator_options)
stub_samples = positional_values.get("stub_samples", stub_samples)
if stub_test:
stub_samples = min([stub_samples, self.imaging_extractor.get_num_samples()])
imaging_extractor = self.imaging_extractor.slice_samples(start_sample=0, end_sample=stub_samples)
else:
imaging_extractor = self.imaging_extractor
# TODO: change to self.get_metadata(use_new_metadata_format=True) when all imaging interfaces are migrated
metadata = metadata or self._get_metadata_for_writing()
add_imaging_to_nwbfile(
imaging=imaging_extractor,
nwbfile=nwbfile,
metadata=metadata,
photon_series_type=photon_series_type,
photon_series_index=photon_series_index,
parent_container=parent_container,
always_write_timestamps=always_write_timestamps,
iterator_type=iterator_type,
iterator_options=iterator_options,
metadata_key=self.metadata_key,
)