v1.1.0: 单声道合并 / 运行期重连(切设备不重启) / 等目标回来(蓝牙重连) / 能量归一修削波 / 电平显式连线不经麦克风 / 面板核心技术区 / 打包修 control 版本+补音频状态.py与web目录
This commit is contained in:
edgevoid
@@ -11,6 +11,7 @@
|
||||
"""
|
||||
import os
|
||||
import subprocess
|
||||
import json
|
||||
|
||||
_ROOT = os.path.dirname(os.path.abspath(__file__))
|
||||
# IR 目录: 环境变量优先(打包/安装场景指 /usr/share/cinema-spatial/hrir),
|
||||
@@ -22,6 +23,162 @@ OUT = os.environ.get("CINEMA_SPATIAL_OUT") or os.path.expanduser(
|
||||
)
|
||||
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")
|
||||
|
||||
|
||||
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 —— 必须是物理设备。
|
||||
@@ -116,15 +273,30 @@ def sink_51() -> list[str]:
|
||||
' 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" ]',
|
||||
' outputs = [ "mixL:Out" "mixR:Out" ]', " }",
|
||||
f' outputs = [ {outs} ]', " }",
|
||||
f" capture.props = {{ {VOL_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}'
|
||||
"audio.channels = 2 audio.position = [ FL FR ] }",
|
||||
+ out_props(mono),
|
||||
" }", " }"]
|
||||
return L
|
||||
|
||||
@@ -149,6 +321,17 @@ def sink_upmix() -> list[str]:
|
||||
' { 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 = [",
|
||||
' { output = "copyFL:Out" input = "mixF:In 1" }',
|
||||
' { output = "copyFR:Out" input = "mixF:In 2" }',
|
||||
@@ -168,18 +351,31 @@ def sink_upmix() -> list[str]:
|
||||
' { 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" ]',
|
||||
' outputs = [ "mixL:Out" "mixR:Out" ]', " }",
|
||||
f' outputs = [ {outs} ]', " }",
|
||||
f" capture.props = {{ {VOL_PROP}node.name = cinema_spatial_up_sink media.class = Audio/Sink "
|
||||
"audio.channels = 2 audio.position = [ FL FR ] }",
|
||||
f' playback.props = {{ node.name = cinema_spatial_up_out {TARGET_PROP}'
|
||||
"audio.channels = 2 audio.position = [ FL FR ] }",
|
||||
+ out_props(mono),
|
||||
" }", " }"]
|
||||
return L
|
||||
|
||||
|
||||
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()
|
||||
|
||||
Reference in New Issue
Block a user