Spaces:
Running
Running
Tilawa Enhancer
S164: fix deep-filter path, deploy jawhar/safi/jalaa/bayan_ve, soundfile+librosa
4fd4350 | #!/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 | |
| # ββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ | |
| 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 | |
| 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]<thresh] += S_mag[:,t]**2 | |
| Use vectorized masking: | |
| inter_mat = (masks < thresh) & (conf_mat > 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()) | |