194 lines
7.2 KiB
Python
194 lines
7.2 KiB
Python
"""P0 golden conformance harness (Python oracle) - SonicForgeStudio.
|
|
|
|
Muc dich: sinh manifest hash cho cac ham DSP thuan (deterministic) lam CHUAN
|
|
DOI CHIEU khi port sang Rust (P1-P3). Chay:
|
|
|
|
python tools/golden_conformance.py --write # ghi manifest
|
|
python tools/golden_conformance.py --check # so voi manifest da ghi
|
|
|
|
Moi case tra ve bytes canonical (so thuc lam tron 6 chu so thap phan) -> sha256.
|
|
Cung input (fixture sinh bang seed co dinh trong code) + cung thuat toan =>
|
|
cung hash; lech => in ro case nao.
|
|
"""
|
|
import argparse
|
|
import hashlib
|
|
import io
|
|
import json
|
|
import os
|
|
import sys
|
|
import tempfile
|
|
|
|
import numpy as np
|
|
|
|
ROOT = os.path.abspath(os.path.join(os.path.dirname(__file__), ".."))
|
|
sys.path.insert(0, ROOT)
|
|
|
|
MANIFEST = os.path.join(ROOT, "md", "api_contract", "golden", "manifest.json")
|
|
SR = 48000
|
|
|
|
|
|
# ── fixtures (seed co dinh, khong dung file ngoai) ───────────────────────────
|
|
|
|
def _tone(freq, dur, sr=SR, amp=0.5):
|
|
t = np.linspace(0, dur, int(sr * dur), endpoint=False)
|
|
return (amp * np.sin(2 * np.pi * freq * t)).astype(np.float32)
|
|
|
|
|
|
def _fixture_stereo():
|
|
rng = np.random.default_rng(1234)
|
|
n = SR * 2
|
|
t = np.arange(n) / SR
|
|
left = (0.4 * np.sin(2 * np.pi * 220 * t) + 0.05 * rng.standard_normal(n)).astype(np.float32)
|
|
right = (0.4 * np.sin(2 * np.pi * 330 * t + 0.7) + 0.05 * rng.standard_normal(n)).astype(np.float32)
|
|
return np.column_stack([left, right]).astype(np.float32)
|
|
|
|
|
|
def _fixture_click_track():
|
|
"""Click track 120 BPM de analyzer/beat khong bi suy bien."""
|
|
n = SR * 4
|
|
y = np.zeros(n, dtype=np.float32)
|
|
step = SR // 2 # 120 BPM
|
|
for i in range(0, n, step):
|
|
y[i:i + 64] = 0.8
|
|
return y
|
|
|
|
|
|
def _write_wav(dirpath, name, data, sr=SR):
|
|
import soundfile as sf
|
|
p = os.path.join(dirpath, name)
|
|
sf.write(p, data, sr, subtype="FLOAT")
|
|
return p
|
|
|
|
|
|
# ── canonical hoa + hash ─────────────────────────────────────────────────────
|
|
|
|
def _canon(obj):
|
|
if isinstance(obj, np.ndarray):
|
|
a = np.asarray(obj, dtype=np.float64)
|
|
return json.dumps({"dtype": str(obj.dtype), "shape": list(a.shape),
|
|
"data": np.round(a.ravel(), 6).tolist()},
|
|
sort_keys=True)
|
|
if isinstance(obj, dict):
|
|
return json.dumps({k: _canon(v) for k, v in sorted(obj.items())}, sort_keys=True)
|
|
if isinstance(obj, (list, tuple)):
|
|
return json.dumps([_canon(v) for v in obj])
|
|
if isinstance(obj, (np.floating, float)):
|
|
return json.dumps(round(float(obj), 6))
|
|
if isinstance(obj, (np.integer, int)):
|
|
return json.dumps(int(obj))
|
|
if isinstance(obj, bytes):
|
|
return json.dumps(list(obj))
|
|
return json.dumps(obj, sort_keys=True)
|
|
|
|
|
|
def _sha(obj):
|
|
return hashlib.sha256(_canon(obj).encode("utf-8")).hexdigest()
|
|
|
|
|
|
# ── cac case ─────────────────────────────────────────────────────────────────
|
|
|
|
def collect():
|
|
cases = {}
|
|
import soundfile as sf
|
|
tmp = tempfile.mkdtemp(prefix="sf_golden_")
|
|
stereo = _fixture_stereo()
|
|
mono = _tone(440.0, 1.0)
|
|
p_stereo = _write_wav(tmp, "stereo.wav", stereo)
|
|
p_mono = _write_wav(tmp, "mono.wav", mono)
|
|
p_click = _write_wav(tmp, "click.wav", _fixture_click_track())
|
|
|
|
from app.core import dsp_utils, audio_editor, analyzer, loudness
|
|
|
|
# dsp_utils
|
|
targets = [0.0, 0.123, 0.5, 1.37, 1.999]
|
|
cases["dsp_utils.find_zero_crossing"] = [
|
|
dsp_utils.find_zero_crossing(stereo[:, 0], SR, tt) for tt in targets]
|
|
|
|
orig = stereo[:, 0].copy()
|
|
edited = np.roll(orig, 37)
|
|
cases["dsp_utils.apply_micro_crossfade"] = dsp_utils.apply_micro_crossfade(
|
|
orig, edited, start_sample=SR // 2, fade_len_ms=10, sr=SR)
|
|
|
|
cases["dsp_utils.generate_peak_waveform"] = dsp_utils.generate_peak_waveform(p_stereo, 400)
|
|
cases["dsp_utils.generate_rms_waveform"] = dsp_utils.generate_rms_waveform(p_stereo, 400)
|
|
|
|
# audio_editor
|
|
seg = audio_editor.cut_and_loop_segment(
|
|
p_stereo, 0.25, 0.75, loop_count=2, fade_in_ms=20, fade_out_ms=20,
|
|
volume_db_change=-3.0)
|
|
cases["audio_editor.cut_and_loop_segment"] = {
|
|
"channels": seg.channels, "frame_rate": seg.frame_rate,
|
|
"sample_width": seg.sample_width, "raw": seg.raw_data}
|
|
|
|
# bit_depth=32 (IEEE float) — KHONG dither, bit-exact. 16/24-bit dung TPDF
|
|
# dither ngau nhien (np.random.default_rng()) nen output KHONG tai lap duoc.
|
|
out_wav = os.path.join(tmp, "mix.wav")
|
|
tracks_meta = [
|
|
{"file_path": p_stereo, "volume": 0.8, "muted": False},
|
|
{"file_path": p_mono, "volume": 0.6, "muted": False},
|
|
{"file_path": p_click, "volume": 0.4, "muted": True},
|
|
]
|
|
audio_editor.mix_multitrack_session(tracks_meta, out_wav, sample_rate=SR, bit_depth=32)
|
|
# Hash theo MAU audio (khong hash byte file): header WAV do pydub/ffmpeg ghi
|
|
# co metadata khong on dinh giua cac lan chay, du samples giong het nhau.
|
|
y_mix, sr_mix = sf.read(out_wav, dtype="float64")
|
|
cases["audio_editor.mix_multitrack_session"] = {"sample_rate": sr_mix, "audio": y_mix}
|
|
|
|
# analyzer
|
|
cases["analyzer.analyze_audio"] = analyzer.analyze_audio(p_click)
|
|
|
|
# loudness
|
|
cases["loudness.integrated_lufs"] = loudness.integrated_lufs(stereo, SR)
|
|
cases["loudness.true_peak_db"] = loudness.true_peak_db(stereo, SR)
|
|
cases["loudness.mono_side_ratio"] = loudness.mono_side_ratio(stereo)
|
|
return cases
|
|
|
|
|
|
def build_manifest():
|
|
import numpy as np
|
|
import scipy
|
|
cases = collect()
|
|
return {
|
|
"schema": 1,
|
|
"generated_by": "tools/golden_conformance.py",
|
|
"env": {"python": sys.version.split()[0], "numpy": np.__version__,
|
|
"scipy": scipy.__version__, "sample_rate": SR},
|
|
"cases": {k: _sha(v) for k, v in cases.items()},
|
|
}
|
|
|
|
|
|
def main():
|
|
ap = argparse.ArgumentParser()
|
|
ap.add_argument("--write", action="store_true")
|
|
ap.add_argument("--check", action="store_true")
|
|
args = ap.parse_args()
|
|
|
|
manifest = build_manifest()
|
|
if args.check:
|
|
with io.open(MANIFEST, "r", encoding="utf-8") as f:
|
|
ref = json.load(f)
|
|
bad = []
|
|
for k, h in ref["cases"].items():
|
|
cur = manifest["cases"].get(k)
|
|
if cur != h:
|
|
bad.append(k)
|
|
extra = [k for k in manifest["cases"] if k not in ref["cases"]]
|
|
for k in bad:
|
|
print("MISMATCH %s\n ref=%s\n cur=%s" % (k, ref["cases"][k], manifest["cases"][k]))
|
|
for k in extra:
|
|
print("NEW-CASE %s" % k)
|
|
print("cases=%d mismatch=%d new=%d" % (len(ref["cases"]), len(bad), len(extra)))
|
|
return 1 if bad else 0
|
|
|
|
os.makedirs(os.path.dirname(MANIFEST), exist_ok=True)
|
|
with io.open(MANIFEST, "w", encoding="utf-8", newline="\n") as f:
|
|
json.dump(manifest, f, indent=1, sort_keys=True)
|
|
print("WROTE %s (%d cases)" % (MANIFEST, len(manifest["cases"])))
|
|
for k, h in sorted(manifest["cases"].items()):
|
|
print(" %-45s %s" % (k, h[:16]))
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|