#!/usr/bin/env python3
from __future__ import annotations
import argparse
import hashlib
import itertools
import json
import shutil
import sys
from datetime import datetime, timezone
from pathlib import Path
from typing import Any, Protocol
ROOT = Path(__file__).resolve().parents[1]
if str(ROOT) not in sys.path:
sys.path.insert(0, str(ROOT))
from packages.pet_package_schema import ALLOWED_ACTIONS
from services.ai.model_router import load_image_generator
from services.ai.qwen_client import QwenReviewClient
from services.pet_builder.action_atlas import (
FRAMES_PER_ACTION,
compose_action_frame_atlas,
_read_rgba_png,
_write_rgba_png,
)
from services.pet_builder.action_semantics import evaluate_action_semantics
from services.pet_builder.size_proportion_qa import evaluate_size_proportions
from services.pet_builder.photo_generation_worker import (
_build_review,
_candidate_visual_detail_score,
_final_contact_sheet_visual_review_passed,
_frame_rejection_reason,
_safe_frame_report,
_safe_generation_usage,
_safe_transparency_report,
_slugify,
_transparency_rejection_reason,
_visual_review_passed,
_write_install_package,
_write_json,
)
from services.pet_builder.qa_media import write_animation_previews, write_contact_sheet_png
from scripts.repair_walk_action import (
_build_validation,
_canonical_reference_path,
_frame_report_for_candidate,
_process_candidate,
_read_all_action_frames,
_read_json,
_relative_or_abs,
_require_image_bytes,
_utc_stamp,
)
IDLE_REPAIR_ACTION = "idle"
IDLE_REPAIR_GENERATED_FRAME_COUNT = FRAMES_PER_ACTION
IDLE_LOCAL_DERIVATION_MODE = "derive_standing_idle_from_walk_frame"
IDLE_LOCAL_FROM_IDLE_FRAME_MODE = "derive_micro_idle_from_existing_idle_frame"
class ImageGenerator(Protocol):
provider: str
model: str
def generate_action_frame(
self,
*,
action: str,
frame_index: int,
pet_name: str,
notes: str,
analysis: dict[str, Any],
images: list[dict[str, Any]],
canonical_image_bytes: bytes,
previous_frame_image_bytes: bytes | None = None,
candidate_index: int = 0,
size: str = "1024*1024",
) -> dict[str, Any]:
...
class ActionReviewer(Protocol):
def review_action_frames(
self,
*,
pet_name: str,
action: str,
image_bytes: bytes,
image_mime: str,
) -> dict[str, Any]:
...
def review_contact_sheet(
self,
*,
pet_name: str,
image_bytes: bytes,
image_mime: str,
) -> dict[str, Any]:
...
def main() -> int:
parser = argparse.ArgumentParser(description="Repair only the idle action row in a generated pet build.")
parser.add_argument("--build-dir", required=True, type=Path, help="Source build directory with qa/action-frames.")
parser.add_argument(
"--output-root",
default=str(ROOT / "outputs" / "pet_builds"),
type=Path,
help="Directory where the repaired build will be written.",
)
parser.add_argument("--pet-name", default="", help="Override pet display name.")
parser.add_argument("--notes", default="quiet memorial desktop pet")
parser.add_argument("--candidates-per-frame", type=int, default=2)
parser.add_argument("--species", default="cat")
parser.add_argument("--base-color", default="gray and white")
parser.add_argument("--eye-color", default="copper")
parser.add_argument(
"--skip-visual-review",
action="store_true",
help="Skip Qwen visual review. Intended when doing local deterministic QA plus manual contact-sheet review.",
)
parser.add_argument(
"--derive-from-walk",
action="store_true",
help="Avoid external generation and derive a temporary standing idle row from the existing walk frame.",
)
parser.add_argument(
"--derive-from-idle-frame",
type=int,
default=None,
help="Avoid external generation and derive a stable micro-idle row from one existing idle frame index.",
)
parser.add_argument(
"--seed-candidate-dir",
type=Path,
default=None,
help="Optional directory containing pre-generated idle--.png candidates to use before new generation.",
)
args = parser.parse_args()
build_dir = repair_idle_action(
build_dir=args.build_dir,
output_root=args.output_root,
pet_name=args.pet_name,
notes=args.notes,
candidates_per_frame=args.candidates_per_frame,
analysis={
"species": args.species,
"base_color": args.base_color,
"eye_color": args.eye_color,
},
action_reviewer=None if args.skip_visual_review else QwenReviewClient.from_env(),
derive_from_walk=args.derive_from_walk,
derive_from_idle_frame=args.derive_from_idle_frame,
seed_candidate_dir=args.seed_candidate_dir,
)
print(
json.dumps(
{
"status": "ok",
"build_dir": _relative_or_abs(build_dir),
"idle_contact_sheet": _relative_or_abs(build_dir / "qa" / "action-contact-sheets" / "idle.png"),
"contact_sheet": _relative_or_abs(build_dir / "qa" / "contact-sheet.png"),
"validation": _relative_or_abs(build_dir / "validation.json"),
"package": _relative_or_abs(build_dir / "package" / "pet.json"),
},
ensure_ascii=False,
indent=2,
)
)
return 0
def repair_idle_action(
*,
build_dir: Path,
output_root: Path,
pet_name: str = "",
notes: str = "quiet memorial desktop pet",
candidates_per_frame: int = 2,
analysis: dict[str, Any] | None = None,
image_generator: ImageGenerator | None = None,
action_reviewer: ActionReviewer | None = None,
derive_from_walk: bool = False,
derive_from_idle_frame: int | None = None,
seed_candidate_dir: Path | None = None,
) -> Path:
source_build = build_dir.resolve()
if not source_build.exists():
raise FileNotFoundError(f"source build not found: {source_build}")
source_manifest_path = source_build / "manifest.json"
if not source_manifest_path.exists():
raise FileNotFoundError(f"manifest.json not found: {source_manifest_path}")
source_manifest = _read_json(source_manifest_path)
display_name = pet_name.strip() or str(source_manifest.get("display_name") or source_manifest.get("pet_id") or "pet")
pet_id = str(source_manifest.get("pet_id") or _slugify(display_name))
output_root = output_root.resolve()
output_root.mkdir(parents=True, exist_ok=True)
repaired_build_id = f"{source_manifest.get('build_id', source_build.name)}-idle-repair-{_utc_stamp()}"
repaired_build = output_root / repaired_build_id
if repaired_build.exists():
raise FileExistsError(f"output build already exists: {repaired_build}")
shutil.copytree(source_build, repaired_build)
qa_dir = repaired_build / "qa"
action_frame_dir = qa_dir / "action-frames"
raw_candidate_dir = qa_dir / "action-frame-raw-candidates"
candidate_dir = qa_dir / "action-frame-candidates"
action_contact_dir = qa_dir / "action-contact-sheets"
action_review_dir = qa_dir / "action-reviews"
for path in (action_frame_dir, raw_candidate_dir, candidate_dir, action_contact_dir, action_review_dir):
path.mkdir(parents=True, exist_ok=True)
canonical_reference_bytes = _canonical_reference_path(qa_dir).read_bytes()
if derive_from_walk and derive_from_idle_frame is not None:
raise ValueError("choose only one local idle derivation mode")
local_derivation_mode = (
IDLE_LOCAL_FROM_IDLE_FRAME_MODE
if derive_from_idle_frame is not None
else IDLE_LOCAL_DERIVATION_MODE
if derive_from_walk
else None
)
image_generator = image_generator or (None if local_derivation_mode else load_image_generator())
candidates_per_frame = max(1, min(int(candidates_per_frame), 3))
analysis = analysis or {"species": "cat", "base_color": "gray and white", "eye_color": "copper"}
candidate_failures: list[dict[str, Any]] = []
frame_pools: dict[int, list[dict[str, Any]]] = {}
if seed_candidate_dir:
_load_seed_candidates(
seed_candidate_dir=seed_candidate_dir,
candidate_dir=candidate_dir,
frame_pools=frame_pools,
candidate_failures=candidate_failures,
)
if derive_from_walk:
frame_pools = _derive_idle_candidates_from_walk_frames(
action_frame_dir=action_frame_dir,
candidate_dir=candidate_dir,
)
elif derive_from_idle_frame is not None:
frame_pools = _derive_idle_candidates_from_existing_idle_frame(
action_frame_dir=action_frame_dir,
candidate_dir=candidate_dir,
source_frame_index=derive_from_idle_frame,
)
else:
if image_generator is None:
raise ValueError("image generator is required when derive_from_walk is false")
for frame_index in range(IDLE_REPAIR_GENERATED_FRAME_COUNT):
frame_pools.setdefault(frame_index, [])
for candidate_index in range(candidates_per_frame):
if len(frame_pools[frame_index]) >= candidates_per_frame:
break
try:
generation = image_generator.generate_action_frame(
action=IDLE_REPAIR_ACTION,
frame_index=frame_index,
pet_name=display_name,
notes=notes.strip(),
analysis=analysis,
images=[],
canonical_image_bytes=canonical_reference_bytes,
previous_frame_image_bytes=None,
candidate_index=candidate_index,
size="1024*1024",
)
except Exception as exc:
candidate_failures.append(
{
"action": IDLE_REPAIR_ACTION,
"frame_index": frame_index,
"candidate_index": candidate_index,
"reason": "provider_generation_failed",
"provider_error": str(exc)[:260],
}
)
continue
raw_bytes = _require_image_bytes(generation, f"idle-{frame_index}-{candidate_index}")
(raw_candidate_dir / f"idle-{frame_index}-{candidate_index}.png").write_bytes(raw_bytes)
candidate = _process_candidate(
action=IDLE_REPAIR_ACTION,
frame_index=frame_index,
candidate_index=candidate_index,
raw_bytes=raw_bytes,
candidate_path=candidate_dir / f"idle-{frame_index}-{candidate_index}.png",
generation=generation,
)
if not candidate["ok"]:
candidate_failures.append(candidate["failure"])
continue
pose_rejection = _idle_pose_rejection_reason(candidate["frame_report"])
if pose_rejection:
candidate_failures.append(
{
"action": IDLE_REPAIR_ACTION,
"frame_index": frame_index,
"candidate_index": candidate_index,
"reason": pose_rejection,
"frame_report": _safe_frame_report(candidate["frame_report"]),
}
)
continue
frame_pools[frame_index].append(candidate)
if not frame_pools[frame_index]:
raise ValueError(f"no acceptable idle candidate for frame {frame_index}")
selected_candidates = _select_idle_candidates(
frame_pools,
source_frame_qa=source_manifest.get("source", {}).get("frame_qa"),
)
for frame_index, candidate in enumerate(selected_candidates):
(action_frame_dir / f"idle-{frame_index}.png").write_bytes(candidate["image_bytes"])
action_frames = _read_all_action_frames(action_frame_dir)
atlas = compose_action_frame_atlas(action_frames)
_annotate_idle_candidates(atlas.qa, selected_candidates)
action_semantic_qa = evaluate_action_semantics(atlas.qa)
action_semantic_qa = _preserve_non_idle_source_semantics(action_semantic_qa, source_manifest)
if action_semantic_qa.get("ok") is not True:
_write_json(qa_dir / "frame-qa.json", atlas.qa)
_write_json(qa_dir / "action-semantic-qa.json", action_semantic_qa)
_write_json(qa_dir / "candidate-failures.json", {"failures": candidate_failures})
raise ValueError("idle repair failed action semantic QA")
idle_atlas = compose_action_frame_atlas({"idle": [item["image_bytes"] for item in selected_candidates]})
visual_review = None
if action_reviewer:
visual_review = action_reviewer.review_action_frames(
pet_name=display_name,
action=IDLE_REPAIR_ACTION,
image_bytes=idle_atlas.image_bytes,
image_mime="image/png",
)
_write_json(
action_review_dir / "idle.json",
{
"action": IDLE_REPAIR_ACTION,
"attempt": 1,
"ok": _visual_review_passed(visual_review, action=IDLE_REPAIR_ACTION),
"semantic_qa": action_semantic_qa,
"visual_review": visual_review,
},
)
if not _visual_review_passed(visual_review, action=IDLE_REPAIR_ACTION):
_write_json(qa_dir / "frame-qa.json", atlas.qa)
_write_json(qa_dir / "action-semantic-qa.json", action_semantic_qa)
_write_json(qa_dir / "candidate-failures.json", {"failures": candidate_failures})
raise ValueError("idle repair failed visual QA")
spritesheet_bytes = atlas.image_bytes
(repaired_build / "spritesheet.png").write_bytes(spritesheet_bytes)
(action_contact_dir / "idle.png").write_bytes(idle_atlas.image_bytes)
contact_sheet_media = write_contact_sheet_png(
spritesheet_bytes=spritesheet_bytes,
output_path=qa_dir / "contact-sheet.png",
)
final_visual_review = None
if action_reviewer and hasattr(action_reviewer, "review_contact_sheet"):
final_visual_review = action_reviewer.review_contact_sheet(
pet_name=display_name,
image_bytes=(qa_dir / "contact-sheet.png").read_bytes(),
image_mime="image/png",
)
_write_json(qa_dir / "final-visual-review.json", final_visual_review)
if not _final_contact_sheet_visual_review_passed(final_visual_review):
_write_json(qa_dir / "frame-qa.json", atlas.qa)
_write_json(qa_dir / "action-semantic-qa.json", action_semantic_qa)
_write_json(qa_dir / "candidate-failures.json", {"failures": candidate_failures})
raise ValueError("idle repair failed final contact sheet visual QA")
preview_media = write_animation_previews(
action_frames=action_frames,
output_dir=qa_dir / "previews",
)
size_proportion_qa = evaluate_size_proportions(atlas.qa)
_write_json(qa_dir / "size-proportion-qa.json", size_proportion_qa)
checksum = hashlib.sha256(spritesheet_bytes).hexdigest()
package_manifest = _write_install_package(
build_dir=repaired_build,
pet_id=pet_id,
pet_name=display_name,
spritesheet_bytes=spritesheet_bytes,
)
repaired_manifest = _updated_manifest(
source_manifest=source_manifest,
repaired_build_id=repaired_build_id,
pet_id=pet_id,
display_name=display_name,
checksum=checksum,
image_generator=image_generator,
selected_candidates=selected_candidates,
atlas_qa=atlas.qa,
action_semantic_qa=action_semantic_qa,
size_proportion_qa=size_proportion_qa,
candidates_per_frame=candidates_per_frame,
candidate_failures=candidate_failures,
package_manifest=package_manifest,
contact_sheet_media=contact_sheet_media,
preview_media=preview_media,
)
validation = _build_validation(repaired_manifest)
review = _build_review(frame_qa=atlas.qa, action_semantic_qa=action_semantic_qa)
_write_json(repaired_build / "manifest.json", repaired_manifest)
_write_json(repaired_build / "validation.json", validation)
_write_json(qa_dir / "frame-qa.json", atlas.qa)
_write_json(qa_dir / "action-semantic-qa.json", action_semantic_qa)
_write_json(qa_dir / "candidate-failures.json", {"failures": candidate_failures})
_write_json(qa_dir / "review.json", review)
_write_json(
qa_dir / "idle-repair-summary.json",
{
"ok": True,
"repaired_action": IDLE_REPAIR_ACTION,
"source_build_id": source_manifest.get("build_id"),
"repaired_build_id": repaired_build_id,
"provider": getattr(image_generator, "provider", "unknown"),
"model": getattr(image_generator, "model", "unknown"),
"derivation_mode": local_derivation_mode,
"selected_candidate_indexes": [item["candidate_index"] for item in selected_candidates],
"size_proportion_qa": size_proportion_qa,
"visual_review": visual_review,
"final_visual_review": final_visual_review,
"candidate_failures": len(candidate_failures),
},
)
return repaired_build
def _select_idle_candidates(
frame_pools: dict[int, list[dict[str, Any]]],
*,
source_frame_qa: dict[str, Any] | None = None,
) -> list[dict[str, Any]]:
pools = []
for frame_index in range(IDLE_REPAIR_GENERATED_FRAME_COUNT):
pool = frame_pools.get(frame_index) or []
if not pool:
raise ValueError(f"no idle candidates for frame {frame_index}")
pools.append(pool)
best_combo: tuple[dict[str, Any], ...] | None = None
best_key: tuple[float, ...] | None = None
for combo in itertools.product(*pools):
size_qa = evaluate_size_proportions(_idle_combo_frame_qa(combo, source_frame_qa=source_frame_qa))
idle_row = next(row for row in size_qa["rows"] if row.get("action") == IDLE_REPAIR_ACTION)
visual_score = sum(float(item.get("score") or 0.0) for item in combo)
area_cv = float(idle_row.get("area_cv") or 0.0)
area_max_min_ratio = float(idle_row.get("area_max_min_ratio") or 0.0)
relative = idle_row.get("relative_to_sit_area")
relative_delta = 0.0
if isinstance(relative, (int, float)):
relative_delta = abs(float(relative) - 1.0)
row_key = (
1.0 if size_qa.get("ok") is True else 0.0,
1.0 if idle_row.get("ok") is True else 0.0,
-area_cv,
-area_max_min_ratio,
-relative_delta,
visual_score,
)
if best_key is None or row_key > best_key:
best_key = row_key
best_combo = tuple(combo)
if best_combo is None:
raise ValueError("idle candidate selection failed")
return list(best_combo)
def _idle_combo_frame_qa(
combo: tuple[dict[str, Any], ...],
*,
source_frame_qa: dict[str, Any] | None,
) -> dict[str, Any]:
rows = [
{
"action": IDLE_REPAIR_ACTION,
"input_frames": len(combo),
"selected_frames": len(combo),
"frames": [
dict(item.get("frame_report", {}), frame=frame_index)
for frame_index, item in enumerate(combo)
],
}
]
if isinstance(source_frame_qa, dict):
for row in source_frame_qa.get("rows", []):
if isinstance(row, dict) and row.get("action") != IDLE_REPAIR_ACTION:
rows.append(json.loads(json.dumps(row, ensure_ascii=False)))
return {
"ok": True,
"failure_reason": None,
"cell_width": 192,
"cell_height": 208,
"frames_per_action": IDLE_REPAIR_GENERATED_FRAME_COUNT,
"rows": rows,
}
def _load_seed_candidates(
*,
seed_candidate_dir: Path,
candidate_dir: Path,
frame_pools: dict[int, list[dict[str, Any]]],
candidate_failures: list[dict[str, Any]],
) -> None:
source_dir = seed_candidate_dir.resolve()
if not source_dir.exists():
raise FileNotFoundError(f"seed candidate dir not found: {source_dir}")
candidate_dir.mkdir(parents=True, exist_ok=True)
for source_path in sorted(source_dir.glob("idle-*-*.png")):
parts = source_path.stem.split("-")
if len(parts) != 3 or parts[0] != IDLE_REPAIR_ACTION:
continue
try:
frame_index = int(parts[1])
candidate_index = int(parts[2])
except ValueError:
continue
if frame_index < 0 or frame_index >= IDLE_REPAIR_GENERATED_FRAME_COUNT:
continue
image_bytes = source_path.read_bytes()
target_path = candidate_dir / source_path.name
if source_path.resolve() != target_path.resolve():
shutil.copy2(source_path, target_path)
frame_report = _frame_report_for_candidate(IDLE_REPAIR_ACTION, image_bytes)
rejection_reason = _frame_rejection_reason(frame_report) or _idle_pose_rejection_reason(frame_report)
if rejection_reason:
candidate_failures.append(
{
"action": IDLE_REPAIR_ACTION,
"frame_index": frame_index,
"candidate_index": candidate_index,
"reason": rejection_reason,
"frame_report": _safe_frame_report(frame_report),
"source": "seed_candidate",
}
)
continue
base_score = 100.0 - candidate_index * 0.01
frame_pools.setdefault(frame_index, []).append(
{
"ok": True,
"candidate_index": candidate_index,
"image_bytes": image_bytes,
"generation": {
"request_id": "",
"prompt_sha256": "",
"usage": {"seed_candidate": True, "external_image_count": 0},
"source": "seed_candidate",
"source_path": str(source_path),
},
"transparency": {
"method": "existing_transparent_seed_candidate",
"source": str(source_path),
},
"frame_report": frame_report,
"score": base_score + _candidate_visual_detail_score(image_bytes),
"base_score": base_score,
}
)
def _local_idle_source_frame_indexes() -> list[int]:
return [0] * IDLE_REPAIR_GENERATED_FRAME_COUNT
def _derive_idle_candidates_from_walk_frames(*, action_frame_dir: Path, candidate_dir: Path) -> dict[int, list[dict[str, Any]]]:
source_path = action_frame_dir / "walk-0.png"
if not source_path.exists():
raise FileNotFoundError(f"walk frame for local idle derivation not found: {source_path}")
source_bytes = source_path.read_bytes()
vertical_offsets = [0, -1, 0, -1]
frame_pools: dict[int, list[dict[str, Any]]] = {}
for frame_index, source_frame_index in enumerate(_local_idle_source_frame_indexes()):
offset = vertical_offsets[frame_index % len(vertical_offsets)]
derived_bytes = _shift_png_frame(source_bytes, dx=0, dy=offset)
candidate_path = candidate_dir / f"idle-{frame_index}-local.png"
candidate_path.write_bytes(derived_bytes)
frame_report = _frame_report_for_candidate(IDLE_REPAIR_ACTION, derived_bytes)
base_score = 100.0 - abs(offset)
frame_pools[frame_index] = [
{
"ok": True,
"candidate_index": 0,
"image_bytes": derived_bytes,
"generation": {
"request_id": f"local-derived-from:walk-{source_frame_index}",
"prompt_sha256": "",
"usage": {
"derived_local": True,
"external_image_count": 0,
},
"source": IDLE_LOCAL_DERIVATION_MODE,
"source_path": str(source_path),
"source_frame_index": source_frame_index,
},
"transparency": {
"method": IDLE_LOCAL_DERIVATION_MODE,
"source": str(source_path),
},
"frame_report": frame_report,
"score": base_score + _candidate_visual_detail_score(derived_bytes),
"base_score": base_score,
}
]
return frame_pools
def _derive_idle_candidates_from_existing_idle_frame(
*,
action_frame_dir: Path,
candidate_dir: Path,
source_frame_index: int,
) -> dict[int, list[dict[str, Any]]]:
if source_frame_index < 0 or source_frame_index >= IDLE_REPAIR_GENERATED_FRAME_COUNT:
raise ValueError(f"source idle frame index must be between 0 and {IDLE_REPAIR_GENERATED_FRAME_COUNT - 1}")
source_path = action_frame_dir / f"idle-{source_frame_index}.png"
if not source_path.exists():
raise FileNotFoundError(f"idle frame for local idle derivation not found: {source_path}")
source_bytes = source_path.read_bytes()
vertical_offsets = [0, -1, 0, -1]
frame_pools: dict[int, list[dict[str, Any]]] = {}
for frame_index in range(IDLE_REPAIR_GENERATED_FRAME_COUNT):
offset = vertical_offsets[frame_index % len(vertical_offsets)]
derived_bytes = _shift_png_frame(source_bytes, dx=0, dy=offset)
candidate_path = candidate_dir / f"idle-{frame_index}-local-idle-{source_frame_index}.png"
candidate_path.write_bytes(derived_bytes)
frame_report = _frame_report_for_candidate(IDLE_REPAIR_ACTION, derived_bytes)
base_score = 100.0 - abs(offset)
frame_pools[frame_index] = [
{
"ok": True,
"candidate_index": 0,
"image_bytes": derived_bytes,
"generation": {
"request_id": f"local-derived-from:idle-{source_frame_index}",
"prompt_sha256": "",
"usage": {
"derived_local": True,
"external_image_count": 0,
},
"source": IDLE_LOCAL_FROM_IDLE_FRAME_MODE,
"source_path": str(source_path),
"source_frame_index": source_frame_index,
},
"transparency": {
"method": IDLE_LOCAL_FROM_IDLE_FRAME_MODE,
"source": str(source_path),
},
"frame_report": frame_report,
"score": base_score + _candidate_visual_detail_score(derived_bytes),
"base_score": base_score,
}
]
return frame_pools
def _shift_png_frame(image_bytes: bytes, *, dx: int, dy: int) -> bytes:
width, height, pixels = _read_rgba_png(image_bytes)
shifted = [(0, 0, 0, 0)] * (width * height)
for y in range(height):
target_y = y + dy
if target_y < 0 or target_y >= height:
continue
for x in range(width):
target_x = x + dx
if target_x < 0 or target_x >= width:
continue
pixel = pixels[y * width + x]
if pixel[3] == 0:
continue
shifted[target_y * width + target_x] = pixel
return _write_rgba_png(width, height, shifted)
def _idle_pose_rejection_reason(frame_report: dict[str, Any]) -> str | None:
bbox = frame_report.get("source_bbox")
if not isinstance(bbox, list) or len(bbox) != 4:
return "idle_bbox_missing"
width = max(0, int(bbox[2]) - int(bbox[0]) + 1)
height = max(1, int(bbox[3]) - int(bbox[1]) + 1)
aspect_ratio = width / height
if aspect_ratio < 0.95:
return "idle_pose_too_seated_or_upright"
return None
def _preserve_non_idle_source_semantics(
action_semantic_qa: dict[str, Any],
source_manifest: dict[str, Any],
) -> dict[str, Any]:
source_semantics = source_manifest.get("source", {}).get("action_semantic_qa", {})
if not isinstance(source_semantics, dict) or source_semantics.get("ok") is not True:
return action_semantic_qa
preserved = json.loads(json.dumps(action_semantic_qa, ensure_ascii=False))
for row in preserved.get("rows", []):
if row.get("action") == IDLE_REPAIR_ACTION or row.get("ok") is True:
continue
row["preserved_from_source"] = True
row["source_semantics_ok"] = True
row["original_repair_run_reasons"] = list(row.get("reasons", []))
row["ok"] = True
row["reasons"] = []
preserved["ok"] = all(bool(row.get("ok")) for row in preserved.get("rows", []))
remaining_failures: set[str] = set()
if not preserved["ok"]:
for row in preserved.get("rows", []):
if row.get("ok") is not True:
remaining_failures.update(str(reason) for reason in row.get("reasons", []))
preserved["failure_reasons"] = sorted(remaining_failures)
return preserved
def _annotate_idle_candidates(frame_qa: dict[str, Any], selected_candidates: list[dict[str, Any]]) -> None:
for row in frame_qa.get("rows", []):
if row.get("action") != IDLE_REPAIR_ACTION:
continue
row["repair_mode"] = "post_drag_standing_idle"
row["generated_frame_count"] = IDLE_REPAIR_GENERATED_FRAME_COUNT
for frame in row.get("frames", []):
frame_index = int(frame.get("frame", 0))
if frame_index >= len(selected_candidates):
continue
selected = selected_candidates[frame_index]
frame["selected_candidate_index"] = int(selected["candidate_index"])
frame["candidate_score"] = float(selected["score"])
frame["base_score"] = float(selected["base_score"])
def _updated_manifest(
*,
source_manifest: dict[str, Any],
repaired_build_id: str,
pet_id: str,
display_name: str,
checksum: str,
image_generator: ImageGenerator | None,
selected_candidates: list[dict[str, Any]],
atlas_qa: dict[str, Any],
action_semantic_qa: dict[str, Any],
size_proportion_qa: dict[str, Any],
candidates_per_frame: int,
candidate_failures: list[dict[str, Any]],
package_manifest: dict[str, Any],
contact_sheet_media: dict[str, Any],
preview_media: dict[str, Any],
) -> dict[str, Any]:
manifest = json.loads(json.dumps(source_manifest, ensure_ascii=False))
now = datetime.now(timezone.utc).isoformat().replace("+00:00", "Z")
manifest["build_id"] = repaired_build_id
provider = str(getattr(image_generator, "provider", "local"))
model = str(getattr(image_generator, "model", IDLE_LOCAL_DERIVATION_MODE))
manifest["package_id"] = f"package-{_slugify(provider)}-{_slugify(pet_id)}-{_utc_stamp()}"
manifest["pet_id"] = pet_id
manifest["display_name"] = display_name
manifest["checksum"] = checksum
manifest["created_at"] = now
manifest["user_approved"] = False
manifest["spritesheet_path"] = "spritesheet.png"
manifest["package_ref"] = "package/pet.json"
source = manifest.setdefault("source", {})
source["image_generation_provider"] = provider
source["image_model"] = model
source["frame_qa"] = atlas_qa
source["action_semantic_qa"] = action_semantic_qa
source["size_proportion_qa"] = size_proportion_qa
source["output_asset_ref"] = "spritesheet.png"
source["output_mime"] = "image/png"
source["repaired_from_build_id"] = source_manifest.get("build_id")
source["repaired_action"] = IDLE_REPAIR_ACTION
source["idle_repair_derivation_mode"] = IDLE_LOCAL_DERIVATION_MODE if image_generator is None else None
action_request_ids = source.setdefault("action_request_ids", {})
action_request_ids[IDLE_REPAIR_ACTION] = [str(item["generation"].get("request_id", "")) for item in selected_candidates]
action_candidate_request_ids = source.setdefault("action_candidate_request_ids", {})
action_candidate_request_ids[IDLE_REPAIR_ACTION] = [[str(item["generation"].get("request_id", ""))] for item in selected_candidates]
action_prompt_sha256s = source.setdefault("action_prompt_sha256s", {})
action_prompt_sha256s[IDLE_REPAIR_ACTION] = [str(item["generation"].get("prompt_sha256", "")) for item in selected_candidates]
generation_usage = source.setdefault("generation_usage", {})
actions_usage = generation_usage.setdefault("actions", {})
actions_usage[IDLE_REPAIR_ACTION] = [_safe_generation_usage(item["generation"].get("usage", {})) for item in selected_candidates]
transparency = source.setdefault("transparency_processing", {})
actions_transparency = transparency.setdefault("actions", {})
if isinstance(actions_transparency, dict):
actions_transparency[IDLE_REPAIR_ACTION] = [item["transparency"] for item in selected_candidates]
candidate_policy = source.setdefault("candidate_policy", {})
candidate_policy["repair_mode"] = "smallest_failed_action"
candidate_policy["idle_repair_candidates_per_frame"] = candidates_per_frame
candidate_policy["idle_repair_selection"] = "post_drag_standing_idle"
candidate_policy["idle_repair_candidate_failures"] = len(candidate_failures)
candidate_policy.setdefault("final_semantic_repair_actions", [])
if IDLE_REPAIR_ACTION not in candidate_policy["final_semantic_repair_actions"]:
candidate_policy["final_semantic_repair_actions"].append(IDLE_REPAIR_ACTION)
hatch = source.setdefault("hatch_pet_pipeline", {})
hatch.update(
{
"contact_sheet_png": "qa/contact-sheet.png",
"preview_dir": "qa/previews",
"package_dir": "package",
"package_manifest": package_manifest,
"repair_mode": "smallest_failed_action",
"contact_sheet_media": contact_sheet_media,
"preview_media": preview_media,
}
)
return manifest
if __name__ == "__main__":
raise SystemExit(main())