From 6f5729baf2e73b82556d98eed353256288370f37 Mon Sep 17 00:00:00 2001 From: lou Date: Sat, 19 Sep 2026 19:37:54 +0800 Subject: [PATCH] =?UTF-8?q?=E5=89=8D=E7=AB=AF=E5=8A=A0=204=20=E4=B8=AA?= =?UTF-8?q?=E7=94=B5=E5=B9=B3=E8=A1=A8(=E5=85=A5L/=E5=87=BAL/=E5=85=A5R/?= =?UTF-8?q?=E5=87=BAR,=20=E5=8F=8C=E5=A3=B0=E9=81=93=E7=94=B1=20flock=20?= =?UTF-8?q?=E5=88=86=E6=A7=BD)=20+=20=E6=B7=B7=E5=93=8D=E6=B9=BF=E5=BA=A6?= =?UTF-8?q?=E6=BB=91=E5=9D=97(=E5=AE=9E=E6=97=B6);=20=E6=B9=BF=E9=87=8F?= =?UTF-8?q?=E6=8A=BD=E6=88=90=E7=8B=AC=E7=AB=8B=20DSP=20=E5=A2=9E=E7=9B=8A?= =?UTF-8?q?=E7=BA=A7(wet=5Fgain.py),=20=E6=B7=B7=E5=93=8D=E4=BB=8D?= =?UTF-8?q?=E7=94=A8=20PipeWire=20convolver=20=E8=B7=91=203=20=E7=A7=92?= =?UTF-8?q?=E9=95=BF=20IR;=20=E5=AE=9E=E6=B5=8B=E6=B9=BF=E9=87=8F=200=20?= =?UTF-8?q?=E6=97=A0=E5=B0=BE=E5=B7=B4=20/=201=20=E6=9C=89=203=20=E7=A7=92?= =?UTF-8?q?=E8=A1=B0=E5=87=8F=E5=B0=BE=E5=B7=B4?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- bin/collaplex-eq | 3 +- bin/collaplex-wet | 4 + dsp/common.py | 112 +++++---- dsp/eq.py | 51 ++-- dsp/wet_gain.py | 60 +++++ dsp/初始化.py | 24 ++ pipewire/20-collaplex-hrtf.conf | 12 +- web/index.html | 405 +++++++++++++++----------------- web/server.py | 152 ++++++------ 测试/test_wet.py | 116 +++++++++ 脚本/安装.sh | 15 +- 11 files changed, 600 insertions(+), 354 deletions(-) create mode 100644 bin/collaplex-wet create mode 100644 dsp/wet_gain.py create mode 100644 dsp/初始化.py create mode 100644 测试/test_wet.py diff --git a/bin/collaplex-eq b/bin/collaplex-eq index 66b61b5..72304fc 100644 --- a/bin/collaplex-eq +++ b/bin/collaplex-eq @@ -1,8 +1,9 @@ #!/bin/sh -# Collaplex 32 段 EQ + 总音量 —— pipe 插件入口 +# Collaplex 32 段 EQ + 总音量 + 房间混响 —— pipe 插件入口 # # pipe 插件的 command 只支持「程序 + 一个参数」, 所以参数全写在这里; # 且路径必须是纯 ASCII(中文路径会静默失效)。 # 用项目 venv 的解释器(带 scipy/numpy)。 export CX_RATE=96000 +export CX_REVERB="__REVERB__" exec "__PY__" "__DSP__" diff --git a/bin/collaplex-wet b/bin/collaplex-wet new file mode 100644 index 0000000..8dac15c --- /dev/null +++ b/bin/collaplex-wet @@ -0,0 +1,4 @@ +#!/bin/sh +# Collaplex 混响湿量增益级 —— pipe 插件入口 +export CX_RATE=96000 +exec "__PY__" "__DSP__" diff --git a/dsp/common.py b/dsp/common.py index 902d1f1..145a95e 100644 --- a/dsp/common.py +++ b/dsp/common.py @@ -1,22 +1,31 @@ """Collaplex 效果 DSP 公共部分: 前端 <-> DSP 的共享内存协议。 -前端(web/server.py) 与 DSP 进程(pipe 插件载荷)通过这一块共享内存通信: +前端(web/server.py) 与 DSP 进程(filter-chain 的 pipe 插件载荷)通过这一块共享内存通信: - 参数区: 32 段 EQ 增益(dB) + 总音量(dB) <- 前端写, DSP 读 - 版本区: 参数版本号 <- 前端写完参数后 +1 - 状态区: 输入/输出 RMS 与峰值(dB) <- DSP 写, 前端读 + 参数区: 32 段 EQ 增益(dB) + 总音量(dB) + 混响湿量(0~1) <- 前端写, DSP 读 + 版本区: 参数版本号 <- 前端写完参数后 +1 + 状态区: 每声道 输入/输出 RMS 与峰值(dB) + 推子 <- DSP 写, 前端读 -版本号最后写, DSP 只在版本号变化时重读参数, 所以前端一次改多个值也不会让 -DSP 读到半更新的数据。 +DSP 是**每声道一个进程**(pipe 插件按声道 fork, 两个进程命令相同), 它们靠 flock +抢声道槽位(0 = 左, 1 = 右)各写自己那份状态, 前端才能拿到双声道共 4 个电平表。 + +版本号最后写, 所以 DSP 只在版本号变化时重读参数, 前端一次改多个值不会读到半更新。 """ from __future__ import annotations +import fcntl import mmap import os import struct EQ_BANDS = 32 +CHANNELS = 2 EQ_PATH = "/dev/shm/collaplex-eq" +LOCK_PATH = "/dev/shm/collaplex-eq.ch" + +GAIN_LIMIT = 15.0 +VOLUME_LIMIT = 12.0 +WET_LIMIT = 1.0 # 1/3 倍频程中心频率(20 Hz 起, 32 段, 覆盖到 25 kHz) EQ_FREQS: list[float] = [ @@ -26,16 +35,14 @@ EQ_FREQS: list[float] = [ 5000.0, 6300.0, 8000.0, 10000.0, 12500.0, 16000.0, 20000.0, 25000.0, ] -GAIN_LIMIT = 15.0 -VOLUME_LIMIT = 12.0 +_PARAMS = struct.Struct("<34d") # 32 段增益 + 总音量 + 湿量 +_VER = struct.Struct(" mmap.mmap: os.close(fd) -def read_params(store: mmap.mmap) -> tuple[list[float], float, int]: - """读参数: (32 段增益 dB, 总音量 dB, 版本号)。""" - store.seek(DATA_OFF) - values = _DATA.unpack(store.read(_DATA.size)) +def read_params(store: mmap.mmap) -> tuple[list[float], float, float, int]: + """读参数: (32 段增益 dB, 总音量 dB, 湿量 0~1, 版本号)。""" + store.seek(PARAMS_OFF) + values = _PARAMS.unpack(store.read(_PARAMS.size)) store.seek(VER_OFF) version = _VER.unpack(store.read(_VER.size))[0] - return list(values[:EQ_BANDS]), float(values[EQ_BANDS]), int(version) + return (list(values[:EQ_BANDS]), float(values[EQ_BANDS]), + float(values[EQ_BANDS + 1]), int(version)) -def write_params(store: mmap.mmap, gains: list[float], volume_db: float) -> int: - """写参数并让版本号 +1, 返回新版本号。""" +def write_params(store: mmap.mmap, gains: list[float], volume_db: float, wet: float) -> int: + """写参数并让版本号 +1(版本号最后写), 返回新版本号。""" if len(gains) != EQ_BANDS: raise ValueError("gains 长度必须是 %d" % EQ_BANDS) store.seek(VER_OFF) version = int(_VER.unpack(store.read(_VER.size))[0]) - store.seek(DATA_OFF) - store.write(_DATA.pack(*[float(g) for g in gains], float(volume_db))) + store.seek(PARAMS_OFF) + store.write(_PARAMS.pack(*[float(g) for g in gains], float(volume_db), float(wet))) version = (version + 1) & 0xFFFFFFFF store.seek(VER_OFF) store.write(_VER.pack(version)) @@ -74,27 +82,45 @@ def write_params(store: mmap.mmap, gains: list[float], volume_db: float) -> int: return version -def read_state(store: mmap.mmap) -> dict[str, float]: - """读状态(DSP 写的电平)。""" +def read_state(store: mmap.mmap) -> dict[str, list[float]]: + """读双声道状态: 每个字段是一个 [左, 右] 列表(dB, 未知时为 -120)。""" store.seek(STATE_OFF) - in_rms, in_peak, out_rms, out_peak, gain_db, _spare = _STATE.unpack(store.read(_STATE.size)) - return { - "in_rms": float(in_rms), - "in_peak": float(in_peak), - "out_rms": float(out_rms), - "out_peak": float(out_peak), - "gain_db": float(gain_db), - } + raw = _STATE.unpack(store.read(_STATE.size)) + keys = ("in_rms", "in_peak", "out_rms", "out_peak", "gain_db") + return {key: [float(raw[ch * 6 + i]) for ch in range(CHANNELS)] + for i, key in enumerate(keys)} -def write_state(store: mmap.mmap, in_rms: float, in_peak: float, out_rms: float, - out_peak: float, gain_db: float) -> None: - """写状态(DSP 侧调用)。""" - store.seek(STATE_OFF) - store.write(_STATE.pack(float(in_rms), float(in_peak), float(out_rms), - float(out_peak), float(gain_db), 0.0)) +def write_state(store: mmap.mmap, channel: int, in_rms: float, in_peak: float, + out_rms: float, out_peak: float, gain_db: float) -> None: + """写本声道的状态(DSP 侧调用)。""" + if not 0 <= channel < CHANNELS: + return + store.seek(STATE_OFF + channel * _CH.size) + store.write(_CH.pack(float(in_rms), float(in_peak), float(out_rms), + float(out_peak), float(gain_db), 0.0)) -def defaults() -> tuple[list[float], float]: - """默认参数: 全平直 + 0 dB 总音量。""" - return [0.0] * EQ_BANDS, 0.0 +def claim_channel() -> int: + """占一个声道槽位并返回它(0 = 左, 1 = 右)。 + + pipe 插件对每个声道 fork 一个进程、命令完全一样, 只能用锁区分: 谁的锁在谁占位。 + 锁随进程消失自动释放, 所以重启后会重新分配(左右可能互换, 电平相近时无影响)。 + """ + for channel in range(CHANNELS): + path = "%s%d" % (LOCK_PATH, channel) + try: + fd = os.open(path, os.O_RDWR | os.O_CREAT, 0o666) + except OSError: + continue + try: + fcntl.flock(fd, fcntl.LOCK_EX | fcntl.LOCK_NB) + return channel # 故意不关 fd: 进程活着锁就一直在 + except OSError: + os.close(fd) + return 0 + + +def defaults() -> tuple[list[float], float, float]: + """默认参数: 全平直 + 0 dB 总音量 + 0.3 混响湿量。""" + return [0.0] * EQ_BANDS, 0.0, 0.3 diff --git a/dsp/eq.py b/dsp/eq.py index acff6fb..4c8c39f 100644 --- a/dsp/eq.py +++ b/dsp/eq.py @@ -2,7 +2,13 @@ 参数实时来自共享内存(dsp/common.py): 32 段增益(dB) + 总音量(dB)。 做法: 每段一个 peaking biquad, 32 段级联成二阶节(sos), 用 scipy.signal.sosfilt -逐块滤波(C 实现, 保留滤波状态, 无额外延迟), 末级乘总音量。 +逐块滤波(C 实现, 保留滤波状态, 无额外延迟), 末级乘总音量, 再做软限幅兜底。 + +混响不在这里: 它由 PipeWire 的 convolver 完成(能跑 3 秒长 IR), 湿量由独立的 +湿量增益级(wet_gain.py)实时控制 —— 图里的 mixer 增益是加载时定死的, 改不了。 + +每声道一个进程(pipe 插件按声道 fork, 命令相同), 靠 flock 抢声道槽位(0 左 / 1 右), +各写自己那份电平状态 —— 前端因此能画双声道共 4 个电平表。 """ from __future__ import annotations @@ -19,15 +25,16 @@ import common # noqa: E402 RATE = int(os.environ.get("CX_RATE", "96000")) BLOCK_BYTES = 4096 -N = BLOCK_BYTES // 4 # 每块样本数(1024) Q = 4.318 # 1/3 倍频程 +CEILING = 0.99 # -0.09 dBFS 绝对上限(只防真削波) +KNEE = 0.06 # 软拐点: 只有峰值超过约 -0.63 dBFS 才介入 +# 注: 原先 ceiling=0.891 / knee=0.25 让介入点低到 -3.9 dBFS, 而正常音乐峰值在 -4~0 dBFS, +# 于是限幅长期工作 -> 中低频谐波失真(听感"沙沙")。阈值贴近 0 后平时完全透明。 + def design_peaking(freq: float, gain_db: float, fs: int, q: float) -> list[float]: - """RBJ cookbook 的 peaking biquad, 返回一行 sos [b0 b1 b2 1 a1 a2]。 - - gain_db = 0 时结果恰好是单位传递函数, 所以 32 段可以一直挂着(不必增删节点)。 - """ + """RBJ cookbook 的 peaking biquad, 返回 sos 一行 [b0 b1 b2 1 a1 a2]。""" amp = 10.0 ** (gain_db / 40.0) w0 = 2.0 * math.pi * freq / fs alpha = math.sin(w0) / (2.0 * q) @@ -42,22 +49,15 @@ def design_peaking(freq: float, gain_db: float, fs: int, q: float) -> list[float def build_sos(gains: list[float], fs: int) -> np.ndarray: - """32 段增益 -> sos 矩阵(32 x 6)。""" + """32 段 peaking 级联成二阶节(平直段的 biquad 恰好是单位响应)。""" return np.array([design_peaking(freq, gain, fs, Q) for freq, gain in zip(common.EQ_FREQS, gains)], dtype=np.float64) -CEILING = 0.99 # -0.09 dBFS 绝对上限(只防真削波) -KNEE = 0.06 # 软拐点: 只有峰值超过约 -0.63 dBFS 才介入 -# 注: 原先 ceiling=0.891 / knee=0.25 让介入点低到 -3.9 dBFS, 而正常音乐峰值在 -4~0 dBFS, -# 于是限幅长期工作 -> 中低频谐波失真(听感"沙沙")。阈值贴近 0 后平时完全透明。 - - def soft_limit(x: np.ndarray, ceiling: float = CEILING, knee: float = KNEE) -> np.ndarray: - """软限幅: 幅值在 ceiling-knee 以内的样本原样通过, 超过的部分用 tanh 平滑渐近 ceiling。 + """软限幅: 幅值在 ceiling-knee 以内的样本原样通过, 超出部分用 tanh 平滑渐近 ceiling。 - 与 np.clip 的关键区别: 不产生平顶。硬削(平顶)听感就是"炸麦", 而且削掉的峰值 - 不增加响度; 软限幅只把峰值压下来, 波形其余部分不动, 所以放大后响度能真正上来。 + 和 np.clip 的区别: 不产生平顶(硬削听起来像炸麦), 只把峰值压下来。 """ threshold = ceiling - knee magnitude = np.abs(x) @@ -71,7 +71,7 @@ def soft_limit(x: np.ndarray, ceiling: float = CEILING, knee: float = KNEE) -> n def level_db(block: np.ndarray) -> tuple[float, float]: - """返回 (RMS dBFS, 峰值 dBFS)。""" + """(RMS dBFS, 峰值 dBFS)。""" if block.size == 0: return -120.0, -120.0 rms = float(np.sqrt(np.mean(block * block))) @@ -83,12 +83,13 @@ def level_db(block: np.ndarray) -> tuple[float, float]: def main() -> None: store = common.open_store() + channel = common.claim_channel() stdin = sys.stdin.buffer stdout = sys.stdout.buffer version = -1 - sos = build_sos(common.defaults()[0], RATE) - zi = np.zeros((sos.shape[0], 2), dtype=np.float64) + sos: np.ndarray | None = None + zi: np.ndarray | None = None volume = 1.0 volume_db = 0.0 @@ -99,10 +100,11 @@ def main() -> None: if len(raw) % 4: raw = raw[: len(raw) - (len(raw) % 4)] - gains, want_volume_db, new_version = common.read_params(store) + gains, want_volume_db, _wet, new_version = common.read_params(store) if new_version != version: version = new_version sos = build_sos(gains, RATE) + zi = np.zeros((sos.shape[0], 2), dtype=np.float64) volume_db = want_volume_db volume = 10.0 ** (volume_db / 20.0) @@ -110,12 +112,15 @@ def main() -> None: if x.size == 0: continue - y, zi = sosfilt(sos, x, zi=zi) - out = soft_limit(y * volume) + if sos is None or zi is None: + out = np.zeros(x.size, dtype=np.float64) + else: + y, zi = sosfilt(sos, x, zi=zi) + out = soft_limit(y * volume) in_rms, in_peak = level_db(x) out_rms, out_peak = level_db(out) - common.write_state(store, in_rms, in_peak, out_rms, out_peak, volume_db) + common.write_state(store, channel, in_rms, in_peak, out_rms, out_peak, volume_db) stdout.write(out.astype(" None: + store = common.open_store() + stdin = sys.stdin.buffer + stdout = sys.stdout.buffer + + version = -1 + wet = 0.0 + wet_prev = 0.0 + + while True: + raw = stdin.read(BLOCK_BYTES) + if not raw: + break + if len(raw) % 4: + raw = raw[: len(raw) - (len(raw) % 4)] + + _gains, _volume_db, want_wet, new_version = common.read_params(store) + if new_version != version: + version = new_version + wet = max(0.0, min(1.0, want_wet)) + + x = np.frombuffer(raw, dtype=" None: + store = common.open_store() + gains, volume_db, wet = common.defaults() + version = common.write_params(store, gains, volume_db, wet) + print("默认参数已写入: 全平直 / 总音量 %+.1f dB / 湿量 %.2f (版本 %d)" + % (volume_db, wet, version)) + + +if __name__ == "__main__": + main() diff --git a/pipewire/20-collaplex-hrtf.conf b/pipewire/20-collaplex-hrtf.conf index 97f23df..5b24864 100644 --- a/pipewire/20-collaplex-hrtf.conf +++ b/pipewire/20-collaplex-hrtf.conf @@ -22,8 +22,10 @@ context.modules = [ { type = builtin label = mixer name = hR control = { "Gain 1" = 1.0 "Gain 2" = 1.0 } } { type = builtin label = convolver name = revL config = { filename = "__REVERB__" gain = 0.126 } } { type = builtin label = convolver name = revR config = { filename = "__REVERB__" gain = 0.126 } } - { type = builtin label = mixer name = wetL control = { "Gain 1" = 1.0 "Gain 2" = __WET__ } } - { type = builtin label = mixer name = wetR control = { "Gain 1" = 1.0 "Gain 2" = __WET__ } } + { type = builtin label = pipe name = wetgL config = { command = "__WET_WRAPPER__" } } + { type = builtin label = pipe name = wetgR config = { command = "__WET_WRAPPER__" } } + { type = builtin label = mixer name = wetL control = { "Gain 1" = 1.0 "Gain 2" = 1.0 } } + { type = builtin label = mixer name = wetR control = { "Gain 1" = 1.0 "Gain 2" = 1.0 } } ] links = [ { output = "iL:Out" input = "cFLL:In" } @@ -36,10 +38,12 @@ context.modules = [ { output = "cFRR:Out" input = "hR:In 2" } { output = "hL:Out" input = "wetL:In 1" } { output = "hL:Out" input = "revL:In" } - { output = "revL:Out" input = "wetL:In 2" } + { output = "revL:Out" input = "wetgL:In" } + { output = "wetgL:Out" input = "wetL:In 2" } { output = "hR:Out" input = "wetR:In 1" } { output = "hR:Out" input = "revR:In" } - { output = "revR:Out" input = "wetR:In 2" } + { output = "revR:Out" input = "wetgR:In" } + { output = "wetgR:Out" input = "wetR:In 2" } ] inputs = [ "iL:In" "iR:In" ] outputs = [ "wetL:Out" "wetR:Out" ] diff --git a/web/index.html b/web/index.html index 714ff89..06f89fb 100644 --- a/web/index.html +++ b/web/index.html @@ -7,244 +7,229 @@ -
-
-

Collaplex 音效

- 滚轮调滑块 · Shift+滚轮微调 · 双击归零 -
- 入 -∞ dBFS - 出 -∞ dBFS -
-
+
+

Collaplex 音效滚轮调节 · Shift+滚轮微调 · 双击归零

+
就绪
+
-
-
-

电平归一化（响度目标）

-
- - -16.0 dB -
-
- -
-
- -
-

总音量放大

-
- - +6.0 dB -
-
- - - — -
-
+
+

电平

+
+
入 L
-∞
+
出 L
-∞
+
入 R
-∞
+
出 R
-∞
+
条长 = 实时 RMS(-60 ~ 0 dBFS),右侧小字 = 本块峰值
+
-
-
-
+
+

控制

+
+ 电平归一化 + + — +
+
+ 混响湿度 + + — +
+
+ 总音量 + + — +
+
+ +
+

32 段 EQ（1/3 倍频程 · ±15 dB）

+
+
+ +
diff --git a/web/server.py b/web/server.py index 10a4f82..da9b479 100644 --- a/web/server.py +++ b/web/server.py @@ -4,33 +4,32 @@ 页面: / 接口: - GET /api/state 参数 + DSP 实时电平 - POST /api/eq {"gains": [32 个 dB]} - POST /api/volume {"db": 数值} - POST /api/loudness {"target_db": 数值, "on": true/false} + GET /api/state 参数 + DSP 双声道电平 + POST /api/eq {"gains": [32 个数]} 每段 dB(±15) + POST /api/volume {"db": -12..12} 总音量 + POST /api/wet {"wet": 0..1} 混响湿量 + POST /api/loudness {"target_db": -30..-6, "on": bool} 响度归一化(DSP 每秒重读该 JSON) """ from __future__ import annotations import json import os import sys +import threading from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from typing import cast -HERE = os.path.dirname(os.path.abspath(__file__)) -sys.path.insert(0, os.path.join(HERE, "..", "dsp")) +sys.path.insert(0, os.path.join(os.path.dirname(os.path.abspath(__file__)), os.pardir, "dsp")) import common # noqa: E402 +PORT = int(sys.argv[1]) if len(sys.argv) > 1 else 8789 +UI_PATH = os.path.join(os.path.dirname(os.path.abspath(__file__)), "index.html") LOUDNESS_JSON = os.path.expanduser("~/.local/state/cinema-spatial/loudness.json") -UI = os.path.join(HERE, "index.html") -DEFAULT_PORT = 8789 def _as_float(value: object, fallback: float = 0.0) -> float: - """把 JSON 取到的值转 float(来源类型不安全, 这里只做值校验)。""" - if isinstance(value, bool): - return fallback + """把 JSON 里取到的值转成 float(类型不安全, 只在值上做校验)。""" if isinstance(value, (int, float)): return float(value) if isinstance(value, str): @@ -41,43 +40,45 @@ def _as_float(value: object, fallback: float = 0.0) -> float: return fallback +def _clamp(value: float, low: float, high: float) -> float: + return max(low, min(high, value)) + + def read_loudness() -> dict[str, object]: - """读响度归一化设置(归一化 DSP 每秒重读这个文件)。""" + fallback: dict[str, object] = {"target_db": -16.0, "on": True} try: with open(LOUDNESS_JSON, "r", encoding="utf-8") as fh: data = json.load(fh) - if isinstance(data, dict): - return cast(dict[str, object], data) except (OSError, ValueError): - pass - return {"on": True, "target_db": -16.0} + return fallback + return cast(dict[str, object], data) if isinstance(data, dict) else fallback -def write_loudness(target_db: float, on: bool) -> None: - os.makedirs(os.path.dirname(LOUDNESS_JSON), exist_ok=True) - with open(LOUDNESS_JSON, "w", encoding="utf-8") as fh: - json.dump({"on": bool(on), "target_db": float(target_db)}, fh) +def write_loudness(data: dict[str, object]) -> None: + """原子写(临时文件 + rename), 归一化 DSP 每秒重读。""" + directory = os.path.dirname(LOUDNESS_JSON) + if directory: + os.makedirs(directory, exist_ok=True) + tmp = LOUDNESS_JSON + ".tmp" + with open(tmp, "w", encoding="utf-8") as fh: + json.dump(data, fh) + fh.flush() + os.fsync(fh.fileno()) + os.replace(tmp, LOUDNESS_JSON) class Handler(BaseHTTPRequestHandler): store = common.open_store() - protocol_version = "HTTP/1.1" + lock = threading.Lock() - def log_message(self, format: str, *args: object) -> None: - pass - - def _send(self, body: bytes, ctype: str, code: int = 200) -> None: + def _send_json(self, payload: object, code: int = 200) -> None: + body = json.dumps(payload, ensure_ascii=False).encode("utf-8") self.send_response(code) - self.send_header("Content-Type", ctype) + self.send_header("Content-Type", "application/json; charset=utf-8") self.send_header("Content-Length", str(len(body))) - self.send_header("Cache-Control", "no-store") self.end_headers() self.wfile.write(body) - def _send_json(self, obj: object, code: int = 200) -> None: - self._send(json.dumps(obj, ensure_ascii=False).encode("utf-8"), - "application/json; charset=utf-8", code) - def _read_json(self) -> dict[str, object]: length = int(self.headers.get("Content-Length") or 0) raw = self.rfile.read(length) if length else b"{}" @@ -90,63 +91,74 @@ class Handler(BaseHTTPRequestHandler): def do_GET(self) -> None: if self.path in ("/", "/index.html"): try: - with open(UI, "rb") as fh: - self._send(fh.read(), "text/html; charset=utf-8") + with open(UI_PATH, "rb") as fh: + body = fh.read() except OSError: - self._send("面板文件缺失".encode("utf-8"), "text/plain; charset=utf-8", 500) + self.send_error(404) + return + self.send_response(200) + self.send_header("Content-Type", "text/html; charset=utf-8") + self.send_header("Content-Length", str(len(body))) + self.end_headers() + self.wfile.write(body) return if self.path == "/api/state": - gains, volume_db, _ver = common.read_params(self.store) - loudness = read_loudness() + with self.lock: + gains, volume_db, wet, _version = common.read_params(self.store) + levels = common.read_state(self.store) self._send_json({ "freqs": common.EQ_FREQS, "gains": gains, "volume_db": volume_db, + "wet": wet, "gain_limit": common.GAIN_LIMIT, "volume_limit": common.VOLUME_LIMIT, - "loudness": { - "target_db": _as_float(loudness.get("target_db", -16.0), -16.0), - "on": bool(loudness.get("on", True)), - }, - "levels": common.read_state(self.store), + "levels": levels, + "loudness": read_loudness(), }) return - self._send_json({"error": "not found"}, 404) + self.send_error(404) def do_POST(self) -> None: data = self._read_json() - if self.path == "/api/eq": - raw_gains = data.get("gains") - if not isinstance(raw_gains, list) or len(raw_gains) != common.EQ_BANDS: - self._send_json({"error": "gains 必须是 %d 个数" % common.EQ_BANDS}, 400) + with self.lock: + gains, volume_db, wet, _version = common.read_params(self.store) + + if self.path == "/api/eq": + raw_gains = data.get("gains") + if not isinstance(raw_gains, list) or len(raw_gains) != common.EQ_BANDS: + self._send_json( + {"ok": False, "error": "gains 需要 %d 个数" % common.EQ_BANDS}, 400) + return + gains = [_clamp(_as_float(g), -common.GAIN_LIMIT, common.GAIN_LIMIT) + for g in cast(list[object], raw_gains)] + elif self.path == "/api/volume": + volume_db = _clamp(_as_float(data.get("db", 0.0)), + -common.VOLUME_LIMIT, common.VOLUME_LIMIT) + elif self.path == "/api/wet": + wet = _clamp(_as_float(data.get("wet", 0.0)), 0.0, common.WET_LIMIT) + elif self.path == "/api/loudness": + current = read_loudness() + target = _clamp(_as_float(data.get("target_db", current.get("target_db", -16.0)), -16.0), + -30.0, -6.0) + on = data.get("on", current.get("on", True)) + write_loudness({"target_db": target, "on": bool(on)}) + self._send_json({"ok": True, "loudness": read_loudness()}) return - items = cast(list[object], raw_gains) - gains = [max(-common.GAIN_LIMIT, min(common.GAIN_LIMIT, _as_float(item))) for item in items] - _g, volume_db, _v = common.read_params(self.store) - version = common.write_params(self.store, gains, volume_db) - self._send_json({"ok": True, "version": version}) - return - if self.path == "/api/volume": - volume_db = max(-common.VOLUME_LIMIT, - min(common.VOLUME_LIMIT, _as_float(data.get("db", 0.0)))) - gains, _old, _v = common.read_params(self.store) - version = common.write_params(self.store, gains, volume_db) - self._send_json({"ok": True, "version": version, "volume_db": volume_db}) - return - if self.path == "/api/loudness": - current = read_loudness() - target_db = _as_float(data.get("target_db", current.get("target_db", -16.0)), -16.0) - on = bool(data.get("on", current.get("on", True))) - write_loudness(max(-40.0, min(-5.0, target_db)), on) - self._send_json({"ok": True, "loudness": read_loudness()}) - return - self._send_json({"error": "not found"}, 404) + else: + self.send_error(404) + return + + common.write_params(self.store, gains, volume_db, wet) + self._send_json({"ok": True, "gains": gains, "volume_db": volume_db, "wet": wet}) + + def log_message(self, format: str, *args: object) -> None: + pass def main() -> None: - port = int(sys.argv[1]) if len(sys.argv) > 1 else DEFAULT_PORT - server = ThreadingHTTPServer(("127.0.0.1", port), Handler) - print("Collaplex 音效控制面板: http://127.0.0.1:%d" % port, flush=True) + server = ThreadingHTTPServer(("127.0.0.1", PORT), Handler) + print("Collaplex 音效面板: http://127.0.0.1:%d" % PORT, flush=True) server.serve_forever() diff --git a/测试/test_wet.py b/测试/test_wet.py new file mode 100644 index 0000000..5590729 --- /dev/null +++ b/测试/test_wet.py @@ -0,0 +1,116 @@ +"""湿度滑块验证: 播脉冲, 测输出里的混响尾巴能量随湿量变化。""" +from __future__ import annotations + +import json +import subprocess +import time +import urllib.request +import wave + +import numpy as np +from scipy.io import wavfile + +RATE = 96000 +IMPULSE = "/tmp/imp.wav" +REC = "/tmp/imp_rec.wav" +API = "http://127.0.0.1:8789" + + +def sink() -> str: + """找数字输出设备的 node.name(含 iec958 的那个)。""" + try: + out = subprocess.run(["pw-dump"], capture_output=True, text=True, timeout=20).stdout + nodes = json.loads(out) + except (OSError, ValueError) as exc: + print(" (pw-dump 失败: %s)" % exc) + return "" + for node in nodes: + props = (node.get("info") or {}).get("props") or {} + if props.get("media.class") != "Audio/Sink": + continue + name = str(props.get("node.name", "")) + if "iec958" in name: + return name + print(" (没找到 iec958 设备, 现有 sink 已打印在上方)") + return "" + + +def post(path: str, body: dict[str, float]) -> None: + req = urllib.request.Request(API + path, data=json.dumps(body).encode(), + headers={"Content-Type": "application/json"}) + urllib.request.urlopen(req, timeout=10).read() + + +def make_impulse() -> None: + """0.1 秒脉冲 + 1.9 秒静音 = 2 秒素材。 + + 加静音尾巴是因为 pw-play 起播和 pw-record 启动都有几百毫秒延迟, + 素材太短会在录音真正开始前就播完(录到全零)。 + """ + n = int(RATE * 3.0) + x = np.zeros(n) + m = int(RATE * 0.1) + t = np.arange(m) / RATE + start = int(RATE * 1.0) # 脉冲放第 1 秒: 等录音真正跑起来再响 + x[start:start + m] = 0.5 * np.sin(2 * np.pi * 1000 * t) * np.hanning(m) + data = (x * 32767).astype(" np.ndarray: + """读回录音文件(取左声道)。""" + try: + _rate, data = wavfile.read(REC) + except OSError: + return np.zeros(1, dtype=np.float64) + arr = np.asarray(data, dtype=np.float64) + return arr[:, 0] if arr.ndim > 1 else arr + + +def tail_db(rec: np.ndarray) -> float: + """从录音里的峰值(脉冲)之后 0.2~1.2 秒算能量 = 混响尾巴。 + + 用峰值自校准, 免得依赖"录音开始时脉冲正好在第 N 秒"。 + """ + if rec.size < RATE: + return -120.0 + peak_idx = int(np.argmax(np.abs(rec))) + start = peak_idx + int(0.2 * RATE) + end = min(peak_idx + int(1.2 * RATE), rec.size) + if end <= start: + return -120.0 + rms = float(np.sqrt(np.mean(rec[start:end] ** 2))) + return 20.0 * np.log10(rms + 1e-12) + + +def main() -> None: + dev = sink() + if not dev: + print("找不到数字输出设备, 退出") + return + print(" 设备: %s" % dev.split(".")[-3]) + make_impulse() + for wet in (0.0, 0.3, 0.6, 1.0): + post("/api/wet", {"wet": wet}) + time.sleep(0.3) + # 先起录音再播放: pw-record 启动有延迟, 反过来会漏掉脉冲 + play = subprocess.Popen(["pw-play", "--target", "collaplex_vsink", IMPULSE], + stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) + time.sleep(0.3) + subprocess.run(["timeout", "3.5", "pw-record", "--target", dev, + "-P", "{ stream.capture.sink = true }", "--rate", str(RATE), + "--channels", "2", "--format", "f32", REC], + stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) + play.wait(timeout=6) + recorded = read_rec() + print(" 湿量 %.2f -> 混响尾巴 %6.1f dBFS" % (wet, tail_db(recorded))) + post("/api/wet", {"wet": 0.3}) + print(" (已恢复湿量 0.3)") + + +if __name__ == "__main__": + main() diff --git a/脚本/安装.sh b/脚本/安装.sh index 9684db2..f9517f0 100755 --- a/脚本/安装.sh +++ b/脚本/安装.sh @@ -39,6 +39,7 @@ echo "数字输出: $TARGET" mkdir -p "$(dirname "$WRAPPER")" sed -e "s|__DSP__|$ROOT/dsp/loudness_norm.py|" "$ROOT/bin/collaplex-loudness-norm" > "$WRAPPER" sed -e "s|__PY__|$ROOT/.venv/bin/python|" -e "s|__DSP__|$ROOT/dsp/eq.py|" \ + -e "s|__REVERB__|$ROOT/reverb/房间混响IR-96k.wav|" \ "$ROOT/bin/collaplex-eq" > "$HOME/.local/bin/collaplex-eq" chmod 755 "$WRAPPER" "$HOME/.local/bin/collaplex-eq" echo "已装入口: $WRAPPER / $HOME/.local/bin/collaplex-eq" @@ -48,11 +49,18 @@ mkdir -p "$CONF_DIR" sed -e "s|__WRAPPER__|$WRAPPER|" -e "s|__HRTF_IN__|$HRTF_IN|" \ "$ROOT/pipewire/10-collaplex-vsink.conf" > "$CONF_DIR/10-collaplex-vsink.conf" sed -e "s|__HRIR__|$ROOT/hrir|" -e "s|__REVERB__|$ROOT/reverb/房间混响IR-96k.wav|" \ - -e "s|__EQ_IN__|$EQ_IN|" -e "s|__WET__|$WET|" \ + -e "s|__EQ_IN__|$EQ_IN|" -e "s|__WET__|$WET|" -e "s|__WET_WRAPPER__|$HOME/.local/bin/collaplex-wet|" \ "$ROOT/pipewire/20-collaplex-hrtf.conf" > "$CONF_DIR/20-collaplex-hrtf.conf" -sed -e "s|__EQ_WRAPPER__|$HOME/.local/bin/collaplex-eq|" -e "s|__TARGET__|$TARGET|" \ +sed -e "s|__PY__|$ROOT/.venv/bin/python|" -e "s|__DSP__|$ROOT/dsp/wet_gain.py|" \ + "$ROOT/bin/collaplex-wet" > "$HOME/.local/bin/collaplex-wet" +chmod 755 "$HOME/.local/bin/collaplex-wet" +sed -e "s|__EQ_WRAPPER__|$HOME/.local/bin/collaplex-eq|" -e "s|__WET_WRAPPER__|$HOME/.local/bin/collaplex-wet|" -e "s|__TARGET__|$TARGET|" \ "$ROOT/pipewire/30-collaplex-eq.conf" > "$CONF_DIR/30-collaplex-eq.conf" cp "$ROOT/pipewire/91-collaplex-clock.conf" "$CONF_DIR/91-clock.conf" +# 共享内存不存在时写入默认参数(全平直 / 0 dB / 湿量 0.3); 已存在则保留用户当前设置 +if [ ! -f /dev/shm/collaplex-eq ]; then + "$ROOT/.venv/bin/python" "$ROOT/dsp/初始化.py" +fi echo "已装配置: 10-collaplex-vsink.conf / 20-collaplex-hrtf.conf(混响湿量 $WET) / 30-collaplex-eq.conf" # ---- 4. 重启前清掉残留 DSP(占住共享槽位会让新实例起不来) ---- @@ -60,7 +68,8 @@ for p in /proc/[0-9]*; do c=$(tr '\0' ' ' < "$p/cmdline" 2>/dev/null) || continue case "$c" in "/usr/bin/python3 "*dsp/loudness_norm.py*) kill -9 "${p#/proc/}" 2>/dev/null || true;; - *dsp/eq.py*) kill -9 "${p#/proc/}" 2>/dev/null || true;; + *"/.venv/bin/python "*dsp/eq.py*) kill -9 "${p#/proc/}" 2>/dev/null || true;; + *"/.venv/bin/python "*dsp/wet_gain.py*) kill -9 "${p#/proc/}" 2>/dev/null || true;; esac done sleep 1