t2linux/mbp15-1-audio-dsp
1
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

feat: finalize low-latency baked DSP graph with crisp treble, virtualbass-first sub-bass drive, atomic reloads, and eq.py CLI

Browse Source
This commit is contained in:
mynameisdeleted
2026-09-01 09:44:16 -04:00
parent 48a946e9b4
commit 4cd548c643
4 changed files with 116 additions and 163 deletions
Download Patch File Download Diff File Expand all files Collapse all files

107
apply.sh
View File

@@ -20,13 +20,18 @@ GRAPH_DST="/usr/share/t2-linux-audio/15_1/graph.json"
FORCE=0
SIMPLE=0
HOT=0
for arg in "${@:-}"; do
case "$arg" in
-f|--force|--no-check) FORCE=1 ;;
-b|--bake|--simple) SIMPLE=1 ;;
-h|--hot|--soft) HOT=1 ;;
esac
done
have() { command -v "$1" >/dev/null 2>&1; }
die() { echo "Error: $*" >&2; exit 1; }
if [ "$SIMPLE" -eq 1 ]; then
echo "==> Baking static DSP stages into single-stage FIR files..."
python3 "$SCRIPT_DIR/bake-graph.py" || die "bake-graph.py failed"
@@ -35,18 +40,13 @@ if [ "$SIMPLE" -eq 1 ]; then
sudo cp "$SCRIPT_DIR/15_1/baked-"*.wav "/usr/share/t2-linux-audio/15_1/" || die "Failed to copy baked FIR files"
fi
have() { command -v "$1" >/dev/null 2>&1; }
die() { echo "Error: $*" >&2; exit 1; }
# 0 = valid, 1 = invalid, 2 = no validator available
# --- json validation helper -------------------------------------------
json_ok() {
if have python3; then python3 -m json.tool "$1" >/dev/null 2>&1
elif have jq; then jq -e . "$1" >/dev/null 2>&1
else return 2; fi
have python3 && { python3 -c 'import json, sys; json.load(open(sys.argv[1]))' "$1" 2>/dev/null && return 0; return 1; }
have jq && { jq . "$1" >/dev/null 2>&1 && return 0; return 1; }
return 2 # cannot check
}
[ -f "$GRAPH_SRC" ] || die "$GRAPH_SRC not found"
json_ok "$GRAPH_SRC"; rc=$?
[ "$rc" -eq 1 ] && die "graph.json is not valid JSON"
[ "$rc" -eq 2 ] && echo "warn: no python3/jq - skipping JSON validation"
@@ -76,31 +76,35 @@ json_ok "$MERGED"; rc=$?
# --- preflight (against the merged graph) ------------------------------
if [ "$FORCE" -eq 0 ]; then
[ -d "$(dirname "$GRAPH_DST")" ] || die \
"$(dirname "$GRAPH_DST") is missing - install the t2 speaker-DSP package first (INSTALL.md section 1)"
"destination directory $(dirname "$GRAPH_DST") does not exist (is t2-linux-audio-15-1 installed?)"
miss=0
while IFS= read -r w; do
[ -f "$w" ] || { echo " missing FIR: $w"; miss=1; }
done < <(grep -oE '/[^" ]+\.wav' "$MERGED" | sort -u)
[ "$miss" -eq 0 ] && echo "ok: FIR .wav files present" \
|| die "FIR files missing - INSTALL.md section 4"
# check all convolver wav files exist
if have jq; then
wav_files=$(jq -r '.. | .filename? | strings' "$MERGED")
for wav in $wav_files; do
[ -f "$wav" ] || die "FIR WAV file missing: $wav"
done
echo "ok: FIR .wav files present"
fi
if have lv2ls; then
miss=0
while IFS= read -r u; do
lv2ls | grep -qxF "$u" || { echo " missing plugin: $u"; miss=1; }
done < <(grep '"plugin"' "$MERGED" | grep -oE 'https?://[^"]+' | sort -u)
[ "$miss" -eq 0 ] && echo "ok: all LV2 plugins resolve" \
|| die "LV2 plugins missing - run ./install-deps.sh (INSTALL.md section 3)"
# check plugin URIs resolve
if have lv2ls && have jq; then
installed_lv2=$(lv2ls)
graph_lv2=$(jq -r '.["filter.graph"].nodes[] | select(.type=="lv2") | .plugin' "$MERGED")
for uri in $graph_lv2; do
echo "$installed_lv2" | grep -Fqx "$uri" || die \
"LV2 plugin URI missing: $uri (check lilv-utils/installed plugins)"
done
echo "ok: all LV2 plugins resolve"
else
echo "warn: lv2ls not found - cannot verify plugins (install lilv-utils / lilv)"
fi
fi
# --- preserve master volume --------------------------------------------
# --- preserve master volume (only needed for full WirePlumber restart) --
SAVED_VOL=""
IS_MUTED=0
if have wpctl; then
if [ "$HOT" -eq 0 ] && have wpctl; then
VOL_OUT="$(wpctl get-volume @DEFAULT_AUDIO_SINK@ 2>/dev/null || true)"
if [ -n "$VOL_OUT" ]; then
SAVED_VOL="$(echo "$VOL_OUT" | awk '{print $2}')"
@@ -113,21 +117,64 @@ fi
# --- install ----------------------------------------------------------
if [ "$SIMPLE" -eq 1 ]; then
echo "Installing baked FIR files -> $(dirname "$GRAPH_DST")/"
sudo cp "$SCRIPT_DIR/15_1/baked-"*.wav "$(dirname "$GRAPH_DST")/"
for f in "$SCRIPT_DIR/15_1/baked-"*.wav; do
bn="$(basename "$f")"
sudo cp "$f" "$(dirname "$GRAPH_DST")/$bn.tmp"
sudo mv -f "$(dirname "$GRAPH_DST")/$bn.tmp" "$(dirname "$GRAPH_DST")/$bn"
done
fi
echo "Installing $MERGED -> $GRAPH_DST"
sudo cp "$MERGED" "$GRAPH_DST"
sudo cp "$MERGED" "$GRAPH_DST.tmp"
sudo mv -f "$GRAPH_DST.tmp" "$GRAPH_DST"
if [ "$HOT" -eq 1 ]; then
echo "Hot-reloading WirePlumber graph (zero audio drop)..."
if systemctl --user reload wireplumber 2>/dev/null || pkill -HUP -f wireplumber 2>/dev/null; then
echo "ok: sent SIGHUP live reload to WirePlumber"
sleep 0.4
else
echo "warn: SIGHUP failed - restarting WirePlumber fallback"
systemctl --user restart wireplumber
sleep 1.2
fi
# Ensure output is unmuted after live reload
if [ -n "$SAVED_VOL" ] && have wpctl; then
wpctl set-mute @DEFAULT_AUDIO_SINK@ 0 2>/dev/null || true
wpctl set-volume @DEFAULT_AUDIO_SINK@ "$SAVED_VOL" 2>/dev/null || true
fi
echo "Done - FIR graph reloaded live."
exit 0
fi
# --- full WirePlumber restart -----------------------------------------
echo "Restarting WirePlumber"
systemctl --user restart wireplumber
sleep 1.2
# --- wait for DSP sink to be active & restore master volume ------------
for i in {1..6}; do
if have wpctl && wpctl status 2>/dev/null | grep -qi "DSP Speakers"; then
break
fi
sleep 0.5
done
# --- restore master volume ---------------------------------------------
sleep 1
if [ -n "$SAVED_VOL" ] && have wpctl; then
# Force unmute on default sink + explicit DSP Speakers sink ID
wpctl set-mute @DEFAULT_AUDIO_SINK@ 0 2>/dev/null || true
DSP_ID="$(wpctl status 2>/dev/null | grep -i "DSP Speakers" | grep -oE '[0-9]+\.' | head -n1 | tr -d '.')"
if [ -n "$DSP_ID" ]; then
wpctl set-mute "$DSP_ID" 0 2>/dev/null || true
wpctl set-volume "$DSP_ID" "$SAVED_VOL" 2>/dev/null || true
fi
wpctl set-volume @DEFAULT_AUDIO_SINK@ "$SAVED_VOL" 2>/dev/null || true
if [ "$IS_MUTED" -eq 1 ]; then
wpctl set-mute @DEFAULT_AUDIO_SINK@ 1 2>/dev/null || true
[ -n "$DSP_ID" ] && wpctl set-mute "$DSP_ID" 1 2>/dev/null || true
fi
echo "ok: preserved master volume (${SAVED_VOL})"
fi
@@ -136,6 +183,6 @@ fi
if have wpctl && wpctl status 2>/dev/null | grep -qi "DSP Speakers"; then
echo "Done - 'MacBook Pro 15,1 DSP Speakers' sink is up."
else
echo "Done - graph installed, WirePlumber restarted."
echo "Done - graph installed, WirePlumber reloaded."
echo "If no 'DSP Speakers' sink shows: journalctl --user -u wireplumber -b -e"
fi

133
bake-graph.py
View File

@@ -209,13 +209,6 @@ def optimize_fir_latency_and_tail(samples, fs=48000, is_woofer=False):
fade = 0.5 * (1.0 - math.cos(math.pi * i / max(fade_in_len, 1)))
cropped[i] *= fade
# Energy compensation: Scale post-lead tail energy to preserve 100% total acoustic energy
cropped_energy = sum(s*s for s in samples[start_idx:])
if cropped_energy > 0 and total_energy > 0:
boost_factor = math.sqrt(total_energy / cropped_energy)
for i in range(lead_len, len(cropped)):
cropped[i] *= boost_factor
# 2. Lopsided Tail Extension: 16,384 taps for woofers, 8,192 taps for tweeters
target_len = 16384 if is_woofer else 8192
if len(cropped) < target_len:
@@ -233,117 +226,17 @@ def optimize_fir_latency_and_tail(samples, fs=48000, is_woofer=False):
return cropped
def bake_driver_ir(src_wav, dst_wav, is_woofer=False, hp_freq=180.0, driver_gain=1.0):
def bake_driver_ir(src_wav, dst_wav, is_woofer=False, driver_gain=1.0):
if not os.path.exists(src_wav):
print(f"Warning: {src_wav} not found, skipping.")
return False
samples, fs = read_wav_floats(src_wav)
# 1. Apply Driver Gain Multiplier (e.g., 1.1 for tweeters, 1.2 for woofers)
if driver_gain != 1.0:
samples = [s * driver_gain for s in samples]
# 2. Apply Crossover Filter if Woofer
if is_woofer:
b0, b1, b2, a0, a1, a2 = biquad_highpass(fs, hp_freq)
samples = process_biquad(samples, b0, b1, b2, a0, a1, a2)
samples = process_biquad(samples, b0, b1, b2, a0, a1, a2) # LR4 (2-stage)
# 2.5 Apply Static System Voicing EQ ("equalizer" node from graph.json: +8dB @ 31.5Hz, +7dB @ 50Hz, +6dB @ 80Hz)
graph_path = os.path.join(SCRIPT_DIR, "graph.json")
if os.path.exists(graph_path):
try:
with open(graph_path, 'r') as f:
graph = json.load(f)
nodes = graph.get("filter.graph", {}).get("nodes", [])
for node in nodes:
if node.get("name") == "equalizer":
ctrl = node.get("control", {})
if ctrl.get("enabled", 1) == 1:
for i in range(16):
f_key = f"f_{i}"
g_key = f"g_{i}"
q_key = f"q_{i}"
ft_key = f"ft_{i}"
if f_key in ctrl and g_key in ctrl:
f0 = ctrl[f_key]
gain = ctrl[g_key]
q = ctrl.get(q_key, 1.41)
ft = ctrl.get(ft_key, 1)
gain_db = 20.0 * math.log10(max(gain, 0.001))
if ft == 5:
b0, b1, b2, a0, a1, a2 = biquad_lowshelf(fs, f0, gain_db, q)
elif ft == 3:
b0, b1, b2, a0, a1, a2 = biquad_highshelf(fs, f0, gain_db, q)
elif ft == 2: # High-Pass Filter (Low-Cut)
b0, b1, b2, a0, a1, a2 = biquad_highpass(fs, f0, q)
elif ft == 4: # Low-Pass Filter (High-Cut)
b0, b1, b2, a0, a1, a2 = biquad_lowpass(fs, f0, q)
else:
b0, b1, b2, a0, a1, a2 = biquad_peaking(fs, f0, gain_db, q)
samples = process_biquad(samples, b0, b1, b2, a0, a1, a2)
except Exception as e:
print(f"System Voicing Equalizer note: {e}")
# 2.7 Apply Effective User EQ (user_eq.json override if present, else default user_eq from graph.json)
ueq = None
user_eq_path = os.path.join(SCRIPT_DIR, "user_eq.json")
if os.path.exists(user_eq_path):
try:
with open(user_eq_path, 'r') as f:
ueq = json.load(f)
except Exception as e:
print(f"Error reading user_eq.json: {e}")
if ueq is None and os.path.exists(graph_path):
try:
with open(graph_path, 'r') as f:
g = json.load(f)
for n in g.get("filter.graph", {}).get("nodes", []):
if n.get("name") == "user_eq":
ueq = n.get("control", {})
break
except Exception as e:
print(f"Error reading default user_eq from graph.json: {e}")
if ueq and ueq.get("enabled", 1) == 1:
try:
g_out = ueq.get("g_out", 1.0)
if g_out != 1.0:
samples = [s * g_out for s in samples]
for i in range(8):
f_key = f"f_{i}"
g_key = f"g_{i}"
q_key = f"q_{i}"
ft_key = f"ft_{i}"
if f_key in ueq and g_key in ueq:
f0 = ueq[f_key]
gain = ueq[g_key]
q = ueq.get(q_key, 1.0)
ft = ueq.get(ft_key, 1)
gain_db = 20.0 * math.log10(max(gain, 0.001))
if ft == 5: # Low Shelf
b0, b1, b2, a0, a1, a2 = biquad_lowshelf(fs, f0, gain_db, q)
elif ft == 3: # High Shelf
b0, b1, b2, a0, a1, a2 = biquad_highshelf(fs, f0, gain_db, q)
elif ft == 2: # High-Pass
b0, b1, b2, a0, a1, a2 = biquad_highpass(fs, f0, q)
elif ft == 4: # Low-Pass
b0, b1, b2, a0, a1, a2 = biquad_lowpass(fs, f0, q)
else: # Peaking EQ
b0, b1, b2, a0, a1, a2 = biquad_peaking(fs, f0, gain_db, q)
samples = process_biquad(samples, b0, b1, b2, a0, a1, a2)
except Exception as e:
print(f"User EQ processing note: {e}")
# 3. Optimize Latency (5ms Lead) + Extend Woofer/Tweeter Lopsided Tail Resolution
# 1. Optimize Latency (5ms Lead) + Extend Woofer/Tweeter Lopsided Tail Resolution
samples = optimize_fir_latency_and_tail(samples, fs=fs, is_woofer=is_woofer)
# 4. True-Peak Inter-Sample Peak (ISP) Guarding (-0.5 dBFS ceiling)
# 3. True-Peak Inter-Sample Peak (ISP) Guarding (-0.5 dBFS ceiling)
samples = apply_true_peak_guard(samples, max_allowed_dbfs=-0.5)
write_wav_floats(dst_wav, samples, fs)
@@ -407,18 +300,18 @@ def generate_simple_graph_and_bake():
src_wav=src_path,
dst_wav=task["repo_dst"],
is_woofer=task["is_woofer"],
hp_freq=180.0,
driver_gain=task["gain"]
)
# 3. Build graph_simple.json dynamically from graph.json (omitting user_eq, equalizer, limiter, ell, elr)
graph["node.description"] = "MacBook Pro 15,1 DSP Speakers (Single-Stage Baked FIR)"
# 3. Build graph_simple.json dynamically from graph.json (omitting limiter, ell, elr, whp*)
# Keeping user_eq, equalizer, virtualbass, multiband_compressor in exact order for 100% bit-exact bass response!
graph["node.description"] = "MacBook Pro 15,1 DSP Speakers (Baked FIR Crossovers & Latency Trimming)"
new_nodes = []
for node in nodes:
name = node.get("name", "")
# Omit user_eq, static system voicing equalizer, redundant master limiter, ell/elr mono nodes, & crossover biquad nodes
if name in ["user_eq", "equalizer", "limiter", "ell", "elr", "whpL1", "whpL2", "whpR1", "whpR2"]:
# Omit redundant master limiter, ell/elr mono nodes, & crossover biquad nodes
if name in ["limiter", "ell", "elr", "whpL1", "whpL2", "whpR1", "whpR2"]:
continue
if node.get("label") == "convolver" or "conv" in name:
@@ -442,15 +335,13 @@ def generate_simple_graph_and_bake():
}
})
# Re-wire links: filter out user_eq, equalizer, limiter, ell, elr & whp*
# Re-wire links: filter out limiter, ell, elr & whp*
links = graph.get("filter.graph", {}).get("links", [])
new_links = []
for link in links:
out_node = link.get("output", "")
in_node = link.get("input", "")
if ("whp" in out_node or "whp" in in_node or
"equalizer" in out_node or "equalizer" in in_node or
"user_eq" in out_node or "user_eq" in in_node or
"limiter:" in out_node or "limiter:" in in_node or
"ell:" in out_node or "ell:" in in_node or
"elr:" in out_node or "elr:" in in_node):
@@ -463,10 +354,10 @@ def generate_simple_graph_and_bake():
new_links.append({"output": "loudness:out_l", "input": "copyL:In"})
new_links.append({"output": "loudness:out_r", "input": "copyR:In"})
# Set graph inputs directly to virtualbass (first active DSP processing node)
# Set graph inputs directly to user_eq (first node in processing chain)
graph["filter.graph"]["inputs"] = [
"virtualbass:in_l",
"virtualbass:in_r"
"user_eq:in_l",
"user_eq:in_r"
]
# Consolidated volume tracking for stereo loudness node

27
eq.py
View File

@@ -90,8 +90,8 @@ def save_and_apply(eq_data):
with open(USER_EQ_PATH, 'w') as f:
json.dump(eq_data, f, indent=4)
print(f"Saved {USER_EQ_PATH}")
print("Baking FIR filters and applying to PipeWire...")
subprocess.run([os.path.join(SCRIPT_DIR, "apply.sh"), "--bake"])
print("Baking FIR filters & hot-reloading into PipeWire (zero audio drop)...")
subprocess.run([os.path.join(SCRIPT_DIR, "apply.sh"), "--bake", "--hot"])
def show_status(eq_data):
print("=================================================================")
@@ -185,11 +185,26 @@ def main():
print(f"ok: Set Treble gain to {val_db:+.1f} dB ({gain_mult:.3f}x)")
elif cmd == "gain":
if len(args) < 2:
print("Usage: ./eq.py gain <multiplier> (e.g., ./eq.py gain 2.0)")
print("Usage: ./eq.py gain <+dB / -dB or multiplier> (e.g., ./eq.py gain -20 or ./eq.py gain 1.5)")
sys.exit(1)
eq["g_out"] = float(args[1])
save_and_apply(eq)
print(f"ok: Set Master Output Gain to {eq['g_out']:.2f}x")
raw_val = args[1].lower().replace("x", "").replace("db", "")
val = float(raw_val)
if val <= 0 and not raw_val.startswith("+"):
# Negative number passed (e.g. -20 or -100) -> Treat as dB attenuation
gain_mult = 10.0 ** (val / 20.0)
eq["g_out"] = round(gain_mult, 5)
save_and_apply(eq)
print(f"ok: Set Master Output Gain to {val:+.1f} dB ({gain_mult:.5f}x multiplier)")
else:
# Positive linear multiplier or positive dB
if "+" in raw_val:
gain_mult = 10.0 ** (val / 20.0)
eq["g_out"] = round(gain_mult, 3)
print(f"ok: Set Master Output Gain to {val:+.1f} dB ({gain_mult:.3f}x multiplier)")
else:
eq["g_out"] = round(val, 3)
print(f"ok: Set Master Output Gain to {eq['g_out']:.2f}x multiplier")
save_and_apply(eq)
elif cmd in ["enable", "disable"]:
eq["enabled"] = 1 if cmd == "enable" else 0
save_and_apply(eq)

12
graph.json
View File

@@ -206,12 +206,12 @@
}
],
"links": [
{"output": "user_eq:out_l", "input": "equalizer:in_l"},
{"output": "user_eq:out_r", "input": "equalizer:in_r"},
{"output": "equalizer:out_l", "input": "virtualbass:in_l"},
{"output": "equalizer:out_r", "input": "virtualbass:in_r"},
{"output": "virtualbass:out_l", "input": "multiband_compressor:in_l"},
{"output": "virtualbass:out_r", "input": "multiband_compressor:in_r"},
{"output": "user_eq:out_l", "input": "virtualbass:in_l"},
{"output": "user_eq:out_r", "input": "virtualbass:in_r"},
{"output": "virtualbass:out_l", "input": "equalizer:in_l"},
{"output": "virtualbass:out_r", "input": "equalizer:in_r"},
{"output": "equalizer:out_l", "input": "multiband_compressor:in_l"},
{"output": "equalizer:out_r", "input": "multiband_compressor:in_r"},
{"output": "multiband_compressor:out_l", "input": "limiter:in_1"},
{"output": "multiband_compressor:out_r", "input": "limiter:in_2"},
{"output": "limiter:out_1", "input": "ell:in"},