"""Voice (Speech-to-Speech) routes at /api/voice.""" import logging import time from quart import Blueprint, jsonify, request from fabledassistant.auth import admin_required, login_required logger = logging.getLogger(__name__) voice_bp = Blueprint("voice", __name__, url_prefix="/api/voice") @voice_bp.route("/status", methods=["GET"]) @login_required async def voice_status(): """Return availability of STT and TTS services.""" from fabledassistant.services.voice_config import get_voice_config from fabledassistant.services.stt import stt_available from fabledassistant.services.tts import tts_available config = await get_voice_config() enabled = config.get("voice_enabled", "false").lower() in ("1", "true", "yes") if not enabled: return jsonify({"enabled": False, "stt": False, "tts": False}) return jsonify({ "enabled": True, "stt": stt_available(), "tts": tts_available(), "stt_model": config.get("voice_stt_model", "base.en"), "tts_backend": "piper", }) @voice_bp.route("/voices", methods=["GET"]) @login_required async def list_voices(): """Return available piper voice IDs and metadata. Scans /opt/piper-voices (bundled) + /data/voices (admin-downloaded) on every call so newly-downloaded voices show up without a restart. Does NOT require tts_available() — even if the active voice failed to load, the catalog is still useful for picking a different one. """ from fabledassistant.services.voice_config import is_voice_enabled if not await is_voice_enabled(): return jsonify({"error": "Voice feature is disabled"}), 503 from fabledassistant.services.tts import list_voices return jsonify({"voices": list_voices()}) @voice_bp.route("/transcribe", methods=["POST"]) @login_required async def transcribe_audio(): """Accept a multipart audio file and return the transcript. Request: multipart/form-data with field 'audio' (WebM/Opus blob) Response: {"transcript": "...", "duration_ms": 123} """ from fabledassistant.services.voice_config import is_voice_enabled if not await is_voice_enabled(): return jsonify({"error": "Voice feature is disabled"}), 503 from fabledassistant.services.stt import stt_available, transcribe if not stt_available(): return jsonify({"error": "STT not available — model may still be loading"}), 503 files = await request.files audio_file = files.get("audio") if audio_file is None: return jsonify({"error": "No audio file provided"}), 400 audio_bytes = audio_file.read() if not audio_bytes: return jsonify({"error": "Empty audio file"}), 400 if len(audio_bytes) > 25 * 1024 * 1024: # 25 MB hard cap return jsonify({"error": "Audio file too large (max 25 MB)"}), 413 mime_type = audio_file.content_type or "audio/webm" form = await request.form context = (form.get("context") or "").strip() or None t0 = time.monotonic() try: transcript = await transcribe(audio_bytes, mime_type, initial_prompt=context) except Exception: logger.exception("STT transcription failed") return jsonify({"error": "Transcription failed"}), 500 duration_ms = round((time.monotonic() - t0) * 1000) return jsonify({"transcript": transcript, "duration_ms": duration_ms}) @voice_bp.route("/synthesise", methods=["POST"]) @login_required async def synthesise_speech(): """Convert text to speech and return WAV bytes. Request body: {"text": "...", "voice": "af_heart", "speed": 1.0} Response: audio/wav bytes """ from fabledassistant.services.voice_config import is_voice_enabled if not await is_voice_enabled(): return jsonify({"error": "Voice feature is disabled"}), 503 from fabledassistant.services.tts import synthesise, tts_available if not tts_available(): return jsonify({"error": "TTS not available — model may still be loading"}), 503 data = await request.get_json() if not data: return jsonify({"error": "JSON body required"}), 400 text = str(data.get("text", "")).strip() if not text: return jsonify({"error": "text is required"}), 400 char_count = len(text) if char_count > 8000: logger.warning( "TTS request rejected: text too long (%d chars, limit 8000). Preview: %r", char_count, text[:120], ) return jsonify({"error": "text too long (max 8000 characters)"}), 400 # Piper voice file basename (e.g. "en_US-amy-medium"). Default is read # from user settings if not in the request body. voice = str(data.get("voice", "")) or "en_US-amy-medium" try: speed = float(data.get("speed", 1.0)) except (TypeError, ValueError): speed = 1.0 # Pull saved settings only when caller didn't override. if "voice" not in data and "speed" not in data: from fabledassistant.services.settings import get_setting from fabledassistant.auth import get_current_user_id try: uid = get_current_user_id() saved_voice = await get_setting(uid, "voice_tts_voice", "") if saved_voice: voice = saved_voice saved_speed = await get_setting(uid, "voice_tts_speed", "") if saved_speed: try: speed = float(saved_speed) except ValueError: pass except Exception: pass logger.info("TTS synthesis start: %d chars, voice=%s, speed=%.2f", char_count, voice, speed) t0 = time.monotonic() try: wav_bytes = await synthesise(text, voice=voice, speed=speed) except Exception: logger.exception( "TTS synthesis failed: %d chars, voice=%s. Preview: %r", char_count, voice, text[:120], ) return jsonify({"error": "Synthesis failed"}), 500 duration_ms = round((time.monotonic() - t0) * 1000) if not wav_bytes: logger.warning( "TTS synthesis returned empty audio: %d chars, voice=%s, %dms. Preview: %r", char_count, voice, duration_ms, text[:120], ) else: logger.info( "TTS synthesis complete: %d chars → %d bytes in %dms (voice=%s)", char_count, len(wav_bytes), duration_ms, voice, ) from quart import Response return Response(wav_bytes, mimetype="audio/wav") # ── Voice library (admin only) ────────────────────────────────────────────── # Browse the piper-voices catalog, download new voices into /data/voices, # remove user-installed voices. Bundled voices in /opt/piper-voices are # read-only and cannot be touched via these endpoints. # These are admin-only because installs consume shared disk and affect # every user on the instance (voices are picked per-user, but the files # themselves are shared). @voice_bp.route("/voices/library", methods=["GET"]) @admin_required async def voice_library(): """Return the piper-voices catalog with install state annotations. Query params: ?refresh=1 — force a fresh fetch from HuggingFace (bypass the 24h in-memory cache). Use sparingly; HF doesn't appreciate hammering. """ from fabledassistant.services import voice_library as lib force = (request.args.get("refresh") or "").lower() in ("1", "true", "yes") try: catalog = await lib.fetch_catalog(force_refresh=force) except Exception: logger.exception("Voice catalog fetch failed") return jsonify({"error": "Failed to fetch voice catalog"}), 502 voices = lib.shape_catalog_for_ui(catalog) return jsonify({"voices": voices, "count": len(voices)}) @voice_bp.route("/voices/install", methods=["POST"]) @admin_required async def install_voice_route(): """Download a voice into /data/voices. Body: {"voice_id": "en_US-amy-medium"} Idempotent — already-installed voices return {"skipped": true} without re-downloading. """ from fabledassistant.services import voice_library as lib data = await request.get_json() voice_id = str((data or {}).get("voice_id") or "").strip() if not voice_id: return jsonify({"error": "voice_id is required"}), 400 try: result = await lib.install_voice(voice_id) except ValueError as e: return jsonify({"error": str(e)}), 400 except KeyError as e: return jsonify({"error": str(e)}), 404 except Exception: logger.exception("Voice install failed: %s", voice_id) return jsonify({"error": "Voice install failed"}), 500 return jsonify(result) @voice_bp.route("/voices/", methods=["DELETE"]) @admin_required async def uninstall_voice_route(voice_id: str): """Remove a /data/voices voice. Bundled voices return 403.""" from fabledassistant.services import voice_library as lib try: result = await lib.uninstall_voice(voice_id) except ValueError as e: return jsonify({"error": str(e)}), 400 except PermissionError as e: return jsonify({"error": str(e)}), 403 except Exception: logger.exception("Voice uninstall failed: %s", voice_id) return jsonify({"error": "Voice uninstall failed"}), 500 return jsonify(result)