nifty-coder/stemsplit-backend

1"""2Utility functions for the voice control optimization system.3 4This module provides common utility functions used throughout the5voice control system.6"""7 8import hashlib9import time10import asyncio11import logging12from typing import Any, Dict, List, Optional, Callable, TypeVar, Union13from functools import wraps14from datetime import datetime, timedelta15 16logger = logging.getLogger(__name__)17 18T = TypeVar('T')19 20 21def generate_fingerprint(data: bytes, algorithm: str = "sha256") -> str:22    """23    Generate a fingerprint hash for binary data.24    25    Args:26        data: Binary data to hash27        algorithm: Hash algorithm to use28        29    Returns:30        Hexadecimal hash string31    """32    if algorithm == "sha256":33        return hashlib.sha256(data).hexdigest()34    elif algorithm == "md5":35        return hashlib.md5(data).hexdigest()36    elif algorithm == "sha1":37        return hashlib.sha1(data).hexdigest()38    else:39        raise ValueError(f"Unsupported hash algorithm: {algorithm}")40 41 42def generate_session_id() -> str:43    """Generate a unique session ID."""44    timestamp = str(int(time.time() * 1000))45    random_part = hashlib.md5(str(time.time()).encode()).hexdigest()[:8]46    return f"session_{timestamp}_{random_part}"47 48 49def calculate_audio_duration(data_size: int, sample_rate: int, channels: int, bit_depth: int) -> float:50    """51    Calculate audio duration from raw audio parameters.52    53    Args:54        data_size: Size of audio data in bytes55        sample_rate: Sample rate in Hz56        channels: Number of audio channels57        bit_depth: Bit depth (8, 16, 24, 32)58        59    Returns:60        Duration in seconds61    """62    bytes_per_sample = bit_depth // 863    total_samples = data_size // (bytes_per_sample * channels)64    return total_samples / sample_rate65 66 67def format_duration(seconds: float) -> str:68    """69    Format duration in seconds to human-readable string.70    71    Args:72        seconds: Duration in seconds73        74    Returns:75        Formatted duration string (e.g., "1m 30s", "2h 15m")76    """77    if seconds < 60:78        return f"{seconds:.1f}s"79    elif seconds < 3600:80        minutes = int(seconds // 60)81        remaining_seconds = seconds % 6082        return f"{minutes}m {remaining_seconds:.0f}s"83    else:84        hours = int(seconds // 3600)85        remaining_minutes = int((seconds % 3600) // 60)86        return f"{hours}h {remaining_minutes}m"87 88 89def format_bytes(bytes_count: int) -> str:90    """91    Format byte count to human-readable string.92    93    Args:94        bytes_count: Number of bytes95        96    Returns:97        Formatted byte string (e.g., "1.5 KB", "2.3 MB")98    """99    for unit in ['B', 'KB', 'MB', 'GB', 'TB']:100        if bytes_count < 1024.0:101            return f"{bytes_count:.1f} {unit}"102        bytes_count /= 1024.0103    return f"{bytes_count:.1f} PB"104 105 106def safe_divide(numerator: float, denominator: float, default: float = 0.0) -> float:107    """108    Safely divide two numbers, returning default if denominator is zero.109    110    Args:111        numerator: Numerator112        denominator: Denominator113        default: Default value if denominator is zero114        115    Returns:116        Division result or default117    """118    return numerator / denominator if denominator != 0 else default119 120 121def clamp(value: float, min_value: float, max_value: float) -> float:122    """123    Clamp a value between minimum and maximum bounds.124    125    Args:126        value: Value to clamp127        min_value: Minimum allowed value128        max_value: Maximum allowed value129        130    Returns:131        Clamped value132    """133    return max(min_value, min(value, max_value))134 135 136def exponential_backoff(attempt: int, base_delay: float = 1.0, max_delay: float = 60.0, jitter: bool = True) -> float:137    """138    Calculate exponential backoff delay.139    140    Args:141        attempt: Attempt number (0-based)142        base_delay: Base delay in seconds143        max_delay: Maximum delay in seconds144        jitter: Whether to add random jitter145        146    Returns:147        Delay in seconds148    """149    import random150    151    delay = min(base_delay * (2 ** attempt), max_delay)152    153    if jitter:154        # Add ±25% jitter155        jitter_range = delay * 0.25156        delay += random.uniform(-jitter_range, jitter_range)157    158    return max(0, delay)159 160 161def retry_with_backoff(162    max_retries: int = 3,163    base_delay: float = 1.0,164    max_delay: float = 60.0,165    exceptions: tuple = (Exception,)166):167    """168    Decorator for retrying functions with exponential backoff.169    170    Args:171        max_retries: Maximum number of retry attempts172        base_delay: Base delay between retries173        max_delay: Maximum delay between retries174        exceptions: Tuple of exceptions to catch and retry175        176    Returns:177        Decorated function178    """179    def decorator(func: Callable[..., T]) -> Callable[..., T]:180        @wraps(func)181        async def async_wrapper(*args, **kwargs) -> T:182            last_exception = None183            184            for attempt in range(max_retries + 1):185                try:186                    if asyncio.iscoroutinefunction(func):187                        return await func(*args, **kwargs)188                    else:189                        return func(*args, **kwargs)190                except exceptions as e:191                    last_exception = e192                    193                    if attempt == max_retries:194                        logger.error(f"Function {func.__name__} failed after {max_retries} retries: {e}")195                        raise e196                    197                    delay = exponential_backoff(attempt, base_delay, max_delay)198                    logger.warning(f"Function {func.__name__} failed (attempt {attempt + 1}), retrying in {delay:.2f}s: {e}")199                    await asyncio.sleep(delay)200            201            # This should never be reached, but just in case202            raise last_exception203        204        @wraps(func)205        def sync_wrapper(*args, **kwargs) -> T:206            last_exception = None207            208            for attempt in range(max_retries + 1):209                try:210                    return func(*args, **kwargs)211                except exceptions as e:212                    last_exception = e213                    214                    if attempt == max_retries:215                        logger.error(f"Function {func.__name__} failed after {max_retries} retries: {e}")216                        raise e217                    218                    delay = exponential_backoff(attempt, base_delay, max_delay)219                    logger.warning(f"Function {func.__name__} failed (attempt {attempt + 1}), retrying in {delay:.2f}s: {e}")220                    time.sleep(delay)221            222            # This should never be reached, but just in case223            raise last_exception224        225        # Return appropriate wrapper based on function type226        if asyncio.iscoroutinefunction(func):227            return async_wrapper228        else:229            return sync_wrapper230    231    return decorator232 233 234class CircuitBreaker:235    """236    Circuit breaker implementation for handling failing services.237    238    States:239    - CLOSED: Normal operation, requests pass through240    - OPEN: Service is failing, requests are rejected immediately241    - HALF_OPEN: Testing if service has recovered242    """243    244    def __init__(245        self,246        failure_threshold: int = 5,247        recovery_timeout: float = 60.0,248        expected_exception: type = Exception249    ):250        """251        Initialize circuit breaker.252        253        Args:254            failure_threshold: Number of failures before opening circuit255            recovery_timeout: Time to wait before trying to recover256            expected_exception: Exception type that counts as failure257        """258        self.failure_threshold = failure_threshold259        self.recovery_timeout = recovery_timeout260        self.expected_exception = expected_exception261        262        self.failure_count = 0263        self.last_failure_time: Optional[datetime] = None264        self.state = "CLOSED"  # CLOSED, OPEN, HALF_OPEN265    266    def __call__(self, func: Callable[..., T]) -> Callable[..., T]:267        """Use circuit breaker as a decorator."""268        @wraps(func)269        async def async_wrapper(*args, **kwargs) -> T:270            # Check if we should attempt the call271            if not self._should_attempt_call():272                raise Exception(f"Circuit breaker is OPEN for {func.__name__}")273            274            try:275                result = await func(*args, **kwargs) if asyncio.iscoroutinefunction(func) else func(*args, **kwargs)276                self._on_success()277                return result278            except self.expected_exception as e:279                self._on_failure()280                raise e281        282        @wraps(func)283        def sync_wrapper(*args, **kwargs) -> T:284            # Check if we should attempt the call285            if not self._should_attempt_call():286                raise Exception(f"Circuit breaker is OPEN for {func.__name__}")287            288            try:289                result = func(*args, **kwargs)290                self._on_success()291                return result292            except self.expected_exception as e:293                self._on_failure()294                raise e295        296        # Return appropriate wrapper297        if asyncio.iscoroutinefunction(func):298            return async_wrapper299        else:300            return sync_wrapper301    302    def _should_attempt_call(self) -> bool:303        """Check if we should attempt the call based on circuit state."""304        if self.state == "CLOSED":305            return True306        elif self.state == "OPEN":307            # Check if recovery timeout has passed308            if (self.last_failure_time and 309                datetime.utcnow() - self.last_failure_time >= timedelta(seconds=self.recovery_timeout)):310                self.state = "HALF_OPEN"311                return True312            return False313        elif self.state == "HALF_OPEN":314            return True315        316        return False317    318    def _on_success(self):319        """Handle successful call."""320        self.failure_count = 0321        self.state = "CLOSED"322    323    def _on_failure(self):324        """Handle failed call."""325        self.failure_count += 1326        self.last_failure_time = datetime.utcnow()327        328        if self.failure_count >= self.failure_threshold:329            self.state = "OPEN"330        elif self.state == "HALF_OPEN":331            self.state = "OPEN"332 333 334def validate_audio_format(format_str: str) -> bool:335    """336    Validate if an audio format string is supported.337    338    Args:339        format_str: Audio format string (e.g., "webm", "wav")340        341    Returns:342        True if format is supported, False otherwise343    """344    supported_formats = {345        "webm", "wav", "mp3", "ogg", "flac", "aac", "m4a", "opus"346    }347    return format_str.lower() in supported_formats348 349 350def validate_language_code(language: str) -> bool:351    """352    Validate if a language code is in the correct format.353    354    Args:355        language: Language code (e.g., "en-US", "fr-FR")356        357    Returns:358        True if language code format is valid, False otherwise359    """360    import re361    # Match pattern like "en-US", "zh-CN", etc.362    pattern = r'^[a-z]{2}-[A-Z]{2}$'363    return bool(re.match(pattern, language))364 365 366def merge_dictionaries(*dicts: Dict[str, Any]) -> Dict[str, Any]:367    """368    Merge multiple dictionaries, with later ones taking precedence.369    370    Args:371        *dicts: Dictionaries to merge372        373    Returns:374        Merged dictionary375    """376    result = {}377    for d in dicts:378        if d:379            result.update(d)380    return result381 382 383def get_nested_value(data: Dict[str, Any], key_path: str, default: Any = None) -> Any:384    """385    Get a nested value from a dictionary using dot notation.386    387    Args:388        data: Dictionary to search389        key_path: Dot-separated key path (e.g., "audio.processing.enabled")390        default: Default value if key not found391        392    Returns:393        Value at key path or default394    """395    keys = key_path.split('.')396    current = data397    398    try:399        for key in keys:400            current = current[key]401        return current402    except (KeyError, TypeError):403        return default404 405 406def set_nested_value(data: Dict[str, Any], key_path: str, value: Any) -> None:407    """408    Set a nested value in a dictionary using dot notation.409    410    Args:411        data: Dictionary to modify412        key_path: Dot-separated key path413        value: Value to set414    """415    keys = key_path.split('.')416    current = data417    418    for key in keys[:-1]:419        if key not in current:420            current[key] = {}421        current = current[key]422    423    current[keys[-1]] = value