intan.io package

This module contains all the I/O utilities for loading, saving, and managing Intan files.

Submodules

canonical_label(s, *, style='lower', allow_fuzzy=True, fuzzy_cutoff=0.88, strict=False)

Canonicalize a raw label string to your official namespace. :rtype: str

• style=”camel”: return CamelCase (e.g., “PinkyFlexion”)

• style=”lower”: return canonical token key (e.g., “pinkyflexion”)

Parameters:

• s (str)

• style (str)

• allow_fuzzy (bool)

• fuzzy_cutoff (float)

• strict (bool)

Return type:

str

load_csv_file(path=None, sample_rate=1000.0, export_basename=None, export_basepath=None, return_dataframe=False)

Load a CSV containing EMG channels (EMG_0 … EMG_N) and optional IMU columns.

Parameters:

• path (str or None) – CSV path. If None, opens a file dialog (like other loaders).

• sample_rate (float) – Fallback EMG sampling rate (Hz) if we cannot infer from a time column.

• export_basename (str or None) – Included in the result to match the RHD loader contract.

• export_basepath (str or None) – Included in the result to match the RHD loader contract.

• return_dataframe (bool) – If True, include the original pandas DataFrame as result[‘dataframe’].

Returns:

A dictionary aligned to the structure of intan.io.load_rhd_file, with keys: - ‘amplifier_data’ : np.ndarray (n_channels, n_samples) float32 - ‘amplifier_channels’ : list[dict] with ‘native_channel_name’, etc. - ‘channel_names’ : list[str] of EMG column names - ‘frequency_parameters’: {‘amplifier_sample_rate’: float} - ‘t_amplifier’ : np.ndarray (n_samples,) seconds - ‘board_adc_data’ : np.ndarray for IMU signals (if present) shape (n_aux, n_samples) - ‘board_adc_channels’ : list[dict] for IMU channel descriptors - ‘export_basename’, ‘export_basepath’ - ‘source_path’ : original file path - ‘meta’ : dict with lightweight details (column maps, etc.) - optionally ‘dataframe’: the raw pandas DataFrame (if return_dataframe=True)

Return type:

dict

find_csv_dir(root)

Return type:

str

Parameters:

root (str)

load_csv_files(root_dir, csv_sample_rate=1000.0, verbose=False)

Parameters:

• root_dir (str)

• csv_sample_rate (float)

• verbose (bool)

intan.io._exceptions

Custom exception classes used throughout the intan package.

These exceptions provide clearer debugging information for: - Invalid or corrupted .rhd files - Mismatched channel definitions - Broken TCP streams from the RHX server - File size or format inconsistencies

Each exception inherits from Exception and includes a brief description.

exception UnrecognizedFileError

Bases: Exception

Exception returned when reading a file as an RHD header yields an invalid magic number (indicating this is not an RHD header file).

exception UnknownChannelTypeError

Bases: Exception

Exception returned when a channel field in RHD header does not have a recognized signal_type value. Accepted values are: 0: amplifier channel 1: aux input channel 2: supply voltage channel 3: board adc channel 4: dig in channel 5: dig out channel

exception FileSizeError

Bases: Exception

Exception returned when file reading fails due to the file size being invalid or the calculated file size differing from the actual file size.

exception QStringError

Bases: Exception

Exception returned when reading a QString fails because it is too long.

exception ChannelNotFoundError

Bases: Exception

Exception returned when plotting fails due to the specified channel not being found.

exception GetSampleRateFailure

Bases: Exception

Exception returned when the TCP socket failed to yield the sample rate as reported by the RHX software.

exception InvalidReceivedDataSize

Bases: Exception

Exception returned when the amount of data received on the TCP socket is not an integer multiple of the excepted data block size.

exception InvalidMagicNumber

Bases: Exception

Exception returned when the first 4 bytes of a data block are not the expected RHX TCP magic number (0x2ef07a08).

intan.io._file_utils

Utility functions for file and path handling in the Intan interface.

This module includes helpers for: - Cross-platform path adjustment (Windows, WSL, Linux) - File presence validation - Reading configuration or labeled trial files - End-of-file checks and error handling - Progress bar display during file loading - Directory scanning for .rhd datasets

Used throughout the intan.io submodule to support flexible loading and validation of data from local or mounted environments.

adjust_path(path)

Adjusts a file path for compatibility with the host operating system.

Automatically detects WSL, Linux, or native Windows and transforms file paths accordingly (e.g., C:` to `/mnt/c/ on WSL).

Parameters:

path (str) – Original file path (Windows-style or POSIX)

Returns:

Transformed path compatible with the current OS.

Return type:

str

check_file_present(file, metrics_file, verbose=False)

Check whether a file is listed in a given metrics CSV.

Parameters:

• file (str) – File path to check.

• metrics_file (pd.DataFrame) – Loaded CSV with a ‘File Name’ column.

• verbose (bool) – If True, print a warning when file is missing.

Returns:

(filename, is_present) where is_present is True if the file is found.

Return type:

tuple

check_end_of_file(filesize, fid)

Validate that the file pointer has reached the end of the file.

Raises:

FileSizeError – If unread bytes remain in the stream.

print_progress(i, target, print_step, percent_done, bar_length=40)

Print an ASCII progress bar in the terminal.

Parameters:

• i (int) – Current iteration index.

• target (int) – Total number of iterations.

• print_step (float) – Update frequency as a percentage.

• percent_done (float) – Last percentage printed (used to avoid overprinting).

• bar_length (int) – Length of the ASCII progress bar.

Returns:

Updated percent_done value.

Return type:

float

read_config_file(config_file)

Parse a simple key=value style configuration file (e.g. TRUECONFIG.txt).

Parameters:

config_file (str) – Path to the config file.

Returns:

Dictionary of key-value settings.

Return type:

dict

get_file_paths(directory, file_type=None, verbose=False)

Scan a directory for files or folders, with optional filtering by extension.

Parameters:

• directory (str) – Root folder to search.

• file_type (str or None) – Extension to search for (e.g. ‘.rhd’). If None, returns subfolders.

• verbose (bool) – If True, print the number of items found.

Returns:

List of matching files or folders.

Return type:

list[pathlib.Path]

load_labeled_file(path=None)

Load a label/notes file containing gesture timing and annotations.

This is used for supervised EMG labeling during offline training.

If no path is provided, opens a GUI file selection dialog.

Parameters:

path (str or None) – Path to the .txt notes file.

Returns:

Labeled samples with columns [“Sample”, “Time”, “Label”],

cleaned and sorted by sample index.

Return type:

pd.DataFrame

load_txt_config(file_path=None, verbose=False)

Parse a simple key=value style configuration file (e.g. config.txt).

Parameters:

• file_path (str) – Path to the config file.

• verbose (bool) – If True, print warnings and info messages.

Returns:

Dictionary of key-value settings.

Return type:

dict

load_yaml_file(file_path=None)

Load configuration from a YAML file.

Parameters:

file_path (str) – Path to the YAML file.

Returns:

Parsed config dictionary.

Return type:

dict

load_json_file(file_path=None)

Load configuration from a JSON file.

Parameters:

file_path (str) – Path to the JSON file.

Returns:

Parsed config dictionary.

Return type:

dict

load_config_file(file_path=None, verbose=False)

Load configuration from a file, supporting .txt, .yaml, and .json formats.

Parameters:

• file_path (str) – Path to the config file.

• verbose (bool) – If True, print debug information.

Returns:

Parsed config dictionary or None if loading failed.

Return type:

dict

labels_from_events(event_path, window_starts, *, strict_segment=False, fs=2000)

Map each window start (absolute sample index) to a label using an events CSV with columns: ‘Sample Index’, ‘Timestamp’, ‘Label’. The Timestamp text is ignored.

strict_segment=True: drops any window whose [start, start+step) crosses an event boundary.

last_event_index(path)

Return type:

Optional[int]

Parameters:

path (str)

stem_without_timestamp(path)

Extract the base name from a file path, stripping any timestamp suffix.

Common patterns handled:

• “IndexExtension_241112_170737.npz” -> “IndexExtension”

• “WristFlexion_2024-01-15_143052.rhd” -> “WristFlexion”

• “recording.rhd” -> “recording”

Parameters:

path (str) – File path or filename

Return type:

str

Returns:

Base name without extension or timestamp suffix

Example

>>> stem_without_timestamp("IndexExtension_241112_170737.npz")
'IndexExtension'

glob_first(pattern)

Return the first file matching a glob pattern, or None if no matches.

Parameters:

pattern (str) – Glob pattern (for example, /data/events/*.event)

Return type:

Optional[str]

Returns:

First matching file path (sorted), or None

Example

>>> glob_first("/data/events/trial*.event")
'/data/events/trial1_emg.event'

find_event_file(root_dir, data_file_path, label='', extensions=('.event', '.txt'))

Auto-discover the event file corresponding to a data recording.

Search order:

1. <root>/events/<stem>_emg.{event,txt}

2. <root>/events/<stem>.{event,txt}

3. <root>/events/<stem_base>_emg.{event,txt} (without timestamp)

4. <root>/events/<stem_base>.{event,txt}

5. Sibling to data file: <parent>/<stem>*.{event,txt}

6. If only one event file exists in events/, use it

Parameters:

• root_dir (str) – Root directory of the experiment

• data_file_path (str) – Path to the data file (.rhd, .npz, etc.)

• label (str) – Optional label prefix to try

• extensions (tuple) – File extensions to search for

Return type:

Optional[str]

Returns:

Path to event file, or None if not found

Example

>>> find_event_file("/data/exp1", "/data/exp1/emg/trial1_241112.rhd")
'/data/exp1/events/trial1_emg.event'

discover_data_files(root_dir, extensions=('.rhd', '.npz'), subdir=None, recursive=True)

Discover all data files in a directory tree.

Parameters:

• root_dir (str) – Root directory to search

• extensions (tuple) – File extensions to find (e.g., (“.rhd”, “.npz”))

• subdir (Optional[str]) – Optional subdirectory to search (e.g., “emg”, “raw”)

• recursive (bool) – If True, search subdirectories

Return type:

list

Returns:

Sorted list of file paths

Example

>>> discover_data_files("/data/exp1", extensions=(".rhd",))
['/data/exp1/raw/trial1.rhd', '/data/exp1/raw/trial2.rhd']

find_dataset_file(root_dir, label='', candidates=None)

Find a training dataset NPZ file in common locations.

Parameters:

• root_dir (str) – Root directory to search

• label (str) – Optional label prefix

• candidates (Optional[list]) – Additional candidate filenames to try

Return type:

Optional[str]

Returns:

Path to dataset file, or None if not found

Example

>>> find_dataset_file("/data/exp1", label="sleeve")
'/data/exp1/sleeve_training_dataset.npz'

list_npz_files(path_or_glob, recursive=False)

Expand dirs/globs/files into a sorted, de-duplicated list of .npz paths.

Return type:

List[str]

Parameters:

• path_or_glob (str | Sequence[str])

• recursive (bool)

find_npz_by_label(npz_dir, label, recursive=True)

Return NPZ paths whose filename contains label (case-insensitive).

Return type:

list[str]

Parameters:

• npz_dir (str)

• label (str)

• recursive (bool)

load_npz_record(path)

Load one NPZ file -> (emg (C,N), label:str). Raises if keys missing.

Return type:

tuple[ndarray, str]

Parameters:

path (str)

save_as_npz(result, file_path=None, use_compressed_format=True, verbose=False)

Save the rhd data as a .npz file. uses the compressed format by default

Parameters:

• result (dict) – Dictionary containing the Open Ephys session data. Must contain keys: ‘amplifier_data’, ‘t_amplifier’, ‘sample_rate’, ‘recording_name’.

• file_path (str, optional) – Path to save the .npz file. If None, uses the recording name.

• verbose (bool)

Return type:

None

Returns:

None

save_as_npz_compressed(result, file_path=None, optimize_dtypes=True, show_progress=False, verbose=True)

Save the rhd data as a compressed .npz file (70-85% smaller than standard NPZ).

This function creates significantly smaller files than save_as_npz() with no loss of data quality. Recommended for all use cases.

Parameters:

• result (dict) – Dictionary containing the Open Ephys session data. Must contain keys: ‘amplifier_data’, ‘t_amplifier’, ‘sample_rate’, ‘recording_name’.

• file_path (str, optional) – Path to save the .npz file. If None, uses the recording name.

• optimize_dtypes (bool, default=True) – Optimize data types to reduce size: - Keep int16 as int16 (ADC data) - Convert float64 to float32 where appropriate

• show_progress (bool, default=False) – Show progress bar during save (requires tqdm).

• verbose (bool, default=True) – Print save status messages.

Returns:

Statistics about the saved file including:

• file_path: Path to saved file

• original_size_mb: Size before optimization

• file_size_mb: Final compressed file size

• compression_ratio: How much smaller (e.g., 4.0 = 4x smaller)

• space_saved_percent: Percentage reduction

Return type:

dict

Examples

save_as_npz_compressed(result)

# With custom path and progress bar stats = save_as_npz_compressed(result, “data.npz”, show_progress=True) print(f”Saved {stats[‘space_saved_percent’]:.1f}% space!”)

load_npz_file(file_path, verbose=False, allow_many=False)

Load an NPZ and reconstruct a convenient dict, normalizing common field aliases.

Return type:

Union[Dict, List[Dict]]

Parameters:

• file_path (str | Sequence[str])

• verbose (bool)

• allow_many (bool)

Supported aliases (all preserved; normalized fields always populated when available):

• EMG matrix: ‘emg’, ‘emg_data’, ‘amplifier_data’ -> normalized to result[‘amplifier_data’]

• Time vector: ‘t’, ‘time_vector’, ‘t_amplifier’ -> normalized to result[‘t_amplifier’]

• Sampling rate (Hz): ‘fs’, ‘sampling_rate’, ‘sample_rate’,

  result[‘frequency_parameters’][‘amplifier_sample_rate’], OR inferred from time vector -> normalized to result[‘_fs_Hz’] & result[‘sample_rate’]

• Channel names: ‘ch_names’, ‘channel_names’, ‘amplifier_channels[*].native_channel_name’

  -> normalized to result[‘channel_names’]

Also passes through any extra keys (e.g., provenance: ‘source_file’, ‘source_gesture’, ‘source_local_index’, ‘source_name’, ‘gestures_combined’, ‘fs_reported’, ‘fs_effective’, ‘align_mode’, ‘dt_target’, ‘t_overlap’, ‘reorder_info’, etc.).

If file_path resolves to multiple files and allow_many=True, returns a list of dicts. Otherwise (legacy behavior) the first match is loaded.

load_npz_files(paths, verbose=False)

Load MANY NPZ files, returning a list of dicts (one per NPZ).

paths may be a directory (loads all *.npz files), a file path, a glob pattern, or a list/tuple containing any combination of those.

Return type:

List[Dict]

Parameters:

• paths (str | Sequence[str])

• verbose (bool)

load_training_dataset(npz_path, verbose=False)

Load a training dataset NPZ with standardized field extraction.

Handles various field naming conventions and returns a normalized dict.

Parameters:

• npz_path (str) – Path to training dataset NPZ file

• verbose (bool) – Print loading info

Returns:

• X: Feature matrix (n_windows, n_features)

• y: Labels as strings (n_windows,)

• y_id: Labels as integers (n_windows,) if available

• class_names: Sorted unique class names

• label_to_id: Dict mapping class name → integer ID

• emg_fs: Sampling frequency (Hz)

• window_ms: Feature window size (ms)

• step_ms: Window step size (ms)

• channel_names: List of channel names

• selected_channels: List of channel indices used

• feature_spec: Feature specification dict

• metadata: Any additional metadata

Return type:

Dict with standardized keys

Example

>>> data = load_training_dataset("training_dataset.npz")
>>> X, y = data["X"], data["y"]
>>> print(f"Loaded {X.shape[0]} samples, {len(data['class_names'])} classes")

load_and_merge_datasets(npz_paths, verbose=False)

Load and concatenate multiple training datasets.

Validates that all datasets have compatible feature dimensions and merges class labels appropriately.

Parameters:

• npz_paths (Sequence[str]) – List of paths to training dataset NPZ files

• verbose (bool) – Print loading info

Returns:

• X: Concatenated feature matrix (total_windows, n_features)

• y: Concatenated string labels (total_windows,)

• metadata: Merged metadata dict

Return type:

Tuple of (X, y, metadata) where

Raises:

ValueError – If feature dimensions don’t match across datasets

Example

>>> X, y, meta = load_and_merge_datasets([
...     "session1_dataset.npz",
...     "session2_dataset.npz",
... ])

save_training_dataset(save_path, X, y, emg_fs, window_ms, step_ms, channel_names, selected_channels=None, feature_spec=None, channel_map=None, channel_map_file=None, modality='emg')

Save a training dataset to NPZ with standardized format.

Parameters:

• save_path (str) – Output file path

• X (ndarray) – Feature matrix (n_windows, n_features)

• y (ndarray) – String labels (n_windows,)

• emg_fs (float) – Sampling frequency (Hz)

• window_ms (int) – Feature window size (ms)

• step_ms (int) – Window step size (ms)

• channel_names (List[str]) – List of channel names in order used

• selected_channels (Optional[List[int]]) – Original channel indices (if subset)

• feature_spec (Optional[Dict]) – Feature specification dict

• channel_map (Optional[str]) – Name of channel mapping used (for reproducibility)

• channel_map_file (Optional[str]) – Path to channel mapping file

• modality (str) – Data modality (default: “emg”)

Return type:

None

Example

>>> save_training_dataset(
...     "training_dataset.npz",
...     X=features, y=labels,
...     emg_fs=2000, window_ms=200, step_ms=50,
...     channel_names=["A-001", "A-002", ...],
... )

intan.io._block_parser

This module provides low-level parsing functions for reading binary .rhd data blocks produced by Intan Technologies hardware. It handles analog/digital signal unpacking, timestamp alignment, and memory preallocation for signal extraction.

Functions include: - Reading amplifier and auxiliary signals - Extracting timestamps - Assembling full data dictionaries

Intended for internal use by the Intan RHX Python interface.

read_uint32(array, arrayIndex)

Read a 4-byte unsigned integer from the data.

read_int32(array, arrayIndex)

Read a 4-byte signed integer from the data.

read_uint16(array, arrayIndex)

Read a 2-byte unsigned integer from the data.

get_timestamp_signed(header)

Determine if timestamps are stored as signed integers.

This depends on the version of the Intan file format. Intan software version 1.2 and later uses signed timestamps.

Parameters:

header (dict) – Parsed header dictionary containing version info.

Returns:

True if timestamps are signed; False if unsigned.

Return type:

bool

read_one_data_block(data, header, indices, fid)

Reads one 60 or 128 sample data block from fid into data, at the location indicated by indices

Parameters:

• data (dict) – Dictionary to store the read data.

• header (dict) – Header information from the file.

• indices (dict) – Indices for each signal type in the data dictionary.

• fid – File object to read from.

read_timestamps(fid, data, indices, num_samples, timestamp_signed)

Reads timestamps from binary file as a NumPy array, indexing them into ‘data’.

Parameters:

• fid – File object to read from.

• data (dict) – Dictionary to store the read data.

• indices (dict) – Indices for each signal type in the data dictionary.

• num_samples (int) – Number of samples to read.

• timestamp_signed (bool) – Flag indicating if timestamps are signed.

read_analog_signals(fid, data, indices, samples_per_block, header)

Reads all analog signal types present in RHD files: amplifier_data, aux_input_data, supply_voltage_data, temp_sensor_data, and board_adc_data, into ‘data’ dict.

Parameters:

• fid – File object to read from.

• data (dict) – Dictionary to store the read data.

• indices (dict) – Indices for each signal type in the data dictionary.

• samples_per_block (int) – Number of samples per block.

• header (dict) – Header information from the file.

read_digital_signals(fid, data, indices, samples_per_block, header)

Reads all digital signal types present in RHD files: board_dig_in_raw and board_dig_out_raw, into ‘data’ dict.

Parameters:

• fid – File object to read from.

• data (dict) – Dictionary to store the read data.

• indices (dict) – Indices for each signal type in the data dictionary.

• samples_per_block (int) – Number of samples per block.

• header (dict) – Header information from the file.

read_analog_signal_type(fid, dest, start, num_samples, num_channels)

Reads data from binary file as a NumPy array, indexing them into ‘dest’, which should be an analog signal type within ‘data’, for example data[‘amplifier_data’] or data[‘aux_input_data’]. Each sample is assumed to be of dtype ‘uint16’.

read_digital_signal_type(fid, dest, start, num_samples, num_channels)

Reads data from binary file as a NumPy array, indexing them into ‘dest’, which should be a digital signal type within ‘data’, either data[‘board_dig_in_raw’] or data[‘board_dig_out_raw’].

Each sample is assumed to be of dtype ‘uint16’, and the data is unpacked into ‘dest’ as a 2D array of shape (num_channels, num_samples).

Parameters:

• fid – File object to read from.

• dest (numpy.ndarray) – Destination array to store the read data.

• start (int) – Starting index for writing data.

• num_samples (int) – Number of samples to read.

• num_channels (int) – Number of channels to read.

read_all_data_blocks(header, num_samples, num_blocks, fid, verbose=True)

Reads all data blocks present in file, allocating memory for and returning ‘data’ dict containing all data.

Parameters:

• header (dict) – Header information from the file.

• num_samples (dict) – Number of samples for each signal type.

• num_blocks (int) – Number of blocks to read.

• fid – File object to read from.

• verbose (bool) – Flag for verbose output.

Returns:

Dictionary containing all read data.

Return type:

data (dict)

initialize_memory(header, num_samples)

Pre-allocates NumPy arrays for each signal type that will be filled during this read, and initializes unique indices for data access to each signal type.

Parameters:

• header (dict) – Header information from the file.

• num_samples (dict) – Number of samples for each signal type.

Returns:

Dictionary with pre-allocated arrays for each signal type. indices (dict): Dictionary with indices for each signal type.

Return type:

data (dict)

advance_indices(indices, samples_per_block)

Advances indices used for data access by suitable values per data block.

Parameters:

• indices (dict) – Dictionary with indices for each signal type.

• samples_per_block (int) – Number of samples per block.

plural(number_of_items)

Utility function to pluralize words based on the number of items.

print_all_channel_names(result)

Searches through all present signal types in ‘result’ dict, and prints the names of these channels. Useful, for example, to determine names of channels that can be plotted.

Parameters:

result (dict) – The result of a call to read_header() or header_to_result().

print_names_in_group(signal_group)

Searches through all channels in this group and print them.

Parameters:

signal_group (list) – The list of channels to search through.

find_channel_in_group(channel_name, signal_group)

Finds a channel with this name in this group, returning whether or not it’s present and, if so, the position of this channel in signal_group.

Parameters:

• channel_name (str) – The name of the channel to search for.

• signal_group (list) – The list of channels to search through.

Returns:

Whether or not the channel was found. channel_index (int): The index of the channel in signal_group.

Return type:

channel_found (bool)

find_channel_in_header(channel_name, header)

Looks through all present signal groups in header, searching for ‘channel_name’. If found, return the signal group and the index of that channel within the group.

Parameters:

• channel_name (str) – The name of the channel to search for.

• header (dict) – The header dictionary containing signal groups.

Returns:

Whether or not the channel was found. signal_group_name (str): The name of the signal group containing the channel. channel_index (int): The index of the channel in the signal group.

Return type:

channel_found (bool)

intan.io._header_parsing

Low-level parser for extracting metadata and channel structure from .rhd files recorded by Intan Technologies hardware.

This module reads: - File version and magic number - Signal groups and their channel maps - Amplifier settings and frequency parameters - Qt-style strings and notes - Impedance settings, board configuration, and reference info

Primary function:

• read_header(fid): returns a fully populated header dictionary

Used internally by intan.io._rhd_loader to build a unified result dictionary for EMG/LFP signal analysis.

read_header(fid)

” Parse the binary file header from an Intan .rhd data file.

This function checks the magic number, reads the file version, evaluates signal settings, channel layouts, impedance settings, and any embedded notes.

Parameters:

fid (file) – Opened file object positioned at the start of the file.

Returns:

Parsed header metadata.

Return type:

dict

read_qstring(fid)

Read a QString (Unicode) from a Qt-generated binary file.

Format: - First 4 bytes: length in bytes (uint32) - If 0xFFFFFFFF, return empty string - Content is 16-bit unicode characters

Parameters:

fid (file) – File object positioned at the string start.

Returns:

Decoded unicode string.

Return type:

str

Raises:

QStringError – If the declared length exceeds file size.

read_notes(header, fid)

Reads notes as QStrings from fid, and stores them as strings in header[‘notes’] dict.

Parameters:

• header (dict) – Header dictionary to store notes.

• fid (file) – Opened file object positioned at the start of the file.

read_version_number(header, fid, verbose=True)

Reads version number (major and minor) from fid. Stores them into header[‘version’][‘major’] and header[‘version’][‘minor’].

Parameters:

• header (dict) – Header dictionary to store version information.

• fid (file) – Opened file object positioned at the start of the file.

• verbose (bool) – If True, print version information to console.

Raises:

UnrecognizedFileError – If the magic number does not match the expected

check_magic_number(fid)

Checks magic number at beginning of file to verify this is an Intan Technologies RHD data file.

Parameters:

fid (file) – Opened file object positioned at the start of the file.

Raises:

UnrecognizedFileError – If the magic number does not match the expected

read_notch_filter_frequency(header, freq, fid)

Reads notch filter mode from fid, and stores frequency (in Hz) in ‘header’ and ‘freq’ dicts.

Parameters:

• header (dict) – Header dictionary to store notch filter frequency.

• freq (dict) – Dictionary to store notch filter frequency.

• fid (file) – Opened file object positioned at the start of the file.

read_channel_structure(header, fid, verbose=False)

Reads signal summary from data file header and stores information for all signal groups and their channels in ‘header’ dict.

Parameters:

• header (dict) – Header dictionary to store channel information.

• fid (file) – Opened file object positioned at the start of the file.

• verbose (bool) – If True, print header summary to console.

read_num_temp_sensor_channels(header, fid)

Stores number of temp sensor channels in header[‘num_temp_sensor_channels’]. Temp sensor data may be saved from versions 1.1 and later.

Parameters:

• header (dict) – Header dictionary to store temp sensor channel count.

• fid (file) – Opened file object positioned at the start of the file.

read_sample_rate(header, fid)

Reads sample rate from fid. Stores it into header[‘sample_rate’].

Parameters:

• header (dict) – Header dictionary to store sample rate.

• fid (file) – Opened file object positioned at the start of the file.

read_freq_settings(freq, fid)

Reads amplifier frequency settings from fid. Stores them in ‘freq’ dict.

Parameters:

• freq (dict) – Dictionary to store frequency settings.

• fid (file) – Opened file object positioned at the start of the file.

read_impedance_test_frequencies(freq, fid)

Reads desired and actual impedance test frequencies from fid, and stores them (in Hz) in ‘freq’ dicts.

Parameters:

• freq (dict) – Dictionary to store impedance test frequencies.

• fid (file) – Opened file object positioned at the start of the file.

read_eval_board_mode(header, fid)

Stores eval board mode in header[‘eval_board_mode’]. Board mode is saved from versions 1.3 and later.

Parameters:

• header (dict) – Header dictionary to store eval board mode.

• fid (file) – Opened file object positioned at the start of the file.

read_reference_channel(header, fid)

Reads name of reference channel as QString from fid, and stores it as a string in header[‘reference_channel’]. Data files v2.0 or later include reference channel.

Parameters:

• header (dict) – Header dictionary to store reference channel name.

• fid (file) – Opened file object positioned at the start of the file.

read_new_channel(fid, signal_group_name, signal_group_prefix, signal_group)

Reads a new channel’s information from fid and returns it as a dict. The channel is identified by its signal group name, prefix, and number.

Parameters:

• fid (file) – Opened file object positioned at the start of the channel.

• signal_group_name (str) – Name of the signal group.

• signal_group_prefix (str) – Prefix of the signal group.

• signal_group (int) – Number of the signal group.

Returns:

Dictionary with channel information. new_trigger_channel (dict): Dictionary with trigger channel info. channel_enabled (bool): Indicates if the channel is enabled. signal_type (int): Type of signal for the channel.

Return type:

new_channel (dict)

append_new_channel(header, new_channel, new_trigger_channel, channel_enabled, signal_type)

Appends ‘new_channel’ to ‘header’ dict depending on if channel is enabled and the signal type.

Parameters:

• header (dict) – Header dictionary to store channel information.

• new_channel (dict) – Dictionary with new channel information.

• new_trigger_channel (dict) – Dictionary with trigger channel info.

• channel_enabled (bool) – Indicates if the channel is enabled.

• signal_type (int) – Type of signal for the channel.

Raises:

UnknownChannelTypeError – If the signal type is unrecognized.

add_num_channels(header)

Adds channel numbers for all signal types to ‘header’ dict.

Parameters:

header (dict) – Header dictionary to store channel counts.

set_num_samples_per_data_block(header)

Determines how many samples are present per data block (60 or 128), depending on version. Data files v2.0 or later have 128 samples per block, otherwise 60.

Parameters:

header (dict) – Header dictionary to store number of samples per block.

set_sample_rates(header, freq)

Determines what the sample rates are for various signal types, and stores them in ‘freq’ dict.

Parameters:

• header (dict) – Header dictionary to store sample rates.

• freq (dict) – Dictionary to store frequency parameters.

set_frequency_parameters(header, freq)

Stores frequency parameters (set in other functions) in header[‘frequency_parameters’]

Parameters:

• header (dict) – Header dictionary to store frequency parameters.

• freq (dict) – Dictionary to store frequency parameters.

initialize_channels(header)

Creates empty lists for each type of data channel and stores them in ‘header’ dict.

Parameters:

header (dict) – Header dictionary to initialize channel lists.

add_signal_group_information(header, fid, signal_group)

Adds information for a signal group and all its channels to ‘header’ dict.

Parameters:

• header (dict) – Header dictionary to store signal group information.

• fid (file) – Opened file object positioned at the start of the signal group.

• signal_group (int) – Number of the signal group.

add_channel_information(header, fid, signal_group_name, signal_group_prefix, signal_group)

Reads a new channel’s information from fid and appends it to ‘header’ dict.

Parameters:

• header (dict) – Header dictionary to store channel information.

• fid (file) – Opened file object positioned at the start of the channel.

• signal_group_name (str) – Name of the signal group.

• signal_group_prefix (str) – Prefix of the signal group.

• signal_group (int) – Number of the signal group.

header_to_result(header, result)

Merge parsed header fields into the global result dictionary.

Parameters:

• header (dict) – Parsed header metadata from read_header.

• result (dict) – Destination dictionary to be populated.

Returns:

Updated result dictionary with signal channel mappings.

Return type:

dict

print_header_summary(header)

Prints summary of contents of RHD header to console.

Parameters:

header (dict) – Header dictionary containing parsed metadata.

data_to_result(header, data, result)

Merges data from all present signals into a common ‘result’ dict. If any signal types have been allocated but aren’t relevant (for example, no channels of this type exist), does not copy those entries into ‘result’.

Parameters:

• header (dict) – Parsed header metadata from read_header.

• data (dict) – Dictionary containing signal data.

• result (dict) – Destination dictionary to be populated.

Returns:

Updated result dictionary with signal data.

Return type:

dict

plural(number_of_items)

Return ‘s’ if the number of items is not 1 (for pluralization).

Parameters:

number_of_items (int) – Quantity to evaluate

Returns:

‘s’ if plural, ‘’ if singular

Return type:

str

get_bytes_per_data_block(header)

Calculate the total number of bytes in each data block of a .rhd file.

This is based on the number of enabled channels and the system used for recording (either 60 or 128 samples per block).

Parameters:

header (dict) – Parsed header metadata with fields like ‘num_amplifier_channels’

Returns:

Number of bytes in one full data block

Return type:

int

bytes_per_signal_type(num_samples, num_channels, bytes_per_sample)

Calculate number of bytes for a specific signal type in a data block.

Parameters:

• num_samples (int or float) – Samples per block (may be fractional)

• num_channels (int) – Number of enabled channels for this signal type

• bytes_per_sample (int) – Number of bytes per sample

Returns:

Number of bytes for this signal type

Return type:

float

calculate_data_size(header, filename, fid, verbose=True)

Determine the size and structure of recorded data in an .rhd file.

Computes how many samples exist, how many data blocks are present, and whether the file appears truncated or corrupt.

Parameters:

• header (dict) – Parsed header from file

• filename (str) – Path to the .rhd file

• fid (file) – Open file object positioned after header

• verbose (bool) – If True, print file duration summary

Returns:

data_present (bool): True if data exists beyond header filesize (int): Full file size in bytes num_blocks (int): Number of data blocks present num_samples (dict): Estimated samples per signal type

Return type:

tuple

calculate_num_samples(header, num_data_blocks)

Estimate the number of samples for each signal type.

Parameters:

• header (dict) – Parsed header with sample structure

• num_data_blocks (int) – Total data blocks in file

Returns:

Signal type → number of samples (e.g., ‘amplifier’: 25600)

Return type:

dict

print_record_time_summary(num_amp_samples, sample_rate, data_present)

Print the estimated duration of the .rhd recording to the console.

Parameters:

• num_amp_samples (int) – Number of amplifier samples in file

• sample_rate (float) – Sampling rate in Hz

• data_present (bool) – Whether any data was detected in the file

intan.io._rhd_loader

This module provides file loaders for Intan Technologies’ .rhd binary files, as well as associated .dat files used in ‘One File Per Signal Type’ format.

It includes GUI-assisted selection, header parsing, signal reconstruction, and helper functions to load and concatenate datasets. The output is a dictionary containing signal data, metadata, and time vectors for each channel type.

Key Functions: - load_rhd_file: Full .rhd file loader with header and data parsing - load_dat_file: Loader for .dat-based datasets with separate header - load_files_from_path: Batch loading and optional concatenation - read_amplifier_file, read_auxiliary_file, etc.: Raw binary readers

load_rhd_file(filepath=None, merge_files=False, sort_files=True, rebuild_time=True, verbose=False)

Load Intan .rhd file(s). If multiple files are selected/provided and merge_files=True, concatenate them along the time axis.

Parameters:

• filepath (str | list[str] | tuple[str] | None) – Path to a single .rhd file, or a sequence of paths. If None, opens a file dialog (multi-select when merge_files=True).

• merge_files (bool) – If True and multiple files are provided/selected, concatenate them.

• sort_files (bool) – Sort paths lexicographically (filenames often encode time).

• rebuild_time (bool) – If True, rebuild a strictly monotonic t_amplifier after concatenation.

• verbose (bool) – Print progress.

Returns:

Parsed signal & metadata dictionary. When concatenating, arrays are stitched along the sample axis and metadata keys are harmonized.

Return type:

dict

read_time_file(path)

Reads int32 timestamp values from a time.dat file.

Parameters:

path (str) – Path to the time.dat file.

Returns:

Array of timestamps in microseconds.

Return type:

np.ndarray

read_amplifier_file(path, num_channels)

Load amplifier signal from a .dat file (One File Per Signal Type format).

Parameters:

• path (str) – Full path to amplifier.dat

• num_channels (int) – Number of amplifier channels recorded

Returns:

Amplifier signals (channels × samples) in µV.

Return type:

np.ndarray

read_auxiliary_file(path, num_channels, scale=3.74e-05)

Reads auxiliary channel data (uint16) and applies scaling.

read_adc_file(path, num_channels, scale=5.0354e-05)

Reads board ADC data (uint16) and applies default scaling.

Parameters:

• path (str) – Path to the board_adc.dat file.

• num_channels (int) – Number of ADC channels recorded.

• scale (float) – Scaling factor for ADC data.

Returns:

Board ADC data (channels × samples) in Volts.

Return type:

np.ndarray

load_dat_file(filepath=None)

Load dataset in ‘One File Per Signal Type’ format using external .dat files.

Requires presence of an info.rhd file in the same directory for channel metadata.

Parameters:

filepath (str or None) – Path to any .dat file in the dataset. Opens file dialog if None.

Returns:

Parsed signal data and metadata.

Return type:

dict

load_per_signal_files(folder_path, header)

Load all .dat files in the specified folder, using the header information

Parameters:

• folder_path (str) – Path to the folder containing .dat files.

• header (dict) – Header information from the .rhd file.

Returns:

Dictionary containing all loaded data and metadata.

Return type:

dict

load_files_from_path(folder_path=None, concatenate=False)

Loads all .rhd files from a specified path or using a file dialog. Concatenates teh data if specified.

Optionally concatenate the data from all files into a single result dictionary.

Parameters:

• folder_path (str or None) – The path to the folder containing the .rhd files.

• concatenate (bool) – Boolean indicating if the data from all files should be concatenated.

Returns:

A list of ‘result’ dictionaries if concatenate is False, otherwise a single ‘result’ dictionary.

Return type:

all_results

apply_notch_filter(header, data, verbose=True)

Checks header to determine if notch filter should be applied, and if so, apply notch filter to all signals in data[‘amplifier_data’].

Parameters:

• header (dict) – The header information of the data file.

• data (dict) – The raw data to be parsed.

• verbose (bool) – If True, print progress messages. Default is True.

parse_data(header, data)

Parses raw data into user readable and interactable forms (for example, extracting raw digital data to separate channels and scaling data to units like microVolts, degrees Celsius, or seconds.)

Parameters:

• header (dict) – The header information of the data file.

• data (dict) – The raw data to be parsed.

scale_timestamps(header, data)

Verifies no timestamps are missing, and scales timestamps to seconds.

Parameters:

• header (dict) – The header information of the data file.

• data (dict) – The raw data to be parsed.

scale_analog_data(header, data)

Scales all analog data signal types (amplifier data, aux input data, supply voltage data, board ADC data, and temp sensor data) to suitable units (microVolts, Volts, deg C).

Parameters:

• header (dict) – The header information of the data file.

• data (dict) – The raw data to be parsed.

extract_digital_data(header, data)

Extracts digital data from raw (a single 16-bit vector where each bit represents a separate digital input channel) to a more user-friendly 16-row list where each row represents a separate digital input channel. Applies to digital input and digital output data.

Parameters:

• header (dict) – The header information of the data file.

• data (dict) – The raw data to be parsed.

parse_event_file(event_files, verbose=False)

Extract all events from event file(s) and return a combined DataFrame.

Parameters:

• event_files (str or list) – Path(s) to the event file(s).

• verbose (bool) – If True, print debug information.

Returns:

DataFrame containing all events with columns:

• ’sample_index’: Sample index of the event (int)

• ’timestamp’: Timestamp string (str or None if missing)

• ’label’: Cleaned label string (str)

Return type:

pd.DataFrame

parse_numeric_args(numeric_args, default_channels=[0, 1, 2, 3])

Parse a channel argument from the command line.

Accepts the string "all", integer lists, or a single slice-like value such as "0:64".

convert_events_to_list(ev_path, window_starts, verbose=False)

Converts event file to a list of labels corresponding to the provided window starts.

lock_params_to_meta(meta, window_ms, step_ms, selected_channels)

Return (window_ms, step_ms, selected_channels, envelope_cut_hz) locked to training meta, if present.

Return type:

Tuple[int, int, Optional[List[int]], float]

Parameters:

• meta (Dict)

• window_ms (int | None)

• step_ms (int | None)

• selected_channels (List[int] | None)

load_metadata_json(root_dir, label='')

Return type:

dict

Parameters:

• root_dir (str)

• label (str)

normalize_name(s)

Return type:

str

Parameters:

s (str)

build_indices_from_mapping(raw_channel_names, mapping_names, *, strict=True)

Return type:

list[int]

Parameters:

• raw_channel_names (list[str])

• mapping_names (list[str])

• strict (bool)

align_channels_by_name(emg, source_names, target_names, *, normalizer=None, missing='error', duplicates='first', return_report=True)

Reorder (C, N) EMG rows to match a target channel-name order.

Parameters:

• emg (np.ndarray) – Array shaped (C, N) (channels x samples).

• source_names (Sequence[str]) – Names for rows of emg in their current order.

• target_names (Sequence[str]) – Desired channel-name order (e.g., training order).

• normalizer (Callable[[str], str], optional) – Function to normalize names before matching (e.g., strip, upper, remove punctuation). Defaults to pyoephys.io.normalize_name if available, else identity.

• missing ({"error","zero","nan"}, optional) –

  What to do when a target channel is not found in source:

  • ”error”: raise RuntimeError (strict).

  • ”zero”: synthesize a zero-filled row.

  • ”nan”: synthesize a NaN-filled row.

• duplicates ({"error","first","last"}, optional) –

  What to do when a source name appears more than once:

  • ”error”: raise RuntimeError.

  • ”first”: use the first occurrence.

  • ”last”: use the last occurrence.

• return_report (bool, optional) – If True, return a dict with details about mapping/missing/duplicates.

Return type:

Tuple[ndarray, List[int], Optional[Dict[str, Any]]]

Returns:

• aligned (np.ndarray) – EMG reordered to (len(target_names), N). If missing!=”error”, rows may be synthesized.

• indices (List[int]) – Source row indices used for each target (=-1 for synthesized rows).

• report (dict or None) – Keys: {“missing”, “extras”, “duplicates”, “index_map”, “used_indices”} (when return_report=True).

Raises:

• RuntimeError – On missing channels (when missing=”error”) or duplicates (when duplicates=”error”).

• ValueError – If shapes/lengths are inconsistent.

select_training_channels_by_name(emg, raw_names, trained_names)

Strict selection: reorder by name, missing/duplicates => errors. Matches old _select_training_channels_by_name semantics.

Return type:

Tuple[ndarray, List[int]]

Parameters:

• emg (ndarray)

• raw_names (Sequence[str])

• trained_names (Sequence[str])

trained_channel_names_from_meta(meta)

Pull training channel names from metadata.

The nested meta["data"]["channel_names"] location is preferred; meta["channel_names"] is retained for compatibility. Returns an empty list if neither location is present.

Return type:

list[str]

Parameters:

meta (dict)

trained_channel_names_from_dataset_npz(root_dir, label='')

Fallback: look inside the training dataset NPZ for channel names. Tries label-specific first, then common defaults.

Return type:

list[str]

Parameters:

• root_dir (str)

• label (str | None)

get_trained_channel_names(root_dir, label='')

High-level: load metadata then fallback to dataset NPZ.

Return type:

list[str]

Parameters:

• root_dir (str)

• label (str)

parse_channel_spec(spec, total=None)

Parse a flexible channel specification into a list of indices.

This function handles multiple common formats for specifying channels from command line arguments or configuration files.

Supported formats:

• None or “” → None (use all channels)

• “all” → list(range(total)) if total provided, else None

• “0:64” → [0, 1, …, 63] (end-exclusive, like Python slice)

• “0:128:2” → [0, 2, 4, …, 126] (with step)

• “0 1 2 3” → [0, 1, 2, 3] (space-separated)

• “0,1,2,3” → [0, 1, 2, 3] (comma-separated)

• “0-7” → [0, 1, …, 7] (inclusive dash range)

• “0:32,64,70-75” → mixed formats combined

• [“0:32”, “64”] → list input also accepted

Parameters:

• spec – Channel specification string, list of strings, or None

• total (Optional[int]) – Total channel count (used for “all” keyword)

Return type:

List[int]

Returns:

Sorted list of unique channel indices, or None if no spec given

Examples

>>> parse_channel_spec("0:8")
[0, 1, 2, 3, 4, 5, 6, 7]

>>> parse_channel_spec("1-4")
[1, 2, 3, 4]

>>> parse_channel_spec("0:8,16,20-22")
[0, 1, 2, 3, 4, 5, 6, 7, 16, 20, 21, 22]

>>> parse_channel_spec("all", total=128)
[0, 1, 2, ..., 127]

load_channel_mapping(mapping_name, mapping_file)

Load a named channel mapping from a JSON file.

The JSON file should contain a dict of named mappings, each being a list of channel names in the desired order.

Parameters:

• mapping_name (str) – Key in the mapping JSON (e.g., “sleeve_halfcount”)

• mapping_file (str) – Path to the JSON file

Return type:

List[str]

Returns:

List of channel names in mapped order

Raises:

• FileNotFoundError – If mapping file doesn’t exist

• KeyError – If mapping name not found in file

Example JSON file:

{
  "sleeve_halfcount": ["A-001", "A-002", "A-003", "B-001"]
}

resolve_channel_selection(raw_channel_names, channels=None, channel_map=None, channel_map_file='custom_channel_mappings.json', strict=True)

Resolve channel selection from either explicit indices or named mapping.

Priority: channel_map > channels > all channels

Parameters:

• raw_channel_names (List[str]) – All channel names from the data source

• channels (Optional[List[int]]) – Explicit channel indices, or None

• channel_map (Optional[str]) – Name of mapping in JSON file, or None

• channel_map_file (str) – Path to mapping JSON

• strict (bool) – If True, raise error for missing channels; else skip them

Return type:

Tuple[Optional[List[int]], List[str]]

Returns:

Tuple of (selected_indices, selected_names) If no selection specified, returns (None, raw_channel_names)

Example

>>> names = ["A-001", "A-002", "A-003", "B-001", "B-002"]
>>> resolve_channel_selection(names, channels=[0, 2, 4])
([0, 2, 4], ['A-001', 'A-003', 'B-002'])

normalize_channel_name_1based(name)

Normalize Intan channel names to 1-based format (A-001, B-002, etc.).

Handles various input formats:

• “A-000” → “A-001” (0-based to 1-based)

• “a_0” → “A-001”

• “B-17” → “B-018” (zero-padded, assumes 0-based input)

• “b 5” → “B-006”

Parameters:

name (str) – Channel name string

Return type:

str

Returns:

Normalized channel name in “X-NNN” format (1-based)

Example

>>> normalize_channel_name_1based("a-000")
'A-001'

parse_channels_spec(specs)

Parse channel specification from CLI arguments.

Accepts: - Single indices: 5 12 - Python slice: 0:128, 0:128:2, :64 - Dash ranges: 1-8 (inclusive) - Comma-separated: 0:64,70,75-80

Returns sorted list of unique channel indices, or None if specs is None.

Return type:

list[int] | None

Examples

>>> parse_channels_spec("0:64")
[0, 1, 2, ..., 63]
>>> parse_channels_spec("5,10,15-20")
[5, 10, 15, 16, 17, 18, 19, 20]
>>> parse_channels_spec(["0:64", "100-110"])
[0, 1, ..., 63, 100, 101, ..., 110]

discover_and_group_files(root_dir, file_type, file_names=None, exclude_pattern=None, merge_pattern=None)

Discover files and group by stem for multi-part recordings.

Parameters:

• root_dir (str) – Root directory to search

• file_type (str) – Type of files to search for (‘rhd’, ‘npz’, ‘csv’)

• file_names (Optional[List[str]]) – Optional list of specific filenames to filter

• exclude_pattern (Optional[str]) – Pattern to exclude from file stems

• merge_pattern (Optional[str]) – Pattern that must be in file stems to include

Returns:

Mapping of file stems to lists of file paths

Return type:

dict

Example

>>> groups = discover_and_group_files("/data", "rhd")
>>> groups
{'recording_1': ['recording_1_part1.rhd', 'recording_1_part2.rhd'],
 'recording_2': ['recording_2.rhd']}

load_single_file(file_type, file_path, root_dir, verbose=False)

Load single file based on type.

Parameters:

• file_type (str) – Type of file (‘rhd’, ‘npz’, ‘csv’)

• file_path (str) – Path to the file

• root_dir (str) – Root directory (used for CSV loading)

• verbose (bool) – Print verbose output

Returns:

Loaded data dictionary

Return type:

dict

load_files_merged(file_type, files, root_dir, verbose=False)

Load and merge multiple files.

Parameters:

• file_type (str) – Type of files (‘rhd’, ‘npz’, ‘csv’)

• files (List[str]) – List of file paths to merge

• root_dir (str) – Root directory (used for CSV loading)

• verbose (bool) – Print verbose output

Returns:

Merged data dictionary

Return type:

dict

file_stem(path)

Get file name without extension.

Parameters:

path (str) – File path

Return type:

str

Returns:

Filename without extension

find_event_for_file(events_dir, data_path, pattern=None)

Find matching event file for a data file.

Tries in order:

1. <events_dir>/<gesture>_emg.event

2. <events_dir>/<gesture>.event

3. Same folder as data file

4. events/ sibling directory

Parameters:

• events_dir (Optional[str]) – Directory containing event files

• data_path (str) – Path to data file (.rhd, .npz, etc)

• pattern (Optional[str]) – Optional pattern to match (e.g., “emg”, “imu”)

Return type:

Optional[str]

Returns:

Path to event file or None if not found