112 lines
4.7 KiB
JavaScript
112 lines
4.7 KiB
JavaScript
import WebSocket from 'ws';
|
|
import { openStt, openTts, streamAnswer } from './providers.mjs';
|
|
import { deviceTools } from './device.mjs';
|
|
import { quickAction } from './quick-actions.mjs';
|
|
|
|
const MAX_AUDIO = 60 * 16000 * 2;
|
|
const STT_CHUNK_BYTES = 3200; // 100 ms of mono PCM16 at 16 kHz.
|
|
const MAX_BUFFERED = 2 * 1024 * 1024;
|
|
|
|
export function handleTurn(client, store) {
|
|
const controller = new AbortController();
|
|
const { signal } = controller;
|
|
const device = deviceTools(client, signal);
|
|
let state = 'waiting', stt, tts, bytes = 0, received = 0, messages = 0, ended = false, terminal = false;
|
|
let pending = [];
|
|
let pendingBytes = 0;
|
|
const timeout = setTimeout(() => fail('timeout'), 120000);
|
|
const send = obj => { if (client.readyState === WebSocket.OPEN) client.send(JSON.stringify(obj)); };
|
|
function cleanup() {
|
|
clearTimeout(timeout);
|
|
controller.abort();
|
|
if (stt && stt.readyState < WebSocket.CLOSING) stt.terminate();
|
|
if (tts?.ws && tts.ws.readyState < WebSocket.CLOSING) tts.ws.terminate();
|
|
}
|
|
function fail(code) {
|
|
if (terminal) return;
|
|
terminal = true; send({ type: 'error', code }); cleanup();
|
|
if (client.readyState === WebSocket.OPEN) client.close();
|
|
}
|
|
function sendChunk(chunk, commit = false) {
|
|
if (!stt || stt.readyState !== WebSocket.OPEN || stt.bufferedAmount > MAX_BUFFERED) return fail('stt_unavailable');
|
|
stt.send(JSON.stringify({ message_type: 'input_audio_chunk', audio_base_64: chunk.toString('base64'), ...(commit ? { commit: true } : {}) }));
|
|
}
|
|
function complete(text, answer) {
|
|
store.recordTurn(text, answer);
|
|
terminal = true;
|
|
send({ type: 'done' });
|
|
cleanup();
|
|
if (client.readyState === WebSocket.OPEN) client.close(1000);
|
|
}
|
|
|
|
async function respond(text, settings) {
|
|
if (state !== 'processing' || terminal) return;
|
|
state = 'responding';
|
|
text = text.trim();
|
|
if (!text) return fail('empty_transcript');
|
|
send({ type: 'transcript', text });
|
|
try {
|
|
const action = quickAction(text);
|
|
if (action) {
|
|
const result = await device.execute(action.name, action.args).catch(() => null);
|
|
if (terminal) return;
|
|
if (action.success(result)) return complete(text, action.done);
|
|
}
|
|
tts = await openTts(settings, signal, audio => {
|
|
if (terminal || client.readyState !== WebSocket.OPEN) return;
|
|
if (client.bufferedAmount > MAX_BUFFERED) return fail('client_too_slow');
|
|
client.send(audio, { binary: true });
|
|
});
|
|
if (terminal) return;
|
|
const answer = action?.failure ?? await streamAnswer(settings, store.getHistory(), text, signal, part => tts.text(part), device.execute);
|
|
if (action) tts.text(answer);
|
|
tts.finish();
|
|
await tts.done;
|
|
if (terminal) return;
|
|
complete(text, answer);
|
|
} catch { fail('upstream_error'); }
|
|
}
|
|
let settings;
|
|
let chain = Promise.resolve();
|
|
client.on('message', (data, binary) => {
|
|
if (++messages > 10000 || (received += data.length) > MAX_AUDIO + 131072) return fail('input_limit');
|
|
chain = chain.then(async () => {
|
|
if (terminal) return;
|
|
if (binary) {
|
|
if (state !== 'recording') return fail('invalid_state');
|
|
const audio = Buffer.from(data);
|
|
if (!audio.length || audio.length % 2 || (bytes += audio.length) > MAX_AUDIO) return fail('invalid_audio');
|
|
if (pendingBytes >= STT_CHUNK_BYTES) {
|
|
sendChunk(Buffer.concat(pending, pendingBytes));
|
|
pending = [];
|
|
pendingBytes = 0;
|
|
}
|
|
pending.push(audio);
|
|
pendingBytes += audio.length;
|
|
return;
|
|
}
|
|
let msg;
|
|
try { msg = JSON.parse(data.toString()); } catch { return fail('invalid_message'); }
|
|
if (device.accept(msg)) return;
|
|
if (msg?.type === 'start' && state === 'waiting') {
|
|
state = 'connecting'; settings = store.getSettings();
|
|
try {
|
|
stt = await openStt(settings, signal, text => { if (!terminal && state === 'recording') send({ type: 'partial', text }); }, text => { if (!ended || terminal) return; respond(text, settings); }, () => fail('stt_error'));
|
|
if (terminal) return;
|
|
state = 'recording';
|
|
} catch { fail('stt_error'); }
|
|
} else if (msg?.type === 'end' && state === 'recording') {
|
|
if (!pendingBytes) return fail('empty_audio');
|
|
ended = true;
|
|
state = 'processing';
|
|
sendChunk(Buffer.concat(pending, pendingBytes), true);
|
|
pending = [];
|
|
pendingBytes = 0;
|
|
} else fail('invalid_state');
|
|
}).catch(() => fail('internal_error'));
|
|
});
|
|
client.on('error', () => { if (!terminal) { terminal = true; cleanup(); } });
|
|
client.on('close', () => { if (!terminal) { terminal = true; cleanup(); } });
|
|
send({ type: 'ready' });
|
|
}
|