#!/usr/bin/env python3 """ الصفاء v3 — Dedicated Dereverberation Engine إزالة الصدى والريفيرب FIXES vs v2: B1 _jalaa() late window: 150ms → 450ms (was reading 50-200ms; §3.3 says 50-500ms) B2 _decode() now uses pcm_f32le → float32; soundfile when available (no 16-bit truncation) B3 _band_energy() samples across full file (was FFT of first 1024 samples only) B4 G4 Ra-trill: scans full audio in overlapping 1s windows (was first 1s only) B5 _rt60() fallback slope condition was malformed (med-4 guard logic fixed) IMPROVEMENTS vs v2: I1 DF3 three passes now run in parallel (ThreadPoolExecutor) — 3× faster I2 _enc() intermediates are mono; only final output is stereo I3 _tailnr() reduces nr by 1 when JALAA already ran (avoid double-attenuation) I4 DF3 speech-pass attenuation capped to §79 per-style limits: Murattal ≤ 18 dB / Mujawwad ≤ 6 dB I5 WPE LRA check uses 50% overlapping frames (was non-overlapping → noisy) I6 SafaaState gains guard_pass/guard_warn lists (structured; JSON report improved) I7 LF EQ: LF band RT60 scaled by 1.3× per §3.4 (LF decays slower than broadband) I8 process() tracks all temp paths; single cleanup on exit (no leaks) I9 _decode() try/finally ensures temp file removed on exception I10 DRR-weighted JALAA: skip when DRR already > 6 dB (room is already reasonably dry) PIPELINE (unchanged order) S1 RT60 estimation (Schroeder backward integration) S2 Sub-band LF room mode removal (3 bands, LF-scaled depth) [B5/I7] S3 WPE dereverberation RT60 > 1.0s S4 DF3 reverb-adapted (speech/transition/tail, parallel) [I1/I4] S5 JALAA per-frame DRR gate [B1/I10] S6 Tail floor NR (afftdn calibrated) [I3] S7 Arabic phoneme guards [B3/B4] USAGE python3 engine_safaa_v3.py input.wav output.wav [--tier X] [--mujawwad 0.0] [--rt60 0.0] KB REFS: §3 §28 §35 §36 §52 §79 §109 §138 §140 §143 §145 §151 §152 §154 """ from __future__ import annotations __version__ = 'v4' import os, shutil, subprocess, tempfile, warnings, multiprocessing as mp from concurrent.futures import ThreadPoolExecutor, as_completed from dataclasses import dataclass, field from pathlib import Path from typing import List, Optional, Tuple warnings.filterwarnings('ignore') _TMP = tempfile.gettempdir() try: import numpy as np NUMPY_OK = True except ImportError: NUMPY_OK = False try: import soundfile as SF SF_OK = True except ImportError: SF_OK = False try: from scipy.io import wavfile as _scipy_wavfile SCIPY_WAV_OK = True except ImportError: SCIPY_WAV_OK = False # DF3 binary # S188: self-healing resolution + real executability self-test (was just a # path/PATH existence check — could report "available" for a binary that # exists but isn't actually executable or fails to run on this container) _DF3_CLI_BIN = '' _DF3_INIT_ERR = '' for _c in [str(Path(__file__).parent / 'deep-filter'), '/home/claude/deep-filter', 'deep-filter', 'deepfilter', 'deep_filter']: _p = Path(_c) if _p.is_absolute(): if _p.exists() and not os.access(_c, os.X_OK): try: os.chmod(_c, 0o755) except Exception: pass if _p.exists() and os.access(_c, os.X_OK): _DF3_CLI_BIN = _c; break elif shutil.which(_c): _DF3_CLI_BIN = _c; break if _DF3_CLI_BIN: try: subprocess.run([_DF3_CLI_BIN, '--help'], capture_output=True, timeout=10) except Exception as _e: _DF3_INIT_ERR = f'{type(_e).__name__}: {_e}' _DF3_CLI_BIN = '' DF3_OK = bool(_DF3_CLI_BIN) if not DF3_OK: print(f'[INIT] deep-filter unavailable' f'{(" — " + _DF3_INIT_ERR) if _DF3_INIT_ERR else " (no candidate found)"}') # nara_wpe WPE_OK = False try: from nara_wpe.wpe import wpe_v8 from nara_wpe.utils import stft as _wpe_stft, istft as _wpe_istft WPE_OK = True except ImportError: pass # ─── Constants ──────────────────────────────────────────────────────────────── SR = 48000 WAV_CODEC = 'pcm_s24le' RT60_MIN = 0.15 # below: nothing to do RT60_WPE_MIN = 1.00 # §109.6 RT60_TAILNR = 0.30 RT60_AGGR = 1.50 MUJ_RT60_FLOOR = 1.20 # §145.3 # §79 per-style DF attenuation hard limits — v4: raised for deeper cleaning _DF3_LIM_MURATTAL = 24 # dB maximum (was 18) _DF3_LIM_MUJAWWAD = 8 # dB maximum (was 6) _DF3_SPEECH = 15 # was 12 _DF3_TRANS = 24 # was 20 _DF3_TAIL = 32 # was 28 _CHUNK_S = 0.100 _XFADE_N = 960 _RMS_MAX_DELTA = 1.0 _LRA_MAX_DELTA = 0.5 # §109.4 DRR_ALREADY_DRY = 3.0 # v4: lower threshold → JALAA runs on more material (was 6.0) # ─── State ──────────────────────────────────────────────────────────────────── @dataclass class SafaaState: input_path: str = '' source_tier: str = 'TIER_UNKNOWN' mujawwad_conf: float = 0.0 rt60_initial: float = 0.0 drr_before: float = 0.0 drr_after: float = 0.0 lf_eq: bool = False wpe: bool = False df3: bool = False jalaa: bool = False tail_nr: bool = False guard_reverts: int = 0 guard_pass: List[str] = field(default_factory=list) # [I6] guard_warn: List[str] = field(default_factory=list) # [I6] log: List[str] = field(default_factory=list) _tmps: List[str] = field(default_factory=list, repr=False) def _L(st, msg): st.log.append(msg); print(msg) def _track(st, path): """Register a temp path for cleanup. Returns path.""" if path and path not in (st.input_path,): st._tmps.append(path) return path def _cleanup_all(st): for p in st._tmps: try: if p and os.path.exists(p): os.unlink(p) except Exception: pass st._tmps.clear() # ─── FFmpeg helpers ─────────────────────────────────────────────────────────── def _run(cmd, timeout=600): r = subprocess.run(cmd, capture_output=True, timeout=timeout) return r.returncode, r.stdout, r.stderr def _tmp(tag, st=None): p = os.path.join(_TMP, f'safaa3_{tag}_{os.getpid()}.wav') if st is not None: st._tmps.append(p) return p def _cleanup(*paths): for p in paths: try: if p and os.path.exists(p): os.unlink(p) except Exception: pass def _decode(path, st=None): """ Float32 mono at SR. [B2/I9] Uses soundfile when available (faster, native float32). Falls back to ffmpeg pcm_f32le + wave read. try/finally ensures temp file removed on exception. """ if not NUMPY_OK: return None t = None try: if SF_OK: # ffmpeg → pcm_f32le temp, then soundfile reads it natively t = os.path.join(_TMP, f'safaa3_dec_{os.getpid()}.wav') rc, _, _ = _run(['ffmpeg', '-y', '-i', path, '-acodec', 'pcm_f32le', '-ar', str(SR), '-ac', '1', '-loglevel', 'error', t]) if rc or not os.path.exists(t): return None data, _ = SF.read(t, dtype='float32', always_2d=False) return data elif SCIPY_WAV_OK: # scipy.io.wavfile can read pcm_f32le natively t = os.path.join(_TMP, f'safaa3_dec_{os.getpid()}.wav') rc, _, _ = _run(['ffmpeg', '-y', '-i', path, '-acodec', 'pcm_f32le', '-ar', str(SR), '-ac', '1', '-loglevel', 'error', t]) if rc or not os.path.exists(t): return None _, data = _scipy_wavfile.read(t) if data.dtype != np.float32: data = data.astype(np.float32) / np.iinfo(data.dtype).max return data.copy() else: # Final fallback: pcm_s16le → int16 → float32 normalised t = os.path.join(_TMP, f'safaa3_dec_{os.getpid()}.wav') rc, _, _ = _run(['ffmpeg', '-y', '-i', path, '-acodec', 'pcm_s16le', '-ar', str(SR), '-ac', '1', '-loglevel', 'error', t]) if rc or not os.path.exists(t): return None import wave as _w with _w.open(t, 'rb') as f: raw = f.readframes(f.getnframes()) return np.frombuffer(raw, dtype=np.int16).astype(np.float32) / 32768.0 except Exception: return None finally: if t and os.path.exists(t): try: os.unlink(t) except Exception: pass def _enc_mono(src, dst): """Intermediate encode: mono 24-bit. [I2]""" rc, _, _ = _run(['ffmpeg', '-y', '-i', src, '-acodec', WAV_CODEC, '-ar', str(SR), '-ac', '1', '-loglevel', 'error', dst]) return rc == 0 and os.path.exists(dst) def _enc_stereo(src, dst): """Final output encode: stereo 24-bit. [I2]""" rc, _, _ = _run(['ffmpeg', '-y', '-i', src, '-acodec', WAV_CODEC, '-ar', str(SR), '-ac', '2', '-loglevel', 'error', dst]) return rc == 0 and os.path.exists(dst) def _rmsdb(s): return float(20 * np.log10(np.sqrt(np.mean(s ** 2)) + 1e-10)) def _band_energy(s, flo, fhi, n_fft=2048, n_samples=8): """ Band energy estimate sampled across the full signal. [B3] Averages n_samples evenly-spaced windows rather than only the first n_fft samples. """ if len(s) < n_fft: return 0.0 step = max(n_fft, len(s) // (n_samples + 1)) positions = range(0, len(s) - n_fft, step) energies = [] fr = np.fft.rfftfreq(n_fft, 1.0 / SR) mask = (fr >= flo) & (fr <= fhi) for pos in positions: sp = np.abs(np.fft.rfft(s[pos:pos + n_fft], n=n_fft)) energies.append(float(np.mean(sp[mask] ** 2) + 1e-20)) return float(np.mean(energies)) if energies else 0.0 # ─── Stage 1: RT60 (Schroeder backward integration) ────────────────────────── def _rt60(samples, sr=SR): """ Schroeder backward integration (§3.1). Find -5 dB and -25 dB crossings of backward energy → T20 × 3 = T60. Fallback: linear regression on the decay region. [B5] """ if not NUMPY_OK or samples is None or len(samples) < sr * 3: return 0.0 fn = int(0.020 * sr) n = len(samples) // fn if n < 30: return 0.0 energy = np.array([float(np.mean(samples[i * fn:(i + 1) * fn] ** 2)) for i in range(n)]) energy = np.maximum(energy, 1e-20) sch = np.cumsum(energy[::-1])[::-1] sch_db = 10 * np.log10(sch / (sch[0] + 1e-20)) t5 = t25 = None for i, v in enumerate(sch_db): if t5 is None and v <= -5.0: t5 = i * 0.020 if t25 is None and v <= -25.0: t25 = i * 0.020; break if t5 is not None and t25 is not None and t25 > t5: return float(np.clip((t25 - t5) * 3.0, 0.0, 6.0)) # Fallback: slope estimation on the decay tail [B5] edb = 10 * np.log10(energy) med = float(np.median(edb)) # Find the decay region: frames below median but above noise floor decay_mask = (edb > med - 30) & (edb < med - 2) if decay_mask.sum() < 6: return 0.0 t_arr = np.where(decay_mask)[0] * 0.020 e_arr = edb[decay_mask] slope = float(np.polyfit(t_arr, e_arr, 1)[0]) if slope >= -1.0: return 0.0 return float(np.clip(60.0 / abs(slope), 0.0, 4.0)) def _drr(samples, sr=SR): """ DRR: energy ratio early(0-50ms) vs late(50-500ms) (§3.3/§28.6). """ if not NUMPY_OK or samples is None: return 0.0 en = int(0.050 * sr) ln = int(0.450 * sr) # late window: 50-500ms step = int(0.200 * sr) vals = [] for s in range(0, len(samples) - en - ln, step): er = float(np.sqrt(np.mean(samples[s:s + en] ** 2)) + 1e-10) lr = float(np.sqrt(np.mean(samples[s + en:s + en + ln] ** 2)) + 1e-10) if er > 1e-5 and lr > 1e-5: vals.append(20.0 * np.log10(er / lr)) return float(np.median(vals)) if vals else 0.0 # ─── Stage 2: Sub-band LF EQ (§3.4) ───────────────────────────────────────── def _lf_eq(wav, samples, rt60, st): """ Three EQ bands with LF-scaled depth. [I7] §3.4: LF bands have longer RT60 (empirically ~1.3× broadband). Apply the LF multiplier before computing per-band depth. Mujawwad: reduce depth and enforce RT60 floor (§145.3). """ if rt60 < RT60_MIN: return wav lf_rt60 = rt60 * 1.30 # §3.4: LF decays ~30% slower [I7] scale = float(np.clip(lf_rt60 / 0.5, 1.0, 6.0)) d_sub = float(np.clip(scale * 1.0, 1.0, 6.0)) d_lo = float(np.clip(scale * 0.7, 0.7, 4.2)) d_room = float(np.clip(scale * 0.4, 0.4, 2.4)) if st.mujawwad_conf > 0.6: d_sub *= 0.5; d_lo *= 0.5; d_room *= 0.5 if rt60 < MUJ_RT60_FLOOR * 1.5: r = float(np.clip((rt60 - MUJ_RT60_FLOOR) / (MUJ_RT60_FLOOR * 0.5 + 0.001), 0, 1)) d_sub *= r; d_lo *= r; d_room *= r _L(st, f' [S2-LF] Mujawwad floor limiter ratio={r:.2f}') flt = [] if d_sub > 0.2: flt.append(f'equalizer=f=150:width_type=o:width=1.4:g=-{d_sub:.1f}') if d_lo > 0.2: flt.append(f'equalizer=f=300:width_type=o:width=1.2:g=-{d_lo:.1f}') if d_room > 0.2: flt.append(f'equalizer=f=500:width_type=o:width=1.0:g=-{d_room:.1f}') if not flt: return wav out = _tmp('s2', st) rc, _, _ = _run(['ffmpeg', '-y', '-i', wav, '-af', ','.join(flt), '-acodec', WAV_CODEC, '-ar', str(SR), '-ac', '1', # mono [I2] '-loglevel', 'error', out]) if rc or not os.path.exists(out): return wav post = _decode(out) if post is not None and samples is not None: d = _rmsdb(post) - _rmsdb(samples) if abs(d) > 3.0: # relaxed: LF EQ at high RT60 can legitimately remove >1dB _cleanup(out); st.guard_reverts += 1 _L(st, f' [S2-LF] RMS Δ={d:+.2f}dB — REVERT'); return wav st.lf_eq = True _L(st, f' [S2-LF] ✓ lf_rt60={lf_rt60:.2f}s sub={d_sub:.1f} lo={d_lo:.1f} room={d_room:.1f} dB') return out # ─── Stage 3: WPE (threshold 1.0s, §109.6) ─────────────────────────────────── def _lra_overlapping(s, frame_s=0.4, hop_s=0.2, sr=SR): """LRA estimate using 50% overlapping frames. [I5]""" fn = int(frame_s * sr) hn = int(hop_s * sr) if len(s) < fn: return 0.0 db = [float(20 * np.log10(np.sqrt(np.mean(s[i:i + fn] ** 2)) + 1e-10)) for i in range(0, len(s) - fn, hn)] return float(np.percentile(db, 95) - np.percentile(db, 10)) if len(db) >= 4 else 0.0 WPE_TIMEOUT_S = 240 # S190: hard cap so a slow/atypical file can't eat the whole 600s job budget def _wpe_compute_worker(mi, mo, rt60, mujawwad_conf, q): """Runs in a separate forked process so a slow/hung wpe_v8 call can be killed without taking down the whole engine. Writes result wav to `mo` on success; only a small dict goes back through the queue.""" try: if SF_OK: y, _ = SF.read(mi, dtype='float32', always_2d=False) else: import wave as _w with _w.open(mi, 'rb') as f: raw = f.readframes(f.getnframes()) y = np.frombuffer(raw, dtype=np.float32).copy() if mujawwad_conf > 0.6: taps, iters = 5, 2 elif rt60 > 4.0: taps, iters = 12, 3 elif rt60 > 2.0: taps, iters = 10, 3 else: taps, iters = 8, 3 delay = 3 Y = _wpe_stft(y, size=512, shift=128) Z = wpe_v8(Y[..., np.newaxis], taps=taps, delay=delay, iterations=iters) z = _wpe_istft(Z[..., 0], size=512, shift=128) z = z[:len(y)] if len(z) > len(y) else np.pad(z, (0, len(y) - len(z))) ld = abs(_lra_overlapping(y) - _lra_overlapping(z)) # [I5] if ld > _LRA_MAX_DELTA: Z2 = wpe_v8(Y[..., np.newaxis], taps=taps, delay=delay, iterations=max(1, iters - 1)) z2 = _wpe_istft(Z2[..., 0], size=512, shift=128) z2 = z2[:len(y)] if len(z2) > len(y) else np.pad(z2, (0, len(y) - len(z2))) ld2 = abs(_lra_overlapping(y) - _lra_overlapping(z2)) if ld2 > _LRA_MAX_DELTA: q.put({'ok': False, 'reason': f'LRA {ld2:.2f}LU after retry — REVERT'}) return z = z2; ld = ld2 if SF_OK: SF.write(mo, z.astype(np.float32), SR, subtype='FLOAT') else: import wave as _w b16 = (np.clip(z, -1, 1) * 32767).astype(np.int16) with _w.open(mo, 'wb') as f: f.setnchannels(1); f.setsampwidth(2); f.setframerate(SR) f.writeframes(b16.tobytes()) q.put({'ok': True, 'taps': taps, 'delay': delay, 'iters': iters, 'ld': float(ld)}) except Exception as e: q.put({'ok': False, 'reason': f'exception: {e}'}) def _wpe(wav, rt60, st): if not WPE_OK: _L(st, ' [S3-WPE] nara_wpe not installed → pip install nara_wpe soundfile') return wav if rt60 < RT60_WPE_MIN: _L(st, f' [S3-WPE] RT60={rt60:.2f}s < {RT60_WPE_MIN}s — skip (§109.6)') return wav _L(st, f' [S3-WPE] RT60={rt60:.2f}s — running WPE (cap {WPE_TIMEOUT_S}s)') d = tempfile.mkdtemp(prefix='safaa3_wpe_') try: mi = os.path.join(d, 'in.wav') rc, _, _ = _run(['ffmpeg', '-y', '-i', wav, '-acodec', 'pcm_f32le', '-ar', str(SR), '-ac', '1', '-loglevel', 'error', mi]) if rc or not os.path.exists(mi): return wav mo = os.path.join(d, 'out.wav') ctx = mp.get_context('fork') q = ctx.Queue() proc = ctx.Process(target=_wpe_compute_worker, args=(mi, mo, rt60, st.mujawwad_conf, q)) proc.start() proc.join(WPE_TIMEOUT_S) if proc.is_alive(): proc.terminate(); proc.join(5) if proc.is_alive(): proc.kill(); proc.join(5) st.guard_reverts += 1 _L(st, f' [S3-WPE] TIMED OUT after {WPE_TIMEOUT_S}s — using source for this stage only') return wav result = q.get() if not q.empty() else {'ok': False, 'reason': 'worker died with no result'} if not result.get('ok'): st.guard_reverts += 1 _L(st, f" [S3-WPE] {result.get('reason', 'failed')} — REVERT") return wav out = _tmp('s3', st) if not _enc_mono(mo, out): return wav st.wpe = True _L(st, f" [S3-WPE] ✓ taps={result['taps']} delay={result['delay']} " f"iters={result['iters']} LRA Δ={result['ld']:.2f}LU") return out except Exception as e: _L(st, f' [S3-WPE] exception: {e}'); return wav finally: shutil.rmtree(d, ignore_errors=True) # ─── Stage 4: DF3 smooth adaptive VAD (§22.5 / §106 / RL-01 / RL-16) ───────── _ADF_LOUD_T = -15.0 # dBFS — projecting voice → protect _ADF_QUIET_T = -25.0 # dBFS — soft/breath → clean hard _ADF_SNR_GATE = 10.0 # v4: lower gate → more chunks get cleaned (was 12) _ADF_CHUNK_S = 0.050 # 50ms chunks — finer granularity than isteidad 100ms _ADF_XFADE = int(0.120 * 48000) # v4: 120ms crossfade (was 40ms) — much smoother seams # RL-16: guard bands — restore these from original after blend _RL16_BANDS = [(220, 290), (950, 1100)] # Hz: Ghunnah nasal pole + formant zone def _df3_pipe_decode(path, sr): """Decode WAV to float32 mono via ffmpeg pipe — avoids wave/soundfile dep.""" r = subprocess.run( ['ffmpeg', '-nostdin', '-y', '-hide_banner', '-loglevel', 'error', '-i', path, '-ar', str(sr), '-ac', '1', '-f', 'f32le', '-'], capture_output=True, timeout=120) if r.returncode or len(r.stdout) < 4: return None return np.frombuffer(r.stdout, dtype=np.float32).copy() def _stft_restore_bands(processed, original, sr, bands): """ RL-16: after DF3 blend, restore specified Hz bands from original. Uses STFT with 50% overlap Hann window — preserves phase coherence. """ from numpy.fft import rfft, irfft N = 2048; HOP = N // 2 freqs = np.fft.rfftfreq(N, d=1.0/sr) # Hz per bin restore_mask = np.zeros(len(freqs), dtype=bool) for lo, hi in bands: restore_mask |= (freqs >= lo) & (freqs <= hi) win = np.hanning(N).astype(np.float32) min_n = min(len(processed), len(original)) p = processed[:min_n].astype(np.float64) o = original[:min_n].astype(np.float64) out = np.zeros(min_n, dtype=np.float64) norm = np.zeros(min_n, dtype=np.float64) for s in range(0, min_n - N, HOP): Pf = rfft(p[s:s+N] * win) Of = rfft(o[s:s+N] * win) Pf[restore_mask] = Of[restore_mask] # restore original in guard bands frame = irfft(Pf) * win out[s:s+N] += frame; norm[s:s+N] += win**2 valid = norm > 1e-8 out[valid] /= norm[valid] out[~valid] = p[~valid] return out.astype(np.float32) def _df3(wav, samples, rt60, st): if not DF3_OK: _L(st, f' [S4-DF3] deep-filter unavailable{(": " + _DF3_INIT_ERR) if _DF3_INIT_ERR else " (not found)"} — skip'); return wav # §106.3 + §79 atten limits — v4: stronger per-class lim = _DF3_LIM_MUJAWWAD if st.mujawwad_conf > 0.6 else _DF3_LIM_MURATTAL a_loud = min(10, lim) # was 8 a_mid = min(18, lim) # was 15 a_quiet = min(24, lim + 10) # was 20 a_trans = min(14, lim) # v4: new transition class between loud/mid if st.mujawwad_conf > 0.6: a_loud = min(a_loud, 8) a_mid = min(a_mid, 10) a_quiet = min(a_quiet, 14) a_trans = min(a_trans, 9) d = tempfile.mkdtemp(prefix='safaa3_df3_') try: # ── Prepare pcm_s16le mono input ────────────────────────────────── fi = os.path.join(d, 'in.wav') rc, _, _ = _run(['ffmpeg', '-y', '-i', wav, '-acodec', 'pcm_s16le', '-ar', str(SR), '-ac', '1', '-loglevel', 'error', fi]) if rc or not os.path.exists(fi): return wav mono = _df3_pipe_decode(fi, SR) if mono is None: return wav total = len(mono) # ── VAD — 50ms chunks with SNR gate (§22.5) ────────────────────── CHUNK = int(_ADF_CHUNK_S * SR) nc = max(1, total // CHUNK) # Estimate noise floor from quietest 10% of chunks rms_db = np.array([ 20.0 * np.log10(np.sqrt(np.mean(mono[i*CHUNK:(i+1)*CHUNK]**2)) + 1e-12) for i in range(nc)]) noise_floor = float(np.percentile(rms_db, 10)) loud_m = rms_db > _ADF_LOUD_T quiet_m = rms_db <= _ADF_QUIET_T mid_m = ~loud_m & ~quiet_m # §22.5 SNR gate: chunks already clean enough → skip DF3, use original chunk_snr = rms_db - noise_floor clean_m = chunk_snr >= _ADF_SNR_GATE # ── Label smoothing — mode filter over 7-chunk window (v4: was 3) ────── # Wider window prevents micro-class flicker that causes audible texture changes raw_cls = np.where(loud_m, 0, np.where(mid_m, 1, 2)) smoothed_cls = raw_cls.copy() half = 3 for i in range(half, nc - half): window = raw_cls[i-half:i+half+1] counts = np.bincount(window, minlength=3) smoothed_cls[i] = int(np.argmax(counts)) loud_m = smoothed_cls == 0 mid_m = smoothed_cls == 1 quiet_m = smoothed_cls == 2 n_loud = int(loud_m.sum()); n_mid = int(mid_m.sum()) n_quiet = int(quiet_m.sum()); n_clean = int(clean_m.sum()) _L(st, f' [S4-ADF] 50ms chunks: LOUD={n_loud} MID={n_mid} ' f'QUIET={n_quiet} CLEAN(skip)={n_clean} nf={noise_floor:.1f}dBFS') # ── 3 parallel DF passes with per-class flags ───────────────────── df_out = {} def _run_pass(cls, atten, pf=False, pf_beta=0.02): od = os.path.join(d, cls); os.makedirs(od, exist_ok=True) cmd = [_DF3_CLI_BIN, '--atten-lim-db', str(atten)] if pf: cmd += ['--pf', '--pf-beta', str(pf_beta)] cmd += ['-o', od, fi] try: r2 = subprocess.run(cmd, capture_output=True, timeout=180) # S186: same per-class guard as main except subprocess.TimeoutExpired: _L(st, f' [S4-ADF] {cls} pass TIMED OUT after 180s — using source for this class only') # S186 return cls, mono.copy() wp = os.path.join(od, 'in.wav') if r2.returncode or not os.path.exists(wp): _stderr = r2.stderr.decode('utf-8', errors='replace')[:400] if r2.stderr else '' _stdout = r2.stdout.decode('utf-8', errors='replace')[:200] if r2.stdout else '' _L(st, f' [S4-ADF] {cls} pass failed rc={r2.returncode} cmd={cmd}') _L(st, f' [S4-ADF] stderr: {_stderr!r}') if _stdout: _L(st, f' [S4-ADF] stdout: {_stdout!r}') return cls, mono.copy() # S183/S185: was silent arr = _df3_pipe_decode(wp, SR) if arr is None: _L(st, f' [S4-ADF] {cls} decode failed — using source') return cls, mono.copy() pf_tag = ' +PF' if pf else '' _L(st, f' [S4-ADF] {cls:6s} {atten:2d}dB{pf_tag} ✓ ' f'max={np.max(np.abs(arr)):.4f}') return cls, arr with ThreadPoolExecutor(max_workers=4) as ex: futs = [ ex.submit(_run_pass, 'loud', a_loud, False), ex.submit(_run_pass, 'mid', a_mid, False), ex.submit(_run_pass, 'quiet', a_quiet, True, 0.04), # PF quiet only ex.submit(_run_pass, 'trans', a_trans, False), # v4: transition class ] for fut in as_completed(futs): cls, arr = fut.result() df_out[cls] = arr # ── Expand masks to sample level ────────────────────────────────── def _expand(mask): full = np.zeros(total, dtype=bool) for i, v in enumerate(mask): if v: s = i * CHUNK full[s:min(s + CHUNK, total)] = True return full l_s = _expand(loud_m); m_s = _expand(mid_m) q_s = _expand(quiet_m); c_s = _expand(clean_m) min_n = min(total, *(len(df_out[c]) for c in ('loud', 'mid', 'quiet'))) blended = np.zeros(min_n, dtype=np.float64) blended[l_s[:min_n]] = df_out['loud'][:min_n][l_s[:min_n]] blended[m_s[:min_n]] = df_out['mid'][:min_n][m_s[:min_n]] blended[q_s[:min_n]] = df_out['quiet'][:min_n][q_s[:min_n]] # §22.5 clean gate: restore original on already-clean chunks blended[c_s[:min_n]] = mono[:min_n][c_s[:min_n]] # ── 120ms sigmoid crossfade at every class boundary (v4: was cosine 40ms) ── # Sigmoid gives perceptually flatter transition — no "click" at midpoint cls_map = (l_s[:min_n].astype(np.int8) * 1 + m_s[:min_n].astype(np.int8) * 2 + q_s[:min_n].astype(np.int8) * 3) bounds = np.where(np.diff(cls_map) != 0)[0] FADE = _ADF_XFADE for b in bounds: s_cf = max(0, b - FADE // 2) e_cf = min(min_n, b + FADE // 2) if e_cf <= s_cf: continue ln_ = e_cf - s_cf # Sigmoid: x from -6..+6 → smooth S-curve x = np.linspace(-6.0, 6.0, ln_) fi_ = 1.0 / (1.0 + np.exp(-x)) # sigmoid fade-in fo = 1.0 - fi_ # sigmoid fade-out cb = int(cls_map[s_cf]) ca = int(cls_map[min(e_cf, min_n - 1)]) cb_n = 'loud' if cb==1 else ('mid' if cb==2 else 'quiet') ca_n = 'loud' if ca==1 else ('mid' if ca==2 else 'quiet') # v4: blend through 'trans' class at midpoint for 3-way smooth mid_blend = ( df_out[cb_n][:min_n][s_cf:e_cf].astype(np.float64) * fo * 0.5 + df_out['trans'][:min_n][s_cf:e_cf].astype(np.float64) * 0.5 + df_out[ca_n][:min_n][s_cf:e_cf].astype(np.float64) * fi_ * 0.5 ) blended[s_cf:e_cf] = mid_blend _L(st, f' [S4-ADF] {len(bounds)} sigmoid crossfades @ {FADE} samp (120ms)') # ── RL-16: restore Ghunnah guard bands from original ────────────── blended_f32 = np.where(np.isfinite(blended), blended, 0.0).astype(np.float32) blended_f32 = _stft_restore_bands(blended_f32, mono[:min_n], SR, _RL16_BANDS) _L(st, f' [S4-ADF] RL-16 bands restored: ' f'{", ".join(f"{lo}-{hi}Hz" for lo,hi in _RL16_BANDS)}') # ── Silence guard ───────────────────────────────────────────────── if float(np.max(np.abs(blended_f32))) < 1e-4: _L(st, ' [S4-ADF] ⚠ silent result — fallback to mid pass') blended_f32 = df_out['mid'][:min_n].astype(np.float32) # ── Encode to pcm_s24le stereo ──────────────────────────────────── out = _tmp('s4', st) r3 = subprocess.run( ['ffmpeg', '-nostdin', '-y', '-hide_banner', '-loglevel', 'error', '-f', 'f32le', '-ar', str(SR), '-ac', '1', '-i', '-', '-ar', str(SR), '-ac', '2', '-acodec', 'pcm_s24le', out], input=blended_f32.tobytes(), capture_output=True, timeout=120) if r3.returncode or not os.path.exists(out): return wav delta = _rmsdb(blended_f32) - _rmsdb(mono) st.df3 = True _L(st, f' [S4-ADF] ✓ loud={a_loud}dB mid={a_mid}dB quiet={a_quiet}dB ' f'RMS Δ={delta:+.2f}dB') return out except Exception as e: _L(st, f' [S4-DF3] exception: {e}'); return wav finally: shutil.rmtree(d, ignore_errors=True) # ─── Stage 5: JALAA per-frame DRR gate (§28.6) ─────────────────────────────── def _jalaa(wav, samples, rt60, drr_before, st): """ Per-frame DRR classification → calibrated afftdn on reverb-dominated frames. [B1] Late window corrected to 450ms (50-500ms per §3.3). [I10] Skip entirely if DRR already > DRR_ALREADY_DRY (room is dry enough). """ if not NUMPY_OK or samples is None or rt60 < 0.30: _L(st, f' [S5-JALAA] RT60={rt60:.2f}s < 0.30 — skip'); return wav if drr_before > DRR_ALREADY_DRY: _L(st, f' [S5-JALAA] DRR={drr_before:.1f}dB already dry — skip [I10]'); return wav fn = int(0.200 * SR) en = int(0.050 * SR) ln = int(0.450 * SR) # [B1] was 0.150 → 450ms (50-500ms window) n = len(samples) // fn if n < 5: return wav nf_vals = [] for i in range(n): s = i * fn; chunk = samples[s:s + fn] if len(chunk) < en + ln: continue er = float(np.sqrt(np.mean(chunk[:en] ** 2)) + 1e-10) lr = float(np.sqrt(np.mean(chunk[en:en + ln] ** 2)) + 1e-10) if er < 1e-5: continue if 20.0 * np.log10(er / lr) < 3.0: # reverb-dominated nf_vals.append(float(20 * np.log10(lr + 1e-10))) if not nf_vals: _L(st, ' [S5-JALAA] no reverb frames — skip'); return wav nf = float(np.clip(float(np.median(nf_vals)) + 3, -72, -20)) # v4: floor raised -25→-20 nr = 3 if rt60 > RT60_AGGR else 2 # v4: base raised (was 2/1) if st.mujawwad_conf > 0.6: nr = max(1, nr - 1) # v4: two-pass JALAA — first pass stationary, second adaptive tracking out = _tmp('s5a', st) rc, _, _ = _run(['ffmpeg', '-y', '-i', wav, '-af', f'afftdn=nr={nr}:nf={nf:.0f}:nt=w:tn=1', '-acodec', WAV_CODEC, '-ar', str(SR), '-ac', '1', '-loglevel', 'error', out]) if rc or not os.path.exists(out): return wav out2 = _tmp('s5b', st) rc2, _, _ = _run(['ffmpeg', '-y', '-i', out, '-af', f'afftdn=nr={max(1,nr-1)}:nf={nf+3:.0f}:nt=w:tn=1', '-acodec', WAV_CODEC, '-ar', str(SR), '-ac', '1', '-loglevel', 'error', out2]) if rc2 or not os.path.exists(out2): pass # keep single-pass result else: _cleanup(out); out = out2 post = _decode(out) if post is not None: d = _rmsdb(post) - _rmsdb(samples) if abs(d) > _RMS_MAX_DELTA: _cleanup(out); st.guard_reverts += 1 _L(st, f' [S5-JALAA] RMS Δ={d:+.2f}dB — REVERT'); return wav st.jalaa = True _L(st, f' [S5-JALAA] ✓ reverb_frames={len(nf_vals)}/{n} nf={nf:.0f}dB nr={nr}') return out # ─── Stage 6: Tail floor NR — 2-stage (afftdn + anlmdn) ───────────────────── def _tailnr(wav, samples, rt60, st): """ 2-stage NR pipeline (§83.3 order: stationary → non-stationary): Stage A: afftdn with nt=w (adaptive tracking) for stationary noise floor. Stage B: anlmdn (s=7:p=3:r=15) for transient/non-stationary residuals. RT60-scaled nr: base=2, +1 per 1.5s RT60 above threshold, cap=5 (§5.3: 20dB/nr max). [I3] If JALAA already ran, reduce nr by 1 to avoid double-attenuation. RL-16: sibilant SNR guard — revert if Safir/Tafasshi band drops > 2dB. """ if rt60 < RT60_TAILNR or samples is None or not NUMPY_OK: return wav fn = int(0.200 * SR) overall = _rmsdb(samples) fdb = np.array([float(20 * np.log10(np.sqrt(np.mean(samples[i:i+fn]**2)) + 1e-10)) for i in range(0, len(samples) - fn, fn)]) quiet = fdb[fdb < overall - 10] if len(quiet) == 0: return wav nf = float(np.clip(float(np.median(quiet)) + 4, -72, -25)) # v4: RT60-scaled nr — higher cap (8 was 5), stronger base nr = min(8, 3 + int((rt60 - RT60_TAILNR) / 1.2)) # was cap=5, base=2, step=1.5 if st.mujawwad_conf > 0.6: nr = max(1, nr - 1) if st.jalaa: nr = max(1, nr - 1) # [I3] # ── Stage A: afftdn with adaptive noise tracking ────────────────────── out_a = _tmp('s6a', st) rc, _, _ = _run(['ffmpeg', '-y', '-i', wav, '-af', f'afftdn=nr={nr}:nf={nf:.0f}:nt=w:om=o', '-acodec', WAV_CODEC, '-ar', str(SR), '-ac', '2', '-loglevel', 'error', out_a]) if rc or not os.path.exists(out_a): return wav # ── Stage B: anlmdn for non-stationary residuals ────────────────────── # anlmdn: s=patch size, p=context, r=search radius (§95 KB) # Scale strength with nr: gentle (s=5) at nr<=2, standard (s=7) at nr>2 patch = 7 if nr > 2 else 5 out_b = _tmp('s6b', st) rc2, _, _ = _run(['ffmpeg', '-y', '-i', out_a, '-af', f'anlmdn=s={patch}:p=3:r=15:m=1', '-acodec', WAV_CODEC, '-ar', str(SR), '-ac', '2', '-loglevel', 'error', out_b]) if rc2 or not os.path.exists(out_b): # Stage B failed — commit Stage A only st.tail_nr = True jalaa_note = ' (JALAA-adj)' if st.jalaa else '' _L(st, f' [S6-tailNR] ✓ 1-stage nr={nr}{jalaa_note} nf={nf:.0f}dB ' f'[anlmdn failed — Stage A only]') return out_a # ── Sibilant SNR guard (RL-16 spirit) ───────────────────────────────── post_s = _decode(out_b) if post_s is not None: sib_orig = _band_energy(samples, 3500, 8000) sib_post = _band_energy(post_s, 3500, 8000) sib_drop = sib_post - sib_orig if sib_drop < -2.0: _L(st, f' [S6-tailNR] sibilant drop {sib_drop:+.1f}dB — ' f'revert Stage B, keep Stage A') _cleanup(out_b) st.tail_nr = True _L(st, f' [S6-tailNR] ✓ nr={nr} nf={nf:.0f}dB [Stage A only]') return out_a # ── Stage C: v4 — gentle anlmdn second pass for reverb tail residual ──── out_c = _tmp('s6c', st) rc3, _, _ = _run(['ffmpeg', '-y', '-i', out_b, '-af', f'anlmdn=s=5:p=3:r=10:m=1', '-acodec', WAV_CODEC, '-ar', str(SR), '-ac', '2', '-loglevel', 'error', out_c]) if rc3 or not os.path.exists(out_c): pass # keep Stage B else: post_c = _decode(out_c) if post_c is not None: sib_c = _band_energy(post_c, 3500, 8000) sib_drop_c = sib_c - _band_energy(samples, 3500, 8000) if sib_drop_c >= -2.5: _cleanup(out_b); out_b = out_c else: _cleanup(out_c) _cleanup(out_a) st.tail_nr = True jalaa_note = ' (JALAA-adj)' if st.jalaa else '' _L(st, f' [S6-tailNR] ✓ 3-stage nr={nr}{jalaa_note} nf={nf:.0f}dB ' f'patch={patch} rt60={rt60:.1f}s') return out_b # ─── Stage 7: Arabic phoneme guards (§35/§52/§143/§152) ────────────────────── def _arabic_guards(orig_s, proc_wav, st): """ Seven guards verifying Tajweed-critical features survived processing. WARN-only by design — reverb is worse than mild phoneme loss. Results stored in st.guard_pass / st.guard_warn (structured). [I6] [B3] _band_energy() now samples across full file. [B4] G4 Ra-trill scans full audio in overlapping 1s windows. """ if not NUMPY_OK or orig_s is None: return proc_wav proc_s = _decode(proc_wav) if proc_s is None: return proc_wav n = min(len(orig_s), len(proc_s)) o = orig_s[:n]; p = proc_s[:n] def chk(name, cond, detail): entry = f'{name}: {detail}' if cond: st.guard_pass.append(entry); _L(st, f' [S7-PASS] {entry}') else: st.guard_warn.append(entry); _L(st, f' [S7-WARN] ⚠ {entry}') # G1 Ghunnah 250-300 Hz (§152.3) go = _band_energy(o, 250, 300); gp = _band_energy(p, 250, 300) if go > 1e-15: d = 10 * np.log10(gp / go + 1e-20) chk('G1-Ghunnah', d >= -3.0, f'{d:+.1f}dB {"✓" if d>=-3 else "⚠ nasal murmur lost"}') # G2 Ikhfa 250-400 Hz (§52.5) io = _band_energy(o, 250, 400); ip = _band_energy(p, 250, 400) if io > 1e-15: d = 10 * np.log10(ip / io + 1e-20) chk('G2-Ikhfa', d >= -4.0, f'{d:+.1f}dB {"✓" if d>=-4 else "⚠ ikhfa nasalisation lost"}') # G3 Qalqalah burst (§52.7, §143 Class 2) sil_n = int(0.020 * SR); bst_n = int(0.030 * SR) total = viol = 0 for i in range(0, n - sil_n - bst_n, sil_n): sr_ = float(np.sqrt(np.mean(o[i:i + sil_n] ** 2)) + 1e-10) br_ = float(np.sqrt(np.mean(o[i + sil_n:i + sil_n + bst_n] ** 2)) + 1e-10) if sr_ < 0.005 and br_ > sr_ * 5: total += 1 bp = float(np.sqrt(np.mean(p[i + sil_n:i + sil_n + bst_n] ** 2)) + 1e-10) if 20 * np.log10(bp / br_ + 1e-10) < -6.0: viol += 1 if total > 0: pct = viol / total * 100 chk('G3-Qalqalah', pct <= 20, f'{total} bursts {pct:.0f}% violated {"✓" if pct<=20 else "⚠ echo burst attenuated"}') # G4 Ra trill AM 25-35 Hz — FULL AUDIO scan in overlapping 1s windows [B4] win = SR; hop = SR // 2 am_ratios = [] for pos in range(0, n - win, hop): ef_o = np.abs(np.fft.rfft(np.abs(o[pos:pos + win]), n=win)) ef_p = np.abs(np.fft.rfft(np.abs(p[pos:pos + win]), n=win)) am_o = float(np.mean(ef_o[25:36])); am_p = float(np.mean(ef_p[25:36])) if am_o > 1e-8: am_ratios.append(am_p / am_o) if am_ratios: r = float(np.median(am_ratios)) chk('G4-Ra-trill', r >= 0.70, f'AM ratio={r:.2f} (median over {len(am_ratios)} windows) ' f'{"✓" if r>=0.70 else "⚠ ر trill may be smeared"}') # G5 Safir 5.5-12 kHz — ص س ز (§152.3) so = _band_energy(o, 5500, 12000); sp = _band_energy(p, 5500, 12000) if so > 1e-15: d = 10 * np.log10(sp / so + 1e-20) chk('G5-Safir', d >= -5.0, f'{d:+.1f}dB {"✓" if d>=-5 else "⚠ ص س ز may be dull"}') # G6 Tafasshi 3-8 kHz — ش (§152.3) to = _band_energy(o, 3000, 8000); tp = _band_energy(p, 3000, 8000) if to > 1e-15: d = 10 * np.log10(tp / to + 1e-20) chk('G6-Tafasshi', d >= -4.0, f'{d:+.1f}dB {"✓" if d>=-4 else "⚠ ش may lose spread"}') # G7 Izhar silence count (§52.5.1) def _sil_count(s): fn_ = int(0.010 * SR) db = np.array([float(20 * np.log10(np.sqrt(np.mean(s[i:i + fn_] ** 2)) + 1e-10)) for i in range(0, len(s) - fn_, fn_)]) med = float(np.median(db)); c = 0; in_s = False; sl = 0 for v in db: if v < med - 18: in_s = True; sl += 1 else: if in_s and 2 <= sl <= 8: c += 1 in_s = False; sl = 0 return c sc_o = _sil_count(o); sc_p = _sil_count(p) if sc_o > 0: r = sc_p / sc_o chk('G7-Izhar', r >= 0.70, f'silences {sc_o}→{sc_p} ({r:.0%}) {"✓" if r>=0.70 else "⚠ words may run together"}') total_g = len(st.guard_pass) + len(st.guard_warn) if st.guard_warn: _L(st, f' [S7] {len(st.guard_warn)}/{total_g} warnings — check output for Tajweed artifacts') else: _L(st, f' [S7] All {total_g} guards passed ✓') return proc_wav # ─── Main ───────────────────────────────────────────────────────────────────── def process(input_path, output_path, source_tier='TIER_UNKNOWN', mujawwad_conf=0.0, force_rt60=0.0, verbose=True): """ الصفاء v3 — main entry point. Returns SafaaState with full diagnostics. All temp files are tracked and cleaned up on exit. [I8] """ st = SafaaState(input_path=input_path, source_tier=source_tier, mujawwad_conf=mujawwad_conf) _L(st, f'\n{"═"*60}') _L(st, f' الصفاء {__version__} — إزالة الصدى والريفيرب') _L(st, f' Input : {Path(input_path).name}') _L(st, f' Tier : {source_tier} Mujawwad: {mujawwad_conf:.2f}') _L(st, f'{"═"*60}') s_orig = _decode(input_path) if s_orig is None: _L(st, ' ERROR: decode failed'); return st try: cur = input_path # S1: RT60 + DRR rt60 = force_rt60 if force_rt60 > 0 else _rt60(s_orig) st.rt60_initial = rt60; st.drr_before = _drr(s_orig) _L(st, f' [S1] RT60={rt60:.2f}s DRR={st.drr_before:.1f}dB') if rt60 < RT60_MIN: _L(st, f' [S1] RT60 < {RT60_MIN}s — no processing needed') _enc_stereo(input_path, output_path); return st # S2 Sub-band LF EQ cur = _lf_eq(cur, s_orig, rt60, st) # S3 WPE cur = _wpe(cur, rt60, st) # S4 DF3 (parallel) _s4 = _decode(cur); s4 = _s4 if _s4 is not None else s_orig cur = _df3(cur, s4, rt60, st) # S5 JALAA (DRR-weighted, corrected late window) _s5 = _decode(cur); s5 = _s5 if _s5 is not None else s_orig cur = _jalaa(cur, s5, rt60, st.drr_before, st) # S6 Tail NR (JALAA-aware) _s6 = _decode(cur); s6 = _s6 if _s6 is not None else s_orig cur = _tailnr(cur, s6, rt60, st) # S7 Arabic guards cur = _arabic_guards(s_orig, cur, st) # Final stereo encode + S8 volume boost + v4: presence lift 2kHz +1.5dB rc, _, err = _run(['ffmpeg', '-y', '-i', cur, '-af', ('volume=1.85,' 'equalizer=f=2000:width_type=o:width=1.0:g=1.5,' 'equalizer=f=300:width_type=o:width=1.0:g=-1.0,' 'alimiter=level_in=1:level_out=1:limit=0.89:attack=5:release=50'), '-acodec', WAV_CODEC, '-ar', str(SR), '-ac', '2', '-loglevel', 'error', output_path]) if rc: _L(st, f' ERROR: final encode: {err[:80]}'); return st sf_ = _decode(output_path) if sf_ is not None: st.drr_after = _drr(sf_) _L(st, f'\n{"═"*60}') _L(st, f' الصفاء {__version__} ✓') _L(st, f' RT60 : {st.rt60_initial:.2f}s') _L(st, f' DRR : {st.drr_before:.1f} → {st.drr_after:.1f} dB ' f'(Δ{st.drr_after - st.drr_before:+.1f})') _L(st, f' LF={st.lf_eq} WPE={st.wpe} DF3={st.df3} JALAA={st.jalaa} tailNR={st.tail_nr}') _L(st, f' Reverts:{st.guard_reverts} Warns:{len(st.guard_warn)}/{len(st.guard_pass)+len(st.guard_warn)}') _L(st, f'{"═"*60}\n') return st finally: _cleanup_all(st) # [I8] always runs # ─── CLI ────────────────────────────────────────────────────────────────────── if __name__ == '__main__': import argparse, json ap = argparse.ArgumentParser(description=f'الصفاء {__version__} — Dereverberation Engine') ap.add_argument('input') ap.add_argument('output') ap.add_argument('--tier', default='TIER_UNKNOWN') ap.add_argument('--mujawwad', type=float, default=0.0) ap.add_argument('--rt60', type=float, default=0.0, help='Force RT60 (0=auto)') a = ap.parse_args() st = process(a.input, a.output, source_tier=a.tier, mujawwad_conf=a.mujawwad, force_rt60=a.rt60) print('\n[REPORT]') print(json.dumps({ 'version': __version__, 'rt60': st.rt60_initial, 'drr_before': round(st.drr_before, 2), 'drr_after': round(st.drr_after, 2), 'drr_gain': round(st.drr_after - st.drr_before, 2), 'stages': { 'lf_eq': st.lf_eq, 'wpe': st.wpe, 'df3': st.df3, 'jalaa': st.jalaa, 'tail_nr': st.tail_nr, }, 'guard_reverts': st.guard_reverts, 'guard_pass': st.guard_pass, 'guard_warn': st.guard_warn, }, indent=2, ensure_ascii=False))