mirror of
https://github.com/Abdess/retroarch_system.git
synced 2026-10-21 03:09:08 -05:00
feat: generate MAME device sets from -listxml
A machine that instantiates a device loads the device's ROMs from <shortname>.zip (src/emu/device.cpp:140-152), and a missing required ROM stops that machine (romload.cpp:529-570, 667-671). The BIOS root scan never sees these sets. mame_hash_scraper --devices --listxml reads them from the -listxml of the release mame.yml pins, refuses another release, and cites each set at the ROM_START block holding its ROMs. The BIOS root merge also wrote refs without the upstream prefix and would have rewritten all 171 of them in mame.yml.
This commit is contained in:
1 parent
7a65a2cf83
commit
8de45b6f45
7 files changed
+974
-13
No files matched your search
@@ -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 <path>:<line>". 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.
|
||||
|
||||
|
||||
@@ -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
|
||||
<device shortname>.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'<mame build="[^"]*\((mame\d+)\)"')
|
||||
_MACHINE = re.compile(rb'<machine name="([^"]+)"([^>]*)>')
|
||||
_ATTR = re.compile(rb'(\w+)="([^"]*)"')
|
||||
_DESCRIPTION = re.compile(rb"<description>(.*?)</description>", re.S)
|
||||
_BIOSSET = re.compile(rb"<biosset\b([^>]*)>")
|
||||
_ROM = re.compile(rb"<rom\b([^>]*)>")
|
||||
|
||||
_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"</machine>", 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"</machine>")
|
||||
buffer = buffer[position:] if position else buffer[-(2 * LISTXML_CHUNK) :]
|
||||
if not tag:
|
||||
raise ValueError("no <mame build=...> 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
|
||||
@@ -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 (
|
||||
|
||||
Reference in new issue
Block a user