import warnings
from typing import Literal
from pydantic import DirectoryPath, validate_call
from pynwb import NWBFile
from ... import (
BrukerTiffImagingInterface,
BrukerTiffMultiPlaneImagingInterface,
BrukerTiffSinglePlaneImagingInterface,
)
from ....basedatainterface import BaseDataInterface
from ....nwbconverter import ConverterPipe
from ....utils import DeepDict, dict_deep_update, get_json_schema_from_method_signature
[docs]
class BrukerTiffConverter(ConverterPipe):
"""Convert a Bruker Prairie View session.
Point this at the session folder (the OME-TIFF files and the ``.xml`` that describes them) and
every channel in it is written: one ``ImagingPlane`` and one ``TwoPhotonSeries`` per channel,
all referring to a single ``Device`` named ``BrukerFluorescenceMicroscope``. Planar and
volumetric, single-channel and multi-channel sessions are all handled; a single-channel session
needs no extra arguments.
``plane_separation_type`` chooses how a volumetric session is written. ``"contiguous"``
(default) writes the volume as one 4D ``TwoPhotonSeries`` per channel; ``"disjoint"`` writes
each depth plane as its own 2D ``TwoPhotonSeries`` and ``ImagingPlane``, carrying that plane's
own focal depth. It has no effect on planar sessions. This replaces the deprecated
``BrukerTiffMultiPlaneConverter``.
"""
display_name = "Bruker TIFF Imaging"
keywords = BrukerTiffImagingInterface.keywords
associated_suffixes = BrukerTiffImagingInterface.associated_suffixes
info = "Auto-channel-enumerated converter for Bruker TIFF imaging data."
[docs]
@classmethod
def get_source_schema(cls) -> dict:
source_schema = get_json_schema_from_method_signature(method=cls.__init__)
source_schema["properties"]["folder_path"][
"description"
] = "Folder containing Bruker .ome.tif files and the matching configuration .xml."
return source_schema
@validate_call
def __init__(
self,
folder_path: DirectoryPath,
plane_separation_type: Literal["contiguous", "disjoint"] = "contiguous",
verbose: bool = False,
):
"""
Parameters
----------
folder_path : DirectoryPath
Folder containing Bruker .ome.tif files and the matching configuration .xml.
plane_separation_type : {"contiguous", "disjoint"}, default: "contiguous"
How to write volumetric data. ``"contiguous"`` writes one 4D ``TwoPhotonSeries`` per
channel; ``"disjoint"`` writes one 2D ``TwoPhotonSeries`` per depth plane per channel.
Has no effect on single-plane (planar) acquisitions.
verbose : bool, default: False
"""
channel_names = BrukerTiffImagingInterface.get_available_channels(folder_path=folder_path)
single_channel = len(channel_names) == 1
# Plane count is folder-level; probe once with the full (unsliced) extractor.
probe_channel = None if single_channel else channel_names[0]
probe = BrukerTiffImagingInterface(folder_path=folder_path, channel_name=probe_channel, verbose=verbose)
num_planes = probe._bruker_extractor.get_num_planes() if probe.imaging_extractor.is_volumetric else 1
disjoint = plane_separation_type == "disjoint" and num_planes > 1
data_interfaces: dict[str, BrukerTiffImagingInterface] = {}
for channel_name in channel_names:
interface_name = "BrukerImaging" if single_channel else f"BrukerImaging_{channel_name}"
channel_argument = None if single_channel else channel_name
if disjoint:
for plane_index in range(num_planes):
data_interfaces[f"{interface_name}_plane{plane_index}"] = BrukerTiffImagingInterface(
folder_path=folder_path,
channel_name=channel_argument,
plane_index=plane_index,
verbose=verbose,
)
else:
data_interfaces[interface_name] = BrukerTiffImagingInterface(
folder_path=folder_path,
channel_name=channel_argument,
verbose=verbose,
)
super().__init__(data_interfaces=data_interfaces, verbose=verbose)
[docs]
class BrukerTiffMultiPlaneConverter(BaseDataInterface):
"""
Deprecated. Use :class:`~neuroconv.converters.BrukerTiffConverter` instead.
Both ``plane_separation_type`` modes are now covered by ``BrukerTiffConverter``:
``"contiguous"`` (one 4D ``TwoPhotonSeries`` per channel) and ``"disjoint"`` (one 2D
``TwoPhotonSeries`` per depth plane), the latter via per-plane selection on the unified
:class:`~neuroconv.datainterfaces.BrukerTiffImagingInterface`.
"""
display_name = "Bruker TIFF Imaging (multiple channels, multiple planes)"
keywords = BrukerTiffMultiPlaneImagingInterface.keywords
associated_suffixes = BrukerTiffMultiPlaneImagingInterface.associated_suffixes
info = "Interface for handling all channels and all planes of Bruker imaging data."
[docs]
@classmethod
def get_source_schema(cls):
source_schema = get_json_schema_from_method_signature(cls)
source_schema["properties"]["folder_path"][
"description"
] = "The folder that contains the Bruker TIF image files (.ome.tif) and configuration files (.xml, .env)."
return source_schema
@validate_call
def __init__(
self,
folder_path: DirectoryPath,
plane_separation_type: Literal["disjoint", "contiguous"],
verbose: bool = False,
):
"""
Initializes the data interfaces for Bruker volumetric imaging data stream.
Parameters
----------
folder_path : DirectoryPath
The path to the folder that contains the Bruker TIF image files (.ome.tif) and configuration files (.xml, .env).
plane_separation_type: {'contiguous', 'disjoint'}
Defines how to write volumetric imaging data. Use 'contiguous' to create the volumetric two photon series,
and 'disjoint' to create separate imaging plane and two photon series for each plane.
verbose : bool, default: False
Controls verbosity.
"""
warnings.warn(
"BrukerTiffMultiPlaneConverter is deprecated and will be removed on or after February 2027. "
"Use BrukerTiffConverter with plane_separation_type instead.",
FutureWarning,
stacklevel=2,
)
self.verbose = verbose
self.data_interface_objects = dict()
streams = BrukerTiffMultiPlaneImagingInterface.get_streams(
folder_path=folder_path,
plane_separation_type=plane_separation_type,
)
channel_streams = streams["channel_streams"]
interface_name = "BrukerImaging"
with warnings.catch_warnings():
warnings.simplefilter("ignore", FutureWarning)
for channel_stream_name in channel_streams:
plane_streams = streams["plane_streams"][channel_stream_name]
for plane_stream in plane_streams:
if len(plane_streams) > 1:
interface_name += plane_stream.replace("_", "")
if plane_separation_type == "contiguous":
self.data_interface_objects[interface_name] = BrukerTiffMultiPlaneImagingInterface(
folder_path=folder_path,
stream_name=plane_stream,
)
elif plane_separation_type == "disjoint":
self.data_interface_objects[interface_name] = BrukerTiffSinglePlaneImagingInterface(
folder_path=folder_path,
stream_name=plane_stream,
)
[docs]
def add_to_nwbfile(
self,
nwbfile: NWBFile,
metadata,
stub_test: bool = False,
stub_samples: int = 100,
):
"""
Add data from multiple data interfaces to the specified NWBFile.
Parameters
----------
nwbfile : NWBFile
The NWBFile object to which the data will be added.
metadata : dict
Metadata dictionary containing information to describe the data being added to the NWB file.
stub_test : bool, optional
If True, only a subset of the data (up to `stub_samples`) will be added for testing purposes. Default is False.
stub_samples : int, default: 100
The number of samples (frames) to use for testing.
"""
for photon_series_index, (interface_name, data_interface) in enumerate(self.data_interface_objects.items()):
data_interface.add_to_nwbfile(
nwbfile=nwbfile,
metadata=metadata,
photon_series_index=photon_series_index,
stub_test=stub_test,
stub_samples=stub_samples,
)
[docs]
class BrukerTiffSinglePlaneConverter(BaseDataInterface):
"""
Primary data interface class for converting Bruker imaging data with multiple channels and a single plane.
"""
display_name = "Bruker TIFF Imaging (multiple channels, single plane)"
keywords = BrukerTiffMultiPlaneImagingInterface.keywords
associated_suffixes = BrukerTiffMultiPlaneImagingInterface.associated_suffixes
info = "Interface for handling multiple channels of a single plane of Bruker imaging data."
[docs]
@classmethod
def get_source_schema(cls):
return get_json_schema_from_method_signature(cls)
@validate_call
def __init__(
self,
folder_path: DirectoryPath,
verbose: bool = False,
):
"""
Initializes the data interfaces for Bruker imaging data stream.
Parameters
----------
folder_path : DirectoryPath
The path to the folder that contains the Bruker TIF image files (.ome.tif) and configuration files (.xml, .env).
verbose : bool, default: False
Controls verbosity.
"""
warnings.warn(
"BrukerTiffSinglePlaneConverter is deprecated and will be removed on or after February 2027."
"Use BrukerTiffImagingInterface instead.",
FutureWarning,
stacklevel=2,
)
from roiextractors.extractors.tiffimagingextractors.brukertiffimagingextractor import (
_determine_imaging_is_volumetric,
)
if _determine_imaging_is_volumetric(folder_path=folder_path):
raise ValueError("For volumetric imaging data use BrukerTiffMultiPlaneConverter.")
self.verbose = verbose
self.data_interface_objects = dict()
streams = BrukerTiffSinglePlaneImagingInterface.get_streams(folder_path=folder_path)
channel_streams = streams["channel_streams"]
interface_name = "BrukerImaging"
with warnings.catch_warnings():
warnings.simplefilter("ignore", FutureWarning)
for channel_stream_name in channel_streams:
if len(channel_streams) > 1:
interface_name += channel_stream_name.replace("_", "")
self.data_interface_objects[interface_name] = BrukerTiffSinglePlaneImagingInterface(
folder_path=folder_path,
stream_name=channel_stream_name,
)
[docs]
def add_to_nwbfile(
self,
nwbfile: NWBFile,
metadata,
stub_test: bool = False,
stub_samples: int = 100,
):
"""
Add data from all instantiated data interfaces to the provided NWBFile.
Parameters
----------
nwbfile : NWBFile
The NWBFile object to which the data will be added.
metadata : dict
Metadata dictionary containing information about the data to be added.
stub_test : bool, optional
If True, only a subset of the data (defined by `stub_samples`) will be added for testing purposes,
by default False.
stub_samples : int, default: 100
The number of samples (frames) to use for testing.
"""
for photon_series_index, (interface_name, data_interface) in enumerate(self.data_interface_objects.items()):
data_interface.add_to_nwbfile(
nwbfile=nwbfile,
metadata=metadata,
photon_series_index=photon_series_index,
stub_test=stub_test,
stub_samples=stub_samples,
)