feat: complete single-stage FIR baking with dynamic graph.json parsing, 5ms lead trimming, and master limiter removal

This commit is contained in:
mynameisdeleted
2026-09-01 09:04:25 -04:00
parent f2a467996a
commit d22bbfd325
3 changed files with 268 additions and 58 deletions

View File

@@ -31,6 +31,8 @@ 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"
GRAPH_SRC="$SCRIPT_DIR/graph_simple.json"
echo "==> Installing baked FIR files -> /usr/share/t2-linux-audio/15_1/"
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; }
@@ -58,14 +60,14 @@ elif [ -f "$OVERRIDE" ]; then
have jq || die "$OVERRIDE exists but jq is not installed"
jq -e . "$OVERRIDE" >/dev/null 2>&1 || die "$OVERRIDE is not valid JSON"
jq -e 'any(.["filter.graph"].nodes[]; .name == "user_eq")' "$GRAPH_SRC" >/dev/null \
|| die "graph.json has no node named user_eq to override"
|| die "$GRAPH_SRC has no node named user_eq to override"
jq --slurpfile ov "$OVERRIDE" \
'(.["filter.graph"].nodes[] | select(.name == "user_eq") | .control) = $ov[0]' \
"$GRAPH_SRC" > "$MERGED" || die "jq merge failed"
echo "ok: merged user_eq.json -> $MERGED"
else
cp "$GRAPH_SRC" "$MERGED"
echo "ok: no user_eq.json - graph.json as-is -> $MERGED"
echo "ok: no user_eq.json - graph as-is -> $MERGED"
fi
json_ok "$MERGED"; rc=$?
@@ -96,6 +98,11 @@ if [ "$FORCE" -eq 0 ]; then
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")/"
fi
echo "Installing $MERGED -> $GRAPH_DST"
sudo cp "$MERGED" "$GRAPH_DST"

View File

@@ -46,6 +46,18 @@ def biquad_highpass(fs, f0, q=0.7071):
a2 = 1.0 - alpha
return b0/a0, b1/a0, b2/a0, 1.0, a1/a0, a2/a0
def biquad_lowpass(fs, f0, q=0.7071):
w0 = 2.0 * math.pi * f0 / fs
alpha = math.sin(w0) / (2.0 * q)
cos_w0 = math.cos(w0)
b0 = (1.0 - cos_w0) / 2.0
b1 = 1.0 - cos_w0
b2 = (1.0 - cos_w0) / 2.0
a0 = 1.0 + alpha
a1 = -2.0 * cos_w0
a2 = 1.0 - alpha
return b0/a0, b1/a0, b2/a0, 1.0, a1/a0, a2/a0
def biquad_lowshelf(fs, f0, gain_db, q=0.7071):
if gain_db == 0.0:
return 1.0, 0.0, 0.0, 1.0, 0.0, 0.0
@@ -63,6 +75,23 @@ def biquad_lowshelf(fs, f0, gain_db, q=0.7071):
a2 = (A + 1.0) + (A - 1.0) * cos_w0 - beta * math.sin(w0)
return b0/a0, b1/a0, b2/a0, 1.0, a1/a0, a2/a0
def biquad_highshelf(fs, f0, gain_db, q=0.7071):
if gain_db == 0.0:
return 1.0, 0.0, 0.0, 1.0, 0.0, 0.0
A = 10.0 ** (gain_db / 40.0)
w0 = 2.0 * math.pi * f0 / fs
alpha = math.sin(w0) / (2.0 * q)
cos_w0 = math.cos(w0)
beta = math.sqrt(A) / q
b0 = A * ((A + 1.0) + (A - 1.0) * cos_w0 + beta * math.sin(w0))
b1 = -2.0 * A * ((A - 1.0) + (A + 1.0) * cos_w0)
b2 = A * ((A + 1.0) + (A - 1.0) * cos_w0 - beta * math.sin(w0))
a0 = (A + 1.0) - (A - 1.0) * cos_w0 + beta * math.sin(w0)
a1 = 2.0 * ((A - 1.0) - (A + 1.0) * cos_w0)
a2 = (A + 1.0) - (A - 1.0) * cos_w0 - beta * math.sin(w0)
return b0/a0, b1/a0, b2/a0, 1.0, a1/a0, a2/a0
def process_biquad(samples, b0, b1, b2, a0, a1, a2):
out = [0.0] * len(samples)
x1 = x2 = y1 = y2 = 0.0
@@ -161,7 +190,7 @@ def optimize_fir_latency_and_tail(samples, fs=48000, is_woofer=False):
cropped[i] *= fade
# Energy compensation: Scale post-lead tail energy to preserve 100% total acoustic energy
cropped_energy = sum(s*s for s in cropped)
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)):
@@ -201,84 +230,185 @@ def bake_driver_ir(src_wav, dst_wav, is_woofer=False, hp_freq=180.0, driver_gain
samples = process_biquad(samples, b0, b1, b2, a0, a1, a2)
samples = process_biquad(samples, b0, b1, b2, a0, a1, a2) # LR4 (2-stage)
# 3. Apply User EQ Boosts (if user_eq.json exists)
# 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 User EQ Boosts ("user_eq.json" if present and enabled)
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)
if ueq.get("enabled", 1) == 1:
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))
b0, b1, b2, a0, a1, a2 = biquad_peaking(fs, f0, gain_db, q)
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}")
# 4. Optimize Latency (5ms Lead) + Extend Woofer/Tweeter Lopsided Tail Resolution
# 3. Optimize Latency (5ms Lead) + Extend Woofer/Tweeter Lopsided Tail Resolution
samples = optimize_fir_latency_and_tail(samples, fs=fs, is_woofer=is_woofer)
write_wav_floats(dst_wav, samples, fs)
print(f"==> Baked {os.path.basename(dst_wav)} ({fs} Hz, {len(samples)} taps, gain={driver_gain}x)")
return True
def generate_simple_graph():
def generate_simple_graph_and_bake():
graph_path = os.path.join(SCRIPT_DIR, "graph.json")
simple_graph_path = os.path.join(SCRIPT_DIR, "graph_simple.json")
if not os.path.exists(graph_path):
print("graph.json not found.")
return
print(f"Error: {graph_path} not found.")
sys.exit(1)
with open(graph_path, 'r') as f:
graph = json.load(f)
# Update description
graph["node.description"] = "MacBook Pro 15,1 DSP Speakers (Single-Stage Baked FIR)"
repo_151 = os.path.join(SCRIPT_DIR, "15_1")
sys_dir = "/usr/share/t2-linux-audio/15_1"
os.makedirs(repo_151, exist_ok=True)
# Replace Convolver filenames in nodes to point to baked WAVs
nodes = graph.get("filter.graph", {}).get("nodes", [])
# 1. Discover all convolver nodes and their input WAV files dynamically from graph.json
convolver_tasks = {} # maps src_filename -> {is_woofer, gain, sys_dst_path, repo_dst_path}
for node in nodes:
if node.get("label") == "convolver" or "conv" in node.get("name", ""):
name = node.get("name", "")
config = node.get("config", {})
gain = config.get("gain", 1.0)
filenames = config.get("filename", [])
is_woofer = ("woofer" in name.lower() or "convlw" in name.lower() or "convrw" in name.lower())
for sys_path in filenames:
basename = os.path.basename(sys_path)
if not basename in convolver_tasks:
if "woofer" in basename.lower():
is_woofer = True
baked_basename = "baked-" + basename
repo_dst_path = os.path.join(repo_151, baked_basename)
sys_dst_path = os.path.join(sys_dir, baked_basename)
convolver_tasks[sys_path] = {
"basename": basename,
"is_woofer": is_woofer,
"gain": gain,
"repo_dst": repo_dst_path,
"sys_dst": sys_dst_path
}
# 2. Bake FIR files dynamically for all discovered WAV targets
for sys_path, task in convolver_tasks.items():
basename = task["basename"]
src_path = os.path.join(repo_151, basename)
if not os.path.exists(src_path) and os.path.exists(sys_path):
src_path = sys_path
if not os.path.exists(src_path) and os.path.exists(os.path.join(SCRIPT_DIR, basename)):
src_path = os.path.join(SCRIPT_DIR, basename)
bake_driver_ir(
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, whp*)
graph["node.description"] = "MacBook Pro 15,1 DSP Speakers (Single-Stage Baked FIR)"
new_nodes = []
for node in nodes:
name = node.get("name", "")
# Omit static biquad EQ nodes that are now baked into FIR
if name in ["user_eq", "equalizer", "whpL1", "whpL2", "whpR1", "whpR2"]:
# Omit user_eq, static system voicing equalizer, redundant master limiter & crossover biquad nodes
if name in ["user_eq", "equalizer", "limiter", "whpL1", "whpL2", "whpR1", "whpR2"]:
continue
repo_151 = os.path.join(SCRIPT_DIR, "15_1")
if name in ["convLT", "convRT"]:
if node.get("label") == "convolver" or "conv" in name:
orig_filenames = node.get("config", {}).get("filename", [])
node["config"]["filename"] = [
os.path.join(repo_151, "baked-tweeters-44k.wav"),
os.path.join(repo_151, "baked-tweeters-48k.wav"),
os.path.join(repo_151, "baked-tweeters-96k.wav")
]
elif name in ["convLW", "convRW"]:
node["config"]["filename"] = [
os.path.join(repo_151, "baked-woofers-44k.wav"),
os.path.join(repo_151, "baked-woofers-48k.wav"),
os.path.join(repo_151, "baked-woofers-96k.wav")
convolver_tasks[p]["sys_dst"] if p in convolver_tasks else os.path.join(sys_dir, "baked-" + os.path.basename(p))
for p in orig_filenames
]
new_nodes.append(node)
# Re-wire links: filter out references to omitted nodes (user_eq, equalizer, whp*)
# Re-wire links: filter out user_eq, equalizer, limiter & 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 "user_eq" in out_node:
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):
continue
new_links.append(link)
# Set graph inputs to virtualbass (first remaining processing node)
# Wire multiband_compressor directly to ell and elr (bypassing redundant master limiter)
new_links.append({"output": "multiband_compressor:out_l", "input": "ell:in"})
new_links.append({"output": "multiband_compressor:out_r", "input": "elr:in"})
# Set graph inputs directly to virtualbass (first active DSP processing node)
graph["filter.graph"]["inputs"] = [
"virtualbass:in_l",
"virtualbass:in_r"
@@ -296,34 +426,7 @@ def main():
print("=================================================================")
print(" SINGLE-STAGE FIR CONVOLVER BAKER & GRAPH SIMPLIFIER")
print("=================================================================")
repo_151 = os.path.join(SCRIPT_DIR, "15_1")
sys_dir = "/usr/share/t2-linux-audio/15_1"
os.makedirs(repo_151, exist_ok=True)
rates = ["44k", "48k", "96k"]
for r in rates:
tw_name = f"tweeters-{r}.wav"
tw_src = os.path.join(repo_151, tw_name)
if not os.path.exists(tw_src) and os.path.exists(os.path.join(sys_dir, tw_name)):
tw_src = os.path.join(sys_dir, tw_name)
if not os.path.exists(tw_src) and os.path.exists(os.path.join(SCRIPT_DIR, tw_name)):
tw_src = os.path.join(SCRIPT_DIR, tw_name)
tw_dst = os.path.join(repo_151, f"baked-tweeters-{r}.wav")
bake_driver_ir(tw_src, tw_dst, is_woofer=False, driver_gain=1.1)
wf_name = f"woofers-{r}.wav"
wf_src = os.path.join(repo_151, wf_name)
if not os.path.exists(wf_src) and os.path.exists(os.path.join(sys_dir, wf_name)):
wf_src = os.path.join(sys_dir, wf_name)
if not os.path.exists(wf_src) and os.path.exists(os.path.join(SCRIPT_DIR, tw_name)):
wf_src = os.path.join(SCRIPT_DIR, wf_name)
wf_dst = os.path.join(repo_151, f"baked-woofers-{r}.wav")
bake_driver_ir(wf_src, wf_dst, is_woofer=True, hp_freq=180.0, driver_gain=1.2)
generate_simple_graph()
generate_simple_graph_and_bake()
print("=================================================================")
print("Done! Baked FIR files & graph_simple.json created.")

View File

@@ -67,6 +67,64 @@ def biquad_highpass(fs, f0, q=0.7071):
a2 = 1.0 - alpha
return b0/a0, b1/a0, b2/a0, 1.0, a1/a0, a2/a0
def biquad_lowpass(fs, f0, q=0.7071):
w0 = 2.0 * math.pi * f0 / fs
alpha = math.sin(w0) / (2.0 * q)
cos_w0 = math.cos(w0)
b0 = (1.0 - cos_w0) / 2.0
b1 = 1.0 - cos_w0
b2 = (1.0 - cos_w0) / 2.0
a0 = 1.0 + alpha
a1 = -2.0 * cos_w0
a2 = 1.0 - alpha
return b0/a0, b1/a0, b2/a0, 1.0, a1/a0, a2/a0
def biquad_peaking(fs, f0, gain_db, q):
if gain_db == 0.0 or gain_db == 1.0:
return 1.0, 0.0, 0.0, 1.0, 0.0, 0.0
A = 10.0 ** (gain_db / 40.0)
w0 = 2.0 * math.pi * f0 / fs
alpha = math.sin(w0) / (2.0 * max(q, 0.01))
b0 = 1.0 + alpha * A
b1 = -2.0 * math.cos(w0)
b2 = 1.0 - alpha * A
a0 = 1.0 + alpha / A
a1 = -2.0 * math.cos(w0)
a2 = 1.0 - alpha / A
return b0/a0, b1/a0, b2/a0, 1.0, a1/a0, a2/a0
def biquad_lowshelf(fs, f0, gain_db, q=0.7071):
if gain_db == 0.0:
return 1.0, 0.0, 0.0, 1.0, 0.0, 0.0
A = 10.0 ** (gain_db / 40.0)
w0 = 2.0 * math.pi * f0 / fs
alpha = math.sin(w0) / (2.0 * q)
cos_w0 = math.cos(w0)
beta = math.sqrt(A) / q
b0 = A * ((A + 1.0) - (A - 1.0) * cos_w0 + beta * math.sin(w0))
b1 = 2.0 * A * ((A - 1.0) - (A + 1.0) * cos_w0)
b2 = A * ((A + 1.0) - (A - 1.0) * cos_w0 - beta * math.sin(w0))
a0 = (A + 1.0) + (A - 1.0) * cos_w0 + beta * math.sin(w0)
a1 = -2.0 * ((A - 1.0) + (A + 1.0) * cos_w0)
a2 = (A + 1.0) + (A - 1.0) * cos_w0 - beta * math.sin(w0)
return b0/a0, b1/a0, b2/a0, 1.0, a1/a0, a2/a0
def biquad_highshelf(fs, f0, gain_db, q=0.7071):
if gain_db == 0.0:
return 1.0, 0.0, 0.0, 1.0, 0.0, 0.0
A = 10.0 ** (gain_db / 40.0)
w0 = 2.0 * math.pi * f0 / fs
alpha = math.sin(w0) / (2.0 * q)
cos_w0 = math.cos(w0)
beta = math.sqrt(A) / q
b0 = A * ((A + 1.0) + (A - 1.0) * cos_w0 + beta * math.sin(w0))
b1 = -2.0 * A * ((A - 1.0) + (A + 1.0) * cos_w0)
b2 = A * ((A + 1.0) + (A - 1.0) * cos_w0 - beta * math.sin(w0))
a0 = (A + 1.0) - (A - 1.0) * cos_w0 + beta * math.sin(w0)
a1 = 2.0 * ((A - 1.0) - (A + 1.0) * cos_w0)
a2 = (A + 1.0) - (A - 1.0) * cos_w0 - beta * math.sin(w0)
return b0/a0, b1/a0, b2/a0, 1.0, a1/a0, a2/a0
def process_biquad(samples, b0, b1, b2, a0, a1, a2):
out = [0.0] * len(samples)
x1 = x2 = y1 = y2 = 0.0
@@ -104,11 +162,53 @@ def main():
# Load baseline woofer IR
orig_ir, fs = read_wav_floats(orig_path)
# Path A: Cascaded Filter Chain (Baseline IR + 180 Hz Crossover High-Pass Biquads)
# Path A: Cascaded Filter Chain (Baseline IR + System Voicing EQ + User EQ + 180 Hz Crossover Biquads)
b0, b1, b2, a0, a1, a2 = biquad_highpass(fs, 180.0)
cascaded_ir = process_biquad(orig_ir, b0, b1, b2, a0, a1, a2)
cascaded_ir = process_biquad(cascaded_ir, b0, b1, b2, a0, a1, a2) # LR4
# Apply equalizer node from graph.json to Path A
graph_path = os.path.join(SCRIPT_DIR, "graph.json")
if os.path.exists(graph_path):
with open(graph_path, 'r') as f:
graph = json.load(f)
for node in graph.get("filter.graph", {}).get("nodes", []):
if node.get("name") == "equalizer":
ctrl = node.get("control", {})
if ctrl.get("enabled", 1) == 1:
g_in = ctrl.get("g_in", 1.0)
g_out = ctrl.get("g_out", 1.0)
if g_in != 1.0: cascaded_ir = [s * g_in for s in cascaded_ir]
if g_out != 1.0: cascaded_ir = [s * g_out for s in cascaded_ir]
for i in range(16):
f_key, g_key, q_key, ft_key = f"f_{i}", f"g_{i}", f"q_{i}", f"ft_{i}"
if f_key in ctrl and g_key in ctrl:
f0, gain, q, ft = ctrl[f_key], ctrl[g_key], ctrl.get(q_key, 1.41), 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: 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)
cascaded_ir = process_biquad(cascaded_ir, b0, b1, b2, a0, a1, a2)
# Apply user_eq.json to Path A
user_eq_path = os.path.join(SCRIPT_DIR, "user_eq.json")
if os.path.exists(user_eq_path):
with open(user_eq_path, 'r') as f:
ueq = json.load(f)
if ueq.get("enabled", 1) == 1:
g_out = ueq.get("g_out", 1.0)
if g_out != 1.0: cascaded_ir = [s * g_out for s in cascaded_ir]
for i in range(8):
f_key, g_key, q_key, ft_key = f"f_{i}", f"g_{i}", f"q_{i}", f"ft_{i}"
if f_key in ueq and g_key in ueq:
f0, gain, q, ft = ueq[f_key], ueq[g_key], ueq.get(q_key, 1.0), ueq.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)
else: b0, b1, b2, a0, a1, a2 = biquad_peaking(fs, f0, gain_db, q)
cascaded_ir = process_biquad(cascaded_ir, b0, b1, b2, a0, a1, a2)
# Path B: Single-Stage Baked FIR
baked_ir, _ = read_wav_floats(baked_path)