diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index e5ccc8c..0b8bf00 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -43,6 +43,22 @@ jobs: - name: Test run: python -m pytest + types: + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v7 + - uses: actions/setup-python@v7 + with: + python-version: "3.13" + cache: pip + cache-dependency-path: pyproject.toml + - name: Install + run: | + python -m pip install --upgrade pip + python -m pip install -e ".[test,deck,fonts]" + - name: Type check + run: python -m pyright + goldens: # The golden records of the stages (tests/golden/*.json) were made with these library versions; with the same # versions installed an output change without a stage or converter version bump fails instead of being skipped. diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md index 6532bd4..807a9d9 100644 --- a/CONTRIBUTING.md +++ b/CONTRIBUTING.md @@ -22,6 +22,7 @@ These are the checks CI runs on every pull request: ruff check . # lint ruff format --check . # formatting (line length 120) lint-imports # the layer contract of pyproject.toml +pyright # type check (the settings of pyproject.toml) pytest # the test suite mkdocs build --strict # the documentation site ``` diff --git a/pyproject.toml b/pyproject.toml index 81c652b..a577ec0 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -26,7 +26,7 @@ classifiers = [ dynamic = ["version"] [project.optional-dependencies] -test = ["pytest>=8", "ruff>=0.15", "jsonschema>=4.18", "import-linter>=2.1"] +test = ["pytest>=8", "ruff>=0.15", "jsonschema>=4.18", "import-linter>=2.1", "pyright>=1.1.414"] fonts = ["fonttools>=4.50", "freetype-py>=2.5", "scipy>=1.13"] docs = ["mkdocs>=1.6,<2", "mkdocs-material>=9.5", "mkdocstrings[python]>=0.29"] # the deck model of music-data: one nnnotes version carries one ournotes-deck release (.github/workflows/deck.yml moves @@ -69,6 +69,12 @@ ignore = ["B023", "B905"] [tool.ruff.lint.flake8-bugbear] extend-immutable-calls = ["nnnotes.content.liveoptions.LiveOptions"] +[tool.pyright] +include = ["src"] +pythonVersion = "3.11" +pythonPlatform = "Linux" +typeCheckingMode = "basic" + [tool.importlinter] root_package = "nnnotes" diff --git a/src/nnnotes/cli/apk.py b/src/nnnotes/cli/apk.py index 25b2f72..2015bc7 100644 --- a/src/nnnotes/cli/apk.py +++ b/src/nnnotes/cli/apk.py @@ -32,8 +32,7 @@ def cmd_settings(args, cfg): def cmd_hca_key(args, cfg): from ..formats import crikey - cfg.require_path("paths", "apk") - k = crikey.find_key(Session(cfg).path("paths", "apk")) + k = crikey.find_key(Session(cfg).required_path("paths", "apk")) r = {"found": bool(k), "digits": len(str(k)) if k else None, "written": None} if args.write and k: d = Path(args.write) diff --git a/src/nnnotes/cli/assets.py b/src/nnnotes/cli/assets.py index 181304e..3536ff9 100644 --- a/src/nnnotes/cli/assets.py +++ b/src/nnnotes/cli/assets.py @@ -64,7 +64,7 @@ from ..source.config import ConfigError, usable_cpus, use from ..source.session import Session from . import common -from .common import nonnegative, print_json +from .common import nonnegative, print_json, usage GIB = 1 << 30 LAYOUT_NAMES = ("original", "cas") @@ -466,7 +466,7 @@ def __init__(self, args, cfg, log=_log): try: self.selection = parse_selection(getattr(args, "select", None)) except ValueError as e: - args.usage(str(e)) + usage(args, str(e)) self.installed = installed() self.calibration = load_calibration(self.root) self.scheduled = calibrated(self.installed.stages, self.calibration) # the stages with calibrated costs @@ -474,7 +474,7 @@ def __init__(self, args, cfg, log=_log): if self.version.get("region"): self.cfg = cfg.for_region(self.version["region"]) self.apk = self.cfg.path("paths", "apk") - self._index = None + self._index: dict | None = None self._problems: list = [] self._apk_names: set | None = None self._facts: dict = {} @@ -495,13 +495,13 @@ def _catalog_version(self): except ValueError: v = self.db.get(ref, self.cfg.get("catalog", "region"), self.cfg.get("catalog", "language")) except (KeyError, ValueError) as e: - self.args.usage(f"--stored-catalog {ref}: {e.args[0] if e.args else e}") + usage(self.args, f"--stored-catalog {ref}: {e.args[0] if e.args else e}") remote, apk = self.db.catalog_bytes(v) return v, remote, apk cat = Session(self.cfg).catalog(bundles=False) src = cat.sources() remote, apk = src["remote"], src.get("apk") - source = cat.source.to_dict() if getattr(cat, "source", None) else None + source = cat.source.to_dict() if cat.source is not None else None vid = catalogdb.version_id(contract.sha256(remote), contract.sha256(apk) if apk is not None else None, source) v = next((x for x in self.db.versions() if x["id"] == vid), None) if v is None: @@ -512,7 +512,7 @@ def _catalog_version(self): language=self.cfg.get("catalog", "language"), apk_version_name=catalogdb.apk_version(self.apk), source=source, - resource_version=source["version"] if source else getattr(cat, "resource_version", None), + resource_version=source["version"] if source else cat.resource_version, ) return v, remote, apk @@ -599,8 +599,9 @@ def views(self) -> list[str]: want = avail if spec == "all" else sorted({"view." + v.strip() for v in spec.split(",") if v.strip()}) unknown = [w[5:] for w in want if w not in avail] if unknown: - self.args.usage( - f"--views: no view {', '.join(unknown)} (views: {', '.join(a[5:] for a in avail) or 'none installed'})" + usage( + self.args, + f"--views: no view {', '.join(unknown)} (views: {', '.join(a[5:] for a in avail) or 'none installed'})", ) if want and self.master() is None: raise self.cfg.missing(*self.cfg.master(self.cfg.get("catalog", "region"))) @@ -616,20 +617,20 @@ def params(self, pipeline: list[str]) -> dict: level = getattr(self.args, "png_level", None) if level is not None: if not 0 <= level <= 9: - self.args.usage(f"--png-level {level}: expected 0 to 9") + usage(self.args, f"--png-level {level}: expected 0 to 9") takers = [n for n in pipeline if "png" in stages[n].PARAMS] if not takers: - self.args.usage("--png-level: no stage of the pipeline encodes PNG") + usage(self.args, "--png-level: no stage of the pipeline encodes PNG") for n in takers: base = stages[n].PARAMS["png"] out.setdefault(n, {})["png"] = {**(base if isinstance(base, dict) else {}), "level": level} level = getattr(self.args, "flac_level", None) if level is not None: if not 0 <= level <= 12: - self.args.usage(f"--flac-level {level}: expected 0 to 12") + usage(self.args, f"--flac-level {level}: expected 0 to 12") takers = [n for n in pipeline if "flac" in stages[n].PARAMS] if not takers: - self.args.usage("--flac-level: no stage of the pipeline encodes FLAC") + usage(self.args, "--flac-level: no stage of the pipeline encodes FLAC") for n in takers: out.setdefault(n, {})["flac"] = {**stages[n].PARAMS["flac"], "level": level} only = getattr(self.args, "only_class", None) @@ -637,8 +638,10 @@ def params(self, pipeline: list[str]) -> dict: classes = sorted({c.strip() for c in only.split(",") if c.strip()}) takers = [n for n in pipeline if "classes" in stages[n].PARAMS] if not classes or not takers: - self.args.usage( - "--only-class: " + ("no class given" if not classes else "no stage of the pipeline selects classes") + usage( + self.args, + "--only-class: " + + ("no class given" if not classes else "no stage of the pipeline selects classes"), ) for n in takers: out.setdefault(n, {})["classes"] = classes @@ -646,7 +649,7 @@ def params(self, pipeline: list[str]) -> dict: try: stages[n].normalize(p) except ValueError as e: - self.args.usage(str(e)) + usage(self.args, str(e)) return out def workers(self) -> int: @@ -727,7 +730,7 @@ def _cache_rel(self, kind: str, entry: dict, loc: dict) -> str: rel = apk_rel(loc["internalId"]) return prefix + "raw/" + rel.lstrip("/") - def _inputs(self, kind: str, entries: list[dict], fetch: bool) -> dict: + def _inputs(self, kind: str, entries: list[dict], fetch: bool) -> Inputs: from ..pipeline.catalogdb import by_id locs = by_id(self.index()) @@ -839,7 +842,7 @@ def previous(self, since: str | None) -> dict | None: if doc is None: runs = sorted(p.name[:-5] for p in (self.root / "runs").glob("*.json") if p.name.startswith(since)) if len(runs) != 1: - self.args.usage(f"--since {since}: {'no run' if not runs else f'{len(runs)} runs'} with this id") + usage(self.args, f"--since {since}: {'no run' if not runs else f'{len(runs)} runs'} with this id") doc = self.store.run(runs[0]) return doc @@ -864,7 +867,7 @@ def plan(self, pipeline: list[str], params: dict, *, since=None, out=None, layou hits = [(n["id"], n["key"]) for n in p.doc["nodes"] if n["status"] == "hit" and self._output(n["stage"])] placed = self.layout_input(dict(hits)) for tid, key in hits: - placed.add(tid, self.store.result(key)) + placed.add(tid, self.store.committed(key, tid)) for name, d in layout_dirs(Path(out), layouts).items(): entries, _ = self.layout_entries(name, placed, store_derived=False) p.doc["summary"]["layouts"][name] = layout.diff(d, entries) @@ -1140,7 +1143,7 @@ def cmd_export(args, cfg): try: names = parse_layouts(args.layout) except ValueError as e: - args.usage(str(e)) + usage(args, str(e)) use(cfg) # the tools the atoms name are part of the keys ws = Workspace(args, cfg) pipeline = ws.pipeline() @@ -1162,11 +1165,11 @@ def cmd_export(args, cfg): try: placement = layout.probe(out, ws.root / "cas" / "sha256", link_mode(args, cfg)) except layout.LayoutError as e: - args.usage(str(e)) + usage(args, str(e)) if args.explain: for line in ws.explain(pipeline, params, orch, names, placement): print(line, file=sys.stderr) - if args.dry_run: + if placement is None: # --dry-run sys.stdout.write(ws.plan(pipeline, params, out=out, layouts=names).text()) return ws.run_index() @@ -1301,7 +1304,7 @@ def cmd_plan(args, cfg): try: layouts = parse_layouts(args.layout) except ValueError as e: - args.usage(str(e)) + usage(args, str(e)) use(cfg) # the tools the atoms name are part of the keys ws = Workspace(args, cfg) pipeline = ws.pipeline() @@ -1341,26 +1344,26 @@ def why(ws: Workspace, p, tid: str, since=None) -> dict: """Why a task of a plan has its status: its reasons, and its key parts beside those of the previous run.""" node = next((n for n in p.doc["nodes"] if n["id"] == tid), None) if node is None: - ws.args.usage(f"--why {tid}: no task with this id in the plan") + usage(ws.args, f"--why {tid}: no task with this id in the plan") new = None if node["key"] is not None: t = next((t for t in p.tasks if t.key == node["key"]), None) doc = None if t is not None else ws.store.result(node["key"]) new = t.key_parts() if t is not None else (doc["keyParts"] if doc else None) prev = ws.previous(since) - old_task = next((t for t in (prev or {}).get("tasks", []) if t["id"] == tid), None) - old = None - if old_task is not None and old_task.get("key"): - doc = ws.store.result(old_task["key"]) - old = doc["keyParts"] if doc else None + old_task = next((t for t in prev.get("tasks", []) if t["id"] == tid), None) if prev is not None else None + previous = old = None + if prev is not None and old_task is not None: + previous = {"run": prev["id"], "key": old_task.get("key"), "status": old_task["status"]} + if old_task.get("key"): + doc = ws.store.result(old_task["key"]) + old = doc["keyParts"] if doc else None return { "task": tid, "status": node["status"], "key": node["key"], "reasons": node["reasons"], - "previous": {"run": prev["id"], "key": old_task.get("key"), "status": old_task["status"]} - if old_task is not None - else None, + "previous": previous, "keyParts": {"old": old, "new": new}, } @@ -1403,7 +1406,7 @@ def cmd_run_stage(args, cfg): try: tasks.append(Task.from_json(contract.loads(f.read_bytes()))) except OSError as e: - args.usage(f"{f}: {e.strerror or e}") + usage(args, f"{f}: {e.strerror or e}") except (IncompatibleTask, ValueError) as e: problems.append(f"{f}: {e}") inst = installed(sorted({t.stage for t in tasks})) diff --git a/src/nnnotes/cli/browse.py b/src/nnnotes/cli/browse.py index cd6f31b..9e384f6 100644 --- a/src/nnnotes/cli/browse.py +++ b/src/nnnotes/cli/browse.py @@ -13,7 +13,7 @@ from .. import __version__ from ..source.addressables import BundleKey, decrypt, remote_path from ..source.catalog import Bundle, Catalog -from ..source.config import ConfigError +from ..source.config import Config, ConfigError from ..source.gameapi import GameApiError from ..source.jp import open_catalog from ..source.session import Session @@ -45,13 +45,27 @@ class Region: label: str cdn: str = field(repr=False) languages: list[str] - config: object = field(default=None, repr=False) + config: Config | None = field(default=None, repr=False) + + +class Server(ThreadingHTTPServer): + """The browser's HTTP server: its regions, the bundle key, the cache, and the catalogs read so far.""" + + def __init__(self, address, regions: list[Region], bundle_key: BundleKey, cache: Path): + super().__init__(address, Handler) + self.regions = {r.name: r for r in regions} + self.bundle_key = bundle_key + self.cache = Path(cache) + self.catalogs: dict[tuple[str, str], tuple[dict, dict]] = {} + self.jp_catalogs: dict[tuple[str, str], Catalog] = {} class Handler(BaseHTTPRequestHandler): """/ -> regions, // -> languages, //// -> listing, ///download/ -> the decrypted bundle.""" + server: Server + def listing(self, label: str, rows) -> None: title = f"nnnotes {__version__} /" + label data = ( @@ -121,11 +135,14 @@ def do_GET(self): entry = bundles[int(ident)] name = entry["internal_id"].rsplit("/", 1)[1] try: - cat = getattr(self.server, "jp_catalogs", {}).get((region.name, language)) + cat = self.server.jp_catalogs.get((region.name, language)) + path = remote_path(entry["internal_id"]) if cat is not None: data = cat.fetch(Bundle(int(ident), entry["internal_id"], name, True)).read_bytes() + elif path is None: + raise ValueError(f"{entry['internal_id']}: not a remote bundle") else: - with urlopen(region.cdn + remote_path(entry["internal_id"]), timeout=60) as response: + with urlopen(region.cdn + path, timeout=60) as response: data = decrypt(response.read(), name, self.server.bundle_key) except (GameApiError, ConfigError, ValueError): self.send_error(502, "Resource could not be downloaded") @@ -160,12 +177,7 @@ def do_GET(self): def serve(regions: list[Region], bundle_key: BundleKey, cache: Path, port: int, host: str = "127.0.0.1") -> None: """Browse the catalogs of `regions` and download decrypted bundles on `host:port` (catalogs cached under /catalogs//).""" - server = ThreadingHTTPServer((host, port), Handler) - server.regions = {r.name: r for r in regions} - server.bundle_key = bundle_key - server.cache = Path(cache) - server.catalogs = {} - server.jp_catalogs = {} + server = Server((host, port), regions, bundle_key, cache) print(f"nnnotes catalog browse: serving on {host}:{port}", flush=True) try: server.serve_forever() diff --git a/src/nnnotes/cli/catalog.py b/src/nnnotes/cli/catalog.py index 1144a5a..bc055ca 100644 --- a/src/nnnotes/cli/catalog.py +++ b/src/nnnotes/cli/catalog.py @@ -9,7 +9,7 @@ from ..source.config import ConfigError, check_catalog_version from ..source.session import Session -from .common import command, group, note, report, write +from .common import command, group, note, report, usage, write CATALOG_FILE = re.compile(r"catalog_([A-Za-z0-9][A-Za-z0-9.-]*)_[A-Za-z0-9-]+\.bin") @@ -114,7 +114,7 @@ def cmd_import(args, cfg): try: remote = Path(args.file).read_bytes() except OSError as e: - args.usage(f"{args.file}: {e.strerror or e}") + usage(args, f"{args.file}: {e.strerror or e}") try: source = None if cfg.provider() == "jp": @@ -133,7 +133,7 @@ def cmd_import(args, cfg): source=source, ) except ValueError as e: - args.usage(str(e)) + usage(args, str(e)) _version(args, v) @@ -146,15 +146,16 @@ def cmd_fetch(args, cfg): from ..source.jp import open_catalog cat = open_catalog(cfg, region, apk=cfg.path("paths", "apk")) - src = cat.sources() + src, source = cat.sources(), cat.source + assert source is not None # a JP catalog names its version source v = _db(cfg).add( src["remote"], src.get("apk"), label=args.label, region=region, language=cfg.get("catalog", "language") or "ja", - source=cat.source.to_dict(), - resource_version=cat.source.version, + source=source.to_dict(), + resource_version=source.version, apk_version_name=catalogdb.apk_version(cfg.path("paths", "apk")), ) _version(args, v) @@ -183,7 +184,7 @@ def cmd_fetch(args, cfg): apk_version_name=catalogdb.apk_version(cfg.path("paths", "apk")), ) except ValueError as e: - args.usage(str(e)) + usage(args, str(e)) _version(args, v) @@ -198,7 +199,7 @@ def cmd_diff(args, cfg): try: vs.append(db.get(ref)) except (KeyError, ValueError) as e: - args.usage(f"{ref}: {e.args[0] if e.args else e}") + usage(args, f"{ref}: {e.args[0] if e.args else e}") store = Store(Session(cfg).store_root()) a, b = vs d = catalogdb.diff(db.index(a), db.index(b), addresses_of(store, a["id"]), addresses_of(store, b["id"])) diff --git a/src/nnnotes/cli/common.py b/src/nnnotes/cli/common.py index cf0894e..40a786b 100644 --- a/src/nnnotes/cli/common.py +++ b/src/nnnotes/cli/common.py @@ -7,6 +7,7 @@ import re import sys from pathlib import Path +from typing import NoReturn from ..core.jsonio import dumps from ..source.catalog import Catalog @@ -110,9 +111,9 @@ def group(sub, name: str, text: str): def command(sub, name: str, func, text: str, *, reads_config: bool = True) -> argparse.ArgumentParser: """The parser of a command that runs `func(args, cfg)` (cfg: the settings, None when `reads_config` is false); - `args.usage(message)` exits with a usage error (status 2) of this command.""" + `usage(args, message)` exits with a usage error (status 2) of this command.""" p = sub.add_parser(name, help=text, description=text, parents=[GLOBALS], formatter_class=Formatter) - p.set_defaults(func=func, usage=p.error, reads_config=reads_config) + p.set_defaults(func=func, parser=p, reads_config=reads_config) return p @@ -172,11 +173,16 @@ def report(args, doc, lines=None) -> None: # ---------------------------------------------------------------- usage errors +def usage(args, message: str) -> NoReturn: + """Exit with a usage error (status 2) of the command `args` were parsed for.""" + argparse.ArgumentParser.error(args.parser, message) + + def known_keys(args, cat: Catalog, keys) -> None: """A usage error (exit 2) naming the keys the catalog does not have.""" unknown = [k for k in keys if not cat.has(k)] if unknown: - args.usage(f"not a key of the catalog: {', '.join(unknown)}") + usage(args, f"not a key of the catalog: {', '.join(unknown)}") def known_row(args, md: Path, table: str, row_id: int, what: str) -> None: @@ -190,7 +196,7 @@ def known_row(args, md: Path, table: str, row_id: int, what: str) -> None: f"master data {md}: no {table}.json (a directory written by `nnnotes master decode`)" ) from None if not found: - args.usage(f"{what} {row_id}: no {table} row with this id") + usage(args, f"{what} {row_id}: no {table} row with this id") def flac_options(args) -> dict: @@ -200,11 +206,11 @@ def flac_options(args) -> dict: if level is None: return {} if level not in FLAC_LEVELS: - args.usage(f"--flac-level {level}: expected 0 to 12") + usage(args, f"--flac-level {level}: expected 0 to 12") if args.format != "flac": - args.usage(f"--flac-level: --format {args.format} writes no FLAC") + usage(args, f"--flac-level: --format {args.format} writes no FLAC") if getattr(args, "no_audio", False): - args.usage("--flac-level: --no-audio decodes no audio") + usage(args, "--flac-level: --no-audio decodes no audio") return {"flac_level": level} @@ -223,7 +229,7 @@ def live_options(args) -> dict: try: return parse_specs(args.live_option) except OptionSpecError as e: - args.usage(str(e)) + usage(args, str(e)) # ---------------------------------------------------------------- argument types and options diff --git a/src/nnnotes/cli/config.py b/src/nnnotes/cli/config.py index 4be3429..a3c8c71 100644 --- a/src/nnnotes/cli/config.py +++ b/src/nnnotes/cli/config.py @@ -6,7 +6,7 @@ from pathlib import Path from ..source.config import DEFAULT_FILE, ENV_CONFIG, ConfigError, config_files, find_file, user_file -from .common import command, group, note, print_json, report, write +from .common import command, group, note, print_json, report, usage, write def _tty() -> bool: @@ -46,7 +46,7 @@ def _value(args, name: str, text: str, stdin_used: list): setting, section = configfile.resolve(name) if text == "-": if stdin_used: - args.usage("only one value can be read from standard input") + usage(args, "only one value can be read from standard input") stdin_used.append(name) if setting.secret and _tty(): import getpass @@ -77,7 +77,7 @@ def cmd_init(args, cfg): for item in args.set or []: name, sep, text = item.partition("=") if not sep: - args.usage(f"--set {name}: give it as SECTION.KEY=VALUE") + usage(args, f"--set {name}: give it as SECTION.KEY=VALUE") section, key, value = _value(args, name.strip(), text, stdin_used) values.append((section, key, value)) region = section.split(".", 1)[1] if section.startswith("servers.") else None diff --git a/src/nnnotes/cli/export.py b/src/nnnotes/cli/export.py index 67b782f..a9bbfb6 100644 --- a/src/nnnotes/cli/export.py +++ b/src/nnnotes/cli/export.py @@ -26,6 +26,7 @@ live_options, out_arg, report, + usage, ) @@ -56,7 +57,7 @@ def cmd_story(args, cfg): try: os.path.relpath(models, out) except ValueError: - args.usage(f"--models {models}: not on the drive of {out} (story.json names it relative to the story)") + usage(args, f"--models {models}: not on the drive of {out} (story.json names it relative to the story)") language = languages.check(cfg.require("catalog", "language")) # the story UI's language md = Session(cfg).master_dir() known_row(args, md, "MasterAdv", args.adv_id, "episode") @@ -92,16 +93,16 @@ def cmd_live(args, cfg): md = Session(cfg).master_dir() known_row(args, md, "MasterLiveMusic", args.music_id, "music") if (args.music_id, args.difficulty) not in all_pairs(md): - args.usage(f"music {args.music_id}: no {args.difficulty} chart (no MasterLiveMusicScore row)") + usage(args, f"music {args.music_id}: no {args.difficulty} chart (no MasterLiveMusicScore row)") try: options = liveoptions.resolve(request, md) except liveoptions.OptionSpecError as e: - args.usage(str(e)) + usage(args, str(e)) cat = Session(cfg).catalog() try: # an unknown --leader-card, a --band without its scene keys live.resolve_band(cat, md, args.music_id, band=args.band, leader_card=args.leader_card) except (KeyError, ValueError) as e: - args.usage(e.args[0] if e.args else type(e).__name__) + usage(args, e.args[0] if e.args else type(e).__name__) r = live.build( cat, md, @@ -129,7 +130,7 @@ def cmd_live2d(args, cfg): try: # a model id or a model key (key,) = webmodel.catalog_models(cat, [args.key]).values() except ValueError as e: - args.usage(str(e)) + usage(args, str(e)) report(args, live2d.extract_model(cat, key, Path(args.out))) @@ -188,7 +189,7 @@ def cmd_shader(args, cfg): try: bundles.append(cat.apk_bundle(s)) except KeyError as e: # none or several bundles match - args.usage(f"--apk-bundle {e.args[0]}") + usage(args, f"--apk-bundle {e.args[0]}") r = shader.dump(bundles, Path(args.out)) report(args, r, lambda r: [f"{r['count']} shaders, {r['variants']} variants", *(f" {n}" for n in r["names"])]) @@ -200,7 +201,7 @@ def cmd_audio(args, cfg): cfg.require_path("paths", "apk") # the HCA keycode is read from the APK cat = Session(cfg).catalog() if not cat.has(f"Cri/Sound/{args.cue_sheet}"): - args.usage(f"cue sheet {args.cue_sheet}: no key Cri/Sound/{args.cue_sheet} in the catalog") + usage(args, f"cue sheet {args.cue_sheet}: no key Cri/Sound/{args.cue_sheet} in the catalog") out = Path(args.out) cues = cri.decode(cat, args.cue_sheet, out, fmt=args.format, **flac) report(args, {"out": str(out), "cues": len(cues)}, lambda r: [f"{r['out']} ({r['cues']} cues)"]) @@ -210,7 +211,7 @@ def cmd_ui(args, cfg): from ..content import ui if args.limit is not None and args.limit < 1: - args.usage("--limit must be positive") + usage(args, "--limit must be positive") index = ui.export_apk_library( cfg, args.out, diff --git a/src/nnnotes/cli/site.py b/src/nnnotes/cli/site.py index 3ea55d0..6b9629a 100644 --- a/src/nnnotes/cli/site.py +++ b/src/nnnotes/cli/site.py @@ -22,6 +22,7 @@ parse_pair, positive, report, + usage, ) @@ -33,16 +34,17 @@ def cmd_build(args, cfg): charts, models = bool(args.pair or args.all), bool(args.live2d or args.all_live2d) stories = bool(args.story or args.all_stories) if not (charts or models or stories or args.player_only or args.reingest_json): - args.usage( + usage( + args, "one of the arguments --pair --all --live2d --all-live2d --story --all-stories --player-only " - "--reingest-json is required" + "--reingest-json is required", ) if models and (args.player_only or args.reingest_json): - args.usage("--player-only and --reingest-json build nothing: leave out --live2d / --all-live2d") + usage(args, "--player-only and --reingest-json build nothing: leave out --live2d / --all-live2d") if stories and (args.player_only or args.reingest_json): - args.usage("--player-only and --reingest-json build nothing: leave out --story / --all-stories") + usage(args, "--player-only and --reingest-json build nothing: leave out --story / --all-stories") if request and not charts: - args.usage("--live-option applies to charts: give --pair or --all") + usage(args, "--live-option applies to charts: give --pair or --all") player = cfg.path("paths", "player") out = Path(args.out) if args.player_only or args.reingest_json: @@ -63,16 +65,17 @@ def cmd_build(args, cfg): base = {"region": regions[0]} if regions else {} unknown = web.unknown_pairs(cfg, args.pair, regions) if args.pair else [] if unknown: - args.usage( + usage( + args, f"chart{'s' if len(unknown) > 1 else ''} {', '.join(f'{m}:{d}' for m, d in unknown)}: no " - f"MasterLiveMusicScore row in the master data of the site's region(s)" + f"MasterLiveMusicScore row in the master data of the site's region(s)", ) if request: # a value a region's master data does not have try: for md in sitefiles.region_masters(cfg, sitefiles.site_regions(cfg, regions)).values(): liveoptions.resolve(request, md) except liveoptions.OptionSpecError as e: - args.usage(str(e)) + usage(args, str(e)) r = {} encoding = args.compress if models: @@ -80,7 +83,7 @@ def cmd_build(args, cfg): try: selected = webmodel.catalog_models(Session(cfg).catalog(bundles=False, **base), args.live2d) except ValueError as e: - args.usage(str(e)) + usage(args, str(e)) r.update( webmodel.build( out, @@ -124,9 +127,10 @@ def cmd_build(args, cfg): unknown = storysite.unknown_stories(cfg, args.story, regions) if args.story else [] if unknown: noun = "stories" if len(unknown) > 1 else "story" - args.usage( + usage( + args, f"{noun} {', '.join(map(str, unknown))}: no MasterAdv row in the master data of the " - f"site's region(s)" + f"site's region(s)", ) r.update( storysite.build( diff --git a/src/nnnotes/cli/voices.py b/src/nnnotes/cli/voices.py index d31e50c..a323dd2 100644 --- a/src/nnnotes/cli/voices.py +++ b/src/nnnotes/cli/voices.py @@ -11,7 +11,7 @@ from ..pipeline.voices import STATUSES from ..source.languages import LANGUAGES from ..source.session import Session -from .common import AUDIO_CHOICES, command, flac_level_arg, flac_options, group, note, out_arg, print_json, write +from .common import AUDIO_CHOICES, command, flac_level_arg, flac_options, group, note, out_arg, print_json, usage, write def _label(doc: dict, ids, language: str) -> str: @@ -45,13 +45,14 @@ def _document(args, cfg, episodes=()) -> tuple[dict, Path | None]: p = Path(args.source_dir) f = p if p.is_file() else p / views.view_path(voices.VIEW, views.MAIN_SUBJECT) if not f.is_file(): - args.usage( + usage( + args, f"--from {p}: no {views.view_path(voices.VIEW, views.MAIN_SUBJECT)} " - "(export assets --views voices writes it)" + "(export assets --views voices writes it)", ) doc = contract.loads(f.read_bytes()) if doc.get("schema") != voices.SCHEMA: - args.usage(f"{f}: not a voices document") + usage(args, f"{f}: not a voices document") return doc, (f.parent.parent if not p.is_file() else None) return voices.index_document(Session(cfg), episodes), None @@ -59,7 +60,7 @@ def _document(args, cfg, episodes=()) -> tuple[dict, Path | None]: def _language(args, cfg) -> str: lang = cfg.get("catalog", "language") or "ja" if lang not in LANGUAGES: - args.usage(f"--language {lang}: one of {', '.join(LANGUAGES)}") + usage(args, f"--language {lang}: one of {', '.join(LANGUAGES)}") return lang @@ -70,12 +71,12 @@ def _filters(args, doc: dict) -> dict: if args.character: chars = voices.character_ids(doc, args.character) if not chars: - args.usage(f"--character {args.character}: no character of the master data has this id or name") + usage(args, f"--character {args.character}: no character of the master data has this id or name") eps = None if args.episode: eps = voices.episode_ids(doc, args.episode) if not eps: - args.usage(f"--episode {args.episode}: no episode of the master data has this id or asset") + usage(args, f"--episode {args.episode}: no episode of the master data has this id or asset") return { "characters": chars, "category": args.category, @@ -155,7 +156,7 @@ def cmd_get(args, cfg) -> None: doc, base = _document(args, cfg, story) rows = voices.rows_for(doc, args.id) if not rows: - args.usage(f"{args.id}: no voice row or sound id of that number") + usage(args, f"{args.id}: no voice row or sound id of that number") r = next((x for x in rows if x["status"] == "ok"), rows[0]) sheet, cue = r["sound"].get("sheet"), r["sound"].get("cue") dst = Path(args.out) diff --git a/src/nnnotes/content/advmedia.py b/src/nnnotes/content/advmedia.py index 3fe5848..c5b9608 100644 --- a/src/nnnotes/content/advmedia.py +++ b/src/nnnotes/content/advmedia.py @@ -119,7 +119,7 @@ def component(self, o) -> dict: "type": "MonoBehaviour", "class": None, "missingScript": True, - **self.value(o, {k: v for k, v in tt.items() if k not in HEADER}), + **{k: self.value(o, v) for k, v in tt.items() if k not in HEADER}, } return super().component(o) diff --git a/src/nnnotes/content/advui.py b/src/nnnotes/content/advui.py index 0b5afa0..cca94e5 100644 --- a/src/nnnotes/content/advui.py +++ b/src/nnnotes/content/advui.py @@ -78,6 +78,7 @@ from __future__ import annotations import re +from collections.abc import Sequence from pathlib import Path from ..core.jsonio import write_json @@ -364,6 +365,14 @@ def _ref_path(v) -> str | None: return v.get("gameObject") or v.get("transform") +def _need_path(v) -> str: + """The node path of a serialized reference that must not be null (RuntimeError for a null one).""" + p = _ref_path(v) + if p is None: + raise RuntimeError(f"a required node reference is null ({v!r})") + return p + + def _plain(v, sprites: list): """A serialized value as plain data: a reference as its node path, a sprite as its name (its reference added to `sprites`), a ScriptableObject or material as its name, a texture as its name; containers element-wise.""" @@ -575,6 +584,14 @@ def _component(node: dict, classes: tuple) -> dict | None: return hit[0] if hit else None +def _need_component(node: dict, classes: tuple) -> dict: + """The one component of `classes` on `node` (RuntimeError when it has none).""" + c = _component(node, classes) + if c is None: + raise RuntimeError(f"{node['path']}: no component of {classes}") + return c + + def extract( cat: Catalog, player: PlayerData, @@ -643,7 +660,7 @@ def records(prefab_nodes: list, keep_text: bool) -> list: if not (rec.get("localizeText") or {}).get("localizeEnabled"): raise NotImplementedError(f"{n['path']}: text without LocalizeText") rec["textStyle"] = styles.record( - n["path"], _component(n, TMP_CLASSES), _component(n, ("LocalizeText",)) + n["path"], _need_component(n, TMP_CLASSES), _component(n, ("LocalizeText",)) ) if not keep_text: del rec["text"] @@ -866,7 +883,8 @@ def records(prefab_nodes: list, keep_text: bool) -> list: if blur is not None: doc["blur"] = blur if game: - tmp_settings = player.names_in(player.resource("TMP Settings"), player.mono(player.resource("TMP Settings"))) + tmp = player.resource("TMP Settings") + tmp_settings = {k: player.names_in(tmp, v) for k, v in player.mono(tmp).items()} doc["fonts"] = font_out doc["tmpSettings"] = { k: tmp_settings[k] @@ -978,9 +996,9 @@ def uses_backdrop_filter(window: list[dict]) -> bool: def backdrop_path(widget: list[dict], comp: dict) -> str: """The node of the centre-talk backdrop (UIAdvWidget BACKDROP_REFS: its CanvasGroup and Image, on one node).""" paths = {_ref_path(comp[k]) for k in BACKDROP_REFS} - if len(paths) != 1 or None in paths: + path = next(iter(paths)) if len(paths) == 1 else None + if path is None: raise RuntimeError(f"UIAdvWidget {BACKDROP_REFS}: {sorted(map(str, paths))}") - path = paths.pop() node = next((n for n in widget if n["path"] == path), None) if node is None: raise RuntimeError(f"{path}: not a widget node") @@ -1102,7 +1120,7 @@ def sprite_asset_record(ex: Exporter, key: str = EMOJI_SPRITE_ASSET_KEY) -> dict tt = o.read_typetree() if tt.get("fallbackSpriteAssets"): raise NotImplementedError(f"{key}: fallback sprite assets") - sheet = ex.deref(o, tt["spriteSheet"]).read_typetree() + sheet = ex.referent(o, tt["spriteSheet"]).read_typetree() chars = [ {"index": i, "unicode": c["m_Unicode"], "name": c["m_Name"], "glyph": c["m_GlyphIndex"], "scale": c["m_Scale"]} for i, c in enumerate(tt["m_SpriteCharacterTable"]) @@ -1384,7 +1402,7 @@ def chat_runtime_texts(episode: dict) -> list[str]: return out -def about(language: str, windows: list[str] = (adv.DEFAULT_TALK_WINDOW,)) -> str: +def about(language: str, windows: Sequence[str] = (adv.DEFAULT_TALK_WINDOW,)) -> str: """ui.json `about`: what the file holds (the widget and its talk windows), in which client language.""" return f"ADV front canvas UI data (UIAdvWidget + {', '.join(windows)}), {language}" @@ -1453,7 +1471,7 @@ def _game_fonts( out_nodes: list, field: str, master_ids: dict, - ui_texts: list[str] = (), + ui_texts: Sequence[str] = (), ) -> tuple[dict, dict]: """`--fonts game`: the localized font / material of every text node (`text.localized`), the TMP font assets, the glyph coverage of the characters the episode shows (in the text field `field`; `master_ids`: @@ -1481,16 +1499,19 @@ def _game_fonts( primary = primary[0] coverage, needed, runtime_units = tmpfont.glyph_coverage(fonts, primary, chars) # the underline character from the primary asset itself (baked, or generated as its dynamic asset would) - hit = fonts.lookup(primary, UNDERLINE_CHARACTER) - in_asset = bool(hit and hit[1] is primary and hit[0] in ("baked", "runtime")) + kind, font = fonts.lookup(primary, UNDERLINE_CHARACTER) or (None, None) + in_asset = font is primary and kind in ("baked", "runtime") if in_asset: - (needed if hit[0] == "baked" else runtime_units).setdefault(primary.name, set()).add(UNDERLINE_CHARACTER) + (needed if kind == "baked" else runtime_units).setdefault(primary.name, set()).add(UNDERLINE_CHARACTER) coverage["underline"] = {"font": primary.name, "character": chr(UNDERLINE_CHARACTER), "inAsset": in_asset} # the missing glyph (tmpfont.missing_glyph): its substitute from the asset that has it tmpfont.check_glyph_variants(fonts, primary, shown) mg = tmpfont.missing_glyph(fonts, primary, chars, tmpfont.missing_glyph_character(ex.player)) if mg: - kind, f = fonts.lookup(primary, mg["unicode"]) + found = fonts.lookup(primary, mg["unicode"]) + if found is None: + raise RuntimeError(f"missing glyph U+{mg['unicode']:04X}: in no font asset of {primary.name}") + kind, f = found if kind != "synthesized": (needed if kind == "baked" else runtime_units).setdefault(f.name, set()).add(mg["unicode"]) coverage["missingGlyph"] = [chr(u) for u in mg["characters"]] diff --git a/src/nnnotes/content/advvideo.py b/src/nnnotes/content/advvideo.py index 0fb7754..ee0cfc0 100644 --- a/src/nnnotes/content/advvideo.py +++ b/src/nnnotes/content/advvideo.py @@ -32,11 +32,11 @@ def usm_asset(cat: Catalog, key: str) -> tuple[dict, Path]: info, raw = None, None for e in cat.dependencies(key): - iid = e["internal_id"] - if remote_path(iid) is None: + if remote_path(e["internal_id"]) is None: continue - if iid.endswith(".bundle"): - env = UnityPy.load(str(cat.fetch(cat.as_bundle(e)))) + bundle = cat.as_bundle(e) + if bundle is not None: + env = UnityPy.load(str(cat.fetch(bundle))) for o in env.objects: if o.type.name == "MonoBehaviour": tt = o.read_typetree() diff --git a/src/nnnotes/content/liveui.py b/src/nnnotes/content/liveui.py index 9438237..17e2739 100644 --- a/src/nnnotes/content/liveui.py +++ b/src/nnnotes/content/liveui.py @@ -253,7 +253,7 @@ def extract( loc = textstyle.localize_text( n["path"], t["m_fontAsset"]["name"], t["m_sharedMaterial"]["material"], swap, lang, mode ) - shown = text_of.get(n["path"], t["m_text"]) + shown = text_of[n["path"]] if n["path"] in text_of else t["m_text"] texts[n["path"]] = {"localized": loc, "text": shown, "setByInitialize": n["path"] in text_of} if styles: record = styles.record(n["path"], t, lt) @@ -268,7 +268,7 @@ def extract( coverage, materials, font_out = _game_fonts(ex, fontset, texts, chars_by_font, packer) else: # name stubs: no font data, no atlases coverage, materials = None, {} - font_out = {f: {"name": f, "note": FONT_NOTE} for f in sorted(chars_by_font)} + font_out: dict = {f: {"name": f, "note": FONT_NOTE} for f in sorted(chars_by_font)} # -- sprites: drawn Images, the difficulty icons, the jacket ---------------------------------------- replaced = {r["gameObject"] for f in ("_jacketImages", "_difficultyImages") for r in canvas[f]} @@ -334,7 +334,8 @@ def extract( # -- the root Animator's controller (evidence) ----------------------------------------------------- an = _comp(by_path[f"{CANVAS}/root"][0], "Animator") controller = an["m_Controller"] if an else None - tmp_settings = player.names_in(player.resource("TMP Settings"), player.mono(player.resource("TMP Settings"))) + tmp = player.resource("TMP Settings") + tmp_settings = {k: player.names_in(tmp, v) for k, v in player.mono(tmp).items()} doc = { "about": f"Live start canvas (Live/UILiveStartCanvas, LightWeight mode): what livescene/scene.json lacks, " diff --git a/src/nnnotes/content/room.py b/src/nnnotes/content/room.py index d3e0ef7..e8ba1ac 100644 --- a/src/nnnotes/content/room.py +++ b/src/nnnotes/content/room.py @@ -42,7 +42,7 @@ from ..core.jsonio import dumps, write_json from ..source.catalog import Catalog -from ..unity.reader import SceneGraph, deref, mesh_arrays, texture_image +from ..unity.reader import SceneGraph, deref, mesh_arrays, referent, texture_image # UnityEngine.Rendering.BlendMode BLEND_ONE, BLEND_ZERO, BLEND_SRC_ALPHA, BLEND_ONE_MINUS_SRC_ALPHA = 1, 0, 5, 10 @@ -372,7 +372,7 @@ def material_for(pptr) -> int: if mkey in mat_cache: return mat_cache[mkey] mat_tt = mat.read_typetree() - shader_tt = deref(mat, mat_tt["m_Shader"]).read_typetree() + shader_tt = referent(mat, mat_tt["m_Shader"]).read_typetree() gm = translate_material(mat_tt, shader_tt) for tex_obj, env_ in main_textures(mat, mat_tt, shader_tt["m_ParsedForm"]["m_Name"]): tid = object_key(tex_obj) @@ -381,7 +381,7 @@ def material_for(pptr) -> int: buf = io.BytesIO() texture_image(tex_obj.read()).save(buf, format="PNG") tex_cache[tid] = glb.texture(buf.getvalue(), tex_tt["m_Name"], _sampler(tex_tt)) - info = {"index": tex_cache[tid]} + info: dict = {"index": tex_cache[tid]} sx, sy = env_["m_Scale"]["x"], env_["m_Scale"]["y"] ox, oy = env_["m_Offset"]["x"], env_["m_Offset"]["y"] if (sx, sy, ox, oy) != (1.0, 1.0, 0.0, 0.0): diff --git a/src/nnnotes/content/score.py b/src/nnnotes/content/score.py index 80bce23..8c4cdfe 100644 --- a/src/nnnotes/content/score.py +++ b/src/nnnotes/content/score.py @@ -166,12 +166,12 @@ def judgement_type(op: int, crit: bool) -> int: MATHF_EPSILON = f32(1.401298e-45) # Mathf.Epsilon (denormal min; 1.17549435e-38 if flush-to-zero is on, assumed off) -def mathf_approximately(a: float, b: float) -> bool: +def mathf_approximately(a: float | np.floating, b: float | np.floating) -> bool: """UnityEngine.Mathf.Approximately (inlined in IsSamePosition / EqualMusicInfoPosition).""" - a, b = f32(a), f32(b) - m = max(abs(a), abs(b)) + x, y = f32(a), f32(b) + m = max(abs(x), abs(y)) tol = max(f32(m * f32(1e-6)), f32(MATHF_EPSILON * f32(8.0))) - return bool(abs(f32(b - a)) < tol) + return bool(abs(f32(y - x)) < tol) # --------------------------------------------------------------------------- SsRoot (ReadNote/ReadEvents) diff --git a/src/nnnotes/content/spot.py b/src/nnnotes/content/spot.py index 92b1280..6c05f16 100644 --- a/src/nnnotes/content/spot.py +++ b/src/nnnotes/content/spot.py @@ -182,7 +182,7 @@ def extract(cat: Catalog, md: Path, spot_id: int, out_dir: Path) -> dict: env = UnityPy.load(*[str(p) for p in cat.fetch_key(sp["_situationAssetPath"])]) graph = SceneGraph(env) - by_cls: dict[str, list] = {} + by_cls: dict[str | None, list] = {} for o in env.objects: if o.type.name == "MonoBehaviour": by_cls.setdefault(script_class(o), []).append(o) diff --git a/src/nnnotes/content/story.py b/src/nnnotes/content/story.py index 1209e69..6404df9 100644 --- a/src/nnnotes/content/story.py +++ b/src/nnnotes/content/story.py @@ -105,6 +105,8 @@ def build( code, needed with the UI); `audio_options`: further cri.decode options of every cue sheet (flac_level, also). With audio, the CRI Lips data is read from the catalog's APK when a voice reaches it (needs_cri_lips; story.json `crilips`, else null).""" + if ui and language is None: + raise ValueError("language: the client language of the UI is needed with ui=True") out_dir = Path(out_dir) out_dir.mkdir(parents=True, exist_ok=True) episode = adv.to_json(adv.extract(cat, master, adv_id)) @@ -121,7 +123,11 @@ def build( media = advmedia.extract(cat, player, master, episode, out_dir, fonts=fonts) scene = advscene.extract(cat, player, episode, out_dir, shaders=media["shaders"]) - ui = advui.extract(cat, player, episode, out_dir, fonts=fonts, language=language, master=master) if ui else None + ui_summary = ( + advui.extract(cat, player, episode, out_dir, fonts=fonts, language=language, master=master) + if ui and language is not None + else None + ) videos = advvideo.extract(cat, episode, out_dir) lips = None if audio: @@ -151,7 +157,7 @@ def build( "stages": scene["stages"], "shaders": scene["count"], "textures": scene["textures"], - "uiSummary": ui, + "uiSummary": ui_summary, "media": {"entries": media["counts"], "textures": media["textures"]}, "videoCount": len(videos["videos"]) if videos else 0, } diff --git a/src/nnnotes/content/storyfonts.py b/src/nnnotes/content/storyfonts.py index 3d944e8..afcd601 100644 --- a/src/nnnotes/content/storyfonts.py +++ b/src/nnnotes/content/storyfonts.py @@ -1123,7 +1123,7 @@ def open_fonts( coverage["missingGlyph"] = [chr(u) for u in sorted(substituted)] sprites = None indices = sprite_needs(rec, shown, set().union(*sprite_drawn.values())) if rec is not None else [] - if not indices: # no sprite to draw: the data holds no sprite asset + if rec is None or not indices: # no sprite to draw: the data holds no sprite asset for t in bindings: t.pop("spriteAsset", None) t.pop("m_tintAllSprites", None) @@ -1348,27 +1348,27 @@ def open_sprite_asset(emoji: EmojiFont | None, game: dict, indices: list[int], n where, width, height = pack_cells(cells) if any(pg for pg, _, _ in where.values()): raise NotImplementedError(f"{name}: sprites on more than one page") - page = f"sprite_{name}" if images else None - arr = np.zeros((height, width, 4), np.uint8) if images else None - glyphs = {} - for gi in glyph_ids: - g = game["glyphs"][str(gi)] - rec = { - "metrics": g["metrics"], + glyphs = { + str(gi): { + "metrics": game["glyphs"][str(gi)]["metrics"], "rect": {"m_X": 0, "m_Y": 0, "m_Width": 0, "m_Height": 0}, - "scale": g["scale"], + "scale": game["glyphs"][str(gi)]["scale"], "atlasIndex": 0, } - if gi in images: - a = images[gi] + for gi in glyph_ids + } + page, textures, pages = None, {}, [] + if images: + page = f"sprite_{name}" + arr = np.zeros((height, width, 4), np.uint8) + for gi, a in images.items(): _, x, y = where[gi] x, y = x + SPRITE_MARGIN, y + SPRITE_MARGIN arr[y : y + a.shape[0], x : x + a.shape[1]] = a + rec = glyphs[str(gi)] rec["rect"] = {"m_X": x, "m_Y": y, "m_Width": a.shape[1], "m_Height": a.shape[0]} rec["packed"] = {"texture": page, "dx": 0, "dy": 0} - glyphs[str(gi)] = rec - textures = ( - { + textures = { page: { "texture": f"{PAGES_DIR}/{_safe(page)}.png", "name": page, @@ -1378,9 +1378,7 @@ def open_sprite_asset(emoji: EmojiFont | None, game: dict, indices: list[int], n "settings": dict(PAGE_SETTINGS), } } - if images - else {} - ) + pages = [(page, arr)] material = _sheet_material(game["material"], f"{name} Material", page, width, height) asset = { "faceInfo": game["faceInfo"], @@ -1394,7 +1392,7 @@ def open_sprite_asset(emoji: EmojiFont | None, game: dict, indices: list[int], n return { "asset": asset, "textures": textures, - "pages": [(page, arr)] if images else [], + "pages": pages, "material": material, "missing": missing, } @@ -1409,7 +1407,7 @@ def game_sprite_asset(ex: Exporter, ui_dir: Path, rec: dict) -> dict: if sh["mipCount"] != 1: raise NotImplementedError(f"{rec['name']}: sprite sheet with {sh['mipCount']} mip levels") o = ex.key_object(advui.EMOJI_SPRITE_ASSET_KEY) - sheet = ex.deref(o, o.read_typetree()["spriteSheet"]) + sheet = ex.referent(o, o.read_typetree()["spriteSheet"]) page = f"sprite_{rec['name']}" dst = f"{PAGES_DIR}/{_safe(page)}.png" (Path(ui_dir) / PAGES_DIR).mkdir(parents=True, exist_ok=True) diff --git a/src/nnnotes/content/storyhost.py b/src/nnnotes/content/storyhost.py index 51d283d..dc6cd33 100644 --- a/src/nnnotes/content/storyhost.py +++ b/src/nnnotes/content/storyhost.py @@ -40,7 +40,7 @@ from ..source import languages, master from ..unity import fontsettings, textstyle, tmpfont from ..unity.export import RECT_PROPS, Exporter, TexelPacker, streamed_frames -from ..unity.reader import SceneGraph, deref, mesh_arrays, script_class +from ..unity.reader import SceneGraph, deref, mesh_arrays, referent, script_class from . import advui, room, spot, storyfonts FORMAT = "ournotes.story-host/1" @@ -158,9 +158,9 @@ def camera_target(nodes: list[dict]) -> dict: if crt is None: raise RuntimeError(f"{nodes[0]['path']}: no CameraTargetRenderer") stage, cam = advui._ref_path(crt["_stage"]), advui._ref_path(crt["_captureCamera"]) - container = _parent(stage) - if not stage or not cam or _parent(cam) != container: + if not stage or not cam or _parent(cam) != _parent(stage): raise RuntimeError("CameraTargetRenderer: stage and capture camera not under one parent") + container = _parent(stage) node = _first(nodes, cam) c = [x for x in node["components"] if x["type"] == "Camera"] d = [x for x in node["components"] if x.get("class") == "UniversalAdditionalCameraData"] @@ -453,7 +453,7 @@ def add(self, key: str, nodes: list[dict], subtrees=None, singles=()) -> list[di if not (rec.get("localizeText") or {}).get("localizeEnabled"): raise NotImplementedError(f"{n['path']}: text without LocalizeText") rec["textStyle"] = self.styles.record( - n["path"], advui._component(n, advui.TMP_CLASSES), advui._component(n, ("LocalizeText",)) + n["path"], advui._need_component(n, advui.TMP_CLASSES), advui._component(n, ("LocalizeText",)) ) del rec["text"] self.nodes.append(rec) @@ -651,7 +651,7 @@ def background(cat, key: str, situation_name: str, ex: Exporter) -> dict: for p in mats: mat = p.deref() tt = mat.read_typetree() - sh = deref(mat, tt["m_Shader"]) + sh = referent(mat, tt["m_Shader"]) rows.append((mat, tt, sh.read_typetree()["m_ParsedForm"]["m_Name"], sh)) tex_index = glb_textures([(mat, tt, name) for mat, tt, name, _ in rows]) if len(volumes) > 1: @@ -715,7 +715,7 @@ def spine_records(env, ex: Exporter) -> tuple[dict, list, dict]: SkeletonAnimation's MeshRenderer sorting order / layer and {atlas page: material name} (SpineAtlasAsset materials in page order), and the shaders the materials name. -> (materials, characters, {name: shader}).""" graph = SceneGraph(env) - by_cls: dict[str, list] = {} + by_cls: dict[str | None, list] = {} mr_by_go = {} for o in env.objects: if o.type.name == "MonoBehaviour": @@ -732,7 +732,7 @@ def pages(sda) -> dict: if a is None: continue att = a.read_typetree() - names = spot._atlas_pages(spot._text_bytes(deref(a, att["atlasFile"]).read()).decode("utf-8")) + names = spot._atlas_pages(spot._text_bytes(referent(a, att["atlasFile"]).read()).decode("utf-8")) mats = [deref(a, m) for m in att["materials"]] if len(names) != len(mats): raise RuntimeError(f"atlas {att.get('m_Name')}: {len(names)} pages vs {len(mats)} materials") @@ -938,13 +938,13 @@ def home_ui(ui: UiDoc, keys: dict) -> tuple[str, dict, list]: _transitionOut, which ends with the cover at 0 as well. -> (overlay root, layout root, opened sequences).""" nodes = ui.prefab(keys["spotWidget"]) - w = advui._component(nodes[0], ("UISpotWidget",)) - overlay = advui._ref_path(w["_afterTalkContents"]) - seqs = [advui._ref_path(w[k]) for k in ("_transitionIn", "_transitionOut", "_fadeIn", "_fadeOut") if w.get(k)] + w = advui._need_component(nodes[0], ("UISpotWidget",)) + overlay = advui._need_path(w["_afterTalkContents"]) + seqs = [advui._need_path(w[k]) for k in ("_transitionIn", "_transitionOut", "_fadeIn", "_fadeOut") if w.get(k)] fade = f"{_parent(overlay)}/Fade" ui.add(keys["spotWidget"], nodes, [overlay, fade, *seqs], _call_targets(nodes, seqs)) ui.add(keys["spotLetterbox"], ui.prefab(keys["spotLetterbox"])) - return overlay, layout_root(nodes, overlay), [advui._ref_path(w["_fadeIn"])] + return overlay, layout_root(nodes, overlay), [advui._need_path(w["_fadeIn"])] def _call_targets(nodes: list[dict], seqs: list[str]) -> list[str]: @@ -965,7 +965,7 @@ def afterlive_ui(ui: UiDoc, keys: dict) -> tuple[str, dict, list, dict]: subtree, the widget's and UISoloLiveMusicResultView's _inAnimation sequences and the nodes their calls target. -> (overlay root, layout root, opened sequences, host.json `afterlive`).""" bnodes = ui.prefab(keys["resultBackground"]) - bw = advui._component(bnodes[0], ("UILiveResultBackgroundWidget",)) + bw = advui._need_component(bnodes[0], ("UILiveResultBackgroundWidget",)) bg_path = advui._ref_path(bw["_backgroundImage"]) recs = [r for r in ui.add(keys["resultBackground"], bnodes) if r["path"] == bg_path and r.get("image")] if len(recs) != 1: @@ -973,8 +973,8 @@ def afterlive_ui(ui: UiDoc, keys: dict) -> tuple[str, dict, list, dict]: sprite = ui.sprite(keys["resultBackgroundSprite"]) recs[0]["image"].update(sprite=sprite["name"], spriteRef=sprite["spriteRef"]) nodes = ui.prefab(keys["resultWidget"]) - rw = advui._component(nodes[0], ("UILiveResultWidget",)) - overlay = advui._ref_path(rw["_advSimpleContents"]) + rw = advui._need_component(nodes[0], ("UILiveResultWidget",)) + overlay = advui._need_path(rw["_advSimpleContents"]) view = [ (n, c) for n in nodes @@ -984,9 +984,9 @@ def afterlive_ui(ui: UiDoc, keys: dict) -> tuple[str, dict, list, dict]: if len(view) != 1: raise RuntimeError(f"{overlay}: {len(view)} UISoloLiveMusicResultView ancestors") seqs = [ - advui._ref_path(bw["_inAnimation"]), - advui._ref_path(rw["_inAnimation"]), - advui._ref_path(view[0][1]["_inAnimation"]), + advui._need_path(bw["_inAnimation"]), + advui._need_path(rw["_inAnimation"]), + advui._need_path(view[0][1]["_inAnimation"]), ] ui.add(keys["resultWidget"], nodes, [overlay, *seqs[1:]], _call_targets(nodes, seqs[1:])) parts = overlay.split("/") @@ -1272,7 +1272,7 @@ def text_bindings(ui: UiDoc, lang: dict, mode: int) -> dict[str, dict]: path = rec["path"] if path in out: raise RuntimeError(f"{path}: two text nodes") - c = advui._component(n, advui.TMP_CLASSES) + c = advui._need_component(n, advui.TMP_CLASSES) t = {k: c[k] for k in advui.TMP_FIELDS if k in c} t.update( { @@ -1323,6 +1323,8 @@ def open_fonts( mode, field = languages.mode(language), languages.column(language)[1:] lang = textstyle.language_fonts(player) bindings = text_bindings(ui, lang, mode) + if ui.doc is None: + raise RuntimeError("the simple UI's texts are read from its written ui.json") shown = ( storyfonts.shown_texts(episode, story_ui, master_dir, field) + storyfonts.shown_texts(NO_EPISODE, ui.doc, master_dir, field) @@ -1440,12 +1442,12 @@ def build( lost = [lang for lang in language_dirs if lang not in font_files] if lost: raise ValueError(f"no font file for {lost}") - if font_of is None: - def font_of(code): - return font_files[code] if code in font_files else storyfonts.no_font_file(code) + def own_font(code): + return font_files[code] if code in font_files else storyfonts.no_font_file(code) - keys = {k: emb_key(cat, ADDRESSES[k]) for k in ("cameraTarget", "talkWindow")} + font_of = font_of or own_font + keys: dict = {k: emb_key(cat, ADDRESSES[k]) for k in ("cameraTarget", "talkWindow")} if host.get("talk") == "area": keys["systemMessage"] = emb_key(cat, ADDRESSES["systemMessage"]) for k in KIND_KEYS[host["kind"]]: diff --git a/src/nnnotes/content/ui.py b/src/nnnotes/content/ui.py index a203011..1e929b8 100644 --- a/src/nnnotes/content/ui.py +++ b/src/nnnotes/content/ui.py @@ -284,7 +284,7 @@ def save_index(): for key in keys: previous = assets.get(key) - if not force and _valid_record(out, previous): + if not force and previous is not None and _valid_record(out, previous): refs = previous.get("dependencies", {}) valid_deps = ( not dependencies diff --git a/src/nnnotes/core/cache.py b/src/nnnotes/core/cache.py index 7844665..f74d054 100644 --- a/src/nnnotes/core/cache.py +++ b/src/nnnotes/core/cache.py @@ -28,6 +28,7 @@ from collections import OrderedDict from importlib import metadata from pathlib import Path +from typing import Any DISK_MAGIC = b"NNC1" DEFAULT_MEMORY_MB = 128 @@ -169,7 +170,7 @@ def _limit(self) -> int: budget = _settings["memory"] return budget // 4 if self._max_item is None else min(self._max_item, budget) - def get(self, k: str): + def get(self, k: str) -> Any: """The value stored under `k`, or None.""" if not _settings["enabled"]: return None diff --git a/src/nnnotes/formats/cri.py b/src/nnnotes/formats/cri.py index 195d05c..c617df5 100644 --- a/src/nnnotes/formats/cri.py +++ b/src/nnnotes/formats/cri.py @@ -161,6 +161,8 @@ def _checked(acb: bytes, cue_sheet: str, layout: str) -> bytes: def _asset_acb(cue_sheet: str, layout: str, objs, t: dict, impl: dict | None) -> bytes: if layout == SPLIT_ACB: return split_acb([text_bytes(objs[c["m_PathID"]].read().m_Script) for c in t["_chunks"]], cue_sheet) + if impl is None: + raise RuntimeError(f"{cue_sheet}: {layout} without its serialized bytes") return embedded_acb(t, impl, cue_sheet) @@ -306,8 +308,7 @@ def decode_acb_bytes( raise NotImplementedError(f"cue sheet {name}: external AWB (streamed waveforms) not supported") _check_format(fmt, also) vgm = vgmstream or tool("vgmstream", "vgmstream-cli") - if ffmpeg is None and (fmt != "wav" or also): - ffmpeg = tool("ffmpeg", "ffmpeg") + encoder = (ffmpeg or tool("ffmpeg", "ffmpeg")) if fmt != "wav" or also else "" # "": no ffmpeg run out_dir = Path(out_dir) work = out_dir / "_work" work.mkdir(parents=True, exist_ok=True) @@ -352,7 +353,7 @@ def one(item) -> Path: subprocess.run([vgm, "-i", "-s", str(i), "-o", str(wav), str(acb_file)], check=True, capture_output=True) if also is not None: r = subprocess.run( - [ffmpeg, *encode_args(wav, out_dir / f"{stem}{also[0]}", also[1])], capture_output=True, text=True + [encoder, *encode_args(wav, out_dir / f"{stem}{also[0]}", also[1])], capture_output=True, text=True ) if r.returncode != 0: raise RuntimeError(f"ffmpeg {also[0]} failed on {name} stream {i}: {r.stderr[-2000:]}") @@ -365,7 +366,7 @@ def one(item) -> Path: "flac": ["-c:a", "flac", "-compression_level", str(int(flac_level))], "ogg": ["-c:a", "libvorbis", "-q:a", "5"], }[fmt] - subprocess.run([ffmpeg, "-y", "-loglevel", "error", "-i", str(wav), *codec, str(dst)], check=True) + subprocess.run([encoder, "-y", "-loglevel", "error", "-i", str(wav), *codec, str(dst)], check=True) wav.unlink() return dst diff --git a/src/nnnotes/formats/usm.py b/src/nnnotes/formats/usm.py index 4e28c73..c71ca3c 100644 --- a/src/nnnotes/formats/usm.py +++ b/src/nnnotes/formats/usm.py @@ -309,7 +309,7 @@ def video_facts(s: Stream) -> dict: if not h and s.codec == "vp9": w, ht, rate, scale, frames = struct.unpack_from(" dict: raise ValueError(f"not an HCA stream: {hca[:4].hex()}") size = struct.unpack_from(">H", hca, 6)[0] head = bytes(b & 0x7F for b in hca[:size]) - info = {"codec": "hca"} + info: dict = {"codec": "hca"} i = head.find(b"fmt\x00") if i < 0: raise ValueError("HCA header without fmt") diff --git a/src/nnnotes/musicdata/__init__.py b/src/nnnotes/musicdata/__init__.py index b1d0ded..53e6fce 100644 --- a/src/nnnotes/musicdata/__init__.py +++ b/src/nnnotes/musicdata/__init__.py @@ -28,9 +28,10 @@ import math import sys from collections import Counter -from collections.abc import Callable +from collections.abc import Callable, Mapping from decimal import Decimal, localcontext from pathlib import Path +from typing import TypeGuard from ..source.languages import LANGUAGES from . import deckdata @@ -233,7 +234,9 @@ def jacket_webp(image, size: int = JACKET_SIZE) -> bytes: img = img.convert("RGB") w, h = img.size if max(w, h) > size: - img = img.resize((max(1, round(w * size / max(w, h))), max(1, round(h * size / max(w, h)))), Image.LANCZOS) + img = img.resize( + (max(1, round(w * size / max(w, h))), max(1, round(h * size / max(w, h)))), Image.Resampling.LANCZOS + ) buf = io.BytesIO() img.save(buf, format="WEBP", quality=JACKET_QUALITY, method=6) return buf.getvalue() @@ -407,7 +410,7 @@ def __init__( self.module, self.seeds, self.workers = module, seeds, 1 if workers == 0 else workers self.aptitude = aptitude self.cache = None if cache is None else Path(cache) - self.counts: dict | None = None # native counters, or None when no cache was requested + self.counts: Mapping[str, object] | None = None # native counters, or None when no cache was requested def info(self) -> dict: """The identity and statistics format of the installed model.""" @@ -489,7 +492,7 @@ def rank_bonus_percents(rows: list[dict], missions) -> list[list]: """The rank bonus percentages [range][rank - 1] of a song's missions from the MasterLiveGekisouRankingScoreBonus rows of its mission pattern (`_count`: the range 1..3, `_rank` 1..5; a later row wins; 0 without a row).""" pattern = mission_pattern(missions) - out = [[0] * RANKS for _ in range(GEKISOU_RANGES)] + out: list[list[int | None]] = [[0] * RANKS for _ in range(GEKISOU_RANGES)] for r in rows: c, k = (r.get("_count") or 0) - 1, (r.get("_rank") or 0) - 1 if r.get("_missionPattern") == pattern and 0 <= c < GEKISOU_RANGES and 0 <= k < RANKS: @@ -497,7 +500,7 @@ def rank_bonus_percents(rows: list[dict], missions) -> list[list]: return out -def _int(v) -> bool: +def _int(v) -> TypeGuard[int]: return isinstance(v, int) and not isinstance(v, bool) @@ -1489,12 +1492,13 @@ def export( } if not full: del doc["master"], doc["charts"] - jackets = sorted({s["jacket"] for s in doc["songs"] if s["jacket"]}) if jacket is not None else [] - images = {name: jacket(name) for name in jackets} + images = {} + if jacket is not None and jackets_dir is not None: + images = {name: jacket(name) for name in sorted({s["jacket"] for s in doc["songs"] if s["jacket"]})} data = deckdata.encode(deckdata._value(doc, "music data")) except deckdata.DeckDataError as e: raise MusicDataError(str(e)) from None - if jackets: + if images and jackets_dir is not None: d = Path(jackets_dir) d.mkdir(parents=True, exist_ok=True) for name, image in images.items(): @@ -1518,7 +1522,7 @@ def export( "unplayable": sum(1 for c in charts if c["deck"] and c["deck"]["unplayable"]), "full": full, "bgm": bgm is not None, - "jackets": len(jackets), + "jackets": len(images), "bytes": len(data), "fileBytes": len(written), "sha256": hashlib.sha256(written).hexdigest(), diff --git a/src/nnnotes/musicdata/deckdata.py b/src/nnnotes/musicdata/deckdata.py index 8556550..9a4a3ed 100644 --- a/src/nnnotes/musicdata/deckdata.py +++ b/src/nnnotes/musicdata/deckdata.py @@ -28,6 +28,7 @@ from collections.abc import Callable from dataclasses import dataclass from pathlib import Path +from typing import Any from ..source.apkset import ApkSet @@ -51,7 +52,7 @@ # (table, columns), in file order. A table with a column list exports those columns; a whole table all of its own. TABLES: tuple[tuple[str, tuple[str, ...] | None], ...] = tuple( - (t, None if c is _WHOLE else tuple(c.split())) + (t, None if c is None else tuple(c.split())) for t, c in ( ( "MasterMemberCard", @@ -321,7 +322,7 @@ def f32_token(v: float) -> Num: return Num(np.format_float_scientific(x, unique=True, trim="0")) -def _value(v, where: str): +def _value(v, where: str) -> Any: if isinstance(v, float): try: return f32_token(v) diff --git a/src/nnnotes/musicdata/replaydata.py b/src/nnnotes/musicdata/replaydata.py index 218c161..85f5641 100644 --- a/src/nnnotes/musicdata/replaydata.py +++ b/src/nnnotes/musicdata/replaydata.py @@ -71,11 +71,13 @@ def _read_package(source: Path, paths: tuple[str, ...], name: str) -> dict[str, return {path: (source / path).read_bytes() for path in paths if (source / path).is_file()} try: with tarfile.open(source, "r:gz") as archive: - return { - member.name.partition("/")[2]: archive.extractfile(member).read() - for member in archive.getmembers() - if member.isfile() and member.name.partition("/")[2] in paths - } + out = {} + for member in archive.getmembers(): + path = member.name.partition("/")[2] + f = archive.extractfile(member) if member.isfile() and path in paths else None + if f is not None: + out[path] = f.read() + return out except (OSError, tarfile.TarError): raise deckdata.DeckDataError(f"{name}: {source} is neither a package directory nor a .tar.gz package") from None @@ -105,7 +107,8 @@ def _package( raise deckdata.DeckDataError( f"{name}: {BUILD_INFO} commit {info.get('commit')!r} differs from the pinned deck model {model['commit']}" ) - hashes = info.get("files") if isinstance(info.get("files"), dict) else {} + files_info = info.get("files") + hashes = files_info if isinstance(files_info, dict) else {} for path in (js, wasm): if path not in package: raise deckdata.DeckDataError(f"{name}: missing {path} in {source}") @@ -113,10 +116,9 @@ def _package( raise deckdata.DeckDataError(f"{name}: {path} SHA-256 differs from {BUILD_INFO}") if not package[wasm].startswith(b"\x00asm\x01\x00\x00\x00"): raise deckdata.DeckDataError(f"{name}: not a WASM v1 module") - paths = tuple(f"{prefix}/{path.rsplit('/', 1)[-1]}" for path in (js, wasm, BUILD_INFO)) - for path, data in zip(paths, (package[js], package[wasm], package[BUILD_INFO])): - files[path] = data - return paths + js_path, wasm_path, info_path = (f"{prefix}/{path.rsplit('/', 1)[-1]}" for path in (js, wasm, BUILD_INFO)) + files[js_path], files[wasm_path], files[info_path] = package[js], package[wasm], package[BUILD_INFO] + return js_path, wasm_path, info_path def bundle( diff --git a/src/nnnotes/pipeline/atoms/astc.py b/src/nnnotes/pipeline/atoms/astc.py index e6a2877..0f1b0d8 100644 --- a/src/nnnotes/pipeline/atoms/astc.py +++ b/src/nnnotes/pipeline/atoms/astc.py @@ -8,6 +8,7 @@ from __future__ import annotations import struct +from collections.abc import Sequence from . import estimate @@ -37,17 +38,18 @@ def level0_size(width: int, height: int, block: tuple[int, int]) -> int: return -(-width // bx) * -(-height // by) * 16 -def container(blocks: bytes, width: int, height: int, block: tuple[int, int]) -> bytes: +def container(blocks: bytes, width: int, height: int, block: Sequence[int]) -> bytes: """The .astc file of a `width` x `height` 2D image whose blocks (of `block` texels) start `blocks`.""" - block = tuple(block) - if block not in BLOCK_SIZES: - raise ValueError(f"ASTC block {block}: not a 2D ASTC block size") + size = tuple(block) + if size not in BLOCK_SIZES: + raise ValueError(f"ASTC block {size}: not a 2D ASTC block size") + bx, by = size if not (0 < width < 1 << 24 and 0 < height < 1 << 24): raise ValueError(f"ASTC image {width}x{height}: sizes must be 1 .. 2^24 - 1") - need = level0_size(width, height, block) + need = level0_size(width, height, (bx, by)) if len(blocks) < need: - raise ValueError(f"ASTC {width}x{height} {block[0]}x{block[1]}: {len(blocks)} bytes, {need} needed") - head = struct.pack(" str: class BundleView(Protocol): def files(self) -> list[str]: """Serialized file names, in bundle order.""" + ... def objects(self) -> list[ObjectInfo]: """Every object, file by file in bundle order, each file's objects in stored order.""" + ... def raw(self, obj: ObjectInfo) -> bytes: """The object's serialized bytes.""" + ... def typetree(self, obj: ObjectInfo) -> dict: """The object read through its typetree, fields in stored order.""" + ... def container(self) -> list[tuple[str, str, int]]: """(container path, serialized file, path id) of every container entry, in stored order.""" + ... def externals(self, file: str) -> list[str]: """The paths of a serialized file's externals (PPtr file id i names externals[i - 1]).""" + ... def unity_version(self, file: str) -> str: """The Unity version a serialized file was written by.""" + ... def texture_input(self, obj: ObjectInfo) -> TextureInput: ... @@ -208,7 +218,7 @@ class UnityBundle: def __init__(self, env): self.env = env self._files = {f.name: f for f in self.env.assets} - self._readers: dict[tuple[str, int], object] = {} + self._readers: dict[tuple[str, int], ObjectReader] = {} self._infos: list[ObjectInfo] = [] for name, f in self._files.items(): for pid, r in f.objects.items(): diff --git a/src/nnnotes/pipeline/atoms/texture.py b/src/nnnotes/pipeline/atoms/texture.py index f96cec9..3e00b60 100644 --- a/src/nnnotes/pipeline/atoms/texture.py +++ b/src/nnnotes/pipeline/atoms/texture.py @@ -69,7 +69,14 @@ def decode(inp: TextureInput): ) try: return parse_image_data( - inp.data, inp.width, inp.height, fmt, tuple(inp.version), inp.platform, inp.platform_blob, True + inp.data, + inp.width, + inp.height, + fmt, + tuple(inp.version), + inp.platform, + inp.platform_blob, # pyright: ignore[reportArgumentType] (UnityPy reads the blob's bytes) + True, ) except NotImplementedError as e: raise Unsupported("unsupported.texture.format", str(e)) from None diff --git a/src/nnnotes/pipeline/contract.py b/src/nnnotes/pipeline/contract.py index 00fb993..f2dd92b 100644 --- a/src/nnnotes/pipeline/contract.py +++ b/src/nnnotes/pipeline/contract.py @@ -469,7 +469,7 @@ def item( raise ValueError(f"item {obj}: `in` is for contained items only") if status in ("contained", "failed") and artifacts: raise ValueError(f"item {obj}: {status} with artifacts") - d = {"object": obj, "status": status} + d: dict = {"object": obj, "status": status} if cls is not None: d["class"] = cls if artifacts: diff --git a/src/nnnotes/pipeline/layout.py b/src/nnnotes/pipeline/layout.py index 9659b15..6bc7495 100644 --- a/src/nnnotes/pipeline/layout.py +++ b/src/nnnotes/pipeline/layout.py @@ -573,7 +573,8 @@ def _why(e: OSError, what: str) -> str: return "store on another file system" if "platform" in str(e): return f"{what} unsupported on this platform" - return f"{what} unsupported on this file system ({errno.errorcode.get(e.errno, e.errno)})" + code = e.errno if e.errno is None else errno.errorcode.get(e.errno, e.errno) + return f"{what} unsupported on this file system ({code})" def probe(out_dir, store_dir, want: str = "auto") -> tuple[str, str]: diff --git a/src/nnnotes/pipeline/modelstages.py b/src/nnnotes/pipeline/modelstages.py index cb5e35c..b7059dc 100644 --- a/src/nnnotes/pipeline/modelstages.py +++ b/src/nnnotes/pipeline/modelstages.py @@ -274,7 +274,8 @@ def run(self, task, store) -> Output: cat = TaskCatalog.from_task(task, store) arts = [] with tempfile.TemporaryDirectory(prefix="nnnotes-live2d-") as d: - summary = live2d.extract_model(cat, task.subject, Path(d)) + # TaskCatalog answers the catalog calls of the model extractor + summary = live2d.extract_model(cat, task.subject, Path(d)) # pyright: ignore[reportArgumentType] for rel, p in _tree(Path(d)): kind = live2d_kind(rel) facts = {k: summary[k] for k in sorted(summary)} if kind == "live2d.model3" else {} @@ -422,7 +423,7 @@ def run(self, task, store) -> Output: written: dict[str, bytes] = {} try: sf = env.get_cab(file) - o = sf.objects[pid] if sf is not None and pid in getattr(sf, "objects", {}) else None + o = getattr(sf, "objects", {}).get(pid) if o is None: raise KeyError(f"{oid} is not in the task's bundles") record = spot.write_skeleton(o, written.__setitem__) diff --git a/src/nnnotes/pipeline/objexport.py b/src/nnnotes/pipeline/objexport.py index bc511d6..2e885bd 100644 --- a/src/nnnotes/pipeline/objexport.py +++ b/src/nnnotes/pipeline/objexport.py @@ -67,6 +67,7 @@ from functools import cache from pathlib import Path from types import SimpleNamespace +from typing import TYPE_CHECKING, cast from . import census as census_mod from . import contract, link @@ -75,6 +76,12 @@ from .contract import Cost, IncompatibleTask, Input from .stages import Output, Pending, Stage +if TYPE_CHECKING: # the exporter's base; exporter_type combines them when UnityPy is loaded + from ..unity.export import Exporter as _ExporterBase + from ..unity.reader import SceneGraph +else: + _ExporterBase = object + EXPORT, CROP, CENSUS, SCRIPTS = "unity.export", "sprite.crop", "unity.census", "link.scripts" SCRIPTS_TASK = contract.task_id(SCRIPTS, "all") @@ -159,7 +166,7 @@ def converter_id(name: str) -> str: def converters_for(classes) -> list[str]: """The converters a bundle holding objects of `classes` may use.""" - out = set(ALWAYS) + out: set[str] = set(ALWAYS) for c in classes: out.update(CLASS_CONVERTERS.get(c, ())) return sorted(out) @@ -297,13 +304,14 @@ def read(self): # ---------------------------------------------------------------- the exporter -class RefExporter: +class RefExporter(_ExporterBase): """unity.export on one bundle. Combined with export.Exporter (exporter_type), whose material, clip and controller code runs here in reference mode: `ref` gives $ref values, `deref` resolves inside the bundle only (a MonoScript of another bundle from the script table), and MonoBehaviour classes come from the task's script table.""" def __init__(self, view, task, store, fns: dict): - super().__init__(None, Path("."), class_of=self._script_name) + # no catalog: the objects come from the task's bundle view + super().__init__(None, Path("."), class_of=self._script_name) # pyright: ignore[reportArgumentType] self.view, self.task, self.store, self.fns = view, task, store, fns self.blob_min = int(task.params["blobMin"]) self.png_level = int(task.params["png"]["level"]) @@ -329,7 +337,7 @@ def __init__(self, view, task, store, fns: dict): self.children_memo: dict[int, tuple] = {} self.heads: dict[int, tuple] = {} self.atlas_entries: dict[tuple, list] = {} - self.graphs: dict[str, object] = {} + self.graphs: dict[str, SceneGraph] = {} self.member: dict[tuple, tuple] = {} self.comp_go: dict[tuple, int] = {} self.roots: list[tuple] = [] @@ -519,9 +527,9 @@ def _special(self, node) -> bool: self.special_memo[id(node)] = (node, r) return r - def tree(self, owner, tt: dict): + def tree(self, owner, tt: dict) -> dict: """The object JSON of `owner`'s typetree `tt` (read with owner's type tree).""" - return self._walk(owner, owner._get_typetree_node(), tt, "") + return cast(dict, self._walk(owner, owner._get_typetree_node(), tt, "")) def _walk(self, owner, node, v, path: str): if not self._special(node): @@ -1210,12 +1218,13 @@ def _split_acb(self, key: tuple, acb: dict) -> str: """The `acb` artifact of a SplitAcbData: its chunk TextAssets (of this bundle) joined and unmasked.""" from ..formats import cri - owner, parts = self.readers[key], [] + owner, parts, chunks = self.readers[key], [], [] for c in acb["chunks"]: k = self._target(owner, c) if isinstance(c, dict) else None if k is None or k not in self.info or self.info[k].class_name != "TextAsset": raise ValueError(f"cue sheet {acb['cueSheet']}: chunk {c} is not a TextAsset of the bundle") parts.append(cri.text_bytes(self._tt(k).get("m_Script", ""))) + chunks.append(k) data = cri.split_acb(parts, acb["cueSheet"]) facts = {"cueSheet": _label(acb["cueSheet"]), "layout": SPLIT_ACB, "chunks": len(parts)} return self._emit( @@ -1226,7 +1235,7 @@ def _split_acb(self, key: tuple, acb: dict) -> str: ACB_KIND, "acb", facts, - refs=[{"rel": "chunk", "object": contract.object_id(*self._target(owner, c))} for c in acb["chunks"]], + refs=[{"rel": "chunk", "object": contract.object_id(*k)} for k in chunks], )[0] def conv_generic(self, info) -> list: diff --git a/src/nnnotes/pipeline/orchestrate.py b/src/nnnotes/pipeline/orchestrate.py index bb534cf..b3b7f62 100644 --- a/src/nnnotes/pipeline/orchestrate.py +++ b/src/nnnotes/pipeline/orchestrate.py @@ -31,12 +31,16 @@ from collections import Counter from dataclasses import dataclass, field from multiprocessing.connection import wait +from typing import TYPE_CHECKING from ..source.config import ConfigError from . import contract from .contract import Task from .stages import Env, Pending, Stage, describe, execute +if TYPE_CHECKING: + from .store import Store + WORKER_BYTES = 70 << 20 # resident memory of an idle worker process (interpreter and imports), per worker # failure codes of tasks (failed items carry the codes of their stages) FAILURE_CODES = ( @@ -247,7 +251,7 @@ class Run: The reports (summary, failures, coverage) come from one reading of every result (scan), kept while the tasks stay the same.""" - store: object + store: Store context: dict selection: list params: dict @@ -263,7 +267,7 @@ def results(self): for tid in sorted(self.tasks): t = self.tasks[tid] if t["status"] != "failed": - yield tid, self.store.result(t["key"]) + yield tid, self.store.committed(t["key"], tid) def scan(self, visit=None, census=None) -> None: """Read every result once, by task id, for the reports; `visit(task id, result)` sees each as well (what @@ -285,6 +289,7 @@ def _tasks_now(self) -> tuple: def _digest(self) -> Digest: if self._scan is None or self._scan[0] != self._tasks_now(): self.scan() + assert self._scan is not None return self._scan[1] def failures(self) -> list[dict]: @@ -370,6 +375,8 @@ def _key(self, oid: str): return self._files.setdefault(file, len(self._files)) * _PATH_IDS + n % _PATH_IDS def add(self, tid: str, doc: dict) -> None: + if self.settled is not None: + raise RuntimeError("the digest is settled: no result can be added") self.artifacts += len(doc.get("artifacts", ())) best, kinds = self.best, self.kinds for it in doc["items"]: @@ -393,7 +400,7 @@ def add(self, tid: str, doc: dict) -> None: def settle(self, census: dict | None) -> None: """Make the coverage document for `census` now and drop the per-object state (no object can be added).""" self.settled = (census, self.coverage(census)) - self.best = self.kinds = self.reasoned = self._files = None + self.best, self.kinds, self.reasoned, self._files = {}, {}, {}, {} def coverage(self, census: dict | None = None) -> dict: """The coverage document (nnnotes.coverage/1): per class the count of every item status and the census @@ -516,7 +523,7 @@ def run( for name in pipeline: self._stage(self.stages[name], norm[name], env, force, run) except ConfigError as e: - e.run = run # what ran before the stop: its results stay valid + e.run = run # pyright: ignore[reportAttributeAccessIssue] (what ran before the stop: its results stay valid) raise finally: self.close() @@ -633,6 +640,7 @@ def _pool_run(self, sched: Scheduler, force, replies: dict, run: Run) -> None: if w.conn not in ready and w.proc.sentinel not in ready: continue t = w.task + assert t is not None # a busy worker has its task reply = None try: if w.conn.poll(): diff --git a/src/nnnotes/pipeline/stages.py b/src/nnnotes/pipeline/stages.py index afdfd0a..95ab75f 100644 --- a/src/nnnotes/pipeline/stages.py +++ b/src/nnnotes/pipeline/stages.py @@ -237,7 +237,8 @@ def execute(task: Task, store, stages: dict[str, Stage], *, force: bool = False) doc = contract.result(task, out.artifacts, out.items) store.commit(doc) own = time.process_time() - cpu - child = None if children is None else max(0.0, children_cpu() - children) + now = children_cpu() + child = None if children is None or now is None else max(0.0, now - children) cost = { "schema": contract.COST, "key": task.key, diff --git a/src/nnnotes/pipeline/store.py b/src/nnnotes/pipeline/store.py index a3218aa..8e8acc5 100644 --- a/src/nnnotes/pipeline/store.py +++ b/src/nnnotes/pipeline/store.py @@ -169,7 +169,7 @@ def __init__(self, root, *, cache=None, fetch=None): self._have: set[str] = set() # content ids this store object wrote or found self._memo: dict | None = None # (path, size, mtime_ns) -> (sha256, size) self._names: dict[str, dict] = {} # kind -> {name: (sha256, size)} - self._docs: dict[str, object] = {} # the documents document() decoded last + self._docs: dict[str, dict] = {} # the documents document() decoded last self._lock = threading.Lock() def __getstate__(self): # a store crosses to worker processes as its settings @@ -287,7 +287,7 @@ def add_chunks(self, chunks, ext: str, media: str | None = None) -> dict: def read(self, sha: str) -> bytes: return self.path(sha).read_bytes() - def document(self, sha: str): + def document(self, sha: str) -> dict: """The JSON document stored as `sha`, decoded; the last two decoded are kept (a catalog index is read at planning and again by the stages it describes) and shared: callers must not change them.""" with self._lock: @@ -311,6 +311,13 @@ def result(self, key: str) -> dict | None: data = _read(self.result_path(key)) return None if data is None else contract.loads(data) + def committed(self, key: str, task: str) -> dict: + """The committed result of a task key that has one (a FileNotFoundError naming `task` when it is missing).""" + doc = self.result(key) + if doc is None: + raise FileNotFoundError(f"{task}: its result {key} is not in the store") + return doc + def has_result(self, key: str) -> bool: return self.result_path(key).is_file() @@ -382,7 +389,7 @@ def identify(self, path, kind: str = "files", name: str | None = None) -> tuple[ } _append(self.root / "inputs" / f"{kind}.jsonl", json.dumps(line, ensure_ascii=False, sort_keys=True) + "\n") with self._lock: - self._memo[memo_key] = hit + self._load_memo()[memo_key] = hit if name is not None: self._names[kind][name] = hit return hit diff --git a/src/nnnotes/pipeline/views.py b/src/nnnotes/pipeline/views.py index 061b25f..926956d 100644 --- a/src/nnnotes/pipeline/views.py +++ b/src/nnnotes/pipeline/views.py @@ -43,7 +43,7 @@ import re import string from collections import OrderedDict -from collections.abc import Iterable, Mapping +from collections.abc import Iterable, Mapping, Sequence from dataclasses import dataclass from importlib import resources from pathlib import Path @@ -112,10 +112,12 @@ class Addresses(Protocol): def keys(self, prefix: str = "") -> list[str]: """The catalog keys starting with `prefix`, sorted.""" + ... def objects(self, key: str) -> list[Obj] | None: """The objects under `key` in the order its bundle stores them (None: not a catalog key; []: a key whose container has no census object).""" + ... class AddressIndex: @@ -215,8 +217,9 @@ def subset(self) -> dict: # ---------------------------------------------------------------- master tables class Tables(Protocol): - def __getitem__(self, name: str) -> list[dict]: + def __getitem__(self, name: str, /) -> list[dict]: """The rows of a master table (KeyError when there is no such table).""" + ... def table_rows(data: bytes) -> list[dict]: @@ -248,7 +251,7 @@ def sha256(self, name: str) -> str: def load_rules(path=None) -> dict: """The view rules: `path`, else the package's viewrules.json; checked (check_rules).""" if path is None: - text = resources.files(__package__).joinpath(RULES_FILE).read_text(encoding="utf-8") + text = resources.files("nnnotes.pipeline").joinpath(RULES_FILE).read_text(encoding="utf-8") else: text = Path(path).read_text(encoding="utf-8") rules = json.loads(text) @@ -463,7 +466,7 @@ def resolve(index: Addresses, address: str, expect=ANY) -> dict: missing.""" key, sub = parse_address(address) objs = index.objects(key) - out = {"address": address} + out: dict = {"address": address} if not objs: out.update(status="missing-key", detail="not a catalog key" if objs is None else "no census objects") return out @@ -679,7 +682,7 @@ def _prefix_pattern(p: str) -> tuple[str, re.Pattern]: return literal, re.compile(re.escape(p).replace(r"\*", "[^/]+")) -def prefix_keys(index: Addresses, prefixes: list[str], exclude: list[str] = ()) -> list[str]: +def prefix_keys(index: Addresses, prefixes: list[str], exclude: Sequence[str] = ()) -> list[str]: """The catalog keys matching a prefix (`*` stands for one path segment) and no exclusion, sorted.""" out = set() for p in prefixes: @@ -810,7 +813,9 @@ def diff(old: dict, new: dict) -> dict: kind = "moved" else: kind = "object" - row, role, each, _ = x or y + entry = x or y + assert entry is not None # the states differ: the entry is in one of the documents + row, role, each, _ = entry c = {"row": row, "role": role, "kind": kind, "old": sx, "new": sy} if each: c["each"] = each diff --git a/src/nnnotes/pipeline/voices.py b/src/nnnotes/pipeline/voices.py index 3803ce7..d0f80e4 100644 --- a/src/nnnotes/pipeline/voices.py +++ b/src/nnnotes/pipeline/voices.py @@ -323,7 +323,8 @@ def __init__(self, rules: dict, index, audio: dict): self.reached: dict[str, set] = {} def __call__(self, sheet, cue) -> dict: - out = {"availability": {"master": sheet is not None and cue is not None, "catalog": None, "exported": False}} + master = sheet is not None and cue is not None + out: dict = {"availability": {"master": master, "catalog": None, "exported": False}} if sheet is None or cue is None: return out present = self.index.objects(self.key.format(sheet=sheet)) is not None @@ -481,7 +482,7 @@ def _code_sources(m: _Master, resolve: _Resolver) -> list[dict]: return out -def _inferred(m: _Master, sheet: str | None, cue: str, near: dict) -> dict: +def _inferred(m: _Master, sheet: str | None, cue: str | None, near: dict) -> dict: out = {} for s in m.v.get("situations", []): if re.fullmatch(s["cue"], cue or ""): @@ -660,7 +661,7 @@ def text_of(tid): "matches": matches, } res, why = sound_gap, None - if res is None: + if sounds is not None: s = sounds.get(vid) if s is None: why = f"{keys['sound']} has no row {vid}" @@ -668,7 +669,7 @@ def text_of(tid): cue = s.get(sd["cue"]) or None row["sound"].update(category=s.get(sd["category"]), sheet=None, cue=cue) res = sheet_gap - if res is None: + if sheets is not None: sh = sheets.get(s.get(sd["sheet"])) if sh is None: why = f"{keys['sheets']} has no row {s.get(sd['sheet'])}" @@ -966,7 +967,7 @@ def _episodes(self, env, index, tables, files: dict, exports: set) -> tuple[dict oid = pick[0].id stable = files.get(contract.parse_object_id(oid)[0]) tid = contract.task_id(EXPORT_STAGE, stable) if stable is not None else None - if tid not in exports: + if tid is None or tid not in exports: continue if tid not in arts: arts[tid] = {a["id"]: a for a in env.result(tid)["artifacts"]} diff --git a/src/nnnotes/schemas/__init__.py b/src/nnnotes/schemas/__init__.py index 87f4af6..ef742d6 100644 --- a/src/nnnotes/schemas/__init__.py +++ b/src/nnnotes/schemas/__init__.py @@ -11,12 +11,13 @@ def names() -> list[str]: """The schema names (`music-data`, `plan`, ...), sorted.""" - return sorted(p.name[: -len(SUFFIX)] for p in resources.files(__package__).iterdir() if p.name.endswith(SUFFIX)) + files = resources.files("nnnotes.schemas") + return sorted(p.name[: -len(SUFFIX)] for p in files.iterdir() if p.name.endswith(SUFFIX)) def load(name: str) -> dict: """The schema `name` (`music-data` for music-data.schema.json).""" - path = resources.files(__package__) / f"{name}{SUFFIX}" + path = resources.files("nnnotes.schemas") / f"{name}{SUFFIX}" if not path.is_file(): raise KeyError(f"no schema {name!r} (schemas: {', '.join(names())})") return json.loads(path.read_text(encoding="utf-8")) diff --git a/src/nnnotes/site/sitefiles.py b/src/nnnotes/site/sitefiles.py index 5622854..158faae 100644 --- a/src/nnnotes/site/sitefiles.py +++ b/src/nnnotes/site/sitefiles.py @@ -336,7 +336,7 @@ def region_meta(cfg: Config, regions: list[str]) -> list[dict]: client languages (`[servers.] languages`).""" out = [] for r in regions: - rec = {"id": r, "name": cfg.get(f"servers.{r}", "name") or r} + rec: dict = {"id": r, "name": cfg.get(f"servers.{r}", "name") or r} langs = cfg.get_list(f"servers.{r}", "languages") if langs: rec["languages"] = langs diff --git a/src/nnnotes/site/web.py b/src/nnnotes/site/web.py index d31b74a..b443994 100644 --- a/src/nnnotes/site/web.py +++ b/src/nnnotes/site/web.py @@ -857,12 +857,13 @@ def __init__(self, player_dir: Path): encoding="utf-8", errors="replace", ) + stdin, stdout, stderr = self.proc.stdin, self.proc.stdout, self.proc.stderr + assert stdin is not None and stdout is not None and stderr is not None # the pipes asked for + self.stdin = stdin self.lines: queue.Queue = queue.Queue() self.stderr: collections.deque = collections.deque(maxlen=50) - threading.Thread(target=self._pump, args=(self.proc.stdout, self.lines.put, True), daemon=True).start() - self._stderr_pump = threading.Thread( - target=self._pump, args=(self.proc.stderr, self.stderr.append, False), daemon=True - ) + threading.Thread(target=self._pump, args=(stdout, self.lines.put, True), daemon=True).start() + self._stderr_pump = threading.Thread(target=self._pump, args=(stderr, self.stderr.append, False), daemon=True) self._stderr_pump.start() @staticmethod @@ -876,8 +877,8 @@ def ask(self, request: dict, timeout: float) -> dict: """The answer to `request`: {"ok": true, "result"} or {"ok": false, "error"}. ReadSetServerError when the process has exited, answers something else or nothing within `timeout` seconds.""" try: - self.proc.stdin.write(json.dumps(request) + "\n") - self.proc.stdin.flush() + self.stdin.write(json.dumps(request) + "\n") + self.stdin.flush() except OSError: pass # exited: the end of its stdout follows try: @@ -899,7 +900,7 @@ def ask(self, request: dict, timeout: float) -> dict: def close(self, wait: float = 10.0) -> None: """End of its stdin: the server exits (ended after `wait` seconds if it does not).""" try: - self.proc.stdin.close() + self.stdin.close() except OSError: pass try: @@ -911,7 +912,7 @@ def kill(self) -> None: self.proc.kill() self.proc.wait() try: - self.proc.stdin.close() + self.stdin.close() except OSError: pass @@ -954,7 +955,8 @@ def read_set(self, live_dir: Path, player_dir: Path, mode: str = "full", setting raise ValueError(f"read set mode {mode}") name = Path(live_dir).name request = {"chart": str(Path(live_dir)), "plan": mode == "plan", **({"settings": settings} if settings else {})} - for attempt in (1, 2): + attempt = 1 + while True: s = self._take() try: doc = s.ask(request, self.timeout) @@ -962,6 +964,7 @@ def read_set(self, live_dir: Path, player_dir: Path, mode: str = "full", setting s.kill() if e.hung or attempt == 2: raise RuntimeError(f"read set of {name} failed: {e}") from None + attempt += 1 continue self._give(s) if not doc["ok"]: @@ -997,7 +1000,8 @@ class ReadSets: def __init__(self, player_dir: Path, log=_log, check_every: int = READ_SET_CHECK): self.player_dir = Path(player_dir) - self.initial = self.mode = None + self.initial: str | None = None + self.mode: str | None = None self.check_every = max(1, check_every) self.log = log self.lock = threading.Lock() # counters, mode @@ -1018,6 +1022,7 @@ def start(self) -> str: self.initial = self.mode = "plan" if "plan" in features else "full" if self.initial == "plan" and "serve" in features: self.servers = ReadSetServers(self.player_dir) + assert self.mode is not None return self.mode def close(self) -> None: @@ -1271,18 +1276,20 @@ def _build_group( "ingest": lambda m, charts, root: ingest_task(m, charts, root, job), } - def submit(name, *a): + def submit_here(name, *a): return ex.submit(call[name], *a) - ctx = None + submit, ctx = submit_here, None else: ctx = mp.get_context("spawn") mgr = ctx.Manager() ex = ProcessPoolExecutor(workers, mp_context=ctx, initializer=_worker_init, initargs=(cfg, mgr.Lock(), job)) - def submit(name, *a): + def submit_pooled(name, *a): return ex.submit(_pool_call, f"{name}_task", *a) + submit = submit_pooled + try: for b, f in [(b, submit("band", b, m)) for b, m in sorted(bands.items())]: try: diff --git a/src/nnnotes/site/webmodel.py b/src/nnnotes/site/webmodel.py index e28c62b..ee2a6f8 100644 --- a/src/nnnotes/site/webmodel.py +++ b/src/nnnotes/site/webmodel.py @@ -102,11 +102,16 @@ # ---------------------------------------------------------------- models and ids -def model_id(key: str) -> str: - """The id of a model key: its .""" +def _model_key(key: str) -> re.Match[str]: m = MODEL_KEY.fullmatch(key) if not m: raise ValueError(f"{key}: not a Live2D model key (Character/Live2D///model/)") + return m + + +def model_id(key: str) -> str: + """The id of a model key: its .""" + m = _model_key(key) if not MODEL_ID.fullmatch(m["name"]): raise ValueError(f"{key}: model name {m['name']!r} is not URL-safe") return m["name"] @@ -349,7 +354,7 @@ def model_facts(key: str, summary: dict, names: dict | None = None) -> dict: """What a model listing needs, from the key and the export summary, plus the NAME_FIELDS of `names` (an entry of model_names).""" facts = { - "group": MODEL_KEY.fullmatch(key)["group"], + "group": _model_key(key)["group"], "canvas": summary["canvas"], "textures": len(summary["textures"]), "nodes": summary["nodes"], diff --git a/src/nnnotes/source/catalog.py b/src/nnnotes/source/catalog.py index 43434e8..eeee5bb 100644 --- a/src/nnnotes/source/catalog.py +++ b/src/nnnotes/source/catalog.py @@ -13,8 +13,10 @@ import threading import urllib.parse import urllib.request +from collections.abc import Callable from dataclasses import dataclass from pathlib import Path +from typing import TypeAlias, TypeVar, cast from ..core.cache import write_atomic as _write_atomic from .addressables import REMOTE_PREFIX, BundleKey, decrypt, parse, parse_locations, relative_path, remote_path @@ -29,6 +31,11 @@ APK_OFFSET_BASE = 1 << 40 # namespace for APK catalog entry offsets +T = TypeVar("T") +# a setting given as its value, or as a function that reads it when a file not in the cache needs it +Lazy: TypeAlias = "T | Callable[[], T] | None" + + @dataclass(frozen=True) class Bundle: offset: int @@ -86,15 +93,15 @@ def __init__( catalog_bytes: bytes, cache_dir: Path, *, - cdn=None, - bundle_key=None, + cdn: Lazy[str] = None, + bundle_key: Lazy[BundleKey] = None, apk: Path | None = None, source=None, client=None, apk_catalog: bytes | None = None, resource_version: str | None = None, ): - self._settings = {"cdn": cdn, "bundle_key": bundle_key} + self._cdn, self._bundle_key = cdn, bundle_key self.cache_dir = Path(cache_dir) self.cache_dir.mkdir(parents=True, exist_ok=True) self.apk = Path(apk) if apk else None @@ -160,8 +167,8 @@ def load( language: str, cache_dir: Path, *, - cdn=None, - bundle_key=None, + cdn: Lazy[str] = None, + bundle_key: Lazy[BundleKey] = None, apk: Path | None = None, version: str = "main", ) -> Catalog: @@ -177,11 +184,11 @@ def load( cache_dir = cache_dir / "international" / hashlib.sha256(cdn.rstrip("/").encode()).hexdigest() / version cat = cache_dir / filename if not cat.exists(): - cdn = _setting(cdn, cat.name) - if not cdn: + root = _setting(cdn, cat.name) + if not root: raise FileNotFoundError(f"{cat} not cached and no CDN base given") cat.parent.mkdir(parents=True, exist_ok=True) - data = download(cdn.rstrip("/") + f"/asset/Android/{cat.name}") + data = download(root.rstrip("/") + f"/asset/Android/{cat.name}") parse(data) # Never install a corrupt/error response as a reusable catalog. _write_atomic(cat, data) return cls( @@ -193,15 +200,20 @@ def load( resource_version=version if version != "main" else None, ) - def _setting(self, name: str, needed_by: str): - """The `cdn` / `bundle_key` given to the catalog, its function called (once) now that `needed_by` needs it.""" - v = self._settings[name] = _setting(self._settings[name], needed_by) - return v + def _cdn_for(self, needed_by: str) -> str | None: + """The `cdn` given to the catalog, its function called (once) now that `needed_by` needs it.""" + self._cdn = _setting(self._cdn, needed_by) + return self._cdn + + def _bundle_key_for(self, needed_by: str) -> BundleKey | None: + """The `bundle_key` given to the catalog, its function called (once) now that `needed_by` needs it.""" + self._bundle_key = _setting(self._bundle_key, needed_by) + return self._bundle_key @property def cdn(self) -> str | None: """The CDN base as given (None while it is given as a function not called yet).""" - v = self._settings["cdn"] + v = self._cdn return None if callable(v) or not v else v.rstrip("/") @staticmethod @@ -331,7 +343,7 @@ def fetch(self, b: Bundle) -> Path: return dst if b.remote: url = self._url(b.internal_id) - key = self._setting("bundle_key", b.name) # before the download: a missing key fails first + key = self._bundle_key_for(b.name) # before the download: a missing key fails first data = self._download(url) else: if self.apk is None: @@ -340,7 +352,7 @@ def fetch(self, b: Bundle) -> Path: with ApkSet(self.apk) as z: self.check_apk(z) data = z.read(APK_AA_DIR + rel) - key = self._setting("bundle_key", b.name) if data[:7] != b"UnityFS" else None + key = self._bundle_key_for(b.name) if data[:7] != b"UnityFS" else None _write_atomic(dst, _unityfs(data, b.name, key)) return dst @@ -348,10 +360,13 @@ def _url(self, internal_id: str) -> str: if self.source is not None: return self.source.url(internal_id) name = internal_id.rsplit("/", 1)[-1] - cdn = self._setting("cdn", name) + cdn = self._cdn_for(name) if not cdn: raise RuntimeError(f"{name} not cached and no CDN base given") - return cdn.rstrip("/") + remote_path(internal_id) + path = remote_path(internal_id) + if path is None: + raise ValueError(f"{internal_id}: not a remote location") + return cdn.rstrip("/") + path def _download(self, url): if self.source is not None: @@ -408,7 +423,7 @@ def apk_bundle(self, name_contains: str) -> Path: dst.parent.mkdir(parents=True, exist_ok=True) self.check_apk(z) data = z.read(names[0]) - key = self._setting("bundle_key", name) if data[:7] != b"UnityFS" else None + key = self._bundle_key_for(name) if data[:7] != b"UnityFS" else None _write_atomic(dst, _unityfs(data, name, key)) return dst @@ -463,11 +478,11 @@ def _urlopen(url: str, timeout: float) -> bytes: return r.read() -def _setting(v, needed_by: str): +def _setting(v: Lazy[T], needed_by: str) -> T | None: """`v`, or what it returns when it is a function; its ConfigError names `needed_by` (what is not cached).""" if not callable(v): - return v + return cast("T | None", v) try: - return v() + return cast("T", v()) except ConfigError as e: raise ConfigError(f"{needed_by} is not in the cache: {e}") from None diff --git a/src/nnnotes/source/config.py b/src/nnnotes/source/config.py index 3fe5627..8ccddd7 100644 --- a/src/nnnotes/source/config.py +++ b/src/nnnotes/source/config.py @@ -20,6 +20,7 @@ import re import shutil import tomllib +from collections.abc import Mapping from pathlib import Path __all__ = ["Config", "ConfigError", "config_files", "user_file"] @@ -41,7 +42,7 @@ def check_catalog_version(value: str) -> str: return value -def user_file(environ: dict[str, str] | None = None) -> Path | None: +def user_file(environ: Mapping[str, str] | None = None) -> Path | None: """The per-user config file: `%APPDATA%\\nnnotes\\nnnotes.toml` on Windows, else `$XDG_CONFIG_HOME/nnnotes/nnnotes.toml` (`~/.config` when XDG_CONFIG_HOME is unset); None when the environment names no such directory.""" @@ -54,7 +55,7 @@ def user_file(environ: dict[str, str] | None = None) -> Path | None: def config_files( - path: str | Path | None = None, environ: dict[str, str] | None = None + path: str | Path | None = None, environ: Mapping[str, str] | None = None ) -> list[tuple[str, Path | None]]: """The config file candidates in lookup order, (how it is named, path or None): `--config`, NNNOTES_CONFIG, the working directory's nnnotes.toml, the per-user file. The first two are used when they are given (and must @@ -68,7 +69,7 @@ def config_files( ] -def find_file(path: str | Path | None = None, environ: dict[str, str] | None = None) -> Path | None: +def find_file(path: str | Path | None = None, environ: Mapping[str, str] | None = None) -> Path | None: """The config file that is read (config_files), or None when there is none.""" for name, file in config_files(path, environ): if file is None: @@ -86,7 +87,7 @@ def template() -> bytes: """The config template shipped with the package (`nnnotes config init` writes it).""" from importlib import resources - return resources.files(__package__).joinpath(TEMPLATE).read_bytes() + return resources.files("nnnotes.source").joinpath(TEMPLATE).read_bytes() def env_name(section: str, key: str) -> str: @@ -112,7 +113,7 @@ def __init__( data: dict | None = None, base: Path | None = None, source: str | None = None, - environ: dict[str, str] | None = None, + environ: Mapping[str, str] | None = None, overrides: dict | None = None, flags: dict | None = None, ): @@ -133,7 +134,7 @@ def load( cls, path: str | Path | None = None, *, - environ: dict[str, str] | None = None, + environ: Mapping[str, str] | None = None, overrides: dict | None = None, flags: dict | None = None, ) -> Config: diff --git a/src/nnnotes/source/configfile.py b/src/nnnotes/source/configfile.py index 1b1d2eb..69ebaae 100644 --- a/src/nnnotes/source/configfile.py +++ b/src/nnnotes/source/configfile.py @@ -309,6 +309,8 @@ def _status(cfg: Config, s: Setting, section: str) -> tuple[str, str | None]: validate(s, cfg.get_list(section, s.key)) elif s.kind == "path": p = cfg.path(section, s.key) + if p is None: + return "unset", None if s.key == "apk": if not p.exists(): return "not found", "no such APK or directory" @@ -321,7 +323,7 @@ def _status(cfg: Config, s: Setting, section: str) -> tuple[str, str | None]: v = cfg.get(section, s.key) validate(s, v) if (section, s.key) == ("catalog", "region") and v not in cfg.regions(): - return "invalid", f"names no [servers.{v}] table and no {env_name('servers.' + v, 'cdn')}" + return "invalid", f"names no [servers.{v}] table and no {env_name(f'servers.{v}', 'cdn')}" except ConfigError as e: return "invalid", str(e).split(": ", 1)[-1] except ValueError as e: diff --git a/src/nnnotes/source/gameapi.py b/src/nnnotes/source/gameapi.py index ed8596f..167be44 100644 --- a/src/nnnotes/source/gameapi.py +++ b/src/nnnotes/source/gameapi.py @@ -155,13 +155,13 @@ def parse_server_list(buf: bytes) -> list[Server]: # ---------------------------------------------------------------- calls -def metadata(client_version: str, request_id: str | None = None) -> list[tuple[str, str]]: +def metadata(client_version: str, request_id: str | None = None) -> tuple[tuple[str, str], ...]: """The metadata of a call: a fresh `x-request-id` unless one is given.""" - return [ + return ( ("x-request-id", request_id or str(uuid.uuid4())), ("x-platform", PLATFORM), ("x-client-version", client_version), - ] + ) def channel_target(root: str) -> tuple[str, bool]: @@ -232,9 +232,10 @@ def call( def failure(name: str, setting: str, code, trailers) -> str: """The one-line error of a failed call: status, a hint, and the game's error code when the server sent one.""" status = code.name if code is not None else "no status" - hint = STATUS_HINT.get(code) + hint = STATUS_HINT.get(code) if code is not None else None game = next((v for k, v in (trailers or ()) if k == ERROR_CODE_TRAILER), None) - extra = ", ".join(x for x in (hint, f"game error code {game}" if game else None, GAME_ERROR_HINT.get(game)) if x) + game_hint = GAME_ERROR_HINT.get(game) if game is not None else None + extra = ", ".join(x for x in (hint, f"game error code {game}" if game else None, game_hint) if x) return f"game API call {name} to {setting} failed: {status}" + (f" ({extra})" if extra else "") diff --git a/src/nnnotes/source/jp.py b/src/nnnotes/source/jp.py index e269c5d..26d40a4 100644 --- a/src/nnnotes/source/jp.py +++ b/src/nnnotes/source/jp.py @@ -45,7 +45,7 @@ def master_version(value): return value -def select_asset(raw, client): +def select_asset(raw, client) -> tuple[str, str] | None: """Select the largest applicable live minimum; no top-level fallback when live has no match.""" if not raw: return None @@ -73,7 +73,7 @@ def select_asset(raw, client): data = max(candidates, key=lambda x: x[0])[1] version, digest = data.get("version"), data.get("Android") numeric_version(version) - if not isinstance(digest, str) or not re.fullmatch(r"[0-9a-f]{32}", digest): + if not isinstance(version, str) or not isinstance(digest, str) or not re.fullmatch(r"[0-9a-f]{32}", digest): raise ValueError() return version, digest except (ValueError, TypeError): diff --git a/src/nnnotes/source/session.py b/src/nnnotes/source/session.py index 081383f..7ad2c88 100644 --- a/src/nnnotes/source/session.py +++ b/src/nnnotes/source/session.py @@ -42,6 +42,13 @@ def path(self, section: str, key: str, kind: str = "file") -> Path | None: raise ConfigError(f"setting {section}.{key}: {kind} {p} not found") return p + def required_path(self, section: str, key: str, kind: str = "file") -> Path: + """The path a required setting names, checked to exist (a ConfigError naming the setting when it is unset).""" + self.cfg.require_path(section, key) + p = self.path(section, key, kind) + assert p is not None # require_path: the setting is set + return p + def store_root(self) -> Path: """The store directory: `[paths] store`, else the store/ directory of `[paths] cache`.""" p = self.cfg.path("paths", "store") @@ -58,15 +65,13 @@ def store_root(self) -> Path: def apk(self, region: str | None = None) -> Path: """`[paths] apk` of `region`, required.""" cfg = self.cfg.for_region(region) if region else self.cfg - cfg.require_path("paths", "apk") - return Session(cfg).path("paths", "apk") + return Session(cfg).required_path("paths", "apk") def master_dir(self, region: str | None = None) -> Path: """The decoded master data of `region`: the --master flag, else `[servers.] master`, else `[paths] master`.""" section, key = self.cfg.master(region or self.cfg.get("catalog", "region")) - self.cfg.require_path(section, key) - return self.path(section, key, "directory") + return self.required_path(section, key, "directory") def master_version(self, region: str | None = None, *, timeout: float = gameapi.TIMEOUT) -> gameapi.MasterVersion: """The master data version `region` serves now (its `[servers.] api`).""" diff --git a/src/nnnotes/unity/export.py b/src/nnnotes/unity/export.py index 0aa99e8..3e77412 100644 --- a/src/nnnotes/unity/export.py +++ b/src/nnnotes/unity/export.py @@ -29,9 +29,11 @@ import struct import zlib from pathlib import Path +from typing import Literal import numpy as np from PIL import Image +from UnityPy.files import ObjectReader from UnityPy.helpers import MeshHelper from ..core import cache @@ -323,7 +325,7 @@ def pixels(self, tex_obj) -> np.ndarray: return self._pixels[k] @staticmethod - def _wrap(mode: int) -> str: + def _wrap(mode: int) -> Literal["wrap", "clip"]: if mode == 0: return "wrap" if mode == 1: @@ -481,7 +483,7 @@ def __init__( self._font_bundles: frozenset | None = None self._unwritten: dict[str, object] = {} # defer_writes: file -> Texture2D self.textures: dict[tuple, dict] = {} - self.shaders: dict[str, object] = {} + self.shaders: dict[str, ObjectReader] = {} self.tex_objs: dict[str, object] = {} # deferred: "file:pathID" -> Texture2D self.sprite_recs: dict[str, dict] = {} # deferred: "file:pathID" -> sprite render record self.clips: dict[str, dict] = {} # livenotes format: id -> clip @@ -490,7 +492,7 @@ def __init__( self._graph_of_file: dict[int, SceneGraph] = {} self._done: dict[tuple, dict] = {} # MonoBehaviour/clip/controller/atlas -> reference once exported self._component_ref: dict[tuple, str] = {} - self._objects: dict[tuple, object] = {} + self._objects: dict[tuple, ObjectReader] = {} self._go_file: dict[int, str] = {} self._loaded: dict[str, tuple] = {} self._clip_recs: dict[tuple, dict] = {} @@ -498,7 +500,7 @@ def __init__( self._names_of: dict[tuple, list] = {} # object -> flat field names self._names_by_class: dict = {} # typeID | ("MonoBehaviour", class) -> names of exported components self._class_of: dict[tuple, str | None] = {} - self._rel_tables: dict[int, dict] = {} + self._rel_tables: dict[tuple, dict] = {} self._gos_at_cache: dict[tuple, list] = {} self._path_sources: list[tuple[str, dict]] = [] self._crcs: dict[str, int] = {} @@ -531,19 +533,36 @@ def load(self, key: str): return env, graph def font_bundles(self) -> frozenset: - """With `font_stubs`: the names of the bundles that hold its font assets.""" + """The names of the bundles that hold the font assets of `font_stubs` (none without it).""" if self._font_bundles is None: - self._font_bundles = self.font_stubs.bundles(self.cat) + self._font_bundles = frozenset() if self.font_stubs is None else self.font_stubs.bundles(self.cat) return self._font_bundles def font_stub(self, o: FontStub) -> dict: """The value of a reference into a font bundle left out (`font_stubs`): a material's record, a font asset's {asset, name}.""" - if o.rec["type"] == "Material": + if o.rec["type"] == "Material" and self.font_stubs is not None: return self.font_stubs.material(None, o.rec["name"]) return {"asset": o.rec["class"], "name": o.rec["name"]} - def deref(self, owner, pptr: dict): + def deref(self, owner, pptr: dict) -> ObjectReader | None: + """The object `pptr` (read from `owner`'s typetree) names; None for a null reference. A reference into a font + bundle left out (`font_stubs`) is an error here: only `ref` writes such a reference, as its record.""" + o = self._lookup(owner, pptr) + if isinstance(o, FontStub): + raise FileNotFoundError( + f"path id {o.path_id} is in a font bundle left out of the closure ({o.rec['name']})" + ) + return o + + def referent(self, owner, pptr: dict) -> ObjectReader: + """The object a reference that must not be null names (KeyError naming `owner` for a null one).""" + o = self.deref(owner, pptr) + if o is None: + raise KeyError(f"{owner.type.name} {owner.path_id}: a required reference is null") + return o + + def _lookup(self, owner, pptr: dict) -> ObjectReader | FontStub | None: if not pptr["m_PathID"]: return None if external_path(owner, pptr) == DEFAULT_RESOURCES: @@ -576,7 +595,7 @@ def value(self, owner, v): return v def ref(self, owner, pptr: dict): - o = self.deref(owner, pptr) + o = self._lookup(owner, pptr) if o is None: return None if isinstance(o, FontStub): @@ -604,7 +623,7 @@ def ref(self, owner, pptr: dict): if t in ("Transform", "RectTransform"): return {"transform": self.graph(o).path(o.path_id)} if _key(o) in self._component_ref: - desc = {"component": t, "gameObject": self._component_ref[_key(o)]} + desc: dict[str, str | None] = {"component": t, "gameObject": self._component_ref[_key(o)]} if t == "MonoBehaviour": desc["class"] = self.script_class(o) return desc @@ -704,7 +723,7 @@ def sprite(self, o) -> dict: return self._deferred_sprite(o, tt, src, owner) if src is None: return {"sprite": tt["m_Name"], "texture": None, "unboundAtlasTags": tt.get("m_AtlasTags")} - tex = self.ref(owner, src["texture"]) + tex = self.texture(self.referent(owner, src["texture"])) rd = tt["m_RD"] vd = rd["m_VertexData"] raw = vd["m_DataSize"] @@ -750,7 +769,7 @@ def sprite(self, o) -> dict: def _deferred_sprite(self, o, tt: dict, src, owner) -> dict: if src is None: raise RuntimeError(f"sprite {tt['m_Name']}: no texture and no atlas") - tex = self.deref(owner, src["texture"]) + tex = self.referent(owner, src["texture"]) if src["alphaTexture"]["m_PathID"] or src.get("secondaryTextures"): raise NotImplementedError(f"sprite {tt['m_Name']}: alpha/secondary textures") raw = src["settingsRaw"] @@ -856,7 +875,7 @@ def _skin(m, tt: dict, vertex_count: int) -> dict: h = MeshHelper.MeshHandler(m) h.process() w, ix = h.m_BoneWeights, h.m_BoneIndices - if not w or len(w) != vertex_count or len(ix) != vertex_count: + if not w or not ix or len(w) != vertex_count or len(ix) != vertex_count: raise RuntimeError(f"mesh {tt['m_Name']}: {len(w or [])} bone weights for {vertex_count} vertices") skin = [] for ws, bs in zip(w, ix): @@ -905,7 +924,7 @@ def _rel_table(self, graph: SceneGraph) -> dict[int, set[str]]: An Animator binding path is relative to the Animator's GameObject, which need not be the hierarchy root, and the clip does not say which GameObject plays it.""" - t = self._rel_tables.get(id(graph)) + t = self._rel_tables.get(("relative", id(graph))) if t is None: t = {0: {""}} for tf_pid in graph.tf: @@ -913,7 +932,7 @@ def _rel_table(self, graph: SceneGraph) -> dict[int, set[str]]: for i in range(1, len(parts)): s = "/".join(parts[i:]) t.setdefault(self._crc_of(s), set()).add(s) - self._rel_tables[id(graph)] = t + self._rel_tables[("relative", id(graph))] = t return t def _all_graphs(self) -> list[SceneGraph]: @@ -1588,7 +1607,7 @@ def controller_states(self, o) -> dict: as its state list (name, speed, loop, cycle offset, clip name) -- what the ADV UI runtime plays.""" rec = self.controller_record(o) tt, tos = rec["tt"], rec["tos"] - clips = [self.deref(o, p).read().m_Name for p in tt["m_AnimationClips"]] + clips = [self.referent(o, p).read().m_Name for p in tt["m_AnimationClips"]] ctrl = tt["m_Controller"] if len(ctrl["m_StateMachineArray"]) != 1 or len(ctrl["m_LayerArray"]) != 1: raise NotImplementedError(f"controller {tt['m_Name']}: several layers") @@ -1647,16 +1666,15 @@ def scriptable(self, o) -> dict: if cls in self.stub_assets: return ref self._done[k] = ref - return {**ref, **self.value(o, {x: y for x, y in tt.items() if x not in HEADER})} + return {**ref, **{x: self.value(o, y) for x, y in tt.items() if x not in HEADER}} def component(self, o) -> dict: tt = o.read_typetree() self._register_names(o, tt) - body = {k: v for k, v in tt.items() if k not in HEADER} out = {"type": o.type.name} if o.type.name == "MonoBehaviour": out["class"] = self.script_class(o) - out.update(self.value(o, body)) + out.update({k: self.value(o, v) for k, v in tt.items() if k not in HEADER}) return out # --- hierarchies ----------------------------------------------------- @@ -1701,17 +1719,20 @@ def walk(tf_pid): return nodes # --- keys -------------------------------------------------------------- - def _container_asset(self, env, internal_id: str): + def _container_asset(self, env, internal_id: str) -> ObjectReader: for o in env.objects: if o.type.name != "AssetBundle": continue tt = o.read_typetree() for path, info in tt["m_Container"]: if path.lower() == internal_id.lower(): - return self.deref(o, info["asset"]) + asset = self.deref(o, info["asset"]) + if asset is None: + raise KeyError(f"{internal_id}: the container entry is a null reference") + return asset raise KeyError(f"{internal_id} not in any AssetBundle container") - def key_object(self, key: str): + def key_object(self, key: str) -> ObjectReader: """The object an addressable key names (its bundle's container asset).""" env, _ = self.load(key) return self._container_asset(env, self.cat.entry(key)["internal_id"]) diff --git a/src/nnnotes/unity/live2d.py b/src/nnnotes/unity/live2d.py index 0529e04..1cecd2d 100644 --- a/src/nnnotes/unity/live2d.py +++ b/src/nnnotes/unity/live2d.py @@ -140,8 +140,8 @@ def expression_json(tt: dict) -> dict: } -def _classes(env) -> dict[str, list]: - by_cls: dict[str, list] = {} +def _classes(env) -> dict[str | None, list]: + by_cls: dict[str | None, list] = {} for o in env.objects: if o.type.name == "MonoBehaviour": by_cls.setdefault(script_class(o), []).append(o) @@ -165,7 +165,7 @@ class RuntimeExport: exporter: Exporter env: object graph: SceneGraph - classes: dict[str, list] + classes: dict[str | None, list] def export_runtime(cat: Catalog, key: str, out_dir: Path) -> RuntimeExport: diff --git a/src/nnnotes/unity/player.py b/src/nnnotes/unity/player.py index b9de4b5..21c36ad 100644 --- a/src/nnnotes/unity/player.py +++ b/src/nnnotes/unity/player.py @@ -21,8 +21,10 @@ import struct from importlib import resources from pathlib import Path +from typing import cast import UnityPy +from UnityPy.files import ObjectReader, SerializedFile from UnityPy.helpers.TypeTreeNode import TypeTreeNode from ..source.apkset import MANIFEST_IN_APK, ApkSet, manifest_version_name @@ -48,7 +50,7 @@ def typetree_key(assembly: str, namespace: str, cls: str) -> str: def typetree_file(unity_version: str): """The embedded typetree file of a Unity version (an importlib.resources Traversable; it may not exist).""" - return resources.files(__package__).joinpath("typetrees", f"{unity_version}.json") + return resources.files("nnnotes.unity").joinpath("typetrees", f"{unity_version}.json") def load_typetrees(unity_version: str) -> dict | None: @@ -149,11 +151,20 @@ def shader(self, name: str, file: str | None = None): raise KeyError(f"shader {name!r}{f' in {file}' if file else ''}: {len(hits)} objects") return hits[0] - def deref(self, owner, pptr: dict): + def referent(self, owner, pptr: dict) -> ObjectReader: + """The object a reference that must not be null names (KeyError naming `owner` for a null one).""" + o = self.deref(owner, pptr) + if o is None: + raise KeyError(f"{owner.type.name} {owner.path_id}: a required reference is null") + return o + + def deref(self, owner, pptr: dict) -> ObjectReader | None: + """The object `pptr` names, in `unity default resources` or through `owner`'s file; None if null.""" if not pptr["m_PathID"]: return None if external_path(owner, pptr) == DEFAULT_RESOURCES: - return self.defaults.files[next(iter(self.defaults.files))].objects[pptr["m_PathID"]] + defaults = cast(SerializedFile, next(iter(self.defaults.files.values()))) + return defaults.objects[pptr["m_PathID"]] return deref(owner, pptr) def script(self, o) -> tuple[str, str, str]: @@ -162,7 +173,7 @@ def script(self, o) -> tuple[str, str, str]: if hit is not None and hit[0] is o: return hit[1] fid, pid = struct.unpack_from(" str: if name not in renderers: feats = [] for fp in tt["m_RendererFeatures"]: - fo = self.deref(o, fp) + fo = self.referent(o, fp) asm, ns, cls = self.script(fo) feats.append( - {"class": f"{ns}.{cls}" if ns else cls, "assembly": asm, **self.names_in(fo, self.mono(fo))} + { + "class": f"{ns}.{cls}" if ns else cls, + "assembly": asm, + **{k: self.names_in(fo, v) for k, v in self.mono(fo).items()}, + } ) tt["m_RendererFeatures"] = feats pp = tt.get("postProcessData") diff --git a/src/nnnotes/unity/reader.py b/src/nnnotes/unity/reader.py index 5887a57..2c640b8 100644 --- a/src/nnnotes/unity/reader.py +++ b/src/nnnotes/unity/reader.py @@ -14,6 +14,7 @@ import UnityPy from UnityPy.classes import PPtr from UnityPy.enums import TextureFormat +from UnityPy.files import ObjectReader from UnityPy.helpers import MeshHelper from UnityPy.helpers.TypeTreeNode import TypeTreeNode @@ -83,7 +84,7 @@ def parse_blob(cls, reader, version): parse_blob._nnnotes_memo = True parse_blob._original = parse - ttn.TypeTreeNode.parse_blob = classmethod(parse_blob) + ttn.TypeTreeNode.parse_blob = classmethod(parse_blob) # pyright: ignore[reportAttributeAccessIssue] _blob_stats["installed"] = True return True @@ -350,7 +351,7 @@ def script_class(obj) -> str | None: pptr = obj.read_typetree(head, check_read=False)["m_Script"] if not pptr["m_PathID"]: return None - return deref(obj, pptr).read().m_ClassName + return referent(obj, pptr).read().m_ClassName except Exception: pass script = obj.read().m_Script @@ -378,8 +379,15 @@ def external_path(owner, pptr: dict) -> str | None: return None if fid == 0 else owner.assets_file.externals[fid - 1].path -def deref(owner, pptr: dict): +def deref(owner: ObjectReader, pptr: dict) -> ObjectReader | None: """ObjectReader that a PPtr read from `owner`'s typetree points at; None if null.""" if not pptr["m_PathID"]: return None + return referent(owner, pptr) + + +def referent(owner: ObjectReader, pptr: dict) -> ObjectReader: + """ObjectReader that a non-null PPtr read from `owner`'s typetree points at (KeyError for a null one).""" + if not pptr["m_PathID"]: + raise KeyError(f"{owner.type.name} {owner.path_id}: a required reference is null") return PPtr(m_FileID=pptr["m_FileID"], m_PathID=pptr["m_PathID"], assetsfile=owner.assets_file).deref() diff --git a/src/nnnotes/unity/tmpfont.py b/src/nnnotes/unity/tmpfont.py index f05d53c..37ff93b 100644 --- a/src/nnnotes/unity/tmpfont.py +++ b/src/nnnotes/unity/tmpfont.py @@ -18,8 +18,11 @@ import importlib.util import io import math +from collections.abc import Callable +from typing import cast import numpy as np +from UnityPy.files import ObjectReader from ..source.config import ConfigError from .export import Exporter, TexelPacker, _key, grow_box @@ -67,7 +70,7 @@ def __init__(self, ex: Exporter, obj): self.glyphs = {g["m_Index"]: g for g in tt["m_GlyphTable"]} self.dynamic = tt["m_AtlasPopulationMode"] == 1 self.atlases = [ex.deref(obj, p) for p in tt["m_AtlasTextures"]] - self.material_obj = ex.deref(obj, tt["m_Material"]) + self.material_obj = ex.referent(obj, tt["m_Material"]) self.fallback_objs = [ex.deref(obj, p) for p in tt["m_FallbackFontAssetTable"]] if any(p["regularTypeface"]["m_PathID"] or p["italicTypeface"]["m_PathID"] for p in tt["m_FontWeightTable"]): raise NotImplementedError(f"{self.name}: font weight table typefaces") @@ -272,7 +275,7 @@ def __init__(self, font: Font): from scipy.ndimage import distance_transform_edt self.ft = freetype - self.edt = distance_transform_edt + self.edt = cast(Callable[[np.ndarray], np.ndarray], distance_transform_edt) tt = font.tt mode = tt["m_AtlasRenderMode"] want = RASTER_MODE["1X"] | RASTER_MODE["SDF"] | RASTER_MODE["NO_HINTING"] | RASTER_MODE["MONO"] @@ -288,6 +291,7 @@ def __init__(self, font: Font): raise NotImplementedError(f"{font.name}: fractional point size {self.point_size}") self.face = freetype.Face(io.BytesIO(font.font_data), tt["m_FaceInfo"]["m_FaceIndex"]) self.face.set_char_size(int(self.point_size) * 64, 0, 72, 72) + self.fine = self.face if self.face.units_per_EM != tt["m_FaceInfo"]["m_UnitsPerEM"]: raise RuntimeError(f"{font.name}: units per EM differ from the face info") @@ -309,7 +313,7 @@ def from_font_file( if oversample < 1 or oversample != int(oversample): raise ValueError(f"oversample {oversample}") g = cls.__new__(cls) - g.ft, g.edt, g.font = freetype, distance_transform_edt, None + g.ft, g.edt, g.font = freetype, cast(Callable[[np.ndarray], np.ndarray], distance_transform_edt), None g.pad, g.spread, g.point_size, g.oversample = int(padding), int(padding) + 1, int(point_size), int(oversample) g.face = freetype.Face(io.BytesIO(data), face_index) g.face.set_char_size(g.point_size * 64, 0, 72, 72) @@ -330,12 +334,12 @@ def cell(self, gi: int, margin: int) -> tuple[dict, tuple, np.ndarray]: m = self.metrics(gi) xmin, ymin, w, h = self.pixel_box(m) rows, cols = h + 2 * margin, w + 2 * margin - self.fine.load_glyph(gi, ft.FT_LOAD_NO_HINTING) - self.fine.glyph.render(ft.FT_RENDER_MODE_MONO) + self.fine.load_glyph(gi, ft.FT_LOAD_FLAGS["FT_LOAD_NO_HINTING"]) + self.fine.glyph.render(ft.FT_RENDER_MODES["FT_RENDER_MODE_MONO"]) bm = self.fine.glyph.bitmap grid = np.zeros((rows * k, cols * k), bool) if bm.rows and bm.width: - if bm.pixel_mode != ft.FT_PIXEL_MODE_MONO: + if bm.pixel_mode != ft.FT_PIXEL_MODES["FT_PIXEL_MODE_MONO"]: raise RuntimeError(f"glyph {gi}: FreeType returned pixel mode {bm.pixel_mode}") raw = np.frombuffer(bytes(bm.buffer), np.uint8).reshape(bm.rows, bm.pitch) bits = np.unpackbits(raw, axis=1)[:, : bm.width].astype(bool)[::-1] # bottom row first @@ -357,7 +361,7 @@ def glyph_index(self, u: int) -> int: return self.face.get_char_index(u) def metrics(self, gi: int) -> dict: - self.face.load_glyph(gi, self.ft.FT_LOAD_NO_HINTING) + self.face.load_glyph(gi, self.ft.FT_LOAD_FLAGS["FT_LOAD_NO_HINTING"]) m = self.face.glyph.metrics return { "m_Width": m.width / 64, @@ -382,12 +386,12 @@ def sdf(self, gi: int) -> tuple[dict, tuple, np.ndarray]: m = self.metrics(gi) xmin, ymin, w, h = self.pixel_box(m) p = self.pad - self.face.load_glyph(gi, ft.FT_LOAD_NO_HINTING) - self.face.glyph.render(ft.FT_RENDER_MODE_MONO) + self.face.load_glyph(gi, ft.FT_LOAD_FLAGS["FT_LOAD_NO_HINTING"]) + self.face.glyph.render(ft.FT_RENDER_MODES["FT_RENDER_MODE_MONO"]) bm = self.face.glyph.bitmap grid = np.zeros((h + 2 * p, w + 2 * p), bool) if bm.rows and bm.width: - if bm.pixel_mode != ft.FT_PIXEL_MODE_MONO: + if bm.pixel_mode != ft.FT_PIXEL_MODES["FT_PIXEL_MODE_MONO"]: raise RuntimeError(f"glyph {gi}: FreeType returned pixel mode {bm.pixel_mode}") raw = np.frombuffer(bytes(bm.buffer), np.uint8).reshape(bm.rows, bm.pitch) bits = np.unpackbits(raw, axis=1)[:, : bm.width].astype(bool)[::-1] # bottom row first @@ -744,7 +748,7 @@ def __init__(self, ex): super().__init__({"format": FORMAT, "fonts": {}, "materials": {}}) self.ex = ex self._set = None - self._mat_objs: dict[str, object] = {} # material name -> its object + self._mat_objs: dict[str, ObjectReader] = {} # material name -> its object def _fontset(self): if self._set is None: @@ -782,6 +786,8 @@ def has_material(self, font: str, name: str) -> bool: def material(self, font: str | None, name: str) -> dict: if name not in self.materials: + if font is None: + raise KeyError(f"material {name}: not read yet, and no font asset to find it by") key = font_key(self.ex.cat, font, name) if not self.ex.cat.has(key): raise KeyError(f"material key {key}") diff --git a/tests/content/test_story_ui.py b/tests/content/test_story_ui.py index 84c3ddb..82ad74d 100644 --- a/tests/content/test_story_ui.py +++ b/tests/content/test_story_ui.py @@ -427,6 +427,8 @@ def key_object(self, key): def deref(self, o, ref): return FakeObj(sheet) if ref == "sheet" else ref + referent = deref + def material(self, o): return {"material": f"{o} record"} diff --git a/tests/pipeline/test_orchestrate.py b/tests/pipeline/test_orchestrate.py index b00abd2..a3e586a 100644 --- a/tests/pipeline/test_orchestrate.py +++ b/tests/pipeline/test_orchestrate.py @@ -353,5 +353,7 @@ def test_a_scan_with_the_census_keeps_only_the_coverage_document(tmp_path): want = run.coverage({"Line": 6}) run.scan(census=lambda: {"Line": 6}) digest = run._scan[1] - assert digest.best is None and run.coverage({"Line": 6}) == want + assert digest.best == digest.kinds == digest.reasoned == {} and run.coverage({"Line": 6}) == want + with pytest.raises(RuntimeError, match="settled"): + digest.add("toy.src:a", {"items": []}) assert run.coverage() == coverage(run.results()) and run._scan[1] is not digest # another census: read again