462 lines
16 KiB
TypeScript
462 lines
16 KiB
TypeScript
/*
|
|
* Veskslop daemon — TTS queue, ducking, capture, and transcription.
|
|
*
|
|
* Lines play one at a time. While another person is speaking, the virtual sink
|
|
* is ducked. Capture reads the client output, splits it into utterances, stores
|
|
* them in the transcript, and runs speech to text when a provider is set.
|
|
*/
|
|
|
|
import { statSync, unlinkSync } from "node:fs";
|
|
import { AudioManager } from "./audio.js";
|
|
import type { CaptureSegment } from "./capture.js";
|
|
import { CapturePipeline } from "./capture.js";
|
|
import type { Stt } from "./stt.js";
|
|
import { Tts, type TtsOptions, type TtsProvider, type TtsVoice } from "./tts.js";
|
|
import type { TranscriptSegment, TranscriptSession, TranscriptStore } from "./transcript.js";
|
|
import { DEFAULT_VAD, type VadConfig } from "./vad.js";
|
|
|
|
export interface AudioServiceOptions {
|
|
cacheDir?: string;
|
|
transcripts: TranscriptStore;
|
|
stt: Stt;
|
|
vad?: VadConfig;
|
|
autoCapture?: boolean;
|
|
getSelfId: () => Promise<string | undefined>;
|
|
getVoiceContext?: () => Promise<{ guildId?: string | null; channelId?: string | null } | undefined>;
|
|
getSpeakerName?: (userId: string) => Promise<string | undefined>;
|
|
onTranscript?: (event: Record<string, unknown>) => void;
|
|
/** Called true when a TTS burst starts and false when the queue drains. */
|
|
onSpeakActivity?: (active: boolean) => void;
|
|
}
|
|
|
|
export interface SpeakResult {
|
|
queued: boolean;
|
|
id: number;
|
|
position: number;
|
|
}
|
|
|
|
interface SpeakerInterval {
|
|
userId: string;
|
|
name?: string;
|
|
start: number;
|
|
end?: number;
|
|
}
|
|
|
|
export class AudioService {
|
|
readonly manager = new AudioManager();
|
|
private readonly tts: Tts;
|
|
private readonly transcripts: TranscriptStore;
|
|
private readonly stt: Stt;
|
|
private readonly vad: VadConfig;
|
|
private readonly queue: Array<{ id: number; text: string; options: TtsOptions }> = [];
|
|
private readonly speakers = new Set<string>();
|
|
private readonly speakerIntervals: SpeakerInterval[] = [];
|
|
private readonly speakerNames = new Map<string, string>();
|
|
private playing = false;
|
|
/** True while a TTS burst holds the voice-processing bypass. */
|
|
private bypassed = false;
|
|
private duckEnabled = true;
|
|
private nextId = 1;
|
|
private ducked = false;
|
|
private selfId?: string;
|
|
private capture?: CapturePipeline;
|
|
private captureAdoptTimer?: ReturnType<typeof setInterval>;
|
|
private session?: TranscriptSession;
|
|
private segmentCount = 0;
|
|
private initialized = false;
|
|
|
|
constructor(private readonly options: AudioServiceOptions) {
|
|
this.tts = new Tts(options.cacheDir);
|
|
this.transcripts = options.transcripts;
|
|
this.stt = options.stt;
|
|
this.vad = options.vad ?? DEFAULT_VAD;
|
|
}
|
|
|
|
async init(): Promise<void> {
|
|
if (this.initialized) return;
|
|
this.initialized = true;
|
|
|
|
await this.manager.detect();
|
|
await this.tts.detect();
|
|
await this.stt.detect();
|
|
if (!this.manager.status().available) {
|
|
console.warn(`[audio] unavailable: ${this.manager.status().error}`);
|
|
return;
|
|
}
|
|
|
|
try {
|
|
await this.manager.ensureDevices();
|
|
console.log(
|
|
`[audio] devices ready: sink=${this.manager.status().sink} source=${this.manager.status().source}`
|
|
);
|
|
} catch (error) {
|
|
console.warn("[audio] device setup failed:", (error as Error).message);
|
|
}
|
|
}
|
|
|
|
status() {
|
|
return {
|
|
...this.manager.status(),
|
|
tts: this.tts.status(),
|
|
stt: this.stt.status(),
|
|
queue: this.queue.length,
|
|
speaking: this.playing,
|
|
duckEnabled: this.duckEnabled,
|
|
capturing: !!this.capture,
|
|
sessionId: this.session?.id ?? null,
|
|
segments: this.segmentCount,
|
|
transcript: this.transcripts.stats()
|
|
};
|
|
}
|
|
|
|
/** Enable or disable ducking the TTS while others speak. */
|
|
setDuckEnabled(on: boolean): void {
|
|
this.duckEnabled = on;
|
|
void this.applyDuck(this.shouldDuck());
|
|
}
|
|
|
|
private shouldDuck(): boolean {
|
|
return this.duckEnabled && this.playing && this.speakers.size > 0;
|
|
}
|
|
|
|
async speak(text: string, options: TtsOptions = {}): Promise<SpeakResult> {
|
|
await this.init();
|
|
|
|
const manager = this.manager.status();
|
|
if (!manager.available) throw new Error(manager.error ?? "audio is not available");
|
|
|
|
const tts = this.tts.status();
|
|
const provider = options.provider ?? tts.defaultProvider;
|
|
if (provider === "api" && !tts.api) {
|
|
throw new Error("API TTS is not configured (VESKSLOP_TTS_API_URL)");
|
|
}
|
|
if (provider === "command" && !tts.command) {
|
|
throw new Error("the custom TTS engine is not configured (VESKSLOP_TTS_CMD)");
|
|
}
|
|
if (provider === "espeak" && !tts.espeak) {
|
|
throw new Error("espeak-ng is not installed");
|
|
}
|
|
if (provider === "local" && !tts.local) {
|
|
throw new Error("local TTS is not configured (VESKSLOP_TTS_CMD)");
|
|
}
|
|
|
|
const trimmed = text.trim();
|
|
if (!trimmed) throw new Error("text is required");
|
|
if (trimmed.length > 1000) throw new Error("text is too long (max 1000 characters)");
|
|
|
|
const id = this.nextId++;
|
|
this.queue.push({ id, text: trimmed, options: { ...options, provider } });
|
|
void this.pump();
|
|
|
|
return { queued: true, id, position: this.queue.length };
|
|
}
|
|
|
|
/**
|
|
* Wait until the wrapper's streaming file has real audio in it, then play it while it
|
|
* grows. The wrapper touches `<file>.ready` after the first word, which is the reliable
|
|
* signal; a size threshold and a timeout cover providers that do not mark readiness.
|
|
*/
|
|
private async playWhenReady(file: string, volume?: number): Promise<void> {
|
|
for (let attempt = 0; attempt < 400; attempt++) {
|
|
try {
|
|
if (statSync(`${file}.ready`).size >= 0) break;
|
|
} catch {
|
|
// no marker yet
|
|
}
|
|
try {
|
|
if (statSync(file).size > 100_000) break;
|
|
} catch {
|
|
// not created yet
|
|
}
|
|
await new Promise(resolve => setTimeout(resolve, 25));
|
|
}
|
|
|
|
try {
|
|
await this.manager.play(file, volume);
|
|
} finally {
|
|
try {
|
|
unlinkSync(`${file}.ready`);
|
|
} catch {
|
|
// best effort
|
|
}
|
|
}
|
|
}
|
|
|
|
/** The TTS output directory, for serving preview files. */
|
|
get ttsDirectory(): string {
|
|
return this.tts.directory;
|
|
}
|
|
|
|
/**
|
|
* Synthesize a line for browser playback. This never touches the call:
|
|
* no queue, no ducking, no virtual microphone, no voice-processing bypass.
|
|
*/
|
|
async preview(text: string, options: TtsOptions = {}): Promise<string> {
|
|
const trimmed = text.trim();
|
|
if (!trimmed) throw new Error("text is required");
|
|
if (trimmed.length > 1000) throw new Error("text is too long (max 1000 characters)");
|
|
|
|
return this.tts.synthesize(trimmed, { ...options, stream: false, onOutputPath: undefined });
|
|
}
|
|
|
|
stop(): void {
|
|
this.queue.length = 0;
|
|
this.manager.stopPlayback();
|
|
}
|
|
|
|
/** Track speakers for ducking and attribution, and auto-capture calls. */
|
|
async onEvent(event: any): Promise<void> {
|
|
if (event?.type === "VOICE_STATE_UPDATE") {
|
|
if (this.options.autoCapture) await this.handleVoiceState(event);
|
|
return;
|
|
}
|
|
|
|
if (event?.type !== "VOICE_SPEAKING") return;
|
|
|
|
if (!this.selfId) this.selfId = await this.options.getSelfId();
|
|
if (this.selfId && event.userId === this.selfId) return;
|
|
|
|
const userId = String(event.userId);
|
|
|
|
if (event.speaking) {
|
|
this.speakers.add(userId);
|
|
this.speakerIntervals.push({ userId, name: await this.speakerName(userId), start: Date.now() });
|
|
} else {
|
|
this.speakers.delete(userId);
|
|
const interval = [...this.speakerIntervals].reverse().find(item => item.userId === userId && !item.end);
|
|
if (interval) interval.end = Date.now();
|
|
}
|
|
|
|
this.pruneIntervals();
|
|
await this.applyDuck(this.shouldDuck());
|
|
}
|
|
|
|
async startCapture(): Promise<{ sessionId: string }> {
|
|
if (this.capture) throw new Error("capture is already running");
|
|
await this.init();
|
|
|
|
const status = this.manager.status();
|
|
if (!status.available) throw new Error(status.hint ?? status.error ?? "audio is not available");
|
|
if (!status.ffmpeg) throw new Error("ffmpeg is missing; install it to capture audio");
|
|
|
|
const context = (await this.options.getVoiceContext?.()) ?? {};
|
|
const session = this.transcripts.startSession(context);
|
|
this.segmentCount = 0;
|
|
|
|
this.capture = new CapturePipeline({
|
|
ffmpeg: process.env.VESKSLOP_FFMPEG ?? "ffmpeg",
|
|
target: `${status.captureSink}.monitor`,
|
|
directory: `${this.tts.directory}/captures`,
|
|
sessionId: session.id,
|
|
vad: this.vad,
|
|
onSegment: segment => this.handleSegment(session, segment),
|
|
onError: error => {
|
|
console.error("[capture] stream error:", error.message);
|
|
this.options.onTranscript?.({ type: "TRANSCRIPT_ERROR", error: error.message });
|
|
}
|
|
});
|
|
|
|
this.capture.start();
|
|
void this.manager.adoptClientStreams();
|
|
this.captureAdoptTimer = setInterval(() => void this.manager.adoptClientStreams(), 5000);
|
|
this.session = session;
|
|
this.options.onTranscript?.({ type: "TRANSCRIPT_SESSION", session });
|
|
|
|
return { sessionId: session.id };
|
|
}
|
|
|
|
stopCapture(): void {
|
|
clearInterval(this.captureAdoptTimer);
|
|
this.captureAdoptTimer = undefined;
|
|
this.capture?.stop();
|
|
this.capture = undefined;
|
|
|
|
if (this.session) {
|
|
this.transcripts.endSession(this.session.id);
|
|
this.options.onTranscript?.({ type: "TRANSCRIPT_SESSION_END", id: this.session.id });
|
|
this.session = undefined;
|
|
}
|
|
|
|
this.speakerIntervals.length = 0;
|
|
}
|
|
|
|
/** Stream call audio to the browser. Returns a stop function. */
|
|
startMonitor(onData: (chunk: Buffer) => void, onError?: (error: Error) => void): () => void {
|
|
return this.manager.startMonitor(onData, onError);
|
|
}
|
|
|
|
ttsVoices(engine?: TtsProvider): Promise<TtsVoice[]> {
|
|
return this.tts.voices(engine);
|
|
}
|
|
|
|
warmTts(engine?: TtsProvider) {
|
|
return this.tts.warm(engine);
|
|
}
|
|
|
|
async dispose(): Promise<void> {
|
|
this.queue.length = 0;
|
|
this.stopCapture();
|
|
await this.manager.dispose();
|
|
}
|
|
|
|
private async handleVoiceState(event: any): Promise<void> {
|
|
if (!this.selfId) this.selfId = await this.options.getSelfId();
|
|
if (!this.selfId || event.userId !== this.selfId) return;
|
|
|
|
if (event.channelId && !this.capture) {
|
|
console.log("[capture] auto-start for", event.channelId);
|
|
await this.startCapture().catch(error => console.warn("[capture] auto-start failed:", (error as Error).message));
|
|
} else if (!event.channelId && this.capture) {
|
|
console.log("[capture] auto-stop");
|
|
this.stopCapture();
|
|
}
|
|
}
|
|
|
|
private handleSegment(session: TranscriptSession, segment: CaptureSegment) {
|
|
this.segmentCount++;
|
|
|
|
const speaker = this.dominantSpeaker(segment.startedAt, segment.endedAt);
|
|
const record: TranscriptSegment = {
|
|
id: `segment-${Date.now()}-${this.segmentCount}`,
|
|
sessionId: session.id,
|
|
startedAt: segment.startedAt,
|
|
endedAt: segment.endedAt,
|
|
durationMs: segment.durationMs,
|
|
speakerId: speaker?.userId ?? null,
|
|
speakerName: speaker?.name ?? null,
|
|
wav: segment.file,
|
|
text: null
|
|
};
|
|
|
|
this.transcripts.addSegment(record);
|
|
this.options.onTranscript?.({ type: "TRANSCRIPT_SEGMENT", segment: record });
|
|
|
|
void this.transcribe(record);
|
|
}
|
|
|
|
private async transcribe(record: TranscriptSegment) {
|
|
if (!record.wav) return;
|
|
|
|
try {
|
|
const text = await this.stt.transcribe(record.wav);
|
|
if (!text) {
|
|
console.warn(`[capture] transcription empty for ${record.wav}`);
|
|
return;
|
|
}
|
|
|
|
this.transcripts.setSegmentText(record.id, text);
|
|
this.options.onTranscript?.({ type: "TRANSCRIPT_TEXT", id: record.id, text });
|
|
} catch (error) {
|
|
console.error("[capture] transcription failed:", (error as Error).message);
|
|
this.options.onTranscript?.({ type: "TRANSCRIPT_ERROR", error: (error as Error).message });
|
|
}
|
|
}
|
|
|
|
private dominantSpeaker(startedAt: string, endedAt: string): SpeakerInterval | undefined {
|
|
const start = Date.parse(startedAt);
|
|
const end = Date.parse(endedAt);
|
|
const totals = new Map<string, { ms: number; interval: SpeakerInterval }>();
|
|
|
|
for (const interval of this.speakerIntervals) {
|
|
const overlap = Math.min(end, interval.end ?? end) - Math.max(start, interval.start);
|
|
if (overlap <= 0) continue;
|
|
|
|
const entry = totals.get(interval.userId) ?? { ms: 0, interval };
|
|
entry.ms += overlap;
|
|
totals.set(interval.userId, entry);
|
|
}
|
|
|
|
let best: SpeakerInterval | undefined;
|
|
let bestMs = 0;
|
|
for (const entry of totals.values()) {
|
|
if (entry.ms > bestMs) {
|
|
best = entry.interval;
|
|
bestMs = entry.ms;
|
|
}
|
|
}
|
|
|
|
return best;
|
|
}
|
|
|
|
private async speakerName(userId: string): Promise<string | undefined> {
|
|
if (this.speakerNames.has(userId)) return this.speakerNames.get(userId);
|
|
const name = await this.options.getSpeakerName?.(userId).catch(() => undefined);
|
|
if (name) this.speakerNames.set(userId, name);
|
|
return name;
|
|
}
|
|
|
|
private pruneIntervals() {
|
|
const cutoff = Date.now() - 60_000;
|
|
while (this.speakerIntervals.length && this.speakerIntervals[0].end && this.speakerIntervals[0].end < cutoff) {
|
|
this.speakerIntervals.shift();
|
|
}
|
|
}
|
|
|
|
private async pump(): Promise<void> {
|
|
if (this.playing) return;
|
|
|
|
const item = this.queue.shift();
|
|
if (!item) {
|
|
await this.applyDuck(false);
|
|
return;
|
|
}
|
|
|
|
this.playing = true;
|
|
if (!this.bypassed) {
|
|
this.bypassed = true;
|
|
try {
|
|
this.options.onSpeakActivity?.(true);
|
|
} catch {
|
|
// Best effort: speech still plays.
|
|
}
|
|
}
|
|
try {
|
|
// Streaming word mode: the wrapper writes the WAV while it synthesizes, so
|
|
// playback starts on the first word instead of after the whole line.
|
|
let streaming: Promise<void> | null = null;
|
|
const options: TtsOptions = { ...item.options };
|
|
if (item.options.stream === true) {
|
|
options.onOutputPath = path => {
|
|
streaming = (async () => {
|
|
await this.applyDuck(this.shouldDuck());
|
|
await this.playWhenReady(path, item.options.volume);
|
|
})();
|
|
};
|
|
}
|
|
|
|
const file = await this.tts.synthesize(item.text, options);
|
|
const pending: Promise<void> | null = streaming;
|
|
if (pending) {
|
|
await pending;
|
|
} else {
|
|
await this.applyDuck(this.shouldDuck());
|
|
await this.manager.play(file, item.options.volume);
|
|
}
|
|
} catch (error) {
|
|
console.error("[audio] speak failed:", (error as Error).message);
|
|
} finally {
|
|
this.playing = false;
|
|
await this.applyDuck(false);
|
|
if (this.bypassed && this.queue.length === 0) {
|
|
this.bypassed = false;
|
|
try {
|
|
this.options.onSpeakActivity?.(false);
|
|
} catch {
|
|
// Best effort.
|
|
}
|
|
}
|
|
void this.pump();
|
|
}
|
|
}
|
|
|
|
private async applyDuck(on: boolean): Promise<void> {
|
|
if (on === this.ducked) return;
|
|
try {
|
|
await this.manager.setDuck(on);
|
|
this.ducked = on;
|
|
} catch (error) {
|
|
console.warn("[audio] duck failed:", (error as Error).message);
|
|
}
|
|
}
|
|
}
|