edgevoid
322 lines
10 KiB
Python
322 lines
10 KiB
Python
#!/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 array
|
|
import json
|
|
import math
|
|
import os
|
|
import re
|
|
import subprocess
|
|
import sys
|
|
import threading
|
|
import time
|
|
from collections import deque
|
|
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"
|
|
|
|
# 电平表: 从物理输出的 monitor 取样(耳机真正在放的信号), 每 100ms 一个点。
|
|
# 指标是 dBFS —— 0dBFS 为数字满刻度, 余量 = -peak(dB)。
|
|
# 注意: 96k 重采样/多路叠加会产生 intersample peak, 峰值超过 0 也算削波。
|
|
LEVEL: deque[dict] = deque(maxlen=240)
|
|
LEVEL_LOCK = threading.Lock()
|
|
CHUNK_MS = 100
|
|
|
|
|
|
def monitor_name() -> str:
|
|
"""当前物理输出设备的 monitor 源(录到的就是耳机在放的东西)。"""
|
|
p = phys_sink()
|
|
return p + ".monitor" if p else ""
|
|
|
|
|
|
def _level_worker() -> None:
|
|
"""持续录物理输出, 按 CHUNK_MS 切块算 RMS/Peak(dBFS)。"""
|
|
rate = 48000
|
|
per = rate * CHUNK_MS // 1000 * 4 # f32 每样本 4 字节
|
|
while True:
|
|
mon = monitor_name()
|
|
if not mon:
|
|
time.sleep(2)
|
|
continue
|
|
proc = None
|
|
try:
|
|
proc = subprocess.Popen(
|
|
["pw-record", "--target", mon, "--rate", str(rate),
|
|
"--channels", "1", "--format", "f32", "-"],
|
|
stdout=subprocess.PIPE, stderr=subprocess.DEVNULL, env=_env())
|
|
while True:
|
|
raw = proc.stdout.read(per) if proc.stdout else b""
|
|
if not raw or len(raw) < per:
|
|
break
|
|
a = array.array("f")
|
|
a.frombytes(raw)
|
|
pk = max(abs(v) for v in a)
|
|
rms = math.sqrt(sum(v * v for v in a) / len(a))
|
|
with LEVEL_LOCK:
|
|
LEVEL.append({
|
|
"peak": round(20 * math.log10(pk), 2) if pk > 1e-9 else -120.0,
|
|
"rms": round(20 * math.log10(rms), 2) if rms > 1e-9 else -120.0,
|
|
})
|
|
except Exception:
|
|
pass
|
|
finally:
|
|
if proc is not None:
|
|
try:
|
|
proc.kill()
|
|
except Exception:
|
|
pass
|
|
time.sleep(0.5)
|
|
|
|
|
|
def start_level_worker() -> None:
|
|
threading.Thread(target=_level_worker, daemon=True).start()
|
|
|
|
|
|
def level_points() -> list[dict]:
|
|
with LEVEL_LOCK:
|
|
return list(LEVEL)
|
|
|
|
|
|
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": next((n["desc"] for n in nodes("Audio/Sink") if n["name"] == ds), ds),
|
|
"default_name": 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())
|
|
elif u.path == "/api/level":
|
|
self._json({"points": level_points(), "monitor": monitor_name()})
|
|
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()
|
|
start_level_worker()
|
|
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())
|