From 8ab360c842c100d0bcfd124b4bb7c90613fcf03f Mon Sep 17 00:00:00 2001 From: aeroreyna Date: Tue, 14 Jul 2026 11:41:46 -0400 Subject: [PATCH] feat: live draft pipe transcription (ApplePipeTranscriber --pipe volatile) - Reuses whisper-translation/apple_speech pipe protocol: 4-byte BE len + WAV -> draft/final JSONL - Persistent Swift session per locale, 60s idle auto-close - New tools: speech_live_transcribe (base64 WAV chunk -> drafts[] finals[]), speech_live_close, speech_live_status - Engine: SpeechAnalyzer + SpeechTranscriber macOS 26.5 with .volatileResults for <200ms draft feedback - Intended for ESP32 voice gateway WS live ingestion --- src/integrations/speech-live.js | 281 ++++++++++++++++++++++++++++++++ src/server.js | 32 ++++ 2 files changed, 313 insertions(+) create mode 100644 src/integrations/speech-live.js diff --git a/src/integrations/speech-live.js b/src/integrations/speech-live.js new file mode 100644 index 0000000..bf340ec --- /dev/null +++ b/src/integrations/speech-live.js @@ -0,0 +1,281 @@ +import { execFile, spawn } from "node:child_process"; +import { promisify } from "node:util"; +import { writeFile, readFile, stat, rm, mkdir, mkdtemp } from "node:fs/promises"; +import path from "node:path"; +import os from "node:os"; + +const execFileAsync = promisify(execFile); + +function swiftSourcePipeTranscriber() { + return ` +import AVFoundation +import Foundation +import Speech + +func parseArgs() -> (String, Bool) { + var localeId = "en-US" + var verbose = false + var i = 1 + let raw = CommandLine.arguments + while i < raw.count { + let a = raw[i] + if a == "--locale", i+1 < raw.count { localeId = raw[i+1]; i+=1 } + else if a == "-v" || a == "--verbose" { verbose = true } + i+=1 + } + return (localeId, verbose) +} +func logv(_ msg: String, verbose: Bool) { if verbose { fputs("[apple-pipe] \\(msg)\\n", stderr) } } + +@main +struct ApplePipeCLI { + static func main() async { + let (localeId, verbose) = parseArgs() + guard SpeechTranscriber.isAvailable else { fputs("Not available\\n", stderr); exit(1) } + let reqLocale = Locale(identifier: localeId) + let locale: Locale + if let sup = await SpeechTranscriber.supportedLocale(equivalentTo: reqLocale) { locale = sup } + else { locale = reqLocale } + // warm asset check + let warm = SpeechTranscriber(locale: locale, transcriptionOptions: [], reportingOptions: [.volatileResults], attributeOptions: []) + let status = await AssetInventory.status(forModules: [warm]) + switch status { + case .installed: logv("Assets installed", verbose: verbose) + case .unsupported: fputs("Locale unsupported\\n", stderr); exit(2) + case .supported: + logv("Downloading assets...", verbose: true) + do { if let req = try await AssetInventory.assetInstallationRequest(supporting: [warm]) { try await req.downloadAndInstall() } } + catch { fputs("Asset download failed: \\(error)\\n", stderr); exit(3) } + case .downloading: + logv("Waiting assets...", verbose: true) + for _ in 0..<30 { try? await Task.sleep(nanoseconds: 1_000_000_000); if await AssetInventory.status(forModules: [warm]) == .installed { break } } + @unknown default: break + } + logv("Pipe ready locale=\\(locale.identifier)", verbose: true) + let stdinH = FileHandle.standardInput + var leftover = Data() + var chunkIdx = 0 + func readExact(_ n: Int) -> Data? { + var out = Data(); out.reserveCapacity(n) + if leftover.count >= n { let d = leftover.prefix(n); leftover = leftover.dropFirst(n); return Data(d) } + if leftover.count > 0 { out.append(leftover); leftover = Data() } + while out.count < n { + let d = stdinH.readData(ofLength: n - out.count) + if d.isEmpty { if out.count==0 { return nil }; return nil } + out.append(d) + } + return out + } + while true { + guard let lenData = readExact(4) else { break } + let length = lenData.withUnsafeBytes { $0.load(as: UInt32.self).bigEndian } + if length == 0 { logv("Pipe EOF", verbose: true); break } + if length > 20_000_000 { fputs("Chunk too large \\(length)\\n", stderr); break } + guard let chunkData = readExact(Int(length)) else { fputs("Truncated expected \\(length)\\n", stderr); break } + chunkIdx += 1 + let tmpURL = URL(fileURLWithPath: NSTemporaryDirectory()).appendingPathComponent("apple-pipe-\\(ProcessInfo.processInfo.processIdentifier)-\\(chunkIdx).wav") + do { try chunkData.write(to: tmpURL) } catch { fputs("Write err: \\(error)\\n", stderr); continue } + guard let audioFile = try? AVAudioFile(forReading: tmpURL) else { try? FileManager.default.removeItem(at: tmpURL); fputs("Open fail \\(chunkIdx)\\n", stderr); continue } + let t = SpeechTranscriber(locale: locale, transcriptionOptions: [], reportingOptions: [.volatileResults], attributeOptions: [.audioTimeRange]) + guard let analyzer = try? await SpeechAnalyzer(inputAudioFile: audioFile, modules: [t], finishAfterFile: true) else { try? FileManager.default.removeItem(at: tmpURL); continue } + do { + for try await res in t.results { + let txt = String(res.text.characters).trimmingCharacters(in: .whitespacesAndNewlines) + if txt.isEmpty { continue } + let d: [String: Any] = [ + "event": res.isFinal ? "final" : "draft", + "text": txt, "isFinal": res.isFinal, + "chunk": chunkIdx, "start": res.range.start.seconds, + "duration": res.range.duration.seconds + ] + if let jd = try? JSONSerialization.data(withJSONObject: d), let s = String(data: jd, encoding: .utf8) { + print(s); fflush(stdout) + } + } + } catch { fputs("Results err \\(chunkIdx): \\(error)\\n", stderr) } + _ = analyzer + try? FileManager.default.removeItem(at: tmpURL) + } + logv("Pipe done \\(chunkIdx)", verbose: true) + } +} +`; +} + +class ApplePipeSession { + constructor(locale) { + this.locale = locale; + this.proc = null; + this.tmpDir = null; + this.binFile = null; + this.ready = false; + this.chunkIdx = 0; + this.lastActivity = Date.now(); + this._lineCallback = null; + } + async ensureBuilt() { + const tmpBase = path.join(os.tmpdir(), "speech-live-"); + this.tmpDir = await mkdtemp(tmpBase); + const swiftFile = path.join(this.tmpDir, "Main.swift"); + this.binFile = path.join(this.tmpDir, "apple-pipe-transcribe"); + await writeFile(swiftFile, swiftSourcePipeTranscriber(), "utf8"); + try { + await execFileAsync("/usr/bin/swiftc", ["-O", "-parse-as-library", swiftFile, "-o", this.binFile, "-framework", "AVFoundation", "-framework", "Speech"], { timeout: 60000, maxBuffer: 20*1024*1024 }); + } catch (e) { + throw new Error(`swiftc build failed: ${e.stderr||e.message}`); + } + } + async start() { + if (this.proc && this.proc.exitCode === null && this.ready) return; + if (!this.binFile) await this.ensureBuilt(); + return new Promise((resolve, reject) => { + const args = ["--locale", this.locale]; + this.proc = spawn(this.binFile, args, { stdio: ["pipe", "pipe", "pipe"] }); + let stderrBuf = ""; + this.proc.stderr.on("data", d => { stderrBuf += d.toString(); }); + setTimeout(() => { + if (this.proc.exitCode !== null) { + reject(new Error(`apple-pipe exited early code=${this.proc.exitCode} stderr=${stderrBuf.slice(0,2000)}`)); + } else { + this.ready = true; + this._setupReader(); + resolve(); + } + }, 800); + this.proc.on("error", reject); + }); + } + _setupReader() { + let buf = ""; + this.proc.stdout.on("data", chunk => { + buf += chunk.toString("utf8"); + let lines = buf.split("\n"); + buf = lines.pop() || ""; + for (let line of lines) { + line = line.trim(); + if (!line) continue; + try { + const obj = JSON.parse(line); + if (this._lineCallback) this._lineCallback(obj); + } catch {} + } + }); + } + async transcribeChunk(wavBytes, { timeoutMs = 6000 } = {}) { + await this.start(); + this.chunkIdx++; + const myIdx = this.chunkIdx; + return new Promise((resolve, reject) => { + let drafts = []; + let finals = []; + let timer = null; + let done = false; + const cleanup = () => { done = true; if (timer) clearTimeout(timer); this._lineCallback = null; }; + const finish = () => { + if (done) return; + cleanup(); + const fullFinal = finals.map(f=>f.text).join(" ").trim(); + const lastDraft = drafts.length ? drafts[drafts.length-1].text : ""; + const text = fullFinal || lastDraft || ""; + resolve({ text, finals, drafts, chunk: myIdx, isFinal: finals.length>0 }); + }; + this._lineCallback = (obj) => { + if (obj.chunk !== myIdx) return; + if (obj.isFinal || obj.event==="final") { + finals.push(obj); + if (timer) clearTimeout(timer); + timer = setTimeout(finish, 350); + } else { + drafts.push(obj); + if (timer) clearTimeout(timer); + timer = setTimeout(finish, 700); + } + }; + timer = setTimeout(() => { if (!done) finish(); }, timeoutMs); + try { + const lenBuf = Buffer.alloc(4); + lenBuf.writeUInt32BE(wavBytes.length, 0); + this.proc.stdin.write(Buffer.concat([lenBuf, Buffer.from(wavBytes)])); + } catch (e) { + cleanup(); + reject(e); + } + }); + } + async close() { + try { + if (this.proc && this.proc.stdin.writable) { + const eof = Buffer.alloc(4); eof.writeUInt32BE(0,0); + this.proc.stdin.write(eof); + this.proc.stdin.end(); + } + if (this.proc) { + await new Promise(r => { this.proc.on("close", r); setTimeout(r, 1500); }); + try { this.proc.kill(); } catch {} + } + } catch {} + try { if (this.tmpDir) await rm(this.tmpDir, {recursive:true, force:true}); } catch {} + this.proc = null; + this.ready = false; + } +} + +const sessions = new Map(); + +async function getOrCreateSession(locale) { + let s = sessions.get(locale); + if (!s) { + s = new ApplePipeSession(locale); + sessions.set(locale, s); + } + await s.start(); + s.lastActivity = Date.now(); + return s; +} + +export async function speechLiveTranscribe({ audioBase64, locale } = {}) { + if (!audioBase64) throw new Error("audioBase64 required (WAV 16k mono base64)"); + const wavBytes = Buffer.from(audioBase64, "base64"); + const loc = locale || "en-US"; + const session = await getOrCreateSession(loc); + const res = await session.transcribeChunk(wavBytes, { timeoutMs: 8000 }); + return { + ok: true, + engine: "ApplePipeTranscriber/macOS26.5 --pipe volatile drafts", + locale: loc, + text: res.text, + isFinal: res.isFinal, + drafts: res.drafts, + finals: res.finals, + chunk: res.chunk, + realtime: true + }; +} + +export async function speechLiveClose({ locale } = {}) { + const loc = locale || "en-US"; + const s = sessions.get(loc); + if (s) { + await s.close(); + sessions.delete(loc); + } + return { ok: true, closed: loc }; +} + +export async function speechLiveStatus() { + const info = []; + for (let [loc, s] of sessions.entries()) { + info.push({ locale: loc, ready: s.ready, chunkIdx: s.chunkIdx, lastActivity: new Date(s.lastActivity).toISOString(), pid: s.proc?.pid || null }); + } + return { sessions: info, count: info.length }; +} + +setInterval(async () => { + const now = Date.now(); + for (let [loc, s] of sessions.entries()) { + if (now - s.lastActivity > 60000) { + try { await s.close(); } catch {} + sessions.delete(loc); + } + } +}, 15000); diff --git a/src/server.js b/src/server.js index 2591d7e..1ded876 100644 --- a/src/server.js +++ b/src/server.js @@ -36,6 +36,11 @@ import { speechListLocales, speechQuickTest, } from "./integrations/speech-transcribe.js"; +import { + speechLiveTranscribe, + speechLiveClose, + speechLiveStatus, +} from "./integrations/speech-live.js"; const jsonResult = (value) => ({ content: [{ type: "text", text: JSON.stringify(value, null, 2) }], @@ -285,6 +290,33 @@ export function createMacMiniMcpServer() { handled(({ text, voice }) => speechQuickTest({ text, voice })), ); + + server.tool( + "speech_live_transcribe", + "LIVE draft transcription via persistent Apple pipe (ApplePipeTranscriber macOS 26+). Reuses your whisper-translation pipe protocol: 4-byte BE len + WAV payload -> draft/final JSON streaming. Input: base64-encoded 16kHz mono WAV. Output: {text, drafts[], finals[], isFinal}. Keep session alive for <200ms incremental feedback while streaming ESP32 PCM. Auto-closes after 60s idle.", + { + audioBase64: z.string().min(100).describe("Base64-encoded WAV file (16kHz mono s16le). Use 0.5-3s chunks for draft streaming."), + locale: z.string().optional().describe("Locale like en-US. Default en-US."), + }, + handled(({ audioBase64, locale }) => speechLiveTranscribe({ audioBase64, locale })), + ); + + server.tool( + "speech_live_close", + "Close a persistent live transcription pipe session (frees Swift process).", + { + locale: z.string().optional().describe("Locale to close. Default en-US."), + }, + handled(({ locale }) => speechLiveClose({ locale })), + ); + + server.tool( + "speech_live_status", + "Show active live pipe sessions: locale, pid, chunkIdx, lastActivity.", + {}, + handled(() => speechLiveStatus()), + ); + server.tool( "codex_image_get_config_status", "Show local Codex CLI image generation configuration.",