fix: a speech model is added with its voices, and Orpheus is sent where it works
An Orpheus id (groq-orpheus-english) did not start with "local-", so generate_tts_audio sent it down the OpenAI path and every call failed. It also could be added with no voice, and the "local-%" filters in /tts/voices and the default lookup hid any non-local voice from learners even when it was added and marked default. services/tts_voices.py is the one table of which voices belong to which model (Kokoro, Orpheus English/Arabic, Fish), the same table the scribe app keeps. Anything the table knows, or anything local-*, goes through the LiteLLM gateway with the options its family needs (Orpheus: wav). Adding a bare model id creates one row per voice with friendly names, so an administrator adds "groq-orpheus-english" and six voices appear to test one by one; a voice from another family is refused, naming the ones that work. Learners are offered every active voice, each saying which model serves it, and /tts/speak answers with the media type the model actually returned. Kitten and Supertonic tables go — those models left the gateway. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01Dv6sqaY6Vq3ChZHMem3cnU
This commit is contained in:
Daniel
parent
6312bf7a00
commit
954b13e7b3
5 changed files with 383 additions and 56 deletions
|
|
@ -14,7 +14,7 @@ from app.database import get_db
|
|||
from app.models.user import User
|
||||
from app.models.ai_model_config import AIModelConfig
|
||||
from app.models.invite import InviteCode
|
||||
from app.services import ai_service, invites, site_settings
|
||||
from app.services import ai_service, invites, site_settings, tts_voices
|
||||
from app.schemas.auth import UserResponse, UserUpdateRole, UserCreate
|
||||
from app.schemas.admin import AIModelConfigCreate, AIModelConfigResponse, AIModelConfigUpdate
|
||||
from app.utils.auth import require_admin, get_current_user, get_password_hash
|
||||
|
|
@ -244,6 +244,20 @@ def create_model(
|
|||
if data.task not in valid_tasks:
|
||||
raise HTTPException(status_code=400, detail=f"Task must be one of: {', '.join(valid_tasks)}")
|
||||
|
||||
if data.task == "tts":
|
||||
if not tts_voices.accepts(data.model_id):
|
||||
model_part, voice_part = tts_voices.split(data.model_id)
|
||||
raise HTTPException(
|
||||
status_code=400,
|
||||
detail=f"{model_part} does not accept the voice {voice_part}. It accepts: "
|
||||
+ ", ".join(v for v, _ in tts_voices.voices_for(model_part)))
|
||||
# A speech model is added with its voices: "groq-orpheus-english"
|
||||
# becomes one row per voice the model accepts, so the administrator
|
||||
# adds a model and the voices appear, ready to be tested one by one.
|
||||
rows = tts_voices.expand(data.model_id, data.name if data.name != data.model_id else None)
|
||||
if len(rows) > 1:
|
||||
return _add_voice_rows(db, data, rows)
|
||||
|
||||
if data.is_default:
|
||||
db.query(AIModelConfig).filter(
|
||||
AIModelConfig.task == data.task,
|
||||
|
|
@ -273,6 +287,32 @@ def create_model(
|
|||
return model
|
||||
|
||||
|
||||
def _add_voice_rows(db: Session, data: AIModelConfigCreate, rows: list[tuple[str, str]]) -> AIModelConfig:
|
||||
"""One row per voice; the first becomes default only if the task had none.
|
||||
|
||||
Returns the first row, which is what the caller shows. Voices already
|
||||
present are left as they are rather than refused, so adding a model twice
|
||||
fills in whatever was missing.
|
||||
"""
|
||||
existing = {m.model_id for m in db.query(AIModelConfig).filter(AIModelConfig.task == "tts").all()}
|
||||
has_default = db.query(AIModelConfig).filter(
|
||||
AIModelConfig.task == "tts", AIModelConfig.is_active == True, AIModelConfig.is_default == True,
|
||||
).count() > 0
|
||||
first = None
|
||||
for model_id, name in rows:
|
||||
if model_id in existing:
|
||||
continue
|
||||
row = AIModelConfig(name=name, model_id=model_id, task="tts", api_key=data.api_key,
|
||||
is_active=data.is_active, is_default=not has_default and first is None)
|
||||
db.add(row)
|
||||
first = first or row
|
||||
if first is None:
|
||||
raise HTTPException(status_code=409, detail=f"Every voice of '{data.model_id}' is already added")
|
||||
db.commit()
|
||||
db.refresh(first)
|
||||
return first
|
||||
|
||||
|
||||
@router.put("/models/{model_id}", response_model=AIModelConfigResponse)
|
||||
def update_model(
|
||||
model_id: int,
|
||||
|
|
@ -423,44 +463,10 @@ class TTSVoiceSearchRequest(BaseModel):
|
|||
region: str | None = None
|
||||
|
||||
|
||||
KOKORO_VOICE_FALLBACKS = [
|
||||
("am_adam", "Kokoro Adam"),
|
||||
("am_michael", "Kokoro Michael"),
|
||||
("af_bella", "Kokoro Bella"),
|
||||
("af_nicole", "Kokoro Nicole"),
|
||||
("bf_emma", "Kokoro Emma"),
|
||||
("bm_lewis", "Kokoro Lewis"),
|
||||
]
|
||||
|
||||
KITTEN_VOICE_FALLBACKS = [
|
||||
("Bella", "Kitten Bella"),
|
||||
("Jasper", "Kitten Jasper"),
|
||||
("Luna", "Kitten Luna"),
|
||||
("Bruno", "Kitten Bruno"),
|
||||
("Rosie", "Kitten Rosie"),
|
||||
("Hugo", "Kitten Hugo"),
|
||||
("Kiki", "Kitten Kiki"),
|
||||
("Leo", "Kitten Leo"),
|
||||
]
|
||||
|
||||
SUPERTONIC_STYLE_FALLBACKS = [
|
||||
("F1", "Supertonic F1"),
|
||||
("F2", "Supertonic F2"),
|
||||
("F3", "Supertonic F3"),
|
||||
("F4", "Supertonic F4"),
|
||||
("F5", "Supertonic F5"),
|
||||
("M1", "Supertonic M1"),
|
||||
("M2", "Supertonic M2"),
|
||||
("M3", "Supertonic M3"),
|
||||
("M4", "Supertonic M4"),
|
||||
("M5", "Supertonic M5"),
|
||||
]
|
||||
|
||||
|
||||
def _kokoro_voice_options(model_name: str) -> list[dict]:
|
||||
base = settings.LOCAL_SPEECH_GATEWAY_URL.rstrip("/")
|
||||
voices = []
|
||||
friendly_names = {voice_id: name for voice_id, name in KOKORO_VOICE_FALLBACKS}
|
||||
friendly_names = {voice_id: name for voice_id, name in tts_voices.KOKORO_VOICES}
|
||||
if base:
|
||||
try:
|
||||
resp = httpx.get(f"{base}/v1/audio/voices", timeout=10)
|
||||
|
|
@ -475,7 +481,7 @@ def _kokoro_voice_options(model_name: str) -> list[dict]:
|
|||
if not voices:
|
||||
voices = [
|
||||
{"voice": voice_id, "name": name, "profile": "kokoro"}
|
||||
for voice_id, name in KOKORO_VOICE_FALLBACKS
|
||||
for voice_id, name in tts_voices.KOKORO_VOICES
|
||||
]
|
||||
|
||||
return [
|
||||
|
|
@ -532,11 +538,12 @@ def search_tts_voices(
|
|||
if model_name == "local-kokoro-tts":
|
||||
voices.extend(_kokoro_voice_options(model_name))
|
||||
continue
|
||||
if model_name == "local-kitten-tts":
|
||||
voices.extend(_static_voice_options(model_name, KITTEN_VOICE_FALLBACKS))
|
||||
continue
|
||||
if model_name == "local-supertonic-tts":
|
||||
voices.extend(_static_voice_options(model_name, SUPERTONIC_STYLE_FALLBACKS))
|
||||
# Kitten and Supertonic were retired from the gateway; Orpheus
|
||||
# and Fish are what it proxies now, and their voices are the
|
||||
# family table's.
|
||||
known = tts_voices.voices_for(model_name)
|
||||
if known:
|
||||
voices.extend(_static_voice_options(model_name, known))
|
||||
continue
|
||||
voices.append({
|
||||
"model_id": model_name,
|
||||
|
|
|
|||
|
|
@ -6,7 +6,7 @@ from sqlalchemy.orm import Session
|
|||
from app.database import get_db
|
||||
from app.models.user import User
|
||||
from app.models.ai_model_config import AIModelConfig
|
||||
from app.services import ai_service
|
||||
from app.services import ai_service, tts_voices
|
||||
from app.utils.auth import get_current_user, check_rate_limit
|
||||
|
||||
router = APIRouter()
|
||||
|
|
@ -32,11 +32,13 @@ def _task_model(db: Session, task: str, fallback: str) -> tuple[str, str | None]
|
|||
|
||||
|
||||
def _default_tts_model(db: Session) -> tuple[str, str | None]:
|
||||
# Whatever the administrator marked default. The "local-%" filter that
|
||||
# used to sit here meant an Orpheus default was ignored in favour of the
|
||||
# hard-coded Kokoro voice below, silently.
|
||||
config = db.query(AIModelConfig).filter(
|
||||
AIModelConfig.task == "tts",
|
||||
AIModelConfig.is_active == True,
|
||||
AIModelConfig.is_default == True,
|
||||
AIModelConfig.model_id.like("local-%"),
|
||||
).first()
|
||||
if config:
|
||||
return config.model_id, config.api_key or None
|
||||
|
|
@ -48,14 +50,18 @@ def get_voices(
|
|||
db: Session = Depends(get_db),
|
||||
current_user: User = Depends(get_current_user),
|
||||
):
|
||||
"""Return LiteLLM-routed TTS models only."""
|
||||
"""Every active voice, each saying which model serves it.
|
||||
|
||||
All of them, not only the locally served ones: a learner is offered every
|
||||
voice an administrator added, whichever model it belongs to.
|
||||
"""
|
||||
query = db.query(AIModelConfig).filter(
|
||||
AIModelConfig.task == "tts",
|
||||
AIModelConfig.is_active == True,
|
||||
AIModelConfig.model_id.like("local-%"),
|
||||
)
|
||||
db_models = query.order_by(AIModelConfig.is_default.desc(), AIModelConfig.name).all()
|
||||
return [{"id": m.model_id, "name": m.name, "is_default": m.is_default} for m in db_models]
|
||||
return [{"id": m.model_id, "name": m.name, "is_default": m.is_default,
|
||||
"model": tts_voices.split(m.model_id)[0]} for m in db_models]
|
||||
|
||||
|
||||
@router.post("/speak")
|
||||
|
|
@ -101,7 +107,7 @@ def text_to_speech(
|
|||
if audio is None:
|
||||
raise HTTPException(status_code=500, detail="TTS generation failed. Check model configuration.")
|
||||
|
||||
return Response(content=audio, media_type="audio/mpeg")
|
||||
return Response(content=audio, media_type=tts_voices.media_type(model_id))
|
||||
|
||||
|
||||
@router.post("/transcribe")
|
||||
|
|
|
|||
|
|
@ -5,6 +5,7 @@ import os
|
|||
import httpx
|
||||
|
||||
from app.config import settings
|
||||
from app.services import tts_voices
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
|
@ -504,12 +505,14 @@ def generate_tts_audio(
|
|||
|
||||
use_model = model_id or "tts-1:alloy"
|
||||
|
||||
# ── Local LiteLLM TTS models ─────────────────────────────────
|
||||
if use_model.startswith("local-"):
|
||||
local_model = use_model
|
||||
local_voice = "am_adam" if use_model.startswith("local-kokoro-tts") else "alloy"
|
||||
if ":" in use_model:
|
||||
local_model, local_voice = use_model.split(":", 1)
|
||||
# ── LiteLLM TTS models ───────────────────────────────────────
|
||||
# local-* and every family tts_voices knows (Orpheus, Fish). Orpheus used
|
||||
# to miss this branch for not starting with "local-" and was sent down the
|
||||
# OpenAI path, where it failed on every call.
|
||||
if tts_voices.is_litellm_tts(use_model):
|
||||
local_model, local_voice = tts_voices.split(use_model)
|
||||
if not local_voice:
|
||||
local_voice = tts_voices.default_voice(local_model)
|
||||
key = api_key or settings.LITELLM_API_KEY
|
||||
base = (settings.LITELLM_API_BASE or "").rstrip("/").removesuffix("/v1")
|
||||
if not base:
|
||||
|
|
@ -519,13 +522,13 @@ def generate_tts_audio(
|
|||
resp = httpx.post(
|
||||
f"{base}/v1/audio/speech",
|
||||
headers={"Authorization": f"Bearer {key}", "Content-Type": "application/json"} if key else {"Content-Type": "application/json"},
|
||||
json={"model": local_model, "input": text, "voice": local_voice, "response_format": "mp3"},
|
||||
json={"model": local_model, "input": text, "voice": local_voice, **tts_voices.request_options(local_model)},
|
||||
timeout=90,
|
||||
)
|
||||
resp.raise_for_status()
|
||||
return resp.content
|
||||
except Exception as e:
|
||||
logger.error(f"Local LiteLLM TTS failed: {e}")
|
||||
logger.error(f"LiteLLM TTS failed for {local_model}:{local_voice}: {e}")
|
||||
return None
|
||||
|
||||
# ── Local Sherpa gateway ───────────────────────────────────
|
||||
|
|
|
|||
146
backend/app/services/tts_voices.py
Normal file
146
backend/app/services/tts_voices.py
Normal file
|
|
@ -0,0 +1,146 @@
|
|||
"""Which voices belong to which LiteLLM speech model.
|
||||
|
||||
A voice is a property of a model, not of the gateway, and the two used to be
|
||||
treated as interchangeable: a model id like ``groq-orpheus-english`` could be
|
||||
added with no voice at all, and ``generate_tts_audio`` then sent it down the
|
||||
OpenAI path because it did not start with ``local-``. Every one of those calls
|
||||
failed, which is what "voice is not set up properly" was.
|
||||
|
||||
The gateway cannot help: ``/model/info`` reports ``audio_speech`` for all of
|
||||
these and carries no voice field. So the mapping lives here, keyed by family,
|
||||
and every list was taken from the provider rather than from documentation —
|
||||
Groq names its own when refused ("voice must be one of the following voices:
|
||||
[...]"), Fish accepts ``alloy`` and refuses the other OpenAI names, and
|
||||
Kokoro's come from the speech gateway's ``/v1/audio/voices`` (admin.py asks it
|
||||
directly and prefers that answer whenever it arrives).
|
||||
|
||||
The same table lives in the scribe app (src/utils/ttsProvider.js). Keep the
|
||||
two in step when a model is added to the gateway.
|
||||
|
||||
Model ids keep the app's ``model:voice`` convention (``local-kokoro-tts:am_adam``).
|
||||
"""
|
||||
|
||||
KOKORO_VOICES = [
|
||||
("am_adam", "Kokoro Adam"), ("am_michael", "Kokoro Michael"),
|
||||
("af_bella", "Kokoro Bella"), ("af_nicole", "Kokoro Nicole"),
|
||||
("bf_emma", "Kokoro Emma"), ("bm_lewis", "Kokoro Lewis"),
|
||||
]
|
||||
ORPHEUS_ENGLISH_VOICES = [
|
||||
("autumn", "Orpheus Autumn"), ("diana", "Orpheus Diana"), ("hannah", "Orpheus Hannah"),
|
||||
("austin", "Orpheus Austin"), ("daniel", "Orpheus Daniel"), ("troy", "Orpheus Troy"),
|
||||
]
|
||||
ORPHEUS_ARABIC_VOICES = [
|
||||
("abdullah", "Orpheus Abdullah"), ("fahad", "Orpheus Fahad"), ("sultan", "Orpheus Sultan"),
|
||||
("lulwa", "Orpheus Lulwa"), ("noura", "Orpheus Noura"), ("aisha", "Orpheus Aisha"),
|
||||
]
|
||||
FISH_VOICES = [("alloy", "Fish Alloy")]
|
||||
|
||||
#: family → (gateway ids that belong to it, its voices, options the request needs)
|
||||
FAMILIES = {
|
||||
"kokoro": {
|
||||
"ids": ("local-kokoro-tts",),
|
||||
"voices": KOKORO_VOICES,
|
||||
"request": {"response_format": "mp3"},
|
||||
"media_type": "audio/mpeg",
|
||||
},
|
||||
"orpheus-english": {
|
||||
"ids": ("groq-orpheus-english", "canopylabs/orpheus-v1-english"),
|
||||
"voices": ORPHEUS_ENGLISH_VOICES,
|
||||
# Groq refuses mp3 for Orpheus; wav is what it returns.
|
||||
"request": {"response_format": "wav"},
|
||||
"media_type": "audio/wav",
|
||||
},
|
||||
"orpheus-arabic": {
|
||||
"ids": ("groq-orpheus-arabic-saudi", "canopylabs/orpheus-arabic-saudi"),
|
||||
"voices": ORPHEUS_ARABIC_VOICES,
|
||||
"request": {"response_format": "wav"},
|
||||
"media_type": "audio/wav",
|
||||
},
|
||||
"fish": {
|
||||
"ids": ("openrouter-fish-s2.1-pro-tts", "fish-audio/s2.1-pro"),
|
||||
"voices": FISH_VOICES,
|
||||
"request": {"response_format": "mp3"},
|
||||
"media_type": "audio/mpeg",
|
||||
},
|
||||
}
|
||||
|
||||
|
||||
def split(model_id: str | None) -> tuple[str, str]:
|
||||
"""``local-kokoro-tts:am_adam`` → (``local-kokoro-tts``, ``am_adam``); no colon → (id, '')."""
|
||||
text = (model_id or "").strip()
|
||||
if ":" in text:
|
||||
model, voice = text.split(":", 1)
|
||||
return model.strip(), voice.strip()
|
||||
return text, ""
|
||||
|
||||
|
||||
def family(model: str | None) -> str | None:
|
||||
model, _ = split(model)
|
||||
key = model.lower()
|
||||
for name, spec in FAMILIES.items():
|
||||
if key in spec["ids"]:
|
||||
return name
|
||||
return None
|
||||
|
||||
|
||||
def is_litellm_tts(model_id: str | None) -> bool:
|
||||
"""Routed through the LiteLLM gateway's /v1/audio/speech.
|
||||
|
||||
``local-*`` is the gateway's own convention for models it serves itself;
|
||||
a known family is one it proxies. Both take the same request.
|
||||
"""
|
||||
model, _ = split(model_id)
|
||||
return model.startswith("local-") or family(model) is not None
|
||||
|
||||
|
||||
def voices_for(model: str | None) -> list[tuple[str, str]]:
|
||||
"""(voice, friendly name) pairs the model accepts, or [] when unknown."""
|
||||
spec = FAMILIES.get(family(model) or "")
|
||||
return list(spec["voices"]) if spec else []
|
||||
|
||||
|
||||
def default_voice(model: str | None) -> str:
|
||||
voices = voices_for(model)
|
||||
if voices:
|
||||
return voices[0][0]
|
||||
# A local model we have no list for still needs something the gateway
|
||||
# will take; alloy is what an OpenAI-shaped server defaults to.
|
||||
return "alloy"
|
||||
|
||||
|
||||
def request_options(model_id: str | None) -> dict:
|
||||
spec = FAMILIES.get(family(model_id) or "")
|
||||
return dict(spec["request"]) if spec else {"response_format": "mp3"}
|
||||
|
||||
|
||||
def media_type(model_id: str | None) -> str:
|
||||
spec = FAMILIES.get(family(model_id) or "")
|
||||
return spec["media_type"] if spec else "audio/mpeg"
|
||||
|
||||
|
||||
def accepts(model_id: str) -> bool:
|
||||
"""A ``model:voice`` pair the model will not refuse.
|
||||
|
||||
A bare id whose family we know is fine (it gets its first voice); a voice
|
||||
from another family is the one thing this says no to.
|
||||
"""
|
||||
model, voice = split(model_id)
|
||||
known = voices_for(model)
|
||||
if not voice or not known:
|
||||
return True
|
||||
return voice in {v for v, _ in known}
|
||||
|
||||
|
||||
def expand(model_id: str, name: str | None = None) -> list[tuple[str, str]]:
|
||||
"""The rows adding ``model_id`` should create: one per voice.
|
||||
|
||||
Adding ``groq-orpheus-english`` yields six ``groq-orpheus-english:<voice>``
|
||||
rows with friendly names, so the admin's "add a model" is what populates
|
||||
the voices — nobody types six ids. An id that already names a voice, or a
|
||||
model with no known list, is one row as given.
|
||||
"""
|
||||
model, voice = split(model_id)
|
||||
known = voices_for(model)
|
||||
if voice or not known:
|
||||
return [(model_id, name or model_id)]
|
||||
return [(f"{model}:{v}", label) for v, label in known]
|
||||
165
backend/tests/test_tts_voices.py
Normal file
165
backend/tests/test_tts_voices.py
Normal file
|
|
@ -0,0 +1,165 @@
|
|||
"""A voice belongs to a model, and a model is added with its voices.
|
||||
|
||||
Disposable SQLite. The rules worth pinning: an Orpheus id is routed through
|
||||
the LiteLLM gateway (it used to fall through to the OpenAI path for not
|
||||
starting with "local-"), a bare model id becomes one row per voice, a voice
|
||||
from another family is refused with the list that would work, and learners
|
||||
are offered every active voice — not only the locally served ones.
|
||||
"""
|
||||
import unittest
|
||||
from unittest.mock import Mock, patch
|
||||
|
||||
import test_quiz_builder as fixtures # noqa: F401 (sets DATABASE_URL before app imports)
|
||||
from fastapi import FastAPI
|
||||
from fastapi.testclient import TestClient
|
||||
from sqlalchemy import create_engine
|
||||
from sqlalchemy.orm import Session
|
||||
from sqlalchemy.pool import StaticPool
|
||||
|
||||
from app.database import Base, get_db
|
||||
from app.models.ai_model_config import AIModelConfig
|
||||
from app.models.user import User
|
||||
from app.routers import admin, tts
|
||||
from app.services import ai_service, tts_voices
|
||||
from app.utils.auth import get_current_user, require_admin
|
||||
|
||||
|
||||
class VoiceTableTests(unittest.TestCase):
|
||||
def test_families_and_their_voices(self):
|
||||
self.assertEqual(tts_voices.family("groq-orpheus-english:hannah"), "orpheus-english")
|
||||
self.assertEqual(tts_voices.family("canopylabs/orpheus-arabic-saudi"), "orpheus-arabic")
|
||||
self.assertEqual(tts_voices.family("openrouter-fish-s2.1-pro-tts"), "fish")
|
||||
self.assertEqual(tts_voices.family("local-kokoro-tts:am_adam"), "kokoro")
|
||||
self.assertIsNone(tts_voices.family("tts-1:alloy"))
|
||||
self.assertEqual([v for v, _ in tts_voices.voices_for("groq-orpheus-english")],
|
||||
["autumn", "diana", "hannah", "austin", "daniel", "troy"])
|
||||
self.assertEqual(tts_voices.voices_for("openrouter-fish-s2.1-pro-tts"), [("alloy", "Fish Alloy")])
|
||||
self.assertEqual(tts_voices.voices_for("something-new"), [])
|
||||
|
||||
def test_litellm_routing_and_request_shape(self):
|
||||
self.assertTrue(tts_voices.is_litellm_tts("groq-orpheus-english:daniel"))
|
||||
self.assertTrue(tts_voices.is_litellm_tts("local-anything:x"))
|
||||
self.assertFalse(tts_voices.is_litellm_tts("tts-1:alloy"))
|
||||
self.assertFalse(tts_voices.is_litellm_tts("elevenlabs/rachel"))
|
||||
# Groq refuses mp3 for Orpheus; the gateway's own models take mp3.
|
||||
self.assertEqual(tts_voices.request_options("groq-orpheus-english"), {"response_format": "wav"})
|
||||
self.assertEqual(tts_voices.media_type("groq-orpheus-english:troy"), "audio/wav")
|
||||
self.assertEqual(tts_voices.request_options("local-kokoro-tts:af_bella"), {"response_format": "mp3"})
|
||||
self.assertEqual(tts_voices.media_type("local-kokoro-tts"), "audio/mpeg")
|
||||
|
||||
def test_accepts_refuses_only_another_familys_voice(self):
|
||||
self.assertTrue(tts_voices.accepts("groq-orpheus-english:hannah"))
|
||||
self.assertFalse(tts_voices.accepts("groq-orpheus-english:am_adam"))
|
||||
self.assertFalse(tts_voices.accepts("local-kokoro-tts:hannah"))
|
||||
self.assertTrue(tts_voices.accepts("groq-orpheus-english")) # gets its first voice
|
||||
self.assertTrue(tts_voices.accepts("local-newmodel-tts:whatever")) # no list: nothing to refuse
|
||||
|
||||
def test_expand_makes_one_row_per_voice(self):
|
||||
rows = tts_voices.expand("groq-orpheus-arabic-saudi")
|
||||
self.assertEqual(len(rows), 6)
|
||||
self.assertEqual(rows[0], ("groq-orpheus-arabic-saudi:abdullah", "Orpheus Abdullah"))
|
||||
# An id that already names a voice, or a model we have no list for, is one row as given.
|
||||
self.assertEqual(tts_voices.expand("local-kokoro-tts:am_adam", "Kokoro Adam"), [("local-kokoro-tts:am_adam", "Kokoro Adam")])
|
||||
self.assertEqual(tts_voices.expand("local-newmodel-tts"), [("local-newmodel-tts", "local-newmodel-tts")])
|
||||
|
||||
|
||||
class GenerateAudioRoutingTests(unittest.TestCase):
|
||||
def _post(self, model_id):
|
||||
calls = []
|
||||
|
||||
def fake_post(url, **kw):
|
||||
calls.append((url, kw.get("json")))
|
||||
resp = Mock(); resp.content = b"audio"; resp.raise_for_status = Mock()
|
||||
return resp
|
||||
|
||||
with patch.object(ai_service.settings, "LITELLM_API_BASE", "http://gateway:4000/v1"), \
|
||||
patch.object(ai_service.settings, "LITELLM_API_KEY", "k"), \
|
||||
patch.object(ai_service.httpx, "post", side_effect=fake_post):
|
||||
out = ai_service.generate_tts_audio("Stridor.", model_id=model_id)
|
||||
return out, calls
|
||||
|
||||
def test_orpheus_goes_to_the_gateway_as_wav(self):
|
||||
out, calls = self._post("groq-orpheus-english:hannah")
|
||||
self.assertEqual(out, b"audio")
|
||||
url, body = calls[0]
|
||||
self.assertEqual(url, "http://gateway:4000/v1/audio/speech")
|
||||
self.assertEqual(body, {"model": "groq-orpheus-english", "input": "Stridor.", "voice": "hannah", "response_format": "wav"})
|
||||
|
||||
def test_bare_model_gets_its_first_voice(self):
|
||||
_, calls = self._post("groq-orpheus-english")
|
||||
self.assertEqual(calls[0][1]["voice"], "autumn")
|
||||
_, calls = self._post("local-kokoro-tts")
|
||||
self.assertEqual(calls[0][1]["voice"], "am_adam")
|
||||
self.assertEqual(calls[0][1]["response_format"], "mp3")
|
||||
|
||||
|
||||
class AdminAddsAModelWithItsVoicesTests(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.engine = create_engine("sqlite://", connect_args={"check_same_thread": False}, poolclass=StaticPool)
|
||||
Base.metadata.create_all(self.engine)
|
||||
self.db = Session(self.engine)
|
||||
self.admin = User(id=1, name="Admin", email="admin@example.test", hashed_password="x", role="admin")
|
||||
self.db.add(self.admin); self.db.commit()
|
||||
app = FastAPI()
|
||||
app.include_router(admin.router, prefix="/admin")
|
||||
app.include_router(tts.router, prefix="/tts")
|
||||
app.dependency_overrides[get_db] = lambda: self.db
|
||||
app.dependency_overrides[require_admin] = lambda: self.admin
|
||||
app.dependency_overrides[get_current_user] = lambda: self.admin
|
||||
self.client = TestClient(app)
|
||||
|
||||
def tearDown(self):
|
||||
self.client.close()
|
||||
|
||||
def _add(self, model_id, **extra):
|
||||
return self.client.post("/admin/models", json={"name": model_id, "model_id": model_id, "task": "tts", **extra})
|
||||
|
||||
def _rows(self):
|
||||
return [(m.model_id, m.name, m.is_default) for m in
|
||||
self.db.query(AIModelConfig).filter(AIModelConfig.task == "tts").order_by(AIModelConfig.id).all()]
|
||||
|
||||
def test_a_bare_speech_model_becomes_its_voices(self):
|
||||
res = self._add("groq-orpheus-english")
|
||||
self.assertEqual(res.status_code, 200, res.text)
|
||||
rows = self._rows()
|
||||
self.assertEqual([r[0] for r in rows], [f"groq-orpheus-english:{v}" for v in ["autumn", "diana", "hannah", "austin", "daniel", "troy"]])
|
||||
self.assertEqual(rows[2][1], "Orpheus Hannah")
|
||||
# The first voice is default only because the task had none.
|
||||
self.assertEqual([r[2] for r in rows], [True, False, False, False, False, False])
|
||||
self.assertEqual(res.json()["model_id"], "groq-orpheus-english:autumn")
|
||||
|
||||
def test_adding_a_second_model_keeps_the_existing_default(self):
|
||||
self._add("local-kokoro-tts:am_adam", is_default=True)
|
||||
self._add("groq-orpheus-english")
|
||||
rows = self._rows()
|
||||
self.assertEqual([r[0] for r in rows if r[2]], ["local-kokoro-tts:am_adam"])
|
||||
self.assertEqual(len(rows), 7)
|
||||
# Again: nothing new to add.
|
||||
self.assertEqual(self._add("groq-orpheus-english").status_code, 409)
|
||||
|
||||
def test_a_voice_from_another_family_is_refused_with_the_right_list(self):
|
||||
res = self._add("local-kokoro-tts:hannah")
|
||||
self.assertEqual(res.status_code, 400)
|
||||
self.assertIn("It accepts: am_adam, am_michael", res.json()["detail"])
|
||||
self.assertEqual(self._rows(), [])
|
||||
|
||||
def test_learners_are_offered_every_active_voice(self):
|
||||
self._add("local-kokoro-tts:am_adam")
|
||||
self._add("groq-orpheus-english")
|
||||
offered = self.client.get("/tts/voices").json()
|
||||
self.assertEqual(len(offered), 7)
|
||||
self.assertEqual({v["model"] for v in offered}, {"local-kokoro-tts", "groq-orpheus-english"})
|
||||
self.assertEqual(offered[0], {"id": "local-kokoro-tts:am_adam", "name": "local-kokoro-tts:am_adam", "is_default": True, "model": "local-kokoro-tts"})
|
||||
|
||||
def test_speak_answers_with_the_media_type_the_model_returns(self):
|
||||
self._add("groq-orpheus-english")
|
||||
with patch.object(tts.ai_service, "generate_tts_audio", return_value=b"RIFF") as gen, \
|
||||
patch.object(tts, "check_rate_limit", lambda **kw: None):
|
||||
res = self.client.post("/tts/speak", json={"text": "Stridor.", "voice": "groq-orpheus-english:troy"})
|
||||
self.assertEqual(res.status_code, 200)
|
||||
self.assertEqual(res.headers["content-type"], "audio/wav")
|
||||
self.assertEqual(gen.call_args.kwargs["model_id"], "groq-orpheus-english:troy")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Loading…
Reference in a new issue