Background / engine_ihya_v3.py
carm5333's picture
S201: add engine_ihya_v3.py (replaces isteidad in v11.2 slot)
90a7ac9 verified
Raw
History Blame Contribute Delete
150 kB
#!/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
# ══════════════════════════════════════════════════════════════════════════════
@dataclass
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)
@dataclass
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())