server.py

import asyncio
import io
import logging
import wave
from functools import partial

from fastapi import FastAPI, HTTPException
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import Response
from fastapi.staticfiles import StaticFiles
from piper import PiperVoice, SynthesisConfig
from pydantic import BaseModel, Field

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

MODEL_PATH = "models/ku/ku_TR/berfin_renas/medium/ku_TR-berfin_renas-medium.onnx"
CONFIG_PATH = "models/ku/ku_TR/berfin_renas/medium/ku_TR-berfin_renas-medium.onnx.json"

app = FastAPI(title="Kurdish TTS")

app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_methods=["POST"],
    allow_headers=["*"],
)

logger.info("Loading voice model...")
voice = PiperVoice.load(MODEL_PATH, config_path=CONFIG_PATH, use_cuda=False)
logger.info("Model ready")


class TTSRequest(BaseModel):
    text: str = Field(..., min_length=1, max_length=500)
    speaker_id: int = Field(default=0, ge=0, le=1)


def _synthesize(text: str, speaker_id: int) -> bytes:
    buf = io.BytesIO()
    with wave.open(buf, "wb") as wav:
        voice.synthesize_wav(text, wav, syn_config=SynthesisConfig(speaker_id=speaker_id))
    buf.seek(0)
    return buf.read()


@app.post("/api/tts")
async def tts(req: TTSRequest):
    text = req.text.strip()
    if not text:
        raise HTTPException(status_code=400, detail="Text cannot be empty")
    loop = asyncio.get_event_loop()
    wav_bytes = await loop.run_in_executor(None, partial(_synthesize, text, req.speaker_id))
    return Response(content=wav_bytes, media_type="audio/wav")


@app.get("/health")
def health():
    return {"status": "ok"}


app.mount("/", StaticFiles(directory="static", html=True), name="static")