mirror of
https://github.com/Abdess/retroarch_system.git
synced 2026-10-18 09:23:26 -05:00
refactor: annotate every remaining signature
This commit is contained in:
1 parent
5961b6bfe7
commit
faa8af0800
52 files changed
+261
-145
No files matched your search
+1
-1
@@ -8,7 +8,7 @@ param(
|
||||
|
||||
$ErrorActionPreference = "Stop"
|
||||
$defaultInstallUrl = "https://raw.githubusercontent.com/Abdess/retrobios/main/install.py"
|
||||
$defaultInstallSha256 = "2e9e93d9809d9276a84209cfd421ad9b0d55853ce9c9d6761beb2ef21044ef2f"
|
||||
$defaultInstallSha256 = "e18309beb7da0a974cff3ef9ca8019292e5798a19ae78fd4711b6b2cca88ebc6"
|
||||
$maximumInstallerBytes = 2MB
|
||||
$installerExit = 0
|
||||
$installer = if ($PSScriptRoot) { Join-Path $PSScriptRoot "install.py" } else { $null }
|
||||
|
||||
+17
-8
@@ -18,6 +18,7 @@ import argparse
|
||||
import concurrent.futures
|
||||
import errno
|
||||
import hashlib
|
||||
import http.client
|
||||
import io
|
||||
import json
|
||||
import os
|
||||
@@ -33,7 +34,9 @@ import urllib.error
|
||||
import urllib.parse
|
||||
import urllib.request
|
||||
import xml.etree.ElementTree as ET
|
||||
from collections.abc import Callable
|
||||
from pathlib import Path, PurePosixPath
|
||||
from typing import BinaryIO, TextIO
|
||||
|
||||
# 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.
|
||||
@@ -232,7 +235,7 @@ class Terminal:
|
||||
A pipe or a log gets plain lines; NO_COLOR and TERM=dumb are honoured.
|
||||
"""
|
||||
|
||||
def __init__(self, stream=None, quiet: bool = False) -> None:
|
||||
def __init__(self, stream: TextIO | None = None, quiet: bool = False) -> None:
|
||||
self.stream = stream or sys.stdout
|
||||
self.quiet = quiet
|
||||
isatty = getattr(self.stream, "isatty", None)
|
||||
@@ -1055,7 +1058,9 @@ def normalize_platform(name: str) -> str:
|
||||
return plat
|
||||
|
||||
|
||||
def _read_limited_json(response, limit: int, label: str) -> object:
|
||||
def _read_limited_json(
|
||||
response: http.client.HTTPResponse, 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:
|
||||
@@ -1499,7 +1504,11 @@ def _for_cores(entry: dict, chosen: set[str]) -> dict | None:
|
||||
return None
|
||||
|
||||
|
||||
def _settle_slot(entry: dict, chosen: set[str], kept_system=None) -> dict:
|
||||
def _settle_slot(
|
||||
entry: dict,
|
||||
chosen: set[str],
|
||||
kept_system: Callable[[str], bool] | None = None,
|
||||
) -> dict:
|
||||
"""A platform file as a pack built for the chosen cores carries it.
|
||||
|
||||
Where a core's own source contradicts the platform and the frontend only
|
||||
@@ -2102,7 +2111,7 @@ def _partial_size(path: Path) -> int:
|
||||
return 0
|
||||
|
||||
|
||||
def _claim_partial(path: Path):
|
||||
def _claim_partial(path: Path) -> BinaryIO | None:
|
||||
"""The partial copy, opened and locked for this run, or None when
|
||||
another run holds it. A link planted at its name is not followed."""
|
||||
flags = os.O_RDWR | os.O_CREAT | getattr(os, "O_NOFOLLOW", 0) | getattr(os, "O_BINARY", 0)
|
||||
@@ -2125,13 +2134,13 @@ def _claim_partial(path: Path):
|
||||
class _Unclosed:
|
||||
"""The locked partial copy, written through without closing it."""
|
||||
|
||||
def __init__(self, handle) -> None:
|
||||
def __init__(self, handle: BinaryIO) -> None:
|
||||
self._handle = handle
|
||||
|
||||
def __enter__(self):
|
||||
def __enter__(self) -> BinaryIO:
|
||||
return self._handle
|
||||
|
||||
def __exit__(self, *_exc) -> None:
|
||||
def __exit__(self, *_exc: object) -> None:
|
||||
self._handle.flush()
|
||||
|
||||
def write(self, data: bytes) -> int:
|
||||
@@ -2882,7 +2891,7 @@ def main() -> None:
|
||||
sys.stdout = real_stdout
|
||||
|
||||
|
||||
def _install(args: argparse.Namespace, real_stdout) -> None:
|
||||
def _install(args: argparse.Namespace, real_stdout: TextIO) -> None:
|
||||
"""One run of the installer, once the options are read."""
|
||||
term = Terminal(real_stdout, quiet=args.quiet)
|
||||
interactive = sys.stdin.isatty() and not args.no_input and not args.quiet
|
||||
|
||||
+1
-1
@@ -18,7 +18,7 @@ esac
|
||||
TEMP_INSTALLER=""
|
||||
TEMP_DIRECTORY=""
|
||||
DEFAULT_INSTALL_URL="https://raw.githubusercontent.com/Abdess/retrobios/main/install.py"
|
||||
DEFAULT_INSTALL_SHA256="2e9e93d9809d9276a84209cfd421ad9b0d55853ce9c9d6761beb2ef21044ef2f"
|
||||
DEFAULT_INSTALL_SHA256="e18309beb7da0a974cff3ef9ca8019292e5798a19ae78fd4711b6b2cca88ebc6"
|
||||
MAX_INSTALLER_BYTES=2097152
|
||||
|
||||
cleanup() {
|
||||
|
||||
@@ -17,6 +17,7 @@ from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import hashlib
|
||||
import http.client
|
||||
import json
|
||||
import os
|
||||
import subprocess
|
||||
@@ -125,7 +126,9 @@ def verify_content(data: bytes, expected: dict) -> bool:
|
||||
return False
|
||||
|
||||
|
||||
def _read_limited(resp, limit: int = MAX_DOWNLOAD) -> bytes | None:
|
||||
def _read_limited(
|
||||
resp: http.client.HTTPResponse, limit: int = MAX_DOWNLOAD
|
||||
) -> bytes | None:
|
||||
"""Read response up to limit bytes. Returns None if exceeded."""
|
||||
data = resp.read(limit + 1)
|
||||
if len(data) > limit:
|
||||
@@ -394,7 +397,7 @@ def generate_issue_body(missing: list[dict], platform: str) -> str:
|
||||
return "\n".join(lines)
|
||||
|
||||
|
||||
def main():
|
||||
def main() -> None:
|
||||
parser = argparse.ArgumentParser(description="Auto-fetch missing BIOS files")
|
||||
parser.add_argument("--platform", "-p", help="Platform to check")
|
||||
parser.add_argument("--all", action="store_true", help="Check all platforms")
|
||||
|
||||
@@ -740,7 +740,7 @@ def print_report(report: dict, scope: str = "all emulators") -> None:
|
||||
print(f" {totals['missing']} missing (need to be sourced)")
|
||||
|
||||
|
||||
def main():
|
||||
def main() -> None:
|
||||
parser = argparse.ArgumentParser(description="Emulator vs platform gap analysis")
|
||||
parser.add_argument("--emulators-dir", default=DEFAULT_EMULATORS_DIR)
|
||||
parser.add_argument("--platforms-dir", default=DEFAULT_PLATFORMS_DIR)
|
||||
|
||||
+5
-3
@@ -185,7 +185,9 @@ def fetch_checksums(release: dict) -> dict[str, str]:
|
||||
return {}
|
||||
|
||||
|
||||
def download_file(url: str, dest: Path, expected_size: int = 0, label: str = ""):
|
||||
def download_file(
|
||||
url: str, dest: Path, expected_size: int = 0, label: str = ""
|
||||
) -> None:
|
||||
"""Download a file with progress indication."""
|
||||
url = _checked_url(url, "asset URL")
|
||||
req = urllib.request.Request(
|
||||
@@ -292,7 +294,7 @@ def fetch_pack(pack: Pack, staging: Path, checksums: dict[str, str]) -> list[Pat
|
||||
return volumes
|
||||
|
||||
|
||||
def show_info(platform: str, release: dict):
|
||||
def show_info(platform: str, release: dict) -> None:
|
||||
"""Show pack information for a platform."""
|
||||
pack = find_pack(release, platform)
|
||||
if not pack:
|
||||
@@ -307,7 +309,7 @@ def show_info(platform: str, release: dict):
|
||||
print(f" {part['name']} ({part.get('size', 0):,} bytes)")
|
||||
|
||||
|
||||
def main():
|
||||
def main() -> None:
|
||||
parser = argparse.ArgumentParser(
|
||||
description="Download BIOS packs from GitHub Releases",
|
||||
formatter_class=argparse.RawDescriptionHelpFormatter,
|
||||
|
||||
@@ -16,7 +16,7 @@ sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
|
||||
from scraper.dat_parser import parse_dat
|
||||
|
||||
from .base_exporter import BaseExporter
|
||||
from .baseline import NativeSystem, Report
|
||||
from .baseline import NativeFile, NativeSystem, Report
|
||||
|
||||
SOURCE_URL = (
|
||||
"https://raw.githubusercontent.com/libretro/libretro-database/master/dat/System.dat"
|
||||
@@ -73,7 +73,7 @@ class Exporter(BaseExporter):
|
||||
return {"System.dat": SOURCE_URL}
|
||||
|
||||
@classmethod
|
||||
def writable(cls, fe, require: str = "") -> bool:
|
||||
def writable(cls, fe: NativeFile, require: str = "") -> bool:
|
||||
"""A clrmamepro rom line is a hash record, and the DAT has no other.
|
||||
|
||||
The original never states a rom it cannot hash, so neither do we.
|
||||
@@ -83,7 +83,7 @@ class Exporter(BaseExporter):
|
||||
return any(fe.hash(h) for h in ("crc32", "md5", "sha1"))
|
||||
|
||||
@staticmethod
|
||||
def _rom_name(fe) -> str:
|
||||
def _rom_name(fe: NativeFile) -> str:
|
||||
"""The name the DAT gives a rom.
|
||||
|
||||
libretro writes the path for some entries (ep128emu/roms/cpc464.rom)
|
||||
|
||||
@@ -49,7 +49,9 @@ def cited_paths(document: dict) -> list[str]:
|
||||
return seen
|
||||
|
||||
|
||||
def audit(name: str, emulators_dir: str, cache_dir: str, offline: bool):
|
||||
def audit(
|
||||
name: str, emulators_dir: str, cache_dir: str, offline: bool
|
||||
) -> list[tuple[str, str]] | None:
|
||||
"""Signals found in the sources a fileless profile cites."""
|
||||
with open(os.path.join(emulators_dir, f"{name}.yml"), encoding="utf-8") as handle:
|
||||
document = yaml_load(handle) or {}
|
||||
|
||||
@@ -348,7 +348,7 @@ def _preserve_large_file_entries(files: dict, db_path: str) -> int:
|
||||
return count
|
||||
|
||||
|
||||
def main():
|
||||
def main() -> int:
|
||||
parser = argparse.ArgumentParser(description="Generate multi-indexed BIOS database")
|
||||
parser.add_argument("--force", action="store_true", help="Force rehash all files")
|
||||
parser.add_argument(
|
||||
@@ -462,7 +462,7 @@ def _collect_all_aliases(files: dict) -> dict:
|
||||
|
||||
aliases = {}
|
||||
|
||||
def _add_alias(name: str, matched_sha1: str):
|
||||
def _add_alias(name: str, matched_sha1: str) -> None:
|
||||
if not name or name in name_to_sha1:
|
||||
return
|
||||
if matched_sha1 not in aliases:
|
||||
|
||||
@@ -14,6 +14,7 @@ import argparse
|
||||
import json
|
||||
import os
|
||||
import sys
|
||||
from collections.abc import Callable
|
||||
from datetime import datetime, timezone
|
||||
from pathlib import Path
|
||||
|
||||
@@ -672,7 +673,9 @@ def generate_readme(db: dict, platforms_dir: str) -> str:
|
||||
return "\n".join(lines) + "\n"
|
||||
|
||||
|
||||
def contributing_text(link, title: str, site_words: str) -> str:
|
||||
def contributing_text(
|
||||
link: Callable[[str], str], title: str, site_words: str
|
||||
) -> str:
|
||||
"""The contribution guide, one text for the repository and the site.
|
||||
|
||||
Two copies drifted apart: four steps to add a platform here, five on the
|
||||
@@ -746,7 +749,7 @@ def generate_contributing() -> str:
|
||||
)
|
||||
|
||||
|
||||
def main():
|
||||
def main() -> None:
|
||||
parser = argparse.ArgumentParser(description="Generate slim README.md")
|
||||
parser.add_argument("--db", default="database.json")
|
||||
parser.add_argument("--platforms-dir", default="platforms")
|
||||
|
||||
@@ -25,6 +25,7 @@ import sys
|
||||
import urllib.error
|
||||
import urllib.parse
|
||||
import urllib.request
|
||||
from collections.abc import Callable
|
||||
from concurrent.futures import ThreadPoolExecutor
|
||||
from datetime import datetime, timezone
|
||||
from pathlib import Path
|
||||
@@ -465,7 +466,7 @@ def generate_stats(stats: dict) -> str:
|
||||
return json.dumps(stats, indent=2) + "\n"
|
||||
|
||||
|
||||
def _json_text(value) -> str:
|
||||
def _json_text(value: object) -> str:
|
||||
"""Stable scalar representation for CSV and SQLite exports."""
|
||||
if value is None:
|
||||
return ""
|
||||
@@ -477,7 +478,7 @@ def _json_text(value) -> str:
|
||||
HASH_ALGORITHMS = ("sha1", "sha256", "md5", "crc32")
|
||||
|
||||
|
||||
def _hash_values(value) -> list[str]:
|
||||
def _hash_values(value: object) -> list[str]:
|
||||
"""Every accepted value of one hash field, lowercase.
|
||||
|
||||
A field holds one value, a list of accepted revisions, or (Recalbox) a
|
||||
@@ -2091,7 +2092,7 @@ def _render_emulator_file(
|
||||
profile: dict,
|
||||
consumers: FileConsumers | None,
|
||||
files: list,
|
||||
_file_available,
|
||||
_file_available: Callable[[dict], bool],
|
||||
anchor: str = "",
|
||||
) -> list[str]:
|
||||
"""Render one row of an emulator profile's file table.
|
||||
@@ -2382,7 +2383,9 @@ def _hash_spans(value: str | list) -> str:
|
||||
)
|
||||
|
||||
|
||||
def _availability_check(db: dict, data_names):
|
||||
def _availability_check(
|
||||
db: dict, data_names: set[str] | None
|
||||
) -> Callable[[dict], bool]:
|
||||
"""A predicate answering whether the collection holds a file.
|
||||
|
||||
It answers the way cross_reference does. A page that called a file
|
||||
@@ -2419,7 +2422,9 @@ def _availability_check(db: dict, data_names):
|
||||
return _file_available
|
||||
|
||||
|
||||
def _emulator_file_summary(files: list[dict], available) -> list[str]:
|
||||
def _emulator_file_summary(
|
||||
files: list[dict], available: Callable[[dict], bool]
|
||||
) -> list[str]:
|
||||
"""The counts above an emulator's file table, and its categories.
|
||||
|
||||
Counted by slot, as verify --emulator and the gaps page count: raze
|
||||
@@ -3742,7 +3747,7 @@ def generate_mkdocs_nav(
|
||||
# Main
|
||||
|
||||
|
||||
def main():
|
||||
def main() -> None:
|
||||
parser = argparse.ArgumentParser(
|
||||
description="Generate MkDocs site from project data"
|
||||
)
|
||||
|
||||
@@ -29,7 +29,7 @@ def list_platforms(include_archived: bool = False) -> list[str]:
|
||||
return list_registered_platforms(PLATFORMS_DIR, include_archived=include_archived)
|
||||
|
||||
|
||||
def main():
|
||||
def main() -> None:
|
||||
parser = argparse.ArgumentParser(description="List available platforms")
|
||||
parser.add_argument("--all", action="store_true", help="Include archived platforms")
|
||||
args = parser.parse_args()
|
||||
|
||||
@@ -51,7 +51,9 @@ def declares(line: str, set_name: str) -> bool:
|
||||
return len(args) > 1 and args[1].strip() == set_name
|
||||
|
||||
|
||||
def audit(name: str, emulators_dir: str, cache_dir: str, offline: bool):
|
||||
def audit(
|
||||
name: str, emulators_dir: str, cache_dir: str, offline: bool
|
||||
) -> tuple[list[Finding], int, int]:
|
||||
"""Findings, plus counts of refs that agree and of unjudgeable ones."""
|
||||
with open(os.path.join(emulators_dir, f"{name}.yml"), encoding="utf-8") as handle:
|
||||
document = yaml_load(handle) or {}
|
||||
|
||||
+1
-1
@@ -491,7 +491,7 @@ def migrate_local(source: Path, target: Path, dry_run: bool) -> tuple:
|
||||
return moved, skipped, errors, existing_hashes
|
||||
|
||||
|
||||
def main():
|
||||
def main() -> None:
|
||||
parser = argparse.ArgumentParser(
|
||||
description="Migrate BIOS files to Manufacturer/Console structure"
|
||||
)
|
||||
|
||||
@@ -27,7 +27,7 @@ def set_offline(value: bool) -> None:
|
||||
global _OFFLINE
|
||||
_OFFLINE = value
|
||||
|
||||
def _values(raw) -> list[str]:
|
||||
def _values(raw: object) -> list[str]:
|
||||
"""A hash field's values: one, a list, or a comma-separated string."""
|
||||
items = raw if isinstance(raw, list) else [raw] if raw else []
|
||||
return [
|
||||
@@ -235,8 +235,8 @@ def preferred_entries(
|
||||
bios_dir: str,
|
||||
base_dest: str,
|
||||
required_only: bool,
|
||||
zip_contents,
|
||||
offline,
|
||||
zip_contents: dict | None,
|
||||
offline: bool | None,
|
||||
) -> dict[str, int]:
|
||||
"""Which declaration wins when several claim the same destination.
|
||||
|
||||
|
||||
@@ -495,7 +495,7 @@ def _intentional_hash_exclusion(
|
||||
return False
|
||||
return True
|
||||
|
||||
def _structural_errors(zf, zip_set: set) -> list[str]:
|
||||
def _structural_errors(zf: zipfile.ZipFile, zip_set: set) -> list[str]:
|
||||
"""Faults in the archive itself, before any file is looked up.
|
||||
|
||||
A duplicate entry, an absolute path or a traversal makes the pack
|
||||
|
||||
+4
-2
@@ -227,7 +227,9 @@ class _Skipped:
|
||||
SKIPPED = _Skipped()
|
||||
|
||||
|
||||
def _refuse_unapplied_flags(parser: argparse.ArgumentParser, args) -> None:
|
||||
def _refuse_unapplied_flags(
|
||||
parser: argparse.ArgumentParser, args: argparse.Namespace
|
||||
) -> None:
|
||||
"""A flag the run would drop is refused before any step.
|
||||
|
||||
--all-variants builds every source itself, so a --source beside it is
|
||||
@@ -311,7 +313,7 @@ def _truth_steps(args: argparse.Namespace, results: dict) -> bool:
|
||||
return all_ok
|
||||
|
||||
|
||||
def main():
|
||||
def main() -> None:
|
||||
parser = argparse.ArgumentParser(description="Run the full retrobios pipeline")
|
||||
parser.add_argument(
|
||||
"--include-archived", action="store_true", help="Include archived platforms"
|
||||
|
||||
+113
-47
@@ -20,9 +20,11 @@ import re
|
||||
import subprocess
|
||||
import sys
|
||||
import tempfile
|
||||
from collections.abc import Callable, Iterable, Iterator, Sequence
|
||||
from dataclasses import asdict, dataclass, field
|
||||
from datetime import date
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
import yaml
|
||||
|
||||
@@ -121,7 +123,7 @@ def parse_source_ref(ref: str) -> tuple[str, int | None, int | None]:
|
||||
return m.group("path"), start, end
|
||||
|
||||
|
||||
def source_ref_values(value) -> list[tuple[str, str]]:
|
||||
def source_ref_values(value: str | dict | None) -> list[tuple[str, str]]:
|
||||
"""Flatten a source_ref into (label, refs) pairs.
|
||||
|
||||
Most entries hold a plain string. `ymir` keys its refs by mode instead,
|
||||
@@ -334,7 +336,9 @@ def is_binary_citation(path: str) -> bool:
|
||||
return head.startswith(".") and "/" not in head
|
||||
|
||||
|
||||
def strip_repo_word(path: str, known=frozenset()) -> str:
|
||||
def strip_repo_word(
|
||||
path: str, known: frozenset[str] | set[str] = frozenset()
|
||||
) -> str:
|
||||
"""Drop a leading project word that names a declared repository.
|
||||
|
||||
`BAM_FPloader FPLoader.cpp:45` follows the `munt ROMInfo.cpp` convention,
|
||||
@@ -355,7 +359,7 @@ def strip_repo_word(path: str, known=frozenset()) -> str:
|
||||
return path
|
||||
|
||||
|
||||
def narrow_by_cited(matches: list[str], cited_dirs) -> list[str]:
|
||||
def narrow_by_cited(matches: list[str], cited_dirs: Iterable[str]) -> list[str]:
|
||||
"""Keep the candidates under the deepest directory the profile cites.
|
||||
|
||||
A bare `version.h` matches BasiliskII/src/include/version.h and
|
||||
@@ -482,7 +486,9 @@ class Citation:
|
||||
in_comment: bool = False
|
||||
|
||||
|
||||
def _walk_document(node, where: str = ""):
|
||||
def _walk_document(
|
||||
node: object, where: str = ""
|
||||
) -> Iterator[tuple[str, str, dict | list, object, Any]]:
|
||||
"""Every citation carrier of the document, with its holder and field path.
|
||||
|
||||
Yields ("ref", field, holder, key, value) for each source_ref key and
|
||||
@@ -509,7 +515,9 @@ def _walk_document(node, where: str = ""):
|
||||
yield from _walk_document(item, spot)
|
||||
|
||||
|
||||
def _prose_external(line: str, path_start: int, known=frozenset()) -> bool:
|
||||
def _prose_external(
|
||||
line: str, path_start: int, known: frozenset[str] | set[str] = frozenset()
|
||||
) -> bool:
|
||||
"""True when the word before the path names an undeclared project.
|
||||
|
||||
`munt ROMInfo.cpp:206-213` in prose follows the source_ref convention:
|
||||
@@ -533,7 +541,7 @@ def _prose_external(line: str, path_start: int, known=frozenset()) -> bool:
|
||||
return not lead or lead[-1] in ",:("
|
||||
|
||||
|
||||
def resolve_project_word(path: str, trees) -> str | None:
|
||||
def resolve_project_word(path: str, trees: Iterable[list[str] | None]) -> str | None:
|
||||
"""Path of `project file` inside a declared tree, None when out of reach.
|
||||
|
||||
m64plus-fz cites `rice Config.cpp` and `mupen64plus-core
|
||||
@@ -561,7 +569,11 @@ def resolve_project_word(path: str, trees) -> str | None:
|
||||
return None
|
||||
|
||||
|
||||
def _prose_runs(text: str, known=frozenset()):
|
||||
def _prose_runs(
|
||||
text: str, known: frozenset[str] | set[str] = frozenset()
|
||||
) -> Iterator[tuple[
|
||||
int, tuple[int, int], tuple[tuple[int, int], ...], tuple[RefPart, ...], str, bool
|
||||
]]:
|
||||
"""Citation runs inside one prose scalar.
|
||||
|
||||
A run is a path:range with its attached continuation ranges, written
|
||||
@@ -642,7 +654,7 @@ def _prose_runs(text: str, known=frozenset()):
|
||||
)
|
||||
|
||||
|
||||
def _scalar_spans(node, spans: list[tuple[int, int]]) -> None:
|
||||
def _scalar_spans(node: yaml.Node, spans: list[tuple[int, int]]) -> None:
|
||||
if isinstance(node, yaml.ScalarNode):
|
||||
spans.append((node.start_mark.index, node.end_mark.index))
|
||||
elif isinstance(node, yaml.MappingNode):
|
||||
@@ -654,7 +666,9 @@ def _scalar_spans(node, spans: list[tuple[int, int]]) -> None:
|
||||
_scalar_spans(item, spans)
|
||||
|
||||
|
||||
def comment_citations(text: str, known=frozenset()) -> list[Citation]:
|
||||
def comment_citations(
|
||||
text: str, known: frozenset[str] | set[str] = frozenset()
|
||||
) -> list[Citation]:
|
||||
"""Citation runs written in the YAML comments of a profile.
|
||||
|
||||
The loader drops comments, so a pin moved without them: zesarux cites
|
||||
@@ -741,7 +755,7 @@ def collect_citations(document: dict, text: str | None = None) -> list[Citation]
|
||||
return citations
|
||||
|
||||
|
||||
def worst_status(statuses) -> str:
|
||||
def worst_status(statuses: Iterable[str]) -> str:
|
||||
"""Severity of an entry is the worst severity among its parts."""
|
||||
worst = "ANCHORED"
|
||||
for status in statuses:
|
||||
@@ -767,7 +781,9 @@ def _find_all(haystack: list[str], needle: list[str]) -> list[int]:
|
||||
return hits
|
||||
|
||||
|
||||
def _map_index(opcodes, index: int) -> int | None:
|
||||
def _map_index(
|
||||
opcodes: list[tuple[str, int, int, int, int]], index: int
|
||||
) -> int | None:
|
||||
for tag, i1, i2, j1, j2 in opcodes:
|
||||
if i1 <= index < i2:
|
||||
return j1 + (index - i1) if tag == "equal" else j1
|
||||
@@ -853,7 +869,7 @@ NUDGE_WINDOW = 60
|
||||
|
||||
|
||||
def nudge_to_declared(
|
||||
pin_lines: list[str], start: int, end: int, tokens
|
||||
pin_lines: list[str], start: int, end: int, tokens: Sequence[str]
|
||||
) -> tuple[int, int] | None:
|
||||
"""Move a cited range onto the nearest line carrying a declared value.
|
||||
|
||||
@@ -880,7 +896,7 @@ def nudge_to_declared(
|
||||
|
||||
|
||||
def resolve_rename(
|
||||
result: CompareResult, path: str, head_paths=()
|
||||
result: CompareResult, path: str, head_paths: Iterable[str] = ()
|
||||
) -> tuple[str | None, list[str]]:
|
||||
"""New path of a moved file, or the candidates when several match.
|
||||
|
||||
@@ -935,7 +951,7 @@ def _narrow(matches: list[str], path: str) -> list[str]:
|
||||
|
||||
|
||||
def locate_by_declared_value(
|
||||
head_lines: list[str], anchored: AnchorResult, tokens
|
||||
head_lines: list[str], anchored: AnchorResult, tokens: Sequence[str]
|
||||
) -> tuple[int, int, str] | None:
|
||||
"""Follow a ref whose subject can still be pinpointed at HEAD.
|
||||
|
||||
@@ -964,7 +980,7 @@ def locate_by_declared_value(
|
||||
return None
|
||||
|
||||
|
||||
def _written_against_head(part, head_lines) -> bool:
|
||||
def _written_against_head(part: RefPart, head_lines: list[str] | None) -> bool:
|
||||
"""Whether the cited line exists at HEAD, which is where it was written."""
|
||||
return (
|
||||
head_lines is not None
|
||||
@@ -974,7 +990,11 @@ def _written_against_head(part, head_lines) -> bool:
|
||||
|
||||
|
||||
def anchor_part(
|
||||
part: RefPart, fetch, rename_getter, describe=None, tokens=()
|
||||
part: RefPart,
|
||||
fetch: Callable[[str, str, int | None, Sequence[str]], list[str] | None],
|
||||
rename_getter: Callable[[str], tuple[str | None, list[str]]],
|
||||
describe: Callable[..., tuple[str | None, str | None, str]] | None = None,
|
||||
tokens: Sequence[str] = (),
|
||||
) -> PartResult:
|
||||
"""Locate one reference part at HEAD, following a rename when needed.
|
||||
|
||||
@@ -1107,7 +1127,9 @@ def anchor_part(
|
||||
SHIFT_SUPPORT = 3
|
||||
|
||||
|
||||
def dominant_shifts(parts, support: int = SHIFT_SUPPORT) -> dict[str, int]:
|
||||
def dominant_shifts(
|
||||
parts: Iterable[PartResult], support: int = SHIFT_SUPPORT
|
||||
) -> dict[str, int]:
|
||||
"""Most common line shift per file, taken from the unambiguous anchors.
|
||||
|
||||
A file whose refs mostly moved by the same amount gives strong evidence
|
||||
@@ -1210,8 +1232,8 @@ def select_repo(profile: dict) -> upstream.Repo | None:
|
||||
|
||||
|
||||
def resolve_pin(
|
||||
profile: dict, repo, cache_dir: str, offline: bool, field: str = "source",
|
||||
mode: str = "", branch: str | None = None,
|
||||
profile: dict, repo: upstream.Repo, cache_dir: str, offline: bool,
|
||||
field: str = "source", mode: str = "", branch: str | None = None,
|
||||
) -> tuple[str | None, str | None]:
|
||||
"""Commit the profile was written at, and how it was obtained.
|
||||
|
||||
@@ -1309,7 +1331,12 @@ def branch_in_url(url: str) -> str | None:
|
||||
SELF_CHECK_CONTEXT = 2
|
||||
|
||||
|
||||
def verify_at_pin(part: RefPart, pin_lines, tokens, hash_tokens=()) -> PartResult:
|
||||
def verify_at_pin(
|
||||
part: RefPart,
|
||||
pin_lines: list[str] | None,
|
||||
tokens: Sequence[str],
|
||||
hash_tokens: Sequence[str] = (),
|
||||
) -> PartResult:
|
||||
"""Check a ref against its own revision instead of against HEAD.
|
||||
|
||||
A profile pinned to a superseded tag documents a program HEAD no longer
|
||||
@@ -1431,7 +1458,7 @@ def version_tag_candidates(core_version: str, repo_name: str = "") -> list[str]:
|
||||
|
||||
|
||||
def detect_pinned_tag(
|
||||
profile: dict, views, cache_dir: str, offline: bool
|
||||
profile: dict, views: list[RepoView], cache_dir: str, offline: bool
|
||||
) -> str | None:
|
||||
"""Tag the profile is pinned to, when the repository has moved past it.
|
||||
|
||||
@@ -1456,7 +1483,9 @@ def detect_pinned_tag(
|
||||
return None
|
||||
|
||||
|
||||
def _resolve_project_words(refs: list, tree_order) -> list:
|
||||
def _resolve_project_words(
|
||||
refs: list, tree_order: Callable[[], Iterable[list[str] | None]]
|
||||
) -> list:
|
||||
"""Point `project file` parts at the declared tree that vendors them.
|
||||
|
||||
tree_order() yields the trees to search, pin before HEAD; it is only
|
||||
@@ -1656,7 +1685,7 @@ def build_report(
|
||||
if len(matches) <= 4:
|
||||
candidates.extend(m for m in matches if m not in candidates)
|
||||
|
||||
def score(lines) -> int:
|
||||
def score(lines: list[str]) -> int:
|
||||
"""How well a repository's cited line matches what the ref means.
|
||||
|
||||
Two repositories can hold the same path with different contents:
|
||||
@@ -1709,7 +1738,7 @@ def build_report(
|
||||
owners[key] = _locate(path, start, tokens)
|
||||
return owners[key]
|
||||
|
||||
def _pin_tree_for(view: RepoView):
|
||||
def _pin_tree_for(view: RepoView) -> list[str] | None:
|
||||
"""Tree at the revision the refs were written against, memoised."""
|
||||
key = view.repo.slug
|
||||
if key not in pin_trees:
|
||||
@@ -1719,7 +1748,7 @@ def build_report(
|
||||
|
||||
mute: set[tuple[str, str]] = set()
|
||||
|
||||
def fetch_from(repo, sha, wanted):
|
||||
def fetch_from(repo: upstream.Repo, sha: str, wanted: str) -> list[str] | None:
|
||||
"""Read a file, treating a repository that refuses as one that lacks it.
|
||||
|
||||
A profile can name several repositories, and a mirror exists exactly
|
||||
@@ -1744,7 +1773,7 @@ def build_report(
|
||||
print(f"{name}: {repo.host}/{repo.slug} muted, {exc}", file=sys.stderr)
|
||||
return None
|
||||
|
||||
def _context_for(view: RepoView):
|
||||
def _context_for(view: RepoView) -> tuple[CompareResult, list[str]]:
|
||||
key = view.repo.slug
|
||||
if key not in context:
|
||||
comparison = upstream.compare(
|
||||
@@ -1793,7 +1822,13 @@ def build_report(
|
||||
|
||||
lines_cache: dict[tuple[str, str, int | None], list[str] | None] = {}
|
||||
|
||||
def fetch(which: str, path: str, start: int | None = None, tokens=(), forced=None):
|
||||
def fetch(
|
||||
which: str,
|
||||
path: str,
|
||||
start: int | None = None,
|
||||
tokens: Iterable[str] = (),
|
||||
forced: RepoView | None = None,
|
||||
) -> list[str] | None:
|
||||
"""Lines for one path at one revision, read once per run.
|
||||
|
||||
The same object is handed back every time so the opcode cache can key
|
||||
@@ -1816,14 +1851,19 @@ def build_report(
|
||||
lines_cache[key] = found
|
||||
return lines_cache[key]
|
||||
|
||||
def describe(path: str, start=None, tokens=(), forced=None):
|
||||
def describe(
|
||||
path: str,
|
||||
start: int | None = None,
|
||||
tokens: Iterable[str] = (),
|
||||
forced: RepoView | None = None,
|
||||
) -> tuple[str | None, str, str]:
|
||||
view, actual = (
|
||||
(forced, path) if forced else resolve_path(path, start, tuple(tokens))
|
||||
)
|
||||
slug = view.repo.slug if view is not primary else None
|
||||
return slug, upstream.raw_url(view.repo, view.head, actual), actual
|
||||
|
||||
def anchor_across_views(part, tokens):
|
||||
def anchor_across_views(part: RefPart, tokens: Sequence[str]) -> PartResult:
|
||||
"""Judge a part against every repository that carries its path.
|
||||
|
||||
A ref is only broken when it fails everywhere. `mame` cites files that
|
||||
@@ -1875,13 +1915,15 @@ def build_report(
|
||||
# a part whose repository already sits on HEAD is judged on
|
||||
# self-consistency instead. Decided per part, by the repository that owns
|
||||
# it: a primary on HEAD says nothing about an upstream that moved on.
|
||||
def self_checked(part, tokens) -> bool:
|
||||
def self_checked(part: RefPart, tokens: Sequence[str]) -> bool:
|
||||
if report.pinned_tag:
|
||||
return True
|
||||
view, _ = resolve_path(part.path, part.start, tuple(tokens))
|
||||
return view.pin == view.head
|
||||
|
||||
def judge(part, flag, tokens, hashes) -> PartResult:
|
||||
def judge(
|
||||
part: RefPart, flag: bool, tokens: Sequence[str], hashes: Sequence[str]
|
||||
) -> PartResult:
|
||||
other = cites_another_version(part, profile.get("core_version"))
|
||||
if other:
|
||||
# ti99sim cites v0.0.11 beside its v0.16.0 pin: no revision of
|
||||
@@ -2076,7 +2118,7 @@ class VersionReport:
|
||||
|
||||
|
||||
def check_version(
|
||||
profile: dict, repo, cache_dir: str, offline: bool
|
||||
profile: dict, repo: upstream.Repo, cache_dir: str, offline: bool
|
||||
) -> VersionReport | None:
|
||||
"""Declared core_version against the latest upstream tag and release."""
|
||||
declared = str(profile.get("core_version") or "")
|
||||
@@ -2345,7 +2387,7 @@ def _match_citation(
|
||||
|
||||
|
||||
def _prose_moves(
|
||||
entry: EntryReport, statuses
|
||||
entry: EntryReport, statuses: tuple[str, ...]
|
||||
) -> dict[int, tuple[int, int, str | None]]:
|
||||
"""Recale targets per part of one prose run, empty when unsafe.
|
||||
|
||||
@@ -2372,7 +2414,9 @@ def _prose_moves(
|
||||
|
||||
|
||||
def _run_after_moves(
|
||||
parts, moves: dict[int, tuple[int, int, str | None]], run: str = ""
|
||||
parts: Sequence[RefPart],
|
||||
moves: dict[int, tuple[int, int, str | None]],
|
||||
run: str = "",
|
||||
) -> str:
|
||||
"""The prose run as it reads once its moved ranges are rewritten.
|
||||
|
||||
@@ -2837,7 +2881,7 @@ def _writing_pins(
|
||||
|
||||
|
||||
def _resolve_at(
|
||||
repo, sha: str, path: str, cache_dir: str, offline: bool
|
||||
repo: upstream.Repo, sha: str, path: str, cache_dir: str, offline: bool
|
||||
) -> tuple[str, object]:
|
||||
"""("found", (path, lines)) or ("unclear"|"missing", reason).
|
||||
|
||||
@@ -2872,7 +2916,11 @@ def _resolve_at(
|
||||
|
||||
|
||||
def _realign_part(
|
||||
part: RefPart, pairs, cache_dir: str, offline: bool, pinned=()
|
||||
part: RefPart,
|
||||
pairs: list[tuple[upstream.Repo, str, str]],
|
||||
cache_dir: str,
|
||||
offline: bool,
|
||||
pinned: Iterable[tuple[upstream.Repo, str]] = (),
|
||||
) -> tuple[str, object] | None:
|
||||
"""One range, anchored from its writing revision to the current pin.
|
||||
|
||||
@@ -2953,7 +3001,7 @@ def _realign_citation(
|
||||
ambiguous: list[str],
|
||||
cache_dir: str,
|
||||
offline: bool,
|
||||
pinned=(),
|
||||
pinned: Iterable[tuple[upstream.Repo, str]] = (),
|
||||
) -> tuple[dict[int, tuple[int, int, str | None]], list[str]]:
|
||||
"""The moves one prose run needs, and what blocks them.
|
||||
|
||||
@@ -3195,7 +3243,7 @@ def build_parser() -> argparse.ArgumentParser:
|
||||
return parser
|
||||
|
||||
|
||||
def select_profiles(profiles: dict, args) -> dict:
|
||||
def select_profiles(profiles: dict, args: argparse.Namespace) -> dict:
|
||||
"""Profiles matching the selection flags."""
|
||||
if args.emulator:
|
||||
if args.emulator not in profiles:
|
||||
@@ -3234,7 +3282,7 @@ def _check_quota(count: int, offline: bool) -> None:
|
||||
raise SystemExit(1)
|
||||
|
||||
|
||||
def _stale_prose_blocks(name: str, path: Path, args) -> bool:
|
||||
def _stale_prose_blocks(name: str, path: Path, args: argparse.Namespace) -> bool:
|
||||
"""Whether prose written under an older pin has to move first.
|
||||
|
||||
Anchored from the current pin, a recale or a bump follows someone else's
|
||||
@@ -3251,7 +3299,10 @@ def _stale_prose_blocks(name: str, path: Path, args) -> bool:
|
||||
return bool(stale)
|
||||
|
||||
|
||||
def _backfill(args, name: str, profile: dict, report: ProfileReport, path: Path) -> None:
|
||||
def _backfill(
|
||||
args: argparse.Namespace, name: str, profile: dict, report: ProfileReport,
|
||||
path: Path,
|
||||
) -> None:
|
||||
"""Write the pin a profile without source_commit was written at."""
|
||||
if not report.pin or profile.get("source_commit"):
|
||||
return
|
||||
@@ -3261,7 +3312,9 @@ def _backfill(args, name: str, profile: dict, report: ProfileReport, path: Path)
|
||||
print(f"{name}: source_commit {report.pin[:7]}")
|
||||
|
||||
|
||||
def _apply_writes(args, name: str, profile: dict, report: ProfileReport) -> None:
|
||||
def _apply_writes(
|
||||
args: argparse.Namespace, name: str, profile: dict, report: ProfileReport
|
||||
) -> None:
|
||||
path = Path(args.emulators_dir) / f"{name}.yml"
|
||||
if not path.is_file():
|
||||
return
|
||||
@@ -3308,7 +3361,10 @@ def _cited_paths(report: ProfileReport) -> list[str]:
|
||||
)
|
||||
|
||||
|
||||
def _print_version(args, profile: dict, report: ProfileReport, repo) -> None:
|
||||
def _print_version(
|
||||
args: argparse.Namespace, profile: dict, report: ProfileReport,
|
||||
repo: upstream.Repo,
|
||||
) -> None:
|
||||
version = check_version(profile, repo, args.cache_dir, args.offline)
|
||||
if version is None:
|
||||
return
|
||||
@@ -3322,7 +3378,10 @@ def _print_version(args, profile: dict, report: ProfileReport, repo) -> None:
|
||||
print(f" {warning}")
|
||||
|
||||
|
||||
def _print_detection(args, profile: dict, report: ProfileReport, repo) -> None:
|
||||
def _print_detection(
|
||||
args: argparse.Namespace, profile: dict, report: ProfileReport,
|
||||
repo: upstream.Repo,
|
||||
) -> None:
|
||||
names, hashes = declared_names(profile), declared_hashes(profile)
|
||||
for path in _cited_paths(report):
|
||||
head_lines = upstream.fetch_file(
|
||||
@@ -3352,7 +3411,9 @@ def _print_detection(args, profile: dict, report: ProfileReport, repo) -> None:
|
||||
print(f" new {kind}: {path} {value}")
|
||||
|
||||
|
||||
def _print_extras(args, profile: dict, report: ProfileReport) -> None:
|
||||
def _print_extras(
|
||||
args: argparse.Namespace, profile: dict, report: ProfileReport
|
||||
) -> None:
|
||||
"""Optional per-profile sections beyond the ref report.
|
||||
|
||||
The ref report mutes a forge that refuses; the extras read the same
|
||||
@@ -3380,7 +3441,10 @@ def _print_extras(args, profile: dict, report: ProfileReport) -> None:
|
||||
print(f" extras skipped: {exc}")
|
||||
|
||||
|
||||
def _print_extras_from(args, profile: dict, report: ProfileReport, repo) -> None:
|
||||
def _print_extras_from(
|
||||
args: argparse.Namespace, profile: dict, report: ProfileReport,
|
||||
repo: upstream.Repo,
|
||||
) -> None:
|
||||
if args.check_version:
|
||||
_print_version(args, profile, report, repo)
|
||||
if args.detect_new_files or args.watch_hashes:
|
||||
@@ -3415,7 +3479,9 @@ def _print_extras_from(args, profile: dict, report: ProfileReport, repo) -> None
|
||||
print(f" {line}")
|
||||
|
||||
|
||||
def _print_triage(args, selected: dict, reports: list[ProfileReport]) -> None:
|
||||
def _print_triage(
|
||||
args: argparse.Namespace, selected: dict, reports: list[ProfileReport]
|
||||
) -> None:
|
||||
"""Profiles ranked by drift, worst first."""
|
||||
ranked = []
|
||||
for report in reports:
|
||||
@@ -3455,7 +3521,7 @@ def _print_triage(args, selected: dict, reports: list[ProfileReport]) -> None:
|
||||
print(f"{score:5d} {report.name:30s} {state}")
|
||||
|
||||
|
||||
def _refuse_flag_combinations(args) -> None:
|
||||
def _refuse_flag_combinations(args: argparse.Namespace) -> None:
|
||||
"""Exit on a flag the chosen mode would not apply, before any request."""
|
||||
if args.accept_changed and not args.emulator:
|
||||
print(
|
||||
|
||||
@@ -52,7 +52,9 @@ def load_registry(registry_path: str = DEFAULT_REGISTRY) -> dict[str, dict]:
|
||||
return data.get("data_directories", {})
|
||||
|
||||
|
||||
def cache_lock(cache_dir: str | Path, shared: bool = False):
|
||||
def cache_lock(
|
||||
cache_dir: str | Path, shared: bool = False
|
||||
) -> contextlib.AbstractContextManager[None]:
|
||||
"""The lock a refresh holds while it swaps cache_dir.
|
||||
|
||||
A reader takes it shared: a pack walking data/sdlpal while another run
|
||||
|
||||
@@ -121,7 +121,7 @@ class AtomPool:
|
||||
hundreds of megabytes and a whole-pool preload buys nothing.
|
||||
"""
|
||||
|
||||
def __init__(self, bios_dir: Path, digest_cache: Path | None = None):
|
||||
def __init__(self, bios_dir: Path, digest_cache: Path | None = None) -> None:
|
||||
self._index: dict[str, tuple[Path, str]] = {}
|
||||
# (md5, size) of each build, kept across runs: TorrentZip bytes are a
|
||||
# function of the members alone, so a recipe hashed once is known
|
||||
|
||||
@@ -6,6 +6,8 @@ the parsers here are the boundary that keeps it that way."""
|
||||
from __future__ import annotations
|
||||
|
||||
import xml.etree.ElementTree as ET
|
||||
from types import ModuleType
|
||||
from typing import Any, TextIO
|
||||
|
||||
try:
|
||||
import yaml
|
||||
@@ -37,7 +39,7 @@ def parse_untrusted_xml(
|
||||
return ET.fromstring(content, parser=parser)
|
||||
return ET.fromstring(content)
|
||||
|
||||
def require_yaml():
|
||||
def require_yaml() -> ModuleType:
|
||||
"""Import and return yaml, exiting if PyYAML is not installed."""
|
||||
try:
|
||||
import yaml as _yaml
|
||||
@@ -49,7 +51,7 @@ def require_yaml():
|
||||
print("Error: PyYAML required (pip install pyyaml)", file=sys.stderr)
|
||||
sys.exit(1)
|
||||
|
||||
def _pick_yaml_loader():
|
||||
def _pick_yaml_loader() -> type[yaml.constructor.SafeConstructor] | None:
|
||||
"""Prefer the libyaml loader when the wheel ships it.
|
||||
|
||||
Reading the emulator profiles is the single most expensive step of every
|
||||
@@ -71,7 +73,9 @@ def _pick_yaml_loader():
|
||||
here uses merge keys (<<), which would count as repeats too.
|
||||
"""
|
||||
|
||||
def construct_mapping(self, node, deep=False):
|
||||
def construct_mapping(
|
||||
self, node: yaml.MappingNode, deep: bool = False
|
||||
) -> dict:
|
||||
mapping = super().construct_mapping(node, deep=deep)
|
||||
if len(mapping) < len(node.value):
|
||||
seen: set = set()
|
||||
@@ -88,7 +92,7 @@ def _pick_yaml_loader():
|
||||
|
||||
_YAML_LOADER = _pick_yaml_loader()
|
||||
|
||||
def yaml_load(stream):
|
||||
def yaml_load(stream: str | TextIO) -> Any:
|
||||
"""Parse YAML from a string or file object, safely and quickly."""
|
||||
if yaml is None:
|
||||
return require_yaml() # exits with the install hint
|
||||
|
||||
@@ -537,7 +537,7 @@ def _append_new_entries(text: str, files: list[dict], original: str) -> str:
|
||||
return text
|
||||
|
||||
|
||||
def _yaml_scalar(value) -> str:
|
||||
def _yaml_scalar(value: object) -> str:
|
||||
"""Quote a scalar so YAML reads back exactly what the source said.
|
||||
|
||||
A MAME ROM named ``01`` written bare re-parses as the integer 1, and the
|
||||
|
||||
@@ -163,7 +163,7 @@ def fetch_bytes(
|
||||
class BaseScraper(ABC):
|
||||
"""Abstract base class for platform BIOS requirement scrapers."""
|
||||
|
||||
def __init__(self, url: str = ""):
|
||||
def __init__(self, url: str = "") -> None:
|
||||
self.url = url
|
||||
self._raw_data: str | None = None
|
||||
|
||||
|
||||
@@ -249,7 +249,7 @@ def _resolve_truncated_md5(md5: str, md5_index: dict[str, str]) -> str:
|
||||
class Scraper(BaseScraper):
|
||||
"""Scraper for batocera-systems Python dict."""
|
||||
|
||||
def __init__(self):
|
||||
def __init__(self) -> None:
|
||||
self.tag = fetch_stable_tag()
|
||||
super().__init__(url=SOURCE_URL.format(tag=self.tag))
|
||||
|
||||
@@ -466,7 +466,7 @@ class Scraper(BaseScraper):
|
||||
return result
|
||||
|
||||
|
||||
def main():
|
||||
def main() -> None:
|
||||
from scripts.scraper.base_scraper import scraper_cli
|
||||
|
||||
scraper_cli(Scraper, "Scrape batocera BIOS requirements")
|
||||
|
||||
@@ -331,7 +331,7 @@ def parse_firmware_database(
|
||||
class Scraper(BaseScraper):
|
||||
"""BizHawk firmware database scraper."""
|
||||
|
||||
def __init__(self):
|
||||
def __init__(self) -> None:
|
||||
self.tag = fetch_github_latest_version(GITHUB_REPO)
|
||||
super().__init__(url=SOURCE_URL.format(tag=self.tag))
|
||||
|
||||
@@ -386,7 +386,7 @@ class Scraper(BaseScraper):
|
||||
}
|
||||
|
||||
|
||||
def main():
|
||||
def main() -> None:
|
||||
scraper_cli(Scraper, "Scrape BizHawk BIOS requirements")
|
||||
|
||||
|
||||
|
||||
@@ -206,7 +206,7 @@ def _extract_md5_from_notes(info: dict) -> dict[str, str]:
|
||||
class Scraper(BaseScraper):
|
||||
"""Scraper for libretro-core-info firmware declarations."""
|
||||
|
||||
def __init__(self):
|
||||
def __init__(self) -> None:
|
||||
super().__init__()
|
||||
self._info_files: dict[str, dict] | None = None
|
||||
|
||||
@@ -303,7 +303,7 @@ class Scraper(BaseScraper):
|
||||
return {"version": version}
|
||||
|
||||
|
||||
def main():
|
||||
def main() -> None:
|
||||
"""CLI entry point."""
|
||||
import argparse
|
||||
|
||||
|
||||
@@ -16,6 +16,7 @@ from __future__ import annotations
|
||||
import argparse
|
||||
import sys
|
||||
import zipfile
|
||||
from collections.abc import Iterator
|
||||
from datetime import datetime, timezone
|
||||
from pathlib import Path
|
||||
|
||||
@@ -50,7 +51,7 @@ def _bios_entries(source: str, dat: LogiqxDat) -> list[dict]:
|
||||
]
|
||||
|
||||
|
||||
def _iter_dat_contents(pack: Path):
|
||||
def _iter_dat_contents(pack: Path) -> Iterator[tuple[str, bytes]]:
|
||||
"""Yield (member name, content bytes) for DAT files in a pack.
|
||||
|
||||
Accepts a ZIP archive or a directory of extracted DATs.
|
||||
|
||||
@@ -221,7 +221,7 @@ class Scraper(BaseScraper):
|
||||
|
||||
def __init__(
|
||||
self, checkbios_url: str = CHECKBIOS_URL, csv_base_url: str = CSV_BASE_URL
|
||||
):
|
||||
) -> None:
|
||||
super().__init__(url=checkbios_url)
|
||||
self.checkbios_url = checkbios_url
|
||||
self.csv_base_url = csv_base_url
|
||||
@@ -492,7 +492,7 @@ class Scraper(BaseScraper):
|
||||
return sorted(cores)
|
||||
|
||||
|
||||
def main():
|
||||
def main() -> None:
|
||||
from scripts.scraper.base_scraper import scraper_cli
|
||||
|
||||
scraper_cli(Scraper, "Scrape emudeck BIOS requirements")
|
||||
|
||||
@@ -125,7 +125,7 @@ def _core_info_files() -> list[str]:
|
||||
class Scraper(BaseScraper):
|
||||
"""Scraper for libretro System.dat."""
|
||||
|
||||
def __init__(self, url: str = SOURCE_URL):
|
||||
def __init__(self, url: str = SOURCE_URL) -> None:
|
||||
super().__init__(url=url)
|
||||
|
||||
def fetch_requirements(self) -> list[BiosRequirement]:
|
||||
@@ -516,7 +516,7 @@ class Scraper(BaseScraper):
|
||||
}
|
||||
|
||||
|
||||
def main():
|
||||
def main() -> None:
|
||||
from scripts.scraper.base_scraper import scraper_cli
|
||||
|
||||
scraper_cli(Scraper, "Scrape libretro BIOS requirements")
|
||||
|
||||
@@ -126,7 +126,7 @@ def system_of(path: str) -> str:
|
||||
class Scraper(BaseScraper):
|
||||
"""Scraper for the MiSTer BiosDB database."""
|
||||
|
||||
def __init__(self, url: str = SOURCE_URL):
|
||||
def __init__(self, url: str = SOURCE_URL) -> None:
|
||||
super().__init__(url=url)
|
||||
|
||||
def _fetch_raw(self) -> str:
|
||||
|
||||
@@ -199,7 +199,7 @@ def board_cores(text: str, board: str, enabled: set[str]) -> set[str]:
|
||||
class Scraper(BaseScraper):
|
||||
"""Scraper for Recalbox es_bios.xml."""
|
||||
|
||||
def __init__(self, tag: str | None = None):
|
||||
def __init__(self, tag: str | None = None) -> None:
|
||||
# A target scraper passes the tag platforms/recalbox.yml pins.
|
||||
self.tag = tag or _fetch_gitlab_stable_tag()
|
||||
super().__init__(url=SOURCE_URL.format(tag=self.tag))
|
||||
@@ -365,7 +365,7 @@ class Scraper(BaseScraper):
|
||||
}
|
||||
|
||||
|
||||
def main():
|
||||
def main() -> None:
|
||||
"""CLI entry point."""
|
||||
from .base_scraper import scraper_cli
|
||||
|
||||
|
||||
@@ -88,7 +88,7 @@ SYSTEM_SLUG_MAP = {
|
||||
class Scraper(BaseScraper):
|
||||
"""Scraper for RetroBat batocera-systems.json."""
|
||||
|
||||
def __init__(self, url: str = SOURCE_URL):
|
||||
def __init__(self, url: str = SOURCE_URL) -> None:
|
||||
super().__init__(url=url)
|
||||
self._parsed: dict | None = None
|
||||
|
||||
@@ -227,7 +227,7 @@ class Scraper(BaseScraper):
|
||||
}
|
||||
|
||||
|
||||
def main():
|
||||
def main() -> None:
|
||||
from scripts.scraper.base_scraper import scraper_cli
|
||||
|
||||
scraper_cli(Scraper, "Scrape retrobat BIOS requirements")
|
||||
|
||||
@@ -51,7 +51,7 @@ _PACKAGE_DIRS = ("emulators", "libretrocores", "ports")
|
||||
class Scraper(BaseScraper):
|
||||
"""Scraper for the RetroPie package list."""
|
||||
|
||||
def __init__(self, url: str = SOURCE_URL):
|
||||
def __init__(self, url: str = SOURCE_URL) -> None:
|
||||
super().__init__(url=url)
|
||||
self._modules: list[str] | None = None
|
||||
self._version = ""
|
||||
|
||||
@@ -147,7 +147,7 @@ def parse_systems(source: str) -> dict[str, dict]:
|
||||
class Scraper(BaseScraper):
|
||||
"""Scraper for the ROCKNIX rocknix-systems script."""
|
||||
|
||||
def __init__(self, url: str | None = None):
|
||||
def __init__(self, url: str | None = None) -> None:
|
||||
self.tag = None if url else fetch_github_latest_version(GITHUB_REPO)
|
||||
super().__init__(url=url or SOURCE_TEMPLATE.format(ref=self.tag))
|
||||
|
||||
|
||||
@@ -122,7 +122,7 @@ FIRMWARE_MIRRORS: dict[str, tuple[str, ...]] = {
|
||||
class Scraper(BaseScraper):
|
||||
"""Scraper for RomM known_bios_files.json."""
|
||||
|
||||
def __init__(self):
|
||||
def __init__(self) -> None:
|
||||
self.tag = fetch_github_latest_version(GITHUB_REPO)
|
||||
super().__init__(url=SOURCE_URL.format(tag=self.tag))
|
||||
self._parsed: dict | None = None
|
||||
@@ -225,7 +225,7 @@ class Scraper(BaseScraper):
|
||||
}
|
||||
|
||||
|
||||
def main():
|
||||
def main() -> None:
|
||||
from scripts.scraper.base_scraper import scraper_cli
|
||||
|
||||
scraper_cli(Scraper, "Scrape RomM BIOS requirements")
|
||||
|
||||
@@ -25,6 +25,7 @@ from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import hashlib
|
||||
import io
|
||||
import json
|
||||
import os
|
||||
import re
|
||||
@@ -32,6 +33,7 @@ import shutil
|
||||
import sys
|
||||
import urllib.request
|
||||
import zipfile
|
||||
from collections.abc import Callable, Iterator
|
||||
from datetime import datetime, timezone
|
||||
from pathlib import Path
|
||||
|
||||
@@ -146,7 +148,9 @@ def _machine_roms(blob: bytes) -> list[dict]:
|
||||
return members
|
||||
|
||||
|
||||
def _scan_listxml(stream, selector) -> dict[str, tuple[str | None, list[dict]]]:
|
||||
def _scan_listxml(
|
||||
stream: io.BufferedIOBase, selector: Callable[[str], bool]
|
||||
) -> dict[str, tuple[str | None, list[dict]]]:
|
||||
"""Stream a MAME -listxml document, keeping the machines *selector* wants.
|
||||
|
||||
The document is 311 MB for MAME 0.289, so it is never held whole: the
|
||||
@@ -178,7 +182,7 @@ def _scan_listxml(stream, selector) -> dict[str, tuple[str | None, list[dict]]]:
|
||||
|
||||
|
||||
def listxml_entries(
|
||||
open_stream, label: str, keep: set[str] | None
|
||||
open_stream: Callable[[], io.BufferedIOBase], label: str, keep: set[str] | None
|
||||
) -> list[dict]:
|
||||
"""Recipes from a MAME -listxml document, parents resolved.
|
||||
|
||||
@@ -291,7 +295,7 @@ def merge_snapshot(
|
||||
return write_snapshot(output, snapshot)
|
||||
|
||||
|
||||
def _iter_dat_contents(pack: Path):
|
||||
def _iter_dat_contents(pack: Path) -> Iterator[tuple[str, bytes]]:
|
||||
"""Yield (member name, content bytes) for DAT files in *pack*."""
|
||||
if pack.is_dir():
|
||||
for path in sorted(pack.rglob("*")):
|
||||
@@ -424,10 +428,10 @@ def _is_listxml(pack: Path) -> bool:
|
||||
return b"<!ELEMENT mame" in handle.read(4096)
|
||||
|
||||
|
||||
def _listxml_opener(pack: Path):
|
||||
def _listxml_opener(pack: Path) -> Callable[[], io.BufferedIOBase]:
|
||||
"""A callable giving a fresh byte stream over the listxml document."""
|
||||
if pack.suffix.lower() == ".zip":
|
||||
def opener():
|
||||
def opener() -> zipfile.ZipExtFile:
|
||||
archive = zipfile.ZipFile(pack)
|
||||
name = next(n for n in archive.namelist() if n.lower().endswith(".xml"))
|
||||
stream = archive.open(name)
|
||||
|
||||
@@ -15,7 +15,7 @@ from pathlib import Path
|
||||
class BaseTargetScraper(ABC):
|
||||
"""Base class for target scrapers."""
|
||||
|
||||
def __init__(self, url: str = ""):
|
||||
def __init__(self, url: str = "") -> None:
|
||||
self.url = url
|
||||
|
||||
@abstractmethod
|
||||
|
||||
@@ -292,7 +292,9 @@ def _arch_from_flag(flag: str) -> str:
|
||||
class Scraper(BaseTargetScraper):
|
||||
"""Cross-references Batocera boards, Config.in, and es_systems to build target lists."""
|
||||
|
||||
def __init__(self, url: str = "https://github.com/batocera-linux/batocera.linux"):
|
||||
def __init__(
|
||||
self, url: str = "https://github.com/batocera-linux/batocera.linux"
|
||||
) -> None:
|
||||
super().__init__(url=url)
|
||||
self.tag = ""
|
||||
|
||||
|
||||
@@ -90,7 +90,7 @@ def _script_to_core(filename: str) -> str | None:
|
||||
class Scraper(BaseTargetScraper):
|
||||
"""Fetches emulator lists for EmuDeck SteamOS and Windows targets."""
|
||||
|
||||
def __init__(self, url: str = "https://github.com/dragoonDorise/EmuDeck"):
|
||||
def __init__(self, url: str = "https://github.com/dragoonDorise/EmuDeck") -> None:
|
||||
super().__init__(url=url)
|
||||
|
||||
def _fetch_cores_for_target(
|
||||
|
||||
@@ -58,7 +58,7 @@ def defconfig_arch(defconfig: str) -> str:
|
||||
class Scraper(BaseTargetScraper):
|
||||
"""Per-board core lists from Recalbox's defconfigs and system.ini files."""
|
||||
|
||||
def __init__(self, url: str = "https://gitlab.com/recalbox/recalbox"):
|
||||
def __init__(self, url: str = "https://gitlab.com/recalbox/recalbox") -> None:
|
||||
super().__init__(url=url)
|
||||
|
||||
def fetch_targets(self) -> dict:
|
||||
|
||||
@@ -86,7 +86,7 @@ _CORE_NAME_RE = re.compile(r"^(.+?)_libretro")
|
||||
class Scraper(BaseTargetScraper):
|
||||
"""Fetches core lists per target from RetroArch buildbot nightly."""
|
||||
|
||||
def __init__(self, url: str = BUILDBOT_URL):
|
||||
def __init__(self, url: str = BUILDBOT_URL) -> None:
|
||||
super().__init__(url=url)
|
||||
|
||||
def _fetch_url(self, url: str) -> str:
|
||||
|
||||
@@ -113,7 +113,7 @@ def _parse_module(content: str) -> tuple[str | None, str]:
|
||||
class Scraper(BaseTargetScraper):
|
||||
"""Fetches RetroPie package availability by parsing scriptmodules."""
|
||||
|
||||
def __init__(self, url: str = GITHUB_API_URL):
|
||||
def __init__(self, url: str = GITHUB_API_URL) -> None:
|
||||
super().__init__(url=url)
|
||||
|
||||
def _list_scriptmodules(self) -> list[str]:
|
||||
|
||||
@@ -127,7 +127,7 @@ def metapackage_cores(package_mk: str, arch: str) -> list[str]:
|
||||
class Scraper(BaseTargetScraper):
|
||||
"""Per-device core lists from ROCKNIX's emulators metapackage."""
|
||||
|
||||
def __init__(self, url: str = "https://github.com/ROCKNIX/distribution"):
|
||||
def __init__(self, url: str = "https://github.com/ROCKNIX/distribution") -> None:
|
||||
super().__init__(url=url)
|
||||
self.tag = ""
|
||||
|
||||
|
||||
@@ -126,7 +126,7 @@ def _system_link(sys_id: str, prefix: str = "") -> str:
|
||||
return f"[{sys_id}]({prefix}systems/{slug}.md#{anchor})"
|
||||
return sys_id
|
||||
|
||||
def _render_yaml_value(lines: list[str], val, indent: int = 4) -> None:
|
||||
def _render_yaml_value(lines: list[str], val: object, indent: int = 4) -> None:
|
||||
"""Render any YAML value as indented markdown."""
|
||||
pad = " " * indent
|
||||
if isinstance(val, dict):
|
||||
@@ -194,7 +194,7 @@ def _pct(n: int, total: int) -> str:
|
||||
return "0%"
|
||||
return f"{n / total * 100:.1f}%"
|
||||
|
||||
def _by_mode(value, mode: str) -> str:
|
||||
def _by_mode(value: str | dict, mode: str) -> str:
|
||||
"""Read a profile field that may be keyed by build mode."""
|
||||
if isinstance(value, dict):
|
||||
return str(value.get(mode) or "") if mode else ""
|
||||
|
||||
@@ -118,7 +118,7 @@ def revision_markdown(profile: dict) -> str:
|
||||
return ", ".join(links)
|
||||
|
||||
|
||||
def _source_ref_markdown(profile: dict, value) -> str:
|
||||
def _source_ref_markdown(profile: dict, value: str | dict | None) -> str:
|
||||
"""Render source_ref values as pinned links when their forge is known.
|
||||
|
||||
A profile can declare two repositories: the libretro port in ``source`` and
|
||||
|
||||
@@ -17,7 +17,7 @@ import os
|
||||
# and a page is only removed once nothing produced it.
|
||||
_produced: set[str] = set()
|
||||
|
||||
def _record(path) -> None:
|
||||
def _record(path: str | Path) -> None:
|
||||
_produced.add(os.path.realpath(str(path)))
|
||||
|
||||
def _undecorated(markdown: str) -> str:
|
||||
@@ -43,7 +43,7 @@ def write_if_changed(path: str, content: str) -> bool:
|
||||
_record(path)
|
||||
return _write_artifact(path, content, normalize=_undecorated)
|
||||
|
||||
def write_decorated(path, content: str) -> bool:
|
||||
def write_decorated(path: str | Path, content: str) -> bool:
|
||||
"""Write the wrapped page, comparing front matter and body together."""
|
||||
_record(path)
|
||||
return _write_artifact(str(path), content)
|
||||
|
||||
+5
-1
@@ -17,6 +17,7 @@ union and is never discarded, the same rule region.py already applies.
|
||||
from __future__ import annotations
|
||||
|
||||
from dataclasses import dataclass, field
|
||||
from typing import TYPE_CHECKING
|
||||
|
||||
import nativemode
|
||||
from common import (
|
||||
@@ -34,6 +35,9 @@ from common import (
|
||||
from packresolve import preferred_entries
|
||||
from validation import outside_gap_scope
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from packextras import ExtraPlacement
|
||||
|
||||
# A profile entry can prove a slot without declaring a hash: Dolphin names no
|
||||
# checksum for the GameCube boot ROM because its source names none, and the
|
||||
# proof is then the path the repository stores the file at.
|
||||
@@ -243,7 +247,7 @@ def _archive_claims(
|
||||
is_standalone: bool,
|
||||
db: dict,
|
||||
base_dest: str,
|
||||
placement,
|
||||
placement: ExtraPlacement | None,
|
||||
zip_contents: dict | None,
|
||||
data_dir_registry: dict | None,
|
||||
) -> list[Claim]:
|
||||
|
||||
@@ -15,6 +15,7 @@ Usage:
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import io
|
||||
import re
|
||||
import struct
|
||||
import sys
|
||||
@@ -99,7 +100,7 @@ def _dos_stamp(info: zipfile.ZipInfo) -> tuple[int, int]:
|
||||
)
|
||||
|
||||
|
||||
def _data_offset(source, info: zipfile.ZipInfo) -> int:
|
||||
def _data_offset(source: io.BufferedReader, info: zipfile.ZipInfo) -> int:
|
||||
"""Where a member's stored bytes begin, read from its own local header."""
|
||||
source.seek(info.header_offset)
|
||||
header = _LOCAL.unpack(source.read(_LOCAL.size))
|
||||
@@ -108,7 +109,9 @@ def _data_offset(source, info: zipfile.ZipInfo) -> int:
|
||||
return info.header_offset + _LOCAL.size + header[10] + header[11]
|
||||
|
||||
|
||||
def _write_part(source, members: list[zipfile.ZipInfo], dest: Path) -> None:
|
||||
def _write_part(
|
||||
source: io.BufferedReader, members: list[zipfile.ZipInfo], dest: Path
|
||||
) -> None:
|
||||
directory = []
|
||||
with open(dest, "wb") as out:
|
||||
for info in members:
|
||||
|
||||
@@ -63,7 +63,7 @@ MAX_FILE_SIZE = 100 * 1024 * 1024
|
||||
|
||||
|
||||
class ValidationResult:
|
||||
def __init__(self, filepath: str):
|
||||
def __init__(self, filepath: str) -> None:
|
||||
self.filepath = filepath
|
||||
self.filename = os.path.basename(filepath)
|
||||
self.checks = [] # (status, message) tuples
|
||||
@@ -72,13 +72,13 @@ class ValidationResult:
|
||||
self.crc32 = ""
|
||||
self.size = 0
|
||||
|
||||
def add_check(self, passed: bool, message: str):
|
||||
def add_check(self, passed: bool, message: str) -> None:
|
||||
self.checks.append(("PASS" if passed else "FAIL", message))
|
||||
|
||||
def add_warning(self, message: str):
|
||||
def add_warning(self, message: str) -> None:
|
||||
self.checks.append(("WARN", message))
|
||||
|
||||
def add_info(self, message: str):
|
||||
def add_info(self, message: str) -> None:
|
||||
self.checks.append(("INFO", message))
|
||||
|
||||
@property
|
||||
@@ -303,7 +303,9 @@ def get_changed_files() -> list[str]:
|
||||
return [f for f in result.stdout.strip().split("\n") if f.startswith("bios/") and f]
|
||||
|
||||
|
||||
def _files_to_validate(args, parser) -> list:
|
||||
def _files_to_validate(
|
||||
args: argparse.Namespace, parser: argparse.ArgumentParser
|
||||
) -> list:
|
||||
"""The files named on the command line, or the changed ones."""
|
||||
if not args.changed:
|
||||
if not args.files:
|
||||
@@ -319,7 +321,7 @@ def _files_to_validate(args, parser) -> list:
|
||||
return files
|
||||
|
||||
|
||||
def _load_database_if_any(path):
|
||||
def _load_database_if_any(path: str) -> dict | None:
|
||||
try:
|
||||
return load_database(path)
|
||||
except FileNotFoundError:
|
||||
@@ -329,7 +331,7 @@ def _load_database_if_any(path):
|
||||
return None
|
||||
|
||||
|
||||
def main():
|
||||
def main() -> None:
|
||||
parser = argparse.ArgumentParser(description="Validate BIOS file contributions")
|
||||
parser.add_argument("files", nargs="*", help="Files to validate")
|
||||
parser.add_argument(
|
||||
|
||||
@@ -35,7 +35,7 @@ class Page:
|
||||
|
||||
|
||||
class _PageParser(HTMLParser):
|
||||
def __init__(self, path: Path):
|
||||
def __init__(self, path: Path) -> None:
|
||||
super().__init__(convert_charrefs=True)
|
||||
self.page = Page(path)
|
||||
self._title_depth = 0
|
||||
@@ -43,7 +43,7 @@ class _PageParser(HTMLParser):
|
||||
self._jsonld_depth = 0
|
||||
self._jsonld_parts: list[str] = []
|
||||
|
||||
def handle_starttag(self, tag: str, attrs) -> None:
|
||||
def handle_starttag(self, tag: str, attrs: list[tuple[str, str | None]]) -> None:
|
||||
data = dict(attrs)
|
||||
if tag == "html":
|
||||
self.page.lang = data.get("lang", "")
|
||||
|
||||
+1
-1
@@ -2077,7 +2077,7 @@ def _run_listing(args: argparse.Namespace, parser: argparse.ArgumentParser) -> N
|
||||
print(target_listing_line(t))
|
||||
|
||||
|
||||
def main():
|
||||
def main() -> None:
|
||||
with contextlib.ExitStack() as held:
|
||||
_main(held)
|
||||
|
||||
|
||||
@@ -3754,7 +3754,7 @@ class SignaturesCarryTheirTypes(unittest.TestCase):
|
||||
priority list of _select_variants)."""
|
||||
|
||||
# Lowered as modules are annotated; never raised.
|
||||
CEILING = 139
|
||||
CEILING = 0
|
||||
COMPLETE = ("artifacts.py", "common.py", "validation.py", "generate_pack.py")
|
||||
|
||||
@staticmethod
|
||||
|
||||
Reference in new issue
Block a user