Add append mode to IntanSplitFile building on #4070

doc/api.rst

@@ -105,6 +105,7 @@ NEO-based
     .. autofunction:: read_blackrock
     .. autofunction:: read_ced
     .. autofunction:: read_intan
+    .. autofunction:: read_split_intan_files
     .. autofunction:: read_maxwell
     .. autofunction:: read_mearec
     .. autofunction:: read_mcsraw

doc/modules/extractors.rst

@@ -127,6 +127,7 @@ For raw recording formats, we currently support:
 * **EDF** :py:func:`~spikeinterface.extractors.read_edf()`
 * **IBL streaming** :py:func:`~spikeinterface.extractors.read_ibl_recording()`
 * **Intan** :py:func:`~spikeinterface.extractors.read_intan()`
+* **Intan split files** :py:func:`~spikeinterface.extractors.read_split_intan_files()`
 * **MaxWell** :py:func:`~spikeinterface.extractors.read_maxwell()`
 * **MCS H5** :py:func:`~spikeinterface.extractors.read_mcsh5()`
 * **MCS RAW** :py:func:`~spikeinterface.extractors.read_mcsraw()`

src/spikeinterface/extractors/extractor_classes.py

@@ -20,6 +20,7 @@
 # sorting/recording/event from neo
 from .neoextractors import *
 from .neoextractors import read_neuroscope
+from .neoextractors.intan import read_split_intan_files, IntanSplitFilesRecordingExtractor
 
 # non-NEO objects implemented in neo folder
 # keep for reference Currently pulling from neoextractor __init__
@@ -200,5 +201,6 @@
         "read_binary",  # convenience function for binary formats
         "read_zarr",
         "read_neuroscope",  # convenience function for neuroscope
+        "read_split_intan_files",  # convenience function for segmented intan files
     ]
 )

src/spikeinterface/extractors/neoextractors/intan.py

@@ -1,9 +1,11 @@
 from __future__ import annotations
+from typing import Literal
 from pathlib import Path
 
 import numpy as np
 
 from spikeinterface.core.core_tools import define_function_from_class
+from spikeinterface.core.segmentutils import ConcatenateSegmentRecording, AppendSegmentRecording
 from .neobaseextractor import NeoBaseRecordingExtractor
 
 
@@ -104,3 +106,106 @@ def _add_channel_groups(self):
 
 
 read_intan = define_function_from_class(source_class=IntanRecordingExtractor, name="read_intan")
+
+
+class IntanSplitFilesRecordingExtractor(ConcatenateSegmentRecording, AppendSegmentRecording):
+    """
+    Class for reading Intan traditional format split files from a folder and
+    concatenating/appending them in temporal order.
+
+    Intan traditional format creates multiple files with time-based naming when recording
+    for extended periods. This class automatically sorts the files by filename and concatenates
+    them to create a continuous recording (monosegment) or appends them to a multisegment recording.
+
+    Parameters
+    ----------
+    folder_path : str or Path
+        Path to the folder containing split Intan files (.rhd or .rhs extensions)
+    mode : "concatenate" | "append": default: "concatenate"
+        The determines whether to concatenate intan files to make a monosegment or to append them
+        to make a multisegment recording
+    stream_id : str, default: None
+        If there are several streams, specify the stream id you want to load.
+    stream_name : str, default: None
+        If there are several streams, specify the stream name you want to load.
+    all_annotations : bool, default: False
+        Load exhaustively all annotations from neo.
+    use_names_as_ids : bool, default: False
+        Determines the format of the channel IDs used by the extractor. If set to True, the channel IDs will be the
+        names from NeoRawIO. If set to False, the channel IDs will be the ids provided by NeoRawIO.
+    ignore_integrity_checks : bool, default: False
+        If True, data that violates integrity assumptions will be loaded. At the moment the only integrity
+        check we perform is that timestamps are continuous. Setting this to True will ignore this check and set
+        the attribute `discontinuous_timestamps` to True in the underlying neo object.
+
+    Examples
+    --------
+    >>> from spikeinterface.extractors import IntanSplitFilesRecordingExtractor
+    >>> recording = IntanSplitFilesRecordingExtractor("/path/to/intan/folder")
+    """
+
+    def __init__(
+        self,
+        folder_path,
+        mode: Literal["append", "concatenate"] = "concatenate",
+        stream_id=None,
+        stream_name=None,
+        all_annotations=False,
+        use_names_as_ids=False,
+        ignore_integrity_checks: bool = False,
+    ):
+
+        if mode not in ("append", "concatenate"):
+            mode_error = (
+                "Possible options for the `mode` argument are 'concatenate' or 'append', you have entered " f"{mode}."
+            )
+            raise ValueError(mode_error)
+
+        folder_path = Path(folder_path)
+
+        if not folder_path.exists() or not folder_path.is_dir():
+            raise ValueError(f"Folder path {folder_path} does not exist or is not a directory")
+
+        # Find all Intan files
+        file_path_list = [p for p in folder_path.iterdir() if p.suffix.lower() in [".rhd", ".rhs"]]
+
+        if not file_path_list:
+            raise ValueError(f"No Intan files (.rhd or .rhs) found in {folder_path}")
+
+        # Sort files by filename (natural sort)
+        file_path_list.sort(key=lambda x: x.name)
+
+        # Read each file and create recording list
+        recording_list = []
+        for file_path in file_path_list:
+            recording = read_intan(
+                file_path,
+                stream_id=stream_id,
+                stream_name=stream_name,
+                all_annotations=all_annotations,
+                use_names_as_ids=use_names_as_ids,
+                ignore_integrity_checks=ignore_integrity_checks,
+            )
+            recording_list.append(recording)
+
+        # Initialize the parent class with the recording list
+        if mode == "concatenate":
+            ConcatenateSegmentRecording.__init__(self, recording_list)
+        elif mode == "append":
+            AppendSegmentRecording.__init__(self, recording_list)
+
+        # Update kwargs to include our specific parameters
+        self._kwargs = dict(
+            folder_path=str(Path(folder_path).resolve()),
+            mode=mode,
+            stream_id=stream_id,
+            stream_name=stream_name,
+            all_annotations=all_annotations,
+            use_names_as_ids=use_names_as_ids,
+            ignore_integrity_checks=ignore_integrity_checks,
+        )
+
+
+read_split_intan_files = define_function_from_class(
+    source_class=IntanSplitFilesRecordingExtractor, name="read_split_intan_files"
+)

src/spikeinterface/extractors/tests/test_neoextractors.py

@@ -43,6 +43,8 @@
     AxonRecordingExtractor,
 )
 
+from spikeinterface.extractors.neoextractors.intan import IntanSplitFilesRecordingExtractor
+
 from spikeinterface.extractors.extractor_classes import KiloSortSortingExtractor
 
 from spikeinterface.extractors.tests.common_tests import (
@@ -184,6 +186,15 @@ class IntanRecordingTestMultipleFilesFormat(RecordingCommonTestSuite, unittest.T
     ]
 
 
+class IntanSplitFilesRecordingTest(RecordingCommonTestSuite, unittest.TestCase):
+    ExtractorClass = IntanSplitFilesRecordingExtractor
+    downloads = ["intan"]
+    entities = [
+        ("intan/test_tetrode_240502_162925", {"mode": "concatenate"}),
+        ("intan/test_tetrode_240502_162925", {"mode": "append"}),
+    ]
+
+
 class NeuroScopeRecordingTest(RecordingCommonTestSuite, unittest.TestCase):
     ExtractorClass = NeuroScopeRecordingExtractor
     downloads = ["neuroscope"]