feat: add region filters and profiles

This commit is contained in:
Abdessamad Derraz committed 2026-08-09 14:15:56 +02:00
1 parent 1c537c3a30
commit 45f89cc6c0
51 files changed
+4594 -262

No files matched your search

+12 -12
View File
@@ -164,9 +164,16 @@ def md5_composite(filepath: str | Path) -> str:
return result
def parse_md5_list(raw: str) -> list[str]:
"""Parse comma-separated MD5 string into normalized lowercase list."""
return [m.strip().lower() for m in raw.split(",") if m.strip()] if raw else []
def parse_md5_list(raw: str | list | None) -> list[str]:
"""Normalize an md5 field into a lowercase list.
Platform YAMLs carry Recalbox multi-hash as one comma-separated string,
emulator profiles carry a YAML list. Both reach here.
"""
if not raw:
return []
values = raw if isinstance(raw, list) else str(raw).split(",")
return [str(m).strip().lower() for m in values if str(m).strip()]
_shared_yml_cache: dict[str, dict] = {}
@@ -436,11 +443,6 @@ def resolve_local_file(
exact, zip_exact, hash_mismatch, not_found.
"""
sha1 = file_entry.get("sha1")
md5_field = file_entry.get("md5", "") or ""
if isinstance(md5_field, list):
md5_raw = ",".join(str(m) for m in md5_field)
else:
md5_raw = str(md5_field)
name = file_entry.get("name", "")
zipped_file = file_entry.get("zipped_file")
aliases = file_entry.get("aliases", [])
@@ -461,9 +463,7 @@ def resolve_local_file(
if hint_base and hint_base not in names_to_try:
names_to_try.append(hint_base)
md5_list = (
[m.strip().lower() for m in md5_raw.split(",") if m.strip()] if md5_raw else []
)
md5_list = parse_md5_list(file_entry.get("md5"))
files_db = db.get("files", {})
by_md5 = db.get("indexes", {}).get("by_md5", {})
by_name = db.get("indexes", {}).get("by_name", {})
@@ -504,7 +504,7 @@ def resolve_local_file(
# declared size must confirm the match.
crc_raw = str(file_entry.get("crc32", "") or "").strip().lower()
declared_size = file_entry.get("size")
if crc_raw and declared_size and not zipped_file and not md5_raw:
if crc_raw and declared_size and not zipped_file and not md5_list:
by_crc32 = db.get("indexes", {}).get("by_crc32", {})
match = by_crc32.get(crc_raw)
if match and match in files_db:
+21 -7
View File
@@ -25,6 +25,7 @@ from common import (
load_emulator_profiles,
load_platform_config,
name_match_size_ok,
parse_md5_list,
require_yaml,
)
@@ -206,6 +207,7 @@ def cross_reference(
by_name = db.get("indexes", {}).get("by_name", {})
by_name_lower = {k.lower(): k for k in by_name}
by_md5 = db.get("indexes", {}).get("by_md5", {})
by_crc32 = db.get("indexes", {}).get("by_crc32", {})
by_path_suffix = db.get("indexes", {}).get("by_path_suffix", {})
db_files = db.get("files", {})
report = {}
@@ -331,20 +333,32 @@ def cross_reference(
path_field, by_name, by_name_lower,
data_names, by_path_suffix, f, db_files,
)
# Try the alternate names the emulator accepts, like
# resolve_local_file does
if source is None:
for alias in f.get("aliases") or []:
source = _resolve_source(
alias, by_name, by_name_lower,
data_names, by_path_suffix, f, db_files,
)
if source is not None:
break
# Try MD5 hash match
if source is None:
md5_raw = f.get("md5", "")
if md5_raw:
for md5_val in md5_raw.split(","):
md5_val = md5_val.strip().lower()
if md5_val and by_md5.get(md5_val):
source = "bios"
break
for md5_val in parse_md5_list(f.get("md5")):
if by_md5.get(md5_val):
source = "bios"
break
# Try SHA1 hash match
if source is None:
sha1 = f.get("sha1", "")
if sha1 and sha1 in db_files:
source = "bios"
# Try CRC32 hash match
if source is None:
crc32 = str(f.get("crc32", "")).lower()
if crc32 and by_crc32.get(crc32):
source = "bios"
if source is None:
source = "missing"
+294 -19
View File
@@ -46,9 +46,11 @@ from common import (
load_database,
load_emulator_profiles,
load_platform_config,
parse_md5_list,
require_yaml,
resolve_local_file,
)
import region as region_mod
from deterministic_zip import rebuild_zip_deterministic
from validation import (
_build_validation_index,
@@ -291,15 +293,7 @@ def resolve_file(
name = file_entry.get("name", "")
sha1 = file_entry.get("sha1")
first_sha1 = (sha1[0] if sha1 else "") if isinstance(sha1, list) else (sha1 or "")
md5_raw = file_entry.get("md5", "")
if isinstance(md5_raw, list):
md5_list = [str(m).strip().lower() for m in md5_raw if str(m).strip()]
else:
md5_list = (
[m.strip().lower() for m in md5_raw.split(",") if m.strip()]
if md5_raw
else []
)
md5_list = parse_md5_list(file_entry.get("md5"))
first_md5 = md5_list[0] if md5_list else ""
cached = fetch_large_file(name, expected_sha1=first_sha1, expected_md5=first_md5)
if cached:
@@ -777,6 +771,8 @@ def _build_readme(
num_systems: int,
source: str = "full",
contributors: list[dict] | None = None,
regions: list[str] | None = None,
fallback_systems: list[str] | None = None,
) -> str:
"""Build a personalized step-by-step README for each platform pack."""
sep = "=" * 50
@@ -952,12 +948,22 @@ def _build_readme(
),
)
if regions:
region_help = (
" - Wrong region? This pack was filtered. Only the\n"
" best-matching BIOS was kept per system. Use the\n"
" unfiltered pack to play imports.\n"
)
else:
region_help = (
" - Wrong region? Some systems have regional BIOS\n"
" variants (USA/EUR/JAP). All are included.\n"
)
footer = (
"TROUBLESHOOTING\n\n"
" - Core says BIOS missing? Check the exact filename\n"
" and make sure it's in the right subfolder.\n"
" - Wrong region? Some systems have regional BIOS\n"
" variants (USA/EUR/JAP). All are included.\n"
f"{region_help}"
" - Need help? https://github.com/Abdess/retrobios/issues\n\n"
f"{sep}\n"
f" https://github.com/Abdess/retrobios\n"
@@ -980,6 +986,30 @@ def _build_readme(
" Independent of platform scraper accuracy.\n\n"
)
region_info = ""
if regions:
pretty = ", ".join(
" ".join(w.title() for w in slug.split("-")) for slug in regions
)
region_info = (
"PACK TYPE: Region Filtered\n\n"
f" Region priority: {pretty}\n\n"
" Only the best-matching BIOS was kept for each system.\n"
)
if fallback_systems:
listed = "\n".join(f" {s}" for s in fallback_systems)
region_info += (
"\n These systems have no BIOS in those regions, so all\n"
" of theirs were kept:\n"
f"{listed}\n"
)
region_info += (
"\n This shrinks the pack. It does not change how cores pick\n"
" a BIOS: most already select per region from fixed filename\n"
" lists driven by the game's region. Loading imports from\n"
" another region may need the unfiltered pack.\n\n"
)
credits = ""
if contributors:
credits = "\nCONTRIBUTORS\n\n"
@@ -988,7 +1018,7 @@ def _build_readme(
credits += f" @{username}\n"
credits += "\n"
return header + source_info + guide + credits + footer
return header + source_info + region_info + guide + credits + footer
def _build_agnostic_rename_readme(
@@ -1028,6 +1058,7 @@ def generate_pack(
precomputed_extras: list[dict] | None = None,
source: str = "full",
flatten: bool = True,
regions: list[str] | None = None,
) -> str | None:
"""Generate a ZIP pack for a platform.
@@ -1045,6 +1076,7 @@ def generate_pack(
version_tag = f"_{version.replace(' ', '')}" if version else ""
req_tag = "_Required" if required_only else ""
source_tag = {"platform": "_Platform", "truth": "_Truth"}.get(source, "")
region_tag_str = f"_{region_mod.region_tag(regions)}" if regions else ""
sys_tag = ""
if system_filter:
@@ -1059,7 +1091,7 @@ def generate_pack(
display_parts.append("_".join(p.title() for p in parts if p))
sys_tag = "_" + "_".join(display_parts)
zip_name = f"{platform_display.replace(' ', '_')}{version_tag}{source_tag}{req_tag}_BIOS_Pack{sys_tag}.zip"
zip_name = f"{platform_display.replace(' ', '_')}{version_tag}{source_tag}{region_tag_str}{req_tag}_BIOS_Pack{sys_tag}.zip"
zip_path = os.path.join(output_dir, zip_name)
os.makedirs(output_dir, exist_ok=True)
@@ -1158,6 +1190,55 @@ def generate_pack(
if best is not None:
preferred_entries[full] = id(best)
# Region selection is decided once, over both the platform baseline and the
# core extras, so the two phases below consult a single set. Grouping uses a
# dedicated extras pass with an empty seen set: it only needs group
# membership, never destinations, and a superset of members is the safe
# direction. Runs only when --region is given.
region_drops: set[str] = set()
region_fallbacks: list[str] = []
if regions:
region_index = region_mod.build_region_index(emu_profiles or {})
region_groups: dict[str, list[tuple[str, str]]] = {}
for sys_id, system in pack_systems.items():
members = region_groups.setdefault(sys_id, [])
for file_entry in system.get("files", []):
dest = _sanitize_path(
file_entry.get("destination", file_entry.get("name", ""))
)
if dest:
members.append((dest, file_entry.get("name", "")))
if source != "platform":
emu_systems = {
n: list(p.get("systems", []))
for n, p in (emu_profiles or {}).items()
}
for fe in _collect_emulator_extras(
config,
emulators_dir,
db,
set(),
base_dest,
emu_profiles,
target_cores=target_cores,
include_all=(source == "truth"),
):
dest = _sanitize_path(fe.get("destination", fe.get("name", "")))
if not dest:
continue
for sys_id in emu_systems.get(
fe.get("source_emulator", ""), ["_extras"]
):
region_groups.setdefault(sys_id, []).append(
(dest, fe.get("name", ""))
)
region_drops = region_mod.resolve_region_drops(
region_groups, region_index, regions
)
region_fallbacks = region_mod.fallback_groups(
region_groups, region_index, regions
)
with zipfile.ZipFile(zip_path, "w", zipfile.ZIP_DEFLATED) as zf:
if source != "truth":
for sys_id, system in sorted(pack_systems.items()):
@@ -1165,6 +1246,8 @@ def generate_pack(
if required_only and file_entry.get("required") is False:
continue
dest = _sanitize_path(file_entry.get("destination", file_entry["name"]))
if region_drops and dest in region_drops:
continue
if not dest:
# EmuDeck-style entries (system:md5 whitelist, no filename).
fkey = f"{sys_id}/{file_entry.get('name', '')}"
@@ -1473,6 +1556,8 @@ def generate_pack(
if required_only and fe.get("required") is False:
continue
dest = _sanitize_path(fe.get("destination", fe["name"]))
if region_drops and dest in region_drops:
continue
if not dest:
continue
# Core extras: _collect_emulator_extras already adjusted
@@ -1572,6 +1657,8 @@ def generate_pack(
platform_name, platform_display, base_dest, total_files, num_systems,
source=source,
contributors=_pack_registry.get(platform_name, {}).get("contributed_by", []),
regions=regions,
fallback_systems=region_fallbacks,
)
zf.writestr("README.txt", readme_text)
@@ -1725,6 +1812,7 @@ def generate_emulator_pack(
standalone: bool = False,
zip_contents: dict | None = None,
required_only: bool = False,
regions: list[str] | None = None,
) -> str | None:
"""Generate a ZIP pack for specific emulator profiles."""
all_profiles = load_emulator_profiles(emulators_dir, skip_aliases=False)
@@ -1768,10 +1856,29 @@ def generate_emulator_pack(
# ZIP naming
display_names = [p.get("emulator", n).replace(" ", "") for n, p in selected]
zip_name = "_".join(display_names) + "_BIOS_Pack.zip"
region_tag_str = f"_{region_mod.region_tag(regions)}" if regions else ""
zip_name = "_".join(display_names) + f"{region_tag_str}_BIOS_Pack.zip"
zip_path = os.path.join(output_dir, zip_name)
os.makedirs(output_dir, exist_ok=True)
# One group per profile: a core's regional alternatives compete only with
# each other. Decided before the ZIP is opened so the loop below reads a
# single set.
region_drops: set[str] = set()
if regions:
region_index = region_mod.build_region_index(dict(selected))
region_groups: dict[str, list[tuple[str, str]]] = {}
for emu_name, profile in sorted(selected):
structure = profile.get("pack_structure")
members = region_groups.setdefault(emu_name, [])
for fe in filter_files_by_mode(profile.get("files", []), standalone):
dest = _resolve_destination(fe, structure, standalone)
if dest:
members.append((dest, fe.get("name", "")))
region_drops = region_mod.resolve_region_drops(
region_groups, region_index, regions
)
total_files = 0
missing_files = []
seen_destinations: set[str] = set()
@@ -1888,6 +1995,8 @@ def generate_emulator_pack(
dest = _resolve_destination(fe, pack_structure, standalone)
if not dest:
continue
if region_drops and dest in region_drops:
continue
if dest.lower() in seen_lower:
continue
@@ -1989,6 +2098,7 @@ def generate_system_pack(
standalone: bool = False,
zip_contents: dict | None = None,
required_only: bool = False,
regions: list[str] | None = None,
) -> str | None:
"""Generate a ZIP pack for all emulators supporting given system IDs."""
profiles = load_emulator_profiles(emulators_dir)
@@ -2036,6 +2146,7 @@ def generate_system_pack(
standalone,
zip_contents,
required_only=required_only,
regions=regions,
)
if result:
# Rename to system-based name
@@ -2092,6 +2203,7 @@ def generate_split_packs(
target_cores: set[str] | None = None,
required_only: bool = False,
source: str = "full",
regions: list[str] | None = None,
) -> list[str]:
"""Generate split packs (one ZIP per system or manufacturer)."""
config = load_platform_config(platform_name, platforms_dir)
@@ -2162,13 +2274,15 @@ def generate_split_packs(
system_filter=group_system_ids,
precomputed_extras=group_extras,
source=source,
regions=regions,
)
if zip_path:
version = config.get("version", config.get("dat_version", ""))
ver_tag = f"_{version.replace(' ', '')}" if version else ""
req_tag = "_Required" if required_only else ""
rgn_tag = f"_{region_mod.region_tag(regions)}" if regions else ""
safe_group = group_name.replace(" ", "_")
new_name = f"{platform_display.replace(' ', '_')}{ver_tag}{source_tag}{req_tag}_{safe_group}_BIOS_Pack.zip"
new_name = f"{platform_display.replace(' ', '_')}{ver_tag}{source_tag}{rgn_tag}{req_tag}_{safe_group}_BIOS_Pack.zip"
new_path = os.path.join(split_dir, new_name)
if new_path != zip_path:
os.rename(zip_path, new_path)
@@ -2399,6 +2513,8 @@ def _validate_args(args, parser):
parser.error("--manifest is incompatible with --emulator")
if args.manifest and args.split:
parser.error("--manifest is incompatible with --split")
if getattr(args, "region", None) and has_from_md5:
parser.error("--region and --from-md5 are mutually exclusive")
def _write_manifest_if_changed(path: str, manifest: dict) -> None:
@@ -2456,11 +2572,18 @@ def _run_manifest_mode(
emu_profiles=emu_profiles,
target_cores=tc,
source=source,
regions=getattr(args, "regions", None),
)
source_suffix = {"platform": "_platform", "truth": "_truth"}.get(source, "")
req_suffix = "_required" if required_only else ""
rgn = getattr(args, "regions", None)
region_suffix = (
f"_{region_mod.region_tag(rgn).lower()}" if rgn else ""
)
out_path = os.path.join(
args.output_dir, f"{representative}{source_suffix}{req_suffix}.json"
args.output_dir,
f"{representative}{source_suffix}{region_suffix}"
f"{req_suffix}.json",
)
_write_manifest_if_changed(out_path, manifest)
print(
@@ -2472,7 +2595,8 @@ def _run_manifest_mode(
if alias_plat != representative:
alias_path = os.path.join(
args.output_dir,
f"{alias_plat}{source_suffix}{req_suffix}.json",
f"{alias_plat}{source_suffix}{region_suffix}"
f"{req_suffix}.json",
)
alias_manifest = dict(manifest)
alias_manifest["platform"] = alias_plat
@@ -2518,6 +2642,8 @@ def _run_verify_packs(args):
sys.exit(1)
all_ok = True
verify_regions = getattr(args, "regions", None)
verify_profiles = load_emulator_profiles(args.emulators_dir)
for platform_name in platforms:
config = load_platform_config(platform_name, args.platforms_dir)
display = config.get("platform", platform_name).replace(" ", "_")
@@ -2561,6 +2687,43 @@ def _run_verify_packs(args):
os.path.join(extract_dir, base_dest)
)
region_drops: set[str] = set()
if verify_regions:
region_index = region_mod.build_region_index(verify_profiles)
region_groups: dict[str, list[tuple[str, str]]] = {}
for sys_id, sys_data in systems.items():
members = region_groups.setdefault(sys_id, [])
for fe in sys_data.get("files", []):
d = _sanitize_path(
fe.get("destination", fe.get("name", ""))
)
if d:
members.append((d, fe.get("name", "")))
# Core extras join the groups so a platform file dropped in
# favour of a better-ranked core file is not reported missing.
from verify import find_undeclared_files as _fud
emu_systems = {
n: list(p.get("systems", []))
for n, p in verify_profiles.items()
}
for u in _fud(
config, args.emulators_dir, verify_db, verify_profiles
):
if not u.get("in_repo"):
continue
raw = u.get("path") or u["name"]
d = _sanitize_path(
f"{raw}{u['name']}" if raw.endswith("/") else raw
)
if not d:
continue
for sid in emu_systems.get(u.get("emulator", ""), ["_extras"]):
region_groups.setdefault(sid, []).append((d, u["name"]))
region_drops = region_mod.resolve_region_drops(
region_groups, region_index, verify_regions
)
missing = []
hash_fail = []
ok = 0
@@ -2569,6 +2732,8 @@ def _run_verify_packs(args):
dest = fe.get("destination", fe.get("name", ""))
if not dest:
continue
if region_drops and _sanitize_path(dest) in region_drops:
continue
fp = (
os.path.join(extract_dir, base_dest, dest)
if base_dest and not is_flat
@@ -2733,6 +2898,7 @@ def _run_platform_packs(
target_cores=tc,
required_only=required_only,
source=source,
regions=getattr(args, "regions", None),
)
print(f" Split into {len(zip_paths)} packs")
else:
@@ -2751,6 +2917,7 @@ def _run_platform_packs(
required_only=required_only,
system_filter=system_filter,
source=source,
regions=getattr(args, "regions", None),
)
if not args.split and zip_path and aliases:
rep_cfg = load_platform_config(representative, args.platforms_dir)
@@ -2776,14 +2943,22 @@ def _run_platform_packs(
print("\nVerifying packs and generating manifests...")
skip_conf = bool(system_filter or args.split)
all_ok = verify_and_finalize_packs(
args.output_dir, db, skip_conformance=skip_conf, data_registry=data_registry
args.output_dir,
db,
skip_conformance=skip_conf,
data_registry=data_registry,
regions=getattr(args, "regions", None),
)
if args.split:
for entry in os.listdir(args.output_dir):
sub = os.path.join(args.output_dir, entry)
if os.path.isdir(sub) and entry.endswith("_Split"):
ok = verify_and_finalize_packs(
sub, db, skip_conformance=True, data_registry=data_registry
sub,
db,
skip_conformance=True,
data_registry=data_registry,
regions=getattr(args, "regions", None),
)
all_ok = all_ok and ok
if not all_ok:
@@ -2858,6 +3033,10 @@ def main():
help="Grouping for --split (default: system)",
)
parser.add_argument("--target", "-t", help="Hardware target (e.g., switch, rpi4)")
parser.add_argument(
"--region",
help="Region priority list, best first (e.g. us,eu,jp)",
)
parser.add_argument(
"--list-targets",
action="store_true",
@@ -2884,6 +3063,15 @@ def main():
)
args = parser.parse_args()
# Parsed before the quick-exit modes: --verify-packs returns early and
# still needs the region priority list to narrow its expectation.
args.regions = []
if args.region:
try:
args.regions = region_mod.parse_requested(args.region)
except ValueError as exc:
parser.error(str(exc))
# Quick-exit modes: --verify-packs alone = verify existing packs only
# Combined with --all-variants, generation runs first then verify
if args.verify_packs and not args.all_variants:
@@ -2975,6 +3163,7 @@ def main():
args.standalone,
zip_contents,
required_only=args.required_only,
regions=getattr(args, "regions", None),
):
sys.exit(1)
return
@@ -2991,6 +3180,7 @@ def main():
args.standalone,
zip_contents,
required_only=args.required_only,
regions=getattr(args, "regions", None),
):
sys.exit(1)
return
@@ -3112,6 +3302,7 @@ def generate_manifest(
emu_profiles: dict | None = None,
target_cores: set[str] | None = None,
source: str = "full",
regions: list[str] | None = None,
) -> dict:
"""Generate a JSON manifest for a platform (same resolution as generate_pack).
@@ -3157,6 +3348,45 @@ def generate_manifest(
manifest_files: list[dict] = []
total_size = 0
region_drops: set[str] = set()
if regions:
region_index = region_mod.build_region_index(emu_profiles)
region_groups: dict[str, list[tuple[str, str]]] = {}
for sys_id, system in pack_systems.items():
members = region_groups.setdefault(sys_id, [])
for file_entry in system.get("files", []):
d = _sanitize_path(
file_entry.get("destination", file_entry.get("name", ""))
)
if d:
members.append((d, file_entry.get("name", "")))
if source != "platform":
emu_systems = {
n: list(p.get("systems", [])) for n, p in emu_profiles.items()
}
for fe in _collect_emulator_extras(
config,
emulators_dir,
db,
set(),
base_dest,
emu_profiles,
target_cores=target_cores,
include_all=(source == "truth"),
):
d = _sanitize_path(fe.get("destination", fe.get("name", "")))
if not d:
continue
for sid in emu_systems.get(
fe.get("source_emulator", ""), ["_extras"]
):
region_groups.setdefault(sid, []).append(
(d, fe.get("name", ""))
)
region_drops = region_mod.resolve_region_drops(
region_groups, region_index, regions
)
# Phase 1: baseline files
if source != "truth":
for sys_id, system in sorted(pack_systems.items()):
@@ -3164,6 +3394,8 @@ def generate_manifest(
dest = _sanitize_path(file_entry.get("destination", file_entry["name"]))
if not dest:
continue
if region_drops and dest in region_drops:
continue
full_dest = f"{base_dest}/{dest}" if base_dest else dest
dedup_key = full_dest
@@ -3234,6 +3466,8 @@ def generate_manifest(
dest = _sanitize_path(fe.get("destination", fe["name"]))
if not dest:
continue
if region_drops and dest in region_drops:
continue
if extras_pfx:
if not dest.startswith(f"{extras_pfx}/"):
full_dest = f"{extras_pfx}/{dest}"
@@ -3703,9 +3937,13 @@ def verify_pack_against_platform(
db: dict | None = None,
emulators_dir: str = "emulators",
emu_profiles: dict | None = None,
regions: list[str] | None = None,
) -> tuple[bool, int, int, list[str]]:
"""Verify a pack ZIP against its platform config and core requirements.
A region priority list narrows the expectation to what the builder would
have packed, using the same selection function.
Checks:
1. Every baseline file declared by the platform exists in the ZIP
at the correct destination path
@@ -3729,6 +3967,37 @@ def verify_pack_against_platform(
if emu_profiles is None:
emu_profiles = load_emulator_profiles(emulators_dir)
region_drops: set[str] = set()
if regions:
region_index = region_mod.build_region_index(emu_profiles)
region_groups: dict[str, list[tuple[str, str]]] = {}
for sys_id, system in config.get("systems", {}).items():
members = region_groups.setdefault(sys_id, [])
for fe in system.get("files", []):
d = _sanitize_path(fe.get("destination", fe.get("name", "")))
if d:
members.append((d, fe.get("name", "")))
if db is not None:
from verify import find_undeclared_files as _fud
emu_systems = {
n: list(p.get("systems", [])) for n, p in emu_profiles.items()
}
for u in _fud(config, emulators_dir, db, emu_profiles):
if not u.get("in_repo"):
continue
raw = u.get("path") or u["name"]
d = _sanitize_path(
f"{raw}{u['name']}" if raw.endswith("/") else raw
)
if not d:
continue
for sys_id in emu_systems.get(u.get("emulator", ""), ["_extras"]):
region_groups.setdefault(sys_id, []).append((d, u["name"]))
region_drops = region_mod.resolve_region_drops(
region_groups, region_index, regions
)
with zipfile.ZipFile(zip_path, "r") as zf:
zip_set = set(zf.namelist())
zip_lower = {n.lower(): n for n in zip_set}
@@ -3779,6 +4048,8 @@ def verify_pack_against_platform(
dest = fe.get("destination", fe.get("name", ""))
if not dest:
continue
if region_drops and _sanitize_path(dest) in region_drops:
continue
expected = f"{base_dest}/{dest}" if base_dest and not is_flat else dest
baseline_checked += 1
@@ -3845,6 +4116,8 @@ def verify_pack_against_platform(
continue
raw_dest = u.get("path") or u["name"]
dest = f"{raw_dest}{u['name']}" if raw_dest.endswith("/") else raw_dest
if region_drops and _sanitize_path(dest) in region_drops:
continue
if extras_pfx and not (is_flat and extras_pfx == base_dest):
if not dest.startswith(f"{extras_pfx}/"):
full = f"{extras_pfx}/{dest}"
@@ -3889,6 +4162,7 @@ def verify_and_finalize_packs(
platforms_dir: str = "platforms",
skip_conformance: bool = False,
data_registry: dict | None = None,
regions: list[str] | None = None,
) -> bool:
"""Verify all packs, inject manifests, generate SHA256SUMS.
@@ -3953,6 +4227,7 @@ def verify_and_finalize_packs(
pname,
platforms_dir,
db=db,
regions=regions,
)
status = "OK" if p_ok else "FAILED"
print(
+3 -3
View File
@@ -31,6 +31,7 @@ from common import (
load_database,
load_emulator_profiles,
load_provenance_snapshots,
parse_md5_list,
require_yaml,
unique_emulator_profiles,
write_if_changed,
@@ -1479,9 +1480,8 @@ def generate_emulator_page(
return True
md5_raw = f.get("md5", "")
if md5_raw:
for md5_val in md5_raw.split(","):
md5_val = md5_val.strip().lower()
if md5_val and by_md5.get(md5_val):
for md5_val in parse_md5_list(md5_raw):
if by_md5.get(md5_val):
return True
sha1 = f.get("sha1", "")
if sha1 and sha1 in db_files:
+245
View File
@@ -0,0 +1,245 @@
"""Region vocabulary and ordered-priority selection for pack generation.
A file's region: value is the region the emulator code selects it for, not the
region of the dump. Signal standard has no field of its own; it survives only
inside slugs whose hardware register fuses the two axes (Saturn SMPC areas).
"""
from __future__ import annotations
WORLD = "world"
# Two levels only: super-region -> member territories.
REGION_TREE: dict[str, frozenset[str]] = {
"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"}),
}
_PARENT: dict[str, str] = {
member: parent for parent, members in REGION_TREE.items() for member in members
}
REGIONS: frozenset[str] = frozenset({WORLD} | set(REGION_TREE) | set(_PARENT))
# Accepted on the CLI and for legacy profile values.
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,
"gb": "uk",
}
def _canonical(raw: str) -> str:
"""Map one token to a canonical region slug."""
key = raw.strip().lower()
key = ALIASES.get(key, key)
if key not in REGIONS:
raise ValueError(f"unknown region: {raw!r}")
return key
def comparable(a: str, b: str) -> bool:
"""True when two regions are equal or in a parent/child relation."""
return a == b or _PARENT.get(a) == b or _PARENT.get(b) == a
def normalize_declared(raw: str | list | None) -> set[str]:
"""Normalise a profile region: value to canonical slugs."""
if raw is None:
return set()
values = raw if isinstance(raw, list) else [raw]
return {_canonical(str(v)) for v in values}
def parse_requested(raw: str) -> list[str]:
"""Parse a comma-separated --region value into an ordered priority list."""
out: list[str] = []
for part in raw.split(","):
if not part.strip():
continue
slug = _canonical(part)
if slug not in out:
out.append(slug)
if not out:
raise ValueError("--region requires at least one region")
return out
def rank(file_regions: set[str], requested: list[str]) -> int:
"""Priority index of a file, lower is better.
Returns len(requested) when no requested region is comparable, the implicit
lowest rank that keeps a group from ever being emptied.
"""
for i, req in enumerate(requested):
if any(comparable(req, fr) for fr in file_regions):
return i
return len(requested)
def region_tag(requested: list[str]) -> str:
"""Pack filename tag for a requested priority list."""
return "_".join(
"".join(part.title() for part in slug.split("-")) for slug in requested
)
def build_region_index(profiles: dict) -> dict[str, dict]:
"""Build a region lookup from emulator profiles.
Keyed by the entry's path when present, by name otherwise. Entries sharing a
key have their regions unioned, so an ambiguous lookup keeps more files
rather than fewer.
"""
index: dict[str, dict] = {}
for emu_name, profile in sorted(profiles.items()):
if profile.get("type") in ("launcher", "alias"):
continue
for f in profile.get("files") or []:
if not isinstance(f, dict):
continue
try:
regions = normalize_declared(f.get("region"))
except ValueError as exc:
raise ValueError(
f"{emu_name}: {f.get('name', '?')}: {exc}"
) from exc
if not regions:
continue
name = f.get("name", "")
path = f.get("path") or ""
# Path-keyed so same-named entries stay separate (Dolphin declares
# three IPL.bin), name-keyed so a candidate identified by name alone
# sees the union and is never dropped on ambiguity.
for key in {path, name} - {""}:
entry = index.setdefault(key, {"regions": set(), "emulators": []})
entry["regions"] |= regions
if emu_name not in entry["emulators"]:
entry["emulators"].append(emu_name)
return index
def lookup_regions(index: dict[str, dict], destination: str, name: str) -> set[str]:
"""Look a candidate file up in the region index.
Tries the full destination, then progressively shorter path suffixes, then
the bare name. Suffix matching bridges a platform entry whose destination
carries an extra prefix to the profile entry that declares the region.
"""
if destination:
entry = index.get(destination)
if entry:
return set(entry["regions"])
parts = destination.split("/")
for i in range(1, len(parts)):
entry = index.get("/".join(parts[i:]))
if entry:
return set(entry["regions"])
entry = index.get(name)
return set(entry["regions"]) if entry else set()
def _competing_ranks(
members: list[tuple[str, str]],
index: dict[str, dict],
requested: list[str],
) -> list[tuple[int, str]]:
"""Rank the members of a group that compete regionally.
Files with no declared region, and files declared world, are excluded: they
never compete and never drop.
"""
ranked: list[tuple[int, str]] = []
for destination, name in members:
regions = lookup_regions(index, destination, name)
if not regions or WORLD in regions:
continue
ranked.append((rank(regions, requested), destination))
return ranked
def resolve_region_drops(
groups: dict[str, list[tuple[str, str]]],
index: dict[str, dict],
requested: list[str],
) -> set[str]:
"""Destinations to skip for a requested region priority list.
Per group, only the best rank actually present survives. A destination kept
by any group is kept overall.
"""
if not requested:
return set()
keep: set[str] = set()
drop: set[str] = set()
for members in groups.values():
ranked = _competing_ranks(members, index, requested)
competing = {dest for _r, dest in ranked}
keep |= {dest for dest, _name in members if dest not in competing}
if not ranked:
continue
best = min(r for r, _ in ranked)
for r, destination in ranked:
if r == best:
keep.add(destination)
else:
drop.add(destination)
return drop - keep
def fallback_groups(
groups: dict[str, list[tuple[str, str]]],
index: dict[str, dict],
requested: list[str],
) -> list[str]:
"""Group IDs where no candidate matched, so the whole group was kept."""
if not requested:
return []
out: list[str] = []
for group_id, members in groups.items():
ranked = _competing_ranks(members, index, requested)
if ranked and min(r for r, _ in ranked) == len(requested):
out.append(group_id)
return sorted(out)
+211
View File
@@ -0,0 +1,211 @@
"""Cross-check profile region: values against dump-catalog provenance.
Advisory only. The emulator source decides what a region: value should be; this
reports where an independent catalogue disagrees, so a hand-written value gets a
second opinion before it is trusted.
Read the limits printed in the header before acting on anything here.
"""
from __future__ import annotations
import argparse
import collections
import json
import re
import sys
from pathlib import Path
sys.path.insert(0, str(Path(__file__).resolve().parent))
import region
from common import load_emulator_profiles, load_database
# No-Intro filename tokens: full English territory names.
NOINTRO = {
"World": "world",
"USA": "north-america",
"Canada": "canada",
"Europe": "europe",
"Japan": "japan",
"Asia": "asia",
"Korea": "south-korea",
"Brazil": "brazil",
"Taiwan": "taiwan",
"China": "china",
"Hong Kong": "hong-kong",
"France": "france",
"Germany": "germany",
"Spain": "spain",
"Italy": "italy",
"Netherlands": "netherlands",
"Sweden": "sweden",
"Norway": "norway",
"Denmark": "denmark",
"Finland": "finland",
"Australia": "australia",
"New Zealand": "new-zealand",
"Russia": "russia",
"UK": "uk",
"Portugal": "portugal",
"Greece": "greece",
"Poland": "poland",
"India": "india",
"Mexico": "mexico",
"Argentina": "argentina",
"Latin America": "latin-america",
}
_PAREN = re.compile(r"\(([^)]+)\)")
LIMITS = """Limits, read before trusting any line below.
TOSEC's country field is the country of ORIGIN, not the regional variant
(TOSEC Naming Convention v4, p.12). A Korean MSX ROM published by Microsoft
US is tagged (US). TOSEC names are therefore NOT read here.
Redump DATs carry no region at all: it exists in the redump.org database but
is never exported, so a Redump entry contributes nothing.
Only No-Intro filename tokens are usable, and only for the fraction of repo
files that carry one. Everything else is silent, not clean.
The emulator source decides. A disagreement is a prompt to reread the code,
never a correction to apply on its own.
"""
def tokens_from_name(name: str) -> set[str]:
"""Extract geography slugs from a No-Intro style DAT entry name."""
found: set[str] = set()
for group in _PAREN.findall(name):
for part in (p.strip() for p in group.split(",")):
slug = NOINTRO.get(part)
if slug:
found.add(slug)
return found
def catalog_regions(db: dict) -> dict[str, tuple[set[str], str]]:
"""Map SHA1 to (geography slugs, source label) from No-Intro provenance."""
out: dict[str, tuple[set[str], str]] = {}
for sha1, record in db["files"].items():
entry = (record.get("provenance") or {}).get("no-intro")
if not entry:
continue
name = entry.get("name", "")
found = tokens_from_name(name)
if found:
out[sha1] = (found, name)
return out
def resolve_sha1(file_entry: dict, db: dict) -> str | None:
"""Resolve a profile file entry to a repo SHA1, or None when ambiguous."""
files = db["files"]
indexes = db["indexes"]
declared = str(file_entry.get("sha1") or "").lower()
if declared and declared in files:
return declared
md5 = str(file_entry.get("md5") or "").lower()
if md5 and md5 in indexes["by_md5"]:
return indexes["by_md5"][md5]
hits = indexes["by_name"].get(file_entry.get("name", ""), [])
if isinstance(hits, str):
hits = [hits]
return hits[0] if len(hits) == 1 else None
def build_report(profiles: dict, db: dict) -> dict:
"""Compare every declared region: against the No-Intro catalogue."""
catalog = catalog_regions(db)
agree = 0
unchecked = 0
conflicts: list[dict] = []
vocabulary: collections.Counter = collections.Counter()
for slugs, _name in catalog.values():
vocabulary.update(slugs)
for emu_name, profile in sorted(profiles.items()):
if profile.get("type") in ("launcher", "alias"):
continue
for f in profile.get("files") or []:
if not isinstance(f, dict) or not f.get("region"):
continue
declared = region.normalize_declared(f["region"])
sha1 = resolve_sha1(f, db)
if not sha1 or sha1 not in catalog:
unchecked += 1
continue
found, name = catalog[sha1]
if region.WORLD in declared or any(
region.comparable(d, c) for d in declared for c in found
):
agree += 1
continue
conflicts.append(
{
"emulator": emu_name,
"file": f.get("name", ""),
"profile": sorted(declared),
"catalog": sorted(found),
"dat_name": name,
}
)
return {
"agree": agree,
"disagree": len(conflicts),
"unchecked": unchecked,
"catalog_files": len(catalog),
"repo_files": len(db["files"]),
"vocabulary": dict(vocabulary.most_common()),
"conflicts": conflicts,
}
def print_report(report: dict) -> None:
print(LIMITS)
print(
f"repo files: {report['repo_files']} "
f"with a usable No-Intro region token: {report['catalog_files']}"
)
print(
f"declared region: values -> agree {report['agree']}, "
f"disagree {report['disagree']}, no catalogue entry {report['unchecked']}"
)
if report["vocabulary"]:
print("\ncatalogue vocabulary present in this collection:")
for slug, count in report["vocabulary"].items():
print(f" {count:5d} {slug}")
if not report["conflicts"]:
print("\nNo disagreement.")
return
print("\nDISAGREEMENTS (reread the emulator source, do not apply blindly):\n")
for c in report["conflicts"]:
print(f" {c['emulator']}: {c['file']}")
print(f" profile: {c['profile']}")
print(f" catalogue: {c['catalog']} ({c['dat_name']})")
def main() -> int:
parser = argparse.ArgumentParser(
description="Cross-check profile region: values against No-Intro provenance"
)
parser.add_argument("--db", default="database.json", help="Path to database.json")
parser.add_argument("--emulators-dir", default="emulators")
parser.add_argument("--json", action="store_true", help="Machine-readable output")
args = parser.parse_args()
db = load_database(args.db)
profiles = load_emulator_profiles(args.emulators_dir)
report = build_report(profiles, db)
if args.json:
print(json.dumps(report, indent=2, sort_keys=True))
else:
print_report(report)
return 0
if __name__ == "__main__":
raise SystemExit(main())
+1 -1
View File
@@ -1495,7 +1495,7 @@ def print_emulator_result(result: dict, verbose: bool = False) -> None:
f" Ground truth: {gt_cov['with_validation']}/{gt_cov['total']} files have emulator validation ({pct}%)"
)
if gt_cov["platform_only"]:
print(f" {gt_cov['platform_only']} platform-only (no emulator profile)")
print(f" {gt_cov['platform_only']} without declared validation")
def main():