From 5a4b11092cedad040e91f68e3e49ef59509dce97 Mon Sep 17 00:00:00 2001 From: lou Date: Sat, 19 Sep 2026 18:34:56 +0800 Subject: [PATCH] =?UTF-8?q?=E5=89=8D=E7=AB=AF:=20=E6=8A=BD=E5=85=AC?= =?UTF-8?q?=E5=85=B1=E5=B1=82(dsp/common.py=20=E5=85=B1=E4=BA=AB=E5=86=85?= =?UTF-8?q?=E5=AD=98=E5=8D=8F=E8=AE=AE)+=2032=E6=AE=B5EQ/=E6=80=BB?= =?UTF-8?q?=E9=9F=B3=E9=87=8F=20DSP(dsp/eq.py,=20scipy=20sosfilt)=20+=20?= =?UTF-8?q?=E6=9C=AC=E5=9C=B0=E6=8E=A7=E5=88=B6=E9=9D=A2=E6=9D=BF(web/,=20?= =?UTF-8?q?=E6=BB=91=E5=9D=97+=E6=BB=9A=E8=BD=AE=E6=8D=95=E6=8D=89);=20?= =?UTF-8?q?=E5=AE=89=E8=A3=85=E8=84=9A=E6=9C=AC=E6=8E=A5=2030-collaplex-eq?= =?UTF-8?q?;=20pyright=20strict=200=20error?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- README.md | 11 ++ bin/collaplex-eq | 8 + dsp/common.py | 100 +++++++++++++ dsp/eq.py | 102 +++++++++++++ pipewire/20-collaplex-hrtf.conf | 2 +- pipewire/30-collaplex-eq.conf | 41 ++++++ pyrightconfig.json | 13 ++ web/index.html | 250 ++++++++++++++++++++++++++++++++ web/server.py | 154 ++++++++++++++++++++ 脚本/安装.sh | 22 ++- 脚本/面板.sh | 4 + 11 files changed, 701 insertions(+), 6 deletions(-) create mode 100644 bin/collaplex-eq create mode 100644 dsp/common.py create mode 100644 dsp/eq.py create mode 100644 pipewire/30-collaplex-eq.conf create mode 100644 pyrightconfig.json create mode 100644 web/index.html create mode 100644 web/server.py create mode 100644 脚本/面板.sh diff --git a/README.md b/README.md index 2a5d5ec..d0ba2dc 100644 --- a/README.md +++ b/README.md @@ -37,6 +37,17 @@ filters2 = [ { type = bq_highshelf freq = 6000 gain = 0.0 } ] # 高频架 ★ 坑: `param_eq` 的端口名**带编号** —— `In 1` / `Out 1`(走 filters1)、`In 2` / `Out 2`(走 filters2)。 写成 `:In` / `:Out` 会报 `unknown input port` 并让**整张图加载失败**(模块静默降级成直通, 节点都不建)。 +## 控制面板 + +```sh +bash 脚本/面板.sh # http://127.0.0.1:8789 +``` + +- 电平归一化(响度目标)、总音量、32 段 EQ **全部是滑块**:滚轮调节、Shift+滚轮微调(0.1 dB)、双击归零 +- EQ 是 1/3 倍频程 32 段(20 Hz … 25 kHz),每段 ±15 dB,**实时生效**(DSP 从共享内存读参数,不用重装) +- 参数落点:`/dev/shm/collaplex-eq`(前端写、DSP 读,版本号保证不成半更新);响度目标写在 + `~/.local/state/cinema-spatial/loudness.json`(归一化 DSP 每秒重读) + ## 实测(2026-09-19) | 指标 | 数据 | diff --git a/bin/collaplex-eq b/bin/collaplex-eq new file mode 100644 index 0000000..66b61b5 --- /dev/null +++ b/bin/collaplex-eq @@ -0,0 +1,8 @@ +#!/bin/sh +# Collaplex 32 段 EQ + 总音量 —— pipe 插件入口 +# +# pipe 插件的 command 只支持「程序 + 一个参数」, 所以参数全写在这里; +# 且路径必须是纯 ASCII(中文路径会静默失效)。 +# 用项目 venv 的解释器(带 scipy/numpy)。 +export CX_RATE=96000 +exec "__PY__" "__DSP__" diff --git a/dsp/common.py b/dsp/common.py new file mode 100644 index 0000000..902d1f1 --- /dev/null +++ b/dsp/common.py @@ -0,0 +1,100 @@ +"""Collaplex 效果 DSP 公共部分: 前端 <-> DSP 的共享内存协议。 + +前端(web/server.py) 与 DSP 进程(pipe 插件载荷)通过这一块共享内存通信: + + 参数区: 32 段 EQ 增益(dB) + 总音量(dB) <- 前端写, DSP 读 + 版本区: 参数版本号 <- 前端写完参数后 +1 + 状态区: 输入/输出 RMS 与峰值(dB) <- DSP 写, 前端读 + +版本号最后写, DSP 只在版本号变化时重读参数, 所以前端一次改多个值也不会让 +DSP 读到半更新的数据。 +""" +from __future__ import annotations + +import mmap +import os +import struct + +EQ_BANDS = 32 +EQ_PATH = "/dev/shm/collaplex-eq" + +# 1/3 倍频程中心频率(20 Hz 起, 32 段, 覆盖到 25 kHz) +EQ_FREQS: list[float] = [ + 20.0, 25.0, 31.5, 40.0, 50.0, 63.0, 80.0, 100.0, + 125.0, 160.0, 200.0, 250.0, 315.0, 400.0, 500.0, 630.0, + 800.0, 1000.0, 1250.0, 1600.0, 2000.0, 2500.0, 3150.0, 4000.0, + 5000.0, 6300.0, 8000.0, 10000.0, 12500.0, 16000.0, 20000.0, 25000.0, +] + +GAIN_LIMIT = 15.0 +VOLUME_LIMIT = 12.0 + +_DATA = struct.Struct("<33d") # 32 段增益 + 总音量 +_VER = struct.Struct(" mmap.mmap: + """打开(不存在则创建)共享内存映射。""" + fd = os.open(EQ_PATH, os.O_RDWR | os.O_CREAT, 0o666) + try: + if os.fstat(fd).st_size < SIZE: + os.ftruncate(fd, SIZE) + return mmap.mmap(fd, SIZE) + finally: + 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)) + store.seek(VER_OFF) + version = _VER.unpack(store.read(_VER.size))[0] + return list(values[:EQ_BANDS]), float(values[EQ_BANDS]), int(version) + + +def write_params(store: mmap.mmap, gains: list[float], volume_db: 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))) + version = (version + 1) & 0xFFFFFFFF + store.seek(VER_OFF) + store.write(_VER.pack(version)) + store.flush() + return version + + +def read_state(store: mmap.mmap) -> dict[str, float]: + """读状态(DSP 写的电平)。""" + 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), + } + + +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 defaults() -> tuple[list[float], float]: + """默认参数: 全平直 + 0 dB 总音量。""" + return [0.0] * EQ_BANDS, 0.0 diff --git a/dsp/eq.py b/dsp/eq.py new file mode 100644 index 0000000..6bcfe6c --- /dev/null +++ b/dsp/eq.py @@ -0,0 +1,102 @@ +"""32 段 EQ + 总音量 —— pipe 插件载荷(单声道块进, 单声道块出)。 + +参数实时来自共享内存(dsp/common.py): 32 段增益(dB) + 总音量(dB)。 +做法: 每段一个 peaking biquad, 32 段级联成二阶节(sos), 用 scipy.signal.sosfilt +逐块滤波(C 实现, 保留滤波状态, 无额外延迟), 末级乘总音量。 +""" +from __future__ import annotations + +import math +import os +import sys + +import numpy as np +from scipy.signal import sosfilt + +sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) + +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 倍频程 + + +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 段可以一直挂着(不必增删节点)。 + """ + amp = 10.0 ** (gain_db / 40.0) + w0 = 2.0 * math.pi * freq / fs + alpha = math.sin(w0) / (2.0 * q) + cos_w0 = math.cos(w0) + b0 = 1.0 + alpha * amp + b1 = -2.0 * cos_w0 + b2 = 1.0 - alpha * amp + a0 = 1.0 + alpha / amp + a1 = -2.0 * cos_w0 + a2 = 1.0 - alpha / amp + return [b0 / a0, b1 / a0, b2 / a0, 1.0, a1 / a0, a2 / a0] + + +def build_sos(gains: list[float], fs: int) -> np.ndarray: + """32 段增益 -> sos 矩阵(32 x 6)。""" + return np.array([design_peaking(freq, gain, fs, Q) + for freq, gain in zip(common.EQ_FREQS, gains)], dtype=np.float64) + + +def level_db(block: np.ndarray) -> tuple[float, float]: + """返回 (RMS dBFS, 峰值 dBFS)。""" + if block.size == 0: + return -120.0, -120.0 + rms = float(np.sqrt(np.mean(block * block))) + peak = float(np.max(np.abs(block))) + rms_db = 20.0 * math.log10(rms) if rms > 1e-9 else -120.0 + peak_db = 20.0 * math.log10(peak) if peak > 1e-9 else -120.0 + return rms_db, peak_db + + +def main() -> None: + store = common.open_store() + 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) + volume = 1.0 + volume_db = 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, want_volume_db, new_version = common.read_params(store) + if new_version != version: + version = new_version + sos = build_sos(gains, RATE) + volume_db = want_volume_db + volume = 10.0 ** (volume_db / 20.0) + + x = np.frombuffer(raw, dtype=" HRTF 核心 -> [本图] 32 段 EQ + 总音量(pipe DSP) -> 数字输出 +context.modules = [ + { name = libpipewire-module-filter-chain + flags = [ nofail ] + args = { + node.description = "Collaplex EQ + 总音量" + media.name = "Collaplex EQ" + filter.graph = { + nodes = [ + { type = builtin label = copy name = c1L } + { type = builtin label = copy name = c1R } + { type = builtin label = pipe name = eqpL config = { command = "__EQ_WRAPPER__" } } + { type = builtin label = pipe name = eqpR config = { command = "__EQ_WRAPPER__" } } + { type = builtin label = copy name = c2L } + { type = builtin label = copy name = c2R } + ] + links = [ + { output = "c1L:Out" input = "eqpL:In" } + { output = "c1R:Out" input = "eqpR:In" } + { output = "eqpL:Out" input = "c2L:In" } + { output = "eqpR:Out" input = "c2R:In" } + ] + inputs = [ "c1L:In" "c1R:In" ] + outputs = [ "c2L:Out" "c2R:Out" ] + } + capture.props = { + media.class = Audio/Sink + node.name = collaplex_eq_in + audio.channels = 2 + audio.position = [ FL FR ] + } + playback.props = { + node.name = collaplex_eq_out + audio.channels = 2 + audio.position = [ FL FR ] + node.target = "__TARGET__" + } + } + } +] diff --git a/pyrightconfig.json b/pyrightconfig.json new file mode 100644 index 0000000..cefc662 --- /dev/null +++ b/pyrightconfig.json @@ -0,0 +1,13 @@ +{ + "include": ["dsp", "web"], + "exclude": ["dsp/loudness_norm.py", "dsp/状态.py"], + "extraPaths": ["dsp"], + "venvPath": ".", + "venv": ".venv", + "typeCheckingMode": "strict", + "pythonVersion": "3.11", + "reportMissingTypeStubs": false, + "reportUnknownMemberType": false, + "reportUnknownArgumentType": false, + "reportUnknownVariableType": false +} diff --git a/web/index.html b/web/index.html new file mode 100644 index 0000000..714ff89 --- /dev/null +++ b/web/index.html @@ -0,0 +1,250 @@ + + + + + +Collaplex 音效 + + + +
+
+

Collaplex 音效

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

电平归一化（响度目标）

+
+ + -16.0 dB +
+
+ +
+
+ +
+

总音量放大

+
+ + +6.0 dB +
+
+ + + — +
+
+
+ +
+
+
+
+
+ + + + diff --git a/web/server.py b/web/server.py new file mode 100644 index 0000000..10a4f82 --- /dev/null +++ b/web/server.py @@ -0,0 +1,154 @@ +"""Collaplex 音效控制面板 —— 本地服务(纯标准库, 无第三方依赖)。 + + python3 web/server.py [端口] 默认 8789 + +页面: / +接口: + GET /api/state 参数 + DSP 实时电平 + POST /api/eq {"gains": [32 个 dB]} + POST /api/volume {"db": 数值} + POST /api/loudness {"target_db": 数值, "on": true/false} +""" +from __future__ import annotations + +import json +import os +import sys +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")) + +import common # noqa: E402 + +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 + if isinstance(value, (int, float)): + return float(value) + if isinstance(value, str): + try: + return float(value) + except ValueError: + return fallback + return fallback + + +def read_loudness() -> dict[str, object]: + """读响度归一化设置(归一化 DSP 每秒重读这个文件)。""" + 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} + + +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) + + +class Handler(BaseHTTPRequestHandler): + store = common.open_store() + protocol_version = "HTTP/1.1" + + def log_message(self, format: str, *args: object) -> None: + pass + + def _send(self, body: bytes, ctype: str, code: int = 200) -> None: + self.send_response(code) + self.send_header("Content-Type", ctype) + 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"{}" + try: + data = json.loads(raw.decode("utf-8")) + except ValueError: + return {} + return cast(dict[str, object], data) if isinstance(data, dict) else {} + + 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") + except OSError: + self._send("面板文件缺失".encode("utf-8"), "text/plain; charset=utf-8", 500) + return + if self.path == "/api/state": + gains, volume_db, _ver = common.read_params(self.store) + loudness = read_loudness() + self._send_json({ + "freqs": common.EQ_FREQS, + "gains": gains, + "volume_db": volume_db, + "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), + }) + return + self._send_json({"error": "not found"}, 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) + 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) + + +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.serve_forever() + + +if __name__ == "__main__": + main() diff --git a/脚本/安装.sh b/脚本/安装.sh index 0521327..ae6bec0 100755 --- a/脚本/安装.sh +++ b/脚本/安装.sh @@ -19,6 +19,7 @@ WET="${WET:-0.3}" VOL_DB="${VOL_DB:-6}" VOL="$(/usr/bin/python3 -c "print(f'{10**(${VOL_DB}/20):.4f}')")" HRTF_IN="collaplex_hrtf_in" +EQ_IN="collaplex_eq_in" # ---- 1. 探测数字输出 ---- TARGET=$(pw-dump | /usr/bin/python3 -c ' @@ -39,23 +40,28 @@ echo "数字输出: $TARGET" # ---- 2. 部署 pipe 插件入口(必须 ASCII 路径) ---- mkdir -p "$(dirname "$WRAPPER")" sed -e "s|__DSP__|$ROOT/dsp/loudness_norm.py|" "$ROOT/bin/collaplex-loudness-norm" > "$WRAPPER" -chmod 755 "$WRAPPER" -echo "已装入口: $WRAPPER" +sed -e "s|__PY__|$ROOT/.venv/bin/python|" -e "s|__DSP__|$ROOT/dsp/eq.py|" \ + "$ROOT/bin/collaplex-eq" > "$HOME/.local/bin/collaplex-eq" +chmod 755 "$WRAPPER" "$HOME/.local/bin/collaplex-eq" +echo "已装入口: $WRAPPER / $HOME/.local/bin/collaplex-eq" # ---- 3. 渲染两份配置 ---- 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|__TARGET__|$TARGET|" -e "s|__WET__|$WET|" -e "s|__VOL__|$VOL|" \ + -e "s|__EQ_IN__|$EQ_IN|" -e "s|__WET__|$WET|" -e "s|__VOL__|$VOL|" \ "$ROOT/pipewire/20-collaplex-hrtf.conf" > "$CONF_DIR/20-collaplex-hrtf.conf" -echo "已装配置: 10-collaplex-vsink.conf / 20-collaplex-hrtf.conf(混响湿量 $WET, 总音量 +${VOL_DB}dB)" +sed -e "s|__EQ_WRAPPER__|$HOME/.local/bin/collaplex-eq|" -e "s|__TARGET__|$TARGET|" \ + "$ROOT/pipewire/30-collaplex-eq.conf" > "$CONF_DIR/30-collaplex-eq.conf" +echo "已装配置: 10-collaplex-vsink.conf / 20-collaplex-hrtf.conf(湿量 $WET, 总音量 +${VOL_DB}dB) / 30-collaplex-eq.conf" # ---- 4. 重启前清掉残留 DSP(占住共享槽位会让新实例起不来) ---- 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;; esac done sleep 1 @@ -72,5 +78,11 @@ 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*) n=$((n+1));; esac done -echo "DSP 进程数: $n (应为 2)" +echo "DSP 进程数: $n (归一化, 应为 2)" +m=0 +for p in /proc/[0-9]*; do + c=$(tr '\0' ' ' < "$p/cmdline" 2>/dev/null) || continue + case "$c" in */dsp/eq.py*) m=$((m+1));; esac +done +echo "EQ 进程数: $m (应为 2)" echo -n "默认输出: "; wpctl inspect @DEFAULT_SINK@ 2>/dev/null | grep -m1 "node.name" | sed 's/.*= //' diff --git a/脚本/面板.sh b/脚本/面板.sh new file mode 100644 index 0000000..1e86709 --- /dev/null +++ b/脚本/面板.sh @@ -0,0 +1,4 @@ +#!/bin/bash +# Collaplex 音效控制面板 —— 起本地服务(默认 8789) +cd "$(dirname "$0")/.." || exit 1 +exec .venv/bin/python web/server.py "${1:-8789}"