xr_ai_voicegate#

Public surface of the xr-ai-voicegate package.

Classes#

`AudioSink`	Consumer-supplied return-audio writer.
`TTSLike`	Duck-typed text-to-speech client used for `say_stop_ack`.
`VoiceGateConfig`	Voice-gate behavior settings.
`VoiceGate`	Speech-input gating state machine.

Functions#

load_voice_gate_config(→ VoiceGateConfig)

Load + parse a voice_gate YAML file into a VoiceGateConfig.

Package Contents#

class xr_ai_voicegate.AudioSink#

Consumer-supplied return-audio writer.

async play_wav(pid: str, wav_bytes: bytes) → None#: Play WAV bytes for participant pid.

class xr_ai_voicegate.TTSLike#

Duck-typed text-to-speech client used for say_stop_ack.

async synthesize(text: str) → bytes#: Synthesize text and return WAV bytes.

class xr_ai_voicegate.VoiceGateConfig#

Voice-gate behavior settings.

magic_phrases: tuple[str, Ellipsis] = ()#

Sentence-boundary opt-in phrases.

A match is valid at transcript start or after ., ?, or !. An empty tuple dispatches every non-STOP STT transcript.

followup_grace_s: float = 5.0#

Seconds in which the next utterance may begin without another phrase.

The utterance may finish after the grace period.

listening_chime: bool = True#

Whether phrase matches may emit a listening chime.

The chime is available only when magic_phrases is non-empty and plays when the consumer calls VoiceGate.play_chime. It defaults to true; set listening_chime: false to disable it.

xr_ai_voicegate.load_voice_gate_config(path: pathlib.Path) → VoiceGateConfig#

Load + parse a voice_gate YAML file into a VoiceGateConfig.

Schema: a top-level mapping with keys magic_phrases (list[str] or bare str), listening_chime (bool), followup_grace_s (float). Missing file or empty file → returns the dataclass defaults (gate disabled / always-on). magic_phrases: null and trailing whitespace in phrases are normalized the same way the inline-block parser did.

class xr_ai_voicegate.VoiceGate( cfg: xr_ai_voicegate.VoiceGateConfig, *, audio_sink: xr_ai_voicegate.AudioSink, tts: xr_ai_voicegate.TTSLike, )#

Speech-input gating state machine.

Event ladder in feed — exactly one event fires per call, in this deterministic priority order:

STOP detected on raw text OR on the magic-phrase-stripped tail → on_stop(pid); closes the follow-up window.
Magic phrase matched AND query non-empty → on_query(pid, query, fresh_match=True); closes the follow-up window.
Follow-up window still open (and not STOP) → on_query(pid, raw_text, fresh_match=False); closes the follow-up window. fresh_match distinguishes this continuation from case 2 so consumers can suppress one-shot side effects (e.g. the listening chime) on the follow-up dispatch.
Magic phrase matched AND query empty → on_phrase_only(pid); opens the follow-up window.
Otherwise → on_drop(pid, raw_text); closes the window defensively.

When magic_phrases is empty the gate is in always-on mode: STOP still wins (interrupts must work without a phrase) and every other utterance dispatches straight to on_query with fresh_match=True. The follow-up / phrase-only / drop branches are inert in this mode — they only make sense once a phrase exists to gate against.

STOP is a closed imperative grammar: direct requests such as stop, stop it, stop talking, be quiet, and shut up may use a bounded set of conversational prefixes and punctuation. Negations, questions, reported speech, unconfigured arbitrary prefixes, and scoped or multi-action commands such as stop monitoring are ordinary queries, not global stops.

Handler exceptions are logged and swallowed so one bad handler does not kill the gate.

on_query(h: Callable[[str, str, bool], Awaitable[None]]) → None#: Register the handler for accepted user queries.

on_stop(h: Callable[[str], Awaitable[None]]) → None#: Register the handler for stop commands.

on_phrase_only(h: Callable[[str], Awaitable[None]]) → None#: Register the handler for a wake phrase without a query payload.

on_drop(h: Callable[[str, str], Awaitable[None]]) → None#: Register the handler for transcripts rejected by the gate.

on_participant_joined(h: Callable[[str], Awaitable[None]]) → None#: Register the handler for participant-joined notifications.

bind( *, on_query: Callable[[str, str, bool], Awaitable[None]], on_stop: Callable[[str], Awaitable[None]], on_phrase_only: Callable[[str], Awaitable[None]] | None = None, on_drop: Callable[[str, str], Awaitable[None]] | None = None, on_participant_joined: Callable[[str], Awaitable[None]] | None = None, ) → None#

Register every handler in one call.

Equivalent to calling the individual on_* setters; offered as a single bind point for external consumers that have all handlers in hand. on_query and on_stop are required because the gate is useless without them; the others are optional and default to no-op when unset.

async participant_joined(pid: str) → None#: Notify the registered handler that participant pid joined.

forget(pid: str) → None#: Discard all gate state for participant pid.

begin_utterance(pid: str) → None#: Remember when a follow-up starts before its transcript is ready.

property wake_ack_enabled: bool#: Whether configured wake phrases should produce a listening chime.

matches_magic_phrase(text: str) → bool#: Return whether text has a phrase at a sentence boundary.

could_match_magic_phrase(text: str) → bool#: Return whether the current partial sentence can become a match.

async feed(pid: str, text: str) → None#

Run one transcript through the event ladder; fires exactly one event.

Not re-entrant per-pid: the follow-up window (_followup_until) is read-then-mutated without locking, so two concurrent feed calls for the same pid can race the window state. Consumers must serialize calls per participant (e.g. a per-pid transcribing flag).

async play_chime(pid: str) → bool#

Emit the listening chime on the consumer’s audio sink.

No-op when the chime is disabled or has not been built yet (the chime is lazily synthesized to match the TTS sample rate the first time observe_tts_wav is called). Returns whether audio was emitted so callers can avoid suppressing a later fallback chime.

format_phrase_help() → str | None#: Return a sentence fragment telling the user how to address the agent given the configured phrases. None when no phrases are configured (the caller picks a generic greeting). The wording carries over from the original _greet implementation.

async say_stop_ack(pid: str) → None#: Synthesize a short canned stop acknowledgement and play it on the consumer’s audio sink. Also observes the WAV so the lazy chime build can pick up the sample rate from this path.

observe_tts_wav(wav_bytes: bytes) → None#: Build the chime at the TTS sample rate the first time a real TTS WAV passes through. No-op once built or when chime is disabled. A malformed WAV header disables the chime so the consumer doesn’t keep paying for failed builds.