#!/usr/bin/env python3 """ Engine Shootout — Quick single-angle comparison across candidate generation engines. Tests one angle/expression/environment across multiple engines to compare quality before committing to a full pipeline run. Usage: python3 engine_shootout.py leviathan/ --character JINX python3 engine_shootout.py leviathan/ --character JINX --angle front --expression furious python3 engine_shootout.py leviathan/ --character JINX --engines qwen_angle,nbp python3 engine_shootout.py leviathan/ --character JINX --threepass python3 engine_shootout.py leviathan/ --character JINX --threepass --lighting "warm practical light" python3 engine_shootout.py leviathan/ --character JINX --dry-run Modes: Default: Run selected engines independently (side-by-side comparison) --threepass: Smart pipeline routing (~$0.04-0.17/angle, ~25-56s): Face angles: Qwen MA → NBP (skin priority, skip SeedVR2) Body angles: Qwen MA → NBP → SeedVR2 (proportionality + upscale) Back/profile: Qwen MA → SeedVR2 Cost: ~$0.24 per default run (4 engines), ~$0.04-0.17 per three-pass run (varies by routing) Time: ~2-3 minutes """ import argparse import json import os import sys import time import urllib.request from pathlib import Path from datetime import datetime, timezone # Shared config loader sys.path.insert(0, str(Path(__file__).resolve().parent.parent / "lib")) from config_loader import load_project_config, load_rendering_directives, get_identity_lock from cost_tracker import CostTracker # ── Engine Registry ────────────────────────────────────────────────────── ENGINES = { "qwen_angle": { "name": "Qwen Multi-Angle 2511", "provider": "fal", "endpoint": "fal-ai/qwen-image-edit-2511-multiple-angles", "cost_est": 0.037, # $0.035/MP × 1.05MP (1024x1024) "description": "Angle geometry via multi-angle LoRA", }, "qwen_edit": { "name": "Qwen Edit 2511 (Standard)", "provider": "fal", "endpoint": "fal-ai/qwen-image-edit-2511", "cost_est": 0.031, # $0.03/MP × 1.05MP (1024x1024) "description": "Location/environment swap while preserving identity", }, "nanobanana_flash": { "name": "Nanobanana Flash (Gemini 2.5 Flash Image)", "provider": "google", "endpoint": "gemini-2.5-flash-image", "cost_est": 0.039, # $0.039/image output "description": "Diversity generation (wardrobe/expression/environment)", }, "nbp": { "name": "NBP (Gemini 3 Pro Image Preview)", "provider": "google", "endpoint": "gemini-3-pro-image-preview", "cost_est": 0.134, # $0.134/image output "description": "Quality pass — best identity preservation, expression control, skin detail", }, "z_turbo": { "name": "Z-Image Turbo + LoRA", "provider": "fal", "endpoint": "fal-ai/z-image/turbo/image-to-image/lora", "cost_est": 0.005, # $0.005/MP × 1.05MP (1024x1024) "description": "Fast/cheap img2img with character LoRA", }, "seedvr2": { "name": "SeedVR2 (Quality Upscale)", "provider": "fal", "endpoint": "fal-ai/seedvr/upscale/image", "cost_est": 0.004, # $0.001/MP × ~4MP (upscaled output) "description": "Non-generative quality enhancement — cannot alter pose/angle", }, } DEFAULT_ENGINES = ["qwen_angle", "qwen_edit", "nanobanana_flash", "nbp"] # ── Engine → pricing model mapping ─────────────────────────────────────── ENGINE_COST_MAP = { "qwen_angle": {"provider": "fal", "model": "qwen_angle", "resolution": "1024x1024"}, "qwen_edit": {"provider": "fal", "model": "qwen_edit", "resolution": "1024x1024"}, "nanobanana_flash": {"provider": "gemini", "model": "gemini-2.5-flash-image"}, "nbp": {"provider": "gemini", "model": "gemini-3-pro-image-preview"}, "z_turbo": {"provider": "fal", "model": "z_image_turbo", "resolution": "1024x1024"}, "seedvr2": {"provider": "fal", "model": "seedvr2", "resolution": "2048x2048"}, # upscaled output ~4MP } def _log_engine_cost(tracker, engine_key, success, elapsed_ms, character=None): """Log an engine API call to the cost tracker.""" if not tracker: return cost_info = ENGINE_COST_MAP.get(engine_key, {}) provider = cost_info.get("provider", "fal") model = cost_info.get("model", engine_key) kwargs = {} # For Gemini image gen, log 1 image output if provider == "gemini" and success: kwargs["images_out"] = 1 # For per-megapixel models, pass resolution if cost_info.get("resolution"): kwargs["resolution"] = cost_info["resolution"] tracker.log( category="reference", provider=provider, model=model, success=success, duration_ms=int(elapsed_ms * 1000) if elapsed_ms else None, detail=f"candidate_gen:{engine_key}" + (f" char={character}" if character else ""), **kwargs, ) def _check_budget(tracker, budget_cap, engine_key): """Check budget before an API call. Returns True if OK, False if over budget.""" if not tracker or not budget_cap: return True ok, remaining, spent = tracker.check_budget(budget_cap) if not ok: est = ENGINES.get(engine_key, {}).get("cost_est", 0) print(f"\n BUDGET CAP HIT: ${spent:.2f} spent of ${budget_cap:.2f} cap " f"(${remaining:.2f} remaining, next call ~${est:.3f})") print(f" Stopping to prevent overspend. Increase budget_cap_usd in project_config.json to continue.") return False return True # ── Angle Map (for Qwen Multi-Angle) ──────────────────────────────────── QWEN_ANGLE_MAP = { "front": {"h": 0, "v": 0, "z": 5}, "three_quarter_right": {"h": 45, "v": 0, "z": 5}, "profile_right": {"h": 90, "v": 0, "z": 5}, "back_right": {"h": 135, "v": 0, "z": 5}, "back": {"h": 180, "v": 0, "z": 5}, "back_left": {"h": 225, "v": 0, "z": 5}, "profile_left": {"h": 270, "v": 0, "z": 5}, "three_quarter_left": {"h": 315, "v": 0, "z": 5}, "low_angle": {"h": 0, "v": -30, "z": 5}, "high_angle": {"h": 0, "v": 30, "z": 0}, "full_body": {"h": 0, "v": 10, "z": 0}, "full_body_three_quarter": {"h": 45, "v": 10, "z": 0}, "closeup_front": {"h": 0, "v": 0, "z": 10}, "closeup_three_quarter":{"h": 45, "v": 0, "z": 10}, } ANGLE_DESCRIPTIONS = { "front": "Front view, eye level, medium shot", "three_quarter_right": "3/4 right view, eye level, medium shot", "profile_right": "Profile right, eye level", "back_right": "Back-right quarter view", "back": "Back view, eye level", "back_left": "Back-left quarter view", "profile_left": "Profile left, eye level", "three_quarter_left": "3/4 left view, eye level, medium shot", "low_angle": "Low angle looking up, full body visible from feet to head", "high_angle": "High angle looking down, full body visible from head to feet", "full_body": "Front view, eye level, full body from head to feet", "full_body_three_quarter": "3/4 right view, eye level, full body from head to feet", "closeup_front": "Close-up, front, headshot", "closeup_three_quarter": "Close-up, 3/4 right, headshot", } # Angles that show significant body — need proportionality prompting in NBP BODY_ANGLES = {"low_angle", "high_angle", "full_body", "full_body_three_quarter"} # Face-heavy angles — skip SeedVR2 to preserve NBP skin texture FACE_ANGLES = {"front", "closeup_front", "closeup_three_quarter", "three_quarter_right", "three_quarter_left"} # ── Engine Implementations ─────────────────────────────────────────────── def _resolve_hero(project_path: Path, character_key: str) -> Path: """Find the canonical hero image via ProjectPaths resolver. Returns the resolved hero Path, or None if not found. """ try: # engine_shootout receives project_path as the full project directory. # Use ProjectPaths.from_root to construct from absolute path. sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) from recoil.core.paths import ProjectPaths, RefNotFoundError pp = ProjectPaths.from_root(project_path) resolved = pp.resolve_ref("char", character_key.lower(), "identity", "hero") return resolved.path except (RefNotFoundError, FileNotFoundError, Exception): return None def _get_character_identity(project_path: Path, character_key: str) -> str: """Extract character identity description from breakdown.json.""" breakdown_path = project_path / "visual" / "breakdown.json" if not breakdown_path.exists(): return character_key bd = json.loads(breakdown_path.read_text()) char_data = bd.get("characters", {}).get(character_key.upper(), {}) visual_desc = char_data.get("visual_description", "") if visual_desc: return visual_desc return char_data.get("display_name", character_key) def _get_rendering_directives(project_path: Path, character_key: str) -> dict: """Load character-specific rendering directives from breakdown.json. Thin wrapper around config_loader.load_rendering_directives(). Returns dict with texture_prompt, texture_negative, mandatory_traits. """ return load_rendering_directives(project_path, character_key) def _get_sample_location(project_path: Path, character_key: str) -> str: """Get a sample location description for the character.""" breakdown_path = project_path / "visual" / "breakdown.json" if not breakdown_path.exists(): return "Dark industrial corridor with worn metal walls and flickering overhead lights" bd = json.loads(breakdown_path.read_text()) char_data = bd.get("characters", {}).get(character_key.upper(), {}) char_episodes = set(char_data.get("episodes", [])) locations = bd.get("locations", {}) for loc_key, loc_data in locations.items(): loc_episodes = set(loc_data.get("episodes", [])) if char_episodes & loc_episodes: descs = loc_data.get("descriptions", []) if descs: return descs[0] if isinstance(descs[0], str) else str(descs[0]) return "Dark industrial corridor with worn metal walls and flickering overhead lights" def _get_lora_url(project_path: Path, character_key: str) -> str: """Get Z-Image LoRA URL from registry.""" registry_path = project_path / "visual" / "lora_registry.json" if not registry_path.exists(): return None reg = json.loads(registry_path.read_text()) char_data = reg.get("characters", {}).get(character_key.lower(), {}) z_image = char_data.get("z_image_t2i", {}) return z_image.get("path") def _get_lora_trigger(project_path: Path, character_key: str) -> str: """Get LoRA trigger word from registry.""" registry_path = project_path / "visual" / "lora_registry.json" if not registry_path.exists(): return character_key.upper() reg = json.loads(registry_path.read_text()) char_data = reg.get("characters", {}).get(character_key.lower(), {}) return char_data.get("trigger", character_key.upper()) def run_qwen_angle(hero_path: Path, output_path: Path, angle: str, **kwargs) -> dict: """Qwen Multi-Angle: generate a specific camera angle.""" import fal_client image_url = fal_client.upload_file(str(hero_path)) angle_params = QWEN_ANGLE_MAP.get(angle, {"h": 315, "v": 0, "z": 5}) # Optional prompts — default to empty (pure geometric rotation). # Text prompts cause Qwen to regenerate content (distorted heads, wrong # proportions) instead of purely rotating geometry. Callers can opt in # via kwargs if character-specific Qwen prompts prove useful later. additional_prompt = kwargs.get("additional_prompt", "") negative_prompt = kwargs.get("negative_prompt", "") arguments = { "image_urls": [image_url], "horizontal_angle": angle_params["h"], "vertical_angle": angle_params["v"], "zoom": angle_params["z"], "lora_scale": 0.9, "image_size": "square_hd", "num_inference_steps": 40, "guidance_scale": 4.5, "num_images": 1, "enable_safety_checker": False, } if additional_prompt: arguments["additional_prompt"] = additional_prompt if negative_prompt: arguments["negative_prompt"] = negative_prompt t0 = time.time() result = fal_client.subscribe( "fal-ai/qwen-image-edit-2511-multiple-angles", arguments=arguments, with_logs=False, ) elapsed = time.time() - t0 if result and "images" in result and result["images"]: img_url = result["images"][0]["url"] output_path.parent.mkdir(parents=True, exist_ok=True) urllib.request.urlretrieve(img_url, str(output_path)) return {"success": True, "elapsed": elapsed, "output": output_path} return {"success": False, "elapsed": elapsed, "error": "No image in response"} def run_qwen_edit(hero_path: Path, output_path: Path, angle: str, expression: str, environment: str, identity: str, **kwargs) -> dict: """Qwen Edit 2511: environment/expression swap while preserving identity.""" import fal_client image_url = fal_client.upload_file(str(hero_path)) angle_desc = ANGLE_DESCRIPTIONS.get(angle, "3/4 left view, eye level") # Load negative prompt from project config if available project_config = kwargs.get("project_config") or {} neg_prompt = project_config.get("negative_prompt", "blurry face, distorted features, deformed eyes, " "asymmetric face, smooth skin, plastic look, low quality, artifacts") prompt = ( f"Change only the background environment to: {environment}. " f"Preserve the person's exact facial features, skin tone, skin texture, " f"expression, and clothing unchanged. " f"Keep the exact same camera angle, framing, and pose. " f"Keep all foreground elements identical. Only replace the background." ) t0 = time.time() result = fal_client.subscribe( "fal-ai/qwen-image-edit-2511", arguments={ "image_urls": [image_url], "prompt": prompt, "negative_prompt": neg_prompt, "image_size": "square_hd", "num_inference_steps": 45, "guidance_scale": 3.5, "acceleration": "none", "output_format": "png", "num_images": 1, "enable_safety_checker": False, }, with_logs=False, ) elapsed = time.time() - t0 if result and "images" in result and result["images"]: img_url = result["images"][0]["url"] output_path.parent.mkdir(parents=True, exist_ok=True) urllib.request.urlretrieve(img_url, str(output_path)) return {"success": True, "elapsed": elapsed, "output": output_path} return {"success": False, "elapsed": elapsed, "error": "No image in response"} def run_gemini(hero_path: Path, output_path: Path, angle: str, expression: str, environment: str, identity: str, model: str = "gemini-2.5-flash-image", lighting: str = None, **kwargs) -> dict: """Gemini (Flash or Pro): generate with identity reference.""" from google import genai from google.genai import types client = genai.Client(api_key=os.environ.get("GOOGLE_API_KEY")) hero_bytes = hero_path.read_bytes() mime = "image/jpeg" if hero_path.suffix.lower() in (".jpg", ".jpeg") else "image/png" angle_desc = ANGLE_DESCRIPTIONS.get(angle, "3/4 left view, eye level") lighting_desc = lighting or "cinematic lighting with modeling" # Detect if this is a three-pass call (input is a pipeline intermediate, not the hero) is_threepass = kwargs.get("threepass", False) # Project config and character-specific rendering directives project_path = kwargs.get("project_path") character_key = kwargs.get("character_key", "") project_config = kwargs.get("project_config") or {} rendering = {} if project_path and character_key: rendering = _get_rendering_directives(project_path, character_key) if project_path and not project_config: project_config = load_project_config(project_path) # Character-specific trait injection (e.g., Kian's electric blue eyes) char_traits = kwargs.get("character_traits", None) trait_block = "" if char_traits: trait_block = ( f"\nMANDATORY CHARACTER TRAITS — these MUST be visible in the output image:\n" f"{char_traits}\n" f"These traits are non-negotiable. If the input image lacks them, ADD them. " f"If they are subtle in the input, make them MORE prominent.\n" ) # Body proportionality block for wide/full-body angles is_body_angle = angle in BODY_ANGLES proportion_block = "" if is_body_angle: proportion_block = ( "\nCRITICAL — ANATOMICAL PROPORTIONS: The head must be correctly proportioned " "relative to the body. An adult human head is approximately 1/7.5 to 1/8 of total " "body height. DO NOT enlarge the head. The head should appear naturally small relative " "to the torso and limbs. Show realistic adult body proportions: shoulders wider than " "the head, torso length approximately 3 head-heights, legs approximately 4 head-heights. " "If the input image has a disproportionately large head, correct it to anatomically " "accurate proportions.\n" ) if is_threepass: # Select lens based on framing — camera body + lenses from project config camera_body = project_config.get("camera_body", "ARRI Alexa LF") candidate_lenses = project_config.get("candidate_lenses", {}) if is_body_angle: lens_spec = candidate_lenses.get("body", "35mm f/2.8 prime") lens_block = ( f"Shot on {camera_body}. {lens_spec} lens. " "Full body in frame with sharp focus across the figure. Natural depth of field." ) else: lens_spec = candidate_lenses.get("face", "85mm f/1.8 prime") lens_block = ( f"Shot on {camera_body}. {lens_spec} lens. " "Sharp focus on eyes — tack-sharp iris fibers, subtle catchlights. " "Shallow depth of field with soft background separation." ) # Wardrobe/armor preservation — applies to all angles wardrobe_block = ( "\nWARDROBE, ARMOR, AND HEADGEAR PRESERVATION:\n" "Maintain the EXACT wardrobe, armor, helmet, headgear, and accessories visible in the input image. " "If the subject wears a helmet or head covering, it MUST remain in the output. " "Do NOT replace a helmet with hair. Do NOT remove headgear. Do NOT expose a bare head if the input shows a covered head.\n" "Preserve material composition: metal shows tool marks, forge patina, and reflective highlights; " "leather shows creasing patterns, grain texture, and wear at stress points; " "fabric shows weave pattern, drape, and natural wrinkles. " "Do NOT simplify, replace, or genericize any clothing or armor elements.\n" ) # Breakdown-driven surface texture (character-specific or human default) texture_prompt = rendering.get("texture_prompt", "") texture_block = "" if texture_prompt: texture_block = ( f"SURFACE TEXTURE — THIS IS CRITICAL:\n" f"{texture_prompt}\n" "DO NOT smooth, beautify, airbrush, or apply any global surface softening.\n" ) # Select identity lock based on character type (human vs non_human) identity_type = rendering.get("identity_type", "human") identity_lock = get_identity_lock(identity_type) prompt_text = ( "This is the SAME PERSON — face identity locked. DO NOT generate a new face.\n\n" f"{trait_block}" f"Change the background/environment to: {environment}.\n" f"The subject's expression: {expression}.\n" f"Camera angle: {angle_desc}.\n" f"Lighting: {lighting_desc}.\n\n" f"{identity_lock}" "Expression muscles may move freely — brows, eyelids, nostrils, lips, jaw — " "these are temporary muscular movements, not identity changes.\n" f"{proportion_block}" f"{wardrobe_block}\n" "Keep the EXACT same camera angle, head position, and body orientation. " "DO NOT rotate the head. DO NOT change the framing or crop.\n\n" f"{lens_block}\n\n" f"{texture_block}\n" f"{project_config.get('quality_guard', '')}.\n" "Single photorealistic photograph. One person only. No text. No split panels." ) else: identity_lock = get_identity_lock(rendering.get("identity_type", "human")) prompt_text = ( "This is the SAME PERSON — face identity locked. DO NOT generate a new face.\n\n" f"{identity_lock}" "Expression muscles may move freely — brows, eyelids, nostrils, lips, jaw — " "these are temporary muscular movements, not identity changes.\n\n" f"Camera angle: {angle_desc}.\n" f"Expression: {expression}.\n" f"Environment: {environment}.\n" f"Lighting: {lighting_desc}.\n\n" f"Shot at {project_config['candidate_lenses']['face']}. Sharp focus on eyes. " "Tack-sharp iris detail. Subtle catchlights in the pupils.\n\n" "DO NOT smooth, beautify, or stylize. Preserve pore texture, freckles, fine lines.\n" f"{project_config['quality_guard']}.\n" "Single photorealistic photograph. One person only. No text. No split panels." ) # Build parts — single input image (no dual-reference) parts = [ types.Part(inline_data=types.Blob(mime_type=mime, data=hero_bytes)), types.Part(text=prompt_text), ] contents = [types.Content(parts=parts)] t0 = time.time() response = client.models.generate_content( model=model, contents=contents, config=types.GenerateContentConfig( response_modalities=["IMAGE", "TEXT"], temperature=1.0, ), ) elapsed = time.time() - t0 # Extract image from response if response.candidates: for part in response.candidates[0].content.parts: if hasattr(part, "inline_data") and part.inline_data and part.inline_data.mime_type.startswith("image/"): output_path.parent.mkdir(parents=True, exist_ok=True) output_path.write_bytes(part.inline_data.data) return {"success": True, "elapsed": elapsed, "output": output_path} # Check for text-only response (safety block etc) text_parts = [] if response.candidates: for part in response.candidates[0].content.parts: if hasattr(part, "text") and part.text: text_parts.append(part.text) error_msg = "; ".join(text_parts) if text_parts else "No image in response" return {"success": False, "elapsed": elapsed, "error": error_msg[:200]} def run_z_turbo(hero_path: Path, output_path: Path, angle: str, expression: str, environment: str, identity: str, lora_url: str = None, trigger: str = "", **kwargs) -> dict: """Z-Image Turbo: img2img with character LoRA for fast/cheap generation.""" import fal_client image_url = fal_client.upload_file(str(hero_path)) angle_desc = ANGLE_DESCRIPTIONS.get(angle, "3/4 left view, eye level") prompt = ( f"{trigger} " f"{angle_desc} of a woman. " f"{expression} expression. " f"Environment: {environment}. " f"Dramatic side lighting. " f"Single photorealistic photograph. Natural skin texture, visible pores, 8K, hyperdetailed." ) arguments = { "image_url": image_url, "prompt": prompt, "strength": 0.55, "image_size": "square_hd", "num_inference_steps": 8, "num_images": 1, "enable_safety_checker": False, } endpoint = "fal-ai/z-image/turbo/image-to-image" if lora_url: endpoint = "fal-ai/z-image/turbo/image-to-image/lora" arguments["loras"] = [{"path": lora_url, "scale": 1.0}] t0 = time.time() result = fal_client.subscribe( endpoint, arguments=arguments, with_logs=False, ) elapsed = time.time() - t0 if result and "images" in result and result["images"]: img_url = result["images"][0]["url"] output_path.parent.mkdir(parents=True, exist_ok=True) urllib.request.urlretrieve(img_url, str(output_path)) return {"success": True, "elapsed": elapsed, "output": output_path} return {"success": False, "elapsed": elapsed, "error": "No image in response"} def run_seedvr2(input_path: Path, output_path: Path, **kwargs) -> dict: """SeedVR2 non-generative quality upscale via fal.ai.""" import fal_client # Upload with retry (safe ASCII filename) data = input_path.read_bytes() mime = "image/jpeg" if input_path.suffix.lower() in (".jpg", ".jpeg") else "image/png" image_url = fal_client.upload(data, mime, file_name="input.png") t0 = time.time() result = fal_client.run("fal-ai/seedvr/upscale/image", arguments={"image_url": image_url}) elapsed = time.time() - t0 if result and "image" in result: img_url = result["image"]["url"] output_path.parent.mkdir(parents=True, exist_ok=True) urllib.request.urlretrieve(img_url, str(output_path)) return {"success": True, "elapsed": elapsed, "output": output_path} return {"success": False, "elapsed": elapsed, "error": "No image in SeedVR2 response"} # ── Dispatch ───────────────────────────────────────────────────────────── ENGINE_RUNNERS = { "qwen_angle": run_qwen_angle, "qwen_edit": run_qwen_edit, "nanobanana_flash": lambda **kw: run_gemini(model="gemini-2.5-flash-image", **kw), "nbp": lambda **kw: run_gemini(model="gemini-3-pro-image-preview", **kw), "z_turbo": run_z_turbo, } # ── HTML Report ────────────────────────────────────────────────────────── def generate_html_report(results: list, hero_path: Path, test_params: dict, output_path: Path): """Generate a comparison HTML page.""" angle = test_params["angle"] expression = test_params["expression"] character = test_params["character"] cards = "" for r in results: engine_info = ENGINES[r["engine_key"]] status_color = "#4ecca3" if r["success"] else "#e94560" status_text = f'{r["elapsed"]:.1f}s' if r["success"] else f'FAILED: {r.get("error", "unknown")[:60]}' if r["success"] and r.get("output_path"): # Use relative path from output dir rel_path = os.path.relpath(r["output_path"], output_path.parent) img_tag = f'{engine_info[' else: img_tag = '
FAILED
{}
'.format(r.get("error", "")[:80]) cards += f"""
{img_tag}
{engine_info["name"]}
{status_text} | ~${engine_info["cost_est"]:.3f}
{engine_info["description"]}
""" hero_rel = os.path.relpath(hero_path, output_path.parent) html = f""" Engine Shootout — {character} — {angle} / {expression}

Engine Shootout: {character}

Generated {datetime.now(timezone.utc).strftime('%Y-%m-%d %H:%M UTC')} | Single-angle comparison across {len(results)} engines

Hero ref
Hero Reference: {hero_path.name}
Used as identity source for all engines
Angle: {angle} ({ANGLE_DESCRIPTIONS.get(angle, angle)})  |  Expression: {expression}  |  Environment: {test_params.get('environment', 'N/A')[:80]}...
{cards}
""" output_path.write_text(html) return output_path # ── Main ───────────────────────────────────────────────────────────────── def main(): parser = argparse.ArgumentParser(description="Engine Shootout — quick single-angle comparison") parser.add_argument("project", help="Project path (e.g., leviathan/)") parser.add_argument("--character", required=True, help="Character key (e.g., JINX)") parser.add_argument("--angle", default="three_quarter_left", help="Test angle (default: three_quarter_left)") parser.add_argument("--expression", default="exhausted", help="Test expression (default: exhausted)") parser.add_argument("--environment", default=None, help="Override environment description") parser.add_argument("--engines", default=None, help="Comma-separated engine keys (default: all)") parser.add_argument("--lighting", default=None, help="Lighting description (default: 'cinematic lighting with modeling')") parser.add_argument("--threepass", action="store_true", help="Sequential pipeline: Qwen MA → NBP (final) or Qwen MA → SeedVR2 (neutral-only)") parser.add_argument("--skip-pass3", action="store_true", help="Skip NBP, go Qwen MA → SeedVR2 (for neutral-only angles)") parser.add_argument("--dry-run", action="store_true", help="Show what would run without generating") parser.add_argument("--hero", default=None, help="Explicit hero image path (overrides auto-detect)") parser.add_argument("--character-traits", default=None, help="Character-specific visual traits to inject into identity prompt") args = parser.parse_args() # Resolve project path script_dir = Path(__file__).resolve().parent if script_dir.name == "tools" and script_dir.parent.name == "recoil": root = script_dir.parent.parent else: root = Path.cwd() project_name = args.project.strip("/").strip("\\") project_path = root / project_name if not project_path.is_dir(): project_path = Path(args.project) if not project_path.is_dir(): print(f"ERROR: Project not found: {project_path}", file=sys.stderr) sys.exit(1) char_upper = args.character.upper() # Resolve hero if args.hero: hero_path = Path(args.hero) else: hero_path = _resolve_hero(project_path, char_upper) if not hero_path or not hero_path.exists(): print(f"ERROR: No hero image found for {char_upper}.", file=sys.stderr) print(f" Use --hero /path/to/image.jpeg to specify explicitly.", file=sys.stderr) sys.exit(1) # Load shared config project_config = load_project_config(project_path) # Cost tracking + budget cap tracker = CostTracker(project_path) budget_cap = project_config.get("budget_cap_usd") # Get character info identity = _get_character_identity(project_path, char_upper) environment = args.environment or _get_sample_location(project_path, char_upper) lora_url = _get_lora_url(project_path, char_upper) trigger = _get_lora_trigger(project_path, char_upper) # Auto-load mandatory_traits from rendering_directives if not overridden via CLI if not args.character_traits: directives = load_rendering_directives(project_path, char_upper) auto_traits = directives.get("mandatory_traits") if auto_traits: args.character_traits = auto_traits # Select engines engine_keys = args.engines.split(",") if args.engines else DEFAULT_ENGINES for key in engine_keys: if key not in ENGINES: print(f"ERROR: Unknown engine '{key}'. Available: {', '.join(ENGINES.keys())}", file=sys.stderr) sys.exit(1) # Output directory timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") out_dir = project_path / "visual" / "lora_candidates" / char_upper / "shootout" / f"{args.angle}_{args.expression}_{timestamp}" print(f"\n{'='*60}") print(f"ENGINE SHOOTOUT — {char_upper}") print(f"{'='*60}") print(f" Hero: {hero_path.name}") print(f" Angle: {args.angle} ({ANGLE_DESCRIPTIONS.get(args.angle, '?')})") print(f" Expression: {args.expression}") print(f" Environment: {environment[:80]}...") print(f" Engines: {len(engine_keys)} ({', '.join(engine_keys)})") if lora_url: print(f" Z-Image LoRA: {trigger} (found)") else: print(f" Z-Image LoRA: NOT FOUND (z_turbo will run without LoRA)") print(f" Output: {out_dir}") print(f" Est. cost: ~${sum(ENGINES[k]['cost_est'] for k in engine_keys):.3f}") print(f"{'='*60}\n") if args.dry_run: print("DRY RUN — nothing generated.") if args.threepass: if args.skip_pass3: print(" MODE: Two-pass (Qwen MA → SeedVR2)") total = ENGINES["qwen_angle"]["cost_est"] + ENGINES["seedvr2"]["cost_est"] else: print(" MODE: Three-pass sequential (Qwen MA → NBP → SeedVR2)") total = ENGINES["qwen_angle"]["cost_est"] + ENGINES["nbp"]["cost_est"] + ENGINES["seedvr2"]["cost_est"] print(f" Est. cost per angle: ~${total:.3f}") else: for key in engine_keys: eng = ENGINES[key] print(f" [{key}] {eng['name']} — ~${eng['cost_est']:.3f} — {eng['description']}") sys.exit(0) # ── Three-Pass Sequential Mode ────────────────────────────────────── if args.threepass: skip_nbp = args.skip_pass3 # Determine pipeline mode skip_seedvr = args.angle in FACE_ANGLES and not skip_nbp if skip_nbp: print("MODE: Two-pass (Qwen MA → SeedVR2, skipping NBP)\n") elif skip_seedvr: print("MODE: Two-pass (Qwen MA → NBP, skipping SeedVR2 — skin texture priority)\n") else: print("MODE: Three-pass sequential (Qwen MA → NBP → SeedVR2)\n") if not os.environ.get("FAL_KEY"): print("ERROR: FAL_KEY environment variable not set.", file=sys.stderr) sys.exit(1) if not skip_nbp and not os.environ.get("GOOGLE_API_KEY"): print("ERROR: GOOGLE_API_KEY environment variable not set.", file=sys.stderr) sys.exit(1) out_dir.mkdir(parents=True, exist_ok=True) results = [] if skip_nbp: total_passes = 2 # Qwen → SeedVR2 elif skip_seedvr: total_passes = 2 # Qwen → NBP (face angles — preserve skin texture) else: total_passes = 3 # Qwen → NBP → SeedVR2 (body/back angles) # Pass 1: Qwen Multi-Angle — angle geometry from hero if not _check_budget(tracker, budget_cap, "qwen_angle"): sys.exit(2) pass1_file = out_dir / f"pass1_qwen_angle_{args.angle}.png" print(f" [Pass 1/{total_passes}] Qwen Multi-Angle → {args.angle}...", end=" ", flush=True) try: r1 = run_qwen_angle(hero_path=hero_path, output_path=pass1_file, angle=args.angle) if r1["success"]: print(f"OK ({r1['elapsed']:.1f}s)") else: print(f"FAILED: {r1.get('error', 'unknown')[:60]}") print("ERROR: Pass 1 failed. Cannot continue pipeline.", file=sys.stderr) sys.exit(1) except Exception as e: print(f"ERROR: {str(e)[:80]}") sys.exit(1) _log_engine_cost(tracker, "qwen_angle", r1["success"], r1["elapsed"], char_upper) results.append({"engine_key": "qwen_angle", "success": r1["success"], "elapsed": r1["elapsed"], "output_path": str(pass1_file), "pass": 1}) time.sleep(2) # Pass 2 (full pipeline): NBP — background swap + expression on Pass 1 output seedvr_input = pass1_file # default: SeedVR2 runs on Pass 1 output if not skip_nbp: if not _check_budget(tracker, budget_cap, "nbp"): sys.exit(2) pass2_file = out_dir / f"pass2_nbp_{args.angle}_{args.expression}.png" print(f" [Pass 2/{total_passes}] NBP → background + expression...", end=" ", flush=True) try: r2 = run_gemini(hero_path=pass1_file, output_path=pass2_file, angle=args.angle, expression=args.expression, environment=environment, identity=identity, model="gemini-3-pro-image-preview", lighting=args.lighting, threepass=True, character_traits=args.character_traits, project_path=project_path, character_key=char_upper, project_config=project_config) if r2["success"]: print(f"OK ({r2['elapsed']:.1f}s)") seedvr_input = pass2_file else: print(f"FAILED: {r2.get('error', 'unknown')[:60]}") print("ERROR: Pass 2 (NBP) failed. Cannot continue pipeline.", file=sys.stderr) sys.exit(1) except Exception as e: print(f"ERROR: {str(e)[:80]}") sys.exit(1) _log_engine_cost(tracker, "nbp", r2["success"], r2["elapsed"], char_upper) results.append({"engine_key": "nbp", "success": r2["success"], "elapsed": r2["elapsed"], "output_path": str(pass2_file), "pass": 2}) time.sleep(2) # Final pass: SeedVR2 — non-generative quality upscale # Skip on face angles to preserve NBP skin texture if not skip_seedvr: if not _check_budget(tracker, budget_cap, "seedvr2"): sys.exit(2) seedvr_pass_num = total_passes seedvr_file = out_dir / f"pass{seedvr_pass_num}_seedvr2_{args.angle}_{args.expression}.png" print(f" [Pass {seedvr_pass_num}/{total_passes}] SeedVR2 → quality upscale...", end=" ", flush=True) try: rs = run_seedvr2(input_path=seedvr_input, output_path=seedvr_file) if rs["success"]: print(f"OK ({rs['elapsed']:.1f}s)") else: print(f"FAILED: {rs.get('error', 'unknown')[:60]}") except Exception as e: rs = {"success": False, "elapsed": 0, "error": str(e)[:200]} print(f"ERROR: {str(e)[:80]}") _log_engine_cost(tracker, "seedvr2", rs.get("success", False), rs.get("elapsed", 0), char_upper) results.append({"engine_key": "seedvr2", "success": rs.get("success", False), "elapsed": rs.get("elapsed", 0), "output_path": str(seedvr_file), "pass": seedvr_pass_num}) # Generate comparison HTML (shows all passes + hero) html_path = out_dir / "comparison.html" generate_html_report( results, hero_path, {"angle": args.angle, "expression": args.expression, "character": char_upper, "environment": environment}, html_path, ) total_time = sum(r["elapsed"] for r in results) total_cost = sum(ENGINES[r["engine_key"]]["cost_est"] for r in results if r["success"]) print(f"\n{'='*60}") if skip_nbp: mode_label = "TWO-PASS (Qwen→SeedVR2)" elif skip_seedvr: mode_label = "TWO-PASS (Qwen→NBP, skin priority)" else: mode_label = "THREE-PASS" print(f"{mode_label} COMPLETE: {sum(1 for r in results if r['success'])}/{total_passes} succeeded") print(f" Total time: {total_time:.1f}s") print(f" Est. cost: ~${total_cost:.3f}") print(f" Comparison: {html_path}") print(f"{'='*60}\n") # Compute hero_relative (path relative to project root for serve.py) try: hero_relative = str(hero_path.resolve().relative_to(project_path.resolve())) except ValueError: hero_relative = None results_json = out_dir / "results.json" results_json.write_text(json.dumps({ "timestamp": datetime.now(timezone.utc).isoformat(), "mode": "threepass_2pass" if skip_nbp else "threepass", "character": char_upper, "hero": str(hero_path), "hero_relative": hero_relative, "angle": args.angle, "expression": args.expression, "environment": environment, "lighting": args.lighting or "cinematic lighting with modeling", "results": results, }, indent=2)) print(f" Reviewer: http://127.0.0.1:8420/shootout_reviewer.html?project={project_name}&character={char_upper}") return html_path # Check API keys need_fal = any(ENGINES[k]["provider"] == "fal" for k in engine_keys) need_google = any(ENGINES[k]["provider"] == "google" for k in engine_keys) if need_fal and not os.environ.get("FAL_KEY"): print("ERROR: FAL_KEY environment variable not set.", file=sys.stderr) sys.exit(1) if need_google and not os.environ.get("GOOGLE_API_KEY"): print("ERROR: GOOGLE_API_KEY environment variable not set.", file=sys.stderr) sys.exit(1) out_dir.mkdir(parents=True, exist_ok=True) # Run each engine results = [] for i, key in enumerate(engine_keys): if not _check_budget(tracker, budget_cap, key): print(f" Skipping remaining engines due to budget cap.") break eng = ENGINES[key] filename = f"{key}_{args.angle}_{args.expression}.png" output_file = out_dir / filename print(f" [{i+1}/{len(engine_keys)}] {eng['name']}...", end=" ", flush=True) runner = ENGINE_RUNNERS[key] try: result = runner( hero_path=hero_path, output_path=output_file, angle=args.angle, expression=args.expression, environment=environment, identity=identity, lora_url=lora_url, trigger=trigger, lighting=args.lighting, project_path=project_path, character_key=char_upper, project_config=project_config, character_traits=args.character_traits, ) if result["success"]: print(f"OK ({result['elapsed']:.1f}s)") else: print(f"FAILED: {result.get('error', 'unknown')[:60]}") _log_engine_cost(tracker, key, result["success"], result.get("elapsed", 0), char_upper) results.append({ "engine_key": key, "success": result["success"], "elapsed": result.get("elapsed", 0), "error": result.get("error"), "output_path": str(result.get("output", output_file)), }) except Exception as e: print(f"ERROR: {str(e)[:80]}") _log_engine_cost(tracker, key, False, 0, char_upper) results.append({ "engine_key": key, "success": False, "elapsed": 0, "error": str(e)[:200], "output_path": None, }) # Rate limit between calls (especially for Gemini) if i < len(engine_keys) - 1: time.sleep(2) # Generate comparison HTML html_path = out_dir / "comparison.html" generate_html_report( results, hero_path, {"angle": args.angle, "expression": args.expression, "character": char_upper, "environment": environment}, html_path, ) # Summary succeeded = sum(1 for r in results if r["success"]) total_cost = sum(ENGINES[r["engine_key"]]["cost_est"] for r in results if r["success"]) total_time = sum(r["elapsed"] for r in results) print(f"\n{'='*60}") print(f"COMPLETE: {succeeded}/{len(results)} succeeded") print(f" Total time: {total_time:.1f}s") print(f" Est. cost: ~${total_cost:.3f}") print(f" Comparison: {html_path}") print(f"{'='*60}\n") # Compute hero_relative (path relative to project root for serve.py) try: hero_relative = str(hero_path.resolve().relative_to(project_path.resolve())) except ValueError: hero_relative = None # Save results JSON results_json = out_dir / "results.json" results_json.write_text(json.dumps({ "timestamp": datetime.now(timezone.utc).isoformat(), "mode": "comparison", "character": char_upper, "hero": str(hero_path), "hero_relative": hero_relative, "angle": args.angle, "expression": args.expression, "environment": environment, "lighting": args.lighting or "cinematic lighting with modeling", "results": results, }, indent=2)) print(f" Reviewer: http://127.0.0.1:8420/shootout_reviewer.html?project={project_name}&character={char_upper}") return html_path if __name__ == "__main__": main()