From 03636629dccd9682d7302e4f13ce401294a4081b Mon Sep 17 00:00:00 2001
From: firestar5683 <168790843+firestar5683@users.noreply.github.com>
Date: Sat, 19 Sep 2026 14:21:41 -0700
Subject: [PATCH] Compare revised openings at fixed gain with matched seed and
gold reference
---
roadscore/tests/test_audio_screening.py | 13 +++++++-
roadscore/tools/review_hook_audio.py | 41 ++++++++++++++++++++++++-
2 files changed, 52 insertions(+), 2 deletions(-)
diff --git a/roadscore/tests/test_audio_screening.py b/roadscore/tests/test_audio_screening.py
index 42392c954e..d05322af28 100644
--- a/roadscore/tests/test_audio_screening.py
+++ b/roadscore/tests/test_audio_screening.py
@@ -5,10 +5,21 @@ from pathlib import Path
import numpy as np
from scipy.io import wavfile
sys.path.insert(0, str(Path(__file__).resolve().parents[1] / 'tools'))
-from review_hook_audio import screen, assembly_check
+from review_hook_audio import screen, assembly_check, compare_initial
class AudioScreeningTests(unittest.TestCase):
+ def test_comparison_uses_matching_prefix_and_preserves_gain(self):
+ rate = 48000
+ wave = np.column_stack([np.sin(np.arange(rate)*2*np.pi*220/rate)]*2).astype('float32')
+ with tempfile.TemporaryDirectory() as folder:
+ current, reference = Path(folder)/'current.wav', Path(folder)/'reference.wav'
+ wavfile.write(current, rate, wave*.2)
+ wavfile.write(reference, rate, np.concatenate([wave*.1, wave]))
+ result = compare_initial(current, reference)
+ self.assertEqual(result['matched_start_seconds'], 1)
+ self.assertAlmostEqual(result['rms_change_db'], 6.0206, places=3)
+
def test_expected_crossfade_is_excluded_but_other_changes_are_reported(self):
parts = [np.zeros((40, 2)), np.ones((40, 2))]
core = np.concatenate(parts)
diff --git a/roadscore/tools/review_hook_audio.py b/roadscore/tools/review_hook_audio.py
index 4e493fc53a..c2191ffd20 100644
--- a/roadscore/tools/review_hook_audio.py
+++ b/roadscore/tools/review_hook_audio.py
@@ -3,11 +3,12 @@ import argparse
import hashlib
import html
import json
+import os
from pathlib import Path
import numpy as np
from scipy.io import wavfile
-from scipy.signal import resample_poly, stft
+from scipy.signal import resample_poly, stft, find_peaks
def load(path):
@@ -84,15 +85,42 @@ def assembly_check(pcm, parts, rate):
'limit': 'Saved windows omit regenerated prefixes. The expected 2s crossfade regions cannot be reconstructed from these files; float32 resume roundoff allowed up to 5e-7.'}
+def contrast(pcm, rate):
+ hop = max(1, rate // 100)
+ envelope = np.array([rms(pcm[i:i+hop]) for i in range(0, len(pcm)-hop+1, hop)])
+ onset = np.maximum(0, np.diff(envelope))
+ threshold = np.median(onset) + 3*np.median(abs(onset-np.median(onset)))
+ peaks, _ = find_peaks(onset, height=max(float(threshold), .001), distance=12)
+ return {'rms': rms(pcm), 'peak': float(np.max(abs(pcm))),
+ 'envelope_variation': float(envelope.std()/max(envelope.mean(), 1e-12)),
+ 'detected_onsets_per_second': float(len(peaks)/(len(pcm)/rate))}
+
+
+def compare_initial(current, other):
+ rate, pcm = load(current)
+ other_rate, original = load(other)
+ if rate != other_rate:
+ return {'error': 'Sample rates differ; comparison skipped'}
+ n = min(len(pcm), len(original))
+ a, b = contrast(pcm[:n], rate), contrast(original[:n], rate)
+ return {'matched_start_seconds': n/rate, 'current': a, 'reference': b,
+ 'rms_change_db': float(20*np.log10(max(a['rms'], 1e-12)/max(b['rms'], 1e-12))),
+ 'reference_sha256': hashlib.sha256(other.read_bytes()).hexdigest(),
+ 'limits': 'Fixed gain, same-length opening comparison. Onsets use 10ms RMS changes with a 120ms minimum separation; density and envelope variation do not establish groove or musical quality.'}
+
+
def main():
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument('directory', type=Path)
+ parser.add_argument('--baseline', type=Path)
+ parser.add_argument('--reference', type=Path)
args = parser.parse_args()
root = args.directory
validation = json.loads((root / 'validation.json').read_text())
report = {'scope': 'All sessions retained, original samples/gain unchanged; no listening judgment.',
'limits': 'Silence is channel-combined RMS below -80 dBFS in 100ms blocks. Repetition compares exact non-silent 1s blocks and zero-lag 4s waveform blocks; it does not measure memorable melody, shifted phrases, or musical quality. RMS/centroid section contrast can reflect level/timbre alone.',
'sessions': []}
+ baseline = json.loads((args.baseline / 'validation.json').read_text()) if args.baseline else None
cards = []
for number, seed in enumerate(validation['sessions'], 1):
folder = root / f'session_{number}'
@@ -114,17 +142,28 @@ def main():
entry['assembly'] = assembly_check(pcm, [wave for _, wave in parts], rate) if parts and all(r == rate for r, _ in parts) else {'status': 'sample rate mismatch or missing windows'}
seams = np.cumsum([len(wave) for _, wave in parts])[:-1]
entry['seams'] = [dict(seconds=float(i/rate), sample_step=float(np.max(abs(pcm[i]-pcm[i-1])))) for i in seams]
+ initial = folder / '00_initial.wav'
+ original = args.baseline / folder.name / '00_initial.wav' if args.baseline else None
+ if initial.exists():
+ if original and original.exists() and baseline['sessions'][number-1]['generation_seed'] == seed['generation_seed']:
+ entry['same_seed_initial_comparison'] = compare_initial(initial, original)
+ if args.reference:
+ entry['gold_initial_comparison'] = compare_initial(initial, args.reference)
report['sessions'].append(entry)
title = f'Session {number}' + (' · Complete' if status == 'technical_pass_listening_pending' else ' · Stopped early' if status == 'quality_failed' else ' · Partial' if core.exists() else ' · Pending')
core_label = 'Full session' if status == 'technical_pass_listening_pending' else 'Available audio (partial)'
choices = ([core] if core.exists() else []) + windows
options = ''.join(f'' for p in choices)
+ if original and original.exists() and 'same_seed_initial_comparison' in entry:
+ options += f''
player = (f''
if choices else 'Session audio is not available yet.
')
metrics = entry.get('core', {})
summary = (f'{metrics["seconds"]:.1f}s · peak {metrics["peak"]:.3f} · {metrics["samples_at_or_above_full_scale"]} full-scale samples'
if metrics and 'error' not in metrics else 'Screening pending')
cards.append(f'{title}
Seed {seed["generation_seed"]}
{player}{summary}
')
+ if args.reference:
+ cards.append(f'Protected GOLD reference
Original 28-second reference, unchanged gain.
')
(root / 'audio_screening.json').write_text(json.dumps(report, indent=2))
page = '''RoadScore · Fresh hook review