feat: Web 控制台(web/) -- 状态总览/模式切换/音量/一键修复失声, 零依赖 stdlib
This commit is contained in:
edgevoid
@@ -66,6 +66,20 @@ mpv-影院 电影.mkv # 只给 mpv 用, 不依赖全局设置(适合 A/B 对比
|
||||
|
||||
也可以直接在桌面环境的"声音设置"里选 **电影院空间音频**。
|
||||
|
||||
## Web 控制台
|
||||
|
||||
浏览器里看状态、切模式、调音量、**一键修复"失声"**（设备名失效时自动重建）。
|
||||
|
||||
```sh
|
||||
bash web/启动.sh # 启动（默认端口 8788，重复执行不会起第二个）
|
||||
python3 web/webui.py --port 8788 --bind 127.0.0.1 # 也可以直接跑
|
||||
```
|
||||
|
||||
打开 <http://127.0.0.1:8788/>。菜单里也有「空间音频控制台」快捷方式。
|
||||
|
||||
界面会实时显示：默认输出、链路采样率、HRTF 模型、**配置输出设备 vs 当前物理输出（不一致会标红）**、
|
||||
输出端实际连到哪个设备、以及两个虚拟声卡和设备的音量。检测到设备名失效时直接点修复即可。
|
||||
|
||||
## 文件说明
|
||||
|
||||
| 文件 | 作用 |
|
||||
@@ -77,6 +91,7 @@ mpv-影院 电影.mkv # 只给 mpv 用, 不依赖全局设置(适合 A/B 对比
|
||||
| `提取HRIR.sh` | 早期版本（已被 `换HRTF.sh` 取代）|
|
||||
| `打包deb.sh` | 打包 `collaplex-cinema-spatial` |
|
||||
| `验收测试.sh` | 安装后自动验证 |
|
||||
| `web/` | Web 控制台（`webui.py` 后端 + `index.html` 前端 + `启动.sh`），零依赖 |
|
||||
| `demo.sh` / `响度校准.sh` / `方向测试.sh` | 开发期验证工具 |
|
||||
| `对比分析.py` / `排查.sh` | 开发期排查工具 |
|
||||
|
||||
|
||||
Binary file not shown.
+138
@@ -0,0 +1,138 @@
|
||||
<!DOCTYPE html>
|
||||
<html lang="zh-CN">
|
||||
<head>
|
||||
<meta charset="utf-8">
|
||||
<meta name="viewport" content="width=device-width,initial-scale=1">
|
||||
<title>电影院空间音频</title>
|
||||
<style>
|
||||
:root{--fg:#e6e6e6;--mut:#8a8a8a;--bg:#0f0f0f;--card:#161616;--bd:#262626;
|
||||
--ok:#4ade80;--bad:#f87171;--warn:#fbbf24}
|
||||
*{box-sizing:border-box}
|
||||
body{margin:0;padding:20px;background:var(--bg);color:var(--fg);
|
||||
font:14px/1.6 system-ui,-apple-system,"Noto Sans CJK SC",sans-serif}
|
||||
h1{font-size:16px;font-weight:600;margin:0 0 4px}
|
||||
.sub{color:var(--mut);font-size:12px;margin-bottom:16px}
|
||||
.card{background:var(--card);border:1px solid var(--bd);border-radius:8px;
|
||||
padding:14px 16px;margin-bottom:12px}
|
||||
.card h2{font-size:13px;font-weight:600;color:var(--mut);margin:0 0 10px;letter-spacing:.05em}
|
||||
.row{display:flex;justify-content:space-between;gap:12px;padding:6px 0;
|
||||
border-bottom:1px solid var(--bd);font-size:13px}
|
||||
.row:last-child{border-bottom:none}
|
||||
.k{color:var(--mut);flex:0 0 auto}
|
||||
.v{font-variant-numeric:tabular-nums;text-align:right;word-break:break-all}
|
||||
.ok{color:var(--ok)}.bad{color:var(--bad)}.warn{color:var(--warn)}
|
||||
.bar{display:flex;gap:8px;flex-wrap:wrap}
|
||||
button{background:#1f1f1f;color:var(--fg);border:1px solid var(--bd);border-radius:6px;
|
||||
padding:8px 14px;font-size:13px;cursor:pointer;font-family:inherit;
|
||||
transition:background .15s,border-color .15s}
|
||||
button:hover{background:#2a2a2a;border-color:#3a3a3a}
|
||||
button.primary{background:#17321f;border-color:#2d5a3f}
|
||||
button.primary:hover{background:#1e4028}
|
||||
button.fix{background:#3a2418;border-color:#5c3a26}
|
||||
button.fix:hover{background:#4a2e1e}
|
||||
#msg{margin-top:12px;font-size:13px;min-height:20px;color:var(--warn)}
|
||||
.nums{display:flex;align-items:center;gap:10px;margin-top:6px}
|
||||
#volval{font-variant-numeric:tabular-nums;min-width:52px;text-align:right;color:var(--ok)}
|
||||
input[type=range]{flex:1;accent-color:#4ade80;height:4px}
|
||||
</style>
|
||||
</head>
|
||||
<body>
|
||||
<h1>电影院空间音频</h1>
|
||||
<div class="sub">PipeWire 虚拟声卡 · SADIE-II 真人 HRTF</div>
|
||||
|
||||
<div class="card"><h2>状态</h2><div id="st"></div></div>
|
||||
|
||||
<div class="card"><h2>模式</h2>
|
||||
<div class="bar">
|
||||
<button class="primary" onclick="mode('on')">立体声上混</button>
|
||||
<button onclick="mode('on51')">5.1 直通</button>
|
||||
<button onclick="mode('off')">关闭（走物理设备）</button>
|
||||
<button class="fix" onclick="doFix()">检测并修复失声</button>
|
||||
</div>
|
||||
</div>
|
||||
|
||||
<div class="card"><h2>虚拟声卡音量</h2>
|
||||
<div class="nums">
|
||||
<input type="range" id="vol" min="0" max="1.5" step="0.01" oninput="onVol(this.value)">
|
||||
<span id="volval">-</span>
|
||||
</div>
|
||||
<div class="sub" style="margin:8px 0 0">降音量是拿响度换安全；削波爆音就往下拉，
|
||||
想更响请改 IR 增益（见 README）。</div>
|
||||
</div>
|
||||
|
||||
<div id="msg"></div>
|
||||
|
||||
<script>
|
||||
const $ = (s) => document.querySelector(s);
|
||||
const esc = (s) => String(s == null ? '' : s).replace(/[&<>"]/g, c =>
|
||||
({'&':'&','<':'<','>':'>','"':'"'}[c]));
|
||||
|
||||
async function api(path, opts) {
|
||||
try { const r = await fetch(path, opts || {}); return await r.json(); }
|
||||
catch (e) { return { error: String(e) }; }
|
||||
}
|
||||
|
||||
function row(k, v, cls) {
|
||||
return '<div class="row"><span class="k">' + esc(k) + '</span>' +
|
||||
'<span class="v ' + (cls || '') + '">' + esc(v) + '</span></div>';
|
||||
}
|
||||
|
||||
function render(s) {
|
||||
if (!s || s.error) {
|
||||
$('#st').innerHTML = row('错误', (s && s.error) || '取不到状态', 'bad');
|
||||
return;
|
||||
}
|
||||
let h = '';
|
||||
h += row('默认输出', s.default || '（无）', s.default_is_virtual ? 'ok' : 'warn');
|
||||
h += row('链路采样率', (s.clock || '?') + ' Hz', s.clock === '96000' ? 'ok' : 'warn');
|
||||
h += row('HRTF 模型', s.hrtf || '-');
|
||||
h += row('配置输出设备', s.target || '（空）', s.target_ok ? 'ok' : 'bad');
|
||||
h += row('当前物理输出', s.physical || '（找不到）', s.physical ? 'ok' : 'bad');
|
||||
h += row('实际连接', (s.route && s.route.length) ? s.route.join(', ') : '（未连接 → 无声）',
|
||||
(s.route && s.route.length) ? 'ok' : 'bad');
|
||||
if (s.device_vol != null) h += row('设备音量', s.device_vol.toFixed(2));
|
||||
if (s.device_desc) h += row('设备', s.device_desc);
|
||||
(s.sinks || []).forEach(function (k) {
|
||||
h += row('声卡 ' + k.key + ' (id ' + k.id + ')', '音量 ' + k.vol.toFixed(2));
|
||||
});
|
||||
$('#st').innerHTML = h;
|
||||
if (s.default_vol != null) {
|
||||
$('#vol').value = s.default_vol;
|
||||
$('#volval').textContent = s.default_vol.toFixed(2);
|
||||
}
|
||||
if (!s.target_ok) msg('配置里的输出设备已失效 —— 点「检测并修复失声」', true);
|
||||
}
|
||||
|
||||
function msg(t, keep) {
|
||||
$('#msg').textContent = t;
|
||||
if (!keep) setTimeout(function () {
|
||||
if ($('#msg').textContent === t) $('#msg').textContent = '';
|
||||
}, 4000);
|
||||
}
|
||||
|
||||
async function mode(m) {
|
||||
const r = await api('/api/mode?mode=' + m, { method: 'POST' });
|
||||
msg(r.message || '完成');
|
||||
if (r.status) render(r.status);
|
||||
}
|
||||
|
||||
async function onVol(v) {
|
||||
$('#volval').textContent = parseFloat(v).toFixed(2);
|
||||
const r = await api('/api/volume?value=' + v, { method: 'POST' });
|
||||
if (r.message) msg(r.message);
|
||||
}
|
||||
|
||||
async function doFix() {
|
||||
msg('正在检测并修复…（约 6 秒）', true);
|
||||
const r = await api('/api/fix', { method: 'POST' });
|
||||
msg(r.message || '完成');
|
||||
setTimeout(refresh, 5000);
|
||||
}
|
||||
|
||||
async function refresh() { render(await api('/api/status')); }
|
||||
|
||||
refresh();
|
||||
setInterval(refresh, 5000);
|
||||
</script>
|
||||
</body>
|
||||
</html>
|
||||
+251
@@ -0,0 +1,251 @@
|
||||
#!/usr/bin/env python3
|
||||
# 电影院空间音频 Web 控制台 (2026-09-13) -- 零依赖, 仅 stdlib
|
||||
# 用法: python3 webui.py [--port 8788] [--bind 127.0.0.1]
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import json
|
||||
import os
|
||||
import re
|
||||
import subprocess
|
||||
import sys
|
||||
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
|
||||
from urllib.parse import 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"
|
||||
|
||||
|
||||
def _env() -> dict[str, str]:
|
||||
e = dict(os.environ)
|
||||
e.setdefault("XDG_RUNTIME_DIR", "/run/user/%d" % os.getuid())
|
||||
return e
|
||||
|
||||
|
||||
def sh(*args: str, timeout: int = 10) -> 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 nodes(kind: str) -> list[dict]:
|
||||
out: list[dict] = []
|
||||
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 vol(nid: object) -> float:
|
||||
m = re.search(r"Volume:\s*([0-9.]+)", sh("wpctl", "get-volume", str(nid)))
|
||||
return float(m.group(1)) if m else -1.0
|
||||
|
||||
|
||||
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 conf_target() -> str:
|
||||
try:
|
||||
with open(CONF, encoding="utf-8") as f:
|
||||
m = re.search(r'node\.target = "([^"]+)"', f.read())
|
||||
return m.group(1) if m else ""
|
||||
except Exception:
|
||||
return ""
|
||||
|
||||
|
||||
def phys_sink() -> str:
|
||||
c = [s["name"] for s in nodes("Audio/Sink")
|
||||
if s["name"] and "cinema_spatial" not in s["name"] and "pro-output" not in s["name"]]
|
||||
return (sorted(c, key=lambda x: (0 if "usb" in x else 1, x)) or [""])[0]
|
||||
|
||||
|
||||
def route_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 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 status() -> dict:
|
||||
ds = default_sink_name()
|
||||
ss: list[dict] = []
|
||||
for n in nodes("Audio/Sink"):
|
||||
if "cinema_spatial" in n["name"]:
|
||||
key = "5.1 直通" if n["name"] == SINK_51 else "立体声上混"
|
||||
ss.append({"key": key, "id": n["id"], "vol": round(vol(n["id"]), 3)})
|
||||
t, p = conf_target(), phys_sink()
|
||||
devs = [n for n in nodes("Audio/Sink") if "cinema_spatial" not in n["name"]]
|
||||
dv = next((round(vol(d["id"]), 3) for d in devs if d["name"] == t), None)
|
||||
return {
|
||||
"default": ds,
|
||||
"default_is_virtual": "cinema_spatial" in ds,
|
||||
"sinks": ss,
|
||||
"clock": clock_rate(),
|
||||
"hrtf": hrtf_name(),
|
||||
"target": t,
|
||||
"target_ok": bool(t) and t == p,
|
||||
"physical": p,
|
||||
"route": route_of("cinema_spatial_up_out"),
|
||||
"default_vol": next((round(vol(n["id"]), 3) for n in nodes("Audio/Sink")
|
||||
if n["name"] == ds), None),
|
||||
"device_vol": dv,
|
||||
"device_desc": next((d["desc"] for d in devs if d["name"] == t), ""),
|
||||
}
|
||||
|
||||
|
||||
def set_volume(val: float) -> str:
|
||||
for n in nodes("Audio/Sink"):
|
||||
if "cinema_spatial" in n["name"]:
|
||||
sh("wpctl", "set-volume", str(n["id"]), "%.3f" % val)
|
||||
return "已设为 %.2f" % val
|
||||
|
||||
|
||||
def set_mode(mode: str) -> str:
|
||||
if mode == "off":
|
||||
p = phys_sink()
|
||||
nid = next((n["id"] for n in nodes("Audio/Sink") if n["name"] == p), None)
|
||||
if nid:
|
||||
sh("wpctl", "set-default", str(nid))
|
||||
return "已切到物理输出"
|
||||
return "找不到物理输出"
|
||||
want = SINK_UP if mode == "on" else SINK_51
|
||||
n = next((x for x in nodes("Audio/Sink") if x["name"] == want), None)
|
||||
if not n:
|
||||
return "虚拟声卡不存在"
|
||||
sh("wpctl", "set-default", str(n["id"]))
|
||||
return "已切到 %s" % ("立体声上混" if mode == "on" else "5.1 直通")
|
||||
|
||||
|
||||
def fix() -> str:
|
||||
gen = os.path.join(ROOT, "生成配置.py")
|
||||
if not os.path.exists(gen):
|
||||
return "找不到 生成配置.py"
|
||||
t, p = conf_target(), phys_sink()
|
||||
if t and t == p:
|
||||
return "配置已正确, 无需修复"
|
||||
if not p:
|
||||
return "找不到物理输出设备"
|
||||
e = _env()
|
||||
e["CINEMA_SPATIAL_IR"] = os.path.join(ROOT, "hrir", "current")
|
||||
subprocess.run([sys.executable, gen], capture_output=True, text=True, timeout=30, env=e)
|
||||
sh("systemctl", "--user", "restart", "pipewire", "pipewire-pulse", "wireplumber", timeout=20)
|
||||
return "已重新探测设备并重建配置"
|
||||
|
||||
|
||||
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/status":
|
||||
self._json(status())
|
||||
else:
|
||||
self._json({"error": "not found"}, 404)
|
||||
|
||||
def do_POST(self) -> None:
|
||||
u = urlparse(self.path)
|
||||
q = parse_qs(u.query)
|
||||
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]
|
||||
|
||||
if u.path == "/api/mode":
|
||||
msg = set_mode(g("mode", "on"))
|
||||
elif u.path == "/api/volume":
|
||||
try:
|
||||
msg = set_volume(max(0.0, min(1.5, float(g("value", "1.0")))))
|
||||
except ValueError:
|
||||
msg = "音量参数无效"
|
||||
elif u.path == "/api/fix":
|
||||
msg = fix()
|
||||
else:
|
||||
self._json({"error": "not found"}, 404)
|
||||
return
|
||||
self._json({"message": msg, "status": status()})
|
||||
|
||||
|
||||
def main() -> int:
|
||||
ap = argparse.ArgumentParser(description="电影院空间音频 Web 控制台")
|
||||
ap.add_argument("--port", type=int, default=8788)
|
||||
ap.add_argument("--bind", default="127.0.0.1")
|
||||
a = ap.parse_args()
|
||||
srv = ThreadingHTTPServer((a.bind, a.port), Handler)
|
||||
print("电影院空间音频控制台: 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())
|
||||
@@ -0,0 +1,18 @@
|
||||
#!/bin/bash
|
||||
# 启动空间音频 Web 控制台(后台运行, 重复执行不会起第二个)
|
||||
PORT="${1:-8788}"
|
||||
P=$(ss -ltnp 2>/dev/null | grep ":$PORT" | grep -oP 'pid=\K[0-9]+' | head -1)
|
||||
if [ -n "$P" ]; then
|
||||
echo "已在运行 (PID $P): http://127.0.0.1:$PORT/"
|
||||
exit 0
|
||||
fi
|
||||
cd "$(dirname "$(readlink -f "$0")")" || exit 1
|
||||
nohup python3 webui.py --port "$PORT" >/tmp/cinema-webui.log 2>&1 &
|
||||
sleep 1
|
||||
if ss -ltn 2>/dev/null | grep -q ":$PORT"; then
|
||||
echo "已启动: http://127.0.0.1:$PORT/"
|
||||
else
|
||||
echo "启动失败, 见 /tmp/cinema-webui.log"
|
||||
cat /tmp/cinema-webui.log
|
||||
exit 1
|
||||
fi
|
||||
Reference in New Issue
Block a user