Source code for neuroconv.datainterfaces.ecephys.intan.intandatainterface

from pathlib import Path

from pydantic import FilePath
from pynwb.ecephys import ElectricalSeries

from ..baserecordingextractorinterface import BaseRecordingExtractorInterface
from ....utils import DeepDict, get_schema_from_hdmf_class




class IntanRecordingInterface(BaseRecordingExtractorInterface):
    """
    Primary data interface for converting Intan amplifier data from .rhd or .rhs files.

    This interface is used for data that comes from the RHD2000/RHS2000 amplifier channels,
    which are the primary neural recording channels.

    If you have other data streams from your Intan system (e.g., analog inputs, auxiliary inputs, DC amplifiers),
    you should use the :py:class:`~neuroconv.datainterfaces.ecephys.intan.intananaloginterface.IntanAnalogInterface`.
    """

    display_name = "Intan Amplifier"
    keywords = ("intan", "amplifier", "rhd", "rhs", "extracellular electrophysiology", "recording")
    associated_suffixes = (".rhd", ".rhs")
    info = "Interface for converting Intan amplifier data."
    stream_id = "0"  # This are the amplifier channels, corresponding to the stream_name 'RHD2000 amplifier channel'



    @classmethod
    def get_source_schema(cls) -> dict:
        source_schema = super().get_source_schema()
        source_schema["properties"]["file_path"]["description"] = "Path to either a .rhd or a .rhs file"
        return source_schema




    @classmethod
    def get_extractor_class(cls):
        from spikeinterface.extractors.extractor_classes import IntanRecordingExtractor

        return IntanRecordingExtractor


    def _initialize_extractor(self, interface_kwargs: dict):
        """Override to add stream_id"""
        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
        self.extractor_kwargs["stream_id"] = self.stream_id

        extractor_class = self.get_extractor_class()
        extractor_instance = extractor_class(**self.extractor_kwargs)
        return extractor_instance

    def __init__(
        self,
        file_path: FilePath,
        verbose: bool = False,
        es_key: str = "ElectricalSeries",
        ignore_integrity_checks: bool = False,
    ):
        """
        Load and prepare raw data and corresponding metadata from the Intan format (.rhd or .rhs files).

        Parameters
        ----------
        file_path : FilePath
            Path to either a rhd or a rhs file

        verbose : bool, default: False
            Verbose
        es_key : str, default: "ElectricalSeries"
        ignore_integrity_checks, bool, default: False.
            If True, data that violates integrity assumptions will be loaded. At the moment the only integrity
            check performed is that timestamps are continuous. If False, an error will be raised if the check fails.
        """

        self.file_path = Path(file_path)

        init_kwargs = dict(
            file_path=self.file_path,
            verbose=verbose,
            es_key=es_key,
            ignore_integrity_checks=ignore_integrity_checks,
        )

        super().__init__(**init_kwargs)



    def get_metadata_schema(self) -> dict:
        metadata_schema = super().get_metadata_schema()
        metadata_schema["properties"]["Ecephys"]["properties"].update(
            ElectricalSeriesRaw=get_schema_from_hdmf_class(ElectricalSeries)
        )
        return metadata_schema




    def get_metadata(self) -> DeepDict:
        metadata = super().get_metadata()
        ecephys_metadata = metadata["Ecephys"]

        # Add device

        system = self.file_path.suffix  # .rhd or .rhs
        device_description = {".rhd": "RHD Recording System", ".rhs": "RHS Stim/Recording System"}[system]

        intan_device = dict(
            name="Intan",
            description=device_description,
            manufacturer="Intan",
        )
        device_list = [intan_device]
        ecephys_metadata.update(Device=device_list)

        electrode_group_metadata = ecephys_metadata["ElectrodeGroup"]
        for electrode_group in electrode_group_metadata:
            electrode_group["device"] = intan_device["name"]
        # Add electrodes and electrode groups
        ecephys_metadata.update(
            ElectricalSeriesRaw=dict(name="ElectricalSeriesRaw", description="Raw acquisition traces."),
        )

        return metadata