123 lines
6.6 KiB
JavaScript
123 lines
6.6 KiB
JavaScript
const { EventEmitter } = require("events");
|
|
const { spawn } = require("child_process");
|
|
const { BoundedQueue } = require("../sessions/bounded_queue");
|
|
|
|
class TranscriptionProvider extends EventEmitter {
|
|
async loadModel() { throw new Error("Model loading is not implemented."); }
|
|
async benchmark() { throw new Error("Benchmarking is not implemented."); }
|
|
async startSession() { throw new Error("Session start is not implemented."); }
|
|
async addTrack() { throw new Error("Track add is not implemented."); }
|
|
async pushAudio() { throw new Error("Audio input is not implemented."); }
|
|
async removeTrack() { throw new Error("Track removal is not implemented."); }
|
|
async stopSession() { throw new Error("Session stop is not implemented."); }
|
|
async health() { return { healthy: false, state: "unavailable" }; }
|
|
}
|
|
|
|
class WhisperWorkerSupervisor extends EventEmitter {
|
|
constructor(options = {}) {
|
|
super();
|
|
this.executable = options.executable;
|
|
this.args = options.args || [];
|
|
this.spawn = options.spawn || spawn;
|
|
this.maxRestarts = options.maxRestarts ?? 3;
|
|
this.restartWindowMs = options.restartWindowMs || 60000;
|
|
this.queue = new BoundedQueue({ maxItems: options.maxQueuedPackets || 250, maxBytes: options.maxQueuedBytes || 5 * 32000, maxAgeMs: 5000 });
|
|
this.child = null;
|
|
this.state = "stopped";
|
|
this.stopping = false;
|
|
this.restartTimes = [];
|
|
this.stdoutBuffer = "";
|
|
}
|
|
start() {
|
|
if (this.child) return;
|
|
if (!this.executable) throw new Error("Whisper worker executable is not configured.");
|
|
this.stopping = false;
|
|
this.state = "starting";
|
|
const child = this.spawn(this.executable, this.args, { stdio: ["pipe", "pipe", "pipe"], windowsHide: true });
|
|
this.child = child;
|
|
child.stdout.on("data", (chunk) => this.onStdout(chunk));
|
|
child.stderr.on("data", (chunk) => this.emit("diagnostic", { level: "warning", message: String(chunk).trim().slice(0, 1000) }));
|
|
child.on("error", (error) => this.emit("error", error));
|
|
child.on("exit", (code, signal) => this.onExit(code, signal));
|
|
child.stdin.on("drain", () => this.flush());
|
|
this.state = "running";
|
|
this.emit("state", { state: this.state });
|
|
this.flush();
|
|
}
|
|
send(metadata, pcm = null) {
|
|
const body = Buffer.from(JSON.stringify(metadata), "utf8");
|
|
const audio = pcm ? Buffer.from(pcm) : Buffer.alloc(0);
|
|
if (body.length > 64 * 1024 || audio.length > 5 * 32000) throw new Error("Worker packet exceeds its limit.");
|
|
const header = Buffer.alloc(8);
|
|
header.writeUInt32LE(body.length, 0);
|
|
header.writeUInt32LE(audio.length, 4);
|
|
const packet = Buffer.concat([header, body, audio]);
|
|
const accepted = this.queue.push(packet, { bytes: packet.length, capturedAt: metadata.captured_at || Date.now() });
|
|
if (accepted) this.flush();
|
|
return accepted;
|
|
}
|
|
flush() {
|
|
if (!this.child || this.state !== "running") return;
|
|
let packet;
|
|
while ((packet = this.queue.shift())) if (!this.child.stdin.write(packet)) break;
|
|
}
|
|
async stop() {
|
|
this.stopping = true;
|
|
this.queue.clear();
|
|
const child = this.child;
|
|
if (!child) { this.state = "stopped"; return; }
|
|
this.sendDirect({ type: "shutdown" });
|
|
await new Promise((resolve) => {
|
|
const timer = setTimeout(() => { child.kill(); resolve(); }, 2000);
|
|
child.once("exit", () => { clearTimeout(timer); resolve(); });
|
|
});
|
|
this.child = null;
|
|
this.state = "stopped";
|
|
}
|
|
sendDirect(metadata) { if (!this.child) return false; const body = Buffer.from(JSON.stringify(metadata)); const header = Buffer.alloc(8); header.writeUInt32LE(body.length, 0); return this.child.stdin.write(Buffer.concat([header, body])); }
|
|
onStdout(chunk) {
|
|
this.stdoutBuffer += String(chunk);
|
|
if (this.stdoutBuffer.length > 256 * 1024) this.stdoutBuffer = this.stdoutBuffer.slice(-64 * 1024);
|
|
let index;
|
|
while ((index = this.stdoutBuffer.indexOf("\n")) >= 0) {
|
|
const line = this.stdoutBuffer.slice(0, index).trim();
|
|
this.stdoutBuffer = this.stdoutBuffer.slice(index + 1);
|
|
if (!line) continue;
|
|
try { this.emit("message", JSON.parse(line)); }
|
|
catch { this.emit("diagnostic", { level: "warning", message: "Whisper worker emitted malformed output." }); }
|
|
}
|
|
}
|
|
onExit(code, signal) {
|
|
this.child = null;
|
|
if (this.stopping) { this.state = "stopped"; return; }
|
|
this.state = "failed";
|
|
this.emit("crash", { code, signal });
|
|
const now = Date.now();
|
|
this.restartTimes = this.restartTimes.filter((time) => now - time < this.restartWindowMs);
|
|
if (this.restartTimes.length >= this.maxRestarts) return;
|
|
this.restartTimes.push(now);
|
|
setTimeout(() => { try { this.start(); } catch (error) { this.emit("error", error); } }, 250).unref?.();
|
|
}
|
|
health() { return { healthy: this.state === "running", state: this.state, queue: this.queue.metrics(), restarts_in_window: this.restartTimes.length }; }
|
|
}
|
|
|
|
class WhisperCppServerProvider extends TranscriptionProvider {
|
|
constructor(supervisor) {
|
|
super(); this.worker = supervisor; this.model = null; this.sessions = new Set();
|
|
supervisor.on("message", (message) => this.emit(message.type || "message", message));
|
|
supervisor.on("crash", (event) => this.emit("provider_error", Object.assign(new Error("Whisper worker crashed."), { details: event })));
|
|
supervisor.on("error", (error) => this.emit("provider_error", error));
|
|
}
|
|
async loadModel(model) { this.worker.start(); this.model = model; this.worker.send({ type: "load_model", model }); return this.health(); }
|
|
async benchmark(options = {}) { this.worker.send({ type: "benchmark", options }); return { accepted: true, model: this.model?.id || null }; }
|
|
async startSession(session) { this.sessions.add(session.id); this.worker.send({ type: "start_session", session }); }
|
|
async addTrack(sessionId, track) { this.worker.send({ type: "add_track", session_id: sessionId, track }); }
|
|
async pushAudio(sessionId, trackId, frame) { return this.worker.send({ type: "audio", session_id: sessionId, track_id: trackId, sequence: frame.sequence, capture_timestamp_us: frame.capture_timestamp_us, captured_at: Date.now() }, frame.pcm); }
|
|
async removeTrack(sessionId, trackId) { this.worker.send({ type: "remove_track", session_id: sessionId, track_id: trackId }); }
|
|
async stopSession(sessionId) { this.sessions.delete(sessionId); this.worker.send({ type: "stop_session", session_id: sessionId }); }
|
|
async health() { return { provider: "whisper_cpp_server", model: this.model?.id || null, sessions: this.sessions.size, ...this.worker.health() }; }
|
|
async stop() { this.sessions.clear(); await this.worker.stop(); }
|
|
}
|
|
|
|
module.exports = { TranscriptionProvider, WhisperWorkerSupervisor, WhisperCppServerProvider };
|