| from __future__ import annotations |
|
|
| from dataclasses import dataclass |
|
|
| from app.errors import OpenAICompatibleError |
|
|
|
|
| @dataclass |
| class EngineRegistry: |
| model_to_engine: dict[str, object] |
|
|
| def get_engine(self, model_name: str) -> object: |
| normalized = model_name.strip().lower() |
| if normalized not in self.model_to_engine: |
| raise OpenAICompatibleError( |
| status_code=400, |
| message=f"Unsupported model '{model_name}'.", |
| param="model", |
| code="model_not_found", |
| ) |
| return self.model_to_engine[normalized] |
|
|
| def list_models(self) -> list[str]: |
| return sorted(self.model_to_engine) |
|
|
| def list_voice_bindings(self, model: str | None = None) -> list[dict[str, str]]: |
| if model: |
| engine = self.get_engine(model) |
| return engine.list_voice_bindings() |
|
|
| seen: set[tuple[str, str]] = set() |
| rows: list[dict[str, str]] = [] |
| for engine in {id(engine): engine for engine in self.model_to_engine.values()}.values(): |
| for row in engine.list_voice_bindings(): |
| key = (row["model"], row["alias"]) |
| if key in seen: |
| continue |
| seen.add(key) |
| rows.append(row) |
| return rows |
|
|
|
|
| _REGISTRY: EngineRegistry | None = None |
|
|
|
|
| def set_engine_registry(registry: EngineRegistry) -> None: |
| global _REGISTRY |
| _REGISTRY = registry |
|
|
|
|
| def get_engine_registry() -> EngineRegistry: |
| if _REGISTRY is None: |
| raise OpenAICompatibleError( |
| status_code=500, |
| message="Engine registry not initialized.", |
| error_type="server_error", |
| code="engine_registry_not_initialized", |
| ) |
| return _REGISTRY |
|
|
