The dictation service — the parts wired into one running thing.
Composes the four pieces of the push-to-talk stack:
hotkey (CGEventTap) → PTT state machine → mic capture →
transcribe (speech backend) → clean paste (clipboard preserved)
The composition — which action leads to capture/transcribe/paste — is a
pure method (_run_action) driven by injected callables, so the whole flow
is testable with fakes: no microphone, no tap, no transcription model. The
live wiring (start/stop) only attaches the real hotkey listener.
Classes
DictationService(
*,
transcribe: Transcribe,
paste: Paste,
hotkey: str = "control",
capture_factory: CaptureFactory = MicCapture,
clock: Callable[[], float] | None = None,
silence_rms: float = 0.15,
on_status: Callable[[str], None] | None = None,
history: bool = True,
model_name: str = "",
on_transcript: Callable[[str], None] | None = None,
on_action: Callable[[Action], None] | None = None,
)
Drive dictation from key events, independent of any window.
Source code in src/diapason/desktop/dictation_service.py
| def __init__(
self,
*,
transcribe: Transcribe,
paste: Paste,
hotkey: str = "control",
capture_factory: CaptureFactory = MicCapture,
clock: Callable[[], float] | None = None,
# Field observation (Carlito's MacBook mic): normal speech lands at
# level ~1.0 on the 0–100 RMS scale, while true silence sits well
# under 0.05. The earlier floor of 0.5 left only a 2× margin to real
# speech — a soft-spoken word could be discarded as silence. 0.15
# keeps ~7× margin to speech while still rejecting an untouched mic.
silence_rms: float = 0.15,
on_status: Callable[[str], None] | None = None,
history: bool = True,
model_name: str = "",
on_transcript: Callable[[str], None] | None = None,
on_action: Callable[[Action], None] | None = None,
) -> None:
# Observer for anything that wants the delivered text (menu bar, UI).
self._on_transcript = on_transcript
# Fires the instant an action is dispatched, BEFORE any device work.
# Opening the input stream costs tens of milliseconds, so feedback
# driven by on_status (which reports after the fact) would lag the
# keypress by exactly the interval the user is trying to confirm.
# Anything hooked here must return immediately.
self._on_action = on_action
self._history = history
self._model_name = model_name
self._transcribe = transcribe
self._paste = paste
self._hotkey = hotkey
self._capture_factory = capture_factory
# Status callback: every stage of a session reports here. Without it
# the pipeline is a black box — a mic without permission, a silent
# buffer, an empty transcript and a failed paste all look identical
# to the user: "nothing happened".
self._on_status = on_status
# Below this RMS (on the 0–100 scale) the buffer is treated as silence
# and never sent to the transcriber. Skipping silence avoids a model
# invocation — and, more importantly, the "you. thanks for watching"
# hallucinations Whisper emits on empty audio.
self._silence_rms = silence_rms
self._ptt = PushToTalk()
self._capture: Optional[MicCapture] = None
self._listener = None
if clock is None:
import time
clock = time.monotonic
self._clock = clock
|
Functions:
float_mono_to_wav(
samples: "Any", sample_rate: int = 16000
) -> bytes
Encode a float32 [-1, 1] mono array as 16-bit PCM WAV bytes (stdlib).
Source code in src/diapason/desktop/dictation_service.py
| def float_mono_to_wav(samples: "Any", sample_rate: int = 16_000) -> bytes: # noqa: F821
"""Encode a float32 [-1, 1] mono array as 16-bit PCM WAV bytes (stdlib)."""
import numpy as np
arr = np.asarray(samples, dtype="float32").reshape(-1)
clipped = np.clip(arr, -1.0, 1.0)
pcm = (clipped * 32767.0).astype("<i2").tobytes()
buf = io.BytesIO()
with wave.open(buf, "wb") as wav:
wav.setnchannels(1)
wav.setsampwidth(2)
wav.setframerate(sample_rate)
wav.writeframes(pcm)
return buf.getvalue()
|