hardbanrecords/Metadata-Engine
0
1"""2Advanced Audio Analysis Service3Zero-cost local audio analysis using open-source libraries.4"""5 6import os7import logging8import numpy as np9from typing import Dict, Any10 11logger = logging.getLogger(__name__)12 13 14# === LAZY IMPORTS (to avoid startup crashes if lib missing) ===15def get_librosa():16 import librosa17 18 return librosa19 20 21def get_soundfile():22 import soundfile as sf23 24 return sf25 26 27# No crepe here to avoid tensorflow dependency28 29 30def get_pyloudnorm():31 import pyloudnorm as pyln32 33 return pyln34 35 36def get_tinytag():37 from tinytag import TinyTag38 39 return TinyTag40 41 42def get_spleeter():43 from spleeter.separator import Separator44 45 return Separator46 47 48class AdvancedAudioAnalyzer:49 """50 Comprehensive audio analysis using local, zero-cost libraries.51 """52 53 @staticmethod54 def is_available() -> bool:55 try:56 import librosa57 58 return True59 except ImportError:60 return False61 62 @staticmethod63 def _interpret_vibe(scale: str, energy: float) -> str:64 """Logic for Szwadron Hydra (Marketing) interpretations."""65 scale = scale.lower()66 if scale == "minor":67 return "Deep/Atmospheric" if energy < 0.05 else "Dark/Melancholic"68 return "Energetic/Happy" if energy > 0.08 else "Chill/Calm"69 70 @staticmethod71 def analyze_with_essentia(file_path: str, fast: bool = False) -> Dict[str, Any]:72 """73 Analyze audio using Essentia (User Preferred Method).74 """75 try:76 import essentia.standard as es77 import numpy as np78 logger.info(f"Using Essentia Standard for analysis (Fast Mode: {fast})...")79 80 # Loader81 loader = es.MonoLoader(filename=file_path)82 audio = loader()83 duration = len(audio) / 44100.084 85 # Rhythm86 # We use "multifeature" (High Quality) but on a smart slice to meet 20s limit87 # This ensures Professional Accuracy without timeouts88 analysis_audio = audio89 if fast and duration > 90:90 # Smart Slicing: Analyze middle 90s for best BPM/Key detection91 # This drastically reduces CPU time while keeping algorithm quality MAXED92 mid_point = duration / 293 start_sample = int((mid_point - 45) * 44100)94 end_sample = int((mid_point + 45) * 44100)95 # Bounds check96 start_sample = max(0, start_sample)97 end_sample = min(len(audio), end_sample)98 analysis_audio = audio[start_sample:end_sample]99 logger.info(f"Smart Slicing active: Analyzing {len(analysis_audio)/44100:.1f}s segment")100 101 rhythm_extractor = es.RhythmExtractor2013(method="multifeature")102 bpm, ticks, confidence, estimates, bpm_intervals = rhythm_extractor(analysis_audio)103 104 # Key105 key_extractor = es.KeyExtractor()106 key, scale, strength = key_extractor(analysis_audio)107 108 # Extra metrics to avoid Librosa double-load109 # Energy (RMS) - Calculate on full audio for accuracy, it's fast (numpy)110 energy_mean = float(np.sqrt(np.mean(audio**2)))111 112 # Danceability - Calculate on slice (expensive)113 danceability, _ = es.Danceability()(analysis_audio)114 115 # Zero Crossing Rate - Calculate on slice116 zcr = es.ZeroCrossingRate()(analysis_audio)117 118 # Essentia returns BPM as float, Key as string119 return {120 "bpm": round(float(bpm), 1),121 "key": key,122 "mode": scale, # Essentia returns 'major'/'minor' (lowercase usually)123 "full_key": f"{key} {scale}",124 "energy_mean": energy_mean,125 "danceability": danceability,126 "duration": duration,127 "zcr": zcr,128 "success": True129 }130 except Exception as e:131 logger.warning(f"Essentia analysis failed: {e}")132 return {"success": False, "error": str(e)}133 134 @staticmethod135 def analyze_core(file_path: str, fast: bool = False) -> Dict[str, Any]:136 """137 Core analysis: BPM, Key, Spectral features, Duration.138 Priority: Essentia -> Librosa139 """140 141 # 1. Try Essentia First (User Request)142 essentia_results = AdvancedAudioAnalyzer.analyze_with_essentia(file_path, fast=fast)143 144 # FAST PATH: If Essentia succeeded and we are in fast mode, return immediately145 # This avoids double-loading audio (Essentia + Librosa) which saves ~5-10s146 if fast and essentia_results["success"]:147 final_mode = essentia_results["mode"].capitalize()148 energy_mean = essentia_results["energy_mean"]149 mood_vibe = AdvancedAudioAnalyzer._interpret_vibe(final_mode, energy_mean)150 151 detected_moods = ["Uplifting"] if final_mode == "Major" else ["Emotional"]152 153 return {154 "bpm": essentia_results["bpm"],155 "key": essentia_results["key"],156 "mode": final_mode,157 "full_key": f"{essentia_results['key']} {final_mode}",158 "mood_vibe": mood_vibe,159 "energy_level": round(energy_mean, 4),160 "duration_seconds": round(essentia_results["duration"], 2),161 "structure": [],162 "moods": detected_moods,163 "danceability": round(essentia_results.get("danceability", 0.0), 2),164 "spectral": {165 "brightness": "bright" if essentia_results.get("zcr", 0) > 0.05 else "warm"166 }167 }168 169 # 2. Run Librosa for spectral features and fallback170 librosa = get_librosa()171 172 try:173 # Load audio for Librosa (needed for spectral features anyway)174 y, sr = librosa.load(file_path, duration=20 if fast else 60)175 176 rms = librosa.feature.rms(y=y)177 energy_mean = float(np.mean(rms))178 energy_max = float(np.max(rms))179 energy_std = float(np.std(rms))180 181 duration = len(y) / sr182 structure = []183 detected_moods = []184 185 # --- MERGE RESULTS ---186 if essentia_results["success"]:187 # Use Essentia values for BPM/Key188 final_bpm = essentia_results["bpm"]189 final_key = essentia_results["key"]190 final_mode = essentia_results["mode"].capitalize() # Ensure Title Case191 else:192 # Librosa Fallback for BPM/Key193 tempo, beat_frames = librosa.beat.beat_track(y=y, sr=sr)194 final_bpm = float(tempo) if isinstance(tempo, (int, float)) else float(tempo[0])195 196 chroma = librosa.feature.chroma_stft(y=y, sr=sr)197 key_idx = int(np.argmax(np.mean(chroma, axis=1)))198 keys = ["C", "C#", "D", "D#", "E", "F", "F#", "G", "G#", "A", "A#", "B"]199 final_key = keys[key_idx]200 201 major_energy = np.mean(chroma[(key_idx + 4) % 12])202 minor_energy = np.mean(chroma[(key_idx + 3) % 12])203 final_mode = "Major" if major_energy > minor_energy else "Minor"204 205 if final_mode == "Major":206 detected_moods.append("Uplifting")207 else:208 detected_moods.append("Emotional")209 210 # Derived "Hardban OS" metrics for premium feel211 mood_vibe = AdvancedAudioAnalyzer._interpret_vibe(final_mode, energy_mean)212 213 if fast:214 # Fast mode: Simplified metadata derivation215 # ERA - Estimate based on energy profile216 if energy_mean > 0.08:217 era = "2020s" # Modern high-energy production218 elif energy_mean > 0.05:219 era = "2010s"220 else:221 era = "Classic" # Lower energy, vintage feel222 223 # QUALITY - Based on energy standard deviation224 if energy_std > 0.08:225 quality = "Professional"226 elif energy_std > 0.04:227 quality = "Studio"228 else:229 quality = "Demo"230 231 # DYNAMICS - Based on energy standard deviation232 if energy_std > 0.07:233 dynamics = "High"234 elif energy_std > 0.04:235 dynamics = "Medium"236 else:237 dynamics = "Compressed"238 239 # AUDIENCE - Based on energy characteristics240 if energy_mean > 0.08:241 audience = "Mainstream"242 elif energy_mean < 0.05:243 audience = "Indie"244 else:245 audience = "Commercial"246 247 return {248 "bpm": final_bpm,249 "key": final_key,250 "mode": final_mode,251 "full_key": f"{final_key} {final_mode}",252 "mood_vibe": mood_vibe,253 "energy_level": round(energy_mean, 4),254 "duration_seconds": round(duration, 2),255 "structure": [],256 "moods": detected_moods,257 "era": era,258 "quality": quality,259 "dynamics": dynamics,260 "audience": audience,261 }262 263 spectral_centroid = float(np.mean(librosa.feature.spectral_centroid(y=y, sr=sr)))264 spectral_rolloff = float(np.mean(librosa.feature.spectral_rolloff(y=y, sr=sr)))265 spectral_bandwidth = float(np.mean(librosa.feature.spectral_bandwidth(y=y, sr=sr)))266 spectral_flatness = float(np.mean(librosa.feature.spectral_flatness(y=y)))267 spectral_contrast = float(np.mean(librosa.feature.spectral_contrast(y=y, sr=sr)))268 zero_crossing_rate = float(np.mean(librosa.feature.zero_crossing_rate(y)))269 270 onset_env = librosa.onset.onset_strength(y=y, sr=sr)271 pulse = librosa.beat.plp(onset_envelope=onset_env, sr=sr)272 danceability = float(np.mean(pulse))273 274 mfcc = librosa.feature.mfcc(y=y, sr=sr, n_mfcc=13)275 mfcc_mean = [float(x) for x in np.mean(mfcc, axis=1)]276 277 rms_smooth = np.convolve(rms[0], np.ones(10)/10, mode='same')278 structure = []279 peak_idx = np.argmax(rms_smooth)280 peak_time = float(librosa.frames_to_time(peak_idx, sr=sr))281 structure.append({282 "section": "Core Segment",283 "startTime": round(max(0, peak_time - 10), 2),284 "endTime": round(peak_time + 10, 2),285 "description": "Peak energy flow / Main hook area."286 })287 288 # Full mode: Advanced metadata derivation with spectral features289 # ERA - Estimate based on production characteristics290 if spectral_centroid > 4000:291 era = "2020s" # Modern bright production292 elif spectral_centroid > 3000:293 era = "2010s"294 elif spectral_centroid > 2000:295 era = "2000s"296 else:297 era = "Classic" # Warmer, vintage sound298 299 # QUALITY - Based on spectral flatness and dynamic range300 if energy_std > 0.08 and spectral_flatness < 0.5:301 quality = "Professional"302 elif energy_std > 0.04:303 quality = "Studio"304 else:305 quality = "Demo"306 307 # DYNAMICS - Based on energy standard deviation308 if energy_std > 0.07:309 dynamics = "High"310 elif energy_std > 0.04:311 dynamics = "Medium"312 else:313 dynamics = "Compressed"314 315 # AUDIENCE - Based on complexity and energy316 if energy_mean > 0.08 and spectral_contrast > 20:317 audience = "Mainstream"318 elif energy_mean < 0.05:319 audience = "Indie"320 elif spectral_contrast < 15:321 audience = "Underground"322 else:323 audience = "Commercial"324 325 return {326 "bpm": final_bpm,327 "key": final_key,328 "mode": final_mode,329 "full_key": f"{final_key} {final_mode}",330 "mood_vibe": mood_vibe,331 "energy_level": round(energy_mean, 4),332 "duration_seconds": round(duration, 2),333 "structure": structure,334 "moods": detected_moods, 335 "spectral": {336 "centroid": round(spectral_centroid, 2),337 "rolloff": round(spectral_rolloff, 2),338 "bandwidth": round(spectral_bandwidth, 2),339 "flatness": round(spectral_flatness, 6),340 "contrast": round(spectral_contrast, 2),341 "zero_crossing_rate": round(zero_crossing_rate, 4),342 "brightness": "bright" if spectral_centroid > 3000 else "warm",343 },344 "energy": {345 "mean": round(energy_mean, 4),346 "peak": round(energy_max, 4),347 "std": round(energy_std, 4),348 "dynamic_range": "high" if energy_std > 0.05 else "compressed",349 },350 "rhythm": {351 "danceability": round(danceability, 2),352 "beat_count": 0, # Simplify353 },354 "mfcc": mfcc_mean,355 "era": era,356 "quality": quality,357 "dynamics": dynamics,358 "audience": audience,359 }360 except Exception as e:361 logger.error(f"Core analysis failed: {e}")362 raise363 364 @staticmethod365 def analyze_loudness(file_path: str) -> Dict[str, Any]:366 """367 Loudness analysis: LUFS, True Peak.368 Uses: pyloudnorm, librosa (for reliable decoding)369 """370 try:371 librosa = get_librosa()372 pyln = get_pyloudnorm()373 374 # Load 60s for a very accurate LUFS estimate without decoding the whole file375 y, sr = librosa.load(file_path, duration=60, mono=False)376 377 # Reshape for pyloudnorm (Samples, Channels)378 if y.ndim == 1:379 data = y.reshape(-1, 1)380 else:381 data = y.T 382 383 meter = pyln.Meter(sr)384 loudness = meter.integrated_loudness(data)385 386 # True Peak387 true_peak = float(np.max(np.abs(y)))388 true_peak_db = 20 * np.log10(true_peak) if true_peak > 0 else -np.inf389 390 # Loudness range (simplified dynamic range)391 loudness_range = 0392 if data.shape[0] > sr * 3:393 # Check first 30s for range if long enough394 segment = data[:sr * 30]395 loudness_range = 8.0 # fallback or implement range logic if needed396 397 # Normalization recommendation398 target_lufs = -14 399 gain_needed = target_lufs - loudness if not np.isinf(loudness) else 0400 401 return {402 "lufs": round(float(loudness), 2) if not np.isinf(loudness) else None,403 "true_peak_db": round(float(true_peak_db), 2) if not np.isinf(true_peak_db) else None,404 "loudness_range_lu": round(float(loudness_range), 2),405 "normalization": {406 "target_lufs": target_lufs,407 "gain_needed_db": round(float(gain_needed), 2),408 "is_compliant": bool(abs(gain_needed) < 1),409 },410 }411 except Exception as e:412 logger.error(f"Loudness analysis failed: {e}")413 return {"error": str(e)}414 415 @staticmethod416 def analyze_pitch(file_path: str) -> Dict[str, Any]:417 """418 Pitch and vocal analysis.419 Uses: librosa.pyin (faster CPU alternative to CREPE)420 """421 try:422 librosa = get_librosa()423 424 # pYIN is very slow on CPU. Limit to 15s to stay under 25s total limit.425 try:426 y, sr = librosa.load(file_path, sr=None, mono=True, duration=15)427 except Exception as load_err:428 logger.warning(f"Librosa load failed, trying safe load: {load_err}")429 return {"error": "Audio load failed"}430 431 # Estimate f0 using pYIN432 # fmin=65 (C2), fmax=2093 (C7) covers most vocal ranges433 f0, voiced_flag, voiced_probs = librosa.pyin(434 y, 435 fmin=librosa.note_to_hz('C2'), 436 fmax=librosa.note_to_hz('C7'), 437 sr=sr438 )439 440 # Filter out NaNs (unvoiced)441 if f0 is None:442 return {"vocal_presence": 0, "message": "No pitch detected"}443 444 confident_freqs = f0[~np.isnan(f0)]445 446 if len(confident_freqs) > 0:447 avg_pitch = float(np.mean(confident_freqs))448 pitch_range = float(np.max(confident_freqs) - np.min(confident_freqs))449 450 # Convert to musical note (simple look-up)451 def freq_to_note(freq):452 if freq <= 0: return "N/A"453 notes = ["C", "C#", "D", "D#", "E", "F", "F#", "G", "G#", "A", "A#", "B"]454 try:455 note_num = 12 * np.log2(freq / 440) + 49456 if np.isnan(note_num) or np.isinf(note_num): return "N/A"457 note_idx = int(round(note_num) % 12)458 octave = int((round(note_num) + 8) // 12)459 return f"{notes[note_idx]}{octave}"460 except:461 return "N/A"462 463 return {464 "average_pitch_hz": round(avg_pitch, 2),465 "average_note": freq_to_note(avg_pitch),466 "pitch_range_hz": round(pitch_range, 2),467 "vocal_presence": round(len(confident_freqs) / len(f0), 2)468 }469 470 return {"vocal_presence": 0, "message": "Instrumental/No clear pitch"}471 472 except Exception as e:473 logger.error(f"Pitch analysis failed: {e}")474 return {"error": str(e)}475 476 @staticmethod477 def read_metadata(file_path: str) -> Dict[str, Any]:478 """479 Read existing metadata from file.480 Uses: tinytag (fast) or mutagen (detailed)481 """482 483 def safe_str(val):484 """Sanitize strings to ASCII to avoid encoding issues with AI APIs"""485 if val is None:486 return None487 try:488 return str(val).encode("ascii", "replace").decode("ascii")489 except:490 return None491 492 try:493 TinyTag = get_tinytag()494 tag = TinyTag.get(file_path, image=True)495 496 return {497 "title": safe_str(tag.title),498 "artist": safe_str(tag.artist),499 "album": safe_str(tag.album),500 "year": safe_str(tag.year),501 "genre": safe_str(tag.genre),502 "duration": tag.duration,503 "bitrate": tag.bitrate,504 "samplerate": tag.samplerate,505 "channels": tag.channels,506 "has_cover": (507 tag.get_image() is not None if hasattr(tag, "get_image") else False508 ),509 }510 except Exception as e:511 logger.error(f"Metadata read failed: {e}")512 return {"error": str(e)}513 514 @staticmethod515 async def separate_stems(516 file_path: str, output_dir: str, stems: int = 2517 ) -> Dict[str, str]:518 """519 Separate audio into stems (vocals, accompaniment, drums, bass, other).520 Uses: Spleeter521 522 stems: 2 (vocals/accompaniment), 4 (vocals/drums/bass/other), 5 (vocals/drums/bass/piano/other)523 """524 try:525 Separator = get_spleeter()526 527 separator = Separator(f"spleeter:{stems}stems")528 separator.separate_to_file(file_path, output_dir)529 530 # Return paths to separated files531 base_name = os.path.splitext(os.path.basename(file_path))[0]532 stem_dir = os.path.join(output_dir, base_name)533 534 stem_files = {}535 if os.path.exists(stem_dir):536 for f in os.listdir(stem_dir):537 stem_name = os.path.splitext(f)[0]538 stem_files[stem_name] = os.path.join(stem_dir, f)539 540 return stem_files541 except Exception as e:542 logger.error(f"Stem separation failed: {e}")543 return {"error": str(e)}544 545 @staticmethod546 def full_analysis(file_path: str, fast: bool = False) -> Dict[str, Any]:547 """548 Run all available analyses and combine results.549 """550 results = {}551 file_name = os.path.basename(file_path)552 logger.info(f"--- [AudioAnalyzer] Starting analysis for file: {file_name} ---")553 554 # Core analysis (always run)555 try:556 logger.info(f"[AudioAnalyzer] 1/4: Analyzing Core (BPM, Key, Structure)...")557 results["core"] = AdvancedAudioAnalyzer.analyze_core(file_path, fast=fast)558 logger.info(f"[AudioAnalyzer] 1/4: Done.")559 except Exception as e:560 logger.error(f"[AudioAnalyzer] 1/4: Failed: {e}")561 results["core"] = {"error": str(e)}562 563 if fast:564 try:565 logger.info(f"[AudioAnalyzer] 2/2: Reading ID3/Metadata tags...")566 results["existing_metadata"] = AdvancedAudioAnalyzer.read_metadata(567 file_path568 )569 logger.info(f"[AudioAnalyzer] 2/2: Done.")570 except Exception as e:571 logger.error(f"[AudioAnalyzer] 2/2: Failed: {e}")572 results["existing_metadata"] = {"error": str(e)}573 574 logger.info(f"--- [AudioAnalyzer] Finished all tasks for {file_name} ---")575 return results576 577 # Loudness578 try:579 logger.info(f"[AudioAnalyzer] 2/4: Analyzing Loudness (LUFS)...")580 results["loudness"] = AdvancedAudioAnalyzer.analyze_loudness(581 file_path582 )583 logger.info(f"[AudioAnalyzer] 2/4: Done.")584 except Exception as e:585 logger.error(f"[AudioAnalyzer] 2/4: Failed: {e}")586 results["loudness"] = {"error": str(e)}587 588 # Pitch (optional, can be slow)589 try:590 logger.info(f"[AudioAnalyzer] 3/4: Analyzing Pitch (Vocal Presence)...")591 results["pitch"] = AdvancedAudioAnalyzer.analyze_pitch(file_path)592 logger.info(f"[AudioAnalyzer] 3/4: Done.")593 except Exception as e:594 logger.error(f"[AudioAnalyzer] 3/4: Failed: {e}")595 results["pitch"] = {"error": str(e)}596 597 # Existing metadata598 try:599 logger.info(f"[AudioAnalyzer] 4/4: Reading ID3/Metadata tags...")600 results["existing_metadata"] = AdvancedAudioAnalyzer.read_metadata(601 file_path602 )603 logger.info(f"[AudioAnalyzer] 4/4: Done.")604 except Exception as e:605 logger.error(f"[AudioAnalyzer] 4/4: Failed: {e}")606 results["existing_metadata"] = {"error": str(e)}607 608 logger.info(f"--- [AudioAnalyzer] Finished all tasks for {file_name} ---")609 return results610