import { EventEmitter } from 'node:events'; import { PipeWireCapture } from './audio-pipewire.js'; import { PipeWirePlayback } from './audio-playback.js'; import { WakeEngine } from './wake-engine.js'; import { VadSegmenter } from './vad.js'; import { isMeaningfulTranscript, isSpeakable, speakableForTts, SentenceBuffer } from './transcript.js'; import { VoiceMetrics } from './voice-metrics.js'; export const POST_PLAYBACK_COOLDOWN_MS = 400; const FAST_COMMANDS = new Map([ ['cancel', 'cancel'], ['stop', 'cancel'], ['never mind', 'cancel'], ['hands off', 'cancel'], ['stop clicking', 'cancel'], ["that's enough", 'cancel'], ['go to sleep', 'sleep'], ['privacy mode', 'sleep'], ['repeat that', 'repeat'], ['dictate this', 'dictate'], ['look at my screen', 'screen'], ['use the computer', 'computer'], ['take the wheel', 'computer'], ['switch to compose', 'compose'], ['switch to imagine', 'imagine'], ['switch to files', 'files'], ['switch to computer', 'computer'], ]); export function fastCommand(text) { return FAST_COMMANDS.get(String(text || '').trim().toLowerCase()) || null; } export class VoiceLoop extends EventEmitter { constructor({ daemon, capture = new PipeWireCapture(), playback = new PipeWirePlayback(), wake = new WakeEngine(), vad = new VadSegmenter(), asr, tts, cooldownMs = POST_PLAYBACK_COOLDOWN_MS, listeningMode = 'conversation', now = () => Date.now(), captureRetryMs = 2000, asrRetryMs = 4000 } = {}) { super(); this.daemon = daemon; this.capture = capture; this.playback = playback; this.wake = wake; this.vad = vad; this.asr = asr; this.tts = tts; this.cooldownMs = cooldownMs; this.listeningMode = listeningMode; this.now = now; this.captureRetryMs = captureRetryMs; this.asrRetryMs = asrRetryMs; this.isSpeaking = false; this.cooldownUntil = 0; this.running = false; this.ptt = false; this._generation = 0; this._speechQueue = Promise.resolve(); this.metrics = new VoiceMetrics(); this._transcriptions = new Set(); capture.on('audio', (chunk) => this.pushAudio(chunk)); capture.on('error', (error) => { this.status.capture = false; this.status.errors.capture = error.message; this.emit('error', error); this._scheduleCaptureRetry(); }); capture.on('close', () => { this.status.capture = false; if (this.running) { this.status.errors.capture = 'Microphone stream closed'; this._scheduleCaptureRetry(); } }); wake.on('unavailable', () => { this.status.wake = false; }); wake.on('error', (error) => { this.status.wake = false; this.emit('error', error); }); this.status = { asr: false, tts: false, capture: false, wake: false, microphone: this.asr != null, speech: this.tts != null, errors: {} }; wake.on('wake', (phrase) => this.wakeHeard(phrase)); vad.on('level', (rms) => daemon?.emit('ListeningLevel', rms)); vad.on('utterance', (audio) => { const task = this.transcribe(audio).catch((error) => this.emit('error', error)); this._transcriptions.add(task); task.finally(() => this._transcriptions.delete(task)).catch(() => {}); }); } async start() { if (this.running) return; if (!this.tts) this.status.errors.tts = 'Spoken replies are disabled'; this.running = true; this._armWake(); this._armCapture(); } _notifyStatus() { try { this.daemon?.emit?.('StateChanged', this.daemon.state || 'ARMED'); } catch {} } _armWake() { try { if (this.listeningMode !== 'ptt') this.wake.start?.(); this.wake.resume(); this.status.wake = this.listeningMode !== 'ptt' && Boolean(this.wake.command || this.wake.detect); } catch (error) { this.status.errors.wake = error.message; this.emit('error', error); } } _armCapture() { if (!this.running || this.status.capture || this.asr == null) return; try { this.capture.start(); this.status.capture = true; delete this.status.errors.capture; this._notifyStatus(); } catch (error) { this.status.errors.capture = error.message; this.emit('error', error); this._scheduleCaptureRetry(); } } _scheduleCaptureRetry() { if (!this.running || this._captureRetry || this.status.capture || this.asr == null) return; this._captureRetry = setTimeout(() => { this._captureRetry = 0; this._armCapture(); }, this.captureRetryMs); this._captureRetry.unref?.(); } _scheduleAsrRetry() { if (!this.running || this._asrRetry || this.status.asr || !this.asr) return; this._asrRetry = setTimeout(() => { this._asrRetry = 0; if (!this.running || this.status.asr || !this.asr) return; this.ensureAsr().catch(() => {}); }, this.asrRetryMs); this._asrRetry.unref?.(); } async ensureAsr() { if (!this.asr) return false; if (this.status.asr) return true; if (this._asrStarting) return this._asrStarting; this._asrStarting = (async () => { try { await this.asr.start(); if (!this.running) return false; this.status.asr = true; delete this.status.errors.asr; this._armCapture(); this._armWake(); this._notifyStatus(); return true; } catch (error) { const message = String(error?.message || error); this.status.errors.asr = message; this.emit('error', new Error(`ASR: ${message}`)); this._scheduleAsrRetry(); return false; } finally { this._asrStarting = null; } })(); return this._asrStarting; } async ensureTts() { if (!this.tts) { this.status.errors.tts = 'Spoken replies are disabled'; return false; } if (this.status.tts) return true; if (this._ttsStarting) return this._ttsStarting; this._ttsStarting = (async () => { try { await this.tts.start(); this.status.tts = true; delete this.status.errors.tts; return true; } catch (error) { const message = String(error?.message || error); this.status.errors.tts = message; this.emit('error', new Error(`TTS: ${message}`)); return false; } finally { this._ttsStarting = null; } })(); return this._ttsStarting; } async parkModels() { if (this._asrRetry) { clearTimeout(this._asrRetry); this._asrRetry = 0; } this._asrStarting = null; this._ttsStarting = null; await this.asr?.stop?.(); if (this.tts && this.tts !== this.asr) await this.tts.stop?.(); this.status.asr = false; this.status.tts = false; } async stop() { this.running = false; if (this._captureRetry) { clearTimeout(this._captureRetry); this._captureRetry = 0; } if (this._asrRetry) { clearTimeout(this._asrRetry); this._asrRetry = 0; } this.interrupt(); this.capture.stop(); this.wake.close(); this.vad.reset(); this.playback.stop(); await Promise.allSettled([this._speechQueue, ...this._transcriptions]); await this.asr?.stop?.(); if (this.tts !== this.asr) await this.tts?.stop?.(); } setPushToTalk(pressed) { if (!pressed) { this._pttHeld = false; this.ptt = false; this.vad.end(); return Promise.resolve(); } this._pttHeld = true; return this.ensureAsr().then((ok) => { if (!this._pttHeld) return; this.ptt = Boolean(ok && this.status.capture); if (this.ptt) this.wakeHeard('push-to-talk'); }).catch((error) => this.emit('error', error)); } pushAudio(chunk) { if (!this.running || this.daemon?.locked) return; const sleeping = this.daemon?.state === 'SLEEPING'; if (!sleeping && (this.isSpeaking || this.now() < this.cooldownUntil)) { this.metrics.feedbackDrop(); return; } if (!sleeping) { try { this.daemon?.emit('ListeningLevel', Math.min(1, Math.max(0, chunk?.length ? 0.01 : 0))); } catch {} } if (!this.ptt && this.listeningMode !== 'ptt') this.wake.push(chunk); if (sleeping) return; if (this.ptt || (this.listeningMode !== 'ptt' && this.daemon?.state === 'LISTENING')) this.vad.push(chunk); } wakeHeard(phrase) { if (!this.running || this.daemon?.locked) return; this.metrics.wakeAccepted(); this.daemon?.emit('WakeHeard', phrase); this.ensureAsr().catch((error) => this.emit('error', error)); this.daemon?.arm?.(); this.emit('wake', phrase); } async transcribe(audio) { if (!this.status.asr || !this.asr?.transcribeAudio) return; const generation = this._generation; const text = await this.asr.transcribeAudio(audio).catch((error) => { this.emit('error', error); return ''; }); if (!this.running || this.daemon?.locked || generation !== this._generation) return; if (!isMeaningfulTranscript(text)) { this.metrics.wakeRejected(); return; } this.metrics.utterance(); this.daemon?.emit('PartialTranscript', text); this.daemon?.emit('FinalTranscript', text); const command = fastCommand(text); if (command === 'cancel' || command === 'sleep') { this.daemon?.cancel?.(); if (command === 'sleep') await this.daemon?.sleep?.(); return; } if (command === 'repeat') { this.daemon?.say?.(this.daemon?.lastReply || 'There is nothing to repeat.'); return; } if (command === 'computer') { this.daemon?.computerGrant?.(false); return; } if (command === 'dictate') { this.emit('dictate'); return; } if (command === 'screen') { await this.daemon?.ask?.('What is on my screen?'); return; } await this.daemon?.ask?.(text); } interrupt() { this._generation += 1; this.playback.stop(); this.isSpeaking = false; this.wake.resume(); } _releaseSpeaking() { this.isSpeaking = false; this.wake.resume(); if (this.daemon?.state === 'SPEAKING') { try { this.daemon.voice?.finishSpeaking?.(); } catch {} if (this.daemon._finishSpeech) this.daemon._finishSpeech(); else this.daemon.setState?.('LISTENING'); } } async speak(text) { const generation = this._generation || 0; if (this.tts && !this.status.tts && typeof this.tts.start === 'function') await this.ensureTts(); if (!isSpeakable(text) || !this.tts?.speak || !this.status.tts) { this._releaseSpeaking(); return; } this.metrics.reply(); const sentences = String(text).match(/[^.!?]+[.!?]+|[^.!?]+$/g)?.map((part) => part.trim()).filter(isSpeakable) || []; if (!sentences.length) { this._releaseSpeaking(); return; } this._speechQueue = this._speechQueue.catch(() => {}).then(async () => { try { if (generation !== this._generation) return; for (const sentence of sentences) { if (generation !== this._generation) return; await this.speakSentence(sentence, generation); } } finally { if (generation === this._generation) this._releaseSpeaking(); } }); return this._speechQueue; } async speakSentence(text, generation) { this.isSpeaking = true; this.wake.pause(); try { const audio = await this.tts.speak(speakableForTts(text)); if (generation !== this._generation) return; await this.playback.play(audio.samples, audio.sampleRate); try { this.daemon?.emit('SpeakingLevel', 0); } catch {} } finally { this.isSpeaking = false; this.cooldownUntil = this.now() + this.cooldownMs; this.wake.resume(); } } } export { SentenceBuffer };