"""Post-processing service for audio mixing, mastering, and effects.""" import os import asyncio from pathlib import Path from typing import Any, TYPE_CHECKING import structlog # Optional audio processing dependencies try: import numpy as np import soundfile as sf import librosa AUDIO_LIBS_AVAILABLE = True except ImportError: AUDIO_LIBS_AVAILABLE = False np = None sf = None librosa = None # Create dummy types for type hints if TYPE_CHECKING: import numpy as np from app.core.config import settings logger = structlog.get_logger(__name__) class PostProcessingService: """Service for post-processing audio (mixing, mastering, effects).""" def __init__(self): """Initialize the post-processing service.""" self.logger = logger.bind(service="post_processing") if not AUDIO_LIBS_AVAILABLE: self.logger.warning("audio_libs_not_available", message="numpy/soundfile/librosa not installed") async def mix_audio( self, instrumental_path: Path, vocal_path: Path, output_path: Path, vocal_volume: float = 0.7, instrumental_volume: float = 0.8, ) -> Path: """ Mix instrumental and vocal tracks. Args: instrumental_path: Path to instrumental audio vocal_path: Path to vocal audio output_path: Path to save mixed audio vocal_volume: Volume level for vocals (0.0-1.0) instrumental_volume: Volume level for instrumental (0.0-1.0) Returns: Path to mixed audio file """ self.logger.info( "mixing_audio", instrumental=str(instrumental_path), vocal=str(vocal_path), ) if os.environ.get("FORCE_SIMULATION", "").lower() == "true" or not AUDIO_LIBS_AVAILABLE: self.logger.warning("simulating_mixing", message="Simulation forced or audio libs missing") import shutil await asyncio.sleep(1) output_path.parent.mkdir(parents=True, exist_ok=True) shutil.copy(instrumental_path, output_path) return output_path def _process(): # Load audio files instrumental, sr_inst = librosa.load(str(instrumental_path), sr=None) vocal, sr_vocal = librosa.load(str(vocal_path), sr=None) # Resample to common sample rate target_sr = max(sr_inst, sr_vocal) if sr_inst != target_sr: instrumental = librosa.resample(instrumental, orig_sr=sr_inst, target_sr=target_sr) if sr_vocal != target_sr: vocal = librosa.resample(vocal, orig_sr=sr_vocal, target_sr=target_sr) # Match lengths (pad shorter track) max_len = max(len(instrumental), len(vocal)) instrumental = np.pad( instrumental, (0, max_len - len(instrumental)), mode="constant" ) vocal = np.pad(vocal, (0, max_len - len(vocal)), mode="constant") # Apply volume adjustments instrumental = instrumental * instrumental_volume vocal = vocal * vocal_volume # Mix tracks mixed = instrumental + vocal # Normalize to prevent clipping max_val = np.abs(mixed).max() if max_val > 1.0: mixed = mixed / max_val # Ensure output directory exists output_path.parent.mkdir(parents=True, exist_ok=True) # Save mixed audio sf.write(str(output_path), mixed, target_sr) return target_sr await asyncio.to_thread(_process) self.logger.info("audio_mixed", output_path=str(output_path)) return output_path async def master_audio( self, audio_path: Path, output_path: Path, normalize: bool = True, apply_compression: bool = True, apply_eq: bool = True, ) -> Path: """ Master audio with compression, EQ, and normalization. Args: audio_path: Path to input audio output_path: Path to save mastered audio normalize: Apply normalization apply_compression: Apply dynamic range compression apply_eq: Apply equalization Returns: Path to mastered audio file """ self.logger.info("mastering_audio", input_path=str(audio_path)) if os.environ.get("FORCE_SIMULATION", "").lower() == "true" or not AUDIO_LIBS_AVAILABLE: self.logger.warning("simulating_mastering", message="Simulation forced or audio libs missing") import shutil await asyncio.sleep(1) output_path.parent.mkdir(parents=True, exist_ok=True) shutil.copy(audio_path, output_path) return output_path def _process(): # Load audio audio, sr = librosa.load(str(audio_path), sr=None) # Apply compression (simple RMS-based compression) if apply_compression: audio = self._apply_compression(audio) # Apply EQ (simple high-pass and low-pass filters) if apply_eq: audio = self._apply_eq(audio, sr) # Normalize if normalize: audio = self._normalize(audio) # Ensure output directory exists output_path.parent.mkdir(parents=True, exist_ok=True) # Save mastered audio sf.write(str(output_path), audio, sr) await asyncio.to_thread(_process) self.logger.info("audio_mastered", output_path=str(output_path)) return output_path def _apply_compression(self, audio: Any, threshold: float = 0.7, ratio: float = 4.0) -> Any: """Apply simple dynamic range compression.""" # Simple RMS-based compression rms = np.sqrt(np.mean(audio**2)) if rms > threshold: gain_reduction = (rms - threshold) / ratio audio = audio * (1.0 - gain_reduction / rms) return audio def _apply_eq(self, audio: np.ndarray, sr: int) -> np.ndarray: """Apply simple equalization.""" # High-pass filter to remove low-frequency noise audio = librosa.effects.preemphasis(audio) return audio def _normalize(self, audio: Any) -> Any: """Normalize audio to prevent clipping.""" max_val = np.abs(audio).max() if max_val > 0: audio = audio / max_val * 0.95 # Leave some headroom return audio async def add_reverb( self, audio_path: Path, output_path: Path, room_size: float = 0.5, ) -> Path: """Add reverb effect to audio.""" # Simple reverb using convolution (would use better reverb in production) audio, sr = librosa.load(str(audio_path), sr=None) # Create simple impulse response for reverb impulse_length = int(sr * room_size) impulse = np.random.randn(impulse_length) * 0.1 impulse = impulse * np.exp(-np.linspace(0, 5, impulse_length)) # Convolve with impulse response reverb_audio = np.convolve(audio, impulse, mode="same") # Mix original and reverb output = audio + reverb_audio * 0.3 # Normalize output = self._normalize(output) output_path.parent.mkdir(parents=True, exist_ok=True) sf.write(str(output_path), output, sr) return output_path # Singleton instance _post_processing_service: PostProcessingService | None = None def get_post_processing_service() -> PostProcessingService: """Get post-processing service instance.""" global _post_processing_service if _post_processing_service is None: _post_processing_service = PostProcessingService() return _post_processing_service