修块间台阶残留: 归一化增益改块内逐样本线性插值(zipper 噪声); 谐波(相对基频) +0/+6dB 由 -12.1/-15.9 降到 -90.1/-70.6 dB; 调试用的测试脚本收进 测试/, 清理全部调试残留

dsp/loudness_norm.py

@@ -181,6 +181,7 @@ def main() -> None:
     # 平滑状态
     energy_smooth = 0.0
     gain_db = 0.0
+    g_prev = 1.0                       # 上一块末尾的线性增益(块间斜坡用)
     _n: int = 0
 
     block_seconds = BLOCK / 4.0 / RATE          # 每块时长(单声道 float32)
@@ -286,7 +287,10 @@ def main() -> None:
                 gain_db = 20.0 * math.log10(g + EPS)     # 状态跟随, 下块再按限速恢复
 
             # ---- 6) 施加增益 ----
-            y = x * g
+            # 块内逐样本线性过渡: 整块共用一个增益会在块边界留下台阶(zipper 噪声),
+            # 表现为中低频的微弱谐波。用"上一块增益 -> 本块增益"的斜坡把它抹平。
+            y = x * np.linspace(g_prev, g, x.size, dtype=np.float64)
+            g_prev = g
             if fade_left > 0:                           # 起播淡入: 线性爬升, 压掉卷积过冲
                 y = y * ((FADE_BLOCKS - fade_left + 1) / float(FADE_BLOCKS))
                 fade_left -= 1

测试/probe3pt.py

@@ -0,0 +1,68 @@
+"""三点电平测量: 虚拟声卡入口 / HRTF 输入(=归一化输出) / 数字输出。用来定位哪一级没处理。"""
+import array
+import json
+import math
+import subprocess
+import time
+import wave
+
+RATE = 48000
+DEVNULL = subprocess.DEVNULL
+SRC = "/tmp/h_white.wav"
+
+
+def ids() -> dict[str, int]:
+    d = json.loads(subprocess.run(["pw-dump"], capture_output=True, text=True).stdout)
+    out = {}
+    for n in d:
+        if n.get("type") != "PipeWire:Interface:Node":
+            continue
+        p = n.get("info", {}).get("props", {}) or {}
+        nm = p.get("node.name") or ""
+        if p.get("media.class") == "Audio/Sink":
+            if nm == "collaplex_vsink":
+                out["vsink"] = n["id"]
+            elif nm == "collaplex_hrtf_in":
+                out["hrtf_in"] = n["id"]
+            elif nm.endswith("iec958-stereo"):
+                out["out"] = n["id"]
+        elif nm == "alsa_input.usb-EDIFIER_Technology_EDIFIER_Fit900NB_4250315939393214-00.mono-fallback":
+            out["mic"] = n["id"]
+    return out
+
+
+def rec(target: int, secs: float, path: str) -> float:
+    r = subprocess.Popen(["pw-record", "--target", str(target), "-P", "{ stream.capture.sink = true }",
+                          "--rate", str(RATE), "--channels", "2", path], stderr=DEVNULL)
+    time.sleep(secs)
+    r.terminate()
+    r.wait()
+    w = wave.open(path)
+    n = w.getnframes()
+    d = array.array("h")
+    d.frombytes(w.readframes(n))
+    if not d:
+        return -99
+    rms = (sum(x * x for x in d) / len(d)) ** 0.5 / 32768
+    return 20 * math.log10(rms) if rms > 0 else -99
+
+
+ID = ids()
+print("节点 id: %s" % ID)
+
+print("\n--- 静音时(应 -99 = 抓的是 sink monitor; 若 -50 左右 = 抓到麦克风) ---")
+for k in ("vsink", "hrtf_in", "out", "mic"):
+    if k in ID:
+        print("  %-8s %.2f dBFS" % (k, rec(ID[k], 1.0, "/tmp/t_%s.wav" % k)))
+
+print("\n--- 播素材时(素材 -12 dBFS) ---")
+p = subprocess.Popen(["pw-play", "--target", "collaplex_vsink", SRC], stderr=DEVNULL)
+time.sleep(0.5)
+res = {}
+for k in ("vsink", "hrtf_in", "out", "mic"):
+    if k in ID:
+        res[k] = rec(ID[k], 1.2, "/tmp/p_%s.wav" % k)
+p.terminate()
+p.wait()
+for k, v in res.items():
+    print("  %-8s %.2f dBFS" % (k, v))

测试/test_bypass.py

@@ -0,0 +1,81 @@
+"""决定性对比: 直接播到设备(旁路链路) vs 经过 Collaplex 链路, 噪声底谁更高。"""
+from __future__ import annotations
+
+import json
+import subprocess
+import time
+import wave
+
+import numpy as np
+from scipy.io import wavfile
+
+RATE = 96000
+SRC = "/tmp/probe_silent.wav"
+REC = "/tmp/rec_bypass.wav"
+SINK = ""
+
+
+def find_digital_sink() -> str:
+    raw = subprocess.run(["pw-dump"], capture_output=True, text=True).stdout
+    for node in json.loads(raw):
+        props = (node.get("info") or {}).get("props") or {}
+        name = str(props.get("node.name", ""))
+        if props.get("media.class") == "Audio/Sink" and "iec958" in name:
+            return name
+    return ""
+
+
+def write_silent() -> None:
+    with wave.open(SRC, "wb") as fh:
+        fh.setnchannels(2)
+        fh.setsampwidth(2)
+        fh.setframerate(RATE)
+        fh.writeframes(np.zeros(RATE * 4 * 2, dtype="<i2").tobytes())
+
+
+def measure(target: str, label: str) -> None:
+    player = subprocess.Popen(["pw-play", "--target", target, SRC],
+                              stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
+    time.sleep(0.4)
+    rec = subprocess.Popen(["timeout", "3", "pw-record", "--target", SINK,
+                            "-P", "{ stream.capture.sink = true }",
+                            "--rate", str(RATE), "--channels", "2", "--format", "f32", REC],
+                           stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
+    rec.wait(timeout=12)
+    player.terminate()
+    player.wait(timeout=5)
+    try:
+        _rate, data = wavfile.read(REC)
+    except (OSError, ValueError):
+        print("  %-28s 读不到录音" % label)
+        return
+    arr = data.astype(np.float64)
+    if arr.ndim > 1:
+        arr = arr[:, 0]
+    if arr.size < 1000:
+        print("  %-28s 录音太短" % label)
+        return
+    rms = float(np.sqrt(np.mean(arr ** 2)))
+    peak = float(np.max(np.abs(arr)))
+    print("  %-28s RMS %8.2f dBFS | 峰值 %8.2f dBFS" % (
+        label, 20 * np.log10(max(rms, 1e-12)), 20 * np.log10(max(peak, 1e-12))))
+
+
+def main() -> None:
+    global SINK
+    SINK = find_digital_sink()
+    if not SINK:
+        print("找不到数字输出")
+        return
+    write_silent()
+    print("设备:", SINK)
+    print("素材: 全零(4 秒), 谁在引入噪声?")
+    print()
+    measure(SINK, "① 旁路(直接播到设备)")
+    measure("collaplex_vsink", "② 经 Collaplex 链路")
+    print()
+    print("判据: ② - ① 的差值 = 链路自己引入的噪声。差值 < 6 dB 基本可忽略。")
+
+
+if __name__ == "__main__":
+    main()

测试/test_clip.py

@@ -0,0 +1,44 @@
+"""复现削波: 喂一个 -6 dBFS 正弦, 总音量 +12 dB, 看输出是被线性放大还是被削平。"""
+from __future__ import annotations
+
+import os
+import subprocess
+import sys
+
+import numpy as np
+
+DSP_DIR = "/home/lou/桌面/工作区/实验/collaplex音效/dsp"
+sys.path.insert(0, DSP_DIR)
+
+import common  # noqa: E402
+
+RATE = 96000
+PY = "/home/lou/桌面/工作区/实验/collaplex音效/.venv/bin/python"
+EQ = os.path.join(DSP_DIR, "eq.py")
+
+
+def run(x: np.ndarray, gains: list[float], vol_db: float) -> np.ndarray:
+    store = common.open_store()
+    common.write_params(store, gains, vol_db)
+    env = dict(os.environ, CX_RATE=str(RATE))
+    proc = subprocess.run([PY, EQ], input=x.astype("<f4").tobytes(),
+                          capture_output=True, timeout=60, env=env)
+    return np.frombuffer(proc.stdout, dtype="<f4").astype(np.float64)
+
+
+flat = [0.0] * 32
+t = np.arange(int(RATE * 0.4)) / RATE
+x = (0.5 * np.sin(2 * np.pi * 1000 * t)).astype(np.float64)   # -6 dBFS 正弦
+
+for vol in (0.0, 6.0, 12.0):
+    y = run(x, flat, vol)
+    seg = y[5000:]
+    peak = float(np.max(np.abs(seg)))
+    at_rail = float(np.mean(np.abs(seg) >= 0.9999)) * 100.0
+    rms = float(np.sqrt(np.mean(seg * seg)))
+    print("总音量 %+5.1f dB -> 峰值 %.4f (%.2f dBFS) | 贴顶样本 %.1f%% | RMS %.4f" % (
+        vol, peak, 20 * np.log10(max(peak, 1e-12)), at_rail, rms))
+
+print()
+print("理论: -6 dBFS 正弦经 +6 dB 应到 %.3f, 经 +12 dB 应到 %.3f" % (
+    0.5 * 10 ** (6 / 20), 0.5 * 10 ** (12 / 20)))

测试/test_e2e_eq.py

@@ -0,0 +1,106 @@
+"""端到端: 虚拟声卡 -> 归一化 -> HRTF -> EQ -> 数字输出, 验证 EQ 段增益与总音量。"""
+from __future__ import annotations
+
+import array
+import json
+import math
+import os
+import subprocess
+import sys
+import time
+import wave
+
+import numpy as np
+
+DSP_DIR = "/home/lou/桌面/工作区/实验/collaplex音效/dsp"
+sys.path.insert(0, DSP_DIR)
+import common  # noqa: E402
+
+RATE = 96000
+WAV = "/tmp/e2e_tone.wav"
+DEVNULL = subprocess.DEVNULL
+store = common.open_store()
+
+
+def nodes() -> dict[str, int]:
+    out = subprocess.run(["pw-dump"], capture_output=True, text=True).stdout
+    res: dict[str, int] = {}
+    for n in json.loads(out):
+        if n.get("type") != "PipeWire:Interface:Node":
+            continue
+        props = (n.get("info") or {}).get("props") or {}
+        name = props.get("node.name") or ""
+        if name:
+            res[name] = int(n["id"])
+    return res
+
+
+def make_tone(freq: float = 1000.0, secs: float = 6.0, db: float = -12.0) -> None:
+    t = np.arange(int(RATE * secs)) / RATE
+    amp = 10.0 ** (db / 20.0) * math.sqrt(2.0)
+    mono = np.clip(amp * np.sin(2.0 * math.pi * freq * t), -1.0, 1.0)
+    inter = np.empty(mono.size * 2, dtype="<i2")
+    inter[0::2] = (mono * 32767).astype("<i2")
+    inter[1::2] = (mono * 32767).astype("<i2")
+    with wave.open(WAV, "wb") as w:
+        w.setnchannels(2)
+        w.setsampwidth(2)
+        w.setframerate(RATE)
+        w.writeframes(inter.tobytes())
+
+
+def play_and_record(target: str, rec_sink: str, secs: float = 6.0) -> np.ndarray:
+    rec_path = "/tmp/e2e_rec.wav"
+    rec = subprocess.Popen(["pw-record", "--target", str(rec_sink), "-P",
+                            "{ stream.capture.sink = true }", "--rate", str(RATE),
+                            "--channels", "2", rec_path],
+                           stdout=DEVNULL, stderr=DEVNULL)
+    time.sleep(0.5)
+    play = subprocess.Popen(["pw-play", "--target", target, WAV], stdout=DEVNULL, stderr=DEVNULL)
+    time.sleep(secs)
+    play.terminate()
+    play.wait()
+    time.sleep(0.3)
+    rec.terminate()
+    rec.wait()
+    with wave.open(rec_path, "rb") as w:
+        data = np.frombuffer(w.readframes(w.getnframes()), dtype="<i2").astype(np.float64) / 32768.0
+    block = data[2::2]
+    return block[int(RATE * 2.0): int(RATE * 2.0) + int(RATE * 2.0)]
+
+
+def rms_db(x: np.ndarray) -> float:
+    return float(20 * np.log10(np.sqrt(np.mean(x * x)) + 1e-12))
+
+
+def main() -> None:
+    ids = nodes()
+    sink_id = next(v for k, v in ids.items() if "iec958" in k)
+    make_tone()
+
+    flat = [0.0] * common.EQ_BANDS
+    common.write_params(store, flat, 0.0)
+    time.sleep(0.3)
+    base = rms_db(play_and_record("collaplex_vsink", sink_id))
+    print("链路节点: %s" % ", ".join(sorted(k for k in ids if k.startswith("collaplex"))))
+    print("平直基线(1 kHz 经 HRTF): %+.2f dBFS" % base)
+
+    gains = flat.copy()
+    gains[17] = 12.0
+    common.write_params(store, gains, 0.0)
+    time.sleep(0.3)
+    boosted = rms_db(play_and_record("collaplex_vsink", sink_id))
+    print("1 kHz 段 +12 dB:        %+.2f dBFS  (差 %+.2f dB)" % (boosted, boosted - base))
+
+    common.write_params(store, flat, 6.0)
+    time.sleep(0.3)
+    vol = rms_db(play_and_record("collaplex_vsink", sink_id))
+    print("总音量 +6 dB:           %+.2f dBFS  (差 %+.2f dB)" % (vol, vol - base))
+
+    common.write_params(store, flat, 0.0)
+    time.sleep(0.2)
+    print("已恢复平直 / 0 dB")
+
+
+if __name__ == "__main__":
+    main()

测试/test_eq.py

@@ -0,0 +1,63 @@
+"""eq.py 自测: 平直直通 / 单段提升 / 总音量 / 邻段不串扰。"""
+from __future__ import annotations
+
+import os
+import subprocess
+import sys
+
+import numpy as np
+
+DSP_DIR = "/home/lou/桌面/工作区/实验/collaplex音效/dsp"
+sys.path.insert(0, DSP_DIR)
+
+import common  # noqa: E402
+
+RATE = 96000
+EQ = os.path.join(DSP_DIR, "eq.py")
+store = common.open_store()
+
+
+def run(signal: np.ndarray, gains: list[float], vol_db: float) -> np.ndarray:
+    common.write_params(store, gains, vol_db)
+    proc = subprocess.run([str(DSP_DIR + "/../.venv/bin/python"), EQ], input=signal.astype("<f4").tobytes(),
+                          capture_output=True, timeout=120,
+                          env={**os.environ, "CX_RATE": str(RATE)})
+    if proc.returncode != 0:
+        print("DSP stderr:", proc.stderr.decode()[:400])
+    return np.frombuffer(proc.stdout, dtype="<f4").astype(np.float64)
+
+
+def tone(freq: float, secs: float = 0.5, amp: float = 0.3) -> np.ndarray:
+    t = np.arange(int(RATE * secs)) / RATE
+    return (amp * np.sin(2 * np.pi * freq * t)).astype(np.float32)
+
+
+def rms_db(x: np.ndarray) -> float:
+    return float(20 * np.log10(np.sqrt(np.mean(x * x)) + 1e-12))
+
+
+flat = [0.0] * common.EQ_BANDS
+x = tone(1000.0)
+tail = slice(4000, None)
+
+print("== 1kHz 正弦, 比较输出/输入 ==")
+print("平直(应 0 dB):          %+.2f dB" % (rms_db(run(x, flat, 0.0)[tail]) - rms_db(x[tail])))
+
+g = flat.copy()
+g[17] = 12.0
+print("1kHz 段 +12(应 ~+12):   %+.2f dB" % (rms_db(run(x, g, 0.0)[tail]) - rms_db(x[tail])))
+
+print("总音量 +6(应 ~+6):      %+.2f dB" % (rms_db(run(x, flat, 6.0)[tail]) - rms_db(x[tail])))
+
+g = flat.copy()
+g[7] = 12.0
+print("100Hz 段 +12(应 ~0):    %+.2f dB" % (rms_db(run(x, g, 0.0)[tail]) - rms_db(x[tail])))
+
+low = tone(100.0)
+g = flat.copy()
+g[7] = 12.0
+print("100Hz 段 +12 测 100Hz:  %+.2f dB" % (rms_db(run(low, g, 0.0)[tail]) - rms_db(low[tail])))
+
+print()
+print("== 状态(共享内存) ==")
+print(common.read_state(store))

测试/test_hr.py

@@ -0,0 +1,111 @@
+"""HRTF 独立实例的接入/输出测试: 播到 collaplex_hrtf_in, 录数字输出。"""
+import array
+import json
+import math
+import random
+import subprocess
+import time
+import wave
+
+RATE = 48000
+DEVNULL = subprocess.DEVNULL
+import sys
+IN_SINK = sys.argv[1] if len(sys.argv) > 1 else "collaplex_hrtf_in"
+
+
+def exit_id() -> int:
+    d = json.loads(subprocess.run(["pw-dump"], capture_output=True, text=True).stdout)
+    for n in d:
+        if n.get("type") == "PipeWire:Interface:Node":
+            p = n.get("info", {}).get("props", {}) or {}
+            if p.get("media.class") == "Audio/Sink" and (p.get("node.name") or "").endswith("iec958-stereo"):
+                return int(n["id"])
+    raise SystemExit("找不到数字输出")
+
+
+def gen_white(path: str, db: float = -12.0, secs: float = 6.0) -> None:
+    n = int(RATE * secs)
+    rnd = random.Random(11)
+    x = [rnd.gauss(0, 1) for _ in range(n)]
+    cur = (sum(v * v for v in x) / n) ** 0.5
+    k = (10 ** (db / 20)) * 32768 / cur
+    d = array.array("h")
+    for v in x:
+        s = int(max(-32760, min(32760, v * k)))
+        d.append(s)
+        d.append(s)
+    with wave.open(path, "wb") as w:
+        w.setnchannels(2)
+        w.setsampwidth(2)
+        w.setframerate(RATE)
+        w.writeframes(d.tobytes())
+
+
+def gen_impulse(path: str, lead: float = 0.5, secs: float = 3.0) -> None:
+    d = array.array("h", [0] * (int(RATE * secs) * 2))
+    i = int(lead * RATE) * 2
+    d[i] = d[i + 1] = 30000
+    with wave.open(path, "wb") as w:
+        w.setnchannels(2)
+        w.setsampwidth(2)
+        w.setframerate(RATE)
+        w.writeframes(d.tobytes())
+
+
+def capture(src: str, rec_secs: float) -> str:
+    out = "/tmp/hr_out.wav"
+    r = subprocess.Popen(["pw-record", "--target", str(exit_id()), "-P", "{ stream.capture.sink = true }",
+                          "--rate", str(RATE), "--channels", "2", out], stderr=DEVNULL)
+    time.sleep(0.5)
+    p = subprocess.Popen(["pw-play", "--target", IN_SINK, src], stderr=DEVNULL)
+    time.sleep(rec_secs)
+    for x in (r, p):
+        x.terminate()
+        x.wait()
+    return out
+
+
+def load(path: str) -> tuple[list[int], list[int], int]:
+    w = wave.open(path)
+    fr = w.getframerate()
+    d = array.array("h")
+    d.frombytes(w.readframes(w.getnframes()))
+    return list(d[0::2]), list(d[1::2]), fr
+
+
+def rms_db(seg: list[int]) -> float:
+    if not seg:
+        return -99
+    v = (sum(x * x for x in seg) / len(seg)) ** 0.5 / 32768
+    return 20 * math.log10(v) if v > 0 else -99
+
+
+def corr(a: list[int], b: list[int]) -> float:
+    n = min(len(a), len(b))
+    a, b = a[:n], b[:n]
+    ma, mb = sum(a) / n, sum(b) / n
+    va = sum((x - ma) ** 2 for x in a)
+    vb = sum((x - mb) ** 2 for x in b)
+    if va <= 0 or vb <= 0:
+        return 0.0
+    return sum((a[i] - ma) * (b[i] - mb) for i in range(n)) / math.sqrt(va * vb)
+
+
+gen_white("/tmp/hr_white.wav")
+gen_impulse("/tmp/hr_imp.wav")
+
+print("【接入→输出: 白噪(左右相同) -12 dBFS 推入 collaplex_hrtf_in】")
+f = capture("/tmp/hr_white.wav", 4.0)
+L, R, fr = load(f)
+s = slice(2 * fr, 4 * fr)
+print("  左 %7.2f dBFS   右 %7.2f dBFS" % (rms_db(L[s]), rms_db(R[s])))
+print("  左右相关系数 %.3f   (直通=1.000; 双耳化后应明显 <1)" % corr(L[s], R[s]))
+
+print("\n【脉冲响应(混响湿量 0.3 的尾巴)】")
+f = capture("/tmp/hr_imp.wav", 2.5)
+L, R, fr = load(f)
+mx = max(range(len(L)), key=lambda i: abs(L[i]))
+print("  峰值 %.1f ms" % (mx / fr * 1000))
+for lo, hi in ((0, 3), (3, 15), (15, 40), (40, 100), (100, 250), (250, 600), (600, 1200), (1200, 2000)):
+    a, b = mx + int(lo * fr / 1000), mx + int(hi * fr / 1000)
+    print("   %5d~%-5d ms: %7.2f dBFS" % (lo, hi, rms_db(L[a:b])))

测试/test_noise.py

@@ -0,0 +1,81 @@
+"""① 块长一对一验证 ② 数字输出底噪测量。"""
+from __future__ import annotations
+
+import json
+import os
+import subprocess
+import sys
+
+import numpy as np
+from scipy.io import wavfile
+
+PROJ = "/home/lou/桌面/工作区/实验/collaplex音效"
+sys.path.insert(0, os.path.join(PROJ, "dsp"))
+
+import common  # noqa: E402
+
+PY = os.path.join(PROJ, ".venv/bin/python")
+EQ = os.path.join(PROJ, "dsp/eq.py")
+REC = "/tmp/noise_probe.wav"
+
+
+def find_digital_sink() -> str:
+    raw = subprocess.run(["pw-dump"], capture_output=True, text=True).stdout
+    for node in json.loads(raw):
+        props = (node.get("info") or {}).get("props") or {}
+        name = str(props.get("node.name", ""))
+        if props.get("media.class") == "Audio/Sink" and "iec958" in name:
+            return name
+    return ""
+
+
+def block_test() -> None:
+    print("=== ① 块长一对一(不等长块不应被填充) ===")
+    store = common.open_store()
+    for nbytes in (4096, 2048, 1024, 512, 384):
+        n = nbytes // 4
+        tone = np.sin(2 * np.pi * 1000 * np.arange(n) / 96000.0)
+        x = (0.3 * tone).astype("<f4")
+        common.write_params(store, [0.0] * 32, 0.0)
+        proc = subprocess.run([PY, EQ], input=x.tobytes(), capture_output=True, timeout=60)
+        ok = len(proc.stdout) == len(x.tobytes())
+        print("  喂 %5d 字节 -> 出 %5d 字节   %s" % (len(x.tobytes()), len(proc.stdout),
+                                                    "✓ 一对一" if ok else "✗ 被填充!"))
+
+
+def noise_test(sink: str) -> None:
+    print()
+    print("=== ② 数字输出底噪(静默 3 秒, 单位 dBFS) ===")
+    if not sink:
+        print("  找不到数字输出, 跳过")
+        return
+    for label, extra in (("静默(无播放)", []),):
+        subprocess.run(["timeout", "3", "pw-record", "--target", sink,
+                        "-P", "{ stream.capture.sink = true }",
+                        "--rate", "96000", "--channels", "2", "--format", "f32", REC, *extra],
+                       stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
+        try:
+            _rate, data = wavfile.read(REC)
+        except (OSError, ValueError):
+            print("  录音失败")
+            return
+        arr = data.astype(np.float64)
+        if arr.ndim > 1:
+            arr = arr[:, 0]
+        if arr.size == 0:
+            print("  空录音")
+            return
+        rms = float(np.sqrt(np.mean(arr ** 2)))
+        peak = float(np.max(np.abs(arr)))
+        print("  %-12s RMS %7.2f dBFS | 峰值 %7.2f dBFS | 样点 %d" % (
+            label, 20 * np.log10(max(rms, 1e-12)), 20 * np.log10(max(peak, 1e-12)), arr.size))
+        print("  (参考: 数字域本底一般应低于 -90 dBFS; 明显高于此值说明链路在自造噪声)")
+
+
+def main() -> None:
+    block_test()
+    noise_test(find_digital_sink())
+
+
+if __name__ == "__main__":
+    main()

测试/test_playing_noise.py

@@ -0,0 +1,101 @@
+"""修复后: 有流但内容为静音/纯音时, 输出端有没有链路自造的噪声。"""
+from __future__ import annotations
+
+import json
+import os
+import subprocess
+import sys
+import time
+import wave
+
+import numpy as np
+from scipy.io import wavfile
+
+PROJ = "/home/lou/桌面/工作区/实验/collaplex音效"
+RATE = 96000
+SILENT = "/tmp/probe_silent.wav"
+TONE = "/tmp/probe_tone.wav"
+REC_S = "/tmp/rec_silent.wav"
+REC_T = "/tmp/rec_tone.wav"
+
+
+def find_digital_sink() -> str:
+    raw = subprocess.run(["pw-dump"], capture_output=True, text=True).stdout
+    for node in json.loads(raw):
+        props = (node.get("info") or {}).get("props") or {}
+        name = str(props.get("node.name", ""))
+        if props.get("media.class") == "Audio/Sink" and "iec958" in name:
+            return name
+    return ""
+
+
+def write_wav(path: str, samples: np.ndarray) -> None:
+    data = (np.clip(samples, -1.0, 1.0) * 32767).astype("<i2")
+    with wave.open(path, "wb") as fh:
+        fh.setnchannels(2)
+        fh.setsampwidth(2)
+        fh.setframerate(RATE)
+        fh.writeframes(np.repeat(data, 2).tobytes())
+
+
+def play_record(src: str, dst: str, secs: float) -> None:
+    player = subprocess.Popen(["pw-play", "--target", "collaplex_vsink", src],
+                              stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
+    time.sleep(0.35)
+    rec = subprocess.Popen(["timeout", str(secs), "pw-record", "--target", sink_name,
+                            "-P", "{ stream.capture.sink = true }",
+                            "--rate", str(RATE), "--channels", "2", "--format", "f32", dst],
+                           stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
+    rec.wait(timeout=secs + 8)
+    player.terminate()
+    player.wait(timeout=5)
+
+
+def analyse(path: str, label: str) -> None:
+    try:
+        _rate, data = wavfile.read(path)
+    except (OSError, ValueError):
+        print("  %s: 读不到录音" % label)
+        return
+    arr = data.astype(np.float64)
+    if arr.ndim > 1:
+        arr = arr[:, 0]
+    if arr.size < 1000:
+        print("  %s: 录音太短" % label)
+        return
+    rms = float(np.sqrt(np.mean(arr ** 2)))
+    peak = float(np.max(np.abs(arr)))
+    print("  %-22s RMS %8.2f dBFS | 峰值 %8.2f dBFS" % (
+        label, 20 * np.log10(max(rms, 1e-12)), 20 * np.log10(max(peak, 1e-12))))
+
+
+sink_name = ""
+
+
+def main() -> None:
+    global sink_name
+    sink_name = find_digital_sink()
+    if not sink_name:
+        print("找不到数字输出")
+        return
+
+    secs = 4.0
+    n = int(RATE * secs)
+    silent = np.zeros(n)
+    tone = 0.1 * np.sin(2 * np.pi * 1000 * np.arange(n) / RATE)   # -20 dBFS 纯音
+    write_wav(SILENT, silent)
+    write_wav(TONE, tone)
+
+    print("数字输出:", sink_name)
+    print()
+    play_record(SILENT, REC_S, 3.0)
+    analyse(REC_S, "播全零素材")
+    play_record(TONE, REC_T, 3.0)
+    analyse(REC_T, "播 -20dBFS 纯音")
+    print()
+    print("判据: 播全零时输出应远低于 -90 dBFS(链路不自造噪声);")
+    print("      纯音那一行若有宽频沙沙, RMS 会明显高于纯音本身应有的 -23 dBFS 太多。")
+
+
+if __name__ == "__main__":
+    main()

测试/test_rms.py

@@ -0,0 +1,73 @@
+"""端到端: 总音量放大对响度与峰值的影响(播素材实测, 数据取自 EQ DSP 的状态区)。"""
+from __future__ import annotations
+
+import json
+import math
+import subprocess
+import time
+import urllib.request
+import wave
+
+import numpy as np
+
+API = "http://127.0.0.1:8789"
+WAV = "/tmp/e2e_white.wav"
+RATE = 96000
+
+
+def make_white() -> None:
+    rng = np.random.default_rng(7)
+    secs = 4.0
+    x = rng.normal(0.0, 1.0, int(RATE * secs))
+    x = x / (np.max(np.abs(x)) + 1e-12) * (10 ** (-12 / 20))   # 峰值 -12 dBFS
+    data = (x * 32767).astype("<i2")
+    with wave.open(WAV, "wb") as fh:
+        fh.setnchannels(2)
+        fh.setsampwidth(2)
+        fh.setframerate(RATE)
+        fh.writeframes(np.repeat(data, 2).tobytes())
+
+
+def post(path: str, payload: dict[str, object]) -> None:
+    req = urllib.request.Request(API + path, data=json.dumps(payload).encode("utf-8"),
+                                 headers={"Content-Type": "application/json"})
+    urllib.request.urlopen(req, timeout=10).read()
+
+
+def state() -> dict[str, object]:
+    raw = urllib.request.urlopen(API + "/api/state", timeout=10).read()
+    return json.loads(raw)
+
+
+def play_and_read() -> dict[str, float]:
+    proc = subprocess.Popen(["pw-play", "--target", "collaplex_vsink", WAV],
+                            stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
+    time.sleep(1.6)
+    levels = state().get("levels")
+    time.sleep(0.2)
+    proc.terminate()
+    proc.wait(timeout=5)
+    if not isinstance(levels, dict):
+        return {"in_rms": -120.0, "out_rms": -120.0, "out_peak": -120.0}
+    return {str(k): float(v) for k, v in levels.items()}
+
+
+def main() -> None:
+    make_white()
+    base: list[float] = []
+    print("素材: 峰值 -12 dBFS 白噪")
+    for db in (0.0, 6.0, 12.0):
+        post("/api/volume", {"db": db})
+        time.sleep(0.5)
+        lv = play_and_read()
+        out_rms = lv.get("out_rms", -120.0)
+        out_peak = lv.get("out_peak", -120.0)
+        base.append(out_rms)
+        delta = out_rms - base[0]
+        print("总音量 %+5.1f dB -> 出 RMS %6.2f dBFS (Δ%+5.2f) | 出峰值 %6.2f dBFS" % (
+            db, out_rms, delta, out_peak))
+    post("/api/volume", {"db": 6.0})
+
+
+if __name__ == "__main__":
+    main()

测试/test_stage_spec.py

@@ -0,0 +1,85 @@
+"""中低频沙沙定位 ③: 逐级旁路, 每级单独喂同一素材, 比频谱。"""
+from __future__ import annotations
+
+import json
+import math
+import subprocess
+import sys
+import time
+
+import numpy as np
+from scipy.io import wavfile
+
+RATE = 96000
+SRC = "/tmp/thd_tone.wav"
+SINK = ""
+
+PROBES = (100.0, 200.0, 300.0, 400.0, 600.0, 800.0)
+
+
+def find_digital_sink() -> str:
+    raw = subprocess.run(["pw-dump"], capture_output=True, text=True).stdout
+    for node in json.loads(raw):
+        props = (node.get("info") or {}).get("props") or {}
+        name = str(props.get("node.name", ""))
+        if props.get("media.class") == "Audio/Sink" and "iec958" in name:
+            return name
+    return ""
+
+
+def capture(target: str, path: str) -> None:
+    player = subprocess.Popen(["pw-play", "--target", target, SRC],
+                              stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
+    time.sleep(0.6)
+    rec = subprocess.Popen(["timeout", "3", "pw-record", "--target", SINK,
+                            "-P", "{ stream.capture.sink = true }",
+                            "--rate", str(RATE), "--channels", "2", "--format", "f32", path],
+                           stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
+    rec.wait(timeout=12)
+    player.terminate()
+    player.wait(timeout=5)
+
+
+def spec_of(path: str) -> tuple[np.ndarray, np.ndarray]:
+    _rate, data = wavfile.read(path)
+    x = data.astype(np.float64)
+    if x.ndim > 1:
+        x = x[:, 0]
+    x = x[8000:8000 + 65536]
+    x = x - float(np.mean(x))
+    spec = np.abs(np.fft.rfft(x * np.hanning(x.size))) ** 2
+    return np.fft.rfftfreq(x.size, 1.0 / RATE), spec
+
+
+def rel_at(freqs: np.ndarray, spec: np.ndarray, f: float, half: float = 15.0) -> float:
+    mask = (freqs >= f - half) & (freqs < f + half)
+    ref = float(np.max(spec)) + 1e-30
+    if not np.any(mask):
+        return -999.0
+    return 10.0 * math.log10((float(np.max(spec[mask])) + 1e-30) / ref)
+
+
+def main() -> None:
+    global SINK
+    SINK = find_digital_sink()
+    stages = (
+        ("旁路(直插设备)", SINK),
+        ("只走 HRTF 级", "collaplex_hrtf_in"),
+        ("走 HRTF+EQ 级", "collaplex_eq_in"),
+        ("全链(含归一化)", "collaplex_vsink"),
+    )
+    print("素材 200 Hz 纯音; 每格 = 该频点相对基频的 dB(越负越干净)")
+    print("  %-18s %s" % ("测点", "  ".join("%7.0fHz" % f for f in PROBES)))
+    for idx, (label, target) in enumerate(stages):
+        path = "/tmp/stage_%d.wav" % idx
+        capture(target, path)
+        try:
+            freqs, spec = spec_of(path)
+        except (OSError, ValueError, IndexError):
+            print("  %-18s 读取失败" % label)
+            continue
+        print("  %-18s %s" % (label, "  ".join("%+8.1f" % rel_at(freqs, spec, f) for f in PROBES)))
+
+
+if __name__ == "__main__":
+    main()

测试/test_thd.py

@@ -0,0 +1,124 @@
+"""中低频沙沙定位: 播 200Hz 纯音, 改总音量, 录输出看谐波/非谐波成分。"""
+from __future__ import annotations
+
+import json
+import os
+import subprocess
+import sys
+import time
+import urllib.request
+import wave
+
+import numpy as np
+from scipy.io import wavfile
+
+PROJ = "/home/lou/桌面/工作区/实验/collaplex音效"
+RATE = 96000
+FREQ = 200.0
+SRC = "/tmp/thd_tone.wav"
+SINK = ""
+API = "http://127.0.0.1:8789"
+
+
+def find_digital_sink() -> str:
+    raw = subprocess.run(["pw-dump"], capture_output=True, text=True).stdout
+    for node in json.loads(raw):
+        props = (node.get("info") or {}).get("props") or {}
+        name = str(props.get("node.name", ""))
+        if props.get("media.class") == "Audio/Sink" and "iec958" in name:
+            return name
+    return ""
+
+
+def write_tone() -> None:
+    secs = 5.0
+    n = int(RATE * secs)
+    t = np.arange(n) / RATE
+    x = (0.1 * np.sin(2 * np.pi * FREQ * t))          # 峰值 -20 dBFS
+    ramp = np.minimum(1.0, np.minimum(np.arange(n) / 2000.0, (n - np.arange(n)) / 2000.0))
+    x = x * ramp
+    data = (np.clip(x, -1, 1) * 32767).astype("<i2")
+    with wave.open(SRC, "wb") as fh:
+        fh.setnchannels(2)
+        fh.setsampwidth(2)
+        fh.setframerate(RATE)
+        fh.writeframes(np.repeat(data, 2).tobytes())
+
+
+def set_volume(db: float) -> None:
+    req = urllib.request.Request(API + "/api/volume", data=json.dumps({"db": db}).encode(),
+                                 headers={"Content-Type": "application/json"})
+    urllib.request.urlopen(req, timeout=10).read()
+
+
+def capture(path: str) -> None:
+    player = subprocess.Popen(["pw-play", "--target", "collaplex_vsink", SRC],
+                              stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
+    time.sleep(0.6)
+    rec = subprocess.Popen(["timeout", "3", "pw-record", "--target", SINK,
+                            "-P", "{ stream.capture.sink = true }",
+                            "--rate", str(RATE), "--channels", "2", "--format", "f32", path],
+                           stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
+    rec.wait(timeout=12)
+    player.terminate()
+    player.wait(timeout=5)
+
+
+def analyse(path: str, label: str) -> None:
+    try:
+        _rate, data = wavfile.read(path)
+    except (OSError, ValueError):
+        print("  %-12s 读不到录音" % label)
+        return
+    x = data.astype(np.float64)
+    if x.ndim > 1:
+        x = x[:, 0]
+    if x.size < 8192:
+        print("  %-12s 录音太短" % label)
+        return
+    x = x[4096:4096 + 65536]
+    x = x - float(np.mean(x))
+    win = np.hanning(x.size)
+    spec = np.abs(np.fft.rfft(x * win)) ** 2
+    freqs = np.fft.rfftfreq(x.size, 1.0 / RATE)
+
+    def band_energy(f0: float, f1: float) -> float:
+        mask = (freqs >= f0) & (freqs < f1)
+        return float(np.sum(spec[mask]))
+
+    def tone_energy(f: float, half: float = 12.0) -> float:
+        return band_energy(f - half, f + half)
+
+    fund = tone_energy(FREQ)
+    harmonics = sum(tone_energy(FREQ * k) for k in range(2, 9))
+    total = float(np.sum(spec)) + 1e-30
+    rest = max(total - fund - harmonics, 0.0)
+    to_db = lambda v: 10.0 * np.log10(v + 1e-30)   # noqa: E731
+
+    print("  %-12s 基频 %7.1f dB | 谐波(2~8次) %7.1f dB 相对基频 %+6.1f dB | "
+          "其余(含噪声) %7.1f dB 相对基频 %+6.1f dB" % (
+              label, to_db(fund), to_db(harmonics), to_db(harmonics) - to_db(fund),
+              to_db(rest), to_db(rest) - to_db(fund)))
+
+
+def main() -> None:
+    global SINK
+    SINK = find_digital_sink()
+    if not SINK:
+        print("找不到数字输出")
+        return
+    write_tone()
+    print("素材: 200 Hz 纯音(峰值 -20 dBFS), 录音后分析 200k 分量之外的成分")
+    print("判据: 谐波/其余成分相对基频越低越干净; 若 +6~+12 dB 时显著抬起, 就是限幅失真")
+    print()
+    for db in (0.0, 6.0, 12.0):
+        set_volume(db)
+        time.sleep(0.5)
+        path = "/tmp/thd_%d.wav" % int(db)
+        capture(path)
+        analyse(path, "总音量 %+d dB" % int(db))
+    set_volume(6.0)
+
+
+if __name__ == "__main__":
+    main()