edgevoid
1e635f251b
面板: - 加内联 SVG favicon(深底 + 五根声波柱), 之前没有 favicon 浏览器显示默认破图标 - 响度归一化卡片: 高度 184 装不下(标题22+推子22+文字行20+间距 ≈ 70px), 而 #ln 只扣 46px -> canvas 溢出、上沿被「目标/判据」那行文字压住。改: 卡片 210px、#ln 扣 74px、 网格 430->456(把这 26px 补给上面两个电平表, 免得它们被压矮)。实测无重叠, #ln 126px ★ 重要纠正(2026-09-14 老板当场抓出): - 物理输出音量是**档位**, 不是"越大越响" —— 响度由 DSP 归一化保证(实测 0.35 就很好) - 上一版错误地把它"对齐"到虚拟声卡(1.0) -> 进削波保护、动态被压扁: 面板波形顶满 而听感反而更小(老 bug 复现)。已撤掉三处: · webui.py 桥启动的"对齐一次" -> 只兜静音 · 音频守护.py 第9项"物理≠虚拟就纠" -> 只在静音/近乎归零(<0.02)时才动手 · fix_phys_volume() 不再写正常档位 - config/组件配置单.json: 改记「档位 0.35」+ 写明教训
1088 lines
47 KiB
Python
1088 lines
47 KiB
Python
#!/usr/bin/env python3
|
||
# Collaplex 音效 Web 控制台 (2026-09-13) -- 零依赖, 仅 stdlib
|
||
# 布局按老板设计稿 collaplex web设计稿.drawio 实现:
|
||
# 顶栏 / 原始电平+处理后电平 / 三个信号开关(竖排) / 音量滑块 / 数字·模拟两条输出 /
|
||
# 采样参数 / 响度统一开关+滑块 / 5.1直通·混响·设备选择 / 状态信息
|
||
# 用法: python3 webui.py [--port 8788] [--bind 127.0.0.1]
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import array
|
||
import json
|
||
import math
|
||
import mmap
|
||
import os
|
||
import re
|
||
import struct
|
||
import subprocess
|
||
import sys
|
||
import threading
|
||
import time
|
||
from collections import deque
|
||
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
|
||
from typing import Any
|
||
from urllib.parse import ParseResult, parse_qs, urlparse
|
||
|
||
HERE = os.path.dirname(os.path.abspath(__file__))
|
||
INDEX = os.path.join(HERE, "index.html")
|
||
ROOT = os.path.dirname(HERE)
|
||
CONF = os.path.expanduser("~/.config/pipewire/pipewire.conf.d/90-cinema-spatial.conf")
|
||
SINK_51 = "cinema_spatial_sink"
|
||
SINK_UP = "cinema_spatial_up_sink"
|
||
CHAIN_OUT = "cinema_spatial_up_out"
|
||
CLI = os.path.join(ROOT, "空间音频")
|
||
HELPER = os.path.join(ROOT, "音频状态.py")
|
||
STATE_DIR = os.path.expanduser("~/.local/state/cinema-spatial")
|
||
VOL_FILE = os.path.join(STATE_DIR, "volume")
|
||
LOUD_FILE = os.path.join(STATE_DIR, "loudness.json")
|
||
|
||
# ★ 电平采集(2026-09-13 实测结论):
|
||
# pw-record --target <**数字节点 ID**> 才会接声卡的 monitor 端口;
|
||
# 给节点**名字**不解析 → 静默回退默认录制源(麦克风); `<名>.monitor` 本机根本不存在。
|
||
# 实测: cinema_spatial_up_sink:monitor_FR |-> pw-record:input_FR, -6dBFS 测试音录到 RMS=0.385
|
||
CHUNK_MS = 100
|
||
RATE = 48000
|
||
KEEP = 120 # 每个电平表保留点数(=12 秒)
|
||
METERS: dict[str, deque[dict[str, float]]] = {
|
||
"raw": deque(maxlen=KEEP), # 原始电平(虚拟声卡捕捉)
|
||
"post": deque(maxlen=KEEP), # 处理后电平(物理输出)
|
||
}
|
||
METER_LOCK = threading.Lock()
|
||
LOG: deque[str] = deque(maxlen=200) # 动作日志(状态信息区)
|
||
|
||
# 两条 route 的硬件增益基数(9-13 实测, 数字 1.0 / 模拟 1.11428 → 差 +0.94dB)
|
||
HW_BASE = {"digital": 1.0, "analog": 1.11428}
|
||
|
||
|
||
def hw_base(kind: str) -> float:
|
||
"""route 硬件增益基数; 卡报回来的异常值(实测某卡报 0.001)当 1.0, 免得补偿算出上千倍."""
|
||
b = HW_BASE.get(kind, 1.0)
|
||
return b if b > 0.05 else 1.0
|
||
|
||
|
||
# ---------------------------------------------------------------- 离线渲染
|
||
# 复用 离线渲染.py 自己的两道锁(渲染锁 + 文件就绪), 面板只是把文件喂进去 + 显示进度。
|
||
RENDER_SCRIPT = os.path.join(ROOT, "离线渲染.py")
|
||
PROGRESS_FILE = os.path.join(STATE_DIR, "render-progress.json")
|
||
UPLOAD_DIR = os.path.join(STATE_DIR, "uploads")
|
||
RENDER: dict[str, Any] = {"proc": None, "files": [], "started": 0.0,
|
||
"log": deque(maxlen=300)}
|
||
RENDER_LOCK = threading.Lock()
|
||
|
||
|
||
def render_status() -> dict[str, Any]:
|
||
"""渲染状态(前端轮询)。"""
|
||
proc = RENDER["proc"]
|
||
running = bool(proc is not None and proc.poll() is None)
|
||
try:
|
||
with open(PROGRESS_FILE, encoding="utf-8") as fh:
|
||
prog: dict[str, Any] = json.load(fh)
|
||
except (OSError, json.JSONDecodeError):
|
||
prog = {}
|
||
return {"running": running, "files": RENDER["files"], "progress": prog,
|
||
"log": list(RENDER["log"])[-28:],
|
||
"took": round(time.time() - RENDER["started"], 1) if RENDER["started"] else 0.0,
|
||
"exit": None if (running or proc is None) else proc.returncode}
|
||
|
||
|
||
def _render_pump(proc: subprocess.Popen[str]) -> None:
|
||
"""把渲染输出收进日志(面板状态区能看)。"""
|
||
assert proc.stdout is not None
|
||
for line in proc.stdout:
|
||
clean = line.replace("\r", "").strip()
|
||
if clean:
|
||
RENDER["log"].append(clean)
|
||
proc.wait()
|
||
|
||
|
||
def render_start(files: list[str], wet: float, target: float, overwrite: bool) -> str:
|
||
"""启动离线渲染(后台线程)。
|
||
|
||
★ 已有任务在跑就**拒绝**, 不排队堆积 —— 与 离线渲染.py 里的渲染锁同一个意思
|
||
(那道锁会再兜一层, 这里先给用户一个爽快的答复)。
|
||
"""
|
||
with RENDER_LOCK:
|
||
proc = RENDER["proc"]
|
||
if proc is not None and proc.poll() is None:
|
||
return "⛔ 已有渲染任务在跑, 等它跑完再扔"
|
||
ok_files = [f for f in files if os.path.isfile(f)]
|
||
if not ok_files:
|
||
return "⛔ 没有可渲染的文件(路径不对, 或者还是目录)"
|
||
skip = len(files) - len(ok_files)
|
||
args = [sys.executable, RENDER_SCRIPT, "--混响", "%.2f" % wet,
|
||
"--目标", "%.1f" % target, "--进度json", PROGRESS_FILE]
|
||
if overwrite:
|
||
args.append("--覆盖")
|
||
args += ok_files
|
||
try:
|
||
os.remove(PROGRESS_FILE)
|
||
except OSError:
|
||
pass
|
||
RENDER["files"] = ok_files
|
||
RENDER["started"] = time.time()
|
||
RENDER["log"].clear()
|
||
p = subprocess.Popen(args, cwd=ROOT, stdout=subprocess.PIPE,
|
||
stderr=subprocess.STDOUT, text=True, bufsize=1)
|
||
RENDER["proc"] = p
|
||
threading.Thread(target=_render_pump, args=(p,), daemon=True).start()
|
||
note = "(另外 %d 个路径无效已略过)" % skip if skip else ""
|
||
return "已开始渲染 %d 个文件%s, 下面看进度" % (len(ok_files), note)
|
||
|
||
|
||
def render_cancel() -> str:
|
||
"""取消渲染(终止进程; 子进程 ffmpeg 由它自己收)。"""
|
||
proc = RENDER["proc"]
|
||
if proc is None or proc.poll() is not None:
|
||
return "当前没有在跑的渲染任务"
|
||
proc.terminate()
|
||
time.sleep(0.6)
|
||
if proc.poll() is None:
|
||
proc.kill()
|
||
RENDER["log"].append("· 已手动取消")
|
||
return "已取消渲染"
|
||
|
||
|
||
def _env() -> dict[str, str]:
|
||
e = dict(os.environ)
|
||
e.setdefault("XDG_RUNTIME_DIR", "/run/user/%d" % os.getuid())
|
||
return e
|
||
|
||
|
||
def log(msg: str) -> None:
|
||
with METER_LOCK:
|
||
LOG.append("%s %s" % (time.strftime("%H:%M:%S"), msg))
|
||
|
||
|
||
def sh(*args: str, timeout: int = 15) -> str:
|
||
try:
|
||
p = subprocess.run(list(args), capture_output=True, text=True, timeout=timeout, env=_env())
|
||
return (p.stdout or "") + (p.stderr or "")
|
||
except Exception:
|
||
return ""
|
||
|
||
|
||
def dump() -> list[dict]:
|
||
try:
|
||
return json.loads(sh("pw-dump") or "[]")
|
||
except Exception:
|
||
return []
|
||
|
||
|
||
def node_list(kind: str) -> list[dict[str, object]]:
|
||
out: list[dict[str, object]] = []
|
||
for o in dump():
|
||
if o.get("type") != "PipeWire:Interface:Node":
|
||
continue
|
||
p = (o.get("info") or {}).get("props") or {}
|
||
if p.get("media.class") == kind:
|
||
out.append({"id": o.get("id"), "name": p.get("node.name", ""),
|
||
"desc": p.get("node.description", "")})
|
||
return out
|
||
|
||
|
||
def node_id(name: str) -> int:
|
||
"""按 node.name 精确取数字 ID (pw-record --target 只认 ID)."""
|
||
for o in dump():
|
||
p = (o.get("info") or {}).get("props") or {}
|
||
if p.get("node.name") == name:
|
||
return int(o.get("id") or -1)
|
||
return -1
|
||
|
||
|
||
def vol_of(nid: int, timeout: int = 15) -> tuple[float, bool]:
|
||
"""(音量, 是否静音). 音量桥轮询时传小 timeout, 免得卡住面板. """
|
||
t = sh("wpctl", "get-volume", str(nid), timeout=timeout)
|
||
m = re.search(r"Volume:\s*([0-9.]+)", t)
|
||
return (float(m.group(1)) if m else -1.0), ("MUTED" in t)
|
||
|
||
|
||
def virtual_sink() -> str:
|
||
"""当前生效的虚拟声卡(默认输出是谁就是谁; 都不是则算上混版)."""
|
||
ds = default_sink_name()
|
||
return ds if "cinema_spatial" in ds else SINK_UP
|
||
|
||
|
||
def default_sink_name() -> str:
|
||
m = re.search(r'node\.name = "([^"]+)"', sh("wpctl", "inspect", "@DEFAULT_AUDIO_SINK@"))
|
||
return m.group(1) if m else ""
|
||
|
||
|
||
def phys_sink() -> str:
|
||
"""物理输出设备名(排除虚拟声卡与显卡 pro-output)."""
|
||
c = [str(s["name"]) for s in node_list("Audio/Sink")
|
||
if s["name"] and "cinema_spatial" not in str(s["name"]) and "pro-output" not in str(s["name"])]
|
||
return (sorted(c, key=lambda x: (0 if "usb" in x else 1, x)) or [""])[0]
|
||
|
||
|
||
def chain_target() -> str:
|
||
"""滤波链的目标设备 = conf 里写的 node.target."""
|
||
try:
|
||
with open(CONF, encoding="utf-8") as f:
|
||
m = re.search(r'node\.target\s*=\s*"([^"]+)"', f.read())
|
||
return m.group(1) if m else ""
|
||
except Exception:
|
||
return ""
|
||
|
||
|
||
def links_of(out_name: str) -> list[str]:
|
||
res: list[str] = []
|
||
cap = False
|
||
for line in sh("pw-link", "-l").splitlines():
|
||
if line.strip().startswith(out_name + ":output_"):
|
||
cap = True
|
||
continue
|
||
if cap:
|
||
m = re.match(r"\s*\|->\s*(\S+)", line)
|
||
if m:
|
||
res.append(m.group(1))
|
||
else:
|
||
cap = False
|
||
return res
|
||
|
||
|
||
def route_kind() -> str:
|
||
"""当前物理设备走在数字(iec958/S/PDIF)还是模拟 route 上."""
|
||
p = phys_sink()
|
||
return "digital" if "iec958" in p or "spdif" in p.lower() else "analog"
|
||
|
||
|
||
def hrtf_name() -> str:
|
||
link = os.path.join(ROOT, "hrir", "current")
|
||
return os.path.basename(os.path.realpath(link)) if os.path.islink(link) else "-"
|
||
|
||
|
||
def clock_rate() -> str:
|
||
m = re.search(r"key:'clock\.rate'\s+value:'(\d+)'", sh("pw-metadata", "-n", "settings"))
|
||
return m.group(1) if m else "?"
|
||
|
||
|
||
def conf_params() -> dict[str, str]:
|
||
"""采样参数: tap 数/位深从 HRIR 的 WAV 头读(配置里没有 length, 那是希尔伯特段的).
|
||
|
||
注意: 配置里 `length = 33` 是 convRL/convRR(希尔伯特)的参数, 不是 HRIR 长度 —— 别抓错.
|
||
"""
|
||
out: dict[str, str] = {"taps": "-", "hrtf_dir": hrtf_name(), "ir": "-", "ir_rate": "-",
|
||
"ir_bits": "-"}
|
||
ir = os.path.join(ROOT, "hrir", "current", "FL_L.wav")
|
||
try:
|
||
out["ir"] = os.path.basename(ir)
|
||
except Exception:
|
||
pass
|
||
try:
|
||
with open(ir, "rb") as f:
|
||
if f.read(4) != b"RIFF" or f.read(4) == b"" or f.read(4) != b"WAVE":
|
||
return out
|
||
while True:
|
||
hdr = f.read(8)
|
||
if len(hdr) < 8:
|
||
break
|
||
cid, csz = hdr[:4], int.from_bytes(hdr[4:8], "little")
|
||
if cid == b"fmt ":
|
||
fmt = f.read(csz)
|
||
nch = int.from_bytes(fmt[2:4], "little")
|
||
out["ir_rate"] = str(int.from_bytes(fmt[4:8], "little"))
|
||
out["ir_bits"] = str(int.from_bytes(fmt[14:16], "little"))
|
||
out["_nch"] = str(nch or 1)
|
||
elif cid == b"data":
|
||
nch = int(out.get("_nch", "1") or 1)
|
||
per = max(1, nch * (int(out["ir_bits"]) // 8 if out["ir_bits"].isdigit() else 2))
|
||
out["taps"] = str(csz // per)
|
||
break
|
||
else:
|
||
f.seek(csz + (csz & 1), os.SEEK_CUR)
|
||
except Exception:
|
||
pass
|
||
out.pop("_nch", None)
|
||
return out
|
||
|
||
|
||
# ---------------- 电平采集 ----------------
|
||
|
||
def _db(x: float) -> float:
|
||
return round(20 * math.log10(x), 2) if x > 1e-9 else -120.0
|
||
|
||
|
||
def _meter_src(key: str) -> tuple[str, list[str]]:
|
||
"""该表该看的信号源.
|
||
|
||
raw = 链的输入(虚拟声卡 monitor) —— 被捕捉的原始音频
|
||
post = 链的输出(output_FL/FR) —— 真正的处理后信号
|
||
|
||
★ 2026-09-13 老板要求: 处理后电平"不要用麦克风捕捉, 要原始信号"。
|
||
pw-record --target 的自动连线会挑错端口(实测 post 两个输入都从 output_FR 来),
|
||
目标一旦解析不到还会静默回退默认录制源 = 麦克风 → 所以 link_meter() 连完必须核。
|
||
"""
|
||
if key == "raw":
|
||
return virtual_sink(), ["monitor_FL", "monitor_FR"]
|
||
return (CHAIN_OUT if node_id(CHAIN_OUT) > 0 else "cinema_spatial_out"), ["output_FL", "output_FR"]
|
||
|
||
|
||
METER_NAME = {"raw": "cinema-meter-raw", "post": "cinema-meter-post"}
|
||
|
||
|
||
def _meter_ports(nid: int, prefix: str) -> dict[str, str]:
|
||
"""某节点上以 prefix 开头的端口: {通道后缀: 端口 id}."""
|
||
res: dict[str, str] = {}
|
||
for o in dump():
|
||
if o.get("type") != "PipeWire:Interface:Port":
|
||
continue
|
||
pr = ((o.get("info") or {}).get("props") or {})
|
||
if int(pr.get("node.id") or -1) != nid:
|
||
continue
|
||
pn = str(pr.get("port.name") or "")
|
||
if pn.startswith(prefix):
|
||
res[pn[len(prefix):]] = str(o.get("id") or "")
|
||
return res
|
||
|
||
|
||
def link_meter(key: str) -> str:
|
||
"""把电平录制进程**显式**连到该看的信号上(端口 id 寻址, 绝不经过麦克风).
|
||
|
||
★ 2026-09-13 老板要求"处理后电平不要用麦克风捕捉, 要原始信号"。
|
||
做法: 录制进程用 --target 0(不自动连线) + 唯一 node.name 启动, 再按端口 **id**
|
||
显式连到目标信号。名字寻址不行 —— 两个 pw-record 进程同名(node.name 都是
|
||
pw-record, application.process.id 还是 None), "pw-record:input_FL" 会打到
|
||
另一个进程的端口上(实测踩过); 而 pw-record 自己找 target 失败时会静默回退
|
||
默认录制源 = 麦克风。
|
||
"""
|
||
src_name, _chans = _meter_src(key)
|
||
sid = rid = -1
|
||
for _ in range(15):
|
||
sid = node_id(src_name)
|
||
rid = node_id(METER_NAME[key])
|
||
if rid > 0 and sid > 0:
|
||
break
|
||
time.sleep(0.2)
|
||
if rid < 0:
|
||
return "电平[%s] 录制节点没进图" % key
|
||
if sid < 0:
|
||
return "电平[%s] 目标信号 %s 不在图里(不连, 宁可空表)" % (key, src_name)
|
||
sport = _meter_ports(sid, "monitor_") or _meter_ports(sid, "output_")
|
||
rport = _meter_ports(rid, "input_")
|
||
ok = 0
|
||
for ch, rp in rport.items():
|
||
sp = sport.get(ch)
|
||
if sp and "fail" not in sh("pw-link", sp, rp).lower():
|
||
ok += 1
|
||
msg = "电平[%s] %d/%d 条 ← %s(只此信号, 无麦克风)" % (key, ok, len(rport), src_name)
|
||
print(msg, flush=True)
|
||
return msg
|
||
|
||
|
||
def _meter_worker(key: str) -> None:
|
||
need = RATE * CHUNK_MS // 1000 * 2 * 4 # 立体声 f32
|
||
while True:
|
||
# 目标信号不在图里就不录(宁可空表, 也绝不让 pw-record 回退到麦克风)
|
||
if node_id(_meter_src(key)[0]) < 0:
|
||
time.sleep(2)
|
||
continue
|
||
proc = None
|
||
t0 = time.monotonic()
|
||
try:
|
||
# ★ --target 0 = 不自动连线; -P 起唯一名 → 之后由 link_meter() 按端口 id 显式连
|
||
proc = subprocess.Popen(
|
||
["pw-record", "--target", "0",
|
||
"-P", "{ node.name = %s node.autoconnect = false }" % METER_NAME[key],
|
||
"--rate", str(RATE), "--channels", "2", "--format", "f32", "-"],
|
||
stdout=subprocess.PIPE, stderr=subprocess.DEVNULL, env=_env())
|
||
try:
|
||
link_meter(key)
|
||
except Exception as exc: # 采集不能因为连线校验挂了而停
|
||
print("电平[%s] 连线异常: %r" % (key, exc), flush=True)
|
||
while True:
|
||
# ★ 每 15s 主动重开一次录制: 重建链路(改混响/参数/设备)后节点 id 会变,
|
||
# 老进程会一直从已消失的目标上录静音 —— 实测电平表会卡在 -120 dBFS
|
||
if time.monotonic() - t0 > 15.0:
|
||
break
|
||
fh = proc.stdout
|
||
raw = fh.read(need) if fh else b""
|
||
if not raw or len(raw) < need:
|
||
break
|
||
a = array.array("f")
|
||
a.frombytes(raw)
|
||
lf = a[0::2]
|
||
rt = a[1::2]
|
||
pk_l = max(abs(v) for v in lf)
|
||
pk_r = max(abs(v) for v in rt)
|
||
rms_l = math.sqrt(sum(v * v for v in lf) / len(lf))
|
||
rms_r = math.sqrt(sum(v * v for v in rt) / len(rt))
|
||
with METER_LOCK:
|
||
# ★ 2026-09-13 老板要求电平分 4 路(原始 L/R + 处理后 L/R):
|
||
# 这里左右**分开**存, 不再 max() 合并成一条线。
|
||
# 旧的 peak/rms(取两声道较大者)保留, 兼容别处调用。
|
||
METERS[key].append({
|
||
"peakL": _db(pk_l), "peakR": _db(pk_r),
|
||
"rmsL": _db(rms_l), "rmsR": _db(rms_r),
|
||
"peak": _db(max(pk_l, pk_r)),
|
||
"rms": _db(max(rms_l, rms_r)),
|
||
})
|
||
except Exception:
|
||
pass
|
||
finally:
|
||
if proc is not None:
|
||
try:
|
||
proc.kill()
|
||
except Exception:
|
||
pass
|
||
time.sleep(0.4)
|
||
|
||
|
||
def start_meters() -> None:
|
||
for key in ("raw", "post"):
|
||
threading.Thread(target=_meter_worker, args=(key,), daemon=True).start()
|
||
|
||
|
||
def meter_points(key: str) -> list[dict[str, float]]:
|
||
with METER_LOCK:
|
||
return list(METERS[key])
|
||
|
||
|
||
# ---------------- 信号开关 ----------------
|
||
|
||
def toggle_state() -> dict[str, bool]:
|
||
"""三个信号开关的真实状态."""
|
||
v = virtual_sink()
|
||
vid = node_id(v)
|
||
_, vmut = vol_of(vid) if vid > 0 else (0.0, False)
|
||
pid = node_id(chain_target() or phys_sink())
|
||
_, pmut = vol_of(pid) if pid > 0 else (0.0, False)
|
||
tgt = chain_target() or phys_sink()
|
||
linked = bool(links_of(CHAIN_OUT)) and any(tgt in x for x in links_of(CHAIN_OUT))
|
||
return {"raw": not vmut, "virtual": linked, "output": not pmut}
|
||
|
||
|
||
def toggle(which: str, on: bool) -> str:
|
||
if which == "raw": # 原始音频信号: 虚拟声卡的静音
|
||
nid = node_id(virtual_sink())
|
||
if nid < 0:
|
||
return "找不到虚拟声卡"
|
||
sh("wpctl", "set-mute", str(nid), "0" if on else "1")
|
||
return "原始音频信号: %s" % ("通" if on else "断")
|
||
if which == "output": # 输出设备信号: 物理输出静音
|
||
nid = node_id(chain_target() or phys_sink())
|
||
if nid < 0:
|
||
return "找不到物理输出"
|
||
sh("wpctl", "set-mute", str(nid), "0" if on else "1")
|
||
return "输出设备信号: %s" % ("通" if on else "断")
|
||
if which == "virtual": # 虚拟声卡信号: 真断开/接回连线
|
||
tgt = chain_target() or phys_sink()
|
||
if not tgt:
|
||
return "找不到目标设备"
|
||
outs = [x for x in sh("pw-link", "-o").split() if x.startswith(CHAIN_OUT + ":output_")]
|
||
ins = [x for x in sh("pw-link", "-i").split() if x.startswith(tgt + ":playback_")]
|
||
acts = []
|
||
for i, o in enumerate(outs):
|
||
if i < len(ins):
|
||
acts.append(((o, ins[i]), "接通"))
|
||
if not outs or not ins:
|
||
return "拿不到链输出/设备端口"
|
||
for (o, i), _ in acts:
|
||
if on:
|
||
sh("pw-link", o, i)
|
||
else:
|
||
sh("pw-link", "-d", o, i)
|
||
return "虚拟声卡信号: %s(%d 路)" % ("通" if on else "断", len(acts))
|
||
return "未知开关"
|
||
|
||
|
||
# ---------------- 响度归一化(pipe 插件挂自研 DSP) ----------------
|
||
# ★ 数据源是 /dev/shm 的共享内存(DSP 进程写的), 不是录音 —— "程序内部状态"(实际增益)
|
||
# 从音频里看不出来, 只能问 DSP。字段布局见 响度归一化/loudness_norm.py 顶部注释。
|
||
# ★ 这一段跟上面那个"响度统一"(数字/模拟两条 route 的硬件增益补偿)不是一回事:
|
||
# 那个统一的是"哪条输出路", 这个统一的是"每个视频/软件自己的响度"。
|
||
LN_SHM = "/dev/shm/collaplex-loudness"
|
||
LN_STRIDE = 32
|
||
LN_SIZE = 80
|
||
LN_TARGET_OFF = 64
|
||
|
||
|
||
def loudness_state() -> dict[str, object]:
|
||
"""响度归一化的实时状态: 4 路电平(原始/处理后 x L/R) + 实际增益(推子) + 目标电平.
|
||
|
||
未启用(共享区不在 / 两个声道都超时没更新)时 ok=False, 前端显示占位。
|
||
"""
|
||
out: dict[str, object] = {
|
||
"ok": False, "target_db": None, "gain_db": None, "loud": None,
|
||
"raw_L": None, "raw_R": None, "post_L": None, "post_R": None,
|
||
}
|
||
if not os.path.exists(LN_SHM):
|
||
return out
|
||
rec: list[tuple[float, float, float, float, float, float]] = []
|
||
try:
|
||
fd = os.open(LN_SHM, os.O_RDONLY)
|
||
try:
|
||
mm = mmap.mmap(fd, LN_SIZE, prot=mmap.PROT_READ)
|
||
try:
|
||
target = struct.unpack_from("<f", mm, LN_TARGET_OFF)[0]
|
||
now = time.monotonic()
|
||
for i in (0, 1):
|
||
b = i * LN_STRIDE
|
||
rec.append((
|
||
struct.unpack_from("<f", mm, b)[0], # [0] e_smooth(响度判据)
|
||
struct.unpack_from("<f", mm, b + 8)[0], # [1] 本块**输入**峰值
|
||
struct.unpack_from("<f", mm, b + 4)[0], # [2] e_block(回退用)
|
||
struct.unpack_from("<f", mm, b + 12)[0], # [3] gain_db
|
||
now - struct.unpack_from("<d", mm, b + 16)[0], # [4] 数据年龄
|
||
struct.unpack_from("<f", mm, b + 28)[0], # [5] ★本块输出峰值(旧版=0)
|
||
))
|
||
finally:
|
||
mm.close()
|
||
finally:
|
||
os.close(fd)
|
||
except Exception:
|
||
return out
|
||
if not rec or min(r[4] for r in rec) > 1.0: # 都超过 1 秒没更新 = 没在跑
|
||
return out
|
||
out["ok"] = True
|
||
out["target_db"] = round(target, 2)
|
||
out["gain_db"] = round(rec[0][3], 2)
|
||
# ★★ 2026-09-13 口径修正(老板报"电平怎么都顶满了"):
|
||
# 旧版: raw = 块能量开方(RMS) 却标成"峰值"; post = 快时标的块能量 + 慢时标的增益
|
||
# → 瞬态时会算出 >0 dBFS 的**不可能**值(实测出现过 +5.5 dBFS), 看着像顶满。
|
||
# 现在: raw = 本块**输入峰值**; post = DSP 上报的**本块输出峰值**(真实值, 受 PEAK_CEIL 限制)。
|
||
# 旧版 DSP(还没写 +28) → post 退回估算, 并截顶到 PEAK_CEIL, 免得又出现不可能值。
|
||
ceil_db = _db(0.985) # DSP 的峰值保护上限 ≈ -0.13 dBFS
|
||
for i, name in ((0, "L"), (1, "R")):
|
||
raw = _db(max(rec[i][1], 0.0))
|
||
out["raw_" + name] = round(raw, 2)
|
||
pko = max(rec[i][5], 0.0)
|
||
if pko > 0.0:
|
||
out["post_" + name] = round(_db(pko), 2)
|
||
else:
|
||
out["post_" + name] = round(min(raw + rec[i][3], ceil_db), 2)
|
||
joint = max(rec[0][0], 0.0) + max(rec[1][0], 0.0)
|
||
out["loud"] = round(_db(math.sqrt(joint)), 2) # 归一化的判据(两声道联合响度)
|
||
return out
|
||
|
||
|
||
# ---------------- 响度统一 ----------------
|
||
|
||
def load_loud() -> dict[str, object]:
|
||
# target_db = 响度归一化(pipe+DSP)的目标电平, DSP 每秒读这个文件;
|
||
# ★ 与上面的 unified/target「响度统一」(数字/模拟两条输出路的增益补偿)不是一回事, 名字撞车
|
||
d: dict[str, object] = {"unified": False, "target": 0.33, "digital": 1.0, "analog": 1.0,
|
||
"target_db": -16.0}
|
||
try:
|
||
with open(LOUD_FILE, encoding="utf-8") as f:
|
||
d.update(json.load(f))
|
||
except Exception:
|
||
pass
|
||
return d
|
||
|
||
|
||
def _num(v: object, dflt: float) -> float:
|
||
"""从 JSON 取数(可能是 str/int/float), 兜底为 float."""
|
||
if isinstance(v, bool):
|
||
return dflt
|
||
if isinstance(v, (int, float)):
|
||
return float(v)
|
||
try:
|
||
return float(str(v))
|
||
except ValueError:
|
||
return dflt
|
||
|
||
|
||
def save_loud(d: dict[str, object]) -> None:
|
||
try:
|
||
os.makedirs(STATE_DIR, exist_ok=True)
|
||
with open(LOUD_FILE, "w", encoding="utf-8") as f:
|
||
json.dump(d, f, ensure_ascii=False, indent=1)
|
||
except OSError:
|
||
pass
|
||
|
||
|
||
def params_state() -> dict[str, object]:
|
||
"""采样参数(HRTF 模型 / tap / 采样率) —— 由 CLI「参数」写."""
|
||
p = os.path.expanduser("~/.local/state/cinema-spatial/params.json")
|
||
try:
|
||
with open(p, encoding="utf-8") as fh:
|
||
d = json.load(fh)
|
||
except (OSError, ValueError):
|
||
d = {}
|
||
return d if isinstance(d, dict) else {}
|
||
|
||
|
||
def param_options() -> dict[str, list[str]]:
|
||
"""可选项: hrir 下能用的模型 / 已裁的 tap 目录 / 允许的采样率."""
|
||
hrir = os.path.join(ROOT, "hrir")
|
||
models: list[str] = []
|
||
taps: list[str] = []
|
||
if os.path.isdir(hrir):
|
||
for n in sorted(os.listdir(hrir)):
|
||
if n.startswith("taps-"):
|
||
taps.append(n[5:])
|
||
elif n not in ("current", "gain") and os.path.isfile(os.path.join(hrir, n, "FL_L.wav")):
|
||
models.append(n)
|
||
taps.sort(key=int, reverse=True)
|
||
return {"hrtf": models, "taps": taps, "rate": ["96000", "48000", "44100"]}
|
||
|
||
|
||
def mono_state() -> dict[str, object]:
|
||
"""单声道设置(自动/开/关) + 链输出节点实际声道数(1 = 合并生效)."""
|
||
p = os.path.expanduser("~/.local/state/cinema-spatial/mono.json")
|
||
s = "自动"
|
||
try:
|
||
with open(p, encoding="utf-8") as fh:
|
||
d = json.load(fh)
|
||
if isinstance(d, dict) and isinstance(d.get("on"), bool):
|
||
s = "开" if d["on"] else "关"
|
||
except (OSError, ValueError):
|
||
pass
|
||
ch = 2
|
||
try:
|
||
raw = subprocess.run(["pw-dump"], capture_output=True, text=True, timeout=8).stdout
|
||
for o in json.loads(raw):
|
||
pr = ((o.get("info") or {}).get("props") or {})
|
||
if pr.get("node.name") == "cinema_spatial_up_out":
|
||
ch = int(pr.get("audio.channels") or 2)
|
||
except (OSError, ValueError, subprocess.SubprocessError):
|
||
pass
|
||
return {"set": s, "channels": ch}
|
||
|
||
|
||
def reverb_state() -> dict[str, object]:
|
||
"""混响运行时状态(on/wet) —— 由 CLI「混响」写, 生成配置.py 读."""
|
||
p = os.path.expanduser("~/.local/state/cinema-spatial/reverb.json")
|
||
try:
|
||
with open(p, encoding="utf-8") as fh:
|
||
d = json.load(fh)
|
||
except (OSError, ValueError):
|
||
return {"on": False, "wet": 0.25}
|
||
return {"on": bool(d.get("on")), "wet": float(d.get("wet") or 0.0)}
|
||
|
||
|
||
def set_speaker(want: float) -> str:
|
||
"""写「扬声器电平」(物理输出音量 = 老板听到的音量).
|
||
|
||
统一响度开着时物理 = 基数 x 目标, 直接写物理会被下次 apply 抹掉 → 写目标;
|
||
基数随机型/route 不同(实测本机数字 0.900, 不是表里的 1.0 → 直接算差 10%),
|
||
所以闭环自校正: 按实测比例算 → 写 → 回读 → 残差修正, 最多 3 轮。
|
||
"""
|
||
d = load_loud()
|
||
kind = route_kind()
|
||
pid = node_id(chain_target() or phys_sink())
|
||
before, _ = vol_of(pid) if pid > 0 else (-1.0, False)
|
||
ratio = 0.0
|
||
if d.get("unified") and before > 0.01 and _num(d.get("target"), 0.0) > 0.01:
|
||
ratio = before / _num(d.get("target"), 1.0)
|
||
for _pass in range(3):
|
||
if d.get("unified"):
|
||
base = ratio if ratio > 0.05 else hw_base(kind)
|
||
d["target"] = max(0.0, min(1.0, want / base))
|
||
else:
|
||
d[kind] = want
|
||
save_loud(d)
|
||
apply_loudness()
|
||
got, _m = vol_of(pid) if pid > 0 else (-1.0, False)
|
||
if got > 0.01 and d.get("unified"):
|
||
ratio = got / max(_num(d.get("target"), 1.0), 1e-6) # 用真实读数校准基数
|
||
if abs(got - want) <= 0.01:
|
||
break
|
||
elif abs(got - want) <= 0.01:
|
||
break
|
||
got, mut = vol_of(pid) if pid > 0 else (-1.0, False)
|
||
return "扬声器电平 %.3f → 物理输出 %.3f (硬件基数 %.3f)%s" % (
|
||
want, got, ratio if ratio > 0 else hw_base(kind), " (静音)" if mut else "")
|
||
|
||
|
||
# ---------------- 菜单音量桥 (2026-09-13) ----------------
|
||
# ★ 老板报"ubuntu 菜单音量绑到输出增益了, 应该绑到扬声器电平上" —— 实测确认:
|
||
# GNOME 顶栏/媒体键改的是 @DEFAULT_AUDIO_SINK@ = **虚拟声卡音量** = 链的**输入增益**,
|
||
# 而那一级会被「响度归一化」自动补回来(所以按键几乎没效果), 压太低还会把 DSP 逼到
|
||
# 顶格(面板推子"一飞冲天")、把整条链的削波余量吃光。
|
||
# 做法(不引入新层级, 不加新设备): 把虚拟声卡那次改动当**用户音量指令**读走 →
|
||
# 按**比例**折算到物理输出(走 set_speaker 的闭环校准) → 再把虚拟声卡**复位到参考值**
|
||
# (削波余量)。于是: 菜单音量 = 扬声器电平; 「输入增益」仍是纯削波余量旋钮。
|
||
BRIDGE_REF: float | None = None # 链输入增益参考值(削波余量); 面板「输入增益」滑块改它
|
||
BRIDGE_MUTED: bool = False # 菜单静音中
|
||
BRIDGE_SAVED: float = 0.0 # 菜单静音前的扬声器电平
|
||
BRIDGE_VID: int = 0 # 虚拟声卡节点 id(缓存, 省掉每轮 pw-dump)
|
||
BRIDGE_PID: int = 0 # 物理输出节点 id(缓存)
|
||
BRIDGE_ID_AT: float = 0.0 # id 上次刷新时刻
|
||
BRIDGE_OK: bool = True
|
||
BRIDGE_ERR: str = ""
|
||
|
||
|
||
def vol_bridge_tick() -> str | None:
|
||
"""轮询一次; 有动作返回日志文本, 否则 None。"""
|
||
global BRIDGE_REF, BRIDGE_MUTED, BRIDGE_SAVED, BRIDGE_VID, BRIDGE_PID, BRIDGE_ID_AT
|
||
now = time.time()
|
||
# id 每 5 秒(或失效时)重解析一次 —— node_id 要跑 pw-dump, 不能每轮都跑
|
||
if BRIDGE_VID <= 0 or now - BRIDGE_ID_AT > 5.0:
|
||
BRIDGE_VID = node_id(SINK_UP)
|
||
BRIDGE_PID = node_id(chain_target() or phys_sink())
|
||
BRIDGE_ID_AT = now
|
||
if BRIDGE_VID <= 0 or BRIDGE_PID <= 0:
|
||
return None
|
||
v, vm = vol_of(BRIDGE_VID, timeout=4)
|
||
if v < 0: # id 已失效(音频栈重启过) → 下轮重解析
|
||
BRIDGE_ID_AT = 0.0
|
||
return None
|
||
if BRIDGE_REF is None: # 首次: 以**实况**为参考
|
||
# ★ 不能采信 VOL_FILE(存档): 重建/别处改动之后实况可能与存档不一致,
|
||
# 那桥会把这个差值当成"菜单音量"折算一次 → 音量无故跳一下(2026-09-13 教训)。
|
||
BRIDGE_REF = v if v > 0.01 else 0.35
|
||
# ★★ 2026-09-14 修正: **不要**在这里把物理输出"对齐"到虚拟声卡 —— 物理输出音量
|
||
# 是一个**档位**(WirePlumber 存档里多少就是多少), 响度由 DSP 归一化负责。
|
||
# 上一版就是这么把老板调好的 0.35 顶到 1.0 的, 结果进削波保护、动态被压扁:
|
||
# 面板波形顶满而听感反而更小(老 bug 复现)。启动时**只兜静音**, 档位一律不动。
|
||
_p0, pm0 = vol_of(BRIDGE_PID, timeout=4)
|
||
if pm0:
|
||
sh("wpctl", "set-mute", str(BRIDGE_PID), "0", timeout=4)
|
||
return "启动检查: 物理输出原本静音 → 已解除(档位不动)"
|
||
return None
|
||
ref = BRIDGE_REF
|
||
p, _pm = vol_of(BRIDGE_PID, timeout=4)
|
||
if p < 0:
|
||
return None
|
||
if vm or v <= 0.001: # 菜单静音 / 音量拉到底
|
||
if not BRIDGE_MUTED:
|
||
BRIDGE_MUTED = True
|
||
BRIDGE_SAVED = p
|
||
set_speaker(0.0)
|
||
return "菜单音量: 静音 → 扬声器电平 0.000"
|
||
return None
|
||
if BRIDGE_MUTED: # 解除静音
|
||
BRIDGE_MUTED = False
|
||
back = BRIDGE_SAVED
|
||
sh("wpctl", "set-volume", str(BRIDGE_VID), "%.3f" % ref, timeout=4)
|
||
if back > 0.01:
|
||
set_speaker(back)
|
||
return "菜单音量: 取消静音 → 扬声器电平 %.3f" % (back if back > 0.01 else p)
|
||
if abs(v - ref) <= 0.02: # 没被动过
|
||
return None
|
||
ratio = v / ref if ref > 0.001 else 1.0 # 外部(菜单/媒体键/系统设置)改的
|
||
want = max(0.0, min(1.5, p * ratio))
|
||
msg = set_speaker(want)
|
||
sh("wpctl", "set-volume", str(BRIDGE_VID), "%.3f" % ref, timeout=4) # 复位削波余量
|
||
return "菜单音量 %.3f (×%.2f) → %s" % (v, ratio, msg)
|
||
|
||
|
||
def vol_bridge_loop() -> None:
|
||
global BRIDGE_OK, BRIDGE_ERR
|
||
while True:
|
||
try:
|
||
m = vol_bridge_tick()
|
||
if m:
|
||
log(m)
|
||
BRIDGE_OK = True
|
||
BRIDGE_ERR = ""
|
||
except Exception as e: # 桥不许把面板带崩
|
||
BRIDGE_OK = False
|
||
BRIDGE_ERR = str(e)
|
||
time.sleep(0.2)
|
||
|
||
|
||
def set_bridge_ref(v: float) -> None:
|
||
"""面板「输入增益」滑块改的就是参考值本身 —— 通知桥一声, 免得它把这次改动
|
||
误当成"菜单音量"再往物理输出折算一遍(桥判据是 v != ref)。"""
|
||
global BRIDGE_REF
|
||
BRIDGE_REF = v
|
||
|
||
|
||
def apply_loudness() -> str:
|
||
"""响度统一: 把两条 route 的硬件增益差补平(+0.94dB), 让数字/模拟切换时响度一致.
|
||
|
||
统一开 -> 总音量接管两条路, 并按 route 硬件基数补偿(数字路 +0.94dB);
|
||
统一关 -> 数字/模拟各走各的滑块(不补偿), 设计稿里那两条线上的滑块这时才生效.
|
||
"""
|
||
d = load_loud()
|
||
kind = route_kind()
|
||
comp = hw_base("analog") / hw_base(kind) # 数字路补 +0.94dB 追上模拟路
|
||
if d.get("unified"):
|
||
val = min(_num(d.get("target"), 0.33) * comp, 1.5)
|
||
note = "统一响度: %.3f(基数补偿 %+0.2fdB)" % (val, 20 * math.log10(comp))
|
||
else:
|
||
val = min(_num(d.get(kind), 1.0), 1.5)
|
||
note = "输出音量(%s路): %.3f" % ("数字" if kind == "digital" else "模拟", val)
|
||
nid = node_id(chain_target() or phys_sink())
|
||
if nid < 0:
|
||
return "找不到物理输出"
|
||
sh("wpctl", "set-volume", str(nid), "%.3f" % val)
|
||
return note
|
||
|
||
|
||
# ---------------- CLI 委托 ----------------
|
||
|
||
def cli(arg: str) -> str:
|
||
if not os.path.exists(CLI):
|
||
return "找不到 空间音频 脚本"
|
||
try:
|
||
out = subprocess.run(["bash", CLI, arg], capture_output=True, text=True, timeout=180, env=_env())
|
||
except subprocess.TimeoutExpired:
|
||
return "执行超时"
|
||
return (out.stdout or out.stderr).strip() or "(无输出)"
|
||
|
||
|
||
# ---------------- 状态 ----------------
|
||
|
||
def helper_sinks() -> list[dict[str, str]]:
|
||
"""可用物理输出设备(委托 音频状态.py sinks: 一处定义, CLI 与面板同一份)."""
|
||
out: list[dict[str, str]] = []
|
||
for ln in sh("python3", HELPER, "sinks").splitlines():
|
||
p = ln.split("\t")
|
||
if len(p) >= 5:
|
||
out.append({"id": p[0], "name": p[1], "desc": p[2], "current": p[3], "kind": p[4]})
|
||
return out
|
||
|
||
|
||
def state() -> dict[str, object]:
|
||
ds = default_sink_name()
|
||
v = virtual_sink()
|
||
vid = node_id(v)
|
||
vv, vm = vol_of(vid) if vid > 0 else (-1.0, False)
|
||
tgt = chain_target() or phys_sink()
|
||
pid = node_id(tgt)
|
||
pv, pm = vol_of(pid) if pid > 0 else (-1.0, False)
|
||
sinks = [s for s in node_list("Audio/Sink") if "cinema_spatial" not in str(s["name"])]
|
||
with METER_LOCK:
|
||
loglines = list(LOG)[-60:]
|
||
kind = route_kind()
|
||
return {
|
||
"title": "Collaplex 音效",
|
||
"mode": "立体声上混" if v == SINK_UP else ("5.1 直通" if v == SINK_51 else "未开"),
|
||
"default": next((s["desc"] for s in node_list("Audio/Sink") if s["name"] == ds), ds),
|
||
"virtual": {"name": v, "vol": round(vv, 3), "muted": vm},
|
||
"physical": {"name": tgt, "desc": next((s["desc"] for s in sinks if s["name"] == tgt), tgt),
|
||
"vol": round(pv, 3), "muted": pm},
|
||
"route": kind,
|
||
"route_label": "数字输出 (S/PDIF)" if kind == "digital" else "模拟输出",
|
||
"clock": clock_rate(),
|
||
"hrtf": hrtf_name(),
|
||
"ir": conf_params(), # 当前实际生效的 IR(tap/文件名) —— 与"请求参数"分开
|
||
"switches": toggle_state(),
|
||
"loud": load_loud(),
|
||
"reverb": reverb_state(),
|
||
"mono": mono_state(),
|
||
"params": params_state(),
|
||
"param_options": param_options(),
|
||
"devices": helper_sinks(),
|
||
"log": loglines,
|
||
"hw_base": HW_BASE,
|
||
}
|
||
|
||
|
||
class Handler(BaseHTTPRequestHandler):
|
||
def log_message(self, format: str, *args: object) -> None: # noqa: A002
|
||
pass
|
||
|
||
def _send(self, code: int, body: bytes, ctype: str) -> 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 _json(self, obj: object, code: int = 200) -> None:
|
||
self._send(code, json.dumps(obj, ensure_ascii=False).encode("utf-8"),
|
||
"application/json; charset=utf-8")
|
||
|
||
def do_GET(self) -> None:
|
||
u = urlparse(self.path)
|
||
if u.path in ("/", "/index.html"):
|
||
try:
|
||
with open(INDEX, "rb") as f:
|
||
self._send(200, f.read(), "text/html; charset=utf-8")
|
||
except FileNotFoundError:
|
||
self._send(404, b"index.html not found", "text/plain; charset=utf-8")
|
||
elif u.path == "/api/state":
|
||
self._json(state())
|
||
elif u.path == "/api/meters":
|
||
self._json({"raw": meter_points("raw"), "post": meter_points("post"),
|
||
"loud": loudness_state(),
|
||
"raw_target": virtual_sink(), "post_target": chain_target() or phys_sink()})
|
||
elif u.path == "/api/render":
|
||
self._json(render_status())
|
||
else:
|
||
self._json({"error": "not found"}, 404)
|
||
|
||
def do_POST(self) -> None:
|
||
u = urlparse(self.path)
|
||
q = parse_qs(u.query)
|
||
# ★ 上传是**二进制** body, 必须在下面对 body 做 JSON 解析之前分出去, 否则文件会坏
|
||
if u.path == "/api/render/upload":
|
||
self._upload(q)
|
||
return
|
||
try:
|
||
n = int(self.headers.get("Content-Length") or 0)
|
||
raw = self.rfile.read(n).decode("utf-8") if n else ""
|
||
if raw:
|
||
for k, v in json.loads(raw).items():
|
||
q[k] = [str(v)]
|
||
except Exception:
|
||
pass
|
||
|
||
def g(k: str, d: str = "") -> str:
|
||
return (q.get(k) or [d])[0]
|
||
|
||
try:
|
||
if u.path == "/api/mode":
|
||
msg = cli({"off": "关", "on": "开", "on51": "开5.1"}.get(g("mode", "on"), "状态"))
|
||
elif u.path == "/api/volume": # 虚拟声卡(链输入增益)
|
||
val = max(0.0, min(1.5, float(g("value", "0.33"))))
|
||
# ★★ 只写**虚拟声卡**(名字以 _sink 结尾的 Audio/Sink)。
|
||
# 绝不能按"名字里含 cinema_spatial"一把全写 —— 那会把上混图的
|
||
# 输入口 cinema_spatial_up_raw(以及归一化输出 up_norm)一起改掉,
|
||
# 而它们**不归用户调**、必须在单位增益。2026-09-13 实测血亏:
|
||
# up_raw 被这个接口写成 0.2068(-13.7 dB) → 全链由 -8.1 dB 掉到
|
||
# -29.9 dB, 耳机听着"一点声音都没有", 而界面、音量账**看着全正常**
|
||
# (旧账只算首尾两级), 排查绕了好几圈。
|
||
for s in node_list("Audio/Sink"):
|
||
if str(s["name"]).endswith("_sink"):
|
||
sh("wpctl", "set-volume", str(s["id"]), "%.3f" % val)
|
||
set_bridge_ref(val) # 这是面板改的 = 参考值本身, 别让桥再折算一次
|
||
try:
|
||
os.makedirs(STATE_DIR, exist_ok=True)
|
||
with open(VOL_FILE, "w", encoding="utf-8") as f:
|
||
f.write("%.3f\n" % val)
|
||
except OSError:
|
||
pass
|
||
msg = "虚拟声卡音量 %.3f" % val
|
||
elif u.path == "/api/mono": # 单声道: 委托 CLI(写状态 → 重建 → 验连线)
|
||
mode = g("mode", "自动").strip()
|
||
if mode not in ("自动", "开", "关"):
|
||
msg = "单声道: 只接受 自动/开/关"
|
||
else:
|
||
msg = (sh("bash", CLI, "单声道", mode, timeout=240) or "").strip() or "(无输出)"
|
||
elif u.path == "/api/speaker": # 扬声器电平 = 物理输出音量(老板听到的音量)
|
||
try:
|
||
want = max(0.0, min(1.5, float(g("value", "1.0"))))
|
||
except ValueError:
|
||
want = 1.0
|
||
msg = set_speaker(want) # 闭环自校正封装在函数里(菜单音量桥也用同一个)
|
||
elif u.path == "/api/route":
|
||
msg = cli("数字" if g("rt", "digital") == "digital" else "模拟")
|
||
elif u.path == "/api/switch":
|
||
msg = toggle(g("which", "raw"), g("on", "1") == "1")
|
||
elif u.path == "/api/loudness":
|
||
d = load_loud()
|
||
if "unified" in q:
|
||
d["unified"] = g("unified") == "1"
|
||
if d["unified"] and "value" not in q:
|
||
# 打开时以当前这条路的音量作目标, 避免音量突跳
|
||
d["target"] = min(1.0, _num(d.get(route_kind()), 0.33))
|
||
# ★ 这里**故意不再接受 `value`**（2026-09-13 修）：它和 /api/speaker
|
||
# (「扬声器电平」)写的是同一个 d["target"]，两个入口各按自己的量纲换算
|
||
# → 拖一个、另一个"跳到别的值"（现场复现：拖扬声器到 0.60，target 被
|
||
# 写成 0.5354，面板那根「统一响度音量」横滑块立刻显示 0.5354）。
|
||
# 物理输出音量现在只有 /api/speaker 一个入口；面板那根横滑块也已改走它。
|
||
if "digital" in q:
|
||
d["digital"] = max(0.0, min(1.5, float(g("digital"))))
|
||
if "analog" in q:
|
||
d["analog"] = max(0.0, min(1.5, float(g("analog"))))
|
||
# 响度归一化的目标电平(面板那根「目标响度」滑块): 只写 json 就够 ——
|
||
# DSP 每秒自己热读, 不用重建链路、也不碰物理音量
|
||
only_tgt = "target_db" in q and len(q) == 1
|
||
if "target_db" in q:
|
||
try:
|
||
d["target_db"] = max(-40.0, min(-6.0, float(g("target_db"))))
|
||
except ValueError:
|
||
pass
|
||
save_loud(d)
|
||
msg = (f"归一化目标 → {_num(d.get('target_db'), -16.0):.1f} dBFS (立即生效)"
|
||
if only_tgt else apply_loudness())
|
||
elif u.path == "/api/render": # 启动离线渲染(后台线程, 不阻塞面板)
|
||
files = [f.strip() for f in g("files", "").split("\n") if f.strip()]
|
||
try:
|
||
wet = max(0.0, min(1.0, float(g("wet", "0.3"))))
|
||
tgt = max(-40.0, min(-6.0, float(g("target", "-14.0"))))
|
||
except ValueError:
|
||
wet, tgt = 0.3, -14.0
|
||
msg = render_start(files, wet, tgt, g("overwrite", "") == "1")
|
||
elif u.path == "/api/render/cancel":
|
||
msg = render_cancel()
|
||
elif u.path == "/api/fix":
|
||
msg = cli("开")
|
||
elif u.path == "/api/device":
|
||
# 设备选择: 委托 CLI「设备」= 重写 node.target → 重建 → 校验 → 实装(约 10s)
|
||
# ★ 参数必须分开传: 设备关键词里可能有空格(如 "Ryzen HD Audio Controller 模拟立体声")
|
||
want = g("name", "").strip()
|
||
if not want:
|
||
msg = "设备选择: 缺 name 参数"
|
||
else:
|
||
msg = (sh("bash", CLI, "设备", want, timeout=200) or "").strip() or "(无输出)"
|
||
elif u.path == "/api/params":
|
||
# 采样参数: 委托 CLI「参数」= 写参数 → (裁 IR) → 重生成配置 → 重启(约 10s)
|
||
key = g("key", "").strip()
|
||
val = g("value", "").strip()
|
||
if key not in ("hrtf", "taps", "rate") or not val:
|
||
msg = "采样参数: 需要 key(hrtf/taps/rate) 与 value"
|
||
else:
|
||
msg = (sh("bash", CLI, "参数", key, val, timeout=300) or "").strip() or "(无输出)"
|
||
elif u.path == "/api/reverb":
|
||
# 混响: 委托 CLI「混响」= 写状态 → 重生成配置 → 重启链路 → 验连线(约 10s)
|
||
on = g("on", "")
|
||
wet = g("wet", "0.25")
|
||
if on not in ("0", "1"):
|
||
msg = "混响: 缺 on 参数(0/1)"
|
||
else:
|
||
out = sh("bash", CLI, "混响", "开" if on == "1" else "关", wet, timeout=240)
|
||
msg = (out or "").strip() or "(无输出)"
|
||
else:
|
||
self._json({"error": "not found"}, 404)
|
||
return
|
||
except Exception as e: # 任何异常都要回给界面
|
||
msg = "执行出错: %s" % e
|
||
log(msg)
|
||
self._json({"message": msg, "state": state()})
|
||
|
||
def _upload(self, q: dict[str, list[str]]) -> None:
|
||
"""接收拖拽进来的文件(localhost 传输很快, 几 MB~几百 MB 都是瞬时的)。
|
||
|
||
上传到 STATE_DIR/uploads/ 再交给渲染 —— 因为浏览器出于安全**不给本地路径**
|
||
(那是 Electron 的扩展), 所以拖进来的文件只能这么进来。
|
||
超大文件(>2GB)建议用路径输入框直接贴路径, 免得白等一次全量传输。
|
||
"""
|
||
try:
|
||
raw_name = (q.get("name") or ["upload.bin"])[0]
|
||
name = re.sub(r"[/\\]", "_", os.path.basename(raw_name)).strip() or "upload.bin"
|
||
n = int(self.headers.get("Content-Length") or 0)
|
||
if n <= 0:
|
||
self._json({"error": "空文件"}, 400)
|
||
return
|
||
if n > 2 * 1024 ** 3:
|
||
self._json({"error": "文件超过 2GB, 请改用路径输入框"}, 413)
|
||
return
|
||
os.makedirs(UPLOAD_DIR, exist_ok=True)
|
||
dst = os.path.join(UPLOAD_DIR, name)
|
||
got = 0
|
||
with open(dst, "wb") as fh:
|
||
left = n
|
||
while left > 0:
|
||
chunk = self.rfile.read(min(1 << 20, left))
|
||
if not chunk:
|
||
break
|
||
fh.write(chunk)
|
||
got += len(chunk)
|
||
left -= len(chunk)
|
||
log("已接收拖入文件: %s (%.1f MB)" % (name, got / 1048576.0))
|
||
self._json({"message": "已接收 %s" % name, "path": dst, "size": got})
|
||
except Exception as e:
|
||
self._json({"error": "上传失败: %s" % e}, 500)
|
||
|
||
|
||
def main() -> int:
|
||
ap = argparse.ArgumentParser(description="Collaplex 音效 Web 控制台")
|
||
ap.add_argument("--port", type=int, default=8788)
|
||
ap.add_argument("--bind", default="127.0.0.1")
|
||
a = ap.parse_args()
|
||
os.makedirs(STATE_DIR, exist_ok=True)
|
||
start_meters()
|
||
threading.Thread(target=vol_bridge_loop, daemon=True).start() # 菜单音量 → 扬声器电平
|
||
log("控制台启动, 电平采集: raw=%s post=%s(均为显式连线, 不经麦克风)"
|
||
% (virtual_sink(), _meter_src("post")[0]))
|
||
log("菜单音量桥已启动: 菜单/媒体键 → 扬声器电平(物理输出), 链输入增益复位到削波余量")
|
||
srv = ThreadingHTTPServer((a.bind, a.port), Handler)
|
||
print("Collaplex 音效 控制台: http://%s:%d/ (Ctrl-C 退出)" % (a.bind, a.port))
|
||
sys.stdout.flush()
|
||
try:
|
||
srv.serve_forever()
|
||
except KeyboardInterrupt:
|
||
print()
|
||
return 0
|
||
|
||
|
||
if __name__ == "__main__":
|
||
sys.exit(main())
|