# ============================================================================== # PORTED FROM STARSEND: lib/asset_manager.py # DATE: 2026-03-29 # ============================================================================== """ asset_manager.py — Reference image loading, caching, mirroring, and expression library. Key principles: - Pristine identity refs (white-bg) are NEVER altered on disk - Mirroring for screen direction happens in-memory only - Grayscale expression library stored in assets/expressions/ - Caches loaded image bytes to avoid repeated disk reads """ import logging from dataclasses import dataclass from pathlib import Path from typing import Optional from io import BytesIO from recoil.core.paths import PIPELINE_ROOT, CONFIG_PATH, ProjectPaths # TODO: move expressions to recoil or projects/ logger = logging.getLogger(__name__) # Naming convention: keystones contain "_hero" (any case) in the filename # e.g. wren_hero.jpeg, torch_hero.png — all live in assets/char//base/ _KEYSTONE_MARKER = "_hero" def _is_keystone(stem: str) -> bool: """Case-insensitive keystone detection.""" return _KEYSTONE_MARKER in stem.lower() # Supported image extensions _IMAGE_EXTS = {".png", ".jpg", ".jpeg", ".webp"} @dataclass class ReferenceImage: """A reference image with metadata for the assembler.""" path: Path label: str weight: int = 5 # 1 (lowest attention) to 10 (highest, closest to prompt) is_mirrored: bool = False # Dynamically flip for screen direction enforcement ref_type: str = "identity" # "identity" | "prop" | "expression" | "scene" | "pose" | "keyframe" def load_bytes(self) -> bytes: """Load image bytes, applying mirror if needed. Never modifies the file on disk.""" from PIL import Image, ImageOps img = Image.open(self.path) if self.is_mirrored: img = ImageOps.mirror(img) buf = BytesIO() # Use original format if possible, fallback to JPEG fmt = "JPEG" if self.path.suffix.lower() in ('.jpg', '.jpeg') else "PNG" img.save(buf, format=fmt, quality=95) return buf.getvalue() @property def mime_type(self) -> str: if self.path.suffix.lower() in ('.jpg', '.jpeg'): return "image/jpeg" return "image/png" class AssetManager: """Manages all reference image assets for the visual pipeline. Handles loading, caching, and organizing reference images across four weight tiers that map to Gemini's recency-bias attention ordering: Scene (1-2) -> Pose (4-5) -> Expression (6-7) -> Prop (7-8) -> Identity (8-10) Images closest to the text prompt (highest weight) receive the most attention from the model. Identity refs are always last before the prompt. """ def __init__(self): self._cache: dict[str, bytes] = {} # path_str -> raw bytes self._expressions_dir = PIPELINE_ROOT / "assets" / "expressions" # TODO: move expressions to recoil or projects/ # ------------------------------------------------------------------ # Cache helpers # ------------------------------------------------------------------ def _cached_bytes(self, path: Path) -> bytes: """Return raw file bytes, using cache to avoid repeated disk reads.""" key = str(path) if key not in self._cache: self._cache[key] = path.read_bytes() return self._cache[key] def clear_cache(self) -> None: """Clear the in-memory byte cache.""" self._cache.clear() def cache_stats(self) -> dict[str, int]: """Return cache entry count and total bytes.""" total = sum(len(v) for v in self._cache.values()) return {"entries": len(self._cache), "total_bytes": total} # ------------------------------------------------------------------ # Identity references (weight 8-10) # ------------------------------------------------------------------ def get_identity_refs(self, character: str, ref_paths: list[Path], max_refs: int = 2, mirror: bool = False, clean: bool = True) -> list[ReferenceImage]: """Build identity reference list for a character. Identity refs get the HIGHEST weight (8-10) -- closest to prompt. Keystones go last (highest weight) as the strongest identity anchor. Keystones are identified by the '_Hero' marker in the filename (e.g. Jinx_Hero.jpeg). Everything else is a pick. ADR-P04 Stack A/B: clean=True (Stack A, default): Standard rembg-clean refs on white bg. clean=False (Stack B): Gray-bg refs with permanent prop geometry preserved. When clean=False, looks in stack_b/ subdirectory first, falls back to standard. Args: character: Character name for labeling. ref_paths: Paths to reference images (from recoil_bridge.get_character_refs). max_refs: Maximum refs to include (default 2 for two-character shots). mirror: Whether to flip images for screen direction. clean: If False, use Stack B refs (gray-bg, permanent props baked in). Returns: List of ReferenceImage sorted by weight ascending (picks 8-9, keystones 10). Picks come first, keystone last -- so the keystone is closest to the prompt. """ # Stack B: remap paths to stack_b/ subdirectory when available if not clean: remapped = [] for p in ref_paths: stack_b_path = p.parent / "stack_b" / p.name if stack_b_path.exists(): remapped.append(stack_b_path) else: remapped.append(p) # Fall back to standard ref ref_paths = remapped keystones: list[Path] = [] picks: list[Path] = [] for p in ref_paths: if not p.exists(): continue if p.suffix.lower() not in _IMAGE_EXTS: continue if _is_keystone(p.stem): keystones.append(p) else: picks.append(p) # Pre-warm cache for all valid paths for p in keystones + picks: self._cached_bytes(p) refs: list[ReferenceImage] = [] # Budget: reserve 1 slot for a keystone if we have one keystone_slots = 1 if keystones else 0 pick_slots = max_refs - keystone_slots # Select picks (up to pick_slots), prioritizing face-forward angles. # Front/three-quarter show the most identity signal; back shows none. _ANGLE_PRIORITY = {"front": 0, "three_quarter": 1, "profile": 2, "back": 3} selected_picks = sorted(picks, key=lambda p: ( min((_ANGLE_PRIORITY.get(tag, 1) for tag in _ANGLE_PRIORITY if tag in p.stem.lower()), default=1), p.name, ))[:pick_slots] # Assign weights to picks: 8 for first, 9 for second for i, p in enumerate(selected_picks): weight = 8 + i # 8, 9 refs.append(ReferenceImage( path=p, label=f"[CHARACTER IDENTITY: {character} - ref {i + 1}]", weight=min(weight, 9), # Cap pick weight at 9 is_mirrored=mirror, ref_type="identity", )) # Keystone always gets weight 10 (strongest identity anchor, last before prompt) if keystones: # Use the first keystone (typically only one per character) ks = keystones[0] refs.append(ReferenceImage( path=ks, label=f"[CHARACTER IDENTITY: {character} - KEYSTONE]", weight=10, is_mirrored=mirror, ref_type="identity", )) # Sort by weight ascending so keystone is last (closest to prompt) refs.sort(key=lambda r: r.weight) return refs # ------------------------------------------------------------------ # Scene/environment references (weight 1-2) # ------------------------------------------------------------------ def get_scene_ref(self, scene_ref_path: Path) -> ReferenceImage: """Build scene/environment reference (lowest weight -- furthest from prompt). Scene refs anchor environment consistency but should not compete with character identity for model attention. Args: scene_ref_path: Path to scene/environment reference image. Returns: ReferenceImage with weight 1-2. """ if scene_ref_path.exists(): self._cached_bytes(scene_ref_path) return ReferenceImage( path=scene_ref_path, label="[SCENE ENVIRONMENT REFERENCE]", weight=2, is_mirrored=False, ref_type="scene", ) # ------------------------------------------------------------------ # Pose/composition references (weight 4-5) # ------------------------------------------------------------------ def get_pose_ref(self, pose_path: Path) -> ReferenceImage: """Build pose/composition reference from Flash exploration or grid extraction. Pose refs anchor body positioning, camera angle, and composition. Medium-low weight so they guide framing without overriding identity. Args: pose_path: Path to pose reference image (extracted grid panel or Flash output). Returns: ReferenceImage with weight 5. """ if pose_path.exists(): self._cached_bytes(pose_path) return ReferenceImage( path=pose_path, label="[POSE AND COMPOSITION REFERENCE]", weight=5, is_mirrored=False, ref_type="pose", ) # ------------------------------------------------------------------ # Expression references (weight 6-7) # ------------------------------------------------------------------ # Canonical emotions in the universal expression library (ADR-C05) _LIBRARY_EMOTIONS = { "joy", "anger", "sorrow", "fear", "determination", "exhaustion", "surprise", "disgust", "neutral", } # Synonym map: descriptive words → library emotion keywords _EMOTION_SYNONYMS = { # fear family "tense": "fear", "dread": "fear", "foreboding": "fear", "anxiety": "fear", "anxious": "fear", "terrified": "fear", "panic": "fear", "horror": "fear", "nervous": "fear", "uneasy": "fear", "claustrophobic": "fear", # anger family "rage": "anger", "fury": "anger", "frustrated": "anger", "hostile": "anger", "aggressive": "anger", "defiant": "anger", "furious": "anger", # sorrow family "grief": "sorrow", "sadness": "sorrow", "mourning": "sorrow", "grim": "sorrow", "melancholy": "sorrow", "loss": "sorrow", "regret": "sorrow", "somber": "sorrow", # determination family "resolve": "determination", "focused": "determination", "deliberate": "determination", "clinical": "determination", "calculated": "determination", "steely": "determination", "decisive": "determination", "committed": "determination", "professional": "determination", # exhaustion family "weary": "exhaustion", "fatigue": "exhaustion", "drained": "exhaustion", "spent": "exhaustion", "depleted": "exhaustion", "tired": "exhaustion", # surprise family "shock": "surprise", "startled": "surprise", "stunned": "surprise", "revelation": "surprise", "reveal": "surprise", "disbelief": "surprise", # disgust family "revulsion": "disgust", "contempt": "disgust", "repulsion": "disgust", # joy family "happy": "joy", "elation": "joy", "relief": "joy", "triumph": "joy", "satisfaction": "joy", "hope": "joy", # neutral family "calm": "neutral", "stoic": "neutral", "blank": "neutral", "flat": "neutral", "measured": "neutral", # tension/suspense family (maps to fear — the closest physical expression) "tension": "fear", "suspense": "fear", "cliffhanger": "fear", "ominous": "fear", "mysterious": "fear", } # Words that push intensity down to "subtle" (restraint, control, internalized) _SUBTLE_MODIFIERS = { "controlled", "restrained", "subtle", "quiet", "internalized", "suppressed", "understated", "micro", "hint", "slight", "muted", } # Words that push intensity up to "extreme" (peak, overwhelming) _EXTREME_MODIFIERS = { "extreme", "peak", "overwhelming", "intense", "maximum", "raw", "uncontrolled", "explosive", "visceral", "primal", } def get_expression_ref( self, emotion: str, intensity: str = "active", mirror: bool = False, ) -> Optional[ReferenceImage]: """Get a grayscale expression reference by emotion and intensity. ADR-C05: Universal expression matrix uses {emotion}_{intensity}.png naming (e.g. anger_active.png, joy_extreme.png). Default intensity is "active" (middle row — good general-purpose expression). Accepts both atomic keywords ("anger") and descriptive phrases ("controlled fear, tactical assessment"). For phrases, scans for the first recognized library emotion keyword. Intensity modifiers in the phrase override the default (e.g. "controlled" → subtle). Args: emotion: Emotion keyword or descriptive phrase. intensity: Intensity level — 'subtle', 'active', or 'extreme'. mirror: Whether to flip for screen direction. Returns: ReferenceImage with weight 7, or None if the expression file doesn't exist (library may not be generated yet). """ if not self._expressions_dir.exists(): return None emotion_lower = emotion.lower().strip().replace(" ", "_") intensity_lower = intensity.lower().strip() # Try direct match first (atomic keyword like "anger") ref = self._try_expression_file(emotion_lower, intensity_lower, emotion, mirror) if ref: return ref # Parse phrase into words for scanning words = emotion.lower().replace(",", " ").replace("-", " ").replace("_", " ").split() words = [w.strip() for w in words if w.strip()] # Detect intensity modifiers from the phrase resolved_intensity = intensity_lower for word in words: if word in self._SUBTLE_MODIFIERS: resolved_intensity = "subtle" break if word in self._EXTREME_MODIFIERS: resolved_intensity = "extreme" break # Scan for library keywords and synonyms for word in words: # Direct library keyword if word in self._LIBRARY_EMOTIONS: ref = self._try_expression_file(word, resolved_intensity, emotion, mirror) if ref: return ref # Synonym lookup mapped = self._EMOTION_SYNONYMS.get(word) if mapped: ref = self._try_expression_file(mapped, resolved_intensity, emotion, mirror) if ref: return ref # Nothing matched — log warning so gaps are visible logger.warning( "Expression ref: no match for '%s' — add keywords to _EMOTION_SYNONYMS", emotion[:60], ) return None def _try_expression_file( self, emotion_key: str, intensity: str, original_emotion: str, mirror: bool, ) -> Optional[ReferenceImage]: """Try to load an expression file by exact emotion key + intensity.""" # Primary: {emotion}_{intensity}.png for ext in (".png", ".jpg", ".jpeg"): candidate = self._expressions_dir / f"{emotion_key}_{intensity}{ext}" if candidate.exists(): self._cached_bytes(candidate) return ReferenceImage( path=candidate, label=f"[FACIAL EXPRESSION TO MATCH: {original_emotion} ({intensity})]", weight=7, is_mirrored=mirror, ref_type="expression", ) # Fallback: {emotion}.png (legacy naming) for ext in (".png", ".jpg", ".jpeg"): candidate = self._expressions_dir / f"{emotion_key}{ext}" if candidate.exists(): self._cached_bytes(candidate) return ReferenceImage( path=candidate, label=f"[FACIAL EXPRESSION TO MATCH: {original_emotion}]", weight=7, is_mirrored=mirror, ref_type="expression", ) return None def list_expressions(self) -> dict: """List available expressions from the universal expression library. Returns: Dict mapping emotion → list of available intensities, e.g.: {"anger": ["subtle", "active", "extreme"], "joy": ["active"], ...} Returns empty dict if the directory doesn't exist or is empty. """ if not self._expressions_dir.exists() or not self._expressions_dir.is_dir(): return {} result: dict = {} for f in sorted(self._expressions_dir.iterdir()): if f.suffix.lower() not in _IMAGE_EXTS: continue stem = f.stem.lower() # Parse {emotion}_{intensity} naming if "_" in stem: parts = stem.rsplit("_", 1) emotion, intensity = parts[0], parts[1] if intensity in ("subtle", "active", "extreme"): result.setdefault(emotion, []).append(intensity) continue # Legacy: bare emotion name result.setdefault(stem, []) return result # ------------------------------------------------------------------ # Prop references (weight 7-8) — ADR-P01, ADR-M08 # ------------------------------------------------------------------ def get_prop_ref(self, prop_id: str, state: str = "default", max_refs: int = 3, project: Optional[str] = None) -> list[ReferenceImage]: """Build prop reference list from turnaround sheet slices. Props get weight 7-8 — between expression (6-7) and identity (8-10). Renders prop ONTO the character when ordering is respected. Reads from projects/{project}/assets/prop/{prop_id}/ looking for: {prop_id}_{state}_front.png, {prop_id}_{state}_side.png, {prop_id}_{state}_3q.png """ prop_dir = ProjectPaths.for_project(project).asset_subject_dir("prop", prop_id) refs = [] for i, view in enumerate(["front", "side", "3q"]): for ext in (".png", ".jpg", ".jpeg"): candidate = prop_dir / f"{prop_id}_{state}_{view}{ext}" if candidate.exists(): self._cached_bytes(candidate) refs.append(ReferenceImage( path=candidate, label=f"[PROP REFERENCE: {prop_id} - {view}]", weight=7 + (1 if i == 0 else 0), # front=8, side/3q=7 ref_type="prop", )) break if len(refs) >= max_refs: break refs.sort(key=lambda r: r.weight) return refs # ------------------------------------------------------------------ # Shot assembly (combine all ref types) # ------------------------------------------------------------------ def build_shot_refs(self, character_refs: list[ReferenceImage], scene_ref: Optional[ReferenceImage] = None, pose_ref: Optional[ReferenceImage] = None, expression_ref: Optional[ReferenceImage] = None, prop_refs: Optional[list[ReferenceImage]] = None, max_total: int = 7) -> list[ReferenceImage]: """Assemble the complete reference list for a shot, respecting weight ordering. Final order (by weight, ascending = lowest attention first): Scene (1-2) -> Pose (4-5) -> Expression (6-7) -> Prop (7-8) -> Identity (8-10) Trims to max_total if needed, cutting from lowest weight first. Identity refs are never trimmed -- they are the most critical. Args: character_refs: Identity references for all characters in the shot. scene_ref: Optional scene/environment reference. pose_ref: Optional pose/composition reference. expression_ref: Optional grayscale expression reference. prop_refs: Optional prop references (weight 7-8, between expression and identity). max_total: Maximum total references (default 7 per config). Returns: Weight-sorted list of ReferenceImage, trimmed to max_total. """ all_refs: list[ReferenceImage] = [] if scene_ref is not None: all_refs.append(scene_ref) if pose_ref is not None: all_refs.append(pose_ref) if expression_ref is not None: all_refs.append(expression_ref) if prop_refs: all_refs.extend(prop_refs) all_refs.extend(character_refs) # Sort by weight ascending (lowest attention first, identity last) all_refs.sort(key=lambda r: r.weight) # Trim from the front (lowest weight) to respect max_total # This preserves identity refs (highest weight) which are most critical if len(all_refs) > max_total: all_refs = all_refs[len(all_refs) - max_total:] return all_refs # ------------------------------------------------------------------ # Diagnostics # ------------------------------------------------------------------ def describe_shot_refs(self, refs: list[ReferenceImage]) -> str: """Return a human-readable summary of a shot's reference stack.""" lines = [f" w={r.weight:2d} {'[M]' if r.is_mirrored else ' '} {r.label} ({r.path.name})" for r in refs] return "\n".join(lines) __all__ = [ # Public symbols (Phase D — MF-3 + DEBT-9). "AssetManager", "ReferenceImage", ] # ====================================================================== # CLI diagnostics # ====================================================================== if __name__ == "__main__": import json import sys config_path = CONFIG_PATH if not config_path.exists(): print(f"Config not found: {config_path}") sys.exit(1) config = json.loads(config_path.read_text()) recoil_root = Path(config["recoil_root"]).expanduser() mgr = AssetManager() # --- Character refs inventory --- project = config.get("default_project", "leviathan") lora_root = recoil_root / project / "visual" / "lora_candidates" print("=" * 60) print("CHARACTER REFERENCE INVENTORY") print("=" * 60) if lora_root.exists(): for char_dir in sorted(lora_root.iterdir()): if not char_dir.is_dir() or char_dir.name.startswith((".", "_")): continue char_name = char_dir.name keystones_dir = char_dir / "keystones" picks_dir = char_dir / "picks" # Collect all ref paths all_paths: list[Path] = [] if keystones_dir.exists(): all_paths.extend( f for f in sorted(keystones_dir.iterdir()) if f.suffix.lower() in _IMAGE_EXTS ) if picks_dir.exists(): all_paths.extend( f for f in sorted(picks_dir.iterdir()) if f.suffix.lower() in _IMAGE_EXTS ) if not all_paths: print(f"\n {char_name}: (no reference images found)") continue # Build identity refs using the manager refs = mgr.get_identity_refs(char_name, all_paths, max_refs=3) keystones = [r for r in refs if r.weight == 10] picks = [r for r in refs if r.weight < 10] print(f"\n {char_name}:") print(f" Total images on disk: {len(all_paths)}") print(f" Keystones ({len(keystones)}):") for r in keystones: size_kb = r.path.stat().st_size / 1024 print(f" {r.path.name} ({size_kb:.0f} KB) weight={r.weight}") print(f" Picks ({len(picks)}):") for r in picks: size_kb = r.path.stat().st_size / 1024 print(f" {r.path.name} ({size_kb:.0f} KB) weight={r.weight}") # Show all available picks on disk (not just selected) if picks_dir.exists(): all_picks_on_disk = sorted( f.name for f in picks_dir.iterdir() if f.suffix.lower() in _IMAGE_EXTS and not _is_keystone(f.stem) ) print(f" All picks on disk ({len(all_picks_on_disk)}):") for name in all_picks_on_disk: print(f" {name}") else: print(f" (lora_candidates directory not found: {lora_root})") # --- Expression library --- print() print("=" * 60) print("EXPRESSION LIBRARY") print("=" * 60) expressions = mgr.list_expressions() if expressions: print(f" Directory: {mgr._expressions_dir}") total = sum(len(v) for v in expressions.values()) print(f" Available: {len(expressions)} emotions, {total} refs") for emotion, intensities in sorted(expressions.items()): if intensities: print(f" {emotion}: {', '.join(intensities)}") else: print(f" {emotion}") else: print(f" Directory: {mgr._expressions_dir}") print(" (empty -- run 'python -m tools.prep_expressions' to populate)") # --- Cache stats --- print() stats = mgr.cache_stats() print(f"Cache: {stats['entries']} entries, {stats['total_bytes'] / 1024:.0f} KB")