mirror of
https://github.com/Abdess/retroarch_system.git
synced 2026-10-22 03:39:14 -05:00
feat: read platform launchers in verify and packs
A names_searched list is one place the launcher fills with the first name it finds, so a core file it would never open once another name of the list is held is not reported missing. A place no declared file and no packed core file fills receives, under the launcher's name, a file the core accepts by content. The layer is applied inside find_undeclared_files, which verify and the pack builder both read, and a test reads their source to keep it there.
This commit is contained in:
Abdessamad Derraz committed
1 parent
8660892d60
commit
b2b8dcd62c
6 files changed
+752
-1
No files matched your search
@@ -0,0 +1,473 @@
|
||||
"""The platform's launcher layer, read by verify and by the pack builder.
|
||||
|
||||
A platform's own launch code often decides which name a file is read under.
|
||||
Batocera's DuckStation generator tries 22 names per region and keeps the
|
||||
first it finds, RetroBat's ares launcher writes the LaserActive image it
|
||||
finds into ares's settings. The platform scraper records that code under
|
||||
``launchers:`` and this module turns the records into places: a set of
|
||||
destinations, relative to the directory core files go to, any one of which
|
||||
the launcher reads for one purpose.
|
||||
|
||||
Two consequences follow, both computed here once for every reader. A core
|
||||
file the launcher would not read because another name of the same place is
|
||||
held is not missing on that platform. A place nothing fills yet receives,
|
||||
under the launcher's name, a file the core accepts by content.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import re
|
||||
from collections.abc import Callable, Iterable
|
||||
from dataclasses import dataclass
|
||||
|
||||
from common import declared_root, name_match_size_ok, sanitize_pack_path
|
||||
|
||||
# The kinds whose names say which file the launcher opens. The others hand
|
||||
# over a directory, copy a file the platform already names, or let the
|
||||
# emulator choose.
|
||||
PLACE_KINDS = ("names_searched", "fixed_name", "core_option")
|
||||
|
||||
_LABELLED = re.compile(r"^([^\s()]+\.[A-Za-z0-9]+) \([^()]*\)$")
|
||||
_PLACEHOLDER = re.compile(r"\{(\w+)\}")
|
||||
_HASH_FIELDS = ("sha1", "md5", "crc32", "sha256")
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class LauncherPlace:
|
||||
"""One file a launcher opens, under any of ``paths``.
|
||||
|
||||
``paths`` are in the order the launcher tries them, relative to the
|
||||
directory core files go to. An unanchored place has no directory of its
|
||||
own: the core resolves the name the launcher passes in the directory it
|
||||
reads the entry from. ``alternatives`` is true when the launcher keeps
|
||||
the first name it finds, false when each path is the one it passes.
|
||||
"""
|
||||
|
||||
emulator: str
|
||||
build: str
|
||||
kind: str
|
||||
slot: str
|
||||
paths: tuple[str, ...]
|
||||
case_insensitive: bool
|
||||
anchored: bool
|
||||
conditional: bool
|
||||
source_ref: str
|
||||
|
||||
def fold(self, name: str) -> str:
|
||||
return name.lower() if self.case_insensitive else name
|
||||
|
||||
def basenames(self) -> set[str]:
|
||||
return {self.fold(p.rsplit("/", 1)[-1]) for p in self.paths}
|
||||
|
||||
@property
|
||||
def stands_for_default(self) -> bool:
|
||||
"""Whether filling the place means the core's default is never read.
|
||||
|
||||
A launcher that always passes the file it found replaces the name
|
||||
the core would open by itself. A record that applies only under an
|
||||
option or a system leaves that default in use elsewhere, and a core
|
||||
option offers a file the user may choose without ever choosing it.
|
||||
"""
|
||||
return self.kind in ("names_searched", "fixed_name") and not self.conditional
|
||||
|
||||
|
||||
def _bios_relative(directory: str, base: str) -> str | None:
|
||||
"""A launcher directory relative to the BIOS root, None when it is not one."""
|
||||
directory = directory.strip().rstrip("/")
|
||||
roots = {base, "<system_directory>"}
|
||||
head, _, rest = directory.partition("/")
|
||||
if directory in roots:
|
||||
return ""
|
||||
if head in roots:
|
||||
relative = rest
|
||||
elif directory.startswith("/") and base and f"/{base}/" in f"{directory}/":
|
||||
relative = f"{directory}/".split(f"/{base}/", 1)[1].rstrip("/")
|
||||
else:
|
||||
return None
|
||||
return None if re.search(r"[\s<>]", relative) else relative
|
||||
|
||||
|
||||
def _directories(record: dict, base: str) -> list[str] | None:
|
||||
"""The directories a record reads, in order. None when the code says none.
|
||||
|
||||
``search_dir`` transcribes the code, conditions included: "bios/chihiro,
|
||||
then bios", "bios/pcsx2/bios if any listed name is there, else bios".
|
||||
A directory the record describes rather than names ("directory of the
|
||||
Chihiro BIOS found") cannot be placed.
|
||||
"""
|
||||
raw = record.get("search_dir")
|
||||
if raw is None:
|
||||
return None
|
||||
text = re.sub(r"\([^()]*\)", "", str(raw))
|
||||
text = re.sub(r"\s+if\s+[^,]*", "", text)
|
||||
found: list[str] = []
|
||||
for part in re.split(r",|\bthen\b|\belse\b", text):
|
||||
if not part.strip():
|
||||
continue
|
||||
relative = _bios_relative(part, base)
|
||||
if relative is None:
|
||||
return []
|
||||
if relative not in found:
|
||||
found.append(relative)
|
||||
return found
|
||||
|
||||
|
||||
def _names(record: dict) -> list[str]:
|
||||
"""The file names a record tries, placeholders expanded in code order."""
|
||||
expands = record.get("expands") or {}
|
||||
names: list[str] = []
|
||||
for raw in record.get("names") or []:
|
||||
name = str(raw).strip()
|
||||
labelled = _LABELLED.match(name)
|
||||
if labelled:
|
||||
name = labelled.group(1)
|
||||
expanded = [name]
|
||||
for key in _PLACEHOLDER.findall(name):
|
||||
values = expands.get(key)
|
||||
if values is None:
|
||||
expanded = []
|
||||
break
|
||||
expanded = [
|
||||
candidate.replace("{" + key + "}", str(value))
|
||||
for candidate in expanded
|
||||
for value in values
|
||||
]
|
||||
for candidate in expanded:
|
||||
if "<" in candidate or "." not in candidate.rsplit("/", 1)[-1]:
|
||||
continue
|
||||
if candidate not in names:
|
||||
names.append(candidate)
|
||||
return names
|
||||
|
||||
|
||||
def launcher_places(config: dict) -> list[LauncherPlace]:
|
||||
"""Every place a platform's launchers open a named file from."""
|
||||
base = str(config.get("base_destination") or "bios")
|
||||
places: list[LauncherPlace] = []
|
||||
for record in config.get("launchers") or []:
|
||||
kind = record.get("kind")
|
||||
if kind not in PLACE_KINDS:
|
||||
continue
|
||||
names = _names(record)
|
||||
directories = _directories(record, base)
|
||||
if not names or directories == []:
|
||||
continue
|
||||
anchored = directories is not None
|
||||
|
||||
def paths_of(name: str) -> tuple[str, ...]:
|
||||
if not anchored:
|
||||
return (name,)
|
||||
return tuple(
|
||||
sanitize_pack_path(f"{d}/{name}" if d else name) for d in directories
|
||||
)
|
||||
|
||||
common = {
|
||||
"emulator": str(record.get("emulator", "")),
|
||||
"build": str(record.get("build", "")),
|
||||
"kind": kind,
|
||||
"slot": str(record.get("slot") or ""),
|
||||
# A filesystem that folds case folds it for the launcher too.
|
||||
"case_insensitive": bool(
|
||||
record.get("case_insensitive") or config.get("case_insensitive_fs")
|
||||
),
|
||||
"anchored": anchored,
|
||||
"conditional": bool(record.get("depends_on_option")),
|
||||
"source_ref": str(record.get("source_ref", "")),
|
||||
}
|
||||
if kind == "names_searched":
|
||||
# Each directory in turn, every name in it.
|
||||
depth = len(directories) if anchored else 1
|
||||
ordered = tuple(paths_of(name)[i] for i in range(depth) for name in names)
|
||||
places.append(LauncherPlace(paths=ordered, **common))
|
||||
continue
|
||||
# A fixed name is the one file passed, and a core option offers
|
||||
# each value as the file the user picks: one place per name.
|
||||
places.extend(LauncherPlace(paths=paths_of(name), **common) for name in names)
|
||||
return places
|
||||
|
||||
|
||||
def declared_destinations(config: dict) -> set[str]:
|
||||
"""Each destination the platform declares, relative to the core file directory.
|
||||
|
||||
Case is kept: a launcher that matches names exactly is not served by a
|
||||
declaration of another case.
|
||||
"""
|
||||
prefix = declared_root(config)
|
||||
found: set[str] = set()
|
||||
for system in config.get("systems", {}).values():
|
||||
for entry in system.get("files", []):
|
||||
slot = sanitize_pack_path(entry.get("destination") or entry.get("name", ""))
|
||||
if prefix and slot.lower().startswith(prefix.lower() + "/"):
|
||||
slot = slot[len(prefix) + 1:]
|
||||
if slot:
|
||||
found.add(slot)
|
||||
return found
|
||||
|
||||
|
||||
def _hashes(entry: dict) -> dict[str, set[str]]:
|
||||
found: dict[str, set[str]] = {}
|
||||
for field in _HASH_FIELDS:
|
||||
value = entry.get(field) or (entry.get("expected") or {}).get(field)
|
||||
if not value:
|
||||
continue
|
||||
values = value if isinstance(value, list) else str(value).split(",")
|
||||
found[field] = {str(v).strip().lower() for v in values if str(v).strip()}
|
||||
return found
|
||||
|
||||
|
||||
def accepts(entry: dict, record: dict) -> bool:
|
||||
"""Whether a core entry takes the database file ``record`` by its content."""
|
||||
hashes = _hashes(entry)
|
||||
if not hashes:
|
||||
return False
|
||||
return any(
|
||||
str(record.get(field, "")).lower() in values
|
||||
for field, values in hashes.items()
|
||||
)
|
||||
|
||||
|
||||
@dataclass
|
||||
class _Candidate:
|
||||
"""A core entry one launcher place stands for, and how it was matched."""
|
||||
|
||||
entry: dict
|
||||
names: set[str]
|
||||
by_content: bool
|
||||
|
||||
|
||||
class LauncherLayer:
|
||||
"""The launcher places of one platform matched to the cores it runs.
|
||||
|
||||
``declared`` holds each destination the platform declares, relative to
|
||||
the directory core files go to, case kept.
|
||||
"""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
config: dict,
|
||||
profiles: dict,
|
||||
relevant: Iterable[str],
|
||||
db: dict,
|
||||
declared: Iterable[str],
|
||||
) -> None:
|
||||
self.places = launcher_places(config)
|
||||
self.declared = set(declared)
|
||||
self.declared_folded = {d.lower() for d in self.declared}
|
||||
self.declared_names = {d.rsplit("/", 1)[-1] for d in self.declared}
|
||||
self.declared_names_folded = {n.lower() for n in self.declared_names}
|
||||
by_name = db.get("indexes", {}).get("by_name", {})
|
||||
self._by_name = by_name
|
||||
self._by_name_folded: dict[str, list[str]] = {}
|
||||
for name, sha1s in by_name.items():
|
||||
self._by_name_folded.setdefault(name.lower(), []).extend(sha1s)
|
||||
self._files = db.get("files", {})
|
||||
self._owners: dict[tuple[str, bool], list[LauncherPlace]] = {}
|
||||
relevant = set(relevant)
|
||||
for place in self.places:
|
||||
standalone = place.build == "standalone"
|
||||
for emulator in (e.strip() for e in place.emulator.split(",")):
|
||||
for key in self._profiles_named(emulator, profiles, relevant):
|
||||
self._owners.setdefault((key, standalone), []).append(place)
|
||||
|
||||
@staticmethod
|
||||
def _profiles_named(name: str, profiles: dict, relevant: set[str]) -> list[str]:
|
||||
"""The profiles a platform's emulator name designates, among those it runs."""
|
||||
return sorted(
|
||||
key for key in relevant
|
||||
if (profile := profiles.get(key)) is not None
|
||||
and profile.get("type") not in ("alias", "launcher")
|
||||
and (key == name or name in {str(c) for c in profile.get("cores") or []})
|
||||
)
|
||||
|
||||
def __bool__(self) -> bool:
|
||||
return bool(self._owners)
|
||||
|
||||
def places_for(self, emulator: str, standalone: bool) -> list[LauncherPlace]:
|
||||
return self._owners.get((emulator, standalone), [])
|
||||
|
||||
def files_named(self, place: LauncherPlace, name: str) -> list[dict]:
|
||||
"""The database files carrying a name, in the place's case rule."""
|
||||
if place.case_insensitive:
|
||||
sha1s = self._by_name_folded.get(name.lower(), [])
|
||||
else:
|
||||
sha1s = self._by_name.get(name, [])
|
||||
return [self._files[s] for s in dict.fromkeys(sha1s) if s in self._files]
|
||||
|
||||
def match(self, place: LauncherPlace, entry: dict) -> _Candidate | None:
|
||||
"""How a core entry belongs to a place: by a name it is read under, or by content."""
|
||||
basenames = place.basenames()
|
||||
own = {
|
||||
place.fold(str(n).rsplit("/", 1)[-1])
|
||||
for n in (entry.get("name", ""), *(entry.get("aliases") or []))
|
||||
if n
|
||||
}
|
||||
names = own & basenames
|
||||
if names:
|
||||
return _Candidate(entry, names, False)
|
||||
content = {
|
||||
place.fold(path.rsplit("/", 1)[-1])
|
||||
for path in place.paths
|
||||
if any(
|
||||
accepts(entry, record)
|
||||
for record in self.files_named(place, path.rsplit("/", 1)[-1])
|
||||
)
|
||||
}
|
||||
if content:
|
||||
return _Candidate(entry, content, True)
|
||||
return None
|
||||
|
||||
def platform_holds(self, place: LauncherPlace) -> bool:
|
||||
"""Whether the platform's own list declares a file the place reads."""
|
||||
if place.anchored:
|
||||
pool = self.declared_folded if place.case_insensitive else self.declared
|
||||
return any(place.fold(path) in pool for path in place.paths)
|
||||
pool = self.declared_names_folded if place.case_insensitive else self.declared_names
|
||||
return bool(place.basenames() & pool)
|
||||
|
||||
def settle(
|
||||
self,
|
||||
candidates: list[tuple[str, bool, dict, dict]],
|
||||
resolve_held: Callable[[dict], dict | None],
|
||||
) -> tuple[list[dict], set[int]]:
|
||||
"""What the launcher layer adds to the report and what it settles.
|
||||
|
||||
``candidates`` holds (profile, standalone, profile entry, report
|
||||
entry) for every core entry the report considered. ``resolve_held``
|
||||
returns the database record of the file the collection holds for a
|
||||
report entry, None when it holds none.
|
||||
|
||||
Returns the report entries to add, one per place nothing filled that
|
||||
a held file the core accepts can fill under the launcher's name, and
|
||||
the ids of the report entries the filled places settle.
|
||||
"""
|
||||
added: list[dict] = []
|
||||
settled: set[int] = set()
|
||||
# Destinations the report already fills with a held file, whichever
|
||||
# core asked for it: the pack puts a file there either way.
|
||||
reported_at: dict[str, dict] = {}
|
||||
for _emulator, _standalone, _entry, report in candidates:
|
||||
if report.get("in_repo"):
|
||||
destination = sanitize_pack_path(report.get("path") or report.get("name", ""))
|
||||
reported_at.setdefault(destination, report)
|
||||
by_layout: dict[tuple[str, bool], list[tuple[dict, dict]]] = {}
|
||||
for emulator, standalone, entry, report in candidates:
|
||||
by_layout.setdefault((emulator, standalone), []).append((entry, report))
|
||||
placed: set[str] = set()
|
||||
for (emulator, standalone), members in sorted(by_layout.items()):
|
||||
for place in self.places_for(emulator, standalone):
|
||||
matched = [
|
||||
(candidate, report)
|
||||
for entry, report in members
|
||||
if (candidate := self.match(place, entry)) is not None
|
||||
]
|
||||
if not matched:
|
||||
continue
|
||||
filled = self._fill(place, matched, reported_at, resolve_held, placed)
|
||||
if filled is None:
|
||||
continue
|
||||
if isinstance(filled, dict):
|
||||
added.append(filled)
|
||||
if place.stands_for_default:
|
||||
settled.update(
|
||||
id(report) for _candidate, report in matched
|
||||
if not report.get("in_repo")
|
||||
)
|
||||
return added, settled
|
||||
|
||||
def _fill(
|
||||
self,
|
||||
place: LauncherPlace,
|
||||
matched: list[tuple[_Candidate, dict]],
|
||||
reported_at: dict[str, dict],
|
||||
resolve_held: Callable[[dict], dict | None],
|
||||
placed: set[str],
|
||||
) -> dict | bool | None:
|
||||
"""True when the place is filled already, a new report entry when one fills it."""
|
||||
if self.platform_holds(place):
|
||||
return True
|
||||
folded_reported = {place.fold(dest): r for dest, r in reported_at.items()}
|
||||
for path in place.paths:
|
||||
name = path.rsplit("/", 1)[-1]
|
||||
key = place.fold(name)
|
||||
if place.anchored:
|
||||
if place.fold(path) in folded_reported:
|
||||
return True
|
||||
elif any(
|
||||
key == place.fold(r.get("name", "")) or key == place.fold(dest.rsplit("/", 1)[-1])
|
||||
for dest, r in folded_reported.items()
|
||||
):
|
||||
return True
|
||||
if not place.anchored:
|
||||
continue
|
||||
source = self._content_for(place, name, key, matched, resolve_held)
|
||||
if source is None:
|
||||
continue
|
||||
report, record = source
|
||||
if place.fold(path) in placed:
|
||||
return True
|
||||
placed.add(place.fold(path))
|
||||
return _launcher_entry(report, name, path, record, place)
|
||||
return None
|
||||
|
||||
def _content_for(
|
||||
self,
|
||||
place: LauncherPlace,
|
||||
name: str,
|
||||
key: str,
|
||||
matched: list[tuple[_Candidate, dict]],
|
||||
resolve_held: Callable[[dict], dict | None],
|
||||
) -> tuple[dict, dict] | None:
|
||||
"""A held file the core accepts for one launcher name, and its entry.
|
||||
|
||||
The file the database knows under that name comes first when the
|
||||
core accepts it: by content when the entry pins one, by name and
|
||||
size when the entry reads it under that name. Otherwise the file the
|
||||
collection holds for an entry read under that name.
|
||||
"""
|
||||
for candidate, report in matched:
|
||||
if key not in candidate.names:
|
||||
continue
|
||||
for record in self.files_named(place, name):
|
||||
if accepts(candidate.entry, record) or (
|
||||
not candidate.by_content
|
||||
and not _hashes(candidate.entry)
|
||||
and name_match_size_ok(candidate.entry, record.get("size"))
|
||||
):
|
||||
return report, record
|
||||
for candidate, report in matched:
|
||||
if key not in candidate.names or candidate.by_content:
|
||||
continue
|
||||
if not report.get("in_repo"):
|
||||
continue
|
||||
record = resolve_held(report)
|
||||
if record is not None:
|
||||
return report, record
|
||||
return None
|
||||
|
||||
|
||||
def _launcher_entry(
|
||||
report: dict, name: str, path: str, record: dict, place: LauncherPlace
|
||||
) -> dict:
|
||||
"""The report entry asking for a file under the name a launcher opens."""
|
||||
entry = {
|
||||
key: report.get(key)
|
||||
for key in (
|
||||
"profile", "emulator", "systems", "system", "region", "variant_group",
|
||||
"required", "hle_fallback", "category",
|
||||
)
|
||||
}
|
||||
entry.update(
|
||||
name=name,
|
||||
path=path,
|
||||
bundled=False,
|
||||
in_repo=True,
|
||||
note="",
|
||||
checks=[],
|
||||
source_ref=None,
|
||||
expected={},
|
||||
sha1=record.get("sha1"),
|
||||
md5=record.get("md5"),
|
||||
launcher=place.source_ref,
|
||||
launcher_of=report.get("name", ""),
|
||||
)
|
||||
return entry
|
||||
Reference in new issue
Block a user