feedBack/tests/test_gp_audio_sync.py
2026-06-16 18:47:13 +02:00

287 lines
11 KiB
Python

"""Tests for lib/gp8_audio_sync.py and lib/gp_autosync.py — the GP8 embedded
audio / score-to-audio auto-sync helpers.
Fixture-free: every test drives a pure helper with hand-built inputs
(in-memory ZIP containers, ElementTree fragments, synthetic warping paths,
crafted tempo-event lists). The librosa-backed chroma/DTW stages need real
audio and are covered by manual validation in the PR; here we pin the timing
math, asset resolution, and sync-point extraction that are easy to drive
without a fixture.
"""
import io
import zipfile
import xml.etree.ElementTree as ET
import numpy as np
import pytest
import gp8_audio_sync as g8
from gp8_audio_sync import (
GpSyncData,
SyncPoint,
build_tempo_map_from_sync,
_parse_gpif,
_resolve_audio_asset,
)
import gp_autosync as ga
from gp_autosync import (
_gp345_tick_to_secs,
_gp345_tempo_events,
_get_tempo_map,
_get_initial_tempo,
_tempo_at_bar,
_safe_normalise,
_extract_sync_points,
)
# ── helpers ───────────────────────────────────────────────────────────────────
def _gpif_bytes(asset_id: str = "abc-123") -> bytes:
root = ET.Element("GPIF")
bt = ET.SubElement(root, "BackingTrack")
ET.SubElement(bt, "AssetId").text = asset_id
return ET.tostring(root)
def _make_gp_zip(asset_id="abc-123", ogg_stems=("abc-123",),
asset_ext=".ogg") -> zipfile.ZipFile:
buf = io.BytesIO()
with zipfile.ZipFile(buf, "w") as zf:
zf.writestr("Content/score.gpif", _gpif_bytes(asset_id))
for stem in ogg_stems:
zf.writestr(f"Content/Assets/{stem}{asset_ext}", b"fake-audio")
buf.seek(0)
return zipfile.ZipFile(buf)
def _fake_gpif_root(n_bars: int, tempo: float = 120.0, time_sig="4/4") -> ET.Element:
root = ET.Element("GPIF")
mt = ET.SubElement(root, "MasterTrack")
autos = ET.SubElement(mt, "Automations")
auto = ET.SubElement(autos, "Automation")
ET.SubElement(auto, "Type").text = "Tempo"
ET.SubElement(auto, "Bar").text = "0"
ET.SubElement(auto, "Value").text = str(tempo)
mbs = ET.SubElement(root, "MasterBars")
for _ in range(n_bars):
mb = ET.SubElement(mbs, "MasterBar")
ET.SubElement(mb, "Time").text = time_sig
return root
# ── GpSyncData.time_at_bar / tempo_at_bar ───────────────────────────────────────
def test_time_at_bar_interpolates_linearly_between_points():
sp = [
SyncPoint(bar=0, time_secs=0.0, modified_tempo=120, original_tempo=120),
SyncPoint(bar=10, time_secs=20.0, modified_tempo=120, original_tempo=120),
]
d = GpSyncData(audio_offset=0.0, audio_asset_id="", sync_points=sp)
assert d.time_at_bar(0) == pytest.approx(0.0)
assert d.time_at_bar(10) == pytest.approx(20.0)
assert d.time_at_bar(5) == pytest.approx(10.0) # exact halfway, any meter
assert d.time_at_bar(2) == pytest.approx(4.0)
def test_time_at_bar_interpolation_is_time_signature_agnostic():
# 3/4 material: linear interp between sync points must not assume 4 beats.
sp = [
SyncPoint(bar=0, time_secs=0.0, modified_tempo=90, original_tempo=90),
SyncPoint(bar=4, time_secs=8.0, modified_tempo=90, original_tempo=90),
]
d = GpSyncData(audio_offset=0.0, audio_asset_id="", sync_points=sp)
assert d.time_at_bar(1) == pytest.approx(2.0)
assert d.time_at_bar(3) == pytest.approx(6.0)
def test_time_at_bar_extrapolates_past_last_point_by_tempo():
sp = [SyncPoint(bar=0, time_secs=0.0, modified_tempo=120, original_tempo=120)]
d = GpSyncData(audio_offset=0.0, audio_asset_id="", sync_points=sp)
# 4 beats/bar at 120 BPM = 2.0 s/bar.
assert d.time_at_bar(3) == pytest.approx(6.0)
# 3/4 (beats_per_bar=3) at 120 BPM = 1.5 s/bar.
assert d.time_at_bar(2, beats_per_bar=3.0) == pytest.approx(3.0)
def test_time_at_bar_empty_returns_zero():
d = GpSyncData(audio_offset=0.0, audio_asset_id="", sync_points=[])
assert d.time_at_bar(5) == 0.0
def test_tempo_at_bar_holds_last_segment():
sp = [
SyncPoint(bar=0, time_secs=0.0, modified_tempo=100, original_tempo=100),
SyncPoint(bar=8, time_secs=10.0, modified_tempo=140, original_tempo=120),
]
d = GpSyncData(audio_offset=0.0, audio_asset_id="", sync_points=sp)
assert d.tempo_at_bar(0) == 100
assert d.tempo_at_bar(7) == 100
assert d.tempo_at_bar(8) == 140
assert d.tempo_at_bar(20) == 140
def test_build_tempo_map_from_sync():
sp = [
SyncPoint(bar=0, time_secs=0.0, modified_tempo=120, original_tempo=120),
SyncPoint(bar=4, time_secs=8.0, modified_tempo=130, original_tempo=120),
]
d = GpSyncData(audio_offset=0.0, audio_asset_id="", sync_points=sp)
assert build_tempo_map_from_sync(d) == [(0, 120.0), (4, 130.0)]
empty = GpSyncData(audio_offset=0.0, audio_asset_id="", sync_points=[])
assert build_tempo_map_from_sync(empty) == [(0, 120.0)]
# ── _parse_gpif / _resolve_audio_asset ────────────────────────────────────────────
def test_parse_gpif_reads_backing_track():
root = _parse_gpif(_gpif_bytes("xyz"))
assert root.find("BackingTrack/AssetId").text == "xyz"
def test_resolve_audio_asset_matches_declared_id():
with _make_gp_zip(asset_id="match", ogg_stems=("other", "match")) as zf:
stem, path = _resolve_audio_asset(zf)
assert stem == "match"
assert path == "Content/Assets/match.ogg"
def test_resolve_audio_asset_falls_back_to_first_when_unmatched():
with _make_gp_zip(asset_id="missing", ogg_stems=("only",)) as zf:
stem, path = _resolve_audio_asset(zf)
assert stem == "only"
assert path == "Content/Assets/only.ogg"
def test_resolve_audio_asset_matches_mp3():
# GP8 commonly embeds the backing track as MP3, not OGG — the resolver
# must match it (regression guard for the OGG-only bug).
with _make_gp_zip(asset_id="trk", ogg_stems=("trk",), asset_ext=".mp3") as zf:
stem, path = _resolve_audio_asset(zf)
assert stem == "trk"
assert path == "Content/Assets/trk.mp3"
def test_resolve_audio_asset_prefers_ogg_when_multiple_formats():
# If the same backing track is present as both OGG and MP3, prefer the
# OGG (lossless copy, no transcode) regardless of ZIP order.
buf = io.BytesIO()
with zipfile.ZipFile(buf, "w") as zf:
zf.writestr("Content/score.gpif", _gpif_bytes("trk"))
zf.writestr("Content/Assets/trk.mp3", b"fake-audio") # first in order
zf.writestr("Content/Assets/trk.ogg", b"fake-audio")
buf.seek(0)
with zipfile.ZipFile(buf) as zf:
stem, path = _resolve_audio_asset(zf)
assert path == "Content/Assets/trk.ogg"
def test_resolve_audio_asset_none_when_no_audio():
buf = io.BytesIO()
with zipfile.ZipFile(buf, "w") as zf:
zf.writestr("Content/score.gpif", _gpif_bytes())
buf.seek(0)
with zipfile.ZipFile(buf) as zf:
stem, path = _resolve_audio_asset(zf)
assert stem == "" and path is None
# ── gp_autosync timing helpers ──────────────────────────────────────────────────
def test_gp345_tick_to_secs_constant_tempo():
te = [(0, 120.0)] # 120 BPM, 960 ticks/quarter -> 1 quarter = 0.5 s
assert _gp345_tick_to_secs(te, 960) == pytest.approx(0.5)
assert _gp345_tick_to_secs(te, 1920) == pytest.approx(1.0)
assert _gp345_tick_to_secs(te, 0) == pytest.approx(0.0)
def test_gp345_tick_to_secs_handles_mid_stream_tempo_change():
# 120 BPM for the first quarter (0.5 s), then 60 BPM (1.0 s/quarter).
te = [(0, 120.0), (960, 60.0)]
assert _gp345_tick_to_secs(te, 1920) == pytest.approx(1.5)
def test_get_tempo_map_and_initial_tempo():
root = _fake_gpif_root(n_bars=2, tempo=144.0)
assert _get_initial_tempo(root) == pytest.approx(144.0)
assert _get_tempo_map(root) == [(0, 144.0)]
def test_get_tempo_map_defaults_when_absent():
assert _get_tempo_map(ET.Element("GPIF")) == [(0, 120.0)]
def test_tempo_at_bar_uses_last_event_at_or_before():
tmap = [(0, 100.0), (4, 150.0)]
assert _tempo_at_bar(tmap, 0) == 100.0
assert _tempo_at_bar(tmap, 3) == 100.0
assert _tempo_at_bar(tmap, 4) == 150.0
assert _tempo_at_bar(tmap, 9) == 150.0
def test_safe_normalise_unit_columns_and_no_nan_on_zero():
m = np.array([[3.0, 0.0], [4.0, 0.0]], dtype=np.float32)
out = _safe_normalise(m)
assert out[:, 0] == pytest.approx([0.6, 0.8]) # 3-4-5 triangle -> unit norm
assert np.all(np.isfinite(out)) # zero column -> no NaN
# zero-energy column is filled with 1/12 (equidistant from all bins)
assert out[:, 1] == pytest.approx([1.0 / 12.0, 1.0 / 12.0])
# ── _extract_sync_points (synthetic DTW path) ────────────────────────────────────
def _identity_setup(n_bars=4, tempo=120.0, sr=22050, hop=4096):
root = _fake_gpif_root(n_bars=n_bars, tempo=tempo)
n = 64
wp = np.array([[i, i] for i in range(n)]) # identity warp path
times = np.arange(n) * hop / sr
return root, wp, times, sr, hop
def test_extract_sync_points_identity_path_maps_bars_to_their_score_time():
root, wp, times, sr, hop = _identity_setup(n_bars=4, tempo=120.0)
pts = _extract_sync_points(wp, root, times, times, sr, hop, n_sync_points=4)
bars = [p.bar for p in pts]
assert bars == [0, 1, 2, 3]
# 4/4 @ 120 BPM -> 2.0 s/bar; identity path -> audio time == score bar start.
for p in pts:
assert p.time_secs == pytest.approx(p.bar * 2.0, abs=hop / sr)
# modified tempo is derived from the score/audio segment ratio and clamped;
# frame quantisation keeps it near (not exactly) the authored 120 BPM.
for p in pts:
assert 20.0 <= p.modified_tempo <= 300.0
assert p.modified_tempo == pytest.approx(120.0, abs=15.0)
def test_extract_sync_points_fills_trailing_modified_tempo():
# Regression: the final sync point must not keep the original-tempo
# placeholder — it carries the last computed segment tempo forward.
root, wp, times, sr, hop = _identity_setup(n_bars=4, tempo=120.0)
pts = _extract_sync_points(wp, root, times, times, sr, hop, n_sync_points=4)
assert pts[-1].modified_tempo == pts[-2].modified_tempo
def test_extract_sync_points_uses_bar_starts_override():
root, wp, times, sr, hop = _identity_setup(n_bars=4, tempo=120.0)
# Override every bar to sit at score-time 0 -> all map to audio frame 0.
override = [0.0, 0.0, 0.0, 0.0]
pts = _extract_sync_points(
wp, root, times, times, sr, hop, n_sync_points=4,
bar_starts_override=override,
)
assert all(p.time_secs == pytest.approx(0.0, abs=hop / sr) for p in pts)
def test_extract_sync_points_handles_zero_sync_points_without_crashing():
# n_sync_points is public; 0 must not raise ZeroDivisionError on the stride.
root, wp, times, sr, hop = _identity_setup(n_bars=4, tempo=120.0)
pts = _extract_sync_points(wp, root, times, times, sr, hop, n_sync_points=0)
assert len(pts) >= 1
def test_extract_sync_points_empty_when_no_bars():
root = ET.Element("GPIF") # no MasterBars
_, wp, times, sr, hop = _identity_setup()
assert _extract_sync_points(wp, root, times, times, sr, hop, 4) == []