diff --git a/CHANGELOG.md b/CHANGELOG.md index 85bb765a..5f3eff3e 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,16 @@ All notable public changes to `sprite-gen` are recorded here. Versions track the `version:` field in `SKILL.md` and `pyproject.toml`. +## Unreleased + +- Added catalog-driven `parts-gen`, `parts-match` and `parts-rig` commands for + illustrated character layers, using parallel generation, measured chroma alpha, + FFT registration and explicit part/composite acceptance gates. +- Exported JSON rigs, HTML layers and deterministic audio RMS lip-sync, blink and + head sway keys for HyperFrames. Interleaved groups preserve global draw order. +- Added synthetic regression coverage for registration failures, generation report + merging, missing layers, long-audio frame timing and deterministic rig export. + ## v1.60.0 - Native Alpha - `sprite-gen gen --transparent` now follows a per-provider transparency strategy declared once on each adapter (`Provider.transparency`). `codex` asks `image_gen` for a genuinely transparent background and publishes the measured alpha (`native`, first choice); `grok` keeps deterministic chroma keying because Grok Imagine returns JPEG only. diff --git a/README.md b/README.md index 1bb7a3e9..50eb55ba 100644 --- a/README.md +++ b/README.md @@ -18,6 +18,23 @@ --- +## Character parts rig + +Generate separate character parts from a base illustration, register them back onto +the base, and export a JSON rig with HTML layers and deterministic GSAP lip-sync, +blinks and head motion. The catalog declares geometry and acceptance thresholds; +failed registration is reported before rig export. + +```bash +sprite-gen parts-gen --catalog catalog.json --out-dir parts +sprite-gen parts-match --catalog catalog.json --parts-dir parts +sprite-gen parts-rig --catalog catalog.json --match-dir parts --audio narration.wav +``` + +See [the catalog and runtime contract](docs/parts-rig.md) for setup, variants and +HyperFrames integration. Inspect the resulting composite: generated parts may +change facial details even when the declared numeric thresholds pass. + ## Breathe A still idle reads as frozen. **Breathe** turns a single pose into a living loop — deterministic squash & stretch baked on top of your curated frames. No regeneration, no re-extraction, no extra art. One sidecar field: diff --git a/SKILL.md b/SKILL.md index 21947f83..e56d2294 100644 --- a/SKILL.md +++ b/SKILL.md @@ -164,6 +164,25 @@ $SPRITE_GEN_ROOT/.venv/bin/python ... 않는다** — 위 venv 절대경로 형식 하나만 쓴다. (`tests/test_entrypoint_interpreter.py` 가 이 두 파일군에 대해 잠근다.) +## Parts rig (illustrated character layers) + +For a generated character split into independently movable body parts, use the +catalog-based parts workflow in [`docs/parts-rig.md`](docs/parts-rig.md). This is +separate from the component-row pixel-art atlas pipeline and its pixel-unfake gates. + +1. Declare the base image, canvas, part boxes, pivots, draw order, groups and variants. +2. `$SPRITE_GEN_ROOT/.venv/bin/python -m sprite_gen.cli parts-gen --catalog catalog.json --out-dir parts` + generates each part from full and cropped references, then keys to measured RGBA. +3. `$SPRITE_GEN_ROOT/.venv/bin/python -m sprite_gen.cli parts-match --catalog catalog.json --parts-dir parts` + registers candidates with FFT template search and gates part agreement and composite coverage. +4. `$SPRITE_GEN_ROOT/.venv/bin/python -m sprite_gen.cli parts-rig --catalog catalog.json --match-dir parts --audio narration.wav` + exports JSON, layer HTML and deterministic lip-sync, blink and sway keys. + +Review both the composite and animation. A declared numeric gate passing does not +prove identity or expression fidelity. Report any relaxed thresholds explicitly. +Keep real character images and catalogs in the character's repository; public +fixtures must be synthetic. + ## Script Map Scripts are explicit pipeline commands, not hidden imports. One job each (stage detail: [`docs/architecture.md`](docs/architecture.md) §2): diff --git a/docs/parts-rig.md b/docs/parts-rig.md new file mode 100644 index 00000000..44c92e99 --- /dev/null +++ b/docs/parts-rig.md @@ -0,0 +1,146 @@ +# Parts Rig — catalog · gen · match · rig contract (SSoT) + +> Status: **contract** (normative). This doc owns the parts-rig feature: how one +> character image is split into generated body-part layers, registered back onto +> the original by pixels, and exported as a JSON rig with a seek-safe HTML runtime. +> Code: `sprite_gen/parts/` (`catalog.py`, `parts_gen.py`, `match.py`, `rig.py`), +> pinned by `tests/parts/test_parts_contract.py` on synthetic shapes only. +> It is independent of the component-row atlas pipeline and of +> [`layer-tracks.md`](layer-tracks.md) (a bake-time row compositor); it shares only +> the landmark discipline — integer coordinates, declared or rejected, never inferred. + +## 0. One sentence + +A **catalog** names every part of one base image (draw order, box, pivot, group, +prompt); **gen** draws each part alone from the base + a crop of its box; **match** +puts each part back where the pixels say it belongs and refuses anything that does +not reproduce the base; **rig** turns the matched layers into `rig.json`, an HTML +fragment, and deterministic GSAP keys (lip-sync, blink, sway). + +## 1. Why generate parts instead of cutting them + +Cutting a flat illustration leaves holes: the face behind the bangs, the eyeball +under the eyelid, the neck under the choker do not exist in the pixels. Generating +each part alone from the same reference fills those occluded areas with the +character's own art, so a layer can move without exposing a gap. The price is that +a generated part can drift (shape, colour, scale) — which is exactly what the +`match` gate measures and refuses. + +## 2. Catalog (`sprite-gen-parts-catalog`, version 1) + +```jsonc +{ + "kind": "sprite-gen-parts-catalog", "version": 1, + "character": "example", + "base": "base.png", // relative to the catalog file; RGBA + "canvas": {"width": 1024, "height": 1536}, // must equal the base image size + "chroma_key": "green", // green | magenta — the part-generation key + "groups": {"head": {"pivot": [512, 700]}}, // rotation/translation origins for the runtime + "parts": [ + {"id": "hair_back", "z": 0, "bbox": [180, 40, 700, 1300], "pivot": [512, 400], "group": "head", + "prompt": "the hair behind the head and shoulders"}, + {"id": "mouth", "z": 12, "bbox": [440, 720, 150, 70], "pivot": [515, 755], "group": "head", + "prompt": "the mouth and lips only", + "variants": {"default": "", "closed": "lips together", "half": "slightly open", "open": "open, teeth visible", "o": "rounded o shape"}, + "tolerance": 0.08} + ] +} +``` + +Rules (validated by `catalog.validate_catalog`, every violation listed, in order): + +- `id` matches `^[a-z][a-z0-9_]{0,31}$` and is unique; `z` is a unique integer (draw order, bottom first). +- `bbox` is `[x, y, w, h]` integers inside the canvas; `pivot` is `[x, y]` integers inside the bbox. +- `group` names a declared group or is omitted (`none`). Groups carry a `pivot`. +- `variants` is an object `variant -> prompt suffix` and must contain `default` (its suffix may be empty). + The runtime recognises `mouth` variants `closed | half | open | o` and eyelid variants `open | half | closed` by name; anything else is a plain swap. +- `tolerance` (0..1 exclusive, default 0.06) is the part's colour gate in `match`; `agree_floor` (0..1], default 0.85) its agreement gate. +- Top-level `composite: {tolerance, coverage}` declares the whole-composite gate (defaults 0.05 / 0.97). +- Coordinates are integers, never floats or booleans. Nothing is inferred from pixels. + +Adding a part or a variant is one catalog entry; `gen`, `match` and `rig` enumerate it. + +## 3. `sprite-gen parts-gen` + +`--catalog --out-dir [--provider codex|grok] [--workers 6] [--only a,b__open]` + +Per job (part × variant), in parallel: the base is flattened onto the chroma colour +and sent as reference 1; a crop of the bbox padded by 25% is reference 2; the prompt +asks for that part only, pixel-faithful, on a flat chroma key; the result is keyed to +RGBA by `gen.generate_image(transparent=True)`. A result with no transparent pixels +is recorded as a failure. Writes `.png` (+ `.raw.png`) and `parts-gen.report.json` +(`ok`, `failed[]`, one record per job with alpha stats). Partial `--only` runs +merge existing job records; `ran` identifies jobs attempted this time and a retained +failure keeps `ok: false`. Exit 1 if any recorded job failed. + +## 4. `sprite-gen parts-match` — the gate + +`--catalog --parts-dir [--out-dir] [--composite-tolerance ]` + +Top-most part first: the candidate is trimmed to its alpha box and **contain-fitted** +(its own aspect ratio, never stretched) into `bbox × scale` for scales `0.50 … 1.10` +in steps of 0.02, inside a window around the bbox (grown by ±8% and by the largest +scale). Pixels already claimed by a higher part are excluded (`free` mask), so a face +is compared only where the bangs do not cover it. + +Two stages per part: + +1. **FFT shortlist** (`register_fft.cost_map`) — for every scale, one masked-SSD cost map + over every offset via FFT correlations (`SSD/(3·255²) − reward·W + miss`), plus the + pure colour-error argmin. A few FFTs per scale regardless of offset count. +2. **Exact rescoring** — each shortlisted placement is scored with the agreement objective + `−(agreeing − 2·disagreeing − 1.5·missed) / bbox area`, where a visible part pixel + *agrees* when its mean RGB distance to the base is ≤ 0.12 and *missed* counts owned + pixels left uncovered. An **owned** pixel is a base pixel inside the bbox, unclaimed, + whose quantized colour (5 bits/channel) the part itself contains — a hair part owes + red pixels, a shirt part does not. Shrinking loses agreement and gains misses; + oversizing pays for every pixel spilled onto something else. + +Reported per part: `score` (alpha-weighted mean RGB distance, gated by the part's +`tolerance`), `agree` (fraction agreeing, gated by the part's `agree_floor`, default +0.85), `scale`, `x`, `y`, `w`, `h`. Variants inherit their default's placement. + +The defaults are stacked bottom-first into `composite.png`; `composite.score` (mean RGB +distance where both are opaque) must be ≤ `catalog.composite.tolerance` (default 0.05) +and `coverage` (base alpha reproduced) ≥ `catalog.composite.coverage` (default 0.97). +Real characters declare looser values than synthetic shapes (a generated hair mass +never reproduces strands pixel-for-pixel); the declaration is the gate, and loosening +it is a visible catalog edit, never a runtime fallback. Any part below its gate, any +missing candidate, or a failing composite makes `ok: false` and is named in `failed[]`. +Outputs `placed/.png` (canvas-sized layers) and `parts-match.report.json`. +Measured 2026-09-08: a 14-part 1024×1536 character registers in ~25 s. + +## 5. `sprite-gen parts-rig` + +`--catalog --match-dir [--out-dir] [--audio narration.mp3 | --duration s] [--fps 30] [--start 0] [--prefix rig] [--asset-prefix]` + +Refuses a match report that is not `ok` or a missing placed layer. Writes: + +- `rig.json` — canvas, groups (pivot + members), parts in z order with placement and + `variants: {name: "placed/.png"}`. +- `rig.html` — `
` with one absolutely positioned canvas-sized `` + per variant (`id="-[__]"`, non-default `opacity:0`), grouped + under wrappers with `data-rig-group=""` and `transform-origin` at the + group pivot. A group interrupted by another layer is split into consecutive z + runs, each with a unique id and explicit z-index. All runs rotate together, + preserving the composite draw order even across transformed stacking contexts. +- `rig-keys.json` / `rig-keys.js` — `window.__rigKeys(tl, start)` stamps GSAP `set` + calls on a paused timeline: mouth variant per frame from the audio's RMS envelope + (ffmpeg → 16 kHz mono → per-frame RMS using exact sample boundaries without cumulative rounding drift, normalized to the clip peak, thresholds + 0.12 / 0.38 / 0.70 → closed / half / open / o), fixed-cadence blinks (3.4 s + phase + seeded from the clip length), closed blinks also hide the corresponding `eye_l` / `eye_r` default layer, + and a slow sine sway on the `head` group (±1.2°). + Same inputs → byte-identical outputs (pinned by test). + +HyperFrames usage: paste `rig.html` inside a scene, load `rig-keys.js`, and call +`window.__rigKeys(tl, sceneStart)` before registering the timeline. Every key is a +`set`, so the runtime stays seek-safe. + +## 6. What this feature does not do + +- No Live2D / Spine runtime. `rig.json` is the source a later exporter could read. +- No mesh warping: parts translate/rotate as rigid layers; expression comes from variant swaps. +- No inference of boxes or pivots from pixels: the catalog is authored (a grid overlay + of the base is the practical way to read coordinates). +- Public-repo hygiene: real character catalogs, bases and parts live with the character + (never in this repository); fixtures here are synthetic shapes. diff --git a/sprite_gen/_modules.py b/sprite_gen/_modules.py index 5943786f..2019c141 100644 --- a/sprite_gen/_modules.py +++ b/sprite_gen/_modules.py @@ -39,6 +39,10 @@ 'serve_curation': 'serve', 'serve_compose': 'serve', 'gif_utils': 'util', + 'catalog': 'parts', + 'parts_gen': 'parts', + 'match': 'parts', + 'rig': 'parts', } diff --git a/sprite_gen/cli.py b/sprite_gen/cli.py index d0e2d228..11247cc1 100644 --- a/sprite_gen/cli.py +++ b/sprite_gen/cli.py @@ -14,6 +14,7 @@ from sprite_gen.frames import cutout, extract, slice_sheet, unpack_atlas from sprite_gen.gen import prepare from sprite_gen.effects import recolor +from sprite_gen.parts import match as parts_match, parts_gen, rig as parts_rig from sprite_gen.serve import serve_compose, serve_curation from sprite_gen.spec import migrate_breathe, migrate_request from sprite_gen.gen.prepare import STYLE_DEFAULT, _outline_config @@ -241,6 +242,24 @@ def _add_correction_loop(p: argparse.ArgumentParser) -> None: compose_layers.add_arguments, compose_layers.run, ), + # Parts rig: generate body parts alone, register them onto the base by pixels, + # export rig.json + HTML runtime. Same rule as compose-layers: the subcommand + # reuses the module's own argument declaration. + "parts-gen": ( + "Generate every parts-catalog part alone (base + crop as refs), in parallel.", + parts_gen.add_arguments, + parts_gen.run, + ), + "parts-match": ( + "Register generated parts onto the base by pixel agreement; gate on tolerance.", + parts_match.add_arguments, + parts_match.run, + ), + "parts-rig": ( + "Build rig.json + HTML runtime + deterministic GSAP keys from matched parts.", + parts_rig.add_arguments, + parts_rig.run, + ), "unpack-atlas": ( "Unpack a composed sprite sheet back into a curator-ready run directory.", _add_unpack_atlas, diff --git a/sprite_gen/parts/__init__.py b/sprite_gen/parts/__init__.py new file mode 100644 index 00000000..342fd61b --- /dev/null +++ b/sprite_gen/parts/__init__.py @@ -0,0 +1,9 @@ +# SPDX-License-Identifier: Apache-2.0 +"""Parts rig — split one character image into generated body-part layers, +register them back onto the original by pixel matching, and export a JSON rig +plus a seek-safe HTML runtime (lip-sync, blink, head sway). + +Contract: `docs/parts-rig.md`. Stages: `catalog` (declaration) → `gen` (per-part +generation from the base + crop) → `match` (pixel registration gate) → `rig` +(rig.json + runtime export). +""" diff --git a/sprite_gen/parts/catalog.py b/sprite_gen/parts/catalog.py new file mode 100644 index 00000000..8a32d82d --- /dev/null +++ b/sprite_gen/parts/catalog.py @@ -0,0 +1,180 @@ +# SPDX-License-Identifier: Apache-2.0 +"""Parts catalog — the declaration half of the parts rig. + +A catalog names every body part of one base image: its draw order, its box on +the base, its pivot, its group, and the prompt that generates it alone. The +catalog is the numeric SSoT that `gen`, `match` and `rig` all read; nothing +downstream infers a part from pixels. This module is filesystem-free and +deterministic: the same catalog always yields the same error list in the same +order, so the CLI can refuse a bad catalog before any generation is paid for. + +Behavior contract: `docs/parts-rig.md` §2. +""" + +from __future__ import annotations + +import json +import re +from pathlib import Path +from typing import Any + +KIND = "sprite-gen-parts-catalog" +VERSION = 1 +PART_ID = re.compile(r"^[a-z][a-z0-9_]{0,31}$") +DEFAULT_VARIANT = "default" +DEFAULT_TOLERANCE = 0.06 +DEFAULT_AGREE_FLOOR = 0.85 +DEFAULT_GROUP = "none" + +# Known variant vocabularies are documentation, not a restriction: any id that +# matches PART_ID is accepted. The runtime looks these names up by convention. +MOUTH_VARIANTS = ("closed", "half", "open", "o") +EYELID_VARIANTS = ("open", "half", "closed") + + +def _is_int(value: Any) -> bool: + return isinstance(value, int) and not isinstance(value, bool) + + +def _bbox_ok(bbox: Any) -> bool: + return (isinstance(bbox, list) and len(bbox) == 4 and all(_is_int(v) for v in bbox) + and bbox[2] > 0 and bbox[3] > 0 and bbox[0] >= 0 and bbox[1] >= 0) + + +def _point_ok(point: Any) -> bool: + return isinstance(point, list) and len(point) == 2 and all(_is_int(v) for v in point) + + +def validate_catalog(catalog: Any) -> list[str]: + """Return every violation, in declaration order. Empty list means valid.""" + errors: list[str] = [] + if not isinstance(catalog, dict): + return ["catalog must be a JSON object"] + if catalog.get("kind") != KIND: + errors.append(f"kind must be {KIND!r}") + if catalog.get("version") != VERSION: + errors.append(f"version must be {VERSION}") + canvas = catalog.get("canvas") + cw = ch = None + if not (isinstance(canvas, dict) and _is_int(canvas.get("width")) and _is_int(canvas.get("height")) + and canvas["width"] > 0 and canvas["height"] > 0): + errors.append("canvas.width / canvas.height must be positive integers") + else: + cw, ch = canvas["width"], canvas["height"] + base = catalog.get("base") + if not (isinstance(base, str) and base): + errors.append("base must be a non-empty path string (relative to the catalog)") + chroma = catalog.get("chroma_key") + if chroma is not None and not (isinstance(chroma, str) and chroma in ("green", "magenta")): + errors.append("chroma_key must be 'green' or 'magenta' when declared") + composite = catalog.get("composite", {}) + if not isinstance(composite, dict): + errors.append("composite must be an object {tolerance, coverage}") + else: + for key, lo, hi in (("tolerance", 0.0, 1.0), ("coverage", 0.0, 1.0)): + value = composite.get(key) + if value is not None and (isinstance(value, bool) or not isinstance(value, (int, float)) or not (lo < value <= hi)): + errors.append(f"composite.{key}: must be a number in ({lo}, {hi}]") + groups = catalog.get("groups", {}) + if not isinstance(groups, dict): + errors.append("groups must be an object") + groups = {} + for name, group in groups.items(): + if not PART_ID.match(str(name)): + errors.append(f"groups.{name}: invalid group id") + if not (isinstance(group, dict) and _point_ok(group.get("pivot"))): + errors.append(f"groups.{name}.pivot must be [x, y] integers") + parts = catalog.get("parts") + if not isinstance(parts, list) or not parts: + errors.append("parts must be a non-empty list") + return errors + seen_ids: set[str] = set() + seen_z: dict[int, str] = {} + for index, part in enumerate(parts): + where = f"parts[{index}]" + if not isinstance(part, dict): + errors.append(f"{where}: must be an object") + continue + pid = part.get("id") + if not (isinstance(pid, str) and PART_ID.match(pid)): + errors.append(f"{where}.id: must match {PART_ID.pattern}") + elif pid in seen_ids: + errors.append(f"{where}.id: duplicate part id {pid!r}") + else: + seen_ids.add(pid) + where = f"parts.{pid}" if isinstance(pid, str) else where + z = part.get("z") + if not _is_int(z): + errors.append(f"{where}.z: must be an integer") + elif z in seen_z: + errors.append(f"{where}.z: duplicate draw order {z} (also {seen_z[z]!r})") + else: + seen_z[z] = pid + bbox = part.get("bbox") + if not _bbox_ok(bbox): + errors.append(f"{where}.bbox: must be [x, y, w, h] integers with w,h > 0") + elif cw is not None and (bbox[0] + bbox[2] > cw or bbox[1] + bbox[3] > ch): + errors.append(f"{where}.bbox: exceeds canvas {cw}x{ch}") + pivot = part.get("pivot") + if not _point_ok(pivot): + errors.append(f"{where}.pivot: must be [x, y] integers") + elif _bbox_ok(bbox) and not (bbox[0] <= pivot[0] <= bbox[0] + bbox[2] + and bbox[1] <= pivot[1] <= bbox[1] + bbox[3]): + errors.append(f"{where}.pivot: must lie inside bbox") + group = part.get("group", DEFAULT_GROUP) + if group != DEFAULT_GROUP and group not in groups: + errors.append(f"{where}.group: unknown group {group!r}") + prompt = part.get("prompt") + if not (isinstance(prompt, str) and prompt.strip()): + errors.append(f"{where}.prompt: must be a non-empty string") + variants = part.get("variants", {DEFAULT_VARIANT: ""}) + if not isinstance(variants, dict) or not variants: + errors.append(f"{where}.variants: must be a non-empty object of variant -> prompt suffix") + else: + if DEFAULT_VARIANT not in variants: + errors.append(f"{where}.variants: must include {DEFAULT_VARIANT!r}") + for vname, suffix in variants.items(): + if not PART_ID.match(str(vname)): + errors.append(f"{where}.variants.{vname}: invalid variant id") + if not isinstance(suffix, str): + errors.append(f"{where}.variants.{vname}: prompt suffix must be a string") + tolerance = part.get("tolerance", DEFAULT_TOLERANCE) + if isinstance(tolerance, bool) or not isinstance(tolerance, (int, float)) or not (0 < tolerance < 1): + errors.append(f"{where}.tolerance: must be a number in (0, 1)") + floor = part.get("agree_floor", DEFAULT_AGREE_FLOOR) + if isinstance(floor, bool) or not isinstance(floor, (int, float)) or not (0 < floor <= 1): + errors.append(f"{where}.agree_floor: must be a number in (0, 1]") + return errors + + +def load_catalog(path: Path | str) -> dict[str, Any]: + """Read + validate. Raises ValueError with every violation listed.""" + path = Path(path) + catalog = json.loads(path.read_text(encoding="utf-8")) + errors = validate_catalog(catalog) + if errors: + raise ValueError("invalid parts catalog " + str(path) + ":\n " + "\n ".join(errors)) + return catalog + + +def parts_by_z(catalog: dict[str, Any]) -> list[dict[str, Any]]: + """Parts in draw order (bottom first).""" + return sorted(catalog["parts"], key=lambda part: part["z"]) + + +def jobs(catalog: dict[str, Any]) -> list[dict[str, Any]]: + """Every (part, variant) generation job, in draw order then declaration order.""" + out: list[dict[str, Any]] = [] + for part in parts_by_z(catalog): + variants = part.get("variants", {DEFAULT_VARIANT: ""}) + for vname, suffix in variants.items(): + out.append({ + "part": part["id"], "variant": vname, "z": part["z"], "bbox": list(part["bbox"]), + "prompt": part["prompt"] + ((" " + suffix.strip()) if suffix.strip() else ""), + "tolerance": float(part.get("tolerance", DEFAULT_TOLERANCE)), + }) + return out + + +def job_name(part: str, variant: str) -> str: + return part if variant == DEFAULT_VARIANT else f"{part}__{variant}" diff --git a/sprite_gen/parts/match.py b/sprite_gen/parts/match.py new file mode 100644 index 00000000..b0b8366d --- /dev/null +++ b/sprite_gen/parts/match.py @@ -0,0 +1,295 @@ +# SPDX-License-Identifier: Apache-2.0 +"""`sprite-gen parts match` — register generated parts onto the base by pixels. + +For every catalog part (top-most draw order first) the candidate image is +trimmed to its own alpha box, then searched over a small scale × integer +offset grid around the catalog bbox for the placement whose alpha-weighted +RGB difference against the base is smallest. Pixels already claimed by a +higher part (its placed alpha) are excluded from a lower part's score, so a +face is compared only where the bangs do not cover it. + +The gate: a part passes when its best normalized difference is at or below its +tolerance; the run passes when every default variant passes AND the full +z-ordered composite of the placed defaults reproduces the base within the +catalog's composite tolerance. Every failure is reported by part with its best +score and placement — nothing is skipped or approximated silently. + +Outputs, under `--out-dir` (the gen output dir by default): + placed/.png part re-sampled to canvas size at its placement + composite.png defaults stacked in z order (for the eye) + parts-match.report.json per-part placement, score, pass/fail; composite score +""" + +from __future__ import annotations + +import argparse +import json +import math +from pathlib import Path +from typing import Any + +from PIL import Image + +from sprite_gen._deps import np +from sprite_gen.parts.catalog import DEFAULT_AGREE_FLOOR, DEFAULT_VARIANT, job_name, jobs, load_catalog, parts_by_z + +COMPOSITE_TOLERANCE = 0.05 +COMPOSITE_COVERAGE = 0.97 +DEFAULT_OFFSET_REACH = 0.08 # fraction of the bbox size searched around the catalog placement + + +def trim_to_alpha(image: Image.Image) -> tuple[Image.Image, tuple[int, int, int, int]]: + """Crop to the alpha bounding box; returns (cropped, box) or raises on fully transparent input.""" + rgba = image.convert("RGBA") + box = rgba.getchannel("A").getbbox() + if box is None: + raise ValueError("part image is fully transparent") + return rgba.crop(box), box + + +def contain(part_size: tuple[int, int], box: tuple[int, int]) -> tuple[int, int]: + """Largest size with the part's own aspect ratio that fits inside `box` (never stretched).""" + pw, ph = part_size + bw, bh = box + k = min(bw / max(pw, 1), bh / max(ph, 1)) + return max(1, int(round(pw * k))), max(1, int(round(ph * k))) + + +def _place(part: Image.Image, canvas: tuple[int, int], x: int, y: int, w: int, h: int) -> np.ndarray: + resampled = part.resize((max(1, w), max(1, h)), Image.Resampling.LANCZOS) + layer = Image.new("RGBA", canvas, (0, 0, 0, 0)) + layer.paste(resampled, (x, y), resampled) + return np.asarray(layer, dtype=np.float32) + + +AGREE_DIFF = 0.12 # per-pixel mean RGB distance (0..1) under which a pixel "agrees" with the base + + +MISS_WEIGHT = 1.5 +PALETTE_BITS = 5 +PALETTE_MIN_MASS = 0.002 + + +def palette_mask(part: Image.Image) -> np.ndarray: + """Boolean lookup over quantized RGB (PALETTE_BITS per channel): which colours the part contains + with at least PALETTE_MIN_MASS of its opaque pixels. Used to decide whether an uncovered base + pixel "belongs" to this part (a hair part owes red pixels, a shirt part does not).""" + data = np.asarray(part.convert("RGBA"), dtype=np.uint8) + opaque = data[..., 3] > 127 + if not opaque.any(): + return np.zeros((1 << PALETTE_BITS,) * 3, dtype=bool) + q = data[..., :3][opaque] >> (8 - PALETTE_BITS) + idx = (q[:, 0].astype(np.int64) << (2 * PALETTE_BITS)) | (q[:, 1].astype(np.int64) << PALETTE_BITS) | q[:, 2] + counts = np.bincount(idx, minlength=1 << (3 * PALETTE_BITS)) + mask = counts >= max(1, int(PALETTE_MIN_MASS * opaque.sum())) + return mask.reshape((1 << PALETTE_BITS,) * 3) + + +def owned_region(base: np.ndarray, free: np.ndarray, bbox: list[int], palette: np.ndarray) -> np.ndarray: + """Base pixels inside the bbox, unclaimed, whose colour is in the part's palette: what the part + is expected to cover. Leaving them uncovered is the "miss" cost that stops a part shrinking.""" + x, y, w, h = bbox + region = np.zeros(base.shape[:2], dtype=np.float32) + sub = base[y:y + h, x:x + w] + q = (sub[..., :3].astype(np.int64) >> (8 - PALETTE_BITS)) + inpal = palette[q[..., 0], q[..., 1], q[..., 2]] + region[y:y + h, x:x + w] = (sub[..., 3] / 255.0) * free[y:y + h, x:x + w] * inpal + return region + + +def score_placement(layer: np.ndarray, base: np.ndarray, free: np.ndarray, area: float, + region: np.ndarray | None = None) -> tuple[float, float, float]: + """(objective, colour, agree) for one placement. + + objective (minimize) = -(agreeing - 2 x disagreeing - MISS_WEIGHT x missed) / bbox area, where + a visible part pixel *agrees* when its mean RGB distance to the base is <= AGREE_DIFF, and + *missed* counts owned-region pixels (base pixels of the part's own colours inside its box) the + placement leaves uncovered. Shrinking loses agreeing pixels and gains misses; oversizing pays + for every pixel spilled onto something else — the optimum reproduces the most base pixels. + colour = alpha-weighted mean RGB distance over the part's visible, unclaimed pixels (the gate). + agree = fraction of those pixels within AGREE_DIFF of the base. + """ + alpha = (layer[..., 3] / 255.0) * free + weight = float(alpha.sum()) + if weight <= 0: + return 2.0, 1.0, 0.0 + diff = np.abs(layer[..., :3] - base[..., :3]).mean(axis=-1) / 255.0 + colour = float((diff * alpha).sum() / weight) + agreeing = float((alpha * (diff <= AGREE_DIFF)).sum()) + disagreeing = float(weight - agreeing) + missed = float((region * (1.0 - layer[..., 3] / 255.0)).sum()) if region is not None else 0.0 + objective = -(agreeing - 2.0 * disagreeing - MISS_WEIGHT * missed) / max(area, 1.0) + return float(objective), colour, float(agreeing / weight) + + +DENSE_SCALES = tuple(round(0.5 + 0.02 * i, 2) for i in range(31)) # 0.50 … 1.10 + + +def register(part: Image.Image, base: np.ndarray, free: np.ndarray, bbox: list[int], + *, reach: float = DEFAULT_OFFSET_REACH, scales: tuple[float, ...] = DENSE_SCALES) -> dict[str, Any]: + """FFT template registration: every offset of every scale, deterministic argmin. + + The candidate is trimmed to its alpha box and contain-fitted into bbox x scale (never + stretched); for each scale one masked-SSD cost map over the search window (bbox grown + by `reach` and by the largest scale) is computed with FFT correlations + (`register_fft.cost_map`), so cost is a few FFTs per scale regardless of how many offsets + exist. The winner is re-scored at full precision for the gate (colour, agree). + """ + from sprite_gen.parts.register_fft import best_offset, cost_map + + trimmed, _ = trim_to_alpha(part) + bx, by, bw, bh = bbox + H, W = base.shape[:2] + rx, ry = max(2, int(round(bw * reach))), max(2, int(round(bh * reach))) + grow = int(math.ceil(max(bw, bh) * (max(scales) - 1.0))) + 2 + wx0, wy0 = max(0, bx - rx - grow), max(0, by - ry - grow) + wx1, wy1 = min(W, bx + bw + rx + grow), min(H, by + bh + ry + grow) + base_w, free_w = base[wy0:wy1, wx0:wx1], free[wy0:wy1, wx0:wx1] + window = (wx1 - wx0, wy1 - wy0) + area = float(bw * bh) + region_w = owned_region(base, free, bbox, palette_mask(trimmed))[wy0:wy1, wx0:wx1] + base_rgb, base_alpha = base_w[..., :3], base_w[..., 3] + # Stage 1 (FFT): per scale, shortlist the offsets that minimise the masked-SSD cost and the + # pure colour error. Stage 2 (exact): re-score every shortlisted placement with the agreement + # objective (agree - 2 x disagree - miss), which is robust to textured parts where a squared + # error alone still rewards shrinking. + candidates: list[tuple[float, float, int, int, int, int, Image.Image]] = [] + for scale in scales: + w, h = contain(trimmed.size, (max(1, int(round(bw * scale))), max(1, int(round(bh * scale))))) + if w > window[0] or h > window[1]: + continue + resampled = trimmed.resize((w, h), Image.Resampling.LANCZOS) + arr = np.asarray(resampled, dtype=np.float32) + cost, ssd, Wm = cost_map(arr[..., :3], arr[..., 3], base_rgb, base_alpha, free_w, region_w, area) + picks = {best_offset(cost)[:2]} + colour_only = np.where(Wm > 0.25 * float(Wm.max() or 1.0), ssd / np.maximum(Wm, 1e-6), np.inf) + picks.add(best_offset(colour_only)[:2]) + for u, v in picks: + candidates.append((scale, 0.0, wx0 + v, wy0 + u, w, h, resampled)) + if not candidates: + raise ValueError("part does not fit inside its search window at any scale") + winner: dict[str, Any] | None = None + for scale, _c, x, y, w, h, resampled in candidates: + layer = Image.new("RGBA", window, (0, 0, 0, 0)) + layer.paste(resampled, (x - wx0, y - wy0), resampled) + obj, colour, agree = score_placement(np.asarray(layer, dtype=np.float32), base_w, free_w, area, region_w) + if winner is None or obj < winner["objective"] or (obj == winner["objective"] and abs(scale - 1.0) < abs(winner["scale"] - 1.0)): + winner = {"objective": obj, "scale": scale, "x": x, "y": y, "w": w, "h": h, "_img": resampled, + "score": colour, "agree": agree} + winner.pop("_img") + winner.update({"score": round(winner["score"], 4), "agree": round(winner["agree"], 4), + "objective": round(float(winner["objective"]), 4)}) + return winner + + +def match_parts(catalog_path: Path, parts_dir: Path, out_dir: Path | None = None, + *, composite_tolerance: float | None = None) -> dict[str, Any]: + catalog = load_catalog(catalog_path) + declared = catalog.get("composite", {}) + if composite_tolerance is None: + composite_tolerance = float(declared.get("tolerance", COMPOSITE_TOLERANCE)) + composite_coverage = float(declared.get("coverage", COMPOSITE_COVERAGE)) + base_img = Image.open((catalog_path.parent / catalog["base"]).resolve()).convert("RGBA") + base = np.asarray(base_img, dtype=np.float32) + canvas = (base_img.width, base_img.height) + out_dir = out_dir or parts_dir + placed_dir = out_dir / "placed" + placed_dir.mkdir(parents=True, exist_ok=True) + free = np.ones(base.shape[:2], dtype=np.float32) + records: list[dict[str, Any]] = [] + placements: dict[str, dict[str, Any]] = {} + # top-most first so a lower part is scored only where it is actually visible + for part in reversed(parts_by_z(catalog)): + name = part["id"] + candidate = parts_dir / f"{name}.png" + record: dict[str, Any] = {"job": name, "part": name, "variant": DEFAULT_VARIANT, "z": part["z"]} + if not candidate.is_file(): + record.update({"ok": False, "error": "missing candidate"}) + records.append(record) + continue + try: + best = register(Image.open(candidate), base, free, part["bbox"]) + except ValueError as exc: + record.update({"ok": False, "error": str(exc)}) + records.append(record) + continue + tolerance = float(part.get("tolerance", 0.06)) + floor = float(part.get("agree_floor", DEFAULT_AGREE_FLOOR)) + ok = bool(best["score"] <= tolerance and best["agree"] >= floor) + record.update({"ok": ok, "placement": best, "tolerance": tolerance, "agree_floor": floor}) + if not ok: + record["error"] = (f"best score {best['score']} exceeds tolerance {tolerance}" if best["score"] > tolerance + else f"only {best['agree']:.0%} of the part agrees with the base (floor {floor:.0%})") + records.append(record) + placements[name] = best + trimmed, _ = trim_to_alpha(Image.open(candidate)) + layer = _place(trimmed, canvas, best["x"], best["y"], best["w"], best["h"]) + Image.fromarray(layer.astype("uint8"), "RGBA").save(placed_dir / f"{name}.png") + free = free * (1.0 - layer[..., 3] / 255.0) + # variants inherit their default's placement (same box, same scale) + for job in jobs(catalog): + if job["variant"] == DEFAULT_VARIANT: + continue + name = job_name(job["part"], job["variant"]) + candidate = parts_dir / f"{name}.png" + record = {"job": name, "part": job["part"], "variant": job["variant"], "z": job["z"]} + placement = placements.get(job["part"]) + if not candidate.is_file(): + record.update({"ok": False, "error": "missing candidate"}) + elif placement is None: + record.update({"ok": False, "error": "default variant did not place"}) + else: + trimmed, _ = trim_to_alpha(Image.open(candidate)) + layer = _place(trimmed, canvas, placement["x"], placement["y"], placement["w"], placement["h"]) + Image.fromarray(layer.astype("uint8"), "RGBA").save(placed_dir / f"{name}.png") + record.update({"ok": True, "placement": dict(placement), "inherited_from": job["part"]}) + records.append(record) + # full composite of the defaults, bottom first + composite = Image.new("RGBA", canvas, (0, 0, 0, 0)) + for part in parts_by_z(catalog): + layer_path = placed_dir / f"{part['id']}.png" + if layer_path.is_file(): + composite.alpha_composite(Image.open(layer_path).convert("RGBA")) + composite.save(out_dir / "composite.png") + comp = np.asarray(composite, dtype=np.float32) + cover = np.minimum(comp[..., 3], base[..., 3]) / 255.0 + diff = np.abs(comp[..., :3] - base[..., :3]).mean(axis=-1) / 255.0 + composite_score = round(float((diff * cover).sum() / max(cover.sum(), 1.0)), 4) + base_alpha = base[..., 3] / 255.0 + coverage = round(float((np.minimum(comp[..., 3] / 255.0, base_alpha)).sum() / max(base_alpha.sum(), 1.0)), 4) + composite_ok = bool(composite_score <= composite_tolerance and coverage >= composite_coverage) + report = {"kind": "sprite-gen-parts-match-report", "version": 1, "catalog": str(catalog_path.resolve()), + "parts_dir": str(parts_dir.resolve()), "parts": records, + "composite": {"score": composite_score, "coverage": coverage, "tolerance": composite_tolerance, + "coverage_floor": composite_coverage, "ok": composite_ok, "path": str(out_dir / "composite.png")}, + "ok": composite_ok and all(r["ok"] for r in records), + "failed": [r["job"] for r in records if not r["ok"]] + ([] if composite_ok else ["composite"])} + (out_dir / "parts-match.report.json").write_text(json.dumps(report, indent=2, ensure_ascii=False) + "\n", encoding="utf-8") + return report + + +def add_arguments(p: argparse.ArgumentParser) -> None: + p.add_argument("--catalog", required=True, type=Path) + p.add_argument("--parts-dir", required=True, type=Path, help="output dir of `parts gen`") + p.add_argument("--out-dir", type=Path, default=None, help="default: --parts-dir") + p.add_argument("--composite-tolerance", type=float, default=None, help="override the catalog's composite.tolerance") + + +def run(*, catalog: Path, parts_dir: Path, out_dir: Path | None = None, + composite_tolerance: float | None = None) -> int: + report = match_parts(Path(catalog), Path(parts_dir), Path(out_dir) if out_dir else None, + composite_tolerance=composite_tolerance) + print(json.dumps({"ok": report["ok"], "failed": report["failed"], "composite": report["composite"]}, ensure_ascii=False)) + return 0 if report["ok"] else 1 + + +def main(argv: list[str] | None = None) -> int: + parser = argparse.ArgumentParser(description="Register generated parts onto the base by pixels.") + add_arguments(parser) + args = parser.parse_args(argv) + return run(catalog=args.catalog, parts_dir=args.parts_dir, out_dir=args.out_dir, + composite_tolerance=args.composite_tolerance) + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/sprite_gen/parts/parts_gen.py b/sprite_gen/parts/parts_gen.py new file mode 100644 index 00000000..b64fd4d9 --- /dev/null +++ b/sprite_gen/parts/parts_gen.py @@ -0,0 +1,195 @@ +# SPDX-License-Identifier: Apache-2.0 +"""`sprite-gen parts gen` — generate every catalog part alone, in parallel. + +Each job sends the provider the whole base image plus a padded crop of the +part's box as references, and asks for that part only on a flat chroma +background. The chroma path (`gen.generate_image(transparent=True)`) keys it +to RGBA; a result with no transparent pixels is a failure, never a layer. + +Outputs, under `--out-dir`: + .png / __.png RGBA part candidates + parts-gen.report.json one record per job (ok / error) + +One failing job does not stop the others (they are independent provider +calls), but the command exits non-zero if any job failed, and the report names +every failure — No Silent Fallback. +""" + +from __future__ import annotations + +import argparse +import json +import sys +from concurrent.futures import ThreadPoolExecutor +from pathlib import Path +from typing import Any + +from PIL import Image + +from sprite_gen._deps import np + +from sprite_gen.gen import generate_image +from sprite_gen.parts.catalog import job_name, jobs, load_catalog + +CHROMA_HEX = {"green": "#00FF00", "magenta": "#FF00FF"} +DEFAULT_WORKERS = 6 +CROP_PAD = 0.25 # fraction of the box added on each side of the reference crop + +PART_PROMPT = ( + "Reference 1 is the full character; reference 2 is a close crop of the region to draw. " + "Draw ONLY this part of that exact character: {prompt} " + "Reproduce it pixel-faithfully from the references: identical shape, position within the crop, " + "line weight, colours and shading, same anime illustration style, same scale and viewing angle. " + "Draw nothing else — no other body parts, no clothing, no background elements. " + "Everything outside the part must be a perfectly flat solid {chroma_name} {chroma_hex} chroma key: " + "no gradient, no shadow, no checkerboard. No text, no watermark." +) + + +def crop_reference(base: Image.Image, bbox: list[int], pad: float = CROP_PAD) -> Image.Image: + x, y, w, h = bbox + px, py = int(round(w * pad)), int(round(h * pad)) + box = (max(0, x - px), max(0, y - py), min(base.width, x + w + px), min(base.height, y + h + py)) + return base.crop(box) + + +def flatten_on_chroma(image: Image.Image, hex_color: str) -> Image.Image: + """The provider sees a chroma-flat reference, so the crop's transparency reads as the key colour.""" + rgb = tuple(int(hex_color[i:i + 2], 16) for i in (1, 3, 5)) + out = Image.new("RGBA", image.size, rgb + (255,)) + out.alpha_composite(image.convert("RGBA")) + return out.convert("RGB") + + +def despill(path: Path, chroma: str) -> int: + """Remove the key hue that survives keying along antialiased edges: a pixel whose key + channel exceeds the other two is clamped to their max. Returns the pixel count touched. + Safe for these subjects because a green (or magenta) cast is never native art here.""" + image = Image.open(path).convert("RGBA") + data = np.asarray(image, dtype=np.int16).copy() + r, g, b, a = data[..., 0], data[..., 1], data[..., 2], data[..., 3] + if chroma == "green": + limit = np.maximum(r, b) + mask = (a > 0) & (g > limit + 6) + data[..., 1] = np.where(mask, limit, g) + else: # magenta: R and B high, G low + limit = g + mask = (a > 0) & (np.minimum(r, b) > limit + 6) + data[..., 0] = np.where(mask, np.minimum(r, limit + (r - limit) // 2), r) + data[..., 2] = np.where(mask, np.minimum(b, limit + (b - limit) // 2), b) + touched = int(mask.sum()) + if touched: + Image.fromarray(data.astype("uint8"), "RGBA").save(path) + return touched + + +def alpha_stats(path: Path) -> dict[str, float]: + image = Image.open(path) + if image.mode != "RGBA": + return {"alpha_zero_pct": 0.0, "opaque_pct": 0.0, "mode": image.mode} + hist = image.getchannel("A").histogram() + total = image.width * image.height + return {"alpha_zero_pct": round(100.0 * hist[0] / total, 2), + "opaque_pct": round(100.0 * sum(hist[250:]) / total, 2), "mode": "RGBA"} + + +def _one(job: dict[str, Any], *, provider: str, base_path: Path, base: Image.Image, chroma: str, + out_dir: Path, workdir: Path) -> dict[str, Any]: + name = job_name(job["part"], job["variant"]) + record: dict[str, Any] = {"job": name, "part": job["part"], "variant": job["variant"], "bbox": job["bbox"]} + try: + ref_dir = workdir / name + ref_dir.mkdir(parents=True, exist_ok=True) + crop_path = ref_dir / "crop.png" + flatten_on_chroma(crop_reference(base, job["bbox"]), CHROMA_HEX[chroma]).save(crop_path) + base_ref = ref_dir / "base.png" + flatten_on_chroma(base, CHROMA_HEX[chroma]).save(base_ref) + prompt = PART_PROMPT.format(prompt=job["prompt"], chroma_name=chroma, chroma_hex=CHROMA_HEX[chroma]) + out = out_dir / f"{name}.png" + result = generate_image(provider, prompt, out, refs=[base_ref, crop_path], transparent=True, + chroma_key=chroma, workdir=ref_dir / "gen") + despilled = despill(out, chroma) + stats = alpha_stats(out) + record.update({"ok": True, "out": str(out), "raw": str(result.raw), "elapsed_seconds": result.elapsed_seconds, + "provider": result.provider, "alpha": stats, "despilled_pixels": despilled}) + if stats["mode"] != "RGBA" or stats["alpha_zero_pct"] <= 0.0: + record.update({"ok": False, "error": "generated part has no transparent pixels after keying"}) + except SystemExit as exc: # generate_image fails loud with SystemExit + record.update({"ok": False, "error": str(exc)}) + except Exception as exc: # noqa: BLE001 — every job failure must land in the report + record.update({"ok": False, "error": f"{type(exc).__name__}: {exc}"}) + return record + + +def generate_parts(catalog_path: Path, out_dir: Path, *, provider: str = "codex", workers: int = DEFAULT_WORKERS, + only: list[str] | None = None, workdir: Path | None = None) -> dict[str, Any]: + catalog = load_catalog(catalog_path) + base_path = (catalog_path.parent / catalog["base"]).resolve() + if not base_path.is_file(): + raise SystemExit(f"parts gen: base image not found: {base_path}") + base = Image.open(base_path).convert("RGBA") + if (base.width, base.height) != (catalog["canvas"]["width"], catalog["canvas"]["height"]): + raise SystemExit(f"parts gen: base is {base.width}x{base.height} but canvas declares " + f"{catalog['canvas']['width']}x{catalog['canvas']['height']}") + chroma = catalog.get("chroma_key", "green") + todo = jobs(catalog) + if only: + wanted = set(only) + todo = [job for job in todo if job["part"] in wanted or job_name(job["part"], job["variant"]) in wanted] + if not todo: + raise SystemExit(f"parts gen: --only matched no catalog job: {sorted(wanted)}") + out_dir.mkdir(parents=True, exist_ok=True) + workdir = workdir or (out_dir / ".work") + workdir.mkdir(parents=True, exist_ok=True) + with ThreadPoolExecutor(max_workers=max(1, workers)) as pool: + records = list(pool.map(lambda job: _one(job, provider=provider, base_path=base_path, base=base, chroma=chroma, + out_dir=out_dir, workdir=workdir), todo)) + # A partial (`--only`) run merges into the previous report so the file always + # describes every job that exists on disk, never just the last invocation. + report_path = out_dir / "parts-gen.report.json" + merged: dict[str, dict[str, Any]] = {} + if only and report_path.is_file(): + try: + for old in json.loads(report_path.read_text(encoding="utf-8")).get("jobs", []): + merged[old["job"]] = old + except (OSError, ValueError): + merged = {} + for record in records: + merged[record["job"]] = record + ordered = [merged[job_name(j["part"], j["variant"])] for j in jobs(catalog) if job_name(j["part"], j["variant"]) in merged] + report = {"kind": "sprite-gen-parts-gen-report", "version": 1, "catalog": str(catalog_path.resolve()), + "base": str(base_path), "provider": provider, "chroma_key": chroma, "jobs": ordered, + "ran": [r["job"] for r in records], + "ok": all(r["ok"] for r in ordered), "failed": [r["job"] for r in ordered if not r["ok"]]} + report_path.write_text(json.dumps(report, indent=2, ensure_ascii=False) + "\n", encoding="utf-8") + return report + + +def add_arguments(p: argparse.ArgumentParser) -> None: + p.add_argument("--catalog", required=True, type=Path, help="parts catalog JSON (base path is relative to it)") + p.add_argument("--out-dir", required=True, type=Path) + p.add_argument("--provider", default="codex", choices=["codex", "grok"]) + p.add_argument("--workers", type=int, default=DEFAULT_WORKERS) + p.add_argument("--only", default=None, help="comma-separated part ids or part__variant names") + + +def run(*, catalog: Path, out_dir: Path, provider: str = "codex", workers: int = DEFAULT_WORKERS, + only: str | None = None) -> int: + report = generate_parts(Path(catalog), Path(out_dir), provider=provider, workers=workers, + only=[s.strip() for s in only.split(",") if s.strip()] if only else None) + print(json.dumps({k: report[k] for k in ("ok", "failed", "provider")}, ensure_ascii=False)) + for record in report["jobs"]: + if not record["ok"]: + print(f"[parts gen] FAILED {record['job']}: {record['error']}", file=sys.stderr) + return 0 if report["ok"] else 1 + + +def main(argv: list[str] | None = None) -> int: + parser = argparse.ArgumentParser(description="Generate every catalog part alone, in parallel.") + add_arguments(parser) + args = parser.parse_args(argv) + return run(catalog=args.catalog, out_dir=args.out_dir, provider=args.provider, workers=args.workers, only=args.only) + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/sprite_gen/parts/register_fft.py b/sprite_gen/parts/register_fft.py new file mode 100644 index 00000000..8a064586 --- /dev/null +++ b/sprite_gen/parts/register_fft.py @@ -0,0 +1,62 @@ +# SPDX-License-Identifier: Apache-2.0 +"""FFT template registration for parts — every offset of one scale in a handful of FFTs. + +For a candidate part `t` (RGB) with mask `m` (alpha), a base window `b`, a `free` +mask (pixels not yet claimed by higher parts) and an owned `region` (base pixels of +the part's own colours it is expected to cover), the per-offset cost is + + cost(u,v) = [ SSD(u,v)/(3*255^2) - REWARD * W(u,v) + MISS_WEIGHT * miss(u,v) ] / area + +where SSD = sum m·free·(t-b)^2 over channels, W = sum m·free (matched pixel mass), +miss = sum region - corr(region, m) (owned pixels left uncovered). Each term is a +cross-correlation, so one scale costs four rfft2 round-trips regardless of how many +offsets are evaluated. Deterministic: same inputs, same argmin (ties → first). +""" + +from __future__ import annotations + +from typing import Any + +from PIL import Image + +from sprite_gen._deps import np + +REWARD = 0.03 # per matched pixel: an exact match contributes -REWARD, so covering is rewarded +MISS_WEIGHT = 0.06 # per owned pixel left uncovered + + +def _corr(f: np.ndarray, g: np.ndarray) -> np.ndarray: + """corr(f, g)[u, v] = sum_xy f[u+x, v+y] * g[x, y] for u in [0, Hf-Hg], v in [0, Wf-Wg].""" + Hf, Wf = f.shape + Hg, Wg = g.shape + F = np.fft.rfft2(f, s=(Hf, Wf)) + G = np.fft.rfft2(g, s=(Hf, Wf)) + full = np.fft.irfft2(F * np.conj(G), s=(Hf, Wf)) + return full[: Hf - Hg + 1, : Wf - Wg + 1] + + +def cost_map(part_rgb: np.ndarray, part_alpha: np.ndarray, base_rgb: np.ndarray, base_alpha: np.ndarray, + free: np.ndarray, region: np.ndarray, area: float) -> tuple[np.ndarray, np.ndarray, np.ndarray]: + """Return (cost, ssd_norm, W) maps over every offset where the part fits inside the window.""" + m = part_alpha.astype(np.float64) / 255.0 + ssd = np.zeros((base_rgb.shape[0] - m.shape[0] + 1, base_rgb.shape[1] - m.shape[1] + 1)) + W = _corr(free.astype(np.float64), m) + # a part pixel over transparent base is a disagreement: fold base alpha into "free" for the colour terms + vis = free.astype(np.float64) * (base_alpha.astype(np.float64) / 255.0) + Wvis = _corr(vis, m) + for c in range(3): + t = part_rgb[..., c].astype(np.float64) + b = base_rgb[..., c].astype(np.float64) + ssd += _corr(vis, m * t * t) - 2.0 * _corr(vis * b, m * t) + _corr(vis * b * b, m) + # pixels placed over empty base count as fully wrong + ssd += (W - Wvis) * 3.0 * 255.0 * 255.0 + ssd_norm = ssd / (3.0 * 255.0 * 255.0) + miss = float(region.sum()) - _corr(region.astype(np.float64), m) + cost = (ssd_norm - REWARD * W + MISS_WEIGHT * miss) / max(area, 1.0) + return cost, ssd_norm, W + + +def best_offset(cost: np.ndarray) -> tuple[int, int, float]: + idx = int(np.argmin(cost)) + u, v = divmod(idx, cost.shape[1]) + return u, v, float(cost[u, v]) diff --git a/sprite_gen/parts/rig.py b/sprite_gen/parts/rig.py new file mode 100644 index 00000000..908c8d0d --- /dev/null +++ b/sprite_gen/parts/rig.py @@ -0,0 +1,258 @@ +# SPDX-License-Identifier: Apache-2.0 +"""`sprite-gen parts rig` — rig.json + a seek-safe HTML runtime from matched parts. + +Reads the catalog and the `parts match` report, and writes: + rig.json z-ordered parts, groups with pivots, variant files (canvas-sized PNGs) + rig.html an HTML fragment: one absolutely positioned per part variant, + group wrappers with transform-origin at the group pivot + rig-keys.js GSAP keys for a paused timeline (`window.__rigKeys(tl, start)`) + rig-keys.json the same keys as data (mouth variant per frame, blinks, sway) + +Every key is derived deterministically: mouth openness from the narration's +RMS envelope (ffmpeg → 16 kHz mono PCM → per-frame RMS, quantized against the +clip's own peak), blinks on a fixed cadence seeded by the clip length, head +sway as a slow sine. The same inputs always produce the same keys, which is +what a frame-stepping renderer (HyperFrames) requires. +""" + +from __future__ import annotations + +import argparse +import hashlib +import html +import itertools +import json +import math +import shutil +import subprocess +from pathlib import Path +from typing import Any + +from sprite_gen._deps import np +from sprite_gen.parts.catalog import DEFAULT_VARIANT, job_name, load_catalog, parts_by_z + +MOUTH_LEVELS = ("closed", "half", "open") +MOUTH_THRESHOLDS = (0.12, 0.38, 0.70) # fraction of the clip's peak RMS → closed | half | open | o +BLINK_PERIOD = 3.4 +BLINK_CLOSE = 0.08 +BLINK_HOLD = 0.06 +SWAY_PERIOD = 5.2 +SWAY_DEGREES = 1.2 + + +def rms_envelope(audio: Path, fps: int, *, sample_rate: int = 16000) -> list[float]: + """Per-frame RMS of the audio, normalized to the clip peak (0..1).""" + if shutil.which("ffmpeg") is None: + raise SystemExit("parts rig: ffmpeg is required to read the audio envelope") + pcm = subprocess.run(["ffmpeg", "-v", "error", "-i", str(audio), "-ac", "1", "-ar", str(sample_rate), + "-f", "s16le", "-"], check=True, capture_output=True).stdout + samples = np.frombuffer(pcm, dtype=np.int16).astype(np.float32) / 32768.0 + if fps <= 0 or fps > sample_rate: + raise ValueError("fps must be between 1 and the audio sample rate") + frames = int(math.ceil(len(samples) * fps / sample_rate)) + env = np.zeros(frames, dtype=np.float32) + for i in range(frames): + chunk = samples[i * sample_rate // fps:(i + 1) * sample_rate // fps] + env[i] = float(np.sqrt(np.mean(chunk * chunk))) if len(chunk) else 0.0 + peak = float(env.max()) if frames else 0.0 + if peak <= 0: + return [0.0] * frames + # light smoothing so a single loud sample does not flap the mouth + kernel = np.array([0.25, 0.5, 0.25], dtype=np.float32) + smoothed = np.convolve(np.pad(env / peak, (1, 1)), kernel, mode="valid") + return [round(float(v), 4) for v in smoothed] + + +def mouth_variant(level: float, available: set[str]) -> str: + if level < MOUTH_THRESHOLDS[0] or "closed" not in available and DEFAULT_VARIANT in available: + return "closed" if "closed" in available else DEFAULT_VARIANT + if level < MOUTH_THRESHOLDS[1]: + return "half" if "half" in available else ("open" if "open" in available else DEFAULT_VARIANT) + if level < MOUTH_THRESHOLDS[2] or "o" not in available: + return "open" if "open" in available else DEFAULT_VARIANT + return "o" + + +def mouth_keys(envelope: list[float], fps: int, available: set[str], *, start: float = 0.0) -> list[dict[str, Any]]: + """Variant changes only (run-length), as (t, variant).""" + keys: list[dict[str, Any]] = [] + current = None + for i, level in enumerate(envelope): + variant = mouth_variant(level, available) + if variant != current: + keys.append({"t": round(start + i / fps, 4), "variant": variant}) + current = variant + return keys + + +def blink_keys(duration: float, *, start: float = 0.0, seed: str = "") -> list[dict[str, Any]]: + """Fixed-cadence blinks with a deterministic phase from the seed.""" + digest = int(hashlib.sha256(seed.encode("utf-8")).hexdigest()[:8], 16) + phase = (digest % 1000) / 1000.0 * BLINK_PERIOD + keys: list[dict[str, Any]] = [] + t = phase + 0.6 + while t + BLINK_CLOSE * 2 + BLINK_HOLD < duration: + keys.append({"t": round(start + t, 4), "variant": "half"}) + keys.append({"t": round(start + t + BLINK_CLOSE, 4), "variant": "closed"}) + keys.append({"t": round(start + t + BLINK_CLOSE + BLINK_HOLD, 4), "variant": "half"}) + keys.append({"t": round(start + t + BLINK_CLOSE * 2 + BLINK_HOLD, 4), "variant": "open"}) + t += BLINK_PERIOD + ((digest >> 8) % 7) * 0.1 + return keys + + +def sway_keys(duration: float, *, start: float = 0.0, step: float = 0.5) -> list[dict[str, Any]]: + keys: list[dict[str, Any]] = [] + t = 0.0 + while t <= duration: + keys.append({"t": round(start + t, 4), "rotation": round(SWAY_DEGREES * math.sin(2 * math.pi * t / SWAY_PERIOD), 3)}) + t += step + return keys + + +def build_rig(catalog_path: Path, match_dir: Path) -> dict[str, Any]: + catalog = load_catalog(catalog_path) + report_path = match_dir / "parts-match.report.json" + if not report_path.is_file(): + raise SystemExit(f"parts rig: no match report at {report_path} — run `parts match` first") + report = json.loads(report_path.read_text(encoding="utf-8")) + if not report.get("ok"): + raise SystemExit(f"parts rig: match report is not passing (failed: {report.get('failed')}) — a rig from unmatched parts is not a rig") + placed = {r["job"]: r for r in report["parts"] if r.get("ok")} + parts_out: list[dict[str, Any]] = [] + for part in parts_by_z(catalog): + variants: dict[str, str] = {} + for vname in part.get("variants", {DEFAULT_VARIANT: ""}): + name = job_name(part["id"], vname) + if name not in placed: + raise SystemExit(f"parts rig: {name} has no passing placement in the match report") + if not (match_dir / "placed" / f"{name}.png").is_file(): + raise SystemExit(f"parts rig: missing placed layer: {name}") + variants[vname] = f"placed/{name}.png" + parts_out.append({"id": part["id"], "z": part["z"], "group": part.get("group", "none"), + "pivot": list(part["pivot"]), "placement": placed[part["id"]]["placement"], + "variants": variants}) + groups = {name: {"pivot": list(g["pivot"]), "members": [p["id"] for p in parts_out if p["group"] == name]} + for name, g in catalog.get("groups", {}).items()} + return {"kind": "sprite-gen-parts-rig", "version": 1, "character": catalog.get("character"), + "canvas": dict(catalog["canvas"]), "base": catalog["base"], "groups": groups, "parts": parts_out, + "match_report": str(report_path.resolve())} + + +def render_html(rig: dict[str, Any], *, prefix: str = "rig", asset_prefix: str = "") -> str: + w, h = rig["canvas"]["width"], rig["canvas"]["height"] + lines = [f'
'] + # A transformed wrapper creates a stacking context. Split a group whenever + # another group interrupts its z run, keeping all runs on the same pivot. + runs: dict[str, int] = {} + for gname, run in itertools.groupby(sorted(rig["parts"], key=lambda p: p["z"]), key=lambda p: p["group"]): + members = list(run) + if gname != "none": + px, py = rig["groups"][gname]["pivot"] + index = runs.get(gname, 0) + runs[gname] = index + 1 + gid = f"{prefix}-g-{gname}" + (f"--{index}" if index else "") + lines.append(f'
') + for part in members: + for vname, rel in part["variants"].items(): + vid = f"{prefix}-{part['id']}" + ("" if vname == DEFAULT_VARIANT else f"__{vname}") + hidden = "" if vname == DEFAULT_VARIANT else "opacity:0;" + lines.append(f' ') + if gname != "none": + lines.append("
") + lines.append("
") + return "\n".join(lines) + "\n" + + +def render_keys_js(keys: dict[str, Any], *, prefix: str = "rig") -> str: + """A function that stamps every key onto a paused GSAP timeline (opacity/rotation sets only).""" + out = ["// generated by sprite-gen parts rig — deterministic; do not hand-edit", + f"window.__rigKeys = function (tl, start) {{ start = start || 0; const P = {json.dumps(prefix)};", + " const show = (part, variant, names, t) => { for (const n of names) tl.set('#' + P + '-' + part + (n === 'default' ? '' : '__' + n), { opacity: n === variant ? 1 : 0 }, start + t); };"] + for track in keys["mouth"]: + names = json.dumps(track["variants"]) + for k in track["keys"]: + out.append(f" show({json.dumps(track['part'])}, {json.dumps(k['variant'])}, {names}, {k['t']});") + for track in keys["blink"]: + names = json.dumps(track["variants"]) + for k in track["keys"]: + out.append(f" show({json.dumps(track['part'])}, {json.dumps(k['variant'])}, {names}, {k['t']});") + if track.get("eye_part"): + visible = 0 if k["variant"] == "closed" else 1 + out.append(f" tl.set('#' + P + '-' + {json.dumps(track['eye_part'])}, {{ opacity: {visible} }}, start + {k['t']});") + for track in keys["sway"]: + for k in track["keys"]: + out.append(f" tl.set({json.dumps('#' + prefix + ' [data-rig-group=' + chr(34) + track['group'] + chr(34) + ']')}, {{ rotation: {k['rotation']} }}, start + {k['t']});") + out.append("};") + return "\n".join(out) + "\n" + + +def build_keys(rig: dict[str, Any], *, audio: Path | None, fps: int, duration: float | None, + start: float = 0.0, mouth_part: str = "mouth", eyelid_parts: tuple[str, ...] = ("eyelid_l", "eyelid_r"), + head_group: str = "head") -> dict[str, Any]: + parts = {p["id"]: p for p in rig["parts"]} + keys: dict[str, Any] = {"fps": fps, "start": start, "mouth": [], "blink": [], "sway": []} + if audio is not None: + env = rms_envelope(audio, fps) + duration = duration or len(env) / fps + if mouth_part in parts: + available = set(parts[mouth_part]["variants"]) + keys["mouth"].append({"part": mouth_part, "variants": sorted(available), + "keys": mouth_keys(env, fps, available, start=start)}) + if duration is None: + raise SystemExit("parts rig: --duration is required when no --audio is given") + seed = f"{audio.name if audio else ''}:{duration:.3f}" + for eyelid in eyelid_parts: + if eyelid in parts and {"open", "half", "closed"} <= set(parts[eyelid]["variants"]): + eye = eyelid.replace("eyelid_", "eye_", 1) + keys["blink"].append({"part": eyelid, "variants": sorted(parts[eyelid]["variants"]), + "eye_part": eye if eye in parts else None, + "keys": blink_keys(duration, start=start, seed=seed)}) + if head_group in rig["groups"]: + keys["sway"].append({"group": head_group, "keys": sway_keys(duration, start=start)}) + keys["duration"] = round(duration, 4) + return keys + + +def add_arguments(p: argparse.ArgumentParser) -> None: + p.add_argument("--catalog", required=True, type=Path) + p.add_argument("--match-dir", required=True, type=Path, help="dir holding parts-match.report.json + placed/") + p.add_argument("--out-dir", type=Path, default=None, help="default: --match-dir") + p.add_argument("--audio", type=Path, default=None, help="narration to drive the mouth (mp3/wav)") + p.add_argument("--duration", type=float, default=None, help="seconds (required without --audio)") + p.add_argument("--fps", type=int, default=30) + p.add_argument("--start", type=float, default=0.0, help="timeline offset for every key") + p.add_argument("--prefix", default="rig", help="DOM id prefix") + p.add_argument("--asset-prefix", default="", help="path prefix for in rig.html") + + +def run(*, catalog: Path, match_dir: Path, out_dir: Path | None = None, audio: Path | None = None, + duration: float | None = None, fps: int = 30, start: float = 0.0, prefix: str = "rig", + asset_prefix: str = "") -> int: + catalog, match_dir = Path(catalog), Path(match_dir) + out_dir = Path(out_dir) if out_dir else match_dir + out_dir.mkdir(parents=True, exist_ok=True) + rig = build_rig(catalog, match_dir) + (out_dir / "rig.json").write_text(json.dumps(rig, indent=2, ensure_ascii=False) + "\n", encoding="utf-8") + (out_dir / "rig.html").write_text(render_html(rig, prefix=prefix, asset_prefix=asset_prefix), encoding="utf-8") + keys = build_keys(rig, audio=Path(audio) if audio else None, fps=fps, duration=duration, start=start) + (out_dir / "rig-keys.json").write_text(json.dumps(keys, indent=1, ensure_ascii=False) + "\n", encoding="utf-8") + (out_dir / "rig-keys.js").write_text(render_keys_js(keys, prefix=prefix), encoding="utf-8") + print(json.dumps({"ok": True, "parts": len(rig["parts"]), "groups": list(rig["groups"]), + "mouth_keys": sum(len(t["keys"]) for t in keys["mouth"]), + "blink_keys": sum(len(t["keys"]) for t in keys["blink"]), + "sway_keys": sum(len(t["keys"]) for t in keys["sway"]), "out_dir": str(out_dir)}, ensure_ascii=False)) + return 0 + + +def main(argv: list[str] | None = None) -> int: + parser = argparse.ArgumentParser(description="Build rig.json + HTML runtime + GSAP keys from matched parts.") + add_arguments(parser) + args = parser.parse_args(argv) + return run(catalog=args.catalog, match_dir=args.match_dir, out_dir=args.out_dir, audio=args.audio, + duration=args.duration, fps=args.fps, start=args.start, prefix=args.prefix, asset_prefix=args.asset_prefix) + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/tests/packaging/test_package_surface.py b/tests/packaging/test_package_surface.py index d7fdb556..8d00cb12 100644 --- a/tests/packaging/test_package_surface.py +++ b/tests/packaging/test_package_surface.py @@ -31,8 +31,11 @@ "gen", "generate_image", "inspect", + "match", + "parts_gen", "prepare", "preview", + "rig", "score", "serve_curation", "slice_sheet", diff --git a/tests/parts/__init__.py b/tests/parts/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/tests/parts/test_parts_contract.py b/tests/parts/test_parts_contract.py new file mode 100644 index 00000000..e1d55804 --- /dev/null +++ b/tests/parts/test_parts_contract.py @@ -0,0 +1,258 @@ +# SPDX-License-Identifier: Apache-2.0 +"""Parts rig contract on synthetic shapes: catalog validation, pixel registration +gate (pass + fail-loud), and rig/key export determinism. No real character data.""" + +from __future__ import annotations + +import json +from pathlib import Path + +import pytest +from PIL import Image, ImageDraw + +from sprite_gen.parts import catalog as cat +from sprite_gen.parts import match, rig + +CANVAS = (160, 200) + + +def _catalog(tmp: Path, **overrides) -> dict: + data = { + "kind": cat.KIND, "version": cat.VERSION, "character": "synthetic", "base": "base.png", + "canvas": {"width": CANVAS[0], "height": CANVAS[1]}, "chroma_key": "green", + "groups": {"head": {"pivot": [80, 120]}}, + "parts": [ + {"id": "body", "z": 0, "bbox": [30, 100, 100, 100], "pivot": [80, 150], "prompt": "the body"}, + {"id": "face", "z": 1, "bbox": [40, 20, 80, 90], "pivot": [80, 65], "group": "head", "prompt": "the face"}, + {"id": "mouth", "z": 2, "bbox": [65, 80, 30, 14], "pivot": [80, 87], "group": "head", "prompt": "the mouth", + "variants": {"default": "", "closed": "lips together", "open": "mouth open"}}, + {"id": "eyelid_l", "z": 3, "bbox": [50, 45, 20, 10], "pivot": [60, 50], "group": "head", "prompt": "left eyelid", + "variants": {"default": "", "open": "", "half": "half closed", "closed": "closed"}}, + ], + } + data.update(overrides) + return data + + +def _draw_base(path: Path) -> None: + img = Image.new("RGBA", CANVAS, (0, 0, 0, 0)) + d = ImageDraw.Draw(img) + d.rectangle((30, 100, 129, 199), fill=(40, 80, 200, 255)) # body + d.ellipse((40, 20, 119, 109), fill=(250, 220, 200, 255)) # face + d.rectangle((65, 80, 94, 93), fill=(200, 60, 80, 255)) # mouth + d.rectangle((50, 45, 69, 54), fill=(30, 30, 30, 255)) # eyelid_l + img.save(path) + + +def _part(path: Path, size: tuple[int, int], color: tuple[int, int, int], *, shape: str = "rect", pad: int = 6) -> None: + """A part drawn alone on transparency, padded so trimming + placement is exercised.""" + img = Image.new("RGBA", (size[0] + pad * 2, size[1] + pad * 2), (0, 0, 0, 0)) + d = ImageDraw.Draw(img) + box = (pad, pad, pad + size[0] - 1, pad + size[1] - 1) + (d.ellipse if shape == "ellipse" else d.rectangle)(box, fill=color + (255,)) + img.save(path) + + +@pytest.fixture +def workspace(tmp_path: Path) -> Path: + _draw_base(tmp_path / "base.png") + (tmp_path / "catalog.json").write_text(json.dumps(_catalog(tmp_path)), encoding="utf-8") + parts = tmp_path / "parts" + parts.mkdir() + _part(parts / "body.png", (100, 100), (40, 80, 200)) + _part(parts / "face.png", (80, 90), (250, 220, 200), shape="ellipse") + _part(parts / "mouth.png", (30, 14), (200, 60, 80)) + _part(parts / "mouth__closed.png", (30, 14), (200, 60, 80)) + _part(parts / "mouth__open.png", (30, 20), (120, 20, 30)) + _part(parts / "eyelid_l.png", (20, 10), (30, 30, 30)) + for v in ("open", "half", "closed"): + _part(parts / f"eyelid_l__{v}.png", (20, 10), (30, 30, 30)) + return tmp_path + + +# --- catalog ----------------------------------------------------------------- + +def test_valid_catalog_has_no_errors(workspace: Path) -> None: + assert cat.validate_catalog(_catalog(workspace)) == [] + + +@pytest.mark.parametrize("mutate, needle", [ + (lambda c: c["parts"].append(dict(c["parts"][0])), "duplicate part id"), + (lambda c: c["parts"][1].__setitem__("z", 0), "duplicate draw order"), + (lambda c: c["parts"][0].__setitem__("bbox", [30, 100, 200, 100]), "exceeds canvas"), + (lambda c: c["parts"][0].__setitem__("pivot", [0, 0]), "inside bbox"), + (lambda c: c["parts"][1].__setitem__("group", "tail"), "unknown group"), + (lambda c: c["parts"][2].__setitem__("variants", {"open": ""}), "must include 'default'"), + (lambda c: c["parts"][0].__setitem__("bbox", [30.0, 100, 100, 100]), "integers"), + (lambda c: c.__setitem__("chroma_key", "blue"), "chroma_key"), +]) +def test_catalog_violations_are_named(workspace: Path, mutate, needle: str) -> None: + data = _catalog(workspace) + mutate(data) + errors = cat.validate_catalog(data) + assert errors and any(needle in e for e in errors), errors + + +def test_jobs_are_draw_ordered_and_named(workspace: Path) -> None: + names = [cat.job_name(j["part"], j["variant"]) for j in cat.jobs(_catalog(workspace))] + assert names[:3] == ["body", "face", "mouth"] + assert "mouth__open" in names and "eyelid_l__closed" in names + + +# --- match ------------------------------------------------------------------- + +def test_match_registers_synthetic_parts_and_passes(workspace: Path) -> None: + report = match.match_parts(workspace / "catalog.json", workspace / "parts") + assert report["ok"], report["failed"] + by = {r["job"]: r for r in report["parts"]} + assert by["body"]["placement"]["x"] == 30 and by["body"]["placement"]["y"] == 100 + assert by["mouth"]["placement"]["scale"] == 1.0 + assert by["mouth__open"]["inherited_from"] == "mouth" + assert report["composite"]["score"] <= 0.02 and report["composite"]["coverage"] >= 0.97 + assert (workspace / "parts" / "placed" / "face.png").is_file() + + +def test_match_fails_loud_on_wrong_part(workspace: Path) -> None: + _part(workspace / "parts" / "face.png", (80, 90), (10, 200, 10), shape="ellipse") # wrong colour + report = match.match_parts(workspace / "catalog.json", workspace / "parts") + assert not report["ok"] + assert "face" in report["failed"] + face = next(r for r in report["parts"] if r["job"] == "face") + assert "exceeds tolerance" in face["error"] + + +def test_match_reports_missing_candidate(workspace: Path) -> None: + (workspace / "parts" / "body.png").unlink() + report = match.match_parts(workspace / "catalog.json", workspace / "parts") + assert not report["ok"] and "body" in report["failed"] + assert next(r for r in report["parts"] if r["job"] == "body")["error"] == "missing candidate" + + +# --- rig --------------------------------------------------------------------- + +def test_rig_refuses_unmatched_report(workspace: Path) -> None: + _part(workspace / "parts" / "face.png", (80, 90), (10, 200, 10), shape="ellipse") + match.match_parts(workspace / "catalog.json", workspace / "parts") + with pytest.raises(SystemExit, match="not passing"): + rig.build_rig(workspace / "catalog.json", workspace / "parts") + + +def test_rig_export_is_deterministic(workspace: Path) -> None: + match.match_parts(workspace / "catalog.json", workspace / "parts") + out_a = workspace / "out_a" + out_b = workspace / "out_b" + for out in (out_a, out_b): + assert rig.run(catalog=workspace / "catalog.json", match_dir=workspace / "parts", out_dir=out, duration=12.0) == 0 + for name in ("rig.json", "rig.html", "rig-keys.js", "rig-keys.json"): + assert (out_a / name).read_bytes() == (out_b / name).read_bytes(), name + data = json.loads((out_a / "rig.json").read_text()) + assert [p["id"] for p in data["parts"]] == ["body", "face", "mouth", "eyelid_l"] + assert data["groups"]["head"]["members"] == ["face", "mouth", "eyelid_l"] + keys = json.loads((out_a / "rig-keys.json").read_text()) + assert keys["mouth"] == [] # no audio → no mouth track + assert keys["blink"][0]["part"] == "eyelid_l" and len(keys["blink"][0]["keys"]) >= 4 + assert keys["sway"][0]["group"] == "head" + html = (out_a / "rig.html").read_text() + assert 'id="rig-mouth__open"' in html and "opacity:0" in html and 'id="rig-g-head"' in html + js = (out_a / "rig-keys.js").read_text() + assert "window.__rigKeys" in js and "rotation" in js + + +def test_mouth_quantization_uses_available_variants() -> None: + assert rig.mouth_variant(0.0, {"default", "closed", "open"}) == "closed" + assert rig.mouth_variant(0.5, {"default", "closed", "open"}) == "open" + assert rig.mouth_variant(0.2, {"default", "closed", "half", "open"}) == "half" + assert rig.mouth_variant(0.9, {"default", "closed", "open", "o"}) == "o" + assert rig.mouth_variant(0.9, {"default", "closed", "open"}) == "open" + + +def test_gen_references_alpha_failures_and_partial_report(workspace: Path, monkeypatch) -> None: + from types import SimpleNamespace + from sprite_gen.parts import parts_gen + calls = [] + fail = {"face"} + + def provider(name, prompt, out, **kwargs): + calls.append((out.stem, kwargs)) + assert len(kwargs["refs"]) == 2 + assert Image.open(kwargs["refs"][0]).size == CANVAS + assert Image.open(kwargs["refs"][1]).width <= CANVAS[0] + assert kwargs["transparent"] and kwargs["chroma_key"] == "green" + if out.stem in fail: + Image.new("RGBA", (20, 20), (220, 200, 180, 255)).save(out) + else: + _part(out, (10, 10), (200, 100, 80)) + return SimpleNamespace(raw=out, elapsed_seconds=0, provider=name) + + monkeypatch.setattr(parts_gen, "generate_image", provider) + out = workspace / "generated" + first = parts_gen.generate_parts(workspace / "catalog.json", out, workers=2) + assert not first["ok"] and first["failed"] == ["face"] + assert len(calls) == len(cat.jobs(_catalog(workspace))) + second = parts_gen.generate_parts(workspace / "catalog.json", out, only=["body"]) + assert not second["ok"] and second["failed"] == ["face"] + assert second["ran"] == ["body"] and len(second["jobs"]) == len(first["jobs"]) + fail.clear() + repaired = parts_gen.generate_parts(workspace / "catalog.json", out, only=["face"]) + assert repaired["ok"] and repaired["failed"] == [] + + +def test_rms_uses_exact_frame_boundaries_for_long_audio(monkeypatch) -> None: + from types import SimpleNamespace + from sprite_gen._deps import np + # 16000 / 30 is fractional: truncating each window adds 5 frames in 141s. + samples = np.zeros(16000 * 141, dtype=np.int16) + samples[16000 * 140:] = 10000 + monkeypatch.setattr(rig.shutil, "which", lambda _: "ffmpeg") + monkeypatch.setattr(rig.subprocess, "run", lambda *a, **k: SimpleNamespace(stdout=samples.tobytes())) + env = rig.rms_envelope(Path("synthetic.wav"), 30) + assert len(env) == 141 * 30 + assert next(i for i, value in enumerate(env) if value > 0) == 140 * 30 - 1 + samples = samples[:100] + assert len(rig.rms_envelope(Path("short.wav"), 30)) == 1 + + +def test_rig_preserves_interleaved_group_draw_order(workspace: Path) -> None: + from html.parser import HTMLParser + data = _catalog(workspace) + # A foreground ungrouped layer interrupts two head layers. + data["parts"][2].pop("group") + (workspace / "catalog.json").write_text(json.dumps(data)) + match.match_parts(workspace / "catalog.json", workspace / "parts") + model = rig.build_rig(workspace / "catalog.json", workspace / "parts") + wrappers, layers = [], [] + class Parser(HTMLParser): + def handle_starttag(self, tag, attrs): + attrs = dict(attrs) + if attrs.get("class") == "rig-group": + wrappers.append(attrs) + if tag == "img": + layers.append(attrs["id"]) + Parser().feed(rig.render_html(model)) + assert layers.index("rig-face") < layers.index("rig-mouth") < layers.index("rig-eyelid_l") + assert len(wrappers) == 2 + assert all(w["data-rig-group"] == "head" for w in wrappers) + assert "z-index:1;" in wrappers[0]["style"] and "z-index:3;" in wrappers[1]["style"] + assert len({w["id"] for w in wrappers}) == 2 + + +def test_rig_refuses_deleted_placed_layer(workspace: Path) -> None: + match.match_parts(workspace / "catalog.json", workspace / "parts") + (workspace / "parts" / "placed" / "mouth__open.png").unlink() + with pytest.raises(SystemExit, match="missing placed layer: mouth__open"): + rig.build_rig(workspace / "catalog.json", workspace / "parts") + + +def test_closed_blink_hides_the_underlying_eye() -> None: + model = {"groups": {}, "parts": [ + {"id": "eye_l", "variants": {"default": "eye.png"}}, + {"id": "eyelid_l", "variants": dict.fromkeys(["default", "open", "half", "closed"], "lid.png")}, + ]} + keys = rig.build_keys(model, audio=None, fps=30, duration=5) + track = keys["blink"][0] + assert track["eye_part"] == "eye_l" + js = rig.render_keys_js(keys) + closed = next(k for k in track["keys"] if k["variant"] == "closed") + opened = next(k for k in track["keys"] if k["variant"] == "open") + assert f'"eye_l", {{ opacity: 0 }}, start + {closed["t"]}' in js + assert f'"eye_l", {{ opacity: 1 }}, start + {opened["t"]}' in js