nifty-coder/stemsplit-backend
0
1"""2Abstract base classes and interfaces for the voice control optimization system.3 4This module defines the contracts that all components must implement,5ensuring consistent behavior and enabling easy testing and extensibility.6"""7 8from abc import ABC, abstractmethod9from typing import Dict, List, Optional, Any, AsyncIterator10from .models import (11 ProviderConfig,12 TranscriptionResult,13 UsageStats,14 ProviderStatus,15 ProcessedAudio,16 CacheStats,17 CostSummary,18 BudgetAlert,19 QuotaStatus,20 AudioSegment21)22 23 24class STTProvider(ABC):25 """26 Abstract base class for speech-to-text providers.27 28 All STT providers must implement this interface to be compatible29 with the provider manager system.30 """31 32 def __init__(self, config: ProviderConfig):33 """Initialize the provider with configuration."""34 self.config = config35 self.name = config.name36 self.provider_type = config.provider_type37 38 @abstractmethod39 async def transcribe_audio(40 self, 41 audio_data: bytes, 42 format: str,43 language: str = "en-US",44 **kwargs45 ) -> TranscriptionResult:46 """47 Transcribe audio data to text.48 49 Args:50 audio_data: Raw audio bytes51 format: Audio format (e.g., "webm", "wav", "mp3")52 language: Language code for transcription53 **kwargs: Provider-specific options54 55 Returns:56 TranscriptionResult with transcribed text and metadata57 58 Raises:59 ProviderError: When transcription fails60 QuotaExceededError: When provider quota is exceeded61 UnsupportedFormatError: When audio format is not supported62 """63 pass64 65 @abstractmethod66 async def transcribe_streaming(67 self,68 audio_stream: AsyncIterator[bytes],69 format: str,70 language: str = "en-US",71 **kwargs72 ) -> AsyncIterator[TranscriptionResult]:73 """74 Transcribe streaming audio data.75 76 Args:77 audio_stream: Async iterator of audio chunks78 format: Audio format79 language: Language code for transcription80 **kwargs: Provider-specific options81 82 Yields:83 TranscriptionResult for each processed chunk84 """85 pass86 87 @abstractmethod88 async def check_health(self) -> bool:89 """90 Check if the provider is healthy and available.91 92 Returns:93 True if provider is healthy, False otherwise94 """95 pass96 97 @abstractmethod98 async def get_quota_status(self) -> Dict[str, QuotaStatus]:99 """100 Get current quota usage status.101 102 Returns:103 Dictionary mapping quota types to their status104 """105 pass106 107 @abstractmethod108 def supports_format(self, format: str) -> bool:109 """110 Check if the provider supports a given audio format.111 112 Args:113 format: Audio format to check114 115 Returns:116 True if format is supported, False otherwise117 """118 pass119 120 @abstractmethod121 def supports_language(self, language: str) -> bool:122 """123 Check if the provider supports a given language.124 125 Args:126 language: Language code to check127 128 Returns:129 True if language is supported, False otherwise130 """131 pass132 133 @abstractmethod134 async def estimate_cost(self, audio_duration: float) -> float:135 """136 Estimate the cost for transcribing audio of given duration.137 138 Args:139 audio_duration: Duration in seconds140 141 Returns:142 Estimated cost in USD143 """144 pass145 146 147class ProviderManagerInterface(ABC):148 """Interface for managing multiple STT providers."""149 150 @abstractmethod151 async def transcribe_audio(152 self, 153 audio_data: bytes, 154 format: str,155 language: str = "en-US",156 preferred_provider: Optional[str] = None157 ) -> TranscriptionResult:158 """159 Transcribe audio using the best available provider.160 161 Args:162 audio_data: Raw audio bytes163 format: Audio format164 language: Language code165 preferred_provider: Preferred provider name (optional)166 167 Returns:168 TranscriptionResult from the selected provider169 """170 pass171 172 @abstractmethod173 async def transcribe_streaming(174 self,175 audio_stream: AsyncIterator[bytes],176 format: str,177 language: str = "en-US",178 preferred_provider: Optional[str] = None179 ) -> AsyncIterator[TranscriptionResult]:180 """181 Transcribe streaming audio using the best available provider.182 183 Args:184 audio_stream: Async iterator of audio chunks185 format: Audio format186 language: Language code187 preferred_provider: Preferred provider name (optional)188 189 Yields:190 TranscriptionResult for each processed chunk191 """192 pass193 194 @abstractmethod195 async def get_provider_status(self) -> Dict[str, ProviderStatus]:196 """197 Get status of all registered providers.198 199 Returns:200 Dictionary mapping provider names to their status201 """202 pass203 204 @abstractmethod205 async def register_provider(self, provider: STTProvider) -> None:206 """207 Register a new STT provider.208 209 Args:210 provider: STTProvider instance to register211 """212 pass213 214 @abstractmethod215 async def unregister_provider(self, provider_name: str) -> None:216 """217 Unregister an STT provider.218 219 Args:220 provider_name: Name of provider to unregister221 """222 pass223 224 @abstractmethod225 async def update_provider_config(226 self, 227 provider_name: str, 228 config: ProviderConfig229 ) -> bool:230 """231 Update configuration for a provider.232 233 Args:234 provider_name: Name of provider to update235 config: New configuration236 237 Returns:238 True if update was successful, False otherwise239 """240 pass241 242 @abstractmethod243 async def get_cache_stats(self) -> Optional[Dict[str, Any]]:244 """245 Get cache performance statistics if cache manager is available.246 247 Returns:248 Cache statistics dictionary or None if no cache manager249 """250 pass251 252 @abstractmethod253 async def get_system_stats(self) -> Dict[str, Any]:254 """255 Get comprehensive system statistics including providers and cache.256 257 Returns:258 Dictionary with provider metrics, cache stats, and system overview259 """260 pass261 262 263class RateLimiterInterface(ABC):264 """Interface for rate limiting and quota management."""265 266 @abstractmethod267 async def check_quota(268 self, 269 provider: str, 270 audio_duration: float271 ) -> QuotaStatus:272 """273 Check if a request would exceed quota limits.274 275 Args:276 provider: Provider name277 audio_duration: Duration of audio in seconds278 279 Returns:280 QuotaStatus indicating current usage and limits281 """282 pass283 284 @abstractmethod285 async def consume_quota(286 self, 287 provider: str, 288 audio_duration: float,289 request_count: int = 1290 ) -> bool:291 """292 Consume quota for a request.293 294 Args:295 provider: Provider name296 audio_duration: Duration of audio in seconds297 request_count: Number of requests (default 1)298 299 Returns:300 True if quota was consumed successfully, False if exceeded301 """302 pass303 304 @abstractmethod305 async def get_usage_stats(306 self, 307 provider: str, 308 time_window: str309 ) -> UsageStats:310 """311 Get usage statistics for a provider and time window.312 313 Args:314 provider: Provider name315 time_window: Time window ("minute", "hour", "day", "month")316 317 Returns:318 UsageStats for the specified period319 """320 pass321 322 @abstractmethod323 async def reset_quota(self, provider: str, quota_type: str) -> None:324 """325 Reset quota counters for a provider.326 327 Args:328 provider: Provider name329 quota_type: Type of quota to reset330 """331 pass332 333 334class AudioProcessorInterface(ABC):335 """Interface for audio processing and optimization."""336 337 @abstractmethod338 async def optimize_for_provider(339 self, 340 audio_data: bytes, 341 provider: str,342 target_format: Optional[str] = None343 ) -> ProcessedAudio:344 """345 Optimize audio for a specific provider.346 347 Args:348 audio_data: Raw audio bytes349 provider: Target provider name350 target_format: Desired output format (optional)351 352 Returns:353 ProcessedAudio optimized for the provider354 """355 pass356 357 @abstractmethod358 async def detect_silence(359 self, 360 audio_data: bytes,361 threshold: float = 0.01362 ) -> List[AudioSegment]:363 """364 Detect silence in audio and return non-silent segments.365 366 Args:367 audio_data: Raw audio bytes368 threshold: Silence detection threshold369 370 Returns:371 List of AudioSegment objects for non-silent parts372 """373 pass374 375 @abstractmethod376 async def apply_noise_reduction(377 self, 378 audio_data: bytes,379 strength: float = 0.5380 ) -> bytes:381 """382 Apply noise reduction to audio data.383 384 Args:385 audio_data: Raw audio bytes386 strength: Noise reduction strength (0.0 to 1.0)387 388 Returns:389 Processed audio bytes with reduced noise390 """391 pass392 393 @abstractmethod394 async def convert_format(395 self,396 audio_data: bytes,397 source_format: str,398 target_format: str,399 **kwargs400 ) -> bytes:401 """402 Convert audio from one format to another.403 404 Args:405 audio_data: Raw audio bytes406 source_format: Current audio format407 target_format: Desired audio format408 **kwargs: Format-specific options409 410 Returns:411 Converted audio bytes412 """413 pass414 415 @abstractmethod416 async def chunk_audio(417 self,418 audio_data: bytes,419 chunk_duration: float,420 overlap: float = 0.0421 ) -> List[AudioSegment]:422 """423 Split audio into chunks of specified duration.424 425 Args:426 audio_data: Raw audio bytes427 chunk_duration: Duration of each chunk in seconds428 overlap: Overlap between chunks in seconds429 430 Returns:431 List of AudioSegment objects432 """433 pass434 435 436class CacheManagerInterface(ABC):437 """Interface for caching transcription results."""438 439 @abstractmethod440 async def get_cached_transcription(441 self, 442 audio_fingerprint: str443 ) -> Optional[TranscriptionResult]:444 """445 Retrieve cached transcription result.446 447 Args:448 audio_fingerprint: Unique fingerprint of audio data449 450 Returns:451 Cached TranscriptionResult or None if not found452 """453 pass454 455 @abstractmethod456 async def cache_transcription(457 self, 458 audio_fingerprint: str, 459 result: TranscriptionResult,460 ttl: int = 3600461 ) -> bool:462 """463 Cache a transcription result.464 465 Args:466 audio_fingerprint: Unique fingerprint of audio data467 result: TranscriptionResult to cache468 ttl: Time to live in seconds469 470 Returns:471 True if caching was successful, False otherwise472 """473 pass474 475 @abstractmethod476 async def generate_audio_fingerprint(self, audio_data: bytes) -> str:477 """478 Generate a unique fingerprint for audio data.479 480 Args:481 audio_data: Raw audio bytes482 483 Returns:484 Unique fingerprint string485 """486 pass487 488 @abstractmethod489 async def get_cache_stats(self) -> CacheStats:490 """491 Get cache performance statistics.492 493 Returns:494 CacheStats with hit rates and usage information495 """496 pass497 498 @abstractmethod499 async def clear_cache(self) -> None:500 """Clear all cached entries."""501 pass502 503 @abstractmethod504 async def evict_expired(self) -> int:505 """506 Remove expired cache entries.507 508 Returns:509 Number of entries evicted510 """511 pass512 513 async def get_detailed_cache_stats(self) -> Dict[str, Any]:514 """515 Get detailed cache performance statistics for optimization analysis.516 517 Returns:518 Dictionary with comprehensive cache metrics519 """520 # Default implementation - can be overridden by concrete classes521 basic_stats = await self.get_cache_stats()522 return {523 "basic_stats": {524 "total_requests": basic_stats.total_requests,525 "cache_hits": basic_stats.cache_hits,526 "cache_misses": basic_stats.cache_misses,527 "hit_rate": basic_stats.hit_rate,528 "evictions": basic_stats.evictions529 }530 }531 532 async def warm_cache(533 self, 534 audio_fingerprints: List[str],535 transcription_callback: Optional[Any] = None536 ) -> Dict[str, bool]:537 """538 Warm the cache by preloading frequently accessed transcriptions.539 540 Args:541 audio_fingerprints: List of audio fingerprints to warm542 transcription_callback: Optional callback to generate transcriptions543 544 Returns:545 Dictionary mapping fingerprints to success status546 """547 # Default implementation - can be overridden by concrete classes548 return {fp: False for fp in audio_fingerprints}549 550 async def optimize_cache_performance(self) -> Dict[str, Any]:551 """552 Perform cache performance optimization operations.553 554 Returns:555 Dictionary with optimization results556 """557 # Default implementation - can be overridden by concrete classes558 return {"optimizations_applied": []}559 560 async def get_cached_transcription_batch(561 self, 562 audio_fingerprints: List[str]563 ) -> Dict[str, Optional[TranscriptionResult]]:564 """565 Retrieve multiple cached transcription results in a single operation.566 567 Args:568 audio_fingerprints: List of unique fingerprints of audio data569 570 Returns:571 Dictionary mapping fingerprints to cached results (None if not found)572 """573 # Default implementation - can be overridden by concrete classes574 results = {}575 for fingerprint in audio_fingerprints:576 results[fingerprint] = await self.get_cached_transcription(fingerprint)577 return results578 579 580class CostMonitorInterface(ABC):581 """Interface for cost monitoring and alerting."""582 583 @abstractmethod584 async def track_usage(585 self, 586 provider: str, 587 audio_duration: float,588 estimated_cost: float,589 success: bool = True590 ) -> None:591 """592 Track usage and cost for a provider.593 594 Args:595 provider: Provider name596 audio_duration: Duration of audio processed597 estimated_cost: Estimated cost of the operation598 success: Whether the operation was successful599 """600 pass601 602 @abstractmethod603 async def get_cost_summary(604 self, 605 time_period: str,606 provider: Optional[str] = None607 ) -> CostSummary:608 """609 Get cost summary for a time period.610 611 Args:612 time_period: Time period ("day", "week", "month")613 provider: Specific provider (optional, all if None)614 615 Returns:616 CostSummary for the specified period617 """618 pass619 620 @abstractmethod621 async def check_budget_alerts(self) -> List[BudgetAlert]:622 """623 Check for budget threshold violations.624 625 Returns:626 List of active budget alerts627 """628 pass629 630 @abstractmethod631 async def set_budget_threshold(632 self,633 provider: str,634 threshold: float,635 time_period: str636 ) -> None:637 """638 Set a budget threshold for alerts.639 640 Args:641 provider: Provider name642 threshold: Budget threshold in USD643 time_period: Time period for the threshold644 """645 pass646 647 @abstractmethod648 async def acknowledge_alert(self, alert_id: str) -> None:649 """650 Acknowledge a budget alert.651 652 Args:653 alert_id: ID of the alert to acknowledge654 """655 pass