Merge pull request #3849 from SpikeInterface/white_matter

WhiteMatterRecordingExtractor

Author: alejoe91

doc/api.rst

@@ -153,6 +153,7 @@ Non-NEO-based
     .. autofunction:: toy_example
     .. autofunction:: read_tridesclous
     .. autofunction:: read_waveclus
+    .. autofunction:: read_whitematter
     .. autofunction:: read_yass
 
 

src/spikeinterface/extractors/extractorlist.py

@@ -39,6 +39,7 @@
 from .cbin_ibl import CompressedBinaryIblExtractor, read_cbin_ibl
 from .iblextractors import IblRecordingExtractor, IblSortingExtractor, read_ibl_recording, read_ibl_sorting
 from .mcsh5extractors import MCSH5RecordingExtractor, read_mcsh5
+from .whitematterrecordingextractor import WhiteMatterRecordingExtractor
 
 # sorting extractors in relation with a sorter
 from .cellexplorersortingextractor import CellExplorerSortingExtractor, read_cellexplorer
@@ -92,6 +93,7 @@
     IblRecordingExtractor,
     MCSH5RecordingExtractor,
     SinapsResearchPlatformRecordingExtractor,
+    WhiteMatterRecordingExtractor,
 ]
 recording_extractor_full_list += neo_recording_extractors_list
 

src/spikeinterface/extractors/tests/test_whitematterrecordingextractor.py

@@ -0,0 +1,59 @@
+import pytest
+import numpy as np
+from pathlib import Path
+
+from spikeinterface.extractors import WhiteMatterRecordingExtractor, BinaryRecordingExtractor
+from spikeinterface.core.numpyextractors import NumpyRecording
+from spikeinterface import get_global_dataset_folder, download_dataset
+
+
+def test_round_trip(tmp_path):
+    num_channels = 10
+    num_samples = 500
+    traces_list = [np.ones(shape=(num_samples, num_channels), dtype="int16")]
+    sampling_frequency = 30_000.0
+    recording = NumpyRecording(traces_list=traces_list, sampling_frequency=sampling_frequency)
+
+    file_path = tmp_path / "test_WhiteMatterRecordingExtractor.raw"
+    BinaryRecordingExtractor.write_recording(recording=recording, file_paths=file_path, dtype="int16", byte_offset=8)
+
+    sampling_frequency = recording.get_sampling_frequency()
+    num_channels = recording.get_num_channels()
+    binary_recorder = WhiteMatterRecordingExtractor(
+        file_path=file_path,
+        sampling_frequency=sampling_frequency,
+        num_channels=num_channels,
+    )
+
+    # Test for full traces
+    assert np.allclose(recording.get_traces(), binary_recorder.get_traces())
+
+    # Test for a sub-set of the traces
+    start_frame = 20
+    end_frame = 40
+    smaller_traces = recording.get_traces(start_frame=start_frame, end_frame=end_frame)
+    binary_smaller_traces = binary_recorder.get_traces(start_frame=start_frame, end_frame=end_frame)
+
+    np.allclose(smaller_traces, binary_smaller_traces)
+
+
+gin_repo = "https://gin.g-node.org/NeuralEnsemble/ephy_testing_data"
+local_folder = get_global_dataset_folder() / "ephy_testing_data"
+remote_path = Path("whitematter") / "HSW_2024_12_12__10_28_23__70min_17sec__hsamp_64ch_25000sps_stub.bin"
+
+
+def test_on_data():
+    file_path = download_dataset(
+        repo=gin_repo, remote_path=remote_path, local_folder=local_folder, update_if_exists=True
+    )
+
+    sampling_frequency = 25_000.0
+    num_channels = 64
+    recording = WhiteMatterRecordingExtractor(
+        file_path=file_path,
+        sampling_frequency=sampling_frequency,
+        num_channels=num_channels,
+    )
+    assert recording.get_sampling_frequency() == sampling_frequency
+    assert recording.get_num_channels() == num_channels
+    assert recording.get_duration() == 1.0

src/spikeinterface/extractors/whitematterrecordingextractor.py

@@ -0,0 +1,64 @@
+from pathlib import Path
+from typing import List, Union, Optional
+
+from spikeinterface.core import BinaryRecordingExtractor
+from spikeinterface.core.core_tools import define_function_from_class
+
+
+class WhiteMatterRecordingExtractor(BinaryRecordingExtractor):
+    """
+    RecordingExtractor for the WhiteMatter binary format.
+
+    The recording format is a raw binary file containing int16 data,
+    with an 8-byte header offset.
+
+    Parameters
+    ----------
+    file_path : Path
+        Path to the binary file.
+    sampling_frequency : float
+        The sampling frequency.
+    num_channels : int
+        Number of channels in the recording.
+    channel_ids : list or None, default: None
+        A list of channel ids. If None, channel_ids = list(range(num_channels)).
+    is_filtered : bool or None, default: None
+        If True, the recording is assumed to be filtered. If None, `is_filtered` is not set.
+    """
+
+    mode = "file"
+
+    # Specific parameters for WhiteMatter format
+    DTYPE = "int16"
+    GAIN_TO_UV = 6.25e3 / 32768
+    OFFSET_TO_UV = 0.0
+    FILE_OFFSET = 8
+    TIME_AXIS = 0
+    # This extractor is based on a single example file without a formal specification from WhiteMatter.
+    # The parameters above are currently assumed to be constant for all WhiteMatter files.
+    # If you encounter issues with this extractor, these assumptions may need to be revisited.
+
+    def __init__(
+        self,
+        file_path: Union[str, Path],
+        sampling_frequency: float,
+        num_channels: int,
+        channel_ids: Optional[List] = None,
+        is_filtered: Optional[bool] = None,
+    ):
+        super().__init__(
+            file_paths=[file_path],
+            sampling_frequency=sampling_frequency,
+            num_channels=num_channels,
+            dtype=self.DTYPE,
+            time_axis=self.TIME_AXIS,
+            file_offset=self.FILE_OFFSET,
+            gain_to_uV=self.GAIN_TO_UV,
+            offset_to_uV=self.OFFSET_TO_UV,
+            is_filtered=is_filtered,
+            channel_ids=channel_ids,
+        )
+
+
+# Define function equivalent for convenience
+read_whitematter = define_function_from_class(source_class=WhiteMatterRecordingExtractor, name="read_whitematter")