feat: rescrape retrobat at its pinned commit

The scraper read emulatorLauncher master on every run, so an extractor
change was diffed together with upstream's commits since the pin. It now
rereads the commit platforms/retrobat.yml pins. --revision master moves
it, and check_freshness and refresh_stale pass that through the
registry's rescrape_args.

The extractors also read master (f681f435): Kronos and YabaSanshiro
fall back to sega_101.bin then mpr-17933.bin when the chosen BIOS is
absent, Ymir writes a forced BIOS only under its override flag, and a
folder created below bios (Ymir's cdb) is a passes_path record. At the
pin the 187 records are unchanged.
This commit is contained in:
Abdessamad Derraz committed 2026-11-11 11:40:05 +01:00
1 parent 5c1d810086
commit 01ca625378
9 files changed
+290 -19

No files matched your search

+3
View File
@@ -585,6 +585,9 @@ platforms:
extract_to: bios/
logo: https://raw.githubusercontent.com/RetroBat-Official/retrobat/main/system/resources/retrobat_logo_notext.png
scraper: retrobat
# The scraper rereads the emulatorLauncher commit retrobat.yml pins;
# check_freshness and refresh_stale move it to the branch head.
rescrape_args: ["--revision", "master"]
source_url: https://raw.githubusercontent.com/RetroBat-Official/emulatorlauncher/master/batocera-systems/Resources/batocera-systems.json
source_format: json
hash_type: md5
+17 -3
View File
@@ -282,7 +282,8 @@ def diff_targets(old: dict, new: dict) -> PlatformDiff:
def _scrape_into_copy(
module: str, source: Path, workdir: Path, timeout: int = 900
module: str, source: Path, workdir: Path, timeout: int = 900,
extra: tuple[str, ...] = (),
) -> tuple[Path, str | None]:
"""Run a scraper's own write path on a throwaway copy of *source*.
@@ -292,7 +293,7 @@ def _scrape_into_copy(
target = workdir / source.name
shutil.copyfile(source, target)
proc = subprocess.run(
[sys.executable, "-m", module, "-o", str(target)],
[sys.executable, "-m", module, "-o", str(target), *extra],
cwd=REPO_ROOT,
capture_output=True,
text=True,
@@ -327,6 +328,17 @@ def _scrapable_platforms(platforms_dir: Path) -> list[tuple[str, str, Path]]:
return rows
def rescrape_args(platforms_dir: Path, name: str) -> tuple[str, ...]:
"""What a rescrape passes so the scraper reads upstream's moving head.
A scraper that rereads the revision its platform file pins by default
(RetroBat) names the ref to move to in the registry's rescrape_args.
"""
with (platforms_dir / "_registry.yml").open(encoding="utf-8") as fh:
registry = (yaml_load(fh) or {}).get("platforms") or {}
return tuple(str(a) for a in (registry.get(name) or {}).get("rescrape_args") or ())
def check_platforms(platforms_dir: Path, workdir: Path, jobs: int) -> list[Finding]:
rows = _scrapable_platforms(platforms_dir)
findings: list[Finding] = []
@@ -334,7 +346,9 @@ def check_platforms(platforms_dir: Path, workdir: Path, jobs: int) -> list[Findi
def run(row: tuple[str, str, Path]) -> Finding:
name, module, path = row
try:
fresh, error = _scrape_into_copy(module, path, workdir)
fresh, error = _scrape_into_copy(
module, path, workdir, extra=rescrape_args(platforms_dir, name)
)
except subprocess.TimeoutExpired:
return Finding("platforms", name, ERROR, detail="scraper timed out")
if error:
+1
View File
@@ -380,6 +380,7 @@ def _command_for(
f"scripts.scraper.{scraper}_scraper",
"-o",
f"platforms/{subject}.yml",
*(str(a) for a in entry.get("rescrape_args") or ()),
]
if area == "targets":
+8 -1
View File
@@ -2,6 +2,7 @@
from __future__ import annotations
import inspect
import json
import re
import sys
@@ -391,10 +392,16 @@ def scraper_cli(
parser.add_argument("--dry-run", action="store_true", help="Show scraped data")
parser.add_argument("--output", "-o", help="Output YAML file")
parser.add_argument("--json", action="store_true", help="Output as JSON")
takes_revision = "revision" in inspect.signature(scraper_class).parameters
if takes_revision:
parser.add_argument(
"--revision",
help="Read this branch head or commit instead of the revision the platform file pins",
)
args = parser.parse_args()
try:
scraper = scraper_class()
scraper = scraper_class(revision=args.revision) if takes_revision else scraper_class()
reqs = scraper.fetch_requirements()
except (OSError, ValueError, RuntimeError) as e:
_exit_on_scrape_error(e)
+91 -10
View File
@@ -1868,13 +1868,27 @@ def _root_writes(src: Source, emulator: str) -> list[Record]:
if root_reach is None:
continue
method = root_reach[0]
# A folder below the root, `string x = Path.Combine(root, "cdb")`
# that the method creates or tests as a directory. A file below it
# is a guarded write's, read by _guarded_write.
folders = {root.group(1): ("bios", root)}
for sub in re.compile(
rf"string (\w+) = Path\.Combine\({root.group(1)}, ((?:{STRING}, )*{STRING})\);"
).finditer(src.text, root.end(), method[1]):
if re.search(
rf"\b(?:\w*CreateDirectory|Directory\.Exists)\({sub.group(1)}\)",
src.text[sub.end() : method[1]],
):
folders[sub.group(1)] = ("/".join(["bios", *_strings(sub.group(2))]), sub)
names = "|".join(map(re.escape, folders))
for write in re.compile(
rf'(\w+)\.WriteValue\("([^"]*)", "([^"]+)", (?:"\'" \+ )?{root.group(1)}(?: \+ "\'")?\);'
rf'(\w+)\.WriteValue\("([^"]*)", "([^"]+)", (?:"\'" \+ )?({names})(?: \+ "\'")?\);'
).finditer(src.text, root.end(), method[1]):
reached = _conditions(src, write.start())
if reached is None:
continue
writer, section, key = write.groups()
writer, section, key, held = write.groups()
folder, declared = folders[held]
target = f"[{section}] {key}" if section else key
records.append(
_record(
@@ -1888,12 +1902,14 @@ def _root_writes(src: Source, emulator: str) -> list[Record]:
[
*reached[2],
src.lines(root.start(), root.end()),
*([src.lines(declared.start(), declared.end())]
if declared is not root else []),
src.lines(write.start(), write.end()),
],
)
),
search_dir="bios",
passes_to=f"{_config_name(src, method, writer)} {target} = bios",
search_dir=folder,
passes_to=f"{_config_name(src, method, writer)} {target} = {folder}",
fails_when_missing=False,
depends_on_option=", ".join(reached[1]) or None,
)
@@ -2376,11 +2392,7 @@ def _option_path_in(src: Source, features: Source, emulator: str) -> list[Record
gate = src.find(
rf'if \(SystemConfig\.isOptSet\("{option}"\)', f"{option} gate", method[0], override.start()
)
gate_block = src.block_at(gate.end())
if not gate_block[0] < override.start() < gate_block[1]:
raise LauncherPatternMissing(
src.path, f"{option} block around line {src.line(override.start())}"
)
gate_block = _gate_span(src, gate, override)
write = re.compile(
rf'(\w+)\.WriteValue\("([^"]*)", "([^"]+)", ([^;\n]*\b{var}\b[^;\n]*)\);'
).search(code, override.end(), method[1])
@@ -2390,6 +2402,8 @@ def _option_path_in(src: Source, features: Source, emulator: str) -> list[Record
guarded = re.compile(rf"if \(File\.Exists\({var}\)[^\n]*\)\s*$").search(
code, method[0], write.start()
)
flag = None if guarded else _flag_guard(code, var, method[0], write.start())
guarded = guarded or flag
skipped = re.findall(
rf'SystemConfig\["{option}"\] != "(\w+)"', code[gate.start() : write.end()]
)
@@ -2397,8 +2411,13 @@ def _option_path_in(src: Source, features: Source, emulator: str) -> list[Record
target = f"{config} [{section}] {key}" if section else f"{config} {key}"
suffix = ", only if the file exists" if guarded else ", written without an existence check"
fails = f"{key} not written" if guarded else "path written without an existence check"
write_span = src.lines(_statement_start(src, write.start()), write.end())
write_span = src.lines(
flag.start() if flag else _statement_start(src, write.start()), write.end()
)
choices, choice_lines = feature_choices(features, emulator, option)
fallback = _fallback_names(src, var, gate_block[1], write.start())
if fallback is not None:
fails = f"{', then '.join(fallback[0])}, instead, else {fails}"
records = []
default = list(
re.compile(rf'string {var} = Path\.Combine\({_BIOS_ROOT}, "([^"]+)"\);').finditer(
@@ -2437,9 +2456,71 @@ def _option_path_in(src: Source, features: Source, emulator: str) -> list[Record
depends_on_option=option,
)
)
if fallback is not None:
names, span = fallback
unset = f"{default[-1].group(1)} absent with {option} unset" if default else None
chosen = f"the {option} file absent" + (
f", {option} not {' or '.join(skipped)}" if skipped else ""
)
records.append(
_record(
emulator,
"standalone",
"names_searched",
names,
_path_parts(src, span, write_span),
case_insensitive=True,
search_dir="bios",
passes_to=f"{target} = <first found>{suffix}",
fails_when_missing=f"{key} not written" if guarded else fails,
depends_on_option=" or ".join(c for c in (unset, chosen) if c),
)
)
return records + _root_writes(src, emulator)
def _gate_span(src: Source, gate: re.Match[str], override: re.Match[str]) -> tuple[int, int]:
"""The span an option gate governs: its block, or the one statement a
braceless if runs, which must be the override."""
paren = gate.start() + len("if ")
after = paren + len(_call_args(src, paren)) + 2
if re.match(r"\s*\{", src.text[after:]):
block = src.block_at(after)
else:
block = (after, _statement_end(src, after))
if not block[0] < override.start() < block[1]:
raise LauncherPatternMissing(
src.path, f"option block around line {src.line(override.start())}"
)
return block
def _flag_guard(code: str, var: str, start: int, end: int) -> re.Match[str] | None:
"""A write under `if (flag)` where flag holds File.Exists of the path."""
flag = re.compile(rf"bool (\w+) = [^;\n]*\bFile\.Exists\({var}\)[^;\n]*;").search(
code, start, end
)
if flag is None or not re.compile(rf"if \({flag.group(1)}\)\s*$", re.M).search(
code, flag.end(), end
):
return None
return flag
def _fallback_names(
src: Source, var: str, start: int, end: int
) -> tuple[list[str], tuple[int, int]] | None:
"""Names tried in bios, first existing one kept, when the chosen file is absent."""
found = re.compile(
rf"if \(([^\n]*!File\.Exists\({var}\)[^\n]*)\)\s*"
rf"{var} = new\[\] \{{ (?P<names>(?:{STRING}(?:, )?)+) \}}\.Select\((?P<arg>\w+) => "
rf"Path\.Combine\({_BIOS_ROOT}, (?P=arg)\)\)\.FirstOrDefault\(File\.Exists\) \?\? {var};"
).search(_code(src), start, end)
if found is None:
return None
return _strings(found.group("names")), src.lines(found.start(), found.end())
OPTION_PATHS = {
f"{GENERATORS}/Kronos.Generator.cs": "kronos",
f"{GENERATORS}/SSF.Generator.cs": "ssf",
+31 -3
View File
@@ -12,6 +12,7 @@ import json
import re
import urllib.error
import urllib.request
from pathlib import Path
try:
from .base_scraper import (
@@ -39,6 +40,7 @@ except ImportError:
from retrobat_launchers import extract_launchers
PLATFORM_NAME = "retrobat"
PLATFORM_FILE = Path(__file__).resolve().parents[2] / "platforms" / "retrobat.yml"
LAUNCHER_REPO = "RetroBat-Official/emulatorlauncher"
SOURCE_URL = (
@@ -132,6 +134,28 @@ def launcher_revision(url: str) -> str:
return found.group(1)
def pinned_revision(platform_file: Path = PLATFORM_FILE) -> str:
"""The emulatorLauncher commit platforms/retrobat.yml was read from.
A rescrape reads this commit unless told to move, so a change to an
extractor shows in the diff alone, without upstream's commits since.
"""
from common import yaml_load
with platform_file.open(encoding="utf-8") as fh:
source = str((yaml_load(fh) or {}).get("source") or "")
return launcher_revision(source)
def resolve_revision(revision: str | None) -> str:
"""The commit to read: the pin when None, a branch's head, or a sha."""
if revision is None:
return pinned_revision()
if re.fullmatch(r"[0-9a-f]{40}", revision):
return revision
return fetch_branch_head(LAUNCHER_REPO, revision)
def fetch_launchers(rev: str) -> list[dict]:
"""Launcher records read from the generators at the commit the BIOS list comes from."""
texts = {
@@ -160,10 +184,14 @@ SYSTEM_SLUG_MAP = {
class Scraper(BaseScraper):
"""Scraper for RetroBat batocera-systems.json."""
def __init__(self, url: str | None = None) -> None:
# batocera-systems.json is read from master: the commit read is recorded.
def __init__(self, url: str | None = None, revision: str | None = None) -> None:
# The BIOS list and the generators are read at one commit, the one
# platforms/retrobat.yml pins unless a revision moves it (--revision
# master reads the branch head). The commit read is recorded.
if url is None:
url = SOURCE_URL.format(rev=fetch_branch_head(LAUNCHER_REPO, "master"))
url = SOURCE_URL.format(rev=resolve_revision(revision))
elif revision is not None:
raise ValueError("pass a url or a revision, not both")
super().__init__(url=url)
self._parsed: dict | None = None
+107
View File
@@ -11,6 +11,7 @@ from __future__ import annotations
import tests # noqa: F401 offline, scratch under tmp/, the suite's own cache
import sys
import tempfile
import textwrap
import unittest
from pathlib import Path
@@ -637,6 +638,57 @@ YMIR = """
}
"""
# emulatorLauncher f681f435: the default falls back to the regional BIOS.
KRONOS_FALLBACK = """
private void SetupConfig(string path)
{
string iniFile = Path.Combine(path, "kronos.ini");
using (var ini = new IniFile(iniFile, IniOptions.KeepEmptyValues))
{
bool hleBios = false;
string bios = Path.Combine(AppConfig.GetFullPath("bios"), "saturn_bios.bin");
if (SystemConfig.isOptSet("kronos_force_bios") && !string.IsNullOrEmpty(SystemConfig["kronos_force_bios"]))
{
if (SystemConfig["kronos_force_bios"] != "hle")
bios = Path.Combine(AppConfig.GetFullPath("bios"), SystemConfig["kronos_force_bios"]);
else
hleBios = true;
}
if (!hleBios && !File.Exists(bios))
bios = new[] { "sega_101.bin", "mpr-17933.bin" }.Select(f => Path.Combine(AppConfig.GetFullPath("bios"), f)).FirstOrDefault(File.Exists) ?? bios;
if (File.Exists(bios))
ini.WriteValue("1.0", "General\\\\Bios", bios.Replace("\\\\", "/"));
}
}
"""
# emulatorLauncher f681f435: a forced BIOS only, written under a flag, and
# a folder below bios for the CD block ROM.
YMIR_OVERRIDE = """
private void SetupConfiguration(string path)
{
string iniFile = Path.Combine(path, "Ymir.toml");
using (var ini = new IniFile(iniFile))
{
string saturnBiosPath = null;
if (SystemConfig.isOptSet("ymir_force_bios") && !string.IsNullOrEmpty(SystemConfig["ymir_force_bios"]))
saturnBiosPath = Path.Combine(AppConfig.GetFullPath("bios"), SystemConfig["ymir_force_bios"]);
bool overrideIpl = saturnBiosPath != null && File.Exists(saturnBiosPath);
ini.WriteValue("System.IPL", "Override", overrideIpl ? "true" : "false");
if (overrideIpl)
ini.WriteValue("System.IPL", "Path", "'" + saturnBiosPath + "'");
string iplPath = AppConfig.GetFullPath("bios");
ini.WriteValue("General.PathOverrides", "IPLROMImages", "'" + iplPath + "'");
string cdbPath = Path.Combine(iplPath, "cdb");
FileTools.TryCreateDirectory(cdbPath);
ini.WriteValue("General.PathOverrides", "CDBlockROMImages", "'" + cdbPath + "'");
}
}
"""
SSF = """
private void SetupConfiguration(string path)
{
@@ -1043,6 +1095,46 @@ class OptionReplacesThePath(unittest.TestCase):
folder["passes_to"], "Ymir.toml [General.PathOverrides] IPLROMImages = bios"
)
def test_kronos_falls_back_to_the_regional_bios(self):
path = f"{G}/Kronos.Generator.cs"
sources = _feature_sources(path, KRONOS_FALLBACK, SATURN_FEATURES)
fixed, option, fallback = rl._option_path_in(
sources[path], sources[rl.FEATURES], "kronos"
)
self.assertEqual(
fixed["fails_when_missing"],
"sega_101.bin, then mpr-17933.bin, instead, else General\\Bios not written",
)
self.assertEqual(option["fails_when_missing"], fixed["fails_when_missing"])
self.assertEqual(fallback["kind"], "names_searched")
self.assertEqual(fallback["names"], ["sega_101.bin", "mpr-17933.bin"])
self.assertEqual(
fallback["passes_to"],
"kronos.ini [1.0] General\\Bios = <first found>, only if the file exists",
)
self.assertEqual(
fallback["depends_on_option"],
"saturn_bios.bin absent with kronos_force_bios unset"
" or the kronos_force_bios file absent, kronos_force_bios not hle",
)
self.assertEqual(fallback["source_ref"], f"{path}:15-19")
def test_ymir_writes_a_forced_bios_under_its_flag(self):
path = f"{G}/Ymir.Generator.cs"
sources = _feature_sources(path, YMIR_OVERRIDE, SATURN_FEATURES)
option, ipl, cdb = rl._option_path_in(sources[path], sources[rl.FEATURES], "ymir")
self.assertEqual(
option["passes_to"],
"Ymir.toml [System.IPL] Path = <bios>/<value>, only if the file exists",
)
self.assertEqual(option["source_ref"], f"{path}:7-13, {rl.FEATURES}:10-13")
self.assertEqual(ipl["passes_to"], "Ymir.toml [General.PathOverrides] IPLROMImages = bios")
self.assertEqual(cdb["search_dir"], "bios/cdb")
self.assertEqual(
cdb["passes_to"], "Ymir.toml [General.PathOverrides] CDBlockROMImages = bios/cdb"
)
self.assertEqual(cdb["source_ref"], f"{path}:15-19")
def test_ssf_has_no_default_name(self):
path = f"{G}/SSF.Generator.cs"
sources = _feature_sources(path, SSF, SATURN_FEATURES)
@@ -1488,6 +1580,21 @@ class Helpers(unittest.TestCase):
src = rl.Source("x.cs", 'void f()\n{\n s = "}";\n}\n')
self.assertEqual(src.lines(*src.block_at(0)), (2, 4))
def test_a_rescrape_reads_the_pinned_commit(self):
from scraper import retrobat_scraper as rs # noqa: PLC0415
pin = "ca486005f598551600d04a33b3156a5d764368d6"
with tempfile.TemporaryDirectory(dir=REPO_ROOT / "tmp") as scratch:
yml = Path(scratch) / "retrobat.yml"
yml.write_text(
"source: https://raw.githubusercontent.com/RetroBat-Official/emulatorlauncher/"
f"{pin}/batocera-systems/Resources/batocera-systems.json\n",
encoding="utf-8",
)
self.assertEqual(rs.pinned_revision(yml), pin)
self.assertEqual(rs.resolve_revision(pin), pin)
self.assertEqual(rs.resolve_revision(None), rs.pinned_revision())
def test_revision_comes_from_the_pinned_url(self):
url = (
"https://raw.githubusercontent.com/RetroBat-Official/emulatorlauncher/"
+11
View File
@@ -17,6 +17,8 @@ import unittest
from pathlib import Path
from unittest import mock
import yaml
REPO_ROOT = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(REPO_ROOT / "scripts"))
@@ -55,6 +57,15 @@ class Planning(unittest.TestCase):
["scripts/verify.py", "--platform", "retrobat"]],
)
def test_a_scraper_reading_its_pin_is_moved_by_rescrape_args(self):
registry = dict(REGISTRY, retrobat={"scraper": "retrobat",
"rescrape_args": ["--revision", "master"]})
jobs, _, _ = rs.plan_jobs([finding("platforms", "retrobat")], registry, {})
self.assertEqual(jobs[0].command[-2:], ["--revision", "master"])
with (REPO_ROOT / "platforms" / "_registry.yml").open(encoding="utf-8") as fh:
real = yaml.safe_load(fh)["platforms"]["retrobat"]
self.assertEqual(real["rescrape_args"], ["--revision", "master"])
def test_a_child_waits_for_its_rescraped_parent(self):
"""retropie inherits retroarch: run beside the parent's rescrape, its
truth read the old parent and its export was refused as stale."""
+21 -2
View File
@@ -650,6 +650,15 @@ class UnreadableReleaseStopsTheScrape(unittest.TestCase):
# Read at a release tag: the pin is the tag, not a branch commit.
TAG_PINNED = {"amberelec", "batocera", "bizhawk", "emuelec", "knulli", "recalbox",
"rocknix", "romm"}
# Reread the commit the platform file pins, moved by the revision the
# registry's rescrape_args names: checked with that revision.
FILE_PINNED = {"retrobat"}
@classmethod
def _construct(cls, module: object, scraper: str) -> object:
if scraper in cls.FILE_PINNED:
return module.Scraper(revision="master")
return module.Scraper()
@staticmethod
def _registry_scrapers() -> list[str]:
@@ -671,7 +680,17 @@ class UnreadableReleaseStopsTheScrape(unittest.TestCase):
with self.subTest(scraper=name), mock.patch(
"urllib.request.urlopen", self._refuse
), self.assertRaises(RuntimeError):
module.Scraper()
self._construct(module, scraper)
def test_a_file_pinned_scraper_reads_its_pin_offline(self):
for scraper in sorted(self.FILE_PINNED):
module = importlib.import_module(f"scraper.{scraper}_scraper")
with self.subTest(scraper=scraper), mock.patch(
"urllib.request.urlopen", self._refuse
):
self.assertEqual(
module.launcher_revision(module.Scraper().url), module.pinned_revision()
)
def test_a_branch_read_records_its_commit(self):
"""System.dat, batocera-systems.json, checkBIOS.sh, the BiosDB and the
@@ -695,7 +714,7 @@ class UnreadableReleaseStopsTheScrape(unittest.TestCase):
name = f"scraper.{scraper}_scraper"
module = importlib.import_module(name)
with self.subTest(scraper=name), mock.patch("urllib.request.urlopen", answer):
instance = module.Scraper()
instance = self._construct(module, scraper)
# RetroDECK reads a tree of manifests and keeps the commit apart.
recorded = getattr(instance, "revision", "") or instance.url
self.assertRegex(recorded, f"(^|/){sha}(/|$)")