intan.ml package

This module contains all the machine learning utilities for Intan data.

Top-level intan.ml package.

This module uses lazy imports so importing intan.ml does not immediately load PyTorch. Accessing symbols (e.g. EMGRealTimePredictor, ModelManager) will import their defining modules on first use.

class EMGRealTimePredictor(device, model, pca, mean, std, label_names, cue_df=None, buffer_sec=1, window_ms=250, step_sec=1, window_offset_samples=0, feature_list=None)

Bases: object

apply_filters(emg)

extract_and_classify(window, already_filtered=True)

get_label_for_sample(sample_idx)

run_prediction_loop(background=True)

stop()

get_message(timeout=0.5)

evaluate_against_events(file_path, window_starts, y_pred, drop_labels=('Unknown', 'Start'), *, label_encoder=None, class_names=None, label_to_id=None, canonicalize=True, alias=None, zero_division=0, return_metrics=False, return_arrays=False, verbose=False)

Print a classification report & confusion matrix using the emg.event file, robust to label casing and int/string mismatches.

If return_metrics=True, returns {‘accuracy’: float, ‘classification_report’: dict, ‘confusion_matrix’: [[…]] , ‘labels’: […]}.

Return type:

Optional[dict]

Parameters:

• file_path (str)

• window_starts (ndarray)

• y_pred (Iterable)

• drop_labels (Tuple[str, ...])

• class_names (Sequence[str] | None)

• label_to_id (Mapping[str, int] | None)

• canonicalize (bool)

• alias (Mapping[str, str] | None)

• zero_division (int)

• return_metrics (bool)

• return_arrays (bool)

• verbose (bool)

classification_report_safe(y_true, y_pred, *, labels=None, zero_division=0, output='text')

Wrapper around sklearn.classification_report with robust handling of kwargs across sklearn versions, and a simple output switch.

Parameters:

• zero_division (int)

• output (str)

confusion_matrix_safe(y_true, y_pred, *, labels=None, normalize=None)

Wrap sklearn.confusion_matrix; tolerate older versions that may lack normalize kw. Returns (cm, labels).

Parameters:

normalize (str | None)

predict_file(root_dir, file_path, label='', window_ms=None, step_ms=None, events_file=None, save_predictions=True, verbose=False)

Offline prediction from single RHD file with optional event comparison.

Parameters:

• root_dir (str) – Directory containing trained model/metadata

• file_path (str) – Path to .rhd file to evaluate

• label (str) – Model label/tag (e.g., ‘128ch’)

• window_ms (Optional[int]) – Override window size (else from metadata)

• step_ms (Optional[int]) – Override step size (else from metadata)

• events_file (Optional[str]) – Optional path to events file for evaluation

• save_predictions (bool) – Save predictions and evaluation to files

• verbose (bool) – Enable verbose logging

Returns:

y_pred, window_starts_sec, starts_samples, eval_results (if events_file)

Return type:

dict with keys

predict_batch(root_dir, rhd_glob=None, rhd_files=None, events_dir=None, label='', window_ms=None, step_ms=None, zero_division=0, save_eval=False, verbose=False)

Batch prediction across multiple RHD files with aggregated metrics.

Parameters:

• root_dir (str) – Directory containing trained model/metadata

• rhd_glob (Optional[str]) – Glob pattern for RHD files (for example, raw/**/*.rhd)

• rhd_files (Optional[list]) – Explicit list of RHD file paths

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

• label (str) – Model label/tag

• window_ms (Optional[int]) – Override window size

• step_ms (Optional[int]) – Override step size

• zero_division (int) – Value for sklearn zero_division parameter

• save_eval (bool) – Save aggregated results to JSON

• verbose (bool) – Enable verbose logging

Returns:

dict with aggregated metrics

predict_from_device(root_dir, label='', seconds=10.0, event_file=None, window_ms=None, step_ms=None, verbose=False)

Record from device and predict gestures (fixed duration).

Parameters:

• root_dir (str) – Directory containing trained model/metadata

• label (str) – Model label/tag

• seconds (float) – Recording duration in seconds

• event_file (Optional[str]) – Optional events file for evaluation

• window_ms (Optional[int]) – Override window size

• step_ms (Optional[int]) – Override step size

• verbose (bool) – Enable verbose logging

Returns:

dict with predictions and optional evaluation results

predict_realtime_stream(root_dir, label='', window_ms=None, step_ms=None, infer_period_s=None, smooth_k=1, seconds_total=None, use_lsl=False, verbose=False)

Real-time streaming prediction from device.

Parameters:

• root_dir (str) – Directory containing trained model/metadata

• label (str) – Model label/tag

• window_ms (Optional[int]) – Override window size

• step_ms (Optional[int]) – Override step size

• infer_period_s (Optional[float]) – Inference period in seconds

• smooth_k (int) – Majority vote window size for smoothing

• seconds_total (Optional[float]) – Total streaming duration (None = infinite)

• use_lsl (bool) – Enable LSL marker publishing

• verbose (bool) – Enable verbose logging

class EMGRealTimePredictor(device, model, pca, mean, std, label_names, cue_df=None, buffer_sec=1, window_ms=250, step_sec=1, window_offset_samples=0, feature_list=None)

Bases: object

apply_filters(emg)

extract_and_classify(window, already_filtered=True)

get_label_for_sample(sample_idx)

run_prediction_loop(background=True)

stop()

get_message(timeout=0.5)