build: type-check the package with pyright

.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.

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
 ```

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"
 

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)

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,15 +466,15 @@ 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
         self.version, self.remote, self.apk_catalog = self._catalog_version()
         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,37 +617,39 @@ 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)
         if only:
             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
         for n, p in out.items():
             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}))

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, /<region>/ -> languages, /<region>/<language>/<dir>/ -> listing,
     /<region>/<language>/download/<offset> -> 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
     <cache>/catalogs/<region>/)."""
-    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()