tinqs/animation
5
0
Fork 0
Code Pull Requests Actions Packages Projects Releases Wiki Activity Issues
Files
main
animation/clothing/garment.py
T
jeremy 3ba86b2ea8 feat: clothing lane, character sources, and DCC bridges 
Bulk import of the working lanes that were living untracked on the PC.

Content:
- characters/  Lena/male body lanes, bakes, texture work, run logs
- clothing/    garment pipeline, configs, gates, contract docs
- garments/    MD-authored garment sources (.zprj/.zpac)
- UAL-Lib/     Universal Animation Library 2 source (.blend/.fbx/.glb)
- tools/       blender_bridge, iclone_bridge, md_bridge, tailor, glm_agent
- docs/, plans/, dev/, .agents/plans/

Repo hygiene:
- .gitattributes: LFS now covers .blend, .zprj, .zpac, .obj, .npy and the
  Reallusion .iAvatar/.ccAvatar/.ccRestore containers. Without this the
  ~3.8 GB in this commit would land as raw blobs. .png/.jpg are left out
  on purpose — ~250 are already tracked raw and converting them would
  rewrite every one without shrinking history.
- .gitignore: exclude /accurig/ (~1 GB AccuRig program files, redistributable
  from Reallusion, nothing authored here) and /dev/null/ (git-lfs hook copies
  dropped by a `>/dev/null` redirect on Windows).

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
2026-08-06 15:55:43 -07:00

1536 lines
66 KiB
Python
Raw Permalink Blame History

#!/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/<name>/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/<name>/ 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/<name>/ --
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 <gate> -- ...
# "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.<mode>` 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/<name>/, 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/<name>/): %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.<mode>` 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.<mode>` 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.<mode>` 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=<Set> BED_GENDER=<0|1> bash <game>/tools/clothing_motion_qa.sh \\
--clips "Idle Walk Dance" --frames 3 [--bless DIR | --diff DIR] \\
--report <work>/qc/g6.json <out-dir>
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": {"<PartName>": {...}, "<PartName>": {...}} 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 "
"<id>_<slot>; 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/<folder>/ (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/<name>/qc/<gate_id>.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/<name>/, 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.<mode> 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())
Reference in New Issue View Git Blame Copy Permalink