"""Voice (Speech-to-Speech) routes at /api/voice.""" import logging import time from quart import Blueprint, jsonify, request from fabledassistant.auth import 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": "kokoro", }) @voice_bp.route("/voices", methods=["GET"]) @login_required async def list_voices(): """Return available Kokoro voice IDs and labels.""" 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, tts_available if not tts_available(): return jsonify({"error": "TTS not available"}), 503 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 voice = str(data.get("voice", "af_heart")) try: speed = float(data.get("speed", 1.0)) except (TypeError, ValueError): speed = 1.0 voice_blend = data.get("voice_blend") if not isinstance(voice_blend, list) or len(voice_blend) < 2: voice_blend = None # When no explicit voice/blend/speed provided, load all voice settings from the user's profile if "voice" not in data and "voice_blend" not in data and "speed" not in data: from fabledassistant.services.settings import get_setting from fabledassistant.auth import get_current_user_id import json as _json 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 saved_blend = await get_setting(uid, "voice_tts_blend", "") if saved_blend: parsed = _json.loads(saved_blend) if isinstance(parsed, list) and len(parsed) >= 2: voice_blend = parsed except Exception: pass blend_desc = f"blend({len(voice_blend)} voices)" if voice_blend else voice logger.info("TTS synthesis start: %d chars, voice=%s, speed=%.2f", char_count, blend_desc, speed) t0 = time.monotonic() try: wav_bytes = await synthesise(text, voice=voice, speed=speed, voice_blend=voice_blend) except Exception: logger.exception( "TTS synthesis failed: %d chars, voice=%s. Preview: %r", char_count, blend_desc, 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, blend_desc, 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, blend_desc, ) from quart import Response return Response(wav_bytes, mimetype="audio/wav")