refactor(cli): run export live2d, audio and shader through the store

docs/architecture.md

@@ -36,9 +36,11 @@ decrypted bundles in the cache directory.
 
 ## Two ways to export
 
-- **One subject.** `export episode`, `export story`, `export live`, `export live2d`, `export spot`, `export room`,
-  `export shader`, `export audio`, `export ui`, `site build` and `music-data` read what one subject needs through the
-  Session and write their output directory directly ([Export stories and lives](export.md)).
+- **One subject.** `export live2d`, `export shader` and `export audio` run the stages of their subject through the
+  store, over the files the subject's catalog keys load, and copy the subject's files from the store into the output
+  directory; a second export of the same subject is a hit of the store. `export episode`, `export story`,
+  `export live`, `export spot`, `export room`, `export ui`, `site build` and `music-data` read what one subject needs
+  through the Session and write their output directory directly ([Export stories and lives](export.md)).
 - **Every asset.** `export assets` splits the work into stages ([stages](stages.md)). A task is one stage applied to
   one subject (a bundle, a content id, a view); its key covers exactly what its output depends on: the stage and
   converter versions, the input contents and the options ([contracts](contracts.md)). Results and their files are

docs/commands.md

@@ -285,7 +285,9 @@ OUT/motions/<motion>.motion3.json motions (and motions/_fades.json)
 OUT/<name>.model3.json            file references, EyeBlink / LipSync groups
 ```
 
-Needs `[paths] apk` (component classes are resolved through the APK). Prints a JSON summary.
+Needs `[paths] apk` (component classes are resolved through the APK). The model is the `live2d.model` task of the
+store ([stages](stages.md)): the bundles of the model's closure are fetched into the cache, the task runs once and
+its files are copied from the store. Prints a JSON summary.
 
 ## export spot
 
@@ -348,6 +350,10 @@ OUT/<name>/<platform>/s<S>p<P>_<stage>_<N>.<ext>   every compiled sub-program (G
 OUT/shaders.json                                   index with the keywords of every variant
 ```
 
+The shaders are the `Shader` artifacts of the `unity.export` tasks of those bundles in the store (the export limited
+to the class `Shader`). A name met again later in the bundle order (the closure in the order the catalog loads it)
+keeps its first shader; `shaders.json` names the bundle file each shader comes from.
+
 ## export audio
 
 ```
@@ -356,9 +362,10 @@ nnnotes export audio CUE_SHEET -o OUT [--format flac|ogg|wav] [--flac-level N]
 
 Decodes the CRI cue sheet of the key `Cri/Sound/<CUE_SHEET>`: one file per stream (a name repeated within the
 sheet gets `<name>_<stream>`), `cues.json` (first stream per name: file, sample rate, channels, samples, loop
-points) and `streams.json` (every stream in order). `flac` (default) keeps the decoded PCM bit-exact; `ogg` is lossy.
-`--flac-level` (0 to 12, default 8) is ffmpeg's FLAC compression level; every level decodes to the same samples.
-The HCA keycode is read from `[paths] apk`.
+points) and `streams.json` (every stream in order). `flac` (default) keeps the decoded PCM bit-exact; `ogg` is lossy
+(encoded with ffmpeg's bit-exact flags: the same samples give the same file). `--flac-level` (0 to 12, default 8) is
+ffmpeg's FLAC compression level; every level decodes to the same samples. The HCA keycode is read from the boot data
+of `[paths] apk`. The sheet is the `cri.audio` task of its ACB in the store, with the format as a parameter.
 
 ## voices
 

docs/configuration.md

@@ -166,11 +166,11 @@ story text); it is not a setting.
 | `servers` | `[bootstrap] api`; the client version: `[client] version` or `[paths] apk` |
 | `export episode` | catalog, master data |
 | `export story` | catalog, `[catalog] language`, master data, `[paths] apk`, `ffmpeg` (audio and videos), `vgmstream` (not with `--no-audio`) |
-| `export live2d` | catalog, `[paths] apk` |
+| `export live2d` | the store (`[paths] store` or `[paths] cache`), catalog, `[paths] apk` |
 | `export spot` | catalog, master data, `[paths] apk` |
 | `export room` | catalog |
-| `export shader` | catalog; `--apk-bundle` also `[paths] apk` |
-| `export audio` | catalog, `[paths] apk`, `vgmstream`, `ffmpeg` (not for `--format wav`) |
+| `export shader` | the store, catalog; `--apk-bundle` also `[paths] apk` |
+| `export audio` | the store, catalog, `[paths] apk`, `vgmstream`, `ffmpeg` (not for `--format wav`) |
 | `apk hca-key` | `[paths] apk` |
 | `apk settings` | `[paths] apk` |
 | `export live` | catalog, `[catalog] language`, master data, `[paths] apk`, `vgmstream`, `ffmpeg` |

docs/stages.md

@@ -111,7 +111,7 @@ bundle.
 
 | Stage | Parameters | Atoms | Artifacts (`<task id>#<role>`) |
 |---|---|---|---|
-| `cri.audio` | `flac.level` (8): ffmpeg's FLAC compression level 0-12 (every level decodes to the same samples) | `hca.decode` (vgmstream), `flac.encode` (ffmpeg) | one FLAC per vgmstream stream (a name repeated within the sheet gets `<name>_<stream>`; facts: stream, name, sample rate, channels, samples, loop points), `cues.json` (the first stream of each name), `streams.json` (every stream in order): the bytes `nnnotes export audio` writes for the sheet (FLAC) |
+| `cri.audio` | `format`: `flac` (default; with `flac.level` (8): ffmpeg's FLAC compression level 0-12, every level decodes to the same samples), `ogg` (Vorbis, ffmpeg's bit-exact flags) or `wav` | `hca.decode` (vgmstream); `flac.encode` or `ogg.encode` (ffmpeg) | one file per vgmstream stream in the format (a name repeated within the sheet gets `<name>_<stream>`; facts: stream, name, sample rate, channels, samples, loop points), `cues.json` (the first stream of each name), `streams.json` (every stream in order): the files `nnnotes export audio` writes for the sheet |
 | `cri.movie` | `format`: `mkv` (default; with `flac.level`) or `webm` | `usm.demux` (nnnotes: `numpy-<v>/2`), `movie.mux` (ffmpeg), `hca.decode` (vgmstream) | per stream as stored and unmasked (a stream stored without its mask is recognised by its structure), named by its kind and, past channel 0, `_<channel>`: `video.<ext>` and `alpha.<ext>` (`ivf`: VP9 in IVF; `m1v`: MPEG-1 elementary stream, CRI's Sofdec.Prime; `h264`: H.264 Annex B; facts: codec, size, display size, frame rate, frames, an alpha stream's alpha type), `audio.<ext>` (`adx`; `hca`, still enciphered as stored; facts: channels, sample rate, samples or blocks, cipher), `subtitle.json` (the records as stored: language, time unit, start, duration, the text decoded as UTF-8 or `textHex`) with `subtitle.srt` and `subtitle.vtt`; `movie.mkv`: every video and alpha stream copied (the alpha as a second video track titled `alpha`), every audio stream as FLAC (HCA decoded by vgmstream with the boot data's key), every subtitle channel as a SubRip track (facts: the tracks); with `format: webm` `movie.webm`, the story's form: the first video stream, VP9 copied, or re-encoded as VP9 (libvpx, CRF 20) when it is not VP9 or has an alpha stream, which becomes its alpha channel cropped to the display size; the first audio stream as Opus; no subtitles |
 
 `cri.movie` in players: a player shows the first video track of `movie.mkv` and no player composites the `alpha`

src/nnnotes/cli/assets.py

@@ -1097,6 +1097,96 @@ def parse_layouts(spec: str) -> list[str]:
     return names
 
 
+# ---------------------------------------------------------------- single subjects
+def _literal(name: str) -> str:
+    """A stable name as a `bundle:` glob that matches that name only."""
+    return "".join(f"[{c}]" if c in "*?[" else c for c in name)
+
+
+def key_selection(index: dict, keys=(), files=()) -> list[str]:
+    """The selection of the bundles and raw files the catalog keys `keys` load (every location of each key and its
+    dependency closure) and of the files `files` (stable names), by stable name."""
+    from ..pipeline import catalogdb
+
+    want, reach = set(keys), catalogdb.closures(index)
+    names = set(files)
+    for loc in index["locations"]:
+        if loc["primaryKey"] in want:
+            names |= {stable for _, stable in reach[loc["id"]]}
+    return parse_selection([f"bundle:{_literal(stable)}" for stable in names])
+
+
+def subject_pipeline(stages: dict[str, Stage], want) -> list[str]:
+    """The stages `want` and the stages they read (Stage.after, transitively), in the order of the pipeline."""
+    need, stack = set(), list(want)
+    while stack:
+        name = stack.pop()
+        if name not in need:
+            need.add(name)
+            stack += [a for a in stages[name].after if a in stages]
+    return [n for n in stages if n in need or n == "catalog.index"]
+
+
+class SubjectRun:
+    """The stages of one command's subjects, run in this process: `run(want, params, keys=, files=)` runs the stages
+    `want` and those they read over the files the catalog keys `keys` load and the files `files` (key_selection);
+    `result(task id)` is a task's result. The store, catalog and settings are those of `export assets`."""
+
+    def __init__(self, args, cfg):
+        self.ws = Workspace(args, cfg)
+        self.ws.run_index()
+
+    @property
+    def store(self):
+        return self.ws.store
+
+    def index(self) -> dict:
+        return self.ws.index()
+
+    def run(self, want, params: dict | None = None, *, keys=(), files=()):
+        from ..pipeline.orchestrate import Orchestrator
+
+        ws = self.ws
+        ws.selection = key_selection(ws.index(), keys, files)
+        pipeline = subject_pipeline(ws.installed.stages, want)
+        facts = ws.facts(pipeline, fetch=True)
+        self.orch = Orchestrator(ws.store, ws.scheduled, workers=0, log=ws.log)
+        self.last = self.orch.run(pipeline, params=params, facts=facts, context=ws.context(), selection=ws.selection)
+        return self.last
+
+    def result(self, tid: str) -> dict:
+        """The result of task `tid` of the last run; when it failed or did not run (a file it reads could not be
+        read), the command exits with status 1 naming why."""
+        t = self.last.tasks.get(tid)
+        if t is None or t["status"] == "failed":
+            why = [f"{f['code']}: {f['message']}" for f in self.last.failures_ if f["task"] == tid]
+            why += [f"{p['name']}: {p['message']}" for p in self.ws.problems()]
+            sys.exit(f"nnnotes: {tid}: no result" + (f" ({'; '.join(why)})" if why else ""))
+        return self.store.committed(t["key"], tid)
+
+    def names(self, stage: str) -> dict:
+        """{subject: catalog names} of a stage that names its tasks' files (Workspace.names), after the last run."""
+        self.ws.done({tid: t["key"] for tid, t in self.last.tasks.items() if t["status"] != "failed"})
+        return self.ws.names(stage)
+
+
+def place(store, doc: dict, out: Path, roles=None) -> list[str]:
+    """Write the artifacts of a result into the directory `out`, each as the file its role names (the part of its
+    id after the task id), relative to `out`; `roles`: only those. Returns the files written, sorted."""
+    from ..pipeline.store import write_file
+
+    written = []
+    for a in doc["artifacts"]:
+        _, role = contract.parse_artifact_id(a["id"])
+        if roles is not None and role not in roles:
+            continue
+        dst = Path(out, *role.split("/"))
+        dst.parent.mkdir(parents=True, exist_ok=True)
+        write_file(dst, store.read(a["content"]["sha256"]))
+        written.append(role)
+    return sorted(written)
+
+
 # ---------------------------------------------------------------- export
 class Totals:
     """What export reports of the results besides the run's counts, from its one reading of each (Run.scan): the

src/nnnotes/cli/export.py

@@ -9,6 +9,7 @@
 from pathlib import Path
 
 from ..core.jsonio import write_json
+from ..pipeline.store import write_file
 from ..source.session import Session
 from . import assets
 from .common import (
@@ -122,16 +123,24 @@ def cmd_live(args, cfg):
 
 
 def cmd_live2d(args, cfg):
+    from ..pipeline import contract
+    from ..pipeline.modelstages import LIVE2D
     from ..site import webmodel
-    from ..unity import live2d
 
     cfg.require_path("paths", "apk")  # the Cubism component classes are read from the APK
-    cat = Session(cfg).catalog()
+    run = assets.SubjectRun(args, cfg)
+    prefix = webmodel.LIVE2D_PREFIX
+    keys = {loc["primaryKey"] for loc in run.index()["locations"] if loc["primaryKey"].startswith(prefix)}
     try:  # a model id or a model key
-        (key,) = webmodel.catalog_models(cat, [args.key]).values()
+        (key,) = webmodel.select([args.key], webmodel.model_keys(sorted(keys))).values()
     except ValueError as e:
         usage(args, str(e))
-    report(args, live2d.extract_model(cat, key, Path(args.out)))
+    run.run([LIVE2D], keys=[key])
+    doc = run.result(contract.task_id(LIVE2D, key))
+    out = Path(args.out)
+    assets.place(run.store, doc, out)
+    (summary,) = [a["semantics"]["facts"] for a in doc["artifacts"] if a["semantics"]["kind"] == "live2d.model3"]
+    report(args, summary)
 
 
 def cmd_spot(args, cfg):
@@ -176,35 +185,82 @@ def cmd_room(args, cfg):
 
 def cmd_shader(args, cfg):
     from ..formats import shader
-
-    cat = Session(cfg).catalog()
+    from ..pipeline import contract
+    from ..pipeline.modelstages import Closures
+    from ..pipeline.objexport import EXPORT
+
+    if not args.key:
+        cfg.require_path("paths", "apk")
+    run = assets.SubjectRun(args, cfg)
+    index = run.index()
     if args.key:
-        known_keys(args, cat, [args.key])
-        bundles = cat.fetch_key(args.key)
+        if not any(loc["primaryKey"] == args.key for loc in index["locations"]):
+            usage(args, f"not a key of the catalog: {args.key}")
+        order = Closures(index).closure_bundles(args.key)  # the bundles of the key's closure, as the catalog loads them
     else:
-        if cat.apk is None:
-            cfg.require_path("paths", "apk")
-        bundles = []
+        order = []
         for s in args.apk_bundle:
-            try:
-                bundles.append(cat.apk_bundle(s))
-            except KeyError as e:  # none or several bundles match
-                usage(args, f"--apk-bundle {e.args[0]}")
-    r = shader.dump(bundles, Path(args.out))
+            names = [b["stable"] for b in index["bundles"] if not b["remote"] and s in b["name"]]
+            if len(names) != 1:
+                usage(args, f"--apk-bundle {s!r}: {len(names)} APK bundles match")
+            order += names
+    run.run([EXPORT], {EXPORT: {"classes": ["Shader"]}}, keys=[args.key] if args.key else (), files=order)
+    out = Path(args.out)
+    names = {b["stable"]: b["name"] for b in index["bundles"]}
+    records: list = []  # shaders.json: the first shader of each name, in bundle and object order
+    for stable in order:
+        doc = run.result(contract.task_id(EXPORT, stable))
+        arts = {a["id"]: a for a in doc["artifacts"]}
+        shaders = [it for it in doc["items"] if it.get("class") == "Shader"]
+        for it in sorted(shaders, key=lambda it: contract.parse_object_id(it["object"])):
+            if it["status"] != "exported":
+                sys.exit(f"nnnotes: shader {it['object']} of {names[stable]}: {it.get('why', {}).get('message')}")
+            obj = it["object"]
+            index_doc = contract.loads(run.store.read(arts[contract.artifact_id(obj, "index")]["content"]["sha256"]))
+            name = index_doc["name"]
+            if any(r["name"] == name for r in records):
+                continue
+            safe = shader.safe_name(name)
+            parsed = run.store.read(arts[contract.artifact_id(obj, "json")]["content"]["sha256"])
+            out.mkdir(parents=True, exist_ok=True)
+            write_file(out / f"{safe}.json", parsed)
+            variants = []
+            for v in index_doc["variants"]:
+                code = run.store.read(arts[contract.artifact_id(obj, v["role"])]["content"]["sha256"])
+                write_file(out.joinpath(*v["file"].split("/")), code)
+                variants.append({k: x for k, x in v.items() if k != "role"})
+            records.append({"name": name, "source": names[stable], "parsed": f"{safe}.json", "variants": variants})
+    r = shader.write_index(records, out)
     report(args, r, lambda r: [f"{r['count']} shaders, {r['variants']} variants", *(f"  {n}" for n in r["names"])])
 
 
 def cmd_audio(args, cfg):
-    from ..formats import cri
+    from ..pipeline import contract
+    from ..pipeline.cristages import AUDIO
 
     flac = flac_options(args)
-    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}"):
-        usage(args, f"cue sheet {args.cue_sheet}: no key Cri/Sound/{args.cue_sheet} in the catalog")
+    cfg.require_path("paths", "apk")  # the HCA keycode is read from the APK's boot data
+    run = assets.SubjectRun(args, cfg)
+    key = f"Cri/Sound/{args.cue_sheet}"
+    if not any(loc["primaryKey"] == key for loc in run.index()["locations"]):
+        usage(args, f"cue sheet {args.cue_sheet}: no key {key} in the catalog")
+    params = {"format": args.format}
+    if "flac_level" in flac:
+        params["flac"] = {"level": flac["flac_level"]}
+    if args.format == "flac":
+        del params["format"]
+    run.run([AUDIO], {AUDIO: params}, keys=[key])
+    subjects = [s for s, names in run.names(AUDIO).items() if key in names]
+    if len(subjects) != 1:
+        sys.exit(f"nnnotes: cue sheet {args.cue_sheet}: {len(subjects)} ACBs have the key {key}")
+    doc = run.result(contract.task_id(AUDIO, subjects[0]))
+    (item,) = doc["items"]
+    if item["status"] != "exported":
+        sys.exit(f"nnnotes: cue sheet {args.cue_sheet}: {item['why']['message']}")
     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)"])
+    assets.place(run.store, doc, out)
+    (cues,) = [a["semantics"]["facts"]["cues"] for a in doc["artifacts"] if a["semantics"]["kind"] == "cri.cues"]
+    report(args, {"out": str(out), "cues": cues}, lambda r: [f"{r['out']}  ({r['cues']} cues)"])
 
 
 def cmd_ui(args, cfg):

src/nnnotes/formats/cri.py

@@ -219,7 +219,8 @@ def decode(
     """Decode every stream of a cue sheet. Returns {streamName: file of its first stream}.
 
     flac (default) keeps the decoded PCM bit-exact (`flac_level`: ffmpeg's compression level; any level decodes to
-    the same samples); ogg re-encodes (lossy). `workers`: streams decoded at once (default default_workers()).
+    the same samples); ogg re-encodes (lossy; bit-exact flags, so the same samples give the same file). `workers`:
+    streams decoded at once (default default_workers()).
     `also` = (extension, ffmpeg codec options): every stream is encoded a second time from the same decoded samples
     into `<stem><extension>` next to its file (encode_args: no metadata, bit-exact flags), e.g. the web format of a
     BGM, so that it need not be transcoded from the first file later; the manifests name the first file only.
@@ -360,14 +361,15 @@ def one(item) -> Path:
         if fmt == "wav":
             dst = out_dir / wav.name
             shutil.move(str(wav), dst)
-        else:
-            dst = out_dir / f"{stem}.{fmt}"
-            codec = {
-                "flac": ["-c:a", "flac", "-compression_level", str(int(flac_level))],
-                "ogg": ["-c:a", "libvorbis", "-q:a", "5"],
-            }[fmt]
+        elif fmt == "flac":
+            dst = out_dir / f"{stem}.flac"
+            codec = ["-c:a", "flac", "-compression_level", str(int(flac_level))]
             subprocess.run([encoder, "-y", "-loglevel", "error", "-i", str(wav), *codec, str(dst)], check=True)
             wav.unlink()
+        else:  # ogg: bit-exact, so that the same samples give the same file
+            dst = out_dir / f"{stem}.ogg"
+            subprocess.run([encoder, *encode_args(wav, dst, ["-c:a", "libvorbis", "-q:a", "5"])], check=True)
+            wav.unlink()
         return dst
 
     with ThreadPoolExecutor(n) as ex: