from pathlib import Path
from pydantic import FilePath
from pynwb import NWBFile
from ._utils import _warn_if_split_siblings_detected
from ....basedatainterface import BaseDataInterface
from ....utils import DeepDict, get_json_schema_from_method_signature
[docs]
class IntanStimInterface(BaseDataInterface):
"""
Data interface for converting electrical stimulation data from Intan .rhs files.
This interface handles the stimulation current channels recorded by the RHS2000
Stim/Recording system. Stimulation data is stored as current in Amperes, with one
stim channel per corresponding amplifier channel.
This interface is only compatible with .rhs files. For the main amplifier channels,
use :py:class:`~neuroconv.datainterfaces.ecephys.intan.intandatainterface.IntanRecordingInterface`.
For other analog streams (ADC, DC amplifier, auxiliary), use
:py:class:`~neuroconv.datainterfaces.ecephys.intan.intananaloginterface.IntanAnalogInterface`.
"""
display_name = "Intan Stimulation"
keywords = ("intan", "stimulation", "stim", "rhs", "current")
associated_suffixes = (".rhs",)
info = "Interface for converting Intan RHS electrical stimulation data."
[docs]
@classmethod
def get_source_schema(cls) -> dict:
source_schema = get_json_schema_from_method_signature(method=cls.__init__)
source_schema["properties"]["file_path"]["description"] = (
"Path to an Intan .rhs file. "
"When ``saved_files_are_split=True``, the file's parent directory is treated as the session "
"folder and all sibling .rhs files are concatenated in filename order."
)
return source_schema
def __init__(
self,
file_path: FilePath,
*,
verbose: bool = False,
metadata_key: str = "TimeSeriesIntanStim",
saved_files_are_split: bool = False,
):
"""
Load and prepare stimulation data from an Intan .rhs file.
Parameters
----------
file_path : FilePath
Path to an Intan .rhs file. Stimulation channels are exclusive to the
RHS Stim/Recording System and are not present in .rhd files. When
``saved_files_are_split=True``, this is any single file in the session folder;
its parent directory is scanned for siblings.
verbose : bool, default: False
Verbose output.
metadata_key : str, default: "TimeSeriesIntanStim"
Key for the TimeSeries metadata in the metadata dictionary.
saved_files_are_split : bool, default: False
Set to True when the recording was saved using Intan RHX's "new save file every N minutes"
option, producing several rotated ``.rhs`` files in one session folder. All sibling
files in ``file_path.parent`` are concatenated in filename order (Intan's default
``{prefix}_YYMMDD_HHMMSS`` naming makes lexicographic order match chronological order).
"""
self._file_path = Path(file_path)
self._stream_name = "Stim channel"
self.metadata_key = metadata_key
self._saved_files_are_split = saved_files_are_split
if saved_files_are_split:
from spikeinterface.extractors import read_split_intan_files
self.recording_extractor = read_split_intan_files(
folder_path=self._file_path.parent,
stream_name=self._stream_name,
all_annotations=True,
)
else:
from spikeinterface.extractors import read_intan
_warn_if_split_siblings_detected(self._file_path, interface_name="IntanStimInterface")
self.recording_extractor = read_intan(
file_path=self._file_path,
stream_name=self._stream_name,
all_annotations=True,
)
super().__init__(
file_path=self._file_path,
verbose=verbose,
)
[docs]
def get_channel_names(self) -> list[str]:
"""
Get a list of channel names from the stimulation recording.
Channel names follow the pattern ``{amplifier_channel}_STIM``
(e.g., ``A-000_STIM``), matching the corresponding amplifier channels.
Returns
-------
list of str
The names of all stimulation channels.
"""
return list(self.recording_extractor.get_channel_ids())
[docs]
def add_to_nwbfile(
self,
nwbfile: NWBFile,
metadata: dict | None = None,
*,
stub_test: bool = False,
iterator_type: str | None = "v2",
iterator_options: dict | None = None,
always_write_timestamps: bool = False,
):
"""
Add stimulation channel data to an NWB file.
Stimulation data are stored as a ``TimeSeries`` in acquisition with
``unit="A"`` (Amperes). The conversion factor is automatically derived
from the ``stim_step_size`` recorded in the .rhs file header.
Parameters
----------
nwbfile : NWBFile
The NWB file to which the stimulation data will be added.
metadata : dict, optional
Metadata dictionary. If None, uses default metadata from ``get_metadata()``.
stub_test : bool, default: False
If True, only writes a small amount of data for testing.
iterator_type : str, optional, default: "v2"
Type of iterator to use for data streaming.
iterator_options : dict, optional
Additional options for the iterator.
always_write_timestamps : bool, default: False
If True, always writes timestamps instead of using sampling rate.
"""
from ....tools.spikeinterface import (
_stub_recording,
add_recording_as_time_series_to_nwbfile,
)
if metadata is None:
metadata = self.get_metadata()
recording = self.recording_extractor
if stub_test:
recording = _stub_recording(recording=recording)
add_recording_as_time_series_to_nwbfile(
recording=recording,
nwbfile=nwbfile,
metadata=metadata,
iterator_type=iterator_type,
iterator_options=iterator_options,
always_write_timestamps=always_write_timestamps,
metadata_key=self.metadata_key,
parent_container="stimulus",
)