Spaces:
Build error
Build error
| """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 | |