固化链路: 虚拟声卡 + 响度归一化 -> 数字输出 (安装脚本可复现部署)

README.md

@@ -1,3 +1,58 @@
 # Collaplex 音效
 
-耳机虚拟环绕（HRTF）音效链路。
+耳机方向的空间音效链路（PipeWire）。
+
+当前链路（2026-09-19 固化）：
+
+```
+全局音频(应用输出) → collaplex_vsink(虚拟声卡) → 响度归一化(pipe DSP) → 数字输出(S/PDIF)
+```
+
+## 安装
+
+把项目里的定义部署到 PipeWire 用户级：
+
+```sh
+bash 脚本/安装.sh
+```
+
+脚本会：探测数字输出设备 → 渲染并安装链路配置到 `~/.config/pipewire/pipewire.conf.d/` →
+装 pipe 插件入口到 `~/.local/bin/` → 重启 PipeWire → 自检（节点 / DSP 进程 / 默认输出）。
+
+**设备名变化或项目挪了位置**，重跑这个脚本即可 —— 定义在项目里，部署可复现。
+
+## 开关与调参
+
+- 实时状态（推子 / 响度 / 输出电平）：`python3 dsp/状态.py`
+- 目标电平（DSP 每秒重读，改完即时生效，不用重建链路）：
+
+```sh
+echo '{"on": true, "target_db": -16}' > ~/.local/state/cinema-spatial/loudness.json
+```
+
+## 文件
+
+| 路径 | 作用 |
+|---|---|
+| `dsp/loudness_norm.py` | 响度归一化 DSP（pipe 插件载荷；左右各一进程，flock 共享能量算同一增益，声像不漂） |
+| `dsp/状态.py` | 读共享内存看实时推子 / 电平 |
+| `bin/collaplex-loudness-norm` | pipe 插件入口模板（纯 ASCII 路径；参数写在这里，配置里只给路径） |
+| `pipewire/10-collaplex-vsink.conf` | 链路定义模板（`__TARGET__` 数字输出、`__WRAPPER__` 入口） |
+| `脚本/安装.sh` | 渲染 + 部署 + 重启 + 自检 |
+
+## 实测（2026-09-19 固化后复测）
+
+| 输入素材 | DSP 推子 | 数字输出端实测 |
+|---|---|---|
+| −12 dBFS | −8.37 dB | −19.31 dBFS |
+| −32 dBFS | −3.58 dB | −18.97 dBFS |
+
+输入差 20.00 dB → 输出端实测差 **0.34 dB**；推子方向相反（响的压、轻的抬），
+目标 −16 dBFS，限速 6 dB/s。
+
+## 注意
+
+- pipe 插件的 `command` 只吃「程序 + 一个参数」，且**不能用中文路径**（否则节点照建、
+  音频绕过、零日志）。所以配置里给的是 `bin/` 里的 ASCII wrapper，由它转发到中文项目路径。
+- `node.target` 写的是设备名，设备重连/换口后可能变；重跑 `脚本/安装.sh` 修正。
+- 播放中的实测要抓 sink 的 monitor：`pw-record --target <sink 节点 id> -P '{ stream.capture.sink = true }'`。

bin/collaplex-loudness-norm

@@ -0,0 +1,15 @@
+#!/bin/sh
+# Collaplex 响度归一化 — pipe 插件入口 (本文件由 脚本/安装.sh 生成, 占位符 __DSP__)
+#
+# ★ pipe 插件的 command 只支持「程序 + 一个参数」(shebang 式 exec):
+#    `env A=B python3 x.py` 会报 "use -[v]S to pass options in shebang lines",
+#    所以参数全部写在这里, 配置里只给本脚本的路径。
+# ★ 本脚本路径必须**纯 ASCII**: pipe 插件给中文路径会一声不吭(节点照建、
+#    音频绕过、零日志)。脚本内部再去 exec 中文路径的 DSP 是没问题的。
+export LN_TARGET_DB=-16
+export LN_GATE_DB=-50
+export LN_MAX_BOOST=20
+export LN_MAX_CUT=20
+export LN_SLEW_DB_S=6
+export LN_RATE=48000
+exec /usr/bin/python3 "__DSP__"

dsp/loudness_norm.py

@@ -0,0 +1,324 @@
+#!/usr/bin/env python3
+"""Collaplex 响度归一化器 — PipeWire filter-chain `pipe` 插件载荷
+
+★ 运行环境: 由 filter-chain 的 pipe 插件 fork, stdin=音频进, stdout=音频出。
+★ 关键机制: filter-chain 会**按声道复制 graph**, 所以左右声道各起一个本进程实例,
+   每个实例只看到**单声道**数据。为保持立体声声像不漂移, 两个实例通过 /dev/shm
+   共享各自的能量值, 用"两声道能量和"作为联合响度, 算出**同一个增益**。
+
+★ 铁律: 本进程**绝不能退出** —— pipe 插件的外部程序一旦退出, 音频立刻变静音。
+   所有异常都吃掉, 出错时退回纯透传。
+
+算法(简化版 EBU R128 思路, 纯 numpy 向量化, 零逐样本循环):
+  块 RMS -> 能量 -> 指数滑动平均(≈400ms 短时响度) -> 门控 ->
+  目标电平求增益(dB) -> 限速平滑 -> 峰值保护 -> 施加
+
+环境变量调参(改配置时注入):
+  LN_TARGET_DB   目标响度(dBFS), 默认 -16
+  LN_GATE_DB     静音门限(dBFS), 默认 -50
+  LN_MAX_BOOST   最大提升(dB), 默认 +12
+  LN_MAX_CUT     最大衰减(dB), 默认 -18
+  LN_SLEW_DB_S   增益变化速率上限(dB/s), 默认 6
+  LN_WINDOW_S    响度平滑窗口(s), 默认 0.4
+  LN_RATE        采样率, 默认 96000
+  LN_BLOCK       每次读取字节数, 默认 4096
+"""
+import ctypes
+import fcntl
+import json
+import math
+import mmap
+import os
+import select
+import signal
+import struct
+import sys
+import time
+
+import numpy as np
+
+# ---------------- 参数 ----------------
+TARGET_DB = float(os.environ.get("LN_TARGET_DB", "-16"))   # 兜底默认(状态文件优先)
+# ★ 目标电平放状态文件: 面板/CLI 改它, DSP 每秒重读一次 -> 实时生效, 不用重建链路。
+LOUDNESS_STATE = os.path.expanduser("~/.local/state/cinema-spatial/loudness.json")
+GATE_DB = float(os.environ.get("LN_GATE_DB", "-50"))
+MAX_BOOST = abs(float(os.environ.get("LN_MAX_BOOST", "12")))
+MAX_CUT = -abs(float(os.environ.get("LN_MAX_CUT", "18")))   # ★ 必须为负, 传正数也自动取负
+SLEW_DB_S = float(os.environ.get("LN_SLEW_DB_S", "6"))
+WINDOW_S = float(os.environ.get("LN_WINDOW_S", "0.4"))
+RATE = float(os.environ.get("LN_RATE", "96000"))
+BLOCK = int(os.environ.get("LN_BLOCK", "4096"))
+PEAK_CEIL = float(os.environ.get("LN_PEAK_CEIL", "0.985"))
+FADE_BLOCKS = int(os.environ.get("LN_FADE_BLOCKS", "4"))       # 起播淡入块数(~170ms @96k/4096)
+FADE_GAP_BLOCKS = int(os.environ.get("LN_FADE_GAP_BLOCKS", "24"))  # 连续静音多少块算"重新起播"
+IDLE_LIMIT = int(os.environ.get("LN_IDLE_LIMIT", "10"))   # 连续无数据多少秒后自行退出
+DEBUG = os.environ.get("LN_DEBUG", "") not in ("", "0")
+
+_ERR = sys.stderr
+
+
+def dbg(msg: str) -> None:
+    if DEBUG:
+        try:
+            _ERR.write(msg + "\n")
+            _ERR.flush()
+        except Exception:
+            pass
+
+EPS = 1e-12
+GATE_E = 10.0 ** (GATE_DB / 10.0)
+
+
+def read_target() -> float:
+    """当前目标电平(dBFS) —— 状态文件优先, 读不到回落到环境变量/默认.
+
+    ★ 面板"响度归一化"推子写的就是这个值; 主循环每秒重读一次, 所以改完立即生效。
+    """
+    try:
+        with open(LOUDNESS_STATE, encoding="utf-8") as fh:
+            d = json.load(fh)
+        if isinstance(d, dict) and d.get("target_db") is not None:
+            return max(-40.0, min(-6.0, float(d["target_db"])))
+    except (OSError, ValueError, TypeError):
+        pass
+    return TARGET_DB
+
+# ---------------- 共享区(两声道交换状态 + 面板读取) ----------------
+# ★ web 面板也读这块内存, 所以**字段只增不改顺序**。
+# 每声道 32 字节, 两声道; 其后是全局参数。
+#   +0   f32  e_smooth  平滑能量(响度判据, ~400ms) —— 归一化的"输入电平"
+#   +4   f32  e_block   本块能量(电平表用, 快响应)
+#   +8   f32  peak      本块峰值(电平表用)
+#   +12  f32  gain_db   本块**实际**施加的增益 dB(含峰值保护) -> 面板推子
+#   +16  f64  t         时间戳(monotonic)
+#   +24  u32  n         已处理块数(判活)
+#   +28  f32  peak_out  ★本块**输出**峰值(2026-09-13 加; 面板"处理后电平"用它,
+#                       旧版这里是 _pad=0, 面板会回退到估算值并截顶到 PEAK_CEIL)
+#   +64  f32  target_db 目标响度(面板推子的基准线)
+#   +68  u32  rate      采样率
+SHM_PATH = "/dev/shm/collaplex-loudness"
+CH_STRIDE = 32
+SHM_SIZE = 80
+TARGET_OFF = 64
+RATE_OFF = 68
+
+
+def set_pdeathsig() -> None:
+    """父进程(pipewire filter-chain)一旦消失, 本进程立即被内核杀掉。
+    ★ 否则本进程会变成孤儿: 继续持有槽位锁, 让新实例拿不到槽位。"""
+    try:
+        libc = ctypes.CDLL("libc.so.6", use_errno=True)
+        libc.prctl(1, signal.SIGKILL)      # PR_SET_PDEATHSIG=1
+    except Exception:
+        pass
+
+
+def claim_slot() -> int:
+    """用 flock 抢一个声道槽位(0/1), 锁随进程退出自动释放。"""
+    for i in (0, 1):
+        try:
+            fd = os.open("%s.lock%d" % (SHM_PATH, i), os.O_CREAT | os.O_RDWR, 0o600)
+            fcntl.flock(fd, fcntl.LOCK_EX | fcntl.LOCK_NB)
+            return i  # fd 不关, 锁一直持有
+        except OSError:
+            try:
+                os.close(fd)
+            except Exception:
+                pass
+    # ★★★ 2026-09-13: 两个槽**都**被占 = 已经有一对 DSP 在跑, 本进程必须退出。
+    #   旧版这里 `return 0` —— 第 3、4 个实例会**一起写槽0**(没有锁保护), 数据被劈开,
+    #   整链静音(老板报"没声音了", 实测 4 个进程抢 2 个槽, 槽1 僵死在 152 秒前)。
+    #   注意 DSP 的父进程是 **pipewire 主进程**而不是 filter-chain 实例, 所以重建链路
+    #   时 pdeathsig 不触发 → 每重建一次就攒一对孤儿, 必须靠这里自我了断。
+    sys.stderr.write("[loudness] 两个槽位都被占用(已有实例在跑) -> 本进程退出\n")
+    sys.stderr.flush()
+    raise SystemExit(0)
+
+
+def open_shm() -> mmap.mmap:
+    fd = os.open(SHM_PATH, os.O_CREAT | os.O_RDWR, 0o600)
+    if os.fstat(fd).st_size < SHM_SIZE:
+        os.ftruncate(fd, SHM_SIZE)
+    return mmap.mmap(fd, SHM_SIZE)
+
+
+STALE_S = 1.0   # 对方声道超过这么久没更新, 视为不存在(防残留值污染联合响度)
+
+
+def read_peer_energy(mm: mmap.mmap, other: int) -> float:
+    """读另一声道的平滑能量; 陈旧(NaN/负/超时)一律当 0。"""
+    base = other * CH_STRIDE
+    try:
+        e = struct.unpack_from("<f", mm, base)[0]
+        t = struct.unpack_from("<d", mm, base + 16)[0]
+    except Exception:
+        return 0.0
+    if e != e or e < 0.0:                    # NaN
+        return 0.0
+    if abs(time.monotonic() - t) > STALE_S:  # 陈旧
+        return 0.0
+    return e
+
+
+def main() -> None:
+    stdin = sys.stdin.buffer
+    stdout = sys.stdout.buffer
+
+    set_pdeathsig()
+    slot = claim_slot()
+    mm = open_shm()
+    other = 1 - slot
+
+    # 全局参数写进共享区(面板推子的基准线)
+    target_db = read_target()
+    try:
+        struct.pack_into("<f", mm, TARGET_OFF, target_db)
+        struct.pack_into("<I", mm, RATE_OFF, int(RATE))
+    except Exception:
+        pass
+    next_target_check = time.monotonic() + 1.0
+
+    # 平滑状态
+    energy_smooth = 0.0
+    gain_db = 0.0
+    _n: int = 0
+
+    block_seconds = BLOCK / 4.0 / RATE          # 每块时长(单声道 float32)
+    alpha = 1.0 - math.exp(-block_seconds / WINDOW_S) if WINDOW_S > 0 else 1.0
+    max_step_db = SLEW_DB_S * block_seconds
+
+    carry = b""                                    # 不足 4 字节的尾巴
+    idle = 0                                       # 连续无数据的秒数
+    # ★ 起播淡入(2026-09-13): 从"长时间静音"回到有声的头几块, 上混的 6 路 HRTF 卷积
+    #   会产生大幅过冲 —— 实测瞬间冲到 +7.8 dBFS(超满刻度 8dB, 削波), 之后指数衰减,
+    #   面板的峰值栏于是"顶满"。淡入让信号平滑爬升即可压掉。
+    #   只在**长静音**后触发(连续 FADE_GAP_BLOCKS 块无信号), 免得音乐里的安静段
+    #   被反复淡入听出"呼吸"。
+    fade_left = 0
+    silent_blocks = 0
+
+    while True:
+        # ★ 用 select 带超时读。为什么必须这样:
+        #   1) filter-chain 死后管道写端不一定关闭, 所以 read() 不一定返回 EOF;
+        #   2) 孤儿进程会被 systemd(subreaper) 收养, ppid 变成 session manager 而不是 1,
+        #      所以 "ppid==1" 也不可靠。
+        #   唯一可靠的判据 = 连续若干秒收不到任何数据(链路已断)。到时自行退出,
+        #   否则每重启一次音频就会攒一对僵尸进程、霸占左右槽位锁。
+        try:
+            ready, _, _ = select.select([stdin], [], [], 1.0)
+        except Exception:
+            ready = [stdin]
+        if not ready:
+            idle += 1
+            if idle >= IDLE_LIMIT or os.getppid() == 1:
+                break
+            continue
+        idle = 0
+
+        # 目标电平热更新: 面板/CLI 改了 loudness.json, 这里每秒跟上(不用重建链路)
+        _now_t = time.monotonic()
+        if _now_t >= next_target_check:
+            next_target_check = _now_t + 1.0
+            _t = read_target()
+            if _t != target_db:
+                target_db = _t
+                try:
+                    struct.pack_into("<f", mm, TARGET_OFF, target_db)
+                except Exception:
+                    pass
+
+        try:
+            raw = stdin.read(BLOCK)
+        except Exception:
+            break
+        if not raw:
+            break
+
+        buf = carry + raw
+        n = len(buf) - (len(buf) % 4)
+        carry = buf[n:]
+        if n == 0:
+            continue
+
+        try:
+            x = np.frombuffer(buf[:n], dtype="<f4")
+            if x.size == 0:
+                continue
+
+            # ---- 1) 块能量 + 滑窗平滑 ----
+            energy = float(np.mean(x.astype(np.float64) ** 2))
+            energy_smooth += alpha * (energy - energy_smooth)
+            if energy_smooth < 0.0:
+                energy_smooth = 0.0
+
+            # ---- 2) 跨声道联合响度(两声道能量和) ----
+            now = time.monotonic()
+            base = slot * CH_STRIDE
+            struct.pack_into("<f", mm, base, energy_smooth)
+            struct.pack_into("<d", mm, base + 16, now)
+            joint = energy_smooth + read_peer_energy(mm, other)
+
+            # ---- 3) 目标增益(dB) ----
+            if joint > GATE_E:
+                if silent_blocks > FADE_GAP_BLOCKS:     # 长静音后重新起播 → 淡入
+                    fade_left = FADE_BLOCKS
+                silent_blocks = 0
+                level_db = 10.0 * math.log10(joint + EPS)
+                want_db = target_db - level_db
+                want_db = max(MAX_CUT, min(MAX_BOOST, want_db))
+            else:
+                silent_blocks += 1
+                want_db = gain_db                       # 静音段保持当前增益
+
+            # ---- 4) 限速平滑(避免抽气/爆音) ----
+            delta = want_db - gain_db
+            if delta > max_step_db:
+                delta = max_step_db
+            elif delta < -max_step_db:
+                delta = -max_step_db
+            gain_db += delta
+
+            # ---- 5) 峰值保护(快降, 只作用于本块) ----
+            g = 10.0 ** (gain_db / 20.0)
+            peak = float(np.max(np.abs(x))) if x.size else 0.0
+            if peak * g > PEAK_CEIL > 0.0:
+                g = PEAK_CEIL / (peak + EPS)
+                gain_db = 20.0 * math.log10(g + EPS)     # 状态跟随, 下块再按限速恢复
+
+            # ---- 6) 施加增益 ----
+            y = x * g
+            if fade_left > 0:                           # 起播淡入: 线性爬升, 压掉卷积过冲
+                y = y * ((FADE_BLOCKS - fade_left + 1) / float(FADE_BLOCKS))
+                fade_left -= 1
+            out_peak = float(np.max(np.abs(y))) if y.size else 0.0   # ★ 输出峰值(面板用, 真实值)
+            if y.dtype != np.float32:
+                y = y.astype("<f4")
+
+            # ---- 7) 写回 + 状态(web 面板读这块内存画 4 路电平/推子) ----
+            _n += 1
+            struct.pack_into("<f", mm, base + 4, energy)          # 本块能量(快响应)
+            struct.pack_into("<f", mm, base + 8, peak)            # 本块峰值(输入)
+            struct.pack_into("<f", mm, base + 12, gain_db)        # 实际增益 -> 推子
+            struct.pack_into("<f", mm, base + 28, out_peak)       # ★ 本块输出峰值 -> 面板"处理后电平"
+            struct.pack_into("<I", mm, base + 24, _n & 0xFFFFFFFF)
+            if DEBUG and _n % 40 == 0:
+                dbg("slot=%d blk=%d raw=%.6f smooth=%.8f joint=%.8f lvl=%.2f want=%.2f g=%.2f peak=%.4f" % (
+                    slot, _n, energy, energy_smooth, joint,
+                    10.0 * math.log10(joint + EPS) if joint > 0 else -999.0,
+                    want_db, gain_db, peak))
+            stdout.write(y.tobytes())
+            stdout.flush()
+        except Exception:
+            # ★ 任何异常都不能让进程退出(退出=静音), 退回纯透传
+            try:
+                stdout.write(buf[:n] if n else raw)
+                stdout.flush()
+            except Exception:
+                pass
+
+
+if __name__ == "__main__":
+    try:
+        main()
+    except Exception:
+        pass

dsp/状态.py

@@ -0,0 +1,128 @@
+#!/usr/bin/env python3
+"""读 Collaplex 响度归一化的共享状态(诊断 / web 面板数据源)
+
+共享区布局(见 loudness_norm.py 顶部注释):
+  每声道 32 字节, 两声道:
+    +0   f32  e_smooth  平滑能量(响度判据) —— 归一化的"输入电平"
+    +4   f32  e_block   本块能量(快响应)
+    +8   f32  peak      本块峰值
+    +12  f32  gain_db   本块实际施加的增益 dB -> 推子
+    +16  f64  t         时间戳(monotonic)
+    +24  u32  n         已处理块数(判活)
+  +64  f32  target_db 目标响度
+  +68  u32  rate      采样率
+
+用法:
+  python3 read_state.py            打印一次
+  python3 read_state.py --watch    每 0.5 秒刷新(诊断动态平衡过程)
+"""
+import json
+import math
+import mmap
+import os
+import struct
+import sys
+import time
+
+SHM_PATH = "/dev/shm/collaplex-loudness"
+CH_STRIDE = 32
+SHM_SIZE = 80
+TARGET_OFF = 64
+RATE_OFF = 68
+STALE_S = 1.0
+
+EPS = 1e-12
+
+
+def db(x: float) -> float:
+    """线性能量/幅度 -> dBFS(0 以下)。"""
+    return 10.0 * math.log10(x + EPS)
+
+
+def read_slot(mm: mmap.mmap, i: int) -> dict[str, float]:
+    base = i * CH_STRIDE
+    e_smooth = struct.unpack_from("<f", mm, base)[0]
+    e_block = struct.unpack_from("<f", mm, base + 4)[0]
+    peak = struct.unpack_from("<f", mm, base + 8)[0]
+    gain = struct.unpack_from("<f", mm, base + 12)[0]
+    t = struct.unpack_from("<d", mm, base + 16)[0]
+    n = struct.unpack_from("<I", mm, base + 24)[0]
+    age = time.monotonic() - t
+    return {
+        "e_smooth": e_smooth,
+        "e_block": e_block,
+        "peak": peak,
+        "gain_db": gain,
+        "n": n,
+        "age": age,
+        "alive": age < STALE_S,
+    }
+
+
+def snapshot() -> dict[str, object]:
+    if not os.path.exists(SHM_PATH):
+        return {"ok": False, "err": "共享区不存在(响度归一化未启用)"}
+    fd = os.open(SHM_PATH, os.O_RDONLY)
+    try:
+        mm = mmap.mmap(fd, SHM_SIZE, prot=mmap.PROT_READ)
+        try:
+            target = struct.unpack_from("<f", mm, TARGET_OFF)[0]
+            rate = struct.unpack_from("<I", mm, RATE_OFF)[0]
+            ch = [read_slot(mm, 0), read_slot(mm, 1)]
+        finally:
+            mm.close()
+    finally:
+        os.close(fd)
+
+    out: dict[str, object] = {
+        "ok": bool(ch[0]["alive"] or ch[1]["alive"]),
+        "target_db": round(target, 2),
+        "rate": rate,
+    }
+    for i, name in ((0, "L"), (1, "R")):
+        c = ch[i]
+        # 原始输入电平 = 快响应块能量; 处理后 = 原始 + 实际增益
+        raw_db = db(c["e_block"])
+        raw_pk = 20.0 * math.log10(c["peak"] + EPS) if c["peak"] > 0 else -120.0
+        out["raw_" + name] = round(raw_db, 2) if c["alive"] else None
+        out["raw_peak_" + name] = round(raw_pk, 2) if c["alive"] else None
+        out["post_" + name] = round(raw_db + c["gain_db"], 2) if c["alive"] else None
+        out["gain_db_" + name] = round(c["gain_db"], 2) if c["alive"] else None
+        # 平滑响度(归一化真正依据的那条)
+        out["loud_" + name] = round(db(c["e_smooth"]), 2) if c["alive"] else None
+    if ch[0]["alive"] and ch[1]["alive"]:
+        joint = ch[0]["e_smooth"] + ch[1]["e_smooth"]
+        out["loud_joint"] = round(db(joint), 2)
+        out["deviation_db"] = round(target - db(joint), 2)   # >0 = 还要提, <0 = 还要压
+    return out
+
+
+def main() -> None:
+    watch = "--watch" in sys.argv
+    try:
+        if not watch:
+            print(json.dumps(snapshot(), ensure_ascii=False, indent=2))
+            return
+        print("每 0.5 秒刷新 (Ctrl-C 退出)")
+        print("%8s %9s %9s %9s %9s %8s" % (
+            "时间", "原始L", "原始R", "处理后L", "处理后R", "增益dB"))
+        t0 = time.time()
+        while time.time() - t0 < 3600:
+            s = snapshot()
+            if not s.get("ok"):
+                print("  (无数据: %s)" % s.get("err", "声道未在跑"))
+            else:
+                fmt = lambda k: ("%9.2f" % s[k]) if s.get(k) is not None else "%9s" % "-"
+                _g = s.get("gain_db_L")
+                gv = float(_g) if isinstance(_g, (int, float)) else 0.0
+                print("%8.1f %s %s %s %s %8.2f" % (
+                    time.time() - t0, fmt("raw_L"), fmt("raw_R"),
+                    fmt("post_L"), fmt("post_R"), gv))
+            sys.stdout.flush()
+            time.sleep(0.5)
+    except KeyboardInterrupt:
+        pass
+
+
+if __name__ == "__main__":
+    main()

pipewire/10-collaplex-vsink.conf

@@ -0,0 +1,41 @@
+# Collaplex 虚拟声卡 + 响度归一化  —— 本文件由 脚本/安装.sh 生成(占位符: __TARGET__ / __WRAPPER__)
+# 链路: 全局音频(应用输出) -> collaplex_vsink(虚拟声卡) -> 响度归一化(pipe DSP) -> 数字输出(S/PDIF)
+context.modules = [
+    {   name = libpipewire-module-filter-chain
+        flags = [ nofail ]
+        args = {
+            node.description = "Collaplex 虚拟声卡 (响度归一化)"
+            media.name       = "Collaplex Virtual Sink"
+            filter.graph = {
+                nodes = [
+                    { type = builtin label = copy name = c1L }
+                    { type = builtin label = copy name = c1R }
+                    { type = builtin label = pipe name = lnL
+                      config = { command = "__WRAPPER__" } }
+                    { type = builtin label = pipe name = lnR
+                      config = { command = "__WRAPPER__" } }
+                    { type = builtin label = copy name = c2L }
+                    { type = builtin label = copy name = c2R }
+                ]
+                links = [
+                    { output = "c1L:Out" input = "lnL:In" }
+                    { output = "c1R:Out" input = "lnR:In" }
+                    { output = "lnL:Out" input = "c2L:In" }
+                    { output = "lnR:Out" input = "c2R:In" }
+                ]
+                inputs  = [ "c1L:In" "c1R:In" ]
+                outputs = [ "c2L:Out" "c2R:Out" ]
+            }
+            capture.props = {
+                media.class    = Audio/Sink
+                node.name      = collaplex_vsink
+                audio.position = [ FL FR ]
+            }
+            playback.props = {
+                node.name      = collaplex_vsink_out
+                audio.position = [ FL FR ]
+                node.target    = "__TARGET__"
+            }
+        }
+    }
+]

脚本/安装.sh

@@ -0,0 +1,93 @@
+#!/bin/bash
+# Collaplex 音效 — 把当前链路固化装到 PipeWire(用户级)
+#
+#   链路: 全局音频(应用输出) -> collaplex_vsink(虚拟声卡) -> 响度归一化(pipe DSP) -> 数字输出
+#
+# 装什么:
+#   $HOME/.config/pipewire/pipewire.conf.d/10-collaplex-vsink.conf   链路定义(由本脚本渲染)
+#   $HOME/.local/bin/collaplex-loudness-norm                         pipe 插件入口(ASCII 路径)
+# 设备名换了 / 项目挪了位置 -> 重跑本脚本即可(这就是"固化"的意义: 定义在项目里, 部署可复现)
+set -eu
+
+ROOT="$(cd "$(dirname "$0")/.." && pwd)"
+CONF_DST="$HOME/.config/pipewire/pipewire.conf.d/10-collaplex-vsink.conf"
+WRAP_DST="$HOME/.local/bin/collaplex-loudness-norm"
+PY=/usr/bin/python3
+
+echo "项目根: $ROOT"
+
+# ---- 1. 探测数字输出设备(跳过 HDMI/显卡) ----
+TARGET="$("$PY" - <<'EOF'
+import json, subprocess
+d = json.loads(subprocess.run(["pw-dump"], capture_output=True, text=True).stdout)
+sinks = []
+for n in d:
+    if n.get("type") != "PipeWire:Interface:Node":
+        continue
+    p = n.get("info", {}).get("props", {}) or {}
+    if p.get("media.class") != "Audio/Sink":
+        continue
+    nm = p.get("node.name") or ""
+    if "hdmi" in nm.lower() or "gb206" in nm.lower() or "radeon" in nm.lower():
+        continue          # 显卡 HDMI, 不是我们要的数字输出
+    sinks.append(nm)
+dig = [s for s in sinks if "iec958" in s or "spdif" in s.lower()]
+print(dig[0] if dig else "")
+EOF
+)"
+if [ -z "$TARGET" ]; then
+    echo "✗ 没找到数字输出(S/PDIF)设备。当前可用 sink:"
+    wpctl status | sed -n '/Sinks:/,/Sources:/p'
+    exit 1
+fi
+echo "数字输出: $TARGET"
+
+# ---- 2. 渲染并安装链路定义 ----
+mkdir -p "$(dirname "$CONF_DST")" "$(dirname "$WRAP_DST")"
+sed -e "s|__TARGET__|$TARGET|" -e "s|__WRAPPER__|$WRAP_DST|" \
+    "$ROOT/pipewire/10-collaplex-vsink.conf" > "$CONF_DST"
+echo "已装配置: $CONF_DST"
+
+# ---- 3. 渲染并安装 pipe 插件入口(指向项目内的 DSP) ----
+sed -e "s|__DSP__|$ROOT/dsp/loudness_norm.py|" \
+    "$ROOT/bin/collaplex-loudness-norm" > "$WRAP_DST"
+chmod 755 "$WRAP_DST"
+echo "已装入口: $WRAP_DST  ->  $ROOT/dsp/loudness_norm.py"
+
+# ---- 4. 让 PipeWire 重新加载 ----
+systemctl --user restart pipewire pipewire-pulse wireplumber
+sleep 6
+
+# ---- 5. 自检 ----
+echo
+echo "=== 自检 ==="
+N=$("$PY" - <<'EOF'
+import json, subprocess
+d = json.loads(subprocess.run(["pw-dump"], capture_output=True, text=True).stdout)
+n = [x for x in d if x.get("type") == "PipeWire:Interface:Node"
+     and (x.get("info", {}).get("props", {}) or {}).get("node.name") == "collaplex_vsink"]
+print(len(n))
+EOF
+)
+[ "$N" = "1" ] && echo "  ✓ 虚拟声卡节点 collaplex_vsink 在" || echo "  ✗ 虚拟声卡节点缺失(个数=$N)"
+
+DP=$(for p in /proc/[0-9]*; do c=$(tr '\0' ' ' < "$p/cmdline" 2>/dev/null) || continue; case "$c" in *dsp/loudness_norm.py*) echo "$p";; esac; done | wc -l)
+[ "$DP" = "2" ] && echo "  ✓ 响度归一化 DSP 进程 2 个(左右各一)" || echo "  ✗ DSP 进程数=$DP(应为 2)"
+
+ID=$("$PY" - <<'EOF'
+import json, subprocess
+d = json.loads(subprocess.run(["pw-dump"], capture_output=True, text=True).stdout)
+for x in d:
+    if x.get("type") == "PipeWire:Interface:Node":
+        if (x.get("info", {}).get("props", {}) or {}).get("node.name") == "collaplex_vsink":
+            print(x["id"])
+EOF
+)
+if [ -n "$ID" ]; then
+    wpctl set-default "$ID" >/dev/null 2>&1 || true
+    echo "  ✓ 默认输出已设为 collaplex_vsink(id $ID)"
+fi
+
+echo
+echo "开关/目标:  python3 \"$ROOT/dsp/状态.py\"          # 看实时推子"
+echo "改目标电平: echo '{\"on\": true, \"target_db\": -16}' > ~/.local/state/cinema-spatial/loudness.json"