edgevoid
d8c3a27821
- 离线渲染.py: pan上混5.1 → sofalizer(H4 SOFA) → 房间IR(afir) → loudnorm → 96k FLAC
视频自动取音轨; 已是5.1的片源不重上混; 实测约 4.4 倍速
两道锁: (1)渲染锁 —— 一轮任务期间拒绝新任务, 不排队堆积
(2)文件就绪检查 —— 相隔1.2s两次stat大小/时间不变且ffprobe能读, 未落盘的跳过
三个入口: 命令行 / 面板「离线渲染」区(拖拽或贴路径, 实时进度+日志+取消) / 右键「打开方式」
- 音频守护.py + config/组件配置单.json: 30s 核对 DSP进程数与槽位新鲜度/链输出连线/
默认设备/虚拟声卡优先级/中间级音量/节点份数/面板/物理设备静音, 扫到就修
(能温和就温和, 链路级才重建), 同类问题 90s 冷却, 连续 3 次修不好转只报警
- 组件配置单.md: 链路拓扑 + 组件表 + 六个必须 + 自愈表
- 面板: 控制滑块从图表卡片独立成「控制」卡片; 数字/模拟输出音量条拉长到各占一半
(合计 = 虚拟声卡开关卡片总高); 新增离线渲染区
- 修: DSP 孤儿抢共享槽位导致重建后无声(claim_slot 抢不到槽位即退出 + 重建前收旧 DSP)
- 修: 打包会把正在用的 PipeWire 配置移开(prune_duplicate_confs 只在自己写到生效目录时清理)
- 修: 离线渲染的 SOFA/混响IR 改为多路径探测(开发目录与 deb 包内布局不同)
- .gitignore: 保留 reverb/ 房间IR(离线渲染要卷积的素材)
431 lines
17 KiB
Python
431 lines
17 KiB
Python
#!/usr/bin/env python3
|
||
"""空间音频状态 / 设备 profile 助手.
|
||
|
||
给「空间音频」切换脚本与 Web 控制台共用. 零依赖(纯 stdlib + pw-dump/wpctl).
|
||
|
||
子命令:
|
||
json 状态 JSON(供程序读)
|
||
check 人读报告(含"音量账")
|
||
phys-sink 当前最合适的物理输出 node.name(USB 优先)
|
||
card-of <sink_node_name> 由 sink 反查所属卡名
|
||
profile <card_name> 打印该卡当前 profile 名 + 索引
|
||
set-profile <card_name> <profile> 把卡切回指定 profile(名字或索引)
|
||
route-kind 当前物理输出走的是 digital(数字/S-PDIF) 还是 analog
|
||
set-route <card_name> digital|analog 按 route 类型切 profile(保住麦克风输入)
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
import math
|
||
import re
|
||
import subprocess
|
||
import sys
|
||
from pathlib import Path
|
||
from typing import Any
|
||
|
||
DEFAULT_CONF = Path.home() / ".config" / "pipewire" / "pipewire.conf.d" / "90-cinema-spatial.conf"
|
||
VOLUME_NODES = ("cinema_spatial_up_sink", "cinema_spatial_up_norm", "cinema_spatial_up_raw",
|
||
"cinema_spatial_up_out", "cinema_spatial_sink", "cinema_spatial_out")
|
||
# ★ 全链节点的顺序(上混那条), 从应用进到物理设备出 —— 每一级都要进音量账。
|
||
CHAIN_UP_NODES = ("cinema_spatial_up_sink", "cinema_spatial_up_norm",
|
||
"cinema_spatial_up_raw", "cinema_spatial_up_out")
|
||
# 中间两级**不归用户调**, 必须是单位增益; 偏离 1.0 就是 bug(见 state() 里的注释)。
|
||
CHAIN_MID_NODES = ("cinema_spatial_up_norm", "cinema_spatial_up_raw")
|
||
# ★ 硬件增益基数(volumeBase)必须按**活动 Route** 读, 别读 EnumRoute:
|
||
# EnumRoute 是能力清单(不带 volumeBase), Route 才是当前/存档状态(带 volumeBase+softVolumes).
|
||
# 实测本机(EDIFIER Fit900NB): 模拟 route volumeBase=1.11428(+0.94 dB),
|
||
# 数字(iec958) route volumeBase=1.000(0 dB) —— 两条都**有**音量控制,
|
||
# 差别只是模拟那条多约 +0.94 dB 硬件增益(不是"数字没有硬件音量", 2026-09-13 修正).
|
||
DIGITAL_HINTS = ("iec958", "spdif")
|
||
|
||
|
||
def run(cmd: list[str]) -> str:
|
||
"""跑命令取 stdout(失败返回空串)."""
|
||
try:
|
||
return subprocess.run(cmd, capture_output=True, text=True, timeout=15).stdout
|
||
except Exception:
|
||
return ""
|
||
|
||
|
||
def pw_dump() -> list[dict[str, Any]]:
|
||
"""pw-dump 的全部对象."""
|
||
out = run(["pw-dump"])
|
||
try:
|
||
return json.loads(out)
|
||
except Exception:
|
||
return []
|
||
|
||
|
||
def nodes(objs: list[dict[str, Any]]) -> dict[str, dict[str, Any]]:
|
||
"""node.name -> 对象."""
|
||
out: dict[str, dict[str, Any]] = {}
|
||
for o in objs:
|
||
if o.get("type") != "PipeWire:Interface:Node":
|
||
continue
|
||
p = o.get("info", {}).get("props", {})
|
||
name = p.get("node.name")
|
||
if name:
|
||
out[name] = o
|
||
return out
|
||
|
||
|
||
def vol_linear(node_id: int) -> float:
|
||
"""wpctl 报的是线性音量(用户看到的那个数)."""
|
||
m = re.search(r"Volume:\s*([0-9.]+)", run(["wpctl", "get-volume", str(node_id)]))
|
||
return float(m.group(1)) if m else 0.0
|
||
|
||
|
||
def db(linear: float) -> float:
|
||
"""线性音量 -> dB."""
|
||
return 20.0 * math.log10(linear) if linear > 0 else -99.0
|
||
|
||
|
||
def conf_target(conf: Path = DEFAULT_CONF) -> str:
|
||
"""读配置里的 node.target."""
|
||
try:
|
||
txt = conf.read_text(encoding="utf-8")
|
||
except OSError:
|
||
return ""
|
||
m = re.search(r'node\.target = "([^"]+)"', txt)
|
||
return m.group(1) if m else ""
|
||
|
||
|
||
def profile_of_card(objs: list[dict[str, Any]], card: str) -> dict[str, Any]:
|
||
"""卡的当前 profile 与硬件音量情况."""
|
||
for o in objs:
|
||
if o.get("type") != "PipeWire:Interface:Device":
|
||
continue
|
||
p = o.get("info", {}).get("props", {})
|
||
if p.get("device.name") != card:
|
||
continue
|
||
prms = o.get("info", {}).get("params", {})
|
||
prof = (prms.get("Profile") or [{}])[0]
|
||
routes: list[dict[str, Any]] = []
|
||
for r in (prms.get("Route") or []):
|
||
pr = r.get("props") or {}
|
||
if not pr:
|
||
continue
|
||
routes.append({"kind": "digital" if "iec958Codecs" in pr else "analog",
|
||
"base": pr.get("volumeBase")})
|
||
pname = str(prof.get("name", ""))
|
||
dig = any(h in pname for h in DIGITAL_HINTS)
|
||
base = next((r["base"] for r in routes if (r["kind"] == "digital") == dig), None)
|
||
return {"name": pname, "index": prof.get("index", -1),
|
||
"description": prof.get("description", ""),
|
||
"active_base": base, "route_bases": routes}
|
||
return {}
|
||
|
||
|
||
def enum_profile(objs: list[dict[str, Any]], card: str, want: str) -> int:
|
||
"""在 EnumProfile 里按名字(或索引)找 index."""
|
||
if want.isdigit():
|
||
return int(want)
|
||
for o in objs:
|
||
if o.get("type") != "PipeWire:Interface:Device":
|
||
continue
|
||
p = o.get("info", {}).get("props", {})
|
||
if p.get("device.name") != card:
|
||
continue
|
||
for it in o.get("info", {}).get("params", {}).get("EnumProfile") or []:
|
||
if it.get("name") == want:
|
||
return int(it.get("index", -1))
|
||
return -1
|
||
|
||
|
||
def card_of_sink(objs: list[dict[str, Any]], sink_name: str) -> str:
|
||
"""由 sink node.name 反查所属卡(device.name)."""
|
||
n = nodes(objs).get(sink_name)
|
||
if not n:
|
||
return ""
|
||
dev_id = n.get("info", {}).get("props", {}).get("device.id")
|
||
for o in objs:
|
||
if o.get("type") == "PipeWire:Interface:Device" and o.get("id") == dev_id:
|
||
return o.get("info", {}).get("props", {}).get("device.name", "")
|
||
return ""
|
||
|
||
|
||
def links_ok() -> tuple[bool, str]:
|
||
"""当前那条虚拟声卡的 filter 输出端是否真的接上了物理设备.
|
||
|
||
★ 属主名是 cinema_spatial[_up]_out(不带 _sink). 必须认当前生效的那条 ——
|
||
5.1 与上混两条同时存在, 只报"找到了"会把 5.1 的连线当成上混的.
|
||
"""
|
||
txt = run(["pw-link", "-l"])
|
||
if not txt:
|
||
return False, "pw-link 无输出"
|
||
want = active_out_name()
|
||
cur = ""
|
||
found: list[str] = []
|
||
for line in txt.splitlines():
|
||
if not line.startswith((" ", "\t")) and ":" in line:
|
||
cur = line.strip()
|
||
if "cinema_spatial" in cur and "|->" in line:
|
||
found.append(cur)
|
||
if want in cur:
|
||
return True, cur
|
||
if found:
|
||
return False, f"接的不是当前那条(要 {want}); 实际: {'; '.join(found)}"
|
||
return False, "未找到 cinema_spatial* 的输出连线"
|
||
|
||
|
||
def default_sink_name() -> str:
|
||
"""当前默认输出的 node.name."""
|
||
m = re.search(r'node\.name = "([^"]+)"', run(["wpctl", "inspect", "@DEFAULT_AUDIO_SINK@"]))
|
||
return m.group(1) if m else ""
|
||
|
||
|
||
def active_out_name() -> str:
|
||
"""当前生效的 filter 输出端名."""
|
||
return "cinema_spatial_up_out" if "up" in default_sink_name() else "cinema_spatial_out"
|
||
|
||
|
||
def state() -> dict[str, Any]:
|
||
"""汇总状态."""
|
||
objs = pw_dump()
|
||
nd = nodes(objs)
|
||
target = conf_target()
|
||
vols = {n: vol_linear(int(nd[n]["id"])) for n in VOLUME_NODES if n in nd}
|
||
card = card_of_sink(objs, target) if target else ""
|
||
prof = profile_of_card(objs, card) if card else {}
|
||
digital = any(h in target.lower() for h in DIGITAL_HINTS) if target else False
|
||
# ★★ 全链 = 链上**每一级**相乘, 不能只算首尾。
|
||
# 2026-09-13 实测血亏: 旧版只乘 up_sink × up_out, 而中间的上混图输入口
|
||
# (cinema_spatial_up_raw) 被写成了 0.2068 —— 全链凭空掉 13.7 dB,
|
||
# **账面显示 -8.1 dB 一切正常, 实际 -29.9 dB 耳机几乎没声音**。
|
||
# 凡是"账对不上听感"的现场, 先怀疑这张账漏了级。
|
||
total = 1.0
|
||
for n in CHAIN_UP_NODES:
|
||
total *= vols.get(n, 1.0)
|
||
mid_off = {n: vols[n] for n in CHAIN_MID_NODES
|
||
if n in vols and abs(vols[n] - 1.0) > 0.02}
|
||
ok, why = links_ok()
|
||
return {
|
||
"conf_target": target,
|
||
"target_alive": target in nd,
|
||
"target_is_digital": digital,
|
||
"card": card,
|
||
"card_profile": prof.get("name", ""),
|
||
"card_profile_index": prof.get("index", -1),
|
||
"card_route_bases": prof.get("route_bases", []),
|
||
"card_active_base": prof.get("active_base"),
|
||
"volumes": vols,
|
||
"chain_linear": total,
|
||
"chain_db": db(total),
|
||
"chain_mid_off": mid_off,
|
||
"links_ok": ok,
|
||
"links_why": why,
|
||
}
|
||
|
||
|
||
def report() -> str:
|
||
"""人读报告: 音量账 + 设备情况."""
|
||
s = state()
|
||
L: list[str] = []
|
||
L.append(f"输出目标 : {s['conf_target'] or '<未设置>'}"
|
||
f"{' ✓ 存在' if s['target_alive'] else ' ✗ 已失效'}")
|
||
if s["card"]:
|
||
L.append(f"物理卡 : {s['card']}")
|
||
L.append(f"当前 profile: {s['card_profile']} (index {s['card_profile_index']}) "
|
||
f"硬件增益基数: {s.get('card_active_base') or '?'}")
|
||
bases = " / ".join(
|
||
f"{'数字(S/PDIF)' if r['kind'] == 'digital' else '模拟'} {r['base']:.5f}"
|
||
f" ({db(r['base']):+.2f} dB)"
|
||
for r in s.get("card_route_bases", []) if r.get("base"))
|
||
if bases:
|
||
L.append(f" route 硬件增益基数: {bases}")
|
||
if s["target_is_digital"]:
|
||
L.append(" · 输出走数字(S/PDIF) route: 实测音质更好(96k/24bit), 其硬件增益基数"
|
||
"比模拟低 0.94 dB(无感) —— 要补偿可设 CINEMA_SPATIAL_VOLUME_DIGITAL")
|
||
L.append("音量账(全链每一级):")
|
||
for k, label in (("cinema_spatial_up_sink", "虚拟声卡(上混)"),
|
||
("cinema_spatial_up_norm", "归一化输出"),
|
||
("cinema_spatial_up_raw", "上混图输入"),
|
||
("cinema_spatial_up_out", "filter 输出端"),
|
||
("cinema_spatial_sink", "虚拟声卡(5.1)"),
|
||
("cinema_spatial_out", "filter 输出端(5.1)")):
|
||
if k in s["volumes"]:
|
||
L.append(f" {label:<14}{k:<26} = {s['volumes'][k]:.3f} ({db(s['volumes'][k]):+.1f} dB)")
|
||
L.append(f" 整链(上混) 四级相乘 = {s['chain_linear']:.3f} ({s['chain_db']:+.1f} dB)")
|
||
if s.get("chain_mid_off"):
|
||
L.append(" !! 链中间级被写成非单位增益(不归用户调, 掉多少听感就掉多少):")
|
||
for n, v in s["chain_mid_off"].items():
|
||
L.append(f" {n} = {v:.3f} ({db(v):+.1f} dB)")
|
||
L.append(" 修: 跑 `空间音频 开`(会自动纠正), 或 wpctl set-volume <该节点id> 1.0")
|
||
L.append(f"连线 : {'✓ ' + s['links_why'] if s['links_ok'] else '✗ ' + s['links_why']}")
|
||
dup = duplicates(pw_dump())
|
||
if dup:
|
||
L.append(f"!! 同名节点重复 {len(dup)} 个 —— 有两份同名 conf 同时加载:")
|
||
L.append(" 音频会进到错的那一份(表现为没声音), 且两份都往同一个物理设备推流")
|
||
for n, c in dup:
|
||
L.append(f" {n} × {c}")
|
||
L.append(" 修: 只留一份 —— 把另一处同名 conf 改名成 .disabled-dup 后重启 pipewire")
|
||
return "\n".join(L)
|
||
|
||
|
||
DUP_WATCH: list[str] = [
|
||
"cinema_spatial_sink", "cinema_spatial_out",
|
||
"cinema_spatial_up_sink", "cinema_spatial_up_norm",
|
||
"cinema_spatial_up_raw", "cinema_spatial_up_out",
|
||
]
|
||
|
||
|
||
def duplicates(objs: list[dict[str, Any]]) -> list[tuple[str, int]]:
|
||
"""同名节点超过 1 份 = 有两份同名 conf 被同时加载(音频进错那一份).
|
||
|
||
★ PipeWire 对 conf.d/*.conf **两份都加载**, 不做"用户级覆盖系统级" —— 所以
|
||
"deb 装到 /usr/share/pipewire/pipewire.conf.d/ 的那份 + 开发目录用户级那份"
|
||
会各建一套实例。2026-09-13 实测症状: 6 个节点各 2 份、两条链的输出都推
|
||
同一个物理设备、归一化实例的 DSP 进程 4 个抢同一对槽位锁、应用流进了没有
|
||
下游消费者的那一份 → 整条链静默(没声音)。
|
||
"""
|
||
cnt: dict[str, int] = {}
|
||
for o in objs:
|
||
if o.get("type") != "PipeWire:Interface:Node":
|
||
continue
|
||
n = str(o.get("info", {}).get("props", {}).get("node.name", ""))
|
||
if n in DUP_WATCH:
|
||
cnt[n] = cnt.get(n, 0) + 1
|
||
return sorted((n, c) for n, c in cnt.items() if c > 1)
|
||
|
||
|
||
def phys_sink(objs: list[dict[str, Any]]) -> str:
|
||
"""最合适的物理输出 sink node.name: USB 优先, 排除虚拟声卡与 HDMI/DP 专业输出."""
|
||
cand: list[str] = []
|
||
for o in objs:
|
||
if o.get("type") != "PipeWire:Interface:Node":
|
||
continue
|
||
p = o.get("info", {}).get("props", {})
|
||
if p.get("media.class") != "Audio/Sink":
|
||
continue
|
||
n = p.get("node.name", "")
|
||
if not n or "cinema_spatial" in n or "pro-output" in n:
|
||
continue
|
||
cand.append(n)
|
||
cand.sort(key=lambda x: (0 if "usb" in x else 1, x))
|
||
return cand[0] if cand else ""
|
||
|
||
|
||
def sinks(objs: list[dict[str, Any]]) -> list[dict[str, str]]:
|
||
"""全部物理输出 sink: id / node.name / 描述 / 是否当前目标 / 类型(usb|pci|pro).
|
||
|
||
★ 保留显卡 Pro/HDMI 输出(手动选它可以把声音送到电视), 但类型标成 pro —— 自动探测
|
||
(phys_sink) 仍跳过 pro, 免得默认落到没接屏的 HDMI 上。
|
||
"""
|
||
cur = conf_target()
|
||
out: list[dict[str, str]] = []
|
||
for o in objs:
|
||
if o.get("type") != "PipeWire:Interface:Node":
|
||
continue
|
||
p = o.get("info", {}).get("props", {})
|
||
if p.get("media.class") != "Audio/Sink":
|
||
continue
|
||
n = str(p.get("node.name", ""))
|
||
if not n or "cinema_spatial" in n:
|
||
continue
|
||
kind = "pro" if "pro-output" in n else ("usb" if "usb" in n else "pci")
|
||
out.append({"id": str(o.get("id", "")), "name": n,
|
||
"desc": str(p.get("node.description", "")) or n,
|
||
"current": "1" if n == cur else "0", "kind": kind})
|
||
out.sort(key=lambda d: (0 if d["kind"] == "usb" else (1 if d["kind"] == "pci" else 2), d["name"]))
|
||
return out
|
||
|
||
|
||
def device_id(objs: list[dict[str, Any]], card: str) -> str:
|
||
"""按 device.name 找设备的数字 id."""
|
||
for o in objs:
|
||
p = o.get("info", {}).get("props", {})
|
||
if o.get("type") == "PipeWire:Interface:Device" and p.get("device.name") == card:
|
||
return str(o["id"])
|
||
return ""
|
||
|
||
|
||
def profiles_of_card(objs: list[dict[str, Any]], card: str) -> list[dict[str, Any]]:
|
||
"""该卡可选的全部 profile(EnumProfile: 能力清单)."""
|
||
for o in objs:
|
||
p = o.get("info", {}).get("props", {})
|
||
if o.get("type") != "PipeWire:Interface:Device" or p.get("device.name") != card:
|
||
continue
|
||
return list(o.get("info", {}).get("params", {}).get("EnumProfile") or [])
|
||
return []
|
||
|
||
|
||
def pick_route_profile(objs: list[dict[str, Any]], card: str,
|
||
kind: str) -> dict[str, Any]:
|
||
"""挑 route 对应的 profile: digital -> iec958-stereo, analog -> analog-stereo.
|
||
|
||
优先带 input:mono-fallback 的那条(别把麦克风切没了).
|
||
"""
|
||
want = "iec958-stereo" if kind == "digital" else "analog-stereo"
|
||
cand = [p for p in profiles_of_card(objs, card)
|
||
if str(p.get("name", "")).startswith("output:") and want in str(p.get("name", ""))]
|
||
if not cand:
|
||
return {}
|
||
cand.sort(key=lambda p: (0 if "input:" in str(p.get("name", "")) else 1,
|
||
int(p.get("index", 99))))
|
||
return cand[0]
|
||
|
||
|
||
def route_kind(objs: list[dict[str, Any]]) -> str:
|
||
"""当前物理输出走 digital / analog / ''(未知) —— 由 sink 名后缀判定."""
|
||
n = phys_sink(objs)
|
||
if any(h in n for h in DIGITAL_HINTS):
|
||
return "digital"
|
||
return "analog" if "analog" in n else ""
|
||
|
||
|
||
def main() -> int:
|
||
args = sys.argv[1:]
|
||
cmd = args[0] if args else "check"
|
||
if cmd == "json":
|
||
print(json.dumps(state(), ensure_ascii=False, indent=1))
|
||
elif cmd == "check":
|
||
print(report())
|
||
elif cmd == "phys-sink":
|
||
print(phys_sink(pw_dump()))
|
||
elif cmd == "card-of":
|
||
print(card_of_sink(pw_dump(), args[1]))
|
||
elif cmd == "profile":
|
||
card = args[1]
|
||
prof = profile_of_card(pw_dump(), card)
|
||
print(f"{prof.get('index', -1)}\t{prof.get('name', '')}")
|
||
elif cmd == "set-profile":
|
||
card, want = args[1], args[2]
|
||
objs = pw_dump()
|
||
idx = enum_profile(objs, card, want)
|
||
if idx < 0:
|
||
print(f"找不到 profile: {card} / {want}")
|
||
return 1
|
||
dev_id = device_id(objs, card)
|
||
if not dev_id:
|
||
print(f"找不到卡: {card}")
|
||
return 1
|
||
run(["wpctl", "set-profile", dev_id, str(idx)])
|
||
print(f"已把 {card} 切到 profile index {idx}")
|
||
elif cmd == "route-kind":
|
||
print(route_kind(pw_dump()))
|
||
elif cmd == "dup":
|
||
for n, c in duplicates(pw_dump()):
|
||
print(f"{n}\t{c}")
|
||
elif cmd == "sinks":
|
||
for d in sinks(pw_dump()):
|
||
print("\t".join([d["id"], d["name"], d["desc"], d["current"], d["kind"]]))
|
||
elif cmd == "set-route":
|
||
card, kind = args[1], args[2]
|
||
objs = pw_dump()
|
||
prof = pick_route_profile(objs, card, kind)
|
||
dev_id = device_id(objs, card)
|
||
if not prof or not dev_id:
|
||
print(f"切不了: card={card} kind={kind} (找不到可用 profile 或卡)")
|
||
return 1
|
||
run(["wpctl", "set-profile", dev_id, str(prof["index"])])
|
||
print(f"已切: {prof.get('name', '')} (index {prof.get('index')})")
|
||
else:
|
||
print(__doc__)
|
||
return 1
|
||
return 0
|
||
|
||
|
||
if __name__ == "__main__":
|
||
raise SystemExit(main())
|