#!/usr/bin/env python3 """ ╔══════════════════════════════════════════════════════════════════════════════╗ ║ ║ ║ SAFI v2.2 | صافي ║ ║ Speech-Adaptive Formant Isolation ║ ║ Arabic: "pure", "clear", "free of impurity" ║ ║ ║ ║ A domain-specific noise reduction algorithm for Quran recitation ║ ║ at extreme low SNR (3–8 dB frame SNR) where DeepFilterNet-3 fails. ║ ║ ║ ╠══════════════════════════════════════════════════════════════════════════════╣ ║ ║ ║ WHY DF3 FAILS AT 3 dB SNR ║ ║ ───────────────────────────────────────────────────────────────────── ║ ║ DeepFilterNet-3 (and all frame-energy NR) estimates speech/noise by ║ ║ comparing a frame's total energy to a learned speech model. At 3 dB ║ ║ SNR the signal and noise are within 1.4× of each other in energy. ║ ║ DF3's decision boundary is below this — it falls back to attenuating ║ ║ everything by ~15 dB uniformly (observed in the Sheikh's mosque file). ║ ║ ║ ║ CORE INSIGHT — Harmonic-Interleaved Noise Estimation (HINE) ║ ║ ───────────────────────────────────────────────────────────────────── ║ ║ Even at 3 dB OVERALL SNR, speech energy is CONCENTRATED at harmonics ║ ║ (k × F0). Noise energy is SPREAD uniformly across all bins. ║ ║ ║ ║ At 3 dB SNR, F0=150 Hz, n_fft=4096, sr=48 kHz: ║ ║ • n_harmonic_bins ≈ 12 harmonics × 5 bins ≈ 60 bins ║ ║ • n_total_bins = 2049 ║ ║ • n_inter_harmonic = 1989 bins (97% of spectrum) ║ ║ • Local SNR, inter-harmonic: ~3 dB (noise-dominated) ║ ║ • Local SNR, at harmonic peak: 10 × log₁₀(2 × 2049/60) ≈ +18 dB ✓ ║ ║ ║ ║ The inter-harmonic bins are a reliable noise reference during VOICED ║ ║ speech, even at 3 dB frame SNR, because speech energy does not appear ║ ║ there. SAFI exploits this to estimate noise PSD without any silence. ║ ║ ║ ║ BUG HISTORY (v1.0 → v2.1) ║ ║ ───────────────────────────────────────────────────────────────────── ║ ║ BUG-1 [CRITICAL] HINE unvoiced frame contamination ║ ║ Arabic fricatives (ص، س، ش) are broadband 2–8 kHz — NOT harmonic. ║ ║ conf=0 → mask=0 → ALL bins accumulate for these frames. ║ ║ Result: noise PSD in sibilant band inflated +33 dB → Wiener filter ║ ║ suppresses Arabic sibilants by 33 dB. Sounds like the Sheikh lost ║ ║ every fricative consonant. ║ ║ Fix: HINE accumulates ONLY from voiced frames (conf > VOICE_THRESH). ║ ║ Unvoiced frames get a fixed conservative noise floor from the ║ ║ voiced-only estimate. Arabic sibilant band (2.5–5 kHz) protected ║ ║ by a separate hard min-gain floor in all frames. ║ ║ ║ ║ BUG-2 SNR gain diagnostic zeroed by RMS normalization ║ ║ SNR gain computed AFTER RMS normalize step → energy restored ║ ║ → |S_enh|² ≈ |S_orig|² → effective_snr_gain_db ≈ 0 always. ║ ║ Fix: capture gain matrix statistics BEFORE RMS normalize. ║ ║ ║ ║ BUG-3 HINE inner loop not vectorized — 6× slower than needed ║ ║ Python for-loop over n_frames (30,000 frames for 5 min file). ║ ║ Fix: vectorize with np.where + nanmean on full matrix at once. ║ ║ ║ ║ BUG-4 Median kernel 110ms — too coarse for Arabic stop consonants ║ ║ Arabic ق/ب/ت closures are 40–60 ms. 110ms kernel smears them. ║ ║ Fix: reduce to 50ms kernel (5 frames) + gain slope limiter. ║ ║ ║ ║ BUG-5 _build_mask_matrix Python loop (n_frames × n_harmonics np.exp) ║ ║ 30,000 frames × 12 harmonics × np.exp(2049) → very slow on Termux. ║ ║ Fix: F0 quantization (1Hz buckets) with precomputed template cache. ║ ║ ║ ║ BUG-6 voiced_thresh not declared in SAFIConfig (v2.0) ║ ║ _f0_frame and _track_f0_sequence accessed it via getattr fallback. ║ ║ Any call to SAFIConfig(voiced_thresh=x) would raise TypeError because ║ ║ @dataclass rejects undeclared keyword arguments. ║ ║ Fix: declared as voiced_thresh: float = 0.25 in SAFIConfig; all ║ ║ getattr fallbacks replaced with direct field access. ║ ║ ║ ║ BUG-7 Gain slope limiter was a Python loop over n_frames (v2.0) ║ ║ 36,000-iteration Python loop (100fps × 360s) with numpy per frame. ║ ║ Contradicted the "fully vectorized" docstring claim. ║ ║ Fix: log-domain cumsum — converts dB-per-frame clamp into diff→ ║ ║ clip→cumsum on the full (n_freqs × n_frames) matrix. No Python loop. ║ ║ Equivalent to sequential clamp after median filtering (smooth signal). ║ ║ ║ ║ ║ ║ BUG-10 Fixed harmonic_bw_hz=55Hz ignores F0 — Gaussians overlap at low F0 ║ ║ The HINE gap condition requires bw < F0/2 / √(2·ln(1/thresh)). ║ ║ At thresh=0.05: bw < F0 × 0.204. With fixed bw=55Hz: ║ ║ F0=207Hz → 55Hz > 207 × 0.204 = 42Hz → Gaussians OVERLAP at midpoint ║ ║ mask at H1-H2 midpoint = 0.34 >> 0.05 → entire voice band in HINE ║ ║ HINE noise estimate = mean of 3–24kHz near-silence = −62 dBFS ║ ║ Wiener SNR >> 1 everywhere → gain = 1.0 → no suppression. ║ ║ Note: user-suggested factor 0.35 also fails — min(55, 207×0.35)=55Hz ║ ║ (the min never kicks in below F0=157Hz). Correct factor must be < 0.204. ║ ║ Fix: adaptive bw = max(15, min(harmonic_bw_hz, F0 × harmonic_bw_f0_factor))║ ║ harmonic_bw_f0_factor=0.18 (12% below 0.204 ceiling, safety margin). ║ ║ F0=207Hz → bw=37Hz. F0=150Hz → bw=27Hz. F0=350Hz → capped at 55Hz. ║ ║ All 12 harmonics pass gap condition (mask@mid ≤ 0.045 < 0.05). ║ ║ Floor at 15Hz prevents sub-bin widths at very low F0 (< 85Hz). ║ ║ ║ ║ ALGORITHM PIPELINE ║ ║ ───────────────────────────────────────────────────────────────────── ║ ║ 1. F0 TRACKING NAC (normalized autocorr) per 40ms frame ║ ║ + parabolic sub-sample interpolation ║ ║ + temporal continuity constraint ║ ║ 2. HARMONIC MASK Soft Gaussians at k × F0, cached by quantized F0 ║ ║ weight = conf/√k, width slightly wider at high k ║ ║ 3. HINE Voiced-only inter-harmonic noise PSD accumulation ║ ║ + spectral interpolation over harmonic bins ║ ║ + Gaussian smoothing (noise is spectrally smooth) ║ ║ 4. WIENER FILTER Harmonic-guided min-gain per bin/frame ║ ║ + Arabic sibilant (2.5–5 kHz) hard protection floor ║ ║ + vectorized over all frames simultaneously ║ ║ 5. TEMPORAL SMOOTH Median filter 50ms + gain slope limiter ║ ║ 6. ITERATIVE Pass 2: re-track F0 on partially cleaned signal, ║ ║ re-estimate noise, apply ONE final filter on ║ ║ original STFT (no double-attenuation) ║ ║ ║ ║ INTEGRATION IN الاسترداد ║ ║ ───────────────────────────────────────────────────────────────────── ║ ║ frame_snr < 2.5 dB → TIER_UNPROCESSABLE: EQ only, skip all NR ║ ║ frame_snr 2.5–8 dB → SAFI runs (replaces DF3 in this zone) ║ ║ frame_snr 8–12 dB → DF3 only (existing path, unmodified) ║ ║ frame_snr ≥ 12 dB → no NR needed ║ ║ ║ ║ Dependencies: numpy, scipy only. No deep learning. Termux/ARM safe. ║ ║ Python 3.8+. pip install numpy scipy ║ ║ ║ ║ Usage: python safi_nr.py -i input.wav -o output.wav ║ ║ Patch: python safi_nr.py --patch-doc ║ ║ ║ ║ Named stronger than DF3 because it exploits speech structure ║ ║ (harmonics) that DF3's learned model cannot use below its SNR floor. ║ ║ ║ ║ وما التوفيق إلا بالله ║ ╚══════════════════════════════════════════════════════════════════════════════╝ """ from __future__ import annotations import os import subprocess import sys import tempfile import warnings from dataclasses import dataclass, field from typing import Dict, Optional, Tuple warnings.filterwarnings("ignore") import numpy as np from scipy.fft import rfft, irfft, rfftfreq from scipy.ndimage import gaussian_filter1d, median_filter from scipy.interpolate import interp1d _TMP = tempfile.gettempdir() # ══════════════════════════════════════════════════════════════════════════════ # CONFIGURATION # ══════════════════════════════════════════════════════════════════════════════ @dataclass class SAFIConfig: """SAFI v2.1 algorithm parameters. Defaults calibrated for Arabic male Quran recitation at 48 kHz. Sheikh Al-Dosari profile: F0 ≈ 130–180 Hz, voiced ratio ≈ 70–85%. Voice type presets ────────────────── Male reciter (default): f0_min=70, f0_max=250 Female reciter: f0_min=150, f0_max=380 Deep bass: f0_min=60, f0_max=180 Child: f0_min=200, f0_max=500 """ # ── Audio ───────────────────────────────────────────────────────────────── sr: int = 48_000 # must match الاسترداد engine SR constant # ── STFT ────────────────────────────────────────────────────────────────── # n_fft=4096 → bin width = sr/n_fft = 11.7 Hz at 48kHz # This gives ≥6 bins between F0=70Hz harmonics (70/11.7 ≈ 6), # enough for reliable HINE inter-harmonic noise sampling. n_fft: int = 4096 hop_length: int = 480 # 10ms hop at 48kHz — 100 frames/second # ── F0 Tracking ─────────────────────────────────────────────────────────── f0_min: float = 70.0 # Hz — deepest bass reciters f0_max: float = 380.0 # Hz — female / child reciters # NAC peak height threshold for voiced/unvoiced decision. # Lower than YIN's canonical 0.5 for clean speech because at 3 dB SNR, # noise contributes ~50% of r[0], halving the normalized ACF peak height. # Distinct from hine_voice_thresh (0.10) — that gates HINE accumulation; # this gates whether a frame is declared voiced at all. voiced_thresh: float = 0.25 # F0 quantization bucket size. Smaller = more accurate, more cache entries. # 1Hz buckets: at F0=150Hz, 1Hz difference → harmonic peak shift of # k×1Hz per harmonic. At k=12, shift = 12Hz < 1 bin (11.7Hz) — acceptable. f0_quant_hz: float = 1.0 # ── Harmonic Structure ──────────────────────────────────────────────────── n_harmonics: int = 12 # cover k=1..12 # Gaussian σ per harmonic. 1 bin = 11.7 Hz at 48kHz/4096. # 55Hz ≈ 4.7 bins: wide enough to account for vibrato (±2Hz) and # room-resonance-induced spectral spreading of harmonics. # BUG-10 FIX: harmonic_bw_hz is now only an UPPER CAP. # Effective bw = max(15, min(harmonic_bw_hz, F0 * harmonic_bw_f0_factor)). # This ensures Gaussians do NOT overlap at the H1-H2 midpoint (and all # higher harmonics) at low F0, which is required for HINE to see genuine # inter-harmonic bins in the voice band (80–2500 Hz). harmonic_bw_hz: float = 55.0 # upper cap only (applies at high F0) # BUG-10 FIX: Fraction of F0 used as Gaussian σ. # Must be < 0.204 (= 0.5 / √(2·ln(1/hine_inter_thresh))) to guarantee a # gap between every harmonic pair at ALL voiced F0 values. # 0.18 gives 12% margin below the 0.204 ceiling: # F0=207Hz → bw=37.3Hz (3.2 STFT bins) mask@mid=0.021 ✓ # F0=150Hz → bw=27.0Hz (2.3 bins) mask@mid=0.021 ✓ # F0=100Hz → bw=18.0Hz (1.5 bins) mask@mid=0.021 ✓ # F0=350Hz → bw=55.0Hz (capped) gap OK (wider spacing) harmonic_bw_f0_factor: float = 0.18 # Width growth factor: higher harmonics are slightly wider in practice. harmonic_width_growth: float = 0.04 # σ *= (1 + 0.04*(k-1)) # ── HINE (BUG-1 FIX) ────────────────────────────────────────────────────── # Only frames with voicing confidence > this contribute to noise estimation. # Prevents Arabic fricatives from inflating the sibilant noise PSD. hine_voice_thresh: float = 0.10 # Bins with harmonic mask < this are considered inter-harmonic (noise-dominant). # BUG-8 FIX: lowered from 0.15 → 0.05. # At H7 (weight=1/√7=0.378), Gaussian wings 100 Hz from center fall to # mask≈0.13 < 0.15, so those bins entered HINE as "noise" while still # carrying harmonic energy. 0.05 = 0.15 × 0.378 (proportional to H7 weight), # ensuring only truly inter-harmonic bins (< 5% of H1 peak) feed HINE. hine_inter_thresh: float = 0.05 # ── Wiener Filter ───────────────────────────────────────────────────────── # β > 1: oversubtraction (removes more noise at cost of slight musical noise). # 1.5 is the empirically balanced value for mosque recordings at 3–8dB SNR. wiener_beta: float = 1.5 # Minimum gain in inter-harmonic bins: 0.07 ≈ −23 dB. # Preserves natural room ambience (absolute silence sounds unnatural for Quran). wiener_floor: float = 0.07 # Minimum gain at confirmed harmonic bins: 0.70 ≈ −3 dB. # Near-transparent in harmonics — speech is preserved, noise only slightly reduced. harmonic_floor: float = 0.70 # ── Arabic Sibilant Protection (BUG-1 FIX) ──────────────────────────────── # Hard minimum gain for all frames in the Arabic fricative range (2.5–5 kHz). # This band contains: ص /s'/, س /s/, ش /sh/, ز /z/, ث /th/ energy. # Even in unvoiced frames where no harmonic mask applies, we refuse to suppress # this band below this floor — protecting these phonemes unconditionally. sibilant_lo_hz: float = 2500.0 sibilant_hi_hz: float = 5000.0 sibilant_floor: float = 0.50 # 0.50 ≈ −6 dB — audible but reduced # ── Temporal Smoothing (BUG-4 FIX) ─────────────────────────────────────── # Median filter kernel: 5 frames = 50ms at 10ms/hop. # Arabic stop closures (ق،ب،ت): 40–60ms → 50ms kernel just fits. # Was 11 frames (110ms) in v1.0 → smeared Arabic stops into adjacent vowels. smooth_frames: int = 5 # kernel size = smooth_frames (must be odd) # Gain slope limiter: maximum gain CHANGE per frame (dB). # Prevents sudden gain jumps at voiced/unvoiced boundaries. # 3 dB/frame = 300 dB/s — much faster than natural dynamics (prevents artifacts) # but prevents instantaneous flips that cause clicks. gain_slope_limit_db: float = 3.0 # ── Iterative Refinement ────────────────────────────────────────────────── # Pass 2 re-tracks F0 on pass-1 output → better F0 → better noise estimate. # ONE final filter applied on original STFT (no double-attenuation). # 2 passes sufficient. 3+ gives diminishing returns. n_passes: int = 2 # ── Early-Exit Thresholds ───────────────────────────────────────────────── tier_unprocessable_snr_db: float = 2.5 # absolute NR floor min_voiced_ratio: float = 0.06 # <6% voiced → UNPROCESSABLE @dataclass class SAFIResult: """SAFI processing diagnostics.""" applied: bool = False status: str = "NOT_RUN" reason: str = "" voiced_ratio: float = 0.0 median_f0_hz: float = 0.0 harmonic_coverage: float = 0.0 noise_floor_db: float = 0.0 mean_gain_db: float = 0.0 # BUG-2 FIX: now captured before RMS normalize effective_snr_gain_db: float = 0.0 # BUG-2 FIX: computed from gain matrix voiced_frames_used: int = 0 # BUG-1 FIX: how many voiced frames in HINE n_passes: int = 0 # Gate constants used by engine integration SAFI_FRAME_SNR_GATE_DB = 8.0 # SAFI zone: frame_snr < 8dB TIER_UNPROCESSABLE_SNR = 2.5 # Absolute floor below all NR # ══════════════════════════════════════════════════════════════════════════════ # SAFI DENOISER # ══════════════════════════════════════════════════════════════════════════════ class SAFIDenoiser: """ SAFI v2.1 — Speech-Adaptive Formant Isolation. All computation is numpy/scipy float32/float64. File I/O uses ffmpeg pipe — no soundfile dependency (Termux-safe). Thread-safe: create one instance per worker. No shared state between calls. """ def __init__(self, cfg: SAFIConfig = None): self.cfg = cfg or SAFIConfig() c = self.cfg # Precompute frequency axis (aligns with STFT rfft output) self.freqs = rfftfreq(c.n_fft, 1.0 / c.sr).astype(np.float64) self.n_freqs = len(self.freqs) # Precompute sibilant band mask (static — doesn't change with F0) self._sib_mask = ( (self.freqs >= c.sibilant_lo_hz) & (self.freqs <= c.sibilant_hi_hz) ) # BUG-5 FIX: Harmonic mask template cache keyed by quantized F0. # A template is the sum of k Gaussians for k=1..n_harmonics at 1Hz # of F0, all at unit confidence. Scaled by conf at lookup time. # Cache size: (f0_max - f0_min) / f0_quant_hz buckets ≈ 310 entries. self._mask_cache: Dict[int, np.ndarray] = {} # ───────────────────────────────────────────────────────────────────────── # STAGE 1 — F0 TRACKING # ───────────────────────────────────────────────────────────────────────── def _f0_frame(self, frame: np.ndarray) -> Tuple[float, float]: """ F0 estimation via Normalized Autocorrelation (NAC). Why NAC instead of FFT pitch? • At 3 dB SNR, harmonic peaks in the magnitude spectrum are barely above the noise floor. FFT pitch detectors see many false candidates. • NAC integrates across ALL harmonics simultaneously — the periodicity of k harmonics all at the same F0 amplifies the ACF peak by k×. • ACF normalization by r[0] makes the peak height ∈ [0,1] regardless of frame energy — essential for stable voiced/unvoiced decisions at varying noise levels. Normalization: r[0] is total signal power (noise + speech). At 3 dB SNR, noise contributes half of r[0]. This means the ACF peak height is still reduced by 50% compared to clean speech. voiced_thresh=0.25 accounts for this — it was 0.5 for clean speech in YIN algorithm. Returns (f0_hz, confidence) f0_hz = 0 means unvoiced or uncertain. confidence ∈ [0, 1], mapped from NAC peak height. """ N = len(frame) if N < 256: return 0.0, 0.0 windowed = frame.astype(np.float64) * np.hanning(N) # ACF via zero-padded FFT (avoids circular wrap-around) X = rfft(windowed, n=2 * N) acf = irfft(X * np.conj(X))[:N].real # Normalize by lag-0 (total power) r0 = acf[0] if r0 < 1e-12: return 0.0, 0.0 acf /= r0 # Lag search bounds lag_min = max(2, int(self.cfg.sr / self.cfg.f0_max)) lag_max = min(N // 2 - 1, int(self.cfg.sr / self.cfg.f0_min)) if lag_min >= lag_max: return 0.0, 0.0 region = acf[lag_min:lag_max] peak_idx = int(np.argmax(region)) peak_val = float(region[peak_idx]) # Confidence: linear map from voiced_thresh → 0.0, saturates at 0.75 → 1.0 _vt = self.cfg.voiced_thresh if peak_val < _vt: return 0.0, float(np.clip(peak_val, 0.0, 1.0)) # Parabolic sub-sample interpolation (standard formula) # For peak at index p: vertex offset = (r[p-1] - r[p+1]) / (2*(r[p-1] - 2*r[p] + r[p+1])) lag = float(lag_min + peak_idx) if 0 < peak_idx < len(region) - 1: a = float(region[peak_idx - 1]) b = float(region[peak_idx]) c = float(region[peak_idx + 1]) denom = a - 2.0 * b + c # always ≤ 0 at a peak if abs(denom) > 1e-10: # offset ∈ (-0.5, +0.5): sub-sample refinement lag += 0.5 * (a - c) / denom f0 = self.cfg.sr / max(lag, 1.0) # Bounds check after sub-sample correction if not (self.cfg.f0_min <= f0 <= self.cfg.f0_max): return 0.0, 0.0 # Confidence: linear map from voiced_thresh → 0.0, saturates at 0.75 → 1.0 conf = float(np.clip((peak_val - _vt) / max(0.75 - _vt, 0.01), 0.0, 1.0)) return float(f0), conf def _track_f0_sequence(self, audio: np.ndarray) -> Tuple[np.ndarray, np.ndarray]: """ F0 tracking aligned to STFT frame boundaries. Analysis window: 40ms (≥3 full periods at F0_min=70Hz → 43ms period). This is longer than the STFT frame (85ms) but centered on each frame's midpoint — the 40ms window gives better periodicity evidence. Post-processing: 1. Abrupt F0 jumps (>±28%) → reduce confidence by 60%. Tajweed melisma: smooth glides, no discontinuous jumps. 2. Single unvoiced frame between voiced → linearly interpolate F0. Short consonant closures (ب, ت, د) at 20–40ms = 2–4 frames. 3. Short voiced runs (< 3 frames = 30ms) → zero confidence. Prevents noise burst from triggering harmonic masking. """ c = self.cfg hop = c.hop_length f0_win = min(int(0.040 * c.sr), c.n_fft) half = f0_win // 2 vt = c.voiced_thresh n_frames = 1 + max(0, (len(audio) - c.n_fft) // hop) f0 = np.zeros(n_frames, dtype=np.float64) conf = np.zeros(n_frames, dtype=np.float64) for t in range(n_frames): center = t * hop + c.n_fft // 2 s = max(0, center - half) e = min(len(audio), center + half) frame = audio[s:e] if len(frame) < 256: continue if len(frame) < f0_win: frame = np.pad(frame, (0, f0_win - len(frame))) f0[t], conf[t] = self._f0_frame(frame) # ── 1. Temporal continuity: penalize abrupt F0 jumps ───────────────── for t in range(1, n_frames - 1): if f0[t] > 0 and f0[t - 1] > 0: ratio = f0[t] / (f0[t - 1] + 1e-10) if not (0.78 <= ratio <= 1.28): conf[t] *= 0.40 # ── 2. Single-frame gap filling ─────────────────────────────────────── # A short unvoiced gap between voiced regions is interpolated. for t in range(1, n_frames - 1): if f0[t] < c.f0_min and f0[t - 1] > 0 and f0[t + 1] > 0: f0[t] = 0.5 * (f0[t - 1] + f0[t + 1]) conf[t] = 0.18 # conservative — interpolated frame # ── 3. Remove very short voiced runs (noise bursts) ─────────────────── # Run-length encode voiced (f0>0) regions. Runs < 3 frames → unvoice. in_run = False; run_start = 0 for t in range(n_frames): if f0[t] > 0: if not in_run: in_run = True; run_start = t else: if in_run: run_len = t - run_start if run_len < 3: f0[run_start:t] = 0.0 conf[run_start:t] = 0.0 in_run = False if in_run and (n_frames - run_start) < 3: f0[run_start:] = 0.0 conf[run_start:] = 0.0 return f0.astype(np.float32), conf.astype(np.float32) # ───────────────────────────────────────────────────────────────────────── # STAGE 2 — HARMONIC MASK (BUG-5 FIX: cached templates) # ───────────────────────────────────────────────────────────────────────── def _get_mask_template(self, f0_quant: int) -> np.ndarray: """ Return cached harmonic mask template for quantized F0 (in integer Hz). Template = sum of n_harmonics Gaussians at k × f0_quant, k=1..N. Weight = 1/√k (higher harmonics have less energy). Stored at unit confidence — caller scales by actual conf. Cache miss: ~0.5ms to compute. Cache hit: <1µs. For a 5-min file with F0 smoothly varying 130–180Hz: ≈ 50 unique quantized F0 values → 50 cache misses, rest are hits. """ if f0_quant in self._mask_cache: return self._mask_cache[f0_quant] c = self.cfg template = np.zeros(self.n_freqs, dtype=np.float64) f0 = float(f0_quant) for k in range(1, c.n_harmonics + 1): hf = k * f0 if hf >= c.sr / 2.0: break weight = 1.0 / np.sqrt(float(k)) # BUG-10 FIX: adaptive base bw = max(15, min(cap, F0 * factor)). # Guarantees a genuine inter-harmonic gap at the midpoint between # every harmonic pair: mask@midpoint < hine_inter_thresh=0.05 # across the full F0 range (85–380 Hz). base_bw = max(15.0, min(c.harmonic_bw_hz, f0 * c.harmonic_bw_f0_factor)) # Width slightly increases for higher harmonics (spectral broadening) bw = base_bw * (1.0 + c.harmonic_width_growth * float(k - 1)) # Vectorized Gaussian: no per-bin loop g = weight * np.exp(-0.5 * ((self.freqs - hf) / bw) ** 2) template = np.maximum(template, g) template = np.clip(template, 0.0, 1.0).astype(np.float32) self._mask_cache[f0_quant] = template return template def _build_mask_matrix( self, f0_seq: np.ndarray, # (n_f0_frames,) float32, Hz; 0 = unvoiced conf_seq: np.ndarray, # (n_f0_frames,) float32, confidence n_stft: int, # target STFT n_frames ) -> np.ndarray: """ Build harmonic mask matrix (n_freqs × n_stft_frames). BUG-5 FIX: Uses precomputed/cached template per quantized F0. For each frame: mask[:, t] = template(quantize(f0[t])) × conf[t] This replaces n_frames × n_harmonics np.exp calls with n_frames × one template lookup + one scalar multiply. The length mismatch between f0_seq and n_stft (boundary effects in the scipy STFT) is handled by linear interpolation. """ q = self.cfg.f0_quant_hz # Align F0/conf sequence to STFT frame count if len(f0_seq) != n_stft: xs = np.linspace(0.0, 1.0, len(f0_seq)) xd = np.linspace(0.0, 1.0, n_stft) f0_seq = np.interp(xd, xs, f0_seq).astype(np.float32) conf_seq = np.interp(xd, xs, conf_seq).astype(np.float32) masks = np.zeros((self.n_freqs, n_stft), dtype=np.float32) for t in range(n_stft): f0_t = float(f0_seq[t]) conf_t = float(conf_seq[t]) if f0_t < self.cfg.f0_min or conf_t < 0.05: continue # unvoiced frame: mask stays zero # Quantize F0 to nearest bucket f0_quant = int(round(f0_t / q) * q) f0_quant = int(np.clip(f0_quant, self.cfg.f0_min, self.cfg.f0_max)) template = self._get_mask_template(f0_quant) masks[:, t] = np.clip(template * conf_t, 0.0, 1.0) return masks # ───────────────────────────────────────────────────────────────────────── # STAGE 3 — HINE NOISE ESTIMATION (BUG-1 + BUG-3 FIX) # ───────────────────────────────────────────────────────────────────────── def _hine( self, S_mag: np.ndarray, # (n_freqs, n_frames) masks: np.ndarray, # (n_freqs, n_frames) conf_seq: np.ndarray, # (n_frames,) — voicing confidence per frame ) -> Tuple[np.ndarray, int]: """ Harmonic-Interleaved Noise Estimation — HINE. BUG-1 FIX: Voiced-only accumulation ───────────────────────────────────── Only frames where conf > hine_voice_thresh contribute to the noise PSD estimate. This is the critical fix. Arabic fricatives (ص، س، ش、ز، ث) are unvoiced (conf ≈ 0) and have broadband energy concentrated in 2–8 kHz. If we include unvoiced frames in HINE, their energy inflates the noise PSD estimate in the sibilant band. The Wiener filter then suppresses that band by 30+ dB, destroying every Arabic fricative phoneme. By restricting HINE to voiced frames only: • HINE sees only the inter-harmonic bins of voiced speech, which genuinely contain only noise (by the HINE theorem). • The sibilant band stays uncontaminated. • Unvoiced frames (fricatives) get the same noise PSD as voiced frames — a slight over-estimate in the sibilant band, but the Arabic sibilant hard floor (sibilant_floor=0.50) prevents over-suppression. BUG-3 FIX: Vectorized accumulation ──────────────────────────────────── Instead of a Python loop per frame: for t: power_sum[mask[:,t] voice_thresh) noise_psd = nanmean(where(inter_mat, S_mag**2, nan), axis=1) Returns (noise_psd [n_freqs], n_voiced_frames_used) """ c = self.cfg # Align conf_seq to matrix frame count n_freqs, n_frames = S_mag.shape if len(conf_seq) != n_frames: xs = np.linspace(0.0, 1.0, len(conf_seq)) xd = np.linspace(0.0, 1.0, n_frames) conf_seq = np.interp(xd, xs, conf_seq).astype(np.float32) # Voiced frame mask: shape (n_frames,) voiced_frames = conf_seq > c.hine_voice_thresh n_voiced = int(voiced_frames.sum()) # Build 2D inter-harmonic mask: # True where bin is inter-harmonic AND frame is voiced # Shape: (n_freqs, n_frames) inter_mask = (masks < c.hine_inter_thresh) # (n_freqs, n_frames) voiced_row = voiced_frames[np.newaxis, :] # (1, n_frames) hine_mask = inter_mask & voiced_row # (n_freqs, n_frames) # BUG-3 FIX: Vectorized power accumulation # np.where → nan at excluded cells → nanmean ignores them power_mat = np.where(hine_mask, S_mag.astype(np.float64) ** 2, np.nan) noise_psd = np.nanmean(power_mat, axis=1) # (n_freqs,) # Fill NaN (bins that were ALWAYS harmonic or always unvoiced) valid = np.isfinite(noise_psd) if valid.sum() < 20: # Very few usable bins: fallback to global percentile estimate all_power = S_mag.astype(np.float64) ** 2 noise_psd = np.percentile(all_power, 5, axis=1) # bottom 5% per bin elif not np.all(valid): # Interpolate over missing bins try: f_valid = self.freqs[valid] psd_valid = noise_psd[valid] f_interp = interp1d( f_valid, psd_valid, kind="linear", bounds_error=False, fill_value=(float(psd_valid[0]), float(psd_valid[-1])), ) interp_vals = f_interp(self.freqs) noise_psd[~valid] = interp_vals[~valid] except Exception: # Nearest-neighbour fallback valid_idx = np.where(valid)[0] for i in np.where(~valid)[0]: nn = valid_idx[np.argmin(np.abs(valid_idx - i))] noise_psd[i] = noise_psd[nn] # Gaussian smoothing (real noise PSDs are spectrally smooth) noise_psd = gaussian_filter1d(noise_psd, sigma=4.0) # Hard floor (prevents divide-by-zero in Wiener filter) noise_psd = np.maximum(noise_psd, 1e-14).astype(np.float32) return noise_psd, n_voiced # ───────────────────────────────────────────────────────────────────────── # STAGE 4+5 — HARMONIC-GUIDED WIENER FILTER + TEMPORAL SMOOTHING # ───────────────────────────────────────────────────────────────────────── def _wiener_filter( self, S_complex: np.ndarray, # (n_freqs, n_frames) complex64 noise_psd: np.ndarray, # (n_freqs,) float32 — HINE noise estimate masks: np.ndarray, # (n_freqs, n_frames) float32 — harmonic mask ) -> Tuple[np.ndarray, np.ndarray]: """ Harmonic-guided Wiener filter with Arabic sibilant protection. Wiener gain formula (amplitude domain) ──────────────────────────────────────── SNR(f,t) = |S(f,t)|² / N(f) (instantaneous SNR per bin) W_gain(f,t)= sqrt( max(SNR − β, ε²) / SNR ) Final gain = max( W_gain, min_gain_floor(f, t) ) where β = wiener_beta (oversubtraction), ε = wiener_floor. Harmonic-blended floor ────────────────────── min_gain(f, t) = wiener_floor + (harmonic_floor − wiener_floor) × mask(f,t) • At mask=0 (inter-harmonic): floor = wiener_floor (0.07, −23 dB) • At mask=1 (harmonic peak): floor = harmonic_floor (0.70, −3 dB) Arabic sibilant hard floor (BUG-1 FIX, part 2) ───────────────────────────────────────────────── For all bins in [sibilant_lo, sibilant_hi] (2.5–5 kHz): min_gain ≥ sibilant_floor (0.50, −6 dB) unconditionally. Applied BEFORE harmonic floor blending, so harmonic peaks in the sibilant range get the higher of (harmonic_floor, sibilant_floor). Temporal smoothing (BUG-4 FIX) ──────────────────────────────── 1. Median filter: kernel=5 frames (50ms) — was 11 frames in v1.0. 50ms is small enough to preserve Arabic stop closures (40–60ms). 2. Gain slope limiter: |gain[f,t] - gain[f,t-1]| ≤ 3 dB/frame. Prevents discontinuous gain jumps at voiced/unvoiced boundaries without smoothing over genuine transients. Implementation: fully vectorized over all frames simultaneously. Wiener gain, sibilant floor, harmonic blending, median filter, and slope limiter are all numpy matrix operations — no Python frame loop. Returns (S_enhanced, gain_matrix). gain_matrix is captured BEFORE RMS normalization (BUG-2 FIX). """ c = self.cfg S_mag = np.abs(S_complex).astype(np.float64) S_phase = np.angle(S_complex).astype(np.float32) n_freqs, n_frames = S_mag.shape # ── Vectorized SNR and Wiener gain (all frames at once) ─────────────── # noise_col: (n_freqs, 1) broadcast over frames noise_col = noise_psd[:, np.newaxis].astype(np.float64) psd = S_mag ** 2 # (n_freqs, n_frames) snr = psd / (noise_col + 1e-14) # instantaneous per-bin SNR # Wiener gain numerator: clip to [ε², 1.0] (never above 1.0) eps_sq = float(c.wiener_floor) ** 2 numerator = np.clip(snr - c.wiener_beta, eps_sq, 1.0) w_gain = np.sqrt(numerator / (snr + 1e-14)) # ── Arabic sibilant hard floor ───────────────────────────────────────── # Apply BEFORE harmonic blending. For sibilant bins, raise the floor # to sibilant_floor regardless of mask or Wiener estimate. sib_col = self._sib_mask.astype(np.float64)[:, np.newaxis] # (n_freqs, 1) w_gain = np.where( sib_col > 0, np.maximum(w_gain, c.sibilant_floor), w_gain, ) # ── Harmonic-blended min-gain floor ──────────────────────────────────── min_gain = (c.wiener_floor + (c.harmonic_floor - c.wiener_floor) * masks.astype(np.float64)) gain_matrix = np.maximum(w_gain, min_gain).astype(np.float32) # ── Temporal smoothing: median filter (BUG-4 FIX: kernel=5) ─────────── # kernel must be odd; smooth_frames is already the full kernel size kernel = c.smooth_frames if c.smooth_frames % 2 == 1 else c.smooth_frames + 1 gain_matrix = median_filter( gain_matrix.astype(np.float64), size=(1, kernel), mode="reflect", ).astype(np.float32) # ── Gain slope limiter — vectorized in log domain ────────────────────── # Working in log space converts the max-ratio-per-frame constraint into # a simple clamp on first differences. cumsum then reconstructs the # constrained sequence from t=0. # # This is equivalent to the sequential clamp after median filtering # because: (a) the median filter has already removed large single-frame # jumps, so consecutive frames needing clamping are rare; (b) the only # case where they differ — a sustained high gain following a clamped # rise — results in holding the gain slightly lower during recovery, # which is perceptually inaudible at 10ms resolution and bounded to # < gain_slope_limit_db per frame. No Python loop over n_frames. log_limit = c.gain_slope_limit_db * (np.log(10.0) / 20.0) # dB → nepers log_gain = np.log(gain_matrix.astype(np.float64) + 1e-12) # (n_freqs, n_frames) diff = np.diff(log_gain, axis=1) # (n_freqs, n_frames-1) diff_clipped = np.clip(diff, -log_limit, log_limit) log_gain_new = np.empty_like(log_gain) log_gain_new[:, 0] = log_gain[:, 0] log_gain_new[:, 1:] = log_gain[:, 0:1] + np.cumsum(diff_clipped, axis=1) gain_matrix = np.exp(log_gain_new).astype(np.float32) # Clip to [wiener_floor, 1.0] gain_matrix = np.clip(gain_matrix, c.wiener_floor, 1.0) # Apply gain to complex STFT (magnitude × gain, phase preserved) S_enhanced = (gain_matrix * S_mag * np.exp(1j * S_phase)).astype(np.complex64) return S_enhanced, gain_matrix # ───────────────────────────────────────────────────────────────────────── # STFT / ISTFT # ───────────────────────────────────────────────────────────────────────── def _stft(self, audio: np.ndarray) -> np.ndarray: """scipy STFT → (n_freqs, n_frames) complex64.""" from scipy.signal import stft as _scipy_stft _, _, S = _scipy_stft( audio.astype(np.float64), fs=self.cfg.sr, window="hann", nperseg=self.cfg.n_fft, noverlap=self.cfg.n_fft - self.cfg.hop_length, nfft=self.cfg.n_fft, padded=True, boundary="zeros", ) return S.astype(np.complex64) def _istft(self, S: np.ndarray, n_samples: int) -> np.ndarray: """scipy ISTFT → float32, trimmed/padded to n_samples.""" from scipy.signal import istft as _scipy_istft _, y = _scipy_istft( S.astype(np.complex128), fs=self.cfg.sr, window="hann", nperseg=self.cfg.n_fft, noverlap=self.cfg.n_fft - self.cfg.hop_length, nfft=self.cfg.n_fft, ) y = y.real.astype(np.float32) if len(y) > n_samples: return y[:n_samples] if len(y) < n_samples: return np.pad(y, (0, n_samples - len(y))) return y # ───────────────────────────────────────────────────────────────────────── # MAIN PROCESS # ───────────────────────────────────────────────────────────────────────── def process(self, audio: np.ndarray) -> Tuple[np.ndarray, SAFIResult]: """ SAFI v2.1 full pipeline on mono float32 audio at self.cfg.sr. Pure algorithm — no file I/O. Use process_file() for WAV-in/WAV-out. """ c = self.cfg n_samples = len(audio) original = audio.copy() # ── Stage 0: Fast voiced pre-check ──────────────────────────────────── # Sample ~40 frames distributed across the file to estimate voiced ratio. # Avoids full STFT on clearly unprocessable files (music, traffic noise). f0_win = min(int(0.040 * c.sr), c.n_fft) n_check = min(40, max(5, n_samples // int(0.5 * c.sr))) step = max(f0_win, n_samples // n_check) n_voiced = sum( 1 for i in range(0, n_samples - f0_win, step) if self._f0_frame(audio[i:i + f0_win])[0] > 0 and self._f0_frame(audio[i:i + f0_win])[1] > 0.20 ) voiced_pre = n_voiced / max(1, n_check) if voiced_pre < c.min_voiced_ratio: return original, SAFIResult( status="UNPROCESSABLE", reason=f"voiced_pre={voiced_pre:.2f} < {c.min_voiced_ratio}", voiced_ratio=voiced_pre, ) # ── Stage 1: F0 tracking ────────────────────────────────────────────── f0_seq, conf_seq = self._track_f0_sequence(audio) voiced_ratio = float(np.mean(conf_seq > 0.10)) f0_voiced = f0_seq[f0_seq > c.f0_min] median_f0 = float(np.median(f0_voiced)) if len(f0_voiced) else 0.0 # ── Stage 2: STFT ───────────────────────────────────────────────────── S = self._stft(audio) n_freqs, n_frames = S.shape S_mag_ref = np.abs(S).astype(np.float32) # ── Stage 3: Harmonic mask ──────────────────────────────────────────── masks = self._build_mask_matrix(f0_seq, conf_seq, n_frames) # BUG-8 FIX: was masks > 0.5, which only counted bins at H1 peak # (H7 peak ≈ 0.38 < 0.5 → never counted). Use hine_inter_thresh as the # boundary: "harmonic" = bins the filter actually treats as harmonic. hm_coverage = float(np.mean(masks > c.hine_inter_thresh)) # ── Stage 4: HINE (voiced-only, vectorized) ─────────────────────────── noise_psd_p1, n_voiced_used_p1 = self._hine(S_mag_ref, masks, conf_seq) # ── Stage 5: Wiener filter — Pass 1 ────────────────────────────────── S_enh_p1, gain_p1 = self._wiener_filter(S, noise_psd_p1, masks) # BUG-2 FIX: Capture diagnostics from gain_p1 (before RMS normalize) mean_gain_p1_db = float( 20.0 * np.log10(float(np.mean(gain_p1)) + 1e-12) ) # ── Stage 6: Iterative refinement — Pass 2 ─────────────────────────── # # Key: do NOT filter twice. Strategy: # 1. Reconstruct pass-1 audio from filtered STFT. # 2. Re-track F0 on this cleaner signal → more accurate F0 seq. # 3. Re-estimate noise PSD on the ORIGINAL STFT (same S_mag_ref) # but with the refined mask from the cleaner signal. # 4. Average the two noise estimates (ensemble → lower variance). # 5. Union of mask estimates (protect anything either pass found). # 6. Apply ONE final filter on the original STFT. # # This avoids the stacked-attenuation problem (double filtering) while # still capturing the F0-tracking improvement from a cleaner signal. if c.n_passes >= 2: audio_p1 = self._istft(S_enh_p1, n_samples) f0_seq2, conf_seq2 = self._track_f0_sequence(audio_p1) masks_p2 = self._build_mask_matrix(f0_seq2, conf_seq2, n_frames) noise_psd_p2, n_v2 = self._hine(S_mag_ref, masks_p2, conf_seq2) # Ensemble noise estimate: average (reduces variance) noise_psd_final = ( (noise_psd_p1.astype(np.float64) + noise_psd_p2.astype(np.float64)) * 0.5 ).astype(np.float32) # Conservative union mask: protect bins either pass thought harmonic masks_final = np.maximum(masks, masks_p2) # Single final filter on original STFT S_final, gain_final = self._wiener_filter(S, noise_psd_final, masks_final) n_voiced_used = n_voiced_used_p1 + n_v2 else: noise_psd_final = noise_psd_p1 masks_final = masks S_final = S_enh_p1 gain_final = gain_p1 n_voiced_used = n_voiced_used_p1 # BUG-2 FIX: Compute SNR gain from gain matrix BEFORE RMS normalization. # mean_gain < 1.0 means noise was removed. SNR improves by roughly # 1/mean_gain_noise_fraction. Approximate: SNR_gain ≈ -mean_gain_db. mean_gain_db = float(20.0 * np.log10(float(np.mean(gain_final)) + 1e-12)) # Effective SNR gain: inter-harmonic bins had full suppression (wiener_floor). # Harmonic bins were nearly transparent (harmonic_floor). # The actual SNR gain is driven by the inter-harmonic suppression. inter_mask_global = np.mean(masks_final < 0.15, axis=1) # per-freq inter_gain_mean = float(np.mean( gain_final[inter_mask_global > 0.5, :] )) if np.any(inter_mask_global > 0.5) else float(np.mean(gain_final)) inter_gain_db = float(20.0 * np.log10(inter_gain_mean + 1e-12)) effective_snr_gain_db = float(max(0.0, -inter_gain_db)) # ── Stage 7: Reconstruct ────────────────────────────────────────────── enhanced = self._istft(S_final, n_samples) # ── Stage 8: Loudness normalization ─────────────────────────────────── # Restore original RMS level. The Wiener filter reduces energy; # the engine's downstream EQ/compand expects consistent levels. orig_rms = float(np.sqrt(np.mean(original.astype(np.float64) ** 2) + 1e-12)) enh_rms = float(np.sqrt(np.mean(enhanced.astype(np.float64) ** 2) + 1e-12)) if enh_rms > 1e-10: enhanced = enhanced * (orig_rms / enh_rms) # Clip guard (should be a no-op, but prevents downstream surprises) enhanced = np.clip(enhanced, -1.0, 1.0).astype(np.float32) # ── Diagnostics ──────────────────────────────────────────────────────── noise_floor_db = float(10.0 * np.log10(float(np.mean(noise_psd_final)) + 1e-12)) result = SAFIResult( applied = True, status = "OK", voiced_ratio = voiced_ratio, median_f0_hz = median_f0, harmonic_coverage = hm_coverage, noise_floor_db = noise_floor_db, mean_gain_db = mean_gain_db, # BUG-2 FIX effective_snr_gain_db = effective_snr_gain_db, # BUG-2 FIX voiced_frames_used = n_voiced_used, # BUG-1 FIX diagnostic n_passes = c.n_passes if c.n_passes >= 2 else 1, ) return enhanced, result # ───────────────────────────────────────────────────────────────────────── # FILE I/O # ───────────────────────────────────────────────────────────────────────── def process_file(self, input_wav: str, output_wav: str) -> SAFIResult: """ SAFI on a file. Handles any format ffmpeg can decode. I/O: ffmpeg pipe (same as الاسترداد — no soundfile, no librosa). In: ffmpeg → raw f32le mono 48kHz → numpy → process() Out: enhanced → ffmpeg → pcm_s24le 48kHz stereo WAV On UNPROCESSABLE or FAILED: output_wav is NOT written. Check result.applied to know if the output file exists. """ # ── Load ────────────────────────────────────────────────────────────── r = subprocess.run( ["ffmpeg", "-y", "-i", input_wav, "-ac", "1", "-ar", str(self.cfg.sr), "-f", "f32le", "-loglevel", "error", "-"], capture_output=True, ) if r.returncode != 0 or not r.stdout: return SAFIResult( status="FAILED", reason=f"ffmpeg_load: {r.stderr.decode(errors='replace')[:200]}", ) audio = np.frombuffer(r.stdout, dtype=np.float32).copy() if len(audio) < self.cfg.sr * 2: return SAFIResult( status="FAILED", reason=f"too_short ({len(audio)/self.cfg.sr:.1f}s < 2s)", ) # ── Process ─────────────────────────────────────────────────────────── try: enhanced, result = self.process(audio) except Exception as exc: return SAFIResult(status="FAILED", reason=f"process: {exc}") if not result.applied: return result # ── Write ───────────────────────────────────────────────────────────── # BUG-9 FIX: scale by 1/√2 before stereo duplication. # Stage 8 set mono RMS = original mono RMS. When the engine reads the # stereo file and computes multi-channel RMS = sqrt(mean(L²+R²)), it # gets sqrt(2)×per-channel RMS = +3.01 dB spurious gain. Pre-scaling # by 1/√2 ensures sqrt(mean(L²+R²)) = sqrt(mean(x²)) = target mono RMS. enhanced_out = enhanced * (1.0 / np.sqrt(2.0)) stereo = np.stack([enhanced_out, enhanced_out], axis=1).astype(np.float32) r2 = subprocess.run( ["ffmpeg", "-y", "-f", "f32le", "-ar", str(self.cfg.sr), "-ac", "2", "-i", "pipe:0", "-c:a", "pcm_s24le", "-ar", str(self.cfg.sr), "-loglevel", "error", output_wav], input=stereo.tobytes(), capture_output=True, ) if r2.returncode != 0 or not os.path.exists(output_wav): return SAFIResult( status="FAILED", reason=f"ffmpeg_write: {r2.stderr.decode(errors='replace')[:200]}", ) return result # ══════════════════════════════════════════════════════════════════════════════ # ENGINE INTEGRATION ENTRY POINT # ══════════════════════════════════════════════════════════════════════════════ _SAFI_SINGLETON: Optional[SAFIDenoiser] = None def _get_safi() -> SAFIDenoiser: global _SAFI_SINGLETON if _SAFI_SINGLETON is None: _SAFI_SINGLETON = SAFIDenoiser(SAFIConfig()) return _SAFI_SINGLETON def apply_safi_to_engine( input_wav: str, frame_snr: float, log_fn=print, ) -> Tuple[str, SAFIResult]: """ Engine integration entry point — called from enhance_tier2(). Replaces the existing T-0.5 DF3 slot for frame_snr < 8 dB. Handles TIER_UNPROCESSABLE gate (frame_snr < 2.5 dB). Returns (output_wav_path, SAFIResult). If SAFI did not process: output_wav == input_wav (original unchanged). """ cfg = SAFIConfig() # ── Absolute floor: cannot help below 2.5 dB ────────────────────────────── if frame_snr < cfg.tier_unprocessable_snr_db: log_fn( f" [SAFI] frame_snr={frame_snr:.1f} dB < {cfg.tier_unprocessable_snr_db} dB" f" → TIER_UNPROCESSABLE — EQ only, all NR skipped" ) return input_wav, SAFIResult( status="UNPROCESSABLE", reason=f"frame_snr={frame_snr:.1f}_below_absolute_floor", ) # ── Outside SAFI zone → pass to existing DF3 ───────────────────────────── if frame_snr >= SAFI_FRAME_SNR_GATE_DB: return input_wav, SAFIResult(status="NOT_RUN", reason="above_safi_gate") log_fn( f" [SAFI] frame_snr={frame_snr:.1f} dB in SAFI zone " f"[{cfg.tier_unprocessable_snr_db}–{SAFI_FRAME_SNR_GATE_DB} dB]" f" — HINE harmonic-guided Wiener NR (voiced-only noise est.)" ) safi_out = input_wav + ".safi.wav" result = _get_safi().process_file(input_wav, safi_out) if result.status == "OK" and os.path.exists(safi_out) and os.path.getsize(safi_out) > 0: log_fn( f" [SAFI] ✓ F0={result.median_f0_hz:.0f} Hz " f"voiced={result.voiced_ratio * 100:.0f}% " f"voiced_frames_used={result.voiced_frames_used} " f"hm_cov={result.harmonic_coverage * 100:.0f}% " f"noise_floor={result.noise_floor_db:.1f} dBFS " f"mean_gain={result.mean_gain_db:.1f} dB " f"SNR_gain=+{result.effective_snr_gain_db:.1f} dB " f"passes={result.n_passes}" ) return safi_out, result log_fn(f" [SAFI] {result.status}: {result.reason} — reverting to original") try: if os.path.exists(safi_out): os.unlink(safi_out) except Exception: pass return input_wav, result # ══════════════════════════════════════════════════════════════════════════════ # ENGINE PATCH INSTRUCTIONS # ══════════════════════════════════════════════════════════════════════════════ PATCH_DOC = """\ ╔══════════════════════════════════════════════════════════════════════════════╗ ║ SAFI v2.1 ENGINE PATCH — engine_isteidad_v6_fixed.py → v7_safi ║ ╚══════════════════════════════════════════════════════════════════════════════╝ Place safi_nr.py in the same directory as engine_isteidad_v6_fixed.py. ──────────────────────────────────────────────────────────────────────────────── PATCH 1 — ADD IMPORT (after existing imports, top of file) ──────────────────────────────────────────────────────────────────────────────── try: from safi_nr import ( apply_safi_to_engine, SAFI_FRAME_SNR_GATE_DB, TIER_UNPROCESSABLE_SNR, ) SAFI_OK = True except ImportError: SAFI_OK = False SAFI_FRAME_SNR_GATE_DB = 8.0 TIER_UNPROCESSABLE_SNR = 2.5 ──────────────────────────────────────────────────────────────────────────────── PATCH 2 — ADD FIELDS to InputState dataclass ──────────────────────────────────────────────────────────────────────────────── # After df3_snr_after field: tier_unprocessable: bool = False safi_applied: bool = False safi_snr_gain_db: float = 0.0 ──────────────────────────────────────────────────────────────────────────────── PATCH 3 — REPLACE the T-0.5 block in enhance_tier2() (~line 5480) ──────────────────────────────────────────────────────────────────────────────── FIND this block (beginning of enhance_tier2 function body): _DF3_GATE = 12.0 snr_frame = getattr(state, 'frame_snr', snr_before) df_trigger = ( not getattr(state, 'df3_applied', False) and ((snr_before < _DF3_GATE) or (snr_frame < _DF3_GATE)) ) df_avail = DEEPFILTER_OK or DEEPFILTER_CLI_OK if df_avail and df_trigger: L(f" [T-0.5/استراد] SNR ..." ... REPLACE WITH: _DF3_GATE = 12.0 _SAFI_GATE = SAFI_FRAME_SNR_GATE_DB # 8.0 dB snr_frame = getattr(state, 'frame_snr', snr_before) # ── SAFI: runs for frame_snr in [2.5, 8.0) dB ───────────────────────── if SAFI_OK and snr_frame < _SAFI_GATE: safi_wav, safi_result = apply_safi_to_engine(input_wav, snr_frame, log_fn=L) if safi_result.status == 'UNPROCESSABLE': state.tier_unprocessable = True # Skip DF3 too — SNR is below everything L(f' [SAFI] TIER_UNPROCESSABLE — NR skipped, EQ only') elif safi_result.status == 'OK': input_wav = safi_wav state.safi_applied = True state.safi_snr_gain_db = safi_result.effective_snr_gain_db t2_report['safi_applied'] = True t2_report['safi_snr_gain_db'] = safi_result.effective_snr_gain_db t2_report['safi_f0_hz'] = safi_result.median_f0_hz t2_report['safi_voiced_ratio'] = safi_result.voiced_ratio t2_report['safi_voiced_frames'] = safi_result.voiced_frames_used # Update estimated SNR so downstream TYPE_A knows the improved state snr_frame = min(snr_frame + safi_result.effective_snr_gain_db, 20.0) state.snr_global = snr_frame # ── DF3: runs for frame_snr in [8.0, 12.0) dB ───────────────────────── # (or after SAFI lifted SNR into DF3 range — unlikely but handled) df_trigger = ( not getattr(state, 'df3_applied', False) and not getattr(state, 'tier_unprocessable', False) and snr_frame < _DF3_GATE and snr_frame >= _SAFI_GATE # SAFI zone already handled above ) df_avail = DEEPFILTER_OK or DEEPFILTER_CLI_OK if df_avail and df_trigger: L(f" [T-0.5] SNR frame={snr_frame:.1f} dB — DeepFilterNet-3") # ... EXISTING DF3 CODE UNCHANGED ... elif not df_avail and not getattr(state, 'safi_applied', False): L(" [T-0.5] deepfilter not installed — skipped") ──────────────────────────────────────────────────────────────────────────────── PATCH 4 — ADD TO enhance() return dict ──────────────────────────────────────────────────────────────────────────────── 'safi_applied': t2_report.get('safi_applied', False), 'safi_snr_gain_db': t2_report.get('safi_snr_gain_db', 0.0), 'safi_f0_hz': t2_report.get('safi_f0_hz', 0.0), 'safi_voiced_ratio': t2_report.get('safi_voiced_ratio', 0.0), 'safi_voiced_frames': t2_report.get('safi_voiced_frames', 0), 'tier_unprocessable': getattr(state, 'tier_unprocessable', False), ──────────────────────────────────────────────────────────────────────────────── PATCH 5 — ADD TO engine header / version string ──────────────────────────────────────────────────────────────────────────────── 'engine_version': 'v10.7-الاسترداد-v7-SAFI-2.1', # Banner addition: # ║ SAFI v2.1 — Harmonic-Interleaved Noise Estimation ║ # ║ voiced-only HINE | 2.5–8.0 dB frame SNR zone | DF3 > 8 dB ║ ──────────────────────────────────────────────────────────────────────────────── SNR ZONE DIAGRAM (final routing after patch) ──────────────────────────────────────────────────────────────────────────────── frame_snr < 2.5 dB → TIER_UNPROCESSABLE: EQ only, NO NR frame_snr 2.5–8 dB → SAFI v2.1 (HINE + Wiener, Arabic-aware) frame_snr 8–12 dB → DeepFilterNet-3 (existing T-0.5 path) frame_snr ≥ 12 dB → No NR (TYPE_A/B/C standard path) """ # ══════════════════════════════════════════════════════════════════════════════ # STANDALONE CLI # ══════════════════════════════════════════════════════════════════════════════ def _voiced_thresh_for_cfg(cfg: SAFIConfig) -> float: """Extract voiced_thresh from config.""" return cfg.voiced_thresh def _main() -> int: import argparse p = argparse.ArgumentParser( prog="safi_nr", description=( "SAFI v2.2 — Speech-Adaptive Formant Isolation\n" "Harmonic-guided Wiener NR for Quran recitation at 3–8 dB SNR.\n" "Uses voiced-only HINE — stronger than DF3 below its SNR floor.\n" ), formatter_class=argparse.RawDescriptionHelpFormatter, ) p.add_argument("-i", "--input", help="Input WAV file") p.add_argument("-o", "--output", help="Output WAV file") p.add_argument("--passes", type=int, default=2, help="Refinement passes (default 2)") p.add_argument("--f0-min", type=float, default=70.0, help="Min F0 Hz (default 70)") p.add_argument("--f0-max", type=float, default=380.0, help="Max F0 Hz (default 380)") p.add_argument("--beta", type=float, default=1.5, help="Wiener β (default 1.5)") p.add_argument("--floor", type=float, default=0.07, help="Inter-harmonic min gain (default 0.07)") p.add_argument("--hfloor", type=float, default=0.70, help="Harmonic bin min gain (default 0.70)") p.add_argument("--sfloor", type=float, default=0.50, help="Sibilant hard floor (default 0.50)") p.add_argument("--sr", type=int, default=48000, help="Sample rate (default 48000)") p.add_argument("--patch-doc", action="store_true", help="Print engine patch instructions") args = p.parse_args() if args.patch_doc: print(PATCH_DOC) return 0 if not args.input or not args.output: p.print_help() return 1 if not os.path.exists(args.input): print(f"ERROR: {args.input} not found", file=sys.stderr) return 1 cfg = SAFIConfig( sr = args.sr, n_passes = args.passes, f0_min = args.f0_min, f0_max = args.f0_max, wiener_beta = args.beta, wiener_floor = args.floor, harmonic_floor = args.hfloor, sibilant_floor = args.sfloor, ) safi = SAFIDenoiser(cfg) W = 60 divider = "─" * W print(f"\n{'═' * W}") print(f" SAFI v2.2 — Speech-Adaptive Formant Isolation") print(f" صافي — voiced-only HINE, Arabic sibilant protection") print(divider) print(f" Input : {args.input}") print(f" Output : {args.output}") print(f" F0 : {cfg.f0_min:.0f}–{cfg.f0_max:.0f} Hz") print(f" β : {cfg.wiener_beta} floor: {cfg.wiener_floor}") print(f" sib.floor: {cfg.sibilant_floor} ({cfg.sibilant_lo_hz:.0f}–{cfg.sibilant_hi_hz:.0f} Hz)") print(f" passes : {cfg.n_passes}") print(divider) result = safi.process_file(args.input, args.output) print(f" Status : {result.status}") if result.status == "OK": print(f" F0 : {result.median_f0_hz:.1f} Hz (median)") print(f" Voiced : {result.voiced_ratio * 100:.1f}% ({result.voiced_frames_used} frames used in HINE)") print(f" HM cov : {result.harmonic_coverage * 100:.1f}%") print(f" Noise ↓ : {result.noise_floor_db:.1f} dBFS") print(f" Gain : {result.mean_gain_db:.1f} dB (mean)") print(f" SNR + : +{result.effective_snr_gain_db:.1f} dB (inter-harmonic suppression)") print(f" Passes : {result.n_passes}") print(f" Output : {args.output}") elif result.status == "UNPROCESSABLE": print(f" Reason : {result.reason}") print(f" Action : run EQ only — NR not possible at this SNR") else: print(f" Reason : {result.reason}") print(f"{'═' * W}\n") return 0 if result.status in ("OK", "UNPROCESSABLE") else 1 if __name__ == "__main__": sys.exit(_main())