#!/usr/bin/env python3 """ recoil_bridge.py — Data access facade for Starsend. Plan-first pattern: checks Starsend extraction output (global_bible.json, ep_NNN_plan.json) first. Falls back to Recoil raw data if plans don't exist. Reference image loading always reads from Recoil. Never modifies any files. """ from pathlib import Path import json import logging from typing import Optional from recoil.core.paths import projects_root, DEFAULT_PROJECT, ProjectPaths logger = logging.getLogger("starsend.bridge") # ── Defaults (mirrors recoil/lib/config_loader.py) ────────────────────── from recoil.core.prompt_config import get_constant def _default_project_config(): """Return default project config with fresh values from prompt_config. Built as a function (not a module-level dict) so that reload() can pick up changed constants instead of serving stale boot-time values. """ return { "camera_body": get_constant("production", "camera_body"), "film_stock": get_constant("production", "film_stock"), "film_style_suffix": get_constant("production", "film_style_suffix"), "quality_guard": get_constant("production", "quality_guard"), "hex_object_map": {}, "candidate_lenses": { "face": "85mm f/1.8 prime", "body": "35mm f/2.8 prime", "close_up": "85mm f/1.4 prime", }, } # ── Helpers ────────────────────────────────────────────────────────────── def _project_dir(project: str) -> Path: """Resolve a project name to its directory under projects_root().""" return projects_root() / project def _read_json(path: Path) -> dict: """Read and parse a JSON file. Raises FileNotFoundError or ValueError.""" if not path.exists(): raise FileNotFoundError(f"File not found: {path}") try: return json.loads(path.read_text(encoding="utf-8")) except json.JSONDecodeError as e: raise ValueError(f"Invalid JSON in {path}: {e}") # ── Visual Pipeline Paths ─────────────────────────────────────────────── def get_bible_path(project: str = None) -> Path: """Resolve per-project bible path under state/visual/.""" proj = project or DEFAULT_PROJECT return ProjectPaths.for_project(proj).global_bible_path def _plan_path(episode: int, project: str = None) -> Path: """Path to an episode plan under projects/{project}/state/visual/plans/.""" filename = f"ep_{episode:03d}_plan.json" proj = project or DEFAULT_PROJECT return ProjectPaths.for_project(proj).plans_dir / filename def _episode_in_range(episode: int, ep_range: list) -> bool: """Check if an episode falls within an inclusive [start, end] range. Breakdown wardrobe/hair phases store episodes as [start, end], meaning episodes start through end inclusive. """ if not ep_range or len(ep_range) != 2: return False return ep_range[0] <= episode <= ep_range[1] def _normalize_plan_shots(data: dict) -> dict: """Add flat fields the orchestrator expects to plan-format shots. Plan shots use nested structures (asset_data.characters, prompt_data. prompt_skeleton.emotion_line, shot_id) but the orchestrator pipeline reads flat keys (characters_in_shot, emotion, id). This copies the nested values to flat keys so both formats work with the same pipeline. Recoil storyboard shots already have these flat fields, so this is a no-op for them. """ for shot in data.get("shots", []): # characters_in_shot: list of char_id strings if "characters_in_shot" not in shot: chars = shot.get("asset_data", {}).get("characters", []) shot["characters_in_shot"] = [ c["char_id"] for c in chars if "char_id" in c ] # emotion: flat string from prompt skeleton if "emotion" not in shot: shot["emotion"] = ( shot.get("prompt_data", {}) .get("prompt_skeleton", {}) .get("emotion_line", "") ) # id: copy shot_id so shot.get("id") works if "id" not in shot: shot["id"] = shot.get("shot_id", 0) return data # ── Public API ─────────────────────────────────────────────────────────── def load_storyboard(episode: int, project: str = None) -> dict: """Load shot data for an episode. Plan-first: returns Starsend episode plan if it exists. Falls back to Recoil storyboard JSON. The returned dict always has a "shots" key. When loaded from a plan, shots are ShotRecord dicts (with routing_data, prompt_data, etc.). When loaded from Recoil, shots are legacy storyboard dicts. Also sets "_source" key: "plan" or "recoil". Raises: FileNotFoundError: If neither plan nor storyboard exists. ValueError: If JSON is malformed. """ # Resolve project for path lookups if project is None: project = DEFAULT_PROJECT # Try Starsend plan first (project-aware path) mp = _plan_path(episode, project) if mp.exists(): data = _read_json(mp) data["_source"] = "plan" _normalize_plan_shots(data) logger.debug("Loaded plan for EP%03d from %s", episode, mp) return data # Fall back to Recoil storyboard filename = f"storyboard_ep_{episode:03d}.json" path = _project_dir(project) / "storyboards" / filename data = _read_json(path) data["_source"] = "recoil" logger.debug("Loaded Recoil storyboard for EP%03d (no plan)", episode) return data def load_breakdown(project: str = None) -> dict: """Load character/location/prop data. Plan-first: returns Starsend global bible if it exists. Falls back to Recoil breakdown.json. Sets "_source" key: "bible" or "recoil". Raises: FileNotFoundError: If neither bible nor breakdown exists. ValueError: If JSON is malformed. """ # Resolve project for path lookups if project is None: project = DEFAULT_PROJECT # Read project-specific bible from state/visual/. proj_bible = ProjectPaths.for_project(project).global_bible_path if proj_bible.exists(): data = _read_json(proj_bible) data["_source"] = "bible" logger.debug("Loaded global bible from %s", proj_bible) return data # Fall back to Recoil breakdown path = _project_dir(project) / "visual" / "breakdown.json" data = _read_json(path) data["_source"] = "recoil" logger.debug("Loaded Recoil breakdown (no bible)") return data def load_project_config(project: str = None) -> dict: """Load project_config.json merged with defaults from config_loader.py pattern. Path: projects/{project}/project_config.json Merges with defaults so that camera_body, film_stock, quality_guard, candidate_lenses, etc. are always present even if the file is partial. Dict values are deep-merged (one level); scalar values are overridden. Returns: dict with all default keys guaranteed present, overridden by file values. Raises: FileNotFoundError: If the project directory does not exist. """ if project is None: project = DEFAULT_PROJECT # Build A Phase 3 (2026-05-09): drop the "visual/" shadow path. The # Project class (recoil/core/project.py:214-239) reads from # `project_config.json` at the project root. The shadow path # `visual/project_config.json` was never materialized for client_video # projects, causing this function to silently return engine defaults # for 3 production tools (generate_location_refs, build_upload_bundle, # backfill_storyboard). config_path = _project_dir(project) / "project_config.json" # Start with defaults (function call ensures fresh values after reload) merged = _default_project_config() if config_path.exists(): try: file_config = _read_json(config_path) except (ValueError, OSError): return merged # Shallow merge top-level keys; deep-merge dicts (one level). # Skip empty-string overrides — project_config files like tartarus and # starsend-test have "quality_guard": "" and "negative_prompt": "" which # would clobber engine defaults from prompt_constants.json. for key, value in file_config.items(): if isinstance(value, dict) and isinstance(merged.get(key), dict): merged[key] = {**merged[key], **value} elif value is not None and value != "": merged[key] = value # krea2-flora Phase 3 — enforce the look XOR cinema_mode binding rule at # config-load time (fail loud, not at first render). Optional fields; # absent → no-op, so existing configs are byte-identical to before. from recoil.pipeline._lib.look_loader import validate_project_binding validate_project_binding(merged) return merged def get_character_refs(character: str, project: str = None) -> list[Path]: """Thin wrapper around core.ref_resolver.get_element_refs.""" from recoil.core.ref_resolver import get_element_refs, MissingCanonicalRefsError if project is None: project = DEFAULT_PROJECT try: refs = get_element_refs(character.upper(), project, "characters") except MissingCanonicalRefsError: return [] result = [] if "hero" in refs: result.append(refs["hero"]) for angle in ("front", "three_quarter", "profile", "back"): p = refs.get(angle) if p and p not in result: result.append(p) return result def get_location_refs(location_key: str, project: str = None) -> list[Path]: """Thin wrapper around core.ref_resolver.get_element_refs.""" from recoil.core.ref_resolver import get_element_refs, MissingCanonicalRefsError if project is None: project = DEFAULT_PROJECT try: refs = get_element_refs(location_key, project, "locations") except MissingCanonicalRefsError: return [] result = [] if "hero" in refs: result.append(refs["hero"]) for v in ("wide", "medium", "closeup"): p = refs.get(v) if p and p not in result: result.append(p) return result def resolve_character_for_episode( character: str, episode: int, project: str = None ) -> dict: """Resolve character visual state for a specific episode. Returns dict with: visual_description: str — base visual description wardrobe_desc: str — wardrobe description for this episode's phase wardrobe_phase: str — wardrobe phase key (e.g. 'jinx_lower_deck_salvager') hair_makeup: str — hair/makeup description for this episode's phase hair_makeup_phase: str — hair/makeup phase key (e.g. 'baseline') height_cm: int|None — character height display_name: str — character display name distinguishing_marks: str — scars/injuries for this phase (bible only) Plan-first: uses global bible phases when available. Falls back to Recoil breakdown.json. Raises: FileNotFoundError: If neither bible nor breakdown exists. KeyError: If the character is not found. """ if project is None: project = DEFAULT_PROJECT bd = load_breakdown(project) source = bd.get("_source", "recoil") char_upper = character.upper() char_data = bd.get("characters", {}).get(char_upper) if char_data is None: raise KeyError( f"Character '{character}' not found. " f"Available: {', '.join(bd.get('characters', {}).keys())}" ) result = { "visual_description": char_data.get("visual_description", ""), "wardrobe_desc": "", "wardrobe_phase": "", "hair_makeup": "", "hair_makeup_phase": "", "height_cm": char_data.get("height_cm"), "display_name": char_data.get("display_name", character), "distinguishing_marks": "", } if source == "bible": # Bible uses "phases" list with start_ep/end_ep for phase in char_data.get("phases", []): if phase.get("start_ep", 0) <= episode <= phase.get("end_ep", float('inf')): result["wardrobe_desc"] = phase.get("wardrobe_description", "") result["wardrobe_phase"] = phase.get("phase_id", "") result["hair_makeup"] = phase.get("hair_makeup", "") result["hair_makeup_phase"] = phase.get("phase_id", "") result["distinguishing_marks"] = phase.get("distinguishing_marks", "") break else: # Recoil breakdown uses nested wardrobe/hair_makeup dicts with episodes: [start, end] wardrobe = char_data.get("wardrobe", {}) for phase_key, phase_data in wardrobe.items(): ep_range = phase_data.get("episodes", []) if _episode_in_range(episode, ep_range): result["wardrobe_desc"] = phase_data.get("description", "") result["wardrobe_phase"] = phase_key break hair_makeup = char_data.get("hair_makeup", {}) for phase_key, phase_data in hair_makeup.items(): ep_range = phase_data.get("episodes", []) if _episode_in_range(episode, ep_range): result["hair_makeup"] = phase_data.get("description", "") result["hair_makeup_phase"] = phase_key break # Try to get height from the Recoil storyboard try: sb = load_storyboard(episode, project) sb_char = sb.get("characters", {}).get(character.lower(), {}) if "height_cm" in sb_char: result["height_cm"] = sb_char["height_cm"] except (FileNotFoundError, ValueError): pass return result def get_all_scenes(storyboard: dict) -> list[list[dict]]: """Split all shots into scenes. Returns list of scene groups. For plans: groups by scene_index field (LLM-assigned narrative scene boundaries). Falls back to location_id if scene_index missing. For Recoil storyboards: uses scene_break_before field. The first shot always starts a scene. """ shots = storyboard.get("shots", []) if not shots: return [] source = storyboard.get("_source", "recoil") scenes: list[list[dict]] = [] current_scene: list[dict] = [] for i, shot in enumerate(shots): if i == 0: current_scene = [shot] elif source in ("plan", "manifest"): # Plan/manifest: scene break on scene_index change (narrative boundaries) prev_idx = shots[i - 1].get("scene_index", 0) cur_idx = shot.get("scene_index", 0) if cur_idx != prev_idx: scenes.append(current_scene) current_scene = [shot] else: current_scene.append(shot) else: # Recoil: explicit scene_break_before flag if shot.get("scene_break_before"): scenes.append(current_scene) current_scene = [shot] else: current_scene.append(shot) if current_scene: scenes.append(current_scene) return scenes def _get_location_id(shot: dict) -> str: """Extract location_id from either plan or Recoil shot format.""" # Plan format asset_data = shot.get("asset_data") if asset_data and isinstance(asset_data, dict): return asset_data.get("location_id", "") # Recoil format return shot.get("location", "") def get_shot_by_id(storyboard: dict, shot_id) -> Optional[dict]: """Get a single shot by ID. Returns None if not found. Handles both formats: - Plan: shot_id is str like "EP001_SH01", or int index (1-based) - Recoil: shot_id is int matching shot["id"] """ source = storyboard.get("_source", "recoil") for shot in storyboard.get("shots", []): if source == "plan": # Match by shot_id string or by shot index if shot.get("shot_id") == shot_id: return shot # Also allow integer lookup by extracting SH number if isinstance(shot_id, int): sid = shot.get("shot_id", "") # EP001_SH03 → 3 if "_SH" in sid: try: sh_num = int(sid.split("_SH")[1]) if sh_num == shot_id: return shot except (ValueError, IndexError): pass else: if shot.get("id") == shot_id: return shot return None # ── CLI Demo ───────────────────────────────────────────────────────────── if __name__ == "__main__": print("=== Recoil Bridge — EP001 Stats ===\n") # Load storyboard try: sb = load_storyboard(1) except FileNotFoundError as e: print(f"ERROR: {e}") raise SystemExit(1) shots = sb.get("shots", []) print(f"Episode: {sb.get('episode')} — \"{sb.get('title')}\"") print(f"Total shots: {len(shots)}") # Characters in this storyboard sb_chars = sb.get("characters", {}) print(f"Characters: {len(sb_chars)} ({', '.join(sb_chars.keys())})") # Scenes scenes = get_all_scenes(sb) print(f"Scenes: {len(scenes)}") for i, scene in enumerate(scenes): print(f" Scene {i}: {len(scene)} shots (IDs {scene[0]['id']}-{scene[-1]['id']})") # Location print(f"Location: {sb.get('location', 'N/A')[:80]}...") print() # Load breakdown try: bd = load_breakdown() except FileNotFoundError as e: print(f"ERROR: {e}") raise SystemExit(1) bd_chars = bd.get("characters", {}) bd_locs = bd.get("locations", {}) print("=== Breakdown Stats ===\n") print(f"Characters: {len(bd_chars)} ({', '.join(bd_chars.keys())})") print(f"Locations: {len(bd_locs)}") # Wardrobe phases for first character from recoil.core.paths import DEFAULT_PROJECT first_char = next(iter(bd_chars), None) if first_char: char_data = bd_chars[first_char] wardrobe = char_data.get("wardrobe", {}) print(f"\n{first_char} wardrobe phases ({len(wardrobe)}):") for phase_key, phase_data in wardrobe.items(): ep_range = phase_data.get("episodes", []) print(f" {phase_key}: episodes {ep_range[0]}-{ep_range[1]}") # Resolve first character for EP001 print() try: char_ep1 = resolve_character_for_episode(first_char.lower(), 1) print(f"{first_char} @ EP001:") print(f" Wardrobe phase: {char_ep1['wardrobe_phase']}") print(f" Hair/makeup: {char_ep1['hair_makeup_phase']}") print(f" Height: {char_ep1['height_cm']}cm") except (FileNotFoundError, KeyError) as e: print(f" ERROR: {e}") # Reference images print() char_refs = get_character_refs(first_char.lower()) print(f"{first_char} refs: {len(char_refs)} images") for ref in char_refs[:5]: print(f" {ref.name}") # Project config print() config = load_project_config() print("=== Project Config ===") print(f"Camera: {config.get('camera_body')}") print(f"Film stock: {config.get('film_stock')}") print(f"Budget cap: ${config.get('budget_cap_usd', 'N/A')}")