diff --git a/dave/bot.mjs b/dave/bot.mjs index 6307ed4..e256688 100644 --- a/dave/bot.mjs +++ b/dave/bot.mjs @@ -19,6 +19,7 @@ // "Speak" voice permissions. See dave/README or run `node bot.mjs --invite`. import fs from 'node:fs'; +import { Readable } from 'node:stream'; import { Client, GatewayIntentBits, @@ -29,6 +30,10 @@ import { entersState, VoiceConnectionStatus, EndBehaviorType, + createAudioPlayer, + createAudioResource, + StreamType, + NoSubscriberBehavior, } from '@discordjs/voice'; import prism from 'prism-media'; @@ -47,10 +52,61 @@ const TOKEN = process.env.DISCORD_BOT_TOKEN || env.DISCORD_BOT_TOKEN; const GUILD_ID = process.env.GUILD_ID || env.GUILD_ID || '1352269198297923648'; const CHANNEL_ID = process.env.CHANNEL_ID || env.CHANNEL_ID || '1352269198914621465'; const RUN_MS = process.env.RUN_MS != null ? Number(process.env.RUN_MS) : 0; // 0 = stay forever +// Python STT+TTS voice-turn endpoint (run `python -m wsai --voice-server`). +const VOICE_ENDPOINT = process.env.WSAI_VOICE_ENDPOINT || env.WSAI_VOICE_ENDPOINT + || 'http://127.0.0.1:8787/api/voice-turn'; +// Ignore utterances shorter than this many PCM bytes (48kHz*2ch*2B = 192000 B/s), +// so key clicks / brief noise don't trigger a turn. ~0.35s. +const MIN_UTTERANCE_BYTES = Number(process.env.WSAI_MIN_UTTERANCE_BYTES || 67000); const t0 = Date.now(); const log = (...a) => console.log(`[+${String(Date.now() - t0).padStart(6)}ms]`, ...a); +// PCM s16le -> WAV container (so the Python side can ffmpeg-decode it). +function wavHeader(dataLen, sampleRate = 48000, channels = 2, bits = 16) { + const blockAlign = channels * bits / 8; + const b = Buffer.alloc(44); + b.write('RIFF', 0); b.writeUInt32LE(36 + dataLen, 4); b.write('WAVE', 8); + b.write('fmt ', 12); b.writeUInt32LE(16, 16); b.writeUInt16LE(1, 20); + b.writeUInt16LE(channels, 22); b.writeUInt32LE(sampleRate, 24); + b.writeUInt32LE(sampleRate * blockAlign, 28); b.writeUInt16LE(blockAlign, 32); + b.writeUInt16LE(bits, 34); b.write('data', 36); b.writeUInt32LE(dataLen, 40); + return b; +} + +// Speak audio into the voice channel. Feed the reply wav bytes through ffmpeg +// (StreamType.Arbitrary) so @discordjs/voice re-encodes to Opus. +let voicePlayer = null; +function playReply(wavBytes) { + if (!voicePlayer || !wavBytes || wavBytes.length === 0) return; + const resource = createAudioResource(Readable.from(wavBytes), { inputType: StreamType.Arbitrary }); + voicePlayer.play(resource); +} + +// One utterance: PCM -> WAV -> POST to Python -> play the reply back. +async function handleUtterance(userId, pcm) { + if (pcm.length < MIN_UTTERANCE_BYTES) { + log(`utterance too short user=${userId} bytes=${pcm.length} — skip`); + return; + } + const wav = Buffer.concat([wavHeader(pcm.length), pcm]); + let resp; + try { + resp = await fetch(VOICE_ENDPOINT, { + method: 'POST', headers: { 'Content-Type': 'audio/wav' }, body: wav, + }); + } catch (e) { + log(`voice-turn POST failed (is \`python -m wsai --voice-server\` running?): ${e.message}`); + return; + } + if (!resp.ok) { log(`voice-turn HTTP ${resp.status}`); return; } + const heard = decodeURIComponent(resp.headers.get('X-Heard') || ''); + const reply = decodeURIComponent(resp.headers.get('X-Reply') || ''); + const buf = Buffer.from(await resp.arrayBuffer()); + log(`heard="${heard}" reply="${reply}" replyWav=${buf.length}B`); + playReply(buf); +} + // ---------- invite URL helper ---------- // A bot cannot add itself to a server; a server admin must click an OAuth2 invite // URL once. Print it so the user can authorise the bot with voice permissions. @@ -118,24 +174,40 @@ client.once('clientReady', async () => { } log(`✅ JOINED & READY. channel=${CHANNEL_ID} — staying connected, listening for speakers…`); - // ---------- receive path (partial M2) ---------- + // Playback path (bot speaks): one player, subscribed to the connection. + voicePlayer = createAudioPlayer({ behaviors: { noSubscriber: NoSubscriberBehavior.Play } }); + voicePlayer.on('error', (e) => log(`player error: ${e.message}`)); + connection.subscribe(voicePlayer); + log(`voice endpoint: ${VOICE_ENDPOINT}`); + + // ---------- receive path: capture each utterance and run the voice turn ---- const receiver = connection.receiver; + const active = new Set(); // userIds with an in-flight subscription (avoid dupes) receiver.speaking.on('start', (userId) => { + if (userId === client.user.id || active.has(userId)) return; // skip self / dupes + active.add(userId); if (!perUser.has(userId)) perUser.set(userId, { opusPackets: 0, pcmFrames: 0 }); log(`SPEAKING start user=${userId}`); const opusStream = receiver.subscribe(userId, { - end: { behavior: EndBehaviorType.AfterSilence, duration: 500 }, + // End the utterance after a short silence so natural pauses don't cut words. + end: { behavior: EndBehaviorType.AfterSilence, duration: 800 }, }); - // Decode Opus -> 48kHz stereo s16le PCM (what faster-whisper will consume, downsampled to 16k in M3). + // Decode Opus -> 48kHz stereo s16le PCM, buffered until the utterance ends. const decoder = new prism.opus.Decoder({ rate: 48000, channels: 2, frameSize: 960 }); + const chunks = []; opusStream.on('data', () => { perUser.get(userId).opusPackets++; }); opusStream.pipe(decoder); - decoder.on('data', () => { - const s = perUser.get(userId); - s.pcmFrames++; - if (s.pcmFrames === 1 || s.pcmFrames % 200 === 0) log(`rtp: user=${userId} opus=${s.opusPackets} pcm=${s.pcmFrames}`); + decoder.on('data', (d) => { + chunks.push(d); + const s = perUser.get(userId); s.pcmFrames++; }); decoder.on('error', (e) => log(`decode error user=${userId}: ${e.message}`)); + decoder.on('end', () => { + active.delete(userId); + const pcm = Buffer.concat(chunks); + log(`utterance end user=${userId} pcm=${pcm.length}B — running voice turn`); + handleUtterance(userId, pcm).catch((e) => log(`voice turn error: ${e.message}`)); + }); }); connection.on(VoiceConnectionStatus.Disconnected, () => { diff --git a/wsai/__main__.py b/wsai/__main__.py index 7caba49..8b43563 100644 --- a/wsai/__main__.py +++ b/wsai/__main__.py @@ -100,6 +100,47 @@ def _run_stt_test(host: str, port: int) -> None: dash.stop() +def _run_voice_server(host: str, port: int) -> None: + """Serve the STT+TTS voice-turn endpoint that the Discord bot (dave/bot.mjs) + calls: it POSTs a captured utterance wav and gets back the reply wav to play + into the voice channel. Both STT and TTS run on the GPU and are pre-warmed. + The same page also shows the live turn feed. Echo mode for now (the reply is + what was heard); the Claude brain can be added as the next slice.""" + import time + + from .backends.melo import MeloTTS + from .backends.whisper import WhisperSTT + from .dashboard import Dashboard + from .monitor import Monitor + + monitor = Monitor() + stt = WhisperSTT() + tts = MeloTTS() + dash = Dashboard(monitor, host=host, port=port, stt=stt, tts=tts) + dash.start() + monitor.set_components({"source": "none", "vision": "none", "stt": "whisper", + "brain": "echo", "tts": "melo"}) + monitor.set_status(running=True, listening=False) + monitor.log("info", "디스코드 음성 서버 시작 — STT+TTS GPU 워밍업 중…") + print("\n STT+TTS 워밍업 중… (모델 로드 + CUDA 예열)") + dash.warm() + sdev = getattr(stt, "resolved_device", None) or "?" + monitor.set_status(listening=True) + monitor.log("info", f"음성 서버 준비 완료 (STT device={sdev}). 디스코드 봇 연결 대기.") + + shown = host if host not in ("0.0.0.0", "") else _lan_ip() + print(f"\n 음성 서버 준비 완료 (STT device: {sdev})") + print(f" 대시보드/상태: http://{shown}:{port}") + print(f" 봇 연결 엔드포인트: http://127.0.0.1:{port}/api/voice-turn\n") + try: + while True: + time.sleep(3600) + except KeyboardInterrupt: + pass + finally: + dash.stop() + + def main() -> None: ap = argparse.ArgumentParser(prog="wsai") ap.add_argument("--voice", action="store_true", help="eyes-free voice loop (STT -> Brain -> TTS)") @@ -108,6 +149,8 @@ def main() -> None: help="keep generating mock demo utterances forever (off by default)") ap.add_argument("--stt-test", action="store_true", help="serve the dashboard with a live GPU STT recognition test") + ap.add_argument("--voice-server", action="store_true", + help="serve STT+TTS voice-turn endpoint for the Discord bot (dave/bot.mjs)") ap.add_argument("--live", action="store_true", help="capture screen + Claude backends") ap.add_argument("--env", action="store_true", help="build from WSAI_* env vars") ap.add_argument("--port", type=int, default=int(os.environ.get("WSAI_DASHBOARD_PORT", "8787")), @@ -126,6 +169,10 @@ def main() -> None: _run_stt_test(args.host, args.port) return + if args.voice_server: + _run_voice_server(args.host, args.port) + return + if args.dashboard: # Default to the eyes-free voice preset for the demo; env can override. settings = Settings.from_env() if args.env else Settings.voice() diff --git a/wsai/backends/melo.py b/wsai/backends/melo.py index 9a3f782..7b0a249 100644 --- a/wsai/backends/melo.py +++ b/wsai/backends/melo.py @@ -131,11 +131,13 @@ class MeloTTS: self.load_ms = info.get("ms") log.info("melo worker ready in %s ms on %s", self.load_ms, info.get("device")) - async def speak(self, reply: Reply) -> None: + async def synth(self, text: str) -> str: + """Synthesize `text` to a wav and return its path (no sink). Reusable by + callers that want the wav directly (e.g. the Discord voice bridge).""" await self._ensure() self._n += 1 out = str(self.out_dir / f"tts-{self._n:06d}.wav") - req = json.dumps({"text": reply.text, "out": out, "speed": self.speed}) + req = json.dumps({"text": text, "out": out, "speed": self.speed}) s = time.monotonic() async with self._lock: assert self._proc and self._proc.stdin and self._proc.stdout @@ -148,7 +150,11 @@ class MeloTTS: if not res.get("ok"): raise RuntimeError(f"melo synth failed: {res.get('error')}") log.debug("synth %d ms (worker %s ms)", int((time.monotonic() - s) * 1000), res.get("ms")) - await self.sink(res["out"], reply) + return res["out"] + + async def speak(self, reply: Reply) -> None: + out = await self.synth(reply.text) + await self.sink(out, reply) async def aclose(self) -> None: if self._proc is not None and self._proc.returncode is None: diff --git a/wsai/dashboard.py b/wsai/dashboard.py index 3905a14..96b05cc 100644 --- a/wsai/dashboard.py +++ b/wsai/dashboard.py @@ -58,9 +58,51 @@ def _make_handler(dash: "Dashboard"): path = self.path.split("?", 1)[0] if path == "/api/stt": self._handle_stt() + elif path == "/api/voice-turn": + self._handle_voice_turn() else: self._send(404, b"not found", "text/plain; charset=utf-8") + def _read_body(self) -> bytes: + try: + length = int(self.headers.get("Content-Length", "0")) + except ValueError: + length = 0 + return self.rfile.read(length) if length > 0 else b"" + + def _handle_voice_turn(self) -> None: + """Discord voice bridge: utterance wav in -> reply wav out. The + recognised/reply text ride along as URL-encoded response headers so + the bot can log them; the body is the reply audio to play back.""" + import urllib.parse + + if dash.stt is None or dash.tts is None: + self._send(503, json.dumps({"ok": False, "error": "voice loop not enabled"}).encode(), + "application/json; charset=utf-8") + return + raw = self._read_body() + if not raw: + self._send(400, json.dumps({"ok": False, "error": "empty upload"}).encode(), + "application/json; charset=utf-8") + return + try: + res = dash.voice_turn(raw) + except Exception as exc: # noqa: BLE001 + log.exception("voice-turn failed") + self._send(500, json.dumps({"ok": False, "error": f"{type(exc).__name__}: {exc}"}, + ensure_ascii=False).encode(), "application/json; charset=utf-8") + return + body = res["wav"] + self.send_response(200 if body else 204) + self.send_header("Content-Type", "audio/wav") + self.send_header("Content-Length", str(len(body))) + self.send_header("X-Heard", urllib.parse.quote(res.get("heard", ""))) + self.send_header("X-Reply", urllib.parse.quote(res.get("reply", ""))) + self.send_header("Cache-Control", "no-store") + self.end_headers() + if body: + self.wfile.write(body) + def _handle_stt(self) -> None: """Accept an uploaded audio blob (mic recording or file), run it through the real GPU STT, and return the recognised text.""" @@ -130,18 +172,19 @@ class Dashboard: """ def __init__(self, monitor: Monitor, host: str = "0.0.0.0", port: int = 8787, - stt=None) -> None: + stt=None, tts=None) -> None: self.monitor = monitor self.host = host self.port = port self.stt = stt + self.tts = tts self._server: ThreadingHTTPServer | None = None self._thread: threading.Thread | None = None self._loop = None self._loop_thread: threading.Thread | None = None def start(self) -> None: - if self.stt is not None: + if self.stt is not None or self.tts is not None: self._start_loop() handler = _make_handler(self) self._server = ThreadingHTTPServer((self.host, self.port), handler) @@ -178,10 +221,72 @@ class Dashboard: return fut.result(timeout=timeout) def warm(self) -> None: - """Pre-start the STT worker (loads + warms the GPU) so the first web - recognition is instant instead of paying model-load + CUDA autotune.""" + """Pre-start the STT/TTS workers (loads + warms the GPU) so the first + recognition/synth is instant instead of paying model-load + CUDA autotune.""" if self.stt is not None: self._submit(self.stt._ensure()) + if self.tts is not None: + self._submit(self.tts._ensure()) + + def voice_turn(self, audio_bytes: bytes) -> dict: + """One Discord voice turn: decode the uploaded utterance, recognise it + on the GPU, produce a reply (echo of what was heard for now), synthesise + it on the GPU, and return {heard, reply, wav} where wav is the reply + audio bytes for the bot to play back into the channel.""" + import os + import subprocess + import time + import uuid + + if self.stt is None or self.tts is None: + raise RuntimeError("voice_turn needs both STT and TTS") + updir = os.path.expanduser("~/.cache/wsai/uploads") + os.makedirs(updir, exist_ok=True) + stem = os.path.join(updir, uuid.uuid4().hex) + src, wav = stem + ".bin", stem + ".wav" + with open(src, "wb") as f: + f.write(audio_bytes) + turn = self.monitor.turn(source="discord") + t0 = time.monotonic() + try: + subprocess.run( + ["ffmpeg", "-y", "-i", src, "-ar", "16000", "-ac", "1", wav], + check=True, capture_output=True, + ) + heard = self._submit(self.stt.transcribe(wav)) or "" + turn.heard(heard or "(빈 결과)") + reply_text = heard.strip() + if not reply_text: + # Nothing recognised (silence/noise): skip TTS, tell the bot. + turn.finish() + return {"heard": heard, "reply": "", "wav": b""} + out_path = self._submit(self.tts.synth(reply_text)) + with open(out_path, "rb") as f: + reply_wav = f.read() + ms = int((time.monotonic() - t0) * 1000) + turn.replied(reply_text) + step = turn.step("STT+TTS(GPU)") + step.ok, step.ms = True, float(ms) + turn._steps.append(step) + turn.finish() + try: + os.remove(out_path) + except OSError: + pass + return {"heard": heard, "reply": reply_text, "wav": reply_wav} + except subprocess.CalledProcessError as exc: + turn.finish(error="ffmpeg decode failed") + err = exc.stderr.decode("utf-8", "replace")[-300:] if exc.stderr else str(exc) + raise RuntimeError(f"ffmpeg: {err}") from exc + except Exception as exc: + turn.finish(error=str(exc)) + raise + finally: + for p in (src, wav): + try: + os.remove(p) + except OSError: + pass def transcribe_upload(self, audio_bytes: bytes) -> dict: """ffmpeg-normalise an uploaded blob to 16 kHz mono wav, transcribe it