Add Plexon LFP interface (#1209)

CHANGELOG.md

@@ -11,6 +11,7 @@
 * `DatasetIOConfiguration` now recommends `chunk_shape = (len(candidate_dataset),)` for datasets with compound dtypes [PR #1146](https://github.com/catalystneuro/neuroconv/pull/1146)
 
 ## Features
+* Added `PlexonLFPInterface` for converting Plexon `FPl-Low Pass Filtered` stream data [#1209](https://github.com/catalystneuro/neuroconv/pull/1209)
 * Use the latest version of ndx-pose for `DeepLabCutInterface` and `LightningPoseDataInterface` [PR #1128](https://github.com/catalystneuro/neuroconv/pull/1128)
 * Added `ImageInterface` for writing large collection of images to NWB and automatically map the images to the correct NWB data types [PR #1190](https://github.com/catalystneuro/neuroconv/pull/1190)
 as used by hdmf >= 3.14.6.

src/neuroconv/datainterfaces/__init__.py

@@ -56,6 +56,7 @@
 from .ecephys.plexon.plexondatainterface import (
     Plexon2RecordingInterface,
     PlexonRecordingInterface,
+    PlexonLFPInterface,
     PlexonSortingInterface,
 )
 from .ecephys.spike2.spike2datainterface import Spike2RecordingInterface
@@ -129,6 +130,7 @@
     EDFRecordingInterface,
     TdtRecordingInterface,
     PlexonRecordingInterface,
+    PlexonLFPInterface,
     Plexon2RecordingInterface,
     PlexonSortingInterface,
     BiocamRecordingInterface,

src/neuroconv/datainterfaces/ecephys/plexon/plexondatainterface.py

@@ -2,6 +2,7 @@
 
 from pydantic import FilePath, validate_call
 
+from ..baselfpextractorinterface import BaseLFPExtractorInterface
 from ..baserecordingextractorinterface import BaseRecordingExtractorInterface
 from ..basesortingextractorinterface import BaseSortingExtractorInterface
 from ....utils import DeepDict
@@ -65,6 +66,69 @@ def get_metadata(self) -> DeepDict:
         return metadata
 
 
+class PlexonLFPInterface(BaseLFPExtractorInterface):
+    """
+    Primary data interface class for converting Plexon LFP data.
+
+    Uses the :py:class:`~spikeinterface.extractors.PlexonRecordingExtractor`.
+    """
+
+    display_name = "Plexon LFP Recording"
+    associated_suffixes = (".plx",)
+    info = "Interface for Plexon low pass filtered data."
+    ExtractorName = "PlexonRecordingExtractor"
+
+    @classmethod
+    def get_source_schema(cls) -> dict:
+        source_schema = super().get_source_schema()
+        source_schema["properties"]["file_path"]["description"] = "Path to the .plx file."
+        return source_schema
+
+    @validate_call
+    def __init__(
+        self,
+        file_path: FilePath,
+        verbose: bool = False,
+        es_key: str = "ElectricalSeriesLF",
+        stream_name: str = "FPl-Low Pass Filtered",
+    ):
+        """
+        Load and prepare data for Plexon.
+
+        Parameters
+        ----------
+        file_path : str or Path
+            Path to the .plx file.
+        verbose : bool, default: False
+            Allows verbosity.
+        es_key : str, default: "ElectricalSeries"
+        stream_name: str, default: "FPl-Low Pass Filtered""
+            Only pass a stream if you modified the channel prefixes in the Plexon file and you know the prefix of
+            the FPllow pass filtered data.
+        """
+
+        # By default, the stream name of Plexon LFP data is "FPl-Low Pass Filter"
+        # But the user might modify it. For that case, we exclude the default stream names
+        # Of the other streams in the Plexon file.
+
+        invalid_stream_names = ["WB-Wideband", "SPKC-High Pass Filtered", "AI-Auxiliary Input"]
+        assert stream_name not in invalid_stream_names, f"Invalid stream name: {stream_name}"
+
+        super().__init__(file_path=file_path, verbose=verbose, es_key=es_key, stream_name=stream_name)
+
+    def get_metadata(self) -> DeepDict:
+        metadata = super().get_metadata()
+        neo_reader = self.recording_extractor.neo_reader
+
+        if hasattr(neo_reader, "raw_annotations"):
+            block_ind = self.recording_extractor.block_index
+            neo_metadata = neo_reader.raw_annotations["blocks"][block_ind]
+            if "rec_datetime" in neo_metadata:
+                metadata["NWBFile"].update(session_start_time=neo_metadata["rec_datetime"])
+
+        return metadata
+
+
 class Plexon2RecordingInterface(BaseRecordingExtractorInterface):
     """
     Primary data interface class for converting Plexon2 data.

src/neuroconv/tools/testing/data_interface_mixins.py

@@ -392,6 +392,7 @@ class RecordingExtractorInterfaceTestMixin(DataInterfaceTestMixin, TemporalAlign
     """
 
     data_interface_cls: Type[BaseRecordingExtractorInterface]
+    is_lfp_interface: bool = False
 
     def check_read_nwb(self, nwbfile_path: str):
         from spikeinterface.extractors import NwbRecordingExtractor
@@ -402,9 +403,15 @@ def check_read_nwb(self, nwbfile_path: str):
 
         if recording.get_num_segments() == 1:
             # Spikeinterface behavior is to load the electrode table channel_name property as a channel_id
+
+            if self.is_lfp_interface:
+                electrical_series_path = f"processing/ecephys/LFP/{electrical_series_name}"
+            else:
+                electrical_series_path = f"acquisition/{electrical_series_name}"
+
             self.nwb_recording = NwbRecordingExtractor(
                 file_path=nwbfile_path,
-                electrical_series_path=f"acquisition/{electrical_series_name}",
+                electrical_series_path=electrical_series_path,
                 use_pynwb=True,
             )
 

tests/test_on_data/ecephys/test_recording_interfaces.py

@@ -27,6 +27,7 @@
     OpenEphysLegacyRecordingInterface,
     OpenEphysRecordingInterface,
     Plexon2RecordingInterface,
+    PlexonLFPInterface,
     PlexonRecordingInterface,
     Spike2RecordingInterface,
     SpikeGadgetsRecordingInterface,
@@ -746,6 +747,15 @@ def check_extracted_metadata(self, metadata: dict):
         assert metadata["NWBFile"]["session_start_time"] == datetime(2013, 11, 19, 13, 48, 13)
 
 
+class TestPlexonLFPInterface(RecordingExtractorInterfaceTestMixin):
+    data_interface_cls = PlexonLFPInterface
+    interface_kwargs = dict(
+        file_path=str(ECEPHY_DATA_PATH / "plexon" / "4chDemoPLX.plx"),
+    )
+    save_directory = OUTPUT_PATH
+    is_lfp_interface = True
+
+
 def is_macos():
     import platform