DeepFilterNet3 AXERA 量化部署包: axmodels (GRU 状态携带) + Python SDK + C++ 可执行文件

.gitattributes

@@ -34,3 +34,6 @@ saved_model/**/* filter=lfs diff=lfs merge=lfs -text
 *.zip filter=lfs diff=lfs merge=lfs -text
 *.zst filter=lfs diff=lfs merge=lfs -text
 *tfevents* filter=lfs diff=lfs merge=lfs -text
+assets/enhanced_sample.wav filter=lfs diff=lfs merge=lfs -text
+assets/noisy_snr0.wav filter=lfs diff=lfs merge=lfs -text
+bin/df3_ax filter=lfs diff=lfs merge=lfs -text

.gitignore

@@ -0,0 +1,8 @@
+# HF 仓库不需要的产物
+__pycache__/
+*.pyc
+outputs/
+*.log
+.msc
+.mv
+._____temp

README.md

@@ -1,3 +1,101 @@
 ---
+language:
+- zh
+- en
 license: mit
+pipeline_tag: audio-to-audio
+tags:
+- speech-enhancement
+- noise-reduction
+- axera
+- ax650
+- deepfilternet
 ---
+
+# DeepFilterNet3.AXERA
+
+DeepFilterNet3 (48kHz 全频带实时语音降噪) 的 AXERA 平台量化部署包 (当前支持 AX650N)。
+
+- 模型: enc (U16) / erb_dec (U16) / df_dec (全 FP32) 3 个子图, GRU 状态跨块携带版
+- 依赖: 仅 numpy + axengine (Python); C++ 为预编译 aarch64 可执行文件
+
+## 目录
+
+```
+├── axmodels/       # enc.axmodel / erb_dec.axmodel / df_dec.axmodel
+├── inference.py    # Python 推理入口 (依赖仅 numpy + axengine)
+├── requirements.txt
+├── assets/         # 演示音频 (带噪输入 + 增强输出样例)
+├── bin/            # C++ 可执行文件 (aarch64)
+└── run.sh          # 一键运行
+```
+
+## Python 运行
+
+```bash
+pip3 install numpy
+# pyaxengine: https://github.com/AXERA-TECH/pyaxengine/releases/latest
+pip3 install axengine-<version>-py3-none-any.whl
+./run.sh input.wav output.wav
+```
+
+或:
+
+```python
+from inference import enhance_audio
+out = enhance_audio(audio_float32_48k, model_dir="axmodels")
+```
+
+## C++ 运行
+
+```bash
+LD_LIBRARY_PATH=<ax_runtime>/lib ./bin/df3_ax \
+  axmodels/enc.axmodel axmodels/erb_dec.axmodel axmodels/df_dec.axmodel \
+  input.wav output.wav
+```
+
+## 模型说明
+
+- 输入: 48kHz 单声道 (PCM16 wav 或 float32 数组)
+- 3 子图流水线: STFT(fft 960/hop 480) -> ERB+unit_norm 特征 -> enc (GRU, 状态携带)
+  -> erb_dec (m 掩码) + df_dec (coefs) -> ERB 增益 + DF 滤波(5 阶, 2 帧前瞻) -> ISTFT
+- 静态 T=99 帧/块, GRU 状态跨块传递 (实时流式语义, 与官方 tract 部署一致)
+- 完整转换源码见 GitHub: [DeepFilterNet3.AXERA](https://github.com/ZY-2012/DeepFilterNet3.AXERA)
+
+## RTF（AX650N 实测, noisy_snr0.wav 10.6s）
+
+| 推理路径 | 耗时 | RTF | 峰值内存 |
+|------|------|------|------|
+| C++ | 1.50 s | 0.142 | — |
+| Python | 1.44 s | 0.136 | 28.9 MB |
+
+> RTF = 推理耗时 / 音频时长（不含模型加载，RTF < 1.0 即可实时）
+
+## 示例结果（AX650N 实测）
+
+| 音频 | 输入 RMS | 输出 RMS | 效果 |
+|------|---------|---------|------|
+| `assets/noisy_snr0.wav`（SNR 0dB 带噪语音） | 0.0640 | 0.0489 | 噪声移除、语音保留 |
+| `assets/enhanced_sample.wav`（上者降噪输出） | — | — | 试听对比用 |
+
+> 其他实测（不在本仓）：干净语音 RMS 0.0287→0.0270（几乎无损伤）;
+> 纯噪声 0.0336→0.0021（约 -24dB 压制）
+
+## 精度（AX650N 实测）
+
+| 指标 | 数值 |
+|------|------|
+| 逐张量 cosine vs ONNX（enc/erb_dec） | ≥ 0.9946（U16 量化） |
+| df_dec 输出 coefs | 0.99999（全 FP32） |
+| 板端端到端 vs 官方 torch 参考 | cosine 0.9857（量化损失仅 0.0005） |
+
+## 参考
+
+- [DeepFilterNet](https://github.com/Rikorose/DeepFilterNet)
+- [DeepFilterNet3.AXERA](https://github.com/ZY-2012/DeepFilterNet3.AXERA)（完整转换源码）
+- [Magnetar](https://github.com/AXERA-TECH/Magnetar) — AXERA 模型部署 agent 工具
+
+## 技术讨论
+
+- GitHub issues
+- QQ 群: 139953715

assets/enhanced_sample.wav

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:87cdbffb148bfac4f52def56a19d0f3b7bd90afac9fddf4963f4e5cbe6c249f9
+size 1017226

assets/noisy_snr0.wav

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:e1e08601f3b7ceb2f36d45c86343e38cd8927a73d7ad5526d6c4687c33aa7186
+size 1017226

axmodels/df_dec.axmodel

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:3dd647f3df37c3442d933ae4aef33a24b8c5d5d072aed782c772a4d747bbe5e8
+size 4193578

axmodels/enc.axmodel

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:87a1a741bffa304511adb0143e54876adda2e2286e702cee435fd62d2f613cd9
+size 1252300

axmodels/erb_dec.axmodel

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:d64fcca474a08263e4f21ff59bbc3e6281413ad359eb2cdaa99e6b3d6575834e
+size 2333248

bin/df3_ax

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:3464f3e384dad5b8b38e9531183fd499b29b3e78625e75d5e640e6816699e963
+size 262360

configuration.json

@@ -0,0 +1,20 @@
+{
+  "model_name": "DeepFilterNet3",
+  "supported_chips": ["AX650N"],
+  "quantization": {
+    "enc": "U16 (MinMax + SmoothQuant 0.5)",
+    "erb_dec": "U16 (MinMax + SmoothQuant 0.5)",
+    "df_dec": "FP32 (输出层精度保底)"
+  },
+  "sample_rate": 48000,
+  "fft_size": 960,
+  "hop_size": 480,
+  "chunk_frames": 99,
+  "gru_stateful": true,
+  "pulsar2_version": "7.0",
+  "axmodel_files": [
+    "axmodels/enc.axmodel",
+    "axmodels/erb_dec.axmodel",
+    "axmodels/df_dec.axmodel"
+  ]
+}

deepfilternet3_ax/__init__.py

@@ -0,0 +1,14 @@
+"""DeepFilterNet3 NPU 推理 SDK (Python, 仅依赖 numpy + axengine)。
+
+用法:
+    from deepfilternet3_ax import DeepFilterNet3
+    enh = DeepFilterNet3("models/")
+    out = enh.enhance(audio_float32_48k)   # (N,) -> (N,)
+
+命令行:
+    python -m deepfilternet3_ax input.wav -o output.wav [--model-dir models/]
+"""
+from .enhance import DeepFilterNet3, enhance_audio
+
+__version__ = "1.0.0"
+__all__ = ["DeepFilterNet3", "enhance_audio"]

deepfilternet3_ax/__main__.py

@@ -0,0 +1,58 @@
+"""命令行降噪: python -m deepfilternet3_ax input.wav -o output.wav
+
+wav 读写用标准库 wave (PCM16), 核心依赖仅 numpy + axengine。
+"""
+import argparse
+import time
+import wave
+from pathlib import Path
+
+import numpy as np
+
+
+def read_wav(path: Path):
+    with wave.open(str(path), "rb") as w:
+        assert w.getnchannels() == 1, "仅支持单声道"
+        sr = w.getframerate()
+        data = np.frombuffer(w.readframes(w.getnframes()), dtype=np.int16)
+    return (data.astype(np.float32) / 32768.0), sr
+
+
+def write_wav(path: Path, audio: np.ndarray, sr: int):
+    pcm = np.clip(audio, -1.0, 1.0)
+    pcm = (pcm * 32767.0).astype(np.int16)
+    with wave.open(str(path), "wb") as w:
+        w.setnchannels(1)
+        w.setsampwidth(2)
+        w.setframerate(sr)
+        w.writeframes(pcm.tobytes())
+
+
+def main():
+    parser = argparse.ArgumentParser(description="DeepFilterNet3 降噪 (AXERA 平台)")
+    parser.add_argument("input", type=Path, help="输入 48kHz 单声道 wav")
+    parser.add_argument("-o", "--output", type=Path, default=None,
+                        help="输出 wav (默认 <输入名>_enhanced.wav)")
+    parser.add_argument("--model-dir", type=Path, default=Path("models"),
+                        help="axmodel 目录 (默认 ./models)")
+    args = parser.parse_args()
+
+    from .enhance import DeepFilterNet3, SR
+
+    audio, sr = read_wav(args.input)
+    if sr != SR:
+        raise SystemExit(f"输入采样率 {sr}Hz 不支持, 请转成 48kHz 单声道")
+
+    enh = DeepFilterNet3(args.model_dir)
+    t0 = time.time()
+    out = enh.enhance(audio)
+    dt = time.time() - t0
+    rtf = dt / (len(audio) / SR)
+
+    out_path = args.output or args.input.with_name(args.input.stem + "_enhanced.wav")
+    write_wav(out_path, out, SR)
+    print(f"enhanced -> {out_path}  ({dt:.2f}s, RTF={rtf:.3f})")
+
+
+if __name__ == "__main__":
+    main()

deepfilternet3_ax/dsp.py

@@ -0,0 +1,233 @@
+"""DeepFilterNet3 numpy DSP 前后处理（SDK 核心, 仅依赖 numpy）。
+
+对齐来源:
+- 特征/STFT: libDF/src/lib.rs (DFState analysis/feat_erb/feat_cplx, erb_fb, norm alpha)
+- 模型语义: df/deepfilternet3.py DfNet.forward (pad_feat/pad_spec crop2/pad2)
+- DF 滤波: df/modules.py DfOp.forward_real_unfold (spec_pad(5, lookahead=2))
+- 流式语义: libDF/src/tract.rs (buf_x/buf_y 环形缓冲, 输出帧 = t-2)
+
+两种对齐变体:
+  batch    — 复刻 DfNet.forward: 特征 crop2/pad2 后进网络, m/coefs(t) 应用于 spec'(t)
+  streaming— 复刻 tract: 原始特征进网络, coefs(t) 滤波 spec(t-2+o), m(t) 应用于 spec(t)
+数值对分 (vs enhance() 输出) 决定最终采用哪种。
+"""
+
+import numpy as np
+
+SR = 48000
+FFT_SIZE = 960
+HOP = 480
+NB_ERB = 32
+NB_DF = 96
+DF_ORDER = 5
+DF_LOOKAHEAD = 2
+MIN_NB_ERB_FREQS = 2
+FREQ_SIZE = FFT_SIZE // 2 + 1  # 481
+
+MEAN_NORM_INIT = (-60.0, -90.0)
+UNIT_NORM_INIT = (0.001, 0.0001)
+
+
+def _freq2erb(f):
+    return 9.265 * np.log(1 + f / (24.7 * 9.265))
+
+
+def _erb2freq(e):
+    return 24.7 * 9.265 * (np.exp(e / 9.265) - 1)
+
+
+def erb_bandwidths(sr=SR, fft_size=FFT_SIZE, nb_bands=NB_ERB,
+                   min_nb_freqs=MIN_NB_ERB_FREQS):
+    """libDF erb_fb(): 每 ERB 带频点宽度 (总宽 = fft_size/2+1)。"""
+    nyq = sr / 2
+    freq_width = sr / fft_size
+    erb_low = _freq2erb(0.0)
+    erb_high = _freq2erb(nyq)
+    step = (erb_high - erb_low) / nb_bands
+    erb = []
+    prev_freq = 0
+    freq_over = 0
+    for i in range(1, nb_bands + 1):
+        f = _erb2freq(erb_low + i * step)
+        fb = int(round(f / freq_width))
+        nb_f = fb - prev_freq - freq_over
+        if nb_f < min_nb_freqs:
+            freq_over = min_nb_freqs - nb_f
+            nb_f = min_nb_freqs
+        else:
+            freq_over = 0
+        erb.append(nb_f)
+        prev_freq = fb
+    erb[nb_bands - 1] += 1
+    too_large = int(np.sum(erb)) - (fft_size // 2 + 1)
+    if too_large > 0:
+        erb[nb_bands - 1] -= too_large
+    assert int(np.sum(erb)) == fft_size // 2 + 1
+    return np.asarray(erb, dtype=np.int64)
+
+
+def norm_alpha(sr=SR, hop=HOP, tau=1.0):
+    """calc_norm_alpha: exp(-dt/tau) 保留 3 位小数 (libDF)。"""
+    alpha = np.exp(-(hop / sr) / tau)
+    a = 1.0
+    precision = 3
+    while a >= 1.0:
+        a = round(float(alpha) * 10 ** precision) / 10 ** precision
+        precision += 1
+    return a
+
+
+class DFStateNp:
+    """libDF DFState 的 numpy 复刻 (流式 STFT/ISTFT + stateful 特征)。"""
+
+    def __init__(self):
+        i = np.arange(FFT_SIZE)
+        self.window = np.sin(0.5 * np.pi * np.sin(0.5 * np.pi * (i + 0.5) / (FFT_SIZE / 2)) ** 2)
+        self.window = self.window.astype(np.float32)
+        self.wnorm = np.float32(1.0 / (FFT_SIZE ** 2 / (2 * HOP)))
+        self.erb_w = erb_bandwidths()
+        self.alpha = norm_alpha()
+        self.band_idx = np.searchsorted(np.cumsum(self.erb_w), np.arange(FREQ_SIZE))  # (481,)
+        self.reset()
+
+    def reset(self):
+        self.analysis_mem = np.zeros(FFT_SIZE - HOP, np.float32)
+        self.synthesis_mem = np.zeros(FFT_SIZE - HOP, np.float32)
+        self.mean_norm_state = np.linspace(*MEAN_NORM_INIT, NB_ERB).astype(np.float32)
+        self.unit_norm_state = np.linspace(*UNIT_NORM_INIT, NB_DF).astype(np.float32)
+
+    def analysis(self, frame):
+        buf = np.concatenate([self.analysis_mem, frame]) * self.window
+        self.analysis_mem = frame.copy()
+        return (np.fft.rfft(buf).astype(np.complex64) * self.wnorm)
+
+    def synthesis(self, spec):
+        # libDF realfft 逆变换不归一化 (wnorm 在 analysis 侧补偿), np.fft 自带 1/n, 需乘回
+        x = np.fft.irfft(spec, n=FFT_SIZE) * FFT_SIZE * self.window
+        out = x[: HOP] + self.synthesis_mem
+        self.synthesis_mem = x[HOP:].copy()
+        return out.astype(np.float32)
+
+    def feat_erb(self, spec):
+        p = spec.real ** 2 + spec.imag ** 2
+        e = np.add.reduceat(p, np.append([0], np.cumsum(self.erb_w)[:-1])) / self.erb_w
+        e = np.log10(e + 1e-10) * 10.0
+        s = self.mean_norm_state
+        s[:] = e * (1 - self.alpha) + s * self.alpha
+        return ((e - s) / 40.0).astype(np.float32)
+
+    def feat_cplx(self, spec):
+        x = spec[:NB_DF].copy()
+        s = self.unit_norm_state
+        s[:] = np.abs(x) * (1 - self.alpha) + s * self.alpha
+        return (x / np.sqrt(s)).astype(np.complex64)
+
+
+def frames_from_audio(audio, state):
+    """audio (N,) float32 -> [spec (481,) complex64] 流式帧序列。
+
+    帧数 = floor(N/hop), 与 libDF python binding 一致 (末尾不足一帧零填充)。
+    """
+    n_frames = len(audio) // HOP
+    specs = []
+    for f in range(n_frames):
+        s = f * HOP
+        frame = audio[s: s + HOP]
+        if len(frame) < HOP:
+            frame = np.pad(frame, (0, HOP - len(frame)))
+        specs.append(state.analysis(frame))
+    return specs
+
+
+def _crop2_pad2(x):
+    """DfNet.forward pad_feat/pad_spec: T 维头部 crop 2 帧 + 尾部补 2 零帧。"""
+    return np.concatenate([x[2:], np.zeros_like(x[:2])], axis=0)
+
+
+def _df_filter(spec_in, coefs, T_axis=0):
+    """spec_pad(5, lookahead=2): 前/后各 pad 2 帧, 逐帧复数乘加。
+
+    spec_in: (T, NB_DF) complex64; coefs: (T, NB_DF, DF_ORDER, 2) float32
+    -> (T, NB_DF) complex64: out(t) = Σ_o coefs(t,o) · spec_in(t-2+o)
+    """
+    T = spec_in.shape[T_axis]
+    padded = np.pad(spec_in, ((2, 2), (0, 0)))
+    out = np.zeros((T, NB_DF), np.complex64)
+    c = coefs[..., 0] + 1j * coefs[..., 1]  # (T, NB_DF, O)
+    for o in range(DF_ORDER):
+        out += padded[o: o + T] * c[..., o]
+    return out
+
+
+def pipeline(specs, enc_fn, erb_dec_fn, df_dec_fn, mode="streaming", T=99):
+    """完整频谱域流水线 (不含 STFT/ISTFT)。
+
+    specs: (T_all, 481) complex64
+    返回 (T_all, 481) complex64 增强谱。
+    mode: "batch" 复刻 DfNet.forward crop 语义; "streaming" 复刻 tract 环形缓冲语义
+    """
+    st = DFStateNp()
+    fb = np.stack([st.feat_erb(s) for s in specs])            # (T_all, 32)
+    fc = np.stack([st.feat_cplx(s) for s in specs])           # (T_all, 96) complex
+    feat_erb = fb[:, None, :]                                  # (T_all, 1, 32)
+    feat_spec = np.stack([fc.real, fc.imag], axis=0)           # (2, T_all, 96)
+
+    if mode == "batch":
+        feat_erb_in = _crop2_pad2(feat_erb)
+        feat_spec_in = _crop2_pad2(feat_spec)
+        spec_work = _crop2_pad2(specs)
+        m, coefs = _chunked(feat_erb_in, feat_spec_in, enc_fn, erb_dec_fn, df_dec_fn, T)
+        spec_m = spec_work * m[:, st.band_idx]
+        spec_e = np.empty_like(spec_work)
+        spec_e[:, :NB_DF] = _df_filter(spec_work[:, :NB_DF], coefs)
+        spec_e[:, NB_DF:] = spec_m[:, NB_DF:]
+        return spec_e
+    else:  # streaming
+        m, coefs = _chunked(feat_erb, feat_spec, enc_fn, erb_dec_fn, df_dec_fn, T)
+        spec_m = specs * m[:, st.band_idx]
+        spec_e = np.empty_like(specs)
+        spec_e[:, :NB_DF] = _df_filter(specs[:, :NB_DF], coefs)
+        spec_e[:, NB_DF:] = spec_m[:, NB_DF:]
+        return spec_e
+
+
+def _chunked(feat_erb, feat_spec, enc_fn, erb_dec_fn, df_dec_fn, T=99):
+    """分块跑 3 子图 (GRU 每块 h0=0)。
+
+    enc_fn(fb(1,1,T,32), fs(1,2,T,96)) -> dict(e0,e1,e2,e3,emb,c0,lsnr)
+    erb_dec_fn(emb,e3,e2,e1,e0) -> m (1,1,T,32)
+    df_dec_fn(emb,c0) -> coefs (1,T,96,10)
+    -> m_all (T_all,32), coefs_all (T_all,96,5,2)
+    """
+    T_all = feat_erb.shape[0]
+    n_ch = (T_all + T - 1) // T
+    ms, cs = [], []
+    for c in range(n_ch):
+        sl = slice(c * T, min((c + 1) * T, T_all))
+        t = sl.stop - sl.start
+        fb = np.zeros((1, 1, T, 32), np.float32)
+        fs = np.zeros((1, 2, T, 96), np.float32)
+        fb[:, 0, :t, :] = feat_erb[sl, 0]          # (t,32)
+        fs[0, :, :t, :] = feat_spec[:, sl]         # (2,t,96)
+        o = enc_fn(fb, fs)
+        m = erb_dec_fn(o["emb"], o["e3"], o["e2"], o["e1"], o["e0"])[0, 0]
+        coefs = df_dec_fn(o["emb"], o["c0"])[0]
+        coefs = coefs.reshape(T, NB_DF, DF_ORDER, 2)
+        ms.append(m[:t])
+        cs.append(coefs[:t])
+    return np.concatenate(ms), np.concatenate(cs)
+
+
+def enhance_offline(audio, enc_fn, erb_dec_fn, df_dec_fn, mode="streaming", T=99):
+    """离线增强 (对齐 enhance()): 输出与输入等长。
+
+    audio: (N,) float32 48kHz 单声道
+    """
+    st = DFStateNp()
+    audio_pad = np.concatenate([audio, np.zeros(FFT_SIZE, np.float32)])
+    specs = np.stack(frames_from_audio(audio_pad, st))          # (T_all, 481)
+    spec_e = pipeline(specs, enc_fn, erb_dec_fn, df_dec_fn, mode, T)
+    st.reset()
+    out = np.concatenate([st.synthesis(s) for s in spec_e]).astype(np.float32)
+    d = FFT_SIZE - HOP
+    return out[d: len(audio) + d]

deepfilternet3_ax/enhance.py

@@ -0,0 +1,92 @@
+"""DeepFilterNet3 端到端增强入口: 48kHz 单声道 float32 音频 -> 降噪后同形音频。
+
+仅依赖 numpy + axengine (NPU-only, 无 onnxruntime/torch 回退)。
+前后处理 (STFT/ERB 特征/掩码/DF 滤波/ISTFT) 为 libDF 的 numpy 复刻,
+已与官方 libDF + torch 参考对分 (cosine=1.0)。
+"""
+from __future__ import annotations
+
+from pathlib import Path
+
+import numpy as np
+
+from . import dsp
+from .session import DfNet3Ax
+
+SR = 48000
+
+
+class DeepFilterNet3:
+    """DeepFilterNet3 NPU 降噪器。
+
+    Attributes:
+        chunk_frames: 每块静态时间帧数 (99, 对应 1s @48k 流式 STFT)
+    """
+
+    chunk_frames = dsp.DFStateNp.__module__ and 99
+
+    def __init__(self, model_dir: str | Path):
+        self.net = DfNet3Ax(model_dir)
+        self.df_state = dsp.DFStateNp()
+
+    # ---- NPU 分块推理 (对齐 cache/validate_dsp2.py 的验证语义) ----
+    def _run_chunks(self, feat_erb, feat_spec, t_all):
+        T = self.chunk_frames
+        ms, cs = [], []
+        h_enc = None  # GRU 状态跨块携带 (stateful axmodel 时生效)
+        h_df = None
+        for c in range((t_all + T - 1) // T):
+            sl = slice(c * T, min((c + 1) * T, t_all))
+            t = sl.stop - sl.start
+            fb = np.zeros((1, 1, T, 32), np.float32)
+            fs = np.zeros((1, 2, T, 96), np.float32)
+            fb[:, 0, :t, :] = feat_erb[sl]
+            fs[0, :, :t, :] = feat_spec[:, sl]
+            o, h_enc = self.net.run_enc(fb, fs, h_enc)
+            m = self.net.run_erb_dec(o["emb"], o["e3"], o["e2"], o["e1"], o["e0"])
+            coefs, h_df = self.net.run_df_dec(o["emb"], o["c0"], h_df)
+            ms.append(m[0, 0][:t])
+            cs.append(coefs[0][:t])
+        m = np.concatenate(ms)
+        coefs = np.concatenate(cs).reshape(-1, 96, 5, 2)
+        return m, coefs
+
+    def enhance(self, audio: np.ndarray, sr: int = SR) -> np.ndarray:
+        """降噪。audio: (N,) float32 48kHz 单声道 -> (N,) float32。
+
+        输入非 48kHz 时建议调用方自行重采样。
+        """
+        if sr != SR:
+            raise ValueError(f"仅支持 48kHz 输入, got {sr}Hz (请先重采样)")
+        audio = np.asarray(audio, dtype=np.float32)
+        if audio.ndim != 1:
+            raise ValueError(f"audio 需为单声道一维数组, got shape {audio.shape}")
+
+        st = dsp.DFStateNp()
+        audio_pad = np.concatenate([audio, np.zeros(dsp.FFT_SIZE, np.float32)])
+        specs = np.stack(dsp.frames_from_audio(audio_pad, st))
+
+        # 特征 (全序列 stateful EMA, 与参考一致)
+        fb = np.stack([st.feat_erb(s) for s in specs])      # (T_all, 32)
+        fc = np.stack([st.feat_cplx(s) for s in specs])     # (T_all, 96) complex
+        feat_erb = fb                                          # (T_all, 32)
+        feat_spec = np.stack([fc.real, fc.imag], axis=0)       # (2, T_all, 96)
+
+        m, coefs = self._run_chunks(feat_erb, feat_spec, len(specs))
+
+        # 后处理: ERB 掩码 + DF 滤波 (流式语义, 与参考一致)
+        gains = m[:, st.band_idx]
+        spec_m = specs * gains
+        spec_e = np.empty_like(specs)
+        spec_e[:, :96] = dsp._df_filter(specs[:, :96], coefs)
+        spec_e[:, 96:] = spec_m[:, 96:]
+
+        st.reset()
+        out = np.concatenate([st.synthesis(s) for s in spec_e]).astype(np.float32)
+        d = dsp.FFT_SIZE - dsp.HOP
+        return out[d: len(audio) + d]
+
+
+def enhance_audio(audio: np.ndarray, model_dir: str | Path, sr: int = SR) -> np.ndarray:
+    """便捷入口: 一次调用完成加载与增强 (重复处理建议复用 DeepFilterNet3)。"""
+    return DeepFilterNet3(model_dir).enhance(audio, sr=sr)

deepfilternet3_ax/session.py

@@ -0,0 +1,123 @@
+"""3 个 axmodel 的 axengine 会话封装 (enc/erb_dec/df_dec 流水线)。"""
+from __future__ import annotations
+
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Any
+
+import numpy as np
+
+T = 99  # 静态时间帧 (1s @ 48kHz 流式 STFT)
+
+
+@dataclass(frozen=True)
+class TensorInfo:
+    name: str
+    shape: tuple[int, ...]
+    dtype: np.dtype
+
+
+def _numpy_dtype(value: Any) -> np.dtype:
+    text = str(value).lower()
+    mapping = (
+        (("tensor(float)", "float32", "fp32", "f32"), np.float32),
+        (("tensor(float16)", "float16", "fp16", "f16"), np.float16),
+        (("tensor(int64)", "int64", "s64"), np.int64),
+        (("tensor(int32)", "int32", "s32"), np.int32),
+        (("tensor(uint16)", "uint16", "u16"), np.uint16),
+        (("tensor(uint8)", "uint8", "u8"), np.uint8),
+    )
+    for aliases, dtype in mapping:
+        if any(alias in text for alias in aliases):
+            return np.dtype(dtype)
+    raise ValueError(f"unsupported runtime tensor dtype: {value}")
+
+
+def _tensor_info(value: Any) -> TensorInfo:
+    shape = getattr(value, "shape", None)
+    if shape is None:
+        shape = getattr(value, "dims", None)
+    if shape is None or any(dim is None for dim in shape):
+        raise ValueError(f"dynamic or missing tensor shape for {value.name}: {shape}")
+    dtype = getattr(value, "dtype", None)
+    if dtype is None:
+        dtype = getattr(value, "type", None)
+    return TensorInfo(
+        name=value.name,
+        shape=tuple(int(dim) for dim in shape),
+        dtype=_numpy_dtype(dtype),
+    )
+
+
+class _AxSession:
+    """单个 axmodel 的 axengine 会话 (参考 AXERA 官方推理工程封装)。"""
+
+    def __init__(self, model_path: str | Path):
+        import axengine
+        self.path = Path(model_path)
+        if not self.path.is_file():
+            raise FileNotFoundError(self.path)
+        self._session = axengine.InferenceSession(str(self.path))
+        self.inputs = [_tensor_info(v) for v in self._session.get_inputs()]
+        self.outputs = [_tensor_info(v) for v in self._session.get_outputs()]
+        self.input_by_name = {v.name: v for v in self.inputs}
+
+    def run(self, feed: dict[str, np.ndarray]) -> dict[str, np.ndarray]:
+        missing = [v.name for v in self.inputs if v.name not in feed]
+        if missing:
+            raise KeyError(f"missing inputs for {self.path.name}: {missing}")
+        prepared = {
+            name: np.ascontiguousarray(np.asarray(feed[name], dtype=meta.dtype))
+            for name, meta in self.input_by_name.items()
+        }
+        values = self._session.run(None, prepared)
+        if isinstance(values, dict):
+            return {name: np.asarray(v) for name, v in values.items()}
+        if not isinstance(values, (list, tuple)):
+            values = [values]
+        if len(values) != len(self.outputs):
+            raise RuntimeError(
+                f"unexpected output count from {self.path.name}: "
+                f"{len(values)} != {len(self.outputs)}")
+        return {meta.name: np.asarray(v) for meta, v in zip(self.outputs, values)}
+
+
+class DfNet3Ax:
+    """DeepFilterNet3 三子图 NPU 流水线 (与 dsp.pipeline 配合)。
+
+    enc:    feat_erb(1,1,T,32)+feat_spec(1,2,T,96) -> e0..e3, emb, c0, lsnr
+    erb_dec: emb,e3,e2,e1,e0 -> m(1,1,T,32)
+    df_dec:  emb,c0 -> coefs(1,T,96,10)
+    """
+
+    def __init__(self, model_dir: str | Path):
+        model_dir = Path(model_dir)
+        self.enc = _AxSession(model_dir / "enc.axmodel")
+        self.erb_dec = _AxSession(model_dir / "erb_dec.axmodel")
+        self.df_dec = _AxSession(model_dir / "df_dec.axmodel")
+        # 状态携带模式检测: axmodel 是否有 h0 输入
+        self.stateful = all("h0" in s.input_by_name for s in (self.enc, self.df_dec))
+
+    def run_enc(self, feat_erb: np.ndarray, feat_spec: np.ndarray,
+                h: np.ndarray | None = None) -> tuple[dict[str, np.ndarray], np.ndarray | None]:
+        feed = {"feat_erb": feat_erb, "feat_spec": feat_spec}
+        if self.stateful:
+            if h is None:
+                h = np.zeros((1, 1, 256), np.float32)
+            feed["h0"] = h
+            out = self.enc.run(feed)
+            return out, np.asarray(out["h_out"])
+        return self.enc.run(feed), None
+
+    def run_erb_dec(self, emb, e3, e2, e1, e0) -> np.ndarray:
+        return self.erb_dec.run({"emb": emb, "e3": e3, "e2": e2, "e1": e1, "e0": e0})["m"]
+
+    def run_df_dec(self, emb, c0, h: np.ndarray | None = None) -> tuple[np.ndarray, np.ndarray | None]:
+        feed = {"emb": emb, "c0": c0}
+        if self.stateful:
+            if h is None:
+                h = np.zeros((2, 1, 256), np.float32)
+            feed["h0"] = h
+            out = self.df_dec.run(feed)
+            return np.asarray(out["coefs"]), np.asarray(out["h_out"])
+        return self.df_dec.run(feed)["coefs"], None

inference.py

@@ -0,0 +1,52 @@
+"""DeepFilterNet3 推理入口 (AXERA 平台, 当前 AX650N) (依赖仅 numpy + axengine)。
+
+用法:
+    python3 inference.py input.wav [-o output.wav] [--model-dir axmodels]
+"""
+import argparse
+import wave
+from pathlib import Path
+
+import numpy as np
+
+SR = 48000
+
+
+def read_wav(path):
+    with wave.open(str(path), "rb") as w:
+        assert w.getnchannels() == 1, "仅支持单声道"
+        sr = w.getframerate()
+        data = np.frombuffer(w.readframes(w.getnframes()), dtype=np.int16)
+    return (data.astype(np.float32) / 32768.0), sr
+
+
+def write_wav(path, audio, sr):
+    pcm = (np.clip(audio, -1.0, 1.0) * 32767.0).astype(np.int16)
+    with wave.open(str(path), "wb") as w:
+        w.setnchannels(1)
+        w.setsampwidth(2)
+        w.setframerate(sr)
+        w.writeframes(pcm.tobytes())
+
+
+def main():
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("input", type=Path)
+    parser.add_argument("-o", "--output", type=Path, default=None)
+    parser.add_argument("--model-dir", type=Path, default=Path("axmodels"))
+    args = parser.parse_args()
+
+    from deepfilternet3_ax import DeepFilterNet3
+
+    audio, sr = read_wav(args.input)
+    if sr != SR:
+        raise SystemExit(f"仅支持 48kHz 输入, got {sr}Hz")
+    enh = DeepFilterNet3(args.model_dir)
+    out = enh.enhance(audio)
+    out_path = args.output or args.input.with_name(args.input.stem + "_enhanced.wav")
+    write_wav(out_path, out, SR)
+    print(f"enhanced -> {out_path}")
+
+
+if __name__ == "__main__":
+    main()

requirements.txt

@@ -0,0 +1,4 @@
+numpy>=1.24
+# pyaxengine (板端 axengine, 从官方 release 下载 wheel 安装):
+#   https://github.com/AXERA-TECH/pyaxengine/releases/latest
+#   pip3 install axengine-<version>-py3-none-any.whl

run.sh

@@ -0,0 +1,16 @@
+#!/usr/bin/env bash
+# DeepFilterNet3 一键降噪 (Python)
+# 用法: ./run.sh input.wav [output.wav]
+set -euo pipefail
+cd "$(dirname "$0")"
+
+INPUT="${1:?用法: ./run.sh input.wav [output.wav]}"
+OUTPUT="${2:-${INPUT%.wav}_enhanced.wav}"
+
+PYTHON_BIN="${PYTHON_BIN:-python3}"
+if [[ -x /root/miniforge3/bin/python ]]; then
+  PYTHON_BIN=/root/miniforge3/bin/python
+fi
+
+PYTHONPATH="$(pwd)${PYTHONPATH:+:$PYTHONPATH}" \
+  "$PYTHON_BIN" inference.py "$INPUT" -o "$OUTPUT" --model-dir "$(pwd)/axmodels"