From d1a811a5c57116d6b0f73fd7fd59ec542c597e00 Mon Sep 17 00:00:00 2001 From: JB Date: Wed, 26 Aug 2026 13:40:47 -0400 Subject: [PATCH] feat(dictate): add voice-reactive listening waveform --- .../listening-level-events-26AUG2026.md | 27 +++++++ docs/dictate/protocol.md | 32 ++++++--- docs/dictate/review-and-fix-plan.md | 2 + docs/todos.md | 17 +++++ extensions/dictate/README.md | 13 ++-- extensions/dictate/index.test.ts | 50 ++++++++++++- extensions/dictate/index.ts | 45 +++++++++++- .../dictate/src/pi_dictate_service/model.py | 7 +- .../src/pi_dictate_service/protocol.py | 4 +- .../src/pi_dictate_service/recorder.py | 30 +++++++- .../dictate/src/pi_dictate_service/server.py | 33 ++++++++- services/dictate/tests/test_model.py | 28 +++++--- services/dictate/tests/test_protocol_audio.py | 25 +++---- services/dictate/tests/test_recorder.py | 72 ++++++++++++++++++- services/dictate/tests/test_server_core.py | 43 +++++++++-- services/dictate/tests/test_unix_server.py | 28 ++++---- 16 files changed, 383 insertions(+), 73 deletions(-) create mode 100644 docs/adr/dictate/listening-level-events-26AUG2026.md diff --git a/docs/adr/dictate/listening-level-events-26AUG2026.md b/docs/adr/dictate/listening-level-events-26AUG2026.md new file mode 100644 index 0000000..77518d4 --- /dev/null +++ b/docs/adr/dictate/listening-level-events-26AUG2026.md @@ -0,0 +1,27 @@ +# Stream bounded audio levels for the dictate indicator + +Date: 26AUG2026 +Status: Accepted + +## Context + +Pi Dictate previously exposed only static service states. Active recording was easy to miss, and the extension had no signal from which to render a voice-reactive waveform. Raw microphone access must remain isolated in the host service. + +## Decision + +Protocol version 2 adds a `level` event while the service is listening. The recorder converts each 100 ms PCM window to a normalized logarithmic RMS level. The service sends the current scalar to the owning client at no more than 10 Hz. + +The extension renders a fixed-width footer status: + +```text +● listening ▁▂▄▇█▆▃▂▅▇▄▂ +``` + +The circle uses Pi's semantic error color. Twelve block cells show recent levels as a scrolling history. Loading, stopping, and transcription remain static text. + +## Consequences + +- The indicator reacts to speech without granting Pi microphone access or sending PCM over the socket. +- Output remains bounded by the existing per-connection queue and frame limits. +- The extension and installed service must upgrade together because protocol version 1 clients do not understand level events. +- Recording causes at most ten footer renders and protocol events per second. diff --git a/docs/dictate/protocol.md b/docs/dictate/protocol.md index d0206e1..1c9d087 100644 --- a/docs/dictate/protocol.md +++ b/docs/dictate/protocol.md @@ -2,7 +2,7 @@ ## Transport and version -Protocol version `1` uses UTF-8, newline-delimited JSON over one persistent Unix stream socket. Each line contains one JSON object. The systemd socket unit owns the socket path and permissions. The service accepts the inherited listening file descriptor and does not bind or unlink the path. +Protocol version `2` uses UTF-8, newline-delimited JSON over one persistent Unix stream socket. Each line contains one JSON object. The systemd socket unit owns the socket path and permissions. The service accepts the inherited listening file descriptor and does not bind or unlink the path. Clients must send `hello` before any other request. A connection negotiates one version only. Coding terms are host-owned service configuration and never cross this protocol. @@ -17,6 +17,7 @@ Clients must send `hello` before any other request. A connection negotiates one | Simultaneous client connections | 16 | | Pending output per client | 65,536 bytes | | Recorder stderr tail | 65,536 bytes | +| Audio-level event rate | 10 per second | | Mono 16 kHz signed-16 PCM | 9,600,000 bytes, five minutes | IDs are non-empty, opaque strings generated by the client. The service compares them only for request correlation, ownership, and stale-generation rejection. It does not interpret IDs as paths. Request objects reject unknown or missing fields. @@ -30,7 +31,7 @@ All examples are one line when transmitted. ### `hello` ```json -{"version":1,"type":"hello","requestId":"hello-1"} +{"version":2,"type":"hello","requestId":"hello-1"} ``` Exact fields: `version`, `type`, `requestId`. @@ -40,7 +41,7 @@ A successful response is `pong` with the same `requestId`. `ready` reports wheth ### `ping` ```json -{"version":1,"type":"ping","requestId":"ping-1"} +{"version":2,"type":"ping","requestId":"ping-1"} ``` Exact fields: `version`, `type`, `requestId`. `hello` is required first. The response is `pong` and may include `ready` and `busy`. @@ -48,7 +49,7 @@ Exact fields: `version`, `type`, `requestId`. `hello` is required first. The res ### `start` ```json -{"version":1,"type":"start","requestId":"start-1","generationId":"generation-1"} +{"version":2,"type":"start","requestId":"start-1","generationId":"generation-1"} ``` Exact fields: `version`, `type`, `requestId`, `generationId`. The service accepts only one global generation. It first emits `state: loading` for a cold model or `state: listening` for a warm model. Recording starts only after model loading succeeds. @@ -56,7 +57,7 @@ Exact fields: `version`, `type`, `requestId`, `generationId`. The service accept ### `stop` ```json -{"version":1,"type":"stop","requestId":"stop-1","generationId":"generation-1"} +{"version":2,"type":"stop","requestId":"stop-1","generationId":"generation-1"} ``` Exact fields: `version`, `type`, `requestId`, `generationId`. Only the owning connection may stop the active listening generation. @@ -64,7 +65,7 @@ Exact fields: `version`, `type`, `requestId`, `generationId`. Only the owning co ### `cancel` ```json -{"version":1,"type":"cancel","requestId":"cancel-1","generationId":"generation-1"} +{"version":2,"type":"cancel","requestId":"cancel-1","generationId":"generation-1"} ``` Exact fields: `version`, `type`, `requestId`, `generationId`. A valid cancellation immediately emits `cancelled`. Cancellation tombstones the generation, so it can never emit a later `result` or begin recording after a cancelled load. @@ -76,21 +77,29 @@ All service events contain `version`, `type`, and a bounded client-generated `re ### `pong` ```json -{"version":1,"type":"pong","requestId":"hello-1","service":"pi-dictate","ready":false} +{"version":2,"type":"pong","requestId":"hello-1","service":"pi-dictate","ready":false} ``` ### `state` ```json -{"version":1,"type":"state","requestId":"start-1","generationId":"generation-1","state":"loading"} +{"version":2,"type":"state","requestId":"start-1","generationId":"generation-1","state":"loading"} ``` `state` is one of `loading`, `listening`, `stopping`, `transcribing`, or `idle`. +### `level` + +```json +{"version":2,"type":"level","requestId":"start-1","generationId":"generation-1","level":0.625} +``` + +While listening, the service emits at most ten `level` events per second. `level` is a finite number from `0` through `1`, derived from the recorder's current 100 ms RMS window on a logarithmic -60 dBFS to 0 dBFS scale. The event carries no PCM samples and stops before recorder shutdown begins. + ### `result` ```json -{"version":1,"type":"result","requestId":"stop-1","generationId":"generation-1","transcript":"example text"} +{"version":2,"type":"result","requestId":"stop-1","generationId":"generation-1","transcript":"example text"} ``` There is at most one result for a generation. An empty transcript is a successful result with `"transcript":""`. @@ -98,7 +107,7 @@ There is at most one result for a generation. An empty transcript is a successfu ### `cancelled` ```json -{"version":1,"type":"cancelled","requestId":"cancel-1","generationId":"generation-1"} +{"version":2,"type":"cancelled","requestId":"cancel-1","generationId":"generation-1"} ``` This is a terminal delivery guarantee: the generation cannot later produce a result. @@ -106,7 +115,7 @@ This is a terminal delivery guarantee: the generation cannot later produce a res ### `error` ```json -{"version":1,"type":"error","requestId":"stop-1","generationId":"generation-1","code":"recorder_timeout","detail":"pw-record did not stop after SIGTERM and was killed"} +{"version":2,"type":"error","requestId":"stop-1","generationId":"generation-1","code":"recorder_timeout","detail":"pw-record did not stop after SIGTERM and was killed"} ``` Stable codes include: @@ -123,6 +132,7 @@ Stable codes include: | no generation | owner `start` | `loading`, then `listening`; warm models may emit `listening` directly | | any active generation | another `start` | `busy`; connection stays open | | `loading` | owner `cancel` | immediate `cancelled`; loading finishes in the worker but recording is suppressed | +| `listening` | recorder audio | rate-limited `level` events for the owning client | | `listening` | owner `stop` | `stopping`; signal recorder, drain stdout to EOF, await child close, convert PCM, then `transcribing` | | `listening` | owner `cancel` or disconnect | stop and drain recorder, discard audio, never infer | | `stopping` | owner `cancel` | immediate `cancelled`; completed audio is discarded | diff --git a/docs/dictate/review-and-fix-plan.md b/docs/dictate/review-and-fix-plan.md index b22a216..8bf58e0 100644 --- a/docs/dictate/review-and-fix-plan.md +++ b/docs/dictate/review-and-fix-plan.md @@ -186,6 +186,8 @@ The current meter updates every 60 ms even when its rendered value has not chang **Fix:** the extension renders state changes reported by the service and does not receive raw audio levels. Use static status text for `loading model...`, `listening...`, `stopping...`, and `transcribing...`. If a meter is wanted later, add a rate-limited level event only after measuring the need. +**Superseded 26AUG2026:** active-listening feedback now needs a voice-reactive waveform. [The accepted follow-up decision](../adr/dictate/listening-level-events-26AUG2026.md) adds a bounded 10 Hz scalar level event without exposing raw audio to Pi. + ### 12. Tests, installation, attribution, and user documentation are missing **Severity:** medium diff --git a/docs/todos.md b/docs/todos.md index 7e123d9..adb8a21 100644 --- a/docs/todos.md +++ b/docs/todos.md @@ -8,3 +8,20 @@ During Dictate Phase 0 on 26AUG2026, repeated overlapping throwaway benchmark pr 2. A later production-adapter check raised `ValueError: bad marshal data (invalid reference)` while importing `torch._tensor`. An immediate standalone `import torch` succeeded, and the cached `_tensor.cpython-312.pyc` passed `marshal.loads()`. The production service serializes model loading and inference and therefore does not intentionally launch concurrent NeMo processes. Before treating concurrent workers as supported, build a repeatable cold-load stress loop, monitor process and cgroup memory, disable bytecode writes, and determine whether the failures come from NeMo/OmegaConf resource pressure or concurrent access to the external environment's bytecode cache. + +## Repair nono self-diagnosis after terminal path drift + +During Dictate waveform testing on 26AUG2026, every Unix socket `bind()` in `services/dictate/tests/test_unix_server.py` failed with `EACCES` or `EPERM`, including when `TMPDIR` pointed inside the repository worktree. The required diagnostic also failed: + +```text +$ nono why --self --path /tmp --op readwrite +nono: Configuration parse error: sandbox state path drifted at reload: serialized resolved=/dev/pts/15, actual resolved=/dev/null +``` + +Reproduce in a non-interactive or reloaded Pi session, determine whether `/dev/pts/*` should be serialized by identity rather than resolved path, and make `nono why --self` report the active policy even after the controlling terminal changes. Also verify whether the Pi profile intentionally denies `AF_UNIX` socket creation in otherwise writable paths; the Dictate service loop tests need local throwaway sockets. + +## Reuse the verified Dictate model during reinstall + +The protocol-v2 reinstall on 26AUG2026 attempted to download the pinned model again even though the exact model already existed under `~/.local/share/pi-dictate/models/`. The sandbox correctly blocked unapproved access to `huggingface.co`, so installation had to be repeated with the installed model passed explicitly through `--model-source`. + +Teach `services/dictate/install.sh` to checksum an existing model first and keep it when it matches `MODEL_SHA256`. Download only when the model is absent or invalid, while preserving the current atomic temporary-file replacement for new model sources. diff --git a/extensions/dictate/README.md b/extensions/dictate/README.md index 164e3ca..ac2d809 100644 --- a/extensions/dictate/README.md +++ b/extensions/dictate/README.md @@ -61,12 +61,15 @@ The active profile is repository-linked on the tested host. Start a new Pi sandb Shortcuts are handled by Pi's main editor. They do not claim to work while a selector, confirmation, or custom modal owns input. If a modal opens after recording starts, a successful result still goes only to the main editor. -The footer shows static service states: +The footer shows service state. While recording, a red circle and twelve-cell scrolling waveform make active capture visible and react to the microphone level: -- `loading model...` -- `listening...` -- `stopping...` -- `transcribing...` +```text +● listening ▁▂▄▇█▆▃▂▅▇▄▂ +``` + +The circle uses Pi's semantic error color and the waveform uses its accent color. The other states remain `loading model...`, `stopping...`, and `transcribing...`. Audio levels are bounded scalar protocol events; raw audio remains inside the host service. + +This indicator requires protocol version 2. Run the installer again after updating the repository so the installed host service and repository-linked extension upgrade together. Only a non-empty finalized result for the current generation changes the editor. Empty speech, cancellation, stale results, disconnects, and failures do not insert text. diff --git a/extensions/dictate/index.test.ts b/extensions/dictate/index.test.ts index 7f04690..5f9be8c 100644 --- a/extensions/dictate/index.test.ts +++ b/extensions/dictate/index.test.ts @@ -77,10 +77,17 @@ function createHarness() { const sockets: FakeSocket[] = []; const socketPaths: string[] = []; const statuses: Array = []; + const themeCalls: Array<{ color: string; text: string }> = []; const notifications: Array<{ message: string; type?: string }> = []; const pasted: string[] = []; const context = { ui: { + theme: { + fg: (color: string, text: string) => { + themeCalls.push({ color, text }); + return text; + }, + }, setStatus: (_key: string, text: string | undefined) => statuses.push(text), notify: (message: string, type?: string) => notifications.push({ message, type }), @@ -134,6 +141,7 @@ function createHarness() { sockets, socketPaths, statuses, + themeCalls, notifications, pasted, context, @@ -213,7 +221,11 @@ test("rejects a mismatched protocol version before sending start", async () => { assert.ok(socket); socket.connect(); const hello = harness.sentFrames(socket)[0]; - socket.frame({ version: 2, type: "pong", requestId: hello?.requestId }); + socket.frame({ + version: PROTOCOL_VERSION + 1, + type: "pong", + requestId: hello?.requestId, + }); assert.equal(harness.sentFrames(socket).length, 1); assert.match(harness.notifications[0]?.message ?? "", /version mismatch/i); @@ -266,19 +278,51 @@ test("renders authoritative service states and sends a bounded stop command", as assert.deepEqual(harness.statuses, [ "loading model...", - "listening...", + "● listening ▁▁▁▁▁▁▁▁▁▁▁▁", "stopping...", "transcribing...", ]); }); +test("scrolls bounded audio levels through the listening waveform", async () => { + const harness = createHarness(); + const { socket, generationId } = await connectAndNegotiate(harness); + serviceFrame(socket, generationId, "state", { state: "listening" }); + assert.deepEqual(harness.themeCalls, [ + { color: "error", text: "●" }, + { color: "dim", text: "listening" }, + { color: "accent", text: "▁▁▁▁▁▁▁▁▁▁▁▁" }, + ]); + serviceFrame(socket, generationId, "level", { level: 0.5 }); + serviceFrame(socket, generationId, "level", { level: 1 }); + + assert.deepEqual(harness.statuses, [ + "● listening ▁▁▁▁▁▁▁▁▁▁▁▁", + "● listening ▁▁▁▁▁▁▁▁▁▁▁▅", + "● listening ▁▁▁▁▁▁▁▁▁▁▅█", + ]); +}); + +test("rejects audio levels outside listening or the normalized range", async () => { + const beforeListening = createHarness(); + const pending = await connectAndNegotiate(beforeListening); + serviceFrame(pending.socket, pending.generationId, "level", { level: 0.5 }); + assert.match(beforeListening.notifications[0]?.message ?? "", /audio level is invalid/i); + + const outOfRange = createHarness(); + const active = await connectAndNegotiate(outOfRange); + serviceFrame(active.socket, active.generationId, "state", { state: "listening" }); + serviceFrame(active.socket, active.generationId, "level", { level: 1.1 }); + assert.match(outOfRange.notifications[0]?.message ?? "", /audio level is invalid/i); +}); + test("ignores stale generations but rejects malformed and oversized frames", async () => { const staleHarness = createHarness(); const { socket: staleSocket, generationId } = await connectAndNegotiate(staleHarness); serviceFrame(staleSocket, "old-generation", "state", { state: "listening" }); assert.deepEqual(staleHarness.statuses, []); serviceFrame(staleSocket, generationId, "state", { state: "listening" }); - assert.deepEqual(staleHarness.statuses, ["listening..."]); + assert.deepEqual(staleHarness.statuses, ["● listening ▁▁▁▁▁▁▁▁▁▁▁▁"]); const malformedHarness = createHarness(); const malformed = await connectAndNegotiate(malformedHarness); diff --git a/extensions/dictate/index.ts b/extensions/dictate/index.ts index 1d51945..5e3ca30 100644 --- a/extensions/dictate/index.ts +++ b/extensions/dictate/index.ts @@ -2,7 +2,7 @@ import type { ExtensionAPI, ExtensionContext } from "@earendil-works/pi-coding-a import { createConnection } from "node:net"; export const DICTATE_SOCKET_PATH = "/run/user/1000/pi-dictate.sock"; -export const PROTOCOL_VERSION = 1; +export const PROTOCOL_VERSION = 2; export const MAX_FRAME_BYTES = 64 * 1024; export const MAX_TRANSCRIPT_BYTES = 32 * 1024; export const MAX_ID_BYTES = 128; @@ -10,6 +10,8 @@ export const MAX_ID_BYTES = 128; const STATUS_KEY = "dictate"; const MAX_ERROR_DETAIL_BYTES = 2 * 1024; const UTF8_DECODER = new TextDecoder("utf-8", { fatal: true }); +const WAVEFORM_BARS = ["▁", "▂", "▃", "▄", "▅", "▆", "▇", "█"] as const; +const WAVEFORM_SAMPLES = 12; type DictateState = | "idle" @@ -97,6 +99,7 @@ export function createDictateExtension(options: DictateExtensionOptions = {}) { let receiveBuffer = Buffer.alloc(0); let nextId = 0; let resultDelivered = false; + let audioLevels: number[] = []; const newId = (kind: string) => `${kind}-${++nextId}`; @@ -104,6 +107,25 @@ export function createDictateExtension(options: DictateExtensionOptions = {}) { activeContext?.ui.setStatus(STATUS_KEY, status); }; + const setListeningStatus = () => { + if (!activeContext) return; + const theme = activeContext.ui.theme; + const waveform = audioLevels + .map( + (level) => + WAVEFORM_BARS[ + Math.min( + WAVEFORM_BARS.length - 1, + Math.floor(level * WAVEFORM_BARS.length), + ) + ], + ) + .join(""); + setStatus( + `${theme.fg("error", "●")} ${theme.fg("dim", "listening")} ${theme.fg("accent", waveform)}`, + ); + }; + const closeSocket = (current: DictateSocket, force: boolean) => { try { if (force) current.destroy(); @@ -121,6 +143,7 @@ export function createDictateExtension(options: DictateExtensionOptions = {}) { negotiated = false; receiveBuffer = Buffer.alloc(0); resultDelivered = false; + audioLevels = []; state = "idle"; setStatus(undefined); activeContext = null; @@ -183,7 +206,8 @@ export function createDictateExtension(options: DictateExtensionOptions = {}) { } if (serviceState === "listening" && state === "connecting") { state = "recording"; - setStatus("listening..."); + audioLevels = Array(WAVEFORM_SAMPLES).fill(0); + setListeningStatus(); return true; } if (serviceState === "stopping" && state === "stopping") { @@ -267,6 +291,22 @@ export function createDictateExtension(options: DictateExtensionOptions = {}) { return; } + if (value.type === "level") { + if ( + state !== "recording" || + typeof value.level !== "number" || + !Number.isFinite(value.level) || + value.level < 0 || + value.level > 1 + ) { + protocolError("audio level is invalid for the current state."); + return; + } + audioLevels = [...audioLevels.slice(1), value.level]; + setListeningStatus(); + return; + } + if (value.type === "result") { if (state !== "transcribing" || typeof value.transcript !== "string") { protocolError("result is invalid for the current state."); @@ -349,6 +389,7 @@ export function createDictateExtension(options: DictateExtensionOptions = {}) { helloRequestId = newId("hello"); negotiated = false; resultDelivered = false; + audioLevels = []; receiveBuffer = Buffer.alloc(0); let current: DictateSocket; diff --git a/services/dictate/src/pi_dictate_service/model.py b/services/dictate/src/pi_dictate_service/model.py index 7f698ee..8390491 100644 --- a/services/dictate/src/pi_dictate_service/model.py +++ b/services/dictate/src/pi_dictate_service/model.py @@ -86,12 +86,13 @@ class NemoParakeetModel: os.environ["NEMO_OFFLINE"] = "1" import numpy - from nemo.collections.asr.models import ASRModel + import torch + from nemo.collections.asr.models import EncDecRNNTBPEModel from omegaconf import open_dict - model = ASRModel.restore_from( + model = EncDecRNNTBPEModel.restore_from( restore_path=str(self._model_path), - map_location="cuda", + map_location=torch.device("cuda"), ) model.to("cuda") model.eval() diff --git a/services/dictate/src/pi_dictate_service/protocol.py b/services/dictate/src/pi_dictate_service/protocol.py index 7ddbe83..c3a5101 100644 --- a/services/dictate/src/pi_dictate_service/protocol.py +++ b/services/dictate/src/pi_dictate_service/protocol.py @@ -1,4 +1,4 @@ -"""Bounded newline-delimited JSON protocol primitives for dictate protocol v1.""" +"""Bounded newline-delimited JSON protocol primitives for dictate protocol v2.""" from __future__ import annotations @@ -6,7 +6,7 @@ from dataclasses import dataclass import json from typing import Any, Mapping -PROTOCOL_VERSION = 1 +PROTOCOL_VERSION = 2 MAX_FRAME_BYTES = 64 * 1024 MAX_TRANSCRIPT_BYTES = 32 * 1024 MAX_ERROR_DETAIL_BYTES = 2 * 1024 diff --git a/services/dictate/src/pi_dictate_service/recorder.py b/services/dictate/src/pi_dictate_service/recorder.py index ce3402d..84fbb33 100644 --- a/services/dictate/src/pi_dictate_service/recorder.py +++ b/services/dictate/src/pi_dictate_service/recorder.py @@ -2,10 +2,13 @@ from __future__ import annotations +from array import array from dataclasses import dataclass +import math import os import signal import subprocess +import sys import threading import time from typing import BinaryIO, Callable, Protocol @@ -24,6 +27,7 @@ PW_RECORD_COMMAND = [ "-", ] MAX_STDERR_BYTES = 64 * 1024 +AUDIO_LEVEL_WINDOW_BYTES = 3_200 # 100 ms of mono 16 kHz signed-16 PCM class Process(Protocol): @@ -95,12 +99,18 @@ class PwRecorder: self._duration_limited = False self._duration_stop_at: float | None = None self._reported = False + self._audio_level = 0.0 @property def buffered_bytes(self) -> int: with self._lock: return len(self._pcm) + @property + def audio_level(self) -> float: + with self._lock: + return self._audio_level + def start(self) -> None: if self._process is not None: raise RecorderError("invalid_state", "recorder is already started") @@ -223,10 +233,15 @@ class PwRecorder: def _drain_stdout(self) -> None: process = self._require_process() + level_window = bytearray() while True: - chunk = process.stdout.read(64 * 1024) + chunk = process.stdout.read(AUDIO_LEVEL_WINDOW_BYTES) if not chunk: return + level_window.extend(chunk) + while len(level_window) >= AUDIO_LEVEL_WINDOW_BYTES: + self._update_audio_level(level_window[:AUDIO_LEVEL_WINDOW_BYTES]) + del level_window[:AUDIO_LEVEL_WINDOW_BYTES] with self._lock: if self._duration_limited: continue @@ -240,6 +255,19 @@ class PwRecorder: if process.poll() is None: self._signal_group(process, signal.SIGTERM) + def _update_audio_level(self, chunk: bytes) -> None: + sample_bytes = len(chunk) - (len(chunk) % 2) + if sample_bytes == 0: + return + samples = array("h") + samples.frombytes(chunk[:sample_bytes]) + if sys.byteorder != "little": + samples.byteswap() + rms = math.sqrt(sum(sample * sample for sample in samples) / len(samples)) / 32768.0 + level = 0.0 if rms == 0 else max(0.0, min(1.0, (20 * math.log10(rms) + 60) / 60)) + with self._lock: + self._audio_level = max(level, self._audio_level * 0.65) + def _drain_stderr(self) -> None: process = self._require_process() while True: diff --git a/services/dictate/src/pi_dictate_service/server.py b/services/dictate/src/pi_dictate_service/server.py index 0304a07..a8ecbc2 100644 --- a/services/dictate/src/pi_dictate_service/server.py +++ b/services/dictate/src/pi_dictate_service/server.py @@ -5,6 +5,7 @@ from __future__ import annotations from collections.abc import Sequence from concurrent.futures import Executor, Future, ThreadPoolExecutor from dataclasses import dataclass +import math import os from queue import Empty, SimpleQueue import selectors @@ -30,9 +31,13 @@ from .recorder import Capture, PwRecorder, RecorderError DEFAULT_IDLE_SECONDS = 15 * 60 MAX_CLIENT_CONNECTIONS = 16 MAX_CONNECTION_OUTPUT_BYTES = MAX_FRAME_BYTES +LEVEL_EVENT_INTERVAL_SECONDS = 0.1 class Recorder(Protocol): + @property + def audio_level(self) -> float: ... + def start(self) -> None: ... def stop(self) -> Capture: ... def cancel(self) -> None: ... @@ -53,6 +58,7 @@ class _Generation: state: str recorder: Recorder | None = None tombstoned: bool = False + last_level_event_at: float | None = None @dataclass(frozen=True) @@ -156,6 +162,8 @@ class DictateServiceCore: if failure is not None: self._async_error(active, failure.code, failure.detail) self._finish_generation(active, notify_idle=True) + return + self._emit_audio_level(active) def drain_events(self) -> list[Outbound]: events, self._events = self._events, [] @@ -216,8 +224,31 @@ class DictateServiceCore: self._finish_generation(active, notify_idle=True) return active.recorder = recorder + active.last_level_event_at = self._clock() self._state(active, "listening") + def _emit_audio_level(self, active: _Generation) -> None: + now = self._clock() + if ( + active.last_level_event_at is not None + and now - active.last_level_event_at < LEVEL_EVENT_INTERVAL_SECONDS + ): + return + level = active.recorder.audio_level if active.recorder is not None else 0.0 + if not math.isfinite(level): + level = 0.0 + active.last_level_event_at = now + self._emit( + active.owner, + { + "version": PROTOCOL_VERSION, + "type": "level", + "requestId": active.request_id, + "generationId": active.generation_id, + "level": round(max(0.0, min(1.0, level)), 3), + }, + ) + def _stop(self, client_id: str, request: Request) -> None: active = self._owned_generation(client_id, request) if active is None: @@ -433,7 +464,7 @@ def inherited_listener(environ: dict[str, str] | None = None) -> socket.socket: def serve(listener: socket.socket, core: DictateServiceCore) -> None: - """Serve protocol v1 on an explicit listener until the core becomes idle.""" + """Serve protocol v2 on an explicit listener until the core becomes idle.""" selector = selectors.DefaultSelector() connections: dict[str, _Connection] = {} sequence = 0 diff --git a/services/dictate/tests/test_model.py b/services/dictate/tests/test_model.py index 945fd48..48f70a5 100644 --- a/services/dictate/tests/test_model.py +++ b/services/dictate/tests/test_model.py @@ -35,13 +35,15 @@ class FakeNemoModel: return [type("Hypothesis", (), {"text": "hello locally"})()] -class FakeASRModel: +class FakeConcreteParakeetModel: restored_path: str | None = None + restored_location: object | None = None restored: FakeNemoModel | None = None @classmethod - def restore_from(cls, restore_path: str, map_location: str) -> FakeNemoModel: + def restore_from(cls, restore_path: str, map_location: object) -> FakeNemoModel: cls.restored_path = restore_path + cls.restored_location = map_location self = FakeNemoModel() cls.restored = self return self @@ -69,28 +71,29 @@ class NemoParakeetModelTests(unittest.TestCase): boost_alpha=0.5, expected_sha256=hashlib.sha256(b"model").hexdigest(), ) - self.assertIsNone(FakeASRModel.restored_path) + self.assertIsNone(FakeConcreteParakeetModel.restored_path) adapter.load() transcript = adapter.transcribe([0.0, 0.5]) self.assertEqual(os.environ["HF_HUB_OFFLINE"], "1") self.assertEqual(os.environ["TRANSFORMERS_OFFLINE"], "1") - self.assertEqual(FakeASRModel.restored_path, str(model_path)) - self.assertTrue(FakeASRModel.restored.eval_called) - self.assertEqual(FakeASRModel.restored.to_device, "cuda") + self.assertEqual(FakeConcreteParakeetModel.restored_path, str(model_path)) + self.assertEqual(FakeConcreteParakeetModel.restored_location, "device:cuda") + self.assertTrue(FakeConcreteParakeetModel.restored.eval_called) + self.assertEqual(FakeConcreteParakeetModel.restored.to_device, "cuda") self.assertEqual(transcript, "hello locally") - self.assertEqual(FakeASRModel.restored.audio, [([0.0, 0.5], "float32")]) + self.assertEqual(FakeConcreteParakeetModel.restored.audio, [([0.0, 0.5], "float32")]) self.assertEqual( - FakeASRModel.restored.decoding_config["greedy"]["boosting_tree"]["key_phrases_list"], + FakeConcreteParakeetModel.restored.decoding_config["greedy"]["boosting_tree"]["key_phrases_list"], ["JSON", "npm test"], ) self.assertEqual( - FakeASRModel.restored.decoding_config["greedy"]["boosting_tree_alpha"], + FakeConcreteParakeetModel.restored.decoding_config["greedy"]["boosting_tree_alpha"], 0.5, ) self.assertEqual( - FakeASRModel.restored.transcribe_options, + FakeConcreteParakeetModel.restored.transcribe_options, {"batch_size": 1, "num_workers": 0, "verbose": False}, ) @@ -123,12 +126,14 @@ class NemoParakeetModelTests(unittest.TestCase): collections = ModuleType("nemo.collections") asr = ModuleType("nemo.collections.asr") models = ModuleType("nemo.collections.asr.models") - models.ASRModel = FakeASRModel + models.EncDecRNNTBPEModel = FakeConcreteParakeetModel asr.models = models collections.asr = asr nemo.collections = collections omegaconf = ModuleType("omegaconf") omegaconf.open_dict = lambda _config: nullcontext() + torch = ModuleType("torch") + torch.device = lambda name: f"device:{name}" return { "nemo": nemo, "nemo.collections": collections, @@ -136,6 +141,7 @@ class NemoParakeetModelTests(unittest.TestCase): "nemo.collections.asr.models": models, "numpy": FakeNumpy("numpy"), "omegaconf": omegaconf, + "torch": torch, } diff --git a/services/dictate/tests/test_protocol_audio.py b/services/dictate/tests/test_protocol_audio.py index 1b2cf3f..89fe6cd 100644 --- a/services/dictate/tests/test_protocol_audio.py +++ b/services/dictate/tests/test_protocol_audio.py @@ -7,6 +7,7 @@ from pi_dictate_service.protocol import ( MAX_FRAME_BYTES, MAX_ID_BYTES, MAX_TRANSCRIPT_BYTES, + PROTOCOL_VERSION, FrameDecoder, ProtocolError, encode_frame, @@ -18,10 +19,10 @@ class FrameDecoderTests(unittest.TestCase): def test_decodes_fragmented_and_multiple_frames(self) -> None: decoder = FrameDecoder() - self.assertEqual(decoder.feed(b'{"version":1'), []) + self.assertEqual(decoder.feed(f'{{"version":{PROTOCOL_VERSION}'.encode()), []) frames = decoder.feed( b',"type":"ping","requestId":"one"}\n' - b'{"version":1,"type":"ping","requestId":"two"}\n' + + f'{{"version":{PROTOCOL_VERSION},"type":"ping","requestId":"two"}}\n'.encode() ) self.assertEqual([frame["requestId"] for frame in frames], ["one", "two"]) @@ -41,14 +42,14 @@ class FrameDecoderTests(unittest.TestCase): class RequestValidationTests(unittest.TestCase): def test_requires_exact_fields_bounded_ids_and_supported_version(self) -> None: request = parse_request( - {"version": 1, "type": "start", "requestId": "r1", "generationId": "g1"} + {"version": PROTOCOL_VERSION, "type": "start", "requestId": "r1", "generationId": "g1"} ) self.assertEqual(request.generation_id, "g1") with self.assertRaises(ProtocolError) as unknown: parse_request( { - "version": 1, + "version": PROTOCOL_VERSION, "type": "start", "requestId": "r1", "generationId": "g1", @@ -61,7 +62,7 @@ class RequestValidationTests(unittest.TestCase): with self.assertRaises(ProtocolError) as long_id: parse_request( { - "version": 1, + "version": PROTOCOL_VERSION, "type": "ping", "requestId": "é" * ((MAX_ID_BYTES // 2) + 1), } @@ -69,7 +70,7 @@ class RequestValidationTests(unittest.TestCase): self.assertEqual(long_id.exception.code, "invalid_request") with self.assertRaises(ProtocolError) as unsupported: - parse_request({"version": 2, "type": "hello", "requestId": "r2"}) + parse_request({"version": PROTOCOL_VERSION + 1, "type": "hello", "requestId": "r2"}) self.assertEqual(unsupported.exception.code, "unsupported_version") self.assertTrue(unsupported.exception.close_connection) @@ -78,12 +79,12 @@ class RequestValidationTests(unittest.TestCase): with self.subTest(request_type=request_type): with self.assertRaises(ProtocolError) as failed: parse_request( - {"version": 1, "type": request_type, "requestId": "request"} + {"version": PROTOCOL_VERSION, "type": request_type, "requestId": "request"} ) self.assertEqual(failed.exception.code, "invalid_request") with self.assertRaises(ProtocolError) as surrogate: - parse_request({"version": 1, "type": "ping", "requestId": "\ud800"}) + parse_request({"version": PROTOCOL_VERSION, "type": "ping", "requestId": "\ud800"}) self.assertEqual(surrogate.exception.request_id, "") huge_integer = b'{"version":' + (b"9" * 5000) + b',"type":"hello","requestId":"r"}\n' @@ -96,13 +97,13 @@ class RequestValidationTests(unittest.TestCase): oversized = "x" * (MAX_ID_BYTES + 1) with self.assertRaises(ProtocolError) as unsupported: parse_request( - {"version": 2, "type": "hello", "requestId": oversized} + {"version": PROTOCOL_VERSION + 1, "type": "hello", "requestId": oversized} ) self.assertEqual(unsupported.exception.request_id, "") encoded = encode_frame( { - "version": 1, + "version": PROTOCOL_VERSION, "type": "error", "requestId": oversized, "generationId": "\ud800", @@ -118,7 +119,7 @@ class RequestValidationTests(unittest.TestCase): with self.assertRaises(ProtocolError) as transcript: encode_frame( { - "version": 1, + "version": PROTOCOL_VERSION, "type": "result", "generationId": "g1", "transcript": "é" * ((MAX_TRANSCRIPT_BYTES // 2) + 1), @@ -128,7 +129,7 @@ class RequestValidationTests(unittest.TestCase): encoded = encode_frame( { - "version": 1, + "version": PROTOCOL_VERSION, "type": "error", "requestId": "r1", "code": "failed", diff --git a/services/dictate/tests/test_recorder.py b/services/dictate/tests/test_recorder.py index cafc85f..077b6a9 100644 --- a/services/dictate/tests/test_recorder.py +++ b/services/dictate/tests/test_recorder.py @@ -1,19 +1,34 @@ import io +import math import subprocess import threading import unittest -from pi_dictate_service.recorder import PW_RECORD_COMMAND, PwRecorder, RecorderError +from pi_dictate_service.recorder import ( + AUDIO_LEVEL_WINDOW_BYTES, + PW_RECORD_COMMAND, + PwRecorder, + RecorderError, +) class TrackingStream(io.BytesIO): - def __init__(self, value: bytes, events: list[str], name: str) -> None: + def __init__( + self, + value: bytes, + events: list[str], + name: str, + max_read_size: int | None = None, + ) -> None: super().__init__(value) self._events = events self._name = name + self._max_read_size = max_read_size self.eof = threading.Event() def read(self, size: int = -1) -> bytes: + if self._max_read_size is not None: + size = min(size, self._max_read_size) value = super().read(size) if not value and not self.eof.is_set(): self._events.append(f"{self._name}-eof") @@ -91,10 +106,11 @@ class PwRecorderTests(unittest.TestCase): stderr_limit: int = 64 * 1024, clock: Clock | None = None, terminate_returncode: int = -15, + stdout_read_size: int | None = None, ) -> tuple[PwRecorder, FakeProcess, ProcessFactory, list[str]]: events: list[str] = [] process = FakeProcess( - TrackingStream(pcm, events, "stdout"), + TrackingStream(pcm, events, "stdout", stdout_read_size), TrackingStream(stderr, events, "stderr"), returncode=returncode, timeout=timeout, @@ -138,6 +154,56 @@ class PwRecorderTests(unittest.TestCase): self.assertIn("stdout-eof", events) self.assertIn("wait", events) + def test_tracks_normalized_audio_level_from_pcm_without_exposing_audio(self) -> None: + recorder, process, _factory, _events = self.make_recorder( + pcm=b"\xff\x7f" * (AUDIO_LEVEL_WINDOW_BYTES // 2), + max_bytes=AUDIO_LEVEL_WINDOW_BYTES, + ) + recorder.start() + self.assertTrue(process.stdout.eof.wait(1)) + + self.assertGreater(recorder.audio_level, 0.99) + self.assertLessEqual(recorder.audio_level, 1.0) + recorder.cancel() + + def test_audio_level_accumulates_fragmented_reads_into_100ms_windows(self) -> None: + loud_sample = (32767).to_bytes(2, "little", signed=True) + pcm = ( + loud_sample * (AUDIO_LEVEL_WINDOW_BYTES // 4) + + b"\x00" * (AUDIO_LEVEL_WINDOW_BYTES // 2) + + loud_sample * 400 + ) + recorder, process, _factory, _events = self.make_recorder( + pcm=pcm, + max_bytes=len(pcm), + stdout_read_size=800, + ) + + recorder.start() + self.assertTrue(process.stdout.eof.wait(1)) + + expected_rms = (0.5**0.5) * 32767 / 32768 + expected_level = (20 * math.log10(expected_rms) + 60) / 60 + self.assertAlmostEqual(recorder.audio_level, expected_level, places=3) + recorder.cancel() + + def test_audio_level_maps_silence_known_rms_decay_and_partial_samples(self) -> None: + recorder = PwRecorder() + + recorder._update_audio_level(b"\x00\x00" * 16) + self.assertEqual(recorder.audio_level, 0.0) + + amplitude = 3277 # approximately -20 dBFS + recorder._update_audio_level(amplitude.to_bytes(2, "little", signed=True) * 16) + self.assertAlmostEqual(recorder.audio_level, 2 / 3, places=3) + + recorder._update_audio_level(b"\x00\x00" * 16) + self.assertAlmostEqual(recorder.audio_level, (2 / 3) * 0.65, places=3) + + previous = recorder.audio_level + recorder._update_audio_level(b"\xff") + self.assertEqual(recorder.audio_level, previous) + def test_accepts_pw_record_code_one_after_intentional_sigterm(self) -> None: recorder, process, _factory, _events = self.make_recorder(terminate_returncode=1) recorder.start() diff --git a/services/dictate/tests/test_server_core.py b/services/dictate/tests/test_server_core.py index bd18fe9..702bc5c 100644 --- a/services/dictate/tests/test_server_core.py +++ b/services/dictate/tests/test_server_core.py @@ -3,7 +3,7 @@ from collections.abc import Sequence from dataclasses import dataclass import unittest -from pi_dictate_service.protocol import MAX_TRANSCRIPT_BYTES, parse_request +from pi_dictate_service.protocol import MAX_TRANSCRIPT_BYTES, PROTOCOL_VERSION, parse_request from pi_dictate_service.recorder import Capture, RecorderError from pi_dictate_service.server import DictateServiceCore, Outbound @@ -73,6 +73,7 @@ class FakeRecorder: self.started = False self.cancelled = False self.capture = Capture(b"\x00\x00\x00\x40", -15, 15, b"") + self.audio_level = 0.0 self.stop_error: RecorderError | None = None self.unexpected: RecorderError | None = None @@ -124,7 +125,7 @@ class CoreTestCase(unittest.TestCase): generation_id: str | None = None, ) -> list[dict[str, object]]: frame: dict[str, object] = { - "version": 1, + "version": PROTOCOL_VERSION, "type": request_type, "requestId": request_id, } @@ -217,6 +218,38 @@ class CancellationTests(CoreTestCase): self.assertEqual(self.core.active_generation_id, None) +class AudioLevelTests(CoreTestCase): + def test_emits_owned_normalized_audio_levels_at_no_more_than_ten_hertz(self) -> None: + recorder = self.load_to_listening() + recorder.audio_level = 0.625 + + events: list[Outbound] = [] + for _ in range(40): + self.clock.advance(0.025) + self.core.poll() + events.extend(self.core.drain_events()) + + self.assertGreaterEqual(len(events), 9) + self.assertLessEqual(len(events), 10) + self.assertEqual({event.client_id for event in events}, {"a"}) + self.assertEqual({event.frame["type"] for event in events}, {"level"}) + self.assertEqual({event.frame["level"] for event in events}, {0.625}) + + def test_clamps_invalid_levels_and_stops_emitting_after_listening(self) -> None: + recorder = self.load_to_listening() + for raw_level, expected in ((2.0, 1.0), (-1.0, 0.0), (float("nan"), 0.0)): + recorder.audio_level = raw_level + self.clock.advance(0.1) + self.core.poll() + level = self.frames(self.core.drain_events(), "a") + self.assertEqual(level[0]["level"], expected) + + self.request("a", "stop", "stop", "g1") + self.clock.advance(0.2) + self.core.poll() + self.assertEqual(self.core.drain_events(), []) + + class StopAndInferenceTests(CoreTestCase): def test_stop_drains_then_converts_before_inference_and_emits_result(self) -> None: self.load_to_listening() @@ -271,11 +304,11 @@ class StopAndInferenceTests(CoreTestCase): executor=self.executor, clock=self.clock, ) - core.handle_request("a", parse_request({"version": 1, "type": "hello", "requestId": "h"})) + core.handle_request("a", parse_request({"version": PROTOCOL_VERSION, "type": "hello", "requestId": "h"})) core.drain_events() - core.handle_request("a", parse_request({"version": 1, "type": "start", "requestId": "s", "generationId": "g"})) + core.handle_request("a", parse_request({"version": PROTOCOL_VERSION, "type": "start", "requestId": "s", "generationId": "g"})) core.drain_events() - core.handle_request("a", parse_request({"version": 1, "type": "stop", "requestId": "t", "generationId": "g"})) + core.handle_request("a", parse_request({"version": PROTOCOL_VERSION, "type": "stop", "requestId": "t", "generationId": "g"})) self.executor.run_next() core.poll() core.drain_events() diff --git a/services/dictate/tests/test_unix_server.py b/services/dictate/tests/test_unix_server.py index 79ce8d4..66173f4 100644 --- a/services/dictate/tests/test_unix_server.py +++ b/services/dictate/tests/test_unix_server.py @@ -73,24 +73,24 @@ class UnixLoopTests(unittest.TestCase): def test_accepts_fragmented_requests_and_writes_persistent_responses(self) -> None: with self.connect() as client: - client.sendall(b'{"version":1,"type":"hello",') + client.sendall(b'{"version":2,"type":"hello",') client.sendall(b'"requestId":"hello"}\n') self.assertEqual(self.receive_frame(client)["type"], "pong") - client.sendall(b'{"version":1,"type":"ping","requestId":"ping"}\n') + client.sendall(b'{"version":2,"type":"ping","requestId":"ping"}\n') pong = self.receive_frame(client) self.assertEqual(pong["requestId"], "ping") self.assertEqual(pong["service"], "pi-dictate") def test_valid_state_error_keeps_open_but_malformed_frame_errors_once_then_closes(self) -> None: with self.connect() as client: - client.sendall(b'{"version":1,"type":"hello","requestId":"hello"}\n') + client.sendall(b'{"version":2,"type":"hello","requestId":"hello"}\n') self.receive_frame(client) client.sendall( - b'{"version":1,"type":"stop","requestId":"stop","generationId":"g"}\n' + b'{"version":2,"type":"stop","requestId":"stop","generationId":"g"}\n' ) self.assertEqual(self.receive_frame(client)["code"], "invalid_state") - client.sendall(b'{"version":1,"type":"ping","requestId":"ping"}\n') + client.sendall(b'{"version":2,"type":"ping","requestId":"ping"}\n') self.assertEqual(self.receive_frame(client)["type"], "pong") client.sendall(b'{broken}\n') @@ -101,9 +101,9 @@ class UnixLoopTests(unittest.TestCase): def test_malicious_json_values_do_not_crash_the_service(self) -> None: with self.connect() as client: - client.sendall(b'{"version":1,"type":[],"requestId":"bad-type"}\n') + client.sendall(b'{"version":2,"type":[],"requestId":"bad-type"}\n') self.assertEqual(self.receive_frame(client)["code"], "invalid_request") - client.sendall(b'{"version":1,"type":"ping","requestId":"\\ud800"}\n') + client.sendall(b'{"version":2,"type":"ping","requestId":"\\ud800"}\n') self.assertEqual(self.receive_frame(client)["code"], "invalid_request") with self.connect() as client: @@ -115,14 +115,14 @@ class UnixLoopTests(unittest.TestCase): self.assertEqual(client.recv(1), b"") with self.connect() as client: - client.sendall(b'{"version":1,"type":"hello","requestId":"healthy"}\n') + client.sendall(b'{"version":2,"type":"hello","requestId":"healthy"}\n') self.assertEqual(self.receive_frame(client)["type"], "pong") def test_closes_non_reading_client_before_output_queue_can_grow_unbounded(self) -> None: with self.connect() as client: - client.sendall(b'{"version":1,"type":"hello","requestId":"hello"}\n') + client.sendall(b'{"version":2,"type":"hello","requestId":"hello"}\n') self.receive_frame(client) - ping = b'{"version":1,"type":"ping","requestId":"p"}\n' + ping = b'{"version":2,"type":"ping","requestId":"p"}\n' try: client.sendall(ping * 1200) except BrokenPipeError: @@ -139,7 +139,7 @@ class UnixLoopTests(unittest.TestCase): self.assertLessEqual(len(received), 64 * 1024) with self.connect() as client: - client.sendall(b'{"version":1,"type":"hello","requestId":"healthy"}\n') + client.sendall(b'{"version":2,"type":"hello","requestId":"healthy"}\n') self.assertEqual(self.receive_frame(client)["type"], "pong") def test_rejects_connections_above_the_fixed_limit(self) -> None: @@ -148,7 +148,7 @@ class UnixLoopTests(unittest.TestCase): time.sleep(0.1) overflow = self.connect() try: - overflow.sendall(b'{"version":1,"type":"hello","requestId":"overflow"}\n') + overflow.sendall(b'{"version":2,"type":"hello","requestId":"overflow"}\n') with self.assertRaises((ConnectionResetError, BrokenPipeError)): if overflow.recv(1) == b"": raise ConnectionResetError @@ -161,12 +161,12 @@ class UnixLoopTests(unittest.TestCase): def test_client_reset_does_not_crash_the_service(self) -> None: reset_client = self.connect() reset_client.setsockopt(socket.SOL_SOCKET, socket.SO_LINGER, struct.pack("ii", 1, 0)) - reset_client.sendall(b'{"version":1,"type":"hello","requestId":"abandoned"}\n') + reset_client.sendall(b'{"version":2,"type":"hello","requestId":"abandoned"}\n') reset_client.close() time.sleep(0.15) with self.connect() as client: - client.sendall(b'{"version":1,"type":"hello","requestId":"hello"}\n') + client.sendall(b'{"version":2,"type":"hello","requestId":"hello"}\n') self.assertEqual(self.receive_frame(client)["type"], "pong")