190 lines
6.6 KiB
Python
190 lines
6.6 KiB
Python
"""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")
|