import warnings
from pathlib import Path
from pydantic import FilePath
from pynwb import NWBFile
from ....basedatainterface import BaseDataInterface
from ....utils import DeepDict, get_json_schema_from_method_signature
[docs]
class IntanAnalogInterface(BaseDataInterface):
"""
Primary data interface for converting non-amplifier analog data streams from Intan .rhd or .rhs files.
This interface handles several types of analog signals that are not the primary neural recording channels,
including auxiliary inputs, ADC inputs, and DC amplifier signals.
If your data consists of the main amplifier channels (neural data), you should use the
:py:class:`~neuroconv.datainterfaces.ecephys.intan.intandatainterface.IntanRecordingInterface`.
"""
display_name = "Intan Analog"
keywords = ("intan", "analog", "auxiliary", "ADC", "DC amplifier", "rhd", "rhs")
associated_suffixes = (".rhd", ".rhs")
info = "Interface for converting Intan non-amplifier analog 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 either a .rhd or a .rhs file"
return source_schema
def __init__(
self,
file_path: FilePath,
*args, # TODO: change to * (keyword only) on or after August 2026
stream_name: str,
verbose: bool = False,
metadata_key: str = "TimeSeriesAnalogIntan",
):
"""
Load and prepare analog data from Intan format (.rhd or .rhs files).
Parameters
----------
file_path : FilePath
Path to either a rhd or a rhs file
stream_name : str
The stream name to load. Valid options include:
- "RHD2000 auxiliary input channel": Auxiliary input channels (e.g., accelerometer data)
- "RHD2000 supply voltage channel": Supply voltage channels
- "USB board ADC input channel": ADC input channels (analog signals -10V to +10V)
- "USB board ADC output channel": ADC output channels (analog signals -10V to +10V)
- "DC Amplifier channel": DC amplifier channels (RHS system only)
verbose : bool, default: False
Verbose output
metadata_key : str, default: "TimeSeriesAnalogIntan"
Key for the TimeSeries metadata in the metadata dictionary.
"""
# Handle deprecated positional arguments
if args:
parameter_names = [
"stream_name",
"verbose",
"metadata_key",
]
num_positional_args_before_args = 1 # file_path
if len(args) > len(parameter_names):
raise TypeError(
f"__init__() takes at most {len(parameter_names) + num_positional_args_before_args + 1} positional arguments but "
f"{len(args) + num_positional_args_before_args + 1} were given. "
"Note: Positional arguments are deprecated and will be removed on or after August 2026. "
"Please use keyword arguments."
)
positional_values = dict(zip(parameter_names, args))
passed_as_positional = list(positional_values.keys())
warnings.warn(
f"Passing arguments positionally to IntanAnalogInterface.__init__() is deprecated "
f"and will be removed on or after August 2026. "
f"The following arguments were passed positionally: {passed_as_positional}. "
"Please use keyword arguments instead.",
FutureWarning,
stacklevel=2,
)
stream_name = positional_values.get("stream_name", stream_name)
verbose = positional_values.get("verbose", verbose)
metadata_key = positional_values.get("metadata_key", metadata_key)
from spikeinterface.extractors import read_intan
self._file_path = Path(file_path)
self._stream_name = stream_name
# Stream type descriptions and time series name mapping
self.stream_info = {
"RHD2000 auxiliary input channel": {
"description": "RHD2000 auxiliary input channels (e.g., accelerometer data)",
"time_series_name": "TimeSeriesIntanAuxiliary",
},
"RHD2000 supply voltage channel": {
"description": "RHD2000 supply voltage channels",
"time_series_name": "TimeSeriesIntanSupplyVoltage",
},
"USB board ADC input channel": {
"description": "USB board ADC input channels (analog signals -10V to +10V)",
"time_series_name": "TimeSeriesIntanADCInput",
},
"USB board ADC output channel": {
"description": "USB board ADC output channels (analog signals -10V to +10V)",
"time_series_name": "TimeSeriesIntanADCOutput",
},
"DC Amplifier channel": {
"description": "DC amplifier channels (RHS system)",
"time_series_name": "TimeSeriesIntanDC",
},
}
# Validate stream_name
if self._stream_name not in self.stream_info:
raise ValueError(
f"Invalid stream_name '{self._stream_name}'. "
f"Valid analog stream names are: {list(self.stream_info.keys())}"
)
# Set time_series_name from stream info and metadata_key from parameter
self._time_series_name = self.stream_info[self._stream_name]["time_series_name"]
self.metadata_key = metadata_key
# Load the recording extractor using stream_name
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,
stream_name=self._stream_name,
verbose=verbose,
)
[docs]
def get_channel_names(self) -> list[str]:
"""
Get a list of channel names from the recording extractor.
Returns
-------
list of str
The names of all channels in the analog recording.
"""
return list(self.recording_extractor.get_channel_ids())
[docs]
def add_to_nwbfile(
self,
nwbfile: NWBFile,
metadata: dict | None = None,
*args, # TODO: change to * (keyword only) on or after August 2026
stub_test: bool = False,
iterator_type: str | None = "v2",
iterator_options: dict | None = None,
iterator_opts: dict | None = None,
always_write_timestamps: bool = False,
):
"""
Add analog channel data to an NWB file.
Parameters
----------
nwbfile : NWBFile
The NWB file to which the analog data will be added
metadata : dict, optional
Metadata dictionary with device information. If None, uses default 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
iterator_opts : dict, optional
Deprecated. Use 'iterator_options' instead.
always_write_timestamps : bool, default: False
If True, always writes timestamps instead of using sampling rate
"""
# Handle deprecated positional arguments
if args:
parameter_names = [
"stub_test",
"iterator_type",
"iterator_options",
"iterator_opts",
"always_write_timestamps",
]
num_positional_args_before_args = 2 # nwbfile, metadata
if len(args) > len(parameter_names):
raise TypeError(
f"add_to_nwbfile() takes at most {len(parameter_names) + num_positional_args_before_args} positional arguments but "
f"{len(args) + num_positional_args_before_args} were given. "
"Note: Positional arguments are deprecated and will be removed on or after August 2026. "
"Please use keyword arguments."
)
positional_values = dict(zip(parameter_names, args))
passed_as_positional = list(positional_values.keys())
warnings.warn(
f"Passing arguments positionally to IntanAnalogInterface.add_to_nwbfile() is deprecated "
f"and will be removed on or after August 2026. "
f"The following arguments were passed positionally: {passed_as_positional}. "
"Please use keyword arguments instead.",
FutureWarning,
stacklevel=2,
)
stub_test = positional_values.get("stub_test", stub_test)
iterator_type = positional_values.get("iterator_type", iterator_type)
iterator_options = positional_values.get("iterator_options", iterator_options)
iterator_opts = positional_values.get("iterator_opts", iterator_opts)
always_write_timestamps = positional_values.get("always_write_timestamps", always_write_timestamps)
from ....tools.spikeinterface import (
_stub_recording,
add_recording_as_time_series_to_nwbfile,
)
# Handle deprecated iterator_opts parameter
if iterator_opts is not None:
warnings.warn(
"The 'iterator_opts' parameter is deprecated and will be removed in May 2026 or after. "
"Use 'iterator_options' instead.",
FutureWarning,
stacklevel=2,
)
if iterator_options is not None:
raise ValueError("Cannot specify both 'iterator_opts' and 'iterator_options'. Use 'iterator_options'.")
iterator_options = iterator_opts
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,
)