#!/usr/bin/env python3 """ prep_character_angles.py — Character reference pipeline. Two paths for generating multi-angle character reference sheets: Path A (Direct Hero): Supply an existing hero image (from MJ or NBP pick). → Qwen Multi-Angle rotation → rembg → white background → Gate 0 Path B (Concepting / Casting Tool): Supply a text description only. → NBP 3x3 grid (9 interpretations, ~$0.13) → User picks best panel → Optional NBP refinement OR export for MJ polish → Selected hero feeds into Path A angle generation Both paths end with: - 4 angle refs saved to projects/{project}/assets/char/{character}/ - Gate 0 validation on the final ref set - Cost tracking Ported from Recoil's test_fullbody_angles.py + batch_generate_refs.py (LoRA training code discarded per ADR). """ import json import logging import os import sys from pathlib import Path from typing import Optional # Project root setup _PROJECT_ROOT = Path(__file__).parent.parent if str(_PROJECT_ROOT) not in sys.path: sys.path.insert(0, str(_PROJECT_ROOT)) from recoil.core.paths import projects_root, ProjectPaths from recoil.core.model_profiles import get_model from recoil.core.prompt_config import get_constant logger = logging.getLogger(__name__) PROJECTS_DIR = projects_root() def _char_refs_dir(project: Optional[str] = None) -> Path: """Return the character ref directory for a project.""" return ProjectPaths.for_project(project).asset_class_dir("char") # Default angles for multi-angle generation DEFAULT_ANGLES = ["front", "profile", "three_quarter", "back"] # QwenMA numeric angle parameters for fal-ai/qwen-image-edit-2511-multiple-angles # horizontal_angle: 0=front, 45=3/4, 90=profile, 180=back # vertical_angle: -30=low angle, 0=eye level, 30=elevated # zoom: 0=wide (full body), 5=medium, 10=close-up QWEN_MA_ANGLES = [ {"name": "front", "horizontal_angle": 0, "vertical_angle": 0, "zoom": 5}, {"name": "three_quarter", "horizontal_angle": 45, "vertical_angle": 0, "zoom": 5}, {"name": "profile", "horizontal_angle": 90, "vertical_angle": 0, "zoom": 5}, {"name": "back", "horizontal_angle": 180, "vertical_angle": 0, "zoom": 5}, ] # ADR-C03: IP-safe codename mapping — prevents famous character names from biasing generation _CODENAME_MAP = { "JINX": "Subject_Alpha", "TORCH": "Subject_Alpha", "KIAN": "Subject_Bravo", "WREN": "Subject_Bravo", "VAREK": "Subject_Gamma", "CHEN": "Subject_Delta", } # ADR-C01: Keywords indicating synthetic/android character (switches diegetic frame) _SYNTHETIC_KEYWORDS = {"chassis", "synthetic skin", "android", "mechanical", "alloy", "optics", "processor"} def _get_codename(char_id: str) -> str: """Map character name to sterile codename for API prompts (ADR-C03).""" return _CODENAME_MAP.get(char_id.upper(), char_id) def _is_synthetic(description: str) -> bool: """Detect android/synthetic characters from description keywords (ADR-C01).""" desc_lower = description.lower() return any(kw in desc_lower for kw in _SYNTHETIC_KEYWORDS) # Angle-specific prompt fragments for Qwen Multi-Angle ANGLE_PROMPTS = { "front": "front view, facing camera directly, symmetrical framing", "profile": "profile view, 90-degree side angle, clean silhouette", "three_quarter": "three-quarter view, 45-degree angle, slight head turn", "back": "back view, facing away from camera, full body from behind", } def check_intake(project: str, character: str) -> Optional[Path]: """Check the project intake folder for a hero image matching the character. Looks for files named like {character}_hero.* or {character}.* in projects/{project}/intake/. Returns the path if found, None otherwise. """ intake_dir = PROJECTS_DIR / project / "intake" if not intake_dir.is_dir(): return None char_lower = character.lower().replace(" ", "_") for pattern in [f"{char_lower}_hero.*", f"{char_lower}.*"]: matches = list(intake_dir.glob(pattern)) # Filter to image files matches = [m for m in matches if m.suffix.lower() in {".png", ".jpg", ".jpeg", ".webp"}] if matches: return matches[0] return None def process_intake(project: str, character: str, intake_path: Path) -> Path: """Move a hero image from intake to the character's ref folder. Copies the image to assets/char/{character}/hero.{ext}, then deletes the original from intake. Returns the destination path. """ char_id = character.upper().replace(" ", "_") char_dir = _char_refs_dir(project) / char_id.lower() char_dir.mkdir(parents=True, exist_ok=True) dest = char_dir / f"hero{intake_path.suffix.lower()}" import shutil shutil.copy2(intake_path, dest) logger.info("Intake: copied %s → %s", intake_path, dest) # Remove from intake intake_path.unlink() logger.info("Intake: removed %s", intake_path) return dest def prep_character( character: str, hero_image: Optional[Path] = None, description: Optional[str] = None, project: Optional[str] = None, angles: Optional[list] = None, remove_bg: bool = True, dry_run: bool = False, gender: Optional[str] = None, backend: str = "gemini", ) -> dict: """Generate multi-angle character reference set. Args: character: Character name/ID (e.g. "TORCH", "WREN"). hero_image: Path A — existing hero image to rotate into angles. If not provided and project is set, checks project intake folder. description: Path B — text description for NBP exploration grid. project: Project name (for output organization + intake folder). angles: Angle names to generate (default: front, profile, 3/4, back). remove_bg: Whether to run rembg on angle outputs. dry_run: Preview operations without API calls. backend: Angle generation backend — "gemini" (NBP grid) or "qwen-ma" (fal.ai). Returns: dict with keys: character, path, angles_generated, gate_0_result, cost """ # Check intake folder if no hero image provided if hero_image is None and project: intake_path = check_intake(project, character) if intake_path: logger.info("Found hero in intake: %s", intake_path) hero_image = process_intake(project, character, intake_path) if hero_image is None and description is None: raise ValueError("Must provide either hero_image (Path A) or description (Path B)") if angles is None: angles = DEFAULT_ANGLES char_id = character.upper().replace(" ", "_") char_dir = _char_refs_dir(project) / char_id.lower() char_dir.mkdir(parents=True, exist_ok=True) total_cost = 0.0 result = { "character": char_id, "path": str(char_dir), "angles_generated": [], "gate_0_result": None, "cost": 0.0, } if dry_run: logger.info("[DRY RUN] Would generate %d angles for %s", len(angles), char_id) result["angles_generated"] = angles result["dry_run"] = True return result # Path B: Concepting via NBP grid if hero_image is None: # Gender comes from bible "gender" field (passed in by caller). # No field = no anchor — correct for genderless characters (aliens, creatures). resolved_gender = gender logger.info("Path B: Concepting %s via NBP 3x3 grid (gender=%s)", char_id, resolved_gender) grid_result = _generate_concept_grid(char_id, description, char_dir, gender=resolved_gender) total_cost += grid_result["cost"] if grid_result["hero_path"] is None: logger.warning("No hero selected from grid. Exiting Path B.") result["cost"] = total_cost result["grid_result"] = grid_result return result hero_image = Path(grid_result["hero_path"]) result["grid_result"] = grid_result # Path A: Generate all angles from hero image logger.info("Generating angles for %s from hero: %s (backend=%s)", char_id, hero_image, backend) if not hero_image.exists(): raise FileNotFoundError(f"Hero image not found: {hero_image}") if backend == "qwen-ma": # QwenMA: upload hero, generate individual angle images via fal.ai import fal_client logger.info("Uploading hero to fal.ai storage...") hero_url = fal_client.upload_file(str(hero_image)) # Map angle names to QwenMA angle configs qwen_angle_map = {a["name"]: a for a in QWEN_MA_ANGLES} qwen_angles_to_run = [qwen_angle_map[a] for a in angles if a in qwen_angle_map] if not qwen_angles_to_run: raise ValueError(f"No matching QwenMA angles for: {angles}") angle_paths = _generate_angles_qwen_ma( hero_url=hero_url, angles=qwen_angles_to_run, output_dir=char_dir, ) # Rename to canonical {char_id}_{angle}.png format for p in angle_paths: angle_name = p.stem # e.g. "front", "profile" canonical = char_dir / f"{char_id.lower()}_{angle_name}.png" if p != canonical: p.rename(canonical) qwen_cost = len(angle_paths) * 0.035 total_cost += qwen_cost result["backend"] = "qwen-ma" result["angles_generated"] = [a["name"] for a in qwen_angles_to_run if (char_dir / f"{char_id.lower()}_{a['name']}.png").exists()] else: # Gemini NBP: generate all angles in a single 2x2 grid grid_result = _generate_angle_grid( char_id=char_id, hero_path=hero_image, output_dir=char_dir, angles=angles, remove_bg=remove_bg, ) total_cost += grid_result.get("cost", 0.0) result["angle_grid"] = grid_result result["backend"] = "gemini" if grid_result.get("grid_path"): # Panels are saved but need approval before rembg processing. # Mark which angles were generated. result["angles_generated"] = grid_result.get("angles", []) result["cost"] = round(total_cost, 4) logger.info( "Character %s angles complete: %d angles, cost: $%.4f, backend: %s", char_id, len(result["angles_generated"]), total_cost, backend, ) return result def _generate_concept_grid( char_id: str, description: str, output_dir: Path, gender: Optional[str] = None, ) -> dict: """Path B: Generate 3x3 NBP grid for character concepting (ADR-C01–C04, C07). Uses diegetic framing ("photographic contact sheet" / "casting call") to force photorealistic output. Switches frame for synthetic characters ("VFX studio"). Uses codename mapping to prevent IP bias (ADR-C03). Returns dict with grid_path, panels (list of paths), hero_path (user-selected), cost. """ from recoil.execution.api_client import get_client # ADR-C03: Use codename to prevent IP contamination codename = _get_codename(char_id) # ADR-C01: Switch diegetic frame based on character type synthetic = _is_synthetic(description) if synthetic: diegetic_frame = ( "A practical VFX studio contact sheet. " "9 different high-budget animatronic prop tests." ) texture_anchor = get_constant("casting", "casting_texture_synthetic") else: diegetic_frame = ( "A casting director's audition photo array. " "9 DIFFERENT actors auditioning for the exact same role. " "Simple headshot-style photos, no borders, no film frames, no sprocket holes." ) texture_anchor = get_constant("casting", "casting_texture_human") # Gender anchor: prepend explicit gender to PHYSICALITY line # NBP treats gender as visual, not grammatical — pronouns in prose don't override defaults if gender and gender.lower() == "female": physicality = f"Female. {description}" elif gender and gender.lower() == "male": physicality = f"Male. {description}" else: physicality = description concept_prompt = ( f"CRITICAL DIRECTIVE: Generate a single image containing a 2x3 grid of photos.\n" f"DIEGETIC FRAMING: {diegetic_frame}\n\n" f"SUBJECT CODENAME: {codename}\n" f"PHYSICALITY: {physicality}\n\n" f"GRID STRUCTURE: 3 columns by 2 rows. Strictly isolated panels, no overlapping elements. " f"Each of the 6 panels must feature a DIFFERENT interpretation of the description " f"(vary the facial structure, styling, and attitude, but keep the age range identical). " f"All panels must perfectly match the core physical traits. " f"Frame each panel as a 3/4 medium-full shot.\n\n" f"PHOTOGRAPHIC ANCHORS:\n" f"- Medium: 35mm motion picture film still, shot on {get_constant('casting', 'casting_camera')}.\n" f"- Lighting: Soft, directional studio lighting. {get_constant('casting', 'casting_background')}.\n" f"- Texture (CRITICAL): {texture_anchor}\n" f"- Negative constraints: {get_constant('casting', 'casting_anti_airbrush')}" ) client = get_client(get_model("production_hq", "image")) from google.genai import types as genai_types # ADR-C04: Locked API configuration for casting grids config = genai_types.GenerateContentConfig( temperature=0.4, response_modalities=["IMAGE", "TEXT"], image_config=genai_types.ImageConfig( aspect_ratio="2:3", ), ) try: api_client = client._get_client() response = api_client.models.generate_content( model=get_model("production_hq", "image"), contents=concept_prompt, config=config, ) except Exception as e: logger.error("Concept grid generation failed: %s", e) return {"grid_path": None, "panels": [], "hero_path": None, "cost": 0.134} # Extract and save grid image grid_path = output_dir / f"{char_id.lower()}_concept_grid.png" cost = 0.134 # NBP cost if response and response.candidates: for candidate in response.candidates: if candidate.content and candidate.content.parts: for part in candidate.content.parts: if hasattr(part, "inline_data") and part.inline_data: grid_path.write_bytes(part.inline_data.data) logger.info("Concept grid saved: %s", grid_path) break # Split grid into panels panels = _split_grid_into_panels(grid_path, output_dir, char_id) return { "grid_path": str(grid_path) if grid_path.exists() else None, "panels": [str(p) for p in panels], "hero_path": None, # User must pick — set via UI or CLI "cost": cost, } def _detect_content_bounds(img) -> tuple: """Detect the actual grid content area, skipping labels and borders. NBP often generates grids with a title label band (e.g., "CASTING CALL: SUBJECT_ALPHA") at the top. The label text has high std deviation (like photo content), so simple thresholding misidentifies it as content. Strategy: scan from top/bottom looking for a "label-then-gap" pattern. A label band is a short run of high-std rows (< 15% of image height) followed by a gap of low-std rows. The real content starts after the gap. The grid's internal dividers (between photo rows/cols) are fine — we only need to find the outer edges of the grid, not each panel. Returns (top, bottom, left, right) pixel coordinates of the content area. """ import numpy as np arr = np.array(img.convert("L")) # grayscale h, w = arr.shape row_stds = np.array([arr[r, :].std() for r in range(h)]) content_threshold = np.median(row_stds) * 0.5 max_label_height = int(h * 0.15) # labels are < 15% of image min_gap_rows = 3 # at least 3 low-std rows to count as a gap # Scan from top: look for [border?][label?][gap][content] top = 0 r = 0 # Skip leading border (low-std rows) while r < max_label_height and row_stds[r] <= content_threshold: r += 1 label_start = r # Skip label (high-std rows, if any) while r < max_label_height and row_stds[r] > content_threshold: r += 1 label_end = r # Check for gap after label gap_count = 0 while r < max_label_height and row_stds[r] <= content_threshold: gap_count += 1 r += 1 # If we found label + gap, content starts after the gap if label_end > label_start and gap_count >= min_gap_rows: top = r # Scan from bottom: same pattern in reverse bottom = h r = h - 1 while r > h - max_label_height and row_stds[r] <= content_threshold: r -= 1 label_end_b = r while r > h - max_label_height and row_stds[r] > content_threshold: r -= 1 label_start_b = r gap_count_b = 0 while r > h - max_label_height and row_stds[r] <= content_threshold: gap_count_b += 1 r -= 1 if label_end_b > label_start_b and gap_count_b >= min_gap_rows: bottom = r + 1 # Left/right: use first/last above-threshold column col_stds = np.array([arr[:, c].std() for c in range(w)]) col_threshold = np.median(col_stds) * 0.5 left = 0 for c in range(min(w // 4, 200)): if col_stds[c] > col_threshold: left = c break right = w for c in range(w - 1, max(w * 3 // 4, w - 200), -1): if col_stds[c] > col_threshold: right = c + 1 break # Sanity check: content area should be at least 60% of the image content_h = bottom - top content_w = right - left if content_h < h * 0.6 or content_w < w * 0.6: logger.warning( "Content detection found suspicious bounds (%d,%d,%d,%d) " "for %dx%d image — falling back to full image", top, bottom, left, right, w, h, ) return 0, h, 0, w if top > 0 or bottom < h or left > 0 or right < w: logger.info( "Detected label/border crop: top=%d, bottom=%d, left=%d, right=%d " "(removed %dpx top, %dpx bottom, %dpx left, %dpx right)", top, bottom, left, right, top, h - bottom, left, w - right, ) return top, bottom, left, right def _split_grid_into_panels( grid_path: Path, output_dir: Path, char_id: str, ) -> list: """Split a 2x3 grid image into 6 individual panels. Handles NBP grids that include a title/label band at the top (e.g., "CASTING CALL: SUBJECT_ALPHA") by detecting and cropping the label area before splitting. """ try: from PIL import Image except ImportError: logger.warning("PIL not available — cannot split grid into panels") return [] if not grid_path.exists(): return [] img = Image.open(grid_path) w, h = img.size # Detect content bounds (skip labels/borders) top, bottom, left, right = _detect_content_bounds(img) content = img.crop((left, top, right, bottom)) cw, ch = content.size # Use round() for column/row boundaries so remainder pixels are # distributed evenly instead of being lost on the right/bottom edge. col_edges = [round(cw * i / 3) for i in range(4)] # 3 columns row_edges = [round(ch * i / 2) for i in range(3)] # 2 rows panels_dir = output_dir / "concept_panels" panels_dir.mkdir(exist_ok=True) panels = [] for row in range(2): for col in range(3): panel_num = row * 3 + col + 1 box = (col_edges[col], row_edges[row], col_edges[col + 1], row_edges[row + 1]) panel = content.crop(box) panel_path = panels_dir / f"{char_id.lower()}_concept_{panel_num:02d}.png" panel.save(panel_path) panels.append(panel_path) logger.info("Split grid into %d panels at %s", len(panels), panels_dir) return panels def _generate_angle_grid( char_id: str, hero_path: Path, output_dir: Path, angles: list, remove_bg: bool = True, ) -> dict: """Generate all character angles in a single 2x2 NBP grid. One API call produces all angles with consistent identity/lighting. Grid is generated on 18% neutral gray background for optimal rembg extraction. After approval, panels are split and processed: rembg → transparent alpha (archival) → white composite (ref stack). Returns dict with grid_path, panels, angles, cost. """ from recoil.execution.api_client import get_client codename = _get_codename(char_id) # Build angle labels for the 2x2 grid grid_size = 2 # 2x2 angle_labels = [] for angle in angles[:grid_size * grid_size]: label = ANGLE_PROMPTS.get(angle, f"{angle} view") angle_labels.append(f"{angle.upper().replace('_', ' ')}: {label}") grid_layout = "\n".join( f" Cell {i + 1}: {label}" for i, label in enumerate(angle_labels) ) prompt = ( f"CRITICAL DIRECTIVE: Generate a single image containing a 2x2 grid of photos.\n" f"REFERENCE IMAGE: The attached photo shows the EXACT character to reproduce. " f"Maintain identical face, body type, clothing, hair, and all distinguishing features.\n\n" f"SUBJECT CODENAME: {codename}\n" f"GRID STRUCTURE: 2 columns by 2 rows. Each cell shows the SAME character " f"from a DIFFERENT camera angle. No borders, no frames, no sprocket holes.\n" f"CELL ASSIGNMENTS:\n{grid_layout}\n\n" f"PHOTOGRAPHIC ANCHORS:\n" f"- Medium: 35mm motion picture film still, shot on {get_constant('casting', 'casting_camera')}.\n" f"- Lighting: Soft, directional studio lighting. {get_constant('casting', 'casting_background')}.\n" f"- FULL BODY head-to-toe in every cell, including feet. Same wardrobe, same props, same person.\n" f"- Texture: {get_constant('casting', 'casting_texture_human_short')}.\n" f"- {get_constant('casting', 'casting_anti_airbrush')}" ) client = get_client(get_model("production_hq", "image")) from google.genai import types as genai_types # Load hero as identity reference hero_bytes = hero_path.read_bytes() mime = "image/jpeg" if hero_path.suffix.lower() in {".jpg", ".jpeg"} else "image/png" hero_part = genai_types.Part.from_bytes(data=hero_bytes, mime_type=mime) config = genai_types.GenerateContentConfig( temperature=0.3, # Lower than casting grid — want consistency not variety response_modalities=["IMAGE", "TEXT"], image_config=genai_types.ImageConfig( aspect_ratio="3:4", ), ) grid_path = output_dir / f"{char_id.lower()}_angle_grid.png" cost = 0.134 try: api_client = client._get_client() response = api_client.models.generate_content( model=get_model("production_hq", "image"), contents=[hero_part, prompt], config=config, ) if response and response.candidates: for candidate in response.candidates: if candidate.content and candidate.content.parts: for part in candidate.content.parts: if hasattr(part, "inline_data") and part.inline_data: grid_path.write_bytes(part.inline_data.data) logger.info("Angle grid saved: %s", grid_path) break except Exception as e: logger.error("Angle grid generation failed: %s", e) return {"grid_path": None, "panels": [], "angles": [], "cost": cost} if not grid_path.exists(): return {"grid_path": None, "panels": [], "angles": [], "cost": cost} # Split into panels (reuse grid splitter with 2x2) panels = _split_angle_grid(grid_path, output_dir, char_id, angles[:grid_size * grid_size]) return { "grid_path": str(grid_path), "panels": [str(p) for p in panels], "angles": angles[:grid_size * grid_size], "cost": cost, } def _find_grid_dividers(content_arr, axis, num_dividers=1): """Find dividing lines in a grid image by detecting low-variance bands. Args: content_arr: numpy array (grayscale) of the content area axis: 0 for horizontal dividers (row scan), 1 for vertical dividers (col scan) num_dividers: expected number of dividers (1 for 2x2 grid) Returns: list of divider center positions (pixel coordinates) """ import numpy as np h, w = content_arr.shape if axis == 0: # Scan rows line_stds = np.array([content_arr[r, :].std() for r in range(h)]) length = h else: # Scan columns line_stds = np.array([content_arr[:, c].std() for c in range(w)]) length = w # Search in the middle 40% (between 30%-70%) search_start = int(length * 0.3) search_end = int(length * 0.7) search_region = line_stds[search_start:search_end] if len(search_region) == 0: return [length // 2] # Smooth with 5-pixel moving average kernel_size = 5 if len(search_region) > kernel_size: kernel = np.ones(kernel_size) / kernel_size smoothed = np.convolve(search_region, kernel, mode='same') else: smoothed = search_region # Find the position of minimum std (most likely divider) divider_relative = int(np.argmin(smoothed)) divider_absolute = search_start + divider_relative return [divider_absolute] def _split_angle_grid( grid_path: Path, output_dir: Path, char_id: str, angles: list, ) -> list: """Split a 2x2 angle grid into individual panels named by angle. Uses gradient-based divider detection with 50/50 fallback. Panels are saved directly to the character ref dir as {char_id}_{angle}.png — the canonical location for identity refs. """ try: from PIL import Image import numpy as np except ImportError: logger.warning("PIL/numpy not available — cannot split grid into panels") return [] if not grid_path.exists(): return [] img = Image.open(grid_path) top, bottom, left, right = _detect_content_bounds(img) content = img.crop((left, top, right, bottom)) cw, ch = content.size content_arr = np.array(content.convert("L")) # Detect actual divider positions h_dividers = _find_grid_dividers(content_arr, axis=0, num_dividers=1) v_dividers = _find_grid_dividers(content_arr, axis=1, num_dividers=1) col_edges = [0] + v_dividers + [cw] row_edges = [0] + h_dividers + [ch] # Validate: each panel should be at least 30% of the dimension min_panel_frac = 0.30 col_valid = True for i in range(len(col_edges) - 1): panel_w = col_edges[i + 1] - col_edges[i] if panel_w < cw * min_panel_frac: logger.warning("Divider detection produced suspicious column split (%.0f%%) — falling back to 50/50", 100 * panel_w / cw) col_valid = False break row_valid = True for i in range(len(row_edges) - 1): panel_h = row_edges[i + 1] - row_edges[i] if panel_h < ch * min_panel_frac: logger.warning("Divider detection produced suspicious row split (%.0f%%) — falling back to 50/50", 100 * panel_h / ch) row_valid = False break grid_size = 2 if not col_valid: col_edges = [round(cw * i / grid_size) for i in range(grid_size + 1)] if not row_valid: row_edges = [round(ch * i / grid_size) for i in range(grid_size + 1)] panels = [] for idx, angle in enumerate(angles): row = idx // grid_size col = idx % grid_size box = (col_edges[col], row_edges[row], col_edges[col + 1], row_edges[row + 1]) panel = content.crop(box) panel_path = output_dir / f"{char_id.lower()}_{angle}.png" panel.save(panel_path) panels.append(panel_path) logger.info("Split angle grid into %d panels at %s (divider detection: col=%s, row=%s)", len(panels), output_dir, col_valid, row_valid) return panels def approve_angle_grid(char_id: str, remove_bg: bool = True, project: Optional[str] = None) -> dict: """Process approved angle grid: rembg → transparent (archival) → white (ref stack). Call this after the director approves the angle grid in the UI. Reads from {char_dir}/{char_id}_{angle}.png (canonical location), writes alpha to {char_dir}/{char_id}_{angle}_alpha.png, overwrites the original with the white-bg composite. """ import io char_dir = _char_refs_dir(project) / char_id.lower() if not char_dir.is_dir(): return {"error": "No character ref dir found. Generate angle grid first."} results = [] for panel_path in sorted(char_dir.glob(f"{char_id.lower()}_*.png")): angle = panel_path.stem.replace(f"{char_id.lower()}_", "") try: from PIL import Image as PILImage img_data = panel_path.read_bytes() if remove_bg: # rembg → transparent alpha alpha_data = _remove_background(img_data) alpha_path = char_dir / f"{char_id.lower()}_{angle}_alpha.png" alpha_path.write_bytes(alpha_data) # Composite onto white for ref stack img = PILImage.open(io.BytesIO(alpha_data)) if img.mode == "RGBA": white_bg = PILImage.new("RGB", img.size, (255, 255, 255)) white_bg.paste(img, mask=img.split()[3]) # Add 5% padding to protect extremities from PIL import ImageOps padding = max(int(white_bg.width * 0.05), 10) white_bg = ImageOps.expand(white_bg, border=padding, fill=(255, 255, 255)) ref_path = char_dir / f"{char_id.lower()}_{angle}.png" white_bg.save(ref_path) results.append({ "angle": angle, "alpha": str(alpha_path), "ref": str(ref_path), }) logger.info("Processed %s: alpha=%s, ref=%s", angle, alpha_path, ref_path) else: ref_path = char_dir / f"{char_id.lower()}_{angle}.png" img.save(ref_path) results.append({"angle": angle, "ref": str(ref_path)}) else: ref_path = char_dir / f"{char_id.lower()}_{angle}.png" panel_path.rename(ref_path) results.append({"angle": angle, "ref": str(ref_path)}) except Exception as e: logger.error("Failed to process angle %s: %s", angle, e) results.append({"angle": angle, "error": str(e)}) return {"character": char_id, "processed": results} def _remove_background(image_data: bytes) -> bytes: """Remove background using rembg. Returns RGBA PNG bytes.""" try: from rembg import remove return remove(image_data) except ImportError: logger.warning("rembg not installed — skipping background removal. pip install rembg") return image_data except Exception as e: logger.warning("rembg failed: %s — returning original image", e) return image_data def _generate_angles_qwen_ma( hero_url: str, angles: list[dict], output_dir: Path, ) -> list[Path]: """Generate multi-angle views via QwenMA LoRA on fal.ai. Each angle dict has: name, horizontal_angle (0-360), vertical_angle (-30 to 30), zoom (0-10). API field is 'image_urls' (list), NOT 'image_url' (singular). Result is in result['image']['url'] or result['images'][0]['url']. Cost: ~$0.035/image. """ import fal_client import urllib.request endpoint = "fal-ai/qwen-image-edit-2511-multiple-angles" saved_paths = [] for angle in angles: name = angle["name"] logger.info( "QwenMA: generating %s (h=%d, v=%d, zoom=%d)", name, angle["horizontal_angle"], angle["vertical_angle"], angle["zoom"], ) try: result = fal_client.subscribe( endpoint, arguments={ "image_urls": [hero_url], "horizontal_angle": angle["horizontal_angle"], "vertical_angle": angle["vertical_angle"], "zoom": angle["zoom"], "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, ) except Exception as e: logger.error("QwenMA generation failed for %s: %s", name, e) continue # Extract image URL from response — handle both response shapes img_url = None if isinstance(result, dict): if result.get("images") and len(result["images"]) > 0: img_url = result["images"][0].get("url") elif result.get("image") and isinstance(result["image"], dict): img_url = result["image"].get("url") if not img_url: logger.warning("QwenMA: no image URL in response for %s", name) continue # Download the generated image out_path = output_dir / f"{name}.png" try: req = urllib.request.Request(img_url) with urllib.request.urlopen(req) as resp: out_path.write_bytes(resp.read()) logger.info("QwenMA: saved %s → %s", name, out_path) saved_paths.append(out_path) except Exception as e: logger.error("QwenMA: download failed for %s: %s", name, e) continue # Write sidecar provenance JSON sidecar_path = out_path.with_suffix(".json") sidecar = { "model": "qwen-ma", "endpoint": endpoint, "source_hero": hero_url, "angle_params": { "horizontal_angle": angle["horizontal_angle"], "vertical_angle": angle["vertical_angle"], "zoom": angle["zoom"], }, "lora_scale": 0.9, "guidance_scale": 4.5, "num_inference_steps": 28, "cost_estimate": 0.035, } sidecar_path.write_text(json.dumps(sidecar, indent=2)) return saved_paths def _run_gate_0(char_id: str, char_dir: Path, angles_generated: list) -> dict: """Run Gate 0 validation on the completed ref set. Gate 0 checks internal consistency of the reference set: - All angles present - Visual identity consistent across angles - No artifacts or background contamination """ ref_paths = [] for angle in angles_generated: p = char_dir / f"{char_id.lower()}_{angle}.png" if p.exists(): ref_paths.append(p) if len(ref_paths) < 2: return { "passed": False, "reason": f"Only {len(ref_paths)} refs generated (need at least 2)", "cost": 0.0, } # Gate 0 uses Flash 3.1 for cost efficiency try: from recoil.pipeline._lib.validation import Validator validator = Validator() result = validator.run_gate_0(char_id, ref_paths) return { "passed": result.passed, "details": result.details, "cost": result.cost, } except ImportError: logger.info("Validation module not yet available — Gate 0 skipped") return { "passed": True, "reason": "Gate 0 skipped (validation module not loaded)", "cost": 0.0, } # ── CLI ────────────────────────────────────────────────────────────── def main(): import argparse logging.basicConfig( level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s", datefmt="%H:%M:%S", ) parser = argparse.ArgumentParser(description="Character reference angle generator") parser.add_argument("character", help="Character name/ID (e.g. jinx, KIAN)") parser.add_argument("--hero", type=Path, help="Path A: existing hero image") parser.add_argument("--description", type=str, help="Path B: text description for concepting") parser.add_argument("--project", type=str, help="Project name") parser.add_argument("--angles", nargs="+", default=None, help="Angles to generate") parser.add_argument("--gender", type=str, choices=["female", "male"], help="Explicit gender anchor (overrides _GENDER_MAP)") parser.add_argument("--backend", choices=["gemini", "qwen-ma"], default="gemini", help="Angle generation backend (default: gemini)") parser.add_argument("--no-rembg", action="store_true", help="Skip background removal") parser.add_argument("--dry-run", action="store_true", help="Preview without API calls") args = parser.parse_args() result = prep_character( character=args.character, hero_image=args.hero, description=args.description, project=args.project, angles=args.angles, remove_bg=not args.no_rembg, dry_run=args.dry_run, gender=args.gender, backend=args.backend, ) print(json.dumps(result, indent=2, default=str)) if __name__ == "__main__": main()