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 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.

class Frame(BaseModel)

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

def to_numpy(self) -> np.ndarray

class CommandToken(BaseModel)

A token predicted by the raw classifier with associated confidence.

class Intent(BaseModel)

Semantic intent reconstructed from a token stream and active context.

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)

def token_buffer(self) -> list[CommandToken]

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 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 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

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, 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. 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, 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.

Returns: Preprocessed data of same shape.

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]]

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.

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 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) -> 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.