diff --git a/scripts/scraper/_hash_merge.py b/scripts/scraper/_hash_merge.py index 6c9f12ce..7c24acad 100644 --- a/scripts/scraper/_hash_merge.py +++ b/scripts/scraper/_hash_merge.py @@ -10,6 +10,7 @@ from __future__ import annotations import json import re import shutil +import textwrap from pathlib import Path from typing import Any @@ -19,6 +20,23 @@ from common import PROFILE_BACKUP_SUFFIX, write_text_atomic, yaml_load _MAME_RELEASE_RE = re.compile(r"^0\.\d+") _MAMEDEV = "https://github.com/mamedev/mame" +DEVICE_SECTION = " # --- Device ROM sets (generated from MAME -listxml) ---" +_FILES_END = re.compile(r"^\S", re.M) + + +def mame_ref_role(profile: dict[str, Any]) -> str: + """Role word a ref read from mamedev/mame carries in this profile. + + A profile that builds from another tree (libretro/mame) cites lines of + mamedev/mame as "upstream :". Written bare, the ref was read + against the libretro port, and every refresh rewrote the 171 prefixed refs + of mame.yml into that form. + """ + upstream = str(profile.get("upstream") or "").rstrip("/") + source = str(profile.get("source") or "").rstrip("/") + if upstream.startswith(_MAMEDEV) and source and source != upstream: + return "upstream " + return "" def merge_mame_profile( @@ -63,9 +81,10 @@ def merge_mame_profile( updated_bios: list[dict] = [] matched_names: set[str] = set() + role = mame_ref_role(profile) for set_name, set_data in hashes.get("bios_sets", {}).items(): contents = _build_contents(set_data.get("roms", [])) - source_ref = _build_source_ref(set_data) + source_ref = _build_source_ref(set_data, role) if set_name in existing_by_name: # Update existing entry: preserve manual fields, update contents @@ -208,6 +227,7 @@ def _diff_mame( updated: list[str] = [] unchanged = 0 + role = mame_ref_role(profile) bios_sets = hashes.get("bios_sets", {}) for set_name, set_data in bios_sets.items(): if set_name not in existing_by_name: @@ -217,7 +237,7 @@ def _diff_mame( old_entry = existing_by_name[set_name] new_contents = _build_contents(set_data.get("roms", [])) old_contents = old_entry.get("contents", []) - new_ref = _build_source_ref(set_data) + new_ref = _build_source_ref(set_data, role) # A driver keeping the same ROMs still moves inside its file between # releases. Without this the hashes stay current while the source_ref @@ -348,12 +368,14 @@ def _build_contents(roms: list[dict]) -> list[dict]: return contents -def _build_source_ref(set_data: dict) -> str: +def _build_source_ref(set_data: dict, role: str = "") -> str: source_file = set_data.get("source_file", "") source_line = set_data.get("source_line") - if source_file and source_line is not None: - return f"{source_file}:{source_line}" - return source_file + if not source_file: + return "" + if source_line is not None: + return f"{role}{source_file}:{source_line}" + return f"{role}{source_file}" def _contents_differ(old: list[dict], new: list[dict]) -> bool: @@ -532,11 +554,190 @@ def _append_new_entries(text: str, files: list[dict], original: str) -> str: lines.append(_format_contents(fe["contents"])) if lines: - text = text.rstrip("\n") + "\n" + "\n".join(lines) + "\n" + # New BIOS root sets go before the generated device section, which + # runs to the end of the files list. + at = _device_section_start(text) + if at is None: + text = text.rstrip("\n") + "\n" + "\n".join(lines) + "\n" + else: + text = text[:at].rstrip("\n") + "\n" + "\n".join(lines) + "\n\n" + text[at:] return text +def _device_section_start(text: str) -> int | None: + """Offset of the generated device section header, if the profile has one.""" + match = re.search(rf"^{re.escape(DEVICE_SECTION)}$", text, re.M) + return match.start() if match else None + + +def _device_contents(roms: list[dict], old: list[dict]) -> list[dict]: + """Contents of a device set, keeping the member descriptions a reader wrote.""" + written = {str(c.get("name")): c.get("description") for c in old or []} + contents = _build_contents(roms) + for member in contents: + if written.get(member["name"]): + member["description"] = written[member["name"]] + elif member.get("description"): + # A biosset description is MAME's text, quoted as written. + member["description"] = f"`{member['description']}`" + return contents + + +def _device_note(description: str) -> str: + """The note of a generated entry. The device name is MAME's own text, + quoted verbatim in backticks: `Sun-1 Workstation Monitor; PROM 1&3: Rev AJ` + is a name, not a sentence.""" + return f"Device ROM set, loaded when a machine instantiates `{description}`." + + +def _zip_scalar(name: str) -> str: + """An archive name as the written entries spell it: bare when YAML keeps it.""" + return name if re.fullmatch(r"[a-z0-9_]+\.zip", name) else _yaml_scalar(name) + + +def _format_device_entry(name: str, device: dict, role: str) -> str: + """One generated device entry, in the field order of the written ones.""" + lines = [ + f" - name: {_zip_scalar(name)}", + " required: false", + " category: bios_zip", + ] + if device.get("source_line"): + lines.append(f' source_ref: "{_build_source_ref(device, role)}"') + lines.append(" note: >-") + lines.extend( + textwrap.wrap( + _device_note(device.get("description") or name[:-4]), + width=78, + initial_indent=" ", + subsequent_indent=" ", + break_on_hyphens=False, + ) + ) + lines.append(_format_contents(_device_contents(device["roms"], []))) + return "\n".join(lines) + + +def merge_mame_devices( + profile_path: str, + devices: dict[str, dict], + write: bool = False, +) -> dict[str, list[str]]: + """Merge MAME device ROM sets into a profile. + + *devices* maps a device shortname to its listxml description, its dumped + ROMs and, when found, source_file and source_line. A set the profile + already lists keeps every field a reader wrote: its contents are refreshed + with the member descriptions kept, and a source_ref is added only where it + has none. A set it lacks is written under the generated section, in name + order, with required false. An entry of another category under a device + name is reported, never touched. + """ + path = Path(profile_path) + text = path.read_text(encoding="utf-8") + profile = yaml_load(text) or {} + role = mame_ref_role(profile) + entries: dict[str, list[dict]] = {} + for entry in profile.get("files") or []: + entries.setdefault(str(entry.get("name")), []).append(entry) + + report: dict[str, list[str]] = { + "added": [], "updated": [], "unchanged": [], "conflict": [], + } + contents_patches: list[dict] = [] + ref_patches: dict[str, str] = {} + new_blocks: dict[str, str] = {} + for set_name, device in sorted(devices.items()): + name = f"{set_name}.zip" + existing = entries.get(name) + if not existing: + new_blocks[name] = _format_device_entry(name, device, role) + report["added"].append(set_name) + continue + if any(e.get("category") != "bios_zip" for e in existing): + report["conflict"].append(set_name) + continue + old = existing[0] + contents = _device_contents(device["roms"], old.get("contents") or []) + changed = False + if _contents_differ(old.get("contents") or [], contents): + contents_patches.append({"name": name, "category": "bios_zip", "contents": contents}) + changed = True + if not old.get("source_ref") and device.get("source_line"): + ref_patches[name] = _build_source_ref(device, role) + changed = True + report["updated" if changed else "unchanged"].append(set_name) + + if write and (contents_patches or ref_patches or new_blocks): + patched = _add_source_refs(text, ref_patches) + patched = _patch_bios_entries(patched, contents_patches) + patched = _insert_device_entries(patched, new_blocks) + shutil.copy2(path, path.with_suffix(PROFILE_BACKUP_SUFFIX)) + write_text_atomic(str(path), patched) + return report + + +def _add_source_refs(text: str, refs: dict[str, str]) -> str: + """Give each named entry a source_ref line, right after its category.""" + if not refs: + return text + lines = text.split("\n") + current = "" + out: list[str] = [] + for line in lines: + head = re.match(r"^ - name:\s*(.+?)\s*$", line) + if head: + current = head.group(1).strip('"').strip("'") + out.append(line) + if current in refs and re.match(r"^ category:", line): + out.append(f' source_ref: "{refs[current]}"') + return "\n".join(out) + + +def _insert_device_entries(text: str, blocks: dict[str, str]) -> str: + """Place new device entries in name order inside the generated section. + + The section opens at its header and runs to the end of the files list. + Entries already there stay in the order they are written, each new one + goes before the first whose name sorts after it. + """ + if not blocks: + return text + start = _device_section_start(text) + if start is None: + files_at = re.search(r"^files:[ \t]*$", text, re.M) + if files_at is None or _FILES_END.search(text, files_at.end()): + raise ValueError( + "files: must be the last top-level key to hold a device section" + ) + text = text.rstrip("\n") + "\n\n" + DEVICE_SECTION + "\n" + start = _device_section_start(text) + body_at = text.index("\n", start) + 1 + end_match = _FILES_END.search(text, body_at) + end = end_match.start() if end_match else len(text) + section = text[body_at:end] + starts = [m.start() for m in re.finditer(r"^ - name:", section, re.M)] + lead = (section[: starts[0]] if starts else section).strip("\n") + present = [ + section[a:b].rstrip("\n") for a, b in zip(starts, starts[1:] + [len(section)]) + ] + + def block_name(block: str) -> str: + return block.split("\n", 1)[0].split(":", 1)[1].strip().strip('"') + + merged: list[str] = [] + pending = sorted(blocks) + for block in present: + while pending and pending[0] < block_name(block): + merged.append(blocks[pending.pop(0)]) + merged.append(block) + merged.extend(blocks[name] for name in pending) + body = (lead + "\n" if lead else "") + "\n\n".join(merged) + "\n" + rest = text[end:] + return text[:body_at] + body + ("\n" + rest if rest else "") + + def _yaml_scalar(value: object) -> str: """Quote a scalar so YAML reads back exactly what the source said. diff --git a/scripts/scraper/mame_devices.py b/scripts/scraper/mame_devices.py new file mode 100644 index 00000000..79edc439 --- /dev/null +++ b/scripts/scraper/mame_devices.py @@ -0,0 +1,279 @@ +"""Device ROM sets of one MAME release, read from its -listxml. + +A machine that instantiates a device loads the device's ROMs from +.zip (src/emu/device.cpp:140-152, searchpath), and a missing +ROM that is neither optional nor undumped stops that machine +(src/emu/romload.cpp:529-570, 667-671). The listxml names every such set with +isdevice="yes", its ROMs and the source file defining the device. The lines +cited for a set are the ROM_START block of that file declaring those ROMs, read +at the release tag. A set whose block cannot be told apart is left without +lines and reported. +""" + +from __future__ import annotations + +import html +import io +import re +import zipfile +from collections.abc import Callable, Iterator +from concurrent.futures import ThreadPoolExecutor +from contextlib import contextmanager +from pathlib import Path + +from .mame_parser import _collect_rom_macros, _expand_macros, _parse_rom_entries + +LISTXML_CHUNK = 8 * 1024 * 1024 + +_BUILD = re.compile(rb']*)>') +_ATTR = re.compile(rb'(\w+)="([^"]*)"') +_DESCRIPTION = re.compile(rb"(.*?)", re.S) +_BIOSSET = re.compile(rb"]*)>") +_ROM = re.compile(rb"]*)>") + +_ROM_START = re.compile(r"ROM_START\s*\(\s*(\w+)\s*\)") +_ROM_END = re.compile(r"\bROM_END\b") +_CRC = re.compile(r"CRC\s*\(\s*([0-9a-fA-F]+)\s*\)") +_COMMENT_OR_LITERAL = re.compile( + r'//[^\n]*|/\*.*?\*/|"(?:\\.|[^"\\\n])*"|\'(?:\\.|[^\'\\\n])*\'', re.S +) +_ROM_NAME = re.compile(r"\bROM_NAME\s*\(\s*(\w+)\s*\)") +_CLASS_END = re.compile(r"^\};", re.M) + +# Why a set carries no line: no block of the file holds its first ROM, +# several remain after every narrowing, or the release lacks the file. +ABSENT = "absent" +AMBIGUOUS = "ambiguous" +NO_FILE = "no file" + + +def _attrs(raw: bytes) -> dict[str, str]: + return { + key.decode(): html.unescape(value.decode()) + for key, value in _ATTR.findall(raw) + } + + +def _device_roms(body: bytes) -> list[dict]: + """Dumped ROMs of one machine element, first of each name kept.""" + bios_names = { + attrs.get("name", ""): attrs.get("description", "") + for attrs in map(_attrs, _BIOSSET.findall(body)) + } + roms: dict[str, dict] = {} + for match in _ROM.finditer(body): + attrs = _attrs(match.group(1)) + name, crc = attrs.get("name", ""), attrs.get("crc", "").lower() + if not name or not crc or name in roms: + continue + rom: dict = { + "name": name, + "size": int(attrs.get("size", "0") or 0), + "crc32": crc, + "sha1": attrs.get("sha1", "").lower(), + } + if attrs.get("status") == "baddump": + rom["bad_dump"] = True + description = bios_names.get(attrs.get("bios", ""), "") + if description: + rom["bios_description"] = description + roms[name] = rom + return list(roms.values()) + + +@contextmanager +def open_listxml(path: Path) -> Iterator[io.BufferedIOBase]: + """Byte stream over a -listxml document, plain or as MAME's release zip.""" + if path.suffix.lower() != ".zip": + with path.open("rb") as stream: + yield stream + return + with zipfile.ZipFile(path) as archive: + names = [n for n in archive.namelist() if n.lower().endswith(".xml")] + if len(names) != 1: + raise ValueError(f"{path}: expected one .xml member, found {len(names)}") + with archive.open(names[0]) as stream: + yield stream + + +def scan_device_sets(stream: io.BufferedIOBase) -> tuple[str, dict[str, dict]]: + """Release tag of a -listxml document and its device sets with dumped ROMs. + + The document is read through a sliding window, never whole: the 0.289 + listxml is 311 MB. + """ + tag = "" + devices: dict[str, dict] = {} + buffer = b"" + while chunk := stream.read(LISTXML_CHUNK): + buffer += chunk + if not tag and (build := _BUILD.search(buffer)): + tag = build.group(1).decode() + position = 0 + while match := _MACHINE.search(buffer, position): + end = buffer.find(b"", match.end()) + if end == -1: + break + attrs = _attrs(match.group(2)) + if attrs.get("isdevice") == "yes": + body = buffer[match.end() : end] + roms = _device_roms(body) + if roms: + description = _DESCRIPTION.search(body) + devices[match.group(1).decode()] = { + "description": html.unescape(description.group(1).decode()) + if description else "", + "sourcefile": attrs.get("sourcefile", ""), + "roms": roms, + } + position = end + len(b"") + buffer = buffer[position:] if position else buffer[-(2 * LISTXML_CHUNK) :] + if not tag: + raise ValueError("no header: not a MAME -listxml document") + return tag, devices + + +def source_path(sourcefile: str) -> str: + """Repository path of a listxml sourcefile. + + The listxml writes device files relative to src/ and driver files + relative to src/mame/. + """ + if sourcefile.startswith(("devices/", "emu/", "lib/")): + return f"src/{sourcefile}" + return f"src/mame/{sourcefile}" + + +def blank_comments(text: str) -> str: + """Text with every comment blanked and its newlines kept. + + Line numbers survive, so a block found in the result is cited at its + line in the file. A ROM_START written inside a comment is not a block. + """ + + def blank(match: re.Match[str]) -> str: + found = match.group(0) + if found[0] in "\"'": + return found + return re.sub(r"[^\n]", " ", found) + + return _COMMENT_OR_LITERAL.sub(blank, text) + + +def rom_region_label(text: str, set_name: str) -> str: + """ROM_START label the device class of *set_name* returns, or "". + + DEFINE_DEVICE_TYPE names the class right before the quoted shortname, and + the class's device_rom_region returns ROM_NAME(label), written out of line + or inside the class body. A class inheriting the method gives nothing. + """ + declaration = re.search( + r"DEFINE_DEVICE_TYPE\w*\s*\((?:[^;]*?,)?\s*(\w+)\s*,\s*\"" + re.escape(set_name) + '"', + text, + ) + if declaration is None: + return "" + cls = re.escape(declaration.group(1)) + method = re.search(rf"\b{cls}::device_rom_region\b[^{{]*\{{([^}}]*)\}}", text) + if method is None: + body = re.search(rf"\bclass\s+{cls}\b", text) + if body is None: + return "" + end = _CLASS_END.search(text, body.end()) + scope = text[body.end() : end.start() if end else len(text)] + method = re.search(r"\bdevice_rom_region\b[^{;]*\{([^}]*)\}", scope) + if method is None: + return "" + label = _ROM_NAME.search(method.group(1)) + return label.group(1) if label else "" + + +def find_rom_block( + lines: list[str], set_name: str, roms: list[tuple[str, str]] +) -> tuple[str | None, str]: + """Lines of the ROM_START block declaring these ROMs, or why there are none. + + The span runs from ROM_START to ROM_END: its label is often not the set + name, and the ROM_START line alone spells neither the name nor a CRC. + + *roms* are (name, crc32) pairs. A block qualifies when it holds the first + ROM's CRC, after comments are blanked and the file's ROM macros expanded. + Several qualifying, the narrowest test that leaves one wins: every ROM by + name and CRC, then every CRC, then the block named after the set, then the + block the device class returns from device_rom_region. Two blocks of one + file can load the same dump under different names, as the Amiga keyboards + do with their old and new MCU programs. + """ + text = blank_comments("\n".join(lines)) + macros = _collect_rom_macros(text) + wanted = {(name, int(crc, 16)) for name, crc in roms} + crcs = {crc for _name, crc in wanted} + first = int(roms[0][1], 16) + blocks: list[tuple[str, str, set[tuple[str, int]], set[int]]] = [] + for start in _ROM_START.finditer(text): + end = _ROM_END.search(text, start.end()) + if end is None: + continue + body = _expand_macros(text[start.end() : end.start()], macros) + held = {int(crc, 16) for crc in _CRC.findall(body)} + if first not in held: + continue + pairs = { + (rom["name"], int(rom["crc32"], 16)) + for rom in _parse_rom_entries(body) if rom["crc32"] + } + first_line = text.count("\n", 0, start.start()) + 1 + last_line = text.count("\n", 0, end.start()) + 1 + blocks.append((f"{first_line}-{last_line}", start.group(1), pairs, held)) + if not blocks: + return None, ABSENT + for narrowed in ( + [b for b in blocks if wanted <= b[2]], + [b for b in blocks if crcs <= b[3]], + blocks, + ): + if len(narrowed) == 1: + return narrowed[0][0], "" + named = [b for b in narrowed if b[1] == set_name] + if len(named) == 1: + return named[0][0], "" + if narrowed: + label = rom_region_label(text, set_name) + returned = [b for b in narrowed if label and b[1] == label] + if len(returned) == 1: + return returned[0][0], "" + break + return None, AMBIGUOUS + + +def locate_sources( + devices: dict[str, dict], + fetch: Callable[[str], list[str] | None], + workers: int = 8, +) -> dict[str, str]: + """Write source_file and source_line into each set, return the unplaced. + + *fetch* returns the lines of one repository path at the release, or None + when the path does not exist there. The result maps each set left without + a line to the reason. + """ + paths = sorted({source_path(d["sourcefile"]) for d in devices.values()}) + with ThreadPoolExecutor(max_workers=workers) as pool: + texts = dict(zip(paths, pool.map(fetch, paths))) + unplaced: dict[str, str] = {} + for name, device in sorted(devices.items()): + path = source_path(device["sourcefile"]) + device["source_file"] = path + lines = texts[path] + if lines is None: + unplaced[name] = NO_FILE + continue + line, reason = find_rom_block( + lines, name, [(rom["name"], rom["crc32"]) for rom in device["roms"]] + ) + if line is None: + unplaced[name] = reason + else: + device["source_line"] = line + return unplaced diff --git a/scripts/scraper/mame_hash_scraper.py b/scripts/scraper/mame_hash_scraper.py index 9a9ba104..f384b396 100644 --- a/scripts/scraper/mame_hash_scraper.py +++ b/scripts/scraper/mame_hash_scraper.py @@ -24,7 +24,16 @@ import yaml from common import source_yaml_paths, write_text_atomic, yaml_load -from ._hash_merge import _build_contents, _build_source_ref, compute_diff, merge_mame_profile +import upstream + +from ._hash_merge import ( + _build_contents, + _build_source_ref, + compute_diff, + merge_mame_devices, + merge_mame_profile, +) +from .mame_devices import locate_sources, open_listxml, scan_device_sets from .mame_parser import parse_mame_source_tree log = logging.getLogger(__name__) @@ -377,7 +386,7 @@ def release_profile( } if known.get(name, {}).get("system"): entry["system"] = known[name]["system"] - entry["source_ref"] = _build_source_ref(data) + entry["source_ref"] = _build_source_ref(data, "upstream ") entry["contents"] = _build_contents(data.get("roms", [])) files.append(entry) return { @@ -423,6 +432,50 @@ def _write_release(args: argparse.Namespace) -> None: print(f"{args.profile}: {len(profile['files'])} BIOS root sets at {args.release}") +def _merge_devices(args: argparse.Namespace) -> None: + """Document the device ROM sets of the release mame.yml is pinned on. + + The listxml must be that release's: the lines are read at its tag and + have to be the lines of the profile's upstream_commit. + """ + profile_path = _EMULATORS_DIR / "mame.yml" + profile = yaml_load(profile_path.read_text(encoding="utf-8")) or {} + with open_listxml(args.listxml) as stream: + tag, devices = scan_device_sets(stream) + version = _parse_version_tag(tag) + if str(profile.get("core_version")) != version: + raise RuntimeError( + f"{args.listxml} is MAME {version}, mame.yml is pinned on " + f"{profile.get('core_version')}: refresh the BIOS root sets first" + ) + repo = upstream.parse_repo(_REPO_URL) + commit = upstream.tag_commit(repo, tag, upstream.CACHE_DIR) + if not commit or commit != profile.get("upstream_commit"): + raise RuntimeError( + f"{tag} is {commit}, mame.yml upstream_commit is " + f"{profile.get('upstream_commit')}" + ) + unplaced = locate_sources( + devices, + lambda path: upstream.fetch_file(repo, commit, path, upstream.CACHE_DIR), + ) + report = merge_mame_devices(str(profile_path), devices, write=not args.dry_run) + + print(f"mame-devices: {len(devices)} device ROM sets in the {tag} listxml") + print(f" {profile_path.name}:") + print(f" + {len(report['added'])} added") + print(f" ~ {len(report['updated'])} updated") + print(f" = {len(report['unchanged'])} unchanged") + for set_name in report["conflict"]: + print(f" ! {set_name}.zip is listed under another category, left as written") + if unplaced: + print(f" {len(unplaced)} without source lines:") + for set_name, reason in sorted(unplaced.items()): + print(f" {set_name} ({devices[set_name]['source_file']}, {reason})") + if args.dry_run: + print("(dry run, no files modified)") + + def build_parser() -> argparse.ArgumentParser: parser = argparse.ArgumentParser( allow_abbrev=False, @@ -444,6 +497,19 @@ def build_parser() -> argparse.ArgumentParser: action="store_true", help="re-fetch even if cache is fresh", ) + devices = parser.add_argument_group( + "device ROM sets", + "document every device set with dumped ROMs from the release's " + "-listxml (e.g. --devices --listxml .cache/dats/mame0289lx.zip)", + ) + devices.add_argument( + "--devices", action="store_true", + help="merge device ROM sets into mame.yml instead of BIOS root sets", + ) + devices.add_argument( + "--listxml", type=Path, + help="MAME -listxml, plain or as the release's mameNNNNlx.zip", + ) release = parser.add_argument_group( "release profile", "write the profile of one MAME release a platform builds " @@ -468,6 +534,13 @@ def main() -> None: ) parser = build_parser() args = parser.parse_args() + if args.devices or args.listxml: + if not (args.devices and args.listxml): + parser.error("--devices and --listxml go together") + if args.release or args.profile or args.json or args.force: + parser.error("--devices takes only --listxml and --dry-run") + _merge_devices(args) + return if args.release or args.profile: missing = [ flag for flag, value in ( diff --git a/tests/test_hash_merge.py b/tests/test_hash_merge.py index 7ebae883..275a07bc 100644 --- a/tests/test_hash_merge.py +++ b/tests/test_hash_merge.py @@ -402,7 +402,7 @@ class AReleaseProfileDescribesItsOwnRelease(unittest.TestCase): self.assertEqual(sorted(files), ["neogeo.zip", "newset.zip"]) self.assertEqual(files["neogeo.zip"]["system"], "snk-neogeo-mvs") self.assertNotIn("system", files["newset.zip"]) - self.assertEqual(files["neogeo.zip"]["source_ref"], "src/mame/neogeo/neogeo.cpp:10") + self.assertEqual(files["neogeo.zip"]["source_ref"], "upstream src/mame/neogeo/neogeo.cpp:10") def test_a_release_profile_is_left_out_of_the_current_refresh(self): from scripts.scraper.mame_hash_scraper import _find_mame_profiles diff --git a/tests/test_mame_devices.py b/tests/test_mame_devices.py new file mode 100644 index 00000000..12cc7069 --- /dev/null +++ b/tests/test_mame_devices.py @@ -0,0 +1,377 @@ +"""Device ROM sets read from a MAME -listxml and merged into a profile. + +Synthetic listxml and driver source, no network: the scan keeps only device +sets with dumped ROMs, the source line is the ROM_START block holding them, +and the merge adds sets, refreshes the contents of written ones without +touching what a reader wrote, and leaves BIOS root sets alone. +""" + +from __future__ import annotations + +import io +import json +import tempfile +import unittest +from pathlib import Path + +from scripts.common import yaml_load +from scripts.scraper._hash_merge import ( + DEVICE_SECTION, + mame_ref_role, + merge_mame_devices, + merge_mame_profile, +) +from scripts.scraper.mame_devices import ( + ABSENT, + AMBIGUOUS, + NO_FILE, + find_rom_block, + locate_sources, + open_listxml, + scan_device_sets, + source_path, +) + +LISTXML = b""" + +]> + +\t +\t\tNeo-Geo MV-6F +\t\t +\t +\t +\t\tQSound +\t\t +\t +\t +\t\tKeyboard & Mouse - New +\t\t +\t\t +\t\t +\t +\t +\t\tUndumped +\t\t +\t +\t +\t\tNo ROM +\t + +""" + +KBD_SOURCE = """// keyboards +ROM_START(kbd_old) +\tROM_REGION(0x0800, "mcu", 0) +\tROM_LOAD("6500-1", 0x0000, 0x0800, CRC(4a3fc332) SHA1(83b21d0c8b93fc9b9b3b287fde4ec8f3badac5a2) BAD_DUMP) +\tROM_LOAD("bad.bin", 0x0800, 0x0010, CRC(00000011) SHA1(0000000000000000000000000000000000000011)) +ROM_END + +ROM_START(kbd_new) +\tROM_REGION(0x0800, "mcu", 0) +\tROM_LOAD("6570-036", 0x0000, 0x0800, CRC(4a3fc332) SHA1(83b21d0c8b93fc9b9b3b287fde4ec8f3badac5a2)) +\tROM_LOAD("bad.bin", 0x0800, 0x0010, CRC(00000011) SHA1(0000000000000000000000000000000000000011)) +ROM_END +""".splitlines() + +QSOUND_SOURCE = """#define QSOUND_ROMS \\ +\tROM_LOAD16_WORD_SWAP("dl-1425.bin", 0x0000, 0x6000, CRC(d6cf5ef5) SHA1(555f50fe5cdf127619da7d854c03f4a244a0c501)) + +// ROM_START( commented ) CRC(d6cf5ef5) +ROM_START( qsound ) +\tROM_REGION16_BE( 0x2000, "dsp", 0 ) +\tQSOUND_ROMS +ROM_END +""".splitlines() + +PROFILE = """emulator: MAME +source: "https://github.com/libretro/mame" +upstream: "https://github.com/mamedev/mame" +upstream_commit: "f34f02505e32c1993c6a782b6814232cbfc74e36" +core_version: "0.289" +files: + # SNK Neo Geo + - name: neogeo.zip + required: true + category: bios_zip + system: snk-neogeo-mvs + source_ref: "upstream src/mame/snk/neogeo.cpp:2400" + contents: + - name: "sp-s2.sp1" + description: "Europe MVS (Ver. 2)" + size: 131072 + crc32: "oldcrc32" + + - name: qsound.zip + required: true + category: bios_zip + system: capcom-cps2 + source_ref: "upstream src/devices/sound/qsound.cpp:121-125" + note: >- + Written by hand. + contents: + - name: "dl-1425.bin" + description: "QSound DSP microcode" + size: 24576 + crc32: "d6cf5ef5" +""" + + +def _devices() -> dict[str, dict]: + tag, devices = scan_device_sets(io.BytesIO(LISTXML)) + assert tag == "mame0289" + return devices + + +def _fetch(path: str) -> list[str] | None: + return { + "src/devices/kbd.cpp": KBD_SOURCE, + "src/devices/sound/qsound.cpp": QSOUND_SOURCE, + }.get(path) + + +class ScanKeepsDeviceSetsWithDumpedRoms(unittest.TestCase): + + def test_only_devices_with_a_crc_are_kept(self): + devices = _devices() + self.assertEqual(sorted(devices), ["kbd_new", "qsound"]) + + def test_rom_fields_bios_description_and_bad_dump(self): + kbd = _devices()["kbd_new"] + self.assertEqual(kbd["description"], "Keyboard & Mouse - New") + self.assertEqual(kbd["sourcefile"], "devices/kbd.cpp") + first, second = kbd["roms"] + self.assertEqual( + (first["name"], first["size"], first["crc32"], first["bios_description"]), + ("6570-036", 2048, "4a3fc332", "Version 2"), + ) + self.assertTrue(second["bad_dump"]) + + def test_a_document_without_a_build_is_refused(self): + with self.assertRaises(ValueError): + scan_device_sets(io.BytesIO(b"")) + + def test_the_release_zip_is_read(self): + import zipfile + + with tempfile.TemporaryDirectory() as tmp: + pack = Path(tmp) / "mame0289lx.zip" + with zipfile.ZipFile(pack, "w") as archive: + archive.writestr("mame0289.xml", LISTXML) + with open_listxml(pack) as stream: + tag, devices = scan_device_sets(stream) + self.assertEqual((tag, sorted(devices)), ("mame0289", ["kbd_new", "qsound"])) + + def test_source_paths(self): + self.assertEqual(source_path("devices/sound/qsound.cpp"), "src/devices/sound/qsound.cpp") + self.assertEqual(source_path("snk/neogeo.cpp"), "src/mame/snk/neogeo.cpp") + + +class SourceLinesAreTheBlockHoldingTheRoms(unittest.TestCase): + + def test_found_through_a_rom_macro_past_a_comment(self): + self.assertEqual( + find_rom_block(QSOUND_SOURCE, "qsound", [("dl-1425.bin", "d6cf5ef5")]), + ("5-8", ""), + ) + + def test_a_block_inside_a_comment_is_not_a_block(self): + source = [ + "/* ROM_START( old )", + '\tROM_LOAD("dl-1425.bin", 0, 0x6000, CRC(d6cf5ef5) SHA1(555f50fe5cdf127619da7d854c03f4a244a0c501))', + "ROM_END */", + "ROM_START( current )", + '\tROM_LOAD("dl-1425.bin", 0, 0x6000, CRC(d6cf5ef5) SHA1(555f50fe5cdf127619da7d854c03f4a244a0c501))', + "ROM_END", + ] + self.assertEqual( + find_rom_block(source, "dsp", [("dl-1425.bin", "d6cf5ef5")]), ("4-6", "") + ) + + def test_same_dump_under_two_names_is_told_apart_by_name(self): + line, _ = find_rom_block( + KBD_SOURCE, "kbd_new", [("6570-036", "4a3fc332"), ("bad.bin", "00000011")] + ) + self.assertEqual(line, "8-12") + + def test_not_found(self): + self.assertEqual( + find_rom_block(KBD_SOURCE, "x", [("x.bin", "deadbeef")]), (None, ABSENT) + ) + + def test_two_blocks_and_no_way_to_choose(self): + self.assertEqual( + find_rom_block(KBD_SOURCE, "other", [("bad.bin", "00000011")]), + (None, AMBIGUOUS), + ) + + def test_the_label_device_rom_region_returns_settles_it(self): + source = KBD_SOURCE + [ + 'DEFINE_DEVICE_TYPE(KBD_X, kbd_x_device, "kbd_x", "Keyboard X")', + "const tiny_rom_entry *kbd_x_device::device_rom_region() const", + "{", + "\treturn ROM_NAME(kbd_old);", + "}", + ] + self.assertEqual( + find_rom_block(source, "kbd_x", [("bad.bin", "00000011")]), ("2-6", "") + ) + + def test_locate_sources_reports_what_it_cannot_place(self): + devices = _devices() + devices["ghost"] = { + "description": "Ghost", "sourcefile": "devices/ghost.cpp", + "roms": [{"name": "g.bin", "size": 1, "crc32": "00000001", "sha1": ""}], + } + unplaced = locate_sources(devices, _fetch, workers=2) + self.assertEqual(unplaced, {"ghost": NO_FILE}) + self.assertEqual(devices["qsound"]["source_file"], "src/devices/sound/qsound.cpp") + self.assertEqual(devices["kbd_new"]["source_line"], "8-12") + + +class MergeIntoTheProfile(unittest.TestCase): + + def setUp(self): + self._tmp = tempfile.TemporaryDirectory() + self.path = Path(self._tmp.name) / "mame.yml" + self.path.write_text(PROFILE, encoding="utf-8") + self.devices = _devices() + locate_sources(self.devices, _fetch, workers=2) + + def tearDown(self): + self._tmp.cleanup() + + def _merge(self) -> dict[str, list[str]]: + return merge_mame_devices(str(self.path), self.devices, write=True) + + def _entries(self) -> dict[str, dict]: + data = yaml_load(self.path.read_text(encoding="utf-8")) + return {f["name"]: f for f in data["files"]} + + def test_a_device_with_roms_is_added_under_the_section(self): + report = self._merge() + self.assertEqual(report["added"], ["kbd_new"]) + text = self.path.read_text(encoding="utf-8") + self.assertLess(text.index(DEVICE_SECTION), text.index("- name: kbd_new.zip")) + entry = self._entries()["kbd_new.zip"] + self.assertIs(entry["required"], False) + self.assertEqual(entry["category"], "bios_zip") + self.assertEqual(entry["source_ref"], "upstream src/devices/kbd.cpp:8-12") + self.assertEqual( + entry["note"], + "Device ROM set, loaded when a machine instantiates `Keyboard & Mouse - New`.", + ) + self.assertEqual(entry["contents"][0]["description"], "`Version 2`") + self.assertEqual(entry["contents"][0]["sha1"], "83b21d0c8b93fc9b9b3b287fde4ec8f3badac5a2") + self.assertIs(entry["contents"][1]["bad_dump"], True) + + def test_devices_without_roms_are_not_written(self): + self._merge() + entries = self._entries() + self.assertNotIn("nodump_dev.zip", entries) + self.assertNotIn("empty_dev.zip", entries) + + def test_an_unplaced_set_is_written_without_a_line(self): + del self.devices["kbd_new"]["source_line"] + self._merge() + self.assertNotIn("source_ref", self._entries()["kbd_new.zip"]) + + def test_a_written_entry_gets_its_contents_and_keeps_the_rest(self): + report = self._merge() + self.assertEqual(report["updated"], ["qsound"]) + entry = self._entries()["qsound.zip"] + self.assertIs(entry["required"], True) + self.assertEqual(entry["system"], "capcom-cps2") + self.assertEqual(entry["note"], "Written by hand.") + self.assertEqual(entry["source_ref"], "upstream src/devices/sound/qsound.cpp:121-125") + member = entry["contents"][0] + self.assertEqual(member["description"], "QSound DSP microcode") + self.assertEqual(member["sha1"], "555f50fe5cdf127619da7d854c03f4a244a0c501") + + def test_a_written_entry_without_a_ref_gets_the_found_one(self): + self.path.write_text( + PROFILE.replace(' source_ref: "upstream src/devices/sound/qsound.cpp:121-125"\n', ""), + encoding="utf-8", + ) + self._merge() + self.assertEqual( + self._entries()["qsound.zip"]["source_ref"], + "upstream src/devices/sound/qsound.cpp:5-8", + ) + + def test_bios_roots_are_left_as_written(self): + before = PROFILE[PROFILE.index(" # SNK"):PROFILE.index(" - name: qsound.zip")] + self._merge() + self.assertIn(before, self.path.read_text(encoding="utf-8")) + + def test_a_second_run_changes_nothing(self): + self._merge() + first = self.path.read_text(encoding="utf-8") + report = self._merge() + self.assertEqual((report["added"], report["updated"]), ([], [])) + self.assertEqual(self.path.read_text(encoding="utf-8"), first) + + def test_new_sets_keep_name_order_inside_the_section(self): + self._merge() + self.devices["aaa_dev"] = dict(self.devices["kbd_new"]) + self.devices["zzz_dev"] = dict(self.devices["kbd_new"]) + self._merge() + text = self.path.read_text(encoding="utf-8") + order = [text.index(f"- name: {n}.zip") for n in ("aaa_dev", "kbd_new", "zzz_dev")] + self.assertEqual(order, sorted(order)) + self.assertTrue(text.endswith(' bad_dump: true\n')) + + def test_another_category_under_a_device_name_is_reported(self): + self.path.write_text( + PROFILE + "\n - name: kbd_new.zip\n category: game_data\n", encoding="utf-8" + ) + report = self._merge() + self.assertEqual(report["conflict"], ["kbd_new"]) + + def test_a_new_bios_root_lands_before_the_device_section(self): + self._merge() + hashes = Path(self._tmp.name) / "h.json" + hashes.write_text(json.dumps({ + "version": "0.289", + "commit": "f34f02505e32c1993c6a782b6814232cbfc74e36", + "bios_sets": {"newbios": { + "source_file": "src/mame/x/x.cpp", "source_line": 7, + "roms": [{"name": "n.bin", "size": 4, "crc32": "12345678"}], + }}, + }), encoding="utf-8") + merge_mame_profile(str(self.path), str(hashes), write=True) + text = self.path.read_text(encoding="utf-8") + self.assertLess(text.index("newbios.zip"), text.index(DEVICE_SECTION)) + self.assertEqual(self._entries()["newbios.zip"]["source_ref"], "upstream src/mame/x/x.cpp:7") + self.assertIn("kbd_new.zip", self._entries()) + + +class RefRole(unittest.TestCase): + + def test_a_port_cites_mamedev_as_upstream(self): + self.assertEqual(mame_ref_role({ + "source": "https://github.com/libretro/mame", + "upstream": "https://github.com/mamedev/mame", + }), "upstream ") + + def test_a_profile_with_one_tree_cites_it_bare(self): + self.assertEqual(mame_ref_role({"emulator": "MAME"}), "") + + +class Cli(unittest.TestCase): + + def test_devices_needs_a_listxml(self): + from scripts.scraper.mame_hash_scraper import build_parser, main # noqa: F401 + import sys + from unittest import mock + + with mock.patch.object(sys, "argv", ["x", "--devices"]), \ + mock.patch("sys.stderr", new=io.StringIO()): + with self.assertRaises(SystemExit) as raised: + main() + self.assertEqual(raised.exception.code, 2) + + +if __name__ == "__main__": + unittest.main() diff --git a/wiki/testing-guide.md b/wiki/testing-guide.md index 206e0208..54ad2ff9 100644 --- a/wiki/testing-guide.md +++ b/wiki/testing-guide.md @@ -45,7 +45,7 @@ Every module in `tests/`, by the area it guards. | Hardware targets and systems | synthetic + target files | `test_extract_targets`, `test_lakka_targets`, `test_libretro_slugs`, `test_system_ids`, `test_system_selection`, `test_target_names`, `test_target_scrapers`, `test_target_systems`, `test_target_verify` | | Installer and downloads | loopback server | `test_check_release`, `test_download`, `test_install`, `test_install_cores`, `test_install_experience`, `test_install_selection` | | Truth and native export | synthetic + cached originals | `test_export_counts`, `test_export_native_layer`, `test_export_report`, `test_exporters`, `test_truth_diff`, `test_truth_layout` | -| Scrapers, upstream and provenance | inline sources, synthetic | `test_check_freshness`, `test_dedup`, `test_fbneo_parser`, `test_find_pinned_revision`, `test_hash_merge`, `test_mame_parser`, `test_mame_ref_audit`, `test_profile_sync`, `test_provenance`, `test_refresh_data_dirs`, `test_refresh_failures`, `test_refresh_stale`, `test_retrodeck_scraper_errors`, `test_romset_recipes`, `test_scraper_contract`, `test_upstream`, `test_upstream_index` | +| Scrapers, upstream and provenance | inline sources, synthetic | `test_check_freshness`, `test_dedup`, `test_fbneo_parser`, `test_find_pinned_revision`, `test_hash_merge`, `test_mame_devices`, `test_mame_parser`, `test_mame_ref_audit`, `test_profile_sync`, `test_provenance`, `test_refresh_data_dirs`, `test_refresh_failures`, `test_refresh_stale`, `test_retrodeck_scraper_errors`, `test_romset_recipes`, `test_scraper_contract`, `test_upstream`, `test_upstream_index` | | Site, docs and repository contracts | rendered pages, sources | `test_contributing_text`, `test_data_dir_refs`, `test_dataclass_defaults`, `test_optional_dependencies`, `test_prose_style`, `test_site_availability`, `test_site_cores`, `test_site_exports`, `test_site_platform_hashes`, `test_site_system_links`, `test_site_validation`, `test_site_write`, `test_validate_pr`, `test_validate_pr_hashes`, `test_workflow_paths` | ## Test architecture @@ -99,6 +99,18 @@ sets from driver files. Fixtures are inline C source snippets containing - Multi-region ROM blocks - Macro expansion and edge cases +**test_mame_devices.** Tests the device ROM sets read from a MAME `-listxml` +and merged into a profile, on an inline listxml and driver source. Covers: + +- Only `isdevice="yes"` sets with a dumped ROM kept, from a plain or zipped listxml +- The `ROM_START` block cited for a set: found through a ROM macro, a block + inside a comment ignored, one dump under two names told apart, the label + `device_rom_region` returns, not found and ambiguous +- Merge: a set added under the generated section in name order, a written + entry's contents refreshed with its manual fields kept, BIOS root sets left + as written, a second run changing nothing, a new BIOS root placed before the + section + **test_fbneo_parser.** Tests the FBNeo C source parser that identifies `BDF_BOARDROM` sets. Same inline fixture approach. diff --git a/wiki/tools.md b/wiki/tools.md index 142e2580..b48fb1f9 100644 --- a/wiki/tools.md +++ b/wiki/tools.md @@ -622,7 +622,7 @@ Further options, by script: | `validate_pr.py` | `--changed` takes the BIOS files the branch changes, `--markdown`, `--json` | | `validate_site.py` | `--site-dir` (rendered site), `--config` (mkdocs.yml), `--max-errors N` | | `scraper/coreinfo_scraper.py` | `--compare-db database.json` lists the `.info` firmware the database lacks | -| `scraper/mame_hash_scraper.py` | `--release mameNNNN --profile FILE --source-commit SHA --core NAME --built-by PLATFORM` writes the profile of a MAME build a platform compiles | +| `scraper/mame_hash_scraper.py` | `--release mameNNNN --profile FILE --source-commit SHA --core NAME --built-by PLATFORM` writes the profile of a MAME build a platform compiles. `--devices --listxml FILE` documents in `mame.yml` every device set with dumped ROMs that the release's `-listxml` names (`isdevice="yes"`), with `--dry-run` to preview | | `scraper/romset_dat_importer.py` | `--cache-dir` (downloaded listxml and DATs) | ## Installation tools @@ -743,9 +743,28 @@ stand alone. | `misterfpga_scraper` | BiosDB_MiSTer bios_db.json.zip | JSON in ZIP | | `coreinfo_scraper` | .info files from libretro-core-info | INI-like | | `bizhawk_scraper` | FirmwareDatabase.cs | C# source | -| `mame_hash_scraper` | mamedev/mame source tree. `--release mameNNNN` writes the profile of one release a platform builds | C source (sparse clone) | +| `mame_hash_scraper` | mamedev/mame source tree. `--release mameNNNN` writes the profile of one release a platform builds. `--devices --listxml FILE` adds the device ROM sets of the release's `-listxml` | C source (sparse clone), listxml | | `fbneo_hash_scraper` | FBNeo source tree | C source (sparse clone) | +A machine that instantiates a device loads the device's ROMs from +`.zip`, and a missing ROM that is neither optional nor +undumped stops that machine. The BIOS root scan never sees these sets, so +`--devices` reads them from the `-listxml` of the release `mame.yml` is pinned +on and refuses another release. Each set is cited at the `ROM_START` to +`ROM_END` block holding its ROMs, in the file the listxml names, read at the +release tag through the upstream cache. When several blocks qualify, the one +matching every ROM by name and CRC wins, then the one named after the set, then +the one the device class returns from `device_rom_region`. A set still +undecided is written without a `source_ref` and listed in the output. New sets +go under the `Device ROM sets (generated from MAME -listxml)` comment at the end +of `files:` with `required: false`. A set the profile already lists keeps every +field a reader wrote and only gets its contents refreshed. + +```bash +python -m scripts.scraper.mame_hash_scraper --devices --listxml .cache/dats/mame0289lx.zip --dry-run +python -m scripts.scraper.mame_hash_scraper --devices --listxml .cache/dats/mame0289lx.zip +``` + Internal modules: `base_scraper.py` (abstract base with `_fetch_raw()` caching and shared CLI), `dat_parser.py` (clrmamepro DAT format parser), `logiqx_parser.py` (Logiqx XML DAT parser, used by the dump catalogs),