Compare commits

...
3 Commits
Author SHA1 Message Date
Iulian ecd686073e wip: incremental ASR outputs in realtime, few bugs 2026-09-10 22:00:20 +03:00
Iulian 65855ace6c wip: incremental ASR 2026-09-09 20:31:03 +03:00
Iulian 0b44d7e1bf feat: live macOS dictation — push-to-talk mic into the realtime ASR session
- /dictate + ctrl+shift+d toggle one shared session: press to start, press again
  to stop, flush and paste into the prompt (a terminal gives no key-release event
  for hold-to-talk without the Kitty keyboard protocol)
- client.recordMic(): spawns ffmpeg AVFoundation capture and appends s16le frames
  as they arrive (no pacer — the mic is already 1x paced); stop() = commit + the
  same quiescence drain transcribe() uses, since is_last_result never flips true
- enumerates mics by slicing the "AVFoundation audio devices:" block off stderr
  (ffmpeg prints the table there, then fails opening the dummy input)
- --device takes an index or name, --lang / ASR_DICTATE_LANGUAGE, --submit sends
  instead of pasting; asr_health now reports the mic inputs it enumerated
- capture death surfaces via the emptyGraceMs path with the TCC hint (System
  Settings → Privacy & Security → Microphone) when ffmpeg exits non-zero at 0 bytes
- env.sh sources the gitignored .env so a shell gets ASR_NIM_URL / ASR_API_KEY
2026-09-09 19:06:06 +03:00
4 changed files with 385 additions and 4 deletions

No files matched your search

+1
View File
@@ -6,5 +6,6 @@ yarn-error.log
*.mp3
*.mp4
asr.json
.env
.DS_Store
*.log
Executable
+8
View File
@@ -0,0 +1,8 @@
#!/usr/bin/env bash
# Load pi-asr local env (secrets live in the gitignored .env).
# Run from the repo root: source ./env.sh
[ -f ./.env ] && { set -a; . ./.env; set +a; } || { echo "[pi-asr] no ./.env found (run this from the repo root)"; return 1 2>/dev/null || exit 1; }
_k="${ASR_API_KEY:-}"
[ -n "$_k" ] && _m="set (${#_k} chars, ${_k:0:5}...)" || _m="UNSET"
echo "[pi-asr] ASR_NIM_URL=${ASR_NIM_URL:-<unset>} key=${_m}"
unset _k _m
+133 -4
View File
@@ -19,8 +19,8 @@ import type { AgentToolResult, ExtensionAPI, ExtensionContext } from "@mariozech
import { Text } from "@mariozechner/pi-tui";
import { Type } from "@sinclair/typebox";
import {
DEFAULT_BASE_URL, RealtimeAsrClient, hasFfmpeg, loadPcm16,
type TranscriptResult, type WordInfo,
DEFAULT_BASE_URL, RealtimeAsrClient, hasFfmpeg, loadPcm16, listAudioInputs,
type MicRecording, type TranscriptResult, type WordInfo,
} from "../src/client.js";
interface TranscribeParams {
@@ -205,11 +205,12 @@ export default function asrExtension(pi: ExtensionAPI) {
async execute(_id, _params, _signal, _onUpdate, ctx): Promise<AgentToolResult<any>> {
const client = new RealtimeAsrClient();
const [live, ready, minted, ffmpeg] = await Promise.all([
const [live, ready, minted, ffmpeg, inputs] = await Promise.all([
client.health("live", 10000).catch((e: any) => ({ error: e?.message ?? String(e) })),
client.health("ready", 10000).catch((e: any) => ({ error: e?.message ?? String(e) })),
client.mint(10000).catch((e: any) => ({ error: e?.message ?? String(e) })),
hasFfmpeg(),
listAudioInputs().catch(() => [] as { index: number; name: string }[]),
]);
const ok = live?.status === "live" && ready?.status === "ready" && !minted?.error;
const key = client.cfg.apiKey;
@@ -229,10 +230,11 @@ export default function asrExtension(pi: ExtensionAPI) {
` ws token single-use, expires in ${secs}s (mint immediately before each connect)`,
].join("\n"),
`ffmpeg ${ffmpeg ? "available (any container/codec accepted)" : "MISSING — only 16kHz mono wav files will work"}`,
`mic inputs ${inputs.length ? inputs.map(d => `[${d.index}] ${d.name}`).join(", ") : ffmpeg ? "none enumerated (check Mic permission / AVFoundation build)" : "n/a (ffmpeg missing)"}`,
`note the realtime decoder runs at ~1x realtime; a 10min file takes ~10min (speed=2 halves it)`,
].join("\n");
if (!ok) throw new Error(`ASR server is not usable:\n${text}`);
return { content: [{ type: "text", text }], details: { ok, live, ready, minted, ffmpeg } };
return { content: [{ type: "text", text }], details: { ok, live, ready, minted, ffmpeg, inputs } };
},
renderResult(result: any, _options: any, theme: any) {
@@ -300,4 +302,131 @@ export default function asrExtension(pi: ExtensionAPI) {
}
},
});
/* ------------------------------------------------------------- /dictate -- *
* macOS live dictation: ffmpeg AVFoundation mic capture -> the same realtime WS client.
* Toggle semantics (press to start, press again to stop + flush) because a plain terminal
* gives no key-release event for hold-to-talk without the Kitty keyboard protocol.
* Shared module state so the /dictate command and the ctrl+shift+d shortcut drive one session.
*/
let active: { rec: MicRecording; submit: boolean; ourTail: string } | null = null;
let starting = false;
/**
* Append-as-you-go composer. Dictation owns ONLY a suffix at the very end of the editor (the words
* heard so far). Each refresh strips exactly the suffix we appended last time from the *current*
* text and re-appends the grown tail — so we never overwrite the user's draft and never revive text
* they deleted: we derive from the live editor, not a start-of-session snapshot.
*/
function paintTail(ctx: ExtensionContext, prevOurs: string, tail: string): string {
const cur = ctx.ui.getEditorText() ?? "";
let prefix = cur;
if (prevOurs) {
if (cur.endsWith(prevOurs)) prefix = cur.slice(0, cur.length - prevOurs.length);
else { const i = cur.lastIndexOf(prevOurs); if (i >= 0) prefix = cur.slice(0, i) + cur.slice(i + prevOurs.length); }
}
const t = tail.trim();
const ours = t ? (prefix === "" || /\s$/.test(prefix) ? "" : " ") + t : "";
ctx.ui.setEditorText(prefix + ours);
return ours;
}
/** Drop the suffix we own, leaving the user's own text untouched (cancel / no-speech / submit). */
function stripTail(ctx: ExtensionContext, ours: string): void {
if (!ours) return;
const cur = ctx.ui.getEditorText() ?? "";
if (cur.endsWith(ours)) ctx.ui.setEditorText(cur.slice(0, cur.length - ours.length));
else { const i = cur.lastIndexOf(ours); if (i >= 0) ctx.ui.setEditorText(cur.slice(0, i) + cur.slice(i + ours.length)); }
}
async function resolveDevice(d?: string | null): Promise<string | null> {
if (!d) return null;
if (/^\d+$/.test(d)) {
const hit = (await listAudioInputs()).find(x => x.index === +d);
if (!hit) throw new Error(`no audio input at index ${d} — list them with: /dictate devices`);
return hit.name;
}
return d;
}
async function toggleDictation(ctx: ExtensionContext, opts: { device?: string | null; language?: string; submit?: boolean } = {}): Promise<void> {
if (!ctx.hasUI) { ctx.ui.notify("dictation needs the interactive TUI", "error"); return; }
// second press while live => stop, flush, finalise the composer with the deduped transcript.
if (active) {
const { rec, submit, ourTail } = active; active = null;
ctx.ui.setStatus("asr", undefined);
try {
const res = await rec.stop();
const text = res.text.trim();
if (rec.error) { stripTail(ctx, ourTail); ctx.ui.notify(`dictation error: ${rec.error}`, "error"); return; }
if (!text) { stripTail(ctx, ourTail); ctx.ui.notify("no speech detected (speak a little longer, or run asr_health)", "warning"); return; }
if (submit) { stripTail(ctx, ourTail); pi.sendUserMessage(text); }
else paintTail(ctx, ourTail, text); // swap our live suffix for the finalised text; user's own text is untouched
ctx.ui.notify(`dictated ${wordCount(text)} words (${res.seconds}s) → ${submit ? "sent" : "added"}`, "info");
} catch (e: any) {
rec.cancel(); stripTail(ctx, ourTail);
ctx.ui.notify(`dictation failed: ${e?.message ?? e}`, "error");
}
return;
}
if (starting) return; // ignore a double-press during mint+handshake
starting = true;
const language = opts.language ?? process.env.ASR_DICTATE_LANGUAGE ?? "en-US";
let device: string | null;
try { device = await resolveDevice(opts.device); }
catch (e: any) { starting = false; ctx.ui.notify(`dictation: ${e?.message ?? e}`, "error"); return; }
// No start-of-session snapshot: dictation appends to the *live* editor text (paintTail/stripTail),
// so any edits or deletions the user makes while we record are preserved and never revived.
const client = new RealtimeAsrClient({ language, punctuation: true });
ctx.ui.setStatus("asr", "● listening …");
let rec: MicRecording;
try {
rec = await client.recordMic({
device,
// stream word-by-word: paintTail owns only our appended suffix, so the user's draft is preserved
// and anything they delete stays deleted (no snapshot is ever written back). liveText is cumulative.
onLive: (live) => { if (active) { active.ourTail = paintTail(ctx, active.ourTail, live); ctx.ui.setStatus("asr", `● ${live.trim().slice(-70) || "…listening"}`); } },
});
} catch (e: any) {
starting = false; ctx.ui.setStatus("asr", undefined);
ctx.ui.notify(`could not start mic capture: ${e?.message ?? e}\n(ffmpeg on PATH w/ AVFoundation; grant the terminal Microphone access in System Settings → Privacy & Security)`, "error");
return;
}
starting = false;
active = { rec, submit: !!opts.submit, ourTail: "" };
active.ourTail = paintTail(ctx, "", rec.liveText()); // reflect anything already recognised without overwriting the user's draft
ctx.ui.notify(`🎙 recording${device ? ` from “${device}”` : " (default mic)"} — speech streams into the prompt; ctrl+shift+d or /dictate to stop`, "info");
}
pi.registerShortcut("ctrl+shift+m", {
description: "Toggle macOS live dictation (press to start, press again to stop and paste into the prompt)",
handler: async (ctx) => { await toggleDictation(ctx); },
});
pi.registerCommand("dictate", {
description: "Live macOS dictation (mic → realtime ASR): /dictate [stop|devices] [--device <name|index>] [--lang xx-YY] [--submit]",
handler: async (args: string, ctx: ExtensionContext) => {
const toks = args.trim().split(/\s+/).filter(Boolean);
const head = (toks[0] ?? "").toLowerCase();
if (head === "devices") {
const ds = await listAudioInputs();
ctx.ui.notify(ds.length ? ds.map(d => `[${d.index}] ${d.name}`).join("\n") : "no audio inputs found (ffmpeg w/ AVFoundation? Mic permission?)", "info");
return;
}
if (head === "stop" || head === "cancel") {
if (!active) { ctx.ui.notify("not currently recording", "info"); return; }
await toggleDictation(ctx);
return;
}
let device: string | undefined, language: string | undefined, submit = false;
for (let i = 0; i < toks.length; i++) {
const t = toks[i];
if (t === "--device") device = toks[++i];
else if (t === "--lang") language = toks[++i];
else if (t === "--submit") submit = true;
}
await toggleDictation(ctx, { device, language, submit });
},
});
}
+243
View File
@@ -277,6 +277,175 @@ export class RealtimeAsrClient {
return { handle, minted };
}
/**
* Push-to-talk live dictation on macOS: spawn ffmpeg AVFoundation mic capture and stream
* s16le frames into a realtime ASR session as they arrive (no artificial pacer — the mic is
* already real-time paced). Returns a handle; call `.stop()` to commit + flush + get the final
* transcript, or `.cancel()` to drop it. `onDelta` drives a live status line.
*
* The completion drain reuses the same quiescence logic as transcribe(): the server emits
* cumulative `.delta`s and a words-bearing then words-less `.completed` per segment, and
* `is_last_result` never flips true, so after commit we wait until no new segment for quietMs.
*/
async recordMic(opts: {
device?: string | null; frameMs?: number; commitMs?: number;
quietMs?: number; emptyGraceMs?: number; finalTimeout?: number; tailMs?: number;
sampleRate?: number; channels?: number; session?: Record<string, any>;
onDelta?: (t: string) => void; onLive?: (live: string) => void; onLevel?: (peak: number) => void; onStart?: (d: { device: string | null }) => void;
signal?: AbortSignal;
} = {}): Promise<MicRecording> {
this.requireKey();
const sampleRate = opts.sampleRate ?? 16000, channels = opts.channels ?? 1;
const frameBytes = Math.max(320, Math.round((sampleRate * (opts.frameMs ?? 100)) / 1000) * 2 * channels);
const minted = await this.mint();
if (opts.signal?.aborted) throw new Error("cancelled");
const handle = await this.open({ token: minted.client_secret?.value ?? null });
handle.send({ type: "transcription_session.update", session: this.sessionConfig(minted, opts.session ?? {}) });
await handle.expect(EV.sessionUpdated, 20000);
const norm = (s: string): string => (s ?? "").replace(/\s+/g, " ").trim();
const deltas: string[] = [];
const finals: Map<string, any> = new Map();
const failures: unknown[] = [];
// Per-segment live model. The endpoint only runs its decoder on `commit` (a bare `append` emits
// nothing until you commit), so the pump below commits periodically. Each item_id accumulates a
// cumulative `.delta` and later a `.completed` whose transcript equals that delta. We render the
// longer of (partial, final) per id, in first-seen order — so a segment can never be duplicated
// and the live view never stalls even if a `.completed` for it arrives before the next delta.
const segOrder: string[] = [];
const segPartial: Map<string, string> = new Map();
const segFinal: Map<string, string> = new Map();
const touchSeg = (id: string): void => { if (!segOrder.includes(id)) segOrder.push(id); };
let lastLive = "";
const liveText = (): string => {
const parts: string[] = [];
for (const id of segOrder) {
const p = norm(segPartial.get(id) ?? "");
const f = norm(segFinal.get(id) ?? "");
const best = f.length > p.length ? f : p; // the cumulative delta already contains the final text
if (best) parts.push(best);
}
return norm(parts.join(" "));
};
const emitLive = (): void => { const s = liveText(); if (s !== lastLive) { lastLive = s; opts.onLive?.(s); } };
const offs = [
this.on(EV.delta, (e: any) => {
const id = e.item_id ?? "_", t = e.delta ?? "";
touchSeg(id); segPartial.set(id, t); deltas.push(t);
rec.partial = t; opts.onDelta?.(t); emitLive();
}),
this.on(EV.completed, (e: any) => {
const key = e.item_id ?? "_";
const prev = finals.get(key);
const nw = e.words_info?.words?.length ?? 0, pw = prev?.words_info?.words?.length ?? 0;
finals.set(key, prev && pw > nw && prev.transcript ? { ...e, words_info: prev.words_info } : e);
const t = norm(e.transcript ?? "");
if (t && t.length > (segFinal.get(key)?.length ?? 0)) segFinal.set(key, t);
emitLive();
}),
this.on(EV.failed, (e: any) => failures.push(e)),
];
let pending: Buffer[] = [], carry: Buffer = Buffer.alloc(0), total = 0, peak = 0, cancelled = false;
let drainP: Promise<TranscriptResult> | null = null;
let commitTimer: ReturnType<typeof setInterval> | null = null;
const rec: MicRecording = {
partial: "", bytes: 0, seconds: 0, level: 0, error: null,
liveText,
stop: () => (drainP ??= drain()),
cancel: () => { cancelled = true; if (commitTimer) { clearInterval(commitTimer); commitTimer = null; } try { handle.send?.({ type: "input_audio_buffer.clear" }); } catch {} for (const f of offs) f(); handle.close(); },
};
function flush(force = false): void {
let buf = carry.length ? Buffer.concat([carry, ...pending]) : Buffer.concat(pending);
pending = [];
if (!force && buf.length < frameBytes) { carry = buf; return; }
carry = Buffer.alloc(0);
for (let i = 0; i < buf.length; i += frameBytes) {
handle.send({ type: "input_audio_buffer.append", audio: buf.subarray(i, Math.min(i + frameBytes, buf.length)).toString("base64") });
}
}
const child = spawn("ffmpeg", micArgs(opts.device, { sampleRate, channels }), { stdio: ["ignore", "pipe", "pipe"] });
let capErr = "";
child.stdout.on("data", (b: Buffer) => {
total += b.length; rec.bytes = total; rec.seconds = +(total / (sampleRate * 2 * channels)).toFixed(2);
const lv = rms16(b); if (lv > peak) { peak = lv; rec.level = lv; } opts.onLevel?.(lv);
pending.push(b); flush();
});
child.stderr.on("data", b => { capErr += b.toString(); });
let capDied: (e: Error) => void = () => {};
const capEnded = new Promise<never>((_, rej) => { capDied = rej; });
child.on("error", e => { const m = (e as any).code === "ENOENT" ? "ffmpeg is not installed / not on PATH" : e.message; rec.error = m; capDied(new Error(m)); });
child.on("close", code => { if (code !== 0 && total === 0) { const m = `mic capture failed (ffmpeg exit ${code}). ${capErr.slice(-300)}\nIf macOS blocked the microphone: System Settings → Privacy & Security → Microphone → allow your terminal app.`; rec.error = m; capDied(new Error(m)); } });
opts.onStart?.({ device: opts.device ?? null });
// Drive the decoder: the server stays silent on `append` alone and only produces `.delta`/`.completed`
// once a `commit` is received, so we commit whenever new audio has arrived since the last commit.
// Each commit flushes a fresh cumulative `.delta` of everything heard so far — that is what makes the
// live transcript actually advance while you speak. drain()'s final commit catches the tail.
const commitMs = Math.max(250, opts.commitMs ?? 500);
let lastCommitBytes = 0;
commitTimer = setInterval(() => {
if (cancelled || drainP) return;
if (total <= lastCommitBytes) return; // no new audio since last commit (silence) — skip the empty decode
lastCommitBytes = total;
try { handle.send({ type: "input_audio_buffer.commit" }); } catch {}
}, commitMs);
function stopCapture(): void { try { child.kill("SIGINT"); } catch {} }
async function drain(): Promise<TranscriptResult> {
if (commitTimer) { clearInterval(commitTimer); commitTimer = null; }
stopCapture();
// let stdout finish flushing the tail, then force-send the last partial frame
await Promise.race([new Promise(r => child.once("close", r)), sleep(400), capEnded.catch(() => 0)]);
flush(true);
handle.send({ type: "input_audio_buffer.commit" });
const t0 = Date.now(), deadline = t0 + (opts.finalTimeout ?? 60000);
const quietMs = opts.quietMs ?? 1200, emptyGraceMs = opts.emptyGraceMs ?? 6000;
let lastCount = 0, lastChange = t0;
while (Date.now() < deadline) {
if (opts.signal?.aborted) break;
const n = finals.size;
if (n !== lastCount) { lastCount = n; lastChange = Date.now(); }
if (failures.length) break;
if (n > 0 && Date.now() - lastChange >= quietMs) break;
if (n === 0 && Date.now() - t0 >= emptyGraceMs) break;
await sleep(120);
}
handle.send({ type: "input_audio_buffer.done" });
await sleep(opts.tailMs ?? 500);
const list = Array.from(finals.values());
const text = list.length
? list.map(f => f.transcript ?? "").filter(Boolean).join(" ").replace(/\s+/g, " ").trim()
: (deltas.at(-1) ?? "").replace(/\s+/g, " ").trim();
for (const f of offs) f();
try { handle.send?.({ type: "input_audio_buffer.clear" }); } catch {}
handle.close();
return build(list, text);
}
function build(list: any[], text: string): TranscriptResult {
return {
text,
segments: list.map(f => ({ transcript: f.transcript ?? "", isLast: f.is_last_result ?? false })),
words: list.flatMap(f => f.words_info?.words ?? []),
failed: failures,
seconds: +(total / (sampleRate * 2 * channels)).toFixed(3),
model: minted.input_audio_transcription?.model ?? null,
sessionId: minted.id ?? "",
protocol: handle.protocol,
};
}
void capEnded.catch(() => {}); // capture-death is surfaced through drain's emptyGraceMs path when .stop() runs
return rec;
}
/**
* One-shot transcribe of raw s16le pcm16 audio.
* `onDelta` receives cumulative partials (great for a live UI), return value has final text.
@@ -444,6 +613,80 @@ export async function hasFfmpeg(): Promise<boolean> {
try { await run("ffmpeg", ["-version"], () => {}); return true; } catch { return false; }
}
/* ------------------------------------------------------- live mic capture -- */
export interface AudioDevice { index: number; name: string }
/** True when this ffmpeg build carries the AVFoundation input (macOS mic capture). */
export async function supportsAvfoundation(): Promise<boolean> {
try {
let out = "";
await run("ffmpeg", ["-hide_banner", "-demuxers"], b => { out += b.toString(); });
return /\bavfoundation\b/i.test(out);
} catch { return false; }
}
/**
* Enumerate AVFoundation audio input devices via `ffmpeg -f avfoundation -list_devices true`.
* The device table is printed to **stderr** and ffmpeg then errors opening the dummy input,
* so we read stderr and slice the "AVFoundation audio devices:" block. Empty array when ffmpeg
* or the AVFoundation demuxer is missing (caller should surface the TCC / install hint).
*/
export async function listAudioInputs(): Promise<AudioDevice[]> {
let stderr = "";
try {
const r = await run("ffmpeg", ["-hide_banner", "-f", "avfoundation", "-list_devices", "true", "-i", "dummy"], () => {});
stderr = r.stderr;
} catch (e: any) {
// expected: ffmpeg exits non-zero after "Error opening input file dummy"
if (!/AVFoundation/i.test(stderr ?? e?.message ?? "")) throw new Error(e?.message ?? String(e));
}
const block = stderr.split(/AVFoundation audio devices:/i)[1] ?? "";
const devs: AudioDevice[] = [];
for (const line of block.split("\n")) {
if (/AVFoundation video devices:/i.test(line)) break;
const m = line.match(/\[(\d+)\]\s+(.*\S)\s*$/);
if (m) devs.push({ index: +m[1], name: m[2] });
}
return devs;
}
/** `-i none:<device|default>` selects the audio half of the [video]:[audio] AVFoundation spec. */
function micArgs(device: string | null | undefined, { sampleRate = 16000, channels = 1 }: { sampleRate?: number; channels?: number } = {}): string[] {
// Testing / power-user override: ASR_MIC_ARGS replaces the ffmpeg argument list verbatim (the `ffmpeg`
// binary itself is still used). e.g. feed a clip in real-time via `-re` as a stand-in for a live mic.
const override = (process.env.ASR_MIC_ARGS ?? "").trim();
if (override) return override.split(/\s+/g).filter(Boolean);
const spec = device && device.trim() && device.trim().toLowerCase() !== "default" ? `none:${device.trim()}` : "none:default";
return ["-hide_banner", "-v", "error", "-f", "avfoundation", "-i", spec, "-ac", String(channels), "-ar", String(sampleRate), "-f", "s16le", "-"];
}
/** Peak-normalised RMS of an s16le buffer, 0..1 (drives the "●" level meter in the UI). */
function rms16(buf: Buffer): number {
let sum = 0, n = 0;
for (let i = 0; i + 1 < buf.length; i += 2) { const v = buf.readInt16LE(i) / 32768; sum += v * v; n++; }
return n ? Math.sqrt(sum / n) : 0;
}
export interface MicRecording {
/** Last cumulative partial the server sent (for a live status line). */
partial: string;
/** Total s16le bytes captured so far. */
bytes: number;
/** Seconds captured so far. */
seconds: number;
/** Peak level 0..1 since recording started (for a VU hint). */
level: number;
/** Capture error surfaced by ffmpeg (missing binary / blocked microphone), if any. */
error: string | null;
/** Deduped, composited live text: finalized segments + the in-progress cumulative partial. */
liveText(): string;
/** Stop capture, commit, flush, and resolve with the final transcript. Idempotent. */
stop(): Promise<TranscriptResult>;
/** Abort without waiting for the final segment (drops the session). */
cancel(): void;
}
function run(cmd: string, args: string[], onData: (b: Buffer) => void): Promise<{ code: number; stderr: string }> {
return new Promise((res, rej) => {
const child = spawn(cmd, args, { stdio: ["ignore", "pipe", "pipe"] });