Add extractors for SiNAPS Research Platform

src/spikeinterface/extractors/extractorlist.py

@@ -45,6 +45,8 @@
 from .herdingspikesextractors import HerdingspikesSortingExtractor, read_herdingspikes
 from .mdaextractors import MdaRecordingExtractor, MdaSortingExtractor, read_mda_recording, read_mda_sorting
 from .phykilosortextractors import PhySortingExtractor, KiloSortSortingExtractor, read_phy, read_kilosort
+from .sinapsrecordingextractor import SinapsResearchPlatformRecordingExtractor, read_sinaps_research_platform
+from .sinapsrecordingh5extractor import SinapsResearchPlatformH5RecordingExtractor, read_sinaps_research_platform_h5
 
 # sorting in relation with simulator
 from .shybridextractors import (
@@ -77,6 +79,7 @@
     CompressedBinaryIblExtractor,
     IblRecordingExtractor,
     MCSH5RecordingExtractor,
+    SinapsResearchPlatformRecordingExtractor,
 ]
 recording_extractor_full_list += neo_recording_extractors_list
 

src/spikeinterface/extractors/sinapsrecordingextractor.py

@@ -0,0 +1,114 @@
+from pathlib import Path
+import numpy as np
+
+from probeinterface import get_probe
+
+from ..core import BinaryRecordingExtractor, ChannelSliceRecording
+from ..core.core_tools import define_function_from_class
+
+
+class SinapsResearchPlatformRecordingExtractor(ChannelSliceRecording):
+    extractor_name = "SinapsResearchPlatform"
+    mode = "file"
+    name = "sinaps_research_platform"
+
+    def __init__(self, file_path, stream_name="filt"):
+        from ..preprocessing import UnsignedToSignedRecording
+
+        file_path = Path(file_path)
+        meta_file = file_path.parent / f"metadata_{file_path.stem}.txt"
+        meta = parse_sinaps_meta(meta_file)
+
+        num_aux_channels = meta["nbHWAux"] + meta["numberUserAUX"]
+        num_total_channels = 2 * meta["nbElectrodes"] + num_aux_channels
+        num_electrodes = meta["nbElectrodes"]
+        sampling_frequency = meta["samplingFreq"]
+
+        probe_type = meta["probeType"]
+        # channel_locations = meta["electrodePhysicalPosition"] # will be depricated soon by Sam, switching to probeinterface
+        num_shanks = meta["nbShanks"]
+        num_electrodes_per_shank = meta["nbElectrodesShank"]
+        num_bits = int(np.log2(meta["nbADCLevels"]))
+
+        # channel_groups = []
+        # for i in range(num_shanks):
+        #     channel_groups.extend([i] * num_electrodes_per_shank)
+
+        gain_ephys = meta["voltageConverter"]
+        gain_aux = meta["voltageAUXConverter"]
+
+        recording = BinaryRecordingExtractor(
+            file_path, sampling_frequency, dtype="uint16", num_channels=num_total_channels
+        )
+        recording = UnsignedToSignedRecording(recording, bit_depth=num_bits)
+
+        if stream_name == "raw":
+            channel_slice = recording.channel_ids[:num_electrodes]
+            renamed_channels = np.arange(num_electrodes)
+            # locations = channel_locations
+            # groups = channel_groups
+            gain = gain_ephys
+        elif stream_name == "filt":
+            channel_slice = recording.channel_ids[num_electrodes : 2 * num_electrodes]
+            renamed_channels = np.arange(num_electrodes)
+            # locations = channel_locations
+            # groups = channel_groups
+            gain = gain_ephys
+        elif stream_name == "aux":
+            channel_slice = recording.channel_ids[2 * num_electrodes :]
+            hw_chans = meta["hwAUXChannelName"][1:-1].split(",")
+            user_chans = meta["userAuxName"][1:-1].split(",")
+            renamed_channels = hw_chans + user_chans
+            # locations = None
+            # groups = None
+            gain = gain_aux
+        else:
+            raise ValueError("stream_name must be 'raw', 'filt', or 'aux'")
+
+        ChannelSliceRecording.__init__(self, recording, channel_ids=channel_slice, renamed_channel_ids=renamed_channels)
+        # if locations is not None:
+        # self.set_channel_locations(locations)
+        # if groups is not None:
+        # self.set_channel_groups(groups)
+
+        self.set_channel_gains(gain)
+        self.set_channel_offsets(0)
+
+        if (stream_name == "filt") | (stream_name == "raw"):
+            if probe_type == "p1024s1NHP":
+                probe = get_probe(manufacturer="sinaps", probe_name="SiNAPS-p1024s1NHP")
+                # now wire the probe
+                channel_indices = np.arange(1024)
+                probe.set_device_channel_indices(channel_indices)
+                self.set_probe(probe, in_place=True)
+            else:
+                raise ValueError(f"Unknown probe type: {probe_type}")
+
+        self._kwargs = {"file_path": str(file_path.absolute())}
+
+
+read_sinaps_research_platform = define_function_from_class(
+    source_class=SinapsResearchPlatformRecordingExtractor, name="read_sinaps_research_platform"
+)
+
+
+def parse_sinaps_meta(meta_file):
+    meta_dict = {}
+    with open(meta_file) as f:
+        lines = f.readlines()
+        for l in lines:
+            if "**" in l or "=" not in l:
+                continue
+            else:
+                key, val = l.split("=")
+                val = val.replace("\n", "")
+                try:
+                    val = int(val)
+                except:
+                    pass
+                try:
+                    val = eval(val)
+                except:
+                    pass
+                meta_dict[key] = val
+    return meta_dict

src/spikeinterface/extractors/sinapsrecordingh5extractor.py

@@ -0,0 +1,149 @@
+from pathlib import Path
+import numpy as np
+
+from probeinterface import get_probe
+
+from ..core.core_tools import define_function_from_class
+from ..core import BaseRecording, BaseRecordingSegment
+from ..preprocessing import UnsignedToSignedRecording
+
+
+class SinapsResearchPlatformH5RecordingExtractor_Unsigned(BaseRecording):
+    extractor_name = "SinapsResearchPlatformH5"
+    mode = "file"
+    name = "sinaps_research_platform_h5"
+
+    def __init__(self, file_path):
+
+        try:
+            import h5py
+
+            self.installed = True
+        except ImportError:
+            self.installed = False
+
+        assert self.installed, self.installation_mesg
+        self._file_path = file_path
+
+        sinaps_info = openSiNAPSFile(self._file_path)
+        self._rf = sinaps_info["filehandle"]
+
+        BaseRecording.__init__(
+            self,
+            sampling_frequency=sinaps_info["sampling_frequency"],
+            channel_ids=sinaps_info["channel_ids"],
+            dtype=sinaps_info["dtype"],
+        )
+
+        self.extra_requirements.append("h5py")
+
+        recording_segment = SiNAPSRecordingSegment(
+            self._rf, sinaps_info["num_frames"], sampling_frequency=sinaps_info["sampling_frequency"]
+        )
+        self.add_recording_segment(recording_segment)
+
+        # set gain
+        self.set_channel_gains(sinaps_info["gain"])
+        self.set_channel_offsets(sinaps_info["offset"])
+        self.num_bits = sinaps_info["num_bits"]
+
+        # set probe
+        if sinaps_info["probe_type"] == "p1024s1NHP":
+            probe = get_probe(manufacturer="sinaps", probe_name="SiNAPS-p1024s1NHP")
+            probe.set_device_channel_indices(np.arange(1024))
+            self.set_probe(probe, in_place=True)
+        else:
+            raise ValueError(f"Unknown probe type: {sinaps_info['probe_type']}")
+
+        # set other properties
+
+        self._kwargs = {"file_path": str(Path(file_path).absolute())}
+
+    def __del__(self):
+        self._rf.close()
+
+
+class SiNAPSRecordingSegment(BaseRecordingSegment):
+    def __init__(self, rf, num_frames, sampling_frequency):
+        BaseRecordingSegment.__init__(self, sampling_frequency=sampling_frequency)
+        self._rf = rf
+        self._num_samples = int(num_frames)
+        self._stream = self._rf.require_group("RealTimeProcessedData")
+
+    def get_num_samples(self):
+        return self._num_samples
+
+    def get_traces(self, start_frame=None, end_frame=None, channel_indices=None):
+        if isinstance(channel_indices, slice):
+            traces = self._stream.get("FilteredData")[channel_indices, start_frame:end_frame].T
+        else:
+            # channel_indices is np.ndarray
+            if np.array(channel_indices).size > 1 and np.any(np.diff(channel_indices) < 0):
+                # get around h5py constraint that it does not allow datasets
+                # to be indexed out of order
+                sorted_channel_indices = np.sort(channel_indices)
+                resorted_indices = np.array([list(sorted_channel_indices).index(ch) for ch in channel_indices])
+                recordings = self._stream.get("FilteredData")[sorted_channel_indices, start_frame:end_frame].T
+                traces = recordings[:, resorted_indices]
+            else:
+                traces = self._stream.get("FilteredData")[channel_indices, start_frame:end_frame].T
+        return traces
+
+
+class SinapsResearchPlatformH5RecordingExtractor(UnsignedToSignedRecording):
+    extractor_name = "SinapsResearchPlatformH5"
+    mode = "file"
+    name = "sinaps_research_platform_h5"
+
+    def __init__(self, file_path):
+        recording = SinapsResearchPlatformH5RecordingExtractor_Unsigned(file_path)
+        UnsignedToSignedRecording.__init__(self, recording, bit_depth=recording.num_bits)
+
+        self._kwargs = {"file_path": str(Path(file_path).absolute())}
+
+
+read_sinaps_research_platform_h5 = define_function_from_class(
+    source_class=SinapsResearchPlatformH5RecordingExtractor, name="read_sinaps_research_platform_h5"
+)
+
+
+def openSiNAPSFile(filename):
+    """Open an SiNAPS hdf5 file, read and return the recording info."""
+
+    import h5py
+
+    rf = h5py.File(filename, "r")
+
+    stream = rf.require_group("RealTimeProcessedData")
+    data = stream.get("FilteredData")
+    dtype = data.dtype
+
+    parameters = rf.require_group("Parameters")
+    gain = parameters.get("VoltageConverter")[0]
+    offset = 0
+
+    nRecCh, nFrames = data.shape
+
+    samplingRate = parameters.get("SamplingFrequency")[0]
+
+    probe_type = str(
+        rf.require_group("Advanced Recording Parameters").require_group("Probe").get("probeType").asstr()[...]
+    )
+    num_bits = int(
+        np.log2(rf.require_group("Advanced Recording Parameters").require_group("DAQ").get("nbADCLevels")[0])
+    )
+
+    sinaps_info = {
+        "filehandle": rf,
+        "num_frames": nFrames,
+        "sampling_frequency": samplingRate,
+        "num_channels": nRecCh,
+        "channel_ids": np.arange(nRecCh),
+        "gain": gain,
+        "offset": offset,
+        "dtype": dtype,
+        "probe_type": probe_type,
+        "num_bits": num_bits,
+    }
+
+    return sinaps_info