mirror of
https://github.com/Abdess/retroarch_system.git
synced 2026-11-13 21:43:55 -06:00
feat: scrape the rom folders of each setup
This commit is contained in:
1 parent
249b189ec9
commit
1aa36258b2
102 files changed
+49969
No files matched your search
@@ -2677,6 +2677,75 @@ SYSTEM_ALIASES: dict[str, str] = {
|
||||
"cps2": "capcom-cps2",
|
||||
"cps3": "capcom-cps3",
|
||||
"pgm": "igs-pgm",
|
||||
# Rom folder names the setups of platforms/romfolders/ give a machine,
|
||||
# each proven by the source_ref of the folder row that names it: the
|
||||
# short tags of Onion, CrossMix-OS, Koriki and the MinUI family, the
|
||||
# number words of Koriki and plumOS, the region and hack folders of
|
||||
# ES-DE, the MSU-1 folders of the ES family, the arcade cores named as
|
||||
# folders.
|
||||
"a2600": "atari-2600",
|
||||
"atari2600": "atari-2600",
|
||||
"fiftytwohundred": "atari-5200",
|
||||
"seventyeighthundred": "atari-7800",
|
||||
"atarixegs": "xegs",
|
||||
"atarijaguarcd": "atari-jaguarcd",
|
||||
"fc": "nintendo-nes",
|
||||
"nesh": "nintendo-nes",
|
||||
"sfc": "nintendo-snes",
|
||||
"supa": "nintendo-snes",
|
||||
"snesh": "nintendo-snes",
|
||||
"snesna": "nintendo-snes",
|
||||
"snesmsu1": "nintendo-snes",
|
||||
"snes-msu1": "nintendo-snes",
|
||||
"msu1": "nintendo-snes",
|
||||
"vb": "nintendo-virtualboy",
|
||||
"gc": "nintendo-gamecube",
|
||||
"gbh": "nintendo-gb",
|
||||
"gb2players": "nintendo-gb",
|
||||
"gbch": "nintendo-gbc",
|
||||
"gbc2players": "nintendo-gbc",
|
||||
"gbah": "nintendo-gba",
|
||||
"mgba": "nintendo-gba",
|
||||
"poke": "nintendo-pokemon-mini",
|
||||
"pkm": "nintendo-pokemon-mini",
|
||||
"md": "sega-megadrive",
|
||||
"genh": "sega-megadrive",
|
||||
"megadriveh": "sega-megadrive",
|
||||
"megadrivejp": "sega-megadrive",
|
||||
"megadrive-japan": "sega-megadrive",
|
||||
"msumd": "sega-megadrive",
|
||||
"mdmsu": "sega-megadrive",
|
||||
"ms": "sega-mastersystem",
|
||||
"mark3": "sega-mastersystem",
|
||||
"gamegearh": "sega-gamegear",
|
||||
"megacdjp": "sega-megacd",
|
||||
"saturnjp": "sega-saturn",
|
||||
"sega32xjp": "sega-32x",
|
||||
"sega32xna": "sega-32x",
|
||||
"thirtytwox": "sega-32x",
|
||||
"segasgone": "sega-sg1000",
|
||||
"ps": "sony-playstation",
|
||||
"pce": "nec-pc-engine",
|
||||
"tg16cd": "nec-pc-engine-cd",
|
||||
"tg-cd": "nec-pc-engine-cd",
|
||||
"turbografxcd": "nec-pc-engine-cd",
|
||||
"neocd": "snk-neogeo-cd",
|
||||
"ngcd": "snk-neogeo-cd",
|
||||
"neogeocdjp": "snk-neogeo-cd",
|
||||
"supervision": "watara-supervision",
|
||||
"cplus4": "commodore-plus4",
|
||||
"cpet": "commodore-pet",
|
||||
"cpc": "amstrad-cpc",
|
||||
"to8": "thomson-to8",
|
||||
"tic": "tic80",
|
||||
"fba": "arcade",
|
||||
"hbmame": "arcade",
|
||||
"advmame": "arcade",
|
||||
"mame-advmame": "arcade",
|
||||
"mame2000": "arcade",
|
||||
"mame2003": "arcade",
|
||||
"mame2003plus": "arcade",
|
||||
"mame2010": "arcade",
|
||||
}
|
||||
|
||||
|
||||
|
||||
@@ -0,0 +1,194 @@
|
||||
"""Rom folders of each setup: where its games live and what each folder holds.
|
||||
|
||||
`platforms/romfolders/<setup>.yml` records what a frontend's own source says
|
||||
about its game folders, read at a pinned revision: the roms root relative to
|
||||
the BIOS folder, each folder with the system name the OS gives it, the
|
||||
extensions it lists and the core it starts by default. Those files hold the
|
||||
OS's facts only. Which repository system a folder is, and which profile its
|
||||
core is, is decided here, through the one alias table in common.py
|
||||
(`SYSTEM_ALIASES`, `_norm_system_id`) and the profiles' `cores:` lists, so a
|
||||
change to either reaches every setup without a rescrape.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import re
|
||||
from dataclasses import dataclass, field
|
||||
from pathlib import Path
|
||||
|
||||
from common import (
|
||||
_norm_system_id,
|
||||
list_registered_platforms,
|
||||
load_emulator_profiles,
|
||||
load_platform_config,
|
||||
native_layer,
|
||||
preferred_profile,
|
||||
upstream_profile_index,
|
||||
)
|
||||
from safeparse import yaml_load
|
||||
|
||||
ROMFOLDERS_DIR = Path("platforms") / "romfolders"
|
||||
SOURCES_DIR = ROMFOLDERS_DIR / "_sources"
|
||||
|
||||
# "<source key> <path>:<line>[-<line>]", several joined by "; ".
|
||||
_REF_PART = re.compile(r"^(?P<key>[a-z0-9][a-z0-9-]*) (?P<path>\S+?)(?::(?P<lines>\d+(?:-\d+)?(?:,\d+(?:-\d+)?)*))?$")
|
||||
|
||||
|
||||
def split_refs(source_ref: str) -> list[tuple[str, str, str | None]]:
|
||||
"""Each (source key, path, lines) a source_ref names, or ValueError."""
|
||||
parts: list[tuple[str, str, str | None]] = []
|
||||
for raw in source_ref.split("; "):
|
||||
match = _REF_PART.match(raw.strip())
|
||||
if not match:
|
||||
raise ValueError(f"source_ref part {raw!r} is not '<source> <path>:<line>'")
|
||||
parts.append((match["key"], match["path"], match["lines"]))
|
||||
return parts
|
||||
|
||||
|
||||
def setup_paths(directory: Path = ROMFOLDERS_DIR) -> list[Path]:
|
||||
"""The setup files of a directory, without the sources file."""
|
||||
return sorted(p for p in Path(directory).glob("*.yml") if not p.name.startswith("_"))
|
||||
|
||||
|
||||
def load_setup(path: Path) -> dict:
|
||||
with Path(path).open(encoding="utf-8") as handle:
|
||||
return yaml_load(handle)
|
||||
|
||||
|
||||
def load_setups(directory: Path = ROMFOLDERS_DIR) -> dict[str, dict]:
|
||||
"""Every setup document of a directory, keyed by its `setup:` id."""
|
||||
return {doc["setup"]: doc for doc in (load_setup(p) for p in setup_paths(directory))}
|
||||
|
||||
|
||||
@dataclass
|
||||
class Resolution:
|
||||
"""What a folder resolved to, and how."""
|
||||
|
||||
systems: list[str]
|
||||
basis: str
|
||||
|
||||
|
||||
@dataclass
|
||||
class SystemResolver:
|
||||
"""Folder and system names to repository system ids.
|
||||
|
||||
The universe is every system a profile or a platform file names. A name
|
||||
resolves, in this order, as one of those ids, as a name a platform's own
|
||||
file gives a system (`native_system`, the folder Batocera calls `psx`),
|
||||
or through `_norm_system_id`, which applies SYSTEM_ALIASES then drops the
|
||||
maker prefix and separators. A name nothing matches stays unresolved:
|
||||
a new abbreviation goes into SYSTEM_ALIASES, never into a table here.
|
||||
|
||||
Ids that normalize alike are one machine spelled several ways
|
||||
(`sega-megadrive`, `sega-mega-drive`, `mega-drive`). Each class answers
|
||||
with one canonical id, the spelling most profiles use, and
|
||||
`equivalents()` gives the whole class back to whoever matches a
|
||||
profile's `systems:` or a manifest's `systems`.
|
||||
"""
|
||||
|
||||
ids: set[str]
|
||||
profile_use: dict[str, int]
|
||||
native: dict[str, dict[str, set[str]]]
|
||||
by_norm: dict[str, set[str]] = field(default_factory=dict)
|
||||
canonical_of: dict[str, str] = field(default_factory=dict)
|
||||
|
||||
def __post_init__(self) -> None:
|
||||
for sid in self.ids:
|
||||
self.by_norm.setdefault(_norm_system_id(sid), set()).add(sid)
|
||||
for members in self.by_norm.values():
|
||||
best = min(members, key=lambda s: (-self.profile_use.get(s, 0), -len(s), s))
|
||||
for sid in members:
|
||||
self.canonical_of[sid] = best
|
||||
|
||||
@classmethod
|
||||
def from_repo(
|
||||
cls, emulators_dir: str = "emulators", platforms_dir: str = "platforms"
|
||||
) -> SystemResolver:
|
||||
profiles = load_emulator_profiles(emulators_dir)
|
||||
use: dict[str, int] = {}
|
||||
for profile in profiles.values():
|
||||
for sid in profile.get("systems") or []:
|
||||
use[str(sid)] = use.get(str(sid), 0) + 1
|
||||
ids = set(use)
|
||||
native: dict[str, dict[str, set[str]]] = {}
|
||||
for platform in list_registered_platforms(platforms_dir, include_archived=True):
|
||||
config = load_platform_config(platform, platforms_dir)
|
||||
ids.update(config.get("systems", {}))
|
||||
names: dict[str, set[str]] = {}
|
||||
for sys_id, system in (native_layer(config) or {}).get("systems", {}).items():
|
||||
for entry in system.get("files", []):
|
||||
name = entry.get("native_system")
|
||||
if isinstance(name, str) and name.strip():
|
||||
names.setdefault(name.strip().lower(), set()).add(sys_id)
|
||||
native[platform] = names
|
||||
return cls(ids=ids, profile_use=use, native=native)
|
||||
|
||||
def canonical(self, sid: str) -> str:
|
||||
return self.canonical_of.get(sid, sid)
|
||||
|
||||
def equivalents(self, sid: str) -> set[str]:
|
||||
"""Every id spelling the same machine as sid, sid included."""
|
||||
return set(self.by_norm.get(_norm_system_id(sid), ())) | {sid}
|
||||
|
||||
def _answer(self, found: set[str], basis: str) -> Resolution:
|
||||
return Resolution(sorted({self.canonical(s) for s in found}), basis)
|
||||
|
||||
def resolve(self, names: list[str], platform: str = "") -> Resolution:
|
||||
"""The systems the first name that resolves designates."""
|
||||
for name in names:
|
||||
if not name:
|
||||
continue
|
||||
lowered = name.strip().lower()
|
||||
if lowered in self.ids:
|
||||
return self._answer({lowered}, "id")
|
||||
own = self.native.get(platform, {}).get(lowered)
|
||||
if own and len(own) == 1:
|
||||
return self._answer(own, f"native name of {platform}")
|
||||
norm = self.by_norm.get(_norm_system_id(lowered))
|
||||
if norm:
|
||||
return self._answer(norm, "normalized id")
|
||||
claimed = [
|
||||
(plat, owners) for plat, table in sorted(self.native.items())
|
||||
if (owners := table.get(lowered)) and len(owners) == 1
|
||||
]
|
||||
if claimed:
|
||||
plat, owners = claimed[0]
|
||||
return self._answer(owners, f"native name of {plat}")
|
||||
return Resolution([], "unresolved")
|
||||
|
||||
|
||||
@dataclass
|
||||
class ProfileResolver:
|
||||
"""An OS's core name to the repository profile that documents it."""
|
||||
|
||||
index: dict[str, set[str]]
|
||||
|
||||
@classmethod
|
||||
def from_repo(cls, emulators_dir: str = "emulators") -> ProfileResolver:
|
||||
return cls(upstream_profile_index(load_emulator_profiles(emulators_dir)))
|
||||
|
||||
def resolve(self, core: str | None) -> str | None:
|
||||
if not core:
|
||||
return None
|
||||
# RetroPie names a libretro package lr-<core> (scriptmodules/libretrocores).
|
||||
bare = core.removeprefix("lr-")
|
||||
spellings = (core, core.lower(), bare, bare.replace("-", "_"), bare.lower().replace("-", "_"))
|
||||
for name in dict.fromkeys(spellings):
|
||||
found = preferred_profile(self.index, name)
|
||||
if found:
|
||||
return found
|
||||
return None
|
||||
|
||||
|
||||
_TAG = re.compile(r"\(([^()]+)\)\s*$")
|
||||
|
||||
|
||||
def folder_names(row: dict) -> list[str]:
|
||||
"""The names a folder row is known by, the OS's system name first.
|
||||
|
||||
A folder named for people and tagged for the frontend, `Sony PlayStation
|
||||
(PS)` on CrossMix-OS, is also known by its tag.
|
||||
"""
|
||||
names = [row.get("name") or "", row["folder"], *(row.get("aliases") or [])]
|
||||
names += [m.group(1).strip() for n in list(names) if (m := _TAG.search(n))]
|
||||
return [n for n in dict.fromkeys(names) if n]
|
||||
@@ -0,0 +1,193 @@
|
||||
"""Scrape each setup's rom folders from its own source, at a pinned revision.
|
||||
|
||||
`platforms/romfolders/_sources/<setup>.yml` names the repositories and
|
||||
the commits to read, the format of the system list, and the facts a parser
|
||||
cannot read from a table (roms root, case rule, folders the OS never scans),
|
||||
each with its `source_ref`. A format parser reads the system list and returns
|
||||
one row per folder. The setup file it writes carries the OS's own names only:
|
||||
`scripts/romfolders.py` resolves them to repository systems and profiles.
|
||||
|
||||
Every fetch goes through `upstream.fetch_file`, cached by revision, so a
|
||||
second run reads nothing from the network and writes the same bytes.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import importlib
|
||||
import pkgutil
|
||||
import sys
|
||||
from collections.abc import Callable
|
||||
from dataclasses import dataclass
|
||||
from pathlib import Path
|
||||
|
||||
_SCRIPTS_DIR = str(Path(__file__).resolve().parents[2])
|
||||
if _SCRIPTS_DIR not in sys.path:
|
||||
sys.path.insert(0, _SCRIPTS_DIR)
|
||||
|
||||
import upstream # noqa: E402
|
||||
from safeparse import yaml_load # noqa: E402
|
||||
|
||||
CACHE_DIR = upstream.CACHE_DIR
|
||||
|
||||
# Keys a setup file carries, in the order they are written.
|
||||
SETUP_KEYS = (
|
||||
"setup", "name", "platform", "sources", "roms_root", "case_sensitive",
|
||||
"case_ref", "skip", "hidden", "media", "notes", "folders",
|
||||
)
|
||||
ROW_KEYS = (
|
||||
"folder", "name", "aliases", "extensions", "core", "emulator", "source_ref",
|
||||
"board_defaults", "arcade", "note",
|
||||
)
|
||||
|
||||
|
||||
class SourceError(RuntimeError):
|
||||
"""A pinned file is absent, or its content no longer has the shape read."""
|
||||
|
||||
|
||||
@dataclass
|
||||
class Fetcher:
|
||||
"""Reads pinned files of one setup's declared sources."""
|
||||
|
||||
sources: dict[str, dict]
|
||||
cache_dir: str = CACHE_DIR
|
||||
offline: bool = False
|
||||
|
||||
def _repo(self, key: str) -> tuple[upstream.Repo, str]:
|
||||
if key not in self.sources:
|
||||
raise SourceError(f"source {key!r} is not declared")
|
||||
spec = self.sources[key]
|
||||
repo = upstream.parse_repo(spec["repo"])
|
||||
if repo is None:
|
||||
raise SourceError(f"source {key!r}: unsupported forge {spec['repo']}")
|
||||
return repo, spec["commit"]
|
||||
|
||||
def text(self, key: str, path: str) -> str:
|
||||
"""One file at the pinned commit, or SourceError when absent."""
|
||||
repo, sha = self._repo(key)
|
||||
lines = upstream.fetch_file(repo, sha, path, self.cache_dir, self.offline)
|
||||
if lines is None:
|
||||
raise SourceError(f"{repo.slug}@{sha[:8]}:{path} does not exist")
|
||||
return "\n".join(lines) + "\n"
|
||||
|
||||
def optional_text(self, key: str, path: str) -> str | None:
|
||||
repo, sha = self._repo(key)
|
||||
lines = upstream.fetch_file(repo, sha, path, self.cache_dir, self.offline)
|
||||
return None if lines is None else "\n".join(lines) + "\n"
|
||||
|
||||
def tree(self, key: str, prefix: str = "") -> list[str]:
|
||||
"""Every blob path under a prefix at the pinned commit."""
|
||||
repo, sha = self._repo(key)
|
||||
if repo.family == "gitlab" and prefix.strip("/"):
|
||||
return self._gitlab_subtree(repo, sha, prefix)
|
||||
paths, truncated = upstream.list_tree(repo, sha, self.cache_dir, self.offline)
|
||||
if truncated:
|
||||
raise SourceError(f"{repo.slug}@{sha[:8]}: the forge truncated the tree")
|
||||
return sorted(p for p in paths if p.startswith(prefix))
|
||||
|
||||
def entries(self, key: str, directory: str) -> list[str]:
|
||||
"""Paths of the files and folders directly under one directory."""
|
||||
repo, sha = self._repo(key)
|
||||
if repo.family != "gitlab":
|
||||
prefix = directory.strip("/") + "/"
|
||||
return sorted({prefix + p[len(prefix):].split("/", 1)[0] for p in self.tree(key, prefix)})
|
||||
return self._gitlab_listing(repo, sha, directory, recursive=False, blobs=False)
|
||||
|
||||
def _gitlab_subtree(self, repo: upstream.Repo, sha: str, prefix: str) -> list[str]:
|
||||
"""GitLab pages its tree 100 entries at a time and Recalbox's whole
|
||||
tree outruns the page cap: list only the directory asked for."""
|
||||
paths = self._gitlab_listing(repo, sha, prefix, recursive=True, blobs=True)
|
||||
return sorted(p for p in paths if p.startswith(prefix))
|
||||
|
||||
def _gitlab_listing(
|
||||
self, repo: upstream.Repo, sha: str, directory: str, *, recursive: bool, blobs: bool
|
||||
) -> list[str]:
|
||||
directory = directory.strip("/")
|
||||
paths: list[str] = []
|
||||
for page in range(1, upstream.MAX_TREE_PAGES + 1):
|
||||
url = (
|
||||
f"{repo.api_base}/projects/{upstream._project(repo)}/repository/tree"
|
||||
f"?ref={sha}&path={directory}&recursive={'true' if recursive else 'false'}"
|
||||
f"&per_page={upstream.TREE_PAGE_SIZE}&page={page}"
|
||||
)
|
||||
payload = upstream._api(url, self.cache_dir, self.offline)
|
||||
if not isinstance(payload, list):
|
||||
raise SourceError(f"{repo.slug}@{sha[:8]}: {directory} could not be listed")
|
||||
paths.extend(
|
||||
entry["path"] for entry in payload
|
||||
if isinstance(entry, dict) and entry.get("path")
|
||||
and (entry.get("type") == "blob" or not blobs)
|
||||
)
|
||||
if len(payload) < upstream.TREE_PAGE_SIZE:
|
||||
return sorted(paths)
|
||||
raise SourceError(f"{repo.slug}@{sha[:8]}: {directory} outruns the page cap")
|
||||
|
||||
|
||||
Parser = Callable[[dict, Fetcher], list[dict]]
|
||||
FORMATS: dict[str, Parser] = {}
|
||||
|
||||
|
||||
def register(name: str) -> Callable[[Parser], Parser]:
|
||||
def wrap(func: Parser) -> Parser:
|
||||
if name in FORMATS:
|
||||
raise ValueError(f"format {name!r} registered twice")
|
||||
FORMATS[name] = func
|
||||
return func
|
||||
|
||||
return wrap
|
||||
|
||||
|
||||
def load_formats() -> dict[str, Parser]:
|
||||
"""Import every parser module of this package once."""
|
||||
if not FORMATS:
|
||||
for info in pkgutil.iter_modules([str(Path(__file__).parent)]):
|
||||
if not info.name.startswith("_"):
|
||||
importlib.import_module(f"{__name__}.{info.name}")
|
||||
return FORMATS
|
||||
|
||||
|
||||
def load_sources(directory: Path) -> dict[str, dict]:
|
||||
"""Each setup's declared sources and facts, keyed by its file name."""
|
||||
out: dict[str, dict] = {}
|
||||
for path in sorted(Path(directory).glob("*.yml")):
|
||||
with path.open(encoding="utf-8") as handle:
|
||||
out[path.stem] = yaml_load(handle) or {}
|
||||
return out
|
||||
|
||||
|
||||
def _clean_row(row: dict) -> dict:
|
||||
unknown = set(row) - set(ROW_KEYS)
|
||||
if unknown:
|
||||
raise ValueError(f"folder {row.get('folder')!r}: unknown keys {sorted(unknown)}")
|
||||
out = {key: row[key] for key in ROW_KEYS if row.get(key) not in (None, [], {}, "")}
|
||||
if out.get("name") == out["folder"]:
|
||||
del out["name"]
|
||||
if "extensions" in out:
|
||||
out["extensions"] = sorted({e.lower() for e in out["extensions"]})
|
||||
if "aliases" in out:
|
||||
out["aliases"] = sorted(set(out["aliases"]) - {out["folder"]})
|
||||
if not out["aliases"]:
|
||||
del out["aliases"]
|
||||
return out
|
||||
|
||||
|
||||
def scrape_setup(setup: str, spec: dict, fetcher: Fetcher) -> dict:
|
||||
"""The setup document: declared facts, then the parsed folders."""
|
||||
formats = load_formats()
|
||||
fmt = spec.get("format")
|
||||
rows: list[dict] = []
|
||||
if fmt:
|
||||
if fmt not in formats:
|
||||
raise SourceError(f"{setup}: no parser for format {fmt!r}")
|
||||
rows = formats[fmt](spec, fetcher)
|
||||
if not rows and not spec.get("allow_empty"):
|
||||
raise SourceError(f"{setup}: format {fmt} read no folder")
|
||||
folders: dict[str, dict] = {}
|
||||
for row in rows:
|
||||
clean = _clean_row(row)
|
||||
key = clean["folder"]
|
||||
if key in folders:
|
||||
raise SourceError(f"{setup}: folder {key!r} read twice")
|
||||
folders[key] = clean
|
||||
doc = {"setup": setup, **{k: spec[k] for k in SETUP_KEYS[1:-1] if k in spec}}
|
||||
doc["folders"] = [folders[k] for k in sorted(folders, key=lambda f: (f.lower(), f))]
|
||||
return doc
|
||||
@@ -0,0 +1,71 @@
|
||||
"""Write platforms/romfolders/<setup>.yml from each setup's pinned source.
|
||||
|
||||
python -m scripts.scraper.romfolders --all
|
||||
python -m scripts.scraper.romfolders --setup batocera --dry-run
|
||||
python -m scripts.scraper.romfolders --all --offline # cache only
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
import yaml
|
||||
|
||||
from . import Fetcher, SourceError, load_sources, scrape_setup
|
||||
from ..base_scraper import _PlatformDumper
|
||||
|
||||
DEFAULT_DIR = Path("platforms") / "romfolders"
|
||||
|
||||
|
||||
def dump_setup(doc: dict) -> str:
|
||||
return yaml.dump(
|
||||
doc, Dumper=_PlatformDumper, sort_keys=False, allow_unicode=True, width=100
|
||||
)
|
||||
|
||||
|
||||
def main() -> int:
|
||||
parser = argparse.ArgumentParser(allow_abbrev=False, description=__doc__.splitlines()[0])
|
||||
which = parser.add_mutually_exclusive_group(required=True)
|
||||
which.add_argument("--setup", action="append", help="setup id (repeatable)")
|
||||
which.add_argument("--all", action="store_true", help="every setup of the sources file")
|
||||
which.add_argument("--list", action="store_true", help="list the setups and their format")
|
||||
parser.add_argument("--sources", type=Path, default=DEFAULT_DIR / "_sources")
|
||||
parser.add_argument("--output-dir", type=Path, default=DEFAULT_DIR)
|
||||
parser.add_argument("--offline", action="store_true", help="read the revision cache only")
|
||||
parser.add_argument("--dry-run", action="store_true", help="print counts, write nothing")
|
||||
args = parser.parse_args()
|
||||
|
||||
sources = load_sources(args.sources)
|
||||
if args.list:
|
||||
for setup, spec in sources.items():
|
||||
print(f"{setup:16} {spec.get('format') or '-':14} {spec.get('name', '')}")
|
||||
return 0
|
||||
wanted = list(sources) if args.all else args.setup
|
||||
unknown = [s for s in wanted if s not in sources]
|
||||
if unknown:
|
||||
print(f"Error: unknown setup(s): {', '.join(unknown)}", file=sys.stderr)
|
||||
return 2
|
||||
failed = 0
|
||||
for setup in wanted:
|
||||
spec = sources[setup]
|
||||
fetcher = Fetcher(spec.get("sources", {}), offline=args.offline)
|
||||
try:
|
||||
doc = scrape_setup(setup, spec, fetcher)
|
||||
except (SourceError, ValueError, KeyError, OSError) as exc:
|
||||
print(f"FAIL {setup}: {exc}", file=sys.stderr)
|
||||
failed += 1
|
||||
continue
|
||||
print(f"{setup}: {len(doc['folders'])} folders")
|
||||
if not args.dry_run:
|
||||
args.output_dir.mkdir(parents=True, exist_ok=True)
|
||||
target = args.output_dir / f"{setup}.yml"
|
||||
text = dump_setup(doc)
|
||||
if not target.exists() or target.read_text(encoding="utf-8") != text:
|
||||
target.write_text(text, encoding="utf-8")
|
||||
return 1 if failed else 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main())
|
||||
@@ -0,0 +1,628 @@
|
||||
"""Rom folder parsers for handheld card OSes and the frontends without a menu.
|
||||
|
||||
Each parser reads the system list the OS ships at the pinned commit and
|
||||
returns one row per folder, in the OS's own spelling:
|
||||
|
||||
- `onion`: Onion's optional packages, `Emu/<SYS>/config.json` and the
|
||||
`launch.sh` beside it. RetroArch-section packages (`RApp/`) add cores on the
|
||||
same folder, the stock menu starts the `Emu/` one.
|
||||
- `spruce`: spruceOS `Emu/<SYS>/config.json`, `default_emulator`.
|
||||
- `crossmix`: CrossMix-OS `Emus/<SYS>/config.json`, first `launchlist` entry
|
||||
with a script, as `load_launcher.sh` picks it.
|
||||
- `koriki`: SimpleMenu section groups, first launcher of `execs`.
|
||||
- `allium`: Allium `consoles.toml`, first entry of `cores`.
|
||||
- `minui`: MinUI, MyMinUI and NextUI paks. A Roms folder starts the pak named
|
||||
by the text in its last parentheses (`getEmuName`), so the folder of a row
|
||||
is the one the skeleton ships, else the tag alone.
|
||||
- `plumos`: plumOS `systems.json`, `directory_aliases` by priority.
|
||||
- `muos`: muOS `core.json`, every `friendly` name of a system.
|
||||
- `gmenu2x`: MiyooCFW, the folders its image ships and the gmenu2x links
|
||||
whose text a repository holds.
|
||||
- `mister`: MiSTer, one `games/<core>/` folder per core the distribution ships.
|
||||
- `romm`: RomM, one library folder per platform slug it plays or keeps BIOS for.
|
||||
- `bizhawk`: BizHawk's per-system base folders.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import re
|
||||
from collections.abc import Iterable
|
||||
from configparser import ConfigParser
|
||||
from pathlib import PurePosixPath
|
||||
|
||||
from . import Fetcher, SourceError, register
|
||||
|
||||
_LIBRETRO = re.compile(r"([A-Za-z0-9_+.\-${}]+?)_libretro\.so")
|
||||
_EMU_EXE = re.compile(r"^\s*EMU_EXE=[\"']?([A-Za-z0-9_+.\-]+)", re.M)
|
||||
|
||||
|
||||
def line_of(text: str, needle: str, start: int = 0) -> int:
|
||||
"""1-based line of the first needle at or after start, 0 when absent."""
|
||||
idx = text.find(needle, start)
|
||||
return text.count("\n", 0, idx) + 1 if idx >= 0 else 0
|
||||
|
||||
|
||||
def ref(key: str, path: str, line: int | str | None = None) -> str:
|
||||
"""A source_ref part. A space in a path is written %20, as in a URL:
|
||||
the ref grammar ends a path at the first space."""
|
||||
path = path.replace("%", "%25").replace(" ", "%20")
|
||||
return f"{key} {path}:{line}" if line else f"{key} {path}"
|
||||
|
||||
|
||||
def extensions(raw: str | Iterable[str] | None) -> list[str]:
|
||||
"""Lowercased extensions with the dot, from a list or a `|`/`,` string."""
|
||||
items = re.split(r"[|,]", raw) if isinstance(raw, str) else list(raw or [])
|
||||
out: list[str] = []
|
||||
for item in items:
|
||||
ext = str(item).strip().lower().lstrip(".")
|
||||
if ext and f".{ext}" not in out:
|
||||
out.append(f".{ext}")
|
||||
return out
|
||||
|
||||
|
||||
def _shell_value(script: str, name: str) -> str | None:
|
||||
match = re.search(rf"^\s*{re.escape(name)}=[\"']?([A-Za-z0-9_+.\-]+)", script, re.M)
|
||||
return match.group(1) if match else None
|
||||
|
||||
|
||||
def libretro_core(script: str) -> str | None:
|
||||
"""The core a launch script names: `<core>_libretro.so`, a shell variable
|
||||
assigned in the same script, or `EMU_EXE=`."""
|
||||
for match in _LIBRETRO.finditer(script):
|
||||
name = match.group(1).split("/")[-1]
|
||||
if "$" not in name:
|
||||
return name
|
||||
var = re.fullmatch(r"\$\{?(\w+)\}?", name)
|
||||
if var:
|
||||
value = _shell_value(script, var.group(1))
|
||||
if value:
|
||||
return value
|
||||
exe = _EMU_EXE.search(script)
|
||||
return exe.group(1) if exe else None
|
||||
|
||||
|
||||
def _json(fetcher: Fetcher, key: str, path: str) -> tuple[dict, str]:
|
||||
text = fetcher.text(key, path)
|
||||
try:
|
||||
return json.loads(text), text
|
||||
except ValueError as exc:
|
||||
raise SourceError(f"{path}: {exc}") from exc
|
||||
|
||||
|
||||
def _rompath_folder(rompath: str) -> str:
|
||||
"""The folder under Roms/ a stock-menu config.json points at."""
|
||||
path = rompath.strip().replace("\\", "/")
|
||||
for prefix in ("../../Roms/", "/mnt/SDCARD/Roms/"):
|
||||
if path.startswith(prefix):
|
||||
return path[len(prefix):].strip("/")
|
||||
raise SourceError(f"rompath {rompath!r} is not under Roms/")
|
||||
|
||||
|
||||
# Onion
|
||||
|
||||
@register("onion")
|
||||
def parse_onion(spec: dict, fetcher: Fetcher) -> list[dict]:
|
||||
params = spec["params"]
|
||||
key = params["source"]
|
||||
pattern = re.compile(r"^static/packages/(Emu|RApp)/([^/]+)/\1/([^/]+)/config\.json$")
|
||||
packages: dict[str, list[dict]] = {}
|
||||
for path in fetcher.tree(key, "static/packages/"):
|
||||
match = pattern.match(path)
|
||||
if not match:
|
||||
continue
|
||||
layer, package = match.group(1), match.group(2)
|
||||
config, text = _json(fetcher, key, path)
|
||||
script_path = f"{path.rsplit('/', 1)[0]}/{config.get('launch', 'launch.sh')}"
|
||||
script = fetcher.optional_text(key, script_path) or ""
|
||||
packages.setdefault(_rompath_folder(config.get("rompath", "")), []).append({
|
||||
"layer": layer, "package": package, "config": config,
|
||||
"core": libretro_core(script), "ref": ref(key, path, line_of(text, '"rompath"')),
|
||||
"script": script_path,
|
||||
})
|
||||
rows = []
|
||||
for folder, found in packages.items():
|
||||
found.sort(key=lambda p: (p["layer"] != "Emu", p["package"]))
|
||||
emu = [p for p in found if p["layer"] == "Emu"]
|
||||
first = emu[0] if emu else found[0]
|
||||
others = [p["package"] for p in found if p is not first]
|
||||
row = {
|
||||
"folder": folder,
|
||||
"extensions": extensions(first["config"].get("extlist")),
|
||||
"source_ref": first["ref"],
|
||||
}
|
||||
if emu:
|
||||
row["core"] = first["core"]
|
||||
row["emulator"] = "retroarch" if first["core"] else "standalone"
|
||||
note = f"package '{first['package']}' (Emu/, the stock menu entry)"
|
||||
if not first["core"]:
|
||||
note += f", launches {first['script']}"
|
||||
else:
|
||||
note = "only RetroArch-section packages (RApp/), the user picks the core"
|
||||
if others:
|
||||
note += f". Other packages on this folder: {', '.join(others)}"
|
||||
row["note"] = note
|
||||
rows.append(row)
|
||||
return rows
|
||||
|
||||
|
||||
# spruceOS
|
||||
|
||||
@register("spruce")
|
||||
def parse_spruce(spec: dict, fetcher: Fetcher) -> list[dict]:
|
||||
params = spec["params"]
|
||||
key = params["source"]
|
||||
skip = set(params.get("skip", []))
|
||||
rows = []
|
||||
for path in fetcher.tree(key, "Emu/"):
|
||||
parts = path.split("/")
|
||||
if len(parts) != 3 or parts[2] != "config.json" or parts[1] in skip:
|
||||
continue
|
||||
config, text = _json(fetcher, key, path)
|
||||
notes = []
|
||||
if config.get("devices"):
|
||||
notes.append(f"only on devices {', '.join(map(str, config['devices']))}")
|
||||
if config.get("ignoreList"):
|
||||
notes.append(f"ignoreList {', '.join(map(str, config['ignoreList']))}")
|
||||
rows.append({
|
||||
"folder": parts[1],
|
||||
"aliases": [str(a) for a in config.get("alternativeFolderNames") or []],
|
||||
"extensions": extensions(config.get("extlist")),
|
||||
"core": config.get("default_emulator"),
|
||||
"source_ref": ref(key, path, line_of(text, '"default_emulator"') or 1),
|
||||
"note": ". ".join(notes),
|
||||
})
|
||||
return rows
|
||||
|
||||
|
||||
# CrossMix-OS
|
||||
|
||||
@register("crossmix")
|
||||
def parse_crossmix(spec: dict, fetcher: Fetcher) -> list[dict]:
|
||||
params = spec["params"]
|
||||
key = params["source"]
|
||||
rows = []
|
||||
for path in fetcher.tree(key, "Emus/"):
|
||||
parts = path.split("/")
|
||||
if len(parts) != 3 or parts[2] != "config.json" or parts[1].startswith("_"):
|
||||
continue
|
||||
config, text = _json(fetcher, key, path)
|
||||
launchers = [e for e in config.get("launchlist") or [] if e.get("launch")]
|
||||
script_name = launchers[0]["launch"] if launchers else config.get("launch", "launch.sh")
|
||||
script_path = f"Emus/{parts[1]}/{script_name}"
|
||||
script = fetcher.optional_text(key, script_path) or ""
|
||||
core = libretro_core(script)
|
||||
note = f"default launcher {launchers[0].get('name') if launchers else script_name} ({script_path})"
|
||||
folder = _rompath_folder(config.get("rompath", ""))
|
||||
rows.append({
|
||||
"folder": folder,
|
||||
"name": parts[1] if parts[1] != folder else None,
|
||||
"extensions": extensions(config.get("extlist")),
|
||||
"core": core,
|
||||
"emulator": "retroarch" if core else "standalone",
|
||||
"source_ref": ref(key, path, line_of(text, '"rompath"') or 1),
|
||||
"note": note if core else f"{note}, names no libretro core",
|
||||
})
|
||||
return rows
|
||||
|
||||
|
||||
# Koriki
|
||||
|
||||
@register("koriki")
|
||||
def parse_koriki(spec: dict, fetcher: Fetcher) -> list[dict]:
|
||||
params = spec["params"]
|
||||
key = params["source"]
|
||||
path = params["sections"]
|
||||
text = fetcher.text(key, path)
|
||||
parser = ConfigParser(interpolation=None, strict=False)
|
||||
parser.optionxform = str # type: ignore[assignment]
|
||||
parser.read_string(text)
|
||||
skip = set(params.get("skip", []))
|
||||
rows = []
|
||||
for section in parser.sections():
|
||||
if section in skip:
|
||||
continue
|
||||
body = parser[section]
|
||||
dirs = [d.strip() for d in body.get("romDirs", "").split(",") if d.strip()]
|
||||
if not dirs:
|
||||
continue
|
||||
rel = [d.replace(params["roms_prefix"], "").strip("/") for d in dirs]
|
||||
execs = [PurePosixPath(e.strip()).name for e in body.get("execs", "").split(",") if e.strip()]
|
||||
core = None
|
||||
if execs:
|
||||
launcher = fetcher.optional_text(key, f"{params['launchers']}/{execs[0]}")
|
||||
core = libretro_core(launcher or "")
|
||||
note = f"launchers {', '.join(execs)}"
|
||||
if len(rel) > 1 or "/" in rel[0]:
|
||||
note += f". romDirs {', '.join(rel)}"
|
||||
rows.append({
|
||||
"folder": rel[0].split("/")[0],
|
||||
"name": section,
|
||||
"extensions": extensions(body.get("romExts")),
|
||||
"core": core or (execs[0] if execs else None),
|
||||
"emulator": "retroarch" if core else "standalone",
|
||||
"source_ref": ref(key, path, line_of(text, f"[{section}]")),
|
||||
"note": note,
|
||||
})
|
||||
return rows
|
||||
|
||||
|
||||
# Allium
|
||||
|
||||
_TOML_KEY = r'(?:"[^"]*"|[A-Za-z0-9_\-]+)'
|
||||
|
||||
|
||||
def _toml_value(raw: str) -> object:
|
||||
raw = raw.strip()
|
||||
if raw.startswith('"') and raw.endswith('"'):
|
||||
return raw[1:-1]
|
||||
if raw.startswith("["):
|
||||
return re.findall(r'"([^"]*)"', raw)
|
||||
if raw in ("true", "false"):
|
||||
return raw == "true"
|
||||
if re.fullmatch(r"-?\d+", raw):
|
||||
return int(raw)
|
||||
raise SourceError(f"TOML value {raw!r} is outside the subset Allium's files use")
|
||||
|
||||
|
||||
def parse_toml_subset(text: str) -> dict:
|
||||
"""The TOML Allium writes: `[[array]]`, `[a.b]` tables, string, integer,
|
||||
boolean and string-array values, arrays that may span lines. Anything
|
||||
else is refused rather than guessed. Each table records the line it
|
||||
starts on under `__line__`."""
|
||||
root: dict = {}
|
||||
current: dict = root
|
||||
lines = text.splitlines()
|
||||
i = 0
|
||||
while i < len(lines):
|
||||
line = lines[i].split("#", 1)[0].strip() if '"' not in lines[i] else lines[i].strip()
|
||||
i += 1
|
||||
if not line or line.startswith("#"):
|
||||
continue
|
||||
if line.startswith("[["):
|
||||
name = line.strip("[]").strip()
|
||||
current = {"__line__": i}
|
||||
root.setdefault(name, []).append(current)
|
||||
continue
|
||||
if line.startswith("["):
|
||||
current = root
|
||||
for part in re.findall(_TOML_KEY, line.strip("[]")):
|
||||
current = current.setdefault(part.strip('"'), {})
|
||||
current["__line__"] = i
|
||||
continue
|
||||
match = re.match(rf"({_TOML_KEY})\s*=\s*(.*)$", line)
|
||||
if not match:
|
||||
raise SourceError(f"TOML line {i}: {line!r} is outside the subset Allium's files use")
|
||||
value = match.group(2)
|
||||
while value.count("[") > value.count("]") and i < len(lines):
|
||||
value += lines[i].strip()
|
||||
i += 1
|
||||
current[match.group(1).strip('"')] = _toml_value(value)
|
||||
return root
|
||||
|
||||
|
||||
@register("allium")
|
||||
def parse_allium(spec: dict, fetcher: Fetcher) -> list[dict]:
|
||||
params = spec["params"]
|
||||
key = params["source"]
|
||||
consoles = parse_toml_subset(fetcher.text(key, params["consoles"]))["consoles"]
|
||||
cores = parse_toml_subset(fetcher.text(key, params["cores"])).get("cores", {})
|
||||
rows = []
|
||||
for console in consoles:
|
||||
patterns = [str(p) for p in console.get("patterns") or []]
|
||||
if not patterns:
|
||||
continue
|
||||
first = (console.get("cores") or [None])[0]
|
||||
core_def = cores.get(first, {}) if first else {}
|
||||
core = core_def.get("retroarch") if isinstance(core_def, dict) else None
|
||||
rows.append({
|
||||
"folder": patterns[0],
|
||||
"name": console["name"],
|
||||
"aliases": patterns[1:],
|
||||
"extensions": extensions(console.get("extensions")),
|
||||
"core": core or first,
|
||||
"emulator": "retroarch" if core else "standalone",
|
||||
"source_ref": ref(key, params["consoles"], console["__line__"]),
|
||||
"note": "a folder named any pattern, or ending in '(<pattern>)', is this console. A file's own extension wins over its folder",
|
||||
})
|
||||
return rows
|
||||
|
||||
|
||||
# MinUI, MyMinUI, NextUI
|
||||
|
||||
_PAK = re.compile(r"^skeleton/(?:[^/]+/)+(?:Emus|paks/Emus)/(?:[^/]+/)?([^/]+)\.pak/launch\.sh$")
|
||||
|
||||
|
||||
def _pak_device(path: str) -> str:
|
||||
"""The device directory a pak sits under (`Emus/<device>/X.pak`,
|
||||
`SYSTEM/<device>/paks/Emus/X.pak`)."""
|
||||
parts = path.split("/")
|
||||
if "paks" in parts:
|
||||
return parts[parts.index("paks") - 1]
|
||||
idx = parts.index("Emus")
|
||||
return parts[idx + 1] if not parts[idx + 1].endswith(".pak") else parts[1]
|
||||
|
||||
|
||||
@register("minui")
|
||||
def parse_minui(spec: dict, fetcher: Fetcher) -> list[dict]:
|
||||
params = spec["params"]
|
||||
key = params["source"]
|
||||
tree = fetcher.tree(key, "skeleton/")
|
||||
shipped: dict[str, str] = {}
|
||||
for path in tree:
|
||||
match = re.match(r"^skeleton/[^/]+/Roms/([^/]+)/", path)
|
||||
if match:
|
||||
name = match.group(1)
|
||||
tag = re.search(r"\(([^()]*)\)\s*$", name)
|
||||
shipped.setdefault(tag.group(1) if tag else name, name)
|
||||
paks: dict[str, dict[str, tuple[str, str]]] = {}
|
||||
for path in tree:
|
||||
match = _PAK.match(path)
|
||||
if not match:
|
||||
continue
|
||||
script = fetcher.text(key, path)
|
||||
core = libretro_core(script)
|
||||
line = line_of(script, "EMU_EXE=") or line_of(script, "_libretro.so") or 1
|
||||
paks.setdefault(match.group(1), {})[_pak_device(path)] = (core or "", ref(key, path, line))
|
||||
rows = []
|
||||
for tag, devices in sorted(paks.items()):
|
||||
counts: dict[str, int] = {}
|
||||
for core, _ in devices.values():
|
||||
counts[core] = counts.get(core, 0) + 1
|
||||
core = min(counts, key=lambda c: (-counts[c], c == "", c))
|
||||
default_ref = next(r for c, r in sorted(devices.values()) if c == core)
|
||||
boards = {
|
||||
device: {"core": c, "source_ref": r}
|
||||
for device, (c, r) in sorted(devices.items()) if c != core and c
|
||||
}
|
||||
folder = shipped.get(tag, tag)
|
||||
rows.append({
|
||||
"folder": folder,
|
||||
"name": tag,
|
||||
"aliases": [tag] if folder != tag else [],
|
||||
"core": core or None,
|
||||
"emulator": "minarch" if core else "standalone",
|
||||
"source_ref": default_ref,
|
||||
"board_defaults": boards,
|
||||
"note": (f"any folder named {tag} or ending in '({tag})' starts {tag}.pak. "
|
||||
f"Devices: {', '.join(sorted(devices))}"),
|
||||
})
|
||||
return rows
|
||||
|
||||
|
||||
# plumOS
|
||||
|
||||
@register("plumos")
|
||||
def parse_plumos(spec: dict, fetcher: Fetcher) -> list[dict]:
|
||||
params = spec["params"]
|
||||
key = params["source"]
|
||||
text = fetcher.text(key, params["path"])
|
||||
systems = json.loads(text)["systems"]
|
||||
taken: dict[str, str] = {}
|
||||
rows = []
|
||||
pos = 0
|
||||
for system in systems:
|
||||
needle = f'"id": "{system["id"]}"'
|
||||
line = line_of(text, needle, pos)
|
||||
pos = max(pos, text.find(needle, pos) + 1)
|
||||
aliases = [a["name"] for a in sorted(system.get("directory_aliases") or [],
|
||||
key=lambda a: a.get("priority", 0))]
|
||||
free = [a for a in aliases if a not in taken]
|
||||
if not free:
|
||||
continue
|
||||
folder = free[0]
|
||||
taken[folder] = system["id"]
|
||||
kind, _, core = (system.get("default_launch_profile") or "").partition(":")
|
||||
notes = [] if system.get("enabled") else ["disabled by default"]
|
||||
shared = [f"{a} (also {taken[a]})" for a in aliases if a in taken and taken[a] != system["id"]]
|
||||
if shared:
|
||||
notes.append(f"shares {', '.join(shared)}, the extension decides")
|
||||
rows.append({
|
||||
"folder": folder,
|
||||
"name": system["id"],
|
||||
"aliases": [a for a in aliases if a != folder],
|
||||
"extensions": extensions(system.get("extensions")),
|
||||
"core": core or None,
|
||||
"emulator": kind or None,
|
||||
"source_ref": ref(key, params["path"], line),
|
||||
"note": ". ".join(notes),
|
||||
})
|
||||
return rows
|
||||
|
||||
|
||||
# muOS
|
||||
|
||||
def _muos_norm(name: str) -> str:
|
||||
"""The key muOS looks a folder up by (common.c: lowercase, drop ' -_+')."""
|
||||
return re.sub(r"[ \-_+]", "", name.lower())
|
||||
|
||||
|
||||
@register("muos")
|
||||
def parse_muos(spec: dict, fetcher: Fetcher) -> list[dict]:
|
||||
params = spec["params"]
|
||||
key = params["source"]
|
||||
text = fetcher.text(key, params["path"])
|
||||
data = json.loads(text)
|
||||
rows = []
|
||||
for system, entry in data.items():
|
||||
friendly = [str(f) for f in entry.get("friendly") or []]
|
||||
if not friendly:
|
||||
continue
|
||||
want = _muos_norm(entry.get("catalogue") or system)
|
||||
folder = next((f for f in friendly if f == want), friendly[0])
|
||||
default = entry.get("default")
|
||||
libretro = entry.get("libretro") or {}
|
||||
so = (libretro.get(default) or {}).get("core") if default in libretro else None
|
||||
core = so[: -len("_libretro.so")] if so and so.endswith("_libretro.so") else default
|
||||
rows.append({
|
||||
"folder": folder,
|
||||
"name": system,
|
||||
"aliases": friendly,
|
||||
"core": core,
|
||||
"emulator": "retroarch" if so else "external",
|
||||
"source_ref": ref(key, params["path"], line_of(text, json.dumps(system) + ":")),
|
||||
"note": "a folder matches when its name, lowercased without ' -_+', is one of the friendly names",
|
||||
})
|
||||
return rows
|
||||
|
||||
|
||||
# MiyooCFW
|
||||
|
||||
@register("gmenu2x")
|
||||
def parse_gmenu2x(spec: dict, fetcher: Fetcher) -> list[dict]:
|
||||
params = spec["params"]
|
||||
rows: dict[str, dict] = {}
|
||||
for link in params.get("links", []):
|
||||
text = fetcher.text(link["source"], link["path"])
|
||||
fields = dict(
|
||||
line.split("=", 1) for line in text.splitlines() if "=" in line
|
||||
)
|
||||
selector = fields.get("selectordir", "").rstrip("/")
|
||||
prefix = params["link_roms_prefix"]
|
||||
if not selector.startswith(prefix):
|
||||
raise SourceError(f"{link['path']}: selectordir {selector!r} is not under {prefix}")
|
||||
folder = selector[len(prefix):].strip("/")
|
||||
rows[folder] = {
|
||||
"folder": folder,
|
||||
"extensions": extensions(fields.get("selectorfilter")),
|
||||
"core": link["core"],
|
||||
"emulator": "standalone",
|
||||
"source_ref": ref(link["source"], link["path"], line_of(text, "selectordir=")),
|
||||
"note": f"gmenu2x link of the {link['core']} package",
|
||||
}
|
||||
key = params["source"]
|
||||
for path in fetcher.tree(key, params["shipped_roms"] + "/"):
|
||||
folder = path[len(params["shipped_roms"]) + 1:].split("/")
|
||||
if len(folder) < 2 or folder[0] in rows:
|
||||
continue
|
||||
rows[folder[0]] = {
|
||||
"folder": folder[0],
|
||||
"source_ref": ref(key, f"{params['shipped_roms']}/{folder[0]}"),
|
||||
"note": "folder the image ships. Its emulator link lives in a release package no repository holds",
|
||||
}
|
||||
return list(rows.values())
|
||||
|
||||
|
||||
# MiSTer
|
||||
|
||||
@register("mister")
|
||||
def parse_mister(spec: dict, fetcher: Fetcher) -> list[dict]:
|
||||
params = spec["params"]
|
||||
key = params["source"]
|
||||
arcade = set(params.get("arcade", []))
|
||||
folders = sorted({p.split("/")[1] for p in fetcher.tree(key, "games/") if p.count("/") >= 2})
|
||||
return [
|
||||
{
|
||||
"folder": folder,
|
||||
"emulator": "mister-core",
|
||||
"source_ref": ref(key, f"games/{folder}/"),
|
||||
"arcade": folder in arcade or None,
|
||||
"note": params["arcade_note"] if folder in arcade else None,
|
||||
}
|
||||
for folder in folders
|
||||
if folder not in params.get("skip", [])
|
||||
]
|
||||
|
||||
|
||||
# RomM
|
||||
|
||||
@register("romm")
|
||||
def parse_romm(spec: dict, fetcher: Fetcher) -> list[dict]:
|
||||
params = spec["params"]
|
||||
key = params["source"]
|
||||
text = fetcher.text(key, params["cores_map"])
|
||||
lines = text.splitlines()
|
||||
maps: dict[str, dict[str, tuple[list[str], int]]] = {}
|
||||
block = None
|
||||
entry = None
|
||||
for number, line in enumerate(lines, 1):
|
||||
start = re.match(r"const (_EJS_(?:NIGHTLY_)?CORES_MAP)\b", line)
|
||||
if start:
|
||||
block = start.group(1)
|
||||
continue
|
||||
if block and line.startswith("}"):
|
||||
block = None
|
||||
continue
|
||||
if not block:
|
||||
continue
|
||||
head = re.match(r'\s+"?([\w-]+)"?:\s*\[(.*)$', line)
|
||||
if head:
|
||||
entry = (head.group(1), number)
|
||||
maps.setdefault(block, {})[entry[0]] = (re.findall(r'"([\w-]+)"', head.group(2)), number)
|
||||
if "]" in head.group(2):
|
||||
entry = None
|
||||
elif entry:
|
||||
maps[block][entry[0]][0].extend(re.findall(r'"([\w-]+)"', line))
|
||||
if "]" in line:
|
||||
entry = None
|
||||
stable = maps.get("_EJS_CORES_MAP", {})
|
||||
nightly = maps.get("_EJS_NIGHTLY_CORES_MAP", {})
|
||||
if not stable:
|
||||
raise SourceError(f"{params['cores_map']}: _EJS_CORES_MAP not found")
|
||||
bios = json.loads(fetcher.text(key, params["bios_fixture"]))
|
||||
bios_slugs = {name.split(":", 1)[0] for name in bios}
|
||||
rows = []
|
||||
for slug in sorted(set(stable) | set(nightly) | bios_slugs):
|
||||
found = stable.get(slug) or nightly.get(slug)
|
||||
if found:
|
||||
note = None if slug in stable else "core only with EmulatorJS nightly cores"
|
||||
rows.append({"folder": slug, "core": found[0][0], "emulator": "emulatorjs",
|
||||
"source_ref": ref(key, params["cores_map"], found[1]), "note": note})
|
||||
else:
|
||||
rows.append({"folder": slug, "source_ref": ref(key, params["bios_fixture"]),
|
||||
"note": "no EmulatorJS core, RomM keeps BIOS for it"})
|
||||
return rows
|
||||
|
||||
|
||||
# BizHawk
|
||||
|
||||
@register("bizhawk")
|
||||
def parse_bizhawk(spec: dict, fetcher: Fetcher) -> list[dict]:
|
||||
params = spec["params"]
|
||||
key = params["source"]
|
||||
entries = fetcher.text(key, params["paths"])
|
||||
loader = fetcher.text(key, params["rom_loader"])
|
||||
config = fetcher.text(key, params["config"])
|
||||
combined = {
|
||||
m.group(1): re.findall(r"VSystemID\.Raw\.(\w+)", m.group(2))
|
||||
for m in re.finditer(r"string (COMBINED_SYSIDS_\w+) = string\.Join\(\"_\", ([^)]*)\)", entries)
|
||||
}
|
||||
ext_fields: dict[str, tuple[list[str], int]] = {}
|
||||
for number, line in enumerate(loader.splitlines(), 1):
|
||||
clean = re.sub(r"/\*.*?\*/", "", line)
|
||||
match = re.search(r"IReadOnlyCollection<string> (\w+) = new\[\] \{ (.*?) \}", clean)
|
||||
if match:
|
||||
ext_fields[match.group(1)] = (re.findall(r'"([^"]+)"', match.group(2)), number)
|
||||
preferred: dict[str, str] = {}
|
||||
picker = config[config.find("CorePickerUIData"):]
|
||||
for match in re.finditer(r"\(\[([^\]]+)\],\s*\[\s*CoreNames\.(\w+)", picker):
|
||||
for sid in re.findall(r"VSystemID\.Raw\.(\w+)", match.group(1)):
|
||||
preferred.setdefault(sid, match.group(2))
|
||||
rows = []
|
||||
for number, line in enumerate(entries.splitlines(), 1):
|
||||
match = re.search(
|
||||
r"CommonEntriesFor\((?:VSystemID\.Raw\.(\w+)|(COMBINED_SYSIDS_\w+)), "
|
||||
r"basePath: Path\.Combine\(\"\.\", \"([^\"]+)\"\)", line)
|
||||
if not match:
|
||||
continue
|
||||
sids = [match.group(1)] if match.group(1) else combined.get(match.group(2), [])
|
||||
exts: list[str] = []
|
||||
refs = [ref(key, params["paths"], number)]
|
||||
for sid in sids:
|
||||
if sid in ext_fields:
|
||||
exts.extend(ext_fields[sid][0])
|
||||
refs.append(ref(key, params["rom_loader"], ext_fields[sid][1]))
|
||||
cores = [preferred[s] for s in sids if s in preferred]
|
||||
note = "the open dialog starts here, nothing is scanned"
|
||||
if cores:
|
||||
note += f". Default core {cores[0]} (CorePickerUIData, CoreNames[0])"
|
||||
rows.append({
|
||||
"folder": match.group(3),
|
||||
"name": sids[0] if sids else None,
|
||||
"aliases": sids[1:],
|
||||
"extensions": extensions(exts),
|
||||
"emulator": "EmuHawk",
|
||||
"source_ref": "; ".join(refs),
|
||||
"note": note,
|
||||
})
|
||||
return rows
|
||||
@@ -0,0 +1,781 @@
|
||||
"""EmulationStation family: the system lists each OS builds its menu from.
|
||||
|
||||
Every parser returns one row per folder under the roms root, with the
|
||||
extensions the list accepts, the core the OS starts by default and the
|
||||
line of the list that says so. Systems the list keeps outside the roms root
|
||||
(an absolute path elsewhere) or marks as no game system are left out: the
|
||||
installer never scans them.
|
||||
|
||||
Formats, by the shape of the source:
|
||||
|
||||
- `batocera-es`: es_systems.yml plus configgen-defaults.yml and its per-board
|
||||
files (Batocera, KNULLI).
|
||||
- `es-xml`: a static es_systems.cfg (AmberELEC, CHAos, PAN4ELEC, ArkOS,
|
||||
dArkOS, RetroBat).
|
||||
- `es-json`: EmuELEC's es_systems.json, turned into the XML at build time.
|
||||
- `elec-conf`: config/emulators/<system>.conf plus the add_emu_core and
|
||||
add_es_system calls of the emulators package (ROCKNIX, JELOS, KOLOS,
|
||||
UnofficialOS, ArchR).
|
||||
- `esde`: ES-DE's es_systems.xml, with the alternative emulators and custom
|
||||
systems an installer writes over it (EmuDeck, RetroDECK).
|
||||
- `recalbox`: one system.ini per system (Recalbox).
|
||||
- `retropie`: platforms.cfg and the addEmulator calls of the scriptmodules.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import re
|
||||
import shlex
|
||||
from collections.abc import Iterator
|
||||
|
||||
from safeparse import parse_untrusted_xml, yaml_load_upstream
|
||||
|
||||
from . import Fetcher, SourceError, register
|
||||
|
||||
|
||||
def _ref(key: str, path: str, *lines: int | str | None) -> str:
|
||||
kept = [str(n) for n in lines if n not in (None, "")]
|
||||
return f"{key} {path}:{','.join(kept)}" if kept else f"{key} {path}"
|
||||
|
||||
|
||||
def _board(core: str | None, emulator: str | None, source_ref: str) -> dict:
|
||||
out = {"core": core or emulator, "emulator": emulator, "source_ref": source_ref}
|
||||
return {k: str(v) for k, v in out.items() if v}
|
||||
|
||||
|
||||
def _top_key_lines(text: str) -> dict[str, int]:
|
||||
"""First line of each top-level YAML key."""
|
||||
out: dict[str, int] = {}
|
||||
for number, line in enumerate(text.splitlines(), 1):
|
||||
match = re.match(r"^([A-Za-z0-9_+.\-]+):", line)
|
||||
if match:
|
||||
out.setdefault(match.group(1), number)
|
||||
return out
|
||||
|
||||
|
||||
def _strip_prefix(path: str, prefixes: list[str]) -> str | None:
|
||||
"""The folder under the roms root, or None for a path outside it."""
|
||||
for prefix in prefixes:
|
||||
if path.startswith(prefix):
|
||||
rest = path[len(prefix):].strip("/\\")
|
||||
return rest.replace("\\", "/") or None
|
||||
return None
|
||||
|
||||
|
||||
def _extensions(raw: str | list | None) -> list[str]:
|
||||
words = raw.split() if isinstance(raw, str) else [str(w) for w in raw or []]
|
||||
return [w if w.startswith(".") else f".{w}" for w in words if w]
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Batocera and KNULLI
|
||||
|
||||
|
||||
def _yaml_or_link(fetcher: Fetcher, key: str, directory: str, name: str) -> tuple[dict, str, str]:
|
||||
"""A configgen file, following a git symlink the raw route serves as text."""
|
||||
path = f"{directory}/{name}"
|
||||
text = fetcher.text(key, path)
|
||||
data = yaml_load_upstream(text)
|
||||
if isinstance(data, str) and data.strip().endswith(".yml"):
|
||||
path = f"{directory}/{data.strip()}"
|
||||
text = fetcher.text(key, path)
|
||||
data = yaml_load_upstream(text)
|
||||
if not isinstance(data, dict):
|
||||
raise SourceError(f"{path} is not a mapping")
|
||||
return data, text, path
|
||||
|
||||
|
||||
@register("batocera-es")
|
||||
def batocera_es(spec: dict, fetcher: Fetcher) -> list[dict]:
|
||||
"""es_systems.yml read the way batocera-es-system writes the XML.
|
||||
|
||||
A system with no emulator and no `force` is never written
|
||||
(es_systems.py `_system_dict_to_xml`), `path` defaults to the key and an
|
||||
absolute path leaves the roms root. The default core is configgen's:
|
||||
the board file first, then configgen-defaults.yml (shared.py
|
||||
ConfiggenDefaults.get).
|
||||
"""
|
||||
params = spec["params"]
|
||||
key = params.get("source", "os")
|
||||
es_path = params["es_systems"]
|
||||
es_text = fetcher.text(key, es_path)
|
||||
systems = yaml_load_upstream(es_text)
|
||||
if not isinstance(systems, dict):
|
||||
raise SourceError(f"{es_path} is not a mapping")
|
||||
es_lines = _top_key_lines(es_text)
|
||||
directory = params["configgen_dir"]
|
||||
defaults, def_text, def_path = _yaml_or_link(fetcher, key, directory, "configgen-defaults.yml")
|
||||
def_lines = _top_key_lines(def_text)
|
||||
boards: dict[str, tuple[dict, dict[str, int], str]] = {}
|
||||
for path in fetcher.tree(key, f"{directory}/configgen-defaults-"):
|
||||
name = path.rsplit("/", 1)[-1]
|
||||
if not name.endswith(".yml"):
|
||||
continue
|
||||
data, text, real = _yaml_or_link(fetcher, key, directory, name)
|
||||
boards[name[len("configgen-defaults-"):-len(".yml")]] = (data, _top_key_lines(text), real)
|
||||
rows: list[dict] = []
|
||||
for name, system in systems.items():
|
||||
if not isinstance(system, dict):
|
||||
continue
|
||||
if not system.get("emulators") and not system.get("force"):
|
||||
continue
|
||||
path = system.get("path", name)
|
||||
if not path or str(path).startswith("/"):
|
||||
continue
|
||||
base = defaults.get(name) or {}
|
||||
emulator, core = base.get("emulator"), base.get("core")
|
||||
board_defaults: dict[str, dict] = {}
|
||||
for board, (data, lines, real) in sorted(boards.items()):
|
||||
override = data.get(name) or {}
|
||||
b_emu, b_core = override.get("emulator", emulator), override.get("core", core)
|
||||
if (b_emu, b_core) != (emulator, core) and b_core:
|
||||
board_defaults[board] = _board(b_core, b_emu, _ref(key, real, lines.get(name)))
|
||||
refs = [_ref(key, es_path, es_lines.get(name))]
|
||||
if name in def_lines:
|
||||
refs.append(_ref(key, def_path, def_lines[name]))
|
||||
rows.append({
|
||||
"folder": str(path),
|
||||
"name": name,
|
||||
"extensions": _extensions(system.get("extensions")),
|
||||
"core": str(core) if core else None,
|
||||
"emulator": str(emulator) if emulator else None,
|
||||
"source_ref": "; ".join(refs),
|
||||
"board_defaults": board_defaults,
|
||||
"arcade": system.get("hardware") == "arcade" or None,
|
||||
})
|
||||
return rows
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Static es_systems.cfg (XML) and EmuELEC's es_systems.json
|
||||
|
||||
|
||||
def _first_lines(text: str, pattern: str) -> dict[str, int]:
|
||||
out: dict[str, int] = {}
|
||||
for number, line in enumerate(text.splitlines(), 1):
|
||||
match = re.search(pattern, line)
|
||||
if match:
|
||||
out.setdefault(match.group(1).strip(), number)
|
||||
return out
|
||||
|
||||
|
||||
def _all_lines(text: str, pattern: str) -> dict[str, list[int]]:
|
||||
"""Every line each name is declared on, in file order."""
|
||||
out: dict[str, list[int]] = {}
|
||||
for number, line in enumerate(text.splitlines(), 1):
|
||||
match = re.search(pattern, line)
|
||||
if match:
|
||||
out.setdefault(match.group(1).strip(), []).append(number)
|
||||
return out
|
||||
|
||||
|
||||
def _name_once(row: dict, lines: dict[str, list[int]], seen: dict[str, int]) -> dict:
|
||||
"""A system name the list gives twice designates neither: the row keeps
|
||||
its folder and the line of its own occurrence."""
|
||||
name = row["name"]
|
||||
index = seen.get(name, 0)
|
||||
seen[name] = index + 1
|
||||
found = lines.get(name, [])
|
||||
row["line"] = found[index] if index < len(found) else None
|
||||
if len(found) > 1:
|
||||
row["note"] = "; ".join(filter(None, [row.get("note"), f"the list names {len(found)} systems {name}"]))
|
||||
row["name"] = None
|
||||
return row
|
||||
|
||||
|
||||
def _pick_default(emulators: list[tuple[str, list[tuple[str, bool]]]], rule: str) -> tuple[str | None, str | None]:
|
||||
"""The emulator and core the frontend starts when the user chose none.
|
||||
|
||||
`flagged`: the core marked default, else the first emulator's first
|
||||
core. `first`: the first emulator's first core, whatever is flagged.
|
||||
"""
|
||||
if rule == "flagged":
|
||||
for emulator, cores in emulators:
|
||||
for core, default in cores:
|
||||
if default:
|
||||
return emulator, core
|
||||
for emulator, cores in emulators:
|
||||
if cores:
|
||||
return emulator, cores[0][0]
|
||||
return emulator, None
|
||||
return None, None
|
||||
|
||||
|
||||
_LIBRETRO_SO = re.compile(r"-L\s+\S*?[/\\]([\w.+-]+?)_libretro\.(?:so|dll|dylib)")
|
||||
|
||||
|
||||
def _libretro_in_command(command: str) -> tuple[str | None, str | None]:
|
||||
match = _LIBRETRO_SO.search(command)
|
||||
return ("retroarch", match.group(1)) if match else (None, None)
|
||||
|
||||
|
||||
def _is_arcade(hardware: str | None, platform: str | None) -> bool | None:
|
||||
words = set(re.split(r"[\s,]+", (platform or "").lower()))
|
||||
return (hardware or "").lower() == "arcade" or "arcade" in words or None
|
||||
|
||||
|
||||
@register("es-xml")
|
||||
def es_xml(spec: dict, fetcher: Fetcher) -> list[dict]:
|
||||
"""A static es_systems.cfg. `boards` names the file another build
|
||||
installs instead (dArkOS copies es_systems.cfg.<chipset>): a system whose
|
||||
default differs there gets that board's default."""
|
||||
params = spec["params"]
|
||||
key = params.get("source", "os")
|
||||
rows = _xml_rows(fetcher.text(key, params["file"]), key, params["file"], params)
|
||||
for board, path in sorted((params.get("boards") or {}).items()):
|
||||
other = {r["name"]: r for r in _xml_rows(fetcher.text(key, path), key, path, params)}
|
||||
for row in rows:
|
||||
alt = other.get(row["name"])
|
||||
if alt and alt["core"] and (alt["core"], alt["emulator"]) != (row["core"], row["emulator"]):
|
||||
row.setdefault("board_defaults", {})[board] = _board(
|
||||
alt["core"], alt["emulator"], alt["source_ref"])
|
||||
return rows
|
||||
|
||||
|
||||
_BARE_AMPERSAND = re.compile(r"&(?!(?:[A-Za-z]+|#\d+|#x[0-9A-Fa-f]+);)")
|
||||
_COMMENT = re.compile(r"<!--.*?-->", re.DOTALL)
|
||||
|
||||
|
||||
def _lenient_xml(text: str) -> str:
|
||||
"""The list as EmulationStation's pugixml reads it, lines kept in place.
|
||||
|
||||
pugixml keeps a bare `&` as text (ArkOS ships `2>&1` in a command) and
|
||||
accepts `--` inside a comment (EmuDeck comments out a command holding
|
||||
`--no-gui`). A comment is blanked to its newlines, so a system left in
|
||||
a comment neither parses nor lends its line to a citation.
|
||||
"""
|
||||
text = _COMMENT.sub(lambda m: "\n" * m.group(0).count("\n"), text)
|
||||
return _BARE_AMPERSAND.sub("&", text)
|
||||
|
||||
|
||||
def _xml_rows(text: str, key: str, path: str, params: dict) -> list[dict]:
|
||||
text = _lenient_xml(text)
|
||||
root = parse_untrusted_xml(text, path)
|
||||
lines = _all_lines(text, r"<name>([^<]+)</name>")
|
||||
seen: dict[str, int] = {}
|
||||
rule = params.get("default", "flagged")
|
||||
rows: list[dict] = []
|
||||
for system in root.iter("system"):
|
||||
name = (system.findtext("name") or "").strip()
|
||||
command = (system.findtext("command") or "").strip()
|
||||
platform = (system.findtext("platform") or "").strip()
|
||||
row = _name_once({"name": name}, lines, seen)
|
||||
if not name or "ignore" in (command, platform):
|
||||
continue
|
||||
folder = _strip_prefix((system.findtext("path") or "").strip(), params["path_prefixes"])
|
||||
if not folder:
|
||||
continue
|
||||
emulators = [
|
||||
(emulator.get("name") or "",
|
||||
[((core.text or "").strip(), core.get("default") == "true")
|
||||
for core in emulator.iter("core") if (core.text or "").strip()])
|
||||
for emulator in system.iter("emulator")
|
||||
]
|
||||
emulator, core = _pick_default(emulators, rule)
|
||||
if not emulators:
|
||||
# No <emulators>: the command starts the core itself (ArkOS
|
||||
# `retroarch -L .../mess_libretro.so %ROM%`).
|
||||
emulator, core = _libretro_in_command(command)
|
||||
rows.append({
|
||||
"folder": folder,
|
||||
"name": row["name"],
|
||||
"extensions": _extensions(system.findtext("extension")),
|
||||
"core": core or emulator,
|
||||
"emulator": emulator,
|
||||
"source_ref": _ref(key, path, row["line"]),
|
||||
"arcade": _is_arcade(system.findtext("hardware"), platform),
|
||||
"note": row.get("note"),
|
||||
})
|
||||
return rows
|
||||
|
||||
|
||||
@register("es-json")
|
||||
def es_json(spec: dict, fetcher: Fetcher) -> list[dict]:
|
||||
params = spec["params"]
|
||||
key = params.get("source", "os")
|
||||
path = params["file"]
|
||||
text = fetcher.text(key, path)
|
||||
lines = _all_lines(text, r'^\s{4,6}"name":\s*"([^"]+)"')
|
||||
seen: dict[str, int] = {}
|
||||
rows: list[dict] = []
|
||||
for system in json.loads(text).get("systems", []):
|
||||
name = system.get("name") or ""
|
||||
row = _name_once({"name": name}, lines, seen)
|
||||
if "ignore" in (system.get("command"), system.get("platform")):
|
||||
continue
|
||||
folder = _strip_prefix(system.get("path") or "", params["path_prefixes"])
|
||||
if not name or not folder:
|
||||
continue
|
||||
emulators = [
|
||||
(emulator.get("name") or "",
|
||||
[(c.get("name") or "", bool(c.get("default"))) for c in emulator.get("cores") or []])
|
||||
for emulator in system.get("emulators") or []
|
||||
]
|
||||
emulator, core = _pick_default(emulators, params.get("default", "flagged"))
|
||||
rows.append({
|
||||
"folder": folder,
|
||||
"name": row["name"],
|
||||
"extensions": _extensions(system.get("extensions")),
|
||||
"core": core or emulator,
|
||||
"emulator": emulator,
|
||||
"source_ref": _ref(key, path, row["line"]),
|
||||
"arcade": _is_arcade(system.get("hardware"), system.get("platform")),
|
||||
"note": row.get("note"),
|
||||
})
|
||||
return rows
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# ROCKNIX family: config/emulators/<system>.conf and the emulators package
|
||||
|
||||
|
||||
def _conf_values(text: str) -> tuple[dict[str, str], dict[str, int]]:
|
||||
values: dict[str, str] = {}
|
||||
lines: dict[str, int] = {}
|
||||
for number, line in enumerate(text.splitlines(), 1):
|
||||
match = re.match(r"^\s*(SYSTEM_[A-Z_]+)=(.*)$", line)
|
||||
if not match:
|
||||
continue
|
||||
raw = match.group(2).strip()
|
||||
try:
|
||||
value = " ".join(shlex.split(raw))
|
||||
except ValueError:
|
||||
value = raw.strip('"')
|
||||
values[match.group(1)] = value
|
||||
lines[match.group(1)] = number
|
||||
return values, lines
|
||||
|
||||
|
||||
def _package_calls(text: str) -> tuple[list[dict], dict[str, list[int]]]:
|
||||
"""add_emu_core and add_es_system calls of makeinstall_target, each
|
||||
with the `case` branches it sits under as (variable, pattern) pairs.
|
||||
An `if` block nests a condition whose truth the build decides: its
|
||||
calls carry it as a branch the default choice does not count as
|
||||
unconditional."""
|
||||
cores: list[dict] = []
|
||||
systems: dict[str, list[int]] = {}
|
||||
stack: list[list[str]] = []
|
||||
inside = False
|
||||
for number, raw in enumerate(text.splitlines(), 1):
|
||||
line = raw.split("#", 1)[0].strip()
|
||||
if raw.startswith("makeinstall_target()"):
|
||||
inside = True
|
||||
continue
|
||||
if not inside or not line:
|
||||
continue
|
||||
if raw.startswith("}"):
|
||||
break
|
||||
match = re.match(r'case\s+"?\$\{?(\w+)\}?"?\s+in', line)
|
||||
if match:
|
||||
stack.append(["case", match.group(1), ""])
|
||||
continue
|
||||
if line == "esac":
|
||||
stack.pop()
|
||||
continue
|
||||
if line.startswith("if ") or line.startswith("[") and "&&" not in line and line.endswith("then"):
|
||||
stack.append(["if", line, ""])
|
||||
continue
|
||||
if line in ("fi",):
|
||||
if stack and stack[-1][0] == "if":
|
||||
stack.pop()
|
||||
continue
|
||||
if line in ("then", "else") or line.startswith("elif "):
|
||||
continue
|
||||
ends = line.endswith(";;")
|
||||
if ends:
|
||||
line = line[:-2].strip()
|
||||
if stack and stack[-1][0] == "case":
|
||||
match = re.match(r"^([^()\s][^()]*)\)\s*(.*)$", line)
|
||||
if match and not line.startswith(("add_", "[", "sed", "cp", "mkdir")):
|
||||
stack[-1][2] = match.group(1).strip()
|
||||
line = match.group(2).strip()
|
||||
words = line.split()
|
||||
branches = [(s[1], s[2]) for s in stack]
|
||||
if words[:1] == ["add_emu_core"] and len(words) >= 4:
|
||||
cores.append({
|
||||
"system": words[1], "emulator": words[2], "core": words[3],
|
||||
"default": len(words) > 4 and words[4] == "true",
|
||||
"branches": branches, "line": number,
|
||||
})
|
||||
elif words[:1] == ["add_es_system"] and len(words) >= 2:
|
||||
systems.setdefault(words[1], []).append(number)
|
||||
if ends and stack and stack[-1][0] == "case":
|
||||
stack[-1][2] = ""
|
||||
return cores, systems
|
||||
|
||||
|
||||
def _board_key(branches: list[tuple[str, str]]) -> str | None:
|
||||
"""A default chosen under one `case` on the device names that device."""
|
||||
if len(branches) == 1 and branches[0][1] and branches[0][1] != "*":
|
||||
variable, pattern = branches[0]
|
||||
if not variable.startswith("["):
|
||||
return f"{variable}={pattern}"
|
||||
return None
|
||||
|
||||
|
||||
@register("elec-conf")
|
||||
def elec_conf(spec: dict, fetcher: Fetcher) -> list[dict]:
|
||||
"""A system ES lists is one add_es_system writes from its conf file
|
||||
(SYSTEM_PATH, SYSTEM_EXTENSION). The default is the core whose
|
||||
add_emu_core says true: unconditional, else the `*)` branch of a
|
||||
device case, the branches for named devices going to board_defaults."""
|
||||
params = spec["params"]
|
||||
key = params.get("source", "os")
|
||||
package = params["package"]
|
||||
cores, systems = _package_calls(fetcher.text(key, package))
|
||||
if not systems:
|
||||
raise SourceError(f"{package}: no add_es_system call")
|
||||
conf_dir = params["conf_dir"]
|
||||
rows: list[dict] = []
|
||||
for name, lines in systems.items():
|
||||
conf_path = f"{conf_dir}/{name}.conf"
|
||||
conf_text = fetcher.optional_text(key, conf_path)
|
||||
if conf_text is None:
|
||||
continue
|
||||
values, value_lines = _conf_values(conf_text)
|
||||
folder = _strip_prefix(values.get("SYSTEM_PATH", ""), params["path_prefixes"])
|
||||
if not folder:
|
||||
continue
|
||||
mine = [c for c in cores if c["system"] == name]
|
||||
flagged = [c for c in mine if c["default"]]
|
||||
plain = [c for c in flagged if not c["branches"]]
|
||||
star = [c for c in flagged if c["branches"] and c["branches"][-1][1] == "*"]
|
||||
pick = (plain or star or flagged or [None])[0]
|
||||
if pick is None:
|
||||
pick = next((c for c in mine if not c["branches"]), mine[0] if mine else None)
|
||||
board_defaults = {}
|
||||
for c in flagged:
|
||||
board = _board_key(c["branches"])
|
||||
if c is not pick and board and (c["emulator"], c["core"]) != (
|
||||
pick and pick["emulator"], pick and pick["core"]):
|
||||
board_defaults[board] = _board(c["core"], c["emulator"], _ref(key, package, c["line"]))
|
||||
refs = [
|
||||
_ref(key, conf_path, value_lines.get("SYSTEM_PATH"), value_lines.get("SYSTEM_EXTENSION")),
|
||||
_ref(key, package, *lines),
|
||||
]
|
||||
if pick:
|
||||
refs.append(_ref(key, package, pick["line"]))
|
||||
note = None
|
||||
if pick and pick["branches"] and not _board_key(pick["branches"]) and pick["branches"][-1][1] != "*":
|
||||
note = "default core chosen under a build condition: " + ", ".join(
|
||||
f"{v} {p}".strip() for v, p in pick["branches"])
|
||||
rows.append({
|
||||
"folder": folder,
|
||||
"name": name,
|
||||
"extensions": _extensions(values.get("SYSTEM_EXTENSION")),
|
||||
"core": pick["core"] if pick else None,
|
||||
"emulator": pick["emulator"] if pick else None,
|
||||
"source_ref": "; ".join(refs),
|
||||
"board_defaults": board_defaults,
|
||||
"arcade": values.get("SYSTEM_HARDWARE") == "arcade" or None,
|
||||
"note": note,
|
||||
})
|
||||
return rows
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# ES-DE (EmuDeck, RetroDECK)
|
||||
|
||||
|
||||
_RA_CORE = re.compile(r"%CORE_RETROARCH%[/\\]([\w.+-]+?)_libretro")
|
||||
_EMULATOR = re.compile(r"%EMULATOR_([A-Za-z0-9_-]+)%")
|
||||
|
||||
|
||||
def _command_core(command: str) -> tuple[str | None, str | None]:
|
||||
match = _RA_CORE.search(command)
|
||||
if match:
|
||||
return "retroarch", match.group(1)
|
||||
match = _EMULATOR.search(command)
|
||||
if match:
|
||||
name = match.group(1).lower()
|
||||
return name, name
|
||||
return None, None
|
||||
|
||||
|
||||
def _esde_systems(text: str, path: str) -> Iterator[tuple[str, str, list[str], list[tuple[str, str]]]]:
|
||||
root = parse_untrusted_xml(_lenient_xml(text), path)
|
||||
for system in root.iter("system"):
|
||||
name = (system.findtext("name") or "").strip()
|
||||
rom_path = (system.findtext("path") or "").strip()
|
||||
commands = [((c.get("label") or "").strip(), (c.text or "").strip())
|
||||
for c in system.findall("command")]
|
||||
yield name, rom_path, _extensions(system.findtext("extension")), commands
|
||||
|
||||
|
||||
def _alternative_emulators(text: str, function: str) -> dict[str, tuple[str, int]]:
|
||||
"""`ESDE_setEmu '<label>' <system>` calls inside one function."""
|
||||
out: dict[str, tuple[str, int]] = {}
|
||||
inside = False
|
||||
for number, line in enumerate(text.splitlines(), 1):
|
||||
if re.search(rf"\b{re.escape(function)}\b\s*\(\)|function\s+{re.escape(function)}\b", line):
|
||||
inside = True
|
||||
continue
|
||||
if not inside:
|
||||
continue
|
||||
if line.rstrip() == "}":
|
||||
break
|
||||
match = re.search(r"ESDE_setEmu\s+['\"]([^'\"]+)['\"]\s+([\w.-]+)", line)
|
||||
if match:
|
||||
out[match.group(2)] = (match.group(1), number)
|
||||
return out
|
||||
|
||||
|
||||
@register("esde")
|
||||
def esde(spec: dict, fetcher: Fetcher) -> list[dict]:
|
||||
"""Each system of ES-DE's list whose path is %ROMPATH%/<folder>.
|
||||
|
||||
The default emulator is the first <command> (FileData.cpp), unless an
|
||||
alternativeEmulator label is written for the system: an installer that
|
||||
sets one names a command of that system by its label. A custom system
|
||||
file replaces the bundled definition of the same name.
|
||||
"""
|
||||
params = spec["params"]
|
||||
systems: dict[str, dict] = {}
|
||||
for item in [params["systems"], *params.get("custom", [])]:
|
||||
key, path = item["source"], item["path"]
|
||||
text = _lenient_xml(fetcher.text(key, path))
|
||||
lines = _first_lines(text, r"<name>([^<]+)</name>")
|
||||
for name, rom_path, extensions, commands in _esde_systems(text, path):
|
||||
folder = _strip_prefix(rom_path, ["%ROMPATH%/", "%ROMPATH%\\"])
|
||||
if not name or not folder:
|
||||
continue
|
||||
systems[name] = {
|
||||
"folder": folder, "extensions": extensions, "commands": commands,
|
||||
"ref": _ref(key, path, lines.get(name)),
|
||||
}
|
||||
alternatives: dict[str, tuple[str, int]] = {}
|
||||
alt = params.get("alternatives")
|
||||
if alt:
|
||||
alternatives = _alternative_emulators(fetcher.text(alt["source"], alt["path"]), alt["function"])
|
||||
if not alternatives:
|
||||
raise SourceError(f"{alt['path']}: no ESDE_setEmu call in {alt['function']}")
|
||||
rows: list[dict] = []
|
||||
for name, system in systems.items():
|
||||
commands = system["commands"]
|
||||
refs = [system["ref"]]
|
||||
chosen = commands[0] if commands else ("", "")
|
||||
note = None
|
||||
if name in alternatives:
|
||||
label, line = alternatives[name]
|
||||
match = next((c for c in commands if c[0] == label), None)
|
||||
refs.append(_ref(alt["source"], alt["path"], line))
|
||||
if match:
|
||||
chosen = match
|
||||
else:
|
||||
note = f"alternative emulator '{label}' names no command of this system"
|
||||
emulator, core = _command_core(chosen[1])
|
||||
extensions = system["extensions"]
|
||||
if "." in extensions:
|
||||
# ES-DE's "." is a file without an extension (SystemData.cpp
|
||||
# getExtension), not a suffix the schema can hold.
|
||||
extensions = [e for e in extensions if e != "."]
|
||||
note = "; ".join(filter(None, [note, "files without an extension are games too ('.' in the list)"]))
|
||||
rows.append({
|
||||
"folder": system["folder"],
|
||||
"name": name,
|
||||
"extensions": extensions,
|
||||
"core": core,
|
||||
"emulator": emulator,
|
||||
"source_ref": "; ".join(refs),
|
||||
"note": note,
|
||||
})
|
||||
return rows
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Recalbox
|
||||
|
||||
|
||||
def _ini_sections(text: str) -> tuple[dict[str, dict[str, str]], dict[str, int]]:
|
||||
sections: dict[str, dict[str, str]] = {}
|
||||
lines: dict[str, int] = {}
|
||||
current: dict[str, str] | None = None
|
||||
name = ""
|
||||
for number, raw in enumerate(text.splitlines(), 1):
|
||||
line = raw.strip()
|
||||
if not line or line.startswith((";", "#")):
|
||||
continue
|
||||
if line.startswith("[") and line.endswith("]"):
|
||||
name = line[1:-1]
|
||||
current = sections.setdefault(name, {})
|
||||
lines.setdefault(name, number)
|
||||
elif current is not None and "=" in line:
|
||||
k, v = line.split("=", 1)
|
||||
current[k.strip()] = v.strip().strip('"')
|
||||
lines.setdefault(f"{name}/{k.strip()}", number)
|
||||
return sections, lines
|
||||
|
||||
|
||||
@register("recalbox")
|
||||
def recalbox(spec: dict, fetcher: Fetcher) -> list[dict]:
|
||||
"""One system.ini per system. The default core is the lowest
|
||||
`priority` (EmulatorManager), and `priority.<board>` changes it on that
|
||||
board. A `.ext!` extension is an archive the frontend hands over
|
||||
unopened."""
|
||||
params = spec["params"]
|
||||
key = params.get("source", "os")
|
||||
rows: list[dict] = []
|
||||
for entry in fetcher.entries(key, params["systems_dir"]):
|
||||
path = f"{entry}/system.ini"
|
||||
text = fetcher.optional_text(key, path)
|
||||
if text is None:
|
||||
continue
|
||||
sections, lines = _ini_sections(text)
|
||||
system = sections.get("system", {})
|
||||
folder = _strip_prefix(system.get("roms.folder", ""), ["%ROOT%/"])
|
||||
if not folder:
|
||||
continue
|
||||
cores = []
|
||||
for section, values in sections.items():
|
||||
parts = section.split(".")
|
||||
if parts[0] != "core" or len(parts) != 2:
|
||||
continue
|
||||
cores.append({
|
||||
"section": section, "emulator": values.get("emulator"), "core": values.get("core"),
|
||||
"priority": int(values.get("priority", "255")),
|
||||
"boards": {k.split(".", 1)[1]: int(v) for k, v in values.items() if k.startswith("priority.")},
|
||||
"extensions": values.get("extensions", "").split(),
|
||||
})
|
||||
if not cores:
|
||||
continue
|
||||
# EmulatorManager walks the cores from the last and keeps a strictly
|
||||
# lower priority: on a tie the last core listed wins.
|
||||
pick = min(reversed(cores), key=lambda c: c["priority"])
|
||||
board_defaults: dict[str, dict] = {}
|
||||
for board in sorted({b for c in cores for b in c["boards"]}):
|
||||
best = min(reversed(cores), key=lambda c: c["boards"].get(board, c["priority"]))
|
||||
if best is not pick:
|
||||
board_defaults[board] = _board(
|
||||
best["core"], best["emulator"],
|
||||
_ref(key, path, lines.get(f"{best['section']}/priority.{board}")))
|
||||
extensions = sorted({e for c in cores for e in c["extensions"]})
|
||||
archive = any(e.endswith("!") for e in extensions)
|
||||
rows.append({
|
||||
"folder": folder,
|
||||
"name": system.get("name"),
|
||||
"extensions": [e.rstrip("!") for e in extensions],
|
||||
"core": pick["core"],
|
||||
"emulator": pick["emulator"],
|
||||
"source_ref": "; ".join([
|
||||
_ref(key, path, lines.get("system/roms.folder")),
|
||||
_ref(key, path, lines.get(f"{pick['section']}/priority"),
|
||||
lines.get(f"{pick['section']}/core")),
|
||||
]),
|
||||
"board_defaults": board_defaults,
|
||||
"arcade": sections.get("properties", {}).get("type") == "arcade" or None,
|
||||
"note": "an extension ending in ! is an archive the frontend does not open" if archive else None,
|
||||
})
|
||||
return rows
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# RetroPie
|
||||
|
||||
|
||||
_MODULE_FIELD = re.compile(r'^rp_module_(id|section|flags)="([^"]*)"', re.MULTILINE)
|
||||
_ADD_EMULATOR = re.compile(r'^\s*addEmulator\s+(\S+)\s+("[^"]*"|\S+)\s+("[^"]*"|\S+)')
|
||||
_LIST = re.compile(r'(?:local\s+)?(\w+)=\(([^)]*)\)')
|
||||
_LOOP = re.compile(r'for (\w+) in (.+?); do')
|
||||
|
||||
|
||||
def _module_valid(flags: list[str], platform: set[str]) -> bool:
|
||||
"""packages.sh rp_registerModule: flags read in order, `!all` drops
|
||||
the module, a platform flag keeps it, `!<platform flag>` drops it."""
|
||||
valid = True
|
||||
for flag in flags:
|
||||
if flag == "!all":
|
||||
valid = False
|
||||
elif flag in platform:
|
||||
valid = True
|
||||
elif flag.startswith("!") and flag[1:] in platform:
|
||||
valid = False
|
||||
return valid
|
||||
|
||||
|
||||
def _emulator_calls(text: str) -> Iterator[tuple[int, str, str, list[str]]]:
|
||||
"""(line, default flag, emulator id, systems) of each addEmulator call,
|
||||
a system variable expanded through the loop or array that sets it."""
|
||||
lines = text.splitlines()
|
||||
arrays = {m.group(1): m.group(2).replace('"', "").split() for m in _LIST.finditer(text)}
|
||||
loops: list[tuple[int, str, list[str]]] = []
|
||||
for index, line in enumerate(lines):
|
||||
match = _LOOP.search(line)
|
||||
if match:
|
||||
items = match.group(2).strip()
|
||||
array = re.fullmatch(r'"?\$\{(\w+)\[@\]\}"?', items)
|
||||
loops.append((index, match.group(1),
|
||||
arrays.get(array.group(1), []) if array else items.replace('"', "").split()))
|
||||
for index, line in enumerate(lines):
|
||||
match = _ADD_EMULATOR.match(line)
|
||||
if not match or "function addEmulator" in line:
|
||||
continue
|
||||
flag, emu, system = match.group(1), match.group(2).strip('"'), match.group(3).strip('"')
|
||||
systems = [system]
|
||||
var = re.fullmatch(r"\$\{?(\w+)\}?", system)
|
||||
if var:
|
||||
found = [items for i, name, items in loops if i < index and name == var.group(1)]
|
||||
if found:
|
||||
systems = found[-1]
|
||||
elif var.group(1) in arrays:
|
||||
systems = arrays[var.group(1)]
|
||||
else:
|
||||
continue
|
||||
yield index + 1, flag, emu, systems
|
||||
|
||||
|
||||
@register("retropie")
|
||||
def retropie(spec: dict, fetcher: Fetcher) -> list[dict]:
|
||||
"""Folders are `<system>` under the roms root (helpers.sh addSystem),
|
||||
with `<system>_exts` from platforms.cfg. The default is set once, by
|
||||
the first installed module whose addEmulator passes 1: a basic install
|
||||
installs the core then the main section in registration order
|
||||
(setup.sh basic_install_setup, packages.sh rp_registerAllModules). A
|
||||
system no basic-install module serves gets its folder from an optional
|
||||
module, so the first opt, then exp, module stands as its default."""
|
||||
params = spec["params"]
|
||||
key = params.get("source", "os")
|
||||
cfg_path = params["platforms_cfg"]
|
||||
cfg = fetcher.text(key, cfg_path)
|
||||
exts: dict[str, tuple[str, int]] = {}
|
||||
for number, line in enumerate(cfg.splitlines(), 1):
|
||||
match = re.match(r'^([\w-]+)_exts="([^"]*)"', line)
|
||||
if match:
|
||||
exts[match.group(1)] = (match.group(2), number)
|
||||
platform = set(params["platform_flags"])
|
||||
modules: list[dict] = []
|
||||
for kind in params["module_dirs"]:
|
||||
for path in fetcher.tree(key, f"scriptmodules/{kind}/"):
|
||||
if not path.endswith(".sh") or path.count("/") != 2:
|
||||
continue
|
||||
text = fetcher.text(key, path)
|
||||
fields = dict(_MODULE_FIELD.findall(text))
|
||||
if not _module_valid(fields.get("flags", "").split(), platform):
|
||||
continue
|
||||
modules.append({"path": path, "id": fields.get("id", ""),
|
||||
"section": fields.get("section", ""), "text": text})
|
||||
order = [m for section in params["install_sections"] for m in modules if m["section"] == section]
|
||||
defaults: dict[str, tuple[str, str, int]] = {}
|
||||
for module in order:
|
||||
for line, flag, emu, systems in _emulator_calls(module["text"]):
|
||||
if flag != "1":
|
||||
continue
|
||||
emu = emu.replace("${md_id}", module["id"]).replace("$md_id", module["id"])
|
||||
for system in systems:
|
||||
defaults.setdefault(system, (emu, module["path"], line))
|
||||
rows: list[dict] = []
|
||||
for system, (raw, line) in sorted(exts.items()):
|
||||
refs = [_ref(key, cfg_path, line)]
|
||||
core = None
|
||||
if system in defaults:
|
||||
core, path, call = defaults[system]
|
||||
refs.append(_ref(key, path, call))
|
||||
emulator = "retroarch" if core and core.startswith("lr-") else core
|
||||
rows.append({
|
||||
"folder": system,
|
||||
"extensions": _extensions(raw),
|
||||
"core": core,
|
||||
"emulator": emulator,
|
||||
"source_ref": "; ".join(refs),
|
||||
})
|
||||
return rows
|
||||
Reference in new issue
Block a user