"""Collection of helper functions related to NWB."""
import uuid
import warnings
from contextlib import contextmanager
from copy import deepcopy
from datetime import datetime
from pathlib import Path
from typing import Literal
from pydantic import FilePath
from pynwb import NWBFile, read_nwb
from pynwb.device import Device, DeviceModel
from pynwb.file import Subject
from . import (
BACKEND_NWB_IO,
BackendConfiguration,
configure_backend,
get_default_backend_configuration,
)
from ._device_types import (
_DEVICE_MODEL_TYPE_SOURCES,
_DEVICE_TYPE_SOURCES,
_build_inline_containers,
_resolve_type,
)
from ._provenance import describe_source_script
from ...utils.dict import DeepDict, load_dict_from_file
from ...utils.json_schema import _validate_device_registry_names, validate_metadata
[docs]
def get_module(nwbfile: NWBFile, name: str, description: str = None):
"""
Check if processing module exists. If not, create it. Then return module.
Parameters
----------
nwbfile : NWBFile
The NWB file to check or add the module to.
name : str
The name of the processing module.
description : str, optional
Description of the module. Only used if creating a new module.
Returns
-------
ProcessingModule
The existing or newly created processing module.
"""
if name in nwbfile.processing:
if description is not None and nwbfile.processing[name].description != description:
existing_description = nwbfile.processing[name].description
warnings.warn(
f"Processing module '{name}' already exists with a different description. "
f"The new description will be ignored.\n"
f" Existing: '{existing_description}'\n"
f" Provided: '{description}'\n"
f"To fix this, ensure all calls to get_module() for '{name}' use the same description, "
f"or omit the description parameter to use the existing one."
)
return nwbfile.processing[name]
else:
if description is None:
description = "No description."
return nwbfile.create_processing_module(name=name, description=description)
def _get_container_by_name(nwbfile: NWBFile, name: str, neurodata_type: str):
"""
Return the container of the given neurodata type carrying the given name.
Used where an interface must link to an object another interface already wrote, since metadata
addresses it by name while the link needs the object itself.
Parameters
----------
nwbfile : NWBFile
The NWB file to search, including its processing modules.
name : str
The name of the container to return.
neurodata_type : str
The class name of the container, such as ``"PoseEstimation"`` or ``"ImageSeries"``.
Returns
-------
The container carrying that name.
Raises
------
ValueError
If no container of that type carries that name, naming the ones that do exist.
"""
containers = {obj.name: obj for obj in nwbfile.all_children() if type(obj).__name__ == neurodata_type}
if name in containers:
return containers[name]
if containers:
raise ValueError(
f"No {neurodata_type} named '{name}' was found in the NWB file. "
f"Available {neurodata_type} containers: {list(containers)}."
)
raise ValueError(
f"No {neurodata_type} named '{name}' was found in the NWB file. No {neurodata_type} containers exist "
"in the file, so ensure the interface that writes it runs first."
)
[docs]
def add_subject_to_nwbfile(nwbfile: NWBFile, metadata: dict | None = None) -> None:
"""
Add the subject described by ``metadata["Subject"]`` to an NWBFile.
Parameters
----------
nwbfile : NWBFile
The NWBFile to add the subject to.
metadata : dict, optional
Metadata dictionary. The subject is read from ``metadata["Subject"]``, whose keys are the
arguments of :py:class:`~pynwb.file.Subject`::
metadata["Subject"] = dict(subject_id="M1", species="Mus musculus", sex="M")
A ``date_of_birth`` stated as an ISO 8601 string is converted. Metadata that names no subject
is not an error: a source that did not record one leaves the file without one.
Raises
------
ValueError
If the NWBFile already holds a subject. An NWBFile describes one subject and pynwb refuses to
replace it, so a second one is a conflict for the caller to resolve rather than something to
overwrite silently.
"""
metadata = metadata or dict()
if "Subject" not in metadata:
return
if nwbfile.subject is not None:
raise ValueError(
f"This NWBFile already holds the subject '{nwbfile.subject.subject_id}' and an NWBFile describes "
f"one subject. The metadata states '{metadata['Subject'].get('subject_id')}'. Write the two subjects "
"to separate files, or drop metadata['Subject'] to keep the one already there."
)
# Copied because the ISO 8601 conversion below writes into the entry, and the metadata belongs to
# the caller.
subject_metadata = deepcopy(metadata["Subject"])
date_of_birth = subject_metadata.get("date_of_birth")
if isinstance(date_of_birth, str):
subject_metadata["date_of_birth"] = datetime.fromisoformat(date_of_birth)
nwbfile.subject = Subject(**subject_metadata)
def _add_device_model_to_nwbfile(
nwbfile: NWBFile,
*,
metadata: dict,
metadata_key: str,
):
"""
Add the ``metadata["DeviceModels"][metadata_key]`` device model to an NWBFile's ``device_models``.
Follows the canonical ``(nwbfile, metadata, metadata_key)`` pattern used across NeuroConv: the key
is resolved against the full metadata. Idempotent on ``name`` (a model already present is returned
unchanged). The entry may carry a ``"type"`` field naming the concrete class (e.g.
``"OpticalFiberModel"``); when omitted, a plain ``pynwb.device.DeviceModel`` is created.
Returns
-------
DeviceModel
The DeviceModel object (either newly created or existing).
"""
_validate_device_registry_names(metadata)
device_models_metadata = metadata.get("DeviceModels", {})
if metadata_key not in device_models_metadata:
raise ValueError(
f"device_model_metadata_key '{metadata_key}' was not found in metadata['DeviceModels'] "
f"(available keys: {list(device_models_metadata)})."
)
device_model_metadata = device_models_metadata[metadata_key]
model_name = device_model_metadata["name"]
if model_name in nwbfile.device_models:
return nwbfile.device_models[model_name]
model_kwargs = {key: value for key, value in device_model_metadata.items() if key != "type"}
# Required by the NWB ``DeviceModel`` but rarely recorded by an acquisition file. Fill any missing
# one at write time rather than forcing every ``get_metadata`` to state a manufacturer its source
# never named. The placeholder is an explicit unknown-marker, as in the ophys and ecephys templates,
# so it reads as "the source did not say" rather than as a value. Mirrors
# ``_add_imaging_plane_to_nwbfile`` and ``_add_electrode_groups_to_nwbfile``.
required_fields = ["manufacturer"]
default_device_model_metadata = {"manufacturer": "unknown"}
for field in required_fields:
model_kwargs.setdefault(field, default_device_model_metadata[field])
model_class = _resolve_type(
device_model_metadata.get("type", "DeviceModel"), sources=_DEVICE_MODEL_TYPE_SOURCES, base_class=DeviceModel
)
model_kwargs = _build_inline_containers(target_class=model_class, kwargs=model_kwargs)
device_model = model_class(**model_kwargs)
nwbfile.add_device_model(device_model)
return device_model
def _get_device_template_entry(*, device_model_metadata_key: str) -> dict:
"""A blank device, ready to be linked to by whatever hangs off it.
The make and catalog specification belong to the model rather than to the instrument: pynwb
deprecated ``Device.manufacturer``, ``model_number`` and ``model_name`` in favor of a linked
``DeviceModel``, so those are offered there and only the serial number of this one instrument
stays here.
"""
return dict(
name=None,
description=None,
serial_number=None,
device_model_metadata_key=device_model_metadata_key,
)
def _get_device_model_template_entry() -> dict:
"""A blank device model: the make and catalog specification, shared by every recording on that instrument.
Optional as a whole. To drop it, delete the entry and the ``device_model_metadata_key`` pointing
at it.
"""
return dict(name=None, manufacturer=None, model_number=None, description=None)
def _add_device_to_nwbfile(
nwbfile: NWBFile,
*,
metadata: dict | None = None,
metadata_key: str | None = None,
device_metadata: dict | None = None,
):
"""
Add a device to an NWBFile. Idempotent on ``name`` (an existing device is returned unchanged).
Two calling forms:
* **Canonical** ``(nwbfile, metadata, metadata_key)`` — the shape all callers should converge on,
matching the rest of NeuroConv. Resolves ``metadata["Devices"][metadata_key]`` and, if the entry
carries a ``device_model_metadata_key``, adds that model (idempotently, on demand) and links it.
* **Transitional** ``(nwbfile, device_metadata)`` — a pre-resolved entry dict, for the deprecated
nested-device fallbacks on the video interfaces, whose entry has no registry key to resolve. This
form goes with them.
In either form the entry may carry a ``"type"`` field naming the concrete class (e.g.
``"OpticalFiber"``); when omitted, a plain ``pynwb.device.Device`` is created.
Returns
-------
Device
The Device object (either newly created or existing).
"""
if metadata_key is not None:
if metadata is None:
raise ValueError("Provide `metadata` with `metadata_key`.")
_validate_device_registry_names(metadata)
# Checked rather than indexed: ``metadata`` is often a ``DeepDict``, where a missing key
# auto-vivifies instead of raising, and the failure then surfaces several lines later as
# ``TypeError: unhashable type: 'DeepDict'`` naming neither the key nor the registry.
devices_metadata = metadata.get("Devices", {})
if metadata_key not in devices_metadata:
raise ValueError(
f"device_metadata_key '{metadata_key}' was not found in metadata['Devices'] "
f"(available keys: {list(devices_metadata)})."
)
device_metadata = devices_metadata[metadata_key]
device_model_metadata_key = device_metadata.get("device_model_metadata_key")
if device_model_metadata_key is not None:
model = _add_device_model_to_nwbfile(
nwbfile=nwbfile, metadata=metadata, metadata_key=device_model_metadata_key
)
device_metadata = {**device_metadata, "model": model}
elif device_metadata is None:
raise ValueError("Provide either `metadata` + `metadata_key` (canonical) or `device_metadata` (transitional).")
device_name = device_metadata["name"]
if device_name in nwbfile.devices:
return nwbfile.devices[device_name]
internal_keys = ("type", "device_model_metadata_key")
device_kwargs = {key: value for key, value in device_metadata.items() if key not in internal_keys}
device_class = _resolve_type(device_metadata.get("type", "Device"), sources=_DEVICE_TYPE_SOURCES, base_class=Device)
device_kwargs = _build_inline_containers(target_class=device_class, kwargs=device_kwargs)
device = device_class(**device_kwargs)
nwbfile.add_device(device)
return device
def _attempt_cleanup_of_existing_nwbfile(nwbfile_path: Path) -> None:
if not nwbfile_path.exists():
return
try:
nwbfile_path.unlink()
# Windows in particular can encounter errors at this step
except PermissionError: # pragma: no cover
message = f"Unable to remove NWB file located at {nwbfile_path.absolute()}! Please remove it manually."
warnings.warn(message=message, stacklevel=2)
[docs]
@contextmanager
def make_or_load_nwbfile(
nwbfile_path: FilePath | None = None,
nwbfile: NWBFile | None = None,
metadata: dict | None = None,
overwrite: bool = False,
backend: Literal["hdf5", "zarr"] = "hdf5",
verbose: bool = False,
):
"""
Context for automatically handling decision of write vs. append for writing an NWBFile.
Parameters
----------
nwbfile_path: FilePath
Path for where to write or load (if overwrite=False) the NWBFile.
If specified, the context will always write to this location.
nwbfile: NWBFile, optional
An in-memory NWBFile object to write to the location.
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.
The default is False (append mode).
backend : "hdf5" or "zarr", default: "hdf5"
The type of backend used to create the file.
verbose: bool, default: True
If 'nwbfile_path' is specified, informs user after a successful write operation.
"""
warnings.warn(
"`make_or_load_nwbfile` is deprecated and will be removed on or after February 2027. "
"Use the `run_conversion` method of an interface or converter, or `configure_and_write_nwbfile` "
"for an NWBFile that has already been assembled.",
FutureWarning,
stacklevel=2,
)
from . import BACKEND_NWB_IO
nwbfile_path_is_provided = nwbfile_path is not None
nwbfile_path_in = Path(nwbfile_path) if nwbfile_path_is_provided else None
nwbfile_is_provided = nwbfile is not None
nwbfile_in = nwbfile if nwbfile_is_provided else None
backend_io_class = BACKEND_NWB_IO[backend]
assert not (nwbfile_path is None and nwbfile is None and metadata is None), (
"You must specify either an 'nwbfile_path', or an in-memory 'nwbfile' object, "
"or provide the metadata for creating one."
)
assert not (overwrite is False and nwbfile_path_in and nwbfile_path_in.exists() and nwbfile is not None), (
"'nwbfile_path' exists at location, 'overwrite' is False (append mode), but an in-memory 'nwbfile' object was "
"passed! Cannot reconcile which nwbfile object to write."
)
if overwrite is False and backend == "zarr":
# TODO: remove when https://github.com/hdmf-dev/hdmf-zarr/issues/182 is resolved
raise NotImplementedError("Appending a Zarr file is not yet supported!")
load_kwargs = dict()
file_initially_exists = nwbfile_path_in.exists() if nwbfile_path_is_provided else False
append_mode = file_initially_exists and not overwrite
if nwbfile_path_is_provided:
load_kwargs.update(path=str(nwbfile_path_in))
if append_mode:
load_kwargs.update(mode="r+", load_namespaces=True)
# Check if the selected backend is the backend of the file in nwfile_path
backends_that_can_read = [
backend_name
for backend_name, backend_io_class in BACKEND_NWB_IO.items()
if backend_io_class.can_read(path=str(nwbfile_path_in))
]
# Future-proofing: raise an error if more than one backend can read the file
assert (
len(backends_that_can_read) <= 1
), "More than one backend is capable of reading the file! Please raise an issue describing your file."
if backend not in backends_that_can_read:
raise IOError(
f"The chosen backend ('{backend}') is unable to read the file! "
f"Please select '{backends_that_can_read[0]}' instead."
)
else:
load_kwargs.update(mode="w")
io = backend_io_class(**load_kwargs)
read_nwbfile = nwbfile_path_is_provided and append_mode
create_nwbfile = not read_nwbfile and not nwbfile_is_provided
nwbfile_loaded_succesfully = True
nwbfile_written_succesfully = True
try:
if nwbfile_is_provided:
nwbfile = nwbfile_in
elif read_nwbfile:
nwbfile = io.read()
elif create_nwbfile:
if metadata is None:
error_msg = "Metadata is required for creating an nwbfile "
raise ValueError(error_msg)
default_metadata = get_default_nwbfile_metadata()
default_metadata.deep_update(metadata)
nwbfile = make_nwbfile_from_metadata(metadata=metadata)
yield nwbfile
except Exception as load_error:
nwbfile_loaded_succesfully = False
raise load_error
finally:
if nwbfile_path_is_provided and nwbfile_loaded_succesfully:
try:
io.write(nwbfile)
if verbose:
print(f"NWB file saved at {nwbfile_path_in}!")
except Exception as write_error:
nwbfile_written_succesfully = False
raise write_error
finally:
io.close()
del io
if not nwbfile_written_succesfully:
_attempt_cleanup_of_existing_nwbfile(nwbfile_path=nwbfile_path_in)
elif nwbfile_path_is_provided and not nwbfile_loaded_succesfully:
# The instantiation of the IO object can itself create a file
_attempt_cleanup_of_existing_nwbfile(nwbfile_path=nwbfile_path_in)
else:
# This is the case where nwbfile is provided but not nwbfile_path
# Note that io never gets created in this case, so no need to close or delete it
pass
# Final attempt to cleanup an unintended file creation, just to be sure
any_load_or_write_error = not nwbfile_loaded_succesfully or not nwbfile_written_succesfully
file_was_freshly_created = not file_initially_exists and nwbfile_path_is_provided and nwbfile_path_in.exists()
attempt_to_cleanup = any_load_or_write_error and file_was_freshly_created
if attempt_to_cleanup:
_attempt_cleanup_of_existing_nwbfile(nwbfile_path=nwbfile_path_in)
def _fetch_backend_from_nwbfile_on_disk(
nwbfile_path: FilePath,
backend: Literal["hdf5", "zarr"] | None = None,
backend_configuration: BackendConfiguration | None = None,
) -> Literal["hdf5", "zarr"]:
"""
Fetch the backend of an NWB file that already exists on disk.
A file can only be opened with the backend it was written with, so a ``backend`` or
``backend_configuration`` that is passed is checked against the file rather than used to select the backend.
Parameters
----------
nwbfile_path: FilePath
Path of the NWB file to inspect.
backend: {"hdf5", "zarr"}, optional
backend_configuration: BackendConfiguration, optional
Returns
-------
backend: {"hdf5", "zarr"}
The backend of the file at ``nwbfile_path``.
"""
nwbfile_path = Path(nwbfile_path)
if not nwbfile_path.exists():
raise FileNotFoundError(f"No NWB file exists at '{nwbfile_path}'!")
backends_that_can_read = [
backend_name
for backend_name, backend_io_class in BACKEND_NWB_IO.items()
if backend_io_class.can_read(path=str(nwbfile_path))
]
if len(backends_that_can_read) == 0:
raise IOError(
f"The file at '{nwbfile_path}' cannot be read by any of the available backends "
f"({list(BACKEND_NWB_IO)})!"
)
# Future-proofing: raise an error if more than one backend can read the file
assert (
len(backends_that_can_read) == 1
), "More than one backend is capable of reading the file! Please raise an issue describing your file."
backend_on_disk = backends_that_can_read[0]
requested_backends = {
"backend": backend,
"backend_configuration.backend": backend_configuration.backend if backend_configuration is not None else None,
}
for parameter_name, requested_backend in requested_backends.items():
if requested_backend is not None and requested_backend != backend_on_disk:
raise ValueError(
f"The file at '{nwbfile_path}' was written with the '{backend_on_disk}' backend, but "
f"`{parameter_name}` is '{requested_backend}'. An existing file can only be opened with the "
f"backend it was written with, so specify '{backend_on_disk}' or leave the backend unspecified."
)
return backend_on_disk
[docs]
def repack_nwbfile(
*,
nwbfile_path: Path,
export_nwbfile_path: Path,
export_backend: Literal["hdf5", "zarr", None] = None,
):
"""
Repack an NWBFile with a new backend configuration.
Parameters
----------
nwbfile_path : Path
Path to the NWB file to be repacked.
export_nwbfile_path : Path
Path to export the repacked NWB file.
export_backend : {"hdf5", "zarr", None}, default: None
The type of backend used to write the repacked file. If None, the same backend as the input file is used.
"""
# The backend of the source file is a property of the file; the export backend is an independent
# choice that defaults to it.
if export_backend is None:
export_backend = _fetch_backend_from_nwbfile_on_disk(nwbfile_path=nwbfile_path)
# Read the file using read_nwb (automatically detects backend)
nwbfile = read_nwb(nwbfile_path)
backend_configuration = get_default_backend_configuration(nwbfile=nwbfile, backend=export_backend)
configure_and_write_nwbfile(
nwbfile=nwbfile,
backend_configuration=backend_configuration,
nwbfile_path=export_nwbfile_path,
)