Add OpenEphysBinaryConverter for multi-stream OpenEphys binary data

CHANGELOG.md

@@ -20,6 +20,7 @@
 * Fixed `get_json_schema_from_method_signature` to resolve PEP 563 string annotations (from `from __future__ import annotations`) before passing them to pydantic. This affected any interface defined in a module with deferred annotations (e.g. `MiniscopeConverter`, or external subclasses from SpikeInterface). [PR #1670](https://github.com/catalystneuro/neuroconv/pull/1670)
 
 ## Features
+* Added `OpenEphysBinaryConverter` for automatic multi-stream OpenEphys binary conversion, following the `SpikeGLXConverterPipe` pattern. Auto-discovers streams and routes neural data to `OpenEphysBinaryRecordingInterface` and analog (ADC/NI-DAQ) data to `OpenEphysBinaryAnalogInterface`. [PR #1686](https://github.com/catalystneuro/neuroconv/pull/1686)
 * Added dict-based metadata pipeline for imaging in `roiextractors.py`, supporting the new `MicroscopySeries`, `ImagingPlanes`, and `Devices` metadata format keyed by `metadata_key`. Old list-based functions are preserved (renamed with `_old_list_format` suffix) and dispatched automatically when `metadata_key` is not provided. [PR #1677](https://github.com/catalystneuro/neuroconv/pull/1677)
 * Added dict-based metadata pipeline for segmentation in `roiextractors.py` (`_add_plane_segmentation_to_nwbfile`, `_add_roi_response_traces_to_nwbfile`) with dual routing in `add_segmentation_to_nwbfile`. Masks are written in the extractor's native format and all traces go into a single `Fluorescence` container. [PR #1692](https://github.com/catalystneuro/neuroconv/pull/1692)
 

docs/conversion_examples_gallery/recording/openephys.rst

@@ -1,6 +1,11 @@
 OpenEphys data conversion
 -------------------------
 
+OpenEphys supports two data formats: the `Binary (.dat) format <https://open-ephys.github.io/gui-docs/User-Manual/Data-formats/Binary-format.html#binaryformat>`_
+and the `Open Ephys (.continuous) format <https://open-ephys.github.io/gui-docs/User-Manual/Data-formats/Open-Ephys-format.html>`_.
+The :py:class:`~neuroconv.datainterfaces.ecephys.openephys.openephysdatainterface.OpenEphysRecordingInterface`
+supports both formats and auto-detects which one to use based on the files present in the folder.
+
 Install NeuroConv with the additional dependencies necessary for reading OpenEphys data.
 
 .. code-block:: bash
@@ -32,3 +37,38 @@ Convert OpenEphys data to NWB using :py:class:`~neuroconv.datainterfaces.ecephys
     >>> # Choose a path for saving the nwb file and run the conversion
     >>> nwbfile_path = f"{path_to_save_nwbfile}"  # This should be something like: "./saved_file.nwb"
     >>> interface.run_conversion(nwbfile_path=nwbfile_path, metadata=metadata)
+
+OpenEphysBinaryConverter
+~~~~~~~~~~~~~~~~~~~~~~~~
+
+For multi-stream OpenEphys binary recordings (e.g. Neuropixels with AP, LFP, and analog streams),
+use :py:class:`~neuroconv.converters.OpenEphysBinaryConverter` to convert all streams at once.
+When AP and LFP streams come from the same probe, the converter automatically shares electrode
+table rows between them.
+
+.. code-block:: python
+
+    >>> from datetime import datetime
+    >>> from zoneinfo import ZoneInfo
+    >>> from neuroconv.converters import OpenEphysBinaryConverter
+    >>>
+    >>> folder_path = f"{ECEPHY_DATA_PATH}/openephysbinary/v0.6.x_neuropixels_with_sync"
+    >>> # Change the folder_path to the appropriate location in your system
+    >>> converter = OpenEphysBinaryConverter(folder_path=folder_path)
+    >>> # Extract what metadata we can from the source files
+    >>> metadata = converter.get_metadata()
+    >>> # For data provenance we add the time zone information to the conversion
+    >>> session_start_time = metadata["NWBFile"]["session_start_time"].replace(tzinfo=ZoneInfo("US/Pacific"))
+    >>> metadata["NWBFile"].update(session_start_time=session_start_time)
+    >>> # Add subject information (required for DANDI upload)
+    >>> metadata["Subject"] = dict(subject_id="subject1", species="Mus musculus", sex="M", age="P30D")
+    >>>
+    >>> # Choose a path for saving the nwb file and run the conversion
+    >>> nwbfile_path = output_folder / "my_openephys_binary_converter_session.nwb"
+    >>> converter.run_conversion(nwbfile_path=nwbfile_path, metadata=metadata)
+
+.. note::
+
+    The ``OpenEphysBinaryConverter`` only supports the Binary (.dat) format. There is currently no converter for the
+    Open Ephys (.continuous) format. If you need multi-stream conversion support for Open Ephys (.continuous) format data, please
+    `open an issue <https://github.com/catalystneuro/neuroconv/issues>`_.

src/neuroconv/converters/__init__.py

@@ -11,6 +11,7 @@
 )
 from ..datainterfaces.ecephys.sortedrecordinginterface import SortedRecordingConverter
 from ..datainterfaces.ecephys.spikeglx.sorted_spikeglx_converter import SortedSpikeGLXConverter
+from ..datainterfaces.ecephys.openephys.openephysbinaryconverter import OpenEphysBinaryConverter
 from ..datainterfaces.ecephys.spikeglx.spikeglxconverter import SpikeGLXConverterPipe
 from ..datainterfaces.ophys.brukertiff.brukertiffconverter import (
     BrukerTiffMultiPlaneConverter,
@@ -20,6 +21,7 @@
 
 converter_list = [
     LightningPoseConverter,
+    OpenEphysBinaryConverter,
     SpikeGLXConverterPipe,
     BrukerTiffMultiPlaneConverter,
     BrukerTiffSinglePlaneConverter,

src/neuroconv/datainterfaces/ecephys/openephys/openephysbinaryconverter.py

@@ -0,0 +1,111 @@
+from pathlib import Path
+
+from pydantic import DirectoryPath, validate_call
+
+from .openephybinarysanaloginterface import OpenEphysBinaryAnalogInterface
+from .openephysbinarydatainterface import OpenEphysBinaryRecordingInterface
+from ....nwbconverter import ConverterPipe
+from ....utils import get_json_schema_from_method_signature
+
+
+class OpenEphysBinaryConverter(ConverterPipe):
+    """
+    Converter for multi-stream OpenEphys binary recording data.
+
+    Auto-discovers all streams in a folder and creates the appropriate interfaces
+    (recording for neural streams, analog for ADC/NI-DAQ streams).
+    """
+
+    display_name = "OpenEphys Binary Converter"
+    keywords = OpenEphysBinaryRecordingInterface.keywords + OpenEphysBinaryAnalogInterface.keywords
+    associated_suffixes = OpenEphysBinaryRecordingInterface.associated_suffixes
+    info = "Converter for multi-stream OpenEphys binary recording data."
+
+    @classmethod
+    def get_source_schema(cls) -> dict:
+        source_schema = get_json_schema_from_method_signature(method=cls.__init__, exclude=["streams"])
+        source_schema["properties"]["folder_path"][
+            "description"
+        ] = "Path to the folder containing OpenEphys binary streams."
+        return source_schema
+
+    @classmethod
+    def get_streams(cls, folder_path: DirectoryPath) -> list[str]:
+        """
+        Get the stream names available in the folder.
+
+        Parameters
+        ----------
+        folder_path : DirectoryPath
+            Path to the folder containing OpenEphys binary streams.
+
+        Returns
+        -------
+        list of str
+            The names of all available streams in the folder.
+        """
+        from spikeinterface.extractors.extractor_classes import (
+            OpenEphysBinaryRecordingExtractor,
+        )
+
+        return OpenEphysBinaryRecordingExtractor.get_streams(folder_path=folder_path)[0]
+
+    @validate_call
+    def __init__(
+        self,
+        folder_path: DirectoryPath,
+        streams: list[str] | None = None,
+        verbose: bool = False,
+    ):
+        """
+        Read all data from multiple streams stored in OpenEphys binary format.
+
+        Parameters
+        ----------
+        folder_path : DirectoryPath
+            Path to the folder containing OpenEphys binary streams.
+        streams : list of str, optional
+            A specific list of streams to load.
+            To see which streams are available, run
+            `OpenEphysBinaryConverter.get_streams(folder_path="path/to/openephys")`.
+            By default, all available streams are loaded.
+        verbose : bool, default: False
+            Whether to output verbose text.
+        """
+        folder_path = Path(folder_path)
+
+        stream_names = streams or self.get_streams(folder_path=folder_path)
+        self._stream_names = stream_names
+
+        non_neural_indicators = ["ADC", "NI-DAQ"]
+        is_non_neural = lambda name: any(indicator in name for indicator in non_neural_indicators)
+        _to_suffix = lambda name: name.rsplit(".", maxsplit=1)[-1].replace("-", "")
+        neural_streams = [name for name in stream_names if not is_non_neural(name)]
+        analog_streams = [name for name in stream_names if is_non_neural(name)]
+
+        data_interfaces = {}
+
+        for stream_name in neural_streams:
+            es_key = "ElectricalSeries" + _to_suffix(stream_name)
+            data_interfaces[stream_name] = OpenEphysBinaryRecordingInterface(
+                folder_path=folder_path,
+                stream_name=stream_name,
+                es_key=es_key,
+            )
+
+        for stream_name in analog_streams:
+            time_series_name = "TimeSeries" + _to_suffix(stream_name)
+            data_interfaces[stream_name] = OpenEphysBinaryAnalogInterface(
+                folder_path=folder_path,
+                stream_name=stream_name,
+                time_series_name=time_series_name,
+            )
+
+        super().__init__(data_interfaces=data_interfaces, verbose=verbose)
+
+    def get_conversion_options_schema(self) -> dict:
+        conversion_options_schema = super().get_conversion_options_schema()
+        conversion_options_schema["properties"].update(
+            {name: interface.get_conversion_options_schema() for name, interface in self.data_interface_objects.items()}
+        )
+        return conversion_options_schema

src/neuroconv/datainterfaces/ecephys/openephys/openephysbinarydatainterface.py

@@ -1,3 +1,4 @@
+import re
 import warnings
 
 from pydantic import DirectoryPath
@@ -165,6 +166,38 @@ def __init__(
         if len(neural_channels) < len(channel_ids):
             self.recording_extractor = recording.select_channels(channel_ids=neural_channels)
 
+        # Set composite channel_name for multi-stream electrode deduplication
+        # When AP and LFP streams exist for the same probe, they record from the
+        # same physical electrodes. Setting composite names (e.g. "AP0,LFP0") on
+        # both streams lets the electrode table builder match them to the same rows,
+        # avoiding duplicate entries. This follows the same approach as SpikeGLX.
+        if stream_name is not None:
+            band_suffixes = {"-AP": "-LFP", "-LFP": "-AP"}
+            current_suffix = None
+            for suffix in band_suffixes:
+                if stream_name.endswith(suffix):
+                    current_suffix = suffix
+                    break
+
+            if current_suffix is not None:
+                companion_suffix = band_suffixes[current_suffix]
+                prefix = stream_name[: -len(current_suffix)]
+                companion_stream = prefix + companion_suffix
+                has_companion = companion_stream in available_streams
+
+                if has_companion:
+                    channel_ids = self.recording_extractor.get_channel_ids()
+                    channel_names = []
+                    for channel_id in channel_ids:
+                        # Extract the numeric part from channel_id (e.g. "AP1" -> "1", "LFP3" -> "3")
+                        match = re.search(r"\d+$", str(channel_id))
+                        channel_number = match.group() if match else str(channel_id)
+                        # Composite name with both bands, alphabetically sorted
+                        channel_name = f"AP{channel_number},LFP{channel_number}"
+                        channel_names.append(channel_name)
+
+                    self.recording_extractor.set_property(key="channel_name", ids=channel_ids, values=channel_names)
+
     def get_metadata(self) -> DeepDict:
         from ._openephys_utils import _get_session_start_time
 

tests/test_on_data/ecephys/test_openephys_binary_converter.py

@@ -0,0 +1,89 @@
+from datetime import datetime
+
+from pynwb import read_nwb
+
+from neuroconv.converters import OpenEphysBinaryConverter
+
+from ..setup_paths import ECEPHY_DATA_PATH
+
+OPENEPHYSBINARY_PATH = ECEPHY_DATA_PATH / "openephysbinary"
+
+
+class TestNeuralAndAnalogMixed:
+    """Test with a single neural stream and a single ADC stream."""
+
+    folder_path = OPENEPHYSBINARY_PATH / "neural_and_non_neural_data_mixed"
+
+    def test_metadata(self):
+        converter = OpenEphysBinaryConverter(folder_path=self.folder_path)
+        metadata = converter.get_metadata()
+
+        assert metadata["NWBFile"]["session_start_time"] == datetime(2022, 7, 25, 15, 30, 0)
+
+    def test_conversion(self, tmp_path):
+        converter = OpenEphysBinaryConverter(folder_path=self.folder_path)
+
+        assert len(converter.data_interface_objects) == 2
+
+        nwbfile_path = tmp_path / "test_neural_and_analog_mixed.nwb"
+        conversion_options = {name: dict(stub_test=True) for name in converter.data_interface_objects}
+        converter.run_conversion(nwbfile_path=nwbfile_path, conversion_options=conversion_options)
+
+        nwbfile = read_nwb(path=nwbfile_path)
+
+        assert "ElectricalSeries0" in nwbfile.acquisition
+        assert "TimeSeries0_ADC" in nwbfile.acquisition
+        assert len(nwbfile.acquisition) == 2
+
+
+class TestMultiStreamWithAnalog:
+    """Test with NI-DAQ analog + Neuropixels AP/LFP streams."""
+
+    folder_path = OPENEPHYSBINARY_PATH / "v0.6.x_neuropixels_with_sync"
+
+    def test_metadata(self):
+        converter = OpenEphysBinaryConverter(folder_path=self.folder_path)
+        metadata = converter.get_metadata()
+
+        assert metadata["NWBFile"]["session_start_time"] == datetime(2023, 8, 30, 23, 41, 36)
+
+    def test_conversion(self, tmp_path):
+        converter = OpenEphysBinaryConverter(folder_path=self.folder_path)
+
+        # NI-DAQ + ProbeA-AP + ProbeA-LFP + ProbeA-APSYNC + ProbeA-LFPSYNC
+        assert len(converter.data_interface_objects) == 5
+
+        nwbfile_path = tmp_path / "test_multi_stream_with_analog.nwb"
+        conversion_options = {name: dict(stub_test=True) for name in converter.data_interface_objects}
+        converter.run_conversion(nwbfile_path=nwbfile_path, conversion_options=conversion_options)
+
+        nwbfile = read_nwb(path=nwbfile_path)
+
+        assert "ElectricalSeriesProbeAAP" in nwbfile.acquisition
+        assert "ElectricalSeriesProbeALFP" in nwbfile.acquisition
+        assert "ElectricalSeriesProbeAAPSYNC" in nwbfile.acquisition
+        assert "ElectricalSeriesProbeALFPSYNC" in nwbfile.acquisition
+        assert "TimeSeriesPXIe6341" in nwbfile.acquisition
+        assert len(nwbfile.acquisition) == 5
+
+        # Both AP and LFP come from the same probe so they share electrode rows.
+        # 384 probe channels + 2 SYNC channels = 386 electrodes
+        assert len(nwbfile.electrodes) == 386
+        ap_electrode_indices = list(nwbfile.acquisition["ElectricalSeriesProbeAAP"].electrodes.data)
+        lfp_electrode_indices = list(nwbfile.acquisition["ElectricalSeriesProbeALFP"].electrodes.data)
+        assert len(ap_electrode_indices) == 384
+        assert len(lfp_electrode_indices) == 384
+        assert ap_electrode_indices == lfp_electrode_indices
+
+    def test_streams_argument_filters_data(self):
+        all_streams = OpenEphysBinaryConverter.get_streams(folder_path=self.folder_path)
+        neural_only = [s for s in all_streams if "NI-DAQ" not in s]
+
+        converter = OpenEphysBinaryConverter(folder_path=self.folder_path, streams=neural_only)
+        conversion_options = {name: dict(stub_test=True) for name in converter.data_interface_objects}
+        nwbfile = converter.create_nwbfile(conversion_options=conversion_options)
+
+        assert "ElectricalSeriesProbeAAP" in nwbfile.acquisition
+        assert "ElectricalSeriesProbeALFP" in nwbfile.acquisition
+        assert "TimeSeriesPXIe6341" not in nwbfile.acquisition
+        assert len(nwbfile.acquisition) == 4