#!/usr/bin/env python3 """ flux2_ma_test.py — A/B test: Flux 2 Multi-Angle vs Qwen Multi-Angle (± SeedVR2). Tests whether Flux 2 MA can replace Qwen MA in the pipeline. Usage: python3 flux2_ma_test.py leviathan/ --character JINX python3 flux2_ma_test.py leviathan/ --character JINX --option-a python3 flux2_ma_test.py leviathan/ --character JINX --angles front,profile_right python3 flux2_ma_test.py leviathan/ --character JINX --flux2-only Modes: Default: A/B comparison — Flux 2 MA vs Qwen MA (± SeedVR2) for all angles --option-a: Hybrid pipeline — Flux 2 MA for all angles (neutral), then NBP expression injection on frontal angles only, SeedVR2 on everything Default test angles: front, 3/4R, profileR, back, high, low, CU-front, CU-3/4 """ import argparse import json import os import sys import time import urllib.request from datetime import datetime from pathlib import Path # ── Angle Map (shared with engine_shootout.py) ─────────────────────── # Flux 2 MA uses same numeric params but vertical range is 0-60° (no negatives) 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": 0, "z": 5}, # Flux 2 can't do negative v "high_angle": {"h": 0, "v": 30, "z": 0}, "closeup_front": {"h": 0, "v": 0, "z": 10}, "closeup_three_quarter":{"h": 45, "v": 0, "z": 10}, } # Qwen supports negative vertical; Flux 2 doesn't 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}, "closeup_front": {"h": 0, "v": 0, "z": 10}, "closeup_three_quarter":{"h": 45, "v": 0, "z": 10}, } ANGLE_LABELS = { "front": "Front", "three_quarter_right": "3/4R", "profile_right": "ProfR", "back_right": "BackR", "back": "Back", "back_left": "BackL", "profile_left": "ProfL", "three_quarter_left": "3/4L", "low_angle": "Low", "high_angle": "High", "closeup_front": "CU-F", "closeup_three_quarter": "CU-3/4", } DEFAULT_TEST_ANGLES = [ "front", "three_quarter_right", "profile_right", "back", "high_angle", "low_angle", "closeup_front", "closeup_three_quarter", ] # Frontal angles where expressions are readable (Option A) EXPRESSION_ANGLES = {"front", "closeup_front", "closeup_three_quarter"} DEFAULT_EXPRESSIONS = [ "neutral — relaxed face, soft gaze, mouth closed", "smiling — genuine warm smile, raised cheeks, crow's feet at eyes", "angry — furrowed brow, clenched jaw, narrowed eyes, flared nostrils", "focused — narrowed eyes, slight frown, intent concentrated gaze", "exhausted — heavy-lidded eyes, slight frown, drained hollow gaze", ] def upload_with_retry(file_path: Path, max_retries: int = 3) -> str: """Upload file to fal.ai with retry logic and auto-resize if too large.""" import fal_client from PIL import Image import io data = file_path.read_bytes() size_mb = len(data) / 1024 / 1024 # If > 2 MB, resize to 1024px max dimension if size_mb > 2.0: img = Image.open(file_path) img.thumbnail((1024, 1024), Image.LANCZOS) buf = io.BytesIO() img.save(buf, format="JPEG", quality=92) data = buf.getvalue() print(f"(resized {size_mb:.1f}MB → {len(data)/1024/1024:.1f}MB)", end=" ", flush=True) mime = "image/jpeg" for attempt in range(max_retries): try: url = fal_client.upload(data, mime, file_name="hero.jpg") return url except Exception as e: if attempt < max_retries - 1: wait = 5 * (attempt + 1) print(f"(upload retry {attempt+1}, wait {wait}s)", end=" ", flush=True) time.sleep(wait) else: raise def resolve_hero(project_path: Path, character_key: str) -> Path: """Find hero image via ProjectPaths resolver. Returns the resolved hero Path, or None if not found. """ try: _recoil_root = str(Path(__file__).resolve().parent.parent) if _recoil_root not in sys.path: sys.path.insert(0, _recoil_root) 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 run_flux2_ma(hero_path: Path, output_path: Path, angle: str, cached_url: str = None) -> dict: """Flux 2 Multi-Angle: generate a specific camera angle.""" import fal_client image_url = cached_url or upload_with_retry(hero_path) params = ANGLE_MAP.get(angle, {"h": 0, "v": 0, "z": 5}) t0 = time.time() result = fal_client.subscribe( "fal-ai/flux-2-lora-gallery/multiple-angles", arguments={ "image_urls": [image_url], "horizontal_angle": params["h"], "vertical_angle": params["v"], "zoom": params["z"], "lora_scale": 1.0, "image_size": "square_hd", "num_inference_steps": 40, "guidance_scale": 2.5, "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": str(result)[:200]} def run_qwen_ma(hero_path: Path, output_path: Path, angle: str, cached_url: str = None) -> dict: """Qwen Multi-Angle: generate a specific camera angle.""" import fal_client image_url = cached_url or upload_with_retry(hero_path) params = QWEN_ANGLE_MAP.get(angle, {"h": 0, "v": 0, "z": 5}) t0 = time.time() result = fal_client.subscribe( "fal-ai/qwen-image-edit-2511-multiple-angles", arguments={ "image_urls": [image_url], "horizontal_angle": params["h"], "vertical_angle": params["v"], "zoom": params["z"], "lora_scale": 0.9, "image_size": "square_hd", "num_inference_steps": 28, "guidance_scale": 4.5, "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": str(result)[:200]} def run_seedvr2(input_path: Path, output_path: Path) -> dict: """SeedVR2 non-generative quality upscale.""" import fal_client 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"} def run_nbp_expression(input_path: Path, output_path: Path, angle: str, expression: str, environment: str, lighting: str = None) -> dict: """NBP (Gemini) expression injection — takes MA output, adds expression + bg.""" from google import genai from google.genai import types client = genai.Client(api_key=os.environ.get("GOOGLE_API_KEY")) img_bytes = input_path.read_bytes() mime = "image/jpeg" if input_path.suffix.lower() in (".jpg", ".jpeg") else "image/png" angle_desc = ANGLE_LABELS.get(angle, angle) lighting_desc = lighting or "cinematic lighting with modeling" prompt_text = ( "This is the SAME PERSON — face identity locked. DO NOT generate a new face.\n\n" 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" "Maintain identical skull structure, brow ridge, nose bridge, nose width, " "cheekbone position, chin shape, ear shape, eye spacing, eye size, iris color, " "skin tone, skin texture, hair color, and hair texture.\n" "Expression muscles may move freely — brows, eyelids, nostrils, lips, jaw — " "these are temporary muscular movements, not identity changes.\n\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" "Shot at 85mm f/1.8. Sharp focus on eyes. Tack-sharp iris detail with visible iris fibers. " "Subtle catchlights in the pupils.\n\n" "DO NOT smooth, beautify, or stylize facial features. No global smoothing. No airbrushing.\n" "Strictly preserve pore texture, freckles, fine lines, and skin imperfections.\n" "Single photorealistic photograph. One person only. No text. No split panels." ) parts = [ types.Part(inline_data=types.Blob(mime_type=mime, data=img_bytes)), types.Part(text=prompt_text), ] t0 = time.time() response = client.models.generate_content( model="gemini-3-pro-image-preview", contents=[types.Content(parts=parts)], config=types.GenerateContentConfig( response_modalities=["IMAGE", "TEXT"], temperature=1.0, ), ) elapsed = time.time() - t0 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} error_text = "No image returned" if response.candidates: for part in response.candidates[0].content.parts: if hasattr(part, "text") and part.text: error_text = part.text[:200] break return {"success": False, "elapsed": elapsed, "error": error_text} def generate_comparison_html(results: list, hero_path: Path, output_dir: Path): """Generate side-by-side HTML comparison page.""" hero_rel = os.path.relpath(hero_path, output_dir) rows_html = "" for r in results: angle = r["angle"] label = ANGLE_LABELS.get(angle, angle) cells = f'{label}
{angle}' for engine in ["flux2_ma", "qwen_ma", "flux2_ma_seedvr2", "qwen_ma_seedvr2"]: entry = r.get(engine) if entry and entry.get("success"): rel = os.path.relpath(entry["output"], output_dir) elapsed = entry.get("elapsed", 0) cells += f'
{elapsed:.1f}s
' elif entry: cells += f'FAIL
{entry.get("error", "")[:40]}' else: cells += '—' rows_html += f"{cells}\n" html = f""" Flux 2 MA vs Qwen MA — A/B Test

Flux 2 MA vs Qwen MA — A/B Comparison

Hero
Hero: {hero_path.name}
Date: {datetime.now().strftime('%Y-%m-%d %H:%M')}
Angles: {len(results)}
{rows_html}
Angle Flux 2 MA Qwen MA Flux 2 MA + SeedVR2 Qwen MA + SeedVR2
""" html_path = output_dir / "comparison.html" html_path.write_text(html) return html_path def generate_option_a_html(results: list, ma_outputs: dict, nbp_outputs: list, hero_path: Path, output_dir: Path, has_seedvr2: bool, angles: list, expressions: list, environment: str): """Generate HTML review page for Option A results.""" hero_rel = os.path.relpath(hero_path, output_dir) # ── Section 1: Angle overview (MA outputs, all angles) ── angle_rows = "" for angle in angles: label = ANGLE_LABELS.get(angle, angle) is_expr = angle in EXPRESSION_ANGLES cells = f'{label}
{angle}' if is_expr: cells += '
+expr' else: cells += '
neutral' cells += '' # MA raw ma_file = output_dir / f"{angle}_flux2_ma.png" if ma_file.exists(): rel = os.path.relpath(ma_file, output_dir) cells += f'' else: cells += 'FAIL' # MA + SeedVR2 svr2_file = output_dir / f"{angle}_flux2_ma_seedvr2.png" if has_seedvr2 and svr2_file.exists(): rel = os.path.relpath(svr2_file, output_dir) cells += f'' else: cells += '—' angle_rows += f"{cells}\n" # ── Section 2: Expression grid (frontal angles × expressions) ── expr_rows = "" for angle in [a for a in angles if a in EXPRESSION_ANGLES]: label = ANGLE_LABELS.get(angle, angle) for expr in expressions: expr_tag = expr.split(" — ")[0] if " — " in expr else expr[:15] safe_expr = expr_tag.replace(" ", "_").lower() cells = f'{label}
{expr_tag}' # NBP raw nbp_file = output_dir / f"{angle}_nbp_{safe_expr}.png" if nbp_file.exists(): rel = os.path.relpath(nbp_file, output_dir) cells += f'' else: cells += 'FAIL' # NBP + SeedVR2 svr2_file = output_dir / f"{angle}_nbp_{safe_expr}_seedvr2.png" if has_seedvr2 and svr2_file.exists(): rel = os.path.relpath(svr2_file, output_dir) cells += f'' else: cells += '—' # MA source (for comparison) ma_file = output_dir / f"{angle}_flux2_ma.png" if ma_file.exists(): rel = os.path.relpath(ma_file, output_dir) cells += f'
MA input
' else: cells += '—' expr_rows += f"{cells}\n" html = f""" Option A — Flux 2 MA + NBP Expressions

Option A — Flux 2 MA + NBP Expression Injection

Hero
Hero: {hero_path.name}
Date: {datetime.now().strftime('%Y-%m-%d %H:%M')}
Pipeline: Flux 2 MA → NBP (frontals) → SeedVR2
Env: {environment[:60]}...

All Angles (Flux 2 MA — neutral)

{angle_rows}
Angle Flux 2 MA (raw) + SeedVR2

Expression Variants (NBP on frontal angles)

{expr_rows}
Angle × Expression NBP (raw) NBP + SeedVR2 MA Source
""" html_path = output_dir / "comparison.html" html_path.write_text(html) return html_path def main(): parser = argparse.ArgumentParser(description="A/B test: Flux 2 MA vs Qwen MA (± SeedVR2)") 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("--angles", default=None, help="Comma-separated angles (default: 6-angle test set)") parser.add_argument("--hero", default=None, help="Explicit hero image path") parser.add_argument("--no-seedvr2", action="store_true", help="Skip SeedVR2 upscale pass") parser.add_argument("--flux2-only", action="store_true", help="Only test Flux 2 MA (skip Qwen)") parser.add_argument("--qwen-only", action="store_true", help="Only test Qwen MA (skip Flux 2)") parser.add_argument("--option-a", action="store_true", help="Option A: Flux 2 MA all angles + NBP expressions on frontals + SeedVR2") parser.add_argument("--expressions", default=None, help="Comma-separated expressions for Option A (default: 5 standard)") parser.add_argument("--environment", default="sparse concrete bunker interior with a single harsh fluorescent tube overhead", help="Background environment for NBP expression pass") parser.add_argument("--lighting", default="cinematic lighting with modeling", help="Lighting style for NBP expression pass") args = parser.parse_args() project_path = Path(args.project) 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) sys.exit(1) # Resolve angles if args.angles: angles = [a.strip() for a in args.angles.split(",")] else: angles = DEFAULT_TEST_ANGLES do_seedvr2 = not args.no_seedvr2 do_flux2 = not args.qwen_only do_qwen = not args.flux2_only option_a = args.option_a # Parse expressions for Option A if args.expressions: expressions = [e.strip() for e in args.expressions.split("|")] else: expressions = DEFAULT_EXPRESSIONS environment = args.environment lighting = args.lighting # Output dir timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") mode_label = "option_a" if option_a else "ab_test" output_dir = (project_path / "visual" / "lora" / char_upper / "candidates" / f"flux2_ma_{mode_label}_{timestamp}") output_dir.mkdir(parents=True, exist_ok=True) if option_a: # Option A: Flux 2 MA all angles → NBP expressions on frontals → SeedVR2 on everything expr_angles = [a for a in angles if a in EXPRESSION_ANGLES] neutral_angles = [a for a in angles if a not in EXPRESSION_ANGLES] nbp_jobs = len(expr_angles) * len(expressions) ma_jobs = len(angles) seedvr2_jobs = ma_jobs + nbp_jobs if do_seedvr2 else 0 total_api_calls = ma_jobs + nbp_jobs + seedvr2_jobs est_cost = (ma_jobs * 0.021 # Flux 2 MA + nbp_jobs * 0.065 # NBP + seedvr2_jobs * 0.001) # SeedVR2 print(f"\n{'='*60}") print(f"OPTION A — HYBRID PIPELINE TEST") print(f"{'='*60}") print(f" Character: {char_upper}") print(f" Hero: {hero_path.name} ({hero_path.stat().st_size / 1024 / 1024:.1f} MB)") print(f" Pipeline: Flux 2 MA → NBP (frontals) → SeedVR2") print(f" Angles: {len(angles)} ({len(expr_angles)} frontal + {len(neutral_angles)} neutral-only)") print(f" Expressions: {len(expressions)} on frontal angles") for expr in expressions: tag = expr.split(" — ")[0] if " — " in expr else expr[:20] print(f" • {tag}") print(f" Environment: {environment[:50]}...") print(f" API calls: ~{total_api_calls}") print(f" Est. cost: ~${est_cost:.2f}") print(f" Output: {output_dir}") print(f"{'='*60}") print() results = [] succeeded = 0 failed = 0 job_num = 0 total_jobs = ma_jobs + nbp_jobs # ── Upload hero once ── print(" Uploading hero...", end=" ", flush=True) hero_url = upload_with_retry(hero_path) print("OK") print() # ── Phase 1: Flux 2 MA for ALL angles (neutral) ── print(" PHASE 1: Flux 2 MA — all angles") print(f" {'─'*50}") ma_outputs = {} # angle → path for i, angle in enumerate(angles): job_num += 1 label = ANGLE_LABELS.get(angle, angle) flux2_file = output_dir / f"{angle}_flux2_ma.png" print(f" [{job_num}/{total_jobs}] {label} — Flux 2 MA...", end=" ", flush=True) try: r = run_flux2_ma(hero_path, flux2_file, angle, cached_url=hero_url) if r["success"]: print(f"OK ({r['elapsed']:.1f}s)") ma_outputs[angle] = flux2_file results.append({ "angle": angle, "expression": "neutral (MA only)", "stage": "flux2_ma", "success": True, "elapsed": r["elapsed"], "output": str(flux2_file), }) succeeded += 1 else: print(f"FAIL: {r.get('error', '')[:60]}") results.append({ "angle": angle, "expression": "neutral (MA only)", "stage": "flux2_ma", "success": False, "error": r.get("error", "")[:200], }) failed += 1 except Exception as e: print(f"ERROR: {str(e)[:60]}") results.append({ "angle": angle, "expression": "neutral (MA only)", "stage": "flux2_ma", "success": False, "error": str(e)[:200], }) failed += 1 if i < len(angles) - 1: time.sleep(1) print() # ── Phase 2: NBP expression pass on frontal angles ── print(" PHASE 2: NBP expression injection — frontal angles") print(f" {'─'*50}") nbp_outputs = [] # list of (angle, expr_tag, path) for angle in expr_angles: if angle not in ma_outputs: print(f" Skipping {angle} — no MA output") continue ma_input = ma_outputs[angle] label = ANGLE_LABELS.get(angle, angle) for expr in expressions: job_num += 1 expr_tag = expr.split(" — ")[0] if " — " in expr else expr[:15] safe_expr = expr_tag.replace(" ", "_").lower() nbp_file = output_dir / f"{angle}_nbp_{safe_expr}.png" print(f" [{job_num}/{total_jobs}] {label} × {expr_tag} — NBP...", end=" ", flush=True) try: r = run_nbp_expression(ma_input, nbp_file, angle, expr, environment, lighting) if r["success"]: print(f"OK ({r['elapsed']:.1f}s)") nbp_outputs.append((angle, expr_tag, nbp_file)) results.append({ "angle": angle, "expression": expr_tag, "stage": "nbp", "success": True, "elapsed": r["elapsed"], "output": str(nbp_file), }) succeeded += 1 else: print(f"FAIL: {r.get('error', '')[:60]}") results.append({ "angle": angle, "expression": expr_tag, "stage": "nbp", "success": False, "error": r.get("error", "")[:200], }) failed += 1 except Exception as e: print(f"ERROR: {str(e)[:60]}") results.append({ "angle": angle, "expression": expr_tag, "stage": "nbp", "success": False, "error": str(e)[:200], }) failed += 1 time.sleep(1) print() # ── Phase 3: SeedVR2 on everything ── if do_seedvr2: print(" PHASE 3: SeedVR2 quality upscale — all outputs") print(f" {'─'*50}") seedvr2_count = 0 # SeedVR2 on neutral-only MA outputs for angle in neutral_angles: if angle not in ma_outputs: continue ma_file = ma_outputs[angle] label = ANGLE_LABELS.get(angle, angle) svr2_file = output_dir / f"{angle}_flux2_ma_seedvr2.png" print(f" {label} neutral — SeedVR2...", end=" ", flush=True) try: r = run_seedvr2(ma_file, svr2_file) if r["success"]: print(f"OK ({r['elapsed']:.1f}s)") seedvr2_count += 1 else: print(f"FAIL") except Exception as e: print(f"ERROR: {str(e)[:60]}") # SeedVR2 on frontal MA outputs (neutral version) for angle in expr_angles: if angle not in ma_outputs: continue ma_file = ma_outputs[angle] label = ANGLE_LABELS.get(angle, angle) svr2_file = output_dir / f"{angle}_flux2_ma_seedvr2.png" print(f" {label} neutral — SeedVR2...", end=" ", flush=True) try: r = run_seedvr2(ma_file, svr2_file) if r["success"]: print(f"OK ({r['elapsed']:.1f}s)") seedvr2_count += 1 else: print(f"FAIL") except Exception as e: print(f"ERROR: {str(e)[:60]}") # SeedVR2 on NBP expression outputs for angle, expr_tag, nbp_file in nbp_outputs: label = ANGLE_LABELS.get(angle, angle) safe_expr = expr_tag.replace(" ", "_").lower() svr2_file = output_dir / f"{angle}_nbp_{safe_expr}_seedvr2.png" print(f" {label} × {expr_tag} — SeedVR2...", end=" ", flush=True) try: r = run_seedvr2(nbp_file, svr2_file) if r["success"]: print(f"OK ({r['elapsed']:.1f}s)") seedvr2_count += 1 else: print(f"FAIL") except Exception as e: print(f"ERROR: {str(e)[:60]}") print(f"\n SeedVR2: {seedvr2_count} upscaled") print() # Generate Option A HTML html_path = generate_option_a_html(results, ma_outputs, nbp_outputs, hero_path, output_dir, do_seedvr2, angles, expressions, environment) # Save results JSON results_json = output_dir / "results.json" results_json.write_text(json.dumps(results, indent=2, default=str)) print(f"{'='*60}") print(f"OPTION A COMPLETE — {char_upper}") print(f"{'='*60}") print(f" Phase 1 (Flux 2 MA): {len(ma_outputs)}/{len(angles)} angles") print(f" Phase 2 (NBP expr): {len(nbp_outputs)}/{nbp_jobs} expressions") print(f" Phase 3 (SeedVR2): {'ON' if do_seedvr2 else 'OFF'}") print(f" Total succeeded: {succeeded}") print(f" Total failed: {failed}") print(f" Comparison: {html_path}") print(f"{'='*60}\n") else: # ── A/B comparison mode (original) ── # Cost estimates jobs = len(angles) engines_per_angle = (1 if do_flux2 else 0) + (1 if do_qwen else 0) seedvr2_mult = 2 if do_seedvr2 else 1 total_api_calls = jobs * engines_per_angle * seedvr2_mult est_cost = jobs * ((0.021 if do_flux2 else 0) + (0.035 if do_qwen else 0) + (0.002 if do_seedvr2 else 0) * engines_per_angle) print(f"\n{'='*60}") print(f"FLUX 2 MA vs QWEN MA — A/B TEST") print(f"{'='*60}") print(f" Character: {char_upper}") print(f" Hero: {hero_path.name} ({hero_path.stat().st_size / 1024 / 1024:.1f} MB)") print(f" Angles: {len(angles)}") print(f" Engines: {'Flux 2 MA' if do_flux2 else ''}{' + ' if do_flux2 and do_qwen else ''}{'Qwen MA' if do_qwen else ''}") print(f" SeedVR2: {'ON' if do_seedvr2 else 'OFF'}") print(f" API calls: ~{total_api_calls}") print(f" Est. cost: ~${est_cost:.2f}") print(f" Output: {output_dir}") print(f"{'='*60}") print() results = [] succeeded = 0 failed = 0 for i, angle in enumerate(angles): label = ANGLE_LABELS.get(angle, angle) print(f" [{i+1}/{len(angles)}] {label} ({angle})") print(f" {'─'*50}") angle_results = {"angle": angle} # ── Flux 2 MA ── if do_flux2: flux2_file = output_dir / f"{angle}_flux2_ma.png" print(f" Flux 2 MA...", end=" ", flush=True) try: r = run_flux2_ma(hero_path, flux2_file, angle) if r["success"]: print(f"OK ({r['elapsed']:.1f}s)") angle_results["flux2_ma"] = r succeeded += 1 if do_seedvr2: seedvr2_file = output_dir / f"{angle}_flux2_ma_seedvr2.png" print(f" Flux 2 MA + SeedVR2...", end=" ", flush=True) r2 = run_seedvr2(flux2_file, seedvr2_file) if r2["success"]: print(f"OK ({r2['elapsed']:.1f}s)") r2["elapsed"] = r["elapsed"] + r2["elapsed"] angle_results["flux2_ma_seedvr2"] = r2 else: print(f"FAIL") angle_results["flux2_ma_seedvr2"] = r2 else: print(f"FAIL: {r.get('error', '')[:60]}") angle_results["flux2_ma"] = r failed += 1 except Exception as e: print(f"ERROR: {str(e)[:60]}") angle_results["flux2_ma"] = {"success": False, "error": str(e)[:200], "elapsed": 0} failed += 1 # ── Qwen MA ── if do_qwen: qwen_file = output_dir / f"{angle}_qwen_ma.png" print(f" Qwen MA...", end=" ", flush=True) try: r = run_qwen_ma(hero_path, qwen_file, angle) if r["success"]: print(f"OK ({r['elapsed']:.1f}s)") angle_results["qwen_ma"] = r succeeded += 1 if do_seedvr2: seedvr2_file = output_dir / f"{angle}_qwen_ma_seedvr2.png" print(f" Qwen MA + SeedVR2...", end=" ", flush=True) r2 = run_seedvr2(qwen_file, seedvr2_file) if r2["success"]: print(f"OK ({r2['elapsed']:.1f}s)") r2["elapsed"] = r["elapsed"] + r2["elapsed"] angle_results["qwen_ma_seedvr2"] = r2 else: print(f"FAIL") angle_results["qwen_ma_seedvr2"] = r2 else: print(f"FAIL: {r.get('error', '')[:60]}") angle_results["qwen_ma"] = r failed += 1 except Exception as e: print(f"ERROR: {str(e)[:60]}") angle_results["qwen_ma"] = {"success": False, "error": str(e)[:200], "elapsed": 0} failed += 1 results.append(angle_results) print() if i < len(angles) - 1: time.sleep(2) # Generate comparison HTML html_path = generate_comparison_html(results, hero_path, output_dir) # Save results JSON results_json = output_dir / "results.json" serializable = [] for r in results: sr = {"angle": r["angle"]} for key in ["flux2_ma", "qwen_ma", "flux2_ma_seedvr2", "qwen_ma_seedvr2"]: if key in r: entry = dict(r[key]) if "output" in entry: entry["output"] = str(entry["output"]) sr[key] = entry serializable.append(sr) results_json.write_text(json.dumps(serializable, indent=2)) print(f"{'='*60}") print(f"TEST COMPLETE — {char_upper}") print(f"{'='*60}") print(f" Succeeded: {succeeded}") print(f" Failed: {failed}") print(f" Comparison: {html_path}") print(f"{'='*60}\n") if __name__ == "__main__": main()