Shaman
127 lines
4.1 KiB
Python
127 lines
4.1 KiB
Python
from fastapi import APIRouter, Request
|
|
from fastapi.responses import Response, StreamingResponse
|
|
from pydantic import BaseModel, ValidationError, field_validator
|
|
|
|
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
|
|
from app.synthesis import SynthesisEngine
|
|
|
|
router = APIRouter()
|
|
|
|
|
|
class SpeechRequest(BaseModel):
|
|
model: str = "UniosTTS"
|
|
voice: str
|
|
input: str
|
|
response_format: str = "mp3"
|
|
speed: float = 1.0
|
|
stream: bool | None = None
|
|
stream_format: str | None = None
|
|
rate: str | None = None
|
|
pitch: str | None = None
|
|
|
|
@field_validator("rate")
|
|
@classmethod
|
|
def _check_rate(cls, v: str | None) -> str | None:
|
|
if v is not None and v not in RATE_VALUES:
|
|
raise ValueError(f"rate must be one of {', '.join(RATE_VALUES)}")
|
|
return v
|
|
|
|
@field_validator("pitch")
|
|
@classmethod
|
|
def _check_pitch(cls, v: str | None) -> str | None:
|
|
if v is not None and v not in PITCH_VALUES:
|
|
raise ValueError(f"pitch must be one of {', '.join(PITCH_VALUES)}")
|
|
return v
|
|
|
|
|
|
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("/v1/audio/voices")
|
|
async def voices(request: Request):
|
|
config: AppConfig = request.app.state.config
|
|
ids: list[str] = []
|
|
for voices_map in config.model_voices.values():
|
|
for v in voices_map:
|
|
if v not in ids:
|
|
ids.append(v)
|
|
ids.sort()
|
|
return {
|
|
"object": "list",
|
|
"data": [{"id": name, "object": "voice"} for name in ids],
|
|
"voices": ids,
|
|
}
|
|
|
|
|
|
@router.post("/v1/audio/speech")
|
|
async def speech(request: Request):
|
|
body = await request.json()
|
|
|
|
try:
|
|
req = SpeechRequest(**body)
|
|
except ValidationError as e:
|
|
msg = str(e.errors()[0].get("msg", "invalid parameter")) if e.errors() else "invalid request"
|
|
return invalid_request_error(msg, status_code=422)
|
|
|
|
if not req.input.strip():
|
|
return invalid_request_error("input must be a non-empty string")
|
|
|
|
if req.response_format not in _SUPPORTED_FORMATS:
|
|
return invalid_request_error(
|
|
f"Unsupported response_format: '{req.response_format}'"
|
|
)
|
|
|
|
if req.speed != 1.0:
|
|
return invalid_request_error(
|
|
"speech rate adjustment is not supported; speed must be 1.0"
|
|
)
|
|
|
|
if req.stream_format == "sse":
|
|
return invalid_request_error("stream_format 'sse' is not supported")
|
|
|
|
config: AppConfig = request.app.state.config
|
|
engine: SynthesisEngine = request.app.state.engine
|
|
|
|
fmt = req.response_format
|
|
sr = config.runtime.sample_rate
|
|
ff = config.runtime.ffmpeg_path
|
|
|
|
is_stream = req.stream is True or req.stream_format == "audio"
|
|
|
|
lang = detect_language(req.input)
|
|
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, req.voice)
|
|
if speaker is None:
|
|
return invalid_request_error(f"voice '{req.voice}' not found for model '{model.model_id}'")
|
|
|
|
segments = prepare_segments(req.input)
|
|
if not segments:
|
|
return invalid_request_error("no speakable text found after processing")
|
|
|
|
if is_stream:
|
|
def _stream_gen():
|
|
for seg in segments:
|
|
seg_pcm = engine.synthesize_segments(
|
|
[seg], speaker, lang, rate=req.rate, pitch=req.pitch
|
|
)
|
|
data, _ = encode(seg_pcm, fmt, sr, ff)
|
|
yield data
|
|
|
|
_, ct = encode(b"\x00" * 4, fmt, sr, ff)
|
|
return StreamingResponse(
|
|
_stream_gen(), media_type=ct, headers={"Transfer-Encoding": "chunked"}
|
|
)
|
|
|
|
pcm = engine.synthesize_segments(segments, speaker, lang, rate=req.rate, pitch=req.pitch)
|
|
data, content_type = encode(pcm, fmt, sr, ff)
|
|
return Response(content=data, media_type=content_type)
|