diff --git a/NOTICE b/NOTICE index 8693d47..8c6e112 100644 --- a/NOTICE +++ b/NOTICE @@ -67,3 +67,22 @@ The browser component is also published as https://github.com/TencentCloud/octop-browser Copyright (c) 2026 Octop Licensed under the MIT License. + +Jonathan Ai voice is adapted from Octop's voice manager: an active +speech-to-text provider, an active text-to-speech provider, and a single +place that dispatches a recording or a spoken reply. Jonathan Ai does not +vendor Octop's browser, Edge, Tencent, or MiMo presets. The default +speech-to-text engine is local faster-whisper and the default voice is +local Piper. OpenAI and xAI are optional and run only after they are +selected. Microphones and speakers, including Bluetooth headsets, are +chosen from the Windows WASAPI devices or the Linux PulseAudio and +PipeWire devices. Push-to-talk is the default. The wake word is off until +the user turns it on, and that check stays on this machine. Model files +are fetched from Hugging Face or the official Piper and Whisper sources +and checked with SHA-256. The optional install extra is voice +(pip install "clawd-codex[voice]"). + +Octop voice +https://github.com/TencentCloud/Octop +Copyright (c) 2026 Octop +Licensed under the MIT License. diff --git a/pyproject.toml b/pyproject.toml index 36a1c37..d8c3b5a 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -43,6 +43,11 @@ dev = [ browser = [ "websocket-client>=1.7.0", ] +voice = [ + "faster-whisper>=1.0.0", + "sounddevice>=0.4.6", + "piper-tts>=1.2.0", +] [project.scripts] clawd = "src.cli:main" diff --git a/src/cli.py b/src/cli.py index fe0aa45..0068cc9 100644 --- a/src/cli.py +++ b/src/cli.py @@ -34,6 +34,7 @@ def main(): clawd memory delete Delete one local memory clawd cron list List local scheduled jobs clawd cron run Run one scheduled job now + clawd voice settings Show local speech settings clawd Start interactive REPL """ ) @@ -106,6 +107,9 @@ def main(): cron_logs = cron_sub.add_parser('logs', help='Show scheduler logs') cron_logs.add_argument('job_id', nargs='?') + voice_parser = subparsers.add_parser('voice', help='Local speech, push-to-talk, and devices') + voice_parser.add_argument('words', nargs='*', help='settings, devices, stt, tts, ptt, wake, say') + args = parser.parse_args() # Handle --version @@ -183,6 +187,8 @@ def main(): return handle_memory(args) elif args.command == 'cron': return handle_cron(args) + elif args.command == 'voice': + return handle_voice(args) # Default: start REPL return start_repl(stream=args.stream) @@ -293,6 +299,15 @@ def handle_cron(args) -> int: return 0 +def handle_voice(args) -> int: + """Show or change speech settings stored on this machine.""" + from src.voice.service import format_voice_command + + words = getattr(args, "words", None) or [] + print(format_voice_command(" ".join(words))) + return 0 + + def start_repl(stream: bool = False): """Start interactive REPL.""" from src.config import get_default_provider diff --git a/src/command_system/__init__.py b/src/command_system/__init__.py index 724de85..43bb88b 100644 --- a/src/command_system/__init__.py +++ b/src/command_system/__init__.py @@ -15,6 +15,7 @@ CRON_COMMAND, INIT_COMMAND, MEMORY_COMMAND, + VOICE_COMMAND, SKILLS_COMMAND, execute_command_async, execute_command_sync, @@ -94,6 +95,7 @@ "COMPACT_COMMAND", "CRON_COMMAND", "MEMORY_COMMAND", + "VOICE_COMMAND", "INIT_COMMAND", "get_builtin_commands", "register_builtin_commands", diff --git a/src/command_system/builtins.py b/src/command_system/builtins.py index 7123db3..6ffefc7 100644 --- a/src/command_system/builtins.py +++ b/src/command_system/builtins.py @@ -636,6 +636,22 @@ def cron_command_call(args: str, context: CommandContext) -> LocalCommandResult: return LocalCommandResult(type="text", value=format_cron_command(args)) + +VOICE_COMMAND = LocalCommand( + name="voice", + description="Listen and speak on this machine, or choose a cloud voice", + argument_hint="settings | devices | stt local|openai|xai | tts local|openai|xai | ptt on|off | wake on|off | say ", + supports_non_interactive=True, +) + + +def voice_command_call(args: str, context: CommandContext) -> LocalCommandResult: + """Show or change local speech settings. Audio stays here unless a cloud provider is selected.""" + del context + from ..voice.service import format_voice_command + + return LocalCommandResult(type="text", value=format_voice_command(args)) + INIT_COMMAND = PromptCommand( name="init", description="Initialize new CLAUDE.md file(s) and optional skills/hooks with codebase documentation", @@ -684,6 +700,8 @@ def execute_command_sync(cmd_name: str, args: str, context: CommandContext) -> t result = memory_command_call(args, context) elif cmd is CRON_COMMAND: result = cron_command_call(args, context) + elif cmd is VOICE_COMMAND: + result = voice_command_call(args, context) else: return False, None, f"Command not implemented for sync execution: {cmd_name}" @@ -702,6 +720,7 @@ def execute_command_sync(cmd_name: str, args: str, context: CommandContext) -> t COMPACT_COMMAND.set_call(compact_command_call) MEMORY_COMMAND.set_call(memory_command_call) CRON_COMMAND.set_call(cron_command_call) +VOICE_COMMAND.set_call(voice_command_call) def get_builtin_commands() -> list[Command]: @@ -716,6 +735,7 @@ def get_builtin_commands() -> list[Command]: COMPACT_COMMAND, MEMORY_COMMAND, CRON_COMMAND, + VOICE_COMMAND, INIT_COMMAND, ] diff --git a/src/desktop/runtime.py b/src/desktop/runtime.py index 0469946..b94cafe 100644 --- a/src/desktop/runtime.py +++ b/src/desktop/runtime.py @@ -1173,6 +1173,7 @@ def _format_palette(self) -> str: - `/memory` — list, recall, or delete local memories (`/memory delete `) - `/cron` — list, run, enable, disable, or delete local schedules (`/cron run `) - Browser tool — local Chrome over CDP. Logins stay in the profile. Submits, purchases, and logins need approval. +- `/voice` — local speech and speakers (`/voice settings`, `/voice say hello`) Type a message to run the existing agent loop. Destructive and network tools prompt for approval before they run. diff --git a/src/desktop/server.py b/src/desktop/server.py index 998ac60..5822b3e 100644 --- a/src/desktop/server.py +++ b/src/desktop/server.py @@ -231,6 +231,18 @@ def handle_api(self, method: str, path: str, query: dict[str, list[str]], body: return 200, runtime.update_status(refresh=True) if path == "/api/update/apply" and method == "POST": return 200, runtime.apply_update() + if path == "/api/voice" and method == "GET": + from src.voice.service import voice_status + + return 200, voice_status() + if path == "/api/voice" and method == "POST": + from src.voice.service import apply_voice_settings + + return 200, apply_voice_settings(body) + if path == "/api/voice/ptt" and method == "POST": + from src.voice.service import push_to_talk_action + + return 200, push_to_talk_action(str(body.get("action") or "")) return 404, {"error": f"unknown route: {method} {path}"} def serve_forever(self) -> None: diff --git a/src/desktop/web/app.js b/src/desktop/web/app.js index 0fb28cd..788e35c 100644 --- a/src/desktop/web/app.js +++ b/src/desktop/web/app.js @@ -679,6 +679,7 @@ function applyRoute() { const showSetup = route === "setup" || (route === "chat" && Boolean(state.status?.needs_setup)); $("setup-modal")?.classList.toggle("hidden", !showSetup); if (route === "library") refreshLibrary().catch(reportError); + if (route === "settings") loadVoiceSettings().catch(reportError); } async function refreshLibrary() { @@ -775,7 +776,90 @@ function renderScheduleList(items) { } } +function fillVoiceDevices(select, devices, kind, selected) { + select.innerHTML = ""; + const fallback = document.createElement("option"); + fallback.value = ""; + fallback.textContent = kind === "input" ? "Default microphone" : "Default speaker"; + select.appendChild(fallback); + for (const device of devices.filter((item) => item.kind === kind)) { + const option = document.createElement("option"); + option.value = device.id; + const marks = [device.host_api]; + if (device.bluetooth) marks.push("Bluetooth"); + if (device.default) marks.push("default"); + option.textContent = `${device.name} (${marks.join(", ")})`; + select.appendChild(option); + } + select.value = selected || ""; +} + +async function loadVoiceSettings() { + try { + const data = await api("/api/voice"); + const settings = data.settings || {}; + $("voice-stt").value = settings.stt_provider || "local"; + $("voice-tts").value = settings.tts_provider || "local"; + $("voice-ptt-enabled").checked = settings.push_to_talk !== false; + $("voice-wake").checked = Boolean(settings.wake_word_enabled); + $("voice-wake-phrase").value = settings.wake_word || "jonathan"; + fillVoiceDevices($("voice-input"), data.devices || [], "input", settings.input_device); + fillVoiceDevices($("voice-output"), data.devices || [], "output", settings.output_device); + $("voice-note").textContent = settings.note || data.device_error || ""; + } catch (err) { + $("voice-note").textContent = err.message; + } +} + +async function saveVoiceSettings() { + const data = await api("/api/voice", { + method: "POST", + body: JSON.stringify({ + stt_provider: $("voice-stt").value, + tts_provider: $("voice-tts").value, + input_device: $("voice-input").value, + output_device: $("voice-output").value, + push_to_talk: $("voice-ptt-enabled").checked, + wake_word_enabled: $("voice-wake").checked, + wake_word: $("voice-wake-phrase").value, + }), + }); + $("voice-note").textContent = data.settings?.note || ""; +} + +function bindVoiceTalk() { + const button = $("voice-ptt"); + if (!button) return; + const start = async (event) => { + event.preventDefault(); + button.classList.add("holding"); + try { + await api("/api/voice/ptt", { method: "POST", body: JSON.stringify({ action: "start" }) }); + } catch (err) { + button.classList.remove("holding"); + addBubble("system", err.message); + } + }; + const stop = async () => { + if (!button.classList.contains("holding")) return; + button.classList.remove("holding"); + try { + const data = await api("/api/voice/ptt", { method: "POST", body: JSON.stringify({ action: "stop" }) }); + if (data.text) { + const box = $("prompt"); + box.value = `${box.value ? `${box.value} ` : ""}${data.text}`; + } + } catch (err) { + addBubble("system", err.message); + } + }; + button.addEventListener("pointerdown", start); + button.addEventListener("pointerup", stop); + button.addEventListener("pointerleave", stop); +} + function bindUi() { + bindVoiceTalk(); $("send").addEventListener("click", () => sendMessage("/api/chat")); $("workers")?.addEventListener("click", () => sendMessage("/api/multi-agent")); $("nav-toggle")?.addEventListener("click", () => { @@ -919,6 +1003,7 @@ function bindUi() { }); } $("settings-key").value = ""; + await saveVoiceSettings(); go("chat"); await refreshStatus(); } catch (err) { diff --git a/src/desktop/web/index.html b/src/desktop/web/index.html index 2a3417d..1566921 100644 --- a/src/desktop/web/index.html +++ b/src/desktop/web/index.html @@ -84,6 +84,7 @@

New chat

+ @@ -165,6 +166,34 @@

Provider & model

+

Voice

+

Speech stays on this machine unless you choose OpenAI or xAI. Install the extra with pip install "clawd-codex[voice]". Push-to-talk uses the selected microphone and speaker, including Bluetooth on Windows WASAPI and Linux PulseAudio or PipeWire.

+ + + + + + + +

Source and updates: https://github.com/GoDeskio/Clawd-Code. Default local tree: ~/Jonathan/Jonathan-Ai.

diff --git a/src/desktop/web/styles.css b/src/desktop/web/styles.css index b0b8484..002537d 100644 --- a/src/desktop/web/styles.css +++ b/src/desktop/web/styles.css @@ -224,6 +224,9 @@ button:disabled { opacity: 0.55; cursor: default; } padding: 8px 10px; } .composer-row { justify-content: space-between; margin-top: 6px; } +#voice-ptt.holding { outline: 2px solid var(--accent); } +label.check { display: flex; align-items: center; } +.dialog input[type="checkbox"] { width: auto; min-height: 0; margin: 0 8px 0 0; } .end { justify-content: flex-end; } .chips { display: flex; flex-wrap: wrap; gap: 4px; margin-bottom: 6px; } .chip { diff --git a/src/voice/devices.py b/src/voice/devices.py new file mode 100644 index 0000000..3ddd3ca --- /dev/null +++ b/src/voice/devices.py @@ -0,0 +1,146 @@ +"""Input and output device selection for Windows WASAPI and Linux PulseAudio. + +Bluetooth headsets and speakers are ordinary devices on those host APIs. +Tests pass a fake backend so PortAudio is never required. +""" + +from __future__ import annotations + +import sys +from dataclasses import asdict, dataclass +from typing import Any, Protocol + +from .models import VOICE_EXTRA_HINT, VoiceError + + +@dataclass(frozen=True) +class AudioDevice: + id: str + name: str + kind: str + host_api: str + is_bluetooth: bool + is_default: bool = False + + def to_dict(self) -> dict[str, Any]: + payload = asdict(self) + payload["bluetooth"] = payload.pop("is_bluetooth") + payload["default"] = payload.pop("is_default") + return payload + + +class AudioBackend(Protocol): + def list_devices(self) -> list[AudioDevice]: + """Return every input and output the host currently exposes.""" + + def read(self, device_id: str) -> bytes: + """Capture one short chunk from the selected input.""" + + def write(self, device_id: str, audio: bytes) -> None: + """Play one buffer on the selected output.""" + + +def is_bluetooth(name: str) -> bool: + lowered = name.lower() + return any(token in lowered for token in ("bluetooth", "bluez", "bt-sco", "airpods")) + + +def host_matches_platform(host_api: str, platform: str) -> bool: + host = host_api.lower() + if platform == "win32": + return "wasapi" in host + if platform.startswith("linux"): + return "pulse" in host or "pipewire" in host or "pipe" in host + return True + + +def filter_for_platform(devices: list[AudioDevice], platform: str | None = None) -> list[AudioDevice]: + system = platform if platform is not None else sys.platform + matched = [device for device in devices if host_matches_platform(device.host_api, system)] + return matched or list(devices) + + +def list_devices(backend: AudioBackend | None = None, *, platform: str | None = None) -> list[AudioDevice]: + if backend is not None: + return filter_for_platform(list(backend.list_devices()), platform) + try: + discovered = _query_sounddevice() + except VoiceError: + raise + except Exception as exc: + raise VoiceError(VOICE_EXTRA_HINT) from exc + return filter_for_platform(discovered, platform) + + +def _query_sounddevice() -> list[AudioDevice]: + try: + import sounddevice as sd + except ImportError as exc: + raise VoiceError(VOICE_EXTRA_HINT) from exc + hostapis = list(sd.query_hostapis()) + devices = list(sd.query_devices()) + defaults = getattr(sd.default, "device", (None, None)) + default_in = defaults[0] if isinstance(defaults, (list, tuple)) and defaults else None + default_out = defaults[1] if isinstance(defaults, (list, tuple)) and len(defaults) > 1 else None + found: list[AudioDevice] = [] + for index, dev in enumerate(devices): + host_name = str(hostapis[int(dev["hostapi"])]["name"]) + name = str(dev["name"]) + if int(dev.get("max_input_channels") or 0) > 0: + found.append( + AudioDevice( + id=str(index), + name=name, + kind="input", + host_api=host_name, + is_bluetooth=is_bluetooth(name), + is_default=index == default_in, + ) + ) + if int(dev.get("max_output_channels") or 0) > 0: + found.append( + AudioDevice( + id=str(index), + name=name, + kind="output", + host_api=host_name, + is_bluetooth=is_bluetooth(name), + is_default=index == default_out, + ) + ) + return found + + +class SoundDeviceBackend: + """PortAudio backend. Windows uses WASAPI; Linux uses PulseAudio or PipeWire.""" + + def list_devices(self) -> list[AudioDevice]: + return _query_sounddevice() + + def read(self, device_id: str) -> bytes: + try: + import sounddevice as sd + except ImportError as exc: + raise VoiceError(VOICE_EXTRA_HINT) from exc + device = int(device_id) if str(device_id).isdigit() else None + audio = sd.rec(int(0.2 * 16000), samplerate=16000, channels=1, dtype="int16", device=device) + sd.wait() + return audio.tobytes() + + def write(self, device_id: str, audio: bytes) -> None: + try: + import sounddevice as sd + except ImportError as exc: + raise VoiceError(VOICE_EXTRA_HINT) from exc + import array + import io + import wave + + with wave.open(io.BytesIO(audio), "rb") as wav: + frames = wav.readframes(wav.getnframes()) + rate = wav.getframerate() + samples = array.array("h") + samples.frombytes(frames) + device = int(device_id) if str(device_id).isdigit() else None + sd.play(samples, samplerate=rate, device=device) + sd.wait() diff --git a/src/voice/engines.py b/src/voice/engines.py new file mode 100644 index 0000000..80e86fc --- /dev/null +++ b/src/voice/engines.py @@ -0,0 +1,213 @@ +"""Speech-to-text and text-to-speech. + +Local faster-whisper and Piper are the defaults. OpenAI and xAI run only +when that provider is selected, and only against their official endpoints. +""" + +from __future__ import annotations + +import json +import urllib.request +import uuid +from pathlib import Path +from typing import Callable + +from ..secret_mask import mask_secrets +from .models import ( + CLOUD_ENDPOINTS, + VOICE_EXTRA_HINT, + VoiceError, + assert_cloud_url, + catalog_model, + verify_installed, +) +from .settings import LOCAL, cloud_api_key, models_dir + +CloudPost = Callable[[str, dict[str, str], bytes, str], bytes] +LocalSpeech = Callable[[bytes, Path, str], str] +LocalVoice = Callable[[str, Path], bytes] + + +def transcribe( + audio: bytes, + settings: dict, + *, + directory: Path | None = None, + local_engine: LocalSpeech | None = None, + transport: CloudPost | None = None, +) -> str: + provider = str(settings.get("stt_provider") or LOCAL) + if provider == LOCAL: + engine = local_engine or _whisper_transcribe + model = catalog_model(str(settings.get("stt_model") or "faster-whisper-tiny")) + if model.role != "stt": + raise VoiceError(f"{model.model_id} is not a speech-to-text model.") + folder = verify_installed(models_dir(directory), model) + return engine(audio, folder, str(settings.get("language") or "en")).strip() + if provider not in CLOUD_ENDPOINTS: + raise VoiceError("Speech provider must be local, openai, or xai.") + return _cloud_transcribe(audio, settings, provider, transport) + + +def synthesize( + text: str, + settings: dict, + *, + directory: Path | None = None, + local_engine: LocalVoice | None = None, + transport: CloudPost | None = None, +) -> bytes: + spoken = " ".join(str(text or "").split()) + if not spoken: + raise VoiceError("There is nothing to say.") + provider = str(settings.get("tts_provider") or LOCAL) + if provider == LOCAL: + engine = local_engine or _piper_synthesize + model = catalog_model(str(settings.get("tts_voice") or "en_US-lessac-medium")) + if model.role != "tts": + raise VoiceError(f"{model.model_id} is not a text-to-speech voice.") + folder = verify_installed(models_dir(directory), model) + return engine(spoken, folder) + if provider not in CLOUD_ENDPOINTS: + raise VoiceError("Voice provider must be local, openai, or xai.") + return _cloud_synthesize(spoken, settings, provider, transport) + + +def _cloud_transcribe(audio: bytes, settings: dict, provider: str, transport: CloudPost | None) -> str: + key = cloud_api_key(provider, settings) + if not key: + raise VoiceError(f"{provider} speech-to-text needs an API key. Audio was not sent.") + url = assert_cloud_url(CLOUD_ENDPOINTS[provider]["stt"]) + language = str(settings.get("language") or "en").split("-")[0] + body, content_type = _multipart( + {"model": "whisper-1", "language": language}, + filename="speech.wav", + content=audio, + mime="audio/wav", + ) + raw = _post(url, key, body, content_type, transport) + try: + payload = json.loads(raw.decode("utf-8")) + except (UnicodeError, json.JSONDecodeError) as exc: + raise VoiceError("Cloud speech-to-text returned an unreadable response.") from exc + return str(payload.get("text") or "").strip() + + +def _cloud_synthesize(text: str, settings: dict, provider: str, transport: CloudPost | None) -> bytes: + key = cloud_api_key(provider, settings) + if not key: + raise VoiceError(f"{provider} text-to-speech needs an API key. The request was not sent.") + url = assert_cloud_url(CLOUD_ENDPOINTS[provider]["tts"]) + payload = json.dumps( + { + "model": "tts-1", + "input": text, + "voice": str(settings.get("cloud_voice") or "alloy"), + "response_format": "wav", + } + ).encode("utf-8") + return _post(url, key, payload, "application/json", transport) + + +def _post(url: str, key: str, body: bytes, content_type: str, transport: CloudPost | None) -> bytes: + headers = {"Authorization": f"Bearer {key}", "Content-Type": content_type} + sender = transport or _urllib_post + try: + return sender(url, headers, body, content_type) + except VoiceError: + raise + except Exception as exc: + raise VoiceError(mask_secrets(str(exc))) from exc + + +def _urllib_post(url: str, headers: dict[str, str], body: bytes, content_type: str) -> bytes: + assert_cloud_url(url) + request = urllib.request.Request(url, data=body, headers=headers, method="POST") + opener = urllib.request.build_opener(_RefuseRedirects) + with opener.open(request, timeout=60) as response: + return response.read() + + +class _RefuseRedirects(urllib.request.HTTPRedirectHandler): + def redirect_request(self, req, fp, code, msg, headers, newurl): # noqa: ANN001 + raise VoiceError("Cloud voice endpoint must not redirect.") + + +def _multipart(fields: dict[str, str], *, filename: str, content: bytes, mime: str) -> tuple[bytes, str]: + boundary = "jonvoice" + uuid.uuid4().hex + chunks: list[bytes] = [] + for key, value in fields.items(): + chunks.append( + f'--{boundary}\r\nContent-Disposition: form-data; name="{key}"\r\n\r\n{value}\r\n'.encode("utf-8") + ) + chunks.append( + ( + f'--{boundary}\r\nContent-Disposition: form-data; name="file"; filename="{filename}"\r\n' + f"Content-Type: {mime}\r\n\r\n" + ).encode("utf-8") + + content + + b"\r\n" + ) + chunks.append(f"--{boundary}--\r\n".encode("utf-8")) + return b"".join(chunks), f"multipart/form-data; boundary={boundary}" + + +def _whisper_transcribe(audio: bytes, model_dir: Path, language: str) -> str: + try: + from faster_whisper import WhisperModel + except ImportError as exc: + raise VoiceError(VOICE_EXTRA_HINT) from exc + import tempfile + + model = WhisperModel(str(model_dir), device="cpu", compute_type="int8") + with tempfile.NamedTemporaryFile(suffix=".wav", delete=False) as handle: + handle.write(audio) + temp_name = handle.name + try: + segments, _info = model.transcribe(temp_name, language=language or None) + return "".join(segment.text for segment in segments) + finally: + Path(temp_name).unlink(missing_ok=True) + + +def _load_piper(): + try: + from piper.voice import PiperVoice + except ImportError: + PiperVoice = None + if PiperVoice is None: + try: + from piper import PiperVoice + except ImportError as exc: + raise VoiceError(VOICE_EXTRA_HINT) from exc + return PiperVoice + + +def _piper_synthesize(text: str, model_dir: Path) -> bytes: + PiperVoice = _load_piper() + import io + import wave + + voice_path = next(model_dir.glob("*.onnx")) + voice = PiperVoice.load(str(voice_path)) + if hasattr(voice, "synthesize_stream_raw"): + pcm = b"".join(voice.synthesize_stream_raw(text)) + sample_rate = int(getattr(voice.config, "sample_rate", 22050) or 22050) + return _pcm16_wav(pcm, sample_rate) + buffer = io.BytesIO() + with wave.open(buffer, "wb") as wav: + voice.synthesize(text, wav) + return buffer.getvalue() + + +def _pcm16_wav(pcm: bytes, sample_rate: int) -> bytes: + import io + import wave + + buffer = io.BytesIO() + with wave.open(buffer, "wb") as wav: + wav.setnchannels(1) + wav.setsampwidth(2) + wav.setframerate(sample_rate) + wav.writeframes(pcm) + return buffer.getvalue() diff --git a/src/voice/models.py b/src/voice/models.py new file mode 100644 index 0000000..b253b47 --- /dev/null +++ b/src/voice/models.py @@ -0,0 +1,231 @@ +"""Official voice-model sources and checksums. + +Models come from Hugging Face or the official Piper and Whisper repositories. +Downloads from other hosts, including mirrors, are refused before any request +is made. Checksums are the SHA-256 digests published with those files. +""" + +from __future__ import annotations + +import hashlib +from dataclasses import dataclass +from pathlib import Path +from urllib.parse import urlsplit + + +class VoiceError(ValueError): + """A voice setting or model request cannot be completed.""" + + +VOICE_EXTRA_HINT = 'Voice extra is not installed. Run: pip install "clawd-codex[voice]"' + +_BLOCKED_MARKERS = ( + "hf-mirror.com", + "hf-mirror", + "modelscope.cn", + "modelscope.com", + "gitee.com", + "ghproxy.com", + "mirror.ghproxy.com", + "aliyuncs.com", + "myqcloud.com", + "tencentcloudapi.com", + "xiaomimimo.com", +) + +_GITHUB_PREFIXES = ( + "/rhasspy/piper", + "/rhasspy/piper-voices", + "/SYSTRAN/faster-whisper", + "/openai/whisper", +) + +_ASSET_HOSTS = frozenset( + { + "github.com", + "objects.githubusercontent.com", + "release-assets.githubusercontent.com", + "github-releases.githubusercontent.com", + } +) + + +@dataclass(frozen=True) +class ModelFile: + name: str + url: str + sha256: str + size: int + + +@dataclass(frozen=True) +class VoiceModel: + model_id: str + role: str + files: tuple[ModelFile, ...] + + +WHISPER_TINY = VoiceModel( + model_id="faster-whisper-tiny", + role="stt", + files=( + ModelFile( + name="model.bin", + url="https://huggingface.co/Systran/faster-whisper-tiny/resolve/main/model.bin", + sha256="dcb76c6586fc06cbdac6dd21f14cfd129cc4cdd9dce19bf4ffa62e59cbe6e6d1", + size=75538270, + ), + ModelFile( + name="config.json", + url="https://huggingface.co/Systran/faster-whisper-tiny/resolve/main/config.json", + sha256="a73a28cdfe1c43ccc7202fa333d1f89c202477271407ae9a7f19afa52039cac8", + size=2249, + ), + ModelFile( + name="tokenizer.json", + url="https://huggingface.co/Systran/faster-whisper-tiny/resolve/main/tokenizer.json", + sha256="fb7b63191e9bb045082c79fd742a3106a12c99513ab30df4a0d47fa6cb6fd0ab", + size=2203239, + ), + ModelFile( + name="vocabulary.txt", + url="https://huggingface.co/Systran/faster-whisper-tiny/resolve/main/vocabulary.txt", + sha256="34ce3fe1c5041027b3f8d42912270993f986dbc4bb34cf27f951e34a1e453913", + size=459861, + ), + ), +) + +PIPER_LESSAC = VoiceModel( + model_id="en_US-lessac-medium", + role="tts", + files=( + ModelFile( + name="en_US-lessac-medium.onnx", + url=( + "https://huggingface.co/rhasspy/piper-voices/resolve/main/" + "en/en_US/lessac/medium/en_US-lessac-medium.onnx" + ), + sha256="5efe09e69902187827af646e1a6e9d269dee769f9877d17b16b1b46eeaaf019f", + size=63201294, + ), + ModelFile( + name="en_US-lessac-medium.onnx.json", + url=( + "https://huggingface.co/rhasspy/piper-voices/resolve/main/" + "en/en_US/lessac/medium/en_US-lessac-medium.onnx.json" + ), + sha256="efe19c417bed055f2d69908248c6ba650fa135bc868b0e6abb3da181dab690a0", + size=4885, + ), + ), +) + +CATALOG: dict[str, VoiceModel] = { + WHISPER_TINY.model_id: WHISPER_TINY, + PIPER_LESSAC.model_id: PIPER_LESSAC, +} + +CLOUD_ENDPOINTS = { + "openai": { + "stt": "https://api.openai.com/v1/audio/transcriptions", + "tts": "https://api.openai.com/v1/audio/speech", + }, + "xai": { + "stt": "https://api.x.ai/v1/audio/transcriptions", + "tts": "https://api.x.ai/v1/audio/speech", + }, +} + + +def catalog_model(model_id: str) -> VoiceModel: + try: + return CATALOG[model_id] + except KeyError as exc: + known = ", ".join(sorted(CATALOG)) + raise VoiceError(f"Unknown voice model {model_id!r}. Known models: {known}.") from exc + + +def assert_model_url(url: str) -> str: + """Refuse every model URL that is not Hugging Face or an official source.""" + parsed = urlsplit(url.strip()) + host = (parsed.hostname or "").lower().rstrip(".") + lowered = url.lower() + if parsed.scheme != "https" or not host: + raise VoiceError(f"Refusing model URL {url!r}.") + if host.endswith(".cn") or any(marker in host or marker in lowered for marker in _BLOCKED_MARKERS): + raise VoiceError(f"Refusing model host {host}. Use Hugging Face or the official project source.") + huggingface = host == "huggingface.co" or host.endswith(".huggingface.co") + if huggingface: + return url + if host == "github.com": + if any(parsed.path.startswith(prefix) for prefix in _GITHUB_PREFIXES): + return url + raise VoiceError("GitHub model downloads are limited to official Piper and Whisper repositories.") + if host in _ASSET_HOSTS: + return url + raise VoiceError(f"Refusing model host {host}. Use Hugging Face or the official project source.") + + +def assert_cloud_url(url: str) -> str: + """Cloud audio is sent only to the selected provider's official endpoint.""" + target = url.strip() + allowed = {item for endpoints in CLOUD_ENDPOINTS.values() for item in endpoints.values()} + if target not in allowed: + raise VoiceError("Refusing voice request to an unknown cloud endpoint.") + return target + + +def file_sha256(path: Path) -> str: + digest = hashlib.sha256() + with path.open("rb") as handle: + for chunk in iter(lambda: handle.read(1024 * 1024), b""): + digest.update(chunk) + return digest.hexdigest() + + +def verify_file(path: Path, expected_sha256: str, expected_size: int | None = None) -> None: + if not path.is_file(): + raise VoiceError( + f"Missing {path.name}. Download it from Hugging Face or the official source with /voice download." + ) + if expected_size is not None and path.stat().st_size != expected_size: + raise VoiceError(f"Checksum mismatch for {path.name}: file size does not match the official file.") + actual = file_sha256(path) + if actual != expected_sha256: + raise VoiceError(f"Checksum mismatch for {path.name}.") + + +def verify_installed(root: Path, model: VoiceModel) -> Path: + folder = root / model.model_id + for item in model.files: + assert_model_url(item.url) + verify_file(folder / item.name, item.sha256, item.size) + return folder + + +def download_model(model_id: str, dest: Path, *, fetch) -> Path: + """Download one catalog model. ``fetch`` is called only after the URL is allowed.""" + model = catalog_model(model_id) + folder = dest / model.model_id + folder.mkdir(parents=True, exist_ok=True) + for item in model.files: + assert_model_url(item.url) + data = fetch(item.url) + digest = hashlib.sha256(data).hexdigest() + if digest != item.sha256 or len(data) != item.size: + raise VoiceError(f"Checksum mismatch for {item.name}. The download was discarded.") + target = folder / item.name + target.write_bytes(data) + return folder + + +def download_url(url: str, dest: Path, *, fetch, expected_sha256: str) -> None: + """Fetch one allowlisted URL. The fetch callback is not used when the host is refused.""" + assert_model_url(url) + data = fetch(url) + digest = hashlib.sha256(data).hexdigest() + if digest != expected_sha256: + raise VoiceError(f"Checksum mismatch for {dest.name}. The download was discarded.") + dest.parent.mkdir(parents=True, exist_ok=True) + dest.write_bytes(data) diff --git a/src/voice/service.py b/src/voice/service.py new file mode 100644 index 0000000..ec42d3a --- /dev/null +++ b/src/voice/service.py @@ -0,0 +1,234 @@ +"""Voice command, settings snapshot, and desktop push-to-talk control.""" + +from __future__ import annotations + +import threading +from pathlib import Path +from typing import Any + +from .devices import AudioBackend, SoundDeviceBackend, list_devices +from .engines import synthesize, transcribe +from .models import VoiceError, catalog_model, download_model +from .session import PushToTalk +from .settings import audio_stays_local, load_settings, public_settings, update_settings, voice_dir + +_CONTROLLER: PushToTalk | None = None +_CONTROLLER_LOCK = threading.Lock() + + +def voice_status( + directory: Path | None = None, + *, + backend: AudioBackend | None = None, + platform: str | None = None, +) -> dict[str, Any]: + settings = load_settings(directory) + device_error = "" + devices: list[dict[str, Any]] = [] + try: + devices = [device.to_dict() for device in list_devices(backend, platform=platform)] + except VoiceError as exc: + device_error = str(exc) + return { + "settings": public_settings(settings), + "devices": devices, + "device_error": device_error, + "providers": [ + {"id": "local", "label": "Local (Whisper and Piper)", "cloud": False}, + {"id": "openai", "label": "OpenAI", "cloud": True}, + {"id": "xai", "label": "xAI", "cloud": True}, + ], + "install": 'pip install "clawd-codex[voice]"', + } + + +def apply_voice_settings(changes: dict[str, Any], directory: Path | None = None) -> dict[str, Any]: + public = update_settings(changes, directory) + return voice_status(directory) | {"settings": public} + + +def speak( + text: str, + *, + directory: Path | None = None, + synthesizer=None, + player=None, + transport=None, +) -> str: + settings = load_settings(directory) + audio = synthesize(text, settings, directory=directory, local_engine=synthesizer, transport=transport) + output = str(settings.get("output_device") or "") + if player is None: + player = SoundDeviceBackend().write + player(output, audio) + if settings.get("tts_provider") == "local": + where = "on this machine" + else: + where = f"with {settings.get('tts_provider')}" + speaker = output or "the default speaker" + return f"Spoke {len(audio)} bytes through {speaker} ({where})." + + +def transcribe_audio( + audio: bytes, + *, + directory: Path | None = None, + local_engine=None, + transport=None, +) -> str: + settings = load_settings(directory) + return transcribe(audio, settings, directory=directory, local_engine=local_engine, transport=transport) + + +def fetch_models(directory: Path | None = None, *, fetch) -> str: + """Download the configured local models from their allowlisted URLs.""" + settings = load_settings(directory) + root = (directory or voice_dir()) / "models" + written = [] + for model_id in (settings["stt_model"], settings["tts_voice"]): + catalog_model(model_id) + download_model(model_id, root, fetch=fetch) + written.append(model_id) + return "Downloaded " + " and ".join(written) + " from Hugging Face after checksum verification." + + +def set_controller(controller: PushToTalk | None) -> None: + global _CONTROLLER + with _CONTROLLER_LOCK: + _CONTROLLER = controller + + +def _default_controller() -> PushToTalk: + backend = SoundDeviceBackend() + + def read_frame(device_id: str) -> bytes: + return backend.read(device_id) + + def _transcribe(audio: bytes) -> str: + return transcribe_audio(audio) + + return PushToTalk(read_frame, _transcribe) + + +def push_to_talk_action(action: str, *, directory: Path | None = None) -> dict[str, Any]: + settings = load_settings(directory) + if not settings.get("push_to_talk", True): + raise VoiceError("Push-to-talk is off. Turn it on with /voice ptt on.") + with _CONTROLLER_LOCK: + controller = _CONTROLLER or _default_controller() + name = str(action or "").strip().lower() + if name == "start": + controller.start(str(settings.get("input_device") or "")) + return {"holding": True, "audio_stays_local": audio_stays_local(settings)} + if name == "stop": + text = controller.stop() + return {"holding": False, "text": text, "audio_stays_local": audio_stays_local(settings)} + raise VoiceError("Push-to-talk action must be start or stop.") + + +def format_voice_command( + args: str, + *, + directory: Path | None = None, + backend: AudioBackend | None = None, + platform: str | None = None, + synthesizer=None, + player=None, + transport=None, + fetch=None, +) -> str: + text = " ".join(str(args or "").split()) + head, _, rest = text.partition(" ") + command = head.lower() + try: + if command in {"", "settings", "status"}: + return _format_status(voice_status(directory, backend=backend, platform=platform)) + if command == "devices": + return _format_devices(voice_status(directory, backend=backend, platform=platform)) + if command == "stt": + public = update_settings({"stt_provider": rest}, directory) + return _provider_line("Speech-to-text", public) + if command == "tts": + public = update_settings({"tts_provider": rest}, directory) + return _provider_line("Text-to-speech", public) + if command == "input": + update_settings({"input_device": rest}, directory) + return f"Microphone set to {rest or 'the default input'}." + if command == "output": + update_settings({"output_device": rest}, directory) + return f"Speaker set to {rest or 'the default output'}." + if command == "ptt": + enabled = update_settings({"push_to_talk": rest or "on"}, directory)["push_to_talk"] + return "Push-to-talk is on." if enabled else "Push-to-talk is off." + if command == "wake": + return _wake_command(rest, directory) + if command == "say": + return speak(rest, directory=directory, synthesizer=synthesizer, player=player, transport=transport) + if command == "download": + if fetch is None: + raise VoiceError("Model download is not configured.") + return fetch_models(directory, fetch=fetch) + if command == "language": + public = update_settings({"language": rest or "en"}, directory) + return f"Language set to {public['language']}." + except VoiceError as exc: + return str(exc) + return ( + "Usage: /voice [settings | devices | stt local|openai|xai | tts local|openai|xai | " + "input | output | ptt on|off | wake on|off [phrase] | say | language ]" + ) + + +def _wake_command(rest: str, directory: Path | None) -> str: + parts = rest.split(maxsplit=1) + if not parts: + raise VoiceError("Usage: /voice wake on|off [phrase]") + changes: dict[str, Any] = {"wake_word_enabled": parts[0]} + if len(parts) == 2: + changes["wake_word"] = parts[1] + public = update_settings(changes, directory) + state = "on" if public["wake_word_enabled"] else "off" + return ( + f"Wake word is {state} ({public['wake_word']}). " + "The check stays on this machine and is off until you enable it." + ) + + +def _provider_line(label: str, public: dict[str, Any]) -> str: + key = "stt_provider" if label.startswith("Speech") else "tts_provider" + stay = "Audio stays on this machine." if public["audio_stays_local"] else public["note"] + return f"{label} set to {public[key]}. {stay}" + + +def _format_status(status: dict[str, Any]) -> str: + settings = status["settings"] + lines = [ + f"Speech-to-text: {settings['stt_provider']} ({settings['stt_model']})", + f"Text-to-speech: {settings['tts_provider']} ({settings['tts_voice']})", + f"Microphone: {settings['input_device'] or 'default'}", + f"Speaker: {settings['output_device'] or 'default'}", + f"Push-to-talk: {'on' if settings['push_to_talk'] else 'off'}", + f"Wake word: {'on' if settings['wake_word_enabled'] else 'off'} ({settings['wake_word']})", + f"Language: {settings['language']}", + settings["note"], + f"Voice extra: {status['install']}", + ] + if status.get("device_error"): + lines.append(status["device_error"]) + return "\n".join(lines) + + +def _format_devices(status: dict[str, Any]) -> str: + devices = status.get("devices") or [] + if not devices: + detail = status.get("device_error") or "No audio devices reported." + return detail + lines = [] + for device in devices: + marks = [device["kind"], device["host_api"]] + if device.get("bluetooth"): + marks.append("Bluetooth") + if device.get("default"): + marks.append("default") + lines.append(f"{device['id']}: {device['name']} ({', '.join(marks)})") + return "\n".join(lines) diff --git a/src/voice/session.py b/src/voice/session.py new file mode 100644 index 0000000..77ed082 --- /dev/null +++ b/src/voice/session.py @@ -0,0 +1,131 @@ +"""Push-to-talk capture and an optional local wake word. + +The microphone is read only while push-to-talk is held, or while the user +has turned the wake word on. Wake-word audio is scored on this machine and +is never handed to a cloud transport. +""" + +from __future__ import annotations + +import threading +import time +from typing import Callable + +from .models import VoiceError + +FrameReader = Callable[[str], bytes] +WakeDetector = Callable[[bytes, str], bool] +Transcriber = Callable[[bytes], str] + + +class PushToTalk: + """Record from ``read_frame`` only while the button is held.""" + + def __init__(self, read_frame: FrameReader, transcribe: Transcriber) -> None: + self._read_frame = read_frame + self._transcribe = transcribe + self._lock = threading.Lock() + self._holding = False + self._device_id = "" + self._frames: list[bytes] = [] + self._thread: threading.Thread | None = None + self._error = "" + + @property + def holding(self) -> bool: + with self._lock: + return self._holding + + def start(self, device_id: str = "") -> None: + with self._lock: + self._frames = [] + self._error = "" + self._device_id = device_id + self._holding = True + if self._thread and self._thread.is_alive(): + return + self._thread = threading.Thread(target=self._loop, name="jonathan-ptt", daemon=True) + self._thread.start() + + def accept(self, frame: bytes) -> None: + """Keep a frame only when push-to-talk is held.""" + if not frame: + return + with self._lock: + if self._holding: + self._frames.append(frame) + + def stop(self) -> str: + thread = None + with self._lock: + self._holding = False + thread = self._thread + if thread is not None: + thread.join(timeout=2) + with self._lock: + audio = b"".join(self._frames) + error = self._error + self._frames = [] + if error: + raise VoiceError(error) + if not audio: + return "" + return self._transcribe(audio) + + def _loop(self) -> None: + while True: + with self._lock: + if not self._holding: + return + device_id = self._device_id + try: + frame = self._read_frame(device_id) + except Exception as exc: + with self._lock: + self._error = str(exc) + self._holding = False + return + with self._lock: + holding = self._holding + if holding and frame: + self._frames.append(frame) + if not holding: + return + time.sleep(0.01) + + +def capture_while_held(frames: list[tuple[bool, bytes]]) -> bytes: + """Join frames whose hold flag is true. Released frames are dropped.""" + chunks: list[bytes] = [] + for held, frame in frames: + if held and frame: + chunks.append(frame) + return b"".join(chunks) + + +def hear_wake_word( + frames: list[bytes], + phrase: str, + detector: WakeDetector, + transcribe: Transcriber, +) -> tuple[bool, str]: + """Score each frame locally. Only audio after the phrase is transcribed.""" + heard_at = None + for index, frame in enumerate(frames): + if detector(frame, phrase): + heard_at = index + break + if heard_at is None: + return False, "" + utterance = b"".join(frames[heard_at + 1 :]) + if not utterance: + return True, "" + return True, transcribe(utterance) + + +def local_phrase_detector(pcm: bytes, phrase: str, *, transcribe_local: Transcriber) -> bool: + """Match a wake phrase with the local transcriber. No cloud transport is available.""" + if not pcm or not phrase: + return False + text = transcribe_local(pcm) + return phrase.casefold() in text.casefold() diff --git a/src/voice/settings.py b/src/voice/settings.py new file mode 100644 index 0000000..913454a --- /dev/null +++ b/src/voice/settings.py @@ -0,0 +1,191 @@ +"""Per-user voice settings stored on this machine. + +Audio stays local unless the saved speech-to-text or text-to-speech provider +is OpenAI or xAI. Keys saved here never appear in the public settings view. +""" + +from __future__ import annotations + +import json +import os +from pathlib import Path +from typing import Any + +from .models import VoiceError + +LOCAL = "local" +OPENAI = "openai" +XAI = "xai" +PROVIDERS = (LOCAL, OPENAI, XAI) + +DEFAULT_SETTINGS: dict[str, Any] = { + "stt_provider": LOCAL, + "tts_provider": LOCAL, + "input_device": "", + "output_device": "", + "push_to_talk": True, + "wake_word_enabled": False, + "wake_word": "jonathan", + "language": "en", + "stt_model": "faster-whisper-tiny", + "tts_voice": "en_US-lessac-medium", + "cloud_voice": "alloy", + "openai_api_key": "", + "xai_api_key": "", +} + +_SECRET_KEYS = frozenset({"openai_api_key", "xai_api_key"}) + + +def voice_dir() -> Path: + override = os.environ.get("JONATHAN_VOICE_DIR", "").strip() + path = Path(override) if override else Path.home() / ".clawd" / "voice" + path.mkdir(parents=True, exist_ok=True) + return path + + +def _user_id() -> str: + from ..agent.memory import current_user_id + + return current_user_id() + + +def settings_path(directory: Path | None = None) -> Path: + root = directory or voice_dir() + root.mkdir(parents=True, exist_ok=True) + return root / f"settings-{_user_id()}.json" + + +def models_dir(directory: Path | None = None) -> Path: + root = directory or voice_dir() + path = root / "models" + path.mkdir(parents=True, exist_ok=True) + return path + + +def _write_private(path: Path, payload: dict[str, Any]) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + text = json.dumps(payload, indent=2, sort_keys=True) + if os.name == "nt": + path.write_text(text, encoding="utf-8") + return + fd = os.open(str(path), os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600) + with os.fdopen(fd, "w", encoding="utf-8") as handle: + handle.write(text) + os.chmod(path, 0o600) + + +def load_settings(directory: Path | None = None) -> dict[str, Any]: + path = settings_path(directory) + stored: dict[str, Any] = {} + if path.is_file(): + try: + loaded = json.loads(path.read_text(encoding="utf-8")) + except (OSError, json.JSONDecodeError): + loaded = {} + if isinstance(loaded, dict): + stored = loaded + merged = dict(DEFAULT_SETTINGS) + merged.update({key: stored[key] for key in DEFAULT_SETTINGS if key in stored}) + return merged + + +def audio_stays_local(settings: dict[str, Any]) -> bool: + return settings.get("stt_provider") == LOCAL and settings.get("tts_provider") == LOCAL + + +def public_settings(settings: dict[str, Any]) -> dict[str, Any]: + """Settings safe to show in the UI and command output.""" + view = {key: settings.get(key) for key in DEFAULT_SETTINGS if key not in _SECRET_KEYS} + view["audio_stays_local"] = audio_stays_local(settings) + view["openai_key_configured"] = bool( + str(settings.get("openai_api_key") or "").strip() or os.environ.get("OPENAI_API_KEY") + ) + view["xai_key_configured"] = bool(str(settings.get("xai_api_key") or "").strip() or os.environ.get("XAI_API_KEY")) + view["note"] = ( + "Audio stays on this machine." + if view["audio_stays_local"] + else "Audio is sent only because a cloud provider is selected." + ) + return view + + +def _saved_openai_key() -> str: + if os.environ.get("OPENAI_API_KEY"): + return os.environ["OPENAI_API_KEY"] + try: + from ..config import get_provider_config + + return str(get_provider_config("openai").get("api_key") or "") + except Exception: + return "" + + +def _coerce_provider(value: Any, label: str) -> str: + name = str(value or "").strip().lower() + if name not in PROVIDERS: + raise VoiceError(f"{label} must be local, openai, or xai.") + return name + + +def _coerce_bool(value: Any) -> bool: + if isinstance(value, bool): + return value + text = str(value).strip().lower() + if text in {"1", "true", "yes", "on"}: + return True + if text in {"0", "false", "no", "off"}: + return False + raise VoiceError(f"Expected on or off, got {value!r}.") + + +def _coerce_phrase(value: Any) -> str: + text = " ".join(str(value or "").split()) + if not text or len(text) > 64 or any(ord(char) > 127 for char in text): + raise VoiceError("Wake word must be 1 to 64 plain characters.") + return text + + +def _coerce_language(value: Any) -> str: + text = str(value or "").strip() + parts = text.split("-") + if not text or len(parts) > 2 or any(not part.isalpha() or len(part) > 8 for part in parts): + raise VoiceError("Language must look like en or en-US.") + return text + + +def _coerce(key: str, value: Any) -> Any: + if key in {"stt_provider", "tts_provider"}: + return _coerce_provider(value, "Speech provider" if key == "stt_provider" else "Voice provider") + if key in {"push_to_talk", "wake_word_enabled"}: + return _coerce_bool(value) + if key == "wake_word": + return _coerce_phrase(value) + if key == "language": + return _coerce_language(value) + if key in _SECRET_KEYS: + return str(value or "") + if key in {"input_device", "output_device", "stt_model", "tts_voice", "cloud_voice"}: + return str(value or "").strip() + raise VoiceError(f"Unknown voice setting {key}.") + + +def update_settings(changes: dict[str, Any], directory: Path | None = None) -> dict[str, Any]: + current = load_settings(directory) + for key, value in changes.items(): + if key not in DEFAULT_SETTINGS: + raise VoiceError(f"Unknown voice setting {key}.") + if key in _SECRET_KEYS and not str(value or "").strip(): + continue + current[key] = _coerce(key, value) + _write_private(settings_path(directory), current) + return public_settings(current) + + +def cloud_api_key(provider: str, settings: dict[str, Any]) -> str: + if provider == OPENAI: + saved = str(settings.get("openai_api_key") or "").strip() + return saved or _saved_openai_key() + if provider == XAI: + return str(settings.get("xai_api_key") or "").strip() or os.environ.get("XAI_API_KEY", "") + return "" diff --git a/tests/test_voice.py b/tests/test_voice.py new file mode 100644 index 0000000..8509290 --- /dev/null +++ b/tests/test_voice.py @@ -0,0 +1,457 @@ +"""Voice tests. Audio engines and PortAudio are mocked; no model is downloaded.""" + +from __future__ import annotations + +import hashlib +import json +import os +import sys +import tempfile +import threading +import unittest +from pathlib import Path +from unittest.mock import MagicMock, patch + +from src.command_system.builtins import execute_command_sync +from src.command_system.engine import create_command_context +from src.desktop.server import DesktopServer +from src.voice import ARCHIVE_NAME, SAMPLE_FILES +from src.voice.devices import AudioDevice, filter_for_platform, is_bluetooth +from src.voice.engines import synthesize, transcribe +from src.voice.models import ( + ModelFile, + VoiceError, + VoiceModel, + assert_cloud_url, + assert_model_url, + download_url, + verify_file, +) +from src.voice.service import ( + format_voice_command, + push_to_talk_action, + set_controller, + speak, + voice_status, +) +from src.voice.session import PushToTalk, capture_while_held, hear_wake_word, local_phrase_detector +from src.voice.settings import load_settings, public_settings, update_settings + + +def _model(payload: bytes, *, model_id: str, role: str, name: str) -> VoiceModel: + return VoiceModel( + model_id=model_id, + role=role, + files=( + ModelFile( + name=name, + url="https://huggingface.co/Systran/faster-whisper-tiny/resolve/main/model.bin", + sha256=hashlib.sha256(payload).hexdigest(), + size=len(payload), + ), + ), + ) + + +class VoiceTestCase(unittest.TestCase): + def setUp(self) -> None: + self.tmp = tempfile.TemporaryDirectory() + self.voice = Path(self.tmp.name) / "voice" + self.voice.mkdir() + self.env = patch.dict( + os.environ, + {"JONATHAN_VOICE_DIR": str(self.voice), "OPENAI_API_KEY": "", "XAI_API_KEY": ""}, + ) + self.env.start() + set_controller(None) + + def tearDown(self) -> None: + set_controller(None) + self.env.stop() + self.tmp.cleanup() + + def _plant(self, model: VoiceModel, payload: bytes) -> None: + folder = self.voice / "models" / model.model_id + folder.mkdir(parents=True) + (folder / model.files[0].name).write_bytes(payload) + + +class TestDefaultsAndSettings(VoiceTestCase): + def test_defaults_are_local_english_and_push_to_talk(self) -> None: + settings = load_settings(self.voice) + self.assertEqual(settings["stt_provider"], "local") + self.assertEqual(settings["tts_provider"], "local") + self.assertEqual(settings["language"], "en") + self.assertTrue(settings["push_to_talk"]) + self.assertFalse(settings["wake_word_enabled"]) + self.assertTrue(public_settings(settings)["audio_stays_local"]) + + def test_cloud_provider_is_explicit_and_keys_stay_out_of_the_view(self) -> None: + secret = "sk-" + ("a" * 24) + public = update_settings( + {"stt_provider": "openai", "openai_api_key": secret, "tts_provider": "xai", "xai_api_key": "xai-secret-value"}, + self.voice, + ) + self.assertEqual(public["stt_provider"], "openai") + self.assertEqual(public["tts_provider"], "xai") + self.assertFalse(public["audio_stays_local"]) + self.assertNotIn(secret, json.dumps(public)) + self.assertNotIn("xai-secret-value", json.dumps(public)) + shown = format_voice_command("settings", directory=self.voice, backend=_EmptyBackend()) + self.assertNotIn(secret, shown) + self.assertIn("cloud provider", shown) + stored = (self.voice / next(self.voice.glob("settings-*.json"))).read_text(encoding="utf-8") + self.assertIn(secret, stored) + + def test_users_do_not_share_settings(self) -> None: + with patch("src.agent.memory.current_user_id", return_value="ada"): + update_settings({"input_device": "ada-mic"}, self.voice) + with patch("src.agent.memory.current_user_id", return_value="bob"): + update_settings({"input_device": "bob-mic"}, self.voice) + bob = load_settings(self.voice) + with patch("src.agent.memory.current_user_id", return_value="ada"): + ada = load_settings(self.voice) + self.assertEqual(ada["input_device"], "ada-mic") + self.assertEqual(bob["input_device"], "bob-mic") + + def test_rejects_unknown_provider(self) -> None: + for name in ("tencent", "edge", "mimo", "browser"): + with self.assertRaises(VoiceError): + update_settings({"stt_provider": name}, self.voice) + + +class TestModels(VoiceTestCase): + def test_refuses_mirrors_before_any_fetch(self) -> None: + calls: list[str] = [] + + def fetch(url: str) -> bytes: + calls.append(url) + return b"nope" + + blocked = ( + "https://hf-mirror.com/Systran/faster-whisper-tiny/resolve/main/model.bin", + "https://www.modelscope.cn/models/whisper/model.bin", + "http://huggingface.co/Systran/faster-whisper-tiny/resolve/main/model.bin", + "https://github.com/evil/piper/releases/model.onnx", + "https://example.cn/model.bin", + "https://mirror.ghproxy.com/https://github.com/rhasspy/piper/model.onnx", + "https://api.xiaomimimo.com/v1/audio", + ) + for url in blocked: + with self.assertRaises(VoiceError): + download_url(url, self.voice / "model.bin", fetch=fetch, expected_sha256="abc") + self.assertEqual(calls, []) + + def test_allows_hugging_face_and_official_piper(self) -> None: + assert_model_url("https://huggingface.co/Systran/faster-whisper-tiny/resolve/main/model.bin") + assert_model_url( + "https://huggingface.co/rhasspy/piper-voices/resolve/main/en/en_US/lessac/medium/en_US-lessac-medium.onnx" + ) + assert_model_url("https://github.com/rhasspy/piper/releases/download/v1/voice.onnx") + assert_cloud_url("https://api.openai.com/v1/audio/transcriptions") + assert_cloud_url("https://api.x.ai/v1/audio/speech") + with self.assertRaises(VoiceError): + assert_cloud_url("https://api.openai.com/v1/chat/completions") + + def test_checksum_mismatch_discards_the_file(self) -> None: + dest = self.voice / "model.bin" + url = "https://huggingface.co/Systran/faster-whisper-tiny/resolve/main/model.bin" + with self.assertRaises(VoiceError) as raised: + download_url(url, dest, fetch=lambda _url: b"not-the-model", expected_sha256="0" * 64) + self.assertIn("Checksum mismatch", str(raised.exception)) + self.assertFalse(dest.exists()) + + def test_checksum_match_keeps_the_file(self) -> None: + payload = b"official-bytes" + dest = self.voice / "nested" / "model.bin" + url = "https://huggingface.co/rhasspy/piper-voices/resolve/main/en/en_US/lessac/medium/en_US-lessac-medium.onnx" + download_url(url, dest, fetch=lambda _url: payload, expected_sha256=hashlib.sha256(payload).hexdigest()) + self.assertEqual(dest.read_bytes(), payload) + with self.assertRaises(VoiceError): + verify_file(dest, "f" * 64, len(payload)) + + +class TestSpeech(VoiceTestCase): + def test_local_transcribe_never_calls_the_network(self) -> None: + payload = b"weights" + model = _model(payload, model_id="tiny-test", role="stt", name="model.bin") + self._plant(model, payload) + calls: list[object] = [] + + def engine(audio: bytes, folder: Path, language: str) -> str: + self.assertEqual(audio, b"mic") + self.assertEqual(language, "en") + self.assertTrue((folder / "model.bin").is_file()) + return "local words" + + def transport(*args: object) -> bytes: + calls.append(args) + return b"{}" + + def exploded(*_args: object, **_kwargs: object) -> bytes: + raise AssertionError("local speech tried to use the network") + + with patch("src.voice.engines.catalog_model", return_value=model), patch( + "urllib.request.urlopen", exploded + ): + text = transcribe( + b"mic", + {"stt_provider": "local", "stt_model": "tiny-test", "language": "en"}, + directory=self.voice, + local_engine=engine, + transport=transport, + ) + self.assertEqual(text, "local words") + self.assertEqual(calls, []) + + def test_bad_checksum_does_not_load_the_engine(self) -> None: + payload = b"nope" + model = _model(b"expected", model_id="tiny-test", role="stt", name="model.bin") + self._plant(model, payload) + called: list[bytes] = [] + with patch("src.voice.engines.catalog_model", return_value=model): + with self.assertRaises(VoiceError): + transcribe( + b"mic", + {"stt_provider": "local", "stt_model": "tiny-test", "language": "en"}, + directory=self.voice, + local_engine=lambda *_args: called.append(b"x") or "no", + ) + self.assertEqual(called, []) + + def test_openai_and_xai_are_used_only_when_selected(self) -> None: + def transport(url: str, headers: dict[str, str], body: bytes, content_type: str) -> bytes: + self.assertIn(url, { + "https://api.openai.com/v1/audio/transcriptions", + "https://api.x.ai/v1/audio/speech", + }) + self.assertIn("Bearer", headers["Authorization"]) + self.assertNotIn("sk-secretsecretsecretsecret", url) + if url.endswith("/speech"): + return b"RIFFcloud" + return b'{"text": "from openai"}' + + with patch("src.voice.settings._saved_openai_key", return_value=""): + text = transcribe( + b"mic", + {"stt_provider": "openai", "openai_api_key": "sk-secretsecretsecretsecret", "language": "en"}, + directory=self.voice, + transport=transport, + local_engine=lambda *_args: (_ for _ in ()).throw(AssertionError("local engine used")), + ) + audio = synthesize( + "hello", + {"tts_provider": "xai", "xai_api_key": "xai-key", "cloud_voice": "alloy"}, + directory=self.voice, + transport=transport, + ) + self.assertEqual(text, "from openai") + self.assertEqual(audio, b"RIFFcloud") + + def test_cloud_without_a_key_sends_nothing(self) -> None: + def transport(*_args: object) -> bytes: + raise AssertionError("audio left the machine") + + with patch("src.voice.settings._saved_openai_key", return_value=""): + with self.assertRaises(VoiceError) as raised: + transcribe( + b"mic", + {"stt_provider": "openai", "openai_api_key": "", "language": "en"}, + transport=transport, + ) + self.assertIn("was not sent", str(raised.exception)) + + def test_speak_plays_local_audio_on_the_selected_device(self) -> None: + payload = b"voice-weights" + model = _model(payload, model_id="piper-test", role="tts", name="voice.onnx") + self._plant(model, payload) + update_settings({"tts_provider": "local", "tts_voice": "piper-test", "output_device": "4"}, self.voice) + played: list[tuple[str, bytes]] = [] + with patch("src.voice.engines.catalog_model", return_value=model): + message = speak( + "hello there", + directory=self.voice, + synthesizer=lambda text, folder: b"RIFF" + text.encode("utf-8"), + player=lambda device, audio: played.append((device, audio)), + ) + self.assertEqual(played, [("4", b"RIFFhello there")]) + self.assertIn("on this machine", message) + + +class TestDevicesAndCapture(VoiceTestCase): + def test_windows_wasapi_and_linux_pulse_include_bluetooth(self) -> None: + devices = [ + AudioDevice("1", "Microphone Array", "input", "MME", False), + AudioDevice("2", "Headset (Bluetooth)", "input", "Windows WASAPI", True), + AudioDevice("3", "Speakers", "output", "Windows WASAPI", False), + AudioDevice("4", "bluez_output.AA_BB.a2dp-sink", "output", "PulseAudio", True), + AudioDevice("5", "PipeWire mic", "input", "PipeWire", False), + ] + self.assertTrue(is_bluetooth("Headset (Bluetooth)")) + self.assertTrue(is_bluetooth("bluez_output.AA_BB")) + windows = filter_for_platform(devices, "win32") + linux = filter_for_platform(devices, "linux") + self.assertTrue(all("wasapi" in device.host_api.lower() for device in windows)) + self.assertTrue(any(device.is_bluetooth and device.kind == "input" for device in windows)) + self.assertTrue(all("pulse" in device.host_api.lower() or "pipe" in device.host_api.lower() for device in linux)) + self.assertTrue(any(device.is_bluetooth for device in linux)) + listed = format_voice_command("devices", directory=self.voice, backend=_Backend(devices), platform="linux") + self.assertIn("Bluetooth", listed) + self.assertIn("bluez_output", listed) + self.assertNotIn("MME", listed) + + def test_push_to_talk_records_only_while_held(self) -> None: + self.assertEqual( + capture_while_held([(False, b"before"), (True, b"aa"), (True, b"bb"), (False, b"after")]), + b"aabb", + ) + started = threading.Event() + reads: list[str] = [] + + def read(device_id: str) -> bytes: + reads.append(device_id) + started.set() + return b"pcm" + + def _transcribe(audio: bytes) -> str: + self.assertTrue(audio.startswith(b"pcm")) + self.assertNotIn(b"after", audio) + return "heard" + + session = PushToTalk(read, _transcribe) + session.accept(b"before") + session.start("bt-mic") + self.assertTrue(started.wait(1)) + self.assertEqual(session.stop(), "heard") + self.assertFalse(session.holding) + count = len(reads) + session.accept(b"after") + self.assertGreaterEqual(count, 1) + self.assertTrue(all(device == "bt-mic" for device in reads)) + + def test_wake_word_stays_local_and_is_optional(self) -> None: + cloud_calls: list[bytes] = [] + + def detector(frame: bytes, phrase: str) -> bool: + self.assertEqual(phrase, "jonathan") + return frame == b"wake" + + def cloud(audio: bytes) -> str: + cloud_calls.append(audio) + return "should not run for the wake frame" + + heard, text = hear_wake_word([b"noise", b"wake", b"utterance"], "jonathan", detector, cloud) + self.assertTrue(heard) + self.assertEqual(cloud_calls, [b"utterance"]) + self.assertEqual(text, "should not run for the wake frame") + missed, empty = hear_wake_word([b"noise"], "jonathan", detector, cloud) + self.assertFalse(missed) + self.assertEqual(empty, "") + self.assertTrue( + local_phrase_detector(b"clip", "jonathan", transcribe_local=lambda _pcm: "hey jonathan") + ) + message = format_voice_command("wake off", directory=self.voice, backend=_EmptyBackend()) + self.assertIn("off", message) + self.assertIn("this machine", message) + + def test_import_does_not_load_optional_audio_packages(self) -> None: + for name in ("faster_whisper", "sounddevice", "piper"): + self.assertNotIn(name, sys.modules) + + +class TestCommandAndDesktop(VoiceTestCase): + def test_voice_command_and_archive_placeholder(self) -> None: + self.assertEqual(ARCHIVE_NAME, "voice") + self.assertTrue(SAMPLE_FILES) + text = format_voice_command("", directory=self.voice, backend=_EmptyBackend()) + self.assertIn("Speech-to-text: local", text) + self.assertIn("clawd-codex[voice]", text) + switched = format_voice_command("stt openai", directory=self.voice, backend=_EmptyBackend()) + self.assertIn("openai", switched) + self.assertIn("cloud provider", switched) + context = create_command_context(self.voice) + ok, result, error = execute_command_sync("voice", "ptt off", context) + self.assertTrue(ok, error) + self.assertIn("off", result or "") + from src.desktop.runtime import _HELP_TEXT + + self.assertIn("/voice", _HELP_TEXT) + + def test_desktop_settings_api(self) -> None: + server = DesktopServer(runtime=MagicMock(), token="token") + status, payload = server.handle_api("GET", "/api/voice", {}, None) + self.assertEqual(status, 200) + self.assertTrue(payload["settings"]["audio_stays_local"]) + self.assertEqual(payload["settings"]["stt_provider"], "local") + status, saved = server.handle_api( + "POST", + "/api/voice", + {}, + {"tts_provider": "local", "input_device": "2", "wake_word_enabled": False}, + ) + self.assertEqual(status, 200) + self.assertEqual(saved["settings"]["input_device"], "2") + self.assertTrue(saved["settings"]["audio_stays_local"]) + + class FakeTalk: + def start(self, device_id: str = "") -> None: + self.device_id = device_id + + def stop(self) -> str: + return "from the headset" + + set_controller(FakeTalk()) + status, started = server.handle_api("POST", "/api/voice/ptt", {}, {"action": "start"}) + self.assertEqual(status, 200) + self.assertTrue(started["holding"]) + self.assertTrue(started["audio_stays_local"]) + status, stopped = server.handle_api("POST", "/api/voice/ptt", {}, {"action": "stop"}) + self.assertEqual(stopped["text"], "from the headset") + html = Path("src/desktop/web/index.html").read_text(encoding="utf-8") + script = Path("src/desktop/web/app.js").read_text(encoding="utf-8") + self.assertIn("voice-stt", html) + self.assertIn("Hold to talk", html) + self.assertIn("Local Whisper", html) + self.assertIn("Local Piper", html) + self.assertIn("/api/voice", script) + + def test_sources_stay_english_and_do_not_call_blocked_hosts(self) -> None: + for path in Path("src/voice").rglob("*.py"): + text = path.read_text(encoding="utf-8") + self.assertFalse(any("\u4e00" <= char <= "\u9fff" for char in text), path.name) + self.assertNotIn("https://hf-mirror", text) + self.assertNotIn("https://asr.tencentcloudapi.com", text) + self.assertNotIn("https://api.xiaomimimo.com", text) + notice = Path("NOTICE").read_text(encoding="utf-8") + self.assertIn("https://github.com/TencentCloud/Octop", notice) + self.assertIn("Piper", notice) + self.assertIn("faster-whisper", notice) + + +class _EmptyBackend: + def list_devices(self) -> list[AudioDevice]: + return [] + + def read(self, device_id: str) -> bytes: + return b"" + + def write(self, device_id: str, audio: bytes) -> None: + return None + + +class _Backend: + def __init__(self, devices: list[AudioDevice]) -> None: + self._devices = devices + + def list_devices(self) -> list[AudioDevice]: + return list(self._devices) + + def read(self, device_id: str) -> bytes: + return b"" + + def write(self, device_id: str, audio: bytes) -> None: + return None + + +if __name__ == "__main__": + unittest.main()