audio / vision

Audio and vision modules: microphone recording, transcription, text processing, and screenshot captioning.

Audio Processing

audio.audio_processing

VAD_CHUNK_SAMPLES: int = 512 module-attribute

Samples per VAD chunk — 250 ms at 16 kHz.

audio_rms(data: bytes, width: int = 2) -> int

Drop-in replacement for audioop.rms() (deprecated in Python 3.11, removed in 3.13).

drain_image_description_task(image_desc_task: Optional[asyncio.Task]) -> None

Consume a finished task result so asyncio does not report it as unhandled later.

find_device_index(p: pyaudio.PyAudio, role: str) -> int | None

Find the correct audio device index based on the platform and role.

Parameters:

Name	Type	Description	Default
p	PyAudio	PyAudio instance.	required
role	str	'microphone' for user mic input or 'loopback' for capturing playback via a virtual cable.	required

Returns:

Type	Description
int | None	The device index if found, otherwise None.

monitor_user_interrupt(tts_interrupt_event: asyncio.Event, can_speak_event: asyncio.Event, threshold: int = 325, silence_limit: float = 0.35, fmt: int | None = None, rate: int = 16000, channels: int = 1, chunk: int = 1024) async

Continuously listen for user speech while the agent is speaking. When RMS exceeds threshold, the tts_interrupt_event is set, signalling playback to stop. The coroutine runs until can_speak_event is set (i.e. the agent finished speaking).

segment_to_wav(audio_data: bytes, filepath: str, rate: int = 16000, channels: int = 1, sample_width: int = 2) -> None

Write raw PCM audio data as a WAV file.

Parameters

audio_data : Raw PCM bytes (int16, little-endian, mono). filepath : Output WAV file path. rate : Sample rate in Hz. channels : Number of channels (default: 1). sample_width : Bytes per sample (default: 2 for int16).

set_transcribe_audit(audit) -> None

Set the SideEffectAudit instance for transcribe_audio calls.

stop_image_description_task(image_desc_task: Optional[asyncio.Task], image_stop_event: Optional[asyncio.Event], timeout: float = 0.25) -> None async

Signal the background caption task to stop, then force-cancel if it lingers.

subdivide_for_vad(data: bytes, sample_width: int) -> list[bytes]

Subdivide a PyAudio read into VAD-sized slices.

PyAudio reads CHUNK=8192 samples (~512 ms), but Silero VAD expects 32 ms chunks (512 samples). This function splits each read into ~16 VAD-sized slices.

Parameters

data : Raw PCM bytes from PyAudio stream.read(). sample_width : Bytes per sample (2 for int16).

Returns

list[bytes] List of VAD_CHUNK_SAMPLES-sized slices. The last slice may be shorter if the input length is not evenly divisible.

transcribe_audio(model: Any, model_name, WAVE_OUTPUT_FILENAME: str, RATE: int = 16000, probability_threshold: float = 0.5, audit=None, skip_silence_gate: bool = False) -> str

Transcribe audio using the configured STT backend (Parakeet-TDT default).

Skips transcription entirely if the WAV file is silent or too short (~200ms), saving ~3s of GPU inference on empty recordings.

Parameters:

Name	Type	Description	Default
model	Any	STT backend instance (Qwen3ASRModel-compatible interface).	required
WAVE_OUTPUT_FILENAME	str	path to the WAV file to transcribe.	required
RATE	int	sample rate, used to determine max_audio_length.	16000
skip_silence_gate	bool	If True, bypass the RMS silence check. Use when VAD has already confirmed the audio contains speech (e.g., VAD streaming segments). Prevents quiet voice from being filtered out after neural detection.	False

Returns:

Type	Description
str	String containing the transcribed text.

transcribe_segment(model: Any, model_name: str, audio_data: bytes, rate: int = 16000, channels: int = 1, sample_width: int = 2, probability_threshold: float = 0.2) -> str | None

Transcribe raw PCM audio bytes by writing a temporary WAV file.

Mirrors the existing transcription flow (write WAV → transcribe → cleanup) but operates on in-memory bytes rather than a pre-written recording buffer.

Parameters

model : STT model instance (ParakeetTDTBackend or similar). model_name : Human-readable model identifier for logging. audio_data : Raw PCM bytes (int16, little-endian, mono). rate : Sample rate in Hz. channels : Number of channels. sample_width : Bytes per sample. probability_threshold : Confidence threshold for transcription output.

Returns

str or None Transcribed text on success, empty string if silent/short, or None on unrecoverable failure.

wav_is_silent(filepath: str, min_rms: int = 30, min_samples: int = 3200) -> bool

Return True if the WAV file is effectively silent or too short to contain speech.

Reads up to ~0.5s of audio at 16kHz (8000 samples), computes RMS amplitude. If the file has fewer than min_samples frames or the RMS falls below min_rms, transcription is skipped entirely -- saves ~3s of GPU inference on empty recordings.

Parameters

filepath : path to a WAV file (16-bit PCM expected). min_rms : minimum RMS amplitude to consider the audio speech. Default 30 is well above ambient noise floor for typical microphones (~5-10). min_samples : minimum number of 16-bit PCM samples. At 16kHz this corresponds to ~200ms of audio -- enough for a short utterance.

Returns

True if the file should be skipped (silent or too short).

Text Processing

audio.text_processing

clean_text(text)

Cleans the text by removing unwanted characters and patterns.

split_buffer_into_sentences(buffer, flush=False)

Split a streaming text buffer into complete sentences.

In streaming mode (flush=False), a sentence boundary is only recognised when there is real whitespace AFTER the punctuation mark. This avoids premature splits on trailing "Dr.", "3.", etc. whose continuation has not yet arrived.

When the stream is finished, call with flush=True so the last fragment is emitted even if it doesn't end with whitespace.

Returns (list_of_sentences, remaining_buffer).

Image Processing

vision.image_processing

is_screenshot_allowed(privacy_mode_enabled: bool | None = True, path: str | Path | None = None) -> tuple[bool, str]

Check whether a screenshot capture is allowed under governance rules.

Backward-compatible: delegates to _check_governance and adds the legacy path-sensitivity check.

When privacy_mode_enabled is None privacy state is unknown — allow by default (defensive, not blocking). False means the user has explicitly disabled screenshot capture via privacy settings.

run_image_description(vision_model, vision_context_length, can_speak_event, stop_event: Optional[asyncio.Event] = None, screenshot_path: str | Path | None = None, privacy_mode_enabled: bool = True, governor: ScreenshotGovernor | None = None) async

Capture a screenshot and generate a description.

Parameters

governor : optional ScreenshotGovernor When passed, enforces the full policy (scope, blocked windows) BEFORE any capture or file I/O. Overrides privacy_mode_enabled.

set_screenshot_governor(gov: ScreenshotGovernor) -> None

Wire the screenshot governor into this module (called once at startup).

set_vision_backend(backend) -> None

Set an optional ModelBackend for vision captioning operations.

write_sentence(sentence: str)

Append a sentence to the in-memory screenshot description.