chore: merge the region reference frames

This commit is contained in:
Abdessamad Derraz committed 2026-10-27 23:07:23 +01:00
commit e6f7ea8f07
794 files changed
+14502 -1807

No files matched your search

+2 -1
View File
@@ -1413,7 +1413,8 @@ class Campaign:
key = self.finding_key(emulator, content, entry, verdict)
previous = self.latest_findings().get(key)
if previous and previous["status"] == status and previous["rule"] == verdict.rule \
and previous["run_shows"] == verdict.value and previous["profile_says"] == says:
and previous["run_shows"] == verdict.value and previous["profile_says"] == says \
and previous["evidence"] == list(verdict.evidence):
return
line = mr.seal_line({
"schema_version": 1, "found_at": now_iso(), "key": key, "emulator": emulator,
+4 -2
View File
@@ -214,12 +214,14 @@ def rule_validation(zeros: Run | None, half: Run | None, none: Run, present: Run
if size is False and "size" in declared:
problems.append(f"the first half of {name} is run, the profile declares a size check")
statement = f"the core {' and '.join(shown)} of {name}"
# What the runs show, for a finding beside what the profile says.
checks = {"content": content, "size": size}
if problems:
return Verdict("validation_contradicted", "validation", said,
return Verdict("validation_contradicted", "validation", checks,
f"{statement}: {'; '.join(problems)}", evidence)
if missing_size:
if not editable:
return Verdict("validation_contradicted", "validation", said,
return Verdict("validation_contradicted", "validation", checks,
f"{statement}: the per-build validation has no size check", evidence)
value = [*(entry.get("validation") or []), "size"]
return Verdict("validation_size_added", "validation", value,
+4
View File
@@ -235,6 +235,10 @@ def _inherited_yaml(platform_name: str, platforms_dir: str) -> tuple[dict, list[
return config, [platform_name]
merged, chain = _inherited_yaml(config["inherits"], platforms_dir)
# A version names one platform's release, never its parent's: Lakka
# read as RetroArch v1.22.2 in its manifest, pack name and site page.
# dat_version stays, it is the version of the inherited list itself.
merged.pop("version", None)
merged.update(
{k: v for k, v in config.items() if k not in ("inherits", "overrides")}
)
+29 -3
View File
@@ -766,6 +766,30 @@ def _bwrap_base(bwrap: str, network: bool) -> list[str]:
return argv
def interpreter_paths() -> list[Path]:
"""The trees the worker's Python lives in, when the sandbox's system
binds do not hold them: CI's setup-python under /opt/hostedtoolcache, a
venv, pyenv or uv. Each link on the way to the interpreter is a path too:
a venv's python points at uv's unversioned name, itself a link to the
versioned tree. Without them the worker could not start in the sandbox,
and every run crashed."""
system = ("/usr/", "/bin/", "/lib")
found: list[Path] = []
candidates = [Path(sys.prefix), Path(sys.base_prefix)]
hop = Path(sys.executable)
for _ in range(16):
candidates.append(hop.parent.parent)
if not hop.is_symlink():
break
target = Path(os.readlink(hop))
hop = target if target.is_absolute() else hop.parent / target
for path in candidates:
path = Path(os.path.abspath(path))
if not str(path).startswith(system) and path not in found and path.exists():
found.append(path)
return found
def sandbox_argv(writable: Sequence[Path], readonly: Sequence[Path] = (),
network: bool = False) -> tuple[list[str], str]:
"""The bubblewrap prefix a third-party program runs under, and its name.
@@ -788,8 +812,10 @@ def sandbox_argv(writable: Sequence[Path], readonly: Sequence[Path] = (),
return [], "environment only: bwrap cannot create a sandbox here"
argv = _bwrap_base(bwrap, network)
for path in readonly:
resolved = Path(path).resolve()
argv += ["--ro-bind", str(resolved), str(resolved)]
# Mounted at the path as the caller names it: a link's own path stays
# usable inside, its target's bytes behind it.
resolved, named = Path(path).resolve(), Path(os.path.abspath(path))
argv += ["--ro-bind", str(resolved), str(named)]
for path in writable:
resolved = Path(path).resolve()
resolved.mkdir(parents=True, exist_ok=True)
@@ -880,7 +906,7 @@ def run_once(args: argparse.Namespace, system_dir: Path, scratch: Path, state_ou
if strace:
command = [strace, "-f", "-qq", "-s", "4096", "-e", "trace=openat,open", "-o", str(trace_path)] + command
writable = [scratch] + [p for p in (png_dir, state_out.parent if state_out else None) if p]
readonly = [Path(__file__), Path(args.core), system_dir]
readonly = [Path(__file__), Path(args.core), system_dir, *interpreter_paths()]
if args.content:
readonly.append(Path(args.content).resolve().parent)
readonly += [Path(value) for _, value in args.env if Path(value).is_absolute() and Path(value).exists()]
+1
View File
@@ -35,6 +35,7 @@ KEY_TYPES: dict[str, dict] = {
**dict.fromkeys((
"installer", "os", "os_name", "stage", "reason", "platform", "message",
"folder", "dest", "system", "name", "bios_dir", "path", "member",
"list_version", "installed_version",
), _TEXT),
"mode": {"enum": ["install", "check", "dry-run", "list"]},
"state": {"enum": ["present", "wrong_hash", "missing"]},
+37 -1
View File
@@ -22,7 +22,7 @@ import ast
import importlib.util
import re
import sys
from pathlib import Path
from pathlib import Path, PurePosixPath
ROOT = Path(__file__).resolve().parent.parent
PAGE = ROOT / "wiki" / "installer.md"
@@ -154,6 +154,40 @@ def route_keys() -> str:
return _table(heading, rows)
def _marker(rule: _install._OsRule) -> str:
"""A card rule's evidence in words: the file or folder, and the line it
holds for a rule that reads one."""
if rule.method == "file_matches":
path, pattern = rule.values
line = re.sub(r"\\s\*\$$", "", re.sub(r"^(\^|\\A)", "", pattern)).replace("\\", "")
return f"`{path}` holding the line `{line}`"
return ", ".join(f"`{value}`" if PurePosixPath(value).suffix else f"`{value}/`"
for value in rule.values)
def _list_words(plat: str) -> str:
name = _install.display_name(plat)
return f"{name}'{'' if name.endswith('s') else 's'} list"
def card_marker_rows() -> list[list[str]]:
"""The files that name a card or a SHARE drive, in the order they are
read: install.py's card rules, then its SHARE drive files."""
rows = [[_marker(rule), rule.name, f"`{rule.bios}`, {_list_words(rule.manifest)}"]
for rule in _install._card_rules()]
hostname = "system.hostname=BATOCERA"
for plat, marker in _install._SHARE_DRIVES:
name = _install.display_name(plat)
evidence = (f"`{marker}` holding `{hostname}` or no hostname line, and no KNULLI file "
"above" if plat == "batocera" else f"`{marker}`")
rows.append([evidence, f"{name} SHARE drive", f"`bios`, {_list_words(plat)}"])
rows.append([f"`{_install._SHARE_DRIVES[0][1]}` with any other hostname, and no KNULLI "
"file above, or the files of two SHARE drives",
"named by neither: the run says which systems it could be",
"the folder of the system named with `--platform`"])
return rows
def render(page: str) -> str:
"""The page with each marked table written from install.py."""
writers = {
@@ -162,6 +196,8 @@ def render(page: str) -> str:
"events": lambda body: _table(["Kind", "Keys", "When"], event_rows(body)),
"pad-routes": lambda body: _table(["Value", "Input", "Set by"], pad_rows(body)),
"route-keys": lambda _body: route_keys(),
"card-markers": lambda _body: _table(["On the card", "System", "Files go to"],
card_marker_rows()),
}
seen = []
+45 -3
View File
@@ -351,6 +351,23 @@ def _platform_pack_stem(
return "_".join(_name_part(n) for n in names) + version_tag
def platform_list_version(config: dict, platform: str) -> str:
"""The version of the list a platform's YAML transcribes: the release
tag the scraper read, or for a list kept on a branch with no release,
the date of the commit it pinned. load_platform_config never passes a
parent's version to a child, Lakka is not RetroArch v1.22.2.
The installer compares it with the version a machine runs, so a missing
value is refused rather than written as a stand-in.
"""
version = str(config.get("version") or "").strip()
if not version:
raise ValueError(
f"{platform}: the platform YAML records no version of its own"
)
return version
# What the builder writes into a platform pack beside the files it carries.
PACK_DOCUMENTS = ("README.txt", "manifest.json")
@@ -2568,6 +2585,9 @@ def _run_manifest_mode(
alias_manifest["display_name"] = alias_cfg.get(
"platform", alias_plat
)
alias_manifest["version"] = platform_list_version(
alias_cfg, alias_plat
)
alias_manifest["base_destination"] = alias_cfg.get(
"base_destination", ""
)
@@ -4600,7 +4620,7 @@ def generate_manifest(
**({"no_game_data": True} if no_game_data else {}),
"platform": platform_name,
"display_name": platform_display,
"version": "1.0",
"version": platform_list_version(config, platform_name),
"generated": now,
"base_destination": base_dest,
"detect": detect,
@@ -4759,12 +4779,31 @@ def pack_systems(install_manifest: dict) -> dict:
("the file list comes from an older release") and offered no Choose
systems, the route of a handheld without Wi-Fi.
"""
return {
files = install_manifest.get("files", [])
entries = [*files, *install_manifest.get("omitted_files", [])]
# What the install list narrows by, the installer's recorded_axes: with
# the pack alone it names that list as the way to --core and --region
# only when the list records them. It named install/misterfpga.json for
# --core, a list that records no core.
records = {"systems"} if any("systems" in entry for entry in entries) else set()
if any(copy.get("cores") for entry in files
for copy in [entry, *entry.get("alternatives", []),
*entry.get("slot_choices", [])]):
records.add("cores")
if records or any("regions" in entry for entry in entries):
records.add("regions")
systems = {
"files": {entry["dest"]: list(entry.get("systems") or [])
for entry in install_manifest.get("files", []) if "systems" in entry},
for entry in files if "systems" in entry},
"system_titles": dict(install_manifest.get("system_titles") or {}),
"system_names": dict(install_manifest.get("system_names") or {}),
"list_records": sorted(records),
}
# The version of the platform's list, which the installer compares with
# the one the machine runs when it has the pack alone.
if install_manifest.get("version"):
systems["list_version"] = install_manifest["version"]
return systems
def annotate_pack_systems(manifest: dict, systems: dict | None) -> None:
@@ -4779,6 +4818,9 @@ def annotate_pack_systems(manifest: dict, systems: dict | None) -> None:
for key in ("system_titles", "system_names"):
if systems.get(key):
manifest[key] = systems[key]
for key in ("list_records", "list_version"):
if key in systems:
manifest[key] = systems[key]
def inject_manifest(zip_path: str, manifest: dict) -> None:
+932 -207
View File
File diff suppressed because it is too large. Load diff
+253 -3
View File
@@ -31,6 +31,7 @@ import yaml
sys.path.insert(0, os.path.dirname(__file__))
import upstream
import version_basis
from common import (
_norm_system_id,
github_token,
@@ -2670,6 +2671,129 @@ def apply_edit(path: Path, new_text: str, expected: dict) -> None:
raise YamlWriteError(f"{path}: structure changed beyond the intended field")
# Key order of a written version_basis, as a reader expects it.
_BASIS_ORDER = ("kind", "why", "repo", "url", "commit", "path", "pattern", "format",
"match", "tag", "distance", "of", "artifact")
def _top_block(lines: list[str], key: str) -> tuple[int, int] | None:
"""Line span of a top-level key and the indented lines under it."""
indices = _top_level_indices(lines, key)
if not indices:
return None
if len(indices) > 1:
raise YamlWriteError(f"{key}: expected one top-level line, found {len(indices)}")
end = indices[0] + 1
while end < len(lines) and (lines[end].startswith((" ", "\t")) or not lines[end].strip()):
end += 1
# Blank lines closing the block belong to the document, not to it.
while end > indices[0] + 1 and not lines[end - 1].strip():
end -= 1
return indices[0], end
def _render_version(value: object) -> list[str]:
if isinstance(value, dict):
return ["core_version:"] + [
f" {mode}: {'null' if v is None else _double_quoted(str(v))}"
for mode, v in value.items()
]
return [f"core_version: {'null' if value is None else _double_quoted(str(value))}"]
def _render_basis(basis: dict) -> list[str]:
def ordered(entry: dict) -> dict:
rank = {key: i for i, key in enumerate(_BASIS_ORDER)}
return dict(sorted(entry.items(), key=lambda item: rank.get(item[0], len(rank))))
def scalar(value: object) -> str:
# Every string quoted: a tag 2407 or 0.10 read back as a number. A
# pattern keeps its backslashes readable in single quotes.
if isinstance(value, bool) or not isinstance(value, str):
return str(value).lower() if isinstance(value, bool) else str(value)
if "\\" in value or '"' in value:
return "'" + value.replace("'", "''") + "'"
return _double_quoted(value)
def block(entry: dict, indent: str) -> list[str]:
# kind, why and repo are words of the schema's enums, written bare.
return [f"{indent}{key}: {value if key in ('kind', 'why', 'repo') else scalar(value)}"
for key, value in ordered(entry).items()]
if "kind" in basis:
return ["version_basis:", *block(basis, " ")]
lines = ["version_basis:"]
for mode, entry in basis.items():
lines += [f" {mode}:", *block(entry, " ")]
return lines
def set_version_fields(text: str, version: object, basis: dict | None) -> str:
"""Rewrite core_version and version_basis, version_basis right after it."""
lines = text.splitlines()
span = _top_block(lines, "core_version")
rendered = _render_version(version)
if span is None:
anchor = _top_block(lines, "upstream_commit") or _top_block(lines, "source_commit") \
or _top_block(lines, "profiled_date") or _top_block(lines, "upstream") \
or _top_block(lines, "source") or _top_block(lines, "alias_of") \
or _top_block(lines, "type")
if anchor is None:
raise YamlWriteError("core_version: no line to write it after")
lines[anchor[1]:anchor[1]] = rendered
else:
lines[span[0]:span[1]] = rendered
if basis is not None:
old = _top_block(lines, "version_basis")
if old is not None:
del lines[old[0]:old[1]]
after = _top_block(lines, "core_version")
lines[after[1]:after[1]] = _render_basis(basis)
return "\n".join(lines) + ("\n" if text.endswith("\n") else "")
def apply_readings(document: dict, readings: list) -> tuple[object, dict | None]:
"""core_version and version_basis as the readings state them.
A reading that could not be made leaves its build mode as written.
"""
version = document.get("core_version")
basis = document.get("version_basis")
basis = dict(basis) if isinstance(basis, dict) else None
if isinstance(version, dict):
version = dict(version)
for reading in readings:
if reading.status not in (version_basis.MATCH, version_basis.DRIFT):
continue
if reading.mode:
if not isinstance(version, dict):
version = {}
version[reading.mode] = reading.read
else:
version = reading.read
if reading.updates and basis is not None:
target = basis if not reading.mode else dict(basis.get(reading.mode) or {})
target.update(dict(reading.updates))
if reading.mode:
basis[reading.mode] = target
return version, basis
def write_version(path: Path, readings: list) -> bool:
"""Set core_version (and a tag basis' tag and distance) from the readings."""
text = path.read_text(encoding="utf-8")
document = yaml.safe_load(text)
version, basis = apply_readings(document, readings)
if version == document.get("core_version") and basis == document.get("version_basis"):
return False
expected = dict(document)
expected["core_version"] = version
if basis is not None:
expected["version_basis"] = basis
apply_edit(path, set_version_fields(text, version, basis), expected)
return True
def backfill_commit(path: Path, sha: str) -> bool:
"""Insert source_commit after profiled_date. False when already present."""
text = path.read_text(encoding="utf-8")
@@ -3313,9 +3437,15 @@ def _needs_review(report: ProfileReport, accept_changed: bool) -> bool:
def bump_commit(
path: Path, report: ProfileReport, accept_changed: bool = False
path: Path, report: ProfileReport, accept_changed: bool = False,
cache_dir: str = DEFAULT_CACHE, offline: bool = False,
) -> bool:
"""Advance source_commit to HEAD when nothing needs a read again."""
"""Advance source_commit to HEAD when nothing needs a read again.
core_version moves with the pin, read again from version_basis at the
new revision. Four bulk advances moved 295 pins and no version: an
advance whose version cannot be read is refused.
"""
if _needs_review(report, accept_changed):
return False
text = path.read_text(encoding="utf-8")
@@ -3375,10 +3505,33 @@ def bump_commit(
new_text = _set_pin(
new_text, document, "upstream_commit", report.upstream_head, "source_commit"
)
readings = version_basis.check(expected, cache_dir, offline, recompute=True)
refused = version_refusal(readings)
if refused:
print(f"{path.stem}: {refused}, its pin is not advanced", file=sys.stderr)
return False
version, basis = apply_readings(expected, readings)
if version != expected.get("core_version") or basis != expected.get("version_basis"):
new_text = set_version_fields(new_text, version, basis)
expected["core_version"] = version
if basis is not None:
expected["version_basis"] = basis
apply_edit(path, new_text, expected)
return True
def version_refusal(readings: list) -> str | None:
"""Why the version at a new pin cannot be stated, or None."""
for reading in readings:
if reading.status == version_basis.NO_BASIS:
return "no version_basis says where its version is read"
if reading.status == version_basis.UNREADABLE:
return f"core_version cannot be read at the new pin: {reading.where}"
if reading.status == version_basis.UNCHECKED:
return f"core_version is {reading.where}, not read from the code at a pin"
return None
def pins_changed(before: dict, after: dict) -> list[str]:
"""The pin fields whose value a write changed."""
return [
@@ -3996,6 +4149,16 @@ def build_parser() -> argparse.ArgumentParser:
"--check-version", action="store_true",
help="compare core_version with the latest tag and release",
)
parser.add_argument(
"--verify-version", action="store_true",
help="read core_version again from version_basis at the pin and "
"exit 1 when the code states another",
)
parser.add_argument(
"--set-version", action="store_true",
help="write core_version from version_basis at the pin, the nearest "
"tag asked of git again",
)
parser.add_argument(
"--detect-new-files", action="store_true",
help="name files upstream loads that the profile does not list",
@@ -4284,7 +4447,8 @@ def _apply_writes(
applied = []
if args.rebase_refs and not report.skipped:
applied = rebase_refs(target, report, args.accept_changed)
moved = bump_commit(target, report, args.accept_changed)
moved = bump_commit(target, report, args.accept_changed,
args.cache_dir, args.offline)
if args.rebase_refs and applied and not moved:
keyed = any(isinstance(profile.get(f), dict) for f in ("source_commit", "upstream_commit"))
reason = (
@@ -4723,6 +4887,32 @@ def _print_triage(
def _refuse_flag_combinations(args: argparse.Namespace) -> None:
"""Exit on a flag the chosen mode would not apply, before any request."""
if args.verify_version or args.set_version:
ignored = [
flag for flag, on in (
("--verify-version", args.verify_version and args.set_version),
("--markdown", args.markdown),
("--fetch-plan", args.fetch_plan),
("--triage", args.triage),
("--strict", args.strict),
("--check-version", args.check_version),
("--detect-new-files", args.detect_new_files),
("--watch-hashes", args.watch_hashes),
("--full-diff", args.full_diff),
("--tree-diff", args.tree_diff),
("--backfill-commits", args.backfill_commits),
("--rebase-refs", args.rebase_refs),
("--attribute-refs", args.attribute_refs),
("--bump-commit", args.bump_commit),
("--realign-prose", args.realign_prose),
("--accept-changed", args.accept_changed),
("--dry-run", args.dry_run and not args.set_version),
) if on
]
if ignored:
mode = "--set-version" if args.set_version else "--verify-version"
print(f"{mode} does not apply {', '.join(ignored)}", file=sys.stderr)
raise SystemExit(1)
if args.fresh and args.offline:
print("--fresh asks the forges again, which --offline forbids.", file=sys.stderr)
raise SystemExit(1)
@@ -4809,6 +4999,63 @@ def _refuse_flag_combinations(args: argparse.Namespace) -> None:
raise SystemExit(1)
def _version_selection(args: argparse.Namespace, profiles: dict) -> dict:
"""--all reaches the alias and test profiles too: each states a version."""
if args.all and not (args.system or args.type_ or args.stale_before or args.limit):
return dict(sorted(profiles.items()))
return select_profiles(profiles, args)
def format_reading(name: str, reading: version_basis.Reading) -> str:
mode = f"[{reading.mode}] " if reading.mode else ""
kind = f"{reading.kind} " if reading.kind else ""
if reading.status == version_basis.DRIFT:
said = f"declared {reading.declared}, code {reading.read}"
elif reading.status == version_basis.MATCH:
said = str(reading.read)
else:
said = f"declared {reading.declared}"
return f"{name}: {mode}{reading.status} {said} ({kind}{reading.where})"
def run_versions(args: argparse.Namespace, profiles: dict) -> None:
"""--verify-version and --set-version: core_version against the code."""
selected = _version_selection(args, profiles)
if args.set_version and not args.dry_run and not args.force:
_refuse_dirty_tree(args.emulators_dir, list(selected))
counts: Counter = Counter()
rows = []
failed = False
for name, profile in selected.items():
readings = version_basis.check(
profile, args.cache_dir, args.offline, profiles, recompute=args.set_version
)
for reading in readings:
counts[reading.status] += 1
rows.append({"profile": name, **asdict(reading)})
if not args.as_json and (not args.changed_only or reading.status != version_basis.MATCH):
print(format_reading(name, reading))
if args.set_version:
path = Path(args.emulators_dir) / f"{name}.yml"
if args.dry_run:
continue
try:
if write_version(path, readings):
print(f"{name}: core_version written")
except YamlWriteError as exc:
print(f"{name}: write refused: {exc}", file=sys.stderr)
failed = True
elif any(r.status in version_basis.FAILING for r in readings):
failed = True
if args.as_json:
print(json.dumps(rows, indent=2))
else:
print(", ".join(f"{count} {status}" for status, count in sorted(counts.items())),
file=sys.stderr)
if failed:
raise SystemExit(1)
def main() -> None:
"""Build one report per selected profile, then apply writes and output."""
args = build_parser().parse_args()
@@ -4818,6 +5065,9 @@ def main() -> None:
# trzy/Supermodel and VICE read as behind their real HEAD.
upstream.MOVING_TTL = 0
profiles = load_emulator_profiles(args.emulators_dir, skip_aliases=False)
if args.verify_version or args.set_version:
run_versions(args, profiles)
return
selected = select_profiles(profiles, args)
_check_quota(len(selected), args.offline)
+10 -6
View File
@@ -158,7 +158,12 @@ def merge_fbneo_profile(
profile = _load_yaml(profile_path)
hashes = _load_json(hashes_path)
profile["core_version"] = hashes.get("version", profile.get("core_version"))
# core_version stays as written: it is the version of the code at the
# profile's own pin, read through version_basis (profile_sync
# --set-version). The hashes come from finalburnneo/FBNeo's tip, a
# revision the libretro/FBNeo pin does not name: writing its version
# left fbneo, fbneo_cps12 and fbneo_neogeo on a release their pin
# never was.
files = profile.get("files", [])
archive_files, non_archive = _split_files(files, lambda f: "archive" in f)
@@ -1060,17 +1065,16 @@ def _backup_and_write_fbneo(path: str, data: dict, hashes: dict) -> None:
"""Write merged FBNeo profile using text-based patching.
FBNeo profiles have individual ROM entries with archive: field.
Patches core_version, the content values of existing entries that
changed, and appends new ROM entries. Advancing core_version alone left
a profile claiming a release whose crc32 it did not carry. Comments and
curated source_refs stay as written.
Patches the content values of existing entries that changed and
appends new ROM entries. core_version is the pin's, never written here.
Comments and curated source_refs stay as written.
"""
p = Path(path)
backup = p.with_suffix(PROFILE_BACKUP_SUFFIX)
shutil.copy2(p, backup)
original = p.read_text(encoding="utf-8")
patched = _patch_core_version(original, data.get("core_version", ""))
patched = original
# Identify new ROM entries by comparing parsed data keys, not text search
existing_data = yaml_load(original) or {}
+26 -70
View File
@@ -11,6 +11,7 @@ import argparse
import json
import logging
import os
import re
import shutil
import subprocess
import sys
@@ -20,7 +21,7 @@ from typing import Any
import yaml
from common import github_token, source_yaml_paths, write_text_atomic, yaml_load
from common import source_yaml_paths, write_text_atomic, yaml_load
from scripts.scraper._hash_merge import compute_diff, merge_fbneo_profile
from scripts.scraper.fbneo_parser import parse_fbneo_source_tree
@@ -51,7 +52,7 @@ def _is_cache_fresh() -> bool:
def _sparse_clone() -> None:
"""Sparse clone FBNeo repo, checking out only src/burn/drv."""
"""Sparse clone FBNeo repo: src/burn/drv, and in cone mode src/burn/version.h."""
if CLONE_DIR.exists():
shutil.rmtree(CLONE_DIR)
@@ -74,7 +75,7 @@ def _sparse_clone() -> None:
)
subprocess.run(
["git", "sparse-checkout", "set", "src/burn/drv", "src/burner/resource.h"],
["git", "sparse-checkout", "set", "src/burn/drv"],
cwd=CLONE_DIR,
check=True,
capture_output=True,
@@ -82,31 +83,28 @@ def _sparse_clone() -> None:
)
# src/burn/version.h, as the libretro core prints it: "v%x.%x.%x.%02x" of
# the BCD BURN_VERSION, so VER_ALPHA 3 reads 1.0.0.03. FBNeo bumps this file
# without tagging: the last tag (v1.0.0.02) was 9 337 commits behind it.
_VERSION_H = re.compile(
r"#define\s+VER_MAJOR\s+(\d+)\s+#define\s+VER_MINOR\s+(\d+)\s+"
r"#define\s+VER_BETA\s+(\d+)\s+#define\s+VER_ALPHA\s+(\d+)"
)
def version_from_header(text: str) -> str:
"""The version src/burn/version.h states."""
found = _VERSION_H.search(text)
if not found:
raise RuntimeError("src/burn/version.h states no VER_MAJOR..VER_ALPHA")
major, minor, beta, alpha = (int(group) for group in found.groups())
return f"v{major}.{minor}.{beta}.{alpha:02d}"
def _extract_version() -> tuple[str, str]:
"""Extract version tag and commit SHA from the cloned repo.
Returns (version, commit_sha). Falls back to resource.h if no tag.
"""
result = subprocess.run(
["git", "describe", "--tags", "--abbrev=0"],
cwd=CLONE_DIR,
capture_output=True,
text=True,
)
# Prefer real version tags over pseudo-tags like "latest"
version = "unknown"
if result.returncode == 0:
tag = result.stdout.strip()
if tag and tag != "latest":
version = tag
# Fallback: resource.h
if version == "unknown":
version = _version_from_resource_h()
# Last resort: use GitHub API for latest real release tag
if version == "unknown":
version = _version_from_api()
"""The version the cloned source states, and its commit."""
header = CLONE_DIR / "src" / "burn" / "version.h"
version = version_from_header(header.read_text(encoding="utf-8", errors="replace"))
sha_result = subprocess.run(
["git", "rev-parse", "HEAD"],
cwd=CLONE_DIR,
@@ -114,49 +112,7 @@ def _extract_version() -> tuple[str, str]:
text=True,
check=True,
)
commit = sha_result.stdout.strip()
return version, commit
def _version_from_api() -> str:
"""The latest real release tag, or an error: "unknown" was written as
core_version and cached for a day."""
import json as json_mod
import urllib.error
import urllib.request
headers = {"User-Agent": "retrobios-scraper/1.0"}
token = github_token()
if token:
headers["Authorization"] = f"Bearer {token}"
req = urllib.request.Request(
"https://api.github.com/repos/finalburnneo/FBNeo/tags?per_page=10", headers=headers
)
try:
with urllib.request.urlopen(req, timeout=10) as resp:
tags = json_mod.loads(resp.read())
except (urllib.error.URLError, OSError, ValueError) as exc:
raise RuntimeError(f"cannot determine the FBNeo version: {exc}") from exc
for tag in tags:
if tag["name"] != "latest" and tag["name"].startswith("v"):
return tag["name"]
raise RuntimeError("cannot determine the FBNeo version: no tag, resource.h or release")
def _version_from_resource_h() -> str:
"""Fallback: parse VER_FULL_VERSION_STR from resource.h."""
resource_h = CLONE_DIR / "src" / "burner" / "resource.h"
if not resource_h.exists():
return "unknown"
text = resource_h.read_text(encoding="utf-8", errors="replace")
for line in text.splitlines():
if "VER_FULL_VERSION_STR" in line:
parts = line.split('"')
if len(parts) >= 2:
return parts[1]
return "unknown"
return version, sha_result.stdout.strip()
def _cleanup() -> None:
+7
View File
@@ -467,6 +467,13 @@ def release_profile(
"source_commit": source_commit,
"upstream_commit": upstream_commit,
"core_version": version,
# Where profile_sync reads it again: the makefile of the libretro/mame
# tree the platform builds, at source_commit.
"version_basis": {
"kind": "file",
"path": "makefile",
"pattern": '#define BARE_BUILD_VERSION "([^"]+)"',
},
"display_name": f"Arcade (MAME {version})",
"cores": cores,
"built_by": built_by,
+5 -5
View File
@@ -215,12 +215,12 @@ class Scraper(BaseScraper):
if not entry["docs"]:
del entry["docs"]
# The BiosDB has no release: the date it was built is the only
# version the list carries, and the manifest needs one.
timestamp = data.get("timestamp")
released = (
datetime.fromtimestamp(timestamp, timezone.utc).strftime("%Y-%m-%d")
if isinstance(timestamp, (int, float))
else ""
)
if not isinstance(timestamp, (int, float)):
raise RuntimeError(f"{self.url}: the BiosDB records no timestamp")
released = datetime.fromtimestamp(timestamp, timezone.utc).strftime("%Y-%m-%d")
return {
"platform": "MiSTer FPGA",
+112
View File
@@ -0,0 +1,112 @@
"""Long lists as one filterable table, each row opening to its details.
MAME's card pages read 13,000 to 16,000 words, one card after another,
before a reader found the archive they came for. Here each entry is one
row: its name, the facts that tell rows apart, and the full card in a
closed block under the name. The table is written as HTML so the block can
hold the card's Markdown. It carries what rehype-stack-tables.mjs gives a
Markdown table: `data-rb-stack` (one block per row on a phone) and the
filter box above, which site.ts wires. Without JavaScript the rows show and
every block opens. A closed block stays in the page, so the browser's find
and the site search still reach every name it holds.
"""
from __future__ import annotations
import html
from dataclasses import dataclass, field
import sitemarkup as markup
# rehype-stack-tables.mjs RECORD_ROWS: from this many rows a table has a
# filter box, and its stacked labels are written by site.ts, not the build.
RECORD_ROWS = 20
# filterPlaceholder in rehype-stack-tables.mjs: the column names offered.
PLACEHOLDER_COLUMNS = 4
@dataclass
class Row:
"""One entry: ``title`` is the inline HTML of its name (a heading, so
the site search lands on it), ``body`` the Markdown lines shown when the
row opens, ``cells`` the HTML of the other columns."""
title: str
body: list[str]
cells: list[str] = field(default_factory=list)
def filter_placeholder(labels: list[str]) -> str:
""""File, contents, status..." as rehype-stack-tables.mjs words it."""
words = [label for label in labels if label][:PLACEHOLDER_COLUMNS]
if not words:
return "Type to filter..."
lowered = [word if index == 0 or not (word[:1].isupper() and word[1:2].islower())
else word[0].lower() + word[1:] for index, word in enumerate(words)]
return f"{', '.join(lowered)}..."
def filter_box(ident: str, heading: str, labels: list[str], rows: int, noun: str = "rows") -> str:
"""The filter box rehype-stack-tables.mjs writes above a long table."""
return (
f'<div class="rb-table-filter"><label for="{ident}">Filter {html.escape(heading)}</label>'
f'<input id="{ident}" type="search" autocomplete="off" '
f'placeholder="{html.escape(filter_placeholder(labels))}">'
f'<span class="rb-table-filter-count" aria-live="polite">{rows:,} {noun}</span></div>'
)
def card_title_and_body(card: list[str]) -> tuple[str, list[str]]:
"""A card's heading line and the lines under it, its wrapping block
(the first line and the last closing tag) taken off."""
title_at = next(index for index, line in enumerate(card) if line.startswith("<h3"))
closing = max(index for index, line in enumerate(card) if line.strip() == "</div>")
body = card[title_at + 1:closing]
while body and not body[0].strip():
body.pop(0)
while body and not body[-1].strip():
body.pop()
return card[title_at], body
def entry_table(rows: list[Row], heads: list[str], ident: str, heading: str,
stacked: list[str] | None = None,
searched: frozenset[str] = frozenset()) -> list[str]:
"""``rows`` under ``heads`` (the first names the column of titles).
``stacked`` gives a column's short label on a phone, as a header's
`title` does in a Markdown table. A column every row leaves empty
(``""``) is not written. The site search skips every column but the
titles and those ``searched`` names: a count or a short hash is no word
anyone types.
"""
keep = [0] + [column for column in range(1, len(heads))
if any(row.cells[column - 1] for row in rows)]
labels = [heads[c] for c in keep]
short = [(stacked or heads)[c] for c in keep]
skip = markup.search_exclude()
# The block holds the box and the table: an open card spans its width.
lines = ["", '<div class="rb-entries-wrap">']
if len(rows) >= RECORD_ROWS:
lines.append(filter_box(ident, heading, labels, len(rows)))
lines.append('<table class="rb-entries" data-rb-stack="true">')
head_cells = ""
for label, name in zip(labels, short):
title = f' title="{html.escape(name)}"' if name != label else ""
head_cells += f'<th scope="col"{title}>{html.escape(label)}</th>'
lines.extend(["<thead>", f"<tr>{head_cells}</tr>", "</thead>", "<tbody>"])
for row in rows:
lines.append("<tr>")
if row.body:
lines.extend(["<td>", '<details class="rb-entry">', f"<summary>{row.title}</summary>",
"", *row.body, "", "</details>", "</td>"])
else:
lines.append(f"<td>{row.title}</td>")
for column in keep[1:]:
label = (f' data-label="{html.escape(short[keep.index(column)])}"'
if len(rows) < RECORD_ROWS else "")
ignored = "" if heads[column] in searched else f" {skip}"
lines.append(f"<td{label}{ignored}>{row.cells[column - 1] or '-'}</td>")
lines.append("</tr>")
lines.extend(["</tbody>", "</table>", "</div>", ""])
return lines
+5 -3
View File
@@ -156,13 +156,15 @@ def image(alt: str, src: str, **attributes: str) -> str:
return f'<img src="{html.escape(src)}" alt="{html.escape(alt)}"{attrs}>'
def hash_value(value: object) -> str:
def hash_value(value: object, copy: bool = False) -> str:
"""A hash printed whole, in the hash style. One element under Starlight:
a span around a code span was 1,400 extra elements on a page of cards."""
a span around a code span was 1,400 extra elements on a page of cards.
*copy* marks it for the copy button the site script adds after it."""
text = str(value).replace("`", "'")
if _engine == "mkdocs":
return f'<span class="rb-hash">`{text}`</span>'
return f'<code class="rb-hash">{html.escape(text, quote=False)}</code>'
marker = ' data-copy=""' if copy else ""
return f'<code class="rb-hash"{marker}>{html.escape(text, quote=False)}</code>'
def block_open(tag: str, attributes: str = "") -> str:
+43 -16
View File
@@ -747,7 +747,7 @@ def short_title(item: dict) -> str:
word = f"issue {issue[1]}" if issue else CHECK_NAMES.get(word, TERMS.get(word, word))
if word not in axis:
axis.append(word)
head = ", ".join(part for part in (f"`{thing}`" if thing else "", game) if part)
head = ", ".join(part for part in (thing, game) if part)
title = f"{head} ({' '.join(axis)})" if axis and head else head or " ".join(axis)
return title or str(item.get("subject") or "")
@@ -901,11 +901,28 @@ def _attr(value: object) -> str:
def _cell(value: object) -> str:
"""Record text placed in raw HTML: escaped, code spans kept, a semicolon
joining two clauses ending a sentence as on the rest of the site."""
"""Record text placed in raw HTML: escaped as text, never read as
Markdown (a run's output is no markup), braces made inert for MDX. A
semicolon joining two clauses ends a sentence, as on the rest of the
site."""
text = " ".join(str(value or "").split())
text = re.sub(r";\s+(\w)", lambda m: ". " + m[1].upper(), text)
return markup.inline_html(text).translate(_BRACES)
return html.escape(text, quote=True).translate(_BRACES)
_SAFE_HREF = re.compile(r"^(?:https?://[^\s\"'<>]+|/[^\s\"'<>]*|\.{0,2}/?[\w.~-][^\s\"'<>:]*)$")
def _href(url: object) -> str | None:
"""A link target from a record: http(s) or relative only, else none."""
text = str(url or "")
return _attr(text) if text and _SAFE_HREF.match(text) and not text.lower().startswith(
("javascript:", "data:", "vbscript:")) else None
def _link(url: object, text: str) -> str:
target = _href(url)
return f'<a href="{target}">{text}</a>' if target else text
def _dots(item: dict) -> str:
@@ -920,7 +937,7 @@ def _dots(item: dict) -> str:
def _chip(item: dict) -> str:
words, kind = RESULT_WORDS.get(item["status"], (item["status"], "muted"))
return f'<span class="rb-badge rb-badge-{kind}">{html.escape(words)}</span>'
return f'<span class="rb-badge rb-badge-{kind}">{_cell(words)}</span>'
def _sentence_html(label: str, value: object, markup_ready: bool = False) -> str:
@@ -931,7 +948,7 @@ def _sentence_html(label: str, value: object, markup_ready: bool = False) -> str
def _details(item: dict) -> list[str]:
core = item["core"]
build = _cell(core["library"]) + (f" at <code>{html.escape(core['commit'][:12])}</code>"
build = _cell(core["library"]) + (f" at <code>{_cell(core['commit'][:12])}</code>"
if core.get("commit") else "")
reps = f"{item['variants']} variant{'s' if item['variants'] != 1 else ''}, " \
f"{item['repetitions']} repetition{'s' if item['repetitions'] != 1 else ''} each"
@@ -945,7 +962,7 @@ def _details(item: dict) -> list[str]:
_sentence_html("Proves", ", ".join(item.get("proves") or [])),
_sentence_html("Ran on", item.get("host_short") or item["host"]),
_sentence_html("Core build", build and f"{build}. {reps}", markup_ready=True),
f'<p><a href="{_attr(item["record_url"])}">Full record</a>{runs}.</p>',
f'<p>{_link(item["record_url"], "Full record")}{runs}.</p>',
]
return [line for line in lines if line]
@@ -966,7 +983,7 @@ def _row(item: dict, title: str, page_host: str, main_core: str) -> list[str]:
elsewhere = (item.get("host_short") or item["host"]) != page_host
title = _cell(title)
if core_label(item) != main_core:
title += f' <small>{html.escape(core_label(item))}</small>'
title += f' <small>{_cell(core_label(item))}</small>'
return [
'<details class="rb-tested-row">',
("<summary>"
@@ -992,6 +1009,13 @@ def by_file(items: list[dict]) -> list[tuple[str, list[dict]]]:
return sorted(groups.items(), key=lambda kv: kv[0] == "")
def folded(item: dict) -> bool:
"""A row the block folds away: a test from before the boot oracle, or a
run that recorded reference frames for the oracle, which a reviewer
reads and a visitor does not."""
return item["status"] == "no verdict" or str(item.get("subject") or "").startswith("references-")
def summary_line(items: list[dict], page_host: str) -> str:
"""N tests, the confirmed and open counts when there are any, the last
run and its computer."""
@@ -1023,15 +1047,15 @@ def tested_section(name: str, items: list[dict]) -> list[str]:
f'<a href="{markup.site_href(f"downloads/measurements/{name}.sqlite")}">SQLite</a>')
ordered = sorted(items, key=lambda i: (i["date"], i["subject"], core_label(i)), reverse=True)
benches = [i for i in ordered if i["kind"] == "bench"]
earlier = [i for i in ordered if i["kind"] != "bench" and i["status"] == "no verdict"]
current = [i for i in ordered if i["kind"] != "bench" and i["status"] != "no verdict"]
earlier = [i for i in ordered if i["kind"] != "bench" and folded(i)]
current = [i for i in ordered if i["kind"] != "bench" and not folded(i)]
lines = ["## Tested", "", '<div class="rb-tested">',
f'<p class="rb-tested-summary">{html.escape(summary_line(items, page_host))} {links}</p>']
f'<p class="rb-tested-summary">{_cell(summary_line(items, page_host))} {links}</p>']
if current or earlier:
lines.append('<div class="rb-tested-head" aria-hidden="true"><span>Test</span>'
"<span>Result</span><span>Variants</span><span>Date</span></div>")
for file, group in by_file(current):
label = f"<code>{html.escape(file)}</code>" if file else "No file varied"
label = f"<code>{_cell(file)}</code>" if file else "No file varied"
lines.append(f'<p class="rb-tested-file">{label}</p>')
for item in group:
lines += _row(item, titles[item["id"]], page_host, main_core)
@@ -1040,9 +1064,12 @@ def tested_section(name: str, items: list[dict]) -> list[str]:
'the same colour for the same result. <span class="rb-dot rb-dot-0 rb-dot-empty">'
"</span> No file staged.</p>")
if earlier:
old = sum(1 for i in earlier if i["status"] == "no verdict")
parts = [f"{old} test{'s' if old != 1 else ''} from before the boot oracle" if old else "",
(f"{len(earlier) - old} reference run{'s' if len(earlier) - old != 1 else ''}"
if len(earlier) - old else "")]
lines += ['<details class="rb-tested-earlier">',
f"<summary>{len(earlier)} earlier test{'s' if len(earlier) != 1 else ''}, "
"before the boot oracle</summary>"]
f"<summary>{_cell(' and '.join(p for p in parts if p))}</summary>"]
for item in earlier:
file = Path(str(item.get("variable") or "")).name
title = f"{file}, {titles[item['id']]}" if file else titles[item["id"]]
@@ -1056,9 +1083,9 @@ def tested_section(name: str, items: list[dict]) -> list[str]:
timing = item.get("timing") or {}
title = _cell(titles[item["id"]])
if core_label(item) != main_core:
title += f' <small>{html.escape(core_label(item))}</small>'
title += f' <small>{_cell(core_label(item))}</small>'
table.append(
f'<tr><td><a href="{_attr(item["record_url"])}">{title}</a></td>'
f'<tr><td>{_link(item["record_url"], title)}</td>'
f'<td>{timing.get("ms_per_frame", 0):.3f}</td><td>{timing.get("fps", 0):.1f}</td>'
f'<td>{timing.get("stdev", 0):.3f}</td><td>{item["repetitions"]}</td></tr>')
lines.append("".join([*table, "</tbody></table>"]))
+27 -12
View File
@@ -425,6 +425,24 @@ def _two_repositories(profile: dict) -> bool:
def _source_ref_markdown(profile: dict, value: str | dict | None) -> str:
"""Render source_ref values as pinned links when their forge is known.
Each part is followed by whose code it cites, in words, when the profile
reads two repositories. source_ref_items() gives the same parts with
that role apart, for a page that names the roles once in a legend.
"""
rendered_groups = []
for label, items in source_ref_items(profile, value):
text = ", ".join(f"{part} ({role})" if role else part for part, role in items)
if label:
text = f"**{label}:** {text}"
rendered_groups.append(text)
return ", ".join(rendered_groups)
def source_ref_items(profile: dict, value: str | dict | None) -> list[tuple[str, list[tuple[str, str]]]]:
"""Each group of source_ref values: its label and its parts, each part
as Markdown with the role_label() of the code it cites ("" when the
profile reads one repository).
A profile can declare two repositories: the libretro port in ``source`` and
the original emulator in ``upstream``. A path links into the one whose
tree SOURCE_PATHS says holds it. Any other path is rendered as plain
@@ -432,18 +450,18 @@ def _source_ref_markdown(profile: dict, value: str | dict | None) -> str:
link to the wrong repository is a false citation. An address inside a
shipped binary reads as one (offset 0x8094-0x80ac in libnes.so).
"""
rendered_groups: list[str] = []
groups: list[tuple[str, list[tuple[str, str]]]] = []
path_re = re.compile(r"^[A-Za-z0-9_.@/+~-]+$")
for label, refs in source_ref_values(value):
candidates = _forge_sources(profile, label)
by_role = role_sources(profile, label)
rendered: list[str] = []
rendered: list[tuple[str, str]] = []
current_role = ""
binary = ""
for part in split_source_ref(refs):
if is_binary_citation(part.path):
text, binary = _binary_citation(part.path, binary)
rendered.append(text)
rendered.append((text, ""))
continue
binary = ""
role, bare = ref_role(part.path)
@@ -473,22 +491,19 @@ def _source_ref_markdown(profile: dict, value: str | dict | None) -> str:
else:
selected = _holder(candidates, part.path)
tag = (f" ({role_label(profile, current_role, part.path)})"
if current_role and _two_repositories(profile) else "")
role = (role_label(profile, current_role, part.path)
if current_role and _two_repositories(profile) else "")
if selected is None:
rendered.append(f"`{display}`{tag}")
rendered.append((f"`{display}`", role))
else:
repo, pin = selected
url = _source_permalink(
repo, pin, real_path, part.start, part.end
)
rendered.append(f"[`{display}`]({url}){tag}")
rendered.append((f"[`{display}`]({url})", role))
text = ", ".join(rendered) if rendered else f"`{refs}`"
if label:
text = f"**{label}:** {text}"
rendered_groups.append(text)
return ", ".join(rendered_groups)
groups.append((label, rendered or [(f"`{refs}`", "")]))
return groups
def _binary_citation(path: str, previous: str) -> tuple[str, str]:
+9 -5
View File
@@ -809,19 +809,23 @@ MOVED_PAGES = {"wiki/getting-started.md": "which-pack.md"}
_CORE_NAME = re.compile(r"[a-z0-9_-]+")
def redirects(profiles: dict, present: set[str]) -> dict[str, str]:
def redirects(profiles: dict, present: set[str],
moved: dict[str, str] | None = None) -> dict[str, str]:
"""Old and alternative addresses, for src/generated/redirects.json.
A page that moved, once its source is gone, and each name a profile's
`cores:` gives the emulator (the buildbot's `mednafen_psx` for
beetle_psx), to the profile's page. A name two profiles claim, or one
that is already a page's address, is left out. Addresses are written
under the site root, without its base."""
under the site root, without its base. ``moved`` adds pages that moved
by generation, old page to new page, the new one with an optional
#fragment."""
def address(rel: str) -> str:
return "/" + page_url(rel)[len(BASE):]
page, _, fragment = rel.partition("#")
return "/" + page_url(page)[len(BASE):] + (f"#{fragment}" if fragment else "")
out = {address(old): address(new) for old, new in MOVED_PAGES.items()
if old not in present and new in present}
out = {address(old): address(new) for old, new in {**MOVED_PAGES, **(moved or {})}.items()
if old not in present and new.partition("#")[0] in present}
pages = {rel.casefold() for rel in present}
claims: dict[str, set[str]] = {}
for key, profile in profiles.items():
+101
View File
@@ -12,6 +12,7 @@ import http.client
import json
import os
import socket
import subprocess
import tempfile
import time
import urllib.error
@@ -1086,3 +1087,103 @@ def commits_touching(
1 for entry in payload
if isinstance(entry, dict) and (entry.get("sha") or entry.get("id")) != base
)
# Commit graphs kept to count the commits between a tag and a pin. A
# treeless bare clone holds commits and tags only: the files a profile reads
# come from fetch_file, by sha.
GIT_DIR = "_git"
def clone_dir(cache_dir: str, repo: Repo) -> Path:
"""Where the commit graph of a repository is kept."""
return Path(cache_dir) / GIT_DIR / repo.host / repo.owner / f"{repo.name}.git"
def _git(args: list[str], cwd: Path | None = None) -> subprocess.CompletedProcess:
env = dict(os.environ, GIT_TERMINAL_PROMPT="0")
return subprocess.run(
["git", *args], cwd=cwd, capture_output=True, text=True, env=env, check=False
)
def _has_commit(clone: Path, sha: str) -> bool:
return _git(["cat-file", "-e", f"{sha}^{{commit}}"], clone).returncode == 0
def _graph(repo: Repo, sha: str, cache_dir: str, offline: bool) -> Path:
"""A clone holding sha and every tag, fetched again once tags may have moved."""
clone = clone_dir(cache_dir, repo)
url = f"https://{repo.host}/{repo.slug}.git"
if not (clone / "HEAD").is_file():
if offline:
raise NotCachedError(f"{repo.slug}: no commit graph in the offline cache")
clone.parent.mkdir(parents=True, exist_ok=True)
result = _git(["clone", "--quiet", "--bare", "--filter=tree:0", url, str(clone)])
if result.returncode:
raise UpstreamError(f"{url}: clone failed: {result.stderr.strip()}")
_git(["fetch", "--quiet", "--filter=tree:0", "--tags", "origin"], clone)
stamp = clone / "FETCH_HEAD"
fresh = stamp.is_file() and time.time() - stamp.stat().st_mtime <= MOVING_TTL
if offline or (fresh and _has_commit(clone, sha)):
if not _has_commit(clone, sha):
raise NotCachedError(f"{repo.slug}@{sha[:8]}: not in the offline commit graph")
return clone
# Tags are fetched again: a tag pushed since the clone can be nearer.
result = _git(
["fetch", "--quiet", "--filter=tree:0", "--force", "--tags", "origin",
"+refs/heads/*:refs/heads/*"], clone,
)
if result.returncode:
raise UpstreamError(f"{url}: fetch failed: {result.stderr.strip()}")
if not _has_commit(clone, sha):
# A commit no branch reaches any more: a rewritten fork, a PR head.
_git(["fetch", "--quiet", "--filter=tree:0", "origin", sha], clone)
if not _has_commit(clone, sha):
raise UpstreamError(f"{url}: {sha[:8]} is on no branch the forge serves")
return clone
def describe(
repo: Repo, sha: str, cache_dir: str, offline: bool = False,
match: str | None = None,
) -> tuple[str, int] | None:
"""The tag nearest to sha in its history, and the commits after it.
git describe counts the commits sha has that the tag lacks, the count a
forge compare reports as ahead. ``match`` is a glob the tag must fit.
None when no tag is reachable from sha.
"""
clone = _graph(repo, sha, cache_dir, offline)
argv = ["describe", "--tags", "--long", "--abbrev=40"]
if match:
argv += ["--match", match]
result = _git([*argv, sha], clone)
if result.returncode:
if any(text in result.stderr for text in (
"No names found", "No tags can describe", "cannot describe",
)):
return None
raise UpstreamError(f"{repo.slug}: describe {sha[:8]}: {result.stderr.strip()}")
tag, _distance, _commit = result.stdout.strip().rsplit("-", 2)
# describe's own count is an estimate across merges (ares: 149 where the
# pin holds 143 commits the tag lacks): the exact count is rev-list's.
counted = _git(["rev-list", "--count", f"refs/tags/{tag}..{sha}"], clone)
if counted.returncode:
raise UpstreamError(f"{repo.slug}: {tag}..{sha[:8]}: {counted.stderr.strip()}")
return tag, int(counted.stdout.strip())
def graph_ancestry(
repo: Repo, base: str, head: str, cache_dir: str, offline: bool = False
) -> tuple[int, int]:
"""(ahead, behind) counted in the commit graph, for a pair a forge
refuses to compare (GitHub answers 422 past a diff size)."""
clone = _graph(repo, head, cache_dir, offline)
counts = []
for spec in (f"{base}..{head}", f"{head}..{base}"):
result = _git(["rev-list", "--count", spec], clone)
if result.returncode:
raise UpstreamError(f"{repo.slug}: {spec}: {result.stderr.strip()}")
counts.append(int(result.stdout.strip()))
return counts[0], counts[1]
+280
View File
@@ -0,0 +1,280 @@
"""Read a profile's core_version from the source its version_basis records.
A profile states the version of the code it was read at. Copied from a
libretro .info or left behind by a pin advance, it named another release:
snes9x said 1.61 at a pin whose snes9x.h states 1.63, and four bulk advances
moved 295 pins without touching one version. version_basis records where the
version is read, so a tool can read it again at any pin:
- ``file``: a pattern in one file of a declared repository at its pin, the
version the program states (a macro, a build file, a package file).
- ``tag``: the nearest tag in the pin's history and the commits after it,
for code that takes its version from git at build time. core_version is
the tag at distance 0, else ``<tag>-<distance>-g<pin7>``, git describe's
form.
- ``alias``: the same binary as another profile, whose version it carries.
- ``package``: an artefact read outside any source tree (an Android package,
a closed binary, a release archive): a version no pin can be asked for.
A pattern reads it from the artefact's name when the name states it.
- ``none``: the code at the pin states no version and no tag reaches it.
core_version is then null.
- ``unverified``: no reading could be made (``why``: no pin, a host the
tools cannot read, a forge gone, no build named as read, no source).
core_version stays as written and is reported unconfirmed.
A profile whose builds live in different repositories keys version_basis by
build mode, the way it keys core_version.
"""
from __future__ import annotations
import re
from dataclasses import dataclass
import upstream
KINDS = ("file", "tag", "alias", "package", "none", "unverified")
# Values copied from a .info or a branch name, which name no version.
PLACEHOLDERS = frozenset({"git", "svn", "unknown", "master", "main", "latest", "dev", "head"})
# Pin field of each repository field a basis may name. A mirror carries the
# source's history, so source_commit pins it too.
PIN_FIELDS = {"source": "source_commit", "source_mirror": "source_commit",
"upstream": "upstream_commit"}
MATCH = "MATCH"
DRIFT = "DRIFT"
UNREADABLE = "UNREADABLE"
NO_BASIS = "NO_BASIS"
UNCHECKED = "UNCHECKED"
FAILING = (DRIFT, UNREADABLE, NO_BASIS)
_FIELD = re.compile(r"\{(\d+)(?::([^}]*))?\}")
class VersionUnreadable(Exception):
"""The basis yields no version at the pin it is asked about."""
@dataclass(frozen=True)
class Reading:
mode: str
kind: str
declared: str | None
read: str | None
where: str
status: str
# What a write sets beside core_version: the tag and distance of a tag basis.
updates: tuple[tuple[str, object], ...] = ()
def bases(profile: dict) -> dict[str, dict]:
"""The basis of each build mode, "" when one covers every build."""
basis = profile.get("version_basis")
if not isinstance(basis, dict):
return {}
if "kind" in basis:
return {"": basis}
return {str(mode): value for mode, value in basis.items() if isinstance(value, dict)}
def declared(profile: dict, mode: str) -> str | None:
"""core_version for a build mode, as written."""
value = profile.get("core_version")
if isinstance(value, dict):
value = value.get(mode)
return None if value is None else str(value)
def _by_mode(profile: dict, field: str, mode: str) -> str | None:
value = profile.get(field)
if isinstance(value, dict):
value = value.get(mode)
return str(value) if value else None
def is_placeholder(value: object) -> bool:
"""A VCS or branch word standing for a version, alone or annotated
("SVN (2015 snapshot)"). The schema refuses the same strings."""
if not isinstance(value, str):
return False
return re.split(r"[\s(]", value.strip(), maxsplit=1)[0].casefold() in PLACEHOLDERS
def format_version(match: re.Match, template: str | None) -> str:
"""The groups of a match laid out by template, `{1}.{2}` or `{4:02d}`.
A format spec applies to the group read as an integer: FBNeo writes its
alpha number on two digits (VER_ALPHA 3 is 1.0.0.03).
"""
if not template:
return match.group(1).strip()
def field(found: re.Match) -> str:
text = match.group(int(found.group(1))) or ""
spec = found.group(2)
return format(int(text, 0), spec) if spec else text.strip()
return _FIELD.sub(field, template)
def read_text_version(text: str, pattern: str, template: str | None) -> tuple[str, int] | None:
"""The version a pattern reads in a file's text, and its line number."""
found = re.search(pattern, text, re.MULTILINE)
if not found:
return None
return format_version(found, template), text.count("\n", 0, found.start()) + 1
def describe_form(tag: str, distance: int, pin: str) -> str:
"""core_version of a tag basis: the tag, or git describe's long form."""
return tag if distance == 0 else f"{tag}-{distance}-g{pin[:7]}"
def _repo_and_pin(
profile: dict, basis: dict, mode: str, cache_dir: str, offline: bool,
) -> tuple[upstream.Repo, str]:
"""The repository a basis reads and the commit it is read at."""
if basis.get("url"):
repo = upstream.parse_repo(str(basis["url"]))
pin = str(basis.get("commit") or "")
where = str(basis["url"])
else:
field = str(basis.get("repo") or "source")
url = _by_mode(profile, field, mode) or ""
repo = upstream.parse_repo(url)
pin = _by_mode(profile, PIN_FIELDS.get(field, ""), mode) or ""
where = url or field
if repo is not None and not pin and field == "upstream" and profile.get("profiled_date"):
# The pin of a second repository falls back on profiled_date,
# as profile_sync resolves it.
pin = upstream.resolve_commit_at(
repo, str(profile["profiled_date"]), cache_dir, offline) or ""
if repo is None:
raise VersionUnreadable(f"{where}: not a forge repository")
if not pin:
raise VersionUnreadable(f"{repo.slug}: no pinned commit")
return repo, pin
def _ancestry(
repo: upstream.Repo, base: str, head: str, cache_dir: str, offline: bool,
) -> tuple[int, int] | None:
"""The forge's compare, else the commit graph: GitHub refuses with 422
a compare whose diff is too large (aps3e's tag and pin)."""
try:
relation = upstream.ancestry(repo, base, head, cache_dir, offline)
except upstream.RateLimitError:
raise
except upstream.UpstreamError:
relation = None
if relation is None:
try:
relation = upstream.graph_ancestry(repo, base, head, cache_dir, offline)
except upstream.NotCachedError:
return None
return relation
def read(
profile: dict, mode: str, basis: dict, cache_dir: str, offline: bool,
profiles: dict | None = None, recompute: bool = False,
) -> Reading:
"""The version the basis yields at the profile's pin, against core_version.
``recompute`` asks git for the nearest tag again, the way a pin advance
has to: a recorded tag only proves its own distance, and a tag released
between the two pins would go unseen.
"""
stated = declared(profile, mode)
kind = str(basis.get("kind") or "")
def reading(value: str | None, where: str, updates: tuple = ()) -> Reading:
status = MATCH if value == stated else DRIFT
return Reading(mode, kind, stated, value, where, status, updates)
def unreadable(reason: str) -> Reading:
return Reading(mode, kind, stated, None, reason, UNREADABLE)
try:
if kind == "none":
if recompute:
# A release tagged since the basis was written gives the
# code a version: the advance has to name it.
repo, pin = _repo_and_pin(profile, basis, mode, cache_dir, offline)
described = upstream.describe(repo, pin, cache_dir, offline)
if described is not None:
return Reading(mode, kind, stated, describe_form(*described, pin),
f"{repo.slug}: tag {described[0]} is now in "
f"{pin[:7]}'s history", DRIFT)
return reading(None, "the code states no version")
if kind == "unverified":
return Reading(mode, kind, stated, None, f"not verified: {basis.get('why')}",
UNCHECKED)
if kind == "package":
artifact = str(basis.get("artifact") or "")
if basis.get("pattern"):
# An artefact named after its version says it in its name:
# advancemame-0.94.0.tar.gz.
found = read_text_version(artifact, str(basis["pattern"]), basis.get("format"))
if found is None:
return unreadable(f"{artifact}: pattern not found")
return reading(found[0], f"named by {artifact}")
return Reading(mode, kind, stated, None, f"read from {artifact}", UNCHECKED)
if kind == "alias":
other = (profiles or {}).get(str(basis.get("of") or ""))
if other is None:
return unreadable(f"alias of {basis.get('of')}: no such profile")
return reading(declared(other, mode) if isinstance(other.get("core_version"), dict)
else declared(other, ""), f"alias of {basis['of']}")
repo, pin = _repo_and_pin(profile, basis, mode, cache_dir, offline)
if kind == "file":
path = str(basis.get("path") or "")
lines = upstream.fetch_file(repo, pin, path, cache_dir, offline)
if lines is None:
return unreadable(f"{repo.slug}@{pin[:7]}:{path}: absent")
found = read_text_version("\n".join(lines), str(basis.get("pattern") or ""),
basis.get("format"))
if found is None:
return unreadable(f"{repo.slug}@{pin[:7]}:{path}: pattern not found")
value, line = found
return reading(value, f"{repo.slug}@{pin[:7]}:{path}:{line}")
if kind == "tag":
if recompute:
described = upstream.describe(repo, pin, cache_dir, offline, basis.get("match"))
if described is None:
return unreadable(f"{repo.slug}@{pin[:7]}: no tag in its history")
tag, distance = described
else:
tag, distance = str(basis.get("tag") or ""), basis.get("distance")
commit = upstream.tag_commit(repo, tag, cache_dir, offline)
if commit is None:
return unreadable(f"{repo.slug}: no tag {tag}")
relation = (0, 0) if commit == pin else _ancestry(
repo, commit, pin, cache_dir, offline)
if relation is None:
return unreadable(f"{repo.slug}: {tag}...{pin[:7]} not comparable")
if relation[1]:
return unreadable(f"{repo.slug}: tag {tag} is not in {pin[:7]}'s history")
if relation[0] != distance:
return Reading(mode, kind, stated, describe_form(tag, relation[0], pin),
f"{repo.slug}: {pin[:7]} is {relation[0]} commits after "
f"{tag}, the basis says {distance}", DRIFT,
(("distance", relation[0]),))
return reading(describe_form(tag, distance, pin),
f"{repo.slug}: tag {tag} + {distance}",
(("tag", tag), ("distance", distance)))
except (upstream.UpstreamError, VersionUnreadable) as exc:
return unreadable(str(exc))
return unreadable(f"unknown kind {kind!r}")
def check(
profile: dict, cache_dir: str, offline: bool, profiles: dict | None = None,
recompute: bool = False,
) -> list[Reading]:
"""One reading per build mode, or a NO_BASIS reading."""
found = bases(profile)
if not found:
return [Reading("", "", declared(profile, ""), None, "no version_basis", NO_BASIS)]
return [read(profile, mode, basis, cache_dir, offline, profiles, recompute)
for mode, basis in found.items()]