#!/usr/bin/env python3 """ Parallel Opus file silence remover Detects and removes 10-100ms of silence from the beginning of Opus files """ import sys import numpy as np from pathlib import Path from multiprocessing import Pool, cpu_count from tqdm import tqdm import logging from typing import Tuple, Optional import argparse from suno_utils.audio import Audio # Audio processing libraries try: import soundfile as sf except ImportError: print("Please install required packages:") print("pip install soundfile numpy tqdm") sys.exit(1) # Configure logging logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(processName)s - %(levelname)s - %(message)s' ) logger = logging.getLogger(__name__) def detect_silence_threshold(audio: np.ndarray, sample_rate: int, silence_thresh_db: float = -50.0, min_silence_ms: float = 10.0, max_silence_ms: float = 100.0) -> Optional[int]: """ Detect silence at the beginning of audio and return the sample index where it ends. Only returns a value if silence is between min_silence_ms and max_silence_ms. Args: audio: Audio signal array sample_rate: Sample rate of the audio silence_thresh_db: Threshold in dB below which audio is considered silence min_silence_ms: Minimum silence duration to consider (ms) max_silence_ms: Maximum silence duration to trim (ms) Returns: Sample index where silence ends, or None if silence is too short, too long, or not found """ # Convert ms to samples min_samples = int(min_silence_ms * sample_rate / 1000) max_samples = int(max_silence_ms * sample_rate / 1000) check_samples = int(max_silence_ms * 2 * sample_rate / 1000) # Handle stereo by converting to mono for silence detection audio_mono = np.mean(audio, axis=1) if len(audio.shape) > 1 else audio # Calculate RMS energy in small windows window_size = int(sample_rate * 0.001) # 1ms windows silence_thresh_linear = 10 ** (silence_thresh_db / 20) # Find the first non-silent sample for i in range(0, min(check_samples, len(audio_mono)), window_size): window_end = min(i + window_size, len(audio_mono)) window = audio_mono[i:window_end] rms = np.sqrt(np.mean(window ** 2)) if rms > silence_thresh_linear: # Only trim if silence is between min and max duration return i if min_samples <= i <= max_samples else None return None def process_single_file(args: Tuple[Path, Path, dict]) -> Tuple[str, bool, str]: """ Process a single Opus file to remove initial silence. Args: args: Tuple of (input_file_path, output_directory, processing_options) Returns: Tuple of (filename, success, message) """ input_path, output_dir, options = args try: # Read the audio file audio_suno= Audio.from_file(str(input_path)) audio = audio_suno.array_float sample_rate = audio_suno.sample_rate # Detect silence silence_end = detect_silence_threshold( audio, sample_rate, silence_thresh_db=options['silence_thresh_db'], min_silence_ms=options['min_silence_ms'], max_silence_ms=options['max_silence_ms'] ) # If silence detected, trim it if silence_end is not None and silence_end > 0: audio_trimmed = audio[silence_end:] # Create output path maintaining directory structure relative_path = input_path.relative_to(options['input_dir']) output_path = output_dir / relative_path output_path.parent.mkdir(parents=True, exist_ok=True) # Save the trimmed audio using Audio class audio_trimmed_obj = Audio.from_array_float(audio_trimmed, sample_rate) audio_trimmed_obj.write_opus(str(output_path)) silence_ms = (silence_end / sample_rate) * 1000 return (str(input_path), True, f"Removed {silence_ms:.1f}ms of silence") else: return (str(input_path), True, "No silence detected, skipped") except Exception as e: return (str(input_path), False, f"Error: {str(e)}") def process_files_parallel(input_dir: Path, output_dir: Path, num_workers: int = None, **options) -> None: """ Process all Opus files in parallel. Args: input_dir: Input directory containing Opus files output_dir: Output directory for processed files num_workers: Number of parallel workers (None = use all CPUs) **options: Additional processing options """ # Collect all Opus files opus_files = list(input_dir.rglob("*.opus")) if not opus_files: logger.warning(f"No .opus files found in {input_dir}") return logger.info(f"Found {len(opus_files)} Opus files to process") # Prepare arguments for parallel processing options['input_dir'] = input_dir process_args = [(f, output_dir, options) for f in opus_files] # Set up worker pool num_workers = num_workers or cpu_count() logger.info(f"Using {num_workers} parallel workers") # Process files with progress bar stats = {'trimmed': 0, 'skipped': 0, 'errors': 0} with Pool(processes=num_workers) as pool: with tqdm(total=len(opus_files), desc="Processing files") as pbar: for filename, success, message in pool.imap_unordered(process_single_file, process_args): if success: if "Removed" in message: stats['trimmed'] += 1 elif "skipped" in message: stats['skipped'] += 1 else: stats['errors'] += 1 logger.error(f"{filename}: {message}") pbar.update(1) pbar.set_postfix({ 'Trimmed': stats['trimmed'], 'Skipped': stats['skipped'], 'Errors': stats['errors'] }) # Final summary logger.info(f"\nProcessing complete:") logger.info(f" Total files: {len(opus_files)}") logger.info(f" Files with silence removed: {stats['trimmed']}") logger.info(f" Files skipped (no silence): {stats['skipped']}") logger.info(f" Errors: {stats['errors']}") def main(): """Main entry point with argument parsing.""" parser = argparse.ArgumentParser( description="Remove initial silence from Opus files in parallel" ) parser.add_argument("input_dir", type=Path, help="Input directory containing Opus files") parser.add_argument("output_dir", type=Path, help="Output directory for processed files") parser.add_argument("--workers", type=int, default=None, help="Number of parallel workers (default: all CPUs)") parser.add_argument("--silence-threshold", type=float, default=-50.0, help="Silence threshold in dB (default: -50.0)") parser.add_argument("--min-silence-ms", type=float, default=10.0, help="Minimum silence duration to remove in ms (default: 10.0)") parser.add_argument("--max-silence-ms", type=float, default=100.0, help="Maximum silence duration to remove in ms (default: 100.0)") args = parser.parse_args() # Validate input directory if not args.input_dir.exists(): print(f"Error: Input directory '{args.input_dir}' does not exist") sys.exit(1) # Create output directory args.output_dir.mkdir(parents=True, exist_ok=True) # Process files process_files_parallel( args.input_dir, args.output_dir, num_workers=args.workers, silence_thresh_db=args.silence_threshold, min_silence_ms=args.min_silence_ms, max_silence_ms=args.max_silence_ms ) if __name__ == "__main__": main()