Shaman
119 lines
3.5 KiB
Python
119 lines
3.5 KiB
Python
import base64
|
|
import time
|
|
from pathlib import Path
|
|
|
|
from fastapi import APIRouter, Request
|
|
from fastapi.responses import FileResponse, JSONResponse
|
|
|
|
from app.audio import encode, _SUPPORTED_FORMATS
|
|
from app.config import AppConfig
|
|
from app.errors import invalid_request_error
|
|
from app.segmentation import detect_language, prepare_segments
|
|
from app.ssml import PITCH_VALUES, RATE_VALUES, strip_sentinels
|
|
from app.synthesis import SynthesisEngine
|
|
|
|
router = APIRouter()
|
|
|
|
_HTML_PATH = Path(__file__).parent / "webui.html"
|
|
|
|
_MODEL_NAME = "UniosTTS"
|
|
|
|
|
|
def _standard_voices(config: AppConfig) -> list[str]:
|
|
ids: list[str] = []
|
|
for voices_map in config.model_voices.values():
|
|
for v in voices_map:
|
|
if v not in ids:
|
|
ids.append(v)
|
|
return sorted(ids)
|
|
|
|
|
|
def _resolve_speaker(config: AppConfig, model_id: str, voice_name: str) -> str | None:
|
|
voices = config.model_voices.get(model_id, {})
|
|
return voices.get(voice_name) or voices.get("default")
|
|
|
|
|
|
@router.get("/webui")
|
|
async def serve_ui():
|
|
return FileResponse(_HTML_PATH, media_type="text/html")
|
|
|
|
|
|
@router.get("/webui/models")
|
|
async def models():
|
|
return {"models": [_MODEL_NAME]}
|
|
|
|
|
|
@router.get("/webui/voices")
|
|
async def voices(request: Request):
|
|
config: AppConfig = request.app.state.config
|
|
return {"voices": _standard_voices(config)}
|
|
|
|
|
|
@router.post("/webui/speech")
|
|
async def speech(request: Request):
|
|
body = await request.json()
|
|
|
|
text = body.get("input", "")
|
|
if not text.strip():
|
|
return invalid_request_error("input must be a non-empty string")
|
|
|
|
voice = body.get("voice", "default")
|
|
fmt = body.get("response_format", "mp3")
|
|
rate = body.get("rate") or None
|
|
pitch = body.get("pitch") or None
|
|
|
|
if fmt not in _SUPPORTED_FORMATS:
|
|
return invalid_request_error(f"Unsupported response_format: '{fmt}'")
|
|
|
|
if rate is not None and rate not in RATE_VALUES:
|
|
return invalid_request_error(
|
|
f"Unsupported rate: '{rate}' (expected one of {', '.join(RATE_VALUES)})"
|
|
)
|
|
|
|
if pitch is not None and pitch not in PITCH_VALUES:
|
|
return invalid_request_error(
|
|
f"Unsupported pitch: '{pitch}' (expected one of {', '.join(PITCH_VALUES)})"
|
|
)
|
|
|
|
config: AppConfig = request.app.state.config
|
|
engine: SynthesisEngine = request.app.state.engine
|
|
|
|
sr = config.runtime.sample_rate
|
|
ff = config.runtime.ffmpeg_path
|
|
|
|
lang = detect_language(text)
|
|
model = engine.find_model(lang)
|
|
if model is None:
|
|
return invalid_request_error(f"no model supports language '{lang}'")
|
|
|
|
speaker = _resolve_speaker(config, model.model_id, voice)
|
|
if speaker is None:
|
|
return invalid_request_error(
|
|
f"voice '{voice}' not found for model '{model.model_id}'"
|
|
)
|
|
|
|
segments = prepare_segments(text)
|
|
if not segments:
|
|
return invalid_request_error("no speakable text found after processing")
|
|
|
|
normalized_text = strip_sentinels(" ".join(seg.text for seg in segments), " · ")
|
|
|
|
t0 = time.perf_counter()
|
|
pcm = engine.synthesize_segments_model(
|
|
segments, model.model_id, speaker, rate=rate, pitch=pitch
|
|
)
|
|
audio_bytes, _ = encode(pcm, fmt, sr, ff)
|
|
elapsed_ms = int((time.perf_counter() - t0) * 1000)
|
|
|
|
return JSONResponse({
|
|
"audio": base64.b64encode(audio_bytes).decode(),
|
|
"format": fmt,
|
|
"language": lang,
|
|
"model": _MODEL_NAME,
|
|
"voice": voice,
|
|
"speaker": speaker,
|
|
"normalized_text": normalized_text,
|
|
"segments": len(segments),
|
|
"synthesis_time_ms": elapsed_ms,
|
|
})
|