API Reference
Auto-generated from docstrings across 45 modules. Regenerate with python tools/build_api_page.py.
subvocal.exceptions
Typed exception hierarchy for the Subvocal SDK.
Every error the SDK raises derives from :class:`SubvocalError`, so integrators can catch one base class at the application boundary::
try: pipeline.step() except SubvocalError as e: logger.error("subvocal failure: %s", e)
Subclasses also inherit the builtin exception type the SDK historically raised in each situation (``RuntimeError``, ``ImportError``, ...), so pre-existing ``except RuntimeError`` handlers keep working across upgrades.
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.
subvocal.paths
Resolution of writable data and model directories.
The SDK writes pipeline traces, correction logs, and trained model weights at runtime. After installation the package directory lives in ``site-packages`` and must never be written to, so all writable paths resolve to a per-user data directory instead. Resolution order:
1. Environment variables ``SUBVOCAL_DATA_DIR`` / ``SUBVOCAL_MODELS_DIR``. 2. The OS-conventional user data directory (via ``platformdirs``), e.g. ``~/Library/Application Support/subvocal`` on macOS or ``~/.local/share/subvocal`` on Linux.
Directories are created on first use.
get_data_dir
def get_data_dir() -> strWritable directory for traces, correction logs, and datasets.
get_models_dir
def get_models_dir() -> strWritable directory for trained and downloaded model weights.
subvocal.core.corrections
Correction-capture loop and fine-tuning hook implementation for Subvocal SDK.
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.
__init__
def __init__(self, log_path: str | None = None)Initializes the manager.
Args: log_path: Path to the JSONL log file. Defaults to 'sdk/data/corrections_log.jsonl'.
log_correction
def log_correction(self, raw_shorthand: str, decoded_intent: str, corrected_intent: str, context: UserContext) -> CorrectionLogEntryLogs a new correction entry to the local JSONL file.
get_corrections
def get_corrections(self) -> list[CorrectionLogEntry]Retrieves all logged corrections from the local file.
clear_logs
def clear_logs(self) -> NoneClears all logged corrections in the file.
class FinetuningHook
Converts correction logs into training datasets for fine-tuning LLMs.
export_to_openai
def export_to_openai(entries: list[CorrectionLogEntry], system_instruction: str = 'Translate silent speech shorthand and context into correct system actions.') -> list[dict[str, Any]]Converts entries to OpenAI chat fine-tuning format.
Format: {"messages": [{"role": "system", "content": ...}, {"role": "user", "content": ...}, {"role": "assistant", "content": ...}]}
export_to_gemini
def export_to_gemini(entries: list[CorrectionLogEntry], system_instruction: str = 'Translate silent speech shorthand and context into correct system actions.') -> list[dict[str, Any]]Converts entries to Google Gemini fine-tuning format.
Format: {"contents": [...], "systemInstruction": {...}}
export_to_jsonl
def export_to_jsonl(data: list[dict[str, Any]], output_path: str) -> intWrites the exported dataset list to a JSONL file.
subvocal.core.interfaces
Abstract base interfaces for subvocal middleware components.
Enables pluggable integration of various: - Hardware sources (real-time stream, replay file, synthetic generators). - LLM providers (Gemini, Claude, GPT, local Llama). - Action executors (browser control, device APIs, notifications). - Context providers (OS, active application, contacts, location).
class HardwareSource(ABC)
Abstract interface for subvocal/sEMG sensor data acquisition.
start
def start(self) -> NoneInitialize connection and start streaming raw sEMG data.
stop
def stop(self) -> NoneStop data acquisition and disconnect hardware clean.
read_frame
def read_frame(self, window_ms: int) -> FrameReads a buffered window of raw sEMG samples.
Args: window_ms: Time duration in milliseconds to buffer and retrieve.
Returns: A Frame containing the Sample points.
is_connected
def is_connected(self) -> boolReturns True if the sensor connection is healthy.
class LLMProvider(ABC)
Abstract interface for turning noisy classified shorthand into semantic Intents.
reconstruct_intent
def reconstruct_intent(self, tokens: list[CommandToken], context: UserContext) -> IntentResolves a noisy shorthand token stream to a structured Intent.
Args: tokens: A sequence of classified CommandTokens. context: The UserContext state snapshot at execution time.
Returns: The reconstructed Intent.
get_provider_name
def get_provider_name(self) -> strReturns the name of the LLM provider (e.g., 'gemini', 'anthropic').
class ActionExecutor(ABC)
Abstract interface for dispatching resolved agentic Actions.
execute
def execute(self, action: Action) -> AnyDispatches the action to device APIs or agent tools.
Args: action: The Action object to execute.
Returns: The result of the execution.
can_execute
def can_execute(self, action: Action) -> boolReturns True if the executor supports this specific action type.
class ContextProvider(ABC)
Abstract interface for retrieving live device or environment context.
get_context
def get_context(self) -> UserContextRetrieves a snapshot of the active user/system context.
Returns: The UserContext populated with active state metadata.
get_provider_name
def get_provider_name(self) -> strReturns the name/identifier of this context provider.
class Classifier(ABC)
Abstract interface for classifying physiological raw signals into command tokens.
predict
def predict(self, frame: Frame | Any) -> CommandToken | NoneClassifies a Frame of raw signals into a CommandToken (applies gating/cooldown if configured).
predict_raw
def predict_raw(self, frame: Frame | Any) -> tuple[str, float, list[float]]Predicts the probability distribution for a Frame of raw signals.
Returns: (predicted_class_label, max_probability, all_probabilities_list)
labels
def labels(self) -> list[str]Returns the list of output labels/classes supported by the classifier.
subvocal.core.llm_providers
Concrete implementations of the LLMProvider interface using urllib.
class BaseLLMProvider(LLMProvider)
Base provider containing prompt formatting and request helpers.
__init__
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)Initializes the base LLM provider.
Args: api_key: API authorization key. Resolves from environment if None. model_name: Name of the LLM model to request. mock_response: Optional mock string response to bypass HTTP calls (for testing). prompt_version: The version string of the prompt template to use. timeout: Per-request HTTP timeout in seconds. max_retries: Additional attempts after the first failure for transient errors (connection failures and retryable HTTP statuses). backoff_seconds: Base delay for exponential backoff between retries.
class ClaudeProvider(BaseLLMProvider)
Anthropic Claude LLM provider client.
get_provider_name
def get_provider_name(self) -> strNo description.
reconstruct_intent
def reconstruct_intent(self, tokens: list[CommandToken], context: UserContext) -> IntentNo description.
class OpenAIProvider(BaseLLMProvider)
OpenAI GPT completions provider client.
get_provider_name
def get_provider_name(self) -> strNo description.
reconstruct_intent
def reconstruct_intent(self, tokens: list[CommandToken], context: UserContext) -> IntentNo description.
class GeminiProvider(BaseLLMProvider)
Google Gemini content generation provider client.
get_provider_name
def get_provider_name(self) -> strNo description.
reconstruct_intent
def reconstruct_intent(self, tokens: list[CommandToken], context: UserContext) -> IntentNo description.
class LlamaProvider(BaseLLMProvider)
Local Llama provider client (running via Ollama).
__init__
def __init__(self, ollama_host: str | None = None, model_name: str | None = None, mock_response: str | None = None, prompt_version: str = 'v1')Initializes the Ollama Llama provider client.
get_provider_name
def get_provider_name(self) -> strNo description.
reconstruct_intent
def reconstruct_intent(self, tokens: list[CommandToken], context: UserContext) -> IntentNo description.
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`).
__init__
def __init__(self, min_confidence: float = 0.0)Initializes the heuristic provider.
Args: min_confidence: Confidence floor below which the intent command is reported as UNKNOWN rather than a low-quality guess.
get_provider_name
def get_provider_name(self) -> strNo description.
reconstruct_intent
def reconstruct_intent(self, tokens: list[CommandToken], context: UserContext) -> IntentNo description.
resolve_provider
def resolve_provider(prefer: str | None = None, **kwargs) -> LLMProviderReturns 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.
subvocal.core.models
Core data models representing the flow from raw physiological signals to actions.
Defines: - Sample: A single point-in-time multi-channel sensor reading. - Frame: A window of Samples buffered for classifier consumption. - CommandToken: A token/shorthand output by a classifier. - Intent: Reconstructed semantic command and arguments. - Action: Executable instruction dispatched to an executor.
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.
to_numpy
def to_numpy(self) -> np.ndarrayConverts the frame samples into a 2D NumPy array.
Returns: np.ndarray of shape (num_samples, num_channels)
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.
subvocal.core.pipeline
SubvocalPipeline class coordinating the hardware, classification, and execution layers.
class PipelineStats
Running counters for a pipeline instance, exposed as ``pipeline.stats``.
as_dict
def as_dict(self) -> dict[str, Any]Returns the counters as a plain dictionary (for logging/export).
class SubvocalPipeline
Orchestrates hardware ingestion, classifier decoding, intent reconstruction, and action execution.
__init__
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)No description.
trace_path
def trace_path(self) -> str | NoneNo description.
trace_path
def trace_path(self, val: str | None) -> NoneNo description.
hardware
def hardware(self) -> HardwareSourceNo description.
hardware
def hardware(self, val: HardwareSource) -> NoneNo description.
classify_fn
def classify_fn(self) -> Callable[[Frame], CommandToken | None]No description.
classify_fn
def classify_fn(self, val: Callable[[Frame], CommandToken | None]) -> NoneNo description.
llm_provider
def llm_provider(self) -> LLMProviderNo description.
llm_provider
def llm_provider(self, val: LLMProvider) -> NoneNo description.
context_provider
def context_provider(self) -> ContextProviderNo description.
context_provider
def context_provider(self, val: ContextProvider) -> NoneNo description.
executor
def executor(self) -> ActionExecutorNo description.
executor
def executor(self, val: ActionExecutor) -> NoneNo description.
policy_engine
def policy_engine(self) -> AnyNo description.
policy_engine
def policy_engine(self, val: Any) -> NoneNo description.
phrase_timeout_seconds
def phrase_timeout_seconds(self) -> floatNo description.
phrase_timeout_seconds
def phrase_timeout_seconds(self, val: float) -> NoneNo description.
dry_run
def dry_run(self) -> boolNo description.
dry_run
def dry_run(self, val: bool) -> NoneNo description.
raise_on_policy_violation
def raise_on_policy_violation(self) -> boolNo description.
raise_on_policy_violation
def raise_on_policy_violation(self, val: bool) -> NoneNo description.
token_buffer
def token_buffer(self) -> list[CommandToken]Returns the current accumulated tokens in the buffer.
clear_buffer
def clear_buffer(self) -> NoneClears the accumulated command token buffer.
step
def step(self, window_ms: int = 100) -> Action | NonePerforms a single step of the pipeline.
Args: window_ms: Buffer window duration in milliseconds to acquire.
Returns: An executed Action if an intent was resolved and dispatched, otherwise None.
process_phrase
def process_phrase(self) -> Action | NoneForces immediate decoding and execution of the accumulated tokens.
Returns: The executed Action if successful, otherwise None.
subvocal.core.prompts
Prompt templates and version manager for intent reconstruction.
class PromptManager
Manages versioned prompt templates for LLM intent decoding.
__init__
def __init__(self, default_version: str = 'v1')No description.
register_template
def register_template(self, version: str, template: str) -> NoneRegister a new custom prompt template version.
get_template
def get_template(self, version: str | None = None) -> strRetrieve a raw prompt template string by version.
format_prompt
def format_prompt(self, noisy_input: str, heuristic_recommendation: str, web_context: str, calendar: str, contacts: str, history: str, version: str | None = None) -> strFormats the selected prompt template with active inputs.
subvocal.core.security
Security policies and authorization gating for subvocal actions.
class AuthorizationPolicy(ABC)
Abstract base class for all subvocal execution policies.
is_authorized
def is_authorized(self, action: Action, context: UserContext) -> boolEvaluate if the given action is authorized to execute.
Args: action: The proposed Action object. context: The UserContext state snapshot at proposal time.
Returns: True if authorized, False otherwise.
class ConfidenceThresholdPolicy(AuthorizationPolicy)
Restricts execution of actions if classifier/reconstruction confidence is too low.
__init__
def __init__(self, threshold: float = 0.8)No description.
is_authorized
def is_authorized(self, action: Action, context: UserContext) -> boolNo description.
class CommandWhitelistPolicy(AuthorizationPolicy)
Restricts commands to a strict whitelist of allowed actions.
__init__
def __init__(self, allowed_commands: list[str])No description.
is_authorized
def is_authorized(self, action: Action, context: UserContext) -> boolNo description.
class ContextBoundPolicy(AuthorizationPolicy)
Blocks execution of sensitive commands unless the active application is safe.
__init__
def __init__(self, sensitive_commands: list[str], safe_applications: list[str])No description.
is_authorized
def is_authorized(self, action: Action, context: UserContext) -> boolNo description.
class PolicyEngine
Orchestrates authorization consensus across multiple active policies.
__init__
def __init__(self, policies: list[AuthorizationPolicy] | None = None)No description.
add_policy
def add_policy(self, policy: AuthorizationPolicy) -> NoneAdd a new policy to the engine.
is_authorized
def is_authorized(self, action: Action, context: UserContext) -> boolReturns True only if all configured policies approve the action.
subvocal.core.testing
Shared test fixtures for Subvocal SDK unit tests.
class MockHardwareSource(HardwareSource)
Generates continuous synthetic sEMG samples (sine + noise).
__init__
def __init__(self, fs: float = 1000.0, num_channels: int = 4)No description.
start
def start(self) -> NoneNo description.
stop
def stop(self) -> NoneNo description.
read_frame
def read_frame(self, window_ms: int) -> FrameNo description.
is_connected
def is_connected(self) -> boolNo description.
class MockLLMProvider(LLMProvider)
Maps shorthand token text patterns to GOTO or CLICK intents.
reconstruct_intent
def reconstruct_intent(self, tokens: list[CommandToken], context: UserContext) -> IntentNo description.
get_provider_name
def get_provider_name(self) -> strNo description.
class MockActionExecutor(ActionExecutor)
Records dispatched actions for assertion in tests.
__init__
def __init__(self)No description.
execute
def execute(self, action: Action) -> AnyNo description.
can_execute
def can_execute(self, action: Action) -> boolNo description.
class MockContextProvider(ContextProvider)
Returns a static browser UserContext snapshot.
get_context
def get_context(self) -> UserContextNo description.
get_provider_name
def get_provider_name(self) -> strNo description.
subvocal.context.manager
Context Manager for retrieving, filtering, and searching user context.
Integrates context querying with custom articulatory phonetic distance matching to enable context-aware silent speech decoding.
class ContextManager
Manages active UserContext and provides phonetic searching utilities.
__init__
def __init__(self, context: UserContext)No description.
update_context
def update_context(self, new_context: UserContext)Update the active user context state.
get_contact_names
def get_contact_names(self) -> list[str]Get list of all contact names in plain text.
get_contact_shorthands
def get_contact_shorthands(self) -> list[str]Get list of all pre-computed contact name shorthands.
get_calendar_titles
def get_calendar_titles(self) -> list[str]Get list of all calendar event titles in plain text.
get_calendar_shorthands
def get_calendar_shorthands(self) -> list[str]Get list of all calendar event title shorthands.
get_visible_element_labels
def get_visible_element_labels(self) -> list[str]Get list of all visible UI element labels in plain text.
get_visible_element_shorthands
def get_visible_element_shorthands(self) -> list[str]Get list of all visible UI element shorthands.
get_all_context_words
def get_all_context_words(self) -> list[str]Compile a list of all potential word tokens present in the active context.
Useful for seeding the dictionary candidate generator in the decoder.
search_contacts
def search_contacts(self, noisy_shorthand: str) -> list[tuple[Contact, float]]Search contacts by comparing noisy shorthand against pre-computed contact shorthands.
Returns: List of tuples (Contact, articulatory_distance) sorted by distance.
search_elements
def search_elements(self, noisy_shorthand: str) -> list[tuple[UIElement, float]]Search visible UI elements using articulatory distance matching.
Returns: List of tuples (UIElement, articulatory_distance) sorted by distance.
subvocal.context.mock_data
Mock user context generator for testing and demonstration.
Generates structured mock data for contacts, calendar events, location, conversation history, and active browser app state, automatically pre-computing compressed phonetic shorthands for all text labels.
generate_mock_context
def generate_mock_context() -> UserContextGenerate a high-fidelity UserContext populated with mock data.
subvocal.context.providers
Modular ContextProvider implementations for the Subvocal SDK.
class StaticContextProvider(ContextProvider)
Simple provider wrapping a static UserContext instance (useful for tests/mock runs).
__init__
def __init__(self, context: UserContext, name: str = 'static_context')No description.
get_context
def get_context(self) -> UserContextNo description.
get_provider_name
def get_provider_name(self) -> strNo description.
class CalendarContextProvider(ContextProvider)
Modular provider for retrieving calendar events.
__init__
def __init__(self, events: list[CalendarEvent] | None = None)No description.
get_provider_name
def get_provider_name(self) -> strNo description.
get_context
def get_context(self) -> UserContextNo description.
class ContactsContextProvider(ContextProvider)
Modular provider for retrieving user contact directory.
__init__
def __init__(self, contacts: list[Contact] | None = None)No description.
get_provider_name
def get_provider_name(self) -> strNo description.
get_context
def get_context(self) -> UserContextNo description.
class LocationContextProvider(ContextProvider)
Modular provider for retrieving current location details.
__init__
def __init__(self, location: LocationInfo | None = None)No description.
get_provider_name
def get_provider_name(self) -> strNo description.
get_context
def get_context(self) -> UserContextNo description.
class AppStateContextProvider(ContextProvider)
Modular provider for retrieving active app state and visible elements.
__init__
def __init__(self, app_state: AppState | None = None)No description.
get_provider_name
def get_provider_name(self) -> strNo description.
get_context
def get_context(self) -> UserContextNo description.
class CompositeContextProvider(ContextProvider)
Aggregates multiple individual ContextProviders into a single UserContext.
__init__
def __init__(self, providers: list[ContextProvider])No description.
get_provider_name
def get_provider_name(self) -> strNo description.
get_context
def get_context(self) -> UserContextNo description.
subvocal.context.schema
Pydantic schemas for the Subvocal user context.
Defines the structure for contacts, calendar events, location, conversation, and app state (including visible interactive elements).
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.
subvocal.hardware.brainflow_compat
class BoardIds(enum.IntEnum)
No description.
class LogLevels(enum.IntEnum)
No description.
class BrainFlowPresets(enum.IntEnum)
No description.
class BrainFlowError(Exception)
No description.
__init__
def __init__(self, message: str, exit_code: int)No description.
class BrainFlowInputParams
No description.
__init__
def __init__(self) -> NoneNo description.
to_json
def to_json(self) -> strNo description.
class BoardShim
Fallback compatible BoardShim class implementing a pure-Python biosensor stream.
__init__
def __init__(self, board_id: int, input_params: BrainFlowInputParams) -> NoneNo description.
prepare_session
def prepare_session(self) -> NoneNo description.
start_stream
def start_stream(self, buffer_size: int = 200000, streamer_params: str = '') -> NoneNo description.
stop_stream
def stop_stream(self) -> NoneNo description.
release_session
def release_session(self) -> NoneNo description.
get_board_data
def get_board_data(self, preset: int = BrainFlowPresets.DEFAULT_PRESET.value) -> np.ndarrayNo description.
get_current_board_data
def get_current_board_data(self, num_samples: int, preset: int = BrainFlowPresets.DEFAULT_PRESET.value) -> np.ndarrayNo description.
is_prepared
def is_prepared(self) -> boolNo description.
get_sampling_rate
def get_sampling_rate(cls, board_id: int, preset: int = BrainFlowPresets.DEFAULT_PRESET.value) -> intNo description.
get_num_rows
def get_num_rows(cls, board_id: int, preset: int = BrainFlowPresets.DEFAULT_PRESET.value) -> intNo description.
get_eeg_channels
def get_eeg_channels(cls, board_id: int, preset: int = BrainFlowPresets.DEFAULT_PRESET.value) -> list[int]No description.
get_exg_channels
def get_exg_channels(cls, board_id: int, preset: int = BrainFlowPresets.DEFAULT_PRESET.value) -> list[int]No description.
get_emg_channels
def get_emg_channels(cls, board_id: int, preset: int = BrainFlowPresets.DEFAULT_PRESET.value) -> list[int]No description.
get_timestamp_channel
def get_timestamp_channel(cls, board_id: int, preset: int = BrainFlowPresets.DEFAULT_PRESET.value) -> intNo description.
get_package_num_channel
def get_package_num_channel(cls, board_id: int, preset: int = BrainFlowPresets.DEFAULT_PRESET.value) -> intNo description.
get_battery_channel
def get_battery_channel(cls, board_id: int, preset: int = BrainFlowPresets.DEFAULT_PRESET.value) -> intNo description.
get_accel_channels
def get_accel_channels(cls, board_id: int, preset: int = BrainFlowPresets.DEFAULT_PRESET.value) -> list[int]No description.
get_gyro_channels
def get_gyro_channels(cls, board_id: int, preset: int = BrainFlowPresets.DEFAULT_PRESET.value) -> list[int]No description.
subvocal.hardware.datasets
Dataset drivers for public electromyography databases (Ninapro, PutEMG, CSL-HDEMG).
class NinaproDriver(HardwareSource)
Streams electromyography signals from Ninapro MATLAB (.mat) files.
__init__
def __init__(self, file_path: str, fs: float = 2000.0, loop: bool = True)Initializes the Ninapro driver.
Args: file_path: Path to the downloaded Ninapro subject MATLAB .mat file. fs: Sampling frequency (default: 2000.0 Hz for DB2/DB3, DB1 is 100 Hz). loop: Boolean representing whether to loop data when EOF is reached.
start
def start(self) -> NoneNo description.
stop
def stop(self) -> NoneNo description.
is_connected
def is_connected(self) -> boolNo description.
read_frame
def read_frame(self, window_ms: int) -> FrameNo description.
class PutEMGDriver(HardwareSource)
Streams electromyography signals from PutEMG HDF5 (.h5) files.
__init__
def __init__(self, file_path: str, fs: float = 5120.0, loop: bool = True)Initializes the PutEMG driver.
Loads h5py dynamically.
start
def start(self) -> NoneNo description.
stop
def stop(self) -> NoneNo description.
is_connected
def is_connected(self) -> boolNo description.
read_frame
def read_frame(self, window_ms: int) -> FrameNo description.
class CSLHDEMGDriver(HardwareSource)
Streams electromyography signals from CSL-HDEMG binary or NumPy (.npy) files.
__init__
def __init__(self, file_path: str, fs: float = 2000.0, loop: bool = True, num_channels: int = 8)Initializes the CSL-HDEMG driver.
Args: file_path: Path to the NumPy (.npy) or raw binary float data file. fs: Sample rate of the recording. loop: Boolean representing whether to loop the data when EOF is reached. num_channels: Number of channels in the raw binary file (ignored for .npy, which encodes shape).
start
def start(self) -> NoneNo description.
stop
def stop(self) -> NoneNo description.
is_connected
def is_connected(self) -> boolNo description.
read_frame
def read_frame(self, window_ms: int) -> FrameNo description.
subvocal.hardware.drivers
Concrete implementations of HardwareSource drivers for the Subvocal SDK.
class FileReplayDriver(HardwareSource)
Replays raw sEMG data from a CSV file, simulating real-time acquisition.
__init__
def __init__(self, file_path: str, fs: float = 1000.0, loop: bool = True)Initializes the file replay driver.
Args: file_path: Path to the recorded CSV file. fs: Sample rate of the recording. loop: Boolean representing whether to loop the data when EOF is reached.
start
def start(self) -> NoneNo description.
stop
def stop(self) -> NoneNo description.
is_connected
def is_connected(self) -> boolNo description.
read_frame
def read_frame(self, window_ms: int) -> FrameNo description.
class SyntheticSignalGenerator(HardwareSource)
Generates continuous physiological sEMG noise and injects command triggers.
__init__
def __init__(self, fs: float = 1000.0, num_channels: int = 8)Initializes the synthetic signal generator.
Args: fs: Sample frequency in Hz. num_channels: Number of channels to generate (maps to 5-zone layouts).
start
def start(self) -> NoneNo description.
stop
def stop(self) -> NoneNo description.
is_connected
def is_connected(self) -> boolNo description.
trigger_command
def trigger_command(self, command: str, duration_ms: int = 300) -> NoneInjects a high-amplitude transient signal profile on selected channels.
Args: command: Command name (e.g., 'gt', 'clk', 'srch', 'typ'). duration_ms: Burst duration in milliseconds.
read_frame
def read_frame(self, window_ms: int) -> FrameNo description.
class OpenBCICytonDriver(HardwareSource)
Acquires raw biosignals from OpenBCI Cyton boards via BrainFlow.
__init__
def __init__(self, port: str | None = None, simulated: bool = True)Initializes the Cyton driver.
Loads BrainFlow dynamically.
start
def start(self) -> NoneNo description.
stop
def stop(self) -> NoneNo description.
is_connected
def is_connected(self) -> boolNo description.
read_frame
def read_frame(self, window_ms: int) -> FrameNo description.
class DelsysTrignoDriver(HardwareSource)
Direct TCP Socket driver connecting to the Delsys Trigno Control Utility.
__init__
def __init__(self, host: str = 'localhost', data_port: int = 50043, cmd_port: int = 50040, num_channels: int = 8, fs: float = 2000.0)Initializes the Delsys socket driver.
start
def start(self) -> NoneNo description.
stop
def stop(self) -> NoneNo description.
is_connected
def is_connected(self) -> boolNo description.
read_frame
def read_frame(self, window_ms: int) -> FrameNo description.
subvocal.emg_core.dsp.brainflow_filter
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.
perform_lowpass
def perform_lowpass(cls, data: np.ndarray, sampling_rate: int, cutoff: float, order: int, filter_type: int, ripple: float) -> NoneNo description.
perform_highpass
def perform_highpass(cls, data: np.ndarray, sampling_rate: int, cutoff: float, order: int, filter_type: int, ripple: float) -> NoneNo description.
perform_bandpass
def perform_bandpass(cls, data: np.ndarray, sampling_rate: int, start_freq: float, stop_freq: float, order: int, filter_type: int, ripple: float) -> NoneNo description.
perform_bandstop
def perform_bandstop(cls, data: np.ndarray, sampling_rate: int, start_freq: float, stop_freq: float, order: int, filter_type: int, ripple: float) -> NoneNo description.
remove_environmental_noise
def remove_environmental_noise(cls, data: np.ndarray, sampling_rate: int, noise_type: int) -> NoneNo description.
perform_rolling_filter
def perform_rolling_filter(cls, data: np.ndarray, period: int, agg_operation: int) -> NoneNo description.
perform_downsampling
def perform_downsampling(cls, data: np.ndarray, period: int, agg_operation: int) -> np.ndarrayNo description.
get_window
def get_window(cls, window_function: int, window_len: int) -> np.ndarrayNo description.
get_psd
def get_psd(cls, data: np.ndarray, sampling_rate: int, window: int) -> tuple[np.ndarray, np.ndarray]No description.
get_psd_welch
def get_psd_welch(cls, data: np.ndarray, nfft: int, overlap: int, sampling_rate: int, window: int) -> tuple[np.ndarray, np.ndarray]No description.
get_band_power
def get_band_power(cls, psd: tuple[np.ndarray, np.ndarray], freq_start: float, freq_end: float) -> floatNo description.
subvocal.emg_core.dsp.features
Feature extraction for sEMG segments.
Implements the TD0/TD10 feature pipeline (validated on the EMG-UKA corpus). This decomposes the raw sEMG signal into: 1. Low-frequency articulatory gestures (using a double moving average filter). 2. High-frequency muscular activity.
Features are computed over sliding window frames, stacked with context, and averaged over the entire segment to create robust representations.
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.ndarrayExtract 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.ndarrayAggregate 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.ndarrayExtract 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.ndarrayExtract features for a batch of segments.
subvocal.emg_core.dsp.filters
DSP filters for sEMG signal preprocessing.
All functions operate on numpy arrays.
PHYSIOLOGICAL RATIONALE FOR FILTER PARAMETERS: Traditional electromyography (EMG) used in physical prosthetics and active gesture recognition typically utilizes a bandpass filter from 20 Hz to 450 Hz. This bandwidth is chosen to capture high-speed motor unit action potentials (MUAPs) generated by large skeletal muscle contractions (e.g., bicep flexion) while rejecting movement artifacts (which lie below 10–20 Hz).
However, for silent speech interface (SSI) surface electromyography (sEMG) around the vocal tract, jaw, and throat: 1. **Low-Velocity Articulatory Movements**: The vocal articulators (tongue, throat musculature, jaw) move at low speeds, producing slow neuromuscular signal envelopes (1–15 Hz) during internal speech imagery/subvocalization. 2. **Rejection of Speech Dynamics**: Standard acoustic speech features operate at higher frequency bands, but silent neuromuscular activations are primary envelope-based targets. 3. **Powerline Interference Gating**: Powerline noise (50 Hz in EU, 60 Hz in US) heavily corrupts biological signals. By constraining the bandpass filter's upper bound to 50.0 Hz, we inherently reject the majority of powerline interference before it can leak into feature extractors. 4. **Nyquist Limit Constraints**: Wearable device power budgets restrict hardware sampling rates (typically 250 Hz). At fs=250 Hz, the Nyquist limit is 125 Hz. A filter cutoff of 50 Hz ensures we satisfy the Nyquist-Shannon sampling theorem with a generous guard band, avoiding aliasing artifacts.
Therefore, we implement and default to the AlterEgo-inspired 1.3–50.0 Hz bandpass filter, but support standard 20–450.0 Hz ranges via parameter config.
remove_dc
def remove_dc(signal: np.ndarray, window_samples: int = 250) -> np.ndarrayRemove 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.ndarrayButterworth 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.ndarrayNotch 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.ndarraySimple 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.ndarrayFull 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.ndarrayPreprocess 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.
subvocal.emg_core.ml.benchmark
Inference benchmarking harness for sEMG classifiers.
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) -> intEstimate 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.
subvocal.emg_core.ml.config_schema
Pydantic schema for reproducible model training configuration.
class TrainingConfig(BaseModel)
Configuration class for reproducible training runs.
subvocal.emg_core.ml.export
Model export and quantization pipeline for sEMG classifiers.
export_to_onnx
def export_to_onnx(user_id: str, model_type: str, export_path: str) -> strExport 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) -> boolCompile 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) -> boolCompile 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.
subvocal.emg_core.ml.infer
class InferenceEngine(Classifier)
Inference engine with debounce and confidence thresholding.
Implements the core.interfaces.Classifier interface.
__init__
def __init__(self, user_id: str, model_type: str = 'rf', confidence_threshold: float = config.CONFIDENCE_THRESHOLD, cooldown_ms: int = config.COOLDOWN_MS)No description.
labels
def labels(self) -> list[str]No description.
predict
def predict(self, frame: Frame | np.ndarray) -> CommandToken | NoneClassify a segment and apply gating logic.
Args: frame: A Frame object or 2D array of shape (segment_length, num_channels)
Returns: CommandToken if successful, else None.
predict_raw
def predict_raw(self, frame: Frame | np.ndarray) -> tuple[str, float, list[float]]Classify a segment/frame without debounce gates.
Returns: (best_command, probability, all_probabilities)
subvocal.emg_core.ml.model_io
Model I/O utilities for saving and loading classifiers.
get_model_path
def get_model_path(user_id: str, model_type: str = 'rf') -> strGet 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') -> strSave 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') -> boolCheck if a model exists for a user.
subvocal.emg_core.ml.pretrain
Pre-training script for deep neural sEMG classifiers on public baseline datasets.
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.
subvocal.emg_core.ml.train
Training pipeline for sEMG command classifiers.
Supports: 1. RandomForest (Standard ML baseline using TD10 aggregated features). 2. SVM (Feature-based Support Vector Machine baseline). 3. 1D CNN (Deep learning model operating on raw time-series segments). 4. GRU (Bidirectional temporal RNN operating on raw time-series segments). 5. Transformer (Attention-based sequential encoder operating on raw segments).
class EMG1DCNN(nn.Module)
1D CNN for sEMG segment classification.
Applies convolutions over the temporal dimension of raw multichannel sEMG.
__init__
def __init__(self, num_channels: int, num_classes: int, segment_length: int = 150)No description.
forward
def forward(self, x)No description.
class EMGGRU(nn.Module)
Bidirectional GRU for temporal tracking of sEMG gestures.
__init__
def __init__(self, num_channels: int, num_classes: int, hidden_size: int = 64, num_layers: int = 2)No description.
forward
def forward(self, x)No description.
class EMGTransformer(nn.Module)
Small Transformer encoder for raw multichannel sEMG sequence classification.
__init__
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)No description.
forward
def forward(self, x)No description.
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.ndarrayPreprocess 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.
subvocal.mcp.server
Model Context Protocol (MCP) reference server for Subvocal Middleware.
class MockContextProvider(ContextProvider)
Exposes mock system context state.
get_context
def get_context(self) -> UserContextNo description.
get_provider_name
def get_provider_name(self) -> strNo description.
class MockActionExecutor(ActionExecutor)
Executes actions and logs execution history.
__init__
def __init__(self)No description.
execute
def execute(self, action: Action) -> AnyNo description.
can_execute
def can_execute(self, action: Action) -> boolNo description.
class SubvocalMCPServer
Zero-dependency Model Context Protocol server communicating via stdio.
__init__
def __init__(self)No description.
handle_request
def handle_request(self, req: dict[str, Any]) -> dict[str, Any] | NoneProcess an incoming JSON-RPC 2.0 request and return the response dictionary.
run
def run(self)Standard input/output reader loop.
main
def main() -> NoneConsole entry point for the ``subvocal-mcp`` stdio server.
subvocal.shorthand.decoder
Hybrid decoder for subvocal compressed shorthand.
Combines an articulatory-distance heuristic alignment algorithm (dynamic programming) with a model-agnostic LLM reconstruction layer.
articulatory_distance
def articulatory_distance(word1: str, word2: str) -> floatCalculates 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 | NoneHelper 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 | NoneReconstructs 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".
subvocal.shorthand.simulator
Noisy channel simulator for subvocal compressed shorthand.
Converts target command phrases to compressed shorthand and applies realistic articulatory sEMG noise (substitutions, insertions, and deletions).
get_articulatory_substitution
def get_articulatory_substitution(char: str) -> strGet 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) -> strApplies 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)
subvocal.shorthand.spec
Shorthand and phonetic specifications for the subvocal interface.
Defines the articulatory phonetic groups representing sEMG jaw/throat activation clusters, and compression rules to construct the shorthand representation.
compress_word
def compress_word(word: str) -> strCompress 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.
subvocal.shorthand.vocab
Target command vocabulary for the Subvocal interface.
Defines a phonetically diverse set of 17 commands optimized for throat/jaw surface Electromyography (sEMG) classification.
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.
subvocal.tts.engine
Text-to-Speech (TTS) engine for subvocal feedback.
Supports native macOS 'say' / 'afplay' utilities, OpenAI Audio API, and pyttsx3.
class TTSEngine
Multi-backend Text-to-Speech generator and player.
__init__
def __init__(self, config: TTSConfig | None = None)No description.
speak
def speak(self, text: str, filename: str | None = None) -> strSynthesize and speak the text, saving the audio to a file.
Args: text: The plain text statement to speak. filename: Target file name (optional). If omitted, a file name is auto-generated based on text hash. Returns: The absolute path of the generated audio file.
subvocal.tts.schema
Pydantic schemas for the Text-to-Speech (TTS) engine.
Defines voices, speeds, formats, and configurations for audio feedback.
class TTSConfig(BaseModel)
Configuration settings for text-to-speech audio feedback.
subvocal.routing.selector
class WorkerNode(Protocol)
Protocol matching standard attributes of routing nodes.
id
def id(self) -> strNo description.
load
def load(self) -> floatNo description.
status
def status(self) -> strNo description.
cpu_usage
def cpu_usage(self) -> floatNo description.
class NodeSelector(ABC)
Base interface for all worker node routing selectors.
select_node
def select_node(self, nodes: list[WorkerNode]) -> WorkerNodeSelects the best node from candidate worker nodes.
Args: nodes: Candidates list.
Returns: The chosen WorkerNode.
Raises: ValueError: If nodes list is empty.
class SessionCountSelector(NodeSelector)
Selects the worker node with the minimum active session load.
select_node
def select_node(self, nodes: list[WorkerNode]) -> WorkerNodeNo description.
class CPULoadSelector(NodeSelector)
Selects the worker node with the lowest CPU load report.
select_node
def select_node(self, nodes: list[WorkerNode]) -> WorkerNodeNo description.
subvocal.runtime.egress
class EgressManager
Coordinates downstream data exports, JSONL tracing, and audio speech feedback.
__init__
def __init__(self, trace_path: str | None = None, tts_engine: Any | None = None)No description.
speak
def speak(self, text: str) -> NonePlays speech feedback to the user via TTS.
write_trace
def write_trace(self, trace_entry: dict) -> NoneAppends a trace entry to the pipeline trace log file.
record_signal
def record_signal(self, output_path: str, frames: list[Any]) -> NoneSaves raw biometric frames to a JSON document for training/fine-tuning models.
subvocal.runtime.ingress
class IngressManager
Manages active hardware source registration, monitoring, and failovers.
__init__
def __init__(self)No description.
register_source
def register_source(self, name: str, source: HardwareSource, is_fallback: bool = False) -> NoneRegisters a biometric input source.
start
def start(self) -> NoneStarts the active ingress source stream.
stop
def stop(self) -> NoneStops all registered ingress sources.
active_name
def active_name(self) -> str | NoneNo description.
get_active_source
def get_active_source(self) -> HardwareSource | NoneNo description.
trigger_failover
def trigger_failover(self) -> NoneFails over dynamically to the registered fallback source.
subvocal.runtime.session
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.
__init__
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)No description.
state
def state(self) -> strReturns the current state of the session state machine.
token_buffer
def token_buffer(self) -> list[CommandToken]Returns the raw reference to the accumulated token buffer.
start
def start(self) -> NoneInitializes and starts the session.
stop
def stop(self) -> NoneGracefully stops and closes the session.
push_frame
def push_frame(self, frame: Frame) -> NoneIngests a new Frame, running stream checks and classification asynchronously on OpsQueue.
process_phrase
def process_phrase(self) -> Action | NoneForces immediate decoding of the current token buffer and runs the resolved action.
subvocal.runtime.store
class SessionStore(ABC)
Abstract interface for session configuration and state storage.
save_session
def save_session(self, session_id: str, state: str, config_yaml: str, metadata: dict[str, Any]) -> NoneSaves or updates a session configuration and state snapshot.
get_session
def get_session(self, session_id: str) -> dict[str, Any] | NoneRetrieves a session snapshot by ID, returning metadata, config, and state.
delete_session
def delete_session(self, session_id: str) -> NoneDeletes a session snapshot by ID.
list_sessions
def list_sessions(self) -> list[dict[str, Any]]Lists all stored sessions.
class InMemorySessionStore(SessionStore)
In-memory session registry for lightweight, transient deployments.
__init__
def __init__(self)No description.
save_session
def save_session(self, session_id: str, state: str, config_yaml: str, metadata: dict[str, Any]) -> NoneNo description.
get_session
def get_session(self, session_id: str) -> dict[str, Any] | NoneNo description.
delete_session
def delete_session(self, session_id: str) -> NoneNo description.
list_sessions
def list_sessions(self) -> list[dict[str, Any]]No description.
class SQLiteSessionStore(SessionStore)
SQLite-backed session registry for persistent edge configurations.
__init__
def __init__(self, db_path: str)No description.
save_session
def save_session(self, session_id: str, state: str, config_yaml: str, metadata: dict[str, Any]) -> NoneNo description.
get_session
def get_session(self, session_id: str) -> dict[str, Any] | NoneNo description.
delete_session
def delete_session(self, session_id: str) -> NoneNo description.
list_sessions
def list_sessions(self) -> list[dict[str, Any]]No description.
subvocal.runtime.worker
class SessionWorker
Manages active sessions and load-tracking across a pool of silent speech sessions.
Equivalent to LiveKit's SessionWorker/AgentWorker.
__init__
def __init__(self, config: SubvocalConfig, max_sessions: int = 10, store: Any | None = None)No description.
load
def load(self) -> floatReturns the current load fraction of the worker (0.0 to 1.0).
status
def status(self) -> strReturns status string based on load.
create_session
def create_session(self, session_id: str, *args, **kwargs) -> SessionCreates, registers, and returns a new Session.
Raises ValueError if session pool capacity is exhausted.
get_session
def get_session(self, session_id: str) -> Session | NoneRetrieves a registered Session by ID, or None if not found.
remove_session
def remove_session(self, session_id: str) -> Session | NoneStops and unregisters a Session by ID.
start_session
def start_session(self, session_id: str) -> NoneStarts a registered session by ID.
close
def close(self) -> NoneGracefully closes all sessions and stops the worker.
subvocal.stream.buffer
class FrameRing
Thread-safe circular queue (ring buffer) storing biometric Frame segments.
__init__
def __init__(self, max_size: int = 100)No description.
push
def push(self, frame: Frame) -> NonePushes a new Frame onto the ring buffer, evicting the oldest if full.
pop
def pop(self) -> Frame | NonePops and returns the oldest Frame from the ring buffer, or None if empty.
clear
def clear(self) -> NoneClears the ring buffer.
get_all
def get_all(self) -> list[Frame]Returns a copy of all current frames in chronological order.
class StreamStats
Windowed tracking for frame arrivals, sample count, gaps, and inter-frame arrival jitter.
__init__
def __init__(self)No description.
observe
def observe(self, frame: Frame) -> NoneObserves a new frame and updates statistics.
get_stats
def get_stats(self) -> dict[str, Any]Returns a snapshot of the current stats.
reset
def reset(self) -> NoneResets all stats to zero.
subvocal.stream.datachannel
class BiometricDataChannelServer
TCP server that broadcasts real-time biometric metrics to connected dashboard/client nodes.
__init__
def __init__(self, host: str = '127.0.0.1', port: int = 8100)No description.
start
def start(self) -> NoneStarts the TCP server in a background thread.
broadcast
def broadcast(self, payload: dict[str, Any]) -> NoneSends a JSON-serialized dictionary to all active listeners.
close
def close(self) -> NoneCloses the server and drops all client connections.
class BiometricDataChannelClient
Helper client to connect to the BiometricDataChannelServer and parse message frames.
__init__
def __init__(self, host: str = '127.0.0.1', port: int = 8100)No description.
connect
def connect(self) -> NoneConnects to the server socket.
read_messages
def read_messages(self)Generator yielding parsed messages from the streaming buffer.
close
def close(self) -> NoneDisconnects the client socket.
subvocal.stream.level
class SignalLevel
Tracks and smooths the biological/electrical signal level of sEMG channels.
Equivalent to LiveKit's AudioLevel.
__init__
def __init__(self, active_level: float = 0.1, min_percentile: float = 40.0, update_interval_ms: int = 400, smooth_intervals: int = 2)No description.
observe
def observe(self, frame: Frame) -> NoneObserves a new Frame, computes its MAV, and updates sliding window metrics.
get_level
def get_level(self, now: float) -> tuple[float, bool]Returns the current smoothed signal level and active status.
Resets level to 0 if signal is stale (no updates received for 2x update_interval).
subvocal.stream.quality
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.
__init__
def __init__(self, clip_max: float = 5.0, increase_factor: float = 0.4, decrease_factor: float = 0.8)No description.
score
def score(self) -> floatReturns the current MOS-like quality score (1.0 - 4.5).
quality
def quality(self) -> strReturns the current mapped quality state.
on_status_changed
def on_status_changed(self, callback: Callable[[str], None]) -> NoneRegisters a callback to be run on quality state transitions.
update
def update(self, frame: Frame) -> NoneEvaluates a Frame, computes metrics, updates the quality score, and checks transitions.
subvocal.stream.tracker
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.
__init__
def __init__(self, samples_required: int = 3, cycles_required: int = 5)No description.
status
def status(self) -> strReturns the current status (active/stopped).
generation
def generation(self) -> intReturns the current tracker worker generation.
on_status_changed
def on_status_changed(self, callback: Callable[[str], None]) -> NoneRegisters a callback to execute on status transitions.
observe
def observe(self, has_activity: bool) -> NoneObserves an activity tick and updates status state using hysteresis.
set_paused
def set_paused(self, paused: bool) -> NonePauses or resumes tracking.
Increments generation to invalidate older checks.
reset
def reset(self) -> NoneResets the tracker to initial stopped state.
stop
def stop(self) -> NonePermanently stops the tracker.