from pathlib import Path
from pydantic import FilePath, validate_call
from ..baselfpextractorinterface import BaseLFPExtractorInterface
from ..baserecordingextractorinterface import BaseRecordingExtractorInterface
from ..basesortingextractorinterface import BaseSortingExtractorInterface
from ....utils import DeepDict
[docs]
class PlexonRecordingInterface(BaseRecordingExtractorInterface):
"""
Primary data interface class for converting Plexon data.
Uses :py:func:`~spikeinterface.extractors.read_plexon` from SpikeInterface.
"""
display_name = "Plexon Recording"
associated_suffixes = (".plx",)
info = "Interface for Plexon recording data."
[docs]
@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 = "ElectricalSeries",
stream_name: str = "WB-Wideband",
):
"""
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, optional
Only pass a stream if you modified the channel prefixes in the Plexon file and you know the prefix of
the wideband data.
"""
invalid_stream_names = ["FPl-Low Pass Filtered", "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)
[docs]
class PlexonLFPInterface(BaseLFPExtractorInterface):
"""
Primary data interface class for converting Plexon LFP data.
Uses :py:func:`~spikeinterface.extractors.read_plexon`.
"""
display_name = "Plexon LFP Recording"
associated_suffixes = (".plx",)
info = "Interface for Plexon low pass filtered data."
[docs]
@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)
[docs]
class Plexon2RecordingInterface(BaseRecordingExtractorInterface):
"""
Primary data interface class for converting Plexon2 data.
Uses :py:func:`~spikeinterface.extractors.read_plexon2` from SpikeInterface.
"""
display_name = "Plexon2 Recording"
associated_suffixes = (".pl2",)
info = "Interface for Plexon2 recording data."
[docs]
@classmethod
def get_source_schema(cls) -> dict:
source_schema = super().get_source_schema()
source_schema["properties"]["file_path"]["description"] = "Path to the .pl2 file."
return source_schema
def _initialize_extractor(self, interface_kwargs: dict):
"""Override to add stream_id parameter."""
self.extractor_kwargs = interface_kwargs.copy()
self.extractor_kwargs.pop("verbose", None)
self.extractor_kwargs.pop("es_key", None)
self.extractor_kwargs["all_annotations"] = True # Handled by base class now
self.extractor_kwargs["stream_id"] = self.stream_id
extractor_class = self.get_extractor_class()
extractor_instance = extractor_class(**self.extractor_kwargs)
return extractor_instance
@validate_call
def __init__(self, file_path: FilePath, verbose: bool = False, es_key: str = "ElectricalSeries"):
"""
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"
"""
# TODO: when neo version 0.14.4 is out or higher change this to stream_name for clarify
import neo
from packaging.version import Version
neo_version = Version(neo.__version__)
if neo_version <= Version("0.13.3"):
self.stream_id = "3"
else:
self.stream_id = "WB"
assert Path(file_path).is_file(), f"Plexon file not found in: {file_path}"
super().__init__(
file_path=file_path,
verbose=verbose,
es_key=es_key,
)
[docs]
class PlexonSortingInterface(BaseSortingExtractorInterface):
"""
Primary data interface class for converting Plexon spiking data.
Uses :py:func:`~spikeinterface.extractors.read_plexon_sorting` from SpikeInterface.
"""
display_name = "Plexon Sorting"
associated_suffixes = (".plx",)
info = "Interface for Plexon sorting data."
[docs]
@classmethod
def get_source_schema(cls) -> dict:
source_schema = super().get_source_schema()
source_schema["properties"]["file_path"]["description"] = "Path to the plexon spiking data (.plx file)."
return source_schema
@validate_call
def __init__(self, file_path: FilePath, verbose: bool = False):
"""
Load and prepare data for Plexon.
Parameters
----------
file_path: FilePath
Path to the plexon spiking data (.plx file).
verbose: bool, default: True
Allows verbosity.
"""
super().__init__(file_path=file_path, verbose=verbose)