""" mosketch_pipeline.py — the full pipeline as one script with subcommands. Pipeline: 1. Identify objects -> from the semantic file (no Qwen) 2. Classify ARAP vs. not -> `classify` subcommand (one Qwen call, all objects) 3. Narrate + deform -> `narrate` + `deform` subcommands, ONLY for objects marked ARAP in step 2 4. Render -> `render` subcommand; every object gets real trajectory translation; ARAP objects additionally get deformation on top Run steps individually for debugging, or use `full` to run everything for one sketch in one process — this loads the Qwen model ONCE and reuses it across classify/narrate/deform, instead of loading it 3 separate times. Examples: # step by step python mosketch_pipeline.py classify --model M --caption-file C --sketch-name S --semantic SEM --out deformation.json python mosketch_pipeline.py narrate --model M --caption-file C --sketch-name S --objects dog --out narratives.json python mosketch_pipeline.py deform --model M --svg S.svg --semantic SEM --traj T --narratives narratives.json --deformation deformation.json --out-dir . python mosketch_pipeline.py render --svg S.svg --semantic SEM --traj T --handles-dir . # everything at once, one model load python mosketch_pipeline.py full --model M --caption-file C --sketch-name S --svg S.svg --semantic SEM --traj T --out-dir . """ import argparse import json import os import re import sys import numpy as np import math import statistics from PIL import Image from datetime import datetime from lib import ( load_strokes_from_svg, load_semantic_assignments, filter_strokes, flatten_strokes, load_object, deduplicate_points, build_mesh, nearest_mesh_vertex, auto_select_handles_deduped, object_bbox_size, arap_deform, load_trajectories, bbox_deltas, get_caption, build_stroke_geometry_text, ) N_KEYFRAMES = 5 MAX_RETRIES = 5 PLAUSIBILITY_THRESHOLD = 4 FAITHFULNESS_THRESHOLD = 4 # both must pass to stop — faithfulness previously only # affected feedback text, never actually gated success QUALITY_THRESHOLD = 4 # same upgrade applied to the new quality criterion — # scored but not gating would repeat the same mistake def faithfulness_passed(score): """ faithfulness_score can be a number 1-5, the string "N/A" (no caption was given to compare against, so there's nothing to fail), or missing entirely (treated as NOT passed — can't confirm it's good, so err toward regenerating the narrative rather than assuming it's fine). """ if score is None: return False if isinstance(score, str): return score.strip().upper() == "N/A" if isinstance(score, (int, float)): return score >= FAITHFULNESS_THRESHOLD return False def unload_model(model): """Frees GPU memory before loading a different model. Necessary because the narrate/deform steps use a text Qwen model and the judge step uses a separate vision-language model (Qwen3-VL) — on hardware with limited VRAM (this project's RTX A4000, 16GB, already documented as a tight fit for a single model), loading both at once risks the same OOM issue that blocked Wan2.2 integration earlier. Load/unload sequentially instead of assuming both fit simultaneously. CONFIRMED BUG (found on real hardware, invisible to all mocked testing since no real GPU was available to catch it): `del model` here only clears THIS function's own local reference — it does nothing to the caller's variable, which stays alive and keeps the whole model resident in VRAM. torch.cuda.empty_cache() then has nothing to actually free, because the refcount never reaches zero. Fixed by returning None — callers MUST reassign their variable to this return value (e.g. `model = unload_model(model)`), or the bug reappears.""" import gc del model gc.collect() try: import torch if torch.cuda.is_available(): torch.cuda.empty_cache() except ImportError: pass return None # Dog3's real, human-reviewed narratives — used BOTH as the few-shot example # in `narrate` and as the fallback default if --narratives is omitted in # `deform`. One constant, one source of truth (previously duplicated across # two separate files under two different names with identical content). DOG3_CAPTION = ("The person throws a frisbee through the air, and the dog sits poised, " "ready to sprint forward and catch it with its mouth in a swift motion.") DOG3_NARRATIVES = { "dog": [ "the dog is sitting alert, watching the frisbee as it is thrown", "the dog is beginning to rise, weight shifting forward, head reaching toward the frisbee", "the dog is mid-leap, body extended, reaching far forward and up toward the frisbee", "the dog is at the peak of its jump, reaching as far as possible toward the frisbee", "the dog is landing after catching the frisbee, body compacting back down", ], "frisbee": [ "the frisbee has just left the thrower's hand, angled slightly upward", "the frisbee is gliding through the air, tilting slightly as it arcs", "the frisbee is near the peak of its arc, angled toward the dog", "the frisbee is descending toward the dog, tilting down slightly", "the frisbee is at the dog's mouth, being caught", ], } SVG_PATH_DEFAULT = "/mnt/user-data/uploads/dog3.svg" SEMANTIC_PATH_DEFAULT = "dog3_semantic.txt" DEFAULT_COLOR = "#444444" DEFAULT_LINEWIDTH = 1.1 OBJECT_COLORS = {"dog": "black", "person": "#3F4C57", "frisbee": "#B0463C"} OBJECT_LINEWIDTH = {"dog": 1.1, "person": 1.1, "frisbee": 1.4} # ============================================================================= # shared: Qwen call + response parsing # ============================================================================= def query_qwen(model, tokenizer, prompt, device, max_new_tokens=500, temperature=0.1): import torch messages = [{"role": "user", "content": prompt}] text = tokenizer.apply_chat_template(messages, tokenize=False, add_generation_prompt=True) inputs = tokenizer([text], return_tensors="pt").to(device) input_token_count = inputs["input_ids"].shape[1] with torch.no_grad(): output_ids = model.generate(**inputs, max_new_tokens=max_new_tokens, temperature=temperature, do_sample=True) generated = output_ids[0][inputs["input_ids"].shape[1]:] output_token_count = generated.shape[0] response_text = tokenizer.decode(generated, skip_special_tokens=True) return response_text, input_token_count, output_token_count def load_qwen_model(model_path): from transformers import AutoModelForCausalLM, AutoTokenizer import torch print(f"loading model from {model_path} ...") tokenizer = AutoTokenizer.from_pretrained(model_path) model = AutoModelForCausalLM.from_pretrained(model_path, torch_dtype=torch.float16, device_map="auto") device = next(model.parameters()).device print("model loaded.") return model, tokenizer, device def parse_json_response(response_text): match = re.search(r"\{.*\}", response_text, re.DOTALL) if not match: raise ValueError("No JSON object found in response:\n" + response_text) return json.loads(match.group(0)) # ============================================================================= # STEP 2: classify — ARAP vs TRAJ_ONLY, one call, all objects # ============================================================================= def build_deformation_prompt(caption, object_names): objects_str = ", ".join(f'"{o}"' for o in object_names) lines = "\n".join( f' "{o}": "ARAP" or "TRAJ_ONLY"' + ("," if i < len(object_names) - 1 else "") for i, o in enumerate(object_names) ) return f"""Scene: "{caption}" Objects in this scene: {objects_str} For each object, decide whether representing it correctly needs NON-RIGID DEFORMATION (its body/shape changes — e.g. limbs moving, a neck reaching, a body crouching or leaning) or whether simple RIGID TRANSLATION (the object moves/rotates as a whole, unchanged in shape, or doesn't move at all) is enough. Answer "ARAP" if the object's shape or body configuration changes at any point in the action, even if its overall position doesn't change. Answer "TRAJ_ONLY" if the object is rigid (a vehicle, tool, projectile, furniture, background element) or is simply carried by its own movement without changing shape. Respond with ONLY a JSON object, no other text, in this exact format: {{ {lines} }} """ def validate_deformation(parsed, object_names): problems = [] for obj in object_names: if obj not in parsed: problems.append(f"'{obj}' missing from response") continue val = str(parsed[obj]).strip().upper() if val not in ("ARAP", "TRAJ_ONLY"): problems.append(f"'{obj}' has invalid value {parsed[obj]!r}, expected ARAP or TRAJ_ONLY") return problems def run_classify(model, tokenizer, device, caption, semantic_path, out_path): assignments = load_semantic_assignments(semantic_path) object_names = list(assignments.keys()) print(f"objects found in {semantic_path}: {object_names}") prompt = build_deformation_prompt(caption, object_names) print("\n--- CLASSIFY PROMPT ---") print(prompt) response, in_tok, out_tok = query_qwen(model, tokenizer, prompt, device, max_new_tokens=250, temperature=0.1) print(f"\ntokens: {in_tok} in / {out_tok} out") print("--- RAW RESPONSE ---") print(response) parsed = parse_json_response(response) problems = validate_deformation(parsed, object_names) print("\n--- PARSED ---") print(json.dumps(parsed, indent=2)) if problems: print("--- VALIDATION PROBLEMS ---") for p in problems: print(f" - {p}") arap_objs = [o for o in object_names if str(parsed.get(o, "")).strip().upper() == "ARAP"] traj_only_objs = [o for o in object_names if o not in arap_objs] print(f"\nARAP: {arap_objs}") print(f"TRAJ_ONLY: {traj_only_objs}") with open(out_path, "w") as f: json.dump(parsed, f, indent=2) print(f"wrote {out_path}") return parsed, arap_objs def run_semantic_handle_selection(model, processor, rest_pose_image, obj_name, caption, unique_points, max_handles=4): """ VLM-guided handle selection replacing KMeans. Given the rest pose image and caption, asks the VLM to identify WHICH named body/object parts need to move to perform the captioned action, and WHERE those parts are located in pixel coordinates. Returns: (joints, anchor_idx, handle_idxs, joint_mesh_indices, part_names) - joints: (M, 2) array of selected handle positions - anchor_idx: index of anchor joint (geometric rule: closest to object centroid) - handle_idxs: list of movable joint indices - joint_mesh_indices: nearest mesh vertex per joint (guaranteed unique) - part_names: {joint_i: "mouth"/"hand"/etc} — the semantic label for each joint Design rationale (from project document Section 8): "semantic handle = 'mouth' → ARAP handle = specific mesh vertex/joint" The VLM reasons about 'mouth', while the existing deformation system operates on numerical geometry. This is caption-aware: we specifically ask which parts are needed for THIS captioned action, not just any meaningful part. """ import torch, json as _json, re as _re prompt = f"""You are analyzing a sketch of "{obj_name}" to determine which body/object parts need to move to perform this action: "{caption}". Look at the image carefully. Identify the specific parts of "{obj_name}" that need to DEFORM OR MOVE to perform the action described. For each part: 1. Give it a SHORT semantic name (e.g. "hand", "mouth", "head", "arm", "leg", "tail", "wing") 2. Estimate its CENTER pixel coordinate in the image (x from left, y from top) Rules: - Only name parts that are DIRECTLY relevant to the captioned action — do not list every part of the body - Maximum {max_handles} parts - Be specific: "right_hand" is better than "body" - The part that STAYS FIXED (e.g. torso, body center) should NOT be listed — only moving parts Respond ONLY with a JSON object, no other text: {{ "parts": [ {{"name": "hand", "x": 70, "y": 155}}, {{"name": "mouth", "x": 150, "y": 148}} ], "reasoning": "one sentence explaining which parts move for this action" }}""" content = [{"type": "image", "image": rest_pose_image}, {"type": "text", "text": prompt}] messages = [{"role": "user", "content": content}] text = processor.apply_chat_template(messages, tokenize=False, add_generation_prompt=True) inputs = processor(text=[text], images=[rest_pose_image], return_tensors="pt").to(model.device) with torch.no_grad(): output_ids = model.generate(**inputs, max_new_tokens=300, temperature=0.2, do_sample=True) generated = output_ids[:, inputs["input_ids"].shape[1]:] response = processor.batch_decode(generated, skip_special_tokens=True)[0] print(f" VLM semantic handle response for '{obj_name}': {response[:300]}") # parse the response parts = [] try: match = _re.search(r"\{.*\}", response, _re.DOTALL) if match: parsed = _json.loads(match.group(0)) parts = parsed.get("parts", []) reasoning = parsed.get("reasoning", "") print(f" reasoning: {reasoning}") except Exception as e: print(f" WARNING: failed to parse VLM semantic handle response: {e}") if not parts: print(f" WARNING: VLM returned no parts for '{obj_name}' — falling back to KMeans") return None # caller falls back to auto_select_handles_deduped # map each named part to nearest unique mesh vertex seen_vertices = {} joints_list, mesh_indices_list, part_names_list = [], [], [] for part in parts[:max_handles]: try: coord = [float(part["x"]), float(part["y"])] except (KeyError, ValueError, TypeError): continue mv = nearest_mesh_vertex(unique_points, coord) if mv in seen_vertices: print(f" NOTE: '{part.get('name')}' collided with already-used vertex, skipping") continue seen_vertices[mv] = True joints_list.append(coord) mesh_indices_list.append(mv) part_names_list.append(part.get("name", f"part_{len(joints_list)}")) if not joints_list: print(f" WARNING: no valid parts after vertex dedup for '{obj_name}' — falling back to KMeans") return None joints = np.array(joints_list) # geometric anchor: joint closest to object centroid (same rule as KMeans version) centroid = joints.mean(axis=0) dists = np.linalg.norm(joints - centroid, axis=1) anchor_idx = int(dists.argmin()) handle_idxs = [i for i in range(len(joints)) if i != anchor_idx] part_names = {i: part_names_list[i] for i in range(len(joints))} print(f" semantic handles for '{obj_name}': " + ", ".join(f"joint_{i}={part_names_list[i]}({'ANCHOR' if i==anchor_idx else 'handle'})" for i in range(len(joints)))) return joints, anchor_idx, handle_idxs, mesh_indices_list, part_names def build_objects_info(svg_path, semantic_path, arap_objects, caption=None, vlm_model=None, vlm_processor=None, rest_pose_image=None): """ Builds mesh/joint info for all ARAP objects. When vlm_model, vlm_processor, rest_pose_image, and caption are all provided, uses VLM-guided SEMANTIC handle selection (Option B) instead of KMeans. Falls back to KMeans automatically if VLM selection fails or returns no parts for a given object. The key addition over the KMeans version: objects_info now carries a `part_names` dict ({joint_i: "hand"/"mouth"/etc}) for each object when semantic selection succeeds — this propagates into the joint legend shown to the generator and judge, so the VLM reasons about named parts ("move the hand joint") rather than bare indices ("move joint_2"). """ objects_info = {} for obj_name in arap_objects: points, slices = load_object(obj_name, svg_path, semantic_path) bbox_size = object_bbox_size(points) unique_points, p2u = deduplicate_points(points, tol=0.35) tri, edges = build_mesh(unique_points) strokes = filter_strokes(load_strokes_from_svg(svg_path), load_semantic_assignments(semantic_path)[obj_name]) part_names = {} # populated by semantic selection, empty for KMeans fallback semantic_result = None # attempt VLM-guided semantic handle selection if all dependencies are available if vlm_model is not None and vlm_processor is not None and \ rest_pose_image is not None and caption: print(f" Attempting VLM semantic handle selection for '{obj_name}'...") semantic_result = run_semantic_handle_selection( vlm_model, vlm_processor, rest_pose_image, obj_name, caption, unique_points, max_handles=4) if semantic_result is not None: joints, anchor_idx, handle_idxs, joint_mesh_indices, part_names = semantic_result else: # fallback: original KMeans approach if vlm_model is not None: # only print if we tried and failed print(f" Falling back to KMeans for '{obj_name}'") joints, anchor_idx, handle_idxs, joint_mesh_indices = auto_select_handles_deduped( strokes, unique_points, k=4) if len(handle_idxs) == 0: print(f"WARNING: '{obj_name}' has no independent handles after dedup, skipping") continue print(f"'{obj_name}': {len(joints)} joints, anchor={anchor_idx}, handles={handle_idxs}, " f"mesh_indices={joint_mesh_indices}, bbox_size={bbox_size:.1f}" + (f", parts={part_names}" if part_names else "")) objects_info[obj_name] = { "joints": joints, "anchor_idx": anchor_idx, "handle_idxs": handle_idxs, "joint_mesh_indices": joint_mesh_indices, "bbox_size": bbox_size, "strokes": strokes, "points": points, "slices": slices, "unique_points": unique_points, "p2u": p2u, "edges": edges, "part_names": part_names, # {} when KMeans used, populated when semantic used } return objects_info def build_deformed_stroke_geometry_text(object_info, kf_targets, n_points=2): """ Reconstructs what a SPECIFIC keyframe's ACTUAL DEFORMED shape looks like, as compact text — runs the SAME ARAP solve used for real rendering, so a stroke's coordinates here are that keyframe's true rendered position, not its rest-pose position. This is what makes the judge's target_coords anchoring correct PER KEYFRAME instead of only correct when a keyframe happens to match rest — e.g. a "mouth" stroke's rest coordinate is wrong context if the head itself has moved by kf3, this reconstructs where that stroke actually is at kf3. kf_targets: {joint_i_str: [x, y]} for ONE keyframe, in the same "joint_i" key format apply_deform_clip_and_write saves (i.e. one entry of deform_outputs[obj_name][kf_key]). COST, not glossed over: re-runs arap_deform (10 iterations) once per object per keyframe purely to build this text — measured ~2,866 tokens for a single 64-stroke object across all 5 keyframes, and that multiplies per object in a multi-object scene. This is real compute and real prompt-length cost on top of everything else already in the judge prompt (images, rest geometry, joint legend, targets, bbox). """ import re as _re unique_points = object_info["unique_points"] edges = object_info["edges"] p2u = object_info["p2u"] slices = object_info["slices"] anchor_idx = object_info["anchor_idx"] joints = object_info["joints"] joint_mesh_indices = object_info["joint_mesh_indices"] handle_mesh_indices, handle_targets = [], [] for key, target in kf_targets.items(): m = _re.match(r"joint_(\d+)", key) if not m: continue joint_i = int(m.group(1)) if joint_i >= len(joint_mesh_indices): continue handle_mesh_indices.append(joint_mesh_indices[joint_i]) handle_targets.append(target) # anchor is always held at rest, same convention as rendering handle_mesh_indices.append(joint_mesh_indices[anchor_idx]) handle_targets.append(joints[anchor_idx].tolist()) if not handle_mesh_indices: return None deformed_unique = arap_deform(unique_points, edges, handle_mesh_indices, np.array(handle_targets), iterations=10) deformed_points = deformed_unique[p2u] lines = [] for i, (start, end) in enumerate(slices): stroke_pts = deformed_points[start:end] idxs = np.linspace(0, len(stroke_pts) - 1, n_points).astype(int) pts = stroke_pts[idxs] pts_str = " -> ".join(f"({x:.0f},{y:.0f})" for x, y in pts) lines.append(f" stroke_{i}: {pts_str}") return "\n".join(lines) def build_all_keyframes_deformed_geometry_text(objects_info, deform_outputs, n_keyframes=N_KEYFRAMES): """ For EVERY object and EVERY keyframe of the CURRENT attempt, reconstruct the actual deformed stroke geometry as text — this is the anchor source for the judge's target_coords grounding: the judge is instructed to find a visually-identified stroke's coordinate HERE, per keyframe, not in the rest-pose geometry (which is only correct when that keyframe happens to match rest). """ sections = [] for obj_name, info in objects_info.items(): obj_targets = (deform_outputs or {}).get(obj_name) if not obj_targets: continue kf_parts = [] for kf in range(n_keyframes): kf_key = f"kf{kf}" if kf_key not in obj_targets: continue shape_text = build_deformed_stroke_geometry_text(info, obj_targets[kf_key], n_points=2) if shape_text: kf_parts.append(f" -- {kf_key} --\n{shape_text}") if kf_parts: sections.append(f'Object "{obj_name}":\n' + "\n".join(kf_parts)) return "\n\n".join(sections) def render_rest_pose_multi(object_names, svg_path, semantic_path, out_path): """ Renders multiple objects' ORIGINAL strokes together, at their real positions in the source SVG (no deformation, no trajectory translation) — this is the "what does the sketch actually look like" image every attempt is grounded in, so Qwen can see what's actually drawable (e.g. whether the dog's back legs even exist as strokes) instead of only reasoning from the caption's text description. """ import matplotlib.pyplot as plt fig, ax = plt.subplots(figsize=(6, 6)) for name in object_names: points, slices = load_object(name, svg_path, semantic_path) for start, end in slices: seg = points[start:end] ax.plot(seg[:, 0], seg[:, 1], color="black", linewidth=1.2) ax.invert_yaxis() ax.set_aspect("equal") ax.set_title("rest pose") fig.savefig(out_path, dpi=150, bbox_inches="tight") plt.close(fig) return out_path def format_joint_feedback_for_object(object_name, joint_feedback): """ Filters the judge's structured joint_feedback list down to entries for THIS object and formats them as explicit, actionable lines. When the judge grounded its suggestion in an actual traced coordinate, delta_px (computed by vlm_judge.compute_joint_feedback_deltas via subtraction, NOT model math) is shown as an exact number the generator can apply directly — e.g. "joint_1 at kf3: hand not near mouth -> move by (dx=+12.0, dy=-8.0) px (target anchored to: mouth stroke in face object)" If the judge couldn't ground a target (target_coords was null, or the joint/keyframe didn't match anything in deform_outputs), delta_px is None and the line falls back to the prose issue only — explicitly labeled as ungrounded rather than silently presenting a guess as if it were a computed number. Returns "" if there's no feedback for this object (nothing to add). """ if not joint_feedback: return "" relevant = [f for f in joint_feedback if f.get("object") == object_name] if not relevant: return "" lines = [] for f in relevant: kf = f.get("keyframe") kf_str = f"kf{kf}" if kf is not None else "unspecified keyframe" issue = f.get("issue", "") delta = f.get("delta_px") if delta is not None: anchor = f.get("anchored_to", "unspecified") lines.append( f" - joint_{f.get('joint')} at {kf_str}: {issue} " f"-> move by (dx={delta['dx']:+.1f}, dy={delta['dy']:+.1f}) px " f"(target anchored to: {anchor})" ) else: lines.append( f" - joint_{f.get('joint')} at {kf_str}: {issue} " f"-> [UNGROUNDED — no traceable/disambiguated coordinate was given; use your own judgment]" ) return "\n".join(lines) def build_object_history_block(object_name, attempt_history, joint_feedback=None, baseline_targets=None): """ Formats THIS object's history across past attempts as text — one entry per past attempt showing its narrative, SCORES (faithfulness/plausibility/quality), the PER-JOINT corrections the judge gave THAT attempt (numeric target_coords/delta_px where grounded, same format as the current attempt's joint_feedback_block), and the flat feedback string, so the generator sees its own full numeric+prose trajectory across multiple attempts — not just the single most recent one (which baseline_targets/previous_narrative/the current joint_feedback_block already cover in full for the LATEST attempt specifically). The MOST RECENT past attempt's joint_feedback is deliberately EXCLUDED from this history text — that attempt's corrections are already shown in full via the current call's joint_feedback_block (built from the SAME `joint_feedback` param passed in here, which is the most recent attempt's computed deltas) — repeating them here would just duplicate that block. History shows joint corrections for OLDER attempts (attempt 1, 2, ... up to but not including the most recent one), which otherwise had no numeric trace anywhere in the prompt at all. Does NOT repeat baseline numeric targets (deform_outputs) here — baseline_targets already gives the generator the MOST RECENT attempt's exact numbers as the thing to refine; restating older attempts' raw target numbers too would give multiple different numeric anchors for the same joint with no clear precedence, the same redundancy problem already solved for baseline_targets vs joint_feedback for the current attempt. Deliberately scoped to ONE object's own entries — keeps history from re-injecting the confirmed cross-object contamination bug (a critique about a DIFFERENT object showing up here) that was fixed elsewhere in this file. """ if not attempt_history: return "" # the LAST entry in attempt_history is the most recent past attempt — its joint_feedback is # already shown via the current joint_feedback_block (same underlying data), so skip it here # to avoid duplicating the same corrections twice in one prompt. most_recent_attempt_number = attempt_history[-1]["attempt"] if attempt_history else None parts = [] for entry in attempt_history: obj_narrative = (entry.get("narratives") or {}).get(object_name) obj_targets = (entry.get("deform_outputs") or {}).get(object_name) # skip this attempt entirely for THIS object if it has no real object-specific data at all # (no targets, no narrative) — same rationale as before: don't fabricate a history entry out # of another object's leftover global feedback text. if not obj_targets and not obj_narrative: continue lines = [f' --- Attempt {entry["attempt"]} ---'] if obj_narrative: lines.append(f" Narrative was: {obj_narrative}") scores = entry.get("scores") if scores: faith, plaus, qual = scores lines.append(f" Scores: faithfulness={faith}, plausibility={plaus}, quality={qual}") if entry["attempt"] != most_recent_attempt_number: entry_joint_feedback_text = format_joint_feedback_for_object(object_name, entry.get("joint_feedback")) if entry_joint_feedback_text: lines.append(f" Per-joint corrections given for that attempt:\n{entry_joint_feedback_text}") if entry.get("feedback"): lines.append(f" Feedback received: {entry['feedback']}") parts.append("\n".join(lines)) if not parts: return "" return ( "\n History for this object across EARLIER attempts (see attached images for what each one " "actually rendered; numeric targets are NOT repeated here — the most recent attempt's exact " "numbers are given separately below as your actual starting point):\n" + "\n".join(parts) + "\n" ) def build_combined_object_section(object_name, joints, anchor_idx, handle_idxs, bbox_size, previous_narrative=None, feedback=None, n_keyframes=N_KEYFRAMES, freeze_narrative=False, strokes=None, joint_feedback=None, baseline_targets=None, history_block="", part_names=None): """ freeze_narrative: if True, previous_narrative is used as a FIXED target pose description (the object's narrative already passed faithfulness — only the numeric deformation needs to improve, not the story). If False (default), the narrative is regenerated fresh, informed by the attached image(s) and feedback, same as before. strokes: if given, the object's actual stroke points are included as TEXT (not just the rendered image) — added specifically because multimodal LLMs can under-attend to image content relative to text; this gives the same geometric information in a text-native form the model is more likely to actually use. Kept deliberately sparse (2 points per stroke, start+end only) since a complex object can have 60+ strokes — measured on real dog3 data: 2 points/stroke costs ~570 tokens for a 64-stroke object vs ~2650 for the full 12 points/stroke used internally for the ARAP mesh. joint_feedback: the FULL joint_feedback list from the judge's verdict (all objects) — filtered down to this object's entries and formatted as precise per-joint lines. Falls back to nothing (not an error) if the judge didn't return joint_feedback (e.g. coordinate context wasn't given to build_judge_prompt) — the object still gets the old flat `feedback` string via previous_block/feedback_line below. baseline_targets: {kf_key: {joint_i_str: [x,y]}} — this object's EXACT numeric targets from the PREVIOUS attempt, given on EVERY retry (not just after a pass) so the generator refines real numbers instead of reconstructing them from images/prose each time — the "guess coordinates from a picture" problem numeric joint_feedback exists to avoid elsewhere. Joints with a GROUNDED joint_feedback correction are excluded here (that correction takes precedence) — this only shows joints without a specific correction, as a "keep unless you have reason to change" anchor, not a fixed target — unlike `pose_lines` used when freeze_narrative=True, which locks the story, not the coordinates. """ cap = round(bbox_size * 0.25, 1) joint_feedback_text = format_joint_feedback_for_object(object_name, joint_feedback) joint_feedback_block = ( f"\n Specific per-joint corrections from the judge (apply these precisely, this is not general " f"guidance):\n{joint_feedback_text}\n" ) if joint_feedback_text else "" baseline_block = "" if baseline_targets: # joints the judge gave a GROUNDED correction for (real delta_px, not ungrounded prose) are # excluded from the raw baseline dump below — joint_feedback_block above is the authoritative # instruction for those specific joints, and repeating the stale pre-correction number here # would be redundant at best and contradictory at worst (two different numbers for the same # joint, no clear precedence). Baseline only shows joints WITHOUT a grounded correction, i.e. # "keep these as they were unless you have your own reason to change them." corrected_joints = { f.get("joint") for f in (joint_feedback or []) if f.get("object") == object_name and f.get("delta_px") is not None } kf_parts = [] for kf, kf_vals in baseline_targets.items(): shown = {j: v for j, v in kf_vals.items() if int(j.replace("joint_", "")) not in corrected_joints} if shown: kf_parts.append(f"{kf}: {{{', '.join(f'{j}={v}' for j, v in shown.items())}}}") if kf_parts: baseline_block = ( f"\n These are the EXACT numeric targets from the PREVIOUS attempt for joints the judge did " f"NOT give a specific correction for above — a real, working starting point, not a guess: " f"{', '.join(kf_parts)}\n" f" Keep these numbers unless you have a genuine reason to change them — do not discard them " f"and reinvent from scratch. For any joint listed in the corrections above instead, follow " f"that correction, not these numbers (that joint is intentionally omitted here).\n" ) joint_lines = "\n".join( f' - joint_{i}' + (f' ({(part_names or {}).get(i)})' if (part_names or {}).get(i) else "") + f': rest position (x={joints[i][0]:.1f}, y={joints[i][1]:.1f})' + (" <-- ANCHOR, must stay at or near this position in EVERY keyframe" if i == anchor_idx else "") for i in range(len(joints)) ) handle_list_str = ', joint_'.join(str(i) for i in handle_idxs) geometry_block = "" if strokes: geometry_text = build_stroke_geometry_text(strokes, n_points=2) geometry_block = ( f"\n This object's ACTUAL drawn strokes (start -> end point of each stroke, same coordinate " f"space as the joints above) — use this to know exactly what is and isn't actually drawn, don't " f"invent motion for parts that have no strokes here:\n{geometry_text}\n" ) if freeze_narrative and previous_narrative: pose_lines = "\n".join(f" kf{i}: {desc}" for i, desc in enumerate(previous_narrative)) feedback_line = f'\n This pose story already matches the intended action — it is FIXED, do not change it. ' \ f'Only the numeric target positions need to improve.' \ + (f' Previous attempt was judged: "{feedback}"' if feedback else "") + \ "\n The attached images show exactly what the previous attempt's target positions " \ "actually looked like when rendered — use them to see specifically what needs to change numerically." return f"""Object: "{object_name}" Joints: {joint_lines} {geometry_block} Target pose across all {n_keyframes} keyframes (FIXED, already correct — do not rewrite): {pose_lines} {feedback_line} {joint_feedback_block} {baseline_block} {history_block} For non-anchor joints (joint_{handle_list_str}), do not move more than {cap} pixels from REST in any keyframe. IMPORTANT: these {n_keyframes} keyframes are SPARSE anchor points spanning the ENTIRE action, NOT consecutive video frames — a large, dramatic difference between consecutive keyframes is NORMAL and EXPECTED, not an error; the actual in-between motion will be generated separately later by a different model. Positions should progress in a DIRECTIONALLY COHERENT way (don't make real progress toward the action and then have a LATER keyframe randomly revert backward without the narrative describing a reason to — e.g. only "landing"/"settling" should move back toward rest). Small, timid, barely-different positions between keyframes are themselves a mistake, not a safe choice.""" previous_block = "" if previous_narrative or feedback: parts = [] if previous_narrative: parts.append(f"Your previous narrative attempt was:\n{json.dumps(previous_narrative, indent=2)}") if feedback: parts.append(f'That attempt was judged and received this critique: "{feedback}"') parts.append("The attached images show exactly what that previous attempt actually looked like when " "rendered. Look at them, understand what specifically was wrong, and revise BOTH the " "narrative and the target positions to fix it — don't just reword the narrative " "superficially while leaving the same underlying problem.") previous_block = "\n " + "\n ".join(parts) + "\n" return f"""Object: "{object_name}" Joints: {joint_lines} {geometry_block} {previous_block} {joint_feedback_block} {baseline_block} {history_block} For non-anchor joints (joint_{handle_list_str}), do not move more than {cap} pixels from REST in any keyframe. IMPORTANT: these {n_keyframes} keyframes are SPARSE anchor points spanning the ENTIRE action, NOT consecutive video frames — a large, dramatic difference between consecutive keyframes is NORMAL and EXPECTED, not an error; the actual in-between motion will be generated separately later by a different model. Positions should progress in a DIRECTIONALLY COHERENT way (don't make real progress toward the action and then have a LATER keyframe randomly revert backward without the narrative describing a reason to — e.g. only "landing"/"settling" should move back toward rest). Small, timid, barely-different positions between keyframes are themselves a mistake, not a safe choice.""" DOG3_COMBINED_FEWSHOT_EXAMPLE = { "dog": { "narrative": [ "the dog is sitting alert, watching the frisbee as it is thrown", "the dog is beginning to rise, weight shifting forward, head reaching toward the frisbee", "the dog is mid-leap, body extended, reaching far forward and up toward the frisbee", "the dog is at the peak of its jump, reaching as far as possible toward the frisbee", "the dog is landing after catching the frisbee, body compacting back down", ], # real dog3 joint rest positions: joint_1=(178.28,136.85) head, joint_2=(228.78,194.94) # tail, joint_3=(189.29,151.35) neck — every value below verified to stay within a # 27px cap of rest. Notice the progression BUILDS UP through kf0->kf3 (increasing # displacement, matching "rising -> leaping -> peak reach") and only SETTLES BACK at # kf4 ("landing") — this is the exact monotonic-then-settle shape that was missing # when a real run produced a kf2 spike with kf3/kf4 reverting toward rest with no # narrative reason to. "targets": { "kf0": {"joint_1": [178.3, 136.8], "joint_2": [228.8, 194.9], "joint_3": [189.3, 151.3]}, "kf1": {"joint_1": [168.0, 127.0], "joint_2": [232.0, 191.0], "joint_3": [184.0, 144.0]}, "kf2": {"joint_1": [160.0, 120.0], "joint_2": [237.0, 186.0], "joint_3": [177.0, 137.0]}, "kf3": {"joint_1": [159.0, 119.0], "joint_2": [240.0, 183.0], "joint_3": [174.0, 134.0]}, "kf4": {"joint_1": [168.0, 128.0], "joint_2": [231.0, 192.0], "joint_3": [185.0, 146.0]}, }, } } def build_combined_narrate_deform_prompt(objects_info, caption, previous_narratives=None, feedback=None, n_keyframes=N_KEYFRAMES, is_retry=False, few_shot=True, freeze_narrative=False, joint_feedback=None, baseline_targets=None, attempt_history=None): sections, example_parts = [], [] for name, info in objects_info.items(): prev_narrative_for_obj = (previous_narratives or {}).get(name) obj_baseline_targets = (baseline_targets or {}).get(name) history_block = build_object_history_block(name, attempt_history, joint_feedback=joint_feedback, baseline_targets=obj_baseline_targets) sections.append(build_combined_object_section( name, info["joints"], info["anchor_idx"], info["handle_idxs"], info["bbox_size"], previous_narrative=prev_narrative_for_obj, feedback=feedback, n_keyframes=n_keyframes, freeze_narrative=freeze_narrative, strokes=info.get("strokes"), joint_feedback=joint_feedback, baseline_targets=obj_baseline_targets, history_block=history_block, part_names=info.get("part_names"), )) kf_examples = ",\n".join( " \"kf%d\": {%s}" % (kf, ", ".join(f'"joint_{i}": [x, y]' for i in info["handle_idxs"])) for kf in range(n_keyframes) ) if freeze_narrative: example_parts.append( f' "{name}": {{\n' f' "targets": {{\n{kf_examples}\n }}\n' f' }}' ) else: example_parts.append( f' "{name}": {{\n' f' "narrative": [<{n_keyframes} short pose description strings, one per keyframe>],\n' f' "targets": {{\n{kf_examples}\n }}\n' f' }}' ) all_sections = "\n\n".join(sections) example_json = "{\n" + ",\n".join(example_parts) + "\n}" image_context = "" if is_retry: image_context = ("The FIRST image attached is the object's original rest pose (undeformed). " "The remaining images are the actual rendered result of your PREVIOUS attempt, " "one per keyframe, in order.") else: image_context = ("The attached image shows the object's original rest pose (undeformed) — use this " "to understand what strokes actually exist and are available to move; do not " "invent motion for body parts that aren't actually drawn.") fewshot_block = "" if few_shot: fewshot_json = json.dumps(DOG3_COMBINED_FEWSHOT_EXAMPLE, indent=2) fewshot_block = f"""Example — for the scene "{DOG3_CAPTION}", a good answer looks like: {fewshot_json} Notice: each narrative keyframe reads as a distinct, substantially different stage of the action — not a near-duplicate of its neighbor, and not a small incremental change from it. The joint targets BUILD UP smoothly (kf0 -> kf1 -> kf2 -> kf3 each moving further than the last) and only settle back toward rest at the FINAL keyframe, matching the narrative's "landing" moment — no keyframe overshoots and then has a later keyframe revert back toward rest without a narrative reason to. Match this style and this kind of numeric consistency for the new scene below. """ if freeze_narrative: output_instruction = ( 'For EACH object above, the narrative/pose story is already fixed (shown above) — ' 'produce ONLY:\n' ' "targets": target (x, y) positions for its non-anchor joints, at every keyframe, ' 'consistent with the fixed pose story above.' ) else: output_instruction = ( "For EACH object above, produce BOTH:\n" ' 1. "narrative": a plain-English pose description for each keyframe. REMEMBER: these are ' "SPARSE keyframes spanning the WHOLE action, not consecutive video frames — each description " "should be a meaningfully, substantially different stage of the action from its neighbors, not " "a small incremental change. Write these like 5 distinct captions for 5 different moments spread " "across an entire action, not like 5 near-duplicate snapshots a split-second apart. Under 20 " "words each.\n" ' 2. "targets": target (x, y) positions for its non-anchor joints, at every keyframe, ' "consistent with your own narrative." ) return f"""{fewshot_block}You are directing a {n_keyframes}-keyframe animated sequence for a hand-drawn sketch, viewed from the side. Coordinate system: x increases rightward, y increases DOWNWARD. IMPORTANT: these {n_keyframes} keyframes are SPARSE anchor points sampled across the ENTIRE action from start to finish — NOT consecutive video frames. Think of them like 5 widely-spaced snapshots of a whole motion, not neighboring frames a fraction of a second apart. Large, dramatic pose changes between consecutive keyframes are normal and expected; a separate model will generate the actual in-between motion frames later. Do not treat these like near-continuous animation frames. Scene: "{caption}" {image_context} {all_sections} {output_instruction} Consider objects together (e.g. a dog reaching toward a frisbee should be spatially consistent with the frisbee's own position) and consider each object's OWN sequence together — these {n_keyframes} keyframes are SPARSE anchor points spanning the WHOLE action, not consecutive video frames, so large differences between consecutive keyframes are expected and correct, not something to avoid. The many actual in-between motion frames will be generated separately later. Only avoid a keyframe making real progress and then a LATER keyframe randomly reverting backward without the narrative describing why. Respond with ONLY one JSON object, no other text, in this exact format: {example_json} """ def run_combined_narrate_deform(model, processor, images, prompt, temperature=0.6): """Same multi-image calling pattern as vlm_judge.run_judge — reused here since both are Qwen3-VL calls with a list of images + one prompt. temperature default raised from 0.1 to 0.6 — CONFIRMED on real hardware (eat2, 3 attempts) that at 0.1 the model reproduced its joint targets as an EXACT copy of the rest-position legend values (not just "close to rest" — bit-identical to the decimal) on attempt 1, before any feedback existed to explain it. Near-zero temperature strongly favors the single highest-probability continuation, and copying a number already visible in-context (the rest-position legend, formatted in the same [x, y] style as the requested targets) is a low-risk, easy completion under numeric uncertainty. This is a hypothesis about mechanism, not confirmed root cause — raising temperature is the cheapest test of it before trying a bigger/different model. """ import torch content = [{"type": "image", "image": img} for img in images] content.append({"type": "text", "text": prompt}) messages = [{"role": "user", "content": content}] text = processor.apply_chat_template(messages, tokenize=False, add_generation_prompt=True) inputs = processor(text=[text], images=images, return_tensors="pt").to(model.device) with torch.no_grad(): output_ids = model.generate(**inputs, max_new_tokens=500 * N_KEYFRAMES, temperature=temperature, do_sample=True) generated = output_ids[:, inputs["input_ids"].shape[1]:] response = processor.batch_decode(generated, skip_special_tokens=True)[0] return response def transform_image_and_get_inverse(image, angle_deg, scale, canvas_size): """ Rotates and scales a PIL image around the canvas center, returning the transformed image AND an exact inverse function mapping a coordinate picked from the TRANSFORMED image back to the ORIGINAL image's coordinate space. This is the one piece of this feature that MUST be exactly correct, since a subtle sign/direction error here would silently produce confidently-wrong coordinates for every joint target, worse than not having the feature at all. The rotation formula used here (x' = x*cos+y*sin, y' = -x*sin+y*cos) was determined EMPIRICALLY against PIL's actual Image.rotate() behavior, NOT derived theoretically and trusted — an initial theoretical derivation (standard CCW rotation matrix) was tested against a real PIL render with a marked point and was WRONG (145px recovery error); this formula is the one that was verified correct via a real end-to-end test: render an image with a known marker point, rotate it with PIL, empirically find where the marker actually ended up, and confirm the inverse function maps that found location back to the original marker location. Any future change to this function should be re-verified the same way, not just algebraically. canvas_size: (width, height) of the image BEFORE transforming — the rotation/scale pivot is the center of this canvas, matching where the sketch itself is centered in the fixed xmin/xmax/ymin/ymax render bounds, not the image's own bounding box after rotation (which would shift as the canvas expands to fit a rotated image). """ cx, cy = canvas_size[0] / 2.0, canvas_size[1] / 2.0 theta = math.radians(angle_deg) cos_t, sin_t = math.cos(theta), math.sin(theta) transformed = image.resize((int(image.width * scale), int(image.height * scale))) if scale != 1.0 else image if angle_deg != 0: transformed = transformed.rotate(angle_deg, resample=Image.BICUBIC, expand=False, center=(transformed.width / 2.0, transformed.height / 2.0), fillcolor=(255, 255, 255)) def inverse(pt): # pt is in the TRANSFORMED image's own pixel coordinates (post-resize, post-rotate) tcx, tcy = transformed.width / 2.0, transformed.height / 2.0 x, y = pt[0] - tcx, pt[1] - tcy # undo scale first (transform applied resize THEN rotate, so inverse undoes in reverse order: # rotate-inverse then scale-inverse — but since both are linear ops about the same center, # order between them doesn't actually matter here; scale-then-rotate-inverse is used to # match the verified formula exactly) x_s, y_s = x / scale, y / scale # undo rotation using the VERIFIED inverse (transpose of the empirically-confirmed forward # matrix [[cos,sin],[-sin,cos]] is [[cos,-sin],[sin,cos]]) x_r = x_s * cos_t - y_s * sin_t y_r = x_s * sin_t + y_s * cos_t # re-center on the ORIGINAL canvas's center return (x_r + cx, y_r + cy) return transformed, inverse def run_combined_narrate_deform_multiview(model, processor, images, prompt, temperature=0.6, views=None, canvas_size=(260, 230)): """ EXPERIMENTAL — generalizes run_combined_narrate_deform to query MULTIPLE rotated/scaled views of the SAME images, then median-aggregates the resulting joint targets per (object, keyframe, joint) back in the ORIGINAL coordinate space. Adapted from the multi-view voting idea in "Handle-based Mesh Deformation Guided By Vision Language Model" (Sun et al. 2025), but applied to the GENERATOR here rather than the judge (see vlm_judge.run_judge_voted for that side) — the hypothesis being tested is that a small/sparse detail (e.g. a hand, a small object) might be picked out more reliably from SOME rotation/scale than others, so combining multiple views could reduce the chance any one view's blind spot dominates the result. IMPORTANT CAVEAT, stated plainly rather than glossed over: unlike the judge-voting case (which resamples the SAME view multiple times), this changes what the model actually SEES between calls — a rotated/scaled image is a genuinely different visual input, closer to the paper's original multi-CAMERA-view setup in spirit. But it is still a 2D sketch, not a 3D mesh with real depth to resolve — there is no geometric ambiguity being resolved here, only a bet that varied presentation might surface details a single fixed view misses. UNTESTED whether this actually helps versus just adding noise from the model reasoning differently about a rotated image (e.g. text/handedness cues in the sketch may become confusing when rotated). views: list of (angle_deg, scale) tuples. If None, defaults to a small combined set: [(0, 1.0), (90, 1.0), (270, 1.0), (0, 1.3), (0, 0.75)] — three rotations (0/90/270, skipping 180 which mostly just flips faces/text unhelpfully) plus two scale variants (zoom in / zoom out) at 0 rotation. This is a starting default, not a validated optimal set. Returns: (aggregated_response_dict_or_None, list_of_per_view_raw_response_strings) aggregated_response_dict is None if EVERY view's response failed to parse — caller should fall back to treating this as a parse failure, same as the single-view case. """ if views is None: views = [(0, 1.0), (90, 1.0), (270, 1.0), (0, 1.3), (0, 0.75)] per_view_parsed = [] raw_responses = [] for angle_deg, scale in views: view_images = [] view_inverses = [] for img in images: transformed_img, inverse_fn = transform_image_and_get_inverse(img, angle_deg, scale, canvas_size) view_images.append(transformed_img) view_inverses.append(inverse_fn) # all images in one call share the SAME transform (same angle/scale), so any one inverse # function is representative — they're mathematically identical per-view inverse_fn = view_inverses[0] if view_inverses else (lambda pt: pt) raw = run_combined_narrate_deform(model, processor, view_images, prompt, temperature=temperature) raw_responses.append(raw) try: parsed = parse_json_response(raw) except (ValueError, json.JSONDecodeError): continue # this view's response excluded from aggregation, not treated as a default # inverse-transform every joint target in this view's response back to original coordinates for obj_name, obj_data in parsed.items(): targets = obj_data.get("targets") if isinstance(obj_data, dict) else None if not targets: continue for kf_key, kf_targets in targets.items(): if not isinstance(kf_targets, dict): continue for joint_key, coord in list(kf_targets.items()): if isinstance(coord, (list, tuple)) and len(coord) == 2: try: x, y = inverse_fn((float(coord[0]), float(coord[1]))) kf_targets[joint_key] = [x, y] except (TypeError, ValueError): pass # malformed coordinate — leave as-is, downstream shape validation will catch it per_view_parsed.append(parsed) if not per_view_parsed: return None, raw_responses # median-aggregate per (object, kf_key, joint_key) across whichever views produced a valid, # correctly-shaped [x, y] pair for that exact combination — same aggregation pattern as # vlm_judge.run_judge_voted, applied to the generator's differently-shaped output. aggregated = {} for parsed in per_view_parsed: for obj_name, obj_data in parsed.items(): if not isinstance(obj_data, dict): continue aggregated.setdefault(obj_name, {"narrative": None, "targets": {}}) if obj_data.get("narrative") and aggregated[obj_name]["narrative"] is None: # narrative text isn't a coordinate — no meaningful "median" of prose; use the # FIRST view's narrative that has one, since all views describe the same intended # action and prose aggregation isn't the point of this feature aggregated[obj_name]["narrative"] = obj_data["narrative"] targets = obj_data.get("targets") or {} for kf_key, kf_targets in targets.items(): if not isinstance(kf_targets, dict): continue aggregated[obj_name]["targets"].setdefault(kf_key, {}) for joint_key, coord in kf_targets.items(): if isinstance(coord, (list, tuple)) and len(coord) == 2: aggregated[obj_name]["targets"][kf_key].setdefault(joint_key, {"xs": [], "ys": []}) try: aggregated[obj_name]["targets"][kf_key][joint_key]["xs"].append(float(coord[0])) aggregated[obj_name]["targets"][kf_key][joint_key]["ys"].append(float(coord[1])) except (TypeError, ValueError): pass final = {} for obj_name, obj_data in aggregated.items(): final[obj_name] = {"narrative": obj_data["narrative"], "targets": {}} for kf_key, kf_targets in obj_data["targets"].items(): final[obj_name]["targets"][kf_key] = {} for joint_key, coords in kf_targets.items(): if coords["xs"]: final[obj_name]["targets"][kf_key][joint_key] = [ statistics.median(coords["xs"]), statistics.median(coords["ys"]) ] return final, raw_responses def apply_deform_clip_and_write(parsed, objects_info, out_dir, sketch_name, n_keyframes=N_KEYFRAMES, frozen_narratives=None): """ Shared post-processing for the combined call's "targets" section: same hard-clip logic as the old run_deform, applied here instead. Returns (narratives_dict, deform_outputs_dict, cap_utilization_dict). frozen_narratives: {obj_name: [...]} — used as a fallback when the response doesn't include a "narrative" key for an object, which happens when freeze_narrative=True was used in the prompt (the model was never asked to produce one, so its absence is expected, not an error — carry the frozen one forward instead of losing it). cap_utilization: {obj_name: mean_fraction_of_cap_used} — CONFIRMED on real hardware (horsecar5's person) that Qwen can propose displacement well within the movement cap without ever being told it did so — the handle selection and clipping were both working correctly, but the actual output was too timid to be visible (e.g. a leg moving only 18% of its allowed range). The hard clip only ever catches OVER the cap; nothing previously caught UNDER-using it. This surfaces that as an explicit number so it can be fed back to Qwen directly. """ narratives_out = {} deform_outputs = {} utilization_by_obj = {} # {obj_name: [fraction, fraction, ...]} across all joints/keyframes for obj_name, info in objects_info.items(): if obj_name not in parsed: print(f" WARNING: '{obj_name}' missing from response entirely, skipping") continue obj_result = parsed[obj_name] utilization_by_obj[obj_name] = [] if "narrative" in obj_result: narratives_out[obj_name] = obj_result["narrative"] elif frozen_narratives and obj_name in frozen_narratives: narratives_out[obj_name] = frozen_narratives[obj_name] else: print(f" WARNING: '{obj_name}' has no narrative in response and no frozen narrative " f"to fall back to — narratives.json will be missing this object") deform_outputs[obj_name] = {} targets = obj_result.get("targets", {}) for kf in range(n_keyframes): kf_key = f"kf{kf}" if kf_key not in targets: print(f" WARNING: '{obj_name}' missing {kf_key} targets, skipping this frame") continue kf_result = targets[kf_key] out = {} joint_targets_this_kf = {} for name, target in kf_result.items(): m = re.match(r"joint_(\d+)", name) if not m: print(f" WARNING: unexpected key '{name}' for '{obj_name}' {kf_key}, skipping") continue joint_i = int(m.group(1)) if joint_i >= len(info["joint_mesh_indices"]): print(f" WARNING: '{obj_name}' joint_{joint_i} out of range, skipping") continue # Qwen occasionally returns a bare scalar (e.g. 195.4) instead of an [x, y] pair for # a joint target. np.array(scalar, dtype=float) does NOT error — it silently makes a # 0-d array, which then broadcasts against `rest` (2-d) into a plausible-looking 2-d # `disp` with no error anywhere in THIS function. The malformed value then gets # written to disk as-is and only crashes 3 steps later in run_render, when it tries # to stack this scalar next to properly-shaped [x,y] entries — as an inhomogeneous # array error that gives no indication which object/joint/keyframe was actually bad. # CONFIRMED on real hardware (eagle4, football7/'person') that this happens on real # Qwen output, not just as a theoretical edge case. target_list = target if isinstance(target, (list, tuple)) else [target] if len(target_list) != 2: print(f" WARNING: '{obj_name}' {kf_key} joint_{joint_i} target = {target!r}, expected " f"an [x, y] pair (got {len(target_list)} value(s)) — skipping this joint for this " f"keyframe rather than writing a malformed value that would crash rendering later") continue mesh_idx = info["joint_mesh_indices"][joint_i] rest = np.array(info["joints"][joint_i]) cap = round(info["bbox_size"] * 0.25, 1) target_arr = np.array(target_list, dtype=float) disp = target_arr - rest dist = np.linalg.norm(disp) utilization_by_obj[obj_name].append(min(dist / cap, 1.0) if cap > 0 else 0.0) if dist > cap: clipped = rest + disp / dist * cap print(f" CLIPPED '{obj_name}' {kf_key} joint_{joint_i}: requested {dist:.1f}px " f"(cap {cap}px) -> clipped to {cap}px, direction preserved") target = clipped.tolist() else: target = target_arr.tolist() out[str(mesh_idx)] = target joint_targets_this_kf[f"joint_{joint_i}"] = target anchor_mesh_idx = info["joint_mesh_indices"][info["anchor_idx"]] out[str(anchor_mesh_idx)] = info["joints"][info["anchor_idx"]].tolist() deform_outputs[obj_name][kf_key] = joint_targets_this_kf out_path = os.path.join(out_dir, f"qwen_{sketch_name}_{obj_name}_kf{kf}.json") with open(out_path, "w") as f: json.dump(out, f, indent=2) print(f" wrote {out_path}") cap_utilization = {} for obj_name, fractions in utilization_by_obj.items(): if fractions: mean_frac = sum(fractions) / len(fractions) cap_utilization[obj_name] = mean_frac print(f" '{obj_name}': mean cap utilization = {mean_frac*100:.0f}% " f"(across {len(fractions)} joint-keyframe pairs)") return narratives_out, deform_outputs, cap_utilization # ============================================================================= # STEP 4: render — compose the full scene (no Qwen call at all) # ============================================================================= def run_render(handles_dir, svg_path, semantic_path, traj_path, out_path=None, frames_dir=None): """ out_path: if given, ALSO saves the combined strip image (all 5 keyframes side by side) here, outside frames_dir. Optional — pass None to keep output confined to frames_dir only. frames_dir: if given, saves each keyframe as its own individual PNG (kf0.png ... kf4.png) plus the combined strip, named after the sketch itself ({sketch_name}.png), all inside this one folder. """ import matplotlib.pyplot as plt sketch_name = os.path.splitext(os.path.basename(svg_path))[0] real_trajectories = load_trajectories(traj_path) object_names = list(real_trajectories.keys()) object_data = {} for name in object_names: points, slices = load_object(name, svg_path, semantic_path) dx_vals, dy_vals = bbox_deltas(real_trajectories[name]) unique_points, p2u = deduplicate_points(points, tol=0.35) tri, edges = build_mesh(unique_points) object_data[name] = { "points": points, "slices": slices, "dx": dx_vals, "dy": dy_vals, "unique_points": unique_points, "p2u": p2u, "edges": edges, } if frames_dir: os.makedirs(frames_dir, exist_ok=True) xmin, xmax, ymin, ymax = 0, 260, 60, 230 # compute each keyframe's drawing data once, reused for both the combined # strip and the individual per-keyframe images keyframe_lines = [] # list of {name: [(seg_x, seg_y), ...]} per keyframe for kf in range(N_KEYFRAMES): lines_this_kf = {} for name in object_names: od = object_data[name] handles_path = os.path.join(handles_dir, f"qwen_{sketch_name}_{name}_kf{kf}.json") if os.path.exists(handles_path): with open(handles_path) as f: spec = json.load(f) handle_indices = [int(k) for k in spec.keys()] handle_targets = np.array([spec[k] for k in spec.keys()]) n_verts = len(od["unique_points"]) bad = [i for i in handle_indices if i >= n_verts] if bad: print(f" ERROR: {handles_path} has out-of-bounds indices {bad} for '{name}' " f"({n_verts} mesh vertices) — likely from a DIFFERENT sketch's mesh. " f"Falling back to translation-only.") deformed_points = od["points"] mode = "translation-only (handles file failed validation)" else: deformed_unique = arap_deform(od["unique_points"], od["edges"], handle_indices, handle_targets, iterations=10) deformed_points = deformed_unique[od["p2u"]] mode = "ARAP" else: deformed_points = od["points"] mode = "translation-only (no handles file found)" moved = deformed_points + np.array([od["dx"][kf], od["dy"][kf]]) lines_this_kf[name] = [moved[start:end] for start, end in od["slices"]] print(f"kf{kf} '{name}': {mode}, points_after_move_range=" f"x[{moved[:,0].min():.1f},{moved[:,0].max():.1f}] " f"y[{moved[:,1].min():.1f},{moved[:,1].max():.1f}]") keyframe_lines.append(lines_this_kf) if frames_dir: fig_i, ax_i = plt.subplots(figsize=(6, 5.5)) for name, segs in lines_this_kf.items(): for seg in segs: ax_i.plot(seg[:, 0], seg[:, 1], color=OBJECT_COLORS.get(name, DEFAULT_COLOR), linewidth=OBJECT_LINEWIDTH.get(name, DEFAULT_LINEWIDTH)) ax_i.set_xlim(xmin, xmax) ax_i.set_ylim(ymax, ymin) ax_i.set_aspect("equal") ax_i.set_title(f"{sketch_name} — kf{kf}", fontsize=12, fontweight="bold") frame_path = os.path.join(frames_dir, f"kf{kf}.png") fig_i.savefig(frame_path, dpi=140, bbox_inches="tight") plt.close(fig_i) print(f" wrote {frame_path}") # combined strip, same as before fig, axes = plt.subplots(1, N_KEYFRAMES, figsize=(24, 5)) for kf in range(N_KEYFRAMES): ax = axes[kf] for name, segs in keyframe_lines[kf].items(): for seg in segs: ax.plot(seg[:, 0], seg[:, 1], color=OBJECT_COLORS.get(name, DEFAULT_COLOR), linewidth=OBJECT_LINEWIDTH.get(name, DEFAULT_LINEWIDTH)) ax.set_xlim(xmin, xmax) ax.set_ylim(ymax, ymin) ax.set_aspect("equal") ax.set_title(f"kf{kf}", fontsize=13, fontweight="bold") plt.tight_layout() if out_path: plt.savefig(out_path, dpi=140, bbox_inches="tight") print(f"wrote {out_path}") if frames_dir: combined_frame_path = os.path.join(frames_dir, f"{sketch_name}.png") plt.savefig(combined_frame_path, dpi=140, bbox_inches="tight") print(f"wrote {combined_frame_path}") if not out_path and not frames_dir: print("WARNING: neither out_path nor frames_dir given, combined strip image not saved anywhere") plt.close(fig) # ============================================================================= # CLI — single input: a sketch name or an SVG path. Everything else is # derived automatically from the directory conventions used throughout # this dataset. Override flags exist for the rare case a path doesn't # match convention, but nothing is required beyond the sketch itself. # ============================================================================= # Confirmed real paths from this dataset, used as defaults so nothing else # needs to be typed per run. If your layout differs, override with the # corresponding --*-dir / --*-file flag below. SVG_DIR_DEFAULT = "/user/HS400/rk01499/my_scratch/sketch/data/raw/60sketches/svg" PROCESSED_DIR_DEFAULT = "/user/HS400/rk01499/my_scratch/sketch/data/processed" CAPTION_FILE_DEFAULT = "/user/HS400/rk01499/my_scratch/sketch/data/raw/60sketches/caption.txt" MODEL_PATH_DEFAULT = "/user/HS400/rk01499/my_scratch/models/qwen2.5-7b/" def resolve_sketch_paths(sketch, svg_dir, processed_dir, caption_file): """ sketch: either a bare sketch name ("dog9") or a path to its SVG ("/path/to/dog9.svg") — either way, everything else (semantic, traj, caption) is derived from the same naming convention used across this dataset: {name}.svg, {name}/{name}_semantic.txt, {name}/{name}_traj.txt, and a lookup in one shared caption.txt. """ name = os.path.splitext(os.path.basename(sketch))[0] svg_path = sketch if sketch.endswith(".svg") else os.path.join(svg_dir, f"{name}.svg") semantic_path = os.path.join(processed_dir, name, f"{name}_semantic.txt") traj_path = os.path.join(processed_dir, name, f"{name}_traj.txt") missing = [p for p in [svg_path, semantic_path, traj_path, caption_file] if not os.path.exists(p)] if missing: raise SystemExit( f"Could not find these expected files for sketch '{name}':\n " + "\n ".join(missing) + "\n\nIf your directory layout differs from the default, pass --svg-dir / " "--processed-dir / --caption-file explicitly." ) caption = get_caption(caption_file, name) return name, svg_path, semantic_path, traj_path, caption def main(): ap = argparse.ArgumentParser( description="Run the full sketch deformation pipeline for one image. " "The only required input is the sketch — everything else " "(semantic assignments, trajectory, caption) is looked up " "automatically from the standard dataset layout.") ap.add_argument("sketch", type=str, help="sketch name (e.g. 'dog9') or path to its .svg file") ap.add_argument("--model", type=str, default=MODEL_PATH_DEFAULT) ap.add_argument("--svg-dir", type=str, default=SVG_DIR_DEFAULT) ap.add_argument("--processed-dir", type=str, default=PROCESSED_DIR_DEFAULT) ap.add_argument("--caption-file", type=str, default=CAPTION_FILE_DEFAULT) ap.add_argument("--out-dir", type=str, default=".") ap.add_argument("--no-fewshot", action="store_true", help="disable the dog3 few-shot example in narrate (for A/B comparison)") ap.add_argument("--narrate-temperature", type=float, default=0.6, help="FLAT sampling temperature for every attempt (default 0.6, raised from the " "original 0.1 — CONFIRMED that 0.1 caused the model to copy rest-position values " "verbatim as targets on attempt 1). Ignored if --narrate-temperature-schedule " "is given instead.") ap.add_argument("--narrate-temperature-schedule", type=str, default=None, help="EXPERIMENTAL — comma-separated per-attempt temperature values, e.g. " "'0.8,0.6,0.5,0.4,0.4' or '0.6,0.4,0.3,0.3,0.3'. Overrides --narrate-temperature " "when given. Hypothesis being tested: high temperature avoids copy-from-rest on " "a cold-start attempt 1 (confirmed); once baseline_targets gives later attempts " "a real number to refine rather than generate from scratch, lower temperature's " "tighter output may help precision more than it risks the original copy-from-rest " "failure — UNTESTED, this is what the schedule exists to check. If fewer values " "are given than MAX_RETRIES, the LAST value is reused for remaining attempts.") ap.add_argument("--force-all-attempts", action="store_true", help="EXPERIMENTAL — always run all MAX_RETRIES attempts, NEVER stop early on a " "pass or on DINOv2 stagnation, then pick the BEST-SCORING attempt at the end " "(ranked by faithfulness_score, tie-broken by plausibility_score). Replaces the " "confirmation-round mechanism (which only ever compared 2 attempts) with a full " "best-of-N comparison across every attempt actually run. Primarily useful when " "combined with --narrate-temperature-schedule, so the WHOLE schedule gets " "exercised rather than potentially stopping before later, lower-temperature " "attempts are ever reached.") ap.add_argument("--judge-votes", type=int, default=1, help="EXPERIMENTAL — call the judge N times per attempt on the SAME prompt/images and " "aggregate (median) the scores and any grounded target_coords across the N " "responses, instead of trusting a single judge call. Based on the multi-view " "voting idea in 'Handle-based Mesh Deformation Guided By Vision Language Model' " "(Sun et al. 2025) — there, uncertainty in a single VLM coordinate prediction is " "reduced by querying from multiple camera views and voting; here, there's only " "one view, so the analogous move is multiple independent judge calls on the SAME " "view, voted the same way. Default 1 = current single-call behavior, unchanged. " "UNTESTED whether this actually helps vs. just costing N times more compute — " "the paper's setting (multiple genuinely different camera views) is not identical " "to this one (same view, resampled), so the noise being averaged out may be " "smaller here than in the original technique.") ap.add_argument("--generator-views", action="store_true", help="EXPERIMENTAL — apply the SAME multi-view voting idea to the GENERATOR " "(narrate+deform) instead of just the judge: query multiple ROTATED/SCALED " "versions of the same images (0°/90°/270° rotations, 0.75x/1.3x scale — see " "run_combined_narrate_deform_multiview for the exact set), inverse-transform " "each view's joint targets back to the ORIGINAL coordinate space, and " "median-aggregate per (object, keyframe, joint). Off by default — the " "coordinate transform math is verified correct (empirically tested against " "real PIL rendering, not just algebra), but whether varying the generator's " "input view actually helps it notice small/sparse details (vs. just adding " "noise from reasoning about an unfamiliar rotated image) is UNTESTED. Also " "costs 5x the generator compute per attempt (one call per view) when enabled.") ap.add_argument("--deform-only", action="store_true", help="run classify + ONE narrate+deform attempt + render, then STOP — no judge, " "no retries. Prints cap utilization directly and saves the render, so you can " "inspect raw generation quality without the judge's assessment as a confound.") args = ap.parse_args() # parse the temperature schedule once, up front — if given, this is a list of per-attempt # values; the attempt loop looks up temperature_schedule[attempt-1] (clamped to the last value # if the schedule is shorter than the number of attempts actually run, e.g. the confirmation # attempt beyond MAX_RETRIES). temperature_schedule = None if args.narrate_temperature_schedule: temperature_schedule = [float(v.strip()) for v in args.narrate_temperature_schedule.split(",")] print(f" narrate temperature SCHEDULE: {temperature_schedule} " f"(overrides flat --narrate-temperature={args.narrate_temperature})") else: print(f" narrate temperature: {args.narrate_temperature} (flat, no schedule given)") name, svg_path, semantic_path, traj_path, caption = resolve_sketch_paths( args.sketch, args.svg_dir, args.processed_dir, args.caption_file) print(f"sketch: {name}") print(f" svg: {svg_path}") print(f" semantic: {semantic_path}") print(f" traj: {traj_path}") print(f" caption: {caption!r}") os.makedirs(args.out_dir, exist_ok=True) # tag every output folder with a run timestamp so re-running the SAME sketch name never # silently overwrites a previous run's output — always unique per invocation, regardless of # which flags/features (temperature, history, confirmation round) differ between runs. run_timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") run_label = f"{name}__{run_timestamp}" json_dir = os.path.join(args.out_dir, "json", run_label) os.makedirs(json_dir, exist_ok=True) print(f" json output dir: {json_dir}") print(f" run label: {run_label}") print("\n########## STEP 1: CLASSIFY ##########") classify_model, classify_tokenizer, classify_device = load_qwen_model(args.model) deformation, arap_objects = run_classify( classify_model, classify_tokenizer, classify_device, caption, semantic_path, os.path.join(json_dir, f"{name}_deformation.json")) classify_model = unload_model(classify_model) if arap_objects: import vimjudge1 as vlm_judge temp_dir = os.path.join(args.out_dir, "P_1", run_label) # render rest pose FIRST — needed as visual input for VLM-guided semantic handle selection rest_pose_image_path = os.path.join(json_dir, f"{name}_rest_pose.png") render_rest_pose_multi(arap_objects, svg_path, semantic_path, rest_pose_image_path) rest_pose_image_for_selection = Image.open(rest_pose_image_path).convert("RGB") # load VLM once for semantic handle selection, then unload before the retry loop # (which reloads it per attempt as before) print("\n########## STEP 1b: SEMANTIC HANDLE SELECTION ##########") sel_model, sel_processor = vlm_judge.load_vlm("Qwen/Qwen2.5-VL-3B-Instruct") objects_info = build_objects_info(svg_path, semantic_path, arap_objects, caption=caption, vlm_model=sel_model, vlm_processor=sel_processor, rest_pose_image=rest_pose_image_for_selection) sel_model = unload_model(sel_model) if not objects_info: print("No objects with valid handles found after mesh setup. Nothing to do.") objects_info = None # rest joint positions were previously computed once and kept ONLY in memory for the rest # of main()'s lifetime — no file ever recorded them, so a question like "is joint_2's target # actually different from its rest position, or just restating rest" was unanswerable after # a run finished. Saved once per sketch (not per attempt, since rest pose doesn't change # attempt to attempt) so it's always available for exactly this kind of check. if objects_info: rest_joints_dump = { obj_name: { f"joint_{i}": {"x": float(j[0]), "y": float(j[1]), "is_anchor": i == info["anchor_idx"], "part_name": info.get("part_names", {}).get(i)} for i, j in enumerate(info["joints"]) } for obj_name, info in objects_info.items() } with open(os.path.join(json_dir, f"{name}_rest_joints.json"), "w") as f: json.dump(rest_joints_dump, f, indent=2) # preprocessed bbox trajectory data — fixed ground truth, given to the # judge as spatial context for EVERY object (ARAP and TRAJ_ONLY alike), # not something the judge critiques or the generator controls real_trajectories = load_trajectories(traj_path) all_object_names = list(real_trajectories.keys()) import dino_similarity dino_model, dino_processor = dino_similarity.load_dino_model() import clip_score clip_model, clip_processor = clip_score.load_clip_model() feedback = None previous_narratives = None # {obj_name: [5 descriptions]} from the last attempt previous_temp_dir = None # where the last attempt's rendered kf0..kf4 images live freeze_narrative = False # only frozen once faithfulness has already passed once consecutive_stagnant = 0 # early-stop if DINOv2 confirms no real change 2 attempts in a row joint_feedback = None # judge's structured per-joint corrections from the last attempt final_verdict = None winning_attempt = None all_attempts_summary = [] # NUMERIC CONTINUITY: every attempt after the first is given the PREVIOUS attempt's actual # numeric targets (baseline_targets) to refine, not just images/prose to reconstruct numbers # from scratch. This applies to EVERY retry, not just a special round after a pass. baseline_targets = None # CONFIRMATION ROUND: when an attempt first passes all three thresholds, don't stop # immediately — run exactly ONE more attempt (which, same as any retry now, gets the # passing attempt's real baseline_targets to refine) to see if it can be beaten, then keep # whichever actually scores higher. The passing attempt is NEVER at risk of being replaced # by something worse — if the confirmation attempt doesn't beat it, the original passing # attempt is kept exactly as if this mechanism didn't exist. Fires ONCE per sketch. first_pass_attempt = None # attempt number of the FIRST attempt that passed all thresholds first_pass_scores = None # that attempt's (faithfulness_score, plausibility_score) confirmation_used = False # True once the one extra confirmation attempt has been consumed # FULL conversation history — every past attempt's rendered images + narrative + feedback, # not just the most recent one (which baseline_targets/previous_narratives/previous_temp_dir # already cover). Lets the generator see its own trajectory across attempts, not just react # to the latest single critique in isolation. Each entry: # {"attempt": int, "temp_dir": str, "deform_outputs": {...}, "narratives": {...}, "feedback": str} # COST, not glossed over: grows every retry — by attempt 5 this is 4 past attempts x 5 images # = 20 extra images on top of rest pose, in a single call. Untested whether a small VLM's # effective attention holds up over that many attached images. attempt_history = [] for attempt in range(1, MAX_RETRIES + 2): # +2, not +1: room for exactly one confirmation # attempt beyond MAX_RETRIES if the pass happens # on the very last regular attempt if not objects_info: break if attempt > MAX_RETRIES and (first_pass_attempt is None or confirmation_used): # only allowed to exceed MAX_RETRIES for the ONE confirmation attempt — never for an # ordinary failed-retry continuation break label = f"{attempt}/{MAX_RETRIES}" if attempt <= MAX_RETRIES else f"{attempt} (CONFIRMATION, beyond normal {MAX_RETRIES})" print(f"\n########## ATTEMPT {label} ##########") attempt_json_dir = os.path.join(json_dir, "attempts", f"attempt_{attempt}") attempt_temp_dir = os.path.join(temp_dir, "attempts", f"attempt_{attempt}") os.makedirs(attempt_json_dir, exist_ok=True) # image list: always the rest pose, then EVERY past attempt's actual rendered keyframes # in order (not just the most recent one) — full conversation history, so Qwen sees the # complete visual trajectory of what it has already tried, not only its single latest # attempt. images = [Image.open(rest_pose_image_path).convert("RGB")] is_retry = attempt > 1 for past in attempt_history: images.extend(vlm_judge.load_keyframe_images(past["temp_dir"])) prompt = build_combined_narrate_deform_prompt( objects_info, caption, previous_narratives=previous_narratives, feedback=feedback, is_retry=is_retry, few_shot=not args.no_fewshot, attempt_history=attempt_history, freeze_narrative=freeze_narrative, joint_feedback=joint_feedback, baseline_targets=baseline_targets) # save the FULL generator prompt too, symmetric with the judge prompt below — otherwise # there's no way to confirm feedback/joint_feedback actually appeared in what Qwen was # shown, only to infer it from whether behavior changed afterward. with open(os.path.join(attempt_json_dir, f"{name}_narrate_deform_prompt.txt"), "w") as f: f.write(prompt) # resolve this attempt's temperature: schedule value if given (clamped to schedule's # last entry if attempt exceeds its length — e.g. the confirmation attempt beyond # MAX_RETRIES), otherwise the flat --narrate-temperature value unchanged. if temperature_schedule: schedule_idx = min(attempt - 1, len(temperature_schedule) - 1) this_attempt_temperature = temperature_schedule[schedule_idx] else: this_attempt_temperature = args.narrate_temperature print(f"\n---------- STEP 2+3: NARRATE+DEFORM (attempt {attempt}, " f"{len(images)} image{'s' if len(images) != 1 else ''}, " f"temperature={this_attempt_temperature}" f"{', narrative FROZEN' if freeze_narrative else ''}" f"{', MULTI-VIEW' if args.generator_views else ''}) ----------") vlm_model, vlm_processor = vlm_judge.load_vlm("Qwen/Qwen2.5-VL-3B-Instruct") if args.generator_views: parsed_or_none, per_view_raws = run_combined_narrate_deform_multiview( vlm_model, vlm_processor, images, prompt, temperature=this_attempt_temperature) # save EVERY view's raw response individually, same rationale as judge-votes — real # model outputs worth inspecting one by one, not just the aggregated result for view_i, raw in enumerate(per_view_raws): with open(os.path.join(attempt_json_dir, f"{name}_narrate_deform_raw_response_view{view_i}.txt"), "w") as f: f.write(raw) response = json.dumps(parsed_or_none) if parsed_or_none is not None else per_view_raws[0] with open(os.path.join(attempt_json_dir, f"{name}_narrate_deform_raw_response.txt"), "w") as f: f.write(f"[AGGREGATED across {len(per_view_raws)} views]\n" + response) if parsed_or_none is None: print(f"FAILED TO PARSE: every one of {len(per_view_raws)} views failed\nraw: {per_view_raws}") vlm_model = unload_model(vlm_model) feedback = "the previous attempt's output could not be parsed; produce valid JSON in the exact requested format" all_attempts_summary.append({"attempt": attempt, "plausibility_score": None, "note": "narrate+deform parse failed"}) continue parsed = parsed_or_none else: response = run_combined_narrate_deform(vlm_model, vlm_processor, images, prompt, temperature=this_attempt_temperature) # save the RAW pre-parse response unconditionally, before anything downstream can fail # or silently collapse it — previously this text existed only transiently in memory and # was discarded the moment parsing succeeded, so a case like identical coordinates across # every keyframe couldn't be traced back to "did Qwen write that itself" vs "did something # downstream produce it" after the fact. Written to the SAME attempt_json_dir the parsed # outputs already live in, so raw and parsed are side by side for direct comparison. with open(os.path.join(attempt_json_dir, f"{name}_narrate_deform_raw_response.txt"), "w") as f: f.write(response) try: parsed = parse_json_response(response) except (ValueError, json.JSONDecodeError) as e: print(f"FAILED TO PARSE: {e}\nraw: {response}") vlm_model = unload_model(vlm_model) feedback = "the previous attempt's output could not be parsed; produce valid JSON in the exact requested format" all_attempts_summary.append({"attempt": attempt, "plausibility_score": None, "note": "narrate+deform parse failed"}) continue narratives_this_attempt, deform_outputs_this_attempt, cap_utilization_this_attempt = apply_deform_clip_and_write( parsed, objects_info, attempt_json_dir, name, frozen_narratives=previous_narratives if freeze_narrative else None) with open(os.path.join(attempt_json_dir, f"{name}_narratives.json"), "w") as f: json.dump(narratives_this_attempt, f, indent=2) # print each ARAP object's ACTUAL joint targets per keyframe — this is the real # signal for "did deformation happen", unlike points_after_move_range in the render # log below, which is the whole object's bbox and can stay constant even when a # small joint (e.g. a hand) moves substantially, since the head/torso/limb extremes # usually dominate the bbox regardless of hand position. print(f"\n---------- attempt {attempt}: actual joint targets per keyframe ----------") for obj_name, obj_targets in deform_outputs_this_attempt.items(): print(f" '{obj_name}':") for kf in range(N_KEYFRAMES): kf_key = f"kf{kf}" if kf_key in obj_targets: print(f" {kf_key}: {obj_targets[kf_key]}") print(f"\n---------- STEP 4: RENDER (attempt {attempt}, no Qwen) ----------") run_render(attempt_json_dir, svg_path, semantic_path, traj_path, frames_dir=attempt_temp_dir) if args.deform_only: print(f"\n########## --deform-only: STOPPING after attempt 1, no judge ##########") print(f"cap_utilization (raw, unfiltered by any threshold):") for obj, frac in (cap_utilization_this_attempt or {}).items(): print(f" {obj}: {frac*100:.1f}% of allowed movement used") print(f"\nInspect the actual render directly at: {attempt_temp_dir}") print(f"(kf0.png ... kf4.png, plus the combined strip)") sys.exit(0) stagnation_result = None if previous_temp_dir: print(f"\n---------- STAGNATION CHECK (attempt {attempt} vs attempt {attempt - 1}) ----------") prev_dino_images = vlm_judge.load_keyframe_images(previous_temp_dir) curr_dino_images = vlm_judge.load_keyframe_images(attempt_temp_dir) stagnation_result = dino_similarity.stagnation_score( dino_model, dino_processor, prev_dino_images, curr_dino_images) stagnant = dino_similarity.is_stagnant(stagnation_result) print(f"mean attempt-to-attempt similarity: {stagnation_result['mean_similarity']:.4f} " f"({'STAGNANT' if stagnant else 'changed'})") consecutive_stagnant = consecutive_stagnant + 1 if stagnant else 0 print(f"\n---------- TEMPORAL CONSISTENCY (attempt {attempt}, diagnostic only) ----------") temporal_images = vlm_judge.load_keyframe_images(attempt_temp_dir) temporal_result = dino_similarity.temporal_consistency(dino_model, dino_processor, temporal_images) clip_result = None if caption: print(f"\n---------- CLIP SCORE (attempt {attempt}) ----------") clip_images = vlm_judge.load_keyframe_images(attempt_temp_dir) clip_result = clip_score.compute_sequence_clip_scores(clip_model, clip_processor, clip_images, caption) print(f"\n---------- STEP 5: JUDGE (attempt {attempt}, images + joint/bbox coordinates) ----------") # reuse the SAME already-loaded VLM for judging — no reload # needed, since narrate+deform and judge are both Qwen3-VL calls now. # Judge sees: rest pose + this attempt's 5 rendered keyframes # (images), PLUS rest-pose stroke text + joint legend + this # attempt's joint targets + every object's fixed bbox trajectory # + PER-KEYFRAME DEFORMED stroke geometry (text) — the last one # is what target_coords grounding actually anchors against: a # visually-identified region's TRUE coordinate at that specific # keyframe, not its rest-pose coordinate (which is only correct # when a keyframe happens to match rest). judge_images = [Image.open(rest_pose_image_path).convert("RGB")] judge_images.extend(vlm_judge.load_keyframe_images(attempt_temp_dir)) deformed_geometry_text = build_all_keyframes_deformed_geometry_text( objects_info, deform_outputs_this_attempt) # save this — it's otherwise invisible after the fact. Needed to verify e.g. whether a # judge's target_coords anchor was itself built from stale/frozen coordinates (a real # failure mode: if deform_outputs_this_attempt is frozen across keyframes, this text # will be too, and a judge "fix" anchored to it would just point back at the same stuck # position rather than actually correcting anything). with open(os.path.join(attempt_json_dir, f"{name}_deformed_geometry_text.txt"), "w") as f: f.write(deformed_geometry_text) past_verdicts = [a for a in all_attempts_summary if "note" not in a] judge_prompt = vlm_judge.build_judge_prompt( name, caption=caption, dino_stagnation=stagnation_result, dino_temporal=temporal_result, clip_scores=clip_result, objects_info=objects_info, deform_outputs=deform_outputs_this_attempt, real_trajectories=real_trajectories, all_object_names=all_object_names, deformed_geometry_text=deformed_geometry_text, narratives=narratives_this_attempt, past_verdicts=past_verdicts) # save the FULL prompt too, not just the response — otherwise there's no way to see # exactly what the judge was shown (only what it said back), which makes it impossible # to distinguish "the judge reasoned badly" from "the judge was given bad/stale input". with open(os.path.join(attempt_json_dir, f"{name}_judge_prompt.txt"), "w") as f: f.write(judge_prompt) judge_result, judge_raw_responses = vlm_judge.run_judge_voted( vlm_model, vlm_processor, judge_images, judge_prompt, n_votes=args.judge_votes, valid_arap_objects=set(objects_info.keys())) vlm_model = unload_model(vlm_model) # save EVERY raw vote, not just one — with judge_votes>1 there are multiple real model # outputs worth being able to inspect individually, not just the synthesized median. for vote_i, raw in enumerate(judge_raw_responses): suffix = f"_vote{vote_i}" if len(judge_raw_responses) > 1 else "" with open(os.path.join(attempt_json_dir, f"{name}_judge_raw_response{suffix}.txt"), "w") as f: f.write(raw) try: # judge_result is a parsed/synthesized dict when voting succeeded (run_judge_voted did # its own parsing internally); it's a raw string when judge_votes<=1 (unchanged # single-call behavior needing parsing here) OR when every vote failed to parse # (run_judge_voted's own fallback) — check the actual TYPE returned, not just the # judge_votes argument, since both cases can legitimately return a string. if isinstance(judge_result, dict): verdict = judge_result else: verdict = vlm_judge.parse_judge_response(judge_result) except (ValueError, json.JSONDecodeError) as e: judge_response = judge_result if isinstance(judge_result, str) else judge_raw_responses[0] print(f"JUDGE FAILED TO PARSE: {e}\nraw: {judge_response}") print("Treating as a failed attempt, retrying without specific feedback.") feedback = "the previous attempt's evaluation could not be parsed; try a clearer, more varied pose progression" previous_narratives = narratives_this_attempt previous_temp_dir = attempt_temp_dir freeze_narrative = False # unknown state — safest to regenerate rather than assume faithfulness held joint_feedback = None # verdict didn't parse, so any joint_feedback in it is unusable/unknown — don't carry stale feedback forward # baseline_targets/attempt_history SHOULD still update here — the JUDGE's verdict # failed to parse, but the generation itself (deform_outputs_this_attempt) is real # and valid; losing numeric continuity because of an unrelated judge-parsing issue # would be a separate, avoidable regression. baseline_targets = deform_outputs_this_attempt attempt_history.append({ "attempt": attempt, "temp_dir": attempt_temp_dir, "deform_outputs": deform_outputs_this_attempt, "narratives": narratives_this_attempt, "feedback": feedback, "scores": None, # no real verdict — can't rank this attempt for best-of-N "joint_feedback": None, # verdict never parsed — no real corrections exist for this attempt }) all_attempts_summary.append({"attempt": attempt, "plausibility_score": None, "note": "judge parse failed"}) continue problems = vlm_judge.validate_judge_response(verdict, valid_arap_objects=set(objects_info.keys())) print("\n--- JUDGE VERDICT ---") print(json.dumps(verdict, indent=2)) if problems: for p in problems: print(f" VALIDATION PROBLEM: {p}") with open(os.path.join(attempt_json_dir, f"{name}_judge_verdict.json"), "w") as f: json.dump(verdict, f, indent=2) final_verdict = verdict winning_attempt = attempt score = verdict.get("plausibility_score") faith_score = verdict.get("faithfulness_score") quality_score = verdict.get("quality_score") all_attempts_summary.append({"attempt": attempt, "plausibility_score": score, "plausibility_notes": verdict.get("plausibility_notes"), "faithfulness_score": faith_score, "faithfulness_notes": verdict.get("faithfulness_notes"), "quality_score": quality_score, "quality_notes": verdict.get("quality_notes"), "overall_verdict": verdict.get("overall_verdict"), "dino_stagnant": dino_similarity.is_stagnant(stagnation_result) if stagnation_result else None}) plausibility_ok = isinstance(score, (int, float)) and score >= PLAUSIBILITY_THRESHOLD faithfulness_ok = faithfulness_passed(faith_score) quality_ok = isinstance(quality_score, (int, float)) and quality_score >= QUALITY_THRESHOLD print(f"\nplausibility_score = {score} (threshold = {PLAUSIBILITY_THRESHOLD}, " f"{'PASS' if plausibility_ok else 'FAIL'})") print(f"faithfulness_score = {faith_score} (threshold = {FAITHFULNESS_THRESHOLD}, " f"{'PASS' if faithfulness_ok else 'FAIL'})") print(f"quality_score = {quality_score} (threshold = {QUALITY_THRESHOLD}, " f"{'PASS' if quality_ok else 'FAIL'})") if plausibility_ok and faithfulness_ok and quality_ok and not args.force_all_attempts: if first_pass_attempt is None: # FIRST time passing — don't stop yet. Remember this as the safe fallback, run # exactly one more attempt to see if it can be beaten, THEN decide. first_pass_attempt = attempt first_pass_scores = (faith_score, score) print(f"Attempt {attempt} passed all thresholds — running ONE confirmation attempt " f"before finalizing, to see if it can be improved on. If the confirmation " f"attempt is not better, attempt {attempt} is kept exactly as-is. (It also " f"gets this attempt's real numeric targets as its baseline, same as any " f"normal retry now would.)") previous_narratives = narratives_this_attempt previous_temp_dir = attempt_temp_dir baseline_targets = deform_outputs_this_attempt joint_feedback = vlm_judge.compute_joint_feedback_deltas( verdict.get("joint_feedback"), deform_outputs_this_attempt) attempt_history.append({ "attempt": attempt, "temp_dir": attempt_temp_dir, "deform_outputs": deform_outputs_this_attempt, "narratives": narratives_this_attempt, "feedback": feedback, "scores": (faith_score, score, quality_score), "joint_feedback": joint_feedback, }) # feedback stays neutral encouragement, not a correction — this attempt already # passed, there's nothing specifically "wrong" to fix, just seeing if variation # produces something even better feedback = ("This attempt already passed all quality thresholds. This is an " "OPTIONAL confirmation attempt: try to match or improve on it, but do " "not discard what is already working.") freeze_narrative = False continue # do NOT break — proceed to the confirmation attempt else: # this IS the confirmation attempt (first_pass_attempt was already set) confirmation_used = True confirmation_scores = (faith_score, score) if confirmation_scores > first_pass_scores: print(f"Confirmation attempt {attempt} scored better " f"(faithfulness={faith_score}, plausibility={score}) than the original " f"passing attempt {first_pass_attempt} " f"(faithfulness={first_pass_scores[0]}, plausibility={first_pass_scores[1]}) " f"— using attempt {attempt} instead.") winning_attempt = attempt else: print(f"Confirmation attempt {attempt} did NOT score better than the original " f"passing attempt {first_pass_attempt} — keeping attempt {first_pass_attempt} " f"as originally found, discarding the confirmation attempt.") winning_attempt = first_pass_attempt break if first_pass_attempt is not None and not confirmation_used and not args.force_all_attempts: # confirmation attempt FAILED thresholds outright (didn't beat the pass AND didn't # even clear the bar itself) — keep the original passing attempt, don't treat this # as a real failure requiring further retries confirmation_used = True print(f"Confirmation attempt {attempt} did not pass thresholds — keeping the original " f"passing attempt {first_pass_attempt} as the final result.") winning_attempt = first_pass_attempt break if consecutive_stagnant >= 2 and not args.force_all_attempts: print(f"\nDINOv2 confirmed NO real change across {consecutive_stagnant} consecutive attempts " f"(attempt {attempt} vs {attempt-1}, and {attempt-1} vs {attempt-2}) — further retries " f"are very unlikely to help. Stopping early and keeping this attempt's result rather " f"than burning through the remaining {MAX_RETRIES - attempt} attempts.") break if args.force_all_attempts and plausibility_ok and faithfulness_ok and quality_ok: print(f"Attempt {attempt} passed all thresholds, but --force-all-attempts is set — " f"continuing to run every remaining attempt rather than stopping early. Best " f"attempt across the FULL run will be selected at the end.") previous_narratives = narratives_this_attempt previous_temp_dir = attempt_temp_dir # NUMERIC CONTINUITY (universal, every retry): carry this attempt's actual numeric # targets forward as the NEXT attempt's starting point to refine, not just images/prose # to reconstruct numbers from scratch. This is the core of the "attempt N+1 works on # attempt N's real output" redesign — replaces relying purely on visual/textual # reconstruction, which this session repeatedly found unreliable (copy-from-rest, # scalar-collapse, frozen joints). baseline_targets = deform_outputs_this_attempt # carry the judge's structured per-joint corrections into the NEXT # attempt's prompt — empty list is valid (judge found nothing # specific to fix), missing key means coordinate context wasn't # given to build_judge_prompt at all; either way, default to None # so format_joint_feedback_for_object just adds nothing. # compute_joint_feedback_deltas turns the judge's target_coords # into an exact pixel delta by SUBTRACTION against this attempt's # actual joint positions — arithmetic, not model math. Entries # where the judge couldn't ground a target (target_coords null) # are left as delta_px=None and fall back to prose-only feedback. joint_feedback = vlm_judge.compute_joint_feedback_deltas( verdict.get("joint_feedback"), deform_outputs_this_attempt) attempt_history.append({ "attempt": attempt, "temp_dir": attempt_temp_dir, "deform_outputs": deform_outputs_this_attempt, "narratives": narratives_this_attempt, "feedback": feedback, "scores": (faith_score, score, quality_score), "joint_feedback": joint_feedback, }) # freeze the narrative on the NEXT attempt only if faithfulness # already passed THIS attempt — no reason to keep regenerating # a story that's already correct, only the numbers need work freeze_narrative = faithfulness_ok if attempt == MAX_RETRIES: print("Below threshold on the final attempt — no retry left, skipping feedback construction.") else: # feedback is now ONE unified sentence describing what's wrong and what needs to # change, instead of pasting plausibility_notes/faithfulness_notes/quality_notes # together as three separately-labeled clauses. The three _notes fields (and the # three scores/thresholds) are UNCHANGED and still drive stopping/freeze logic — # this only changes what prose text gets sent back to the generator as feedback. # overall_verdict is the judge's own synthesized summary (already part of the # schema, previously unused for feedback) — using that instead of splicing notes # avoids asking the model to write three separate critiques when one coherent one # is what the generator actually needs to act on. feedback = verdict.get( "overall_verdict", verdict.get("plausibility_notes", "the pose progression needs to look more plausible")) # DINOv2 objective override: if the images barely changed from the # previous attempt, say so explicitly and forcefully — this is the # exact failure mode confirmed on real hardware (cannon1 ran all # MAX_RETRIES with no real change), where the judge's own text # critique was never specific enough for Qwen to act on. An # objective embedding-distance number doesn't have that problem. if stagnation_result and dino_similarity.is_stagnant(stagnation_result): feedback = (f"CRITICAL: your last attempt was measured as nearly IDENTICAL to the one " f"before it (DINOv2 similarity {stagnation_result['mean_similarity']:.3f}) — " f"you are NOT making real changes. You MUST produce substantially different " f"target positions this time, not a superficial rewording. Original feedback: {feedback}") print(f"Below threshold — retrying with feedback: {feedback!r} " f"(narrative will be {'FROZEN' if freeze_narrative else 'regenerated'})") # the line above only ever showed the flat plausibility/faithfulness/quality string — # joint_feedback is a SEPARATE variable also being carried into the next prompt (see # build_combined_narrate_deform_prompt's joint_feedback= argument), and was never # visible in this log even when it had real content. Printing it explicitly here so # it's possible to verify from the log whether per-joint correction is actually # happening on a given attempt, instead of having to infer it from final results. if joint_feedback: grounded = [f for f in joint_feedback if f.get("delta_px") is not None] ungrounded = [f for f in joint_feedback if f.get("delta_px") is None] print(f" joint_feedback being sent to next attempt: {len(grounded)} grounded " f"(with computed delta_px), {len(ungrounded)} ungrounded") for f in joint_feedback: tag = "GROUNDED" if f.get("delta_px") is not None else "UNGROUNDED" print(f" [{tag}] {f.get('object')}.joint_{f.get('joint')} @ kf{f.get('keyframe')}: " f"{f.get('issue')} (delta_px={f.get('delta_px')})") else: print(" joint_feedback being sent to next attempt: none (empty or not returned by judge)") else: if args.force_all_attempts: # ran every attempt to completion by design — pick the best-scoring one now. # Ranked by faithfulness_score first (tie-break: plausibility_score), same # convention used elsewhere this session. Entries with scores=None (a judge parse # failure) are excluded — nothing to rank them by. ranked = [e for e in attempt_history if e.get("scores") is not None] if ranked: def _rank_key(e): faith, plaus, qual = e["scores"] faith_key = faith if isinstance(faith, (int, float)) else -1 # "N/A" sorts lowest return (faith_key, plaus) best_entry = max(ranked, key=_rank_key) winning_attempt = best_entry["attempt"] print(f"\n--force-all-attempts: ran all {len(attempt_history)} attempts, best-scoring " f"is attempt {winning_attempt} (faithfulness={best_entry['scores'][0]}, " f"plausibility={best_entry['scores'][1]}, quality={best_entry['scores'][2]}).") else: print(f"\n--force-all-attempts: ran all attempts but NONE had a parseable verdict to " f"rank — falling back to the last attempt's result.") else: print(f"\nReached MAX_RETRIES ({MAX_RETRIES}) without meeting both thresholds. " f"Using the last attempt's result.") # print a compact table so the score progression across attempts is # visible in one place, not just scattered through the full log print("\n########## ATTEMPT SUMMARY ##########") for a in all_attempts_summary: print(f" attempt {a['attempt']}: plausibility={a.get('plausibility_score')} " f"faithfulness={a.get('faithfulness_score')} quality={a.get('quality_score')} " f"dino_stagnant={a.get('dino_stagnant')} " f"— {a.get('plausibility_notes') or a.get('note', '')}") if winning_attempt: import shutil winning_json = os.path.join(json_dir, "attempts", f"attempt_{winning_attempt}") winning_temp = os.path.join(temp_dir, "attempts", f"attempt_{winning_attempt}") for f in os.listdir(winning_json): shutil.copy2(os.path.join(winning_json, f), os.path.join(json_dir, f)) for f in os.listdir(winning_temp): src = os.path.join(winning_temp, f) if os.path.isfile(src): shutil.copy2(src, os.path.join(temp_dir, f)) print(f"\ncopied winning attempt ({winning_attempt}) to the top-level " f"json/{run_label}/ and P_1/{run_label}/ locations") print(f"all {len(all_attempts_summary)} attempts preserved under " f"json/{run_label}/attempts/ and P_1/{run_label}/attempts/ for comparison") else: print("\nNo ARAP objects — skipping narrate/deform/judge entirely.") temp_dir = os.path.join(args.out_dir, "P_1", run_label) print("\n########## RENDER (no Qwen) ##########") run_render(json_dir, svg_path, semantic_path, traj_path, frames_dir=temp_dir) if __name__ == "__main__": main()