#!/usr/bin/env python """garment.py -- one entry point for the clothing lane (MD -> Blender -> ariki-game). Design: .agents/plans/clothing-pipeline-unification-2026-07-31.md Contract: clothing/PIPELINE-CONTRACT.md (agent A owns this file only) A DRIVER, NOT A REWRITE. Every stage shells out to the tool that already owns it; the only stage implemented in-process is `register` (emit-not-edit codegen). python clothing/garment.py configs/piupiu_sb.json # census..export python clothing/garment.py configs/x.json --only census python clothing/garment.py configs/x.json --from fit --to skin python clothing/garment.py configs/x.json --qc deep # sign-off sweep python clothing/garment.py configs/x.json --list python clothing/garment.py configs/x.json --dry-run python clothing/garment.py --selftest Stage order (three worlds, one list): draft drape publish | census prepare fit reduce bake skin export | register import verify +---- MD bridge ----+ +-------- Blender headless -------------+ +---- ariki-game ----+ Default range is census..export -- Blender only. MD and game stages run only when asked for by name via --from/--to/--only. Range rules (predictable, no surprise stages): * neither flag -> census..export * --from X (X after export) -> X..verify * --to Y (Y before census) -> draft..Y * otherwise the missing end stays at its default. WHERE resolved.json LIVES (the one open question in the brief -- verified, not guessed) ---------------------------------------------------------------------------------------- `work//resolved.json`, exactly as the contract specifies. Verified safe against the real code rather than assumed: * garment_pipeline.py:1005 -- `cfg_path = ARGS.config if os.path.isabs(ARGS.config) else os.path.join(HERE, ARGS.config)`. An ABSOLUTE --config path is accepted verbatim, so a config outside configs/ is fine. * Relative paths INSIDE a config are resolved against HERE (the clothing/ directory), never against the config file's own directory -- see the texture branch at garment_pipeline.py:265 (`os.path.join(HERE, tex)`). `work_path()` (:36-39) likewise keys off `CFG["name"]` under `HERE/work/`, not off the config location. * Therefore moving a config from configs/ to work// changes NOTHING about how its paths resolve *for garment_pipeline.py*. The `configs/.resolved/` fallback described in the brief is therefore NOT needed and is not implemented -- one location, no ambiguity for the other four agents. INTEGRATION FIX (2026-07-31): the paragraph above is true of garment_pipeline.py and FALSE of the other two consumers. tools/tailor/draft_garment.py and gates/g1_drape.py resolve relative paths against the CONFIG FILE'S OWN directory first -- work// -- so `"texture": "../tools/tailor/textures/piupiu.png"` would land in clothing/work/tools/. `absolutize()` now rewrites every known path field to an absolute path (base: clothing/, matching garment_pipeline's HERE) before resolved.json is written, so all three rules agree. See PATH_FIELDS. QC MODES (--qc light|deep) -------------------------- Two ways to use this pipeline, two costs. `light` is the iterate-on-a-garment loop: one clip, two frames, one anim pack, gate failures REPORT instead of stopping, so you get the whole picture of a work-in-progress garment in seconds. `deep` is the sign-off sweep: three clips, eight frames each, both packs, the synthetic extremes on top, and a failing gate stops the run. A mode is a PROFILE -- a small dict of SAMPLING/COST knobs merged into the resolved config at resolve time. A profile may only touch the keys in QC_PROFILE_KEYS (clips/frames_per_clip/anim_pack/pose_source/max_render_frames + verify clips/frames + gate_stop). It can never touch a THRESHOLD -- depth_mm, max_verts, max_verts_rest, gape budgets, islands, bands, budget are the config author's alone, so switching modes changes how hard you look, never what counts as a defect. The knobs a mode does own it OVERRIDES (that is the point of a mode); a config that wants to keep its own value for one of them lists it in `qc.pin`. Gates stay mode-unaware: they receive the same resolved.json they always did, with different numbers in it. See QC_PROFILES and apply_qc_mode(). Python 3 stdlib only. """ import argparse import json import os import re import shutil import subprocess import sys import time # -- layout ------------------------------------------------------------------- HERE = os.path.dirname(os.path.abspath(__file__)) # .../animation/clothing REPO = os.path.dirname(HERE) # .../animation CONFIGS = os.path.join(HERE, "configs") WORK = os.path.join(HERE, "work") GATES = os.path.join(HERE, "gates") PIPELINE = os.path.join(HERE, "garment_pipeline.py") MD_BRIDGE = os.path.join(REPO, "tools", "md_bridge.py") DRAFT_GARMENT = os.path.join(REPO, "tools", "tailor", "draft_garment.py") GAME_REPO = os.environ.get( "ARIKI_GAME", os.path.join(os.path.dirname(REPO), "ariki-game")) OUTFIT_CATALOG = os.path.join(GAME_REPO, "src", "Character", "OutfitCatalog.cs") TARGETED_REIMPORT = os.path.join(GAME_REPO, "tools", "targeted_reimport.sh") MOTION_QA = os.path.join(GAME_REPO, "tools", "clothing_motion_qa.sh") BLENDER = os.environ.get( "BLENDER", r"C:/Program Files/Blender Foundation/Blender 5.1/blender.exe") # G6 baseline blessing: --bless-baseline, or GARMENT_BLESS=1 in the environment. BLESS_BASELINE = False # -- stages ------------------------------------------------------------------- MD_STAGES = ["draft", "drape", "publish"] BLENDER_STAGES = ["census", "prepare", "fit", "reduce", "bake", "skin", "export"] GAME_STAGES = ["register", "import", "verify"] STAGES = MD_STAGES + BLENDER_STAGES + GAME_STAGES DEFAULT_FROM = "census" DEFAULT_TO = "export" # Gate map (PIPELINE-CONTRACT "Orchestrator gate map"). # stage -> list of (gate_id, script, kind, expect_key, extra_flags) # expect_key None => gate always runs (g8). # kind "blender" => run under blender --background --python -- ... # "plain" => run under system python. # CONFIRMED as-built (2026-07-31, integrator): g5_posed_sweep.py imports bpy and opens # 20_fit.blend/50_skin.blend, so it is the one Blender gate. g1/g2/g8 are stdlib(+PIL) # and read JSON/PNG/OutfitCatalog.cs, so they are plain-python gates. # Override per gate with resolved config `gates: {"g5": {"kind": "plain"}}`. GATE_MAP = { "drape": [("g1", "g1_drape.py", "plain", "bands", [])], "census": [("g2", "g2_census.py", "plain", "islands", [])], "fit": [("g5", "g5_posed_sweep.py", "blender", "penetration", ["--rest"])], "skin": [("g5", "g5_posed_sweep.py", "blender", "penetration", [])], "register": [("g8", "g8_catalog_lint.py", "plain", None, [])], } # -- QC modes ----------------------------------------------------------------- # light = the iterate loop (fast, reports everything, never stops the run). # deep = the sign-off sweep (every clip, more frames, synthetic extremes, stops). # # A profile owns SAMPLING/COST knobs only. Thresholds (depth_mm, max_verts, # max_verts_rest, gape.*, islands, bands, budget) belong to the config author and are # untouchable here -- a mode changes how hard you look, never what counts as a defect. ANIM_DIR = os.path.join(GAME_REPO, "assets", "quaternius", "anim") UAL1 = os.path.join(ANIM_DIR, "UAL1.glb").replace("\\", "/") UAL2 = os.path.join(ANIM_DIR, "UAL2.glb").replace("\\", "/") QC_MODES = ("light", "deep") DEFAULT_QC_MODE = "light" # The ONLY keys a profile (builtin or a config's `qc.` block) may set. QC_ALLOWED_PENETRATION = ("clips", "frames_per_clip", "anim_pack", "pose_source", "max_render_frames") QC_ALLOWED_VERIFY = ("clips", "frames") QC_ALLOWED_TOP = ("gate_stop",) QC_PROFILE_KEYS = { "expect.penetration": QC_ALLOWED_PENETRATION, "verify": QC_ALLOWED_VERIFY, "": QC_ALLOWED_TOP, } QC_PROFILES = { # ~13 s of gate time on piupiu_sb_test: ONE pack (UAL2 is the one that carries # Walk_Fwd_Loop, which is what the game aliases "Walk" onto -- loading UAL1 too # costs ~7 s and buys nothing when Idle/Dance are not sampled). "light": { "expect": {"penetration": { "clips": ["Walk"], "anim_pack": [UAL2], "frames_per_clip": 2, "max_render_frames": 2, # "auto", not "clips": if Walk somehow fails to resolve the gate falls # back to synthetic extremes rather than erroring. It never passes for # lack of animation, in any mode. "pose_source": "auto", }}, "verify": {"clips": "Walk", "frames": 1}, "gate_stop": False, }, "deep": { "expect": {"penetration": { "clips": ["Idle", "Walk", "Dance"], "anim_pack": [UAL1, UAL2], "frames_per_clip": 8, "max_render_frames": 12, # "both" = the clips, then the builtin synthetic extremes appended # (g5_posed_sweep.py). Harsher than any shipped clip, by design. "pose_source": "both", }}, "verify": {"clips": "Idle Walk Dance", "frames": 3}, "gate_stop": True, }, } QC_LIGHT_REMINDER = "QC mode: light -- run --qc deep before sign-off" MD_NO_SESSION = ( "no Marvelous Designer bridge session.\n" " Start MD, then click Plugin > TinqsMDBridge (once per MD session).\n" " MD's UI freezes while the session is live -- that is expected.\n" " Verify with: python tools/md_bridge.py --ping") OUTFIT_SLOTS = {"Body", "Arms", "Legs", "Feet", "HeadGear", "Accessories"} def log(msg): print("[garment] " + str(msg), flush=True) def warn(msg): print("[garment] WARNING: " + str(msg), flush=True) def die(msg, code=1): print("[garment] ERROR: " + str(msg), file=sys.stderr, flush=True) raise SystemExit(code) # -- config resolution -------------------------------------------------------- def deep_merge(base, over): """Child over parent. Objects merge per key; arrays and scalars replace whole.""" if isinstance(base, dict) and isinstance(over, dict): out = dict(base) for k, v in over.items(): out[k] = deep_merge(base[k], v) if k in base else v return out return over # Fields whose relative paths are made ABSOLUTE before resolved.json is written. # WHY (this bit is load-bearing -- see the module docstring's caveat above): # garment_pipeline.py resolves relative config paths against HERE (clothing/), so for # IT the location of resolved.json is irrelevant. But two other consumers do not use # that rule: # * tools/tailor/draft_garment.py -- resolves against the CONFIG FILE'S OWN directory, # then the repo root. # * clothing/gates/g1_drape.py:_resolve -- same (config_dir, REPO_ROOT, cwd). # resolved.json lives in work//, so "../tools/tailor/textures/piupiu.png" would # resolve to clothing/work/tools/... and silently miss. Absolutising here makes every # consumer agree regardless of its own rule, and is a no-op for already-absolute paths. PATH_FIELDS = ( ("source",), ("body",), ("parts", "*", "texture"), ("md", "avatar_fbx"), ("md", "zfab"), ("md", "texture"), ("md", "snapshot"), ("md", "presim_snapshot"), ("md", "export_dir"), ("export", "out_dir"), ) def abs_path(value, base=None): """Relative -> absolute against `base` (default clothing/). Absolute is untouched. Non-strings, empty strings and whitespace are returned unchanged so a config can still carry `"texture": null` or `""` without this turning it into a directory. """ if not isinstance(value, str) or not value.strip(): return value p = os.path.expanduser(value) if os.path.isabs(p): return os.path.normpath(p).replace("\\", "/") return os.path.normpath(os.path.join(base or HERE, p)).replace("\\", "/") def _abs_at(node, path, base): """Apply abs_path at `path` inside `node`. '*' expands over a dict's values.""" if not isinstance(node, dict): return key, rest = path[0], path[1:] if key == "*": for v in node.values(): _abs_at(v, rest, base) return if key not in node: return if rest: _abs_at(node[key], rest, base) else: node[key] = abs_path(node[key], base) def absolutize(resolved, base=None): """Make every known path field absolute (against clothing/). Mutates and returns.""" for path in PATH_FIELDS: _abs_at(resolved, path, base) return resolved def _read_json(path): try: with open(path, "r", encoding="utf-8") as f: return json.load(f) except FileNotFoundError: die("config not found: %s" % path) except ValueError as exc: die("config is not valid JSON: %s\n %s" % (path, exc)) def find_config(spec): """Accept a config path relative to cwd, clothing/, clothing/configs/, or the repo.""" if os.path.isabs(spec): return spec tried = [] for base in ("", HERE, CONFIGS, REPO): cand = os.path.join(base, spec) if base else spec if os.path.exists(cand): return cand tried.append(os.path.abspath(cand)) die("config not found: %s\n tried:\n %s" % (spec, "\n ".join(tried))) def resolve_config(path): """Load `path`, apply its `extends` chain, return (resolved, chain, child_raw). `extends` is a single parent, path relative to configs/ (absolute also accepted). Cycles are detected and reported with the full chain. """ path = os.path.abspath(path) chain = [] seen = [] node = path docs = [] while node: real = os.path.normcase(os.path.realpath(node)) if real in seen: die("`extends` cycle: %s -> %s" % (" -> ".join(os.path.basename(c) for c in chain), os.path.basename(node))) seen.append(real) chain.append(node) doc = _read_json(node) docs.append(doc) parent = doc.get("extends") if not parent: break node = parent if os.path.isabs(parent) else os.path.join(CONFIGS, parent) if not os.path.exists(node): die("`extends` target not found: %s\n (referenced by %s; parents resolve " "relative to %s)" % (node, chain[-1], CONFIGS)) resolved = {} for doc in reversed(docs): # root parent first resolved = deep_merge(resolved, doc) resolved.pop("extends", None) absolutize(resolved) child = docs[0] name = child.get("name") or resolved.get("name") if not name: die("config has no `name` (needed for work//): %s" % path) if not child.get("name") and len(docs) > 1: warn("child config has no `name`; inherited '%s' from the extends chain " "-- work dir will be shared with the parent" % name) resolved["name"] = name resolved["_resolved"] = { "config": path, "chain": [os.path.relpath(c, HERE).replace("\\", "/") for c in chain], "generated": time.strftime("%Y-%m-%dT%H:%M:%S"), } return resolved, chain, child # -- QC mode profiles --------------------------------------------------------- def sanitize_qc_profile(block, label): """Keep only the keys a QC profile is allowed to set. -> (clean, dropped[]). Anything else in a config's `qc.` block -- a threshold, a path, a whole other `expect` sub-block -- is DROPPED with a note. A QC mode must never be able to change what counts as a defect, only how hard the pipeline looks for it. """ clean, dropped = {}, [] b = block if isinstance(block, dict) else {} if block is not None and not isinstance(block, dict): return clean, ["%s is not an object -- ignored" % label] exp = b.get("expect") if isinstance(b.get("expect"), dict) else {} pen = exp.get("penetration") if isinstance(exp.get("penetration"), dict) else {} keep = {k: v for k, v in pen.items() if k in QC_ALLOWED_PENETRATION} dropped += ["%s.expect.penetration.%s" % (label, k) for k in pen if k not in QC_ALLOWED_PENETRATION] dropped += ["%s.expect.%s" % (label, k) for k in exp if k != "penetration"] if keep: clean["expect"] = {"penetration": keep} ver = b.get("verify") if isinstance(b.get("verify"), dict) else {} vkeep = {k: v for k, v in ver.items() if k in QC_ALLOWED_VERIFY} dropped += ["%s.verify.%s" % (label, k) for k in ver if k not in QC_ALLOWED_VERIFY] if vkeep: clean["verify"] = vkeep for k in QC_ALLOWED_TOP: if k in b: clean[k] = bool(b[k]) dropped += ["%s.%s" % (label, k) for k in b if k not in ("expect", "verify") + QC_ALLOWED_TOP] return clean, dropped def normalize_pins(pins): """`qc.pin` entries -> canonical dotted paths. Accepts the shorthand ("clips" == "expect.penetration.clips") and the full path ("verify.frames"). Unknown names are returned as-is so the caller can warn. """ out = [] for p in (pins or []): p = str(p) if p in QC_ALLOWED_PENETRATION: out.append("expect.penetration." + p) elif p.startswith("verify.") or p.startswith("expect.penetration."): out.append(p) else: out.append(p) return out def qc_profile_for(mode, qc_cfg): """Builtin profile <- config `qc.` block. -> (profile, dropped[]).""" over, dropped = sanitize_qc_profile((qc_cfg or {}).get(mode), "qc.%s" % mode) return deep_merge(QC_PROFILES[mode], over), dropped def apply_qc_mode(resolved, cli_mode=None, cli_gate_stop=None, quiet=False): """Merge the QC-mode profile into `resolved`. Returns the `_resolved.qc_mode` record. Precedence, lowest to highest: builtin QC_PROFILES[mode] -> config `qc.` block -> CLI against the config's own `expect`/`verify`: * THRESHOLDS (everything not in QC_PROFILE_KEYS) are never touched -- the config author's values stand in both modes. * COST KNOBS are owned by the active mode and override the config's own value; that is what makes `--qc light` faster than the config as written. A config that wants to keep its own value for a knob lists it in `qc.pin`, and then its value wins over the profile. `expect.penetration` is only touched when it already exists: its presence is what ARMS G3/G5 (see gates_for), so a profile must not conjure a gate the author never asked for. """ qc_cfg = resolved.get("qc") or {} mode, source = DEFAULT_QC_MODE, "default" if qc_cfg.get("default_mode"): mode, source = str(qc_cfg["default_mode"]).strip().lower(), "config qc.default_mode" if mode not in QC_PROFILES: die("config qc.default_mode is %r -- known modes: %s" % (qc_cfg.get("default_mode"), ", ".join(QC_MODES))) if cli_mode: mode, source = str(cli_mode).strip().lower(), "--qc" if mode not in QC_PROFILES: die("unknown --qc mode %r -- known: %s" % (cli_mode, ", ".join(QC_MODES))) profile, dropped = qc_profile_for(mode, qc_cfg) pins = normalize_pins(qc_cfg.get("pin")) known = (["expect.penetration." + k for k in QC_ALLOWED_PENETRATION] + ["verify." + k for k in QC_ALLOWED_VERIFY]) for p in pins: if p not in known: dropped.append("qc.pin[%s] is not a profile knob -- ignored" % p) applied, pinned, skipped = {}, [], [] pen = (profile.get("expect") or {}).get("penetration") or {} exp = resolved.get("expect") if pen: if isinstance(exp, dict) and isinstance(exp.get("penetration"), dict): target = exp["penetration"] for k, v in pen.items(): path = "expect.penetration." + k if path in pins and k in target: pinned.append(path) continue target[k] = v applied[path] = v else: skipped.append("expect.penetration absent -- G3/G5 not armed, " "penetration knobs not applied") ver = profile.get("verify") or {} if ver: target = resolved.setdefault("verify", {}) for k, v in ver.items(): path = "verify." + k if path in pins and k in target: pinned.append(path) continue target[k] = v applied[path] = v gate_stop = bool(profile.get("gate_stop", True)) gate_stop_source = "profile" if cli_gate_stop is not None: gate_stop, gate_stop_source = bool(cli_gate_stop), "CLI" record = { "mode": mode, "source": source, "gate_stop": gate_stop, "gate_stop_source": gate_stop_source, "applied": applied, "pinned": sorted(set(pinned)), "notes": dropped + skipped, } if not quiet: for n in record["notes"]: warn("qc: %s" % n) return record def qc_banner(record): """The run banner lines: mode, what it means, and the effective knob values.""" a = record["applied"] lines = ["QC mode: %s (%s) -- gate failures %s%s" % (record["mode"], record["source"], "STOP the run" if record["gate_stop"] else "REPORT only", "" if record["gate_stop_source"] == "CLI" else (" [--gate-stop to stop]" if not record["gate_stop"] else " [--no-gate-stop to continue]"))] def group(prefix, keys): bits = [] for k in keys: v = a.get(prefix + k) if v is None: continue if k == "anim_pack": v = [os.path.basename(str(p)) for p in (v if isinstance(v, (list, tuple)) else [v])] bits.append("%s=%s" % (k, v if not isinstance(v, str) else repr(v))) return bits pen = group("expect.penetration.", QC_ALLOWED_PENETRATION) if pen: lines.append(" g3/g5 " + " ".join(pen)) ver = group("verify.", QC_ALLOWED_VERIFY) if ver: lines.append(" verify " + " ".join(ver)) if record["pinned"]: lines.append(" pinned by config (profile did not override): " + ", ".join(record["pinned"])) return lines def write_resolved(resolved, dry_run=False): d = os.path.join(WORK, resolved["name"]) p = os.path.join(d, "resolved.json") if dry_run: log("dry-run: would write %s" % p) return d, p os.makedirs(os.path.join(d, "qc"), exist_ok=True) with open(p, "w", encoding="utf-8") as f: json.dump(resolved, f, indent=1, ensure_ascii=False) log("resolved config -> %s" % p) return d, p # -- command building --------------------------------------------------------- def blender_cmd(script, args): return [BLENDER, "--background", "--python", script, "--"] + args def stage_commands(stage, resolved, workdir, resolved_path, md_timeout): """Return the list of commands for a stage. `register` returns [] (in-process).""" if stage in BLENDER_STAGES: return [blender_cmd(PIPELINE, ["--config", resolved_path, "--stage", stage])] if stage in MD_STAGES: script = os.path.join(workdir, "md_script.py") return [ [sys.executable, MD_BRIDGE, "--ping"], [sys.executable, DRAFT_GARMENT, "--config", resolved_path, "--emit", script, "--stage", stage], [sys.executable, MD_BRIDGE, "--file", script, "--timeout", str(md_timeout)], ] if stage == "register": return [] if stage == "import": return [["bash", TARGETED_REIMPORT, "--report", os.path.join(workdir, "qc", "g7.json")] + expected_gltfs(resolved)] if stage == "verify": return [verify_plan(resolved, workdir)["argv"]] die("unknown stage %r" % stage) # -- verify (G6) -------------------------------------------------------------- def gender_bit(resolved): """export.gender -> ClothingTestBed's BED_GENDER (ClothingTestBed.cs:27, 0=M 1=F).""" word = str((resolved.get("export") or {}).get("gender", "Female")) return 1 if word.strip().lower().startswith("f") else 0 def verify_plan(resolved, workdir): """Build agent E's clothing_motion_qa.sh invocation, exactly as that script documents. BED_SET= BED_GENDER=<0|1> bash /tools/clothing_motion_qa.sh \\ --clips "Idle Walk Dance" --frames 3 [--bless DIR | --diff DIR] \\ --report /qc/g6.json Baseline mode is decided from the filesystem, not from a flag nobody remembers: baseline dir exists -> --diff (regression tripwire) GARMENT_BLESS=1 / --bless-baseline -> --bless (freeze the current frames) neither -> NO bless/diff. The run still carries the HARD signal (outfit load errors on /console + engine stdout) and captures the evidence frames a human or VLM signs off on. Blessing on the first run would freeze a defect as the baseline. Overridable from the config's optional `verify` block: slug, baseline_dir, out_dir, clips, frames. """ exp = resolved.get("export") or {} v = resolved.get("verify") or {} set_name = str(exp.get("set", "")) # Directory slug: agent E's own example is lowercase (`clothing/baselines/kapahaka`, # `.game-cli/clothing-qa/kapahaka` for BED_SET=Kapahaka). Keep that convention so a # hand-run and a pipeline run share one baseline; override with verify.slug. slug = str(v.get("slug") or set_name or resolved["name"]).lower() baseline = v.get("baseline_dir") or os.path.join(HERE, "baselines", slug) out_dir = v.get("out_dir") or os.path.join(GAME_REPO, ".game-cli", "clothing-qa", slug) clips = v.get("clips") or "Idle Walk Dance" if isinstance(clips, (list, tuple)): clips = " ".join(str(c) for c in clips) frames = str(v.get("frames", 3)) baseline = os.path.abspath(baseline) out_dir = os.path.abspath(out_dir) argv = ["bash", MOTION_QA, "--clips", clips, "--frames", frames] if BLESS_BASELINE or os.environ.get("GARMENT_BLESS") == "1": mode = "bless" argv += ["--bless", baseline] elif os.path.isdir(baseline): mode = "diff" argv += ["--diff", baseline] else: mode = "first" argv += ["--report", os.path.join(workdir, "qc", "g6.json"), out_dir] return {"argv": argv, "env": {"BED_SET": set_name, "BED_GENDER": str(gender_bit(resolved))}, "mode": mode, "baseline": baseline, "out_dir": out_dir} def verify_note(plan): """The prominent first-sign-off banner, or None once a baseline exists.""" if plan["mode"] != "first": return None bar = "=" * 78 return "\n".join([ "", bar, " first sign-off: review frames in %s," % plan["out_dir"], " then re-run verify with GARMENT_BLESS=1 to bless", "", " No baseline at %s, so this run carried the HARD signal only" % plan["baseline"], " (outfit load errors). Pixel-diff regression cover starts after blessing.", " Blessing before looking freezes a defect as the thing every future run matches.", bar, ""]) def stage_env(stage, resolved, workdir): """Environment overrides a stage needs. Only `verify` has any (BED_* -> the test bed).""" if stage == "verify": return verify_plan(resolved, workdir)["env"] return None def expected_gltfs(resolved): """The GLTFs `export` writes: {out_dir}/{Gender}_{Set}_{Slot}.gltf (pipeline :976).""" exp = resolved.get("export", {}) out_dir = exp.get("out_dir", "") gender = exp.get("gender", "Female") set_name = exp.get("set", "") slots = [] for part in (resolved.get("parts") or {}).values(): slot = (part or {}).get("slot") if slot and slot not in slots: slots.append(slot) return [("%s/%s_%s_%s.gltf" % (out_dir.rstrip("/\\"), gender, set_name, s)) for s in slots] def show(cmd, env=None): def q(a): a = str(a) return '"%s"' % a if (" " in a or not a) else a pre = "".join("%s=%s " % (k, q(v)) for k, v in sorted((env or {}).items())) return pre + " ".join(q(a) for a in cmd) def run(cmd, dry_run, env=None): if dry_run: print(" " + show(cmd, env)) return 0 log("$ " + show(cmd, env)) exe = cmd[0] if not os.path.isabs(exe) and shutil.which(exe) is None: warn("%r is not on PATH" % exe) return 127 if os.path.isabs(exe) and not os.path.exists(exe): warn("not found: %s" % exe) return 127 child_env = None if env: child_env = dict(os.environ) child_env.update({k: str(v) for k, v in env.items()}) try: return subprocess.call(cmd, env=child_env) except OSError as exc: warn("could not run %s: %s" % (exe, exc)) return 127 # -- register (emit-not-edit) ------------------------------------------------- def fmt_float(v): v = float(v) s = "%.2f" % v return (s if float(s) == v else repr(v)) + "f" def cs_str(s): return '"%s"' % str(s).replace("\\", "\\\\").replace('"', '\\"') def catalog_entries(resolved): """Normalise the `catalog` block into one dict per Add() line. Accepted shapes (all derive slot/set/gender from `parts`/`export`): "catalog": {"id":..., "displayName":..., ...} flat, one part "catalog": {"items": [ {...}, {...} ]} explicit list "catalog": {"": {...}, "": {...}} per part """ cat = resolved.get("catalog") if not cat: return [], ["config has no `catalog` block -- nothing to register"] parts = resolved.get("parts") or {} exp = resolved.get("export", {}) set_name = exp.get("set") gender_word = str(exp.get("gender", "Female")) gender = 1 if gender_word.lower().startswith("f") else 0 notes = [] if not set_name: notes.append("export.set is missing -- `set:` will be empty") raw = [] if isinstance(cat.get("items"), list): raw = [(e.get("part"), e) for e in cat["items"]] elif any(k in parts for k in cat): raw = [(k, v) for k, v in cat.items() if k in parts] for k in cat: if k not in parts: notes.append("catalog key %r matches no part -- ignored" % k) else: slots = [p.get("slot") for p in parts.values() if p.get("slot")] uniq = sorted(set(slots)) if len(uniq) <= 1: raw = [(next(iter(parts), None), cat)] else: notes.append( "flat `catalog` but %d distinct slots (%s) -- ids derived as " "_; use per-part or `catalog.items` to control them" % (len(uniq), ", ".join(uniq))) for pname, p in parts.items(): e = dict(cat) e["id"] = "%s_%s" % (cat.get("id", "item"), str(p.get("slot", "")).lower()) raw.append((pname, e)) entries = [] for pname, e in raw: part = parts.get(pname, {}) if pname else {} suffix = e.get("slot_suffix") or e.get("slotSuffix") or part.get("slot") slot = e.get("slot") or part.get("slot") if slot not in OUTFIT_SLOTS: if suffix in OUTFIT_SLOTS or slot is None: notes.append("part %r slot %r is not an OutfitSlot -- set catalog.slot " "explicitly (one of %s)" % (pname, slot, ", ".join(sorted(OUTFIT_SLOTS)))) else: notes.append("slot %r is a file suffix, not an OutfitSlot; assuming " "OutfitSlot.Legs for part %r" % (slot, pname)) slot = "Legs" missing = [k for k in ("id", "displayName") if not e.get(k)] if missing: notes.append("catalog entry for part %r is missing %s" % (pname, "/".join(missing))) entries.append({ "id": e.get("id", "TODO_id"), "displayName": e.get("displayName", "TODO"), "slot": slot or "Body", "suffix": suffix or slot or "Body", "charisma": e.get("charisma", 0.0), "workSpeed": e.get("workSpeed", 0.0), "category": e.get("category", "casual"), "set": set_name or "", "gender": gender, }) return entries, notes def basedir_cases(): """Parse OutfitCatalog.BaseDirFor's switch arms -> {setId: dir}. {} if unreadable.""" try: with open(OUTFIT_CATALOG, "r", encoding="utf-8") as f: src = f.read() except OSError: return None m = re.search(r"BaseDirFor\(string setId\)\s*=>\s*setId switch\s*\{(.*?)\n\s*\};", src, re.S) if not m: return None return dict(re.findall(r'"([^"]+)"\s*=>\s*"([^"]+)"', m.group(1))) def stage_register(resolved, workdir, dry_run): entries, notes = catalog_entries(resolved) exp = resolved.get("export", {}) set_name = exp.get("set", "") out_dir = str(exp.get("out_dir", "")).replace("\\", "/").rstrip("/") # BaseDirFor wants a res:// path, so derive it from where out_dir actually SITS in # the game repo -- not from its basename. basename alone produced # "res://assets/quaternius/outfits/export" for a scratch out_dir, which points at # nothing; g8 would then flag the mismatch with no hint of the cause. want_dir, outside = None, False game_root = os.path.abspath(GAME_REPO) if out_dir: try: rel = os.path.relpath(os.path.abspath(out_dir), game_root) except ValueError: # different drive rel = ".." if not rel.startswith(".."): want_dir = "res://" + rel.replace("\\", "/") else: outside = True want_dir = "res://assets/quaternius/outfits/%s" % os.path.basename(out_dir) notes.append( "export.out_dir (%s) is OUTSIDE the game repo (%s) -- the BaseDirFor " "res:// path above is a placeholder guess. Export into " "assets/quaternius/outfits// (or copy the .gltf+.bin there) " "before pasting this." % (out_dir, game_root.replace("\\", "/"))) else: notes.append("export.out_dir is missing -- cannot derive a BaseDirFor path") want_dir = "res://assets/quaternius/outfits/TODO" lines = [] lines.append("// %s -- generated by clothing/garment.py register; PASTE, do not import." % resolved["name"]) lines.append("// Target: %s" % OUTFIT_CATALOG.replace("\\", "/")) lines.append("") cases = basedir_cases() if cases is None: notes.append("could not parse BaseDirFor from %s -- verify the case by hand" % OUTFIT_CATALOG) if cases is None or cases.get(set_name) != want_dir: if cases is not None and set_name in cases: lines.append("// BaseDirFor: set %r maps to %s but export.out_dir is %s" % (set_name, cases[set_name], want_dir)) lines.append("// 1. BaseDirFor() switch -- add this arm (OutfitCatalog.cs ~:88):") lines.append(' %s => "%s",' % (cs_str(set_name), want_dir)) lines.append("") else: lines.append('// BaseDirFor already maps "%s" -> %s (no switch edit needed)' % (set_name, want_dir)) lines.append("") lines.append("// Initialize() -- add inside the set's section (OutfitCatalog.cs ~:116):") for e in entries: lines.append( ' Add(%s, %s, OutfitSlot.%s, %s, %s, %s, %s, set: %s, gender: %d);' % (cs_str(e["id"]), cs_str(e["displayName"]), e["slot"], cs_str(e["suffix"]), fmt_float(e["charisma"]), fmt_float(e["workSpeed"]), cs_str(e["category"]), cs_str(e["set"]), e["gender"])) lines.append("") lines.append("// Assets these lines resolve to:") for p in expected_gltfs(resolved): mark = "OK " if os.path.exists(p) else "MISSING" lines.append("// [%s] %s" % (mark, p)) for n in notes: lines.append("// NOTE: %s" % n) text = "\n".join(lines) + "\n" out = os.path.join(workdir, "register.cs.txt") print(text) if dry_run: log("dry-run: would write %s" % out) return 0 with open(out, "w", encoding="utf-8") as f: f.write(text) log("register lines -> %s" % out) for n in notes: warn("register: %s" % n) # Emitting always succeeds; judging the result is g8's job (gate contract). return 0 # -- gates -------------------------------------------------------------------- def gates_for(stage, resolved): out = [] overrides = resolved.get("gates") or {} expect = resolved.get("expect") or {} for gate_id, script, kind, key, extra in GATE_MAP.get(stage, []): if key is not None and key not in expect: continue kind = (overrides.get(gate_id) or {}).get("kind", kind) out.append((gate_id, os.path.join(GATES, script), kind, extra)) return out def gate_cmd(script, kind, extra, resolved_path, workdir): args = ["--config", resolved_path, "--work", workdir] + list(extra) if kind == "blender": return blender_cmd(script, args) return [sys.executable, script] + args def qc_path(workdir, gate_id, extra): """Contract: work//qc/.json. --rest mode may write g3.json instead.""" cands = [os.path.join(workdir, "qc", "%s.json" % gate_id)] if "--rest" in extra: cands.insert(0, os.path.join(workdir, "qc", "g3.json")) found = [c for c in cands if os.path.exists(c)] if not found: return "-" return os.path.relpath(max(found, key=os.path.getmtime), HERE).replace("\\", "/") def run_gates(stage, resolved, resolved_path, workdir, dry_run, gate_stop, rows): """Returns True to keep going, False to stop the run.""" for gate_id, script, kind, extra in gates_for(stage, resolved): label = gate_id + (" --rest" if "--rest" in extra else "") if not dry_run and not os.path.exists(script): warn("gate %s not built yet (%s) -- skipping" % (label, script)) rows.append((stage + " :gate", "SKIP", label, "not built", "-")) continue if dry_run: print(" [gate %s]" % label) t0 = time.time() rc = run(gate_cmd(script, kind, extra, resolved_path, workdir), dry_run) secs = "%.1fs" % (time.time() - t0) if dry_run: continue qc = qc_path(workdir, gate_id, extra) if rc == 0: log("gate %s PASS (%s)" % (label, secs)) rows.append((stage + " :gate", "PASS", label, qc, secs)) elif rc == 3: warn("gate %s could not evaluate (exit 3) -- continuing" % label) rows.append((stage + " :gate", "WARN", label, qc, secs)) elif rc == 127: rows.append((stage + " :gate", "SKIP", label, "not runnable", secs)) else: kind_txt = "FAIL" if rc == 2 else "FAIL(exit %d)" % rc warn("gate %s %s (%s)" % (label, kind_txt, secs)) rows.append((stage + " :gate", kind_txt, label, qc, secs)) if gate_stop: return False return True # -- run ---------------------------------------------------------------------- def select_range(args): if args.only: only = [s.strip() for s in args.only.split(",") if s.strip()] bad = [s for s in only if s not in STAGES] if bad: die("unknown stage(s): %s\n known: %s" % (", ".join(bad), " ".join(STAGES))) return sorted(only, key=STAGES.index) start = args.from_stage or DEFAULT_FROM end = args.to_stage or DEFAULT_TO for s in (start, end): if s not in STAGES: die("unknown stage %r\n known: %s" % (s, " ".join(STAGES))) i, j = STAGES.index(start), STAGES.index(end) if args.from_stage and not args.to_stage and i > j: j = len(STAGES) - 1 if args.to_stage and not args.from_stage and j < i: i = 0 if i > j: die("--from %s comes after --to %s" % (start, end)) return STAGES[i:j + 1] def md_preflight(resolved_path, workdir, dry_run, md_timeout): """--ping, then emit the script. Returns (ok, cmds_left) -- degrades gracefully.""" if dry_run: return True, None if not os.path.exists(MD_BRIDGE): warn("md_bridge.py not found at %s" % MD_BRIDGE) return False, None log("$ " + show([sys.executable, MD_BRIDGE, "--ping"])) if subprocess.call([sys.executable, MD_BRIDGE, "--ping"]) != 0: print("[garment] ERROR: " + MD_NO_SESSION, file=sys.stderr, flush=True) return False, None if not os.path.exists(DRAFT_GARMENT): warn("draft_garment.py not built yet (%s) -- MD stages unavailable" % DRAFT_GARMENT) return False, None return True, None def run_md_stage(stage, resolved_path, workdir, dry_run, md_timeout): cmds = stage_commands(stage, None, workdir, resolved_path, md_timeout) if dry_run: for c in cmds: print(" " + show(c)) return 0 ok, _ = md_preflight(resolved_path, workdir, dry_run, md_timeout) if not ok: return 3 emit = cmds[1] rc = run(emit, False) if rc != 0: # draft_garment.py may not accept --stage (the contract does not mention it); # retry the contract-exact form before giving up. log("retrying emit without --stage (contract-exact form)") rc = run(emit[:-2], False) if rc != 0: warn("draft_garment.py failed for stage %s" % stage) return rc return run(cmds[2], False) def do_run(args, resolved, chain): stages = select_range(args) qc = args.qc_record gate_stop = qc["gate_stop"] workdir, resolved_path = write_resolved(resolved, args.dry_run) log("config %s (%s)" % (resolved["name"], " <- ".join( os.path.basename(c) for c in chain))) log("stages: %s" % " ".join(stages)) for line in qc_banner(qc): log(line) if args.dry_run: log("DRY RUN -- nothing is executed") rows = [] stopped = False for stage in stages: if stage == "bake" and not resolved.get("bake"): log("stage bake SKIPPED (set \"bake\": true to enable)") rows.append(("bake", "SKIP", "-", "not requested", "-")) continue print("") log("=== stage %s ===" % stage) t0 = time.time() if stage in MD_STAGES: rc = run_md_stage(stage, resolved_path, workdir, args.dry_run, args.md_timeout) elif stage == "register": if args.dry_run: print(" (in-process codegen -> %s)" % os.path.join(workdir, "register.cs.txt")) rc = stage_register(resolved, workdir, args.dry_run) else: cmds = stage_commands(stage, resolved, workdir, resolved_path, args.md_timeout) env = stage_env(stage, resolved, workdir) rc = 0 for c in cmds: if stage in ("import", "verify") and not args.dry_run \ and not os.path.exists(c[1]): warn("%s not built yet (%s) -- skipping stage %s" % (os.path.basename(c[1]), c[1], stage)) rc = 127 break rc = run(c, args.dry_run, env) if rc != 0: break if stage == "verify": note = verify_note(verify_plan(resolved, workdir)) if note: print(note, flush=True) dt = time.time() - t0 # import/verify ARE gates G7/G6 -- agent E's scripts speak the gate exit code # contract (0/2/3) and write qc/g7.json, qc/g6.json themselves. Treat their # exit 2 as a gate failure so --no-gate-stop behaves the same everywhere, # instead of aborting the run as if the tool had crashed. egate = {"import": "g7", "verify": "g6"}.get(stage) secs = "%.1fs" % dt if args.dry_run: rows.append((stage, "DRY", "-", "-", "-")) elif rc == 0: log("stage %s OK (%.1fs)" % (stage, dt)) rows.append((stage, "OK", egate or "-", qc_path(workdir, egate, []) if egate else "-", secs)) elif rc == 127: rows.append((stage, "SKIP", "-", "tool missing", secs)) elif egate and rc in (2, 3): if rc == 3: warn("gate %s could not evaluate (exit 3, %.1fs) -- continuing" % (egate, dt)) rows.append((stage, "WARN", egate, qc_path(workdir, egate, []), secs)) else: warn("gate %s FAIL (%.1fs)" % (egate, dt)) rows.append((stage, "FAIL", egate, qc_path(workdir, egate, []), secs)) if gate_stop: stopped = True break else: warn("stage %s FAILED (exit %d, %.1fs)" % (stage, rc, dt)) rows.append((stage, "FAIL(%d)" % rc, "-", "-", secs)) summary(rows, qc) return rc if not run_gates(stage, resolved, resolved_path, workdir, args.dry_run, gate_stop, rows): stopped = True break summary(rows, qc) if stopped: warn("run stopped by a failing gate (use --no-gate-stop to continue)") return 2 failed = [r[2] for r in rows if str(r[1]).startswith("FAIL")] if failed: warn("run finished with failing gate(s): %s" % ", ".join(sorted(set(failed)))) return 2 return 0 def summary(rows, qc=None): if not rows: return print("") log("run summary") hdr = ("stage", "status", "gate", "qc / detail", "secs") n = len(hdr) w = [max(len(str(r[i])) for r in rows + [hdr]) for i in range(n)] print(" " + " ".join("%-*s" % (w[i], hdr[i]) for i in range(n))) print(" " + " ".join("-" * w[i] for i in range(n))) for r in rows: print(" " + " ".join("%-*s" % (w[i], str(r[i])) for i in range(n))) total = 0.0 for r in rows: s = str(r[4]) if s.endswith("s") and s[:-1].replace(".", "", 1).isdigit(): total += float(s[:-1]) print(" " + " ".join("-" * w[i] for i in range(n))) print(" " + " ".join(["%-*s" % (w[0], "total")] + ["%-*s" % (w[i], "") for i in (1, 2, 3)] + ["%-*s" % (w[4], "%.1fs" % total)])) if qc: print("") if qc["mode"] == "light": log(QC_LIGHT_REMINDER) else: log("QC mode: deep (%s) -- full sweep, gate failures %s" % (qc["source"], "stop the run" if qc["gate_stop"] else "report only")) def do_list(resolved, args): sel = set(select_range(args)) if resolved else set() print("[garment] stages (* = would run):") for group, names in (("MD bridge", MD_STAGES), ("Blender", BLENDER_STAGES), ("ariki-game", GAME_STAGES)): print(" %-11s %s" % (group + ":", "")) for s in names: mark = "*" if s in sel else " " note = "" if s == "bake" and resolved is not None and not resolved.get("bake"): note = " (skipped: \"bake\": true not set)" gs = [] for gate_id, script, kind, extra in (gates_for(s, resolved) if resolved else []): built = "" if os.path.exists(script) else " [not built]" gs.append("%s%s%s" % (gate_id, " --rest" if "--rest" in extra else "", built)) if resolved is not None and not gs and s in GATE_MAP: keys = [k for _, _, _, k, _ in GATE_MAP[s] if k] if keys: note += " (gate off: no expect.%s)" % "/".join(sorted(set(keys))) print(" %s %-9s %s%s" % (mark, s, ("-> gate " + ", ".join(gs)) if gs else "", note)) if resolved is not None: for line in qc_banner(args.qc_record) if getattr(args, "qc_record", None) else []: print("[garment] " + line) print("[garment] name=%s work=%s" % (resolved["name"], os.path.join(WORK, resolved["name"]))) print("[garment] blender=%s%s" % (BLENDER, "" if os.path.exists(BLENDER) else " [NOT FOUND]")) return 0 # -- selftest ----------------------------------------------------------------- def selftest(): """Assert extends resolution against clothing/configs/tests/extends_child.json.""" child = os.path.join(CONFIGS, "tests", "extends_child.json") if not os.path.exists(child): die("selftest fixture missing: %s" % child) r, chain, _ = resolve_config(child) fails = [] checks = [0] def check(label, got, want): checks[0] += 1 if got != want: fails.append("%s: got %r, want %r" % (label, got, want)) check("chain length", len(chain), 2) check("name from child", r["name"], "extends_child") check("scalar override", r["weld_threshold"], 0.0009) check("inherited scalar", r["shell_mm"], 4) check("inherited source", r["source"], "C:/parent/source.fbx") # objects merge per key check("align merged key kept", r["align"]["top_bone"], "spine_01") check("align overridden key", r["align"]["z_nudge"], 0.042) # arrays replace whole check("array replaced", r["align"]["xy_nudge"], [0.01, 0.0]) check("part array replaced", r["parts"]["Skirt"]["islands"], [2]) # nested part override: fit.offset_mm changes, fit.mask inherited check("nested part override", r["parts"]["Skirt"]["fit"]["offset_mm"], 7) check("nested part inherited", r["parts"]["Skirt"]["fit"]["mask"], "full") check("part added by child", r["parts"]["Belt"]["slot"], "Body") check("part kept from parent", r["parts"]["Skirt"]["slot"], "Legs") check("export merged", r["export"]["gender"], "Female") check("export overridden", r["export"]["set"], "TestSetChild") check("extends stripped", "extends" in r, False) # deep_merge unit checks check("dm arrays", deep_merge({"a": [1, 2]}, {"a": [3]}), {"a": [3]}) check("dm objects", deep_merge({"a": {"x": 1, "y": 2}}, {"a": {"y": 3}}), {"a": {"x": 1, "y": 3}}) check("dm type change", deep_merge({"a": {"x": 1}}, {"a": 5}), {"a": 5}) # -- path absolutization (resolved.json lives in work//, not configs/) ----- H = HERE.replace("\\", "/") raw = { "source": "../tools/tailor/x.fbx", "body": "C:/abs/body.glb", "parts": {"A": {"texture": "../tools/tailor/textures/piupiu.png"}, "B": {"texture": "C:/abs/t.png"}, "C": {"slot": "Legs"}}, "md": {"avatar_fbx": "md/a.fbx", "zfab": "md/g.zfab", "texture": "md/t.png", "snapshot": "work/x/qc/drape.png", "presim_snapshot": "work/x/qc/pre.png", "export_dir": "work/x/md"}, "export": {"out_dir": "work/x/export", "gender": "Female"}, } a = absolutize(json.loads(json.dumps(raw))) check("abs source", a["source"], os.path.normpath(os.path.join(HERE, "../tools/tailor/x.fbx")).replace("\\", "/")) check("abs body untouched", a["body"], "C:/abs/body.glb") check("abs parts.*.texture", a["parts"]["A"]["texture"], os.path.normpath(os.path.join(HERE, "../tools/tailor/textures/piupiu.png") ).replace("\\", "/")) check("abs texture already absolute", a["parts"]["B"]["texture"], "C:/abs/t.png") check("abs part without texture survives", a["parts"]["C"], {"slot": "Legs"}) for k in ("avatar_fbx", "zfab", "texture", "snapshot", "presim_snapshot", "export_dir"): check("abs md.%s" % k, a["md"][k], H + "/" + raw["md"][k]) check("abs export.out_dir", a["export"]["out_dir"], H + "/work/x/export") check("abs export.gender untouched", a["export"]["gender"], "Female") # every absolutized path must be absolute, and idempotent under a second pass check("abs idempotent", absolutize(json.loads(json.dumps(a))), a) check("abs all absolute", all(os.path.isabs(p) for p in ( a["source"], a["body"], a["parts"]["A"]["texture"], a["md"]["snapshot"], a["export"]["out_dir"])), True) # empties/None are left alone rather than becoming the clothing/ dir itself check("abs empty string", abs_path(""), "") check("abs None", abs_path(None), None) # and the real resolver applies it check("resolver absolutizes", os.path.isabs(r["source"]), True) # -- verify plan (agent E's clothing_motion_qa.sh CLI) -------------------------- vr = {"name": "vt", "export": {"set": "KapahakaSBTest", "gender": "Female"}, "verify": {"baseline_dir": os.path.join(HERE, "baselines", "__nope__")}} plan = verify_plan(vr, os.path.join(WORK, "vt")) check("verify BED_GENDER female", plan["env"]["BED_GENDER"], "1") check("verify BED_SET", plan["env"]["BED_SET"], "KapahakaSBTest") check("verify no baseline -> no diff/bless", plan["mode"], "first") check("verify has no --diff", "--diff" in plan["argv"], False) check("verify has no --bless", "--bless" in plan["argv"], False) check("verify clips default", plan["argv"][plan["argv"].index("--clips") + 1], "Idle Walk Dance") check("verify frames default", plan["argv"][plan["argv"].index("--frames") + 1], "3") check("verify report last-but-one", plan["argv"][-2].replace("\\", "/"), (os.path.join(WORK, "vt", "qc", "g6.json")).replace("\\", "/")) check("verify out-dir is positional last", plan["argv"][-1], plan["out_dir"]) check("verify note present", "GARMENT_BLESS=1" in (verify_note(plan) or ""), True) vr_m = dict(vr, export={"set": "S", "gender": "Male"}) check("verify BED_GENDER male", verify_plan(vr_m, "w")["env"]["BED_GENDER"], "0") # an existing baseline dir flips it to --diff vr_d = dict(vr, verify={"baseline_dir": HERE}) check("verify existing baseline -> --diff", verify_plan(vr_d, "w")["mode"], "diff") check("verify --diff arg", verify_plan(vr_d, "w")["argv"][ verify_plan(vr_d, "w")["argv"].index("--diff") + 1], os.path.abspath(HERE)) check("verify note silent once blessed", verify_note(verify_plan(vr_d, "w")), None) # -- QC modes (--qc light|deep) ------------------------------------------------- def qcfg(**kw): """A resolved-shaped config with real thresholds, for profile-merge checks.""" base = { "name": "qct", "expect": { "penetration": {"depth_mm": 1.0, "max_verts": 0, "max_verts_rest": 3, "clips": ["Idle", "Walk"], "frames_per_clip": 6, "gape": {"Top": {"band_z": [1.0, 1.3], "max_exposed_verts": 4}}}, "islands": {"count": 1}, }, } base.update(kw) return json.loads(json.dumps(base)) # default mode is light, and light is the fast profile c = qcfg() rec = apply_qc_mode(c, None, None, quiet=True) check("qc default mode", rec["mode"], "light") check("qc default source", rec["source"], "default") check("qc light clips", c["expect"]["penetration"]["clips"], ["Walk"]) check("qc light frames_per_clip", c["expect"]["penetration"]["frames_per_clip"], 2) check("qc light max_render_frames", c["expect"]["penetration"]["max_render_frames"], 2) check("qc light one pack", len(c["expect"]["penetration"]["anim_pack"]), 1) check("qc light pack is UAL2", os.path.basename(c["expect"]["penetration"]["anim_pack"][0]), "UAL2.glb") check("qc light verify clips", c["verify"]["clips"], "Walk") check("qc light verify frames", c["verify"]["frames"], 1) # light implies no-gate-stop; deep stops check("qc light gate_stop", rec["gate_stop"], False) check("qc light gate_stop from profile", rec["gate_stop_source"], "profile") # user thresholds SURVIVE the profile -- a mode changes how hard we look, never # what counts as a defect pen = c["expect"]["penetration"] check("qc keeps depth_mm", pen["depth_mm"], 1.0) check("qc keeps max_verts", pen["max_verts"], 0) check("qc keeps max_verts_rest", pen["max_verts_rest"], 3) check("qc keeps gape budget", pen["gape"]["Top"]["max_exposed_verts"], 4) check("qc keeps other expect blocks", c["expect"]["islands"], {"count": 1}) # deep profile c = qcfg() rec = apply_qc_mode(c, "deep", None, quiet=True) check("qc deep mode", rec["mode"], "deep") check("qc deep clips", c["expect"]["penetration"]["clips"], ["Idle", "Walk", "Dance"]) check("qc deep frames_per_clip", c["expect"]["penetration"]["frames_per_clip"], 8) check("qc deep both packs", [os.path.basename(p) for p in c["expect"]["penetration"]["anim_pack"]], ["UAL1.glb", "UAL2.glb"]) check("qc deep synthetic pass", c["expect"]["penetration"]["pose_source"], "both") check("qc deep verify clips", c["verify"]["clips"], "Idle Walk Dance") check("qc deep verify frames", c["verify"]["frames"], 3) check("qc deep gate_stop", rec["gate_stop"], True) check("qc deep keeps thresholds", c["expect"]["penetration"]["max_verts_rest"], 3) # config qc.default_mode is honoured, and --qc beats it c = qcfg(qc={"default_mode": "deep"}) rec = apply_qc_mode(c, None, None, quiet=True) check("qc config default_mode", rec["mode"], "deep") check("qc config default_mode source", rec["source"], "config qc.default_mode") check("qc config default_mode applies", c["expect"]["penetration"]["frames_per_clip"], 8) c = qcfg(qc={"default_mode": "deep"}) rec = apply_qc_mode(c, "light", None, quiet=True) check("qc CLI beats config default", rec["mode"], "light") check("qc CLI beats config source", rec["source"], "--qc") check("qc CLI beats config applies", c["expect"]["penetration"]["frames_per_clip"], 2) # per-garment profile override: config qc. merges ON TOP of the builtin c = qcfg(qc={"light": {"expect": {"penetration": {"frames_per_clip": 4}}}}) rec = apply_qc_mode(c, "light", None, quiet=True) check("qc config block overrides builtin", c["expect"]["penetration"]["frames_per_clip"], 4) check("qc config block keeps rest of builtin", c["expect"]["penetration"]["clips"], ["Walk"]) check("qc config block recorded", rec["applied"]["expect.penetration.frames_per_clip"], 4) # ... and it can flip gate_stop for this garment c = qcfg(qc={"light": {"gate_stop": True}}) check("qc config block gate_stop", apply_qc_mode(c, "light", None, quiet=True)["gate_stop"], True) # a config's qc block may NOT touch a threshold: dropped, with a note c = qcfg(qc={"light": {"expect": {"penetration": {"max_verts": 999, "depth_mm": 9.0}, "islands": {"count": 99}}, "verify": {"out_dir": "/nope"}, "bake": True}}) rec = apply_qc_mode(c, "light", None, quiet=True) check("qc profile cannot raise max_verts", c["expect"]["penetration"]["max_verts"], 0) check("qc profile cannot change depth_mm", c["expect"]["penetration"]["depth_mm"], 1.0) check("qc profile cannot touch islands", c["expect"]["islands"], {"count": 1}) check("qc profile cannot set verify.out_dir", "out_dir" in c.get("verify", {}), False) check("qc profile cannot set bake", c.get("bake"), None) check("qc drops are reported", sorted(rec["notes"]), sorted([ "qc.light.expect.penetration.max_verts", "qc.light.expect.penetration.depth_mm", "qc.light.expect.islands", "qc.light.verify.out_dir", "qc.light.bake"])) # CLI --gate-stop / --no-gate-stop beat the profile check("qc --gate-stop in light", apply_qc_mode(qcfg(), "light", True, quiet=True)["gate_stop"], True) check("qc --gate-stop source", apply_qc_mode(qcfg(), "light", True, quiet=True)["gate_stop_source"], "CLI") check("qc --no-gate-stop in deep", apply_qc_mode(qcfg(), "deep", False, quiet=True)["gate_stop"], False) # `qc.pin` keeps the config's own value for a cost knob c = qcfg(qc={"pin": ["clips", "verify.frames"]}) c["verify"] = {"frames": 5} rec = apply_qc_mode(c, "light", None, quiet=True) check("qc pin keeps config clips", c["expect"]["penetration"]["clips"], ["Idle", "Walk"]) check("qc pin keeps config verify.frames", c["verify"]["frames"], 5) check("qc pin does not freeze the rest", c["expect"]["penetration"]["frames_per_clip"], 2) check("qc pinned recorded", rec["pinned"], ["expect.penetration.clips", "verify.frames"]) # a config with no expect.penetration must NOT have the gate conjured for it c = {"name": "n", "expect": {"islands": {"count": 1}}} rec = apply_qc_mode(c, "deep", None, quiet=True) check("qc does not arm g5", "penetration" in c["expect"], False) check("qc notes the unarmed gate", any("not armed" in n for n in rec["notes"]), True) # the mode record is what lands in resolved.json c = qcfg() rec = apply_qc_mode(c, "deep", None, quiet=True) r2, _, _ = resolve_config(child) r2["_resolved"]["qc_mode"] = rec check("qc_mode recorded in _resolved", r2["_resolved"]["qc_mode"]["mode"], "deep") check("qc_mode json round-trips", json.loads(json.dumps(r2["_resolved"]["qc_mode"]))["applied"][ "expect.penetration.frames_per_clip"], 8) check("qc banner names the mode", "QC mode: deep" in qc_banner(rec)[0], True) check("qc light banner says report-only", "REPORT only" in qc_banner(apply_qc_mode(qcfg(), "light", None, quiet=True))[0], True) check("qc light reminder text", "run --qc deep before sign-off" in QC_LIGHT_REMINDER, True) # every knob a builtin profile sets must be one a profile is ALLOWED to set for m in QC_MODES: prof = QC_PROFILES[m] check("qc %s only cost knobs (penetration)" % m, sorted(prof["expect"]["penetration"]), sorted(QC_ALLOWED_PENETRATION)) check("qc %s only cost knobs (verify)" % m, sorted(prof["verify"]), sorted(QC_ALLOWED_VERIFY)) check("qc %s top-level keys" % m, sorted(k for k in prof if k not in ("expect", "verify")), sorted(QC_ALLOWED_TOP)) # light must be strictly cheaper than deep on every sampling knob lp = QC_PROFILES["light"]["expect"]["penetration"] dp = QC_PROFILES["deep"]["expect"]["penetration"] check("light fewer clips", len(lp["clips"]) < len(dp["clips"]), True) check("light fewer frames", lp["frames_per_clip"] < dp["frames_per_clip"], True) check("light fewer packs", len(lp["anim_pack"]) < len(dp["anim_pack"]), True) check("light fewer renders", lp["max_render_frames"] < dp["max_render_frames"], True) check("light pack carries Walk", "UAL2" in lp["anim_pack"][0], True) for f in fails: print("[garment] SELFTEST FAIL: " + f, file=sys.stderr) if fails: print("[garment] selftest: %d check(s) failed" % len(fails)) return 1 log("selftest: all %d checks passed (extends chain, deep-merge, array replacement, " "path absolutization, verify/G6 plan, QC modes)" % checks[0]) return 0 # -- main --------------------------------------------------------------------- def main(argv=None): ap = argparse.ArgumentParser( prog="garment.py", description="Unified clothing pipeline: MD -> Blender -> ariki-game.", formatter_class=argparse.RawDescriptionHelpFormatter, epilog="stages: " + " ".join(STAGES) + "\ndefault range: %s..%s" % (DEFAULT_FROM, DEFAULT_TO)) ap.add_argument("config", nargs="?", help="path to a garment config JSON") ap.add_argument("--from", dest="from_stage", metavar="STAGE") ap.add_argument("--to", dest="to_stage", metavar="STAGE") ap.add_argument("--only", metavar="STAGE[,STAGE...]") ap.add_argument("--list", action="store_true", help="print stages + gates for this config and exit") ap.add_argument("--dry-run", action="store_true", help="print the exact commands without executing") ap.add_argument("--qc", metavar="MODE", choices=list(QC_MODES), help="QC mode: light (fast iterate loop; default) or deep " "(sign-off sweep). Overrides the config's qc.default_mode.") ap.add_argument("--no-gate-stop", action="store_true", help="continue past a failing gate (the default in --qc light)") ap.add_argument("--gate-stop", action="store_true", help="stop on a failing gate (the default in --qc deep); use it " "to get deep's stop-on-fail behaviour inside light") ap.add_argument("--blender", metavar="PATH", help="override the Blender executable") ap.add_argument("--bless-baseline", action="store_true", help="verify stage: pass --bless to clothing_motion_qa.sh, freezing " "the captured frames as the G6 baseline (same as GARMENT_BLESS=1). " "LOOK at the frames first -- a blessed defect becomes the standard.") ap.add_argument("--md-timeout", type=int, default=900, help="md_bridge socket timeout, seconds (default: %(default)s)") ap.add_argument("--selftest", action="store_true", help="assert extends/deep-merge against configs/tests fixtures") args = ap.parse_args(argv) global BLENDER, BLESS_BASELINE if args.blender: BLENDER = args.blender BLESS_BASELINE = bool(args.bless_baseline) if args.selftest: return selftest() if args.only and (args.from_stage or args.to_stage): die("--only cannot be combined with --from/--to") if args.gate_stop and args.no_gate_stop: die("--gate-stop and --no-gate-stop are opposites -- pick one") # None = "the QC mode decides"; True/False = an explicit CLI override. args.cli_gate_stop = True if args.gate_stop else (False if args.no_gate_stop else None) if not args.config: if args.list: return do_list(None, args) ap.error("a config path is required (or use --selftest)") path = find_config(args.config) resolved, chain, _ = resolve_config(path) # QC mode is applied at RESOLVE time, so resolved.json -- the one file every stage # and gate is handed -- already carries the mode's numbers. Gates stay mode-unaware. args.qc_record = apply_qc_mode(resolved, args.qc, args.cli_gate_stop) resolved["_resolved"]["qc_mode"] = args.qc_record if args.list: return do_list(resolved, args) if not os.path.exists(BLENDER) and not args.dry_run: warn("Blender not found at %s (set $BLENDER or --blender)" % BLENDER) return do_run(args, resolved, chain) if __name__ == "__main__": sys.exit(main())