""" Progress Reporting Utilities Provides standardized progress callback types and stage name constants for consistent progress reporting across all Voice Tools services. """ from typing import Callable # Standard progress callback signature # Parameters: (stage: str, current: float, total: float) # - stage: User-friendly description of current processing stage # - current: Current progress value (0.0-1.0 for modern format) # - total: Total progress value (1.0 for percentage, >1.0 for legacy format) ProgressCallback = Callable[[str, float, float], None] # Standardized stage names for Speaker Separation SPEAKER_SEPARATION_STAGES = { "segmentation": "Detecting speech segments", "embeddings": "Computing speaker embeddings", "discrete_diarization": "Identifying speakers", "speaker_counting": "Counting speakers", } # Standardized stage names for Voice Denoising VOICE_DENOISING_STAGES = [ "Detecting voice activity", # VAD processing "Reducing noise", # Noise reduction "Concatenating segments", # Segment assembly ] # Standardized stage names for Speaker Extraction SPEAKER_EXTRACTION_STAGES = [ "Loading reference audio", # Reference processing "Computing embeddings", # Embedding generation "Matching voice segments", # Similarity comparison "Extracting segments", # Audio extraction ] def interpret_progress(current: float, total: float) -> float: """ Convert progress callback parameters to 0.0-1.0 percentage. Handles both modern format (total=1.0, current is percentage) and legacy format (total>1.0, current is count). Args: current: Current progress value total: Total progress value Returns: float: Progress percentage between 0.0 and 1.0 """ if total == 1.0: # Modern format: current is already a percentage return max(0.0, min(1.0, current)) elif total > 0: # Legacy format: calculate percentage return max(0.0, min(1.0, current / total)) else: return 0.0 def safe_progress_callback( callback: ProgressCallback | None, stage: str, current: float, total: float ) -> None: """ Safely invoke progress callback with exception handling. Progress reporting is non-critical, so exceptions are caught and logged rather than propagated. Args: callback: Optional progress callback function stage: Stage description current: Current progress total: Total progress """ if callback: try: callback(stage, current, total) except Exception as e: import logging logger = logging.getLogger(__name__) logger.warning(f"Progress callback failed: {e}") # Continue processing - progress reporting is non-critical