cinema-spatial/生成配置.py

#!/usr/bin/env python3
"""生成 PipeWire filter-chain 配置（2026-09-12）

两个虚拟声卡:
  1) cinema_spatial_sink     5.1 输入      -> 双耳 HRTF         (多声道片源)
  2) cinema_spatial_up_sink  立体声输入    -> 上混5.1 -> HRTF  (浏览器/音乐等, 全局默认用这个)

上混链照抄 PipeWire 自带的 sink-upmix-5.1-filter.conf(矩阵式):
  FL/FR 直通; FC = (FL+FR)*0.707 低通; LFE = 同源再低通+衰减;
  RL/RR = (FL-FR)*0.707 反相 + 希尔伯特卷积(产生相位差/延迟)
"""
import os
import math
import subprocess
import json

_ROOT = os.path.dirname(os.path.abspath(__file__))
# IR 目录: 环境变量优先(打包/安装场景指 /usr/share/cinema-spatial/hrir),
# 否则用项目内的软链 hrir/current (开发时换 HRTF 只改软链)。
IR = os.environ.get("CINEMA_SPATIAL_IR") or os.path.join(_ROOT, "hrir", "current")

OUT = os.environ.get("CINEMA_SPATIAL_OUT") or os.path.expanduser(
    "~/.config/pipewire/pipewire.conf.d/90-cinema-spatial.conf"
)
DIRS: list[str] = ["FL", "FR", "FC", "LFE", "BL", "BR"]

# 房间混响(镜像法生成的真实房间 IR; 生成器在 声学追踪引擎/src/镜像法IR.py)
REVERB_IR = os.environ.get("CINEMA_SPATIAL_REVERB_IR") or os.path.join(
    _ROOT, "reverb", "房间混响IR-96k.wav")
REVERB_STATE = os.path.expanduser("~/.local/state/cinema-spatial/reverb.json")

# 采样参数(HRTF 模型 / tap 数 / 采样率) —— 运行时状态, 由 CLI「参数」写
PARAMS = os.path.expanduser("~/.local/state/cinema-spatial/params.json")
CLOCK_OUT = os.environ.get("CINEMA_SPATIAL_CLOCK") or os.path.expanduser(
    "~/.config/pipewire/pipewire.conf.d/91-clock.conf")


# 单声道模式(单声道蓝牙音响/耳机): 状态文件三态, 缺省自动判目标设备声道数
MONO_STATE = os.path.expanduser("~/.local/state/cinema-spatial/mono.json")

# 响度归一化(2026-09-13): 在图的最前端插一级 pipe 插件, 挂自研 DSP 做全系统响度统一。
# 状态: ~/.local/state/cinema-spatial/loudness.json  {"on": bool, "target_db": float}
LOUDNESS_STATE = os.path.expanduser("~/.local/state/cinema-spatial/loudness.json")
# ★ LN_ENTRY 必须是**纯 ASCII 路径**: pipe 插件的 command 走 shebang 式 exec, 实测传中文
#   路径时它一声不吭 —— 节点建得出来、音频直接绕过去、DSP 根本不 fork、日志无任何错。
#   wrapper 内部再去 exec 中文路径的 DSP 是没问题的。
LN_ENTRY = (os.environ.get("CINEMA_SPATIAL_LN")
            or os.path.expanduser("~/.local/bin/collaplex-loudness-norm"))


def loudness_cfg() -> tuple[bool, float]:
    """响度归一化开关 + 目标电平(dBFS).

    判定顺序: 环境变量 CINEMA_SPATIAL_LOUDNESS(0/1) > 状态文件 > 默认**开**。

    ★ 只插在**立体声上混**图里 —— 那才是抖音/网页/音乐走的路; 5.1 图放的是专业混音
      片源, 响度本来就规矩, 不掺和。
    """
    on = True
    target = -16.0
    try:
        with open(LOUDNESS_STATE, encoding="utf-8") as fh:
            d = json.load(fh)
        if isinstance(d, dict):
            on = bool(d.get("on", True))
            target = float(d.get("target_db") or -16.0)
    except (OSError, ValueError, TypeError):
        pass
    env = os.environ.get("CINEMA_SPATIAL_LOUDNESS")
    if env in ("0", "1"):
        on = env == "1"
    return on, target


def loudness_instance() -> list[str]:
    """独立的响度归一化 filter-chain 实例(**串在上混图前面**)。

    ★ 为什么单开一级, 而不是把 pipe 节点塞进上混图:
      实测塞进那个 40+ 节点的大图里 —— 不管放图入口、copy 之后、还是输出末端 ——
      pipe 都会被正常 fork、数据也在稳定流动(每 0.5s 47 个块), 但它收到的永远是
      恒定 -43 dBFS 的"空数据", 音频直接绕过去, 日志里没有任何报错。同一个 pipe
      放在 copy->pipe->copy 的小图里完全正常(实测收到 -12.04 dBFS 的真实节目电平)。
      拆成两级串联绕开这个坑; 归一化本来就是独立一级, 开关/调参也更干净
      (改它不用重建那张大图)。

    ★ command 只能是"程序 + **一个**参数" —— pipe 走 shebang 式 exec, 多一个就炸
      (env A=B python3 x.py 会报 "use -[v]S to pass options in shebang lines")。
      参数全在 wrapper 里, 这里只给脚本路径, 而且**必须是纯 ASCII 路径**:
      实测传中文路径时 pipe 一声不吭 —— 节点照建、音频绕过、无任何日志。

    ★ capture 用原来的 sink 名(cinema_spatial_up_sink): 用户/CLI 视角不变,
      所有软件照样往这个默认 sink 推流, 归一化在它后面悄悄做掉。
    """
    return [
        "  { name = libpipewire-module-filter-chain",
        "    flags = [ nofail ]",
        "    args = {",
        '      node.description = "电影院空间音频 (响度归一化)"',
        '      media.name       = "Cinema Spatial Loudness"',
        "      filter.graph = {",
        "        nodes = [",
        "          { type = builtin label = copy name = c1L }",
        "          { type = builtin label = copy name = c1R }",
        "          { type = builtin label = pipe name = lnL",
        f'            config = {{ command = "{LN_ENTRY}" }} }}',
        "          { type = builtin label = pipe name = lnR",
        f'            config = {{ command = "{LN_ENTRY}" }} }}',
        "          { type = builtin label = copy name = c2L }",
        "          { type = builtin label = copy name = c2R }",
        "        ]",
        "        links = [",
        '          { output = "c1L:Out" input = "lnL:In" }',
        '          { output = "c1R:Out" input = "lnR:In" }',
        '          { output = "lnL:Out" input = "c2L:In" }',
        '          { output = "lnR:Out" input = "c2R:In" }',
        "        ]",
        '        inputs  = [ "c1L:In" "c1R:In" ]',
        '        outputs = [ "c2L:Out" "c2R:Out" ]',
        "      }",
        f"      capture.props  = {{ {VOL_PROP}{PRIO_PROP}node.name = cinema_spatial_up_sink  "
        "media.class = Audio/Sink  audio.channels = 2  audio.position = [ FL FR ] }",
        "      playback.props = { node.name = cinema_spatial_up_norm  "
        'node.target = "cinema_spatial_up_raw"  audio.channels = 2  '
        "audio.position = [ FL FR ] }",
        "    }",
        "  }",
    ]


def mono_cfg() -> bool:
    """是否把链的输出合并成单声道.

    为什么需要: 单声道蓝牙音响只有 1 个 MONO 端口, 链若坚持输出 FL/FR 就链接失败 →
    设备掉线 → 回退守卫滚回 -> 之后一直连不上(2026-09-13 老板实测)。
    判定顺序: 环境变量 CINEMA_SPATIAL_MONO > 状态文件 {"on": bool} > 自动(目标设备 1 声道)。
    """
    env = os.environ.get("CINEMA_SPATIAL_MONO")
    if env in ("0", "1"):
        return env == "1"
    try:
        with open(MONO_STATE, encoding="utf-8") as fh:
            d = json.load(fh)
        if isinstance(d, dict) and isinstance(d.get("on"), bool):
            return bool(d["on"])
    except (OSError, ValueError):
        pass
    return target_channels() == 1


def target_name() -> str:
    """node.target 里的设备名(为空 = 交给 WirePlumber 路由)."""
    return TARGET_PROP.split('"')[1] if '"' in TARGET_PROP else ""


def target_channels() -> int:
    """目标设备有几个声道(pw-dump 读 audio.channels; 读不到按 2)."""
    tgt = target_name()
    if not tgt:
        return 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") == tgt:
                return int(pr.get("audio.channels") or 2)
    except (OSError, ValueError, subprocess.SubprocessError):
        pass
    return 2


def mono_block(indent: str) -> list[str]:
    """单声道合并级: L/R 各 -6dB 相加(避免相关信号 +6dB 溢出)."""
    return [f'{indent}{{ type = builtin label = mixer name = monoOut '
            'control = { "Gain 1" = 0.5 "Gain 2" = 0.5 } }']


def mono_links(indent: str, srcs: tuple[str, str]) -> list[str]:
    a, b = srcs
    return [f'{indent}{{ output = "{a}" input = "monoOut:In 1" }}',
            f'{indent}{{ output = "{b}" input = "monoOut:In 2" }}']


def out_props(mono: bool) -> str:
    """链输出节点的声道属性."""
    if mono:
        return "audio.channels = 1  audio.position = [ MONO ] }"
    return "audio.channels = 2  audio.position = [ FL FR ] }"


def params_cfg() -> dict[str, object]:
    """读采样参数; 缺省/坏文件都当空(→ 回落到 hrir/current 软链)."""
    try:
        with open(PARAMS, encoding="utf-8") as fh:
            d = json.load(fh)
    except (OSError, ValueError):
        return {}
    return d if isinstance(d, dict) else {}


def ir_dir() -> str:
    """选 IR 目录: 环境变量 > 参数里的 taps-<N> > 参数里的模型名 > 软链 hrir/current."""
    env = os.environ.get("CINEMA_SPATIAL_IR")
    if env:
        return env
    p = params_cfg()
    taps = int(p.get("taps") or 0)          # type: ignore[arg-type]
    cut = os.path.join(_ROOT, "hrir", "taps-%d" % taps)
    if taps and os.path.isdir(cut):
        return cut
    name = str(p.get("hrtf") or "")
    if name and os.path.isdir(os.path.join(_ROOT, "hrir", name)):
        return os.path.join(_ROOT, "hrir", name)
    return os.path.join(_ROOT, "hrir", "current")


def write_clock(rate: int) -> str:
    """写用户级主时钟(同名覆盖 deb 装的 /usr/share 那份, 不用 sudo)."""
    txt = (
        "# 主时钟 %d Hz —— 由 空间音频 参数 rate 生成\n"
        "#\n"
        "# 为什么必须显式写: PipeWire 的 convolver 按\"当前时钟率\"重采样 IR,\n"
        "# 时钟若低于 IR 的采样率, 等于白换高采样率 IR。\n"
        "context.properties = {\n"
        "    default.clock.rate          = %d\n"
        "    default.clock.allowed-rates = [ %d ]\n"
        "}\n" % (rate, rate, rate))
    os.makedirs(os.path.dirname(CLOCK_OUT), exist_ok=True)
    with open(CLOCK_OUT, "w", encoding="utf-8") as fh:
        fh.write(txt)
    return CLOCK_OUT


def reverb_cfg() -> tuple[bool, float]:
    """(是否开混响, 湿量) —— 读运行时状态文件; 缺省关, 没有 IR 也当关."""
    if not os.path.exists(REVERB_IR):
        return False, 0.0
    try:
        with open(REVERB_STATE, encoding="utf-8") as fh:
            d = json.load(fh)
    except (OSError, ValueError):
        return False, 0.0
    wet = float(d.get("wet") or 0.0)
    on = bool(d.get("on")) and wet > 0.001
    return on, max(0.0, min(wet, 1.0))


def reverb_block(indent: str, wet: float) -> list[str]:
    """混响级: HRTF 输出 -> 卷积房间 IR -> 与干信号按湿量混合.

    dry 走 Gain 1(恒 1.0), wet 走 Gain 2(湿量) —— 湿量是生成期常量, 改湿量=重新生成.
    """
    return [
        f'{indent}{{ type = builtin label = convolver name = revL config = {{ filename = "{REVERB_IR}" }} }}',
        f'{indent}{{ type = builtin label = convolver name = revR config = {{ filename = "{REVERB_IR}" }} }}',
        f'{indent}{{ type = builtin label = mixer name = wetL control = {{ "Gain 1" = 1.0 "Gain 2" = {wet:.4f} }} }}',
        f'{indent}{{ type = builtin label = mixer name = wetR control = {{ "Gain 1" = 1.0 "Gain 2" = {wet:.4f} }} }}',
    ]


def reverb_links(indent: str) -> list[str]:
    """干路 mixL/mixR 直入混音器, 湿路经房间 IR 卷积入混音器."""
    return [
        f'{indent}{{ output = "mixL:Out" input = "revL:In" }}',
        f'{indent}{{ output = "mixR:Out" input = "revR:In" }}',
        f'{indent}{{ output = "mixL:Out" input = "wetL:In 1" }}',
        f'{indent}{{ output = "revL:Out" input = "wetL:In 2" }}',
        f'{indent}{{ output = "mixR:Out" input = "wetR:In 1" }}',
        f'{indent}{{ output = "revR:Out" input = "wetR:In 2" }}',
    ]


def default_sink_name() -> str:
    """当前默认物理输出的 node.name —— 必须是物理设备。

    切换到虚拟声卡之后 @DEFAULT_AUDIO_SINK@ 会指向虚拟声卡自己, 那会让
    node.target 变成自我循环, 所以这种情况要去 Sinks 列表里挑第一个物理输出。
    """
    def props(cmd: list[str]) -> str:
        try:
            out = subprocess.run(cmd, capture_output=True, text=True, timeout=10).stdout
        except Exception:
            return ""
        for line in out.splitlines():
            if "node.name" in line and '"' in line:
                return line.split('"')[1]
        return ""

    cur = props(["wpctl", "inspect", "@DEFAULT_AUDIO_SINK@"])
    if cur and "cinema_spatial" not in cur:
        return cur

    try:
        status = subprocess.run(["wpctl", "status"], capture_output=True, text=True, timeout=10).stdout
    except Exception:
        return ""
    in_sinks = False
    first = ""
    for line in status.splitlines():
        if "Sinks:" in line:
            in_sinks = True
            continue
        if "Sources:" in line:
            break
        if not in_sinks or "cinema_spatial" in line:
            continue
        for tok in line.split():
            if tok.endswith(".") and tok[:-1].isdigit():
                name = props(["wpctl", "inspect", tok[:-1]])
                if not name or "cinema_spatial" in name:
                    continue
                if "usb-" in name:
                    return name          # USB 设备优先(耳机/外置声卡)
                if not first:
                    first = name
    return first


# node.target —— filter-chain 的输出端必须送到"真实设备":
#   - @DEFAULT_SINK@ 这类 pactl 占位符 PipeWire 不认(会当成设备名, 结果连到别的设备)
#   - 完全不写则 WirePlumber 可能连错(实测连到 HDMI)
# 所以默认探测物理输出(优先 USB), 换设备后重跑本脚本即可。
# 也可 export CINEMA_SPATIAL_TARGET=<node.name> 强制指定。
_TARGET = os.environ.get("CINEMA_SPATIAL_TARGET") or default_sink_name()
TARGET_PROP = f'node.target = "{_TARGET}"  ' if _TARGET else ""

# 虚拟声卡的音量。★ 必须写进配置, 否则重建后不保留:
# filter-chain 重建时新节点没有历史音量, 会回到 1.0 → 整链直接削波爆音。
# 0.25 ≈ -12dB, 给电影那种高峰值因子瞬态留足余量(见 README 第 10、11 条)。
VOLUME = os.environ.get("CINEMA_SPATIAL_VOLUME", "0.25")
VOL_PROP = f"node.volume = {VOLUME}  " if VOLUME else ""
# ★★ 虚拟声卡必须带**高于所有物理设备**的 priority.session, 否则默认节点会被抢走:
#   实测本机 HDMI(pro-output-3) = 1196、USB EDIFIER = 1108、板载模拟 = 1009,
#   而虚拟声卡原本**没有这个属性**(None) → 每次"重新选择默认节点"(显示器唤醒、
#   设备插拔、WirePlumber 重启)都是 HDMI 赢; 从 GNOME 手动选回来, 下一个事件又抢走
#   (老板报"虚拟声卡老是被 hdmi 抢走, 我点都点不回来")。设成 2000 就压过全部物理口。
SINK_PRIO = 2000
PRIO_PROP = f"priority.session = {SINK_PRIO}  "


def hrir_block(indent: str) -> list[str]:
    """12 个卷积器 + 6 个分发 + 左右耳混音。输入端口: cp<dir>:In"""
    L: list[str] = []
    for d in DIRS:
        for e in ("L", "R"):
            L.append(f'{indent}{{ type = builtin label = convolver name = c{d}_{e} '
                     f'config = {{ filename = "{IR}/{d}_{e}.wav" }} }}')
    for d in DIRS:
        L.append(f'{indent}{{ type = builtin label = copy name = cp{d} }}')
    for e in ("L", "R"):
        # ★★ 6 路相加必须做**能量归一**(1/sqrt(6) = -7.8 dB), 不能每路都 1。
        #   2026-09-13 实测: 每路 gain=1 时上混链净增益 +6 dB —— 归一化 DSP 已经把
        #   输出压在峰值保护线(-0.13 dBFS), 这里再抬 6 dB 就顶穿满刻度(起播瞬间实测
        #   +4.2 ~ +7.8 dBFS, 就是老板看到的"电平怎么都顶满")。
        #   1/sqrt(6) 是"不相关信号相加"的能量守恒系数, 净增益落到 -1.8 dB, 留出余量。
        g6 = "%.4f" % (1.0 / math.sqrt(6.0))
        gains = " ".join(f'"Gain {i}" = {g6}' for i in range(1, 7))
        L.append(f'{indent}{{ type = builtin label = mixer name = mix{e} control = {{ {gains} }} }}')
    return L


def hrir_links(indent: str) -> list[str]:
    L: list[str] = []
    for d in DIRS:
        for e in ("L", "R"):
            L.append(f'{indent}{{ output = "cp{d}:Out" input = "c{d}_{e}:In" }}')
    for i, d in enumerate(DIRS, start=1):
        L.append(f'{indent}{{ output = "c{d}_L:Out" input = "mixL:In {i}" }}')
    for i, d in enumerate(DIRS, start=1):
        L.append(f'{indent}{{ output = "c{d}_R:Out" input = "mixR:In {i}" }}')
    return L


def sink_51() -> list[str]:
    L = ["  { name = libpipewire-module-filter-chain", "    flags = [ nofail ]", "    args = {",
         '      node.description = "电影院空间音频 (5.1 虚拟环绕)"',
         '      media.name       = "Cinema Spatial 5.1"', "      filter.graph = {", "        nodes = ["]
    L += hrir_block("          ")
    on, wet = reverb_cfg()
    outs = '"mixL:Out" "mixR:Out"'
    srcs = ("mixL:Out", "mixR:Out")
    mono = mono_cfg()
    if on:
        L += reverb_block("          ", wet)
        outs = '"wetL:Out" "wetR:Out"'
        srcs = ("wetL:Out", "wetR:Out")
    if mono:
        L += mono_block("          ")
        outs = '"monoOut:Out"'
    L += ["        ]", "        links = ["]
    L += hrir_links("          ")
    if on:
        L += reverb_links("          ")
    if mono:
        L += mono_links("          ", srcs)
    L += ["        ]",
          '        inputs  = [ "cpFL:In" "cpFR:In" "cpFC:In" "cpLFE:In" "cpBL:In" "cpBR:In" ]',
          f'        outputs = [ {outs} ]', "      }",
          f"      capture.props  = {{ {VOL_PROP}{PRIO_PROP}node.name = cinema_spatial_sink  media.class = Audio/Sink  "
          "audio.channels = 6  audio.position = [ FL FR FC LFE BL BR ] }",
          f'      playback.props = {{ node.name = cinema_spatial_out  {TARGET_PROP}'
          + out_props(mono),
          "    }", "  }"]
    return L


def sink_upmix() -> list[str]:

    L = ["  { name = libpipewire-module-filter-chain", "    flags = [ nofail ]", "    args = {",
         '      node.description = "电影院空间音频 (立体声上混)"',
         '      media.name       = "Cinema Spatial Upmix"', "      filter.graph = {", "        nodes = ["]
    # 响度归一化**不在**这张图里做 —— 它跑在独立的 filter-chain 实例上(见
    # loudness_instance), 由 cinema_spatial_up_sink 收流、再把结果推给这张图。
    L += ["          # ---- 第一段: 立体声矩阵上混成 5.1 (照抄官方 upmix) ----",
          "          { type = builtin label = copy name = copyFL }",
         "          { type = builtin label = copy name = copyFR }",
         "          { type = builtin label = copy name = copyOFL }",
         "          { type = builtin label = copy name = copyOFR }",
         '          { type = builtin label = mixer name = mixF control = { "Gain 1" = 0.707 "Gain 2" = 0.707 } }',
         "          { type = builtin label = param_eq name = eq_FC_LFE config = {",
         "              filters1 = [ { type = bq_lowpass freq = 12000 } { type = bq_lowpass freq = 12000 } ]",
         "              filters2 = [ { type = bq_highshelf freq = 0 gain = -20.0 } "
         "{ type = bq_lowpass freq = 120 } { type = bq_lowpass freq = 120 } ] } }",
         '          { type = builtin label = mixer name = subR control = { "Gain 1" = 0.707 "Gain 2" = -0.707 } }',
         '          { type = builtin label = convolver name = convRL config = { gain = 1.0  delay = 0.012  filename = "/hilbert"  length = 33 } }',
         '          { type = builtin label = convolver name = convRR config = { gain = -1.0  delay = 0.012  filename = "/hilbert"  length = 33 } }',
         "          # ---- 第二段: 5.1 -> 双耳 HRTF ----"]
    L += hrir_block("          ")
    on, wet = reverb_cfg()
    outs = '"mixL:Out" "mixR:Out"'
    srcs = ("mixL:Out", "mixR:Out")
    mono = mono_cfg()
    if on:
        L += reverb_block("          ", wet)
        outs = '"wetL:Out" "wetR:Out"'
        srcs = ("wetL:Out", "wetR:Out")
    if mono:
        L += mono_block("          ")
        outs = '"monoOut:Out"'
    L += ["        ]", "        links = ["]
    L += ['          { output = "copyFL:Out"   input = "mixF:In 1" }',
          '          { output = "copyFR:Out"   input = "mixF:In 2" }',
          '          { output = "copyFL:Out"   input = "copyOFR:In" }',
          '          { output = "copyFR:Out"   input = "copyOFL:In" }',
          '          { output = "mixF:Out"     input = "eq_FC_LFE:In 1" }',
          '          { output = "mixF:Out"     input = "eq_FC_LFE:In 2" }',
          '          { output = "copyFL:Out"   input = "subR:In 1" }',
          '          { output = "copyFR:Out"   input = "subR:In 2" }',
          '          { output = "subR:Out"     input = "convRL:In" }',
          '          { output = "subR:Out"     input = "convRR:In" }',
          "          # 上混产出的 6 路接进 HRTF 的分发",
          '          { output = "copyOFL:Out"  input = "cpFL:In" }',
          '          { output = "copyOFR:Out"  input = "cpFR:In" }',
          '          { output = "eq_FC_LFE:Out 1" input = "cpFC:In" }',
          '          { output = "eq_FC_LFE:Out 2" input = "cpLFE:In" }',
          '          { output = "convRL:Out"   input = "cpBL:In" }',
          '          { output = "convRR:Out"   input = "cpBR:In" }']
    L += hrir_links("          ")
    if on:
        L += reverb_links("          ")
    if mono:
        L += mono_links("          ", srcs)
    L += ["        ]",
          '        inputs  = [ "copyFL:In" "copyFR:In" ]',
          f'        outputs = [ {outs} ]', "      }",
          # ★ 这一级不再是默认 sink —— sink 名让给前面的归一化实例, 由它的
          #   node.target = "cinema_spatial_up_raw" 把流推到这里。
          "      capture.props  = { node.volume = 1.0  node.name = cinema_spatial_up_raw  "
          "media.class = Audio/Sink  audio.channels = 2  audio.position = [ FL FR ] }",
          f'      playback.props = {{ node.name = cinema_spatial_up_out  {TARGET_PROP}'
          + out_props(mono),
          "    }", "  }"]
    return L


# ★★ PipeWire 对每个 conf.d 目录里的 *.conf **全都加载**, 不做"用户级覆盖系统级" ——
#    所以"deb 装了一份到 /usr/share/pipewire/pipewire.conf.d/ 之后, 开发时又写了一份
#    用户级" = 同名实例建两遍、节点重名(实测 6 个节点各 2 份), 音频进哪一份由调度决定,
#    常常进了没有下游消费者的那一份 → 整条链静默。症状: "两边看着都在, 就是没声音"。
#    (2026-09-13 实测) 写入后顺手把**别处**的同名 conf 挪开: 能改名就改名, root 的只告警。
CONF_DIRS: list[str] = [
    "/usr/share/pipewire/pipewire.conf.d",
    "/etc/pipewire/pipewire.conf.d",
    os.path.expanduser("~/.config/pipewire/pipewire.conf.d"),
]


def _prune_one(name: str, keep: str) -> None:
    """把 keep 之外的同名 conf 挪成 .disabled-dup(root 的挪不动就告警)."""
    mine = os.path.realpath(keep)
    for d in CONF_DIRS:
        p = os.path.join(d, name)
        if not os.path.isfile(p) or os.path.realpath(p) == mine:
            continue
        try:
            os.rename(p, p + ".disabled-dup")
            print(f"  · 移开重复配置 {p} -> {name}.disabled-dup")
        except OSError as e:
            print(f"  ! 重复配置挪不动: {p} ({e.strerror})")
            print("    它会与本配置建出两份同名声卡 = 音频进错实例(没声音)。请执行一次:")
            print(f"    sudo mv {p} {p}.disabled-dup")


def prune_duplicate_confs() -> None:
    """移开 OUT/CLOCK_OUT 之外的同名 conf.

    ★ 91-clock.conf 同理: 改采样率写的是用户级那份, 而 deb 装在系统级也有一份 →
      两个 default.clock.rate 同时加载 = 时钟不确定。所以两份都要"同名只留一份"。

    ★★ 但**只有在写真正生效的位置时**才该清理: 若 OUT 落在 CONF_DIRS 之外(打包时
      写到临时构建目录、或测试时写到 /tmp), 那用户目录/系统目录里的配置一份都不该动。
      2026-09-13 打包正是栽在这里 —— 生成包内 conf 时把用户级和系统级两份都移开,
      虚拟声卡整个消失(看门狗先报"链输出没有任何连线, 且找不到物理设备")。
      需要强制跳过时设 CINEMA_SPATIAL_KEEP_CONFS=1。
    """
    if os.environ.get("CINEMA_SPATIAL_KEEP_CONFS") == "1":
        return
    live = {os.path.realpath(d) for d in CONF_DIRS}
    if os.path.dirname(os.path.realpath(OUT)) not in live:
        return
    _prune_one(os.path.basename(OUT), OUT)
    if os.path.isfile(CLOCK_OUT):
        _prune_one(os.path.basename(CLOCK_OUT), CLOCK_OUT)


def main() -> int:
    global IR
    IR = ir_dir()
    _p = params_cfg()
    _rate = int(_p.get("rate") or 0)        # type: ignore[arg-type]
    if _rate:
        write_clock(_rate)
    print("IR 目录: " + IR)
    if _p:
        print("采样参数: " + json.dumps(_p, ensure_ascii=False))
    os.makedirs(os.path.dirname(OUT), exist_ok=True)
    body: list[str] = ["# 电影院空间音频  自动生成, 改 IR/增益请改 生成配置.py 重跑", "context.modules = ["]
    body += sink_51()
    # 响度归一化: 独立一级, 串在上混图前面(开关见 ~/.local/state/cinema-spatial/loudness.json)
    if loudness_cfg()[0]:
        body += loudness_instance()
    body += sink_upmix()
    body += ["]", ""]
    with open(OUT, "w", encoding="utf-8") as fh:
        fh.write("\n".join(body))
    prune_duplicate_confs()
    print(f"已生成 {OUT} ({len(body)} 行)")
    print(f"输出目标: {TARGET_PROP.replace('node.target = ', '').strip() or '自动(WirePlumber 路由到默认输出)'}")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())