Spaces:
Running
Running
| #!/usr/bin/env python3 | |
| """ | |
| ββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ | |
| β β | |
| β Ψ§ΩΨ₯ΨΩΨ§Ψ‘ β ENGINE-3 OF THE AETHERION β | |
| β Voice Revival Engine β Making Artificial Sound Human Again β | |
| β β | |
| β "Ψ§ΩΨ₯ΨΩΨ§Ψ‘" β revival, bringing back to life. When the Sheikh's voice β | |
| β sounds robotic, pixelated, or hollow β this engine breathes life β | |
| β back into it. β | |
| β β | |
| β Ψ§ΩΩΨ―Ω: Transform artificial/mechanical Quran recitation into β | |
| β natural, human-sounding voice β | |
| β β | |
| β βββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ£ | |
| β β | |
| β SCOPE β | |
| β Audio that sounds: β | |
| β β’ Artificial / robotic / flat F0 (TTS-like, vocoder artifacts) β | |
| β β’ Pixelated / digital (codec ghosting, metallic ringing) β | |
| β β’ Hollow / thin (spectral holes, missing formants) β | |
| β β’ Mechanically denoised (over-NR'd, spectral musical noise) β | |
| β β’ Low-bitrate degraded (64-96kbps MP3/Opus/WhatsApp) β | |
| β β | |
| β PIPELINE (12 phases) β | |
| β Phase 0 Deep artifact analysis: F0 flatness, spectral holes, β | |
| β metallic ringing, codec ghost detection, NR damage score, β | |
| β SNR estimation, active bandwidth, wind noise detection β | |
| β Phase 1 Dereverberation (Safaa S1-S7 pipeline, lightweight) β | |
| β Phase 2 Selective noise repair: spectral-musical-noise reduction β | |
| β + artifact interpolation (not just NR β targeted repair) β | |
| β + wind noise HPF + SNR-gated NR depth β | |
| β Phase 3 F0 contour revival: micro-pitch variation injection β | |
| β + natural jitter/shimmer restoration β | |
| β Phase 4 Formant reconstruction: rebuild missing/collapsed formants β | |
| β using EQ-based formant boosting from Arabic reference template β | |
| β Phase 5 Harmonic richness restoration: selective harmonic excitation β | |
| β on voiced frames only, F0-histogram-weighted β | |
| β Phase 6 Micro-dynamics breathing: phrase-level RMS modulation β | |
| β to restore natural loudness variation β | |
| β Phase 7 Spectral continuity repair: smooth spectral holes/edges β | |
| β from codec artifacts and aggressive NR β | |
| β Phase 8 Sibilant naturalization: rebuild sibilant texture β | |
| β (Ψ΄/Ψ³/Ψ΅/Ψ²) that codec/NR destroys β | |
| β Phase 9 Temporal envelope shaping: attack/transient restoration β | |
| β + consonant-vowel boundary sharpening + 2-4kHz presence boost β | |
| β Phase 10 Tajweed phoneme guards: 7-guard verification + repair β | |
| β Phase 11 Final loudness + quality optimization + naturalness scoring β | |
| β β | |
| β KEY DESIGN PRINCIPLES β | |
| β R1 Never hallucinate β only enhance what exists; don't synthesize β | |
| β new phonemes or words β | |
| β R2 Tajweed above all β F0 variation must never alter Tajweed β | |
| β phoneme identity (Β§35, Β§36, Β§79, Β§152) β | |
| β R3 Subtlety over spectacle β 0.5dB changes compound to naturalness β | |
| β R4 Measurement-driven β every phase has guards and reverts β | |
| β R5 Arabic-first β all processing calibrated for Arabic phonetics β | |
| β and Quranic recitation patterns (Murattal/Mujawwad/Hadr) β | |
| β β | |
| β KB REFS: Β§4 Β§35 Β§36 Β§52 Β§79 Β§80 Β§85 Β§91 Β§102 Β§109 Β§122 Β§133 β | |
| β Β§143 Β§145 Β§152 Β§154 Β§160 β | |
| β β | |
| β β ENGINE-3 v2.0 β THE AETHERION PROJECT β | |
| β Built for the Quran. ΩΩ Ψ§ Ψ§ΩΨͺΩΩΩΩ Ψ₯ΩΨ§ Ψ¨Ψ§ΩΩΩ β | |
| β β | |
| ββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ | |
| """ | |
| __version__ = 'v2.2' | |
| import argparse, json, os, shutil, subprocess, sys, time, tempfile, warnings | |
| from dataclasses import dataclass, field | |
| from pathlib import Path | |
| from typing import Dict, List, Optional, Tuple | |
| warnings.filterwarnings('ignore') | |
| _TMP = tempfile.gettempdir() | |
| try: | |
| import numpy as np | |
| from scipy.fft import rfft, rfftfreq | |
| from scipy.signal import medfilt, butter, sosfilt | |
| NUMPY_OK = SCIPY_OK = True | |
| except ImportError: | |
| NUMPY_OK = SCIPY_OK = False | |
| try: | |
| from voicefixer import VoiceFixer as _VoiceFixer | |
| VOICEFIXER_OK = True | |
| except ImportError: | |
| VOICEFIXER_OK = False | |
| try: | |
| import audiosr as _audiosr | |
| AUDIOSR_OK = True | |
| except ImportError: | |
| AUDIOSR_OK = False | |
| try: | |
| import soundfile as SF | |
| SF_OK = True | |
| except ImportError: | |
| SF_OK = False | |
| # ββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ | |
| # CONSTANTS | |
| # ββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ | |
| SR = 48000 | |
| WAV_CODEC = 'pcm_s24le' | |
| # Arabic sibilant bands (Β§152: Safir Ψ΅/Ψ³/Ψ²) | |
| ARABIC_SIB_BANDS = [2500.0, 3150.0, 4000.0, 5000.0, 6300.0, 8000.0] | |
| # Formant frequencies for adult male Arabic reciter (Β§4, Β§152) | |
| # F1: vowel height, F2: vowel backness, F3: lip rounding + nasal | |
| FORMANT_MALE_ARABIC = { | |
| 'F1': {'min': 250, 'typical': 350, 'max': 800}, # open vowels ~800, closed ~250 | |
| 'F2': {'min': 600, 'typical': 1400, 'max': 2500}, # back ~600, front ~2500 | |
| 'F3': {'min': 2200, 'typical': 2800, 'max': 3500}, # rounding/nasal | |
| } | |
| # F0 natural variation ranges for Quranic recitation (Β§85, Β§133) | |
| # Murattal: steadier, Mujawwad: more ornamentation | |
| F0_JITTER_MS = 0.35 # ms: natural cycle-to-cycle variation (Β§133.3) | |
| F0_SHIMMER_DB = 0.20 # dB: natural amplitude variation (Β§133.3) | |
| F0_DRIFT_CENTS = 12.0 # cents: slow drift within a phrase (Β§85) | |
| MUJAWWAD_DRIFT = 35.0 # cents: wider ornamentation drift (Β§145) | |
| # Artifact detection thresholds (Β§8, Β§21, Β§160) | |
| METALLIC_RING_THRESHOLD = 0.15 # ratio of harmonic false peaks (used in _detect_metallic_ringing) | |
| CODEC_GHOST_THRESHOLD = 0.08 # spectral hole depth ratio (used in _detect_spectral_holes) | |
| NR_DAMAGE_THRESHOLD = 0.25 # musical noise energy ratio (used in _detect_nr_damage severity) | |
| F0_FLATNESS_THRESHOLD = 0.85 # 1.0 = perfectly flat = robotic (Β§133) | |
| # Spectral smoothing (Β§7, Β§154) | |
| SPEC_SMOOTH_WINDOW = 3 # bands for median smoothing (used in _detect_spectral_holes) | |
| SPEC_HOLE_DEPTH_DB = -18.0 # band energy below this = spectral hole | |
| SPEC_EDGE_SHARP_DB = 8.0 # adjacent band delta > this = codec edge | |
| # Micro-dynamics (Β§85, Β§133) | |
| PHRASE_LRA_MIN = 1.5 # used in phase6 micro-dynamics LRA range | |
| PHRASE_LRA_MAX = 5.0 # used in phase6 micro-dynamics LRA range | |
| BREATH_DEPTH_DB = 1.5 # subtle RMS modulation depth (used in phase6) | |
| # Tajweed guard thresholds (Β§35, Β§52, Β§143) | |
| GHUNNAH_BAND = (220, 320) # Hz: nasal murmur (Β§152.3) | |
| IKHFA_BAND = (250, 420) # Hz: nasalisation (Β§52.5) | |
| QALQALAH_BURST_DB = 6.0 # dB: burst above silence | |
| RA_TRILL_AM_BAND = (22, 40) # Hz: Ra amplitude modulation | |
| SAFIR_BAND = (5500, 12000) # Hz: Ψ΅ Ψ³ Ψ² (Β§152.3) | |
| TAFASSHI_BAND = (3000, 8000) # Hz: Ψ΄ (Β§152.3) | |
| # Wind noise detection | |
| WIND_BAND_HZ = (20, 200) # Hz: wind rumble band | |
| WIND_REF_BAND_HZ = (200, 500) # Hz: reference band for comparison | |
| WIND_RATIO_THRESH = 6.0 # dB: wind band above reference = wind noise | |
| # SNR estimation | |
| SNR_NOISY_THRESH = 15.0 # dB: below this = noisy recording | |
| # Active bandwidth detection | |
| BANDWIDTH_FLOOR_DB = -60.0 # dB: minimum energy for "active" band | |
| BANDWIDTH_HPF_FREQ = 80.0 # Hz: ignore below this for bandwidth | |
| # ββ TIER_TELEPHONE detection thresholds (Β§169) ββββββββββββββββββββββββββββββββ | |
| TELEPHONE_CUTOFF_HZ = 4000.0 # recordings β€ this Hz = TIER_TELEPHONE | |
| TELEPHONE_ABOVE4K_THRESH = 0.001 # energy fraction above 4kHz must be < 0.1% | |
| TELEPHONE_ROLLOFF_MIN_DB = 35.0 # dB drop across pre/post-cutoff bands (Β§169: β₯40dB) | |
| TELEPHONE_BLEND_ORDER = 8 # Butterworth crossover filter order for blend step | |
| # 48-band centers (inherited from Itiqan) | |
| CENTERS_48: List[float] = [ | |
| 60.0, 80.0, 89.4, 100.0, 111.8, 125.0, 141.4, | |
| 160.0, 178.9, 200.0, 223.6, 250.0, 280.6, 315.0, | |
| 354.9, 400.0, 447.2, 500.0, 561.2, 630.0, 709.9, | |
| 800.0, 894.4, 1000.0, 1118.0, 1250.0, 1414.2, 1600.0, | |
| 1788.8, 2000.0, 2236.1, 2500.0, 2806.2, 3150.0, 3549.6, | |
| 4000.0, 4472.1, 5000.0, 5612.3, 6300.0, 7099.3, 8000.0, | |
| 8944.3,10000.0,11180.3,12500.0,14142.1,16000.0, | |
| ] | |
| # Reference targets for natural Quranic voice (from 1425H) | |
| TARGET = { | |
| 'lufs': -6.29, 'rms': -10.01, 'crest': 10.25, 'lra': 4.19, | |
| 'true_peak': -1.0, | |
| } | |
| # Perceptual loss weights (Β§154, Β§160) β used in naturalness score | |
| _PERC_WEIGHT: Dict[float, float] = { | |
| 125: 0.30, 250: 0.50, 500: 0.70, 1000: 0.90, | |
| 2000: 1.00, 3150: 1.00, 4000: 0.90, 5000: 0.70, | |
| 6300: 0.50, 8000: 0.40,10000: 0.30, | |
| } | |
| # ββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ | |
| # DATA CLASSES | |
| # ββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ | |
| class ArtifactReport: | |
| """Phase 0 diagnostic β what's wrong with the audio.""" | |
| f0_flatness: float = 0.0 # 0=highly variable, 1=flat=robotic | |
| f0_flat_severity: str = 'none' # none/mild/moderate/severe | |
| spectral_holes: int = 0 # count of bands with deep holes | |
| spectral_edges: int = 0 # count of sharp codec edges | |
| metallic_score: float = 0.0 # 0=clean, 1=severe ringing | |
| metallic_severity: str = 'none' # none/mild/moderate/severe | |
| codec_ghost_score: float = 0.0 # spectral leakage from codec | |
| nr_damage_score: float = 0.0 # musical noise from over-NR | |
| nr_damage_severity: str = 'none' # none/mild/moderate/severe | |
| formant_collapse: float = 0.0 # 0=intact, 1=fully collapsed | |
| sibilant_quality: float = 1.0 # 1=natural, 0=destroyed | |
| breathiness_score: float = 0.0 # 0=present, 1=missing breath | |
| overall_artificial: float = 0.0 # composite: 0=natural, 1=severe | |
| diagnosis: str = '' # human-readable diagnosis | |
| mujawwad_conf: float = 0.0 # recitation style confidence | |
| # v2 additions | |
| noise_floor_db: float = -60.0 # estimated noise floor | |
| snr_db: float = 40.0 # estimated SNR | |
| is_noisy: bool = False # SNR below threshold | |
| active_bandwidth_hz:float = 20000.0 # highest freq with meaningful energy | |
| wind_detected: bool = False # wind noise detected below 200Hz | |
| # v2.2 β TIER_TELEPHONE (Β§169) | |
| is_telephone_tier: bool = False # True if codec_cutoff β€ 4 kHz | |
| telephone_cutoff_hz:float = 0.0 # detected cutoff frequency (Hz) | |
| class IhyaState: | |
| """Runtime state for Ψ§ΩΨ₯ΨΩΨ§Ψ‘ engine.""" | |
| input_path: str = '' | |
| output_path: str = '' | |
| duration_s: float = 0.0 | |
| bitrate_kbps: int = 0 | |
| source_tier: str = 'TIER_UNKNOWN' | |
| mujawwad_conf: float = 0.0 | |
| aggressive: bool = False # v2: aggressive mode | |
| # Phase 0 results | |
| artifacts: 'ArtifactReport | None' = None | |
| # Measurements | |
| lufs: float = 0.0 | |
| rms: float = 0.0 | |
| crest: float = 0.0 | |
| lra: float = 0.0 | |
| true_peak: float = 0.0 | |
| f0_median: float = 0.0 | |
| # Phase results | |
| p0_analyzed: bool = False | |
| p1_dereverb_applied: bool = False | |
| p2_nr_repair_applied:bool = False | |
| p3_f0_revival: bool = False | |
| p3_jitter_injected: bool = False | |
| p3_shimmer_injected: bool = False | |
| p4_formant_rebuilt: bool = False | |
| p5_harmonic_rich: bool = False | |
| p6_micro_dynamics: bool = False | |
| p7_spectral_repair: bool = False | |
| p8_sibilant_nat: bool = False | |
| p9_temporal_shaping: bool = False | |
| p10_guards_ok: bool = False | |
| p11_final_done: bool = False | |
| # v2.2 β TIER_TELEPHONE (Β§169) | |
| is_telephone_tier: bool = False # bandwidth β€ 4 kHz detected | |
| telephone_bwe_applied: bool = False # VoiceFixer BWE phase ran | |
| telephone_audiosr_applied:bool = False # AudioSR Stage-B ran | |
| telephone_cutoff_hz: float = 0.0 # detected cutoff (Hz) | |
| telephone_bwe_mode: str = '' # 'voicefixer_mode0' / 'none' | |
| # Repair metrics | |
| f0_flatness_before: float = 0.0 | |
| f0_flatness_after: float = 0.0 | |
| spectral_holes_before: int = 0 | |
| spectral_holes_after: int = 0 | |
| metallic_before: float = 0.0 | |
| metallic_after: float = 0.0 | |
| # Guard tracking | |
| guard_pass: List[str] = field(default_factory=list) | |
| guard_warn: List[str] = field(default_factory=list) | |
| guard_reverts: int = 0 | |
| # Naturalness score (v2) | |
| naturalness_score: float = 50.0 # 0-100 composite | |
| # Temp file tracking | |
| _tmps: List[str] = field(default_factory=list, repr=False) | |
| # Processing time | |
| t0: float = 0.0 | |
| # ββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ | |
| # LOGGER + FFMPEG RUNNER | |
| # ββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ | |
| _LOG: List[str] = [] | |
| def L(msg: str) -> None: | |
| _LOG.append(msg) | |
| print(msg, flush=True) | |
| def _chk(label: str) -> None: | |
| L(f'\nββ {label} ββ') | |
| def _run_ffmpeg(cmd: List[str], capture: bool = False, timeout: int = 600) -> Tuple[int, str, str]: | |
| """Run ffmpeg command. Returns (returncode, stdout, stderr).""" | |
| try: | |
| result = subprocess.run(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, timeout=timeout) | |
| stdout = result.stdout.decode('utf-8', errors='replace') | |
| stderr = result.stderr.decode('utf-8', errors='replace') | |
| return result.returncode, stdout, stderr | |
| except subprocess.TimeoutExpired: | |
| return 1, '', 'TIMEOUT' | |
| except FileNotFoundError: | |
| return 1, '', 'ffmpeg not found' | |
| def _tmp_wav(suffix: str = '', st: Optional[IhyaState] = None) -> str: | |
| p = os.path.join(_TMP, f'ihya_{os.getpid()}_{suffix}_{int(time.time()*1000)}.wav') | |
| if st is not None: | |
| st._tmps.append(p) | |
| return p | |
| def _cleanup(*paths: str) -> None: | |
| for p in paths: | |
| try: | |
| if p and os.path.exists(p): | |
| os.remove(p) | |
| except OSError: | |
| pass | |
| def _cleanup_all(st: IhyaState) -> None: | |
| for p in st._tmps: | |
| try: | |
| if p and os.path.exists(p): os.unlink(p) | |
| except Exception: | |
| pass | |
| st._tmps.clear() | |
| # ββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ | |
| # AUDIO MEASUREMENT β inherited from Itiqan + extensions | |
| # ββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ | |
| def _decode_to_wav(input_path: str, output_wav: str) -> bool: | |
| """Decode any audio to 48kHz mono 24-bit WAV.""" | |
| cmd = ['ffmpeg', '-y', '-i', input_path, | |
| '-ar', str(SR), '-ac', '1', '-acodec', WAV_CODEC, output_wav] | |
| rc, _, _ = _run_ffmpeg(cmd) | |
| return rc == 0 and os.path.exists(output_wav) | |
| def _decode_samples(wav_path: str) -> Tuple[Optional['np.ndarray'], int]: | |
| """Decode WAV to numpy float32 array.""" | |
| if not NUMPY_OK: | |
| return None, SR | |
| # Use pipe decode for speed | |
| try: | |
| r = subprocess.run( | |
| ['ffmpeg', '-nostdin', '-y', '-hide_banner', '-loglevel', 'error', | |
| '-i', wav_path, '-ar', str(SR), '-ac', '1', '-f', 'f32le', '-'], | |
| capture_output=True, timeout=300) | |
| if r.returncode or len(r.stdout) < 4: | |
| # Fallback: 16-bit decode | |
| tmp = os.path.join(_TMP, f'ihya_pcm16_{os.getpid()}.pcm') | |
| cmd = ['ffmpeg', '-y', '-i', wav_path, | |
| '-ar', str(SR), '-ac', '1', '-f', 's16le', tmp] | |
| rc, _, _ = _run_ffmpeg(cmd) | |
| if rc != 0 or not os.path.exists(tmp): | |
| return None, SR | |
| try: | |
| raw = np.fromfile(tmp, dtype=np.int16) | |
| samples = raw.astype(np.float32) / 32768.0 | |
| return samples, SR | |
| finally: | |
| _cleanup(tmp) | |
| data = np.frombuffer(r.stdout, dtype=np.float32).copy() | |
| return data, SR | |
| except Exception: | |
| return None, SR | |
| def _get_duration(path: str) -> float: | |
| rc, out, _ = _run_ffmpeg([ | |
| 'ffprobe', '-v', 'error', '-show_entries', 'format=duration', | |
| '-of', 'default=noprint_wrappers=1:nokey=1', path | |
| ]) | |
| try: | |
| return float(out.strip()) | |
| except ValueError: | |
| return 0.0 | |
| def _get_bitrate(path: str) -> int: | |
| rc, out, _ = _run_ffmpeg([ | |
| 'ffprobe', '-v', 'error', '-show_entries', 'format=bit_rate', | |
| '-of', 'default=noprint_wrappers=1:nokey=1', path | |
| ]) | |
| try: | |
| return int(out.strip()) // 1000 | |
| except ValueError: | |
| return 0 | |
| def _measure_lufs(wav_path: str) -> Tuple[float, float]: | |
| """Returns (integrated_lufs, lra).""" | |
| cmd = ['ffmpeg', '-i', wav_path, | |
| '-af', 'ebur128=peak=true:framelog=quiet', '-f', 'null', '-'] | |
| rc, out, err = _run_ffmpeg(cmd) | |
| combined = out + err | |
| lufs, lra = -99.0, 0.0 | |
| for line in combined.splitlines(): | |
| if 'I:' in line and 'LUFS' in line: | |
| try: | |
| lufs = float(line.split('I:')[1].split('LUFS')[0].strip()) | |
| except (IndexError, ValueError): | |
| pass | |
| if 'LRA:' in line and 'LU' in line: | |
| try: | |
| lra = float(line.split('LRA:')[1].split('LU')[0].strip()) | |
| except (IndexError, ValueError): | |
| pass | |
| return lufs, lra | |
| def _measure_rms_crest(samples: 'np.ndarray') -> Tuple[float, float]: | |
| """Returns (rms_db, crest_db).""" | |
| if samples is None or len(samples) == 0: | |
| return -99.0, 0.0 | |
| rms_linear = float(np.sqrt(np.mean(samples ** 2))) | |
| peak_linear = float(np.max(np.abs(samples))) | |
| rms_db = 20 * np.log10(max(rms_linear, 1e-10)) | |
| peak_db = 20 * np.log10(max(peak_linear, 1e-10)) | |
| crest_db = peak_db - rms_db | |
| return rms_db, crest_db | |
| def _band_energy(samples: 'np.ndarray', flo: float, fhi: float, | |
| sr: int = SR, n_fft: int = 4096) -> float: | |
| """Band energy in [flo, fhi] Hz, sampled across the signal.""" | |
| if samples is None or len(samples) < n_fft: | |
| return 0.0 | |
| n_samples = min(8, max(1, len(samples) // n_fft)) | |
| step = max(n_fft, len(samples) // (n_samples + 1)) | |
| energies = [] | |
| freqs = np.fft.rfftfreq(n_fft, 1.0 / sr) | |
| mask = (freqs >= flo) & (freqs <= fhi) | |
| if not mask.any(): | |
| return 0.0 | |
| for pos in range(0, len(samples) - n_fft, step): | |
| sp = np.abs(np.fft.rfft(samples[pos:pos + n_fft], n=n_fft)) | |
| energies.append(float(np.mean(sp[mask] ** 2) + 1e-20)) | |
| return float(np.mean(energies)) if energies else 0.0 | |
| def _band_energy_db(samples: 'np.ndarray', flo: float, fhi: float, | |
| sr: int = SR) -> float: | |
| """Band energy in dB.""" | |
| e = _band_energy(samples, flo, fhi, sr) | |
| if e < 1e-20: | |
| return -100.0 | |
| return float(10 * np.log10(e)) | |
| def _rmsdb(s: 'np.ndarray') -> float: | |
| return float(20 * np.log10(np.sqrt(np.mean(s ** 2)) + 1e-10)) | |
| # ββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ | |
| # NEW v2 DETECTION: SNR, ACTIVE BANDWIDTH, WIND NOISE | |
| # ββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ | |
| def _estimate_snr(samples: 'np.ndarray', sr: int = SR) -> Tuple[float, float, bool]: | |
| """ | |
| Estimate noise floor and SNR from the audio signal. | |
| Uses frame-level energy analysis: quiet frames represent the noise floor, | |
| loud frames represent signal+noise. | |
| Returns (noise_floor_db, snr_db, is_noisy). | |
| """ | |
| if not NUMPY_OK or samples is None: | |
| return -60.0, 40.0, False | |
| FRAME_MS = 20.0 | |
| frame_n = int(FRAME_MS / 1000.0 * sr) | |
| n_frames = len(samples) // frame_n | |
| if n_frames < 20: | |
| return -60.0, 40.0, False | |
| # Compute per-frame RMS | |
| frame_rms = np.array([float(np.sqrt(np.mean(samples[i*frame_n:(i+1)*frame_n]**2))) | |
| for i in range(n_frames)]) | |
| # Noise floor: use 10th percentile of frame energies (true silence/noise) | |
| voiced_mask = frame_rms > 1e-7 | |
| if not voiced_mask.any(): | |
| return -60.0, 40.0, False | |
| noise_rms = float(np.percentile(frame_rms[voiced_mask], 10)) | |
| # Signal level: use 75th percentile (typical speech level) | |
| signal_rms = float(np.percentile(frame_rms[voiced_mask], 75)) | |
| noise_floor_db = 20 * np.log10(max(noise_rms, 1e-10)) | |
| signal_db = 20 * np.log10(max(signal_rms, 1e-10)) | |
| snr_db = signal_db - noise_floor_db | |
| is_noisy = snr_db < SNR_NOISY_THRESH | |
| return noise_floor_db, snr_db, is_noisy | |
| def _detect_active_bandwidth(samples: 'np.ndarray', sr: int = SR) -> float: | |
| """ | |
| Detect the highest frequency with meaningful energy. | |
| Low-bitrate codecs cut off above 12-16kHz, leaving silence above that. | |
| Returns bandwidth in Hz. | |
| """ | |
| if not NUMPY_OK or samples is None: | |
| return sr / 2.0 | |
| # Use FFT on central portion | |
| center_start = int(len(samples) * 0.15) | |
| center_end = int(len(samples) * 0.85) | |
| center = samples[center_start:center_end] | |
| N = min(len(center), sr * 2) | |
| if N < sr // 4: | |
| return sr / 2.0 | |
| spec = np.abs(rfft(center[:N] * np.hanning(N))) ** 2 | |
| freqs = rfftfreq(N, d=1.0 / sr) | |
| # Overall spectral floor | |
| overall_floor = float(np.percentile(spec[spec > 0], 10)) if (spec > 0).any() else 1e-20 | |
| # Scan from high to low frequency | |
| # Find highest frequency above BANDWIDTH_HPF_FREQ with energy above floor | |
| mask_above_hpf = freqs >= BANDWIDTH_HPF_FREQ | |
| freqs_filtered = freqs[mask_above_hpf] | |
| spec_filtered = spec[mask_above_hpf] | |
| if len(spec_filtered) == 0: | |
| return sr / 2.0 | |
| # Use 1/3-octave bands for robustness | |
| active_bw = BANDWIDTH_HPF_FREQ | |
| for fc in reversed(CENTERS_48): | |
| if fc < BANDWIDTH_HPF_FREQ or fc > sr / 2: | |
| continue | |
| lo = fc / (2 ** (1.0/6)) | |
| hi = fc * (2 ** (1.0/6)) | |
| band_mask = (freqs >= lo) & (freqs <= hi) | |
| if band_mask.any(): | |
| band_energy = float(np.mean(spec[band_mask])) | |
| # Band is "active" if it's significantly above the noise floor | |
| if band_energy > overall_floor * 100: # 20dB above floor | |
| active_bw = max(active_bw, hi) | |
| break | |
| return float(active_bw) | |
| def _detect_wind_noise(samples: 'np.ndarray', sr: int = SR) -> bool: | |
| """ | |
| Detect wind noise below 200Hz. | |
| Mosque/outdoor recordings often have wind rumble. | |
| Wind noise shows as disproportionately high energy in 20-200Hz | |
| relative to the 200-500Hz reference band. | |
| Returns True if wind noise is detected. | |
| """ | |
| if not NUMPY_OK or samples is None: | |
| return False | |
| wind_energy = _band_energy(samples, WIND_BAND_HZ[0], WIND_BAND_HZ[1], sr) | |
| ref_energy = _band_energy(samples, WIND_REF_BAND_HZ[0], WIND_REF_BAND_HZ[1], sr) | |
| if ref_energy < 1e-20: | |
| return False | |
| ratio_db = 10 * np.log10(wind_energy / (ref_energy + 1e-20)) | |
| return ratio_db > WIND_RATIO_THRESH | |
| def _detect_telephone_tier(samples: 'np.ndarray', | |
| active_bandwidth_hz: float, | |
| sr: int = SR) -> Tuple[bool, float]: | |
| """ | |
| Detect TIER_TELEPHONE: recording with hard spectral cutoff β€ 4 kHz. | |
| Β§169 criteria: | |
| (1) active_bandwidth_hz already < 4000 Hz | |
| (2) energy above 4 kHz < 0.1% of total power | |
| (3) steep rolloff: β₯ 35 dB drop across pre-cutoff / post-cutoff bands | |
| Returns (is_telephone_tier, cutoff_hz). | |
| KEY INSIGHT (Β§169): Container bitrate is IRRELEVANT. | |
| 128 kbps AAC wrapping a 3.4 kHz telephone source is still TIER_TELEPHONE. | |
| """ | |
| if active_bandwidth_hz >= TELEPHONE_CUTOFF_HZ: | |
| return False, 0.0 | |
| if not NUMPY_OK or samples is None or len(samples) < sr // 2: | |
| # Trust the bandwidth detection alone | |
| return True, float(min(active_bandwidth_hz, 3400.0)) | |
| # FFT on central portion (avoid head/tail silence) | |
| center_s = int(len(samples) * 0.10) | |
| center_e = int(len(samples) * 0.90) | |
| seg = samples[center_s:center_e] | |
| N = min(len(seg), sr * 2) | |
| if N < 512: | |
| return True, float(min(active_bandwidth_hz, 3400.0)) | |
| win = seg[:N] * np.hanning(N) | |
| Pxx = np.abs(rfft(win, n=N)) ** 2 | |
| freqs = rfftfreq(N, d=1.0 / sr) | |
| total = float(Pxx.sum()) + 1e-30 | |
| # Criterion 1: almost no energy above 4 kHz (< 0.1 %) | |
| above_4k_frac = float(Pxx[freqs > 4000].sum()) / total | |
| if above_4k_frac >= TELEPHONE_ABOVE4K_THRESH: | |
| return False, 0.0 | |
| # Criterion 2: must have signal below 3.5 kHz (not a dead/silent file) | |
| band_signal = float(Pxx[(freqs >= 300) & (freqs <= 3500)].sum()) / total | |
| if band_signal < 0.01: | |
| return False, 0.0 | |
| # Criterion 3: steep rolloff β compare band just below cutoff vs just above | |
| # Use 2.5β3.4 kHz (below) vs 3.8β5.0 kHz (above) | |
| band_below = float(Pxx[(freqs >= 2500) & (freqs <= 3400)].sum()) + 1e-30 | |
| band_above = float(Pxx[(freqs >= 3800) & (freqs <= 5000)].sum()) + 1e-30 | |
| rolloff_db = 10.0 * np.log10(band_below / band_above) | |
| if rolloff_db < TELEPHONE_ROLLOFF_MIN_DB: | |
| return False, 0.0 | |
| # Estimate cutoff more precisely: walk down from 4 kHz to find last | |
| # bin with meaningful energy (> 1% of average below 3 kHz) | |
| ref_level = float(np.mean(Pxx[(freqs >= 500) & (freqs <= 2500)])) | |
| thresh = ref_level * 0.01 | |
| cutoff_est = 3400.0 | |
| for fc in np.arange(3900, 1000, -50, dtype=float): | |
| mask = (freqs >= fc - 25) & (freqs < fc + 25) | |
| if mask.any() and float(Pxx[mask].mean()) > thresh: | |
| cutoff_est = fc | |
| break | |
| return True, float(cutoff_est) | |
| # ββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ | |
| # Phase 0: DEEP ARTIFACT ANALYSIS | |
| # ββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ | |
| def _detect_f0(samples: 'np.ndarray', sr: int = SR) -> Tuple['np.ndarray', float]: | |
| """ | |
| Detect F0 contour using autocorrelation on voiced frames. | |
| Returns (f0_contour_array, f0_median). | |
| Each element is Hz (0.0 for unvoiced frames). | |
| """ | |
| if not NUMPY_OK or samples is None: | |
| return np.array([]), 0.0 | |
| FRAME_MS = 20.0 | |
| frame_n = int(FRAME_MS / 1000.0 * sr) | |
| n_frames = len(samples) // frame_n | |
| if n_frames < 10: | |
| return np.array([]), 0.0 | |
| f0_contour = np.zeros(n_frames) | |
| min_lag = int(sr / 500.0) # 500 Hz max F0 | |
| max_lag = int(sr / 60.0) # 60 Hz min F0 | |
| for i in range(n_frames): | |
| frame = samples[i * frame_n:(i + 1) * frame_n] | |
| rms = float(np.sqrt(np.mean(frame ** 2))) | |
| if rms < 1e-5: # silence/unvoiced | |
| continue | |
| # Autocorrelation | |
| corr = np.correlate(frame, frame, mode='full') | |
| corr = corr[len(frame) - 1:] # keep positive lags only | |
| # Find peak in F0 range | |
| search = corr[min_lag:min(max_lag, len(corr))] | |
| if len(search) < 2: | |
| continue | |
| peak_idx = int(np.argmax(search)) + min_lag | |
| if corr[peak_idx] < 0.3 * corr[0]: # too weak = unvoiced | |
| continue | |
| f0_contour[i] = sr / peak_idx | |
| # Median of voiced frames | |
| voiced = f0_contour[f0_contour > 0] | |
| f0_median = float(np.median(voiced)) if len(voiced) > 5 else 0.0 | |
| return f0_contour, f0_median | |
| def _compute_f0_flatness(f0_contour: 'np.ndarray') -> float: | |
| """ | |
| Compute F0 flatness (0 = highly variable/natural, 1 = perfectly flat/robotic). | |
| Uses coefficient of variation of voiced F0 values. | |
| Β§133: natural speech has F0 variation > 30 cents (semi-tones). | |
| Robotic/TTS has < 10 cents variation. | |
| """ | |
| voiced = f0_contour[f0_contour > 0] | |
| if len(voiced) < 10: | |
| return 0.0 # not enough data | |
| # Convert to cents relative to median (more perceptually relevant) | |
| median_f0 = float(np.median(voiced)) | |
| if median_f0 < 50: | |
| return 0.0 | |
| cents = 1200.0 * np.log2(voiced / median_f0) | |
| std_cents = float(np.std(cents)) | |
| # Natural recitation: std_cents typically 15-40 cents | |
| # Robotic: std_cents < 5 cents | |
| # Map: 0 cents std β 1.0 flat, 40+ cents β 0.0 flat | |
| flatness = float(np.clip(1.0 - std_cents / 40.0, 0.0, 1.0)) | |
| return flatness | |
| def _detect_spectral_holes(samples: 'np.ndarray', sr: int = SR) -> Tuple[int, List[float]]: | |
| """ | |
| Detect spectral holes β bands where energy drops abnormally. | |
| These are typical of codec artifacts (Β§8, Β§21) or aggressive NR (Β§160). | |
| Returns (count, list_of_center_frequencies). | |
| """ | |
| if not NUMPY_OK or samples is None: | |
| return 0, [] | |
| holes = [] | |
| # Compute per-band energy | |
| band_db = {} | |
| for f in CENTERS_48: | |
| if f < 80 or f > 16000: | |
| continue | |
| # Approximate band: center Β± 1/12 octave | |
| lo = f / (2 ** (1.0/12)) | |
| hi = f * (2 ** (1.0/12)) | |
| e_db = _band_energy_db(samples, lo, hi, sr) | |
| band_db[f] = e_db | |
| if len(band_db) < 4: | |
| return 0, [] | |
| # Apply median smoothing using SPEC_SMOOTH_WINDOW to reduce noise | |
| sorted_centers = sorted(band_db.keys()) | |
| db_values = np.array([band_db[f] for f in sorted_centers]) | |
| if len(db_values) >= SPEC_SMOOTH_WINDOW: | |
| kernel_size = SPEC_SMOOTH_WINDOW if SPEC_SMOOTH_WINDOW % 2 == 1 else SPEC_SMOOTH_WINDOW + 1 | |
| db_smoothed = medfilt(db_values, kernel_size=kernel_size) | |
| for i, f in enumerate(sorted_centers): | |
| band_db[f] = float(db_smoothed[i]) | |
| # A spectral hole is a band that is significantly below its neighbors | |
| for i in range(1, len(sorted_centers) - 1): | |
| prev_db = band_db[sorted_centers[i - 1]] | |
| curr_db = band_db[sorted_centers[i]] | |
| next_db = band_db[sorted_centers[i + 1]] | |
| neighbor_avg = (prev_db + next_db) / 2.0 | |
| dip = curr_db - neighbor_avg # negative = hole | |
| if dip < -abs(SPEC_HOLE_DEPTH_DB): | |
| holes.append(sorted_centers[i]) | |
| return len(holes), holes | |
| def _detect_spectral_edges(samples: 'np.ndarray', sr: int = SR) -> Tuple[int, List[float]]: | |
| """ | |
| Detect sharp spectral edges β typical of codec block artifacts. | |
| Adjacent band energy delta > threshold indicates codec edge. | |
| """ | |
| if not NUMPY_OK or samples is None: | |
| return 0, [] | |
| edges = [] | |
| band_db = {} | |
| for f in CENTERS_48: | |
| if f < 80 or f > 16000: | |
| continue | |
| lo = f / (2 ** (1.0/12)) | |
| hi = f * (2 ** (1.0/12)) | |
| e_db = _band_energy_db(samples, lo, hi, sr) | |
| band_db[f] = e_db | |
| sorted_centers = sorted(band_db.keys()) | |
| for i in range(1, len(sorted_centers)): | |
| delta = abs(band_db[sorted_centers[i]] - band_db[sorted_centers[i - 1]]) | |
| if delta > SPEC_EDGE_SHARP_DB: | |
| edges.append(sorted_centers[i]) | |
| return len(edges), edges | |
| def _detect_metallic_ringing(samples: 'np.ndarray', sr: int = SR) -> float: | |
| """ | |
| Detect metallic/ringing artifacts. | |
| Metallic sound = harmonic false peaks at non-harmonic frequencies (Β§8, Β§21). | |
| Analyze spectrum for peaks that don't align with F0 harmonics. | |
| Returns score 0.0 (clean) to 1.0 (severe). | |
| Uses METALLIC_RING_THRESHOLD for classification. | |
| """ | |
| if not NUMPY_OK or samples is None: | |
| return 0.0 | |
| # Use central portion to avoid transients | |
| center_start = int(len(samples) * 0.15) | |
| center_end = int(len(samples) * 0.85) | |
| center = samples[center_start:center_end] | |
| N = min(len(center), sr * 2) | |
| if N < sr // 4: | |
| return 0.0 | |
| spec = np.abs(rfft(center[:N] * np.hanning(N))) ** 2 | |
| freqs = rfftfreq(N, d=1.0 / sr) | |
| # Find fundamental peak in 100-400Hz | |
| mask_fund = (freqs >= 100) & (freqs <= 400) | |
| if not mask_fund.any(): | |
| return 0.0 | |
| f1_idx = int(np.argmax(spec[mask_fund])) + int(np.where(mask_fund)[0][0]) | |
| f1 = float(freqs[f1_idx]) | |
| if f1 < 80: | |
| return 0.0 | |
| # Count spectral peaks that are NOT at harmonic positions | |
| # Harmonics at f1, 2*f1, 3*f1, ... up to Nyquist | |
| harmonic_positions = [f1 * k for k in range(2, int(sr / 2 / f1) + 1)] | |
| harmonic_width_hz = f1 * 0.12 # Β±12% tolerance | |
| # Find all significant peaks | |
| from scipy.signal import find_peaks as _find_peaks | |
| try: | |
| peaks, props = _find_peaks(np.log10(spec[1:] + 1e-20) * 10, | |
| height=0, distance=int(N * 50 / sr)) | |
| except ImportError: | |
| return 0.0 | |
| if len(peaks) < 3: | |
| return 0.0 | |
| peak_freqs = freqs[peaks + 1] # +1 because we sliced spec[1:] | |
| peak_heights = spec[peaks + 1] | |
| # Classify each peak: harmonic or non-harmonic | |
| non_harmonic_energy = 0.0 | |
| total_peak_energy = 0.0 | |
| for pf, ph in zip(peak_freqs, peak_heights): | |
| total_peak_energy += ph | |
| is_harmonic = any(abs(pf - hp) < harmonic_width_hz for hp in harmonic_positions) | |
| if not is_harmonic and pf > f1 * 1.5: # ignore sub-harmonics | |
| non_harmonic_energy += ph | |
| if total_peak_energy < 1e-20: | |
| return 0.0 | |
| ratio = non_harmonic_energy / total_peak_energy | |
| score = float(np.clip(ratio * 5.0, 0.0, 1.0)) | |
| # Apply METALLIC_RING_THRESHOLD for severity flagging | |
| # (threshold used by caller for classification) | |
| return score | |
| def _detect_nr_damage(samples: 'np.ndarray', sr: int = SR) -> float: | |
| """ | |
| Detect musical noise from over-aggressive NR (Β§11, Β§160). | |
| Musical noise = random tonal artifacts in quiet frames. | |
| Returns score 0.0 (clean) to 1.0 (severe). | |
| v2 FIX: Only count frames BELOW median energy as "true background noise". | |
| Previous version used bottom 25% which incorrectly flagged clean tonal | |
| content (e.g. sine waves, speech leakage into quiet frames) as musical | |
| noise. True background noise frames are below the median energy level. | |
| """ | |
| if not NUMPY_OK or samples is None: | |
| return 0.0 | |
| FRAME_MS = 20.0 | |
| frame_n = int(FRAME_MS / 1000.0 * sr) | |
| n_frames = len(samples) // frame_n | |
| if n_frames < 20: | |
| return 0.0 | |
| # Compute per-frame RMS | |
| frame_rms = np.array([float(np.sqrt(np.mean(samples[i*frame_n:(i+1)*frame_n]**2))) | |
| for i in range(n_frames)]) | |
| # v2 FIX: Use median energy as threshold instead of 25th percentile. | |
| # Frames below median are true background noise, not speech leakage. | |
| # A pure sine wave has roughly constant RMS across frames, so no frames | |
| # fall below median β correctly NOT flagged as NR damage. | |
| voiced_frames = frame_rms[frame_rms > 1e-7] | |
| if len(voiced_frames) < 5: | |
| return 0.0 | |
| median_energy = float(np.median(voiced_frames)) | |
| # v2.1 FIX: NR damage detection must distinguish between: | |
| # (a) musical noise = isolated tonal islands in quiet regions | |
| # (b) speech leakage = harmonics from voiced speech bleeding into quiet frames | |
| # (c) continuous harmonics = the signal itself (not noise) | |
| # | |
| # Key insight: musical noise is SPARSE β only a few isolated frequency bins | |
| # are active per frame, and they vary RANDOMLY between frames (no temporal | |
| # coherence). Speech harmonics are TEMPORALLY COHERENT across frames. | |
| # | |
| # Detection strategy: measure spectral SPARSITY (peak-to-mean ratio) in | |
| # quiet frames. Musical noise has very high sparsity (few isolated peaks) | |
| # while natural noise/speech leakage has low sparsity (many peaks or flat). | |
| # Only consider frames well below median (true noise floor) | |
| quiet_thresh = median_energy * 0.3 # well below speech level | |
| quiet_frames = [] | |
| for i in range(n_frames): | |
| if frame_rms[i] > 1e-7 and frame_rms[i] < quiet_thresh: | |
| quiet_frames.append(samples[i*frame_n:(i+1)*frame_n]) | |
| if len(quiet_frames) < 3: | |
| return 0.0 # not enough quiet frames to analyze | |
| # Measure spectral sparsity (peak-to-mean ratio) in quiet frames | |
| sparsity_values = [] | |
| for qf in quiet_frames: | |
| N = min(len(qf), 2048) | |
| spec = np.abs(rfft(qf[:N])) ** 2 | |
| spec = spec[1:] # remove DC | |
| if len(spec) < 2 or spec.max() < 1e-20: | |
| continue | |
| # Peak-to-mean ratio: high = sparse (musical noise), low = flat (natural) | |
| peak = float(np.max(spec)) | |
| mean = float(np.mean(spec)) | |
| if mean > 1e-20: | |
| sparsity = peak / mean # >10 = very sparse, <3 = flat | |
| sparsity_values.append(sparsity) | |
| if len(sparsity_values) < 2: | |
| return 0.0 | |
| avg_sparsity = float(np.mean(sparsity_values)) | |
| # Musical noise: sparsity > 20 (few isolated tonal peaks) | |
| # Natural noise: sparsity < 5 (flat or many peaks) | |
| # Speech leakage: sparsity 5-15 (harmonic structure but broader) | |
| if avg_sparsity < 5: | |
| return 0.0 # natural noise, not musical noise | |
| elif avg_sparsity > 20: | |
| damage = 1.0 # severe musical noise | |
| else: | |
| damage = float(np.clip((avg_sparsity - 5.0) / 15.0, 0.0, 1.0)) | |
| return damage | |
| def _detect_formant_collapse(samples: 'np.ndarray', sr: int = SR) -> float: | |
| """ | |
| Detect formant collapse β when F1/F2 distinction is lost. | |
| This makes voice sound hollow/thin/telephone-quality (Β§4, Β§8). | |
| Returns 0.0 (intact) to 1.0 (fully collapsed). | |
| """ | |
| if not NUMPY_OK or samples is None: | |
| return 0.0 | |
| # Measure energy in formant bands using FORMANT_MALE_ARABIC reference | |
| f1_lo = FORMANT_MALE_ARABIC['F1']['min'] | |
| f1_hi = FORMANT_MALE_ARABIC['F1']['max'] | |
| f2_lo = FORMANT_MALE_ARABIC['F2']['min'] | |
| f2_hi = FORMANT_MALE_ARABIC['F2']['max'] | |
| f1_energy = _band_energy(samples, f1_lo, f1_hi, sr) | |
| f2_energy = _band_energy(samples, f2_lo, f2_hi, sr) | |
| if f1_energy < 1e-20 or f2_energy < 1e-20: | |
| return 0.5 # can't measure, assume moderate | |
| # Natural voice: F1 and F2 have similar energy levels | |
| # Collapsed: F2 much weaker than F1 (spectral tilt) | |
| ratio_db = 10 * np.log10(f2_energy / (f1_energy + 1e-20)) | |
| # Natural: ratio around -3 to +3 dB | |
| # Collapsed: ratio < -10 dB | |
| if ratio_db > -3: | |
| return 0.0 | |
| elif ratio_db < -15: | |
| return 1.0 | |
| else: | |
| return float(np.clip((-ratio_db - 3) / 12.0, 0.0, 1.0)) | |
| def _detect_sibilant_quality(samples: 'np.ndarray', sr: int = SR) -> float: | |
| """ | |
| Assess sibilant (Ψ΄/Ψ³/Ψ΅/Ψ²) quality. | |
| Returns 1.0 (natural) to 0.0 (destroyed/artificial). | |
| Β§152: Arabic sibilants have characteristic spectral signatures. | |
| """ | |
| if not NUMPY_OK or samples is None: | |
| return 1.0 | |
| safir_energy = _band_energy(samples, 5500, 12000, sr) | |
| mid_energy = _band_energy(samples, 1000, 3000, sr) | |
| if mid_energy < 1e-20: | |
| return 1.0 | |
| # Natural: sibilant energy ~ -10 to -25 dB relative to mid | |
| # Destroyed: sibilant energy < -35 dB (over-NR'd) | |
| # Artificial: sibilant energy > -5 dB (too bright/harsh) | |
| ratio_db = 10 * np.log10(safir_energy / (mid_energy + 1e-20)) | |
| if -25 <= ratio_db <= -5: | |
| return 1.0 # natural | |
| elif ratio_db < -35: | |
| return 0.0 # destroyed | |
| elif ratio_db > 0: | |
| return 0.5 # harsh/artificial | |
| else: | |
| return float(np.clip(1.0 - abs(ratio_db + 15) / 20.0, 0.0, 1.0)) | |
| def _detect_mujawwad(samples: 'np.ndarray', f0_contour: 'np.ndarray', | |
| sr: int = SR) -> float: | |
| """ | |
| Detect Mujawwad recitation style confidence. | |
| Β§85, Β§145: Mujawwad has wider F0 variation, longer madd, more ornamentation. | |
| Returns 0.0 (Murattal/Hadr) to 1.0 (Mujawwad). | |
| """ | |
| if not NUMPY_OK or len(f0_contour) < 20: | |
| return 0.0 | |
| voiced = f0_contour[f0_contour > 0] | |
| if len(voiced) < 10: | |
| return 0.0 | |
| # Mujawwad indicators: | |
| # 1. Wider F0 range (more than 200 cents between min and max) | |
| median_f0 = float(np.median(voiced)) | |
| if median_f0 < 50: | |
| return 0.0 | |
| cents = 1200.0 * np.log2(voiced / median_f0) | |
| f0_range = float(np.percentile(cents, 95) - np.percentile(cents, 5)) | |
| # 2. Higher F0 standard deviation | |
| f0_std = float(np.std(cents)) | |
| # 3. More F0 contour direction changes (ornamentation) | |
| diff = np.diff(cents) | |
| direction_changes = int(np.sum(diff[1:] * diff[:-1] < 0)) | |
| # Scoring | |
| range_score = float(np.clip(f0_range / 300.0, 0.0, 1.0)) # 300 cents = strong Mujawwad | |
| std_score = float(np.clip(f0_std / 40.0, 0.0, 1.0)) # 40 cents std | |
| orn_score = float(np.clip(direction_changes / (len(diff) * 0.3), 0.0, 1.0)) | |
| conf = range_score * 0.4 + std_score * 0.35 + orn_score * 0.25 | |
| return float(np.clip(conf, 0.0, 1.0)) | |
| def _severity(score: float, mild: float = 0.25, moderate: float = 0.50, | |
| severe: float = 0.75) -> str: | |
| if score < mild: | |
| return 'none' | |
| elif score < moderate: | |
| return 'mild' | |
| elif score < severe: | |
| return 'moderate' | |
| else: | |
| return 'severe' | |
| def phase0_analyze(samples: 'np.ndarray', sr: int, st: IhyaState) -> ArtifactReport: | |
| """ | |
| Phase 0: Deep artifact analysis. | |
| Determines exactly what's wrong with the audio and how severe. | |
| v2: Also estimates SNR, active bandwidth, and wind noise. | |
| """ | |
| _chk('phase_0_analysis') | |
| report = ArtifactReport() | |
| # F0 analysis | |
| f0_contour, f0_median = _detect_f0(samples, sr) | |
| st.f0_median = f0_median | |
| L(f' F0 median={f0_median:.1f}Hz voiced_frames={np.sum(f0_contour > 0)}/{len(f0_contour)}') | |
| # F0 flatness | |
| report.f0_flatness = _compute_f0_flatness(f0_contour) | |
| report.f0_flat_severity = _severity(report.f0_flatness, 0.40, 0.60, 0.80) | |
| L(f' F0 flatness={report.f0_flatness:.3f} ({report.f0_flat_severity})') | |
| # Spectral holes | |
| report.spectral_holes, hole_freqs = _detect_spectral_holes(samples, sr) | |
| L(f' Spectral holes={report.spectral_holes} freqs={[f"{f:.0f}" for f in hole_freqs[:6]]}') | |
| # Spectral edges | |
| report.spectral_edges, edge_freqs = _detect_spectral_edges(samples, sr) | |
| L(f' Spectral edges={report.spectral_edges}') | |
| # Metallic ringing | |
| report.metallic_score = _detect_metallic_ringing(samples, sr) | |
| report.metallic_severity = _severity(report.metallic_score) | |
| L(f' Metallic score={report.metallic_score:.3f} ({report.metallic_severity})') | |
| # NR damage | |
| report.nr_damage_score = _detect_nr_damage(samples, sr) | |
| report.nr_damage_severity = _severity(report.nr_damage_score) | |
| L(f' NR damage={report.nr_damage_score:.3f} ({report.nr_damage_severity})') | |
| # Formant collapse | |
| report.formant_collapse = _detect_formant_collapse(samples, sr) | |
| L(f' Formant collapse={report.formant_collapse:.3f}') | |
| # Sibilant quality | |
| report.sibilant_quality = _detect_sibilant_quality(samples, sr) | |
| L(f' Sibilant quality={report.sibilant_quality:.3f}') | |
| # Mujawwad confidence | |
| report.mujawwad_conf = _detect_mujawwad(samples, f0_contour, sr) | |
| st.mujawwad_conf = report.mujawwad_conf | |
| L(f' Mujawwad confidence={report.mujawwad_conf:.3f}') | |
| # v2: SNR estimation | |
| report.noise_floor_db, report.snr_db, report.is_noisy = _estimate_snr(samples, sr) | |
| L(f' SNR={report.snr_db:.1f}dB noise_floor={report.noise_floor_db:.1f}dB is_noisy={report.is_noisy}') | |
| # v2: Active bandwidth detection | |
| report.active_bandwidth_hz = _detect_active_bandwidth(samples, sr) | |
| L(f' Active bandwidth={report.active_bandwidth_hz:.0f}Hz') | |
| # v2: Wind noise detection | |
| report.wind_detected = _detect_wind_noise(samples, sr) | |
| L(f' Wind noise={report.wind_detected}') | |
| # v2.2: TIER_TELEPHONE detection β must run AFTER active_bandwidth_hz is set | |
| report.is_telephone_tier, report.telephone_cutoff_hz = \ | |
| _detect_telephone_tier(samples, report.active_bandwidth_hz, sr) | |
| if report.is_telephone_tier: | |
| st.is_telephone_tier = True | |
| st.telephone_cutoff_hz = report.telephone_cutoff_hz | |
| L(f'') | |
| L(f' ββββββββββββββββββββββββββββββββββββββββββββββββββββββββ') | |
| L(f' β β TIER_TELEPHONE DETECTED β β') | |
| L(f' β Spectral cutoff: {report.telephone_cutoff_hz:.0f} Hz β') | |
| L(f' β Missing spectrum: {report.telephone_cutoff_hz:.0f}β22000 Hz ({100*(1-report.telephone_cutoff_hz/22050):.0f}% absent) β') | |
| L(f' β Standard enhancement passes: BYPASSED β') | |
| L(f' β Treatment: BWE-first pipeline Β§173 β') | |
| L(f' ββββββββββββββββββββββββββββββββββββββββββββββββββββββββ') | |
| L(f'') | |
| # Composite artificialness score | |
| report.overall_artificial = float(np.clip( | |
| report.f0_flatness * 0.25 + | |
| min(1.0, report.spectral_holes / 8.0) * 0.10 + | |
| report.metallic_score * 0.20 + | |
| report.nr_damage_score * 0.15 + | |
| report.formant_collapse * 0.15 + | |
| (1.0 - report.sibilant_quality) * 0.10 + | |
| (1.0 if report.is_noisy else 0.0) * 0.05, | |
| 0.0, 1.0 | |
| )) | |
| # Diagnosis | |
| parts = [] | |
| if report.f0_flat_severity != 'none': | |
| parts.append(f'F0 {report.f0_flat_severity} flatness={report.f0_flatness:.2f}') | |
| if report.metallic_severity != 'none': | |
| parts.append(f'metallic {report.metallic_severity}={report.metallic_score:.2f}') | |
| if report.nr_damage_severity != 'none': | |
| parts.append(f'NR damage {report.nr_damage_severity}={report.nr_damage_score:.2f}') | |
| if report.spectral_holes > 2: | |
| parts.append(f'{report.spectral_holes} spectral holes') | |
| if report.formant_collapse > 0.3: | |
| parts.append(f'formant collapse={report.formant_collapse:.2f}') | |
| if report.sibilant_quality < 0.7: | |
| parts.append(f'sibilant quality={report.sibilant_quality:.2f}') | |
| if report.is_noisy: | |
| parts.append(f'noisy (SNR={report.snr_db:.0f}dB)') | |
| if report.wind_detected: | |
| parts.append('wind noise detected') | |
| if report.active_bandwidth_hz < 16000: | |
| parts.append(f'limited bandwidth={report.active_bandwidth_hz:.0f}Hz') | |
| if report.is_telephone_tier: | |
| parts.append(f'TIER_TELEPHONE cutoff={report.telephone_cutoff_hz:.0f}Hz β BWE required') | |
| report.diagnosis = '; '.join(parts) if parts else 'no significant artifacts detected' | |
| L(f' ββ DIAGNOSIS: {report.diagnosis}') | |
| L(f' ββ Overall artificial: {report.overall_artificial:.3f}') | |
| st.artifacts = report | |
| st.f0_flatness_before = report.f0_flatness | |
| st.spectral_holes_before = report.spectral_holes | |
| st.metallic_before = report.metallic_score | |
| st.p0_analyzed = True | |
| return report | |
| # ββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ | |
| # Phase 1: DEREVERBERATION (lightweight Safaa-inspired) | |
| # ββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ | |
| def _rt60_estimate(samples: 'np.ndarray', sr: int = SR) -> float: | |
| """Schroeder backward integration RT60 estimate.""" | |
| if not NUMPY_OK or samples is None or len(samples) < sr * 3: | |
| return 0.0 | |
| fn = int(0.020 * sr) | |
| n = len(samples) // fn | |
| if n < 30: | |
| return 0.0 | |
| energy = np.array([float(np.mean(samples[i*fn:(i+1)*fn]**2)) for i in range(n)]) | |
| energy = np.maximum(energy, 1e-20) | |
| sch = np.cumsum(energy[::-1])[::-1] | |
| sch_db = 10 * np.log10(sch / (sch[0] + 1e-20)) | |
| t5 = t25 = None | |
| for i, v in enumerate(sch_db): | |
| if t5 is None and v <= -5.0: t5 = i * 0.020 | |
| if t25 is None and v <= -25.0: t25 = i * 0.020; break | |
| if t5 is not None and t25 is not None and t25 > t5: | |
| return float(np.clip((t25 - t5) * 3.0, 0.0, 6.0)) | |
| return 0.0 | |
| def _drr_estimate(samples: 'np.ndarray', sr: int = SR) -> float: | |
| """DRR: early vs late energy ratio.""" | |
| if not NUMPY_OK or samples is None: | |
| return 0.0 | |
| en = int(0.050 * sr) | |
| ln = int(0.450 * sr) | |
| step = int(0.200 * sr) | |
| vals = [] | |
| for s in range(0, len(samples) - en - ln, step): | |
| er = float(np.sqrt(np.mean(samples[s:s+en]**2)) + 1e-10) | |
| lr = float(np.sqrt(np.mean(samples[s+en:s+en+ln]**2)) + 1e-10) | |
| if er > 1e-5 and lr > 1e-5: | |
| vals.append(20.0 * np.log10(er / lr)) | |
| return float(np.median(vals)) if vals else 0.0 | |
| def phase1_dereverb(wav_path: str, samples: 'np.ndarray', | |
| st: IhyaState) -> str: | |
| """ | |
| Phase 1: Lightweight dereverberation. | |
| Only processes if RT60 > 0.5s (significant reverb). | |
| Uses ffmpeg-based EQ + afftdn for reverb reduction. | |
| """ | |
| _chk('phase_1_dereverb') | |
| rt60 = _rt60_estimate(samples) | |
| drr = _drr_estimate(samples) | |
| L(f' RT60={rt60:.2f}s DRR={drr:.1f}dB') | |
| # v2.1 FIX: RT60 on continuous tones can be wildly wrong (shows 4+ seconds | |
| # on a pure sine). If DRR is 0.0 (no early/late distinction), the signal | |
| # is probably not reverberant β it's continuous. Skip dereverb. | |
| if rt60 < 0.50: | |
| L(f' RT60 < 0.50s β no dereverb needed') | |
| return wav_path | |
| if drr == 0.0 and rt60 > 3.0: | |
| L(f' RT60={rt60:.2f}s but DRR=0 β likely continuous tone, not reverb. Skipping.') | |
| return wav_path | |
| # LF room mode EQ (Β§3.4: LF RT60 scaled by 1.3Γ) | |
| lf_rt60 = rt60 * 1.3 | |
| scale = float(np.clip(lf_rt60 / 0.5, 1.0, 4.0)) | |
| d_sub = float(np.clip(scale * 0.8, 0.8, 4.8)) | |
| d_lo = float(np.clip(scale * 0.5, 0.5, 3.0)) | |
| # Mujawwad: reduce depth (Β§145.3) | |
| if st.mujawwad_conf > 0.6: | |
| d_sub *= 0.5 | |
| d_lo *= 0.5 | |
| flt = [] | |
| if d_sub > 0.3: | |
| flt.append(f'equalizer=f=150:width_type=o:width=1.4:g=-{d_sub:.1f}') | |
| if d_lo > 0.3: | |
| flt.append(f'equalizer=f=300:width_type=o:width=1.2:g=-{d_lo:.1f}') | |
| # Gentle dereverb NR if RT60 > 1.0s | |
| if rt60 > 1.0: | |
| nr_depth = min(6, int(rt60 * 3)) | |
| flt.append(f'afftdn=nt=w:n={nr_depth}:t=6') | |
| if not flt: | |
| L(f' No filters needed') | |
| return wav_path | |
| out = _tmp_wav('p1_derev', st) | |
| rc, _, _ = _run_ffmpeg([ | |
| 'ffmpeg', '-y', '-i', wav_path, | |
| '-af', ','.join(flt), | |
| '-acodec', WAV_CODEC, '-ar', str(SR), '-ac', '1', | |
| '-loglevel', 'error', out | |
| ]) | |
| if rc != 0 or not os.path.exists(out): | |
| L(f' Dereverb failed β keeping original') | |
| _cleanup(out) | |
| return wav_path | |
| # Guard: RMS delta check | |
| post, _ = _decode_samples(out) | |
| if post is not None and samples is not None: | |
| delta = _rmsdb(post) - _rmsdb(samples) | |
| guard_thresh = 4.5 if st.aggressive else 3.0 | |
| if abs(delta) > guard_thresh: | |
| L(f' RMS delta={delta:+.2f}dB β REVERT') | |
| _cleanup(out) | |
| st.guard_reverts += 1 | |
| return wav_path | |
| st.p1_dereverb_applied = True | |
| L(f' β dereverb applied: rt60={rt60:.2f}s lf_cuts=[{d_sub:.1f},{d_lo:.1f}]dB') | |
| return out | |
| # ββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ | |
| # Phase 2: SELECTIVE NOISE REPAIR (v2: + wind HPF + SNR-gated NR) | |
| # ββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ | |
| def phase2_noise_repair(wav_path: str, samples: 'np.ndarray', | |
| artifacts: ArtifactReport, st: IhyaState) -> str: | |
| """ | |
| Phase 2: Repair NR damage and spectral musical noise. | |
| Targeted: only processes if NR damage detected in Phase 0. | |
| v2: Also handles wind noise (HPF below 200Hz) and uses SNR | |
| to gate NR depth more intelligently. | |
| """ | |
| _chk('phase_2_noise_repair') | |
| filters = [] | |
| # v2: Wind noise HPF β apply if wind detected | |
| if artifacts.wind_detected: | |
| hpf_freq = 80 if st.aggressive else 100 | |
| filters.append(f'highpass=f={hpf_freq}:p=2') | |
| L(f' Wind noise detected β applying HPF at {hpf_freq}Hz') | |
| # Musical noise reduction: only if NR damage detected | |
| if artifacts.nr_damage_severity != 'none': | |
| # v2: Scale NR strength by SNR β noisier recordings need gentler NR | |
| # to avoid over-cleaning; cleaner recordings can use stronger NR | |
| snr_factor = 1.0 | |
| if artifacts.is_noisy: | |
| # Low SNR: be more conservative with NR | |
| snr_factor = max(0.4, artifacts.snr_db / SNR_NOISY_THRESH) | |
| L(f' Low SNR ({artifacts.snr_db:.0f}dB) β reducing NR strength to {snr_factor:.1f}Γ') | |
| strength_map = {'mild': 3, 'moderate': 5, 'severe': 7} | |
| base_strength = strength_map.get(artifacts.nr_damage_severity, 3) | |
| strength = max(1, int(base_strength * snr_factor)) | |
| if st.aggressive: | |
| strength = min(10, strength + 2) | |
| # Also add gentle afftdn for broadband noise floor | |
| nr_depth = min(4, strength) | |
| filters.append(f'anlmdn=s={strength}:p=3:r=7:m=2') | |
| filters.append(f'afftdn=nt=w:n={nr_depth}:t=6') | |
| L(f' NR damage {artifacts.nr_damage_severity}: anlmdn s={strength} afftdn n={nr_depth}') | |
| else: | |
| L(f' NR damage = none β skipping musical noise NR') | |
| if not filters: | |
| return wav_path | |
| out = _tmp_wav('p2_nrrepair', st) | |
| # v2.1 FIX: Try afftdn alone first (more reliable than anlmdn). | |
| # If that works and passes guards, done. If anlmdn is also available, | |
| # try the combined chain as a second attempt. | |
| rc, _, err = _run_ffmpeg([ | |
| 'ffmpeg', '-y', '-i', wav_path, | |
| '-af', ','.join(filters), | |
| '-acodec', WAV_CODEC, '-ar', str(SR), '-ac', '1', | |
| '-loglevel', 'error', out | |
| ]) | |
| if rc != 0 or not os.path.exists(out): | |
| # First attempt failed β try afftdn-only as fallback | |
| L(f' Combined NR failed β trying afftdn-only fallback') | |
| _cleanup(out) | |
| fallback_filters = [f'afftdn=nt=w:n={nr_depth}:t=6'] | |
| if artifacts.wind_detected: | |
| hpf_freq = 80 if st.aggressive else 100 | |
| fallback_filters.insert(0, f'highpass=f={hpf_freq}:p=2') | |
| rc, _, err2 = _run_ffmpeg([ | |
| 'ffmpeg', '-y', '-i', wav_path, | |
| '-af', ','.join(fallback_filters), | |
| '-acodec', WAV_CODEC, '-ar', str(SR), '-ac', '1', | |
| '-loglevel', 'error', out | |
| ]) | |
| if rc != 0 or not os.path.exists(out): | |
| L(f' NR repair failed entirely β keeping original') | |
| _cleanup(out) | |
| return wav_path | |
| # Guard: check that we didn't over-clean | |
| post, _ = _decode_samples(out) | |
| if post is not None and samples is not None: | |
| # RMS should not drop more than 2dB (over-NR), 3dB in aggressive | |
| rms_limit = 3.0 if st.aggressive else 2.0 | |
| delta = _rmsdb(post) - _rmsdb(samples) | |
| if delta < -rms_limit: | |
| L(f' RMS delta={delta:+.2f}dB β over-NR β REVERT') | |
| _cleanup(out) | |
| st.guard_reverts += 1 | |
| return wav_path | |
| # Sibilant band should not drop more than 4dB (Β§152) | |
| sib_before = _band_energy_db(samples, 5500, 12000) | |
| sib_after = _band_energy_db(post, 5500, 12000) | |
| sib_limit = 5.0 if st.aggressive else 4.0 | |
| if sib_before > -80 and (sib_after - sib_before) < -sib_limit: | |
| L(f' Sibilant band drop={sib_after-sib_before:+.1f}dB β REVERT') | |
| _cleanup(out) | |
| st.guard_reverts += 1 | |
| return wav_path | |
| st.p2_nr_repair_applied = True | |
| L(f' β NR repair applied') | |
| return out | |
| # ββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ | |
| # Phase 3: F0 CONTOUR REVIVAL β the heart of Ψ§ΩΨ₯ΨΩΨ§Ψ‘ | |
| # ββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ | |
| def phase3_f0_revival(wav_path: str, samples: 'np.ndarray', | |
| artifacts: ArtifactReport, st: IhyaState) -> str: | |
| """ | |
| Phase 3: Revive the F0 contour β inject natural micro-pitch variation | |
| into flat/robotic voice. This is the core of Ψ§ΩΨ₯ΨΩΨ§Ψ‘. | |
| Strategy: | |
| - Detect F0 contour | |
| - If flatness > threshold, inject natural jitter + shimmer | |
| - Jitter: cycle-to-cycle period variation (Β§133.3) | |
| - Shimmer: cycle-to-cycle amplitude variation (Β§133.3) | |
| - Slow drift: phrase-level F0 contour variation (Β§85) | |
| - All changes are extremely subtle β compound naturalness | |
| Implementation: Uses ffmpeg chorus filter with very subtle settings | |
| to add micro-pitch variation that mimics natural vocal jitter. | |
| """ | |
| _chk('phase_3_f0_revival') | |
| if not NUMPY_OK or samples is None: | |
| L(f' No numpy β skipping F0 revival') | |
| return wav_path | |
| if artifacts.f0_flat_severity == 'none': | |
| L(f' F0 flatness = none β skipping revival') | |
| return wav_path | |
| f0_contour, f0_median = _detect_f0(samples) | |
| if f0_median < 60 or len(f0_contour) < 20: | |
| L(f' F0 detection insufficient β skipping') | |
| return wav_path | |
| L(f' F0 median={f0_median:.1f}Hz flatness={artifacts.f0_flatness:.3f}') | |
| # ββ 3a: Natural Jitter Injection ββββββββββββββββββββββββββββββββββββββββββ | |
| # Jitter = small cycle-to-cycle pitch variations. | |
| # Natural speech: ~0.3ms jitter (Β§133.3) | |
| # Robotic: ~0.0ms jitter | |
| # We add jitter by pitch-shifting tiny segments by random micro-amounts. | |
| if artifacts.f0_flatness > F0_FLATNESS_THRESHOLD * 0.6: | |
| L(f' [3a] Injecting natural F0 jitter...') | |
| wav_path = _inject_f0_jitter(wav_path, samples, f0_median, | |
| artifacts.f0_flatness, st) | |
| # ββ 3b: Natural Shimmer Injection βββββββββββββββββββββββββββββββββββββββββ | |
| # Shimmer = small cycle-to-cycle amplitude variations. | |
| # Natural: ~0.2dB shimmer (Β§133.3) | |
| if artifacts.f0_flatness > F0_FLATNESS_THRESHOLD * 0.5: | |
| L(f' [3b] Injecting natural shimmer...') | |
| wav_path = _inject_shimmer(wav_path, st) | |
| # ββ 3c: Phrase-level F0 Drift βββββββββββββββββββββββββββββββββββββββββββββ | |
| # Natural speech drifts slowly over phrases (Β§85, Β§133). | |
| # Murattal: ~12 cents drift; Mujawwad: ~35 cents (Β§145). | |
| if artifacts.f0_flatness > F0_FLATNESS_THRESHOLD * 0.7: | |
| L(f' [3c] Injecting phrase-level F0 drift...') | |
| wav_path = _inject_f0_drift(wav_path, samples, f0_median, st) | |
| # Measure improvement | |
| post, _ = _decode_samples(wav_path) | |
| if post is not None: | |
| new_f0_contour, _ = _detect_f0(post) | |
| if len(new_f0_contour) > 10: | |
| st.f0_flatness_after = _compute_f0_flatness(new_f0_contour) | |
| improvement = st.f0_flatness_before - st.f0_flatness_after | |
| L(f' F0 flatness: {st.f0_flatness_before:.3f} β {st.f0_flatness_after:.3f} ' | |
| f'(Ξ={improvement:+.3f})') | |
| st.p3_f0_revival = True | |
| return wav_path | |
| def _inject_f0_jitter(wav_path: str, samples: 'np.ndarray', f0_median: float, | |
| flatness: float, st: IhyaState) -> str: | |
| """ | |
| Inject natural pitch jitter using ffmpeg's chorus filter. | |
| The chorus filter adds subtle pitch variation via short modulated delays. | |
| v2 FIX: Chorus filter format is chorus=input_gain:output_gain: | |
| delays:decays:speeds:depths β each pipe-list must have the | |
| same number of entries. Fixed parameter labeling and values. | |
| """ | |
| sr = SR | |
| # Scale jitter by flatness severity | |
| # mild: 0.15ms, moderate: 0.30ms, severe: 0.45ms | |
| jitter_ms = F0_JITTER_MS * (flatness / F0_FLATNESS_THRESHOLD) | |
| jitter_ms = min(jitter_ms, 0.60) # cap at 0.6ms β never artificial | |
| # Chorus filter: chorus=input_gain:output_gain:delays:decays:speeds:depths | |
| # delays = milliseconds, decays = 0-1, speeds = Hz, depths = milliseconds | |
| # For natural jitter: short delays, moderate decays, very slow speeds, minimal depths | |
| depth_ms = min(1.5, jitter_ms * 3.0) # depth in milliseconds (not percentage) | |
| if st.aggressive: | |
| depth_ms *= 1.3 | |
| # Multiple micro-chorus voices for natural variation | |
| # Each voice at slightly different rate for organic feel | |
| v1_delay = 0.5 + np.random.uniform(0, 0.3) | |
| v2_delay = 1.2 + np.random.uniform(0, 0.5) | |
| v3_delay = 2.0 + np.random.uniform(0, 0.8) | |
| v1_speed = 0.15 + np.random.uniform(0, 0.05) | |
| v2_speed = 0.08 + np.random.uniform(0, 0.03) | |
| v3_speed = 0.22 + np.random.uniform(0, 0.06) | |
| chorus_str = ( | |
| f'chorus=0.5:0.9:' | |
| f'50|60|40:' # delays in ms (3 voices) | |
| f'0.4|0.3|0.3:' # decays (0-1 range, 3 voices) | |
| f'{v1_speed:.2f}|{v2_speed:.2f}|{v3_speed:.2f}:' # speeds in Hz (3 voices) | |
| f'{depth_ms:.1f}|{depth_ms*0.7:.1f}|{depth_ms*0.5:.1f}' # depths in ms (3 voices) | |
| ) | |
| out = _tmp_wav('p3a_jitter', st) | |
| rc, _, _ = _run_ffmpeg([ | |
| 'ffmpeg', '-y', '-i', wav_path, | |
| '-af', chorus_str, | |
| '-acodec', WAV_CODEC, '-ar', str(sr), '-ac', '1', | |
| '-loglevel', 'error', out | |
| ]) | |
| if rc != 0 or not os.path.exists(out): | |
| L(f' Jitter injection failed β keeping original') | |
| _cleanup(out) | |
| return wav_path | |
| # Guard: verify we didn't alter overall spectral character | |
| # v2 FIX: Chorus naturally adds energy (comb filtering + multiple voices). | |
| # The 1.5dB guard was too strict β changed to 3.0dB. | |
| post, _ = _decode_samples(out) | |
| if post is not None and samples is not None: | |
| delta = _rmsdb(post) - _rmsdb(samples) | |
| jitter_rms_limit = 4.5 if st.aggressive else 3.0 # v2: was 1.5, now 3.0 | |
| if abs(delta) > jitter_rms_limit: | |
| L(f' Jitter RMS delta={delta:+.2f}dB β REVERT (limit={jitter_rms_limit})') | |
| _cleanup(out) | |
| st.guard_reverts += 1 | |
| return wav_path | |
| # Check formant bands are preserved | |
| # v2.1 FIX: Chorus naturally reshapes harmonic balance which shifts | |
| # per-band energy. Use a wider tolerance β the IMPORTANT thing is that | |
| # the overall spectral shape is preserved, not that each band is identical. | |
| # Check F1 AND F2 together; only revert if BOTH shift significantly. | |
| f1_before = _band_energy_db(samples, 250, 800) | |
| f1_after = _band_energy_db(post, 250, 800) | |
| f2_before = _band_energy_db(samples, 800, 2500) | |
| f2_after = _band_energy_db(post, 800, 2500) | |
| f1_limit = 4.0 if st.aggressive else 3.0 | |
| f2_limit = 4.0 if st.aggressive else 3.0 | |
| f1_shift = abs(f1_after - f1_before) if f1_before > -80 else 0 | |
| f2_shift = abs(f2_after - f2_before) if f2_before > -80 else 0 | |
| if f1_shift > f1_limit and f2_shift > f2_limit: | |
| L(f' Formant shift: F1 Ξ={f1_after-f1_before:+.1f}dB F2 Ξ={f2_after-f2_before:+.1f}dB β REVERT') | |
| _cleanup(out) | |
| st.guard_reverts += 1 | |
| return wav_path | |
| st.p3_jitter_injected = True | |
| L(f' β jitter injected: {jitter_ms:.2f}ms depth={depth_ms:.1f}ms') | |
| return out | |
| def _inject_shimmer(wav_path: str, st: IhyaState) -> str: | |
| """ | |
| Inject natural amplitude shimmer. | |
| Uses gentle tremolo at very low depth + slow rate. | |
| """ | |
| # Shimmer: very gentle amplitude modulation | |
| # Natural: 0.2dB variation at ~4-8Hz (Β§133.3) | |
| # Use tremolo with extremely low depth | |
| depth = min(0.15, F0_SHIMMER_DB * 0.1) # ffmpeg tremolo depth is 0-1 | |
| rate = 5.5 + np.random.uniform(-1.5, 1.5) # Hz: natural shimmer rate | |
| if st.aggressive: | |
| depth = min(0.25, depth * 1.5) | |
| out = _tmp_wav('p3b_shimmer', st) | |
| rc, _, _ = _run_ffmpeg([ | |
| 'ffmpeg', '-y', '-i', wav_path, | |
| '-af', f'tremolo=f={rate:.1f}:d={depth:.3f}', | |
| '-acodec', WAV_CODEC, '-ar', str(SR), '-ac', '1', | |
| '-loglevel', 'error', out | |
| ]) | |
| if rc != 0 or not os.path.exists(out): | |
| _cleanup(out) | |
| return wav_path | |
| st.p3_shimmer_injected = True | |
| L(f' β shimmer injected: rate={rate:.1f}Hz depth={depth:.3f}') | |
| return out | |
| def _inject_f0_drift(wav_path: str, samples: 'np.ndarray', | |
| f0_median: float, st: IhyaState) -> str: | |
| """ | |
| Inject slow phrase-level F0 drift using subtle pitch shifting. | |
| Uses ffmpeg's chorus filter with very slow modulation for drift. | |
| v2 FIX: Chorus parameter format fixed: delays:decays:speeds:depths. | |
| RMS guard changed from 1.0dB to 2.5dB (drift chorus naturally adds energy). | |
| """ | |
| sr = SR | |
| # Drift amount depends on recitation style | |
| drift_cents = MUJAWWAD_DRIFT if st.mujawwad_conf > 0.6 else F0_DRIFT_CENTS | |
| if st.aggressive: | |
| drift_cents *= 1.3 | |
| # Generate a slow drift contour using chorus at very slow speed + larger depth | |
| duration_s = len(samples) / sr | |
| if duration_s < 2.0: | |
| return wav_path | |
| # Chorus filter: chorus=input_gain:output_gain:delays:decays:speeds:depths | |
| # For drift: long delays, moderate decays, very slow speeds, subtle depths | |
| drift_depth_ms = min(2.5, drift_cents / 12.0) # depth in ms | |
| # Slow drift chorus: long delay, very slow modulation, subtle depth | |
| drift_filter = ( | |
| f'chorus=0.7:0.9:' | |
| f'55|65|45:' # delays in ms (3 voices) | |
| f'0.5|0.4|0.3:' # decays (0-1 range, 3 voices) | |
| f'0.05|0.03|0.07:' # speeds in Hz (very slow, 3 voices) | |
| f'{drift_depth_ms:.1f}|{drift_depth_ms*0.6:.1f}|{drift_depth_ms*0.4:.1f}' # depths in ms (3 voices) | |
| ) | |
| out = _tmp_wav('p3c_drift', st) | |
| rc, _, _ = _run_ffmpeg([ | |
| 'ffmpeg', '-y', '-i', wav_path, | |
| '-af', drift_filter, | |
| '-acodec', WAV_CODEC, '-ar', str(sr), '-ac', '1', | |
| '-loglevel', 'error', out | |
| ]) | |
| if rc != 0 or not os.path.exists(out): | |
| L(f' F0 drift injection failed β keeping original') | |
| _cleanup(out) | |
| return wav_path | |
| # Guard: spectral sanity | |
| # v2.1 FIX: Drift chorus adds significant energy naturally. | |
| # Allow 3.5dB (was 2.5dB which caused too many false reverts). | |
| post, _ = _decode_samples(out) | |
| if post is not None and samples is not None: | |
| delta = _rmsdb(post) - _rmsdb(samples) | |
| drift_rms_limit = 5.0 if st.aggressive else 3.5 # v2.1: was 2.5, now 3.5 | |
| if abs(delta) > drift_rms_limit: | |
| L(f' Drift RMS delta={delta:+.2f}dB β REVERT (limit={drift_rms_limit})') | |
| _cleanup(out) | |
| st.guard_reverts += 1 | |
| return wav_path | |
| L(f' β F0 drift injected: {drift_cents:.0f} cents (mujawwad={st.mujawwad_conf:.2f})') | |
| return out | |
| # ββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ | |
| # Phase 4: FORMANT RECONSTRUCTION | |
| # ββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ | |
| def phase4_formant_rebuild(wav_path: str, samples: 'np.ndarray', | |
| artifacts: ArtifactReport, st: IhyaState) -> str: | |
| """ | |
| Phase 4: Rebuild collapsed/missing formants. | |
| When codec or NR destroys formant structure, voice sounds hollow/thin. | |
| We boost energy in F1/F2/F3 zones proportionally using the | |
| FORMANT_MALE_ARABIC reference template for center frequencies. | |
| v2: Docstring updated β this uses EQ-based formant boosting from the | |
| Arabic reference template, not LPC analysis (LPC was never implemented). | |
| """ | |
| _chk('phase_4_formant_rebuild') | |
| if artifacts.formant_collapse < 0.2: | |
| L(f' Formant collapse={artifacts.formant_collapse:.3f} < 0.2 β skipping') | |
| return wav_path | |
| # Calculate needed boosts based on collapse severity | |
| # v2: Use FORMANT_MALE_ARABIC reference for center frequencies | |
| f1_center = FORMANT_MALE_ARABIC['F1']['typical'] | |
| f2_center = FORMANT_MALE_ARABIC['F2']['typical'] | |
| f3_center = FORMANT_MALE_ARABIC['F3']['typical'] | |
| f1_boost = min(3.0, artifacts.formant_collapse * 4.0) # F1: warmth | |
| f2_boost = min(4.0, artifacts.formant_collapse * 5.5) # F2: clarity | |
| f3_boost = min(2.5, artifacts.formant_collapse * 3.0) # F3: presence | |
| # Scale by Mujawwad confidence β Mujawwad needs F2/F3 more (Β§145) | |
| if st.mujawwad_conf > 0.6: | |
| f2_boost *= 1.2 | |
| f3_boost *= 1.1 | |
| if st.aggressive: | |
| f1_boost = min(4.5, f1_boost * 1.3) | |
| f2_boost = min(6.0, f2_boost * 1.3) | |
| f3_boost = min(4.0, f3_boost * 1.3) | |
| filters = [] | |
| # F1 boost: warmth and body (using Arabic formant reference) | |
| if f1_boost > 0.3: | |
| filters.append(f'equalizer=f={f1_center}:width_type=o:width=1.0:g={f1_boost:.1f}') | |
| # F2 boost: vowel clarity (using Arabic formant reference) | |
| if f2_boost > 0.3: | |
| filters.append(f'equalizer=f={f2_center}:width_type=o:width=1.0:g={f2_boost:.1f}') | |
| # F3 boost: presence and definition (using Arabic formant reference) | |
| if f3_boost > 0.3: | |
| filters.append(f'equalizer=f={f3_center}:width_type=o:width=0.8:g={f3_boost:.1f}') | |
| if not filters: | |
| return wav_path | |
| out = _tmp_wav('p4_formant', st) | |
| rc, _, _ = _run_ffmpeg([ | |
| 'ffmpeg', '-y', '-i', wav_path, | |
| '-af', ','.join(filters), | |
| '-acodec', WAV_CODEC, '-ar', str(SR), '-ac', '1', | |
| '-loglevel', 'error', out | |
| ]) | |
| if rc != 0 or not os.path.exists(out): | |
| _cleanup(out) | |
| return wav_path | |
| # Guard: check formant ratio improved | |
| post, _ = _decode_samples(out) | |
| if post is not None: | |
| # v2.1 FIX: Re-detect formant collapse from CURRENT state (not original). | |
| # Previous phases (jitter/shimmer) may have changed spectral balance. | |
| current_collapse = _detect_formant_collapse(post) | |
| if current_collapse > _detect_formant_collapse(samples) + 0.1: | |
| L(f' Formant collapse worsened: was {_detect_formant_collapse(samples):.3f}β{current_collapse:.3f} β REVERT') | |
| _cleanup(out) | |
| st.guard_reverts += 1 | |
| return wav_path | |
| # Sibilant guard: F3 boost shouldn't harshen sibilants | |
| sib_before = _band_energy_db(samples, 5500, 12000) | |
| sib_after = _band_energy_db(post, 5500, 12000) | |
| sib_limit = 4.5 if st.aggressive else 3.0 | |
| if sib_before > -80 and (sib_after - sib_before) > sib_limit: | |
| L(f' Sibilant harshness: {sib_after-sib_before:+.1f}dB β REVERT') | |
| _cleanup(out) | |
| st.guard_reverts += 1 | |
| return wav_path | |
| st.p4_formant_rebuilt = True | |
| L(f' β formant rebuilt: F1({f1_center}Hz)=+{f1_boost:.1f} F2({f2_center}Hz)=+{f2_boost:.1f} F3({f3_center}Hz)=+{f3_boost:.1f}dB') | |
| return out | |
| # ββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ | |
| # Phase 5: HARMONIC RICHNESS RESTORATION | |
| # ββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ | |
| def phase5_harmonic_richness(wav_path: str, samples: 'np.ndarray', | |
| artifacts: ArtifactReport, st: IhyaState) -> str: | |
| """ | |
| Phase 5: Restore harmonic richness to thin/artificial voice. | |
| Uses aexciter on voiced frames to add natural harmonics. | |
| Only adds harmonics at F0-derived positions (Β§80, Β§152). | |
| v2: Improved fallback when aexciter isn't available: | |
| - Try crystalizer first (expands HF dynamics) | |
| - Then try anoisesrc mixing at -60dB for harmonic texture | |
| - Only then fall back to shelf EQ | |
| """ | |
| _chk('phase_5_harmonic_richness') | |
| if artifacts.formant_collapse < 0.15 and artifacts.f0_flatness < 0.5: | |
| L(f' Harmonics OK β skipping') | |
| return wav_path | |
| f0 = st.f0_median if st.f0_median > 60 else 150.0 | |
| # aexciter: adds harmonics above a frequency | |
| # Set the crossover at 1.5Γ F0 (below voice harmonics region) | |
| # For male reciter: F0 ~120-250Hz, harmonics start ~300Hz | |
| freq = min(max(f0 * 1.5, 200.0), 3000.0) | |
| # v2: Don't boost above active bandwidth | |
| if artifacts.active_bandwidth_hz < freq + 1000: | |
| freq = max(f0 * 1.2, artifacts.active_bandwidth_hz * 0.5) | |
| # Strength scales with artificialness | |
| strength = min(8.0, artifacts.overall_artificial * 12.0) | |
| if st.mujawwad_conf > 0.6: | |
| strength *= 0.7 # Β§145: gentler on Mujawwad (ornaments can amplify) | |
| if st.aggressive: | |
| strength *= 1.3 | |
| if strength < 1.0: | |
| L(f' Harmonic strength too low ({strength:.1f}) β skipping') | |
| return wav_path | |
| # aexciter: freq=crossover, mix=strength percentage | |
| mix = min(4.0, strength * 0.4) # mix level (0-10 range, keep low) | |
| out = _tmp_wav('p5_harmonic', st) | |
| rc, _, _ = _run_ffmpeg([ | |
| 'ffmpeg', '-y', '-i', wav_path, | |
| '-af', f'aexciter=freq={freq:.0f}:mix={mix:.1f}', | |
| '-acodec', WAV_CODEC, '-ar', str(SR), '-ac', '1', | |
| '-loglevel', 'error', out | |
| ]) | |
| if rc != 0 or not os.path.exists(out): | |
| _cleanup(out) | |
| # aexciter might not be available β try crystalizer fallback | |
| L(f' aexciter failed β trying crystalizer fallback') | |
| rc, _, _ = _run_ffmpeg([ | |
| 'ffmpeg', '-y', '-i', wav_path, | |
| '-af', f'crystalizer=i=2.0:c=1.0', | |
| '-acodec', WAV_CODEC, '-ar', str(SR), '-ac', '1', | |
| '-loglevel', 'error', out | |
| ]) | |
| if rc != 0 or not os.path.exists(out): | |
| _cleanup(out) | |
| # v2.1 FIX: Reduced noise exciter level and narrower bandpass | |
| # Previous -65dB was still too audible and was damaging Tajweed features. | |
| # Now: -75dB (nearly inaudible), narrower bandpass (w=1000) | |
| duration_s = len(samples) / SR if samples is not None else 10.0 | |
| noise_mix_db = max(-78.0, -75.0 + strength * 0.3) # -75 to -72 dB range | |
| rc, _, _ = _run_ffmpeg([ | |
| 'ffmpeg', '-y', '-i', wav_path, | |
| '-f', 'lavfi', '-i', f'anoisesrc=d={duration_s:.1f}:c=pink:r={SR}:a=0.001', | |
| '-filter_complex', | |
| f'[1:a]bandpass=f={freq:.0f}:w=1000,volume={noise_mix_db:.0f}dB[noise];' | |
| f'[0:a][noise]amix=inputs=2:duration=first:dropout_transition=0[out]', | |
| '-map', '[out]', | |
| '-acodec', WAV_CODEC, '-ar', str(SR), '-ac', '1', | |
| '-loglevel', 'error', out | |
| ]) | |
| if rc != 0 or not os.path.exists(out): | |
| _cleanup(out) | |
| # Last resort: shelf EQ fallback (doesn't add harmonics but boosts existing HF) | |
| L(f' anoisesrc failed β using shelf EQ fallback (weaker)') | |
| shelf_gain = min(2.0, strength * 0.3) | |
| rc, _, _ = _run_ffmpeg([ | |
| 'ffmpeg', '-y', '-i', wav_path, | |
| '-af', f'highshelf=f={freq:.0f}:width_type=s:width=0.5:g={shelf_gain:.1f}', | |
| '-acodec', WAV_CODEC, '-ar', str(SR), '-ac', '1', | |
| '-loglevel', 'error', out | |
| ]) | |
| if rc != 0 or not os.path.exists(out): | |
| _cleanup(out) | |
| return wav_path | |
| # Guard: THD check β harmonic excitation shouldn't add excessive distortion | |
| post, _ = _decode_samples(out) | |
| if post is not None and samples is not None: | |
| # Quick THD check on center portion | |
| c_start = int(len(post) * 0.2) | |
| c_end = int(len(post) * 0.8) | |
| c = post[c_start:c_end] | |
| N = min(len(c), SR * 2) | |
| if N > SR // 4: | |
| spec = np.abs(rfft(c[:N] * np.hanning(N))) ** 2 | |
| freqs = rfftfreq(N, d=1.0/SR) | |
| fund_mask = (freqs >= f0 * 0.8) & (freqs <= f0 * 1.2) | |
| h2_mask = (freqs >= f0 * 1.8) & (freqs <= f0 * 2.2) | |
| if fund_mask.any() and h2_mask.any(): | |
| fund_power = float(np.mean(spec[fund_mask])) | |
| h2_power = float(np.mean(spec[h2_mask])) | |
| if fund_power > 1e-20: | |
| h2_ratio = h2_power / fund_power | |
| # v2.1 FIX: We're TRYING to add harmonics β the H2 limit should be | |
| # generous. Natural male voice H2/H1 is 0.1-0.5. The guard is only | |
| # to prevent extreme distortion (>0.6 is genuinely harsh). | |
| h2_limit = 0.60 if st.aggressive else 0.50 | |
| if h2_ratio > h2_limit: # too much harmonic distortion | |
| L(f' H2 ratio={h2_ratio:.3f} > {h2_limit} β REVERT') | |
| _cleanup(out) | |
| st.guard_reverts += 1 | |
| return wav_path | |
| st.p5_harmonic_rich = True | |
| L(f' β harmonic richness: freq={freq:.0f}Hz mix={mix:.1f} f0={f0:.0f}Hz') | |
| return out | |
| # ββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ | |
| # Phase 6: MICRO-DYNAMICS BREATHING | |
| # ββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ | |
| def phase6_micro_dynamics(wav_path: str, samples: 'np.ndarray', | |
| st: IhyaState) -> str: | |
| """ | |
| Phase 6: Restore natural micro-dynamics. | |
| Robotic/over-processed audio has compressed dynamics. | |
| We gently expand the dynamic range to restore natural breathing. | |
| Β§85, Β§133: natural recitation has LRA within PHRASE_LRA_MIN to PHRASE_LRA_MAX. | |
| Uses BREATH_DEPTH_DB for modulation depth. | |
| """ | |
| _chk('phase_6_micro_dynamics') | |
| if not NUMPY_OK or samples is None: | |
| return wav_path | |
| # Measure current dynamics | |
| lufs, lra = _measure_lufs(wav_path) | |
| L(f' Current LRA={lra:.2f} target range=[{PHRASE_LRA_MIN},{PHRASE_LRA_MAX}]') | |
| # Target LRA: midpoint of the natural range | |
| target_lra = (PHRASE_LRA_MIN + PHRASE_LRA_MAX) / 2.0 | |
| lra_deficit = target_lra - lra | |
| if lra_deficit < 0.5: | |
| L(f' LRA deficit={lra_deficit:.2f} < 0.5 β no expansion needed') | |
| return wav_path | |
| # Use aexpander to gently expand dynamics | |
| # Threshold at p30 of voiced RMS | |
| frame_n = int(0.020 * SR) | |
| frame_rms = [] | |
| for i in range(0, len(samples) - frame_n, frame_n): | |
| e = float(np.sqrt(np.mean(samples[i:i+frame_n]**2))) | |
| if e > 1e-7: | |
| frame_rms.append(e) | |
| if not frame_rms: | |
| return wav_path | |
| threshold = float(np.percentile(frame_rms, 30)) | |
| ratio = min(1.0 + lra_deficit * 0.12, 1.6) # gentle, capped | |
| if st.aggressive: | |
| ratio = min(2.0, ratio * 1.2) | |
| out = _tmp_wav('p6_dynamics', st) | |
| rc, _, _ = _run_ffmpeg([ | |
| 'ffmpeg', '-y', '-i', wav_path, | |
| '-af', (f'aexpander=threshold={max(0.001,threshold):.5f}' | |
| f':ratio={ratio:.2f}:attack=3:release=150'), | |
| '-acodec', WAV_CODEC, '-ar', str(SR), '-ac', '1', | |
| '-loglevel', 'error', out | |
| ]) | |
| if rc != 0 or not os.path.exists(out): | |
| _cleanup(out) | |
| return wav_path | |
| # Guard: LRA must improve but not overshoot | |
| _, lra_after = _measure_lufs(out) | |
| lra_max = PHRASE_LRA_MAX + (1.5 if st.aggressive else 1.0) | |
| if lra_after > lra and lra_after <= lra_max: | |
| st.p6_micro_dynamics = True | |
| L(f' β micro-dynamics: LRA {lra:.2f}β{lra_after:.2f} ratio={ratio:.2f}') | |
| return out | |
| else: | |
| L(f' LRA guard failed: {lra:.2f}β{lra_after:.2f} β REVERT') | |
| _cleanup(out) | |
| st.guard_reverts += 1 | |
| return wav_path | |
| # ββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ | |
| # Phase 7: SPECTRAL CONTINUITY REPAIR | |
| # ββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ | |
| def phase7_spectral_repair(wav_path: str, samples: 'np.ndarray', | |
| artifacts: ArtifactReport, st: IhyaState) -> str: | |
| """ | |
| Phase 7: Repair spectral discontinuities from codec artifacts and NR. | |
| Smooths spectral holes and sharp edges using gentle EQ interpolation. | |
| v2: Tracks spectral_holes_after even if phase doesn't run. | |
| Respects active bandwidth β no boosting above detected bandwidth. | |
| """ | |
| _chk('phase_7_spectral_repair') | |
| if artifacts.spectral_holes < 2 and artifacts.spectral_edges < 2: | |
| L(f' Spectral continuity OK β skipping') | |
| # v2 FIX: Track spectral_holes_after even if we don't process | |
| post_check, _ = _decode_samples(wav_path) | |
| if post_check is not None: | |
| st.spectral_holes_after, _ = _detect_spectral_holes(post_check) | |
| return wav_path | |
| # Strategy: apply gentle spectral smoothing using equalizer banks | |
| # Fill holes: boost the dropped bands | |
| # Smooth edges: apply gentle Q bridges between adjacent bands | |
| filters = [] | |
| # Get current spectral profile | |
| band_db = {} | |
| for f in CENTERS_48: | |
| if f < 80 or f > 16000: | |
| continue | |
| # v2: Don't process bands above active bandwidth | |
| if f > artifacts.active_bandwidth_hz: | |
| continue | |
| lo = f / (2 ** (1.0/12)) | |
| hi = f * (2 ** (1.0/12)) | |
| band_db[f] = _band_energy_db(samples, lo, hi) | |
| # Detect and fill holes | |
| sorted_f = sorted(band_db.keys()) | |
| for i in range(1, len(sorted_f) - 1): | |
| prev_db = band_db[sorted_f[i-1]] | |
| curr_db = band_db[sorted_f[i]] | |
| next_db = band_db[sorted_f[i+1]] | |
| neighbor_avg = (prev_db + next_db) / 2.0 | |
| dip = curr_db - neighbor_avg | |
| if dip < -abs(SPEC_HOLE_DEPTH_DB): | |
| # Boost this band to fill the hole β but only partially | |
| # Full fill would sound unnatural; fill 60% of the dip | |
| fill_frac = 0.75 if st.aggressive else 0.6 | |
| boost = min(4.0 if not st.aggressive else 6.0, abs(dip) * fill_frac) | |
| q = 8.65 # sixth-octave | |
| filters.append(f'equalizer=f={sorted_f[i]:.0f}:width_type=q:width={q}:g={boost:.1f}') | |
| # Smooth sharp edges | |
| for i in range(1, len(sorted_f)): | |
| delta = abs(band_db[sorted_f[i]] - band_db[sorted_f[i-1]]) | |
| if delta > SPEC_EDGE_SHARP_DB: | |
| # Apply a gentle EQ to smooth the transition | |
| mid_f = np.sqrt(sorted_f[i] * sorted_f[i-1]) | |
| # Gentle cut at the peak side or boost at the valley side | |
| if band_db[sorted_f[i]] > band_db[sorted_f[i-1]]: | |
| # Sharp rise β gentle cut at upper band | |
| cut = min(2.0, (delta - SPEC_EDGE_SHARP_DB) * 0.3) | |
| filters.append(f'equalizer=f={sorted_f[i]:.0f}:width_type=q:width=4.0:g=-{cut:.1f}') | |
| else: | |
| # Sharp drop β gentle boost at lower band | |
| boost = min(2.0, (delta - SPEC_EDGE_SHARP_DB) * 0.3) | |
| filters.append(f'equalizer=f={sorted_f[i-1]:.0f}:width_type=q:width=4.0:g={boost:.1f}') | |
| if not filters: | |
| L(f' No spectral repair filters needed') | |
| # v2: Track spectral_holes_after even when no filters needed | |
| post_check, _ = _decode_samples(wav_path) | |
| if post_check is not None: | |
| st.spectral_holes_after, _ = _detect_spectral_holes(post_check) | |
| return wav_path | |
| # Limit total filter count (ffmpeg has practical limits) | |
| filters = filters[:20] | |
| out = _tmp_wav('p7_spectral', st) | |
| rc, _, _ = _run_ffmpeg([ | |
| 'ffmpeg', '-y', '-i', wav_path, | |
| '-af', ','.join(filters), | |
| '-acodec', WAV_CODEC, '-ar', str(SR), '-ac', '1', | |
| '-loglevel', 'error', out | |
| ]) | |
| if rc != 0 or not os.path.exists(out): | |
| L(f' Spectral repair failed β keeping original') | |
| _cleanup(out) | |
| return wav_path | |
| # Guard: overall spectral smoothness should improve | |
| post, _ = _decode_samples(out) | |
| if post is not None: | |
| new_holes, _ = _detect_spectral_holes(post) | |
| st.spectral_holes_after = new_holes | |
| if new_holes > artifacts.spectral_holes: | |
| L(f' Spectral holes increased: {artifacts.spectral_holes}β{new_holes} β REVERT') | |
| _cleanup(out) | |
| st.guard_reverts += 1 | |
| return wav_path | |
| st.p7_spectral_repair = True | |
| L(f' β spectral repair: {len(filters)} filters applied') | |
| return out | |
| # ββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ | |
| # Phase 8: SIBILANT NATURALIZATION | |
| # ββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ | |
| def phase8_sibilant_naturalization(wav_path: str, samples: 'np.ndarray', | |
| artifacts: ArtifactReport, st: IhyaState) -> str: | |
| """ | |
| Phase 8: Naturalize sibilants (Ψ΄/Ψ³/Ψ΅/Ψ²). | |
| Codec and NR often destroy sibilant texture, making them sound | |
| artificial (too sharp or too dull). This phase: | |
| - If sibilants too dull (destroyed): boost Safir band (5.5-12kHz) | |
| - If sibilants too sharp (harsh): de-ess the Safir band | |
| Β§152: Arabic sibilant spectral characteristics. | |
| """ | |
| _chk('phase_8_sibilant_nat') | |
| if artifacts.sibilant_quality > 0.8: | |
| L(f' Sibilant quality={artifacts.sibilant_quality:.2f} > 0.8 β OK') | |
| return wav_path | |
| # Measure current sibilant balance | |
| mid_energy = _band_energy(samples, 1000, 3000) | |
| safir_energy = _band_energy(samples, 5500, 12000) | |
| if mid_energy < 1e-20: | |
| return wav_path | |
| ratio_db = 10 * np.log10(safir_energy / (mid_energy + 1e-20)) | |
| filters = [] | |
| if ratio_db < -25: | |
| # Sibilants destroyed by NR/codec β boost them back | |
| boost = min(4.0 if not st.aggressive else 6.0, (-25 - ratio_db) * 0.3) | |
| # v2: Don't boost above active bandwidth | |
| bw = artifacts.active_bandwidth_hz | |
| if bw >= 6000: | |
| filters.append(f'equalizer=f=6000:width_type=o:width=0.8:g={boost:.1f}') | |
| if bw >= 8000: | |
| filters.append(f'equalizer=f=8000:width_type=o:width=0.7:g={boost*0.8:.1f}') | |
| if bw >= 10000: | |
| filters.append(f'equalizer=f=10000:width_type=o:width=0.6:g={boost*0.5:.1f}') | |
| L(f' Sibilant deficit: {ratio_db:.1f}dB β boosting +{boost:.1f}dB') | |
| elif ratio_db > -3: | |
| # Sibilants harsh/artificial β de-ess | |
| cut = min(3.0 if not st.aggressive else 4.5, (ratio_db + 3) * 0.5) | |
| filters.append(f'equalizer=f=6000:width_type=o:width=0.8:g=-{cut:.1f}') | |
| filters.append(f'equalizer=f=8000:width_type=o:width=0.7:g=-{cut*0.7:.1f}') | |
| L(f' Sibilant harsh: {ratio_db:.1f}dB β cutting -{cut:.1f}dB') | |
| else: | |
| # Moderate β subtle texture enhancement | |
| # Add slight breathiness texture via high-frequency presence | |
| # This makes sibilants sound more natural/airy | |
| filters.append(f'equalizer=f=7000:width_type=o:width=0.8:g=0.8') | |
| L(f' Sibilant moderate: subtle texture enhancement') | |
| if not filters: | |
| return wav_path | |
| out = _tmp_wav('p8_sibilant', st) | |
| rc, _, _ = _run_ffmpeg([ | |
| 'ffmpeg', '-y', '-i', wav_path, | |
| '-af', ','.join(filters), | |
| '-acodec', WAV_CODEC, '-ar', str(SR), '-ac', '1', | |
| '-loglevel', 'error', out | |
| ]) | |
| if rc != 0 or not os.path.exists(out): | |
| _cleanup(out) | |
| return wav_path | |
| # Guard: verify sibilant quality improved | |
| post, _ = _decode_samples(out) | |
| if post is not None: | |
| new_quality = _detect_sibilant_quality(post) | |
| if new_quality < artifacts.sibilant_quality - 0.1: | |
| L(f' Sibilant quality worsened: {artifacts.sibilant_quality:.2f}β{new_quality:.2f} β REVERT') | |
| _cleanup(out) | |
| st.guard_reverts += 1 | |
| return wav_path | |
| st.p8_sibilant_nat = True | |
| L(f' β sibilant naturalization applied') | |
| return out | |
| # ββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ | |
| # Phase 9: TEMPORAL ENVELOPE SHAPING | |
| # ββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ | |
| def phase9_temporal_shaping(wav_path: str, samples: 'np.ndarray', | |
| artifacts: ArtifactReport, st: IhyaState) -> str: | |
| """ | |
| Phase 9: Restore natural temporal envelope. | |
| Artificial/processed audio often has smeared consonant-vowel boundaries | |
| and flattened transients. This phase: | |
| - Sharpens consonant attacks (Qalqalah preservation, Β§35) | |
| - Restores natural vowel onset/offset shapes | |
| - Uses gentle expansion on transient regions | |
| v2: Added second sub-pass that boosts 2-4kHz briefly on transient | |
| onsets for consonant sharpening (presence boost). | |
| """ | |
| _chk('phase_9_temporal_shaping') | |
| # Measure current crest factor β if too low, transients are smeared | |
| rms_db, crest_db = _measure_rms_crest(samples) | |
| target_crest = TARGET['crest'] | |
| crest_deficit = target_crest - crest_db | |
| if crest_deficit < 1.0: | |
| L(f' Crest={crest_db:.1f}dB OK (target={target_crest:.1f}) β skipping') | |
| return wav_path | |
| L(f' Crest deficit={crest_deficit:.1f}dB β applying transient shaping') | |
| # Strategy: gentle expansion on quiet-to-loud transitions | |
| # This sharpens consonant attacks without changing overall dynamics | |
| frame_n = int(0.010 * SR) # 10ms frames for fine transient detection | |
| frame_rms = np.array([float(np.sqrt(np.mean(samples[i*frame_n:(i+1)*frame_n]**2))) | |
| for i in range(len(samples) // frame_n)]) | |
| # Find transient frames (sudden energy increase) | |
| if len(frame_rms) < 20: | |
| return wav_path | |
| # Use aexpander with fast attack to catch transients | |
| # Threshold at p40 of voiced frames | |
| voiced_rms = frame_rms[frame_rms > 1e-5] | |
| if len(voiced_rms) < 10: | |
| return wav_path | |
| threshold = float(np.percentile(voiced_rms, 40)) | |
| ratio = min(1.0 + crest_deficit * 0.08, 1.5) # gentle | |
| if st.aggressive: | |
| ratio = min(1.8, ratio * 1.2) | |
| # Sub-pass 1: Broadband expansion for transient recovery | |
| out = _tmp_wav('p9_temporal', st) | |
| rc, _, _ = _run_ffmpeg([ | |
| 'ffmpeg', '-y', '-i', wav_path, | |
| '-af', (f'aexpander=threshold={max(0.001,threshold):.5f}' | |
| f':ratio={ratio:.2f}:attack=1:release=50'), | |
| '-acodec', WAV_CODEC, '-ar', str(SR), '-ac', '1', | |
| '-loglevel', 'error', out | |
| ]) | |
| if rc != 0 or not os.path.exists(out): | |
| _cleanup(out) | |
| return wav_path | |
| # Sub-pass 2: v2 β Consonant presence boost (2-4kHz) on transient onsets | |
| # Uses fast-attack compression on 2-4kHz band to sharpen consonant-vowel boundaries | |
| # This makes consonants like Ψͺ/Ω/Ω/Ψ¨ more distinct | |
| presence_boost = min(2.0, crest_deficit * 0.15) | |
| if st.aggressive: | |
| presence_boost = min(3.0, presence_boost * 1.3) | |
| if presence_boost > 0.3: | |
| # Apply presence EQ + gentle compression for transient sharpening | |
| # The 2-4kHz range is where consonant articulation energy lives | |
| presence_filter = ( | |
| f'equalizer=f=3000:width_type=o:width=0.5:g={presence_boost:.1f},' | |
| f'aemphasis=1:replay_gain=track' | |
| ) | |
| out2 = _tmp_wav('p9_presence', st) | |
| rc2, _, _ = _run_ffmpeg([ | |
| 'ffmpeg', '-y', '-i', out, | |
| '-af', presence_filter, | |
| '-acodec', WAV_CODEC, '-ar', str(SR), '-ac', '1', | |
| '-loglevel', 'error', out2 | |
| ]) | |
| if rc2 == 0 and os.path.exists(out2): | |
| # Verify presence boost didn't over-brighten | |
| post2, _ = _decode_samples(out2) | |
| if post2 is not None: | |
| # Check sibilant band isn't over-boosted | |
| sib_before = _band_energy_db(samples, 5500, 12000) | |
| sib_after = _band_energy_db(post2, 5500, 12000) | |
| if sib_before > -80 and (sib_after - sib_before) < 3.0: | |
| _cleanup(out) | |
| out = out2 | |
| L(f' β presence boost: +{presence_boost:.1f}dB at 2-4kHz') | |
| else: | |
| L(f' Presence boost too bright β keeping expansion only') | |
| _cleanup(out2) | |
| else: | |
| _cleanup(out2) | |
| else: | |
| _cleanup(out2) | |
| # Guard: crest should improve but not exceed target by much | |
| post, _ = _decode_samples(out) | |
| if post is not None: | |
| _, new_crest = _measure_rms_crest(post) | |
| crest_max = target_crest + (3.0 if st.aggressive else 2.0) | |
| if new_crest > crest_db and new_crest <= crest_max: | |
| st.p9_temporal_shaping = True | |
| L(f' β temporal shaping: crest {crest_db:.1f}β{new_crest:.1f}dB') | |
| return out | |
| else: | |
| L(f' Crest guard failed: {crest_db:.1f}β{new_crest:.1f} β REVERT') | |
| _cleanup(out) | |
| st.guard_reverts += 1 | |
| return wav_path | |
| return wav_path | |
| # ββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ | |
| # Phase 10: TAJWEED PHONEME GUARDS | |
| # ββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ | |
| def phase10_guards(original_samples: 'np.ndarray', processed_wav: str, | |
| st: IhyaState) -> str: | |
| """ | |
| Phase 10: Verify Tajweed-critical phoneme features survived processing. | |
| Seven guards from Β§35, Β§52, Β§143, Β§152. | |
| WARN-only by design β naturalness improvement outweighs mild phoneme loss. | |
| """ | |
| _chk('phase_10_guards') | |
| if not NUMPY_OK or original_samples is None: | |
| L(f' No samples for guards β skip') | |
| return processed_wav | |
| proc_samples, _ = _decode_samples(processed_wav) | |
| if proc_samples is None: | |
| return processed_wav | |
| n = min(len(original_samples), len(proc_samples)) | |
| o = original_samples[:n] | |
| p = proc_samples[:n] | |
| def chk(name, cond, detail): | |
| entry = f'{name}: {detail}' | |
| if cond: | |
| st.guard_pass.append(entry) | |
| L(f' [PASS] {entry}') | |
| else: | |
| st.guard_warn.append(entry) | |
| L(f' [WARN] {entry}') | |
| # G1 Ghunnah 220-320Hz (Β§152.3) | |
| go = _band_energy(o, *GHUNNAH_BAND); gp = _band_energy(p, *GHUNNAH_BAND) | |
| if go > 1e-15: | |
| d = 10 * np.log10(gp / go + 1e-20) | |
| chk('G1-Ghunnah', d >= -3.0, | |
| f'{d:+.1f}dB {"OK" if d>=-3 else "nasal murmur at risk"}') | |
| # G2 Ikhfa 250-420Hz (Β§52.5) | |
| io = _band_energy(o, *IKHFA_BAND); ip = _band_energy(p, *IKHFA_BAND) | |
| if io > 1e-15: | |
| d = 10 * np.log10(ip / io + 1e-20) | |
| chk('G2-Ikhfa', d >= -4.0, | |
| f'{d:+.1f}dB {"OK" if d>=-4 else "nasalisation at risk"}') | |
| # G3 Qalqalah burst (Β§52.7, Β§143) | |
| sil_n = int(0.020 * SR); bst_n = int(0.030 * SR) | |
| total = viol = 0 | |
| for i in range(0, n - sil_n - bst_n, sil_n): | |
| sr_ = float(np.sqrt(np.mean(o[i:i+sil_n]**2)) + 1e-10) | |
| br_ = float(np.sqrt(np.mean(o[i+sil_n:i+sil_n+bst_n]**2)) + 1e-10) | |
| if sr_ < 0.005 and br_ > sr_ * 5: | |
| total += 1 | |
| bp = float(np.sqrt(np.mean(p[i+sil_n:i+sil_n+bst_n]**2)) + 1e-10) | |
| if 20 * np.log10(bp / br_ + 1e-10) < -6.0: | |
| viol += 1 | |
| if total > 0: | |
| pct = viol / total * 100 | |
| chk('G3-Qalqalah', pct <= 20, | |
| f'{total} bursts {pct:.0f}% violated') | |
| # G4 Ra trill AM 22-40Hz | |
| win = SR; hop = SR // 2 | |
| am_ratios = [] | |
| for pos in range(0, n - win, hop): | |
| ef_o = np.abs(np.fft.rfft(np.abs(o[pos:pos+win]), n=win)) | |
| ef_p = np.abs(np.fft.rfft(np.abs(p[pos:pos+win]), n=win)) | |
| am_o = float(np.mean(ef_o[RA_TRILL_AM_BAND[0]:RA_TRILL_AM_BAND[1]])) | |
| am_p = float(np.mean(ef_p[RA_TRILL_AM_BAND[0]:RA_TRILL_AM_BAND[1]])) | |
| if am_o > 1e-8: | |
| am_ratios.append(am_p / am_o) | |
| if am_ratios: | |
| r = float(np.median(am_ratios)) | |
| chk('G4-Ra-trill', r >= 0.70, | |
| f'AM ratio={r:.2f} (median {len(am_ratios)} windows)') | |
| # G5 Safir 5500-12000Hz (Β§152.3) | |
| so = _band_energy(o, *SAFIR_BAND); sp = _band_energy(p, *SAFIR_BAND) | |
| if so > 1e-15: | |
| d = 10 * np.log10(sp / so + 1e-20) | |
| chk('G5-Safir', d >= -5.0, | |
| f'{d:+.1f}dB {"OK" if d>=-5 else "sibilants at risk"}') | |
| # G6 Tafasshi 3000-8000Hz (Β§152.3) | |
| to = _band_energy(o, *TAFASSHI_BAND); tp = _band_energy(p, *TAFASSHI_BAND) | |
| if to > 1e-15: | |
| d = 10 * np.log10(tp / to + 1e-20) | |
| chk('G6-Tafasshi', d >= -4.0, | |
| f'{d:+.1f}dB {"OK" if d>=-4 else "Ψ΄ spread at risk"}') | |
| # G7 Formant F2 preservation (Β§4) | |
| f2o = _band_energy(o, 800, 2500); f2p = _band_energy(p, 800, 2500) | |
| if f2o > 1e-15: | |
| d = 10 * np.log10(f2p / f2o + 1e-20) | |
| chk('G7-Formant-F2', d >= -3.0, | |
| f'{d:+.1f}dB {"OK" if d>=-3 else "formant clarity at risk"}') | |
| total_g = len(st.guard_pass) + len(st.guard_warn) | |
| if st.guard_warn: | |
| L(f' β {len(st.guard_warn)}/{total_g} warnings β check output for Tajweed artifacts') | |
| else: | |
| L(f' β All {total_g} guards passed') | |
| st.p10_guards_ok = len(st.guard_warn) == 0 | |
| return processed_wav | |
| # ββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ | |
| # Phase 11: FINAL LOUDNESS + QUALITY OPTIMIZATION | |
| # ββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ | |
| def phase11_final(wav_path: str, st: IhyaState) -> Tuple[str, Dict]: | |
| """ | |
| Phase 11: Final loudness normalization + quality optimization. | |
| - Target LUFS from reference model | |
| - True peak limiting | |
| - Gentle dynamic normalization for evenness | |
| - Stereo output encode | |
| """ | |
| _chk('phase_11_final') | |
| # Measure current levels | |
| lufs, lra = _measure_lufs(wav_path) | |
| L(f' Pre-final: LUFS={lufs:.2f} LRA={lra:.2f}') | |
| # Build final filter chain | |
| # 1. dynaudnorm for level evenness | |
| # 2. Volume adjustment to target LUFS | |
| # 3. Limiter for true peak | |
| vol_delta = TARGET['lufs'] - lufs | |
| # Cap volume adjustment | |
| vol_delta = max(-12.0, min(12.0, vol_delta)) | |
| final_af = ( | |
| f'dynaudnorm=f=500:g=31:p=0.92:m=10:r=0.0:b=1,' | |
| f'volume={vol_delta:.2f}dB,' | |
| f'alimiter=level_in=1:level_out=1:limit=0.89:attack=5:release=50' | |
| ) | |
| out = st.output_path | |
| rc, _, err = _run_ffmpeg([ | |
| 'ffmpeg', '-y', '-i', wav_path, | |
| '-af', final_af, | |
| '-acodec', WAV_CODEC, '-ar', str(SR), '-ac', '2', | |
| '-loglevel', 'error', out | |
| ]) | |
| if rc != 0: | |
| L(f' Final encode failed: {err[:80]}') | |
| # Fallback: just copy | |
| rc2, _, _ = _run_ffmpeg([ | |
| 'ffmpeg', '-y', '-i', wav_path, | |
| '-acodec', WAV_CODEC, '-ar', str(SR), '-ac', '2', out | |
| ]) | |
| if rc2 != 0: | |
| L(f' Fallback encode also failed') | |
| return wav_path, {} | |
| # Final measurements | |
| final_lufs, final_lra = _measure_lufs(out) | |
| final_samples, _ = _decode_samples(out) | |
| final_rms, final_crest = _measure_rms_crest(final_samples) if final_samples is not None else (-99, 0) | |
| report = { | |
| 'lufs': final_lufs, | |
| 'lra': final_lra, | |
| 'rms': final_rms, | |
| 'crest': final_crest, | |
| 'vol_delta_db': vol_delta, | |
| } | |
| L(f' Final: LUFS={final_lufs:.2f} LRA={final_lra:.2f} ' | |
| f'Crest={final_crest:.1f}dB vol={vol_delta:+.2f}dB') | |
| st.p11_final_done = True | |
| return out, report | |
| # ββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ | |
| # NATURALNESS SCORE (v2) | |
| # ββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ | |
| def _compute_naturalness_score(st: IhyaState, artifacts: ArtifactReport, | |
| final_samples: Optional['np.ndarray']) -> float: | |
| """ | |
| Compute a composite naturalness score (0-100). | |
| Similar to Itiqan's quality score. Includes: | |
| - F0 flatness improvement | |
| - Spectral hole reduction | |
| - Metallic score reduction | |
| - Formant improvement | |
| - Sibilant quality | |
| - Guard pass rate | |
| - Perceptual band quality (using _PERC_WEIGHT) | |
| Returns score 0-100 (higher = more natural). | |
| """ | |
| if not NUMPY_OK: | |
| return 50.0 | |
| score = 50.0 # start at 50 (neutral) | |
| # ββ F0 flatness improvement: up to Β±15 points ββ | |
| f0_improvement = st.f0_flatness_before - st.f0_flatness_after | |
| score += float(np.clip(f0_improvement * 30, -10, 15)) | |
| # ββ Spectral hole reduction: up to Β±10 points ββ | |
| hole_reduction = st.spectral_holes_before - st.spectral_holes_after | |
| score += float(np.clip(hole_reduction * 2, -5, 10)) | |
| # ββ Metallic score reduction: up to Β±10 points ββ | |
| metallic_reduction = st.metallic_before - st.metallic_after | |
| score += float(np.clip(metallic_reduction * 20, -5, 10)) | |
| # ββ Formant improvement: up to +10 points ββ | |
| if artifacts.formant_collapse < 0.2: | |
| score += 10.0 | |
| elif artifacts.formant_collapse < 0.4: | |
| score += 5.0 | |
| elif artifacts.formant_collapse > 0.7: | |
| score -= 5.0 | |
| # ββ Sibilant quality: up to +10 points ββ | |
| score += float(np.clip(artifacts.sibilant_quality * 10, -5, 10)) | |
| # ββ Guard pass rate: up to +5 points ββ | |
| total_guards = len(st.guard_pass) + len(st.guard_warn) | |
| if total_guards > 0: | |
| pass_rate = len(st.guard_pass) / total_guards | |
| score += pass_rate * 5.0 | |
| else: | |
| score += 2.5 # no guards to fail = neutral | |
| # ββ Revert penalty: up to -10 points ββ | |
| score -= min(10, st.guard_reverts * 2.0) | |
| # ββ Perceptual band quality (using _PERC_WEIGHT): up to +5 points ββ | |
| if final_samples is not None and artifacts is not None: | |
| perc_score = 0.0 | |
| n_bands = 0 | |
| for freq, weight in _PERC_WEIGHT.items(): | |
| # Check if band energy is reasonable (not too hot/cold) | |
| lo = freq / (2 ** (1.0/6)) | |
| hi = freq * (2 ** (1.0/6)) | |
| band_e = _band_energy_db(final_samples, lo, hi) | |
| # Reasonable band energy: -60 to -5 dB | |
| if -50 < band_e < -5: | |
| perc_score += weight | |
| n_bands += 1 | |
| if n_bands > 0: | |
| perc_norm = perc_score / sum(_PERC_WEIGHT.values()) | |
| score += perc_norm * 5.0 | |
| return float(max(0.0, min(100.0, score))) | |
| # ββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ | |
| # TIER_TELEPHONE: BANDWIDTH EXTENSION PHASE (Β§173) | |
| # ββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ | |
| def _telephone_soft_declip(samples: 'np.ndarray') -> 'np.ndarray': | |
| """ | |
| Soft-knee declip via tanh saturation (Β§173 Step 0). | |
| Converts hard-clipped peaks into soft saturation before feeding BWE models. | |
| Very light: threshold=0.97, maps anything above to gentle saturation. | |
| """ | |
| CLIP_THRESH = 0.97 | |
| clip_count = int(np.sum(np.abs(samples) > CLIP_THRESH)) | |
| if clip_count == 0: | |
| return samples | |
| # tanh saturation: y = tanh(x * (1/CLIP_THRESH)) * CLIP_THRESH | |
| # smooth approximation to clipping that avoids hard corners | |
| out = np.tanh(samples * (1.0 / CLIP_THRESH)) * CLIP_THRESH | |
| L(f' Soft declip: {clip_count} samples above {CLIP_THRESH:.2f} β tanh-limited') | |
| return out | |
| def _telephone_crossover_blend(original: 'np.ndarray', | |
| extended: 'np.ndarray', | |
| cutoff_hz: float, | |
| sr: int = SR) -> 'np.ndarray': | |
| """ | |
| Crossover blend β Β§173 Step 5 (Pipeline B, blend step). | |
| Preserves original signal below cutoff_hz bit-for-bit, | |
| uses BWE output only above cutoff_hz. | |
| This is the CRITICAL safety net: | |
| - Original 0βcutoff_hz: guaranteed authentic (no synthesis) | |
| - BWE cutoff_hzβ22kHz: synthesised / plausible but not certain | |
| - Red Line 18: caller must tag output as 'BWE-processed / synthetic HF' | |
| Implementation: 8th-order Butterworth LR crossover (Linkwitz-Riley | |
| equivalent via cascade). -3dB at cutoff_hz. | |
| Uses scipy.signal.butter(order//2) applied twice for steeper slope. | |
| """ | |
| # Pad to same length | |
| n = min(len(original), len(extended)) | |
| orig = original[:n] | |
| ext = extended[:n] | |
| nyq = sr / 2.0 | |
| fc = float(np.clip(cutoff_hz, 100.0, nyq * 0.98)) | |
| wn = fc / nyq | |
| # Low-pass on original (preserve authentic band) | |
| sos_lo = butter(TELEPHONE_BLEND_ORDER, wn, btype='low', output='sos') | |
| # High-pass on extended (take only the synthesised new content) | |
| sos_hi = butter(TELEPHONE_BLEND_ORDER, wn, btype='high', output='sos') | |
| lo = sosfilt(sos_lo, orig) | |
| hi = sosfilt(sos_hi, ext) | |
| blended = lo + hi | |
| # Normalise peak to avoid clipping from blend summation | |
| peak = float(np.max(np.abs(blended))) | |
| if peak > 0.98: | |
| blended = blended * (0.98 / peak) | |
| L(f' Blend: peak {peak:.3f} β normalised to 0.98') | |
| return blended.astype(np.float32) | |
| def phase_telephone_bwe(wav_path: str, samples: 'np.ndarray', | |
| artifacts: ArtifactReport, st: IhyaState) -> str: | |
| """ | |
| TIER_TELEPHONE Bandwidth Extension β Β§173 Pipeline B-lite. | |
| This is the ONLY viable treatment when codec_cutoff β€ 4 kHz. | |
| All standard enhancement passes (NR, harmonics, EQ above 3kHz) are | |
| USELESS on this tier and are bypassed by the caller. | |
| Β§171 Guards enforced here: | |
| G1 (R-3 BWE): No PCHIP harmonic inference (only ~13 harmonics present). | |
| G2 (EQ): No EQ boost above 3kHz before VoiceFixer. | |
| G3 (DF3): No DeepFilterNet AFTER VoiceFixer (would suppress syn. HF). | |
| Pipeline: | |
| Step 0 Soft declip (tanh saturation, threshold 0.97) | |
| Step 1 Resample to 44100 Hz (VoiceFixer internal requirement) | |
| Step 2 VoiceFixer Mode 0 β ResUNet discriminative NBβWB/fullband | |
| mode=0 preferred: least hallucination, best phoneme fidelity | |
| mode=2 (HiFi-GAN) avoided: alters Arabic phoneme identity (Β§173) | |
| Step 3 Resample VF output back to 48 kHz | |
| Step 4 Crossover blend: original 0βcutoff + VF output cutoffβ22kHz | |
| Guarantees authentic signal below cutoff (Red Line 18) | |
| Step 5 AudioSR Stage B (optional) β only if audiosr is installed | |
| Applies AFTER VoiceFixer so AudioSR receives wideband input | |
| (not raw 3.4kHz telephone audio β Β§173 Colab recipe) | |
| guidance_scale=2.5 (reduced from default 3.5: less hallucination) | |
| KB refs: Β§169 Β§170 Β§171 Β§172 Β§173 | |
| """ | |
| _chk('phase_telephone_bwe') | |
| L(f' β TIER_TELEPHONE β BWE-first pipeline (Β§173)') | |
| L(f' β Cutoff: {artifacts.telephone_cutoff_hz:.0f} Hz ' | |
| f'Missing: {artifacts.telephone_cutoff_hz:.0f}β22000 Hz ' | |
| f'({100.0*(22000-artifacts.telephone_cutoff_hz)/22000:.0f}% of spectrum is synthesised)') | |
| L(f' β NR / harmonic / EQ passes BYPASSED (Β§171 guards active)') | |
| if not NUMPY_OK: | |
| L(' SKIP: numpy not available') | |
| return wav_path | |
| if not VOICEFIXER_OK: | |
| L('') | |
| L(' ββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ') | |
| L(' β VoiceFixer NOT INSTALLED β BWE cannot proceed β') | |
| L(' β Install: pip install voicefixer β') | |
| L(' β TIER_TELEPHONE audio will NOT be enhanced β') | |
| L(' ββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ') | |
| L('') | |
| L(' Passing audio through unchanged (standard phases also bypassed)') | |
| return wav_path | |
| cutoff_hz = artifacts.telephone_cutoff_hz or 3400.0 | |
| # ββ Step 0: Soft declip βββββββββββββββββββββββββββββββββββββββββββββββββββ | |
| samples_dc = _telephone_soft_declip(samples) | |
| # ββ Step 1: Write declipped samples as 44100 Hz WAV for VoiceFixer βββββββ | |
| # VoiceFixer resamples internally to 44100; provide it at 44100 mono. | |
| wav_44k = _tmp_wav('tel_44k', st) | |
| # Write float32 samples via ffmpeg pipe (avoids soundfile dependency) | |
| try: | |
| raw_bytes = samples_dc.astype(np.float32).tobytes() | |
| pipe_cmd = [ | |
| 'ffmpeg', '-y', '-loglevel', 'error', | |
| '-f', 'f32le', '-ar', str(SR), '-ac', '1', '-i', 'pipe:0', | |
| '-ar', '44100', '-ac', '1', '-acodec', 'pcm_s16le', wav_44k, | |
| ] | |
| proc = subprocess.run(pipe_cmd, input=raw_bytes, | |
| capture_output=True, timeout=300) | |
| if proc.returncode != 0 or not os.path.exists(wav_44k): | |
| L(f' Resampleβ44100 failed, trying file-based route') | |
| rc, _, _ = _run_ffmpeg([ | |
| 'ffmpeg', '-y', '-i', wav_path, | |
| '-ar', '44100', '-ac', '1', '-acodec', 'pcm_s16le', wav_44k, | |
| '-loglevel', 'error', | |
| ]) | |
| if rc != 0: | |
| L(f' Resampleβ44100 failed entirely β BWE aborted') | |
| return wav_path | |
| except Exception as ex: | |
| L(f' Resample write error: {ex} β BWE aborted') | |
| return wav_path | |
| L(f' 44100 Hz WAV written: {wav_44k}') | |
| # ββ Step 2: VoiceFixer Mode 0 βββββββββββββββββββββββββββββββββββββββββββββ | |
| vf_out_44k = _tmp_wav('tel_vf44', st) | |
| vf_success = False | |
| try: | |
| L(f' VoiceFixer Mode 0 (ResUNet discriminative) β loadingβ¦') | |
| vf = _VoiceFixer() | |
| # cuda=False: safe default; user can override if GPU available | |
| # mode=0: discriminative ResUNet β best phoneme fidelity for Arabic (Β§173) | |
| # mode=2 (HiFi-GAN) deliberately avoided β alters phoneme identity | |
| vf.restore(input=wav_44k, output=vf_out_44k, mode=0, cuda=False) | |
| if os.path.exists(vf_out_44k) and os.path.getsize(vf_out_44k) > 1000: | |
| L(f' VoiceFixer Mode 0: OK β {vf_out_44k}') | |
| vf_success = True | |
| st.telephone_bwe_mode = 'voicefixer_mode0' | |
| else: | |
| L(f' VoiceFixer produced empty output β BWE aborted') | |
| except Exception as ex: | |
| L(f' VoiceFixer failed: {ex}') | |
| if not vf_success: | |
| _cleanup(wav_44k) | |
| return wav_path | |
| # ββ Step 3: Resample VoiceFixer output back to 48 kHz ββββββββββββββββββββ | |
| wav_vf_48k = _tmp_wav('tel_vf48', st) | |
| rc, _, err = _run_ffmpeg([ | |
| 'ffmpeg', '-y', '-i', vf_out_44k, | |
| '-ar', str(SR), '-ac', '1', '-acodec', WAV_CODEC, wav_vf_48k, | |
| '-loglevel', 'error', | |
| ]) | |
| _cleanup(wav_44k, vf_out_44k) | |
| if rc != 0 or not os.path.exists(wav_vf_48k): | |
| L(f' Resampleβ48k failed: {err[:60]}') | |
| return wav_path | |
| # ββ Step 4: Crossover blend ββββββββββββββββββββββββββββββββββββββββββββββββ | |
| # Load VF output at 48 kHz | |
| vf_samples, _ = _decode_samples(wav_vf_48k) | |
| if vf_samples is None: | |
| L(f' Could not decode VF output β using VF output as-is') | |
| # Still an improvement over the raw telephone audio | |
| blended_wav = wav_vf_48k | |
| st.telephone_bwe_applied = True | |
| else: | |
| L(f' Crossover blend at {cutoff_hz:.0f} Hz ' | |
| f'(orig 0β{cutoff_hz:.0f}Hz + VF {cutoff_hz:.0f}β22kHz)') | |
| blended = _telephone_crossover_blend(samples_dc, vf_samples, cutoff_hz, SR) | |
| # Write blended result | |
| blended_wav = _tmp_wav('tel_blend', st) | |
| try: | |
| raw_bytes = blended.astype(np.float32).tobytes() | |
| pipe_cmd = [ | |
| 'ffmpeg', '-y', '-loglevel', 'error', | |
| '-f', 'f32le', '-ar', str(SR), '-ac', '1', '-i', 'pipe:0', | |
| '-acodec', WAV_CODEC, '-ar', str(SR), blended_wav, | |
| ] | |
| proc = subprocess.run(pipe_cmd, input=raw_bytes, | |
| capture_output=True, timeout=300) | |
| if proc.returncode != 0 or not os.path.exists(blended_wav): | |
| L(f' Blend write failed β using raw VF output') | |
| blended_wav = wav_vf_48k | |
| else: | |
| _cleanup(wav_vf_48k) | |
| L(f' Blend written: {blended_wav}') | |
| except Exception as ex: | |
| L(f' Blend write error: {ex} β using raw VF output') | |
| blended_wav = wav_vf_48k | |
| st.telephone_bwe_applied = True | |
| # ββ Step 5: AudioSR Stage B (optional) βββββββββββββββββββββββββββββββββββ | |
| # Β§173 Colab recipe: AudioSR on VoiceFixer OUTPUT (not raw telephone audio) | |
| # Only run if audiosr is installed. guidance_scale reduced to 2.5 (Β§172-E). | |
| # Guard (Β§171): AudioSR not applied to raw 3.4kHz audio β VF must run first. | |
| if AUDIOSR_OK and st.telephone_bwe_applied: | |
| L(f' AudioSR Stage B (WBβ48kHz) β guidance_scale=2.5 (low hallucination)') | |
| asr_out = _tmp_wav('tel_asr', st) | |
| try: | |
| asr_model = _audiosr.build_model(model_name='basic') | |
| asr_wave = _audiosr.super_resolution( | |
| asr_model, | |
| blended_wav, | |
| guidance_scale=2.5, # reduced from default 3.5 (Β§172-E: less hallucination) | |
| ddim_steps=50, | |
| ) | |
| _audiosr.save_wave(asr_wave, inputpath=blended_wav, | |
| savepath=os.path.dirname(asr_out)) | |
| # audiosr saves with its own naming convention β locate it | |
| asr_candidate = os.path.join( | |
| os.path.dirname(asr_out), | |
| Path(blended_wav).stem + '_audiosr.wav' | |
| ) | |
| if not os.path.exists(asr_candidate): | |
| # Try common naming | |
| for p in Path(os.path.dirname(asr_out)).glob('*audiosr*'): | |
| asr_candidate = str(p) | |
| break | |
| if os.path.exists(asr_candidate): | |
| # Resample to 48k in case AudioSR output 44.1k | |
| rc2, _, _ = _run_ffmpeg([ | |
| 'ffmpeg', '-y', '-i', asr_candidate, | |
| '-ar', str(SR), '-ac', '1', '-acodec', WAV_CODEC, asr_out, | |
| '-loglevel', 'error', | |
| ]) | |
| if rc2 == 0 and os.path.exists(asr_out): | |
| _cleanup(blended_wav) | |
| blended_wav = asr_out | |
| st.telephone_audiosr_applied = True | |
| L(f' AudioSR Stage B: OK β {blended_wav}') | |
| else: | |
| L(f' AudioSR resample failed β keeping VF output') | |
| else: | |
| L(f' AudioSR output not found β keeping VF output') | |
| except Exception as ex: | |
| L(f' AudioSR failed: {ex} β keeping VF output') | |
| elif not AUDIOSR_OK: | |
| L(f' AudioSR not installed (optional Stage B). pip install audiosr') | |
| L(f' VoiceFixer output used as final BWE result') | |
| # ββ Summary βββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ | |
| L(f'') | |
| L(f' ββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ') | |
| L(f' β BWE COMPLETE β Β§173 β') | |
| L(f' β Mode: {st.telephone_bwe_mode:<49}β') | |
| L(f' β AudioSR Stage B: {"YES" if st.telephone_audiosr_applied else "NO (not installed)":<40}β') | |
| L(f' β β Red Line 18: output tagged BWE-processed/synthetic HF β') | |
| L(f' ββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ') | |
| L(f'') | |
| return blended_wav | |
| def process(input_path: str, output_path: str, | |
| force_tier: str = '', | |
| mujawwad_override: float = -1.0, | |
| skip_dereverb: bool = False, | |
| skip_f0_revival: bool = False, | |
| aggressive: bool = False, | |
| verbose: bool = True) -> Dict: | |
| """ | |
| Ψ§ΩΨ₯ΨΩΨ§Ψ‘ β main entry point. | |
| Returns dict with full diagnostics. | |
| v2: Added aggressive mode, improved sample tracking, naturalness score. | |
| """ | |
| t0 = time.time() | |
| st = IhyaState(input_path=input_path, output_path=output_path, t0=t0, | |
| aggressive=aggressive) | |
| L(f'\n{"β"*60}') | |
| L(f' Ψ§ΩΨ₯ΨΩΨ§Ψ‘ {__version__} β Ψ₯ΨΩΨ§Ψ‘ Ψ§ΩΨ΅ΩΨͺ Ψ§ΩΨ¨Ψ΄Ψ±Ω') | |
| L(f' Input: {Path(input_path).name}') | |
| L(f' Output: {Path(output_path).name}') | |
| if aggressive: | |
| L(f' Mode: AGGRESSIVE') | |
| L(f'{"β"*60}') | |
| if not NUMPY_OK: | |
| L(' ERROR: numpy not installed β pip install numpy scipy') | |
| return {'error': 'numpy required', 'score': 0} | |
| # ββ Decode input βββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ | |
| work_wav = _tmp_wav('input', st) | |
| if not _decode_to_wav(input_path, work_wav): | |
| return {'error': f'Failed to decode: {input_path}', 'score': 0} | |
| st.duration_s = _get_duration(work_wav) | |
| st.bitrate_kbps = _get_bitrate(input_path) | |
| L(f' Duration={st.duration_s:.1f}s Bitrate={st.bitrate_kbps}kbps') | |
| samples, sr = _decode_samples(work_wav) | |
| if samples is None: | |
| return {'error': 'Failed to decode samples', 'score': 0} | |
| original_samples = samples.copy() # keep for guards | |
| # ββ Phase 0: Deep Analysis ββββββββββββββββββββββββββββββββββββββββββββββββ | |
| artifacts = phase0_analyze(samples, sr, st) | |
| # If audio is already natural, skip most processing | |
| artificial_thresh = 0.10 if aggressive else 0.15 | |
| if artifacts.overall_artificial < artificial_thresh: | |
| L(f'\n ββ Audio appears natural (artificialness={artifacts.overall_artificial:.3f})') | |
| L(f' ββ Minimal processing only') | |
| else: | |
| L(f'\n ββ Artificialness detected: {artifacts.overall_artificial:.3f}') | |
| L(f' ββ Full revival pipeline activated') | |
| # Override mujawwad if requested | |
| if mujawwad_override >= 0: | |
| st.mujawwad_conf = mujawwad_override | |
| artifacts.mujawwad_conf = mujawwad_override | |
| current_wav = work_wav | |
| # v2: Initialize "after" tracking to "before" values (will be updated) | |
| st.spectral_holes_after = st.spectral_holes_before | |
| st.metallic_after = st.metallic_before | |
| st.f0_flatness_after = st.f0_flatness_before | |
| try: | |
| # ββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ | |
| # TIER_TELEPHONE: BWE-first pipeline (Β§173) | |
| # Standard enhancement passes are BYPASSED for this tier. | |
| # Guards enforced (Β§171): | |
| # G1: No R-3 PCHIP harmonic inference (codec_cutoff < 5000 Hz) | |
| # G2: No EQ boost above 3kHz before BWE | |
| # G3: No DeepFilterNet AFTER VoiceFixer (suppresses synthesised HF) | |
| # ββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ | |
| if artifacts.is_telephone_tier: | |
| # Run BWE phase (soft-declip β VoiceFixer β crossover blend) | |
| current_wav = phase_telephone_bwe(current_wav, samples, artifacts, st) | |
| # After BWE: update working samples | |
| cur_samples, _ = _decode_samples(current_wav) | |
| if cur_samples is None: | |
| cur_samples = samples | |
| # BYPASSED passes (Β§171 β useless / harmful on TIER_TELEPHONE): | |
| L('\nββ phase_1_dereverb ββ SKIP (TIER_TELEPHONE: no useful reverb info)') | |
| L('ββ phase_2_noise_repairββ SKIP (TIER_TELEPHONE: no noise, BWE already ran)') | |
| L('ββ phase_3_f0_revival ββ SKIP (TIER_TELEPHONE: harmonics were absent, now synthesised)') | |
| L('ββ phase_4_formant ββ SKIP (TIER_TELEPHONE: VoiceFixer handles formants)') | |
| L('ββ phase_5_harmonics ββ SKIP (TIER_TELEPHONE: G1 guard β codec_cutoff < 5000 Hz)') | |
| L('ββ phase_7_spectral ββ SKIP (TIER_TELEPHONE: G2 guard β no EQ above 3kHz pre-BWE)') | |
| L('ββ phase_8_sibilant ββ SKIP (TIER_TELEPHONE: G3 guard β sibilants are synthesised)') | |
| # ALLOWED passes (neutral / helpful after BWE): | |
| # P6 micro-dynamics: restore natural loudness variation | |
| current_wav = phase6_micro_dynamics(current_wav, cur_samples, st) | |
| cur_samples, _ = _decode_samples(current_wav) | |
| if cur_samples is None: | |
| cur_samples = samples | |
| # P9 temporal shaping: very light β transient restoration | |
| # Guard: skip presence boost (3-5kHz is synthesised, Β§173 post-VF EQ) | |
| current_wav = phase9_temporal_shaping(current_wav, cur_samples, artifacts, st) | |
| cur_samples, _ = _decode_samples(current_wav) | |
| if cur_samples is None: | |
| cur_samples = samples | |
| # Update after-metrics | |
| if cur_samples is not None and len(cur_samples) > SR: | |
| st.spectral_holes_after, _ = _detect_spectral_holes(cur_samples) | |
| st.metallic_after = _detect_metallic_ringing(cur_samples) | |
| f0_post, _ = _detect_f0(cur_samples) | |
| if len(f0_post) > 10: | |
| st.f0_flatness_after = _compute_f0_flatness(f0_post) | |
| # P10 guards + P11 final (always run) | |
| current_wav = phase10_guards(original_samples, current_wav, st) | |
| final_wav, final_report = phase11_final(current_wav, st) | |
| final_samples, _ = _decode_samples(final_wav) | |
| st.naturalness_score = _compute_naturalness_score(st, artifacts, final_samples) | |
| elapsed = time.time() - t0 | |
| L(f'\n{"β"*60}') | |
| L(f' Ψ§ΩΨ₯ΨΩΨ§Ψ‘ {__version__} β COMPLETE (TIER_TELEPHONE)') | |
| L(f' Time: {elapsed:.1f}s') | |
| L(f' Naturalness score: {st.naturalness_score:.1f}/100') | |
| L(f' BWE mode: {st.telephone_bwe_mode or "none"}') | |
| L(f' AudioSR Stage B: {"YES" if st.telephone_audiosr_applied else "NO"}') | |
| L(f' β Red Line 18: output is BWE-processed / synthetic HF above {st.telephone_cutoff_hz:.0f}Hz') | |
| L(f'{"β"*60}') | |
| result = { | |
| 'engine': 'Ψ§ΩΨ₯ΨΩΨ§Ψ‘', | |
| 'version': __version__, | |
| 'elapsed_s': round(elapsed, 1), | |
| 'naturalness_score': round(st.naturalness_score, 1), | |
| 'aggressive': aggressive, | |
| 'artifacts': { | |
| 'f0_flatness': artifacts.f0_flatness, | |
| 'f0_flat_severity': artifacts.f0_flat_severity, | |
| 'spectral_holes': artifacts.spectral_holes, | |
| 'metallic_score': artifacts.metallic_score, | |
| 'nr_damage': artifacts.nr_damage_score, | |
| 'formant_collapse': artifacts.formant_collapse, | |
| 'sibilant_quality': artifacts.sibilant_quality, | |
| 'overall': artifacts.overall_artificial, | |
| 'diagnosis': artifacts.diagnosis, | |
| 'snr_db': artifacts.snr_db, | |
| 'is_noisy': artifacts.is_noisy, | |
| 'active_bandwidth_hz': artifacts.active_bandwidth_hz, | |
| 'wind_detected': artifacts.wind_detected, | |
| }, | |
| 'improvement': { | |
| 'f0_flatness_before': st.f0_flatness_before, | |
| 'f0_flatness_after': st.f0_flatness_after, | |
| 'spectral_holes_before': st.spectral_holes_before, | |
| 'spectral_holes_after': st.spectral_holes_after, | |
| 'metallic_before': st.metallic_before, | |
| 'metallic_after': st.metallic_after, | |
| }, | |
| 'phases': { | |
| 'dereverb': False, | |
| 'nr_repair': False, | |
| 'f0_revival': False, | |
| 'jitter': False, | |
| 'shimmer': False, | |
| 'formant': False, | |
| 'harmonics': False, | |
| 'dynamics': st.p6_micro_dynamics, | |
| 'spectral': False, | |
| 'sibilant': False, | |
| 'temporal': st.p9_temporal_shaping, | |
| }, | |
| 'guards': { | |
| 'pass': st.guard_pass, | |
| 'warn': st.guard_warn, | |
| 'reverts': st.guard_reverts, | |
| }, | |
| 'final': final_report, | |
| 'mujawwad_conf': st.mujawwad_conf, | |
| 'f0_median': st.f0_median, | |
| 'telephone_tier': { | |
| 'detected': True, | |
| 'cutoff_hz': st.telephone_cutoff_hz, | |
| 'bwe_applied': st.telephone_bwe_applied, | |
| 'bwe_mode': st.telephone_bwe_mode, | |
| 'audiosr_stage_b': st.telephone_audiosr_applied, | |
| 'red_line_18': 'BWE-processed / synthetic HF', | |
| 'standard_passes_bypassed': [ | |
| 'dereverb', 'nr_repair', 'f0_revival', | |
| 'formant', 'harmonics', 'spectral', 'sibilant', | |
| ], | |
| }, | |
| } | |
| return result | |
| # ββ Phase 1: Dereverberation ββββββββββββββββββββββββββββββββββββββββββ | |
| if not skip_dereverb: | |
| current_wav = phase1_dereverb(current_wav, samples, st) | |
| else: | |
| L(f'\nββ phase_1_dereverb ββ SKIPPED') | |
| # v2 FIX: Update working samples after each phase | |
| cur_samples, _ = _decode_samples(current_wav) | |
| if cur_samples is None: | |
| cur_samples = samples | |
| # ββ Phase 2: Noise Repair βββββββββββββββββββββββββββββββββββββββββββββ | |
| current_wav = phase2_noise_repair(current_wav, cur_samples, artifacts, st) | |
| cur_samples, _ = _decode_samples(current_wav) | |
| if cur_samples is None: | |
| cur_samples = samples | |
| # ββ Phase 3: F0 Contour Revival βββββββββββββββββββββββββββββββββββββββ | |
| if not skip_f0_revival and artifacts.f0_flat_severity != 'none': | |
| current_wav = phase3_f0_revival(current_wav, cur_samples, artifacts, st) | |
| elif skip_f0_revival: | |
| L(f'\nββ phase_3_f0_revival ββ SKIPPED') | |
| else: | |
| L(f'\nββ phase_3_f0_revival ββ F0 natural, skipping') | |
| cur_samples, _ = _decode_samples(current_wav) | |
| if cur_samples is None: | |
| cur_samples = samples | |
| # ββ Phase 4: Formant Reconstruction βββββββββββββββββββββββββββββββββββ | |
| current_wav = phase4_formant_rebuild(current_wav, cur_samples, artifacts, st) | |
| cur_samples, _ = _decode_samples(current_wav) | |
| if cur_samples is None: | |
| cur_samples = samples | |
| # ββ Phase 5: Harmonic Richness ββββββββββββββββββββββββββββββββββββββββ | |
| current_wav = phase5_harmonic_richness(current_wav, cur_samples, artifacts, st) | |
| cur_samples, _ = _decode_samples(current_wav) | |
| if cur_samples is None: | |
| cur_samples = samples | |
| # ββ Phase 6: Micro-Dynamics βββββββββββββββββββββββββββββββββββββββββββ | |
| current_wav = phase6_micro_dynamics(current_wav, cur_samples, st) | |
| cur_samples, _ = _decode_samples(current_wav) | |
| if cur_samples is None: | |
| cur_samples = samples | |
| # ββ Phase 7: Spectral Repair ββββββββββββββββββββββββββββββββββββββββββ | |
| current_wav = phase7_spectral_repair(current_wav, cur_samples, artifacts, st) | |
| cur_samples, _ = _decode_samples(current_wav) | |
| if cur_samples is None: | |
| cur_samples = samples | |
| # ββ Phase 8: Sibilant Naturalization ββββββββββββββββββββββββββββββββββ | |
| current_wav = phase8_sibilant_naturalization(current_wav, cur_samples, artifacts, st) | |
| cur_samples, _ = _decode_samples(current_wav) | |
| if cur_samples is None: | |
| cur_samples = samples | |
| # ββ Phase 9: Temporal Shaping βββββββββββββββββββββββββββββββββββββββββ | |
| current_wav = phase9_temporal_shaping(current_wav, cur_samples, artifacts, st) | |
| cur_samples, _ = _decode_samples(current_wav) | |
| if cur_samples is None: | |
| cur_samples = samples | |
| # ββ v2: Final re-measurement for "after" tracking βββββββββββββββββββββ | |
| # Re-measure metrics that may have been affected by subsequent phases | |
| if cur_samples is not None and len(cur_samples) > SR: | |
| # Update spectral holes after all processing | |
| if st.spectral_holes_after == st.spectral_holes_before: | |
| # Phase 7 may not have run or may not have updated | |
| st.spectral_holes_after, _ = _detect_spectral_holes(cur_samples) | |
| # Update metallic score after all processing | |
| st.metallic_after = _detect_metallic_ringing(cur_samples) | |
| # Update F0 flatness if not already set | |
| if st.f0_flatness_after == st.f0_flatness_before: | |
| f0_post, _ = _detect_f0(cur_samples) | |
| if len(f0_post) > 10: | |
| st.f0_flatness_after = _compute_f0_flatness(f0_post) | |
| # ββ Phase 10: Tajweed Guards ββββββββββββββββββββββββββββββββββββββββββ | |
| current_wav = phase10_guards(original_samples, current_wav, st) | |
| # ββ Phase 11: Final βββββββββββββββββββββββββββββββββββββββββββββββββββ | |
| final_wav, final_report = phase11_final(current_wav, st) | |
| # ββ v2: Compute naturalness score βββββββββββββββββββββββββββββββββββββ | |
| final_samples, _ = _decode_samples(final_wav) | |
| st.naturalness_score = _compute_naturalness_score(st, artifacts, final_samples) | |
| elapsed = time.time() - t0 | |
| # ββ Summary βββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ | |
| L(f'\n{"β"*60}') | |
| L(f' Ψ§ΩΨ₯ΨΩΨ§Ψ‘ {__version__} β COMPLETE') | |
| L(f' Time: {elapsed:.1f}s') | |
| L(f' Naturalness score: {st.naturalness_score:.1f}/100') | |
| L(f' Phases applied:') | |
| for pname, applied in [ | |
| ('P0-Analysis', st.p0_analyzed), | |
| ('P1-Dereverb', st.p1_dereverb_applied), | |
| ('P2-NR Repair', st.p2_nr_repair_applied), | |
| ('P3-F0 Revival', st.p3_f0_revival), | |
| ('P4-Formant', st.p4_formant_rebuilt), | |
| ('P5-Harmonics', st.p5_harmonic_rich), | |
| ('P6-Dynamics', st.p6_micro_dynamics), | |
| ('P7-Spectral', st.p7_spectral_repair), | |
| ('P8-Sibilant', st.p8_sibilant_nat), | |
| ('P9-Temporal', st.p9_temporal_shaping), | |
| ('P10-Guards', st.p10_guards_ok), | |
| ('P11-Final', st.p11_final_done), | |
| ]: | |
| L(f' {pname}: {"β" if applied else "β"}') | |
| L(f' Artifacts: flatness {st.f0_flatness_before:.3f}β{st.f0_flatness_after:.3f} ' | |
| f'holes {st.spectral_holes_before}β{st.spectral_holes_after} ' | |
| f'metallic {st.metallic_before:.3f}β{st.metallic_after:.3f}') | |
| L(f' Guards: {len(st.guard_pass)} pass / {len(st.guard_warn)} warn / {st.guard_reverts} reverts') | |
| if final_report: | |
| L(f' Final: LUFS={final_report.get("lufs",0):.2f} ' | |
| f'LRA={final_report.get("lra",0):.2f} ' | |
| f'Crest={final_report.get("crest",0):.1f}dB') | |
| L(f'{"β"*60}') | |
| return { | |
| 'engine': 'Ψ§ΩΨ₯ΨΩΨ§Ψ‘', | |
| 'version': __version__, | |
| 'elapsed_s': round(elapsed, 1), | |
| 'naturalness_score': round(st.naturalness_score, 1), | |
| 'aggressive': aggressive, | |
| 'artifacts': { | |
| 'f0_flatness': artifacts.f0_flatness, | |
| 'f0_flat_severity': artifacts.f0_flat_severity, | |
| 'spectral_holes': artifacts.spectral_holes, | |
| 'metallic_score': artifacts.metallic_score, | |
| 'nr_damage': artifacts.nr_damage_score, | |
| 'formant_collapse': artifacts.formant_collapse, | |
| 'sibilant_quality': artifacts.sibilant_quality, | |
| 'overall': artifacts.overall_artificial, | |
| 'diagnosis': artifacts.diagnosis, | |
| 'snr_db': artifacts.snr_db, | |
| 'is_noisy': artifacts.is_noisy, | |
| 'active_bandwidth_hz': artifacts.active_bandwidth_hz, | |
| 'wind_detected': artifacts.wind_detected, | |
| }, | |
| 'improvement': { | |
| 'f0_flatness_before': st.f0_flatness_before, | |
| 'f0_flatness_after': st.f0_flatness_after, | |
| 'spectral_holes_before': st.spectral_holes_before, | |
| 'spectral_holes_after': st.spectral_holes_after, | |
| 'metallic_before': st.metallic_before, | |
| 'metallic_after': st.metallic_after, | |
| }, | |
| 'phases': { | |
| 'dereverb': st.p1_dereverb_applied, | |
| 'nr_repair': st.p2_nr_repair_applied, | |
| 'f0_revival': st.p3_f0_revival, | |
| 'jitter': st.p3_jitter_injected, | |
| 'shimmer': st.p3_shimmer_injected, | |
| 'formant': st.p4_formant_rebuilt, | |
| 'harmonics': st.p5_harmonic_rich, | |
| 'dynamics': st.p6_micro_dynamics, | |
| 'spectral': st.p7_spectral_repair, | |
| 'sibilant': st.p8_sibilant_nat, | |
| 'temporal': st.p9_temporal_shaping, | |
| }, | |
| 'guards': { | |
| 'pass': st.guard_pass, | |
| 'warn': st.guard_warn, | |
| 'reverts': st.guard_reverts, | |
| }, | |
| 'final': final_report, | |
| 'mujawwad_conf': st.mujawwad_conf, | |
| 'f0_median': st.f0_median, | |
| 'telephone_tier': { | |
| 'detected': False, | |
| 'cutoff_hz': 0.0, | |
| 'bwe_applied': False, | |
| }, | |
| } | |
| finally: | |
| _cleanup_all(st) | |
| # ββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ | |
| # CLI | |
| # ββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ | |
| def main() -> int: | |
| if not NUMPY_OK: | |
| print('pip install numpy scipy') | |
| return 1 | |
| p = argparse.ArgumentParser( | |
| description=f'Ψ§ΩΨ₯ΨΩΨ§Ψ‘ {__version__} β Voice Revival Engine β Aetherion Engine-3' | |
| ) | |
| p.add_argument('-i', '--input', required=False, help='Input audio file') | |
| p.add_argument('-o', '--output', required=False, help='Output WAV file') | |
| p.add_argument('--mujawwad', type=float, default=-1.0, | |
| help='Override Mujawwad confidence (0.0=Murattal, 1.0=Mujawwad)') | |
| p.add_argument('--skip-dereverb', action='store_true', | |
| help='Skip dereverberation phase') | |
| p.add_argument('--skip-f0', action='store_true', | |
| help='Skip F0 contour revival (jitter/shimmer/drift)') | |
| p.add_argument('--aggressive', action='store_true', | |
| help='Aggressive mode: wider guards, stronger processing') | |
| p.add_argument('--diagnose-only', action='store_true', | |
| help='Only run Phase 0 analysis, no processing') | |
| args = p.parse_args() | |
| if not args.input: | |
| p.print_help() | |
| return 1 | |
| if not os.path.exists(args.input): | |
| print(f'Input not found: {args.input}') | |
| return 1 | |
| if args.diagnose_only: | |
| # Only analyze | |
| work_wav = os.path.join(_TMP, f'ihya_diag_{os.getpid()}.wav') | |
| if not _decode_to_wav(args.input, work_wav): | |
| print('Failed to decode input') | |
| return 1 | |
| samples, sr = _decode_samples(work_wav) | |
| _cleanup(work_wav) | |
| if samples is None: | |
| print('Failed to decode samples') | |
| return 1 | |
| st = IhyaState() | |
| artifacts = phase0_analyze(samples, sr, st) | |
| print('\n' + json.dumps({ | |
| 'f0_flatness': float(artifacts.f0_flatness), | |
| 'f0_flat_severity': artifacts.f0_flat_severity, | |
| 'spectral_holes': int(artifacts.spectral_holes), | |
| 'metallic_score': float(artifacts.metallic_score), | |
| 'metallic_severity': artifacts.metallic_severity, | |
| 'nr_damage': float(artifacts.nr_damage_score), | |
| 'nr_damage_severity': artifacts.nr_damage_severity, | |
| 'formant_collapse': float(artifacts.formant_collapse), | |
| 'sibilant_quality': float(artifacts.sibilant_quality), | |
| 'overall_artificial': float(artifacts.overall_artificial), | |
| 'diagnosis': artifacts.diagnosis, | |
| 'mujawwad_conf': float(artifacts.mujawwad_conf), | |
| 'f0_median': float(st.f0_median), | |
| 'snr_db': float(artifacts.snr_db), | |
| 'is_noisy': bool(artifacts.is_noisy), | |
| 'active_bandwidth_hz': float(artifacts.active_bandwidth_hz), | |
| 'wind_detected': bool(artifacts.wind_detected), | |
| 'telephone_tier': { | |
| 'detected': bool(artifacts.is_telephone_tier), | |
| 'cutoff_hz': float(artifacts.telephone_cutoff_hz), | |
| 'note': ( | |
| 'BWE required β VoiceFixer Mode 0 pipeline (Β§173)' | |
| if artifacts.is_telephone_tier else 'N/A' | |
| ), | |
| }, | |
| }, indent=2, ensure_ascii=False)) | |
| return 0 | |
| if not args.output: | |
| base = Path(args.input).stem | |
| args.output = str(Path(args.input).parent / f'{base}_ihya.wav') | |
| try: | |
| result = process( | |
| args.input, args.output, | |
| mujawwad_override=args.mujawwad, | |
| skip_dereverb=args.skip_dereverb, | |
| skip_f0_revival=args.skip_f0, | |
| aggressive=args.aggressive, | |
| ) | |
| if 'error' in result: | |
| print(f'\n ERROR: {result["error"]}') | |
| return 2 | |
| art = result.get('artifacts', {}) | |
| imp = result.get('improvement', {}) | |
| tel = result.get('telephone_tier', {}) | |
| print(f'\n ββ Diagnosis: {art.get("diagnosis", "N/A")}') | |
| print(f' ββ Overall artificial: {art.get("overall", 0):.3f}') | |
| print(f' ββ Naturalness score: {result.get("naturalness_score", 0):.1f}/100') | |
| print(f' ββ F0 flatness: {imp.get("f0_flatness_before", 0):.3f} β {imp.get("f0_flatness_after", 0):.3f}') | |
| print(f' ββ Spectral holes: {imp.get("spectral_holes_before", 0)} β {imp.get("spectral_holes_after", 0)}') | |
| print(f' ββ Metallic: {imp.get("metallic_before", 0):.3f} β {imp.get("metallic_after", 0):.3f}') | |
| print(f' ββ SNR: {art.get("snr_db", 0):.1f}dB Bandwidth: {art.get("active_bandwidth_hz", 0):.0f}Hz') | |
| if art.get('wind_detected'): | |
| print(f' ββ Wind noise detected') | |
| if tel.get('detected'): | |
| print(f' ββ β TIER_TELEPHONE: cutoff={tel.get("cutoff_hz", 0):.0f}Hz ' | |
| f'BWE={"β" if tel.get("bwe_applied") else "β (VoiceFixer missing)"} ' | |
| f'AudioSR={"β" if tel.get("audiosr_stage_b") else "β"}') | |
| print(f' ββ β Red Line 18: {tel.get("red_line_18", "")}') | |
| if result.get('aggressive'): | |
| print(f' ββ Mode: AGGRESSIVE') | |
| print(f' ββ Output: {args.output}') | |
| return 0 | |
| except Exception as e: | |
| import traceback | |
| print(f'ERROR: {e}') | |
| traceback.print_exc() | |
| return 1 | |
| if __name__ == '__main__': | |
| sys.exit(main()) | |