#!/usr/bin/env python3 """ generate_storyboard_keyframes.py — Full Visual Production Pipeline Reads a storyboard JSON and runs the complete pipeline per shot: Step 1: T2I keyframe generation (fal.ai Flux 2 + LoRA) Step 2: Crop border artifacts (triptych mode only) Step 3: Gemini NBP upscale Step 4: WAN 2.2 FLF video generation (fal.ai) Usage: python3 generate_storyboard_keyframes.py leviathan/ --episode 1 python3 generate_storyboard_keyframes.py leviathan/ --episode 1 --shots 3,5,8 python3 generate_storyboard_keyframes.py leviathan/ --episode 1 --stage keyframes python3 generate_storyboard_keyframes.py leviathan/ --episode 1 --stage upscale python3 generate_storyboard_keyframes.py leviathan/ --episode 1 --stage video python3 generate_storyboard_keyframes.py leviathan/ --episode 1 --dry-run python3 generate_storyboard_keyframes.py leviathan/ --episode 1 --skip-existing python3 generate_storyboard_keyframes.py leviathan/ --episode 1 --qc python3 generate_storyboard_keyframes.py leviathan/ --episode 1 --qc --qc-retries 3 Env vars: FAL_KEY — fal.ai API key (required for T2I and video) GOOGLE_API_KEY — Gemini API key (required for upscale stage and --qc) Dependencies: pip install fal-client google-genai Pillow requests """ import argparse import json import os import sys import time from datetime import datetime from pathlib import Path from typing import Dict, List, Optional # ── LoRA Registry (loaded from per-project lora_registry.json) ─────────── # Add tools to path for train_lora import _tools_dir = str(Path(__file__).resolve().parent) if _tools_dir not in sys.path: sys.path.insert(0, _tools_dir) from train_lora import load_registry, get_inference_config from asset_naming import get_asset_code, build_asset_name, char_tag_from_list from cost_tracker import CostTracker from recoil.core.model_profiles import get_model # Add lib to path for prompt_compiler import _lib_dir = str(Path(__file__).resolve().parent.parent / "lib") if _lib_dir not in sys.path: sys.path.insert(0, _lib_dir) from prompt_compiler import ( compile as compile_prompt, build_layers as build_prompt_layers, PreviousShotContext, OverrideStore, _load_breakdown, _load_project_config, _build_spatial_directive, _strip_vfx_language, ) from prompt_validators import validate_verb_strength # Optional: Gemini QC (loaded when --qc flag is used) HAS_GEMINI_QC = False try: from gemini_qc import run_keyframe_qc, resolve_character_refs HAS_GEMINI_QC = True except ImportError: pass # QC retry budget — max regen attempts per shot after auto_reject QC_MAX_RETRIES = 2 # Module-level registry, populated in main() after project_dir is known LORA_REGISTRY: Dict[str, dict] = {} # Module-level breakdown + config, populated in main() BREAKDOWN: dict = {} PROJECT_CONFIG: dict = {} OVERRIDE_STORE: Optional[OverrideStore] = None # ── Model Configurations ───────────────────────────────────────────────── MODEL_CONFIGS = { "flux2": { "endpoint": "fal-ai/flux-2/lora", "label": "Flux 2 Dev", "defaults": { "width": 768, "height": 1344, "steps": 28, "guidance": 2.5, }, "lora_key": "t2i_path", # key in LORA_REGISTRY for this model's LoRA "supports_img2img": True, # can pass image_url + strength "supports_guidance": True, }, "z_image": { "endpoint": "fal-ai/z-image/turbo/lora", "endpoint_img2img": "fal-ai/z-image/turbo/image-to-image/lora", "label": "Z-Image Turbo", "defaults": { "width": 768, "height": 1344, "steps": 8, "guidance": None, # Z-Image doesn't use guidance_scale }, "lora_key": "z_image_t2i_path", "supports_img2img": True, # via dedicated image-to-image/lora endpoint "supports_guidance": False, }, } # Active model config — set in main() based on --model flag ACTIVE_MODEL = MODEL_CONFIGS["z_image"] T2I_DEFAULTS = { "width": 768, "height": 1344, "steps": 28, "guidance": 2.5, "seed": 42, } VIDEO_ENDPOINT = "fal-ai/wan/v2.2-a14b/image-to-video/lora" VIDEO_DEFAULTS = { "steps": 40, "guidance": 5.0, "seed": 42, "num_frames": 49, "fps": 16, "resolution": "720p", "aspect_ratio": "9:16", } UPSCALE_CROP_PX = 4 # Triptych border artifact removal UPSCALE_MODEL = get_model("upscale", "image") # ── Legacy Triptych Constants (strip approach — deprecated) ────────────── TRIPTYCH_WIDTH = 1536 # Each panel ~512x912 TRIPTYCH_HEIGHT = 912 TRIPTYCH_CROP_PX = 12 # Wider trim for model-rendered divider misalignment TRIPTYCH_PANEL_BUDGET = 40 # Max words per panel description # ── Generation Type Detection ───────────────────────────────────────────── # # Pipeline architecture (Feb 2026): # Hero frame generated via Flux T2I + LoRA (always). # First/last frames derived via Gemini NBP from hero image (nbp_derive: True). # NBP gets the storyboard's first_frame/last_frame text as the target description # and the hero image as environment/style reference. # Frame count is driven by storyboard content: # - first_frame exists → derive first via NBP # - last_frame exists → derive last via NBP # - neither → hero only # Fallback: if no GOOGLE_API_KEY, falls back to Flux img2img chain. # # LoRA scale: 1.0 solo, 0.5 dual (codified in lora_registry.json). GENERATION_TYPES = { "triptych_split_flf": {"keyframes": ["first", "mid", "last"], "video_calls": 2, "nbp_derive": True}, "standard_flf": {"keyframes": ["first", "mid", "last"], "video_calls": 1, "nbp_derive": True}, "held_frame_push": {"keyframes": ["first"], "video_calls": 0, "nbp_derive": False}, "held_frame_static": {"keyframes": ["first"], "video_calls": 0, "nbp_derive": False}, # Legacy aliases (v2 storyboards) "wan_i2v": {"keyframes": ["first", "last"], "video_calls": 1, "nbp_derive": True}, "wan_flf_reaction": {"keyframes": ["first", "mid", "last"], "video_calls": 2, "nbp_derive": True}, "held_frame_with_push": {"keyframes": ["first"], "video_calls": 0, "nbp_derive": False}, } def determine_generation_type(shot: dict) -> str: """Determine how to generate this shot based on storyboard data. V3 storyboards have explicit 'generation_approach' field. V2 storyboards fall back to 'generation_type' or inference. """ # V3: explicit generation_approach (preferred) gen_approach = shot.get("generation_approach") if gen_approach and gen_approach in GENERATION_TYPES: return gen_approach # V2: explicit generation_type gen_type = shot.get("generation_type") if gen_type and gen_type in GENERATION_TYPES: return gen_type # Infer from shot characteristics (v2 fallback) shot_type = shot.get("shot_type", "MS").upper() has_last = bool(shot.get("last_frame")) motion = shot.get("camera_movement", "static").lower() if shot_type in ("ECU", "CU") and motion == "static" and not has_last: return "held_frame_static" if has_last: action = shot.get("action", "").lower() occlusion_cues = ["spins", "turns", "wrenches", "opens", "slams", "rips", "expression changes", "realization", "recognition"] if any(cue in action for cue in occlusion_cues): return "triptych_split_flf" return "standard_flf" if motion in ("dolly", "push", "crane", "zoom", "track"): return "held_frame_push" return "held_frame_static" # ── Character Detection ─────────────────────────────────────────────────── def detect_characters(shot: dict, storyboard: dict) -> List[str]: """Detect which characters are in a shot.""" # Check explicit field chars_in_shot = shot.get("characters_in_shot", []) if chars_in_shot: return [c.lower() for c in chars_in_shot] # Scan text fields for character names search_text = " ".join([ shot.get("first_frame", ""), shot.get("last_frame", ""), shot.get("subject", ""), shot.get("name", ""), shot.get("script_excerpt", ""), shot.get("action", ""), ]).lower() storyboard_chars = storyboard.get("characters", {}) found = [] for char_name in storyboard_chars: if char_name.lower() in search_text: found.append(char_name.lower()) return found # ── Prompt Building ─────────────────────────────────────────────────────── # Track previous shot context (layers + metadata) for transition-aware prompts PREVIOUS_SHOT_CONTEXT: Optional[PreviousShotContext] = None # Episode number, set in main() CURRENT_EPISODE: int = 0 def get_lora_path_for_model(char_reg: dict) -> Optional[str]: """Get the LoRA path for the active model from a character's registry entry.""" lora_key = ACTIVE_MODEL["lora_key"] return char_reg.get(lora_key) def build_t2i_prompt(shot: dict, storyboard: dict, frame_type: str = "first") -> dict: """Build prompt for T2I using the prompt compiler. Returns: dict with "prompt", "negative_prompt", "prompt_hash", "layers", etc. For backward compat, string access still works via result["prompt"]. """ global PREVIOUS_SHOT_CONTEXT # Mid-frame (FMLF anchor) — route through compiler as "hero" frame type. # Mid IS the hero/peak-action frame, just generated first in FMLF order. # Previously this bypassed the compiler entirely — no spatial, no overrides, # no VFX stripping, no wardrobe/environment resolution. if frame_type == "mid": frame_type = "hero" inherit = PREVIOUS_SHOT_CONTEXT if not shot.get("scene_break_before") else None model_key = [k for k, v in MODEL_CONFIGS.items() if v is ACTIVE_MODEL][0] overrides = OVERRIDE_STORE.list_all() if OVERRIDE_STORE else None result = compile_prompt( shot=shot, breakdown=BREAKDOWN, episode=CURRENT_EPISODE, storyboard=storyboard, model=model_key, lora_registry=LORA_REGISTRY, overrides=overrides, project_config=PROJECT_CONFIG, previous_shot_context=inherit, frame_type=frame_type, ) # Pre-generation verb strength check — warn about weak/static prompts # before spending API credits. VerbStrengthValidator checks Layer 3 (action_pose) # for physical micro-details. "mid-stride pivots" > "stands in corridor". verb_warning = validate_verb_strength(result, shot, frame_type) if verb_warning: shot_id = shot.get("id", 0) print(f" WEAK VERB (S{shot_id:02d} {frame_type}): {verb_warning['detail_count']} micro-details " f"(need {2 if frame_type == 'hero' else 1}+). " f"{verb_warning['suggestion'][:100]}", file=sys.stderr) return result def synthesize_mid_frame(first_frame: str, action: str, shot: dict) -> str: """Synthesize a mid-action frame description from shot data.""" emotion = shot.get("emotion", "intense effort") mid_parts = [] if action: mid_parts.append(action.strip()) else: mid_parts.append(first_frame.strip()) mid_parts.append( f"Frozen mid-motion, peak action. {emotion}. " "Motion blur on extremities, debris suspended in air." ) return " ".join(mid_parts) # Triptych banned phrases — trigger illustration/comic mode instead of photographic # "triptych" is the ONLY safe term. See INNOVATIONS.md Section 4. import re as _re _TRIPTYCH_BANNED = [ (r"comic\s+strip", "triptych"), (r"storyboard\s+strip", "triptych"), (r"comic\s+panel[s]?", "triptych panels"), (r"comic\s+book\s+style", "cinematic"), (r"cartoon\s+style", "cinematic"), (r"manga\s+style", "cinematic"), (r"illustrated\s+panel[s]?", "photographic panels"), (r"drawn\s+panel[s]?", "photographic panels"), (r"anime\s+style", "cinematic"), ] _TRIPTYCH_BANNED_COMPILED = [(_re.compile(p, _re.IGNORECASE), repl) for p, repl in _TRIPTYCH_BANNED] def _sanitize_triptych_prompt(prompt: str) -> tuple: """Strip banned phrases from triptych prompts that trigger illustration mode. Returns: (sanitized_prompt, list_of_replacements_made) """ replacements = [] sanitized = prompt for pattern, replacement in _TRIPTYCH_BANNED_COMPILED: match = pattern.search(sanitized) if match: replacements.append(f"'{match.group()}' → '{replacement}'") sanitized = pattern.sub(replacement, sanitized) return sanitized, replacements def _truncate_panel_content(text: str, max_words: int = TRIPTYCH_PANEL_BUDGET) -> str: """Truncate text to word budget, preferring sentence boundaries.""" words = text.split() if len(words) <= max_words: return text # Try to cut at a sentence boundary within budget truncated = " ".join(words[:max_words]) # Find last sentence-ending punctuation within the truncated text for i in range(len(truncated) - 1, max(len(truncated) - 40, 0), -1): if truncated[i] in ".!": return truncated[: i + 1] # No sentence boundary found — hard cut with ellipsis-free trim return truncated.rstrip(",;: ") def _compact_character_ref(char_name: str, storyboard: dict, lora_registry: dict) -> str: """Build 'TRIGGER, core visual traits' (~10 words) for per-panel embedding. Takes the trigger word + first ~8 words of the character's visual description from the storyboard character data. """ trigger = lora_registry.get(char_name, {}).get("trigger", "") # Get visual description from storyboard characters char_data = storyboard.get("characters", {}).get(char_name, {}) visual = char_data.get("visual", "") if visual: # Take first ~8 words of visual description for core traits visual_words = visual.split()[:8] core_visual = " ".join(visual_words).rstrip(",;:. ") if trigger: return f"{trigger}, {core_visual}" return core_visual return trigger or char_name def compose_triptych_prompt(shot: dict, storyboard: dict) -> str: """Auto-compose a triptych strip prompt using the prompt compiler's resolved layers. When a storyboard marks a shot as triptych_split_flf but doesn't provide a triptych_prompt, this function builds one from the validated template (see INNOVATIONS.md, Section 4). Template structure: - Spatial directive (Layer 0) — character position/facing - Shared DNA: compiler-resolved subject + wardrobe + environment - Left panel = first_frame (anticipation) - Center panel = hero_frame or action (peak action) - Right panel = last_frame (aftermath) - Compiler-resolved camera/film/quality footer Routes through the prompt compiler for: - Breakdown data resolution (character visual, wardrobe, environment, lighting) - Override system (global/character/episode/shot-scoped) - Edge continuity (inherits layers from previous shot) - VFX language stripping (post-compose) - Spatial directives (character position/facing) - Model-aware quality guard LoRA triggers are embedded per-panel based on character presence in each panel's prose. This prevents character bleeding into environment-only panels. Callers should NOT prepend triggers — they are already positioned. """ episode = storyboard.get("episode", 1) model_key = [k for k, v in MODEL_CONFIGS.items() if v is ACTIVE_MODEL][0] # Build resolved layers via the prompt compiler layers = build_prompt_layers( shot=shot, breakdown=BREAKDOWN, episode=episode, storyboard=storyboard, model=model_key, lora_registry=LORA_REGISTRY, project_config=PROJECT_CONFIG, frame_type="hero", ) # Extract compiler-resolved data subject = layers.subject or "the character" wardrobe = layers.wardrobe_props or "" environment = layers.environment or "" lighting = layers.lighting or "" camera_lens = layers.camera_lens or "" film_style = layers.film_style or "" quality_guard = layers.quality_guard or "" color_objects = layers.color_objects or "" # Build spatial directive (Layer 0) spatial = _build_spatial_directive(shot, model=model_key) # ── Panel content from shot prose fields ── first_frame = (shot.get("first_frame") or "").strip() hero_frame = (shot.get("hero_frame") or "").strip() action = (shot.get("action") or "").strip() last_frame = (shot.get("last_frame") or "").strip() center_content = hero_frame or action or "Peak action moment, maximum exertion" left_content = first_frame or "Anticipation, preparation, coiled tension" right_content = last_frame or "Aftermath, result, settling" # ── Per-panel LoRA trigger positioning ── # Instead of prepending ALL triggers globally (which forces character into # every panel), embed triggers only in panels where a character is mentioned. # Triptych = single image/prompt/LoRA load, so the LoRA is always active — # but trigger POSITION in the prompt controls WHERE the model places the character. char_triggers = {} # {char_name: trigger_word} for char_name in (shot.get("characters_in_shot") or []): cn = char_name.lower() if cn in LORA_REGISTRY and get_lora_path_for_model(LORA_REGISTRY[cn]): char_triggers[cn] = LORA_REGISTRY[cn]["trigger"] # Also check storyboard characters if characters_in_shot is empty if not char_triggers: for cn in storyboard.get("characters", {}): cn_lower = cn.lower() if cn_lower in LORA_REGISTRY and get_lora_path_for_model(LORA_REGISTRY[cn_lower]): char_triggers[cn_lower] = LORA_REGISTRY[cn_lower]["trigger"] # Human presence indicators — if a panel contains ANY of these, it likely # depicts a character (not pure environment). We check these instead of just # the character name because storyboard prose often uses "figure", "she", etc. _HUMAN_INDICATORS = { "figure", "person", "woman", "man", "girl", "boy", "child", "face", "hand", "hands", "arm", "arms", "shoulder", "shoulders", "body", "torso", "leg", "legs", "foot", "feet", "fist", "jaw", "chin", "cheek", "cheekbone", "cheekbones", "neck", "mouth", "she", "her", "he", "his", "their", "them", "braces", "grips", "drives", "wrenches", "leverages", "muscles", } def _panel_has_character(panel_text: str, char_name: str) -> bool: """Check if a panel depicts a character (by name or human presence indicators).""" text_lower = panel_text.lower() # Direct name match if char_name in text_lower: return True # Human presence indicators words = set(text_lower.split()) return bool(words & _HUMAN_INDICATORS) def _prepend_triggers_to_panel(panel_text: str) -> str: """Prepend relevant character triggers to a panel description.""" panel_triggers = [] for cn, trigger in char_triggers.items(): if _panel_has_character(panel_text, cn): panel_triggers.append(trigger) if panel_triggers: return ", ".join(panel_triggers) + ", " + panel_text return panel_text left_content = _prepend_triggers_to_panel(left_content) center_content = _prepend_triggers_to_panel(center_content) right_content = _prepend_triggers_to_panel(right_content) # ── Assemble using INNOVATIONS.md template ── # Key: "triptych" language keeps output photographic/cinematic. # "storyboard strip" or "comic strip" triggers illustration mode. parts = [] if spatial: parts.append(spatial) parts.append( "A horizontal triptych of three vertical panels showing a continuous action sequence," ) parts.append(f"left to right, of {subject}.") if wardrobe: parts.append(wardrobe + ".") if environment: parts.append(environment + ".") if lighting: parts.append(lighting + ".") parts.append(f"\nLeft panel — ANTICIPATION: {left_content}") parts.append(f"\nCenter panel — PEAK ACTION: {center_content}") parts.append(f"\nRight panel — AFTERMATH: {right_content}") footer_parts = [] if camera_lens: footer_parts.append(camera_lens) if film_style: footer_parts.append(film_style) if color_objects: footer_parts.append(color_objects) if quality_guard: footer_parts.append(quality_guard) footer_parts.append("consistent character and environment across all panels") parts.append(f"\nAll three panels: {'. '.join(footer_parts)}.") composed = " ".join(parts) # Apply VFX language stripping (same gate as prompt compiler) composed, vfx_stripped = _strip_vfx_language(composed) if vfx_stripped: print(f" strip: VFX STRIPPED — {', '.join(vfx_stripped)}") return composed # ── Crop Regions (for punch-in continuity) ─────────────────────────────── CROP_REGIONS = { # (left%, top%, right%, bottom%) as fractions of image size "full": (0.0, 0.0, 1.0, 1.0), "upper_third": (0.0, 0.0, 1.0, 0.33), "lower_third": (0.0, 0.67, 1.0, 1.0), "center": (0.25, 0.25, 0.75, 0.75), "face": (0.2, 0.05, 0.8, 0.4), "wrist_left": (0.4, 0.35, 0.85, 0.65), "hands": (0.15, 0.4, 0.85, 0.7), } def crop_region_from_image(image_path: str, region: str, output_path: str) -> str: """Crop a region from a local image for punch-in continuity. Args: image_path: Path to source image. region: Region preset name from CROP_REGIONS. output_path: Where to save the cropped image. Returns: Path to the cropped image. """ from PIL import Image img = Image.open(image_path) w, h = img.size coords = CROP_REGIONS.get(region, CROP_REGIONS["center"]) left = int(w * coords[0]) top = int(h * coords[1]) right = int(w * coords[2]) bottom = int(h * coords[3]) cropped = img.crop((left, top, right, bottom)) Path(output_path).parent.mkdir(parents=True, exist_ok=True) cropped.save(output_path) return output_path def _determine_frame_order(shot: dict, no_fmlf: bool = False) -> List[str]: """Determine generation order for a shot's frames. Hero-first workflow: - If shot has hero_action: hero → first → last - Triptych: strip generation (unchanged) - Standard FLF: mid → first → last (FMLF default — anchor-first) - Held frame: first only FMLF is now the DEFAULT for all standard_flf shots. Mid (peak action) is generated first as the anchor, then first and last are derived via img2img. Use no_fmlf=True to fall back to legacy first → last (2 frames). """ gen_type = determine_generation_type(shot) type_info = GENERATION_TYPES[gen_type] # NBP-derive shots handled separately (hero + NBP derivation path) if type_info.get("nbp_derive"): return type_info["keyframes"] # Hero-first: if verb states exist, generate hero first if shot.get("hero_action"): return ["hero", "first", "last"] # FMLF default: mid is the anchor (peak action), generated first. # First (anticipation) and last (aftermath) are derived from mid via img2img. if not no_fmlf and type_info["keyframes"] == ["first", "last"]: return ["mid", "first", "last"] # Legacy fallback (--no-fmlf): use original frame order return type_info["keyframes"] # ── fal.ai API Calls ───────────────────────────────────────────────────── def generate_t2i( prompt: str, characters: List[str], width: int = T2I_DEFAULTS["width"], height: int = T2I_DEFAULTS["height"], steps: int = T2I_DEFAULTS["steps"], guidance: float = T2I_DEFAULTS["guidance"], seed: int = T2I_DEFAULTS["seed"], image_url: Optional[str] = None, strength: Optional[float] = None, negative_prompt: str = "", ) -> dict: """Generate a T2I image via fal.ai (Flux 2 or Z-Image Turbo) + LoRA. Args: image_url: Source image URL for img2img conditioning (spatial coherence). strength: How much to deviate from source (0.0=exact copy, 1.0=fully new). Only used when image_url is provided. negative_prompt: Negative prompt (only passed to models that support it). Returns: fal.ai result dict with 'images' key. """ import fal_client endpoint = ACTIVE_MODEL["endpoint"] lora_key = ACTIVE_MODEL["lora_key"] # Use dedicated img2img endpoint if available and we have a source image use_img2img = (image_url and strength is not None and ACTIVE_MODEL.get("supports_img2img", False)) if use_img2img and "endpoint_img2img" in ACTIVE_MODEL: endpoint = ACTIVE_MODEL["endpoint_img2img"] # Build LoRA list using the active model's LoRA paths # Use engine-specific scales from registry when available (e.g., flux2_scale_solo for Flux 2 Dev) is_flux2 = ACTIVE_MODEL.get("endpoint", "").startswith("fal-ai/flux-2") loras = [] num_chars = len([c for c in characters if c in LORA_REGISTRY and get_lora_path_for_model(LORA_REGISTRY[c])]) for char in characters: reg = LORA_REGISTRY.get(char) if reg: lora_path = get_lora_path_for_model(reg) if lora_path: if num_chars <= 1: scale = reg.get("flux2_scale_solo", reg["scale_solo"]) if is_flux2 else reg["scale_solo"] else: scale = reg.get("flux2_scale_dual", reg["scale_dual"]) if is_flux2 else reg["scale_dual"] # GATE: Flux 2 scale blowout — 1.3 causes muddy textures with Flux 2 Dev if is_flux2 and scale > 1.1: print(f" SCALE BLOWOUT BLOCKED: {char} LoRA scale {scale} for Flux 2 Dev " f"(max 1.1, recommended 1.0). Use flux2_scale_solo in registry.", file=sys.stderr) raise ValueError( f"Flux 2 Dev LoRA scale {scale} for {char} exceeds safe max (1.1). " f"Set flux2_scale_solo: 1.0 in lora_registry.json." ) loras.append({"path": lora_path, "scale": scale}) # GATE: Dual LoRA total scale cap — above 1.0 causes image corruption if len(loras) > 1: total_scale = sum(l["scale"] for l in loras) if total_scale > 1.0: char_scales = ", ".join(f"{l['path'].split('/')[-1]}={l['scale']}" for l in loras) print(f" DUAL LORA CAP BLOCKED: Total scale {total_scale:.2f} exceeds 1.0 " f"({char_scales}). Reduce to 0.5/0.5 or lower.", file=sys.stderr) raise ValueError( f"Dual LoRA total scale {total_scale:.2f} exceeds max (1.0). " f"Set scale_dual: 0.5 for each character in lora_registry.json." ) args = { "prompt": prompt, "image_size": {"width": width, "height": height}, "num_inference_steps": steps, "seed": seed, "num_images": 1, "output_format": "png", "enable_safety_checker": False, } # Only include guidance_scale if the model supports it if ACTIVE_MODEL["supports_guidance"] and guidance is not None: args["guidance_scale"] = guidance # Only include negative_prompt for models that support it if negative_prompt and ACTIVE_MODEL.get("supports_guidance", False): args["negative_prompt"] = negative_prompt if loras: args["loras"] = loras # img2img conditioning: pass source image for spatial coherence if use_img2img: args["image_url"] = image_url args["strength"] = strength result = fal_client.subscribe(endpoint, arguments=args) # Attach generation metadata for traceability meta = { "prompt": prompt, "loras": loras, "seed": seed, "steps": steps, "width": width, "height": height, "endpoint": endpoint, "model": next((k for k, v in MODEL_CONFIGS.items() if v["endpoint"] == endpoint or v.get("endpoint_img2img") == endpoint), "unknown"), } if guidance is not None and ACTIVE_MODEL["supports_guidance"]: meta["guidance"] = guidance if image_url: meta["img2img_source"] = image_url meta["img2img_strength"] = strength result["_gen_meta"] = meta return result def generate_video_flf( first_url: str, last_url: str, prompt: str, steps: int = VIDEO_DEFAULTS["steps"], guidance: float = VIDEO_DEFAULTS["guidance"], seed: int = VIDEO_DEFAULTS["seed"], num_frames: int = VIDEO_DEFAULTS["num_frames"], fps: int = VIDEO_DEFAULTS["fps"], negative_prompt: str = "", ) -> dict: """Generate FLF video via fal.ai WAN 2.2. Args: negative_prompt: Negative prompt (WAN 2.2 supports this). Returns: fal.ai result dict with 'video' key. """ import fal_client args = { "prompt": prompt, "image_url": first_url, "end_image_url": last_url, "num_inference_steps": steps, "guidance_scale": guidance, "seed": seed, "num_frames": num_frames, "frames_per_second": fps, "resolution": VIDEO_DEFAULTS["resolution"], "aspect_ratio": VIDEO_DEFAULTS["aspect_ratio"], "enable_safety_checker": False, } if negative_prompt: args["negative_prompt"] = negative_prompt result = fal_client.subscribe(VIDEO_ENDPOINT, arguments=args) return result # ── File Download ───────────────────────────────────────────────────────── def download_file(url: str, output_path: str) -> str: """Download a file from URL to local path.""" import requests resp = requests.get(url, timeout=120) resp.raise_for_status() Path(output_path).parent.mkdir(parents=True, exist_ok=True) with open(output_path, "wb") as f: f.write(resp.content) return output_path def upload_file(path: str) -> str: """Upload a local file to fal.ai CDN.""" import fal_client return fal_client.upload_file(path) # ── NBP Frame Derivation ───────────────────────────────────────────────── NBP_DERIVE_MODEL = get_model("exploration", "image") NBP_DERIVE_DELAY = 5.0 # seconds between Gemini calls (RPM limit) def derive_frame_nbp( hero_image_path: str, shot: dict, frame_type: str, storyboard: dict, output_path: str, client=None, tracker: Optional["CostTracker"] = None, ) -> Optional[str]: """Derive a first/last frame from a hero image using Gemini NBP reasoning. Instead of generating a single triptych strip, this sends the hero (peak action) image to Gemini NBP with a reasoning prompt asking it to generate the anticipation (first) or aftermath (last) moment of the same scene. NBP can reason about causality: "what did this scene look like 1 second before this peak moment?" — maintaining environment, lighting, and character identity from the reference image. Args: hero_image_path: Path to the hero frame image. shot: Shot data from storyboard. frame_type: "first" (anticipation) or "last" (aftermath). storyboard: Full storyboard data. output_path: Where to save the derived image. client: Existing genai.Client (created if None). tracker: CostTracker instance for cost logging (optional). Returns: Path to derived image, or None on failure. """ import io from PIL import Image if client is None: from google import genai api_key = os.environ.get("GOOGLE_API_KEY") if not api_key: raise RuntimeError("GOOGLE_API_KEY not set (required for NBP frame derivation)") client = genai.Client(api_key=api_key) from google.genai import types # Load and encode the hero image img = Image.open(hero_image_path) buf = io.BytesIO() fmt = "PNG" if Path(hero_image_path).suffix.lower() == ".png" else "JPEG" img.save(buf, format=fmt, quality=95) image_bytes = buf.getvalue() mime = "image/png" if fmt == "PNG" else "image/jpeg" # Build the derivation prompt — direct and specific, not vague temporal references. # NBP is a reasoning model: tell it exactly what you want to see, not "1 second before." lighting = shot.get("lighting", "") if frame_type == "first": frame_content = (shot.get("anticipation_action") or shot.get("first_frame") or "") elif frame_type == "last": frame_content = (shot.get("aftermath_action") or shot.get("last_frame") or "") else: raise ValueError(f"derive_frame_nbp only supports 'first' or 'last', got '{frame_type}'") if not frame_content: print(f" {frame_type}: no first_frame/last_frame content in storyboard — skipping") return None prompt = ( f"Use this reference image for environment, lighting, color palette, " f"photographic style, grain, and aspect ratio ONLY.\n\n" f"Generate a NEW image matching this exact description:\n" f"{frame_content}\n\n" f"REQUIREMENTS:\n" f"- Match the reference image's environment, lighting ({lighting[:80]}), and vertical 9:16 framing\n" f"- Match the reference image's photographic style and film grain exactly\n" f"- The content and composition must match the description above, NOT the reference image\n" f"- Photorealistic, cinematic" ) input_filename = Path(hero_image_path).name t0 = time.time() try: response = client.models.generate_content( model=NBP_DERIVE_MODEL, contents=[ prompt, types.Part.from_bytes(data=image_bytes, mime_type=mime), ], config=types.GenerateContentConfig( response_modalities=["IMAGE", "TEXT"], ), ) except Exception as e: elapsed_ms = int((time.time() - t0) * 1000) if tracker: tracker.log( category="nbp_derive", provider="gemini", model=NBP_DERIVE_MODEL, images_out=0, duration_ms=elapsed_ms, detail=f"NBP derive {frame_type}: {input_filename} — {str(e)[:100]}", success=False, ) print(f" {frame_type}: NBP derive FAILED — {e}") return None elapsed_ms = int((time.time() - t0) * 1000) # Extract token usage tokens_in = 0 tokens_out = 0 if hasattr(response, 'usage_metadata') and response.usage_metadata: tokens_in = getattr(response.usage_metadata, 'prompt_token_count', 0) or 0 tokens_out = getattr(response.usage_metadata, 'candidates_token_count', 0) or 0 # Extract image from response if response.candidates: for part in response.candidates[0].content.parts: if part.inline_data and part.inline_data.mime_type.startswith("image/"): Path(output_path).parent.mkdir(parents=True, exist_ok=True) with open(output_path, "wb") as f: f.write(part.inline_data.data) if tracker: tracker.log( category="nbp_derive", provider="gemini", model=NBP_DERIVE_MODEL, images_out=1, tokens_in=tokens_in, tokens_out=tokens_out, duration_ms=elapsed_ms, detail=f"NBP derive {frame_type}: {input_filename}", success=True, ) print(f" {frame_type}: NBP derived → {Path(output_path).name} ({elapsed_ms}ms)") return output_path # No image in response text_parts = [] if response.candidates: for part in response.candidates[0].content.parts: if part.text: text_parts.append(part.text) if text_parts: print(f" {frame_type}: NBP returned text instead of image: {' '.join(text_parts)[:200]}") else: print(f" {frame_type}: NBP returned no image or text") if tracker: tracker.log( category="nbp_derive", provider="gemini", model=NBP_DERIVE_MODEL, images_out=0, tokens_in=tokens_in, tokens_out=tokens_out, duration_ms=elapsed_ms, detail=f"NBP derive {frame_type}: {input_filename} — no image returned", success=False, ) return None # ── Per-Shot Pipeline ───────────────────────────────────────────────────── def _find_shot_result(all_results: list, shot_id: int) -> Optional[dict]: """Find a processed shot result by shot_id.""" for r in all_results: if r.get("shot_id") == shot_id: return r return None def process_shot( shot: dict, storyboard: dict, output_dir: Path, stage: str, args, gemini_client=None, stats: dict = None, all_results_ref: list = None, tracker: "CostTracker | None" = None, ) -> dict: """Run the full pipeline for a single shot. Args: shot: Shot data from storyboard. storyboard: Full storyboard. output_dir: Flat assets directory (storyboards/assets/ep_NNN/). stage: How far to run ('keyframes', 'upscale', 'video', 'full'). args: CLI args namespace. gemini_client: Reusable Gemini client. stats: Mutable stats dict. all_results_ref: List of all shot results so far (for continuity lookups). Returns: Shot result dict for manifest. """ if all_results_ref is None: all_results_ref = [] shot_id = shot["id"] shot_name = shot.get("name", f"shot_{shot_id:02d}") gen_type = determine_generation_type(shot) characters = detect_characters(shot, storyboard) keyframe_types = GENERATION_TYPES[gen_type]["keyframes"] video_calls = GENERATION_TYPES[gen_type]["video_calls"] use_nbp = GENERATION_TYPES[gen_type].get("nbp_derive", False) # Asset naming (v3) or legacy naming (v2) prj = get_asset_code(Path(args.project_dir).resolve()) char_tag = char_tag_from_list(characters) if hasattr(args, 'new_take') and args.new_take: from asset_naming import next_take_number take = next_take_number(Path(args.project_dir).resolve(), storyboard["episode"], shot_id, char_tag) asset_base = build_asset_name(prj, storyboard["episode"], shot_id, take, char_tag) else: asset_base = shot.get("asset_name") if not asset_base: asset_base = build_asset_name(prj, storyboard["episode"], shot_id, 1, char_tag) # Suffix map: keyframe type → file suffix SUFFIX_MAP = {"first": "_f1", "mid": "_f2", "last": "_f3"} result = { "shot_id": shot_id, "name": shot_name, "generation_type": gen_type, "asset_name": asset_base, "characters": characters, "status": "pending", "keyframes": {}, "upscaled": {}, "videos": {}, } # Check LoRA availability for active model blocked_chars = [] for char in characters: reg = LORA_REGISTRY.get(char) if reg and not get_lora_path_for_model(reg): blocked_chars.append(char) if blocked_chars and all(c in blocked_chars for c in characters): result["status"] = "blocked_lora" result["blocked_reason"] = f"No {ACTIVE_MODEL['label']} LoRA for: {', '.join(blocked_chars)}" print(f" BLOCKED — no {ACTIVE_MODEL['label']} LoRA for {', '.join(blocked_chars)}") if stats: stats["blocked"] += 1 return result if blocked_chars: result["status"] = "partial_lora" result["partial_reason"] = f"Missing {ACTIVE_MODEL['label']} LoRA: {', '.join(blocked_chars)}" print(f" PARTIAL — missing {ACTIVE_MODEL['label']} LoRA for {', '.join(blocked_chars)}") # ── STEP 1: T2I KEYFRAME GENERATION ── if use_nbp: # Hero-first + NBP derivation: generate hero frame via Flux T2I + LoRA, # then use Gemini NBP to reason about before/after moments. # This replaces the old single-strip triptych approach which suffered from # panel bleed and divider misalignment at any dimension. SUFFIX_MAP_EXT = {"first": "_f1", "mid": "_f2", "last": "_f3", "hero": "_hero"} # Step 1: Generate hero frame via Flux T2I (peak action) hero_suffix = SUFFIX_MAP_EXT["mid"] # Hero = mid = peak action = _f2 hero_path = str(output_dir / f"{asset_base}{hero_suffix}.png") hero_url = None if args.skip_existing and Path(hero_path).exists(): print(f" hero: exists (skipped)") result["keyframes"]["mid"] = hero_path else: # Build hero prompt via compiler compile_result = build_t2i_prompt(shot, storyboard, "hero") prompt = compile_result["prompt"] negative_prompt = compile_result.get("negative_prompt", "") # Check for location ref or continuity source img2img_url = None img2img_strength = None location_ref_url = shot.get("reference_image_url") punch_in = shot.get("continuity_from") if punch_in and not getattr(args, 'no_img2img', False): punch_shot_id = punch_in.get("shot_id") punch_frame = punch_in.get("frame", "hero") punch_region = punch_in.get("region", "center") punch_strength = punch_in.get("strength", 0.30) prev_result = _find_shot_result(all_results_ref, punch_shot_id) if prev_result: prev_path = prev_result.get("keyframes", {}).get(punch_frame) if prev_path and Path(prev_path).exists(): crop_path = str(output_dir / f"{asset_base}_crop_src.png") crop_region_from_image(prev_path, punch_region, crop_path) img2img_url = upload_file(crop_path) img2img_strength = punch_strength print(f" hero: generating T2I [{ACTIVE_MODEL['label']}] (punch-in from S{punch_shot_id:02d}, strength={punch_strength})...") if not img2img_url and location_ref_url and not getattr(args, 'no_img2img', False): img2img_url = location_ref_url img2img_strength = args.location_ref_strength print(f" hero: generating T2I [{ACTIVE_MODEL['label']}] (location ref, strength={img2img_strength})...") elif not img2img_url: print(f" hero: generating T2I [{ACTIVE_MODEL['label']}]...") _frame_w = args.width if hasattr(args, 'width') else T2I_DEFAULTS["width"] _frame_h = args.height if hasattr(args, 'height') else T2I_DEFAULTS["height"] t0_t2i = time.monotonic() t2i_ok = False try: t2i_result = generate_t2i( prompt=prompt, characters=characters, width=_frame_w, height=_frame_h, steps=args.steps, guidance=args.guidance, seed=args.seed, image_url=img2img_url, strength=img2img_strength, negative_prompt=negative_prompt, ) t2i_ok = True if not t2i_result.get("images"): raise RuntimeError("fal.ai returned empty images array") img_url = t2i_result["images"][0]["url"] download_file(img_url, hero_path) hero_url = img_url print(f" hero: saved → {Path(hero_path).name}") if "_gen_meta" in t2i_result: result["generation_meta"] = t2i_result["_gen_meta"] if stats: stats["keyframes"] += 1 result["keyframes"]["mid"] = hero_path except Exception as e: print(f" hero: FAILED — {e}") result["status"] = "keyframe_failed" if stats: stats["failed"] += 1 finally: elapsed_t2i = int((time.monotonic() - t0_t2i) * 1000) if tracker: _model_key = "z_image_turbo" if "z_image" in ACTIVE_MODEL.get("endpoint", "") else "flux2" _lora_count = sum(1 for c in characters if c in LORA_REGISTRY and get_lora_path_for_model(LORA_REGISTRY[c])) tracker.log( category="generation", provider="fal", model=_model_key, resolution=f"{_frame_w}x{_frame_h}", loras=_lora_count, episode=args.episode, shot_id=shot["id"], duration_ms=elapsed_t2i, success=t2i_ok, detail=f"T2I hero (triptych→NBP): ep{args.episode:02d}/shot_{shot['id']:02d}", ) if not t2i_ok: return result time.sleep(args.delay) # Step 2: Derive first (anticipation) and last (aftermath) via Gemini NBP # NBP reasons about "what did this scene look like before/after the peak moment" # maintaining environment, character, and camera consistency from the hero image. if not gemini_client: api_key = os.environ.get("GOOGLE_API_KEY") if api_key: from google import genai gemini_client = genai.Client(api_key=api_key) else: print(f" WARNING: GOOGLE_API_KEY not set — cannot derive first/last via NBP") for derive_type in ["first", "last"]: suffix = SUFFIX_MAP_EXT[derive_type] derive_path = str(output_dir / f"{asset_base}{suffix}.png") if args.skip_existing and Path(derive_path).exists(): print(f" {derive_type}: exists (skipped)") result["keyframes"][derive_type] = derive_path continue if not gemini_client: print(f" {derive_type}: SKIPPED — no Gemini client") continue print(f" {derive_type}: deriving via NBP from hero ({NBP_DERIVE_MODEL})...") derived = derive_frame_nbp( hero_image_path=hero_path, shot=shot, frame_type=derive_type, storyboard=storyboard, output_path=derive_path, client=gemini_client, tracker=tracker, ) if derived: result["keyframes"][derive_type] = derived if stats: stats["keyframes"] += 1 else: print(f" {derive_type}: NBP derivation failed — shot will have hero only") if stats: stats["failed"] += 1 time.sleep(NBP_DERIVE_DELAY) else: # Per-frame T2I generation with hero-first + continuity chain hero_frame_url = None # Captured after generating hero frame first_frame_url = None # Captured after generating first frame mid_frame_url = None # Captured after generating mid frame (FMLF anchor) # Check for location reference image (img2img conditioning for first frame) location_ref_url = shot.get("reference_image_url") # Determine frame generation order frame_order = _determine_frame_order(shot, no_fmlf=getattr(args, 'no_fmlf', False)) # Check for continuity sources from previous shots punch_in = shot.get("continuity_from") same_angle = shot.get("same_angle_from") # Suffix map extension for hero frame SUFFIX_MAP_EXT = {"first": "_f1", "mid": "_f2", "last": "_f3", "hero": "_hero"} for frame_type in frame_order: suffix = SUFFIX_MAP_EXT.get(frame_type, SUFFIX_MAP.get(frame_type, f"_{frame_type}")) raw_path = str(output_dir / f"{asset_base}{suffix}.png") if args.skip_existing and Path(raw_path).exists(): print(f" {frame_type}: exists (skipped)") result["keyframes"][frame_type] = raw_path continue # Mid frame: route through prompt compiler with "hero" frame_type # (hero = peak moment of the shot, which is what mid represents) if frame_type == "mid": compile_result = build_t2i_prompt(shot, storyboard, "hero") prompt = compile_result["prompt"] negative_prompt = compile_result.get("negative_prompt", "") else: compile_result = build_t2i_prompt(shot, storyboard, frame_type) prompt = compile_result["prompt"] negative_prompt = compile_result.get("negative_prompt", "") # Determine img2img conditioning based on frame type + continuity img2img_url = None img2img_strength = None if frame_type == "hero": # Hero frame: punch-in from previous shot or fresh T2I if punch_in and not args.no_img2img: # Look up previous shot's frame in manifest punch_shot_id = punch_in.get("shot_id") punch_frame = punch_in.get("frame", "hero") punch_region = punch_in.get("region", "center") punch_strength = punch_in.get("strength", 0.30) # Try to find the frame in already-processed results prev_result = _find_shot_result(all_results_ref, punch_shot_id) if prev_result: prev_path = prev_result.get("keyframes", {}).get(punch_frame) if prev_path and Path(prev_path).exists(): crop_path = str(output_dir / f"{asset_base}_crop_src.png") crop_region_from_image(prev_path, punch_region, crop_path) img2img_url = upload_file(crop_path) img2img_strength = punch_strength print(f" {frame_type}: generating T2I [{ACTIVE_MODEL['label']}] (punch-in from S{punch_shot_id:02d} {punch_frame} {punch_region}, strength={punch_strength})...") else: print(f" {frame_type}: punch-in source not found, falling back to fresh T2I") else: print(f" {frame_type}: punch-in shot S{punch_shot_id:02d} not yet processed") if not img2img_url and location_ref_url and not args.no_img2img: img2img_url = location_ref_url img2img_strength = args.location_ref_strength print(f" {frame_type}: generating T2I [{ACTIVE_MODEL['label']}] (location ref, strength={img2img_strength})...") elif not img2img_url: print(f" {frame_type}: generating T2I [{ACTIVE_MODEL['label']}]...") elif frame_type == "first": if same_angle and not args.no_img2img: # Same-angle continuation: img2img from previous shot's last frame sa_shot_id = same_angle.get("shot_id") sa_frame = same_angle.get("frame", "last") sa_strength = same_angle.get("strength", 0.35) prev_result = _find_shot_result(all_results_ref, sa_shot_id) if prev_result: prev_path = prev_result.get("keyframes", {}).get(sa_frame) if prev_path and Path(prev_path).exists(): img2img_url = upload_file(prev_path) img2img_strength = sa_strength print(f" {frame_type}: generating T2I [{ACTIVE_MODEL['label']}] (same-angle from S{sa_shot_id:02d} {sa_frame}, strength={sa_strength})...") if not img2img_url and mid_frame_url and not args.no_img2img: # FMLF: first (anticipation) derives from mid (anchor/peak action) img2img_url = mid_frame_url img2img_strength = args.img2img_strength print(f" {frame_type}: generating T2I [{ACTIVE_MODEL['label']}] (img2img from mid anchor, strength={img2img_strength})...") elif not img2img_url and hero_frame_url and not args.no_img2img: # Hero-first: first derives from hero img2img_url = hero_frame_url img2img_strength = 0.40 print(f" {frame_type}: generating T2I [{ACTIVE_MODEL['label']}] (img2img from hero, strength=0.40)...") elif not img2img_url and location_ref_url and not args.no_img2img: img2img_url = location_ref_url img2img_strength = args.location_ref_strength print(f" {frame_type}: generating T2I [{ACTIVE_MODEL['label']}] (location ref, strength={img2img_strength})...") elif not img2img_url: print(f" {frame_type}: generating T2I [{ACTIVE_MODEL['label']}]...") elif frame_type == "last": if mid_frame_url and not args.no_img2img: # FMLF: last (aftermath) derives from mid (anchor/peak action) img2img_url = mid_frame_url img2img_strength = args.img2img_strength print(f" {frame_type}: generating T2I [{ACTIVE_MODEL['label']}] (img2img from mid anchor, strength={img2img_strength})...") elif hero_frame_url and not args.no_img2img: # Hero-first: last derives from hero img2img_url = hero_frame_url img2img_strength = 0.40 print(f" {frame_type}: generating T2I [{ACTIVE_MODEL['label']}] (img2img from hero, strength=0.40)...") elif first_frame_url and not args.no_img2img: # Fallback: img2img from first frame (legacy FLF behavior) img2img_url = first_frame_url img2img_strength = args.img2img_strength print(f" {frame_type}: generating T2I [{ACTIVE_MODEL['label']}] (img2img from first, strength={img2img_strength})...") else: print(f" {frame_type}: generating T2I [{ACTIVE_MODEL['label']}]...") elif frame_type == "mid": # Mid frame IS the anchor (peak action) — generated fresh or with location ref only. # First and last are derived FROM mid, not the other way around. if location_ref_url and not args.no_img2img: img2img_url = location_ref_url img2img_strength = args.location_ref_strength print(f" {frame_type}: generating T2I [{ACTIVE_MODEL['label']}] (ANCHOR, location ref, strength={img2img_strength})...") else: print(f" {frame_type}: generating T2I [{ACTIVE_MODEL['label']}]...") else: print(f" {frame_type}: generating T2I [{ACTIVE_MODEL['label']}]...") if getattr(args, 'debug_prompts', False): print(f" PROMPT: {prompt[:200]}...") t0_t2i = time.monotonic() t2i_ok = False _frame_w = args.width if hasattr(args, 'width') else T2I_DEFAULTS["width"] _frame_h = args.height if hasattr(args, 'height') else T2I_DEFAULTS["height"] try: t2i_result = generate_t2i( prompt=prompt, characters=characters, width=_frame_w, height=_frame_h, steps=args.steps, guidance=args.guidance, seed=args.seed, image_url=img2img_url, strength=img2img_strength, negative_prompt=negative_prompt, ) t2i_ok = True if not t2i_result.get("images"): raise RuntimeError("fal.ai returned empty images array") img_url = t2i_result["images"][0]["url"] download_file(img_url, raw_path) result["keyframes"][frame_type] = raw_path print(f" {frame_type}: saved → {Path(raw_path).name}") # Capture frame URLs for chaining if frame_type == "hero": hero_frame_url = img_url elif frame_type == "mid": mid_frame_url = img_url elif frame_type == "first": first_frame_url = img_url # Store generation metadata per frame if "_gen_meta" in t2i_result: if "generation_meta" not in result: result["generation_meta"] = {} result["generation_meta"][frame_type] = t2i_result["_gen_meta"] # Store prompt hash from compiler if compile_result and compile_result.get("prompt_hash"): if "generation_meta" not in result: result["generation_meta"] = {} if frame_type not in result["generation_meta"]: result["generation_meta"][frame_type] = {} result["generation_meta"][frame_type]["prompt_hash"] = compile_result["prompt_hash"] if stats: stats["keyframes"] += 1 except Exception as e: print(f" {frame_type}: FAILED — {e}") result["status"] = "keyframe_failed" if stats: stats["failed"] += 1 finally: elapsed_t2i = int((time.monotonic() - t0_t2i) * 1000) if tracker: _model_key = "z_image_turbo" if "z_image" in ACTIVE_MODEL.get("endpoint", "") else "flux2" _lora_count = sum(1 for c in characters if c in LORA_REGISTRY and get_lora_path_for_model(LORA_REGISTRY[c])) tracker.log( category="generation", provider="fal", model=_model_key, resolution=f"{_frame_w}x{_frame_h}", loras=_lora_count, episode=args.episode, shot_id=shot["id"], duration_ms=elapsed_t2i, success=t2i_ok, detail=f"T2I {frame_type} frame: ep{args.episode:02d}/shot_{shot['id']:02d}", ) if not t2i_ok: return result time.sleep(args.delay) if stage == "keyframes": if result["status"] == "pending": result["status"] = "keyframes_done" return result # ── STEP 2+3: UPSCALE ── for frame_type, raw_path in result["keyframes"].items(): if frame_type == "strip": continue # Don't upscale the source strip — only the split panels suffix = SUFFIX_MAP.get(frame_type, f"_{frame_type}") hq_path = str(output_dir / f"{asset_base}{suffix}_up.png") if args.skip_existing and Path(hq_path).exists(): print(f" {frame_type} upscale: exists (skipped)") result["upscaled"][frame_type] = hq_path continue print(f" {frame_type}: upscaling via Gemini NBP...") t0_up = time.monotonic() up_ok = False try: from upscale_gemini import upscale_image upscale_image(raw_path, output_path=hq_path, client=gemini_client) up_ok = True result["upscaled"][frame_type] = hq_path elapsed_up = time.monotonic() - t0_up print(f" {frame_type}: upscaled in {elapsed_up:.1f}s → {Path(hq_path).name}") if stats: stats["upscaled"] += 1 except Exception as e: print(f" {frame_type}: upscale FAILED — {e}") # Fall back to raw keyframe for video result["upscaled"][frame_type] = raw_path if result["status"] not in ("partial_lora", "blocked_lora"): result["status"] = "upscale_failed" finally: elapsed_up_ms = int((time.monotonic() - t0_up) * 1000) if tracker: tracker.log( category="upscale", provider="gemini", model=UPSCALE_MODEL, images_out=1, episode=args.episode, shot_id=shot["id"], duration_ms=elapsed_up_ms, success=up_ok, detail=f"Gemini NBP upscale {frame_type}: ep{args.episode:02d}/shot_{shot['id']:02d}", ) time.sleep(5) # Gemini rate limit if stage == "upscale": if result["status"] == "pending": result["status"] = "upscale_done" return result # ── STEP 4: VIDEO GENERATION ── if video_calls == 0: result["status"] = "complete" return result motion_prompt = shot.get("motion_prompt", shot.get("action", "Subtle motion")) # WAN 2.2 supports negative prompts — pull from project config video_negative_prompt = (PROJECT_CONFIG or {}).get("negative_prompt", "") if gen_type in ("standard_flf", "wan_i2v"): # Standard FLF: first → last (hero used as mid if available) first_hq = result["upscaled"].get("first") or result["upscaled"].get("hero") last_hq = result["upscaled"].get("last") if not first_hq or not last_hq: print(f" video: SKIPPED — missing keyframes") result["status"] = "video_skipped" return result video_path = str(output_dir / f"{asset_base}_flf.mp4") if args.skip_existing and Path(video_path).exists(): print(f" video: exists (skipped)") result["videos"]["flf"] = video_path result["status"] = "complete" return result print(f" video: FLF first→last...") t0_vid = time.monotonic() vid_ok = False try: first_url = upload_file(first_hq) last_url = upload_file(last_hq) vid_result = generate_video_flf( first_url, last_url, motion_prompt, steps=args.video_steps, guidance=args.video_guidance, seed=args.seed, negative_prompt=video_negative_prompt, ) vid_url = vid_result.get("video", {}).get("url") if vid_url: download_file(vid_url, video_path) result["videos"]["flf"] = video_path vid_ok = True elapsed_vid = time.monotonic() - t0_vid print(f" video: done in {elapsed_vid:.1f}s → {Path(video_path).name}") if stats: stats["videos"] += 1 else: print(f" video: FAILED — no video URL in response") result["status"] = "video_failed" except Exception as e: print(f" video: FAILED — {e}") result["status"] = "video_failed" finally: elapsed_vid_ms = int((time.monotonic() - t0_vid) * 1000) if tracker: tracker.log( category="video", provider="fal", model="wan_2.2_i2v", episode=args.episode, shot_id=shot["id"], duration_ms=elapsed_vid_ms, success=vid_ok, detail=f"WAN FLF video: ep{args.episode:02d}/shot_{shot['id']:02d}", ) time.sleep(args.delay) elif gen_type in ("triptych_split_flf", "wan_flf_reaction"): # Split FLF: first→mid + mid→last first_hq = result["upscaled"].get("first") mid_hq = result["upscaled"].get("mid") last_hq = result["upscaled"].get("last") if not first_hq or not mid_hq or not last_hq: print(f" video: SKIPPED — missing keyframes for split FLF") result["status"] = "video_skipped" return result # Upload all three print(f" video: uploading keyframes...") first_url = upload_file(first_hq) mid_url = upload_file(mid_hq) last_url = upload_file(last_hq) # Segment 1: first → mid seg1_path = str(output_dir / f"{asset_base}_seg1.mp4") if not (args.skip_existing and Path(seg1_path).exists()): print(f" video: seg1 first→mid...") t0_seg1 = time.monotonic() seg1_ok = False try: seg1_result = generate_video_flf( first_url, mid_url, motion_prompt + " First half of action.", steps=args.video_steps, guidance=args.video_guidance, seed=args.seed, negative_prompt=video_negative_prompt, ) seg1_vid_url = seg1_result.get("video", {}).get("url") if seg1_vid_url: download_file(seg1_vid_url, seg1_path) result["videos"]["seg1"] = seg1_path seg1_ok = True elapsed_seg1 = time.monotonic() - t0_seg1 print(f" seg1: done in {elapsed_seg1:.1f}s") if stats: stats["videos"] += 1 else: result["status"] = "video_failed" except Exception as e: print(f" seg1: FAILED — {e}") result["status"] = "video_failed" finally: elapsed_seg1_ms = int((time.monotonic() - t0_seg1) * 1000) if tracker: tracker.log( category="video", provider="fal", model="wan_2.2_i2v", episode=args.episode, shot_id=shot["id"], duration_ms=elapsed_seg1_ms, success=seg1_ok, detail=f"WAN FLF seg1 (first->mid): ep{args.episode:02d}/shot_{shot['id']:02d}", ) time.sleep(args.delay) else: print(f" seg1: exists (skipped)") result["videos"]["seg1"] = seg1_path # Segment 2: mid → last seg2_path = str(output_dir / f"{asset_base}_seg2.mp4") if not (args.skip_existing and Path(seg2_path).exists()): print(f" video: seg2 mid→last...") t0_seg2 = time.monotonic() seg2_ok = False try: seg2_result = generate_video_flf( mid_url, last_url, motion_prompt + " Second half of action.", steps=args.video_steps, guidance=args.video_guidance, seed=args.seed, negative_prompt=video_negative_prompt, ) seg2_vid_url = seg2_result.get("video", {}).get("url") if seg2_vid_url: download_file(seg2_vid_url, seg2_path) result["videos"]["seg2"] = seg2_path seg2_ok = True elapsed_seg2 = time.monotonic() - t0_seg2 print(f" seg2: done in {elapsed_seg2:.1f}s") if stats: stats["videos"] += 1 else: result["status"] = "video_failed" except Exception as e: print(f" seg2: FAILED — {e}") result["status"] = "video_failed" finally: elapsed_seg2_ms = int((time.monotonic() - t0_seg2) * 1000) if tracker: tracker.log( category="video", provider="fal", model="wan_2.2_i2v", episode=args.episode, shot_id=shot["id"], duration_ms=elapsed_seg2_ms, success=seg2_ok, detail=f"WAN FLF seg2 (mid->last): ep{args.episode:02d}/shot_{shot['id']:02d}", ) time.sleep(args.delay) else: print(f" seg2: exists (skipped)") result["videos"]["seg2"] = seg2_path if result["status"] == "pending": result["status"] = "complete" return result # ── Results HTML ────────────────────────────────────────────────────────── def write_results_html(results: list, output_dir: Path, episode: int, storyboard: dict): """Write an HTML page showing all generated keyframes and videos.""" html_parts = [ '', f'Keyframes — Episode {episode}', '', f'

Keyframes + Video — Episode {episode}: {storyboard.get("title", "")}

', f'

Generated {datetime.now().strftime("%Y-%m-%d %H:%M")}

', ] for r in results: status_class = r["status"].replace(" ", "_") html_parts.append(f'
') html_parts.append(f'
') html_parts.append(f' Shot {r["shot_id"]:02d}: {r["name"]}') html_parts.append(f' {r["status"]}') html_parts.append(f'
') html_parts.append(f'
Type: {r["generation_type"]} | Characters: {", ".join(r["characters"]) or "none"}
') html_parts.append(f'
') # Keyframes (raw) for frame_type, path in r.get("keyframes", {}).items(): if path and Path(path).exists(): html_parts.append(f'
{frame_type} (raw)
') # Upscaled for frame_type, path in r.get("upscaled", {}).items(): if path and Path(path).exists() and "_hq" in path: html_parts.append(f'
{frame_type} (HQ)
') # Videos for vid_name, path in r.get("videos", {}).items(): if path and Path(path).exists(): html_parts.append(f'
{vid_name}
') html_parts.append(f'
') if r.get("blocked_reason"): html_parts.append(f'
{r["blocked_reason"]}
') html_parts.append(f'
') html_parts.append('') html_path = output_dir / "results.html" html_path.write_text("\n".join(html_parts)) return str(html_path) # ── Main ────────────────────────────────────────────────────────────────── def main(): parser = argparse.ArgumentParser( description="Full visual production pipeline: T2I → upscale → video" ) parser.add_argument("project_dir", help="Project directory (e.g., leviathan/)") parser.add_argument("--episode", "-e", type=int, required=True, help="Episode number") parser.add_argument("--shots", help="Shot filter (e.g., '1-5' or '3,5,8,9,10')") parser.add_argument("--stage", choices=["keyframes", "upscale", "video", "full"], default="full", help="Pipeline stage to run up to (default: full)") parser.add_argument("--skip-existing", action="store_true", help="Skip existing outputs") parser.add_argument("--dry-run", action="store_true", help="Show plan without executing") parser.add_argument("--seed", type=int, default=T2I_DEFAULTS["seed"], help="Random seed") parser.add_argument("--steps", type=int, default=T2I_DEFAULTS["steps"], help="T2I inference steps") parser.add_argument("--guidance", type=float, default=T2I_DEFAULTS["guidance"], help="T2I guidance scale") parser.add_argument("--video-steps", type=int, default=VIDEO_DEFAULTS["steps"], help="Video inference steps") parser.add_argument("--video-guidance", type=float, default=VIDEO_DEFAULTS["guidance"], help="Video guidance scale") parser.add_argument("--delay", type=float, default=2.0, help="Seconds between fal.ai calls") parser.add_argument("--new-take", action="store_true", help="Create new take (T02, T03...) instead of overwriting T01") parser.add_argument("--model", choices=list(MODEL_CONFIGS.keys()), default="z_image", help="T2I model to use (default: z_image)") parser.add_argument("--img2img-strength", type=float, default=0.40, help="Denoise strength for frame img2img conditioning (0.0=exact copy, 1.0=fully new). Default: 0.40") parser.add_argument("--location-ref-strength", type=float, default=0.35, help="Denoise strength for location reference img2img. Default: 0.35") parser.add_argument("--no-img2img", action="store_true", help="Disable img2img conditioning (generate all frames independently)") parser.add_argument("--no-fmlf", action="store_true", help="Disable FMLF (anchor-first) mode — fall back to first+last only for standard FLF shots") parser.add_argument("--debug-prompts", action="store_true", help="Print compiled prompts during generation") parser.add_argument("--qc", action="store_true", help="Run Gemini QC on keyframes after generation, before upscale. " "Auto-rejects trigger regen with new seed (max --qc-retries). " "Requires GOOGLE_API_KEY.") parser.add_argument("--qc-retries", type=int, default=QC_MAX_RETRIES, help=f"Max regen attempts per shot on QC reject (default: {QC_MAX_RETRIES})") parser.add_argument("--qc-model", default=None, help="Override Gemini model for QC (default: gemini-3-pro-preview)") args = parser.parse_args() # Normalize stage alias if args.stage == "full": args.stage = "video" # Set active model configuration global ACTIVE_MODEL ACTIVE_MODEL = MODEL_CONFIGS[args.model] # Override defaults from model config if user didn't specify model_defaults = ACTIVE_MODEL["defaults"] if args.steps == T2I_DEFAULTS["steps"] and model_defaults.get("steps"): args.steps = model_defaults["steps"] if args.guidance == T2I_DEFAULTS["guidance"] and model_defaults.get("guidance") is not None: args.guidance = model_defaults["guidance"] elif not ACTIVE_MODEL["supports_guidance"]: args.guidance = None # Resolve project directory project_dir = Path(args.project_dir).resolve() if not project_dir.exists(): engine_dir = Path(__file__).resolve().parent.parent.parent project_dir = engine_dir / args.project_dir if not project_dir.exists(): print(f"ERROR: Project directory not found: {args.project_dir}", file=sys.stderr) sys.exit(1) # Load LoRA registry from per-project JSON global LORA_REGISTRY raw_registry = load_registry(project_dir) for char_name, char_data in raw_registry.items(): LORA_REGISTRY[char_name] = get_inference_config(raw_registry, char_name) # Load breakdown + project config for prompt compiler global BREAKDOWN, PROJECT_CONFIG, OVERRIDE_STORE, PREVIOUS_SHOT_CONTEXT, CURRENT_EPISODE BREAKDOWN = _load_breakdown(project_dir) PROJECT_CONFIG = _load_project_config(project_dir) OVERRIDE_STORE = OverrideStore(project_dir) PREVIOUS_SHOT_CONTEXT = None CURRENT_EPISODE = args.episode # Load storyboard ep_str = f"{args.episode:03d}" storyboard_path = project_dir / "storyboards" / f"storyboard_ep_{ep_str}.json" if not storyboard_path.exists(): print(f"ERROR: Storyboard not found: {storyboard_path}", file=sys.stderr) sys.exit(1) try: with open(storyboard_path) as f: storyboard = json.load(f) except json.JSONDecodeError as e: print(f"ERROR: Invalid JSON in {storyboard_path}: {e}", file=sys.stderr) sys.exit(1) shots = storyboard.get("shots", []) if not shots: print("ERROR: No shots in storyboard", file=sys.stderr) sys.exit(1) # Filter shots if args.shots: if "-" in args.shots and "," not in args.shots: start, end = map(int, args.shots.split("-")) shots = [s for s in shots if start <= s["id"] <= end] elif "," in args.shots: ids = set(map(int, args.shots.split(","))) shots = [s for s in shots if s["id"] in ids] else: shot_id = int(args.shots) shots = [s for s in shots if s["id"] == shot_id] # Output directory (flat — all assets for an episode in one dir) base_dir = project_dir / "storyboards" / "assets" / f"ep_{ep_str}" base_dir.mkdir(parents=True, exist_ok=True) # Summary banner img2img_mode = "disabled" if args.no_img2img else f"strength={args.img2img_strength}" print(f"{'=' * 62}") print(f" VISUAL PRODUCTION PIPELINE — Episode {args.episode}") print(f"{'=' * 62}") print(f" Model: {ACTIVE_MODEL['label']} ({ACTIVE_MODEL['endpoint']})") print(f" Storyboard: {storyboard_path.name}") print(f" Shots: {len(shots)}") print(f" Stage: {args.stage}") print(f" Seed: {args.seed} | T2I steps: {args.steps} | Video steps: {args.video_steps}") fmlf_mode = "OFF (legacy first+last)" if args.no_fmlf else "ON (mid→first→last)" print(f" img2img: {img2img_mode} | Location ref: strength={args.location_ref_strength}") print(f" FMLF: {fmlf_mode}") if args.qc: print(f" QC: Gemini keyframe QC (retries: {args.qc_retries})") print(f" Output: {base_dir.relative_to(project_dir)}") print(f"{'=' * 62}") print() # Analyze shots lora_key = ACTIVE_MODEL["lora_key"] type_counts = {} for shot in shots: gt = determine_generation_type(shot) type_counts[gt] = type_counts.get(gt, 0) + 1 chars = detect_characters(shot, storyboard) blocked = [c for c in chars if c in LORA_REGISTRY and not get_lora_path_for_model(LORA_REGISTRY[c])] status = "BLOCKED" if blocked and len(blocked) == len(chars) else ("PARTIAL" if blocked else "OK") lora_info = f" [LoRA: {', '.join(c for c in chars if c in LORA_REGISTRY and get_lora_path_for_model(LORA_REGISTRY[c]))}]" if chars else "" ref_info = " [ref]" if shot.get("reference_image_url") else "" asset = shot.get("asset_name", "") print(f" Shot {shot['id']:2d}: {gt:<25s} {status:<8s} {asset:<30s}{lora_info}{ref_info}") print() no_fmlf = getattr(args, 'no_fmlf', False) for gt, count in type_counts.items(): is_nbp = GENERATION_TYPES[gt].get("nbp_derive", False) if is_nbp: # Hero + NBP: 1 T2I hero + up to 2 NBP derivations per shot # Actual NBP count depends on whether first_frame/last_frame exist kf_count = 1 * count nbp_count = 0 for s in shots: if determine_generation_type(s) == gt: if s.get("anticipation_action") or s.get("first_frame"): nbp_count += 1 if s.get("aftermath_action") or s.get("last_frame"): nbp_count += 1 vid_count = GENERATION_TYPES[gt]["video_calls"] * count print(f" {gt}: {count} shots → {kf_count} T2I hero + {nbp_count} NBP derive + {vid_count} videos") else: sample_shot = next(s for s in shots if determine_generation_type(s) == gt) kf_count = len(_determine_frame_order(sample_shot, no_fmlf=no_fmlf)) * count vid_count = GENERATION_TYPES[gt]["video_calls"] * count fmlf_note = " (FMLF)" if not no_fmlf and GENERATION_TYPES[gt]["keyframes"] == ["first", "last"] else "" print(f" {gt}: {count} shots → {kf_count} keyframes{fmlf_note} + {vid_count} videos") print() if args.dry_run: # ── Prompt Preview ── print("=" * 62) print(" DRY RUN — PROMPT PREVIEW") print("=" * 62) print() no_fmlf = getattr(args, 'no_fmlf', False) for shot in shots: shot_id = shot["id"] gen_type = determine_generation_type(shot) type_info = GENERATION_TYPES[gen_type] use_nbp = type_info.get("nbp_derive", False) frame_order = _determine_frame_order(shot, no_fmlf=no_fmlf) characters = detect_characters(shot, storyboard) print(f" {'─' * 58}") print(f" Shot {shot_id}: {shot.get('name', '')} [{gen_type}]") print(f" Frame order: {' → '.join(frame_order)}") print(f" Characters: {', '.join(characters) if characters else 'none'}") # LoRA info lora_key = ACTIVE_MODEL["lora_key"] for char in characters: if char in LORA_REGISTRY: lora_path = get_lora_path_for_model(LORA_REGISTRY[char]) trigger = LORA_REGISTRY[char].get("trigger", "?") reg = LORA_REGISTRY[char] is_flux2 = ACTIVE_MODEL.get("endpoint", "").startswith("fal-ai/flux-2") if len(characters) == 1: scale = reg.get("flux2_scale_solo", reg.get("scale_solo", "?")) if is_flux2 else reg.get("scale_solo", "?") else: scale = reg.get("flux2_scale_dual", reg.get("scale_dual", "?")) if is_flux2 else reg.get("scale_dual", "?") status = "OK" if lora_path else "MISSING" print(f" LoRA [{char}]: trigger={trigger} scale={scale} [{status}]") if lora_path: print(f" path: {lora_path[:80]}...") # img2img conditioning info if shot.get("continuity_from"): print(f" Continuity from: shot {shot['continuity_from'].get('shot_id')} " f"region={shot['continuity_from'].get('region', 'center')}") if shot.get("same_angle_from"): print(f" Same angle from: shot {shot['same_angle_from']}") if shot.get("reference_image_url"): print(f" Location ref: {shot['reference_image_url'][:80]}") print() # Hero + NBP derivation preview if use_nbp: first_content = (shot.get("anticipation_action") or shot.get("first_frame") or "") last_content = (shot.get("aftermath_action") or shot.get("last_frame") or "") nbp_targets = [] if first_content: nbp_targets.append("first") if last_content: nbp_targets.append("last") nbp_label = f"Gemini NBP ({', '.join(nbp_targets)})" if nbp_targets else "hero only (no first/last content)" print(f" [HERO + NBP DERIVATION]") print(f" Pipeline: Flux T2I (hero) → {nbp_label}") print(f" NBP model: {NBP_DERIVE_MODEL}") # Show hero prompt try: hero_result = build_t2i_prompt(shot, storyboard, frame_type="hero") hero_prompt = hero_result.get("prompt", "") if isinstance(hero_result, dict) else str(hero_result) hero_neg = hero_result.get("negative_prompt", "") if isinstance(hero_result, dict) else "" word_count = len(hero_prompt.split()) print(f" [HERO] ({word_count} words):") words = hero_prompt.split() line = " " for w in words: if len(line) + len(w) + 1 > 90: print(line) line = " " + w else: line += (" " if len(line) > 6 else "") + w if line.strip(): print(line) if hero_neg: print(f" NEG: {hero_neg[:120]}") except Exception as e: print(f" [HERO] COMPILE ERROR: {e}") # Show what NBP will derive if first_content: first_truncated = _truncate_panel_content(first_content, max_words=50) print(f" [NBP → FIRST] target: {first_truncated[:120]}") else: print(f" [NBP → FIRST] SKIP (no first_frame content)") if last_content: last_truncated = _truncate_panel_content(last_content, max_words=50) print(f" [NBP → LAST] target: {last_truncated[:120]}") else: print(f" [NBP → LAST] SKIP (no last_frame content)") print() # Per-frame compiled prompts (non-triptych) if not use_nbp: for frame_type in frame_order: try: result = build_t2i_prompt(shot, storyboard, frame_type=frame_type) prompt = result.get("prompt", "") if isinstance(result, dict) else str(result) neg = result.get("negative_prompt", "") if isinstance(result, dict) else "" word_count = len(prompt.split()) print(f" [{frame_type.upper()}] ({word_count} words):") # Word-wrap words = prompt.split() line = " " for w in words: if len(line) + len(w) + 1 > 90: print(line) line = " " + w else: line += (" " if len(line) > 6 else "") + w if line.strip(): print(line) if neg: print(f" NEG: {neg[:120]}") print() except Exception as e: print(f" [{frame_type.upper()}] COMPILE ERROR: {e}") print() print("=" * 62) print(" DRY RUN — no generation performed.") print("=" * 62) sys.exit(0) # Check API keys if not os.environ.get("FAL_KEY"): print("ERROR: FAL_KEY not set", file=sys.stderr) sys.exit(1) gemini_client = None # Initialize Gemini client if needed for upscale, QC, or NBP frame derivation has_nbp = any(GENERATION_TYPES.get(determine_generation_type(s), {}).get("nbp_derive") for s in shots) needs_gemini = args.stage in ("upscale", "video") or has_nbp if needs_gemini: api_key = os.environ.get("GOOGLE_API_KEY") if not api_key: if args.stage in ("upscale", "video"): print("ERROR: GOOGLE_API_KEY not set (required for upscale stage)", file=sys.stderr) sys.exit(1) elif has_nbp: print("WARNING: GOOGLE_API_KEY not set — shots will generate hero only (no NBP first/last derivation), falling back to img2img chain", file=sys.stderr) else: from google import genai gemini_client = genai.Client(api_key=api_key) # Add tools to path so upscale_gemini is importable tools_dir = str(Path(__file__).resolve().parent) if tools_dir not in sys.path: sys.path.insert(0, tools_dir) # Gemini QC initialization qc_enabled = False if args.qc: if not HAS_GEMINI_QC: print("ERROR: --qc requires gemini_qc.py (import failed)", file=sys.stderr) sys.exit(1) if not os.environ.get("GOOGLE_API_KEY"): print("ERROR: --qc requires GOOGLE_API_KEY environment variable", file=sys.stderr) sys.exit(1) qc_enabled = True print(f" QC: Gemini keyframe QC enabled (model: {args.qc_model or 'default'}, " f"max retries: {args.qc_retries})") # Cost tracking tracker = CostTracker(project_dir) # Process shots stats = {"keyframes": 0, "upscaled": 0, "videos": 0, "blocked": 0, "failed": 0, "qc_pass": 0, "qc_marginal": 0, "qc_reject": 0, "qc_regen": 0} all_results = [] for i, shot in enumerate(shots): print(f" [{i+1}/{len(shots)}] Shot {shot['id']:2d}: {shot.get('name', '')}") # QC retry loop: generate keyframes, run QC, regen on reject current_seed = args.seed qc_attempt = 0 result = None while qc_attempt <= args.qc_retries: # Override seed for regen attempts original_seed = args.seed if qc_attempt > 0: args.seed = current_seed result = process_shot( shot, storyboard, base_dir, args.stage, args, gemini_client, stats, all_results_ref=all_results, tracker=tracker, ) # Restore original seed args.seed = original_seed # If not keyframes stage or generation failed/blocked, skip QC if not qc_enabled or result["status"] in ("blocked_lora", "keyframe_failed", "failed"): break # QC only applies after keyframe generation, before upscale if not result.get("keyframes"): break # Run Gemini QC on each generated keyframe qc_results = {} worst_verdict = "pass" # track worst across all frames for frame_type, frame_path in result["keyframes"].items(): if frame_type == "strip": continue # Skip triptych source strip if not Path(frame_path).exists(): continue # Resolve character refs for comparison chars_in_shot = shot.get("characters_in_shot", []) ref_paths = [] if chars_in_shot and HAS_GEMINI_QC: ref_paths = resolve_character_refs( chars_in_shot[0], project_dir, storyboard ) qc_result = run_keyframe_qc( frame_path, shot, storyboard, ref_paths, frame_type=frame_type, tracker=tracker, episode=args.episode, model_override=args.qc_model, ) qc_results[frame_type] = qc_result verdict = qc_result.get("qc_result", "error") score = qc_result.get("overall_score", 0) regen = qc_result.get("regen_recommendation", "keep") print(f" {frame_type} QC: {verdict.upper()} ({score}/10) [{regen}]") # Track worst verdict if verdict == "fail" or regen == "regen_required": worst_verdict = "fail" elif verdict == "marginal" and worst_verdict != "fail": worst_verdict = "marginal" result["qc"] = qc_results # Decision based on worst verdict across all frames if worst_verdict == "pass": stats["qc_pass"] += 1 break elif worst_verdict == "marginal": stats["qc_marginal"] += 1 # Accept marginals — flag but don't burn regen budget print(f" QC: MARGINAL — accepted (flagged for review)") break elif worst_verdict == "fail" and qc_attempt < args.qc_retries: # Regen with different seed qc_attempt += 1 stats["qc_regen"] += 1 current_seed = args.seed + (qc_attempt * 1000) # Collect reject reasons reasons = [] for ft, qr in qc_results.items(): for issue in qr.get("issues", []): if issue.get("severity") in ("major", "moderate"): reasons.append(f"{ft}:{issue.get('category', '?')}") reason_str = ", ".join(reasons[:3]) if reasons else "QC fail" print(f" QC: REJECT ({reason_str}), regen #{qc_attempt} (seed={current_seed})...") else: # Final rejection — exhausted retries stats["qc_reject"] += 1 result["status"] = "qc_rejected" print(f" QC: REJECTED after {qc_attempt} retries") break all_results.append(result) # Track previous shot context for transition-aware prompts model_key = [k for k, v in MODEL_CONFIGS.items() if v is ACTIVE_MODEL][0] PREVIOUS_SHOT_CONTEXT = PreviousShotContext( layers=build_prompt_layers( shot, BREAKDOWN, args.episode, storyboard, model=model_key, lora_registry=LORA_REGISTRY, project_config=PROJECT_CONFIG, frame_type="last", ), shot_type=shot.get("shot_type", ""), camera_angle=shot.get("camera_angle", "eye"), camera_side=shot.get("spatial", {}).get("camera_side", ""), camera_movement=shot.get("camera_movement", "static"), screen_direction=shot.get("spatial", {}).get("screen_direction", ""), shot_id=shot.get("id", 0), ) # Write manifest (cumulative — merges with existing) manifest_path = base_dir / "manifest.json" existing_manifest = {} if manifest_path.is_file(): try: with open(manifest_path) as f: existing_manifest = json.load(f) except (json.JSONDecodeError, OSError): existing_manifest = {} # Index existing shots by (shot_id, asset_name) for merging existing_shots = {} for s in existing_manifest.get("shots", []): key = (s.get("shot_id"), s.get("asset_name", "")) existing_shots[key] = s # Merge new results — new takes get added, same takes get updated for r in all_results: key = (r.get("shot_id"), r.get("asset_name", "")) existing_shots[key] = r merged_shots = sorted(existing_shots.values(), key=lambda s: (s.get("shot_id", 0), s.get("asset_name", ""))) manifest = { "episode": args.episode, "storyboard": storyboard_path.name, "stage": args.stage, "qc_enabled": qc_enabled, "generated_at": datetime.now().isoformat(), "settings": { "seed": args.seed, "t2i_steps": args.steps, "t2i_guidance": args.guidance, "video_steps": args.video_steps, "video_guidance": args.video_guidance, }, "shots": merged_shots, } with open(manifest_path, "w") as f: json.dump(manifest, f, indent=2) # Write results HTML html_path = write_results_html(all_results, base_dir, args.episode, storyboard) # Summary print() print(f"{'=' * 62}") print(f" RESULTS") print(f"{'=' * 62}") print(f" Keyframes: {stats['keyframes']}") print(f" Upscaled: {stats['upscaled']}") print(f" Videos: {stats['videos']}") print(f" Blocked: {stats['blocked']}") print(f" Failed: {stats['failed']}") if qc_enabled: print(f" QC pass: {stats['qc_pass']}") print(f" QC marginal: {stats['qc_marginal']}") print(f" QC rejected: {stats['qc_reject']}") print(f" QC regens: {stats['qc_regen']}") print(f" Manifest: {manifest_path}") print(f" Results: {html_path}") print(f"{'=' * 62}") sys.exit(0 if stats["failed"] == 0 else 1) if __name__ == "__main__": main()