Files
libretro/scripts/pipeline.py
T
Abdessamad Derraz e9da4d72ed fix: write one truth model per target, not alias
generate_truth named a narrowed model's directory after the target
as typed, so three aliases of one target wrote three copies, and
the pipeline rebuilt the slug by hand to diff it. One function now
names the directory from the canonical target for both.
2026-10-18 02:55:45 +02:00

705 lines
26 KiB
Python

#!/usr/bin/env python3
"""Run the full retrobios pipeline.
Steps:
1. generate_db.py --force (rebuild database.json from bios/)
1b. provenance_report.py (dump-catalog coverage from provenance/)
1c. romset_recipes.py (archive identification, reconstruction targets)
2. refresh_data_dirs.py (update Dolphin Sys, PPSSPP, etc.)
2b2. check_release_assets.py (large-files release vs gitignored files, online)
3. verify.py --all (check all platforms)
4. generate_pack.py --all (build ZIP packs)
4b. generate install manifests
4c. generate target manifests
5. consistency check (verify counts == pack counts)
8. generate_readme.py (rebuild README.md + CONTRIBUTING.md)
9. generate_site.py (rebuild MkDocs pages)
Usage:
python scripts/pipeline.py # active platforms
python scripts/pipeline.py --include-archived # all platforms
python scripts/pipeline.py --skip-packs # steps 1-3 only
python scripts/pipeline.py --skip-docs # skip steps 8-9
python scripts/pipeline.py --offline # skip step 2
"""
from __future__ import annotations
import argparse
import contextlib
import os
import re
import subprocess
import sys
import time
from pathlib import Path
# The build scripts use syntax CPython 3.12 introduced (generate_readme's
# f-strings, generate_pack's parenthesized context managers). On an older
# interpreter the run got through the packs and failed hours in, at step 7.
if sys.version_info < (3, 12):
sys.exit(f"pipeline.py needs Python 3.12 or newer, this is {sys.version.split()[0]}")
sys.path.insert(0, str(Path(__file__).parent))
from common import (
ArtifactLockBusy,
artifact_lock,
hold_artifact_lock,
list_registered_platforms,
truth_target_dir,
)
def run(cmd: list[str], label: str) -> tuple[bool, str]:
"""Run a command, echoing its output as it comes. Returns (success, output).
The output used to be captured and printed when the step ended, stderr
after stdout: a twenty-minute pack build showed nothing, and a warning
landed far from the line it was about. One stream, read line by line.
"""
print(f"\n--- {label} ---", flush=True)
start = time.monotonic()
repo_root = str(Path(__file__).resolve().parent.parent)
lines: list[str] = []
with subprocess.Popen(
cmd, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True,
cwd=repo_root, env={**os.environ, "PYTHONUNBUFFERED": "1"},
) as proc:
for line in proc.stdout:
lines.append(line)
print(line, end="", flush=True)
returncode = proc.wait()
elapsed = time.monotonic() - start
ok = returncode == 0
print(f"--- {label}: {'OK' if ok else 'FAILED'} ({elapsed:.1f}s) ---")
return ok, "".join(lines)
def parse_verify_counts(output: str) -> dict[str, tuple[int, int]]:
"""Extract per-group OK/total from verify output.
Matches: "Label: X/Y OK ..." or "Label: X/Y present ..."
Returns {group_label: (ok, total)}.
"""
counts = {}
for line in output.splitlines():
m = re.match(r"^(.+?):\s+(\d+)/(\d+)\s+(OK|present)", line)
if m:
label = m.group(1).strip()
ok, total = int(m.group(2)), int(m.group(3))
for name in label.split(" / "):
counts[name.strip()] = (ok, total)
return counts
def _full_pack_label(label: str) -> str:
"""The pack name, or "" for a reduced variant such as [source=platform]."""
meta = re.search(r"\s+\[([^\]]+)\]$", label)
if meta is None:
return label
return label[: meta.start()] if meta.group(1) == "source=full" else ""
def parse_pack_counts(output: str) -> dict[str, tuple[int, int]]:
"""Extract per-pack OK/total from generate_pack output.
Returns {pack_label: (ok, total)}.
"""
counts = {}
current_label = ""
for line in output.splitlines():
m = re.match(r"Generating (?:shared )?pack for (.+)\.\.\.", line)
if m:
# Labels carry execution metadata such as ``[source=full]``. Only
# the full, unreduced pack is what verify describes: with
# --all-variants the six variants shared one label and the last
# (platform, required) was compared to the full report.
current_label = _full_pack_label(m.group(1).strip())
continue
if not current_label:
continue
if "files packed" not in line:
continue
# "622 files packed (359 baseline + 263 from cores), 358/359 files OK"
ok_m = re.search(r"(\d+)/(\d+) files OK", line)
if ok_m:
counts[current_label] = (int(ok_m.group(1)), int(ok_m.group(2)))
return counts
def parse_pack_exclusions(output: str) -> dict[str, int]:
"""Extract intentional unsafe-omission counts from pack output."""
exclusions: dict[str, int] = {}
current_label = ""
for line in output.splitlines():
label_match = re.match(r"Generating (?:shared )?pack for (.+)\.\.\.", line)
if label_match:
# The same reading as parse_pack_counts: a reduced variant must
# not overwrite the full pack's count under the shared label.
current_label = _full_pack_label(label_match.group(1).strip())
continue
if not current_label or "files packed" not in line:
continue
excluded_match = re.search(r"(\d+) unsafe excluded", line)
exclusions[current_label] = (
int(excluded_match.group(1)) if excluded_match else 0
)
return exclusions
def _match_key(label: str) -> set[str]:
"""Comparable identity for a platform label.
Display labels and registry ids differ in punctuation and spacing
(``MiSTer FPGA`` against ``misterfpga``), and grouped packs join their
members with either separator.
"""
return {
re.sub(r"[^a-z0-9]+", "", part.strip().lower())
for part in label.replace("+", "/").split("/")
} - {""}
def check_consistency(verify_output: str, pack_output: str) -> bool:
"""Verify that check counts match between verify and pack for each platform."""
v = parse_verify_counts(verify_output)
p = parse_pack_counts(pack_output)
excluded = parse_pack_exclusions(pack_output)
print("\n--- 5/8 consistency check ---")
all_ok = True
if not v or not p:
# Nothing parsed is nothing compared: a changed output format would
# otherwise pass the check without a single platform in it.
print(f" parsed {len(v)} verify and {len(p)} pack results: nothing to compare")
all_ok = False
for p_label in sorted(p):
if not any(_match_key(p_label) & _match_key(v_label) for v_label in v):
print(f" {p_label}: pack with no verify result")
all_ok = False
for v_label, (v_ok, v_total) in sorted(v.items()):
# Match by normalized name overlap. Platform display labels and
# registry IDs legitimately differ in punctuation and spacing
# (notably ``MiSTer FPGA`` vs ``misterfpga``).
p_match = None
v_names = _match_key(v_label)
for p_label in p:
if v_names & _match_key(p_label):
p_match = p_label
break
if p_match:
p_ok, p_total = p[p_match]
p_excluded = excluded.get(p_match, 0)
if v_total != p_total:
print(f" {v_label}: MISMATCH total verify {v_total} != pack {p_total}")
all_ok = False
elif p_ok == v_ok:
print(
f" {v_label}: verify {v_ok}/{v_total} == pack {p_ok}/{p_total} OK"
)
else:
# Both ways are a disagreement. A pack exclusion never covers
# a file verify counts OK: an existence platform ships a file
# whose hash disagrees, and a hash platform's verify reports
# it untested, so the counts can only differ by a defect.
print(
f" {v_label}: MISMATCH pack {p_ok}/{p_total} OK "
f"({p_excluded} unsafe excluded) != verify {v_ok}/{v_total}"
)
all_ok = False
else:
print(f" {v_label}: {v_ok}/{v_total} (no separate pack)")
all_ok = False
status = "OK" if all_ok else "FAILED"
print(f"--- consistency check: {status} ---")
return all_ok
class _Skipped:
"""A step that did not run. Truthy, so it never fails the pipeline, and
distinct, so the summary does not report it as work that was done."""
def __bool__(self) -> bool:
return True
def __repr__(self) -> str:
return "SKIPPED"
SKIPPED = _Skipped()
def _refuse_unapplied_flags(
parser: argparse.ArgumentParser, args: argparse.Namespace
) -> None:
"""A flag the run would drop is refused before any step.
--all-variants builds every source itself, so a --source beside it is
dropped; --skip-packs builds none, so both are.
"""
if args.all_variants and args.source != "full":
parser.error("--source is incompatible with --all-variants")
if args.skip_packs:
for flag, on in (("--source", args.source != "full"),
("--all-variants", args.all_variants)):
if on:
parser.error(f"{flag} is incompatible with --skip-packs")
def _truth_steps(args: argparse.Namespace, results: dict) -> bool:
"""Steps 2c to 2e: truth model, its diff, the native export."""
all_ok = True
# Step 2c: Generate truth YAMLs
# A targeted run writes its model in a subdirectory of its own, and the
# diff has to read the same one or it compares a narrowed model against
# the whole scrape.
truth_dir = Path(args.output_dir) / "truth"
if args.target:
# generate_truth --all writes every registered platform's model.
truth_dir = Path(truth_target_dir(
str(truth_dir),
list_registered_platforms("platforms", include_archived=True),
args.target,
))
if args.with_truth or args.with_export:
truth_cmd = [
sys.executable,
"scripts/generate_truth.py",
"--all",
"--output-dir",
str(Path(args.output_dir) / "truth"),
]
if args.target:
truth_cmd.extend(["--target", args.target])
ok, _ = run(truth_cmd, "2c generate truth")
# A native file is one per platform, never per target: the export
# corrects it against the full model, which a targeted run does not
# write.
if ok and args.with_export and args.target:
ok, _ = run(truth_cmd[:-2], "2c generate truth (export model)")
results["generate_truth"] = ok
all_ok = all_ok and ok
else:
results["generate_truth"] = SKIPPED
# Step 2d: Diff truth vs scraped
if args.with_truth or args.with_export:
diff_cmd = [sys.executable, "scripts/diff_truth.py", "--all"]
diff_cmd.extend(["--truth-dir", str(truth_dir)])
ok, _ = run(diff_cmd, "2d diff truth")
results["diff_truth"] = ok
all_ok = all_ok and ok
else:
results["diff_truth"] = SKIPPED
# Step 2e: Export native formats
if args.with_export:
export_cmd = [
sys.executable,
"scripts/export_native.py",
"--all",
"--output-dir",
str(Path(args.output_dir) / "upstream"),
"--truth-dir",
str(Path(args.output_dir) / "truth"),
]
# Seven of the formats carry code, so the export patches the
# platform's own file rather than regenerating it. Offline that
# file has to be in the cache already.
if not args.offline:
export_cmd.append("--fetch")
ok, _ = run(export_cmd, "2e export native")
results["export_native"] = ok
all_ok = all_ok and ok
else:
results["export_native"] = SKIPPED
return all_ok
def main() -> None:
parser = argparse.ArgumentParser(description="Run the full retrobios pipeline")
parser.add_argument(
"--include-archived", action="store_true", help="Include archived platforms"
)
parser.add_argument(
"--skip-packs",
action="store_true",
help="Only regenerate DB and verify, skip pack generation",
)
parser.add_argument(
"--skip-docs", action="store_true", help="Skip README and site generation"
)
parser.add_argument(
"--offline", action="store_true", help="Skip data directory refresh"
)
parser.add_argument(
"--output-dir", default="dist", help="Pack output directory (default: dist/)"
)
parser.add_argument("--target", "-t", help="Hardware target (e.g., switch, rpi4)")
parser.add_argument("--source", choices=["platform", "truth", "full"], default="full")
parser.add_argument("--all-variants", action="store_true")
parser.add_argument(
"--check-buildbot",
action="store_true",
help="Check buildbot system directory for changes",
)
parser.add_argument(
"--with-truth",
action="store_true",
help="Generate truth YAMLs and diff against scraped",
)
parser.add_argument(
"--with-export",
action="store_true",
help="Export native formats (implies --with-truth)",
)
args = parser.parse_args()
_refuse_unapplied_flags(parser, args)
# A second run on the same output directory is refused before any work:
# the database rebuild alone takes minutes, and the reader holding the
# directory would otherwise see the answer only after all of it. The
# hold lasts until this process exits: taken step by step, it let another
# run purge the packs between the build and the integrity check, which
# then reported every platform SKIP and passed.
held = contextlib.ExitStack()
if not args.skip_packs:
try:
held.enter_context(hold_artifact_lock(args.output_dir))
except ArtifactLockBusy as exc:
print(f"ERROR: {exc}")
sys.exit(1)
results = {}
all_ok = True
total_start = time.monotonic()
# Step 1: Generate database
ok, out = run(
[
sys.executable,
"scripts/generate_db.py",
"--force",
"--bios-dir",
"bios",
"--output",
"database.json",
],
"1/8 generate database",
)
results["generate_db"] = ok
if not ok:
print("\nDatabase generation failed, aborting.")
sys.exit(1)
# Step 1b: Dump-catalog coverage (offline, reads provenance/ snapshots)
ok, _ = run(
[sys.executable, "scripts/provenance_report.py"],
"1b provenance report",
)
results["provenance"] = ok
all_ok = all_ok and ok
# Step 1c: Which emulator version each arcade archive corresponds to, and
# which pinned archive the collection could rebuild from ROMs it holds.
# Read-only: writing reconstructions is an explicit --write invocation.
ok, out = run(
[sys.executable, "scripts/romset_recipes.py"],
"1c romset recipes",
)
results["romset_recipes"] = ok
all_ok = all_ok and ok
# Step 2: Refresh data directories
if not args.offline:
ok, out = run(
[sys.executable, "scripts/refresh_data_dirs.py"],
"2/8 refresh data directories",
)
results["refresh_data"] = ok
all_ok = all_ok and ok
else:
print("\n--- 2/8 refresh data directories: SKIPPED (--offline) ---")
results["refresh_data"] = SKIPPED
# Step 2a: Refresh MAME BIOS hashes
if not args.offline:
ok, _ = run(
[sys.executable, "-m", "scripts.scraper.mame_hash_scraper"],
"2a refresh MAME hashes",
)
results["mame_hashes"] = ok
all_ok = all_ok and ok
else:
print("\n--- 2a refresh MAME hashes: SKIPPED (--offline) ---")
results["mame_hashes"] = SKIPPED
# Step 2a2: Refresh FBNeo BIOS hashes
if not args.offline:
ok, _ = run(
[sys.executable, "-m", "scripts.scraper.fbneo_hash_scraper"],
"2a2 refresh FBNeo hashes",
)
results["fbneo_hashes"] = ok
all_ok = all_ok and ok
else:
print("\n--- 2a2 refresh FBNeo hashes: SKIPPED (--offline) ---")
results["fbneo_hashes"] = SKIPPED
# Step 2b: Check buildbot system directory (non-blocking)
if args.check_buildbot and not args.offline:
ok, _ = run(
[sys.executable, "scripts/check_buildbot_system.py"],
"2b check buildbot system",
)
results["check_buildbot"] = ok
elif args.check_buildbot:
print("\n--- 2b check buildbot system: SKIPPED (--offline) ---")
# Step 2b2: The release must serve the bytes the manifests describe. The
# installer compares Content-Length with the manifest size, so an asset
# uploaded before its local file was rebuilt fails every install of it.
if not args.offline:
ok, _ = run(
[sys.executable, "scripts/check_release_assets.py"],
"2b2 check release assets",
)
results["check_release_assets"] = ok
all_ok = all_ok and ok
else:
print("\n--- 2b2 check release assets: SKIPPED (--offline) ---")
results["check_release_assets"] = SKIPPED
all_ok = _truth_steps(args, results) and all_ok
# Step 3: Verify
verify_cmd = [sys.executable, "scripts/verify.py", "--all", "--strict"]
if args.include_archived:
verify_cmd.append("--include-archived")
if args.target:
verify_cmd.extend(["--target", args.target])
ok, verify_output = run(verify_cmd, "3/8 verify all platforms")
results["verify"] = ok
all_ok = all_ok and ok
# Step 3a: One content answering to two identities. Walks the tree rather
# than the index, which keeps a single path per content and therefore
# cannot see a file copied under another machine's name.
ok, _ = run(
[sys.executable, "scripts/identity.py", "--strict"],
"3a/8 file identity",
)
results["identity"] = ok
all_ok = all_ok and ok
# Step 3b: Destinations both layers claim, and how each was settled. The
# ones the pack settles by itself must stay at zero cost; the rest name an
# upstream declaration no build can repair, so this reports and never gates.
# A crash is not a report: without --strict slots.py exits 0 on every
# conflict, so a non-zero code means it never finished reading.
ok, _ = run(
[sys.executable, "scripts/slots.py"],
"3b/8 slot arbitration",
)
results["slots"] = ok
all_ok &= ok
# Step 4: Generate packs
pack_output = ""
if not args.skip_packs:
# Purge stale packs: leftover ZIPs from previous builds would be
# picked up by pack verification and shipped in releases.
out_dir = Path(args.output_dir)
if out_dir.is_dir():
try:
with artifact_lock(str(out_dir)):
stale = [
p for p in out_dir.iterdir()
if p.is_file() and (
p.suffix == ".zip"
or ".zip." in p.name
or p.name == "SHA256SUMS.txt"
)
]
for p in stale:
p.unlink()
if stale:
print(f"Purged {len(stale)} stale pack file(s) from {out_dir}/")
except ArtifactLockBusy as exc:
print(f"ERROR: {exc}")
sys.exit(1)
pack_cmd = [
sys.executable,
"scripts/generate_pack.py",
"--all",
"--output-dir",
args.output_dir,
]
if args.include_archived:
pack_cmd.append("--include-archived")
if args.offline:
pack_cmd.append("--offline")
else:
# Step 2 refreshed the trees once: refreshing again here let the
# pack and the manifest describe two revisions of a moving tree.
pack_cmd.append("--no-refresh-data")
if args.target:
pack_cmd.extend(["--target", args.target])
if args.source != "full":
pack_cmd.extend(["--source", args.source])
if args.all_variants:
pack_cmd.append("--all-variants")
ok, pack_output = run(pack_cmd, "4/8 generate packs")
results["generate_packs"] = ok
all_ok = all_ok and ok
else:
print("\n--- 4/8 generate packs: SKIPPED (--skip-packs) ---")
results["generate_packs"] = SKIPPED
# Step 4b: Generate install manifests
if not args.skip_packs:
manifest_cmd = [
sys.executable,
"scripts/generate_pack.py",
"--all",
"--manifest",
"--output-dir",
"install",
]
if args.include_archived:
manifest_cmd.append("--include-archived")
if args.offline:
manifest_cmd.append("--offline")
else:
# Step 2 refreshed the trees once: refreshing again here let the
# pack and the manifest describe two revisions of a moving tree.
manifest_cmd.append("--no-refresh-data")
if args.target:
manifest_cmd.extend(["--target", args.target])
ok, _ = run(manifest_cmd, "4b/8 generate install manifests")
results["generate_manifests"] = ok
all_ok = all_ok and ok
else:
print("\n--- 4b/8 generate install manifests: SKIPPED (--skip-packs) ---")
results["generate_manifests"] = SKIPPED
# Step 4c: Generate target manifests
if not args.skip_packs:
target_cmd = [
sys.executable,
"scripts/generate_pack.py",
"--manifest-targets",
"--output-dir",
"install/targets",
]
ok, _ = run(target_cmd, "4c/8 generate target manifests")
results["generate_target_manifests"] = ok
all_ok = all_ok and ok
else:
print("\n--- 4c/8 generate target manifests: SKIPPED (--skip-packs) ---")
results["generate_target_manifests"] = SKIPPED
# Step 5: Consistency check. verify describes the full pack; a run that
# built only a reduced one (--source platform or truth) has nothing to
# compare it with, and says so instead of failing on an empty parse.
full_pack_built = args.source == "full" or args.all_variants
if pack_output and verify_output and not full_pack_built:
print("\n--- 5/8 consistency check: SKIPPED (no full pack built) ---")
results["consistency"] = SKIPPED
elif pack_output and verify_output:
ok = check_consistency(verify_output, pack_output)
results["consistency"] = ok
all_ok = all_ok and ok
else:
print("\n--- 5/8 consistency check: SKIPPED ---")
results["consistency"] = SKIPPED
# Step 6: Pack integrity (extract + hash verification)
if not args.skip_packs and not full_pack_built:
# The check reads the full pack's name; a reduced pack holds fewer
# files than the platform declares by design.
print("\n--- 6/8 pack integrity: SKIPPED (no full pack built) ---")
results["pack_integrity"] = SKIPPED
elif not args.skip_packs:
integrity_cmd = [
sys.executable,
"scripts/generate_pack.py",
"--all",
"--verify-packs",
"--output-dir",
args.output_dir,
]
if args.include_archived:
integrity_cmd.append("--include-archived")
# Step 4 built with the target, so step 6 has to check against the
# same expectation. Without it a targeted run reported every system
# the target removed as missing from a pack that was exactly right.
if args.target:
integrity_cmd.extend(["--target", args.target])
ok, _ = run(integrity_cmd, "6/8 pack integrity")
results["pack_integrity"] = ok
all_ok = all_ok and ok
else:
print("\n--- 6/8 pack integrity: SKIPPED (--skip-packs) ---")
results["pack_integrity"] = SKIPPED
# Step 7: Generate README
if not args.skip_docs:
ok, _ = run(
[
sys.executable,
"scripts/generate_readme.py",
"--db",
"database.json",
"--platforms-dir",
"platforms",
],
"7/8 generate readme",
)
results["generate_readme"] = ok
all_ok = all_ok and ok
else:
print("\n--- 7/8 generate readme: SKIPPED (--skip-docs) ---")
results["generate_readme"] = SKIPPED
# Step 8: Generate site pages
if not args.skip_docs:
ok, _ = run(
[sys.executable, "scripts/generate_site.py"],
"8/8 generate site",
)
results["generate_site"] = ok
all_ok = all_ok and ok
else:
print("\n--- 8/8 generate site: SKIPPED (--skip-docs) ---")
results["generate_site"] = SKIPPED
# Summary
total_elapsed = time.monotonic() - total_start
print(f"\n{'=' * 60}")
for step, outcome in results.items():
if outcome is SKIPPED:
label = "SKIPPED"
else:
label = "OK" if outcome else "FAILED"
print(f" {step:.<40} {label}")
print(f" {'total':.<40} {total_elapsed:.1f}s")
print(f"{'=' * 60}")
print(f" Pipeline {'COMPLETE' if all_ok else 'FINISHED WITH ERRORS'}")
print(f"{'=' * 60}")
held.close()
sys.exit(0 if all_ok else 1)
if __name__ == "__main__":
from common import run_cli
run_cli(main)