import math
import warnings
from typing import Literal
import numpy as np
import psutil
from pydantic import FilePath
from pynwb import NWBFile
from pynwb.base import Images
from pynwb.image import GrayscaleImage
from pynwb.ophys import (
Fluorescence,
ImageSegmentation,
ImagingPlane,
OnePhotonSeries,
PlaneSegmentation,
RoiResponseSeries,
TwoPhotonSeries,
)
from roiextractors import (
ImagingExtractor,
SegmentationExtractor,
)
from .imagingextractordatachunkiterator import ImagingExtractorDataChunkIterator
from .roiextractors_pending_deprecation import (
_add_devices_to_nwbfile_old_list_format,
_add_photon_series_to_nwbfile_old_list_format,
_add_segmentation_to_nwbfile_old_list_format,
get_nwb_segmentation_metadata,
)
from ..hdmf import SliceableDataChunkIterator
from ..nwb_helpers import (
BACKEND_NWB_IO,
HDF5BackendConfiguration,
ZarrBackendConfiguration,
configure_backend,
get_default_backend_configuration,
get_default_nwbfile_metadata,
get_module,
make_nwbfile_from_metadata,
)
from ..nwb_helpers._device_types import _build_inline_containers
from ..nwb_helpers._metadata_and_file_helpers import (
_add_device_to_nwbfile,
_fetch_backend_from_nwbfile_on_disk,
configure_and_write_nwbfile,
)
from ...utils import (
calculate_regular_series_rate,
dict_deep_update,
)
from ...utils._metadata_translation import (
_ophys_block_is_old,
_translate_old_metadata,
)
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 plural dict-valued keys under 'Ophys'
('ImagingPlanes', 'MicroscopySeries', 'PlaneSegmentations', 'RoiResponses').
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", {})
dict_based_keys = {"ImagingPlanes", "MicroscopySeries", "PlaneSegmentations", "RoiResponses"}
if dict_based_keys & ophys.keys():
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": "PlaceholderMicroscope",
},
}
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,
},
},
"PlaneSegmentations": {
default_metadata_key: {
"name": "PlaneSegmentation",
"description": "Segmented ROIs",
"imaging_plane_metadata_key": default_metadata_key,
},
},
"RoiResponses": {
default_metadata_key: {
"raw": {
"name": "RoiResponseSeries",
"unit": "n.a.",
},
"deconvolved": {
"name": "Deconvolved",
"unit": "n.a.",
},
"neuropil": {
"name": "Neuropil",
"unit": "n.a.",
},
"denoised": {
"name": "Denoised",
"unit": "n.a.",
},
"baseline": {
"name": "Baseline",
"unit": "n.a.",
},
"dff": {
"name": "DfOverF",
"unit": "n.a.",
},
},
},
"SegmentationImages": {
default_metadata_key: {
"correlation": {
"name": "correlation_image",
},
"mean": {
"name": "mean_image",
},
},
},
}
return metadata
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()
# These are required by the NWB ``ImagingPlane`` object but an interface often knows only some of
# them (a name and a device link, say). Fill any missing one from the central placeholder template
# at write time rather than forcing every ``get_metadata`` to emit a placeholder it cannot justify.
# The templates are explicit unknown-markers (``np.nan``, ``"unknown"``), so defaulting them states
# that the source did not say, not a fabricated value. Mirrors ``_add_electrode_groups_to_nwbfile``.
required_fields = ["name", "excitation_lambda", "indicator", "location", "optical_channel"]
default_imaging_plane = _get_ophys_metadata_placeholders()["Ophys"]["ImagingPlanes"]["default_metadata_key"]
for field in required_fields:
imaging_plane_kwargs.setdefault(field, default_imaging_plane[field])
# The same rule one level down: an interface that knows a channel's name but not its emission
# wavelength states the name alone, and the entry is completed here rather than rejected by
# ``OpticalChannel``. Only a lone channel takes the default name, since two channels defaulted to the
# same name would collide inside the imaging plane.
default_optical_channel = default_imaging_plane["optical_channel"][0]
optical_channels = imaging_plane_kwargs["optical_channel"]
if len(optical_channels) > 1:
default_optical_channel = {field: value for field, value in default_optical_channel.items() if field != "name"}
imaging_plane_kwargs["optical_channel"] = [
{**default_optical_channel, **optical_channel} for optical_channel in optical_channels
]
# 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. An entry naming no device gets the placeholder: a plain Device carrying the one
# field NWB requires, built here rather than through the registry writer, since there is no registry
# entry to key it against. Reused by name, so several planes naming no device land on one device. A
# keyed entry resolves against the caller's ``metadata``, which goes down whole because the device
# may name its model with ``device_model_metadata_key``, resolved against ``metadata["DeviceModels"]``.
device_metadata_key = imaging_plane_kwargs.pop("device_metadata_key", None)
if device_metadata_key is None:
placeholder = _get_ophys_metadata_placeholders()["Devices"]["default_metadata_key"]
placeholder_name = placeholder["name"]
if placeholder_name not in nwbfile.devices:
nwbfile.create_device(**placeholder)
imaging_plane_kwargs["device"] = nwbfile.devices[placeholder_name]
else:
imaging_plane_kwargs["device"] = _add_device_to_nwbfile(
nwbfile=nwbfile, metadata=metadata, metadata_key=device_metadata_key
)
# ``optical_channel`` is written inline as a list of dicts and built into OpticalChannel objects
# by the shared primitive, which reads the target type off ImagingPlane's own constructor spec.
imaging_plane_kwargs = _build_inline_containers(target_class=ImagingPlane, kwargs=imaging_plane_kwargs)
imaging_plane = ImagingPlane(**imaging_plane_kwargs)
nwbfile.add_imaging_plane(imaging_plane)
return imaging_plane
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()
# Required by the NWB photon-series object; default any the interface did not supply from the central
# placeholder template rather than raising. See ``_add_imaging_plane_to_nwbfile``. The default name is
# the generic ``MicroscopySeries``: an interface that knows what it is writing states its own name,
# as ``BaseImagingExtractorInterface`` does.
required_fields = ["name", "unit"]
default_series = _get_ophys_metadata_placeholders()["Ophys"]["MicroscopySeries"]["default_metadata_key"]
for field in required_fields:
photon_series_kwargs.setdefault(field, default_series[field])
# 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 _add_plane_segmentation_to_nwbfile(
*,
segmentation_extractor: SegmentationExtractor,
nwbfile: NWBFile,
metadata: dict,
metadata_key: str,
) -> NWBFile:
"""
Add a PlaneSegmentation to an NWBFile using dict-based metadata.
Masks are written in the extractor's native format (image, pixel, or voxel) as
determined by ``_roi_masks.mask_tpe``. All extractor properties (including
acceptance/rejection status and quality metrics) are written as columns on the
PlaneSegmentation table via the roiextractors property system.
Parameters
----------
segmentation_extractor : SegmentationExtractor
The segmentation extractor to get the results from.
nwbfile : NWBFile
The NWB file to add the plane segmentation to.
metadata : dict
The full metadata dictionary with dict-based format.
metadata_key : str
The key in ``metadata["Ophys"]["PlaneSegmentations"]`` identifying the segmentation.
Returns
-------
NWBFile
The NWBFile with the added PlaneSegmentation.
"""
plane_seg_metadata = metadata["Ophys"]["PlaneSegmentations"][metadata_key].copy()
# Required by the NWB ``PlaneSegmentation`` object; default any the interface did not supply from the
# central placeholder template. ``name`` is defaulted where the segmentation is built, which is also
# where reusing a defaulted name is rejected rather than silently collapsing two segmentations.
# See ``_add_imaging_plane_to_nwbfile``.
required_fields = ["description"]
default_plane_seg = _get_ophys_metadata_placeholders()["Ophys"]["PlaneSegmentations"]["default_metadata_key"]
for field in required_fields:
plane_seg_metadata.setdefault(field, default_plane_seg[field])
# Resolve imaging plane
imaging_plane_metadata_key = plane_seg_metadata.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,
)
# Get or create ImageSegmentation container
ophys_module = get_module(nwbfile, "ophys", description="contains optical physiology processed data")
image_segmentation_name = "ImageSegmentation"
if image_segmentation_name in ophys_module.data_interfaces:
image_segmentation = ophys_module[image_segmentation_name]
else:
image_segmentation = ImageSegmentation(name=image_segmentation_name)
ophys_module.add(image_segmentation)
# The name defaults to the neurodata type being written, as the photon series does. Reuse by name is
# how two interfaces deliberately share one segmentation, so it stays allowed for a name the caller
# stated; a name that was defaulted cannot express that intent, and reusing it would silently drop the
# second interface's ROIs, so it is an error instead.
name_was_defaulted = "name" not in plane_seg_metadata
plane_segmentation_name = plane_seg_metadata.setdefault("name", "PlaneSegmentation")
if plane_segmentation_name in image_segmentation.plane_segmentations:
if name_was_defaulted:
raise ValueError(
f"A PlaneSegmentation named '{plane_segmentation_name}' is already in the file, and "
f"metadata['Ophys']['PlaneSegmentations']['{metadata_key}'] does not name its own. Give it "
"a 'name' to write a second segmentation, or use 1 metadata key to share one."
)
# If PlaneSegmentation already exists, return early
return nwbfile
# Extract ROI data
roi_ids = segmentation_extractor.get_roi_ids()
# Detect native mask format from the extractor
# TODO: open a discussion on roiextractors to expose mask_tpe as a public API
native_mask_type = segmentation_extractor._roi_masks.mask_tpe # e.g. "nwb-image_mask"
mask_type = native_mask_type.replace("nwb-", "").replace("_mask", "") # "image", "pixel", or "voxel"
if mask_type == "image":
image_or_pixel_masks = segmentation_extractor.get_roi_image_masks()
else:
image_or_pixel_masks = segmentation_extractor.get_roi_pixel_masks()
# Build PlaneSegmentation object
plane_seg_metadata["imaging_plane"] = imaging_plane
plane_segmentation = PlaneSegmentation(**plane_seg_metadata)
# Add ROIs
roi_names = [str(roi_id) for roi_id in roi_ids]
roi_indices = list(range(len(roi_ids)))
plane_segmentation.add_column(name="roi_name", description="The unique identifier for each ROI.")
if mask_type == "image":
image_mask_array = image_or_pixel_masks.T
for roi_index, roi_name in zip(roi_indices, roi_names):
image_mask = image_mask_array[roi_index]
plane_segmentation.add_roi(id=roi_index, roi_name=roi_name, image_mask=image_mask)
else:
mask_type_kwarg = f"{mask_type}_mask"
pixel_masks = image_or_pixel_masks
for roi_index, roi_name in zip(roi_indices, roi_names):
pixel_mask = pixel_masks[roi_index]
pixel_mask_to_write = [tuple(x) for x in pixel_mask]
plane_segmentation.add_roi(id=roi_index, roi_name=roi_name, **{mask_type_kwarg: pixel_mask_to_write})
# Add all extractor properties as columns (acceptance, quality metrics, etc.). The quality metrics
# below are named the same way by every segmenter that reports them, so their descriptions are known
# here rather than left empty; this is the same set the old list-based path describes.
known_property_descriptions = {
"snr": "Signal-to-noise ratio for each component",
"r_values": "Spatial correlation values for each component",
"cnn_preds": "CNN classifier predictions for component quality",
}
available_properties = segmentation_extractor.get_property_keys()
for property_key in available_properties:
values = segmentation_extractor.get_property(key=property_key, ids=roi_ids)
description = known_property_descriptions.get(property_key, "")
plane_segmentation.add_column(name=property_key, description=description, data=values)
image_segmentation.add_plane_segmentation(plane_segmentations=[plane_segmentation])
return nwbfile
def _trace_is_all_zero(trace, iterator_options: dict) -> bool:
"""Check whether a trace holds nothing but zeros without materializing all of it.
The trace is read through the same ``SliceableDataChunkIterator`` the writer uses, so a buffer is the
unit that reaches memory here and the peak is never worse than the peak of writing the trace. The scan
returns on the first buffer that holds anything, so a trace with data costs one buffer instead of a
full read.
"""
for buffer in SliceableDataChunkIterator(trace, **iterator_options):
if np.any(buffer.data):
return False
return True
def _add_roi_response_traces_to_nwbfile(
*,
segmentation_extractor: SegmentationExtractor,
nwbfile: NWBFile,
metadata: dict,
metadata_key: str,
iterator_options: dict | None = None,
) -> NWBFile:
"""
Add ROI response traces to an NWBFile using dict-based metadata.
Adds all traces as ``RoiResponseSeries`` inside a single ``Fluorescence`` container,
without splitting into ``Fluorescence`` and ``DfOverF``. This follows the direction
of nwb-schema#616 and ndx-microscopy's single-container pattern
(``MicroscopyResponseSeriesContainer``).
The same ``metadata_key`` is used to look up both the ``RoiResponses`` entry and the
``PlaneSegmentations`` entry, coupling the two implicitly.
If the caller wrote no ``RoiResponses`` entry for ``metadata_key``, placeholder metadata is used
for all available traces, and an extractor with no trace data writes nothing. The entry the
placeholder template carries under ``default_metadata_key`` counts as not written. If the caller
did write one but the extractor has no trace data, a ``ValueError`` is raised.
Parameters
----------
segmentation_extractor : SegmentationExtractor
The segmentation extractor containing trace data.
nwbfile : NWBFile
The NWB file to add traces to.
metadata : dict
The full metadata dictionary with dict-based format. Not modified by this function.
metadata_key : str
The key used to look up both ``metadata["Ophys"]["RoiResponses"]`` and
``metadata["Ophys"]["PlaneSegmentations"]``.
iterator_options : dict, optional
Options for the data chunk iterator.
Returns
-------
NWBFile
The NWBFile with the added traces.
"""
iterator_options = iterator_options or dict()
# Get traces from extractor, filter None/empty
traces_dict = segmentation_extractor.get_traces_dict()
traces_to_add = {
trace_name: trace
for trace_name, trace in traces_dict.items()
if trace is not None and math.prod(trace.shape) != 0
}
roi_responses = metadata.get("Ophys", {}).get("RoiResponses", {})
user_provided_roi_responses = metadata_key in roi_responses
# The placeholder template names a RoiResponses entry under the default key, so an entry there says
# nothing about what the caller asked for.
user_provided_roi_responses_metadata = user_provided_roi_responses and metadata_key != "default_metadata_key"
if user_provided_roi_responses_metadata and not traces_to_add:
raise ValueError("RoiResponses metadata was provided but the segmentation extractor has no trace data.")
if not traces_to_add:
return nwbfile
# Use user-provided metadata or fall back to placeholders
if user_provided_roi_responses:
roi_responses_metadata = roi_responses[metadata_key].copy()
if user_provided_roi_responses_metadata:
requested_traces = set(roi_responses_metadata.keys())
available_traces = set(traces_to_add.keys())
missing_traces = requested_traces - available_traces
if missing_traces:
warnings.warn(
f"RoiResponses metadata specifies traces {missing_traces} "
f"but the segmentation extractor has no data for them. "
f"These traces will be skipped."
)
else:
roi_responses_metadata = _get_ophys_metadata_placeholders()["Ophys"]["RoiResponses"]["default_metadata_key"]
# An all-zero trace is a valid output of a segmentation pipeline -suite2p writes one for `spks` when
# nothing was deconvolved- but it carries no information, so it is not written. A trace the caller named
# is written whatever it holds, as discarding something stated by hand is worse than an empty series.
caller_named_traces = set(roi_responses_metadata) if user_provided_roi_responses_metadata else set()
all_zero_traces = [
trace_name
for trace_name, trace in traces_to_add.items()
if trace_name not in caller_named_traces and _trace_is_all_zero(trace=trace, iterator_options=iterator_options)
]
if all_zero_traces:
warnings.warn(
f"These traces hold only zeros and are not written: {sorted(all_zero_traces)}. "
f"Name them in metadata['Ophys']['RoiResponses'] to write them anyway.",
UserWarning,
stacklevel=2,
)
traces_to_add = {
trace_name: trace for trace_name, trace in traces_to_add.items() if trace_name not in all_zero_traces
}
# Nothing left to write, so the Fluorescence container is not created either.
if not traces_to_add:
return nwbfile
# Resolve PlaneSegmentation via the same metadata_key, defaulting its name the same way the
# segmentation writer does.
plane_segmentation_name = metadata["Ophys"]["PlaneSegmentations"][metadata_key].get("name", "PlaneSegmentation")
ophys_module = get_module(nwbfile, "ophys", description="contains optical physiology processed data")
image_segmentation = ophys_module["ImageSegmentation"]
plane_segmentation = image_segmentation.plane_segmentations[plane_segmentation_name]
# Create ROI table region
roi_ids = segmentation_extractor.get_roi_ids()
available_roi_names = list(plane_segmentation["roi_name"][:])
roi_names = [str(roi_id) for roi_id in roi_ids]
region = [available_roi_names.index(roi_name) for roi_name in roi_names]
imaging_plane_name = plane_segmentation.imaging_plane.name
roi_table_region = plane_segmentation.create_roi_table_region(
region=region,
description=f"The ROIs for {imaging_plane_name}.",
)
# Resolve timestamps
timestamps_were_set = segmentation_extractor.has_time_vector()
if timestamps_were_set:
timestamps = segmentation_extractor.get_timestamps()
else:
timestamps = segmentation_extractor.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(segmentation_extractor.get_sampling_frequency())
timestamps_are_regular = True
starting_time = 0.0
# All traces go into a single Fluorescence container, matching the pattern from
# ndx-microscopy (single MicroscopyResponseSeriesContainer) and avoiding the
# Fluorescence/DfOverF split that nwb-schema#616 proposes to remove.
fluorescence_name = "Fluorescence"
if fluorescence_name in ophys_module.data_interfaces:
fluorescence = ophys_module[fluorescence_name]
else:
fluorescence = Fluorescence(name=fluorescence_name)
ophys_module.add(fluorescence)
for trace_name, trace_data in traces_to_add.items():
# Skip traces not in metadata
if trace_name not in roi_responses_metadata:
continue
# Copy before defaulting; this entry is a live reference into the caller's metadata dict.
trace_metadata = roi_responses_metadata[trace_name].copy()
# Required by the NWB ``RoiResponseSeries`` object; default any the interface did not supply from
# the central placeholder template, falling back to an arbitrary trace template for a trace type
# the template does not name. ``name`` is deliberately not defaulted: series are reused by name
# below, so two unnamed traces would silently collapse into one.
# See ``_add_imaging_plane_to_nwbfile``.
required_fields = ["unit"]
default_roi_responses = _get_ophys_metadata_placeholders()["Ophys"]["RoiResponses"]["default_metadata_key"]
default_trace = default_roi_responses.get(
trace_name, next(v for v in default_roi_responses.values() if isinstance(v, dict))
)
for field in required_fields:
trace_metadata.setdefault(field, default_trace[field])
# Skip if series already exists
series_name = trace_metadata["name"]
if series_name in fluorescence.roi_response_series:
continue
roi_response_series_kwargs = trace_metadata.copy()
roi_response_series_kwargs["data"] = SliceableDataChunkIterator(trace_data, **iterator_options)
roi_response_series_kwargs["rois"] = roi_table_region
if timestamps_are_regular:
roi_response_series_kwargs["starting_time"] = starting_time
roi_response_series_kwargs["rate"] = rate
else:
roi_response_series_kwargs["timestamps"] = timestamps
roi_response_series = RoiResponseSeries(**roi_response_series_kwargs)
fluorescence.add_roi_response_series(roi_response_series)
return nwbfile
def _add_summary_images_to_nwbfile(
*,
segmentation_extractor: SegmentationExtractor,
nwbfile: NWBFile,
metadata: dict,
metadata_key: str,
) -> NWBFile:
"""
Add summary images (e.g. mean and correlation) to an NWBFile.
Images are added to a single ``Images`` container named ``"SegmentationImages"``
in the ``ophys`` processing module. If the extractor has no images, this is a no-op.
Parameters
----------
segmentation_extractor : SegmentationExtractor
The segmentation extractor containing image data.
nwbfile : NWBFile
The NWB file to add images to.
metadata : dict
The full metadata dictionary. Image metadata is looked up under
``metadata["Ophys"]["SegmentationImages"][metadata_key]``.
metadata_key : str
The key identifying which segmentation's image metadata to use.
Returns
-------
NWBFile
The NWBFile with the added summary images.
"""
images_dict = segmentation_extractor.get_images_dict()
images_to_add = {img_name: img for img_name, img in images_dict.items() if img is not None}
if not images_to_add:
return nwbfile
# Look up per-image metadata for this segmentation
segmentation_images_metadata = metadata.get("Ophys", {}).get("SegmentationImages", {})
user_provided_images_metadata = (
metadata_key in segmentation_images_metadata and metadata_key != "default_metadata_key"
)
if user_provided_images_metadata:
images_metadata = segmentation_images_metadata[metadata_key]
requested_images = set(images_metadata.keys())
available_images = set(images_to_add.keys())
missing_images = requested_images - available_images
if missing_images:
warnings.warn(
f"SegmentationImages metadata specifies images {missing_images} "
f"but the segmentation extractor has no data for them. "
f"These images will be skipped."
)
else:
# The caller stated nothing for this key, so every image the extractor holds is written. The
# placeholders only name ``correlation`` and ``mean``, so letting them decide instead would
# drop any other summary image the source produced.
placeholders = _get_ophys_metadata_placeholders()["Ophys"]["SegmentationImages"]["default_metadata_key"]
supplied = segmentation_images_metadata.get(metadata_key, dict())
images_metadata = {
img_type: supplied.get(img_type, placeholders.get(img_type, dict())) for img_type in images_to_add
}
# ``images_metadata`` decides what is written, so an entry naming none of the images the extractor
# holds writes nothing. Resolve that before the container is built: ``Images`` requires at least one
# image, and an empty one makes the file invalid.
images_to_write = {
img_type: img_data for img_type, img_data in images_to_add.items() if img_type in images_metadata
}
if not images_to_write:
return nwbfile
# Get or create the single shared Images container
container_name = "SegmentationImages"
container_description = "Summary images for segmentation."
ophys_module = get_module(nwbfile=nwbfile, name="ophys", description="contains optical physiology processed data")
if container_name not in ophys_module.data_interfaces:
ophys_module.add(Images(name=container_name, description=container_description))
image_collection = ophys_module.data_interfaces[container_name]
for img_type, img_data in images_to_write.items():
image_metadata = images_metadata[img_type]
image_name = image_metadata.get("name", img_type)
image_description = image_metadata.get("description", f"Summary image: {img_type}.")
# Skip if an image with this name already exists in the container
if image_name in image_collection.images:
continue
# NWB uses width x height (columns, rows); roiextractors uses height x width (rows, columns)
image_collection.add_image(GrayscaleImage(name=image_name, data=img_data.T, description=image_description))
return nwbfile
def _add_segmentation_to_nwbfile(
*,
segmentation_extractor: SegmentationExtractor,
nwbfile: NWBFile,
metadata: dict,
metadata_key: str,
iterator_options: dict | None = None,
) -> NWBFile:
"""
Add segmentation data to an NWBFile using dict-based metadata.
Orchestrates adding the PlaneSegmentation and ROI response traces.
Parameters
----------
segmentation_extractor : SegmentationExtractor
The segmentation extractor containing all segmentation data.
nwbfile : NWBFile
The NWB file to add segmentation data to.
metadata : dict
The full metadata dictionary with dict-based format.
metadata_key : str
The key used across ``PlaneSegmentations`` and ``RoiResponses``.
iterator_options : dict, optional
Options for the data chunk iterator.
Returns
-------
NWBFile
The NWBFile with the added segmentation data.
"""
_add_plane_segmentation_to_nwbfile(
segmentation_extractor=segmentation_extractor,
nwbfile=nwbfile,
metadata=metadata,
metadata_key=metadata_key,
)
_add_roi_response_traces_to_nwbfile(
segmentation_extractor=segmentation_extractor,
nwbfile=nwbfile,
metadata=metadata,
metadata_key=metadata_key,
iterator_options=iterator_options,
)
_add_summary_images_to_nwbfile(
segmentation_extractor=segmentation_extractor,
nwbfile=nwbfile,
metadata=metadata,
metadata_key=metadata_key,
)
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.
Parameters
----------
imaging : ImagingExtractor
An imaging extractor object from roiextractors.
Raises
------
MemoryError
"""
element_size_in_bytes = imaging.get_dtype().itemsize
sample_shape = imaging.get_sample_shape()
num_samples = imaging.get_num_samples()
traces_size_in_bytes = num_samples * math.prod(sample_shape) * element_size_in_bytes
available_memory_in_bytes = psutil.virtual_memory().available
if traces_size_in_bytes > available_memory_in_bytes:
message = (
f"Memory error, full TwoPhotonSeries data is {human_readable_size(traces_size_in_bytes, binary=True)} but "
f"only {human_readable_size(available_memory_in_bytes, binary=True)} are available! "
"Please use iterator_type='v2'."
)
raise MemoryError(message)
def _imaging_frames_to_hdmf_iterator(
imaging: ImagingExtractor,
iterator_type: str | None = "v2",
iterator_options: dict | None = None,
):
"""
Private auxiliary method to wrap frames from an ImagingExtractor into a DataChunkIterator.
Parameters
----------
imaging : ImagingExtractor
The imaging extractor to get the data from.
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. 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.
Returns
-------
iterator
The frames of the imaging extractor wrapped in an iterator for chunked writing.
"""
assert iterator_type in ["v2", None], "'iterator_type' must be either 'v2' (recommended) or None."
iterator_options = dict() if iterator_options is None else iterator_options
if iterator_type is None:
_check_if_imaging_fits_into_memory(imaging=imaging)
series = imaging.get_series()
# (samples, height, width) planar, (samples, height, width, planes) volumetric. The same rule
# the v2 iterator applies in ImagingExtractorDataChunkIterator._get_data.
transpose_axes = (0, 2, 1) if series.ndim == 3 else (0, 2, 1, 3)
return series.transpose(transpose_axes)
return ImagingExtractorDataChunkIterator(imaging_extractor=imaging, **iterator_options)
[docs]
def add_imaging_to_nwbfile(
imaging: ImagingExtractor,
nwbfile: NWBFile,
metadata: dict | None = None,
*args, # TODO: change to * (keyword only) on or after February 2027
photon_series_type: Literal["TwoPhotonSeries", "OnePhotonSeries"] = "TwoPhotonSeries",
photon_series_index: int = 0,
iterator_type: str | None = "v2",
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
The extractor object containing the imaging data.
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.
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
Additional options for controlling the iteration process, by default None.
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".
always_write_timestamps : bool, default: False
Set to True to always write timestamps.
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
-------
NWBFile
The NWB file with the imaging data added
"""
# TODO: Remove this block in September 2026 or after when positional arguments are no longer supported.
if args:
parameter_names = [
"photon_series_type",
"photon_series_index",
"iterator_type",
"iterator_options",
"parent_container",
"always_write_timestamps",
]
num_positional_args_before_args = 3 # imaging, nwbfile, metadata
if len(args) > len(parameter_names):
raise TypeError(
f"add_imaging_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 September 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_imaging_to_nwbfile is deprecated "
f"and will be removed in September 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)
iterator_type = positional_values.get("iterator_type", iterator_type)
iterator_options = positional_values.get("iterator_options", iterator_options)
parent_container = positional_values.get("parent_container", parent_container)
always_write_timestamps = positional_values.get("always_write_timestamps", always_write_timestamps)
if metadata is None:
metadata = _get_ophys_metadata_placeholders()
# Old-shaped metadata is converted here, the last public function before the private writers, so
# everything below sees one format. The old format addresses a series by position in
# ``Ophys[photon_series_type]`` and the new one by key, so the entry this call writes is named here
# and handed to the translator; when the caller gave no key, the series' own name becomes it.
ophys_metadata = metadata.get("Ophys", {})
if isinstance(ophys_metadata, dict) and _ophys_block_is_old(ophys_metadata):
series_list = ophys_metadata.get(photon_series_type, [])
if metadata_key is None and photon_series_index < len(series_list):
metadata_key = series_list[photon_series_index].get("name")
metadata = _translate_old_metadata(
metadata,
metadata_key=metadata_key,
photon_series_type=photon_series_type,
photon_series_index=photon_series_index,
)
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
[docs]
def write_imaging_to_nwbfile(
imaging: ImagingExtractor,
nwbfile_path: FilePath | None = None,
nwbfile: NWBFile | None = None,
metadata: dict | None = None,
overwrite: bool = False,
verbose: bool = False,
photon_series_type: Literal["TwoPhotonSeries", "OnePhotonSeries"] = "TwoPhotonSeries",
*,
iterator_type: str | None = "v2",
iterator_options: dict | None = None,
backend: Literal["hdf5", "zarr"] | None = None,
backend_configuration: HDF5BackendConfiguration | ZarrBackendConfiguration | None = None,
append_on_disk_nwbfile: bool = False,
) -> NWBFile | None:
"""
Primary method for writing an ImagingExtractor object to an NWBFile.
Parameters
----------
imaging : ImagingExtractor
The imaging extractor object to be written to nwb.
nwbfile_path : FilePath, optional
Path for where to write the NWBFile.
If not provided, only adds data to the in-memory nwbfile without writing to disk.
**Deprecated**: Using this function without nwbfile_path is deprecated.
Use ``add_imaging_to_nwbfile`` instead.
nwbfile : NWBFile, optional
If passed, this function will fill the relevant fields within the NWBFile object.
E.g., calling::
write_imaging_to_nwbfile(imaging=my_imaging_extractor, nwbfile=my_nwbfile)
will result in the appropriate changes to the my_nwbfile object.
metadata : dict, optional
Metadata dictionary with information used to create the NWBFile when one does not exist or overwrite=True.
overwrite : bool, default: False
Whether to overwrite the NWBFile if one exists at the nwbfile_path.
verbose : bool, default: False
If 'nwbfile_path' is specified, informs user after a successful write operation.
photon_series_type : {"TwoPhotonSeries", "OnePhotonSeries"}, default: "TwoPhotonSeries"
The type of photon series to add.
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. 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.
backend : {"hdf5", "zarr"}, optional
The type of backend to use when writing the file.
If a ``backend_configuration`` is not specified, the default type will be "hdf5".
If a ``backend_configuration`` is specified, then the type will be auto-detected.
backend_configuration : HDF5BackendConfiguration or ZarrBackendConfiguration, optional
The configuration model to use when configuring the datasets for this backend.
append_on_disk_nwbfile : bool, default: False
Whether to append to an existing NWBFile on disk. If True, the ``nwbfile`` parameter must be None.
Returns
-------
NWBFile or None
The NWBFile object when writing a new file or using an in-memory nwbfile.
Returns None when appending to an existing file on disk (append_on_disk_nwbfile=True).
**Deprecated**: Returning NWBFile in append mode is deprecated and will return None on or after June 2026.
"""
# Handle deprecated usage without nwbfile_path
if nwbfile_path is None:
warnings.warn(
"Using 'write_imaging_to_nwbfile' without 'nwbfile_path' to only add data to an in-memory nwbfile "
"is deprecated and will be removed on or after June 2026. Use 'add_imaging_to_nwbfile' instead.",
DeprecationWarning,
stacklevel=2,
)
if nwbfile is None:
raise ValueError(
"Either 'nwbfile_path' or 'nwbfile' must be provided. "
"To add data to an in-memory nwbfile, use 'add_imaging_to_nwbfile' instead."
)
add_imaging_to_nwbfile(
imaging=imaging,
nwbfile=nwbfile,
metadata=metadata,
photon_series_type=photon_series_type,
iterator_type=iterator_type,
iterator_options=iterator_options,
)
return nwbfile
iterator_options = iterator_options or dict()
if metadata is None:
metadata = dict()
if hasattr(imaging, "nwb_metadata"):
metadata = dict_deep_update(imaging.nwb_metadata, metadata, append_list=False)
appending_to_in_memory_nwbfile = nwbfile is not None
file_initially_exists = nwbfile_path.exists()
allowed_to_modify_existing = overwrite or append_on_disk_nwbfile
if file_initially_exists and not allowed_to_modify_existing:
raise FileExistsError(
f"The file at '{nwbfile_path}' already exists. Set overwrite=True to overwrite the existing file "
"or append_on_disk_nwbfile=True to append to the existing file."
)
if append_on_disk_nwbfile and appending_to_in_memory_nwbfile:
raise ValueError(
"Cannot append to an existing file on disk while also providing an in-memory NWBFile. "
"Either set append_on_disk_nwbfile=False to write the in-memory NWBFile to disk, "
"or remove the nwbfile parameter to append to the existing file on disk."
)
# An append is bound to the backend of the file on disk; a new file gets its backend from the caller
if append_on_disk_nwbfile:
backend = _fetch_backend_from_nwbfile_on_disk(
nwbfile_path=nwbfile_path, backend=backend, backend_configuration=backend_configuration
)
# Determine if we're writing a new file or appending
writing_new_file = not append_on_disk_nwbfile
if writing_new_file:
# Writing mode: create or use provided nwbfile and write
if nwbfile is None:
nwbfile = make_nwbfile_from_metadata(metadata=metadata)
add_imaging_to_nwbfile(
imaging=imaging,
nwbfile=nwbfile,
metadata=metadata,
photon_series_type=photon_series_type,
iterator_type=iterator_type,
iterator_options=iterator_options,
)
configure_and_write_nwbfile(
nwbfile=nwbfile,
nwbfile_path=nwbfile_path,
backend=backend,
backend_configuration=backend_configuration,
)
if verbose:
print(f"NWB file saved at {nwbfile_path}!")
return nwbfile
else:
# Append mode: read existing file, add data, write back
warnings.warn(
"Returning an NWBFile object when using append_on_disk_nwbfile=True is deprecated "
"and will return None on or after June 2026.",
DeprecationWarning,
stacklevel=2,
)
IO = BACKEND_NWB_IO[backend]
with IO(path=str(nwbfile_path), mode="r+", load_namespaces=True) as io:
nwbfile = io.read()
add_imaging_to_nwbfile(
imaging=imaging,
nwbfile=nwbfile,
metadata=metadata,
photon_series_type=photon_series_type,
iterator_type=iterator_type,
iterator_options=iterator_options,
)
if backend_configuration is None:
backend_configuration = get_default_backend_configuration(nwbfile=nwbfile, backend=backend)
configure_backend(nwbfile=nwbfile, backend_configuration=backend_configuration)
io.write(nwbfile)
if verbose:
print(f"NWB file saved at {nwbfile_path}!")
return nwbfile
def _segmentation_extractor_has_data(segmentation_extractor: SegmentationExtractor) -> bool:
"""
Whether a segmentation extractor holds anything to write: ROIs, traces or summary images.
Parameters
----------
segmentation_extractor : SegmentationExtractor
The extractor to inspect.
Returns
-------
bool
False only when the extractor reports no ROIs, no traces and no summary images.
"""
if segmentation_extractor.get_num_rois() > 0:
return True
traces = segmentation_extractor.get_traces_dict().values()
if any(trace is not None and math.prod(trace.shape) != 0 for trace in traces):
return True
return any(image is not None for image in segmentation_extractor.get_images_dict().values())
[docs]
def add_segmentation_to_nwbfile(
segmentation_extractor: SegmentationExtractor,
nwbfile: NWBFile,
metadata: dict | None = None,
*args, # TODO: change to * (keyword only) on or after February 2027
plane_segmentation_name: str | None = None,
background_plane_segmentation_name: str | None = None,
include_background_segmentation: bool = False,
include_roi_centroids: bool = True,
include_roi_acceptance: bool | None = None,
mask_type: Literal["image", "pixel", "voxel"] = "image",
iterator_options: dict | None = None,
# TODO: move metadata_key after metadata once positional args removed (September 2026)
metadata_key: str | None = None,
) -> NWBFile:
"""
Add segmentation data from a SegmentationExtractor object to an NWBFile.
Supports both old list-based metadata and new dict-based metadata (via ``metadata_key``).
Parameters
----------
segmentation_extractor : SegmentationExtractor
The extractor object containing segmentation data.
nwbfile : NWBFile
The NWB file where the segmentation data will be added.
metadata : dict, optional
Metadata for the NWBFile, by default None.
plane_segmentation_name : str, optional
The name of the PlaneSegmentation object to be added, by default None.
Used with the old list-based metadata format.
Deprecated: in the dict-based format, use ``metadata_key`` for pattern discovery and edit
``metadata["Ophys"]["PlaneSegmentations"][metadata_key]["name"]`` directly. Will be removed
when the old list-based metadata format is removed.
background_plane_segmentation_name : str, optional
The name of the background PlaneSegmentation, if any, by default None.
Used with the old list-based metadata format.
Deprecated: same guidance as ``plane_segmentation_name``.
include_background_segmentation : bool, optional
If True, includes background plane segmentation, by default False.
Used with the old list-based metadata format.
include_roi_centroids : bool, optional
If True, includes the centroids of the regions of interest (ROIs), by default True.
include_roi_acceptance : bool, optional
Deprecated and ignored. ROI acceptance is now written automatically as a
column on the PlaneSegmentation table whenever the segmentation extractor
exposes acceptance/rejection through its property system.
mask_type : str
Type of mask to use for segmentation; can be either "image" or "pixel", by default "image".
iterator_options : dict, optional
Options for iterating over the data, by default None.
metadata_key : str, optional
The key in ``metadata["Ophys"]["PlaneSegmentations"]`` identifying the segmentation.
When provided, uses the new dict-based metadata format.
Returns
-------
NWBFile
The NWBFile with the added segmentation data.
"""
# TODO: Remove this block in September 2026 or after when positional arguments are no longer supported.
if args:
parameter_names = [
"plane_segmentation_name",
"background_plane_segmentation_name",
"include_background_segmentation",
"include_roi_centroids",
"include_roi_acceptance",
"mask_type",
"iterator_options",
]
num_positional_args_before_args = 3 # segmentation_extractor, nwbfile, metadata
if len(args) > len(parameter_names):
raise TypeError(
f"add_segmentation_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 September 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_segmentation_to_nwbfile is deprecated "
f"and will be removed in September 2026 or after. "
f"The following arguments were passed positionally: {passed_as_positional}. "
"Please use keyword arguments instead.",
FutureWarning,
stacklevel=2,
)
plane_segmentation_name = positional_values.get("plane_segmentation_name", plane_segmentation_name)
background_plane_segmentation_name = positional_values.get(
"background_plane_segmentation_name", background_plane_segmentation_name
)
include_background_segmentation = positional_values.get(
"include_background_segmentation", include_background_segmentation
)
include_roi_centroids = positional_values.get("include_roi_centroids", include_roi_centroids)
include_roi_acceptance = positional_values.get("include_roi_acceptance", include_roi_acceptance)
mask_type = positional_values.get("mask_type", mask_type)
iterator_options = positional_values.get("iterator_options", iterator_options)
if include_roi_acceptance is not None:
warnings.warn(
"`include_roi_acceptance` is deprecated and has no effect. ROI acceptance is now "
"written automatically as a column on the PlaneSegmentation table whenever the "
"segmentation extractor exposes acceptance/rejection through its property system. "
"This parameter will be removed on or after February 2027.",
DeprecationWarning,
stacklevel=2,
)
# Without this the writer either fails inside the mask handling with an AttributeError naming a
# private attribute or, where masks are absent, writes an imaging plane and an empty table that read
# as a successful conversion. See https://github.com/catalystneuro/neuroconv/issues/1401.
if not _segmentation_extractor_has_data(segmentation_extractor=segmentation_extractor):
raise ValueError(
f"{type(segmentation_extractor).__name__} contains no segmentation data: it reports 0 ROIs, "
"no traces and no summary images. This is usually an empty result from the segmentation "
"pipeline, or a file whose data is not where the format expected it. Writing it would produce "
"an NWB file holding an imaging plane, a device and an empty ROI table and nothing else, which "
"is indistinguishable from a successful conversion."
)
if metadata is None:
metadata = _get_ophys_metadata_placeholders()
# As in ``add_imaging_to_nwbfile``: the old format addresses a plane segmentation by name and the new
# one by key, so the entry this call writes is named here. Traces and summary images are keyed by the
# same handle, which is why the translator needs to know which name it is.
ophys_metadata = metadata.get("Ophys", {})
if isinstance(ophys_metadata, dict) and _ophys_block_is_old(ophys_metadata):
segmentation_list = ophys_metadata.get("ImageSegmentation", {}).get("plane_segmentations", [])
addressed_name = plane_segmentation_name
if addressed_name is None and segmentation_list:
addressed_name = segmentation_list[0].get("name")
if metadata_key is None:
metadata_key = addressed_name
metadata = _translate_old_metadata(metadata, metadata_key=metadata_key, plane_segmentation_name=addressed_name)
# The old format's defaults declare six trace roles on every segmentation, so a translated block
# asks for traces from extractors that expose none, `InscopixSegmentationInterface` among them.
# That is boilerplate rather than a request, and the old writer answered it by writing nothing, so
# the block goes here rather than letting the writer reject metadata the caller never wrote.
traces = segmentation_extractor.get_traces_dict().values()
if not any(trace is not None and math.prod(trace.shape) != 0 for trace in traces):
metadata["Ophys"] = {key: value for key, value in metadata["Ophys"].items() if key != "RoiResponses"}
if _is_dict_based_metadata(metadata):
metadata_key = metadata_key or "default_metadata_key"
nwbfile = _add_segmentation_to_nwbfile(
segmentation_extractor=segmentation_extractor,
nwbfile=nwbfile,
metadata=metadata,
metadata_key=metadata_key,
iterator_options=iterator_options,
)
else:
nwbfile = _add_segmentation_to_nwbfile_old_list_format(
segmentation_extractor=segmentation_extractor,
nwbfile=nwbfile,
metadata=metadata,
plane_segmentation_name=plane_segmentation_name,
background_plane_segmentation_name=background_plane_segmentation_name,
include_background_segmentation=include_background_segmentation,
include_roi_centroids=include_roi_centroids,
mask_type=mask_type,
iterator_options=iterator_options,
)
return nwbfile
[docs]
def write_segmentation_to_nwbfile(
segmentation_extractor: SegmentationExtractor,
nwbfile_path: FilePath | None = None,
nwbfile: NWBFile | None = None,
metadata: dict | None = None,
overwrite: bool = False,
verbose: bool = False,
include_background_segmentation: bool = False,
include_roi_centroids: bool = True,
include_roi_acceptance: bool | None = None,
mask_type: Literal["image", "pixel", "voxel"] = "image",
*,
iterator_options: dict | None = None,
backend: Literal["hdf5", "zarr"] | None = None,
backend_configuration: HDF5BackendConfiguration | ZarrBackendConfiguration | None = None,
append_on_disk_nwbfile: bool = False,
) -> NWBFile | None:
"""
Primary method for writing a SegmentationExtractor object to an NWBFile.
Parameters
----------
segmentation_extractor : SegmentationExtractor
The segmentation extractor object to be written to nwb.
nwbfile_path : FilePath, optional
Path for where to write the NWBFile.
If not provided, only adds data to the in-memory nwbfile without writing to disk.
**Deprecated**: Using this function without nwbfile_path is deprecated.
Use ``add_segmentation_to_nwbfile`` instead.
nwbfile : NWBFile, optional
If passed, this function will fill the relevant fields within the NWBFile object.
E.g., calling::
write_segmentation_to_nwbfile(segmentation_extractor=my_segmentation_extractor, nwbfile=my_nwbfile)
will result in the appropriate changes to the my_nwbfile object.
metadata : dict, optional
Metadata dictionary with information used to create the NWBFile when one does not exist or overwrite=True.
overwrite : bool, default: False
Whether to overwrite the NWBFile if one exists at the nwbfile_path.
verbose : bool, default: False
If 'nwbfile_path' is specified, informs user after a successful write operation.
include_background_segmentation : bool, default: False
Whether to include the background plane segmentation and fluorescence traces in the NWB file. If False,
neuropil traces are included in the main plane segmentation rather than the background plane segmentation.
include_roi_centroids : bool, default: True
Whether to include the ROI centroids on the PlaneSegmentation table.
If there are a very large number of ROIs (such as in whole-brain recordings), you may wish to disable this for
faster write speeds.
include_roi_acceptance : bool, optional
Deprecated and ignored. ROI acceptance is now written automatically as a
column on the PlaneSegmentation table whenever the segmentation extractor
exposes acceptance/rejection through its property system.
mask_type : {"image", "pixel", "voxel"}, default: "image"
There are three types of ROI masks in NWB, 'image', 'pixel', and 'voxel'.
* 'image' masks have the same shape as the reference images the segmentation was applied to, and weight each pixel
by its contribution to the ROI (typically boolean, with 0 meaning 'not in the ROI').
* 'pixel' masks are instead indexed by ROI, with the data at each index being the shape of the image by the number
of pixels in each ROI.
* 'voxel' masks are instead indexed by ROI, with the data at each index being the shape of the volume by the number
of voxels in each ROI.
Specify your choice between these two as mask_type='image', 'pixel', 'voxel'
iterator_options : dict, optional
A dictionary with options for the internal iterators that process the data.
backend : {"hdf5", "zarr"}, optional
The type of backend to use when writing the file.
If a ``backend_configuration`` is not specified, the default type will be "hdf5".
If a ``backend_configuration`` is specified, then the type will be auto-detected.
backend_configuration : HDF5BackendConfiguration or ZarrBackendConfiguration, optional
The configuration model to use when configuring the datasets for this backend.
append_on_disk_nwbfile : bool, default: False
Whether to append to an existing NWBFile on disk. If True, the ``nwbfile`` parameter must be None.
Returns
-------
NWBFile or None
The NWBFile object when writing a new file or using an in-memory nwbfile.
Returns None when appending to an existing file on disk (append_on_disk_nwbfile=True).
**Deprecated**: Returning NWBFile in append mode is deprecated and will return None on or after June 2026.
"""
iterator_options = iterator_options or dict()
if include_roi_acceptance is not None:
warnings.warn(
"`include_roi_acceptance` is deprecated and has no effect. ROI acceptance is now "
"written automatically as a column on the PlaneSegmentation table whenever the "
"segmentation extractor exposes acceptance/rejection through its property system. "
"This parameter will be removed on or after February 2027.",
DeprecationWarning,
stacklevel=2,
)
segmentation_extractors = [segmentation_extractor]
if metadata is not None and not isinstance(metadata, list):
metadata = [metadata]
metadata_base_list = [get_nwb_segmentation_metadata(seg_extractor) for seg_extractor in segmentation_extractors]
# Updating base metadata with new:
for num, data in enumerate(metadata_base_list):
metadata_input = metadata[num] if metadata else {}
metadata_base_list[num] = dict_deep_update(metadata_base_list[num], metadata_input, append_list=False)
metadata_base_common = metadata_base_list[0]
# Handle deprecated usage without nwbfile_path
if nwbfile_path is None:
warnings.warn(
"Using 'write_segmentation_to_nwbfile' without 'nwbfile_path' to only add data to an in-memory nwbfile "
"is deprecated and will be removed on or after June 2026. Use 'add_segmentation_to_nwbfile' instead.",
DeprecationWarning,
stacklevel=2,
)
if nwbfile is None:
raise ValueError(
"Either 'nwbfile_path' or 'nwbfile' must be provided. "
"To add data to an in-memory nwbfile, use 'add_segmentation_to_nwbfile' instead."
)
_ = get_module(nwbfile=nwbfile, name="ophys", description="contains optical physiology processed data")
for seg_extractor, seg_metadata in zip(segmentation_extractors, metadata_base_list):
add_segmentation_to_nwbfile(
segmentation_extractor=seg_extractor,
nwbfile=nwbfile,
metadata=seg_metadata,
include_background_segmentation=include_background_segmentation,
include_roi_centroids=include_roi_centroids,
mask_type=mask_type,
iterator_options=iterator_options,
)
return nwbfile
appending_to_in_memory_nwbfile = nwbfile is not None
file_initially_exists = nwbfile_path.exists()
allowed_to_modify_existing = overwrite or append_on_disk_nwbfile
if file_initially_exists and not allowed_to_modify_existing:
raise FileExistsError(
f"The file at '{nwbfile_path}' already exists. Set overwrite=True to overwrite the existing file "
"or append_on_disk_nwbfile=True to append to the existing file."
)
if append_on_disk_nwbfile and appending_to_in_memory_nwbfile:
raise ValueError(
"Cannot append to an existing file on disk while also providing an in-memory NWBFile. "
"Either set append_on_disk_nwbfile=False to write the in-memory NWBFile to disk, "
"or remove the nwbfile parameter to append to the existing file on disk."
)
# An append is bound to the backend of the file on disk; a new file gets its backend from the caller
if append_on_disk_nwbfile:
backend = _fetch_backend_from_nwbfile_on_disk(
nwbfile_path=nwbfile_path, backend=backend, backend_configuration=backend_configuration
)
# Determine if we're writing a new file or appending
writing_new_file = not append_on_disk_nwbfile
if writing_new_file:
# Writing mode: create or use provided nwbfile and write
if nwbfile is None:
nwbfile = make_nwbfile_from_metadata(metadata=metadata_base_common)
_ = get_module(nwbfile=nwbfile, name="ophys", description="contains optical physiology processed data")
for seg_extractor, seg_metadata in zip(segmentation_extractors, metadata_base_list):
add_segmentation_to_nwbfile(
segmentation_extractor=seg_extractor,
nwbfile=nwbfile,
metadata=seg_metadata,
include_background_segmentation=include_background_segmentation,
include_roi_centroids=include_roi_centroids,
mask_type=mask_type,
iterator_options=iterator_options,
)
configure_and_write_nwbfile(
nwbfile=nwbfile,
nwbfile_path=nwbfile_path,
backend=backend,
backend_configuration=backend_configuration,
)
if verbose:
print(f"NWB file saved at {nwbfile_path}!")
return nwbfile
else:
# Append mode: read existing file, add data, write back
warnings.warn(
"Returning an NWBFile object when using append_on_disk_nwbfile=True is deprecated "
"and will return None on or after June 2026.",
DeprecationWarning,
stacklevel=2,
)
IO = BACKEND_NWB_IO[backend]
with IO(path=str(nwbfile_path), mode="r+", load_namespaces=True) as io:
nwbfile = io.read()
_ = get_module(nwbfile=nwbfile, name="ophys", description="contains optical physiology processed data")
for seg_extractor, seg_metadata in zip(segmentation_extractors, metadata_base_list):
add_segmentation_to_nwbfile(
segmentation_extractor=seg_extractor,
nwbfile=nwbfile,
metadata=seg_metadata,
include_background_segmentation=include_background_segmentation,
include_roi_centroids=include_roi_centroids,
mask_type=mask_type,
iterator_options=iterator_options,
)
if backend_configuration is None:
backend_configuration = get_default_backend_configuration(nwbfile=nwbfile, backend=backend)
configure_backend(nwbfile=nwbfile, backend_configuration=backend_configuration)
io.write(nwbfile)
if verbose:
print(f"NWB file saved at {nwbfile_path}!")
return nwbfile