API Reference

class SubvocalError(Exception)

Base class for all errors raised by the Subvocal SDK.

class ConfigurationError(SubvocalError, ValueError)

Invalid or missing configuration (API keys, parameters, paths).

class HardwareError(SubvocalError, RuntimeError)

A hardware source failed: not connected, stream fault, or parse error.

class MissingDependencyError(SubvocalError, ImportError)

An optional dependency is required for the requested feature.

The message always names the pip extra that provides it.

class ProviderError(SubvocalError, RuntimeError)

An LLM provider call failed after exhausting retries.

class DecodingError(SubvocalError, RuntimeError)

Shorthand-to-intent decoding produced no usable result.

class PolicyViolationError(SubvocalError, PermissionError)

An action was rejected by the authorization policy engine.

Raised only when the pipeline is configured with ``raise_on_policy_violation=True``; by default rejections are traced and reported through callbacks without raising.

class CalibrationError(SubvocalError, RuntimeError)

Per-user classifier calibration or model loading failed.

get_data_dir

def get_data_dir() -> str

Writable directory for traces, correction logs, and datasets.

get_models_dir

def get_models_dir() -> str

Writable directory for trained and downloaded model weights.

class HardwareConfig(BaseModel)

No description.

class DSPConfig(BaseModel)

No description.

class ClassifierConfig(BaseModel)

No description.

class ProviderConfig(BaseModel)

No description.

class PolicyConfig(BaseModel)

No description.

class TelemetryConfig(BaseModel)

No description.

class RuntimeConfig(BaseModel)

No description.

class AuthConfig(BaseModel)

No description.

class SubvocalConfig(BaseModel)

No description.

merge_env_overrides

def merge_env_overrides(config_dict: dict[str, Any]) -> dict[str, Any]

Inspects environment variables for SUBVOCAL_ keys and merges them into the config dict.

Only variables that use the ``__`` section separator (e.g. ``SUBVOCAL_HARDWARE__SAMPLE_RATE``) are treated as config overrides, since every configurable field lives inside a named section. Flat variables such as ``SUBVOCAL_DATA_DIR`` / ``SUBVOCAL_MODELS_DIR`` are reserved for :mod:`subvocal.paths` and are ignored here.

load_config

def load_config(path: str | None = None) -> SubvocalConfig

Loads a SubvocalConfig from a YAML file (if provided) and merges environment overrides.

class CorrectionLogEntry(BaseModel)

Represents a single correction entry logged when a user fixes an incorrect reconstruction.

class CorrectionManager

Manages local storage and retrieval of user correction log entries.

def __init__(self, log_path: str | None = None)

def log_correction(self, raw_shorthand: str, decoded_intent: str, corrected_intent: str, context: UserContext) -> CorrectionLogEntry

def get_corrections(self) -> list[CorrectionLogEntry]

def clear_logs(self) -> None

class FinetuningHook

Converts correction logs into training datasets for fine-tuning LLMs.

def export_to_openai(entries: list[CorrectionLogEntry], system_instruction: str = 'Translate silent speech shorthand and context into correct system actions.') -> list[dict[str, Any]]

def export_to_gemini(entries: list[CorrectionLogEntry], system_instruction: str = 'Translate silent speech shorthand and context into correct system actions.') -> list[dict[str, Any]]

def export_to_jsonl(data: list[dict[str, Any]], output_path: str) -> int

class HardwareSource(ABC)

Abstract interface for subvocal/sEMG sensor data acquisition.

def start(self) -> None

def stop(self) -> None

def read_frame(self, window_ms: int) -> Frame

def is_connected(self) -> bool

class LLMProvider(ABC)

Abstract interface for turning noisy classified shorthand into semantic Intents.

def reconstruct_intent(self, tokens: list[CommandToken], context: UserContext) -> Intent

def get_provider_name(self) -> str

class ActionExecutor(ABC)

Abstract interface for dispatching resolved agentic Actions.

def execute(self, action: Action) -> Any

def can_execute(self, action: Action) -> bool

class ContextProvider(ABC)

Abstract interface for retrieving live device or environment context.

def get_context(self) -> UserContext

def get_provider_name(self) -> str

class Classifier(ABC)

Abstract interface for classifying physiological raw signals into command tokens.

def predict(self, frame: Frame | Any) -> CommandToken | None

def predict_raw(self, frame: Frame | Any) -> tuple[str, float, list[float]]

def labels(self) -> list[str]

class BaseLLMProvider(LLMProvider)

Base provider containing prompt formatting and request helpers.

def __init__(self, api_key: str | None = None, model_name: str | None = None, mock_response: str | None = None, prompt_version: str = 'v1', timeout: float = 10.0, max_retries: int = 2, backoff_seconds: float = 0.5)

class ClaudeProvider(BaseLLMProvider)

Anthropic Claude LLM provider client.

def get_provider_name(self) -> str

def reconstruct_intent(self, tokens: list[CommandToken], context: UserContext) -> Intent

class OpenAIProvider(BaseLLMProvider)

OpenAI GPT completions provider client.

def get_provider_name(self) -> str

def reconstruct_intent(self, tokens: list[CommandToken], context: UserContext) -> Intent

class GeminiProvider(BaseLLMProvider)

Google Gemini content generation provider client.

def get_provider_name(self) -> str

def reconstruct_intent(self, tokens: list[CommandToken], context: UserContext) -> Intent

class LlamaProvider(BaseLLMProvider)

Local Llama provider client (running via Ollama).

def __init__(self, ollama_host: str | None = None, model_name: str | None = None, mock_response: str | None = None, prompt_version: str = 'v1')

def get_provider_name(self) -> str

def reconstruct_intent(self, tokens: list[CommandToken], context: UserContext) -> Intent

class HeuristicProvider(LLMProvider)

Fully offline intent reconstruction using the articulatory-distance decoder.

Resolves shorthand phrases against the command vocabulary and the active user context with no network access or API keys. Useful as a development default, an air-gapped deployment mode, and a graceful fallback when no LLM credentials are configured (see :func:`resolve_provider`).

def __init__(self, min_confidence: float = 0.0)

def get_provider_name(self) -> str

def reconstruct_intent(self, tokens: list[CommandToken], context: UserContext) -> Intent

resolve_provider

def resolve_provider(prefer: str | None = None, **kwargs) -> LLMProvider

Returns the best available LLM provider for this environment.

Selection order: an explicit ``prefer`` name ("anthropic", "openai", "gemini", "ollama", "heuristic"), then the first provider whose API key is present in the environment, then the offline :class:`HeuristicProvider`.

Args: prefer: Optional provider name to force. **kwargs: Passed through to the chosen provider constructor.

class Sample(BaseModel)

A single multichannel biometric time-series sample.

def validate_channels(cls, v: list[float]) -> list[float]

class Frame(BaseModel)

A window/segment of Samples buffered for signal processing or ML inference.

def validate_frame_ordering(self) -> 'Frame'

def to_numpy(self) -> np.ndarray

class CommandToken(BaseModel)

A token predicted by the raw classifier with associated confidence.

def validate_confidence(cls, v: float) -> float

class Intent(BaseModel)

Semantic intent reconstructed from a token stream and active context.

def validate_confidence(cls, v: float) -> float

class Action(BaseModel)

The executable instruction dispatched to the system or agent executor.

class PipelineStats

Running counters for a pipeline instance, exposed as ``pipeline.stats``.

def as_dict(self) -> dict[str, Any]

class SubvocalPipeline

Orchestrates hardware ingestion, classifier decoding, intent reconstruction, and action execution.

def __init__(self, hardware: HardwareSource, classify_fn: Callable[[Frame], CommandToken | None], llm_provider: LLMProvider, context_provider: ContextProvider, executor: ActionExecutor, phrase_timeout_seconds: float = 1.5, policy_engine: Any | None = None, dry_run: bool = False, trace_path: str | None = None, raise_on_policy_violation: bool = False, on_token: Callable[[CommandToken], None] | None = None, on_intent: Callable[[Intent], None] | None = None, on_action: Callable[[Action, str], None] | None = None, on_error: Callable[[Exception], None] | None = None, telemetry_service: Any | None = None)

def trace_path(self) -> str | None

def trace_path(self, val: str | None) -> None

def hardware(self) -> HardwareSource

def hardware(self, val: HardwareSource) -> None

def classify_fn(self) -> Callable[[Frame], CommandToken | None]

def classify_fn(self, val: Callable[[Frame], CommandToken | None]) -> None

def llm_provider(self) -> LLMProvider

def llm_provider(self, val: LLMProvider) -> None

def context_provider(self) -> ContextProvider

def context_provider(self, val: ContextProvider) -> None

def executor(self) -> ActionExecutor

def executor(self, val: ActionExecutor) -> None

def policy_engine(self) -> Any

def policy_engine(self, val: Any) -> None

def phrase_timeout_seconds(self) -> float

def phrase_timeout_seconds(self, val: float) -> None

def dry_run(self) -> bool

def dry_run(self, val: bool) -> None

def raise_on_policy_violation(self) -> bool

def raise_on_policy_violation(self, val: bool) -> None

def token_buffer(self) -> list[CommandToken]

def inject_token(self, token: CommandToken) -> None

def clear_buffer(self) -> None

def step(self, window_ms: int = 100) -> Action | None

def process_phrase(self) -> Action | None

class PromptManager

Manages versioned prompt templates for LLM intent decoding.

def __init__(self, default_version: str = 'v1')

def register_template(self, version: str, template: str) -> None

def get_template(self, version: str | None = None) -> str

def format_prompt(self, noisy_input: str, heuristic_recommendation: str, web_context: str, calendar: str, contacts: str, history: str, version: str | None = None) -> str

class AuthorizationPolicy(ABC)

Abstract base class for all subvocal execution policies.

def is_authorized(self, action: Action, context: UserContext) -> bool

class ConfidenceThresholdPolicy(AuthorizationPolicy)

Restricts execution of actions if classifier/reconstruction confidence is too low.

def __init__(self, threshold: float = 0.8)

def is_authorized(self, action: Action, context: UserContext) -> bool

class CommandWhitelistPolicy(AuthorizationPolicy)

Restricts commands to a strict whitelist of allowed actions.

def __init__(self, allowed_commands: list[str])

def is_authorized(self, action: Action, context: UserContext) -> bool

class ContextBoundPolicy(AuthorizationPolicy)

Blocks execution of sensitive commands unless the active application is safe.

def __init__(self, sensitive_commands: list[str], safe_applications: list[str])

def is_authorized(self, action: Action, context: UserContext) -> bool

class PolicyEngine

Orchestrates authorization consensus across multiple active policies.

def __init__(self, policies: list[AuthorizationPolicy] | None = None)

def add_policy(self, policy: AuthorizationPolicy) -> None

def is_authorized(self, action: Action, context: UserContext) -> bool

class MockHardwareSource(HardwareSource)

Generates continuous synthetic sEMG samples (sine + noise).

def __init__(self, fs: float = 1000.0, num_channels: int = 4)

def start(self) -> None

def stop(self) -> None

def read_frame(self, window_ms: int) -> Frame

def is_connected(self) -> bool

class MockLLMProvider(LLMProvider)

Maps shorthand token text patterns to GOTO or CLICK intents.

def reconstruct_intent(self, tokens: list[CommandToken], context: UserContext) -> Intent

def get_provider_name(self) -> str

class MockActionExecutor(ActionExecutor)

Records dispatched actions for assertion in tests.

def __init__(self)

def execute(self, action: Action) -> Any

def can_execute(self, action: Action) -> bool

class MockContextProvider(ContextProvider)

Returns a static browser UserContext snapshot.

def get_context(self) -> UserContext

def get_provider_name(self) -> str

class ContextManager

Manages active UserContext and provides phonetic searching utilities.

def __init__(self, context: UserContext)

def update_context(self, new_context: UserContext)

def get_contact_names(self) -> list[str]

def get_contact_shorthands(self) -> list[str]

def get_calendar_titles(self) -> list[str]

def get_calendar_shorthands(self) -> list[str]

def get_visible_element_labels(self) -> list[str]

def get_visible_element_shorthands(self) -> list[str]

def get_all_context_words(self) -> list[str]

def search_contacts(self, noisy_shorthand: str) -> list[tuple[Contact, float]]

def search_elements(self, noisy_shorthand: str) -> list[tuple[UIElement, float]]

generate_mock_context

def generate_mock_context() -> UserContext

Generate a high-fidelity UserContext populated with mock data.

class StaticContextProvider(ContextProvider)

Simple provider wrapping a static UserContext instance (useful for tests/mock runs).

def __init__(self, context: UserContext, name: str = 'static_context')

def get_context(self) -> UserContext

def get_provider_name(self) -> str

class CalendarContextProvider(ContextProvider)

Modular provider for retrieving calendar events.

def __init__(self, events: list[CalendarEvent] | None = None)

def get_provider_name(self) -> str

def get_context(self) -> UserContext

class ContactsContextProvider(ContextProvider)

Modular provider for retrieving user contact directory.

def __init__(self, contacts: list[Contact] | None = None)

def get_provider_name(self) -> str

def get_context(self) -> UserContext

class LocationContextProvider(ContextProvider)

Modular provider for retrieving current location details.

def __init__(self, location: LocationInfo | None = None)

def get_provider_name(self) -> str

def get_context(self) -> UserContext

class AppStateContextProvider(ContextProvider)

Modular provider for retrieving active app state and visible elements.

def __init__(self, app_state: AppState | None = None)

def get_provider_name(self) -> str

def get_context(self) -> UserContext

class CompositeContextProvider(ContextProvider)

Aggregates multiple individual ContextProviders into a single UserContext.

def __init__(self, providers: list[ContextProvider])

def get_provider_name(self) -> str

def get_context(self) -> UserContext

class Contact(BaseModel)

Represents a user contact contact.

class CalendarEvent(BaseModel)

Represents a calendar entry.

class LocationInfo(BaseModel)

Represents the user's current spatial coordinates and address.

class Message(BaseModel)

Represents a conversation turn in user-agent history.

class UIElement(BaseModel)

Represents an interactive user interface element visible on screen.

class AppState(BaseModel)

Represents the active application and on-screen context.

class UserContext(BaseModel)

Root model for the complete user context state.

class BoardIds(enum.IntEnum)

No description.

class LogLevels(enum.IntEnum)

No description.

class BrainFlowPresets(enum.IntEnum)

No description.

class BrainFlowError(Exception)

No description.

def __init__(self, message: str, exit_code: int)

class BrainFlowInputParams

No description.

def __init__(self) -> None

def to_json(self) -> str

class BoardShim

Fallback compatible BoardShim class implementing a pure-Python biosensor stream.

def __init__(self, board_id: int, input_params: BrainFlowInputParams) -> None

def prepare_session(self) -> None

def start_stream(self, buffer_size: int = 200000, streamer_params: str = '') -> None

def stop_stream(self) -> None

def release_session(self) -> None

def get_board_data(self, preset: int = BrainFlowPresets.DEFAULT_PRESET.value) -> np.ndarray

def get_current_board_data(self, num_samples: int, preset: int = BrainFlowPresets.DEFAULT_PRESET.value) -> np.ndarray

def is_prepared(self) -> bool

def get_sampling_rate(cls, board_id: int, preset: int = BrainFlowPresets.DEFAULT_PRESET.value) -> int

def get_num_rows(cls, board_id: int, preset: int = BrainFlowPresets.DEFAULT_PRESET.value) -> int

def get_eeg_channels(cls, board_id: int, preset: int = BrainFlowPresets.DEFAULT_PRESET.value) -> list[int]

def get_exg_channels(cls, board_id: int, preset: int = BrainFlowPresets.DEFAULT_PRESET.value) -> list[int]

def get_emg_channels(cls, board_id: int, preset: int = BrainFlowPresets.DEFAULT_PRESET.value) -> list[int]

def get_timestamp_channel(cls, board_id: int, preset: int = BrainFlowPresets.DEFAULT_PRESET.value) -> int

def get_package_num_channel(cls, board_id: int, preset: int = BrainFlowPresets.DEFAULT_PRESET.value) -> int

def get_battery_channel(cls, board_id: int, preset: int = BrainFlowPresets.DEFAULT_PRESET.value) -> int

def get_accel_channels(cls, board_id: int, preset: int = BrainFlowPresets.DEFAULT_PRESET.value) -> list[int]

def get_gyro_channels(cls, board_id: int, preset: int = BrainFlowPresets.DEFAULT_PRESET.value) -> list[int]

class NinaproDriver(HardwareSource)

Streams electromyography signals from Ninapro MATLAB (.mat) files.

def __init__(self, file_path: str, fs: float = 2000.0, loop: bool = True)

def start(self) -> None

def stop(self) -> None

def is_connected(self) -> bool

def read_frame(self, window_ms: int) -> Frame

class PutEMGDriver(HardwareSource)

Streams electromyography signals from PutEMG HDF5 (.h5) files.

def __init__(self, file_path: str, fs: float = 5120.0, loop: bool = True)

def start(self) -> None

def stop(self) -> None

def is_connected(self) -> bool

def read_frame(self, window_ms: int) -> Frame

class CSLHDEMGDriver(HardwareSource)

Streams electromyography signals from CSL-HDEMG binary or NumPy (.npy) files.

def __init__(self, file_path: str, fs: float = 2000.0, loop: bool = True, num_channels: int = 8)

def start(self) -> None

def stop(self) -> None

def is_connected(self) -> bool

def read_frame(self, window_ms: int) -> Frame

class GaddyDriver(HardwareSource)

Streams EMG from the Gaddy & Klein silent-speech dataset (Zenodo 4064409).

Dataset layout (per Zenodo): each utterance provides ``{id}_emg.npy`` shape ``(T, 8)`` raw EMG at 1000 Hz (ACL 2021 resampled to 800 Hz), ``{id}_audio.flac`` / ``_audio_clean.flac``, and ``{id}_info.json`` (contains ``text``/``transcript``/``prompt``). Utterances are sampled for both silent (``ES``) and vocalized (``EV``) modes; this driver indexes all ``*_emg.npy`` recursively and can filter by split.

Streaming emulates hardware by sequentially concatenating utterances and returning windowed :class:`Frame` objects.

Args: data_dir: Root directory of the extracted Zenodo archive (contains ``*_emg.npy`` files recursively). fs: Sampling frequency in Hz. Raw data is 1000 Hz; ACL 2021 model resamples to 800 Hz. Defaults to 1000.0. loop: Whether to loop to the first utterance when EOF is reached. split: Optional filter — ``"silent"`` or ``"vocalized"``. When set, only utterances whose path or ``info.json`` mode contains the term are retained. ``None`` retains all.

Raises: FileNotFoundError: If ``data_dir`` does not exist, with download instructions for https://zenodo.org/records/4064409.

def __init__(self, data_dir: str, fs: float = 800.0, loop: bool = True, split: str | None = None) -> None

def start(self) -> None

def stop(self) -> None

def is_connected(self) -> bool

def read_frame(self, window_ms: int) -> Frame

def list_utterances(self) -> list[str]

def get_transcript(self, utterance_id: str | None = None) -> str

def get_audio_path(self, utterance_id: str) -> str | None

def get_info(self, utterance_id: str) -> dict[str, object]

class MetaEMGDriver(HardwareSource)

Driver for Meta's sEMG-RD emg2pose / surface-typing datasets.

Open-sourced Dec 2024 (blog_), Nature 2025 (Sivakumar & Landau et al. "Generic hand pose tracking..." + Salter et al. NeurIPS 2024 emg2pose ``arXiv:2412.02725``). Wristband is the sEMG-RD: 16 bipolar channels per wrist at 2 kHz (12-bit), paired with hand-pose labels at 2 kHz from a 26-camera motion-capture rig (19 markers/hand → IK joint angles, linearly interpolated to 2 kHz).

Layout (HDF5 per recording, 25k files / 193 participants / 370 h):

* ``/emg`` → ``(T, 16)`` float32 (also accepted: ``raw_emg``, ``semg``) * ``/pose`` → ``(T, D)`` float32 (also accepted: ``joint_angles``, ``hand_pose``, ``joints``; typically 63 = 21 joints × 3 or 84-dim)

Also supports unpacked NPY layout: ``*_emg.npy`` paired with ``*_pose.npy`` / ``*_joint_angles.npy``.

Streaming concatenates recordings sequentially and exposes :class:`Frame` (EMG) via the :class:`HardwareSource` interface; pose is available via :meth:`load_pair`, :meth:`get_pose`, and :meth:`read_frame_with_pose`.

Args: data_dir: Root of the extracted archive (may contain ``train`` / ``val`` / ``test`` subfolders; see blog). fs: Sampling frequency in Hz (sEMG-RD is 2000.0). split: Dataset split — ``"train"`` / ``"val"`` / ``"test"``. Sanitized to ``[A-Za-z0-9_-]`` and used to scope to ``data_dir/split`` if that subdirectory exists, otherwise filtered by path substring. ``"all"`` disables filtering. loop: Whether to loop when recordings are exhausted.

Raises: FileNotFoundError: If ``data_dir`` is missing or contains no recordings, with instructions to download from https://ai.meta.com/blog/open-sourcing-surface-electromyography-datasets-neurips-2024 and https://github.com/facebookresearch/emg2pose.

.. _blog: https://ai.meta.com/blog/open-sourcing-surface-electromyography-datasets-neurips-2024

def __init__(self, data_dir: str, fs: float = 2000.0, split: str = 'train', loop: bool = True) -> None

def start(self) -> None

def stop(self) -> None

def is_connected(self) -> bool

def read_frame(self, window_ms: int) -> Frame

def list_recordings(self) -> list[str]

def load_pair(self, idx: int | str) -> tuple[np.ndarray, np.ndarray]

def get_emg(self, idx: int | str) -> np.ndarray

def get_pose(self, idx: int | str) -> np.ndarray

def read_frame_with_pose(self, window_ms: int) -> tuple[Frame, np.ndarray]

class FileReplayDriver(HardwareSource)

Replays raw sEMG data from a CSV file, simulating real-time acquisition.

def __init__(self, file_path: str, fs: float = 1000.0, loop: bool = True)

def start(self) -> None

def stop(self) -> None

def is_connected(self) -> bool

def read_frame(self, window_ms: int) -> Frame

class SyntheticSignalGenerator(HardwareSource)

Generates continuous physiological sEMG noise and injects command triggers.

def __init__(self, fs: float = 1000.0, num_channels: int = 8)

def start(self) -> None

def stop(self) -> None

def is_connected(self) -> bool

def trigger_command(self, command: str, duration_ms: int = 300) -> None

def read_frame(self, window_ms: int) -> Frame

class OpenBCICytonDriver(HardwareSource)

Acquires raw biosignals from OpenBCI Cyton boards via BrainFlow.

def __init__(self, port: str | None = None, simulated: bool = True)

def start(self) -> None

def stop(self) -> None

def is_connected(self) -> bool

def read_frame(self, window_ms: int) -> Frame

class DelsysTrignoDriver(HardwareSource)

Direct TCP Socket driver connecting to the Delsys Trigno Control Utility.

def __init__(self, host: str = 'localhost', data_port: int = 50043, cmd_port: int = 50040, num_channels: int = 8, fs: float = 2000.0)

def start(self) -> None

def stop(self) -> None

def is_connected(self) -> bool

def read_frame(self, window_ms: int) -> Frame

class _BaseTransformerEncoder(nn.Module)

Shared Transformer base: Linear -> pos -> 4-layer Transformer -> 1-layer head -> L2.

def __init__(self, in_dim: int, seq_len: int = 150, d_model: int = 256, nhead: int = 8, num_layers: int = 4, dim_feedforward: int = 512, dropout: float = 0.1, embed_dim: int = 256, use_cls: bool = True) -> None

def forward(self, x: Any) -> Any

class EMGEncoder(_BaseTransformerEncoder)

EMG Transformer encoder: 4 layers, dim 256, 1-layer projection to 256-d, L2 norm.

Args: in_channels: Number of EMG channels (default 8). seq_len: Time length of EMG segment (default 150). d_model: Transformer hidden dim (default 256). nhead: Attention heads (default 8). num_layers: Transformer layers (default 4 per CPEP). dim_feedforward: FFN dim (default 512). dropout: Dropout. embed_dim: Output embedding dim (default 256 per CPEP). use_cls: Use [CLS] token as sequence summary (paper uses CLS).

Input: Tensor shape ``(B,C,T)`` or ``(B,T,C)`` or ``(C,T)``. Output: L2-normalized embedding ``(B, embed_dim)``.

def __init__(self, in_channels: int = 8, seq_len: int = 150, d_model: int = 256, nhead: int = 8, num_layers: int = 4, dim_feedforward: int = 512, dropout: float = 0.1, embed_dim: int = 256, use_cls: bool = True, **kwargs: Any) -> None

class PoseEncoder(_BaseTransformerEncoder)

Pose Transformer encoder: mirror of EMGEncoder for pose (4 layers, dim 256).

Args: pose_dim: Input pose dimensionality per frame (e.g. 63 for 21 joints x3, 42, or flattened heatmap dims). Default 63. seq_len: Temporal length (default 150 to match EMG). d_model: Hidden dim 256. nhead: Heads 8. num_layers: 4. dim_feedforward: 512. dropout: 0.1. embed_dim: 256 (shared space). use_cls: Use [CLS] token.

Input: Tensor shape ``(B,T,D)`` or ``(B,D,T)`` or ``(T,D)`` where D==pose_dim. Both layouts are accepted (auto-detected).

Output: L2-normalized embedding ``(B, embed_dim)``.

def __init__(self, pose_dim: int = 63, seq_len: int = 150, d_model: int = 256, nhead: int = 8, num_layers: int = 4, dim_feedforward: int = 512, dropout: float = 0.1, embed_dim: int = 256, use_cls: bool = True, **kwargs: Any) -> None

class EMGEncoder

Stub when torch not installed.

def __init__(self, *args: Any, **kwargs: Any) -> None

def forward(self, x: Any) -> Any

def parameters(self) -> Any

def to(self, *a: Any, **k: Any) -> Any

def eval(self) -> Any

def train(self, *a: Any, **k: Any) -> Any

class PoseEncoder

Stub when torch not installed.

def __init__(self, *args: Any, **kwargs: Any) -> None

def forward(self, x: Any) -> Any

def parameters(self) -> Any

def to(self, *a: Any, **k: Any) -> Any

def eval(self) -> Any

def train(self, *a: Any, **k: Any) -> Any

class CPEPFramework(nn.Module)

Contrastive Pose-EMG Pretraining framework (CPEP).

Wraps :class:`EMGEncoder` and :class:`PoseEncoder` with a learnable temperature ``tau`` (init 0.02) and symmetric InfoNCE. During contrastive training the pose encoder is frozen (paper: only EMG encoder + projection head are updated) — controlled by ``freeze_pose``.

Args: emg_channels: EMG input channels (default 8). pose_dim: Pose per-frame dim (default 63). seq_len: EMG temporal length (default 150). pose_seq_len: Pose temporal length (default seq_len if None). d_model: Transformer hidden dim (default 256). embed_dim: Shared embedding dim (default 256 per CPEP). nhead: Attention heads (default 8). num_layers: Transformer layers (default 4). dim_feedforward: FFN dim (default 512). dropout: Dropout (default 0.1). tau_init: Initial temperature (default 0.02). freeze_pose: If True, pose encoder params are frozen (non-trainable) — matches CPEP Sec. 2.3. use_cls: Use [CLS] token (default True, per paper).

Attributes: emg_encoder: :class:`EMGEncoder` pose_encoder: :class:`PoseEncoder` tau: Learnable temperature scalar ``nn.Parameter``.

Reference: Cui et al., arXiv:2509.04699, Sec. 2.3.

def __init__(self, emg_channels: int = 8, pose_dim: int = 63, seq_len: int = 150, pose_seq_len: int | None = None, d_model: int = 256, embed_dim: int = 256, nhead: int = 8, num_layers: int = 4, dim_feedforward: int = 512, dropout: float = 0.1, tau_init: float = 0.02, freeze_pose: bool = True, use_cls: bool = True) -> None

def forward(self, emg: Any, pose: Any) -> Any

def encode_emg(self, emg: Any) -> Any

def encode_pose(self, pose: Any) -> Any

def get_tau(self) -> float

def zero_shot_predict(self, query_emg: Any, gallery_pose_embs: Any, gallery_labels: Any, k: int = 10, metric: str = 'cosine') -> np.ndarray

def classify_zero_shot(self, *a: Any, **k: Any) -> np.ndarray

def zero_shot_classify(self, *a: Any, **k: Any) -> np.ndarray

def predict_zero_shot(self, *a: Any, **k: Any) -> np.ndarray

class CPEPFramework

Stub when torch not installed.

def __init__(self, *args: Any, **kwargs: Any) -> None

def forward(self, emg: Any, pose: Any) -> Any

def encode_emg(self, emg: Any) -> Any

def encode_pose(self, pose: Any) -> Any

def zero_shot_predict(self, *a: Any, **k: Any) -> Any

l2_normalize_embeddings

def l2_normalize_embeddings(x: Any, dim: int = -1, eps: float = 1e-08) -> Any

L2-normalize embeddings along ``dim`` (torch or numpy).

pose_emg_contrastive_loss

def pose_emg_contrastive_loss(emg_emb: Any, pose_emb: Any, tau: Any = 0.02) -> Any

Symmetric InfoNCE for pose-EMG alignment (CPEP).

Both embeddings are L2-normalized and cosine similarity is scaled by temperature ``tau`` (learnable, init 0.02 per paper):

s_ij = (u_i·v_j)/tau, u=norm(emg), v=norm(pose)

Loss averages two CE losses over batch:

L = 0.5*(CE(logits, arange(N)) + CE(logits.T, arange(N))) logits = u @ v.T / tau

Args: emg_emb: ``(N,D)`` embeddings (Tensor or ndarray). pose_emb: ``(N,D)`` embeddings. tau: Temperature scalar — float or 0-d Tensor (learnable). Clamped to ``>=1e-4`` to avoid division blow-up. Init 0.02.

Returns: Scalar loss Tensor (torch) or float (numpy fallback).

Reference: CPEP Sec. 2.3, CLIP symmetric InfoNCE.

knn_classify

def knn_classify(train_embs: Any, train_labels: Any, query_embs: Any, k: int = 5, metric: str = 'cosine') -> np.ndarray

Zero-shot kNN classification in embedding space (CPEP ZS protocol).

For each query embedding, retrieve TopK nearest training embeddings by cosine similarity (L2-normalized dot) or Euclidean, then majority vote. Paper uses ``k=10``.

Args: train_embs: Gallery embeddings ``(G,D)`` (ndarray or Tensor). train_labels: Gallery labels ``(G,)`` (array-like). query_embs: Query embeddings ``(Q,D)``. k: Number of neighbors (default 5; CPEP uses 10). Clamped to ``min(k,G)``. metric: ``"cosine"`` (default, as CPEP) or ``"euclidean"``.

Returns: Predicted labels ``(Q,)`` ndarray with same dtype as train_labels.

Handles small data (G < k, single sample, etc.) and works without sklearn/torch (pure numpy).

embedding_knn_predict

def embedding_knn_predict(*a: Any, **k: Any) -> np.ndarray

No description.

zero_shot_predict

def zero_shot_predict(*a: Any, **k: Any) -> np.ndarray

No description.

class SAL(nn.Module)

Spatial Adaptation Layer (SAL) per Pereira et al., arXiv:2409.08058.

Learnable, prependable input transform that warps a new session's biosignals back to the original spatial frame. For HD-sEMG grids (default 8×16) a 2-D affine (``2×3``, 6 DoF) is applied via ``affine_grid`` + ``grid_sample``; the translation-only variant keeps 2 DoF ``(tx, ty)``. For any montage a per-channel affine ``x * scale + shift`` (``2*C`` params, identity init) is always present — the paper's 7 params are realised as 6 affine + global scale bias, here decomposed as per-channel gain/bias + spatial warp.

Args: num_channels: Number of EMG channels (default 8; 128 for 8×16 HD). use_affine: If True, learn full 2×3 affine (6 params). Ignored if ``use_translation_only`` is True. use_translation_only: If True, only 2 translation params.

Input: ``(B,C,T)`` or ``(B,T,C)`` Tensor (or ``(C,T)`` single sample), or ``np.ndarray`` of same shapes (handled via torch conversion).

Output: Tensor/ndarray of same shape as input.

Reference: Pereira et al., arXiv:2409.08058, Sec. 2–3.

def __init__(self, num_channels: int = 8, use_affine: bool = True, use_translation_only: bool = False) -> None

def get_affine_params(self) -> Any

def forward(self, x: Any) -> Any

class LBN(nn.Module)

Learnable Baseline Normalisation (LBN) per Pereira et al.

Per-channel learnable bias subtraction ``x - bias``. Bias is initialised to zero (identity). Inspired by baseline normalisation in Gaddy & Klein 2020 / JNE 2024 [4] where resting baseline is removed per channel; here the baseline is learned jointly with SAL and the backbone.

Args: num_channels: Number of EMG channels.

Reference: Pereira et al., arXiv:2409.08058, Sec. 3.2; baseline norm [4].

def __init__(self, num_channels: int = 8) -> None

def forward(self, x: Any) -> Any

class SAL_LBN(nn.Module)

Combined SAL + LBN adaptation block.

Sequentially applies :class:`SAL` then :class:`LBN`: ``forward(x) == LBN(SAL(x))``. Prependable to any model: ``logits = backbone(sal_lbn(emg))``.

Args: num_channels: Number of EMG channels. use_affine: Passed to :class:`SAL`. use_translation_only: Passed to :class:`SAL`.

Reference: Pereira et al., arXiv:2409.08058 — SAL/LBN combined.

def __init__(self, num_channels: int = 8, use_affine: bool = True, use_translation_only: bool = False) -> None

def forward(self, x: Any) -> Any

def get_affine_params(self) -> Any

class SAL

SAL numpy fallback (torch not installed).

Same API as torch SAL but using numpy arrays. Per-channel ``x * scale + shift`` with learnable 2*C params; affine 2×3 and translation 2-vector stored as numpy. Grid warp is no-op in pure numpy (per-channel only) to keep fallback lightweight.

def __init__(self, num_channels: int = 8, use_affine: bool = True, use_translation_only: bool = False) -> None

def get_affine_params(self) -> np.ndarray

def forward(self, x: np.ndarray) -> np.ndarray

def parameters(self) -> list[Any]

def train(self, mode: bool = True) -> None

def eval(self) -> None

def to(self, device: str) -> Any

class LBN

LBN numpy fallback.

def __init__(self, num_channels: int = 8) -> None

def forward(self, x: np.ndarray) -> np.ndarray

def parameters(self) -> list[Any]

def train(self, mode: bool = True) -> None

def eval(self) -> None

def to(self, device: str) -> Any

class SAL_LBN

SAL_LBN numpy fallback.

def __init__(self, num_channels: int = 8, use_affine: bool = True, use_translation_only: bool = False) -> None

def forward(self, x: np.ndarray) -> np.ndarray

def get_affine_params(self) -> np.ndarray

def parameters(self) -> list[Any]

def train(self, mode: bool = True) -> None

def eval(self) -> None

def to(self, device: str) -> Any

adapt_sal_lbn

def adapt_sal_lbn(model: Any, sal_lbn: Any, train_loader: Any, epochs: int = 5, lr: float = 0.001, device: str = 'cpu', criterion: Any | None = None) -> dict[str, Any]

Supervised adaptation of SAL/LBN on a new session.

Freezes ``model`` and optimises only ``sal_lbn`` parameters via a supervised loss (``CrossEntropy`` for classification by default).

The ``sal_lbn`` block is prepended: ``logits = model(sal_lbn(x))``. Typical usage (Pereira et al., Sec. 4 — <2 min calibration data)::

sal_lbn = SAL_LBN(num_channels=8) adapt_sal_lbn(backbone, sal_lbn, calib_loader, epochs=5, lr=1e-3) logits = backbone(sal_lbn(emg_new_session))

Args: model: Backbone ``nn.Module`` (e.g. CNN/GRU). Frozen in-place (``requires_grad=False``) for the duration. sal_lbn: :class:`SAL` | :class:`LBN` | :class:`SAL_LBN` instance. train_loader: Iterable yielding ``(x, y)`` batches or dicts with ``x``/``y``. ``x`` shape ``(B,C,T)`` or ``(B,T,C)``. epochs: Number of adaptation epochs (default 5, paper uses 3-5). lr: Learning rate for SAL/LBN (default 1e-3). device: Torch device. criterion: Loss fn; defaults to ``CrossEntropyLoss``. May be ``MSELoss`` for regression targets.

Returns: Dict with ``loss_history``, ``epochs``, ``lr``, ``final_loss``.

Raises: MissingDependencyError: if torch not installed.

Reference: Pereira et al., arXiv:2409.08058, Sec. 4 — session adaptation.

class GaussianClassificationModel

Per-class Gaussian classifier with shared-variance semantics.

Attributes: means: ``(C,D)`` per-class means (subject-specific after transfer). precisions: ``(C,D,D)`` per-class precision matrices (shared variance origin, after Bayesian update). covariances: ``(C,D,D)`` derived covariances (inv precisions). classes: Unique label values ``(C,)`` (original dtype preserved). priors: Class priors ``(C,)`` summing to 1 (uniform if None). n_classes: Number of classes. n_features: Feature dimension D. class_to_idx: Mapping label -> index.

Prediction is QDA/LDA-like via Gaussian log-likelihood with precision form:

log p(x|k) = -0.5*(x-mu_k)^T Lambda_k (x-mu_k) +0.5*log|Lambda_k| -0.5*D*log(2pi)

plus log prior.

Reference: Yoneda et al., Sec. II-B GCM.

def shared_covariance(self) -> np.ndarray

def variances(self) -> np.ndarray

def predict(self, X: np.ndarray) -> np.ndarray

def predict_proba(self, X: np.ndarray) -> np.ndarray

class VarianceTransferGCM

End-to-end variance-transfer GCM wrapper (Yoneda et al.).

Convenience class that holds the source posterior and target model, delegating to :func:`pretrain_variance_transfer`, :func:`transfer_to_target`, and :func:`predict`.

Example: vt = VarianceTransferGCM() post = vt.pretrain(source_data, source_labels) model = vt.transfer(post, calib_data, calib_labels, w_s=1.0) preds = vt.predict(model, test_data) # or vt.predict(test_data)

# or functional: post = pretrain_variance_transfer(src_data, src_labels) model = transfer_to_target(post, calib_data, calib_labels, w_s=1.0) preds = predict(model, test_data)

Attributes: posterior: Last computed posterior dict (or None). model: Last fitted :class:`GaussianClassificationModel` (or None).

Reference: Yoneda et al., EMBC 2024 / arXiv:2505.15381.

def __init__(self) -> None

def pretrain(self, source_data: list[np.ndarray], source_labels: list[np.ndarray]) -> dict[str, Any]

def transfer(self, posterior: dict[str, Any] | None, target_calib_data: np.ndarray, target_calib_labels: np.ndarray, w_s: float = 1.0) -> GaussianClassificationModel

def adapt(self, target_calib_data: np.ndarray, target_calib_labels: np.ndarray, w_s: float = 1.0, posterior: dict[str, Any] | None = None) -> GaussianClassificationModel

def fit_target(self, posterior: dict[str, Any], target_calib_data: np.ndarray, target_calib_labels: np.ndarray, w_s: float = 1.0) -> GaussianClassificationModel

def predict(self, test_data: np.ndarray, model: GaussianClassificationModel | None = None) -> np.ndarray

def transfer_to_target(self, posterior: dict[str, Any], target_calib_data: np.ndarray, target_calib_labels: np.ndarray, w_s: float = 1.0) -> GaussianClassificationModel

def pretrain_variance_transfer(self, source_data: list[np.ndarray], source_labels: list[np.ndarray]) -> dict[str, Any]

def fit(self, *a: Any, **k: Any) -> GaussianClassificationModel

pretrain_variance_transfer

def pretrain_variance_transfer(source_data: list[np.ndarray], source_labels: list[np.ndarray]) -> dict[str, Any]

Pre-train shared variance posterior from multiple source subjects.

Pools EMG feature vectors per class across subjects to estimate per-class Wishart posterior for shared precision. Means remain subject-specific -> not pooled beyond class scatter.

Args: source_data: List of feature arrays per source subject. Each element shape ``(N_s, D)`` (2D) or ``(N_s,)`` for D=1. Can also be 1D per trial — stacked internally. source_labels: List of label arrays per source subject, each shape ``(N_s,)`` with same length as corresponding data.

Returns: Posterior dict with keys: ``classes`` (np.ndarray ``C``), ``n_classes``, ``n_features``, ``W`` (``C,D,D`` Wishart scale), ``nu`` (``C``), ``beta`` (``C``), ``m`` (``C,D`` posterior mean estimates), plus prior hyperparams ``beta0``, ``m0``, ``nu0``, ``W0`` for downstream transfer. Suitable to pass as ``posterior`` to :func:`transfer_to_target`.

Reference: Yoneda et al., Sec. II-C1 Pre-training with source subjects. Prior hyperparams fixed: m0=0, beta0=1, nu0=D+1, W0=I.

transfer_to_target

def transfer_to_target(posterior: dict[str, Any], target_calib_data: np.ndarray, target_calib_labels: np.ndarray, w_s: float = 1.0) -> GaussianClassificationModel

Transfer shared variance to target subject with limited calibration.

Bayesian update using scaled source posterior as prior for shared precision, plus target calibration data (e.g. 1 trial). Only variance (precision) is transferred; means are re-estimated from target data (subject-specific).

The ``w_s`` coefficient controls uncertainty of the source posterior: larger ``w_s`` = less uncertainty = more influence. Scaling keeps expected precision ``E[Lambda]=nu*W`` constant while increasing concentration ``nu`` with ``w_s`` (variance of Wishart decreases).

nu_scaled = max(D+1, w_s * nu_src) W_scaled = W_src * (nu_src / nu_scaled) # keep mean fixed

For each class, target posterior:

beta_post = beta0 + N_t m_post = (beta0*m0 + N_t*x_bar_t)/beta_post W_post^-1 = W_scaled^-1 + S_t + (beta0*N_t/beta_post)*(x_bar_t-m0)(x_bar_t-m0)^T nu_post = nu_scaled + N_t precision = nu_post * W_post

Handles small data (N_t=0 or 1) with regularization; unseen classes get posterior from scaled prior alone.

Args: posterior: Dict from :func:`pretrain_variance_transfer`. target_calib_data: Calibration features ``(N_t, D)`` or ``(N_t, )``. Single-trial calibration typical (e.g. few samples per class). target_calib_labels: Labels ``(N_t,)`` matching data rows. w_s: Weight coefficient ``>0`` controlling transferred amount. ``1.0`` default (no scaling). Larger -> stronger transfer.

Returns: :class:`GaussianClassificationModel` fitted to target (subject-specific means + variance-transferred precisions).

Reference: Yoneda et al., Sec. II-C2 Transfer learning to target, Eq. 13-22.

predict

def predict(model: GaussianClassificationModel, test_data: np.ndarray) -> np.ndarray

Predict labels for test_data using a :class:`GaussianClassificationModel`.

Args: model: Fitted :class:`GaussianClassificationModel`. test_data: Features ``(N,D)`` or ``(D,)`` single sample. Can be list of arrays; converted via ``np.asarray``.

Returns: Predicted labels ``(N,)`` ndarray.

class EMGBench

EMGBench evaluation harness (NeurIPS 2024).

Args: datasets: List of 9 EMGBench dataset slugs (e.g. ``"ninapro-db5"``, ``"capgmyo-db-b"``). Sanitized to ``[A-Za-z0-9._-]``. If empty, :data:`DEFAULT_DATASETS` is used. root_dir: Root directory containing per-dataset subfolders (``root_dir/<dataset>/...``) or the ``DatasetsProcessed_hdf5`` parent. Traversal-sanitized.

Example::

bench = EMGBench(datasets=DEFAULT_DATASETS, root_dir="DatasetsProcessed_hdf5") loso = bench.evaluate_loso(my_model_fn, dataset="ninapro-db5") adapt = bench.evaluate_adaptation(my_model_fn, dataset="ninapro-db5", n_shot=5) all_res = bench.evaluate_all(my_model_fn, mode="loso")

def __init__(self, datasets: list[str], root_dir: str) -> None

def evaluate_loso(self, model_fn: Callable[..., Any], dataset: str) -> dict[str, Any]

def evaluate_adaptation(self, model_fn: Callable[..., Any], dataset: str, n_shot: int = 1) -> dict[str, Any]

def evaluate_all(self, model_fn: Callable[..., Any], mode: str = 'loso', n_shot: int = 1) -> dict[str, Any]

def run_all(self, model_fn: Callable[..., Any], mode: str = 'loso', n_shot: int = 1) -> dict[str, Any]

summarize_results

def summarize_results(results: dict[str, Any]) -> dict[str, float]

Aggregate an :meth:`EMGBench.evaluate_all` result to mean/std.

Args: results: Dict returned by :meth:`EMGBench.evaluate_all`.

Returns: Dict with ``mean``, ``std``, ``min``, ``max`` over dataset means.

aggregate_loso_scores

def aggregate_loso_scores(per_dataset_results: dict[str, dict[str, Any]]) -> dict[str, float]

Backwards-compatible alias for :func:`summarize_results` on per-dataset dict.

class FilterTypes(enum.IntEnum)

No description.

class AggOperations(enum.IntEnum)

No description.

class WindowOperations(enum.IntEnum)

No description.

class DetrendOperations(enum.IntEnum)

No description.

class NoiseTypes(enum.IntEnum)

No description.

class DataFilter

Pure Python/SciPy re-implementation of BrainFlow's DataFilter processing methods.

def perform_lowpass(cls, data: np.ndarray, sampling_rate: int, cutoff: float, order: int, filter_type: int, ripple: float) -> None

def perform_highpass(cls, data: np.ndarray, sampling_rate: int, cutoff: float, order: int, filter_type: int, ripple: float) -> None

def perform_bandpass(cls, data: np.ndarray, sampling_rate: int, start_freq: float, stop_freq: float, order: int, filter_type: int, ripple: float) -> None

def perform_bandstop(cls, data: np.ndarray, sampling_rate: int, start_freq: float, stop_freq: float, order: int, filter_type: int, ripple: float) -> None

def remove_environmental_noise(cls, data: np.ndarray, sampling_rate: int, noise_type: int) -> None

def perform_rolling_filter(cls, data: np.ndarray, period: int, agg_operation: int) -> None

def perform_downsampling(cls, data: np.ndarray, period: int, agg_operation: int) -> np.ndarray

def get_window(cls, window_function: int, window_len: int) -> np.ndarray

def get_psd(cls, data: np.ndarray, sampling_rate: int, window: int) -> tuple[np.ndarray, np.ndarray]

def get_psd_welch(cls, data: np.ndarray, nfft: int, overlap: int, sampling_rate: int, window: int) -> tuple[np.ndarray, np.ndarray]

def get_band_power(cls, psd: tuple[np.ndarray, np.ndarray], freq_start: float, freq_end: float) -> float

extract_features_td10

def extract_features_td10(segment: np.ndarray, sample_rate: float = 250.0, frame_size_ms: int = 27, frame_shift_ms: int = 10, context: int = 10) -> np.ndarray

Extract TD10 features for all channels.

Returns: 2D array: (num_frames, num_channels * 5 * (2*context + 1))

extract_features_td10_segment

def extract_features_td10_segment(segment: np.ndarray, sample_rate: float = 250.0, frame_size_ms: int = 27, frame_shift_ms: int = 10, context: int = 10) -> np.ndarray

Aggregate frame-level TD10 features into a single 1D segment feature vector.

Computes mean and standard deviation across frames. For 4 channels with context=10: 4 * 5 * 21 * 2 = 840 dimensional vector.

extract_features

def extract_features(segment: np.ndarray, sample_rate: float = 250.0) -> np.ndarray

Extract features from a multi-channel segment using TD10 pipeline.

extract_features_batch

def extract_features_batch(segments: list[np.ndarray], sample_rate: float = 250.0) -> np.ndarray

Extract features for a batch of segments.

remove_dc

def remove_dc(signal: np.ndarray, window_samples: int = 250) -> np.ndarray

Remove DC offset by subtracting a rolling mean.

Args: signal: 1D array of samples for a single channel. window_samples: Window size in samples (default 250 = 1s at 250Hz).

Returns: DC-removed signal.

bandpass

def bandpass(signal: np.ndarray, fs: float = 250.0, low: float = 1.3, high: float = 50.0, order: int = 4) -> np.ndarray

Butterworth bandpass filter for raw sEMG.

Args: signal: 1D array of samples. fs: Sampling frequency in Hz. low: Low cutoff frequency. high: High cutoff frequency. order: Filter order.

Returns: Bandpass-filtered signal.

notch_60hz

def notch_60hz(signal: np.ndarray, fs: float = 250.0, freq: float = 60.0, Q: float = 30.0) -> np.ndarray

Notch filter to remove power line interference.

Args: signal: 1D array of samples. fs: Sampling frequency. freq: Notch frequency (60 Hz US, 50 Hz EU). Q: Quality factor (higher = narrower notch).

Returns: Notch-filtered signal.

smooth

def smooth(signal: np.ndarray, window: int = 3) -> np.ndarray

Simple moving average smoothing.

Args: signal: 1D array of samples. window: Window size in samples.

Returns: Smoothed signal.

preprocess_channel

def preprocess_channel(signal: np.ndarray, fs: float = 250.0, low: float = 1.3, high: float = 50.0, notch_freq: float = _DEFAULT_NOTCH_FREQ, notch_q: float = _DEFAULT_NOTCH_Q, apply_bandpass: bool = True, apply_notch: bool = True) -> np.ndarray

Full preprocessing pipeline for a single channel.

Args: signal: 1D float array of raw ADC values. fs: Sample rate. low: Low cutoff frequency for bandpass. high: High cutoff frequency for bandpass. notch_freq: Notch filter frequency (e.g., 60.0 for US, 50.0 for EU). notch_q: Notch filter Q factor. apply_bandpass: Whether to apply bandpass. apply_notch: Whether to apply notch filter.

Returns: Preprocessed signal.

preprocess_multichannel

def preprocess_multichannel(data: np.ndarray, fs: float = 250.0, low: float = 1.3, high: float = 50.0, notch_freq: float = _DEFAULT_NOTCH_FREQ, notch_q: float = _DEFAULT_NOTCH_Q, apply_bandpass: bool = True, apply_notch: bool = True) -> np.ndarray

Preprocess all channels.

Args: data: 2D array of shape (num_samples, num_channels). fs: Sample rate. low: Low cutoff frequency for bandpass. high: High cutoff frequency for bandpass. notch_freq: Notch filter frequency (e.g., 60.0 for US, 50.0 for EU). notch_q: Notch filter Q factor.

Returns: Preprocessed data of same shape.

extract_handcrafted_features

def extract_handcrafted_features(segment: np.ndarray, fs: float = 250) -> np.ndarray

Extract 112 handcrafted sEMG features for a segment.

Args: segment: Array of shape ``(T, C)`` – ``T`` time samples, ``C`` channels. Typical subvocal segment is ``(150, 4)`` at ``fs=250 Hz``. fs: Sampling rate in Hz.

Returns: Array of shape ``(112,)`` (``C * 28`` for ``C=4``). For ``C != 4`` the output is ``(C*28,)`` padded/truncated to 112 to satisfy the adaptor ``input_dim=112`` contract.

References: Mohapatra et al., ACL 2025; Jou et al., 2006; Gaddy & Klein, 2020.

extract_handcrafted_timevarying

def extract_handcrafted_timevarying(segment: np.ndarray, fs: float = 250, window_ms: int = 50, step_ms: int = 20) -> np.ndarray

Time-varying 112-d features via sliding windows.

Splits *segment* into overlapping windows of ``window_ms`` with step ``step_ms`` and extracts :func:`extract_handcrafted_features` per window.

Args: segment: Array of shape ``(T, C)``. fs: Sampling rate in Hz. window_ms: Window length in milliseconds. step_ms: Step / hop in milliseconds.

Returns: Array of shape ``(num_windows, 112)``.

References: Mohapatra et al., ACL 2025; Jou et al., 2006.

compute_covariance_matrix

def compute_covariance_matrix(segment: np.ndarray) -> np.ndarray

Compute channel covariance ``(C, C)`` for a segment.

Args: segment: Array ``(T, C)`` — ``T`` time samples, ``C`` channels. Convention follows :mod:`subvocal.emg_core.dsp.filters` (``(num_samples, num_channels)``). A ``(C, T)`` array with ``C < T`` is **not** transposed automatically; pass ``(T, C)``.

Returns: Covariance matrix ``(C, C)``, symmetric positive semi-definite. Shape is ``(C, C)`` where ``C = segment.shape[1]``.

References: Gowda & Miller, ACL 2026 Findings, Sec. 3.1 (sample covariance on mV-normalised sEMG windows); J Neural Eng 2024, Eq. 1.

compute_spd_matrix

def compute_spd_matrix(segment: np.ndarray, eps: float = 1e-06) -> np.ndarray

Compute SPD regularised covariance ``(C, C)``.

Adds ``eps * I`` to the sample covariance so the result is strictly symmetric positive-definite even when the segment is rank-deficient (e.g., constant channel or ``T < C``).

Args: segment: Array ``(T, C)``. eps: Ridge regulariser added to the diagonal (default ``1e-6``). Must be non-negative; ``1e-6`` follows ACL 2026 (``1e-6*I``).

Returns: SPD matrix ``(C, C)``.

References: Gowda & Miller, ACL 2026 Findings, Sec. 3.2 (``C + eps I``); J Neural Eng 2024, Eq. 2.

compute_spd_timevarying

def compute_spd_timevarying(segment: np.ndarray, fs: float = 250, window_ms: int = 50, step_ms: int = 20, eps: float = 1e-06) -> np.ndarray

Sliding-window SPD sequence ``(num_windows, C, C)``.

Splits *segment* into overlapping windows of length ``window_ms`` with hop ``step_ms`` and computes an SPD matrix per window via :func:`compute_spd_matrix`. Overlap (``step < window``) yields dense temporal sampling for the downstream SPD-GRU (ACL 2026, Sec. 4).

Args: segment: Array ``(T, C)``. fs: Sampling rate in Hz. window_ms: Window length in ms (default 50 ms → 13 samples @250 Hz rounded; actual ``window_samples = round(fs*window_ms/1000)``). step_ms: Hop / shift in ms (default 20 ms). eps: SPD regulariser per window.

Returns: Array ``(num_windows, C, C)``. If ``T < window_samples`` the segment is zero-padded to produce a single window (consistent with :func:`subvocal.emg_core.dsp.handcrafted.extract_handcrafted_timevarying`).

References: Gowda & Miller, ACL 2026 Findings, Sec. 4.1 (50 ms / 20 ms sliding SPD); J Neural Eng 2024, Sec. 3.3 (overlap windowing).

spd_logm

def spd_logm(spd: np.ndarray, eps: float = 1e-06) -> np.ndarray

Matrix logarithm via eigendecomposition (``V log(D) V^T``).

Uses :func:`scipy.linalg.eigh` which exploits symmetry/sparsity of SPD matrices and avoids the ``O(C^3)`` Schur bottleneck of :func:`scipy.linalg.logm` (ACL 2026, App. B: “sparse spectral domain”).

Eigenvalues are clamped to ``eps`` before ``log`` for numerical stability on rank-deficient inputs (also covers explicit ``eps*I`` regularisation).

Args: spd: SPD matrix ``(C, C)`` or stack ``(..., C, C)`` (e.g., ``(num_windows, C, C)`` or ``(B, T, C, C)``). eps: Minimum eigenvalue before log (default ``1e-6``).

Returns: Matrix logarithm(s) with same shape as *spd*.

References: Gowda & Miller, ACL 2026 Findings, Eq. 4; J Neural Eng 2024, Eq. 5 (affine-invariant log map at identity).

spd_flatten_upper

def spd_flatten_upper(spd: np.ndarray) -> np.ndarray

Flatten symmetric matrix to upper-triangular vector.

For ``(C, C)`` returns ``(K,)`` where ``K = C*(C+1)//2``; for ``(..., C, C)`` returns ``(..., K)``. Order is row-major ``np.triu_indices(C)`` (ACL 2026, Sec. 4.2: “upper-tri vectorisation”).

Off-diagonal terms are **not** scaled by ``sqrt(2)``; the plain upper-tri is used so that :func:`spd_logm` outputs can be consumed directly by a linear projection. If the norm-preserving variant is needed, multiply off-diagonal entries by ``sqrt(2)`` after calling.

Args: spd: Symmetric matrix ``(C, C)`` or stack ``(..., C, C)``. Input is typically a tangent-space matrix from :func:`spd_logm`.

Returns: Vectorised array.

References: Gowda & Miller, ACL 2026 Findings, Sec. 4.2.

spd_riemannian_features

def spd_riemannian_features(spd: np.ndarray, eps: float = 1e-06) -> np.ndarray

Tangent-space Riemannian features: ``flatten_upper(logm(spd))``.

Convenience wrapper combining :func:`spd_logm` and :func:`spd_flatten_upper`. For a stack ``(..., C, C)`` returns ``(..., K)`` with ``K = C*(C+1)//2`` (e.g., ``C=4 → K=10``).

Args: spd: SPD matrix or stack ``(..., C, C)``. eps: Eigenvalue clamp for the log map.

Returns: Riemannian feature vector(s) ``(..., K)`` in the tangent space at identity.

References: Gowda & Miller, ACL 2026 Findings, Sec. 4.2, Fig. 2; J Neural Eng 2024, Sec. 3.4.

get_riemannian_features

def get_riemannian_features(segment: np.ndarray, eps: float = 1e-06) -> np.ndarray

Riemannian features directly from a raw segment.

Computes ``spd_riemannian_features(compute_spd_matrix(segment))``.

Args: segment: Array ``(T, C)``. eps: Regulariser for SPD and log map.

Returns: Vector ``(K,)`` with ``K = C*(C+1)//2``.

References: ACL 2026 Findings, Sec. 4.2; J Neural Eng 2024, Sec. 3.4.

extract_spd_features

def extract_spd_features(segment: np.ndarray, eps: float = 1e-06) -> np.ndarray

Alias for :func:`get_riemannian_features`.

get_riemannian_features_timevarying

def get_riemannian_features_timevarying(segment: np.ndarray, fs: float = 250, window_ms: int = 50, step_ms: int = 20, eps: float = 1e-06) -> np.ndarray

Time-varying Riemannian features ``(num_windows, K)``.

Computes sliding SPD stack via :func:`compute_spd_timevarying` then maps each SPD to tangent space.

Args: segment: Array ``(T, C)``. fs: Sampling rate. window_ms: Window length in ms. step_ms: Hop in ms. eps: SPD regulariser / eigenvalue clamp.

Returns: Array ``(num_windows, K)``.

References: Gowda & Miller, ACL 2026 Findings, Sec. 4.1–4.2.

class EMGTokenizer

Neuromuscular Contraction Tokenizer (NCT) + VQ.

Discretizes EMG ``(T, C)`` into contraction tokens via sliding-window segmentation into physiologically grounded primitives (~30-60 ms) and vector quantization against a learned codebook.

Args: codebook_size: Vocabulary size K (default 512). token_dim: Codebook vector dimension D (default 64). window_size: Primitive window in samples (default 32; ~32 ms @1000 Hz or ~128 ms @250 Hz, within 30-100 ms contraction range). stride: Hop between windows (default 16, 50% overlap for overlap-add reconstruction). seed: RNG seed for codebook init (default 0).

def __init__(self, codebook_size: int = 512, token_dim: int = 64, window_size: int = 32, stride: int = 16, seed: int = 0) -> None

def get_codebook(self) -> np.ndarray

def get_vocab(self) -> dict[int, np.ndarray]

def encode(self, signal: np.ndarray) -> np.ndarray

def decode(self, tokens: np.ndarray) -> np.ndarray

def decode_with_channels(self, tokens: np.ndarray, num_channels: int) -> np.ndarray

def update_codebook(self, new_codebook: np.ndarray) -> None

def vocabulary_size(self) -> int

class AEMGFramework

AEMG self-supervised framework: masked modeling over contraction tokens.

Wraps an :class:`EMGTokenizer` and performs masked language modeling over token sentences (``collective tokens``). Randomly masks a fraction of tokens and predicts the masked vocabulary ids, learning a universal EMG vocabulary shared across subjects.

Args: tokenizer: EMGTokenizer instance. If None, a default is created. mask_ratio: Fraction of tokens to mask (default 0.15, BERT-style; spec says mask tokens and predict collective tokens). mask_token_id: Token id used for masked positions. Defaults to ``codebook_size`` (extra id) or 0 if codebook_size is max. vocab_size: Explicit vocab size; defaults to tokenizer.codebook_size. seed: RNG seed.

Torch is optional: if available, a small Transformer encoder is instantiated for masked prediction; otherwise pretrain_step returns a dummy heuristic loss.

def __init__(self, tokenizer: EMGTokenizer | None = None, mask_ratio: float = 0.15, mask_token_id: int | None = None, vocab_size: int | None = None, seed: int = 0) -> None

def get_vocab(self) -> dict[int, np.ndarray]

def get_codebook(self) -> np.ndarray

def tokenize(self, signal: np.ndarray) -> np.ndarray

def pretrain_step(self, tokens: np.ndarray, optimizer: Any | None = None) -> dict[str, Any]

def encode_and_mask(self, signal: np.ndarray) -> dict[str, Any]

class CyRoPE(nn.Module)

Cylindrical Rotary Position Embedding.

Factorizes temporal (linear) and spatial (annular for 8×16 forearm grid) positional encodings via rotary embeddings. ``dim`` is split into temporal and spatial halves.

Args: dim: Embedding dimension ``D`` (split into temporal vs spatial). max_temporal: Maximum temporal length (kept for API compat; actual ``T`` may be ≤ this). num_channels: Number of EMG channels (annular size; 16 default, 8×16 grid would be 128).

Forward: ``(B, T, C, D) -> (B, T, C, D)`` with rotary applied per position.

def __init__(self, dim: int, max_temporal: int = 1000, num_channels: int = 16) -> None

def forward(self, x: Any) -> Any

class CyRoPE

Stub — raises MissingDependencyError when torch is absent.

def __init__(self, dim: int, max_temporal: int = 1000, num_channels: int = 16) -> None

def forward(self, x: Any) -> Any

class SPECTREEncoder(nn.Module)

SPECTRE encoder: depthwise CNN + Transformer with CyRoPE + masked head.

Args: num_channels: Input EMG channels. embed_dim: Embedding width ``D``. depth: Number of Transformer layers. n_clusters: Output pseudo-label vocabulary size (head ``D -> n_clusters``). nhead: Attention heads. dropout: Dropout prob. max_temporal: Max temporal length for RoPE.

def __init__(self, num_channels: int = 16, embed_dim: int = 128, depth: int = 4, n_clusters: int = 64, nhead: int = 4, dropout: float = 0.1, max_temporal: int = 1000) -> None

def forward(self, x: Any, mask: Any | None = None) -> Any

def ssl_loss(self, x_or_logits: Any, labels: Any | None = None, mask: Any | None = None, **kwargs: Any) -> Any

def compute_ssl_loss(self, *args: Any, **kwargs: Any) -> Any

def masked_loss(self, *args: Any, **kwargs: Any) -> Any

class SPECTREEncoder

Stub — raises MissingDependencyError when torch is absent.

def __init__(self, num_channels: int = 16, embed_dim: int = 128, depth: int = 4, n_clusters: int = 64, nhead: int = 4, dropout: float = 0.1, max_temporal: int = 1000) -> None

def forward(self, x: Any, mask: Any | None = None) -> Any

def ssl_loss(self, *args: Any, **kwargs: Any) -> Any

stft_kmeans_pseudolabels

def stft_kmeans_pseudolabels(signal: np.ndarray, fs: float = 1000, n_fft: int = 64, hop: int = 16, n_clusters: int = 64) -> np.ndarray

Compute STFT per channel and cluster magnitudes to pseudo-labels.

Per-channel STFT (``scipy.signal.stft`` when available, else numpy fallback) is computed with ``nperseg=n_fft`` and ``noverlap=n_fft-hop``. Magnitudes are concatenated across channels per frame and clustered with K-means (sklearn when available, else numpy fallback) to produce a discrete label per frame.

Args: signal: Array ``(T, C)`` — ``T`` samples, ``C`` channels. fs: Sampling rate in Hz. n_fft: FFT size / window length. hop: Hop / stride in samples. n_clusters: Number of K-means clusters.

Returns: labels: ``(num_frames,)`` int array with values in ``[0, n_clusters)``.

Raises: ValueError: if signal is not 2-D or params invalid. ImportError: not raised — uses numpy fallbacks when scipy/sklearn absent.

class _TransformerBlock(nn.Module)

Bidirectional Transformer block with pre-norm and RoPE.

def __init__(self, embed_dim: int = 128, num_heads: int = 4, mlp_ratio: float = 12.0, dropout: float = 0.1) -> None

def forward(self, x: Any) -> Any

class TinyMyoEncoder(nn.Module)

TinyMyo encoder — channel-independent patching + masked reconstruction.

Args: num_channels: EMG channels (default 4). patch_size: Temporal patch length (default 10). embed_dim: Embedding dimension (default 128). depth: Number of Transformer blocks (default 8). num_heads: Attention heads (default 4). mask_ratio: Fraction of patches masked during training (default 0.5, SimMIM).

def __init__(self, num_channels: int = 4, patch_size: int = 10, embed_dim: int = 128, depth: int = 8, num_heads: int = 4, mask_ratio: float = 0.5) -> None

def forward(self, x: Any) -> Any

def reconstruct(self, encoded: Any) -> Any

def count_parameters(self) -> int

def estimate_flops(self, seq_len: int = 150) -> int

def estimate_FLOPs(self, seq_len: int = 150) -> int

class TinyMyoFoundation(nn.Module)

Foundation wrapper: encoder + task heads (classification, regression, speech).

Args: num_channels: Passed to encoder. patch_size: Passed to encoder. embed_dim: Passed to encoder. depth: Passed to encoder. num_heads: Passed to encoder. mask_ratio: Passed to encoder. num_classes: Classes for classification head (default 8). regression_dim: Output dim for regression head (default 1). vocab_size: Vocab size for speech/CTC head (default 40). encoder: Optional pre-built TinyMyoEncoder instance.

def __init__(self, num_channels: int = 4, patch_size: int = 10, embed_dim: int = 128, depth: int = 8, num_heads: int = 4, mask_ratio: float = 0.5, num_classes: int = 8, regression_dim: int = 1, vocab_size: int = 40, encoder: Any | None = None) -> None

def forward(self, x: Any, task: str = 'classification') -> Any

def pretrain_step(self, x: Any, optimizer: Any | None = None) -> dict[str, Any]

def finetune_step(self, x: Any, y: Any, task: str = 'classification', optimizer: Any | None = None, criterion: Any | None = None) -> dict[str, Any]

def count_parameters(self) -> int

def count_trainable_parameters(self) -> int

def estimate_flops(self, seq_len: int = 150) -> int

def estimate_FLOPs(self, seq_len: int = 150) -> int

class TinyMyoEncoder

Stub — raises MissingDependencyError when torch is absent.

def __init__(self, num_channels: int = 4, patch_size: int = 10, embed_dim: int = 128, depth: int = 8, num_heads: int = 4, mask_ratio: float = 0.5) -> None

def forward(self, x: Any) -> Any

def reconstruct(self, encoded: Any) -> Any

def count_parameters(self) -> int

def estimate_flops(self, seq_len: int = 150) -> int

def estimate_FLOPs(self, seq_len: int = 150) -> int

class TinyMyoFoundation

Stub — raises MissingDependencyError when torch is absent.

def __init__(self, num_channels: int = 4, patch_size: int = 10, embed_dim: int = 128, depth: int = 8, num_heads: int = 4, mask_ratio: float = 0.5, num_classes: int = 8, regression_dim: int = 1, vocab_size: int = 40, encoder: Any | None = None) -> None

def forward(self, x: Any, task: str = 'classification') -> Any

def pretrain_step(self, x: Any, optimizer: Any | None = None) -> dict[str, Any]

def finetune_step(self, x: Any, y: Any, task: str = 'classification', optimizer: Any | None = None, criterion: Any | None = None) -> dict[str, Any]

def count_parameters(self) -> int

def estimate_flops(self, seq_len: int = 150) -> int

def estimate_FLOPs(self, seq_len: int = 150) -> int

pretrain_step

def pretrain_step(model: Any, x: Any, optimizer: Any | None = None) -> dict[str, Any]

Functional pretrain step for any TinyMyoFoundation/TinyMyoEncoder.

If model is TinyMyoEncoder, computes masked MSE directly.

finetune_step

def finetune_step(model: Any, x: Any, y: Any, task: str = 'classification', optimizer: Any | None = None, criterion: Any | None = None) -> dict[str, Any]

Functional fine-tune step.

pretrain_step

def pretrain_step(model: Any, x: Any, optimizer: Any | None = None) -> dict[str, Any]

No description.

finetune_step

def finetune_step(model: Any, x: Any, y: Any, task: str = 'classification', optimizer: Any | None = None, criterion: Any | None = None) -> dict[str, Any]

No description.

class EMGAdaptor(nn.Module)

Two-layer MLP adaptor: EMG features → LLM embedding.

Architecture (ACL 2025): ``Linear(input_dim → hidden_dim) → ReLU → Dropout → Linear(hidden_dim → output_dim)``.

Handles both raw sEMG segments ``(T, 4)`` / ``(B, T, 4)`` (converted via handcrafted extraction) and pre-computed feature vectors ``(112,)`` / ``(B, 112)`` / ``(768,)``.

Args: input_dim: Feature dimension (112 handcrafted, 768 speech-encoder). hidden_dim: Hidden layer width (default 768). output_dim: LLM embedding dim (3072 for Llama-3.2-3B). dropout: Dropout probability between layers.

def __init__(self, input_dim: int = 112, hidden_dim: int = 768, output_dim: int = 3072, dropout: float = 0.1) -> None

def forward(self, x: Any) -> Any

class EMGAdaptor

Stub – raises MissingDependencyError when torch is absent.

def __init__(self, input_dim: int = 112, hidden_dim: int = 768, output_dim: int = 3072, dropout: float = 0.1) -> None

def forward(self, x: Any) -> Any

class EMGAdaptorProvider(LLMProvider)

Wrapper coupling an :class:`EMGAdaptor` with a frozen LLM.

Compatible with :mod:`subvocal.core.llm_providers` ``LLMProvider`` interface (``get_provider_name`` / ``reconstruct_intent``). Also exposes :meth:`encode_emg_features` (plus aliases ``encode`` / ``adapt``) for direct EMG→LLM embedding.

Args: adaptor: Pre-initialised adaptor. If ``None``, a default ``EMGAdaptor(input_dim=112, hidden_dim=768, output_dim=3072)`` is created (requires ``torch``). llm_name: Identifier of the frozen LLM (e.g. ``"meta-llama/Llama-3.2-3B"``). llm_provider: Optional underlying :class:`subvocal.core.interfaces.LLMProvider` to delegate ``reconstruct_intent`` to. If ``None``, a heuristic fallback is used. device: Torch device for the adaptor.

def __init__(self, adaptor: Any | None = None, llm_name: str = 'meta-llama/Llama-3.2-3B', llm_provider: Any | None = None, device: str = 'cpu') -> None

def encode_emg_features(self, emg_input: np.ndarray | Any, fs: float = 250.0) -> Any

def encode(self, emg_input: np.ndarray | Any, fs: float = 250.0) -> Any

def adapt(self, emg_input: np.ndarray | Any, fs: float = 250.0) -> Any

def encode_emg(self, emg_input: np.ndarray | Any, fs: float = 250.0) -> Any

def get_provider_name(self) -> str

def reconstruct_intent(self, tokens: Any, context: Any) -> Any

estimate_flops

def estimate_flops(model_type: str, num_channels: int = 4, segment_length: int = 150, num_classes: int = 4, hidden_size: int = 64, num_layers: int = 2) -> int

Estimate floating point operations (FLOPs) per forward inference pass.

Formulas derived from structural layers (convolutions, linear projections, attention, and recurrences).

run_benchmark

def run_benchmark(user_id: str, model_type: str, num_runs: int = 200) -> dict[str, Any]

Benchmark model inference latency, disk footprint, and parameter size.

Estimates computational FLOPs and electrical energy footprint.

class TrainingConfig(BaseModel)

Configuration class for reproducible training runs.

export_to_onnx

def export_to_onnx(user_id: str, model_type: str, export_path: str) -> str

Export a trained PyTorch model to ONNX format.

Args: user_id: The ID of the user. model_type: "cnn", "gru", or "transformer". export_path: Destination path for the exported ONNX model.

export_to_coreml

def export_to_coreml(user_id: str, model_type: str, export_path: str) -> bool

Compile PyTorch/ONNX model to Apple Core ML format.

Note: requires coremltools. If missing, handles gracefully.

export_to_tflite

def export_to_tflite(user_id: str, model_type: str, export_path: str) -> bool

Compile PyTorch model to TFLite format via ONNX → tf2onnx → TFLite.

Requires: tensorflow, tf2onnx. If missing, raises NotImplementedError. Returns True on success, raises on failure.

quantize_model_int8

def quantize_model_int8(user_id: str, model_type: str, threshold: float = 0.05) -> dict[str, Any]

Perform dynamic int8 quantization on a PyTorch model and run accuracy regression checks.

Args: user_id: User calibration dataset ID to evaluate against. model_type: "cnn", "gru", or "transformer". threshold: Maximum allowed accuracy drop (e.g. 0.05 = 5%).

Returns: Dict detailing quantization metrics.

class InferenceEngine(Classifier)

Inference engine with debounce and confidence thresholding.

Implements the core.interfaces.Classifier interface.

def __init__(self, user_id: str, model_type: str = 'rf', confidence_threshold: float = config.CONFIDENCE_THRESHOLD, cooldown_ms: int = config.COOLDOWN_MS)

def labels(self) -> list[str]

def predict(self, frame: Frame | np.ndarray) -> CommandToken | None

def predict_raw(self, frame: Frame | np.ndarray) -> tuple[str, float, list[float]]

class LISA(nn.Module)

LLM Integrated Scoring Adjustment (LISA).

Rescores acoustic beam-search hypotheses with a frozen LLM prior.

Args: llm_provider: LLM provider for scoring. May be any :class:`subvocal.core.interfaces.LLMProvider` or object with ``score(text)``. If ``None``, a deterministic heuristic is used (still satisfies the API). alpha: Interpolation weight for LLM score (0..1). ``0`` = acoustic only, ``1`` = LLM only. Default 0.5 per MONA. beta: Length-bonus weight (default 0.3). Added as ``beta * (len(text)/100)`` or word-count normalized; set 0 to disable.

def __init__(self, llm_provider: Any | None = None, alpha: float = 0.5, beta: float = 0.3) -> None

def rescore(self, hypotheses: list[str], beam_scores: list[float]) -> list[tuple[str, float]]

def forward(self, hypotheses: list[str], beam_scores: list[float]) -> list[tuple[str, float]]

def beam_search_rescore(self, logits: Any, vocab: list[str], beam_width: int = 10, lm_weight: float = 0.5) -> str

class LISA

Stub — LISA scorer without torch (rescore works, beam_search requires torch).

def __init__(self, llm_provider: Any | None = None, alpha: float = 0.5, beta: float = 0.3) -> None

def rescore(self, hypotheses: list[str], beam_scores: list[float]) -> list[tuple[str, float]]

def forward(self, hypotheses: list[str], beam_scores: list[float]) -> list[tuple[str, float]]

def beam_search_rescore(self, logits: Any, vocab: list[str], beam_width: int = 10, lm_weight: float = 0.5) -> str

lisa_rescore

def lisa_rescore(hypotheses: list[dict[str, Any]], llm_provider: Any | None = None, weight: float = 0.5) -> list[dict[str, Any]]

Re-rank hypotheses list with LISA LLM scoring (functional API).

Functional wrapper around :class:`LISA` for dict-based beam outputs (common in ASR/EMG decoders). Each hypothesis dict must contain a text field and a score field; flexible key aliases are accepted.

Args: hypotheses: List of dicts, each with text and acoustic score. Accepted text keys: ``text``, ``hypothesis``, ``transcript``, ``hyp``, ``str``. Accepted score keys: ``score``, ``acoustic_score``, ``beam_score``, ``acoustic``, ``log_prob``. Additional keys are preserved in output. llm_provider: LLM provider for scoring. May be an :class:`subvocal.core.interfaces.LLMProvider`, a provider name string (e.g., ``"heuristic"``, ``"openai"``), or ``None`` for heuristic fallback. weight: LLM interpolation weight (0..1). ``0`` = acoustic only, ``1`` = LLM only. Default 0.5.

Returns: Re-ranked list of dicts sorted descending by combined score. Each dict is shallow-copied and augmented with ``llm_score``, ``combined_score`` (or ``lisa_score`` alias).

Raises: ValueError: if weight not in [0,1] or hypotheses not a list. TypeError: if hypotheses contains non-dict items.

Example: >>> from subvocal.core.llm_providers import HeuristicProvider >>> hyps = [{"text": "hello world", "score": -0.2}, {"text": "hallo world", "score": -0.5}] >>> reranked = lisa_rescore(hyps, HeuristicProvider(), weight=0.5) >>> reranked[0]["text"] 'hello world'

get_model_path

def get_model_path(user_id: str, model_type: str = 'rf') -> str

Get the path to a user's model file.

Saves RF/SVM models as joblib, and CNN/GRU/Transformer models as .pth.

save_model

def save_model(model_data: dict[str, Any], user_id: str, model_type: str = 'rf') -> str

Save a model and its associated metadata to disk.

load_model

def load_model(user_id: str, model_type: str = 'rf') -> dict[str, Any]

Load a model and its metadata from disk.

model_exists

def model_exists(user_id: str, model_type: str = 'rf') -> bool

Check if a model exists for a user.

class BatchSampler

Undersample LibriSpeech to 50% per epoch (MONA Algorithm 1).

Each epoch, retains 100% of EMG indices (silent + vocal) and a random 50% subset of LibriSpeech indices (without replacement), shuffles the combined pool, and yields batches.

This keeps EMG and LibriSpeech balanced per batch despite LibriSpeech being much larger (~960 h vs ~10 h EMG), matching MONA's ``undersample LS to 50% per epoch`` description.

Args: emg_indices: List of EMG sample indices (e.g., 0..N_emg-1) or int count, or None (inferred from dataset sizes). librispeech_indices: List of LibriSpeech sample indices or int count. batch_size: Batch size (default 32). librispeech_ratio: Fraction of LibriSpeech to keep per epoch (default 0.5 per MONA). shuffle: Whether to shuffle the combined pool each epoch. seed: Base random seed; epoch is added for per-epoch determinism. drop_last: Whether to drop last incomplete batch. dataset_size: Total dataset size (alternative to explicit indices). librispeech_size: Number of LibriSpeech samples (used with dataset_size). dataset: Optional dataset object (unused but accepted for API compat; if it exposes ``emg_indices`` / ``librispeech_indices`` attributes, they are used).

Example: ``sampler = BatchSampler(emg_indices=list(range(100)), librispeech_indices=list(range(100,600)), batch_size=32)`` ``for batch in sampler: train_step(batch)`` ``sampler.set_epoch(epoch)`` # call each epoch for new subsample

Notes: Compatible with ``torch.utils.data.DataLoader`` via manual batching or by passing ``batch_sampler=sampler``. When used with PyTorch, each iteration yields a ``list[int]`` batch of indices.

def __init__(self, emg_indices: list[int] | int | None = None, librispeech_indices: list[int] | int | None = None, batch_size: int = 32, librispeech_ratio: float = 0.5, shuffle: bool = True, seed: int | None = None, drop_last: bool = False, dataset_size: int | None = None, librispeech_size: int | None = None, dataset: Any | None = None, **kwargs: Any) -> None

def set_epoch(self, epoch: int) -> None

def num_emg(self) -> int

def num_librispeech(self) -> int

def num_librispeech_sampled(self) -> int

def get_combined_indices(self) -> list[int]

cross_contrastive_loss

def cross_contrastive_loss(emg_latents: Any, audio_latents: Any, temperature: float = 0.1) -> Any

Cross-modal contrastive loss (crossCon) between EMG and audio latents.

Bidirectional InfoNCE: for batch size B, similarity matrix S[i,j] = cosine(emg_i, audio_j) / temperature. Positives are diagonal. Loss = (CE(S, labels) + CE(S^T, labels)) / 2.

Latents may be (B, D) or (B, T, D) (temporal mean-pooled).

Args: emg_latents: Tensor of shape (B, D) or (B, T, D). audio_latents: Tensor of shape (B, D) or (B, T, D). temperature: Softmax temperature, ``>0`` (default 0.1 per MONA).

Returns: Scalar Tensor loss (requires grad). 0 if batch size < 2 still differentiable.

Raises: MissingDependencyError: if torch not installed. ValueError: if shapes mismatched or temperature invalid.

supervised_temporal_contrastive_loss

def supervised_temporal_contrastive_loss(latents: Any, labels: Any, temperature: float = 0.1) -> Any

Supervised temporal contrastive loss (supTcon).

SupCon-style: positives share the same label, negatives differ. For each anchor i, loss pulls positives j (|labels_j==labels_i, j!=i) closer in latent space than negatives.

Temporal handling: if latents is (B, T, D), mean-pools over T. Labels may be (B,) or (B, 1).

Args: latents: Tensor (B, D) or (B, T, D). labels: Tensor (B,) long/int or array-like. temperature: Softmax temperature (>0, default 0.1).

Returns: Scalar Tensor loss. Returns 0 if no anchor has positives.

Raises: MissingDependencyError: if torch not installed.

dtw_align

def dtw_align(latents_a: Any, latents_b: Any) -> Any

Align ``latents_a`` to ``latents_b`` via Dynamic Time Warping in latent space.

Latent DTW per MONA: warps silent EMG (or vocal EMG) to audio temporal structure in the shared encoder latent space. Uses squared Euclidean distance in latent space; warps ``a`` to ``b``'s time axis by averaging frames of ``a`` that map to each frame of ``b`` along the optimal path.

Supports ``torch.Tensor`` (GPU/CPU) or ``np.ndarray``. If torch is not available, numpy fallback is used transparently. Batched inputs ``(B, T, D)`` are handled per-batch element.

Args: latents_a: Tensor/array of shape ``(T_a, D)`` or ``(B, T_a, D)`` — source to warp (e.g., silent EMG latents). latents_b: Tensor/array of shape ``(T_b, D)`` or ``(B, T_b, D)`` — target time axis (e.g., audio latents).

Returns: Warped version of ``latents_a`` aligned to ``latents_b`` time axis: * 2-D inputs ``(T_a,D)`` / ``(T_b,D)`` → ``(T_b, D)`` * 3-D batched ``(B, T_a,D)`` / ``(B, T_b,D)`` → ``(B, T_b, D)`` Returned type matches input type (Tensor if inputs were Tensor, otherwise ndarray). Device/dtype preserved for Tensor outputs.

Raises: ValueError: if feature dims mismatch or batch dims mismatch.

mona_loss

def mona_loss(emg_silent: Any, emg_vocal: Any, audio: Any, labels: Any, weights: dict[str, float] | None = None) -> dict[str, Any]

Combined MONA objective with crossCon, supTcon and CTC components.

Computes::

total = w_cross * crossCon + w_sup * supTcon + w_ctc * ctc

where crossCon is averaged over (silent↔audio) and (vocal↔audio) when both EMG modalities are present, supTcon is the supervised temporal contrastive loss on EMG latents (mean-pooled if temporal), and ctc is an optional CTC loss if logits are provided.

Args: emg_silent: Silent EMG latents Tensor (B, D) or (B, T, D) or None. emg_vocal: Vocal EMG latents Tensor (B, D) or (B, T, D) or None. audio: Audio (LibriSpeech) latents Tensor (B, D) or (B, T, D) or None. labels: Ground-truth labels Tensor (B,) or (B, L) for CTC; may be list/ndarray; None => supTcon/ctc skipped. weights: Dict with keys ``crossCon``/``cross``, ``supTcon``/``sup``, ``ctc``. Missing keys default to 1.0. Also accepts ``total`` not needed.

Returns: Dict with keys ``crossCon``, ``supTcon``, ``ctc``, ``total``. Values are scalar Tensors (if torch available) or floats (fallback). Always contains ``total``.

Raises: MissingDependencyError: if torch not installed and tensor inputs given (fallback returns float dict without raising, to keep training loop alive in non-torch tests).

generate_synthetic_dataset

def generate_synthetic_dataset(user_id: str = 'pretrained', num_classes: int = 5, segments_per_class: int = 40)

Generate a reproducible multi-channel sEMG dataset to simulate a public dataset.

Each class is synthesized with unique harmonic and frequency patterns.

run_pretraining

def run_pretraining()

Run pre-training pipeline for all deep architectures.

class SPDGRU(nn.Module)

SPD-GRU CTC decoder.

Maps an SPD sequence to per-frame phoneme logits suitable for :class:`torch.nn.CTCLoss`.

Args: num_channels: Number of sEMG channels ``C`` (default 4). The Riemannian input dimension is ``K = C*(C+1)//2`` (e.g., 10). hidden_size: GRU hidden width (default 64, ACL 2026 Sec. 5). num_layers: Number of stacked GRU layers (default 3 per Gowda 2025/2026; dropout applied between layers). num_phonemes: Phoneme vocabulary size including CTC blank (default 40; blank is index 0, cf. Gowda 2025 Table 1). dropout: Dropout probability between GRU layers and before the final projection (default 0.2).

Input: ``spd_seq`` — Tensor ``(B,T,C,C)`` or ``(T,C,C)`` (single sequence). ``numpy`` arrays are auto-converted (detached CPU).

Output: Logits ``(B,T,num_phonemes)`` — raw, **not** log-softmaxed. Apply ``log_softmax(-1)`` and permute to ``(T,B,V)`` for :class:`torch.nn.CTCLoss`.

References: Gowda & Miller, ACL 2025 & Findings of ACL 2026; J Neural Eng 2024.

def __init__(self, num_channels: int = 4, hidden_size: int = 64, num_layers: int = 3, num_phonemes: int = 40, dropout: float = 0.2) -> None

def forward(self, spd_seq: Any) -> Any

class SPDGRU

Stub – raises MissingDependencyError when torch is absent.

def __init__(self, num_channels: int = 4, hidden_size: int = 64, num_layers: int = 3, num_phonemes: int = 40, dropout: float = 0.2) -> None

def forward(self, spd_seq: Any) -> Any

ctc_loss

def ctc_loss(log_probs: Any, targets: Any, input_lengths: Any, target_lengths: Any, blank: int = 0, zero_infinity: bool = True) -> Any

CTC loss wrapper (log_probs ``(T,N,V)`` or logits ``(B,T,V)``).

Accepts both ``(T,N,V)`` log-probs (standard) and ``(B,T,V)`` logits (SPDGRU output) for convenience; orientation and logit vs log-prob are auto-detected.

Args: log_probs: Log-probabilities ``(T, N, V)`` or logits ``(B,T,V)``. If ``(B,T,V)`` is detected (via ``input_lengths``), it is converted via ``log_softmax`` and permuted to ``(T,B,V)``. targets: Concatenated target indices ``(sum(target_lengths),)``. input_lengths: Lengths ``(N,)`` of log_probs per batch element. target_lengths: Lengths ``(N,)`` of targets per batch element. blank: Blank label index (default 0 per Gowda 2025). zero_infinity: Whether to zero infinite losses (default True).

Returns: Scalar loss tensor.

References: Graves et al., 2006; Gowda 2025, Sec. 5.3 (CTC for phonemes).

ctc_loss_from_logits

def ctc_loss_from_logits(logits: Any, targets: Any, input_lengths: Any, target_lengths: Any, blank: int = 0, zero_infinity: bool = True) -> Any

Convenience: compute CTC loss directly from logits ``(B,T,V)`` or ``(T,C,C)`` style.

Args: logits: Raw logits ``(B,T,V)`` or ``(T,V)``. If ``(T,V)`` batch is 1. Also accepts SPDGRU logits ``(B,T,num_phonemes)``. targets: Concatenated targets ``(sum(target_lengths),)``. input_lengths: ``(N,)`` or scalar. target_lengths: ``(N,)`` or scalar. blank: Blank index. zero_infinity: Zero infinities.

Returns: Scalar loss.

greedy_decode

def greedy_decode(logits: Any, blank_id: int = 0, collapse_repeated: bool = True) -> list[list[int]]

Greedy CTC decoding (argmax + collapse).

Args: logits: Tensor ``(B,T,V)`` or ``(T,V)`` or ``(T,C,C)``-style? Actually expects phoneme logits. Can also be ``(B,T,V)`` numpy. blank_id: CTC blank index (default 0). collapse_repeated: Whether to collapse repeated non-blank tokens.

Returns: List of length ``B`` (or 1) with decoded index sequences (blanks removed).

References: Graves et al., CTC greedy search; Gowda 2025, Sec. 5.4.

train_step

def train_step(model: Any, optimizer: Any, logits: Any, targets: Any, input_lengths: Any, target_lengths: Any, blank: int = 0) -> Any

Single CTC training step (forward → loss → backward → step).

Args: model: :class:`SPDGRU` instance (unused except for zero_grad validation; logits may be pre-computed). optimizer: Torch optimizer (e.g., ``AdamW``). logits: Raw logits ``(B,T,V)`` from ``model(spd_seq)``. targets: Concatenated target ``(sum(target_lengths),)``. input_lengths: ``(N,)`` input lengths. target_lengths: ``(N,)`` target lengths. blank: CTC blank index.

Returns: Scalar loss tensor (detached for logging).

References: Gowda 2025, Sec. 6 (AdamW, CTC training).

train_spd_gru

def train_spd_gru(model: Any, optimizer: Any, spd_seq: Any, targets: Any, input_lengths: Any, target_lengths: Any, blank: int = 0) -> Any

End-to-end training helper: ``spd_seq → logits → CTC loss → step``.

Args: model: :class:`SPDGRU`. optimizer: Torch optimizer. spd_seq: SPD sequence ``(B,T,C,C)`` or ``(T,C,C)``, Tensor or ndarray. targets: Concatenated phoneme targets. input_lengths: Input lengths ``(N,)``. target_lengths: Target lengths ``(N,)``. blank: Blank index.

Returns: Tuple ``(loss, logits)`` where ``loss`` is detached scalar and ``logits`` are ``(B,T,V)``.

References: Gowda 2025, Sec. 6.

class SpeechNet(nn.Module)

Tiny depthwise-separable CNN (~15k params) for sEMG commands.

Args: num_channels: EMG channels (14 for SilentWear, 4-8 generic). num_classes: Output classes (8 commands + rest if needed). segment_length: Temporal length T (default 150). dropout: Dropout prob between blocks.

def __init__(self, num_channels: int = 4, num_classes: int = 8, segment_length: int = 150, dropout: float = 0.2) -> None

def forward(self, x: Any) -> Any

def count_parameters(self) -> int

def count_trainable_parameters(self) -> int

def count_parameters_total(self) -> int

def estimate_energy(self) -> float

def estimate_energy_uj(self) -> float

def estimate_latency(self) -> float

def estimate_latency_ms(self) -> float

def estimate_flops(self) -> int

class SpeechNet

Stub — raises MissingDependencyError when torch is absent.

def __init__(self, num_channels: int = 4, num_classes: int = 8, segment_length: int = 150, dropout: float = 0.2) -> None

def forward(self, x: Any) -> Any

def count_parameters(self) -> int

def estimate_energy(self) -> float

def estimate_latency(self) -> float

finetune_inter_session

def finetune_inter_session(model: Any, train_loader: Any, val_loader: Any | None = None, epochs: int = 5, lr: float = 0.0005, weight_decay: float = 0.0001, freeze_backbone: bool = True, device: str = 'cpu') -> dict[str, Any]

Incremental fine-tuning for a new session (<10 min data).

SilentWear protocol: freeze depthwise-separable backbone, fine-tune only the linear head (or last block + head) with low LR for few epochs, using calibration data collected in <10 min (~50-150 utterances).

Args: model: SpeechNet instance (or any nn.Module with .classifier). train_loader: DataLoader yielding (B, C, T), labels. val_loader: Optional validation loader. epochs: Few epochs (3-10, default 5) to avoid overfit on tiny set. lr: Low learning rate (1e-4 - 1e-3). weight_decay: Weight decay. freeze_backbone: If True, freeze blocks 1-3, train head only. device: Torch device.

Returns: Dict with train loss history and val accuracy if available.

train_speechnet

def train_speechnet(num_channels: int = 4, num_classes: int = 8, segment_length: int = 150, train_data: Any | None = None, train_labels: Any | None = None, epochs: int = 20, batch_size: int = 16, lr: float = 0.001, weight_decay: float = 0.0001, device: str = 'cpu', seed: int = 42) -> tuple[Any, dict[str, Any]]

Train SpeechNet from scratch (convenience wrapper).

If train_data/train_labels are None, returns untrained model. Otherwise builds a DataLoader and trains with AdamW + CrossEntropy.

Returns: (model, history_dict)

class EMG1DCNN(nn.Module)

1D CNN for sEMG segment classification.

Applies convolutions over the temporal dimension of raw multichannel sEMG.

def __init__(self, num_channels: int, num_classes: int, segment_length: int = 150)

def forward(self, x)

class EMGGRU(nn.Module)

Bidirectional GRU for temporal tracking of sEMG gestures.

def __init__(self, num_channels: int, num_classes: int, hidden_size: int = 64, num_layers: int = 2)

def forward(self, x)

class EMGTransformer(nn.Module)

Small Transformer encoder for raw multichannel sEMG sequence classification.

def __init__(self, num_channels: int, num_classes: int, segment_length: int = 150, d_model: int = 64, nhead: int = 4, num_layers: int = 2, dim_feedforward: int = 128)

def forward(self, x)

load_dataset

def load_dataset(user_id: str) -> tuple[list[np.ndarray], list[str]]

Load the calibration dataset for a user.

Returns: segments: list of raw 2D arrays, each (segment_length, num_channels) labels: list of label strings corresponding to each segment

preprocess_segments

def preprocess_segments(segments: list[np.ndarray], fs: float, low: float, high: float, notch_freq: float = config.NOTCH_FREQ, notch_q: float = config.NOTCH_Q) -> np.ndarray

Preprocess raw segments and stack into a 3D numpy array of shape (N, C, T).

train_model

def train_model(user_id: str, model_type: str = 'rf', test_size: float = 0.2, config_obj: TrainingConfig | None = None) -> dict[str, Any]

Train a sEMG model for a user and save it to disk.

Args: user_id: The ID of the user. model_type: "rf", "svm", "cnn", "gru", or "transformer". test_size: Split ratio for testing/validation. config_obj: Optional TrainingConfig parameters object.

Returns: A dictionary containing training metrics and results.

calibrate_model

def calibrate_model(user_id: str, pretrained_model_type: str, calibration_config: TrainingConfig | None = None) -> dict[str, Any]

Calibrate / fine-tune a pre-trained model for a new user.

For PyTorch models (CNN, GRU, Transformer), it loads pre-trained weights, replaces/adjusts the classification head to match the user's classes, freezes base layers (optionally), and fine-tunes on user calibration data. For classical models (RF, SVM), it fits a new pipeline on user data.

class MockContextProvider(ContextProvider)

Exposes mock system context state.

def get_context(self) -> UserContext

def get_provider_name(self) -> str

class MockActionExecutor(ActionExecutor)

Executes actions and logs execution history.

def __init__(self)

def execute(self, action: Action) -> Any

def can_execute(self, action: Action) -> bool

class SubvocalMCPServer

Zero-dependency Model Context Protocol server communicating via stdio.

def __init__(self)

def handle_request(self, req: dict[str, Any]) -> dict[str, Any] | None

def run(self)

main

def main() -> None

Console entry point for the ``subvocal-mcp`` stdio server.

articulatory_distance

def articulatory_distance(word1: str, word2: str) -> float

Calculates Levenshtein distance customized for sEMG articulatory phonetic groups.

Substitutions between letters in the same articulatory group cost 0.25 (highly likely sEMG error). Substitutions across different articulatory groups cost 1.0. Deletion in candidate (extra character in query) costs 1.0. Insertion in candidate (skipped character in query shorthand) costs 0.4.

find_best_shorthand_match

def find_best_shorthand_match(noisy_token: str, candidate_words: list[str]) -> list[tuple[str, float]]

Scores a noisy shorthand token against a list of candidate words.

Compresses candidates to shorthand and computes articulatory distance. Returns a list of tuples (candidate, distance) sorted by distance.

heuristic_decode_phrase

def heuristic_decode_phrase(noisy_phrase: str, web_context_words: list[str] | None = None, calendar_words: list[str] | None = None, contacts_words: list[str] | None = None, ui_elements: list[str] | None = None, contacts: list[str] | None = None, calendar_events: list[str] | None = None) -> tuple[str, float]

Decodes a full noisy shorthand phrase using articulatory heuristics.

Uses Command-Aware Context Prioritization and Phrase-Level Matching. If structured contexts are provided, they are prioritized based on the command. Otherwise, falls back to flat context words.

call_llm_api

def call_llm_api(provider: str, api_key: str, model: str | None, prompt: str) -> str | None

Helper to perform model-agnostic LLM calls via urllib without external SDK dependencies.

reconstruct_intent_llm

def reconstruct_intent_llm(noisy_input: str, heuristic_candidate: str, web_context: str | None = None, calendar: str | None = None, contacts: str | None = None, history: str | None = None) -> str | None

Reconstructs the target command phrase from noisy shorthand using a model-agnostic LLM.

Discovers API keys dynamically from the environment.

hybrid_decode

def hybrid_decode(noisy_phrase: str, web_context_words: list[str] | None = None, calendar_words: list[str] | None = None, contacts_words: list[str] | None = None, history: str | None = None, ui_elements: list[str] | None = None, contacts: list[str] | None = None, calendar_events: list[str] | None = None) -> tuple[str, float, str]

Combines heuristic alignment and LLM disambiguation for premium accuracy.

Returns: A tuple of (decoded_phrase, confidence, method_used) Method used is either "heuristic" or "llm".

get_articulatory_substitution

def get_articulatory_substitution(char: str) -> str

Get a random alternative character from the same articulatory group to simulate sEMG confusion.

simulate_emg_noise

def simulate_emg_noise(shorthand_word: str, sub_rate: float = 0.1, del_rate: float = 0.05, ins_rate: float = 0.05) -> str

Applies deletion, insertion, and substitution noise to a shorthand word.

- Substitution: Replaces char with another from the same articulatory group (sEMG confusion). - Deletion: Drops the character (packet drop / weak signal). - Insertion: Inserts a random character (muscle twitch / swallow artifact).

phrase_to_noisy_shorthand

def phrase_to_noisy_shorthand(phrase: str, sub_rate: float = 0.1, del_rate: float = 0.05, ins_rate: float = 0.05) -> tuple[str, str]

Convert a full phrase to clean shorthand and simulated noisy shorthand.

Returns: A tuple of (clean_shorthand_phrase, noisy_shorthand_phrase)

compress_word

def compress_word(word: str) -> str

Compress a single English word into its skeletal shorthand format.

Rules: 1. Lowercase the word. 2. Check if the word has a common abbreviation. 3. Otherwise: a. Keep the first letter. b. Remove subsequent vowels (a, e, i, o, u, y) and 'h'. c. Deduplicate consecutive identical consonants (e.g. 'gg' -> 'g').

word_to_articulatory_sequence

def word_to_articulatory_sequence(shorthand_word: str) -> list[str]

Convert a shorthand word to its articulatory group sequence.

Non-alphabet characters (like numbers and punctuation) are preserved as-is.

get_command_list

def get_command_list() -> list[str]

Return a list of all command names in the vocabulary.

get_command_details

def get_command_details(name: str) -> dict[str, Any]

Retrieve details for a specific command.

class TTSEngine

Multi-backend Text-to-Speech generator and player.

def __init__(self, config: TTSConfig | None = None)

def speak(self, text: str, filename: str | None = None) -> str

class TTSConfig(BaseModel)

Configuration settings for text-to-speech audio feedback.

def validate_audio_format(cls, v: str) -> str

def validate_output_dir(cls, v: str) -> str

class WorkerNode(Protocol)

Protocol matching standard attributes of routing nodes.

def id(self) -> str

def load(self) -> float

def status(self) -> str

def cpu_usage(self) -> float

class NodeSelector(ABC)

Base interface for all worker node routing selectors.

def select_node(self, nodes: list[WorkerNode]) -> WorkerNode

class SessionCountSelector(NodeSelector)

Selects the worker node with the minimum active session load.

def select_node(self, nodes: list[WorkerNode]) -> WorkerNode

class CPULoadSelector(NodeSelector)

Selects the worker node with the lowest CPU load report.

def select_node(self, nodes: list[WorkerNode]) -> WorkerNode

class EgressManager

Coordinates downstream data exports, JSONL tracing, and audio speech feedback.

def __init__(self, trace_path: str | None = None, tts_engine: Any | None = None)

def speak(self, text: str) -> None

def write_trace(self, trace_entry: dict) -> None

def record_signal(self, output_path: str, frames: list[Any]) -> None

class IngressManager

Manages active hardware source registration, monitoring, and failovers.

def __init__(self)

def register_source(self, name: str, source: HardwareSource, is_fallback: bool = False) -> None

def start(self) -> None

def stop(self) -> None

def active_name(self) -> str | None

def get_active_source(self) -> HardwareSource | None

def trigger_failover(self) -> None

class Session

Represents an active silent speech processing session.

Co-ordinates real-time signal level monitoring, stream tracking, quality scoring, and intent decoding on a background OpsQueue thread.

def __init__(self, id: str, config: SubvocalConfig, hardware: HardwareSource, classify_fn: Callable[[Frame], CommandToken | None], llm_provider: LLMProvider, context_provider: ContextProvider, executor: ActionExecutor, policy_engine: Any | None = None, trace_path: str | None = None, on_state_changed: Callable[[str], None] | None = None, on_token: Callable[[CommandToken], None] | None = None, on_intent: Callable[[Intent], None] | None = None, on_action: Callable[[Action, str], None] | None = None, on_error: Callable[[Exception], None] | None = None, telemetry: Any | None = None, data_channel: Any | None = None)

def state(self) -> str

def token_buffer(self) -> list[CommandToken]

def start(self) -> None

def stop(self) -> None

def push_frame(self, frame: Frame) -> None

def process_phrase(self) -> Action | None

class SessionStore(ABC)

Abstract interface for session configuration and state storage.

def save_session(self, session_id: str, state: str, config_yaml: str, metadata: dict[str, Any]) -> None

def get_session(self, session_id: str) -> dict[str, Any] | None

def delete_session(self, session_id: str) -> None

def list_sessions(self) -> list[dict[str, Any]]

class InMemorySessionStore(SessionStore)

In-memory session registry for lightweight, transient deployments.

def __init__(self)

def save_session(self, session_id: str, state: str, config_yaml: str, metadata: dict[str, Any]) -> None

def get_session(self, session_id: str) -> dict[str, Any] | None

def delete_session(self, session_id: str) -> None

def list_sessions(self) -> list[dict[str, Any]]

class SQLiteSessionStore(SessionStore)

SQLite-backed session registry for persistent edge configurations.

def __init__(self, db_path: str)

def save_session(self, session_id: str, state: str, config_yaml: str, metadata: dict[str, Any]) -> None

def get_session(self, session_id: str) -> dict[str, Any] | None

def delete_session(self, session_id: str) -> None

def list_sessions(self) -> list[dict[str, Any]]

class SessionWorker

Manages active sessions and load-tracking across a pool of silent speech sessions.

Exposes ``id``, ``load``, ``cpu_usage``, and ``status`` so an instance satisfies the routing ``WorkerNode`` protocol and can be ranked by the selectors in :mod:`subvocal.routing`. Equivalent to LiveKit's SessionWorker/AgentWorker.

def __init__(self, config: SubvocalConfig, max_sessions: int = 10, store: Any | None = None, worker_id: str | None = None)

def load(self) -> float

def cpu_usage(self) -> float

def status(self) -> str

def create_session(self, session_id: str, *args, **kwargs) -> Session

def get_session(self, session_id: str) -> Session | None

def remove_session(self, session_id: str) -> Session | None

def start_session(self, session_id: str) -> None

def close(self) -> None

class FrameRing

Thread-safe circular queue (ring buffer) storing biometric Frame segments.

def __init__(self, max_size: int = 100)

def push(self, frame: Frame) -> None

def pop(self) -> Frame | None

def clear(self) -> None

def get_all(self) -> list[Frame]

class StreamStats

Windowed tracking for frame arrivals, sample count, gaps, and inter-frame arrival jitter.

def __init__(self)

def observe(self, frame: Frame) -> None

def get_stats(self) -> dict[str, Any]

def reset(self) -> None

class BiometricDataChannelServer

TCP server that broadcasts real-time biometric metrics to connected dashboard/client nodes.

def __init__(self, host: str = '127.0.0.1', port: int = 8100)

def start(self) -> None

def broadcast(self, payload: dict[str, Any]) -> None

def close(self) -> None

class BiometricDataChannelClient

Helper client to connect to the BiometricDataChannelServer and parse message frames.

def __init__(self, host: str = '127.0.0.1', port: int = 8100)

def connect(self) -> None

def read_messages(self)

def close(self) -> None

class SignalLevel

Tracks and smooths the biological/electrical signal level of sEMG channels.

Equivalent to LiveKit's AudioLevel.

def __init__(self, active_level: float = 0.1, min_percentile: float = 40.0, update_interval_ms: int = 400, smooth_intervals: int = 2)

def observe(self, frame: Frame) -> None

def get_level(self, now: float) -> tuple[float, bool]

class SignalQualityScorer

Scores sEMG signal quality using physical metrics (saturation, drift, dropouts, SNR).

Outputs a MOS-like score mapped to EXCELLENT, GOOD, POOR, and LOST states. Equivalent to LiveKit's connectionquality Scorer.

def __init__(self, clip_max: float = 5.0, increase_factor: float = 0.4, decrease_factor: float = 0.8)

def score(self) -> float

def quality(self) -> str

def on_status_changed(self, callback: Callable[[str], None]) -> None

def update(self, frame: Frame) -> None

class StreamTracker

Monitors whether a biometric sEMG stream is active or stopped.

Applies a hysteresis threshold to prevent bouncing between states. Equivalent to LiveKit's StreamTracker.

def __init__(self, samples_required: int = 3, cycles_required: int = 5)

def status(self) -> str

def generation(self) -> int

def on_status_changed(self, callback: Callable[[str], None]) -> None

def observe(self, has_activity: bool) -> None

def set_paused(self, paused: bool) -> None

def reset(self) -> None

def stop(self) -> None

class ActionGrants(BaseModel)

Capability-scoped credentials detailing authorized actions and constraints.

class GrantsPolicy(AuthorizationPolicy)

Enforces capability-scoped authorization checks on proposed Actions.

def is_authorized(self, action: Action, context: UserContext) -> bool

set_context_grants

def set_context_grants(grants: ActionGrants) -> contextvars.Token

Sets the given ActionGrants in the current thread/coroutine execution context.

get_context_grants

def get_context_grants() -> ActionGrants | None

Retrieves active ActionGrants from the current context.

clear_context_grants

def clear_context_grants(token: contextvars.Token) -> None

Resets the context grants to their previous state.

generate_token

def generate_token(grants: ActionGrants, secret_key: str, expires_in: int = 3600) -> str

Generates a secure HMAC-signed base64 token carrying capability claims.

verify_token

def verify_token(token: str, secret_key: str) -> ActionGrants | None

Verifies the HMAC signature and expiration of a base64 token, returning decoded ActionGrants.

ensure_permission

def ensure_permission(action_type: str, confidence: float = 1.0) -> None

Checks the active context grants and raises a PermissionError if the action is unauthorized.

class TelemetryService

Interface for telemetry exporting, based on LiveKit telemetry service design.

def session_started(self, session_id: str, config: SubvocalConfig) -> None

def session_ended(self, session_id: str, stats: Any) -> None

def phrase_detected(self, session_id: str, phrase_id: str, duration_seconds: float) -> None

def intent_resolved(self, session_id: str, phrase_id: str, intent_name: str, confidence: float) -> None

def action_executed(self, session_id: str, action_type: str, status: str) -> None

def action_blocked(self, session_id: str, action_type: str, reason: str) -> None

def provider_retry(self, session_id: str, provider_name: str, attempt: int, error: str) -> None

def quality_changed(self, session_id: str, quality_score: float, quality_state: str) -> None

def error_occurred(self, session_id: str, error_type: str) -> None

class NullTelemetry(TelemetryService)

Default telemetry implementation that does nothing.

class PrometheusTelemetry(TelemetryService)

Telemetry service that exports metrics to Prometheus.

Prometheus cardinality: session_id is high cardinality (unbounded, one series per session). To keep series count bounded (Prometheus recommends <10k series per metric, ideally <100k total), session_id is NOT used as a Prometheus label. Metrics are aggregated globally or by low-cardinality labels (intent, action_type, etc.). If per-session debugging is required, use logs or hash session_id via _hash_session_id() into a small bucket, but do not expose raw session_id. Documented limit: if you must add session_id, hash and cap distinct values <1000 and ensure series are removed on session_ended to avoid leak.

def __init__(self, config: SubvocalConfig)

def session_started(self, session_id: str, config: SubvocalConfig) -> None

def session_ended(self, session_id: str, stats: Any) -> None

def phrase_detected(self, session_id: str, phrase_id: str, duration_seconds: float) -> None

def intent_resolved(self, session_id: str, phrase_id: str, intent_name: str, confidence: float) -> None

def action_executed(self, session_id: str, action_type: str, status: str) -> None

def action_blocked(self, session_id: str, action_type: str, reason: str) -> None

def provider_retry(self, session_id: str, provider_name: str, attempt: int, error: str) -> None

def quality_changed(self, session_id: str, quality_score: float, quality_state: str) -> None

def error_occurred(self, session_id: str, error_type: str) -> None

def shutdown(self) -> None

class OpsQueue

A thread-safe serialized work queue running on a single background worker thread.

Equivalent to LiveKit's OpsQueue.

def __init__(self, name: str, min_size: int = 128, maxsize: int | None = None, flush_on_stop: bool = False, custom_logger: logging.Logger | None = None)

def start(self) -> None

def stop(self) -> None

def enqueue(self, fn: Callable, *args, **kwargs) -> bool

def qsize(self) -> int

def is_full(self) -> bool

class IncrementalDispatcher

Allows multiple consumers to consume items thread-safely as they become available.

Producers can add items at any time. Equivalent to LiveKit's IncrementalDispatcher.

def __init__(self)

def add(self, item) -> None

def done(self) -> None

def for_each(self, callback: Callable) -> None

class ChangeNotifier

Registry for keyed observers to receive change notifications asynchronously.

Equivalent to LiveKit's ChangeNotifier.

def __init__(self)

def add_observer(self, key: str, on_changed: Callable[[], None]) -> None

def remove_observer(self, key: str) -> None

def has_observers(self) -> bool

def notify_changed(self) -> None

class ChangeNotifierManager

Manages a registry of named ChangeNotifier instances.

Equivalent to LiveKit's ChangeNotifierManager.

def __init__(self)

def get_notifier(self, key: str) -> ChangeNotifier | None

def get_or_create_notifier(self, key: str) -> ChangeNotifier

def remove_notifier(self, key: str, force: bool = False) -> None

class Debouncer

Resettable delay timer to debounce fast-firing events.

Equivalent to LiveKit's Debouncer.

def __init__(self, after_seconds: float)

def add(self, fn: Callable[[], None]) -> None

def set_duration(self, after_seconds: float) -> None

def cancel(self) -> None