#!/usr/bin/env python3 """ Base TTS Provider Abstract Class for OpenVoiceUI. This module defines the abstract interface that all TTS providers must implement. It provides a consistent API for generating speech, listing available voices, and retrieving provider information. Author: OpenVoiceUI Date: 2026-02-11 """ from abc import ABC, abstractmethod from typing import Any, Dict, List, Optional from dataclasses import dataclass, field @dataclass class TTSVoice: """ Represents a single voice available from a TTS provider. Attributes: id: Unique identifier for the voice (e.g., 'M1', 'your-hume-voice-id') name: Human-readable name (e.g., 'Male Voice 1', 'Custom Voice') language: Language code (e.g., 'en-US', 'en', 'es') gender: Gender of the voice ('male', 'female', 'neutral', or None) description: Optional description of the voice characteristics """ id: str name: str language: str = 'en' gender: Optional[str] = None description: Optional[str] = None def to_dict(self) -> Dict[str, Any]: """Convert the voice to a dictionary representation.""" return { 'id': self.id, 'name': self.name, 'language': self.language, 'gender': self.gender, 'description': self.description } @dataclass class TTSProviderInfo: """ Metadata about a TTS provider. Attributes: name: Provider name (e.g., 'supertonic', 'hume') display_name: Human-readable name (e.g., 'Supertonic TTS', 'Hume EVI') version: Provider version string cost_per_minute: Cost in USD per minute of generated audio quality: Quality rating ('low', 'medium', 'high', 'premium') latency: Expected latency ('instant', 'fast', 'medium', 'slow') features: List of feature strings (e.g., ['emotion-aware', 'multi-language']) requires_api_key: Whether the provider requires an API key is_online: Whether the provider requires internet connectivity status: Current status ('active', 'inactive', 'error') """ name: str display_name: str version: str cost_per_minute: float quality: str latency: str features: List[str] requires_api_key: bool is_online: bool status: str = 'active' def to_dict(self) -> Dict[str, Any]: """Convert the provider info to a dictionary representation.""" return { 'name': self.name, 'display_name': self.display_name, 'version': self.version, 'cost_per_minute': self.cost_per_minute, 'quality': self.quality, 'latency': self.latency, 'features': self.features, 'requires_api_key': self.requires_api_key, 'is_online': self.is_online, 'status': self.status } class TTSProvider(ABC): """ Abstract base class for Text-to-Speech providers. All TTS providers must inherit from this class and implement the required methods. This ensures a consistent interface across different TTS backends. Required Methods: - generate_speech(text, **kwargs): Convert text to audio bytes - list_voices(): Return list of available voice names - get_info(): Return provider metadata (name, status, capabilities) Example: >>> class MyTTS(TTSProvider): ... def generate_speech(self, text, **kwargs): ... # Implementation here ... return audio_bytes ... def list_voices(self): ... return ['voice1', 'voice2'] ... def get_info(self): ... return {'name': 'MyTTS', 'status': 'active'} """ @abstractmethod def generate_speech(self, text: str, **kwargs) -> bytes: """ Generate speech audio from the given text. Args: text: The text to synthesize into speech. **kwargs: Provider-specific parameters (voice, speed, lang, etc.) Returns: bytes: Raw audio data (usually WAV format) that can be written to a file or sent via HTTP with Content-Type: audio/wav. Raises: ValueError: If text is empty or parameters are invalid. RuntimeError: If speech generation fails. Example: >>> audio = provider.generate_speech("Hello world", voice='M1') >>> with open('output.wav', 'wb') as f: ... f.write(audio) """ pass @abstractmethod def list_voices(self) -> List[str]: """ Return a list of available voice names for this provider. Returns: List[str]: List of voice identifiers (e.g., ['M1', 'M2', 'F1']). These IDs should be valid values for a 'voice' parameter in generate_speech(). Example: >>> provider.list_voices() ['M1', 'M2', 'F1', 'F2'] """ pass @abstractmethod def get_info(self) -> Dict[str, Any]: """ Return metadata about this TTS provider. Returns: Dict with at minimum: - 'name': str - Provider display name - 'status': str - 'active', 'inactive', or 'error' - 'description': str - Brief description of the provider - 'capabilities': dict - Optional feature flags - 'streaming': bool - Supports streaming audio - 'ssml': bool - Supports SSML markup - 'custom_voices': bool - Supports custom voice cloning - 'languages': List[str] - Supported language codes Example: >>> provider.get_info() { 'name': 'Supertonic', 'status': 'active', 'description': 'Local ONNX-based TTS engine', 'capabilities': { 'streaming': False, 'ssml': False, 'custom_voices': True, 'languages': ['en', 'ko', 'es', 'pt', 'fr'] } } """ pass def is_available(self) -> bool: """ Check if this provider is currently available for use. Returns: bool: True if the provider is active and ready, False otherwise. Default implementation checks if get_info()['status'] == 'active'. Subclasses can override for more complex availability checks. Example: >>> if provider.is_available(): ... audio = provider.generate_speech("Hello") """ return self.get_info().get('status', 'inactive') == 'active' def validate_text(self, text: str) -> None: """ Validate that text is suitable for speech generation. Args: text: Text to validate. Raises: ValueError: If text is None, empty, or only whitespace. Example: >>> provider.validate_text("Hello world") # OK >>> provider.validate_text("") # Raises ValueError """ if text is None: raise ValueError("Text cannot be None") if not isinstance(text, str): raise ValueError(f"Text must be a string, got {type(text).__name__}") if not text.strip(): raise ValueError("Text cannot be empty or contain only whitespace") def validate_voice(self, voice: str) -> bool: """ Check if a given voice name is valid for this provider. Args: voice: Voice identifier to validate. Returns: bool: True if the voice is available, False otherwise. Example: >>> provider.validate_voice('M1') True >>> provider.validate_voice('invalid') False """ return voice in self.list_voices() def get_default_voice(self) -> Optional[str]: """ Return the default voice for this provider. Returns: The first voice in list_voices(), or None if no voices available. Example: >>> provider.get_default_voice() 'M1' """ voices = self.list_voices() return voices[0] if voices else None def __repr__(self) -> str: """String representation of the provider.""" info = self.get_info() return f"{self.__class__.__name__}(name='{info.get('name', 'Unknown')}', status='{info.get('status', 'unknown')}')" class TTSProviderError(Exception): """ Base exception class for TTS provider errors. This exception is raised when a TTS provider encounters an error that is specific to the provider implementation. """ def __init__(self, provider_name: str, message: str): """ Initialize the exception. Args: provider_name: Name of the provider that raised the error. message: Error message describing what went wrong. """ self.provider_name = provider_name super().__init__(f"[{provider_name}] {message}") class TTSGenerationError(TTSProviderError): """ Raised when speech generation fails. This can occur due to invalid input, network issues, or problems with the TTS service. """ pass class TTSConfigurationError(TTSProviderError): """ Raised when the provider is misconfigured. This can occur due to missing API keys, invalid paths, or other configuration issues. """ pass class TTSVoiceNotFoundError(TTSProviderError): """ Raised when a requested voice is not available. This occurs when a voice ID is specified that doesn't exist in the provider's voice catalog. """ pass # =========================================================================== # Voice-clone PROVIDER-SEAM (docs/jambot/ovu-voice-clone-seam-spec.md §2) # =========================================================================== # A provider that can create a voice clone implements CloneCapableTTSProvider. # POST /api/tts/clone selects providers through the registry (catalog feature # "voice-cloning" + a class implementing this ABC + availability) instead of a # hard-coded tuple, so a new provider (e.g. the GPU lane's qwen3-gpu) plugs in # without touching the route. #: Error enum, spec §2.4, plus the six codes of the GPU wire (mac-claude's #: voice-clone-gpu-wire-v0 §4, accepted by host 2026-11-07). Each code maps to #: one HTTP status in the route. CLONE_ERROR_CODES = ( 'unavailable', 'busy', 'bad_audio', 'quota', 'consent_mismatch', 'timeout', 'internal', # GPU wire v0 §4 additions 'unauthorized', # upstream bearer bad/revoked: a CONFIG alarm 'tenant_mismatch', # this container holds another tenant's token 'bad_request', # client bug: missing tenant_id, bad name/voice_id 'not_found', # unknown or foreign id (no oracle on other tenants) 'disabled', # taken down; never restore 'request_conflict', # same request_id, different audio or name ) _CLONE_ERROR_RETRYABLE = { 'unavailable': True, 'busy': True, 'bad_audio': False, 'quota': False, 'consent_mismatch': False, 'timeout': True, 'internal': False, 'unauthorized': False, 'tenant_mismatch': False, 'bad_request': False, 'not_found': False, 'disabled': False, 'request_conflict': False, } class CloneError(Exception): """A typed clone failure (spec §2.4). Raised by ``CloneCapableTTSProvider.create_clone`` (or derived from a ``CloneResult`` with ``status == 'failed'``). The route maps ``code`` to an HTTP status; anything that is NOT a CloneError keeps today's behaviour (RuntimeError -> 500 with its message, anything else -> opaque 500). Args: code: one of CLONE_ERROR_CODES. message: human-readable detail (never audio bytes, never an access code). retryable: defaults per code (unavailable/busy/timeout retryable). retry_after: seconds, sent as the Retry-After header when set (always sent for ``busy``; the route defaults it). """ def __init__(self, code: str, message: str = '', retryable: Optional[bool] = None, retry_after: Optional[int] = None): if code not in CLONE_ERROR_CODES: raise ValueError( f"Unknown CloneError code {code!r}; expected one of {CLONE_ERROR_CODES}" ) self.code = code self.message = message or code self.retryable = (_CLONE_ERROR_RETRYABLE[code] if retryable is None else bool(retryable)) self.retry_after = retry_after super().__init__(f"[{code}] {self.message}") def to_dict(self) -> Dict[str, Any]: return {'code': self.code, 'retryable': self.retryable, 'message': self.message} @classmethod def from_dict(cls, err: Optional[Dict[str, Any]]) -> 'CloneError': err = err or {} code = err.get('code', 'internal') if code not in CLONE_ERROR_CODES: code = 'internal' return cls(code, err.get('message', ''), err.get('retryable'), err.get('retry_after')) @dataclass class CloneRequest: """One clone request (spec §2.2). ``tenant_id`` comes from the OVU container env, NEVER from the client body. ``audio_url`` is not in spec §2.2 but is required by the fal.ai qwen3 provider, which takes a URL rather than a file; it is the public URL of the same saved upload (or the caller's URL on the JSON path). Fields marked TODO are carried so the shape is stable, but nothing in this release populates or enforces them (access codes: spec §1, consent: §3, duration caps: §3.4). """ tenant_id: str request_id: str name: str audio_path: Optional[str] = None audio_url: Optional[str] = None audio_sha256: Optional[str] = None # TODO spec §2.2/§4 duration_s: Optional[float] = None # TODO spec §3.4 (ffprobe) reference_text: Optional[str] = None language: Optional[str] = None consent_id: Optional[str] = None # TODO spec §3 code_id: Optional[str] = None # TODO spec §1 #: The phrase the speaker was asked to recite (spec §3.3). When set it is #: the preferred reference_text for providers that need a transcript #: (qwen3-gpu). Populated by the consent/recitation PR; None until then. recitation_phrase: Optional[str] = None @dataclass class CloneResult: """Outcome of ``create_clone`` / ``get_clone_status`` (spec §2.2). ``metadata`` is the provider's own record, verbatim (e.g. the dict the legacy ``clone_voice`` returned); the route reads provider-specific extras such as ``embedding_size`` from it so its success envelope is unchanged. """ status: str # 'ready' | 'pending' | 'failed' provider: str name: str voice_id: Optional[str] = None job_id: Optional[str] = None created_at: str = '' clone_time_ms: int = 0 voice_ref: str = '' # opaque provider handle, stored verbatim error: Optional[Dict[str, Any]] = None metadata: Dict[str, Any] = field(default_factory=dict) @classmethod def from_legacy(cls, provider: str, requested_name: str, record: Dict[str, Any], voice_ref: str = '') -> 'CloneResult': """Wrap the dict a legacy ``clone_voice`` returned, field-for-field the way the pre-seam route read it (voice_id default '', name default = requested name, created_at default '', clone_time_ms default 0).""" record = record or {} return cls( status='ready', provider=provider, name=record.get('name', requested_name), voice_id=record.get('voice_id', ''), created_at=record.get('created_at', ''), clone_time_ms=record.get('clone_time_ms', 0), voice_ref=voice_ref, metadata=record, ) @dataclass class CloneInfo: """One stored clone as returned by ``list_clones`` (spec §2.1).""" voice_id: str name: str provider: str created_at: str = '' tenant_id: Optional[str] = None consent_id: Optional[str] = None audio_sha256: Optional[str] = None metadata: Dict[str, Any] = field(default_factory=dict) def to_dict(self) -> Dict[str, Any]: return { 'voice_id': self.voice_id, 'name': self.name, 'provider': self.provider, 'created_at': self.created_at, 'tenant_id': self.tenant_id, 'consent_id': self.consent_id, 'audio_sha256': self.audio_sha256, } class CloneCapableTTSProvider(TTSProvider): """A TTSProvider that can create voice clones (spec §2.1). Subclasses set the class attributes below and implement ``create_clone``, ``list_clones`` and ``disable_clone``. ``get_clone_status`` is only needed when ``async_clone`` is True. ``health`` defaults to wrapping ``health_check()``/``is_available()``. Class attributes: clone_label: human name used in input-validation errors. clone_input: accepted reference inputs, subset of ('bytes','path','url'). The route rejects a request that carries none of them (400). async_clone: True if create_clone may return status 'pending'. min_ref_seconds / max_ref_seconds: reference length bounds or None. clone_languages: languages a clone can speak (empty = unknown). """ clone_label: str = '' clone_input: tuple = ('path',) async_clone: bool = False min_ref_seconds: Optional[int] = None max_ref_seconds: Optional[int] = None clone_languages: tuple = () def clone_capabilities(self) -> Dict[str, Any]: """The ``get_info()['capabilities']`` clone block (spec §2.1).""" return { 'clone': True, 'clone_input': list(self.clone_input), 'async_clone': bool(self.async_clone), 'max_ref_seconds': self.max_ref_seconds, 'min_ref_seconds': self.min_ref_seconds, 'languages': list(self.clone_languages), } def clone_request_error(self, req: CloneRequest) -> Optional[str]: """Return a 400 message if ``req`` carries no input this provider accepts, else None. Checked by the route BEFORE create_clone, so a bad request spends nothing.""" label = self.clone_label or self.__class__.__name__ accepts = set(self.clone_input) if 'path' in accepts and req.audio_path: return None if 'url' in accepts and req.audio_url: return None if 'path' in accepts: return f'{label} requires audio file upload (multipart form)' return f'{label} requires audio_url' @abstractmethod def create_clone(self, req: CloneRequest) -> CloneResult: """Create a clone. Sync providers return status 'ready'; async ones may return 'pending' + job_id. Raise CloneError for typed failures.""" def get_clone_status(self, job_id: str) -> CloneResult: """Poll an async clone job. Sync providers have no jobs.""" raise CloneError( 'internal', f'{self.clone_label or self.__class__.__name__} clones synchronously; ' f'there is no job {job_id!r}', ) def synthesize(self, text: str, voice_id: Optional[str] = None, **kwargs) -> bytes: """Alias of generate_speech (spec §2.1) — no new behaviour.""" if voice_id is not None: kwargs['voice'] = voice_id return self.generate_speech(text, **kwargs) @abstractmethod def list_clones(self, tenant_id: str) -> List[CloneInfo]: """Clones belonging to ``tenant_id`` only.""" @abstractmethod def disable_clone(self, tenant_id: str, voice_id: str, reason: str = '') -> None: """Soft-retire a clone (rename/flag). NEVER a hard delete.""" def health(self) -> Dict[str, Any]: """{'state': 'ok'|'degraded'|'down', 'detail': str, 'load': float|None}.""" check = getattr(self, 'health_check', None) if callable(check): try: h = check() or {} except Exception as e: # a health probe must never raise return {'state': 'down', 'detail': str(e), 'load': None} return {'state': 'ok' if h.get('ok') else 'down', 'detail': str(h.get('detail', '')), 'load': None} ok = self.is_available() return {'state': 'ok' if ok else 'down', 'detail': '' if ok else 'not available', 'load': None} def soft_retire_clone_dir(base_dir, voice_id: str, reason: str = '') -> 'Path': """Rename ``base_dir/voice_id`` to ``.retired`` (never delete). Same rule as DELETE /api/tts/voices/: only ``clone_*`` ids, and the path must stay inside ``base_dir``. Writes a small ``retired.json`` note with the reason into the renamed folder. Returns the new path. """ import json as _json import time as _time from pathlib import Path if not voice_id or not voice_id.startswith('clone_'): raise CloneError('internal', f'Can only retire cloned voices (clone_*), got {voice_id!r}') base = Path(base_dir) voice_dir = base / voice_id try: voice_dir.resolve().relative_to(base.resolve()) except ValueError: raise CloneError('internal', f'Invalid voice_id {voice_id!r}') if not voice_dir.is_dir(): raise CloneError('internal', f'Voice {voice_id} not found') renamed = voice_dir.with_name(voice_dir.name + '.retired') voice_dir.rename(renamed) try: (renamed / 'retired.json').write_text(_json.dumps({ 'voice_id': voice_id, 'reason': reason, 'retired_at': _time.strftime('%Y-%m-%dT%H:%M:%SZ', _time.gmtime()), }, indent=2)) except OSError: pass return renamed __all__ = [ 'TTSProvider', 'TTSVoice', 'TTSProviderInfo', 'TTSProviderError', 'TTSGenerationError', 'TTSConfigurationError', 'TTSVoiceNotFoundError', 'CLONE_ERROR_CODES', 'CloneError', 'CloneRequest', 'CloneResult', 'CloneInfo', 'CloneCapableTTSProvider', 'soft_retire_clone_dir', ]