feat: resolve_local_file data directory fallback

This commit is contained in:
Abdessamad Derraz
2026-03-29 11:08:31 +02:00
parent 500e33d1ff
commit c513d6c0ad
4 changed files with 67 additions and 9 deletions

View File

@@ -292,6 +292,7 @@ def resolve_local_file(
zip_contents: dict | None = None,
dest_hint: str = "",
_depth: int = 0,
data_dir_registry: dict | None = None,
) -> tuple[str | None, str]:
"""Resolve a BIOS file to its local path using database.json.
@@ -445,10 +446,28 @@ def resolve_local_file(
canonical_entry = {"name": canonical}
result = resolve_local_file(
canonical_entry, db, zip_contents, dest_hint, _depth=_depth + 1,
data_dir_registry=data_dir_registry,
)
if result[0]:
return result[0], "mame_clone"
# Data directory fallback: scan data/ caches for matching filename
if data_dir_registry:
for _dd_key, dd_entry in data_dir_registry.items():
cache_dir = dd_entry.get("local_cache", "")
if not cache_dir or not os.path.isdir(cache_dir):
continue
for try_name in names_to_try:
candidate = os.path.join(cache_dir, try_name)
if os.path.isfile(candidate):
return candidate, "data_dir"
if "/" in try_name:
basename_candidate = os.path.join(
cache_dir, try_name.rsplit("/", 1)[-1],
)
if os.path.isfile(basename_candidate):
return basename_candidate, "data_dir"
return None, "not_found"

View File

@@ -230,7 +230,8 @@ def _register_path(dest: str, seen_files: set[str],
def resolve_file(file_entry: dict, db: dict, bios_dir: str,
zip_contents: dict | None = None,
dest_hint: str = "") -> tuple[str | None, str]:
dest_hint: str = "",
data_dir_registry: dict | None = None) -> tuple[str | None, str]:
"""Resolve a BIOS file with storage tiers and release asset fallback.
Wraps common.resolve_local_file() with pack-specific logic for
@@ -244,7 +245,8 @@ def resolve_file(file_entry: dict, db: dict, bios_dir: str,
return None, "external"
path, status = resolve_local_file(file_entry, db, zip_contents,
dest_hint=dest_hint)
dest_hint=dest_hint,
data_dir_registry=data_dir_registry)
if path and status != "hash_mismatch":
return path, status
@@ -545,7 +547,10 @@ def generate_pack(
total_files += 1
continue
local_path, status = resolve_file(file_entry, db, bios_dir, zip_contents)
local_path, status = resolve_file(
file_entry, db, bios_dir, zip_contents,
data_dir_registry=data_registry,
)
if status == "external":
file_ext = os.path.splitext(file_entry["name"])[1] or ""
@@ -688,7 +693,10 @@ def generate_pack(
if _has_path_conflict(full_dest, seen_destinations, seen_parents):
continue
local_path, status = resolve_file(fe, db, bios_dir, zip_contents)
local_path, status = resolve_file(
fe, db, bios_dir, zip_contents,
data_dir_registry=data_registry,
)
if status in ("not_found", "external", "user_provided"):
continue
@@ -1009,7 +1017,10 @@ def generate_emulator_pack(
continue
archive_entry = {"name": archive_name}
local_path, status = resolve_file(archive_entry, db, bios_dir, zip_contents)
local_path, status = resolve_file(
archive_entry, db, bios_dir, zip_contents,
data_dir_registry=data_registry,
)
if local_path and status not in ("not_found",):
if local_path.endswith(".zip"):
_normalize_zip_for_pack(local_path, archive_dest, zf)
@@ -1050,8 +1061,10 @@ def generate_emulator_pack(
continue
dest_hint = fe.get("path", "")
local_path, status = resolve_file(fe, db, bios_dir, zip_contents,
dest_hint=dest_hint)
local_path, status = resolve_file(
fe, db, bios_dir, zip_contents,
dest_hint=dest_hint, data_dir_registry=data_registry,
)
if status == "external":
file_ext = os.path.splitext(fe["name"])[1] or ""

View File

@@ -501,6 +501,7 @@ def verify_platform(
emulators_dir: str = DEFAULT_EMULATORS_DIR,
emu_profiles: dict | None = None,
target_cores: set[str] | None = None,
data_dir_registry: dict | None = None,
) -> dict:
"""Verify all BIOS files for a platform, including cross-reference gaps."""
mode = config.get("verification_mode", "existence")
@@ -540,6 +541,7 @@ def verify_platform(
for file_entry in system.get("files", []):
local_path, resolve_status = resolve_local_file(
file_entry, db, zip_contents,
data_dir_registry=data_dir_registry,
)
if mode == "existence":
result = verify_entry_existence(
@@ -965,7 +967,10 @@ def verify_emulator(
if archive and archive not in seen_archives:
seen_archives.add(archive)
archive_entry = {"name": archive}
local_path, _ = resolve_local_file(archive_entry, db, zip_contents)
local_path, _ = resolve_local_file(
archive_entry, db, zip_contents,
data_dir_registry=data_registry,
)
required = any(
f.get("archive") == archive and f.get("required", True)
for f in files
@@ -999,6 +1004,7 @@ def verify_emulator(
dest_hint = file_entry.get("path", "")
local_path, resolve_status = resolve_local_file(
file_entry, db, zip_contents, dest_hint=dest_hint,
data_dir_registry=data_registry,
)
name = file_entry.get("name", "")
required = file_entry.get("required", True)
@@ -1269,6 +1275,7 @@ def main():
# Load emulator profiles once for cross-reference (not per-platform)
emu_profiles = load_emulator_profiles(args.emulators_dir)
data_registry = load_data_dir_registry(args.platforms_dir)
target_cores_cache: dict[str, set[str] | None] = {}
if args.target:
@@ -1300,7 +1307,10 @@ def main():
for group_platforms, representative in groups:
config = load_platform_config(representative, args.platforms_dir)
tc = target_cores_cache.get(representative) if args.target else None
result = verify_platform(config, db, args.emulators_dir, emu_profiles, target_cores=tc)
result = verify_platform(
config, db, args.emulators_dir, emu_profiles,
target_cores=tc, data_dir_registry=data_registry,
)
names = [load_platform_config(p, args.platforms_dir).get("platform", p) for p in group_platforms]
group_results.append((result, names))
for p in group_platforms:

View File

@@ -2477,6 +2477,22 @@ class TestE2E(unittest.TestCase):
# Subdirectory destination should be added
self.assertIn("subcore/bios/present_req.bin", extra_dests)
def test_167_resolve_local_file_data_dir_fallback(self):
"""resolve_local_file finds files in data directories when not in bios/."""
data_dir = os.path.join(self.root, "data", "test-data")
os.makedirs(data_dir, exist_ok=True)
data_file = os.path.join(data_dir, "data_only.bin")
with open(data_file, "wb") as f:
f.write(b"DATA_DIR_CONTENT")
registry = {"test-data": {"local_cache": data_dir}}
fe = {"name": "data_only.bin"}
path, status = resolve_local_file(fe, self.db, data_dir_registry=registry)
self.assertIsNotNone(path)
self.assertEqual(os.path.basename(path), "data_only.bin")
self.assertEqual(status, "data_dir")
def test_90_registry_install_metadata(self):
"""Registry install section is accessible."""
import yaml