kmoody/libretro
1
0 Fork 0
mirror of https://github.com/Abdess/retroarch_system.git synced 2026-10-11 05:53:23 -05:00
Code Issues Packages Projects Releases Wiki Activity
Files
527526a8e2be1b251e18c2ee9f3e2dcbb5adac06
libretro/install.py
T

2295 lines
85 KiB
Python
Raw Blame History

#!/usr/bin/env python3
"""Universal BIOS installer for retrogaming platforms.
Self-contained script using only Python stdlib. Downloads missing BIOS files
from the retrobios repository and places them in the correct location for
the detected emulator platform.
Usage:
python install.py
python install.py --platform retroarch --dest ~/custom/bios
python install.py --check
python install.py --list-platforms
python install.py --platform retroarch --system sony-playstation --region us
"""
from __future__ import annotations
import argparse
import concurrent.futures
import hashlib
import json
import os
import platform
import re
import shutil
import sys
import tempfile
import threading
import urllib.error
import urllib.parse
import urllib.request
import xml.etree.ElementTree as ET
from pathlib import Path, PurePosixPath
# Manifests are read from the same ref the bootstrap verified this installer
# against, so a file list and the code reading it always come from one commit.
# RETROBIOS_REF pins an installation to a tag when reproducibility matters.
DEFAULT_RELEASE_REF = "main"
RELEASE_REF = os.environ.get("RETROBIOS_REF", DEFAULT_RELEASE_REF)
DEFAULT_BASE_URL = (
"https://raw.githubusercontent.com/Abdess/retrobios/"
+ urllib.parse.quote(RELEASE_REF, safe="")
)
_LOOPBACK_HOSTS = ("localhost", "127.0.0.1", "::1")
def _checked_base_url(value: str) -> str:
"""Refuse a base URL that is neither HTTPS nor loopback.
This base serves the manifest and the files it declares, so whoever
controls it controls the expected hashes too and verification stops
proving anything. install.sh and install.ps1 make the same check on the
URL they fetch the installer from.
Plain HTTP to loopback stays allowed: nothing sits between the two ends
to intercept it, and it is how the installer is exercised end to end.
"""
parsed = urllib.parse.urlparse(value)
if parsed.scheme == "https":
return value
host = (parsed.hostname or "").lower()
if parsed.scheme == "http" and host in _LOOPBACK_HOSTS:
return value
print(
f"Error: RETROBIOS_BASE_URL must use HTTPS, got {value!r}",
file=sys.stderr,
)
sys.exit(1)
BASE_URL = _checked_base_url(os.environ.get("RETROBIOS_BASE_URL", DEFAULT_BASE_URL))
MANIFEST_URL = f"{BASE_URL}/install/{{platform}}.json"
TARGETS_URL = f"{BASE_URL}/install/targets/{{platform}}.json"
RAW_FILE_URL = f"{BASE_URL}/{{path}}"
RELEASE_URL = (
"https://github.com/Abdess/retrobios/releases/download/large-files/{asset}"
)
MAX_RETRIES = 3
MAX_MANIFEST_BYTES = 16 * 1024 * 1024
MAX_TARGETS_BYTES = 4 * 1024 * 1024
MAX_MANIFEST_FILES = 100_000
MAX_DOWNLOAD_SIZE = 1024 * 1024 * 1024
MAX_TOTAL_DOWNLOAD_SIZE = 64 * 1024 * 1024 * 1024
# Why a manifest leaves a file out, as the installer words it. The manifest
# schema and the generator are held to this list by the tests.
OMISSION_REASONS = {
"hash_mismatch": "hash mismatch",
"not_found": "not found",
"external": "external",
"user_provided": "user provided",
"no_platform_slug": "no platform folder",
}
_SHA1_RE = re.compile(r"^[0-9a-fA-F]{40}$")
_SHA256_RE = re.compile(r"^[0-9a-fA-F]{64}$")
# The region vocabulary of scripts/region.py, which this file cannot import.
# A test holds the two copies equal.
WORLD_REGION = "world"
REGION_TREE: dict[str, frozenset] = {
"north-america": frozenset({"canada"}),
"latin-america": frozenset({"brazil", "mexico", "argentina"}),
"europe": frozenset({
"uk", "france", "germany", "italy", "spain", "netherlands", "portugal",
"greece", "poland", "russia", "sweden", "norway", "denmark", "finland",
}),
"asia": frozenset({
"japan", "south-korea", "china", "taiwan", "hong-kong", "singapore",
"india", "asia-ntsc", "asia-pal",
}),
"oceania": frozenset({"australia", "new-zealand"}),
}
REGION_ALIASES: dict[str, str] = {
"jp": "japan",
"ntsc-j": "japan",
"us": "north-america",
"usa": "north-america",
"na": "north-america",
"ntsc-u": "north-america",
"eu": "europe",
"pal": "europe",
"kr": "south-korea",
"korea": "south-korea",
"auto": WORLD_REGION,
"gb": "uk",
}
_REGION_PARENT = {
member: parent for parent, members in REGION_TREE.items() for member in members
}
REGIONS = frozenset({WORLD_REGION} | set(REGION_TREE) | set(_REGION_PARENT))
MAX_SELECTION_TAGS = 1000
SELECTION_FIELDS = ("systems", "regions", "region_groups", "region_system_groups")
# The files cores load for systems the platform does not list (game engines,
# computers): no system of the platform owns them, and choosing systems
# leaves them out unless this name is chosen too.
OTHER_SYSTEMS = "other"
# Platforms with a manifest in install/. Manifest URLs are case sensitive,
# so user input is normalized against this list before any fetch.
AVAILABLE_PLATFORMS = (
"retroarch", "batocera", "recalbox", "retrobat", "emudeck",
"lakka", "retrodeck", "rocknix", "romm", "bizhawk", "misterfpga",
"retropie",
)
# Fallback BIOS destination per platform when --platform is forced
# but auto-detection finds nothing on the machine.
DEFAULT_DESTS = {
"batocera": Path("/userdata/bios"),
"recalbox": Path("/recalbox/share/bios"),
"lakka": Path("/storage/system"),
"retrodeck": Path.home() / "retrodeck",
"emudeck": Path.home() / "Emulation" / "bios",
"rocknix": Path("/storage/roms/bios"),
"misterfpga": Path("/media/fat/games"),
"retropie": Path.home() / "RetroPie" / "BIOS",
"romm": Path("/romm/library/bios"),
}
def _os_default_dests(os_type: str) -> dict[str, Path]:
"""Where a platform keeps its BIOS on this OS, when it is not detected.
The same folders the platform's detection looks for: a platform named on
the command line but not found must not land in ~/bios, which none of
them reads.
"""
home = Path.home()
if os_type == "windows":
profile = Path(os.environ.get("USERPROFILE", str(home)))
appdata = Path(os.environ.get("APPDATA", str(home / "AppData" / "Roaming")))
return {
"retroarch": appdata / "RetroArch" / "system",
# RetroBat's installer defaults to C:\RetroBat (installer.iss).
"retrobat": Path("C:/RetroBat/bios"),
"bizhawk": profile / "BizHawk" / "Firmware",
}
if os_type == "darwin":
return {
"retroarch": home / "Library" / "Application Support" / "RetroArch" / "system",
}
return {
"retroarch": home / ".config" / "retroarch" / "system",
"bizhawk": home / ".config" / "BizHawk" / "Firmware",
}
# Set when the run is interrupted, so queued downloads end instead of draining
# the pool: every future is submitted before the first one completes.
_stop = threading.Event()
class _Interrupted(Exception):
"""Raised inside a worker once the run has been stopped."""
# The three RetroArch packages published for Android, newest ABI first.
ANDROID_RETROARCH_PACKAGES = (
"com.retroarch.aarch64",
"com.retroarch",
"com.retroarch.ra32",
)
def _is_android() -> bool:
"""Android reports itself as Linux, so the runtime is named instead.
Termux inherits the two variables the Android runtime exports, and
/system/build.prop covers a shell that was started without them.
"""
if os.environ.get("ANDROID_ROOT") and os.environ.get("ANDROID_DATA"):
return True
return Path("/system/build.prop").exists()
def detect_os() -> str:
"""Return normalized OS identifier.
RETROBIOS_OS names the platform outright (linux, wsl, windows, darwin,
android), for a run that has to behave like another host: the PowerShell
wrapper tests drive a Windows layout on a Linux runner.
"""
forced = os.environ.get("RETROBIOS_OS", "").strip().lower()
if forced in ("linux", "wsl", "windows", "darwin", "android"):
return forced
system = platform.system().lower()
if system == "linux":
if _is_android():
return "android"
proc_version = Path("/proc/version")
if proc_version.exists():
try:
content = proc_version.read_text(encoding="utf-8", errors="replace")
if "microsoft" in content.lower():
return "wsl"
except OSError:
pass
return "linux"
if system == "darwin":
return "darwin"
if system == "windows":
return "windows"
return system
def _parse_os_release() -> dict[str, str]:
"""Parse /etc/os-release KEY=value format."""
result: dict[str, str] = {}
path = Path("/etc/os-release")
if not path.exists():
return result
try:
for line in path.read_text(encoding="utf-8", errors="replace").splitlines():
line = line.strip()
if "=" not in line or line.startswith("#"):
continue
key, _, value = line.partition("=")
value = value.strip('"').strip("'")
result[key] = value
except OSError:
pass
return result
def _expand_retroarch_path(value: str, app_dir: Path) -> Path:
"""Expand the notations RetroArch writes into its config values.
fill_pathname_expand_special (libretro-common/file/file_path.c) maps a
leading '~' to the home directory and a leading ':' to the application
directory, dropping the two leading characters in both cases.
"""
if value[:1] == "~":
return Path(str(Path.home())) / value[2:]
if value[:1] == ":":
return app_dir / value[2:]
return Path(os.path.expandvars(os.path.expanduser(value)))
def _parse_retroarch_system_dir(
cfg_path: Path, app_dir: Path | None = None
) -> Path | None:
"""Parse system_directory from retroarch.cfg.
app_dir is RetroArch's application directory, used to expand ':' values.
It defaults to the directory holding the config, which is where a portable
install keeps both.
"""
if not cfg_path.exists():
return None
app_dir = app_dir or cfg_path.parent
try:
for line in cfg_path.read_text(encoding="utf-8", errors="replace").splitlines():
line = line.strip()
if line.startswith("system_directory"):
_, _, value = line.partition("=")
value = value.strip().strip('"').strip("'")
if not value or value == "default":
return app_dir / "system"
return _expand_retroarch_path(value, app_dir)
except OSError:
pass
return None
def _shell_unquote(value: str) -> str:
"""Resolve a shell right-hand side into its effective string value.
Handles concatenated quoted/unquoted segments the way bash does
(e.g. "/run/media/deck/EmuSD"/Emulation), expands variables in
double-quoted and unquoted segments, and stops at unquoted
whitespace or a comment.
"""
parts: list[str] = []
i = 0
n = len(value)
while i < n:
c = value[i]
if c == '"':
end = value.find('"', i + 1)
if end == -1:
parts.append(os.path.expandvars(value[i + 1:]))
break
parts.append(os.path.expandvars(value[i + 1:end]))
i = end + 1
elif c == "'":
end = value.find("'", i + 1)
if end == -1:
parts.append(value[i + 1:])
break
parts.append(value[i + 1:end])
i = end + 1
elif c.isspace() or c == "#":
break
else:
end = i
while end < n and value[end] not in "\"'#" and not value[end].isspace():
end += 1
parts.append(os.path.expandvars(value[i:end]))
i = end
return os.path.expanduser("".join(parts))
def _parse_bash_var(path: Path, key: str) -> str | None:
"""Extract the effective value of key= from a bash/shell file."""
if not path.exists():
return None
try:
for line in path.read_text(encoding="utf-8", errors="replace").splitlines():
line = line.strip()
if line.startswith(f"{key}="):
_, _, value = line.partition("=")
return _shell_unquote(value)
except OSError:
pass
return None
def _parse_json_path(path: Path, *keys: str) -> str | None:
"""Extract a nested string value from a JSON file."""
if not path.exists():
return None
try:
data = json.loads(path.read_text(encoding="utf-8", errors="replace"))
for key in keys:
data = data[key]
return data if isinstance(data, str) and data else None
except (OSError, json.JSONDecodeError, KeyError, TypeError):
return None
def _parse_ps1_var(path: Path, key: str) -> str | None:
"""Extract value of $key= or key= from a PowerShell file."""
if not path.exists():
return None
normalized = key.lstrip("$")
try:
for line in path.read_text(encoding="utf-8", errors="replace").splitlines():
line = line.strip()
check = line.lstrip("$")
if check.startswith(f"{normalized}="):
_, _, value = check.partition("=")
return value.strip('"').strip("'")
except OSError:
pass
return None
def _detect_embedded() -> list[tuple[str, Path]]:
"""Check for embedded Linux retrogaming OSes."""
found: list[tuple[str, Path]] = []
os_release = _parse_os_release()
os_id = os_release.get("ID", "").lower()
if os_id == "rocknix":
found.append(("rocknix", Path("/storage/roms/bios")))
return found
# MiSTer keeps the SD card at /media/fat with the main binary at its root
# (Main_MiSTer file_io.cpp:1150-1156), and cores read from games/
# (file_io.h:168)
if Path("/media/fat/MiSTer").exists():
found.append(("misterfpga", Path("/media/fat/games")))
return found
if Path("/etc/knulli-release").exists():
found.append(("batocera", Path("/userdata/bios")))
return found
if os_id == "lakka":
found.append(("lakka", Path("/storage/system")))
return found
if Path("/etc/batocera-version").exists():
found.append(("batocera", Path("/userdata/bios")))
return found
if (
Path("/recalbox/recalbox.version").exists()
or Path("/usr/bin/recalbox-settings").exists()
):
found.append(("recalbox", Path("/recalbox/share/bios")))
return found
if Path("/opt/muos").exists() or Path("/mnt/mmc/MUOS/").exists():
found.append(("retroarch", Path("/mnt/mmc/MUOS/bios")))
return found
if Path("/home/ark").exists() and Path("/opt/system").exists():
found.append(("retroarch", Path("/roms/bios")))
return found
if Path("/mnt/vendor/bin/dmenu.bin").exists():
found.append(("retroarch", Path("/mnt/mmc/bios")))
return found
return found
def _android_storage_root() -> Path:
"""Shared storage root, the parent RetroArch builds its defaults from."""
value = os.environ.get("EXTERNAL_STORAGE", "").strip()
return Path(value) if value else Path("/storage/emulated/0")
def _detect_android() -> list[tuple[str, Path]]:
"""Find the RetroArch system directory on Android.
The config is read from the app external files directory first, then from
the internal one (platform_unix.c:1327-1395). A system_directory left at
default resolves against <shared storage>/RetroArch, which is also where
the directory sits when no config has been written yet
(platform_unix.c:2836-2851).
"""
found: list[tuple[str, Path]] = []
storage = _android_storage_root()
default_root = storage / "RetroArch"
for package in ANDROID_RETROARCH_PACKAGES:
for cfg in (
storage / "Android" / "data" / package / "files" / "retroarch.cfg",
Path("/data/data") / package / "files" / "retroarch.cfg",
):
system_dir = _parse_retroarch_system_dir(cfg, default_root)
if system_dir:
found.append(("retroarch", system_dir))
return found
if default_root.exists():
found.append(("retroarch", default_root / "system"))
return found
def _lnk_target(lnk_path: Path, exe_name: str) -> Path | None:
"""Read the target path of a Windows shortcut.
Shortcuts embed the target as a plain local path; scanning for it avoids
parsing the whole .lnk structure for the one field that matters here.
"""
try:
if lnk_path.stat().st_size > 64 * 1024:
return None
blob = lnk_path.read_bytes()
except OSError:
return None
pattern = re.compile(
rb"[A-Za-z]:\\[ -~]{0,260}?" + re.escape(exe_name.encode()), re.IGNORECASE
)
match = pattern.search(blob)
if not match:
return None
return Path(match.group().decode("ascii", "replace").replace("\\", "/"))
def launchbox_root(os_type: str) -> Path | None:
"""Locate the LaunchBox installation directory.
LaunchBox installs wherever the user points its installer and writes no
uninstall registry key, so the Start menu shortcut is the only record of
that choice. Everything else falls back to the default location.
"""
if os_type not in ("windows", "wsl"):
return None
candidates: list[Path] = []
appdata = os.environ.get("APPDATA", "")
if appdata:
lnk = (
Path(appdata)
/ "Microsoft"
/ "Windows"
/ "Start Menu"
/ "Programs"
/ "LaunchBox"
/ "LaunchBox.lnk"
)
if lnk.exists():
target = _lnk_target(lnk, "LaunchBox.exe")
if target:
# The shortcut points at Core/LaunchBox.exe
candidates.append(target.parent.parent)
candidates.append(target.parent)
userprofile = os.environ.get("USERPROFILE", "")
if userprofile:
candidates.append(Path(userprofile) / "LaunchBox")
for root in candidates:
if (root / "Data" / "Emulators.xml").exists():
return root
return None
def launchbox_emulators(emulators_xml: Path) -> dict[str, Path]:
"""Map each emulator LaunchBox knows about to its installation directory.
LaunchBox stores its emulator list in Data/Emulators.xml; ApplicationPath
is either absolute or relative to the LaunchBox root directory. Keys are
the lowercase executable names.
"""
found: dict[str, Path] = {}
if not emulators_xml.exists():
return found
try:
if emulators_xml.stat().st_size > 10 * 1024 * 1024:
return found
raw = emulators_xml.read_text(encoding="utf-8", errors="replace")
# ElementTree expands internal entities; LaunchBox writes no doctype
if "<!DOCTYPE" in raw.upper():
return found
tree = ET.ElementTree(ET.fromstring(raw))
except (ET.ParseError, OSError):
return found
lb_root = emulators_xml.parent.parent
for emu in tree.getroot().iter("Emulator"):
app_path = (emu.findtext("ApplicationPath") or "").replace("\\", "/")
if not app_path:
continue
exe = Path(app_path)
if not exe.is_absolute():
exe = lb_root / exe
if exe.parent.is_dir():
found.setdefault(exe.name.lower(), exe.parent)
return found
def _launchbox_retroarch_system_dir(emulators_xml: Path) -> Path | None:
"""Resolve the system dir of a RetroArch referenced by LaunchBox.
Mirrors the plugin's UpdateSystemPath: read system_directory from the
retroarch.cfg sitting next to the executable, else assume system/.
"""
emu_dir = launchbox_emulators(emulators_xml).get("retroarch.exe")
if emu_dir is None:
return None
system_dir = _parse_retroarch_system_dir(emu_dir / "retroarch.cfg", emu_dir)
return system_dir or emu_dir / "system"
def _pcsx2_bios_dir(emu_dir: Path) -> Path:
"""Resolve the BIOS folder of a PCSX2 installed under LaunchBox.
EmuFolders (pcsx2/Pcsx2Config.cpp) enters portable mode when portable.ini
or portable.txt sits next to the executable, and then takes the data root
from the contents of portable.txt, relative to the executable. Otherwise
the data root is Documents/PCSX2. Bios defaults to "bios" below it and the
Bios key of PCSX2.ini overrides that.
"""
documents = Path(os.environ.get("USERPROFILE", str(Path.home()))) / "Documents"
portable_txt = emu_dir / "portable.txt"
portable = (emu_dir / "portable.ini").exists() or portable_txt.exists()
if portable:
subpath = ""
try:
subpath = portable_txt.read_text(
encoding="utf-8", errors="replace"
).strip()
except OSError:
pass
data_root = emu_dir / subpath if subpath else emu_dir
else:
data_root = documents / "PCSX2"
ini = data_root / "inis" / "PCSX2.ini"
try:
for line in ini.read_text(encoding="utf-8", errors="replace").splitlines():
if line.startswith("Bios = "):
value = line[len("Bios = "):].strip()
candidate = Path(value)
return candidate if candidate.is_absolute() else data_root / value
except OSError:
pass
return data_root / "bios"
def _xemu_bios_dir(emu_dir: Path) -> Path:
"""Resolve the BIOS folder of a xemu managed by LaunchBox.
Follows the plugin: xemu.toml next to the executable else the roaming
copy, with bootrom_path and flashrom_path naming a file whose directory
holds the images. Both default to bios/ under the executable.
"""
appdata = os.environ.get("APPDATA", "")
toml = emu_dir / "xemu.toml"
if not toml.exists() and appdata:
toml = Path(appdata) / "xemu" / "xemu" / "xemu.toml"
if toml.exists():
try:
for line in toml.read_text(encoding="utf-8", errors="replace").splitlines():
if line.startswith(("bootrom_path", "flashrom_path")):
parts = line.split("'")
if len(parts) > 1 and Path(parts[1]).exists():
return Path(parts[1]).parent
except OSError:
pass
return emu_dir / "bios"
def launchbox_bios_dirs(root: Path) -> dict[str, Path]:
"""Map emulator ids to the BIOS directory LaunchBox expects for them.
Only emulators whose destination LaunchBox itself computes are listed;
the rest keep the default locations declared in the manifest.
"""
emulators = launchbox_emulators(root / "Data" / "Emulators.xml")
dirs: dict[str, Path] = {}
emu_dir = emulators.get("pcsx2.exe")
if emu_dir:
dirs["pcsx2"] = _pcsx2_bios_dir(emu_dir)
emu_dir = emulators.get("xemu.exe")
if emu_dir:
dirs["xemu"] = _xemu_bios_dir(emu_dir)
emu_dir = emulators.get("dolphin.exe")
if emu_dir and (emu_dir / "portable.txt").exists():
# portable.txt beside the executable moves the user directory to
# User (SetUserDirectory in Source/Core/UICommon/UICommon.cpp,
# PORTABLE_USER_DIR in Source/Core/Common/CommonPaths.h)
dirs["dolphin"] = emu_dir / "User"
return dirs
def detect_frontends(os_type: str, home: Path | None = None) -> list[str]:
"""Detect frontends that reference emulators without owning a BIOS dir.
ES-DE resolves its application data directory to $ESDE_APPDATA_DIR or
<home>/ES-DE (es-core FileSystemUtil.cpp getAppDataDirectory).
"""
home = home or Path.home()
frontends: list[str] = []
esde_env = os.environ.get("ESDE_APPDATA_DIR", "")
if (esde_env and Path(esde_env).is_dir()) or (home / "ES-DE").is_dir():
frontends.append("esde")
if launchbox_root(os_type) is not None:
frontends.append("launchbox")
return frontends
def detect_platforms(os_type: str) -> list[tuple[str, Path]]:
"""Detect installed emulator platforms and their BIOS directories."""
found: list[tuple[str, Path]] = []
if os_type == "android":
found.extend(_detect_android())
if os_type in ("linux", "wsl"):
found.extend(_detect_embedded())
# EmuDeck (Linux/SteamOS)
home = Path.home()
emudeck_settings = home / ".config" / "EmuDeck" / "settings.sh"
if emudeck_settings.exists():
emu_path = _parse_bash_var(emudeck_settings, "emulationPath")
if emu_path:
bios_dir = Path(emu_path) / "bios"
found.append(("emudeck", bios_dir))
# RetroDECK: retrodeck.json since the cfg-to-json migration, which
# renames the old retrodeck.cfg to retrodeck.bak (global.sh:149-153)
retrodeck_conf_dir = home / ".var" / "app" / "net.retrodeck.retrodeck" / "config" / "retrodeck"
retrodeck_json = retrodeck_conf_dir / "retrodeck.json"
retrodeck_cfg = retrodeck_conf_dir / "retrodeck.cfg"
if retrodeck_json.exists():
rd_home = _parse_json_path(retrodeck_json, "paths", "rd_home_path")
found.append(("retrodeck", Path(rd_home) if rd_home else home / "retrodeck"))
elif retrodeck_cfg.exists():
rd_home = _parse_bash_var(retrodeck_cfg, "rdhome")
found.append(("retrodeck", Path(rd_home) if rd_home else home / "retrodeck"))
# RetroArch Flatpak
flatpak_cfg = home / ".var" / "app" / "org.libretro.RetroArch" / "config" / "retroarch" / "retroarch.cfg"
ra_dir = _parse_retroarch_system_dir(flatpak_cfg)
if ra_dir:
found.append(("retroarch", ra_dir))
# RetroArch Snap
snap_cfg = home / "snap" / "retroarch" / "current" / ".config" / "retroarch" / "retroarch.cfg"
ra_dir = _parse_retroarch_system_dir(snap_cfg)
if ra_dir:
found.append(("retroarch", ra_dir))
# RetroArch native
native_cfg = home / ".config" / "retroarch" / "retroarch.cfg"
ra_dir = _parse_retroarch_system_dir(native_cfg)
if ra_dir:
found.append(("retroarch", ra_dir))
if os_type == "darwin":
home = Path.home()
mac_cfg = home / "Library" / "Application Support" / "RetroArch" / "retroarch.cfg"
ra_dir = _parse_retroarch_system_dir(mac_cfg)
if ra_dir:
found.append(("retroarch", ra_dir))
if os_type in ("windows", "wsl"):
# EmuDeck Windows
home = Path.home()
emudeck_ps1 = Path(os.environ.get("APPDATA", "")) / "EmuDeck" / "settings.ps1"
if emudeck_ps1.exists():
emu_path = _parse_ps1_var(emudeck_ps1, "$emulationPath")
if emu_path:
found.append(("emudeck", Path(emu_path) / "bios"))
# RetroArch Windows
appdata = os.environ.get("APPDATA", "")
if appdata:
win_cfg = Path(appdata) / "RetroArch" / "retroarch.cfg"
ra_dir = _parse_retroarch_system_dir(win_cfg)
if ra_dir:
found.append(("retroarch", ra_dir))
# RetroArch from Steam: a portable install under the default library
for var in ("ProgramFiles(x86)", "ProgramFiles"):
program_files = os.environ.get(var, "")
if not program_files:
continue
steam_cfg = (
Path(program_files) / "Steam" / "steamapps" / "common"
/ "RetroArch" / "retroarch.cfg"
)
ra_dir = _parse_retroarch_system_dir(steam_cfg)
if ra_dir and not any(name == "retroarch" for name, _ in found):
found.append(("retroarch", ra_dir))
break
# Portable RetroArch referenced by LaunchBox
lb_root = launchbox_root(os_type)
if lb_root and not any(name == "retroarch" for name, _ in found):
system_dir = _launchbox_retroarch_system_dir(
lb_root / "Data" / "Emulators.xml"
)
if system_dir:
found.append(("retroarch", system_dir))
return found
def normalize_platform(name: str) -> str:
"""Return the canonical platform id for user-supplied input."""
plat = name.strip().lower()
if plat not in AVAILABLE_PLATFORMS:
print(f"Unknown platform '{name}'. Available platforms:", file=sys.stderr)
for p in AVAILABLE_PLATFORMS:
print(f" {p}", file=sys.stderr)
sys.exit(1)
return plat
def _read_limited_json(response, limit: int, label: str) -> object:
"""Read a bounded UTF-8 JSON response."""
raw_length = response.headers.get("Content-Length") if response.headers else None
if raw_length:
try:
if int(raw_length) > limit:
raise ValueError(f"{label} exceeds {limit} bytes")
except ValueError as exc:
if "exceeds" in str(exc):
raise
payload = response.read(limit + 1)
if len(payload) > limit:
raise ValueError(f"{label} exceeds {limit} bytes")
try:
return json.loads(payload.decode("utf-8"))
except (UnicodeDecodeError, json.JSONDecodeError) as exc:
raise ValueError(f"invalid {label}: {exc}") from exc
def _safe_relative_path(value: object, field: str) -> PurePosixPath:
"""Validate a manifest-controlled relative POSIX path."""
if not isinstance(value, str) or not value or len(value) > 1024:
raise ValueError(f"invalid {field}")
if (
"\\" in value
or "\x00" in value
or "//" in value
or value.endswith("/")
or re.match(r"^[A-Za-z]:", value)
):
raise ValueError(f"unsafe {field}: {value!r}")
path = PurePosixPath(value)
if path.is_absolute() or any(part in ("", ".", "..") for part in path.parts):
raise ValueError(f"unsafe {field}: {value!r}")
return path
def _within(candidate: Path, root: Path) -> bool:
"""Whether *candidate* stays under *root* once both are resolved."""
try:
(root / candidate.name).resolve().relative_to(root)
except (OSError, ValueError):
return False
return True
def _safe_target_dir(value: object, field: str) -> str:
"""Validate a manifest-controlled standalone-copy directory.
These name emulator install directories, so unlike a BIOS destination they
are legitimately absolute and outside the BIOS tree. What they must never
do is climb: a '..' anywhere turns "copy next to the emulator" into "write
wherever the manifest likes".
"""
if not isinstance(value, str) or not value or len(value) > 2048:
raise ValueError(f"invalid {field}")
if "\x00" in value:
raise ValueError(f"unsafe {field}: {value!r}")
parts = re.split(r"[\\/]", value)
if any(part == ".." for part in parts):
raise ValueError(f"unsafe {field}: {value!r}")
return value
def _destination_path(root: Path, value: object) -> Path:
"""Place a manifest destination below *root*.
The manifest cannot climb: _safe_relative_path refuses absolute paths,
drive letters and '..'. Symbolic links met under the root are the
user's own layout and are followed. EmuDeck links bios/shadps4/sys_modules
into shadPS4's data directory, and the file has to land where the
emulator reads it.
"""
relative = _safe_relative_path(value, "dest")
return root.resolve() / Path(*relative.parts)
def _check_tags(entry: dict, field: str, where: str) -> None:
"""A selection field is absent or a short list of short strings."""
value = entry.get(field)
if value is None:
return
if (
not isinstance(value, list)
or len(value) > MAX_SELECTION_TAGS
or not all(isinstance(tag, str) and 0 < len(tag) <= 256 for tag in value)
):
raise ValueError(f"invalid {field} for {where}")
def _validate_manifest(data: object, plat: str) -> dict:
"""Validate the untrusted install-manifest boundary using stdlib only."""
if not isinstance(data, dict):
raise ValueError("manifest root must be an object")
if data.get("manifest_version") not in (1, 2):
raise ValueError("unsupported manifest_version")
if data.get("platform") != plat:
raise ValueError("manifest platform does not match request")
files = data.get("files")
if not isinstance(files, list) or len(files) > MAX_MANIFEST_FILES:
raise ValueError("invalid manifest files list")
seen_destinations: set[str] = set()
total_size = 0
for index, entry in enumerate(files):
if not isinstance(entry, dict):
raise ValueError(f"files[{index}] must be an object")
dest = str(_safe_relative_path(entry.get("dest"), f"files[{index}].dest"))
if dest in seen_destinations:
raise ValueError(f"duplicate manifest destination: {dest}")
seen_destinations.add(dest)
size = entry.get("size")
if isinstance(size, bool) or not isinstance(size, int) or not (0 <= size <= MAX_DOWNLOAD_SIZE):
raise ValueError(f"invalid size for {dest}")
total_size += size
if total_size > MAX_TOTAL_DOWNLOAD_SIZE:
raise ValueError("manifest total size exceeds safety limit")
sha1 = entry.get("sha1", "")
sha256 = entry.get("sha256", "")
if sha1 and (not isinstance(sha1, str) or not _SHA1_RE.fullmatch(sha1)):
raise ValueError(f"invalid SHA1 for {dest}")
if sha256 and (
not isinstance(sha256, str) or not _SHA256_RE.fullmatch(sha256)
):
raise ValueError(f"invalid SHA256 for {dest}")
if not sha1 and not sha256:
raise ValueError(f"missing content hash for {dest}")
release_asset = entry.get("release_asset")
repo_path = entry.get("repo_path")
if release_asset:
asset = _safe_relative_path(release_asset, f"files[{index}].release_asset")
if len(asset.parts) != 1:
raise ValueError(f"release asset must be a basename: {release_asset}")
elif repo_path:
source = _safe_relative_path(repo_path, f"files[{index}].repo_path")
if source.parts[0] != "bios":
raise ValueError(f"repo_path outside bios/: {repo_path}")
else:
raise ValueError(f"no download source for {dest}")
cores = entry.get("cores")
if cores is not None and (
not isinstance(cores, list) or not all(isinstance(core, str) for core in cores)
):
raise ValueError(f"invalid cores list for {dest}")
for field in SELECTION_FIELDS:
_check_tags(entry, field, dest)
declared_total_files = data.get("total_files")
if declared_total_files is not None and declared_total_files != len(files):
raise ValueError("manifest total_files does not match files list")
declared_total_size = data.get("total_size")
if declared_total_size is not None and declared_total_size != total_size:
raise ValueError("manifest total_size does not match file sizes")
omitted = data.get("omitted_files", [])
if not isinstance(omitted, list) or len(omitted) > MAX_MANIFEST_FILES:
raise ValueError("invalid omitted_files list")
seen_omitted: set[str] = set()
for index, entry in enumerate(omitted):
if not isinstance(entry, dict):
raise ValueError(f"omitted_files[{index}] must be an object")
dest = str(
_safe_relative_path(
entry.get("dest"), f"omitted_files[{index}].dest"
)
)
if dest in seen_destinations or dest in seen_omitted:
raise ValueError(f"duplicate or conflicting omitted destination: {dest}")
seen_omitted.add(dest)
if not isinstance(entry.get("name"), str) or not entry["name"]:
raise ValueError(f"invalid omitted file name for {dest}")
if not isinstance(entry.get("system", ""), str):
raise ValueError(f"invalid omitted system for {dest}")
if not isinstance(entry.get("required"), bool):
raise ValueError(f"invalid omitted required flag for {dest}")
if entry.get("reason") not in OMISSION_REASONS:
raise ValueError(f"invalid omission reason for {dest}")
cores = entry.get("cores")
if cores is not None and (
not isinstance(cores, list)
or not all(isinstance(core, str) for core in cores)
):
raise ValueError(f"invalid omitted cores list for {dest}")
for field in SELECTION_FIELDS:
_check_tags(entry, field, dest)
declared_total_omitted = data.get("total_omitted")
if (
declared_total_omitted is not None
and declared_total_omitted != len(omitted)
):
raise ValueError("manifest total_omitted does not match omitted_files")
copies = data.get("standalone_copies", [])
if not isinstance(copies, list) or len(copies) > 10_000:
raise ValueError("invalid standalone_copies")
for index, entry in enumerate(copies):
if not isinstance(entry, dict):
raise ValueError(f"standalone_copies[{index}] must be an object")
# The shape do_standalone_copies reads, checked here: a note with a
# list for detect, a copy naming neither file nor pattern, or a list
# for emulator passed and crashed the copy step after the download.
where = f"standalone_copies[{index}]"
if "note" in entry:
detect = entry.get("detect", {})
if not isinstance(entry["note"], str) or not isinstance(detect, dict):
raise ValueError(f"invalid note entry at {where}")
for values in detect.values():
if not isinstance(values, list) or not all(
isinstance(value, str) for value in values
):
raise ValueError(f"invalid note entry at {where}")
continue
if "file" not in entry and "pattern" not in entry:
raise ValueError(f"{where} names neither a file nor a pattern")
if not isinstance(entry.get("emulator", ""), str):
raise ValueError(f"invalid emulator at {where}")
if "file" in entry:
_safe_relative_path(
entry["file"], f"standalone_copies[{index}].file"
)
if "pattern" in entry:
pattern = entry["pattern"]
if (
not isinstance(pattern, str)
or not pattern
or len(pattern) > 256
or "/" in pattern
or "\\" in pattern
or ".." in pattern
):
raise ValueError(f"invalid standalone copy pattern: {pattern!r}")
targets = entry.get("targets", {})
if targets and not isinstance(targets, dict):
raise ValueError(f"invalid standalone copy targets at index {index}")
for values in (targets or {}).values():
if not isinstance(values, list) or len(values) > 100:
raise ValueError(f"invalid standalone copy targets at index {index}")
for value in values:
_safe_target_dir(value, f"standalone_copies[{index}].targets")
return data
def _validate_targets(data: object) -> dict[str, dict]:
"""Validate and normalize legacy list-valued target manifests.
A null core list means the target publishes no core inventory. That is a
known target with no filter, not a broken manifest: rejecting it would
discard every other target on the platform.
"""
if not isinstance(data, dict) or len(data) > 10_000:
raise ValueError("invalid targets manifest")
normalized: dict[str, dict] = {}
for target, value in data.items():
if not isinstance(target, str) or not target or len(target) > 128:
raise ValueError("invalid target name")
if isinstance(value, dict):
cores = value.get("cores")
else:
cores = value
if cores is None:
normalized[target] = {"cores": None}
continue
if not isinstance(cores, list) or len(cores) > 10_000 or not all(
isinstance(core, str) and 0 < len(core) <= 256 for core in cores
):
raise ValueError(f"invalid core list for target {target}")
normalized[target] = {"cores": cores}
return normalized
def fetch_manifest(plat: str) -> dict:
"""Download platform manifest JSON."""
url = MANIFEST_URL.format(platform=plat)
try:
with urllib.request.urlopen(url, timeout=30) as resp:
return _validate_manifest(
_read_limited_json(resp, MAX_MANIFEST_BYTES, "manifest"), plat
)
except (urllib.error.URLError, urllib.error.HTTPError, OSError, ValueError) as exc:
print(f" Failed to fetch manifest for {plat}: {exc}", file=sys.stderr)
sys.exit(1)
def fetch_targets(plat: str) -> "dict | None":
"""Download the target core list.
Empty on 404, where the platform publishes no targets. None when the list
could not be read: a timeout says nothing about the platform, and
answering "no targets" for it sent users to the full pack.
"""
url = TARGETS_URL.format(platform=plat)
try:
with urllib.request.urlopen(url, timeout=30) as resp:
return _validate_targets(
_read_limited_json(resp, MAX_TARGETS_BYTES, "targets manifest")
)
except urllib.error.HTTPError as exc:
if exc.code == 404:
return {}
print(f" Error: failed to fetch targets for {plat}: {exc}", file=sys.stderr)
return None
except (urllib.error.URLError, OSError, ValueError) as exc:
print(f" Error: failed to fetch targets for {plat}: {exc}", file=sys.stderr)
return None
def _filter_by_target(
files: list[dict], target_cores: list[str]
) -> list[dict]:
"""Keep files where cores is None or overlaps with target_cores."""
result: list[dict] = []
target_set = set(target_cores)
for f in files:
cores = f.get("cores")
if cores is None or any(c in target_set for c in cores):
result.append(f)
return result
def canonical_region(raw: str) -> str:
"""One region name or alias, spelled the way manifests spell it."""
key = raw.strip().lower()
key = REGION_ALIASES.get(key, key)
if key not in REGIONS:
raise ValueError(f"unknown region: {raw.strip()}")
return key
def _region_comparable(a: str, b: str) -> bool:
return a == b or _REGION_PARENT.get(a) == b or _REGION_PARENT.get(b) == a
def _region_rank(file_regions: set[str], requested: list[str]) -> int:
for index, wanted in enumerate(requested):
if any(_region_comparable(wanted, region) for region in file_regions):
return index
return len(requested)
def region_drops(
entries: list[dict], requested: list[str], systems: list[str] | None = None
) -> set[str]:
"""Destinations a region priority list withdraws.
The pack builder's rule (region.resolve_region_drops), replayed over the
groups the manifest records. Per group, the best-ranked regional file
stays; a world file beats unmatched regional ones the code does not search
before it (`priority`, lower first); when nothing matches and no world
file exists, every regional file stays, so a group is never emptied. A file with no region never competes and always stays.
A core extra competes whatever systems are chosen; a platform's own
declaration competes only when its system is kept, as in the pack.
"""
if not requested:
return set()
kept_systems = set(systems) if systems else None
groups: dict[str, list[tuple[set, str, object]]] = {}
for entry in entries:
regions = entry.get("regions")
if not regions:
continue
joined = list(entry.get("region_groups") or [])
joined += [
group for group in entry.get("region_system_groups") or []
if kept_systems is None or group in kept_systems
]
priority = entry.get("priority")
if not isinstance(priority, int) or isinstance(priority, bool):
priority = None
for group in joined:
groups.setdefault(group, []).append((set(regions), entry["dest"], priority))
keep: set[str] = set()
drop: set[str] = set()
for members in groups.values():
world = {dest for regions, dest, _p in members if WORLD_REGION in regions}
ranks = {dest: priority for _regions, dest, priority in members}
regional = [
(_region_rank(regions, requested), dest)
for regions, dest, _p in members
if WORLD_REGION not in regions
]
keep |= world
if not regional:
continue
matched = [(rank, dest) for rank, dest in regional if rank < len(requested)]
if matched:
best = min(rank for rank, _dest in matched)
keep |= {dest for rank, dest in matched if rank == best}
drop |= {dest for _rank, dest in regional}
elif world:
world_ranks = [ranks[dest] for dest in world]
best_world = min(world_ranks) if None not in world_ranks else None
for _rank, dest in regional:
ours = ranks[dest]
if best_world is not None and ours is not None and ours < best_world:
keep.add(dest)
else:
drop.add(dest)
else:
keep |= {dest for _rank, dest in regional}
return drop - keep
def entry_systems(entry: dict) -> list[str]:
"""The platform systems a file or an omission belongs to.
An omission's `system` is spelled the way its profile spells it; only
`systems` names the platform's own, the ones --system accepts.
"""
return list(entry.get("systems") or [])
def narrow(
files: list[dict],
omitted: list[dict],
systems: list[str],
cores: list[str],
regions: list[str],
) -> tuple[list[dict], list[dict]]:
"""Keep what the chosen systems, cores and regions need.
A core choice keeps the platform's own list and narrows only the files
cores load, the rule --target already follows. Regions are then decided
the way the pack builder decides them under --system: over the chosen
systems' declarations and every core extra. The system filter comes last.
"""
if cores:
files = _filter_by_target(files, cores)
omitted = _filter_by_target(omitted, cores)
if regions:
dropped = region_drops(files + omitted, regions, systems)
files = [f for f in files if f["dest"] not in dropped]
omitted = [o for o in omitted if o["dest"] not in dropped]
if systems:
wanted = set(systems)
files = [f for f in files if _chosen(f, wanted)]
omitted = [o for o in omitted if _chosen(o, wanted)]
return files, omitted
def _chosen(entry: dict, wanted: set[str]) -> bool:
systems = entry_systems(entry)
if not systems:
return OTHER_SYSTEMS in wanted
return bool(wanted & set(systems))
def records_selection(files: list[dict]) -> bool:
"""Whether a manifest carries what narrowing reads; older ones do not."""
return any("systems" in entry for entry in files)
def available_choices(files: list[dict], omitted: list[dict]) -> dict[str, list[str]]:
"""The systems, cores and regions a manifest lets one narrow to."""
systems = {s for e in files + omitted for s in entry_systems(e)}
ordered = sorted(systems)
if any(not entry_systems(f) for f in files):
ordered.append(OTHER_SYSTEMS)
cores = {c for f in files for c in (f.get("cores") or [])}
regions = {
r for e in files + omitted for r in (e.get("regions") or [])
if r != WORLD_REGION
}
return {
"systems": ordered,
"cores": sorted(cores),
"regions": sorted(regions),
}
def resolve_choices(values: list[str], known: list[str], label: str) -> list[str]:
"""Names given on the command line, matched without regard to case.
Each value is one name or a comma-separated list. A whole value that is
itself a name wins over splitting it: Recalbox names one system
`msx1,msx2,msxturbor`. An unknown name is refused, never ignored, since
carrying on would install something other than what was asked.
"""
by_lower = {name.lower(): name for name in known}
chosen: list[str] = []
unknown: list[str] = []
for value in values:
value = value.strip()
tokens = [value] if value.lower() in by_lower else value.split(",")
for token in (t.strip() for t in tokens):
if not token:
continue
name = by_lower.get(token.lower())
if name is None:
unknown.append(token)
elif name not in chosen:
chosen.append(name)
if unknown:
available = ", ".join(known) if known else "none"
raise ValueError(
f"unknown {label}: {', '.join(unknown)} (available: {available})"
)
if values and not chosen:
raise ValueError(f"--{label} needs at least one name")
return chosen
def resolve_regions(values: list[str]) -> list[str]:
"""A region priority list, best first, from names or aliases."""
chosen: list[str] = []
for value in values:
for token in value.split(","):
if token.strip():
region = canonical_region(token)
if region not in chosen:
chosen.append(region)
if values and not chosen:
raise ValueError("--region needs at least one region")
return chosen
def parse_selection(text: str, count: int, ordered: bool = False) -> list[int]:
"""Numbers typed at a prompt: `3`, `1,4`, `2-5`, or `all`, 1-based.
Returned sorted, or in the typed order when the order is a priority.
"""
if text.strip().lower() in ("all", "*"):
return list(range(1, count + 1))
chosen: list[int] = []
for token in (t.strip() for t in text.split(",")):
if not token:
continue
low, sep, high = token.partition("-")
try:
start = int(low)
end = int(high) if sep else start
except ValueError:
raise ValueError(f"not a number or range: {token}") from None
if start > end or start < 1 or end > count:
raise ValueError(f"out of range: {token} (1-{count})")
for number in range(start, end + 1):
if number not in chosen:
chosen.append(number)
return chosen if ordered else sorted(chosen)
def _digest_file(path: Path, algorithms: tuple[str, ...]) -> dict[str, str]:
"""Compute several digests of a file in a single read.
Checking a 3 GB collection against both a SHA-1 and a SHA-256 used to walk
every file twice, single threaded. Hashing is cheap next to the I/O, so
the read is what has to happen once.
"""
hashers = {name: hashlib.new(name) for name in algorithms}
with open(path, "rb") as fh:
for chunk in iter(lambda: fh.read(1024 * 1024), b""):
for hasher in hashers.values():
hasher.update(chunk)
return {name: hasher.hexdigest() for name, hasher in hashers.items()}
def _sha1_file(path: Path) -> str:
"""Compute SHA1 of a file."""
return _digest_file(path, ("sha1",))["sha1"]
def _sha256_file(path: Path) -> str:
"""Compute SHA256 of a file."""
return _digest_file(path, ("sha256",))["sha256"]
def _classify_local(entry: dict, dest: Path) -> str:
"""Return 'up_to_date' or 'mismatched' for a file already on disk."""
expected = {
name: entry.get(name, "").lower()
for name in ("sha256", "sha1")
if entry.get(name)
}
if not expected:
return "up_to_date"
actual = _digest_file(dest, tuple(expected))
if all(actual[name] == value for name, value in expected.items()):
return "up_to_date"
return "mismatched"
def check_local(
files: list[dict], bios_path: Path, jobs: int = 8
) -> tuple[list[dict], list[dict], list[dict]]:
"""Check which files exist locally and have correct hashes.
Returns (to_download, up_to_date, mismatched).
"""
to_download: list[dict] = []
present: list[tuple[dict, Path]] = []
for f in files:
dest = _destination_path(bios_path, f["dest"])
if dest.exists():
present.append((f, dest))
else:
to_download.append(f)
up_to_date: list[dict] = []
mismatched: list[dict] = []
if present:
with concurrent.futures.ThreadPoolExecutor(max_workers=jobs) as pool:
verdicts = pool.map(
lambda item: (item[0], _classify_local(item[0], item[1])), present
)
for entry, verdict in verdicts:
if verdict == "up_to_date":
up_to_date.append(entry)
else:
mismatched.append(entry)
return to_download, up_to_date, mismatched
def _download_one(
f: dict, bios_path: Path, verbose: bool = False
) -> tuple[str, bool]:
"""Download a single file. Returns (dest, success)."""
if _stop.is_set():
return str(f.get("dest", "?")), False
try:
dest = _destination_path(bios_path, f["dest"])
except ValueError:
return str(f.get("dest", "?")), False
dest.parent.mkdir(parents=True, exist_ok=True)
if f.get("release_asset"):
url = RELEASE_URL.format(asset=urllib.parse.quote(f["release_asset"], safe=""))
else:
url = RAW_FILE_URL.format(path=urllib.parse.quote(f["repo_path"], safe="/"))
for attempt in range(1, MAX_RETRIES + 1):
tmp_path: Path | None = None
try:
with urllib.request.urlopen(url, timeout=60) as resp:
expected_size = f["size"]
raw_length = resp.headers.get("Content-Length") if resp.headers else None
if raw_length and int(raw_length) != expected_size:
raise ValueError(
f"Content-Length {raw_length} != expected {expected_size}"
)
with tempfile.NamedTemporaryFile(
mode="wb",
dir=dest.parent,
prefix=f".{dest.name}.",
suffix=".part",
delete=False,
) as out:
tmp_path = Path(out.name)
downloaded = 0
while True:
if _stop.is_set():
raise _Interrupted()
chunk = resp.read(1024 * 1024)
if not chunk:
break
downloaded += len(chunk)
if downloaded > expected_size or downloaded > MAX_DOWNLOAD_SIZE:
raise ValueError("download exceeded declared size")
out.write(chunk)
if downloaded != expected_size:
raise ValueError(
f"downloaded {downloaded} bytes; expected {expected_size}"
)
expected_sha256 = f.get("sha256", "")
expected_sha1 = f.get("sha1", "")
if expected_sha256 and _sha256_file(tmp_path) != expected_sha256.lower():
if verbose:
print(f" SHA256 mismatch on attempt {attempt}", file=sys.stderr)
tmp_path.unlink(missing_ok=True)
continue
if expected_sha1:
actual = _sha1_file(tmp_path)
if actual != expected_sha1.lower():
if verbose:
print(f" SHA1 mismatch on attempt {attempt}", file=sys.stderr)
tmp_path.unlink(missing_ok=True)
continue
os.replace(tmp_path, dest)
return f["dest"], True
except _Interrupted:
# Stopped mid-transfer: drop the scratch file rather than leaving
# it beside the destination, and do not retry.
if tmp_path is not None:
tmp_path.unlink(missing_ok=True)
return f["dest"], False
except (
urllib.error.URLError,
urllib.error.HTTPError,
OSError,
ValueError,
) as exc:
if verbose:
print(f" Attempt {attempt} failed: {exc}", file=sys.stderr)
if tmp_path is not None:
tmp_path.unlink(missing_ok=True)
return f["dest"], False
def _locations(paths: list[Path]) -> str:
"""Where the run wrote, naming every destination when it served several."""
return ", ".join(str(p) for p in paths) if paths else "nowhere"
def _plural(count: int, word: str) -> str:
"""Count and noun, so a run of one file does not read as machine output."""
return f"{count} {word}" if count == 1 else f"{count} {word}s"
def _report_outcome(
paths: list[Path],
downloaded: int,
up_to_date: int,
errors: int,
omitted: int,
interrupted: bool,
) -> None:
"""Close the run on what happened, and leave with a status that says so.
A run that installed nothing used to end on the word Done and a zero exit,
which reads as success to a person and to a script alike.
"""
if interrupted:
print(f"\nStopped. {_plural(downloaded, 'file')} installed.")
print("Run the same command again to carry on from here. Nothing is lost.")
sys.exit(130)
if errors and not downloaded:
print(f"\nNothing was installed. {_plural(errors, 'file')} failed.")
print(f"Location that was being written to: {_locations(paths)}")
print("Run the same command again once that is resolved. "
"Files already installed are kept.")
sys.exit(2)
if errors:
print(f"\nFinished with errors. {_plural(downloaded, 'file')} installed, "
f"{_plural(errors, 'file')} failed.")
print(f"Location: {_locations(paths)}")
print("Run the same command again to retry only those.")
sys.exit(1)
print(f"\nDone. {_plural(downloaded, 'file')} installed, "
f"{_plural(up_to_date, 'file')} already up to date.")
print(f"Location: {_locations(paths)}")
if omitted:
print(f"{_plural(omitted, 'file')} are not in the collection yet "
"and were skipped.")
print("To check this later, run the same command with --check.")
def _check_free_space(bios_path: Path, needed: int) -> None:
"""Stop before writing when the volume cannot hold the download.
Reported against the nearest existing ancestor, since the destination
itself may have just been created on a volume that is already full.
"""
probe = bios_path
while not probe.exists() and probe != probe.parent:
probe = probe.parent
try:
free = shutil.disk_usage(probe).free
except OSError:
return
if free >= needed:
return
print(f"\nNot enough space on the drive holding {bios_path}.", file=sys.stderr)
print(f" Needed: {format_size(needed)}. Free: {format_size(free)}.",
file=sys.stderr)
print(f" Free up {format_size(needed - free)}, or use --dest to install "
"to another drive.", file=sys.stderr)
sys.exit(1)
def download_files(
files: list[dict], bios_path: Path, jobs: int = 8, verbose: bool = False
) -> tuple[list[str], bool]:
"""Download files in parallel.
Returns the destinations that failed and whether the run was interrupted.
Every future is submitted up front, so a bare KeyboardInterrupt would let
the pool drain the queue in full: the stop flag is what makes Ctrl+C end
the run rather than only stop the reporting.
"""
failed: list[str] = []
total = len(files)
interrupted = False
_stop.clear()
with concurrent.futures.ThreadPoolExecutor(max_workers=jobs) as pool:
future_map = {
pool.submit(_download_one, f, bios_path, verbose): f
for f in files
}
done_count = 0
try:
for future in concurrent.futures.as_completed(future_map):
done_count += 1
dest, success = future.result()
status = "ok" if success else "FAILED"
print(f" [{done_count}/{total}] {dest} {status}")
if not success:
failed.append(dest)
except KeyboardInterrupt:
interrupted = True
_stop.set()
print("\n Stopping. Files already installed are kept.")
return failed, interrupted
def do_standalone_copies(
manifest: dict, bios_path: Path, os_type: str,
extra_dirs: dict[str, Path] | None = None,
) -> tuple[int, int, int]:
"""Copy BIOS files to standalone emulator directories.
Supports:
- file: single file copy
- pattern: glob match (e.g. "scph*.bin")
- note: informational message when detect path exists
- WSL fallback to linux targets
extra_dirs maps the emulator id of an entry to a root that replaces the
default per-OS locations, for setups such as LaunchBox that keep their
emulators outside them. The layout below the root is the same.
Returns (copied, skipped, failed): skipped is a target directory that
does not exist, failed a copy the system refused (permission, disk
full, read-only mount), which used to be counted as "dir not found".
"""
from fnmatch import fnmatch
copies = manifest.get("standalone_copies", [])
if not copies:
return 0, 0, 0
copied = 0
skipped = 0
failed = 0
for entry in copies:
# Note entries: print message if emulator detected
if "note" in entry:
detect_paths = entry.get("detect", {}).get(os_type, [])
if not detect_paths and os_type == "wsl":
detect_paths = entry.get("detect", {}).get("linux", [])
for dp in detect_paths:
expanded = Path(os.path.expandvars(os.path.expanduser(dp)))
if expanded.is_dir():
print(f" {entry['note']}")
break
continue
# Resolve source files
if "pattern" in entry:
sources = [
f for f in bios_path.rglob("*")
if fnmatch(f.name, entry["pattern"]) and f.is_file()
]
else:
src = bios_path / entry["file"]
sources = [src] if src.exists() else []
if not sources:
continue
# Resolve target directories with WSL fallback
targets = entry.get("targets", {}).get(os_type, [])
if not targets and os_type == "wsl":
targets = entry.get("targets", {}).get("linux", [])
target_dirs = [
Path(os.path.expandvars(os.path.expanduser(t))) for t in targets
]
extra_root = (extra_dirs or {}).get(entry.get("emulator", ""))
if extra_root is not None:
subdir = PurePosixPath(entry.get("file", "")).parent
extra = extra_root if str(subdir) == "." else extra_root / subdir
if extra not in target_dirs:
target_dirs.append(extra)
for target_dir in target_dirs:
if not target_dir.is_dir():
skipped += len(sources)
continue
resolved_dir = target_dir.resolve()
for src in sources:
dest = target_dir / src.name
# A symlink already sitting at the destination would redirect
# the write outside the directory the user opted into, and a
# crafted source name would climb out of it.
if dest.is_symlink() or not _within(dest, resolved_dir):
skipped += 1
continue
try:
shutil.copy2(src, dest)
copied += 1
except OSError as exc:
print(f" ERROR: {dest}: {exc.strerror or exc}")
failed += 1
return copied, skipped, failed
def format_size(n: int) -> str:
"""Human-readable file size."""
if n < 1024:
return f"{n} B"
if n < 1024 * 1024:
return f"{n / 1024:.1f} KB"
if n < 1024 * 1024 * 1024:
return f"{n / (1024 * 1024):.1f} MB"
return f"{n / (1024 * 1024 * 1024):.1f} GB"
def _prompt_manual_platform(os_type: str) -> list[tuple[str, Path]]:
"""Ask for a platform and a BIOS directory when nothing was detected."""
print("\nChoose the platform to install for:")
for i, name in enumerate(AVAILABLE_PLATFORMS, 1):
print(f" {i}) {name}")
print(" q) quit")
print()
plat = ""
while not plat:
try:
choice = input("> ").strip().lower()
except (EOFError, KeyboardInterrupt):
print()
sys.exit(0)
if choice == "q":
sys.exit(0)
if choice.isdigit() and 1 <= int(choice) <= len(AVAILABLE_PLATFORMS):
plat = AVAILABLE_PLATFORMS[int(choice) - 1]
elif choice in AVAILABLE_PLATFORMS:
plat = choice
default_dest = _default_dest(os_type, plat)
try:
raw = input(f"BIOS directory [{default_dest}]: ").strip()
except (EOFError, KeyboardInterrupt):
print()
sys.exit(0)
dest = Path(raw).expanduser() if raw else default_dest
if os_type == "windows" and raw:
dest = Path(os.path.expandvars(raw))
return [(plat, dest)]
def _default_dest(os_type: str, platform_name: str) -> Path:
"""Fall back path for a platform named on the command line but not found.
A home directory is meaningless for RetroArch on Android, where Termux
owns the home and the frontend reads from shared storage.
"""
if os_type == "android" and platform_name == "retroarch":
return _android_storage_root() / "RetroArch" / "system"
return _os_default_dests(os_type).get(
platform_name, DEFAULT_DESTS.get(platform_name, Path.home() / "bios")
)
def _manual_usage_hint(os_type: str) -> list[str]:
"""How to name the platform and directory when nothing is detected."""
if os_type == "windows":
return [
" Use --platform and --dest to specify manually, for example:",
" & ([ScriptBlock]::Create((iwr -useb "
"https://raw.githubusercontent.com/Abdess/retrobios/main/install.ps1))) "
"--platform retroarch --dest 'C:\\RetroArch\\system'",
]
if os_type == "android":
return [
" Use --platform and --dest to specify manually, for example:",
" python3 install.py --platform retroarch "
"--dest /storage/emulated/0/RetroArch/system",
]
return [
" Use --platform and --dest to specify manually, for example:",
" python3 install.py --platform retroarch --dest ~/.config/retroarch/system",
]
def _prompt_platform_choice(
platforms: list[tuple[str, Path]],
) -> list[tuple[str, Path]]:
"""Prompt user to choose among detected platforms."""
print("\nInstall for:")
for i, (name, path) in enumerate(platforms, 1):
print(f" {i}) {name.capitalize()} ({path})")
if len(platforms) > 1:
print(f" {len(platforms) + 1}) All")
print()
while True:
try:
choice = input("> ").strip()
except (EOFError, KeyboardInterrupt):
print()
sys.exit(0)
if not choice:
continue
try:
idx = int(choice)
except ValueError:
continue
if 1 <= idx <= len(platforms):
return [platforms[idx - 1]]
if idx == len(platforms) + 1 and len(platforms) > 1:
return platforms
def _ask(prompt: str) -> str:
try:
return input(prompt).strip()
except (EOFError, KeyboardInterrupt):
print()
sys.exit(0)
def _sized(files: list[dict]) -> str:
return f"{_plural(len(files), 'file')}, {format_size(sum(f.get('size', 0) for f in files))}"
def _pick(title: str, rows: list[tuple[str, list[dict]]], ordered: bool = False) -> list[str]:
"""Numbered list of names; an empty answer keeps every one."""
if not rows:
return []
print(f"\n{title}")
width = len(str(len(rows)))
name_width = min(max(len(name) for name, _files in rows), 32)
for index, (name, files) in enumerate(rows, 1):
print(f" {index:>{width}}) {name:<{name_width}} {_sized(files)}")
hint = "order of preference, e.g. 2,1" if ordered else "e.g. 1,3-5"
while True:
text = _ask(f"Numbers ({hint}), Enter for all, q to quit: ")
if not text:
return []
if text.lower() == "q":
sys.exit(0)
try:
chosen = parse_selection(text, len(rows), ordered=ordered)
except ValueError as exc:
print(f" {exc}")
continue
if chosen:
return [rows[number - 1][0] for number in chosen]
def _prompt_custom_selection(
plat: str, files: list[dict], omitted: list[dict]
) -> tuple[list[str], list[str], list[str]] | None:
"""Offer to narrow the install by system, core and region.
Enter installs everything, the default most runs want. Returns None
then, or the three choices; each empty list keeps everything on its axis.
"""
print(f"\n{plat.capitalize()}: {_sized(files)}.")
print(" Enter) install everything")
print(" c) choose systems, cores and regions")
print(" q) quit")
while True:
answer = _ask("> ").lower()
if answer == "":
return None
if answer == "q":
sys.exit(0)
if answer == "c":
break
choices = available_choices(files, omitted)
if OTHER_SYSTEMS in choices["systems"]:
print(f"\n'{OTHER_SYSTEMS}' holds what cores load for systems this platform")
print("does not list, such as game engines and extra computers.")
systems = _pick("Systems:", [
(name, [f for f in files if _chosen(f, {name})])
for name in choices["systems"]
])
kept, kept_omitted = narrow(files, omitted, systems, [], [])
cores: list[str] = []
core_names = available_choices(kept, kept_omitted)["cores"]
if core_names:
print("\nThe platform's own files for these systems are always kept;")
print("the extra files each core loads can be narrowed.")
cores = _pick("Cores:", [
(name, [f for f in kept if name in (f.get("cores") or [])])
for name in core_names
])
kept, kept_omitted = narrow(kept, kept_omitted, [], cores, [])
regions: list[str] = []
region_names = available_choices(kept, kept_omitted)["regions"]
if region_names:
print("\nSome systems have one BIOS per region. Keeping only the")
print("preferred region removes the others where a match exists.")
regions = _pick("Regions:", [
(name, [f for f in kept if name in (f.get("regions") or [])])
for name in region_names
], ordered=True)
return systems, cores, regions
def _same_choice_hint(plat: str, systems: list[str], cores: list[str], regions: list[str]) -> str:
"""The options that repeat a choice made at the prompts."""
def quoted(value: str) -> str:
return f'"{value}"' if any(c in value for c in " ;&|") else value
parts = [f"--platform {plat}"]
parts += [f"--system {quoted(name)}" for name in systems]
parts += [f"--core {quoted(name)}" for name in cores]
if regions:
parts.append(f"--region {','.join(regions)}")
return " ".join(parts)
def _print_choices(
plat: str, files: list[dict], omitted: list[dict], systems: bool
) -> None:
"""What --system or --core accepts for a platform, with what each holds."""
choices = available_choices(files, omitted)
names = choices["systems"] if systems else choices["cores"]
if not names:
print(f" {plat} records no {'systems' if systems else 'cores'}")
return
width = min(max(len(name) for name in names), 40)
for name in names:
held = (
[f for f in files if _chosen(f, {name})]
if systems
else [f for f in files if name in (f.get("cores") or [])]
)
print(f" {name:<{width}} {_sized(held)}")
def _requested_narrowing(
args: argparse.Namespace, plat: str, files: list[dict], omitted: list[dict]
) -> tuple[list[str], list[str], list[str]] | None:
"""The systems, cores and regions named on the command line, checked."""
if not (args.system or args.core or args.region):
return None
if not records_selection(files):
# Without the fields, a region filter would match nothing and install
# everything: refused rather than ignored.
print(
f"Error: the {plat} file list records no systems or regions; "
"narrowing needs a newer one",
file=sys.stderr,
)
sys.exit(1)
choices = available_choices(files, omitted)
try:
systems = resolve_choices(args.system or [], choices["systems"], "system")
cores = resolve_choices(args.core or [], choices["cores"], "core")
regions = resolve_regions(args.region or [])
except ValueError as exc:
print(f"Error: {plat}: {exc}", file=sys.stderr)
sys.exit(1)
return systems, cores, regions
def main() -> None:
"""Entry point."""
parser = argparse.ArgumentParser(
description="Download missing BIOS files for retrogaming emulators.",
)
parser.add_argument(
"--platform",
help="target platform (retroarch, batocera, emudeck, ...)",
)
parser.add_argument(
"--dest",
type=Path,
help="override BIOS destination directory",
)
parser.add_argument(
"--target",
help="hardware target for core filtering (switch, rpi4, ...)",
)
parser.add_argument(
"--check",
action="store_true",
help="check existing files without downloading",
)
parser.add_argument(
"--list-platforms",
action="store_true",
help="list detected platforms and exit",
)
parser.add_argument(
"--list-targets",
action="store_true",
help="list available targets for a platform and exit",
)
parser.add_argument(
"--jobs", "-j",
type=int,
default=8,
help="parallel download threads (default: 8)",
)
parser.add_argument(
"--verbose", "-v",
action="store_true",
help="verbose output",
)
parser.add_argument(
"--standalone-copies",
action="store_true",
help="opt in to copies into detected standalone-emulator directories",
)
parser.add_argument(
"--system",
action="append",
metavar="NAME",
help="install only these systems (repeat or comma-separate; see --list-systems)",
)
parser.add_argument(
"--core",
action="append",
metavar="NAME",
help="keep only the extra files these cores load (see --list-cores)",
)
parser.add_argument(
"--region",
action="append",
metavar="LIST",
help="preferred regions, best first, e.g. us,eu,jp: keep one BIOS per "
"region where a system has several",
)
parser.add_argument(
"--list-systems",
action="store_true",
help="list a platform's systems with their file counts and exit",
)
parser.add_argument(
"--list-cores",
action="store_true",
help="list the cores whose extra files a platform installs and exit",
)
parser.add_argument(
"--no-input",
action="store_true",
help="never ask a question; fail where an answer is needed",
)
args = parser.parse_args()
if not 1 <= args.jobs <= 32:
parser.error("--jobs must be between 1 and 32")
if args.region:
try:
resolve_regions(args.region)
except ValueError as exc:
parser.error(f"{exc} (known: {', '.join(sorted(REGIONS))})")
for option, values in (("--system", args.system), ("--core", args.core)):
if values is not None and not any(
token.strip() for value in values for token in value.split(",")
):
parser.error(f"{option} needs at least one name")
interactive = sys.stdin.isatty() and not args.no_input
listing = args.list_targets or args.list_systems or args.list_cores
print("RetroBIOS\n")
os_type = detect_os()
# Early exit for listing
if args.list_platforms:
print("Available platforms:")
for p in AVAILABLE_PLATFORMS:
print(f" {p}")
detected = detect_platforms(os_type)
if detected:
print("\nDetected on this system:")
for name, path in detected:
print(f" {name}: {path}")
return
# Platform detection or override
if args.platform:
args.platform = normalize_platform(args.platform)
if args.platform and args.dest:
platforms = [(args.platform, args.dest)]
elif args.platform:
print("Detecting platform...")
detected = detect_platforms(os_type)
matched = [(n, p) for n, p in detected if n == args.platform]
if matched:
platforms = matched
else:
default_dest = _default_dest(os_type, args.platform)
print(
f" Platform '{args.platform}' not detected, "
f"using default path: {default_dest}"
)
platforms = [(args.platform, default_dest)]
elif args.dest:
print(f" Using destination: {args.dest}")
platforms = [("retroarch", args.dest)]
else:
print("Detecting platform...")
frontends = detect_frontends(os_type)
if "esde" in frontends:
print(" Found ES-DE (BIOS files go to each emulator, not to ES-DE itself)")
if "launchbox" in frontends:
print(" Found LaunchBox")
platforms = detect_platforms(os_type)
if not platforms:
print(" No supported platform detected.")
if not interactive or listing:
for line in _manual_usage_hint(os_type):
print(line)
sys.exit(1)
platforms = _prompt_manual_platform(os_type)
for name, path in platforms:
print(f" Found {name.capitalize()} at {path}")
if len(platforms) > 1 and not listing and interactive:
platforms = _prompt_platform_choice(platforms)
total_downloaded = 0
total_up_to_date = 0
total_errors = 0
total_omitted = 0
total_interrupted = False
installed_paths: list[Path] = []
for plat_name, bios_path in platforms:
if bios_path not in installed_paths:
installed_paths.append(bios_path)
print(f"\nFetching file index for {plat_name}...")
manifest = fetch_manifest(plat_name)
files = manifest.get("files", [])
omitted_files = manifest.get("omitted_files", [])
if args.list_targets:
targets = fetch_targets(plat_name)
if targets is None:
sys.exit(1)
if not targets:
print(f" No targets available for {plat_name}")
else:
for t in sorted(targets.keys()):
cores = targets[t].get("cores")
label = "no core list" if cores is None else f"{len(cores)} cores"
print(f" {t} ({label})")
continue
# Target filtering
if args.target:
targets = fetch_targets(plat_name)
if targets is None:
sys.exit(1)
target_info = targets.get(args.target)
if target_info is None:
# Carrying on would install every file, which is the opposite
# of what a filter was asked for: a 4 GB download instead of
# the subset the target needs. Refuse rather than ignore.
print(
f"Error: unknown target '{args.target}' for {plat_name}.",
file=sys.stderr,
)
if targets:
print(
f" available: {', '.join(sorted(targets))}", file=sys.stderr
)
else:
print(
f" {plat_name} publishes no target list", file=sys.stderr
)
sys.exit(1)
elif target_info.get("cores") is None:
print(
f" Target '{args.target}' publishes no core list; "
"installing every file"
)
else:
target_cores = target_info["cores"]
before = len(files)
files = _filter_by_target(files, target_cores)
omitted_files = _filter_by_target(omitted_files, target_cores)
print(f" Filtered {before} -> {len(files)} files for target {args.target}")
selection = _requested_narrowing(args, plat_name, files, omitted_files)
if (
selection is None
and interactive
and not args.check
and not listing
and records_selection(files)
):
selection = _prompt_custom_selection(plat_name, files, omitted_files)
if selection is not None:
print("\n Same choice without questions: "
+ _same_choice_hint(plat_name, *selection))
if selection is not None:
before = len(files)
files, omitted_files = narrow(files, omitted_files, *selection)
print(f" Narrowed {before} -> {len(files)} files")
if args.list_systems or args.list_cores:
_print_choices(plat_name, files, omitted_files, args.list_systems)
continue
total_size = sum(f.get("size", 0) for f in files)
print(f" {len(files)} files ({format_size(total_size)})")
if omitted_files:
required_omitted = sum(
1 for entry in omitted_files if entry.get("required", True)
)
reasons: dict[str, int] = {}
for entry in omitted_files:
reason = entry.get("reason", "unknown")
reasons[reason] = reasons.get(reason, 0) + 1
reason_summary = ", ".join(
f"{OMISSION_REASONS.get(reason, reason)}: {count}"
for reason, count in sorted(reasons.items())
)
print(
f" Safety notice: {len(omitted_files)} unavailable or unsafe "
f"entries omitted ({required_omitted} required; {reason_summary})."
)
total_omitted += len(omitted_files)
print("\nChecking existing files...")
to_download, up_to_date, mismatched = check_local(
files, bios_path, jobs=args.jobs
)
present = len(up_to_date) + len(mismatched)
print(
f" {present}/{len(files)} present "
f"({len(up_to_date)} verified, {len(mismatched)} wrong hash)"
)
# Mismatched files need re-download
to_download.extend(mismatched)
if args.check:
if to_download:
print(f"\n {len(to_download)} files need downloading.")
else:
print("\n All files up to date.")
continue
if to_download:
dl_size = sum(f.get("size", 0) for f in to_download)
print(f"\nDownloading {len(to_download)} files ({format_size(dl_size)})...")
print(" Press Ctrl+C to stop. Running the same command again "
"carries on from here.")
try:
bios_path.mkdir(parents=True, exist_ok=True)
except OSError as exc:
print(f"\nCannot create {bios_path}.", file=sys.stderr)
print(f" {exc.strerror}.", file=sys.stderr)
print(" On Batocera, Recalbox and ROCKNIX the BIOS folder "
"belongs to root: run the same command with sudo.",
file=sys.stderr)
sys.exit(1)
_check_free_space(bios_path, dl_size)
failed, interrupted = download_files(
to_download, bios_path, jobs=args.jobs, verbose=args.verbose
)
total_downloaded += len(to_download) - len(failed)
total_errors += len(failed)
if interrupted:
total_interrupted = True
break
else:
print("\n All files up to date.")
total_up_to_date += len(up_to_date)
# Standalone copies
if (
manifest.get("standalone_copies")
and not args.check
and args.standalone_copies
):
print("\nStandalone emulators:")
lb_root = launchbox_root(os_type)
extra_dirs = launchbox_bios_dirs(lb_root) if lb_root else None
copied, skipped, copy_failed = do_standalone_copies(
manifest, bios_path, os_type, extra_dirs
)
if copied or skipped or copy_failed:
print(
f" {copied} copied, {skipped} skipped (dir not found), "
f"{copy_failed} failed"
)
total_errors += copy_failed
elif manifest.get("standalone_copies") and not args.check:
print(
"\nStandalone copies skipped "
"(use --standalone-copies to opt in)."
)
if not args.check and not listing:
_report_outcome(
installed_paths,
total_downloaded,
total_up_to_date,
total_errors,
total_omitted,
total_interrupted,
)
if __name__ == "__main__":
main()
Reference in new issue View Git Blame Copy Permalink