cinema-spatial/音频状态.py

#!/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())