Spaces:
Running
Running
Add BarScript (experimental preview) model + ura difficulty; vendored sc2 subpackage
d27cb1d verified | """End-to-end acceptance for the Space: real weights, real music, real app entry. | |
| The other two suites stop short of this on purpose. ``test_sc2_vendor.py`` | |
| proves the vendored 2.0 tree is the one the release package describes, and | |
| ``test_app_wiring.py`` proves the routing by replaying every model as a | |
| recorder over a synthetic mel. Neither one ever decodes real music, so | |
| neither can say whether a chart comes out of ``app.generate_chart`` intact. | |
| This file drives ``app.generate_chart`` itself -- the same generator the | |
| frontend consumes, with a ``gr.FileData`` upload -- on held-out songs, and | |
| inspects the artifacts it actually produced. | |
| Two questions: | |
| 1. **Does 2.0 produce a well-formed chart on every difficulty?** For each | |
| song x {easy, normal, hard, oni, ura} the emitted TJA is re-parsed with | |
| ``softchart.sc2.trace`` and checked for: a single course section whose | |
| ``COURSE:`` header resolves back to the course that was asked for (this is | |
| what makes ``COURSE:Edit`` -> ``ura`` a semantic check and not a string | |
| comparison); a bar count consistent with the tempo and excerpt length; | |
| 96-digit bars whose every note sits on the shipped /16 lattice; balanced, | |
| properly ordered drumroll spans; and a note total that agrees with the | |
| count the UI reported. | |
| The suite also RECORDS the difficulty ladder rather than assuming it. The | |
| model card (§2.1) says the ladder is not nested, and ura carries 208 | |
| training charts against ~1 100 elsewhere, so "ura is harder than oni" is | |
| an open question. ``--strict-ladder`` turns the ladder into a hard | |
| failure; by default it is reported, because a monotonicity failure here is | |
| a MODEL result to be published, not a wiring bug to be hidden. | |
| 2. **Do the published models still behave exactly as before?** Same songs, | |
| same controls, run through both the current ``app.py`` and | |
| ``fixtures/app_baseline.py`` -- the app as it stood BEFORE the 2.0/ura | |
| change, pinned by sha256 to the same source the wiring suite's golden | |
| trace was recorded from. Both run with REAL Hub weights, and the | |
| comparison is over the full TJA text plus the reported metrics. The | |
| wiring suite compares call arguments under mocks; this one compares | |
| generated charts. | |
| Fixtures. Preferred source is the held-out ``test`` split of | |
| ``JacobLinCool/taiko-1000-parsed`` if it is in the local HuggingFace cache: | |
| those songs are real music the models never trained on, and they ship | |
| authored charts for all five courses, which gives the difficulty ladder a | |
| reference to be read against. Otherwise pass ``--audio-dir`` with any | |
| directory of audio files (the ladder is then reported without a reference). | |
| Nothing is downloaded. | |
| python spaces/scripts/test_space_e2e.py | |
| python spaces/scripts/test_space_e2e.py --songs 2 --skip-legacy | |
| python spaces/scripts/test_space_e2e.py --audio-dir ../other --json report.json | |
| Runs on CPU. Legacy models are read from the local HuggingFace cache; export | |
| ``HF_HUB_OFFLINE=1`` to prove no network is involved. | |
| """ | |
| import argparse | |
| import collections | |
| import glob | |
| import hashlib | |
| import io | |
| import json | |
| import os | |
| import statistics | |
| import sys | |
| import tempfile | |
| import time | |
| SPACES = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) | |
| if SPACES not in sys.path: | |
| sys.path.insert(0, SPACES) | |
| FIXTURES = os.path.join(os.path.dirname(os.path.abspath(__file__)), "fixtures") | |
| BASELINE_APP = os.path.join(FIXTURES, "app_baseline.py") | |
| GOLDEN = os.path.join(FIXTURES, "legacy_call_trace.json") | |
| ALL_COURSES = ("easy", "normal", "hard", "oni", "ura") | |
| # Published models replayed against the baseline. v1.7 is the default the UI | |
| # recommends and v1.5 is the oldest still selectable, so between them they | |
| # cover both ends of the published range; both are full-size, so a difference | |
| # cannot hide behind a small model's noise. | |
| LEGACY_MODELS = ("v1.7", "v1.5") | |
| LEGACY_COURSES = ("hard", "oni") | |
| # The TJA export lattice: write_tja_slots emits SUB=96 digits per 4/4 bar. | |
| SUB = 96 | |
| # The deployment grid (sc2_loader.SHIP_GRID_DENOM). 96/16 = 6, so a note that | |
| # is on the /16 lattice sits at a digit index divisible by 6. | |
| SHIP_DENOM = 16 | |
| SPAN_OPEN = {"5", "6", "7"} # roll, big roll, balloon | |
| SPAN_CLOSE = "8" | |
| HIT_CHARS = {"1", "2", "3", "4"} | |
| # The TJA `COURSE:` literal each course must be WRITTEN as, measured over all | |
| # 1 155 songs of JacobLinCool/taiko-1000-parsed: the 229 ura charts are headed | |
| # `Edit` (223) or the numeric `4` (6), and the string "Ura" occurs zero times | |
| # anywhere in the corpus. | |
| # | |
| # The literal is asserted separately from the re-parse because | |
| # ``softchart.sc2.trace`` is deliberately lenient -- its alias table maps | |
| # "ura" -> ura -- so a chart written `COURSE:Ura` round-trips through this | |
| # repo's own parser and would pass a semantic-only check while still being | |
| # the exact defect the header rewrite exists to prevent. | |
| CORPUS_COURSE_LITERAL = {"easy": "Easy", "normal": "Normal", "hard": "Hard", | |
| "oni": "Oni", "ura": "Edit"} | |
| RESULTS = [] | |
| def check(name, ok, detail=""): | |
| RESULTS.append((name, bool(ok), detail)) | |
| print(f"{'PASS' if ok else 'FAIL'} {name}" + (f" -- {detail}" if detail else "")) | |
| return bool(ok) | |
| def note(name, detail): | |
| """Record an observation that is reported but never fails the suite.""" | |
| print(f"NOTE {name} -- {detail}") | |
| # --------------------------------------------------------------------------- | |
| # fixtures | |
| def dataset_test_shards(): | |
| pattern = os.path.expanduser( | |
| "~/.cache/huggingface/hub/datasets--JacobLinCool--taiko-1000-parsed/" | |
| "snapshots/*/data/test-*.parquet") | |
| return sorted(glob.glob(pattern)) | |
| def songs_from_dataset(out_dir, n_songs, seconds): | |
| """Held-out songs with authored charts, decoded from the local cache. | |
| Only songs carrying BOTH an oni and an ura chart are taken: those are the | |
| ones that can say what the authored oni->ura step actually looks like on | |
| the same music the model is being asked about. | |
| """ | |
| import soundfile as sf | |
| import pyarrow.parquet as pq | |
| picked = {} | |
| seen_titles = set() | |
| for shard in dataset_test_shards(): | |
| table = pq.read_table(shard) | |
| for i in range(table.num_rows): | |
| if len(picked) >= n_songs: | |
| break | |
| meta = table["metadata"][i].as_py() | |
| title = (meta.get("TITLE") or "").strip() | |
| if not title or title in seen_titles: | |
| continue | |
| authored = {} | |
| for course in ALL_COURSES: | |
| chart = table[course][i].as_py() | |
| if not chart: | |
| continue | |
| times = sorted( | |
| ev["timestamp"] for seg in chart["segments"] | |
| for ev in seg["notes"] | |
| if ev["note_type"] in ("Don", "Ka", "DonBig", "KaBig")) | |
| span = (times[-1] - times[0]) if len(times) > 1 else 0.0 | |
| authored[course] = { | |
| "level": chart["level"], "n_notes": len(times), | |
| "nps": (len(times) / span) if span else 0.0} | |
| if "ura" not in authored or "oni" not in authored: | |
| continue | |
| try: | |
| bpm = float(meta["BPM"]) | |
| except (TypeError, ValueError, KeyError): | |
| continue | |
| audio = table["audio"][i].as_py() | |
| wav, sr = sf.read(io.BytesIO(audio["bytes"]), dtype="float32", | |
| always_2d=True) | |
| wav = wav.mean(axis=1) | |
| # Start 20 % in: intros are often sparse or silent, and a chart | |
| # generated over silence would test nothing. | |
| start = int(len(wav) * 0.20) | |
| seg = wav[start:start + int(seconds * sr)] | |
| if len(seg) < int(seconds * sr * 0.9): | |
| continue | |
| seen_titles.add(title) | |
| slug = f"song{len(picked):02d}" | |
| path = os.path.join(out_dir, slug + ".wav") | |
| sf.write(path, seg, sr) | |
| picked[slug] = {"title": title, "path": path, "bpm": bpm, | |
| "seconds": len(seg) / sr, "authored": authored, | |
| "source": "taiko-1000-parsed/test"} | |
| if len(picked) >= n_songs: | |
| break | |
| return picked | |
| def songs_from_dir(audio_dir, out_dir, n_songs, seconds): | |
| """Fallback fixture: any directory of audio, with no authored reference.""" | |
| import librosa | |
| import soundfile as sf | |
| files = [] | |
| for ext in ("*.ogg", "*.mp3", "*.wav", "*.flac", "*.m4a"): | |
| files.extend(glob.glob(os.path.join(audio_dir, ext))) | |
| picked = {} | |
| for path in sorted(files)[:n_songs]: | |
| wav, sr = librosa.load(path, sr=None, mono=True) | |
| start = int(len(wav) * 0.20) | |
| seg = wav[start:start + int(seconds * sr)] | |
| if len(seg) < int(seconds * sr * 0.5): | |
| continue | |
| slug = f"song{len(picked):02d}" | |
| out = os.path.join(out_dir, slug + ".wav") | |
| sf.write(out, seg, sr) | |
| picked[slug] = {"title": os.path.basename(path), "path": out, | |
| "bpm": None, "seconds": len(seg) / sr, "authored": {}, | |
| "source": os.path.abspath(audio_dir)} | |
| return picked | |
| def build_fixtures(out_dir, n_songs, seconds, audio_dir=None): | |
| if audio_dir: | |
| songs = songs_from_dir(audio_dir, out_dir, n_songs, seconds) | |
| origin = f"--audio-dir {audio_dir}" | |
| elif dataset_test_shards(): | |
| songs = songs_from_dataset(out_dir, n_songs, seconds) | |
| origin = "taiko-1000-parsed test split (held out)" | |
| else: | |
| return {}, "none" | |
| return songs, origin | |
| # --------------------------------------------------------------------------- | |
| # driving the app | |
| def run_app(module, song, course, model, *, level=8, bpm=0.0, | |
| auto_plan_on=False, use_beat=True, use_planner=False, | |
| sampling=False, temperature=0.7, top_p=0.95, drum_volume=0.8): | |
| """Consume ``generate_chart`` and capture the artifacts it produced. | |
| The app deletes its working directory in a ``finally`` once the generator | |
| is closed, so the TJA has to be read while it is still suspended at the | |
| completion yield. | |
| """ | |
| import gradio as gr | |
| name = os.path.basename(song["path"]) | |
| payload = gr.FileData(path=song["path"], orig_name=name, | |
| mime_type="audio/wav").model_dump() | |
| started = time.time() | |
| stages = [] | |
| out = {"model": model, "course": course, "song": song["title"]} | |
| for event in module.generate_chart( | |
| payload, name, course, level, bpm, auto_plan_on, use_beat, | |
| use_planner, sampling, temperature, top_p, drum_volume, model): | |
| stages.append(event.get("stage")) | |
| if event["kind"] == "error": | |
| out.update(kind="error", detail=event.get("detail")) | |
| break | |
| if event["kind"] == "complete": | |
| with open(event["files"]["tja"]["path"], encoding="utf-8") as f: | |
| tja = f.read() | |
| out.update(kind="complete", tja=tja, metrics=dict(event["metrics"]), | |
| files={k: os.path.basename(v["path"]) | |
| for k, v in event["files"].items()}) | |
| break | |
| out["stages"] = stages | |
| out["seconds"] = round(time.time() - started, 2) | |
| return out | |
| # --------------------------------------------------------------------------- | |
| # chart inspection | |
| def inspect_tja(text, course): | |
| """Structural facts about one generated TJA, from a real re-parse.""" | |
| from softchart.sc2 import trace as T | |
| facts = {} | |
| charts = T.parse_tja(text) # raises on anything it will not guess about | |
| facts["n_sections"] = len(charts) | |
| chart = charts[0] | |
| facts["parsed_course"] = chart.course | |
| facts["level"] = chart.level | |
| facts["n_bars"] = len(chart.bars) | |
| facts["bar_lengths"] = sorted(set(len(b.digits) for b in chart.bars)) | |
| header = [ln for ln in text.splitlines() if ln.startswith("COURSE:")] | |
| facts["course_header"] = header[0] if header else None | |
| # Every character that is not '0' must land on the shipped lattice. | |
| off_lattice = 0 | |
| counts = collections.Counter() | |
| for bar in chart.bars: | |
| stride = len(bar.digits) // SHIP_DENOM if len(bar.digits) else 0 | |
| for pos, ch in enumerate(bar.digits): | |
| if ch == "0": | |
| continue | |
| counts[ch] += 1 | |
| if not stride or pos % stride: | |
| off_lattice += 1 | |
| facts["off_lattice_notes"] = off_lattice | |
| facts["note_chars"] = dict(counts) | |
| facts["n_hits"] = sum(v for k, v in counts.items() if k in HIT_CHARS) | |
| # Span balance, in emission order across the whole chart. | |
| depth = 0 | |
| bad_order = 0 | |
| n_spans = 0 | |
| for bar in chart.bars: | |
| for ch in bar.digits: | |
| if ch in SPAN_OPEN: | |
| if depth: | |
| bad_order += 1 | |
| depth += 1 | |
| n_spans += 1 | |
| elif ch == SPAN_CLOSE: | |
| if not depth: | |
| bad_order += 1 | |
| else: | |
| depth -= 1 | |
| facts["n_spans"] = n_spans | |
| facts["unclosed_spans"] = depth | |
| facts["misordered_spans"] = bad_order | |
| # A full trace conversion is the strongest re-parse available: it resolves | |
| # every note to an exact bar-local Fraction and a wall-clock time. | |
| tr = T.chart_to_trace(chart) | |
| times = T.trace_to_times(tr) | |
| facts["n_trace_notes"] = len(times) | |
| facts["bpm"] = tr.bpm | |
| facts["offset"] = tr.offset | |
| return facts | |
| def note_times(text): | |
| """Wall-clock times of the hit notes, for density statistics.""" | |
| from softchart.sc2 import trace as T | |
| chart = T.parse_tja(text)[0] | |
| out = [] | |
| for entry in T.trace_to_times(T.chart_to_trace(chart)): | |
| t = entry[0] if isinstance(entry, (tuple, list)) else entry.get("t") | |
| kind = entry[1] if isinstance(entry, (tuple, list)) else entry.get("type") | |
| if str(kind) in HIT_CHARS: | |
| out.append(float(t)) | |
| return sorted(out) | |
| # --------------------------------------------------------------------------- | |
| # test 1: BarScript on all five difficulties | |
| def test_sc2_all_courses(songs, records, strict_ladder=False): | |
| import app as APP | |
| from softchart import sc2_loader | |
| if not sc2_loader.is_available(): | |
| check("BarScript release package present", False, | |
| f"no package at {sc2_loader.default_package_dir()}") | |
| return | |
| check("BarScript release package present", True, sc2_loader.default_package_dir()) | |
| per_song = {} | |
| for slug, song in sorted(songs.items()): | |
| per_song[slug] = {} | |
| for course in ALL_COURSES: | |
| tag = f"BarScript {song['title'][:22]} / {course}" | |
| result = run_app(APP, song, course, APP.SC2_MODEL) | |
| records.append({"phase": "sc2", "slug": slug, **{ | |
| k: v for k, v in result.items() if k != "tja"}}) | |
| if result["kind"] != "complete": | |
| check(f"{tag}: completes", False, str(result.get("detail"))[:160]) | |
| continue | |
| facts = inspect_tja(result["tja"], course) | |
| records[-1]["facts"] = facts | |
| metrics = result["metrics"] | |
| ok = check(f"{tag}: completes", True, | |
| f"{metrics['notes']} notes, {result['seconds']}s") | |
| ok &= check(f"{tag}: single course section", | |
| facts["n_sections"] == 1, f"{facts['n_sections']}") | |
| # Two independent header checks. The literal is what a Taiko | |
| # simulator reads; the re-parse is what it means. | |
| ok &= check(f"{tag}: COURSE header is the literal the corpus uses", | |
| facts["course_header"] | |
| == f"COURSE:{CORPUS_COURSE_LITERAL[course]}", | |
| f"{facts['course_header']!r}") | |
| ok &= check(f"{tag}: COURSE header resolves to the course asked for", | |
| facts["parsed_course"] == course, | |
| f"{facts['course_header']} -> {facts['parsed_course']}") | |
| ok &= check(f"{tag}: bars are {SUB}-digit", | |
| facts["bar_lengths"] == [SUB], f"{facts['bar_lengths']}") | |
| ok &= check(f"{tag}: every note on the /{SHIP_DENOM} lattice", | |
| facts["off_lattice_notes"] == 0, | |
| f"{facts['off_lattice_notes']} off-lattice") | |
| ok &= check(f"{tag}: no unclosed drumroll", | |
| facts["unclosed_spans"] == 0 and | |
| facts["misordered_spans"] == 0, | |
| f"open={facts['unclosed_spans']} " | |
| f"misordered={facts['misordered_spans']}") | |
| ok &= check(f"{tag}: re-parsed note count matches the UI", | |
| facts["n_hits"] == metrics["notes"], | |
| f"{facts['n_hits']} vs {metrics['notes']}") | |
| ok &= check(f"{tag}: span count matches the UI", | |
| facts["n_spans"] == metrics["spans"], | |
| f"{facts['n_spans']} vs {metrics['spans']}") | |
| # Bar count must follow from tempo and length, or the lattice is | |
| # anchored wrong and every barline is in the wrong place. | |
| bar_sec = 4.0 * 60.0 / float(metrics["bpm"]) | |
| expected = (song["seconds"] - abs(facts["offset"])) / bar_sec | |
| ok &= check(f"{tag}: bar count follows tempo and length", | |
| abs(facts["n_bars"] - expected) <= 1.5, | |
| f"{facts['n_bars']} bars, expected ~{expected:.1f} " | |
| f"at {metrics['bpm']} BPM") | |
| ok &= check(f"{tag}: reported as slot-exact on the /16 grid", | |
| metrics["timing"] == "slot-exact" | |
| and "/16" in metrics.get("chart_grid", ""), | |
| f"{metrics['timing']}, {metrics.get('chart_grid')}") | |
| times = note_times(result["tja"]) | |
| span = (times[-1] - times[0]) if len(times) > 1 else 0.0 | |
| per_song[slug][course] = { | |
| "n_notes": facts["n_hits"], "n_spans": facts["n_spans"], | |
| "nps": (len(times) / span) if span else 0.0, | |
| "min_gap_ms": (min((b - a) for a, b in zip(times, times[1:])) | |
| * 1000.0) if len(times) > 1 else None} | |
| records.append({"phase": "sc2_ladder", "generated": per_song, | |
| "authored": {slug: song["authored"] | |
| for slug, song in songs.items()}}) | |
| ladder_report(songs, per_song, strict_ladder) | |
| return per_song | |
| def ladder_report(songs, per_song, strict_ladder): | |
| """Report the generated difficulty ladder against the authored one.""" | |
| print("\n--- difficulty ladder (BarScript, generated vs authored)") | |
| header = f"{'song':26s} " + " ".join(f"{c:>13s}" for c in ALL_COURSES) | |
| print(header) | |
| ura_gt_oni = [] | |
| monotone = [] | |
| for slug, song in sorted(songs.items()): | |
| got = per_song.get(slug) or {} | |
| if not got: | |
| continue | |
| row = f"{song['title'][:25]:26s} " | |
| for course in ALL_COURSES: | |
| entry = got.get(course) | |
| row += f"{entry['n_notes']:6d}/{entry['nps']:5.2f} " if entry else f"{'--':>13s} " | |
| print(row) | |
| if song["authored"]: | |
| arow = f"{' (authored)':26s} " | |
| for course in ALL_COURSES: | |
| a = song["authored"].get(course) | |
| arow += f"{a['n_notes']:6d}/{a['nps']:5.2f} " if a else f"{'--':>13s} " | |
| print(arow) | |
| counts = [got[c]["n_notes"] for c in ALL_COURSES if c in got] | |
| if len(counts) == len(ALL_COURSES): | |
| monotone.append(all(x < y for x, y in zip(counts, counts[1:]))) | |
| ura_gt_oni.append(got["ura"]["n_notes"] > got["oni"]["n_notes"]) | |
| if not ura_gt_oni: | |
| return | |
| n_ura = sum(ura_gt_oni) | |
| n_mono = sum(monotone) | |
| detail = (f"ura denser than oni on {n_ura}/{len(ura_gt_oni)} songs; " | |
| f"fully monotone easy<normal<hard<oni<ura on " | |
| f"{n_mono}/{len(monotone)}") | |
| authored = [s for s in songs.values() if s["authored"].get("ura") | |
| and s["authored"].get("oni")] | |
| if authored: | |
| a_ura = sum(1 for s in authored | |
| if s["authored"]["ura"]["n_notes"] | |
| > s["authored"]["oni"]["n_notes"]) | |
| detail += f"; authored reference: ura denser on {a_ura}/{len(authored)}" | |
| if strict_ladder: | |
| check("BarScript: ura is harder than oni on every song", | |
| n_ura == len(ura_gt_oni), detail) | |
| check("BarScript: the difficulty ladder is monotone on every song", | |
| n_mono == len(monotone), detail) | |
| else: | |
| note("BarScript difficulty ladder (reported, not asserted -- MODEL_CARD " | |
| "§2.1 says the ladder is not nested)", detail) | |
| # --------------------------------------------------------------------------- | |
| # test 2: published models unchanged, on real audio and real weights | |
| def load_baseline(): | |
| """Import the pre-change app, pinned to the wiring suite's provenance. | |
| The baseline mounts its static assets from ``Path(__file__).with_name | |
| ("static")``, so it is staged in a scratch directory next to a link to | |
| the Space's real ``static/`` rather than imported from ``fixtures/``. | |
| """ | |
| import importlib.util | |
| import shutil | |
| if not os.path.isfile(BASELINE_APP): | |
| return None, "fixtures/app_baseline.py is missing" | |
| with open(BASELINE_APP, "rb") as f: | |
| digest = hashlib.sha256(f.read()).hexdigest() | |
| expected = None | |
| if os.path.isfile(GOLDEN): | |
| with open(GOLDEN, encoding="utf-8") as f: | |
| expected = json.load(f).get("source_sha256") | |
| if expected and digest != expected: | |
| return None, (f"baseline sha256 {digest[:16]} does not match the " | |
| f"golden trace's source {expected[:16]}") | |
| stage = tempfile.mkdtemp(prefix="softchart-baseline-") | |
| staged = os.path.join(stage, "app_baseline.py") | |
| shutil.copyfile(BASELINE_APP, staged) | |
| os.symlink(os.path.join(SPACES, "static"), os.path.join(stage, "static")) | |
| spec = importlib.util.spec_from_file_location("app_pre_sc2", staged) | |
| module = importlib.util.module_from_spec(spec) | |
| sys.modules["app_pre_sc2"] = module | |
| spec.loader.exec_module(module) | |
| return module, digest | |
| def test_legacy_unchanged(songs, records): | |
| import app as APP | |
| baseline, info = load_baseline() | |
| if baseline is None: | |
| check("legacy baseline available", False, info) | |
| return | |
| check("legacy baseline available", True, | |
| f"fixtures/app_baseline.py sha256 {info[:16]}, matches the golden " | |
| f"trace source") | |
| for model in LEGACY_MODELS: | |
| for course in LEGACY_COURSES: | |
| for slug, song in sorted(songs.items()): | |
| tag = f"{model} {song['title'][:22]} / {course}" | |
| try: | |
| now = run_app(APP, song, course, model) | |
| was = run_app(baseline, song, course, model) | |
| except Exception as exc: # noqa: BLE001 | |
| check(f"{tag}: runs", False, f"{type(exc).__name__}: {exc}") | |
| continue | |
| if now["kind"] != "complete" or was["kind"] != "complete": | |
| check(f"{tag}: both apps complete", False, | |
| f"now={now['kind']} baseline={was['kind']} " | |
| f"{now.get('detail') or was.get('detail')}") | |
| continue | |
| same_tja = now["tja"] == was["tja"] | |
| # The metrics panel legitimately GAINED fields in this change | |
| # (the model name, the TJA course literal, the plan source). | |
| # What may not change is any value the baseline already | |
| # reported, so the shared keys are compared exactly and the | |
| # added ones are recorded separately. | |
| shared = set(was["metrics"]) & set(now["metrics"]) | |
| same_shared = all(was["metrics"][k] == now["metrics"][k] | |
| for k in shared) | |
| added = sorted(set(now["metrics"]) - set(was["metrics"])) | |
| dropped = sorted(set(was["metrics"]) - set(now["metrics"])) | |
| records.append({ | |
| "phase": "legacy", "slug": slug, "model": model, | |
| "course": course, "identical_tja": same_tja, | |
| "identical_shared_metrics": same_shared, | |
| "metrics_added": added, "metrics_dropped": dropped, | |
| "notes": now["metrics"]["notes"], | |
| "sha256": hashlib.sha256( | |
| now["tja"].encode("utf-8")).hexdigest()[:16]}) | |
| check(f"{tag}: TJA byte-identical to the pre-change app", | |
| same_tja, | |
| f"{len(now['tja'])} bytes, sha " | |
| f"{records[-1]['sha256']}" if same_tja else | |
| _first_difference(was["tja"], now["tja"])) | |
| check(f"{tag}: every metric the old app reported is unchanged", | |
| same_shared and not dropped, | |
| f"{len(shared)} shared keys equal" | |
| + (f", added {added}" if added else "") | |
| + (f", DROPPED {dropped}" if dropped else "") | |
| if same_shared and not dropped else | |
| _dict_difference(was["metrics"], now["metrics"])) | |
| def _first_difference(a, b): | |
| for i, (x, y) in enumerate(zip(a, b)): | |
| if x != y: | |
| return f"first differs at byte {i}: {a[i:i+40]!r} vs {b[i:i+40]!r}" | |
| return f"lengths {len(a)} vs {len(b)}" | |
| def _dict_difference(a, b): | |
| keys = sorted(set(a) | set(b)) | |
| return "; ".join(f"{k}: {a.get(k)!r} -> {b.get(k)!r}" | |
| for k in keys if a.get(k) != b.get(k)) | |
| # --------------------------------------------------------------------------- | |
| def main(): | |
| parser = argparse.ArgumentParser( | |
| description=__doc__, | |
| formatter_class=argparse.RawDescriptionHelpFormatter) | |
| parser.add_argument("--songs", type=int, default=6, | |
| help="number of songs (default 6)") | |
| parser.add_argument("--seconds", type=float, default=60.0, | |
| help="excerpt length in seconds (default 60)") | |
| parser.add_argument("--audio-dir", | |
| help="use this directory of audio instead of the " | |
| "held-out dataset split") | |
| parser.add_argument("--skip-legacy", action="store_true", | |
| help="skip the published-model regression") | |
| parser.add_argument("--skip-sc2", action="store_true", | |
| help="skip the BarScript five-difficulty run") | |
| parser.add_argument("--strict-ladder", action="store_true", | |
| help="fail if the generated ladder is not monotone") | |
| parser.add_argument("--json", help="write a machine-readable report here") | |
| args = parser.parse_args() | |
| workdir = tempfile.mkdtemp(prefix="softchart-e2e-") | |
| records = [] | |
| try: | |
| songs, origin = build_fixtures(workdir, args.songs, args.seconds, | |
| args.audio_dir) | |
| if not songs: | |
| print("no audio fixtures available: the taiko-1000-parsed test " | |
| "split is not in the local HuggingFace cache and no " | |
| "--audio-dir was given") | |
| return 2 | |
| print(f"fixtures: {len(songs)} songs x {args.seconds:.0f}s from {origin}") | |
| for slug, song in sorted(songs.items()): | |
| print(f" {slug} {song['title'][:40]:42s} {song['bpm'] or '?'} BPM") | |
| if not args.skip_sc2: | |
| print("\n--- test_sc2_all_courses") | |
| test_sc2_all_courses(songs, records, args.strict_ladder) | |
| if not args.skip_legacy: | |
| print("\n--- test_legacy_unchanged") | |
| test_legacy_unchanged(songs, records) | |
| finally: | |
| import shutil | |
| shutil.rmtree(workdir, ignore_errors=True) | |
| failed = [n for n, ok, _ in RESULTS if not ok] | |
| print(f"\n{len(RESULTS) - len(failed)}/{len(RESULTS)} checks passed") | |
| for name in failed: | |
| print(f" FAILED: {name}") | |
| if args.json: | |
| with open(args.json, "w", encoding="utf-8") as f: | |
| json.dump({"checks": [{"name": n, "ok": ok, "detail": d} | |
| for n, ok, d in RESULTS], | |
| "records": records}, f, indent=1, ensure_ascii=False) | |
| print(f"report -> {args.json}") | |
| return 1 if failed else 0 | |
| if __name__ == "__main__": | |
| sys.exit(main()) | |