#!/usr/bin/env python3 """Rooftop Spatial Pipeline Plan — end-to-end 정보 흐름 dry-run (plan v1 APPROVED). 이미지 생성 0, Gemini 호출 0, DB write 0, production 0 수정. 입력: source grounding bible + spatial bg base plates + DB 14 L05 shots. 출력: 5 inference cards + camera_slot_plan + base_plate_plan + shot_background_payloads + HTML. Codex Q1-Q5 결정 반영. 기존 experiment_rooftop_spatial_bg.py classifier 재사용 (local adapter). """ from __future__ import annotations import argparse import html import json import shutil import sys import time import uuid from dataclasses import dataclass, field, asdict from datetime import datetime from pathlib import Path from typing import Optional _REPO_ROOT = Path(__file__).resolve().parents[2] _BACKEND_ROOT = _REPO_ROOT / "backend" _SCRIPTS_DIR = _BACKEND_ROOT / "scripts" for _p in (str(_BACKEND_ROOT), str(_SCRIPTS_DIR)): if _p not in sys.path: sys.path.insert(0, _p) def _load_backend_env() -> None: import os as _os env_path = _BACKEND_ROOT / ".env" if not env_path.exists(): return for raw in env_path.read_text(encoding="utf-8").splitlines(): line = raw.strip() if not line or line.startswith("#") or "=" not in line: continue k, v = line.split("=", 1) k = k.strip() v = v.strip() if (v.startswith('"') and v.endswith('"')) or (v.startswith("'") and v.endswith("'")): v = v[1:-1] if k and k not in _os.environ: _os.environ[k] = v _load_backend_env() from app.core.database import SessionLocal # noqa: E402 from app.models.project import SceneStill # noqa: E402 # Codex Q2: 기존 classifier 재사용 (local adapter — generation/client path X). import experiment_rooftop_spatial_bg as bg # noqa: E402 PROJECT_ID = "6cb862d9-590c-4dce-86e6-d10c2977db19" EPISODE_ID = "08ad2cd3-3e96-4d84-808f-869ee628473c" L05_SHORT_ID = "L05" DEFAULT_OUTPUT_DIR = Path("scripts_output/rooftop_spatial_pipeline_plan") DEFAULT_SOURCE_RUN = Path( "scripts_output/rooftop_source_grounding/codex_entry_sanity_gemini_ok" ) DEFAULT_SPATIAL_RUN = Path( "scripts_output/rooftop_spatial_bg_experiment/20260523_1956_d22e35" ) # Codex Q4: 9-grid enum. ZONE_GRID_9 = [ "foreground_left", "foreground_center", "foreground_right", "midground_left", "midground_center", "midground_right", "background_left", "background_center", "background_right", ] # Codex Q5: placeholder enum. PLACEHOLDER_KINDS = ["dotted_box", "translucent_shape", "color_box", "none"] SLOT_ROLES = ["shared_base", "shot_specific", "rejected_manual_review"] DESIGN_CONFIDENCE = ["direct", "strong_inference", "weak_inference", "unknown"] HTML_QUOTE_LIMIT = 140 # entity kind 매핑 (visible_entities 의 short_id prefix 기반) ENTITY_KIND_PREFIX = {"C": "character", "P": "prop", "O": "owned", "L": "location"} # --------------------------------------------------------------------------- # Data classes # --------------------------------------------------------------------------- @dataclass class IdentityCard: name: str tone: dict scale_density: dict condition_age: dict evidence_quotes: list[str] confidence_majority: str @dataclass class TopologyNode: id: str label: str kind: str # living_kitchen|bedroom|bathroom|entry|outer confidence: str evidence: list[str] @dataclass class TopologyEdge: from_node: str to_node: str kind: str # door|opening|window_view|stairs confidence: str # direct|inferred|unknown evidence: str @dataclass class TopologyCard: nodes: list[TopologyNode] edges: list[TopologyEdge] unknowns: list[str] @dataclass class DesignInferenceItem: item_id: str category: str # door|window|fixture|furniture|finish|lighting|prop description: str source_evidence_ids: list[str] confidence: str derived_from: str # Codex Blocking 1: plate scope. 빈 list = global (모든 plate 에 적용, identity tone 류). applies_to_nodes: list[str] = field(default_factory=list) @dataclass class ForbiddenDriftItem: cue: str reason: str evidence: str @dataclass class StateOverlay: state: str environmental_marks: list[str] evidence: str confidence: str @dataclass class CameraSlot: slot_id: str used_by_shots: list[str] base_space_node: str camera_anchor: str looking_toward: str visible_nodes: list[str] offscreen_nodes: list[str] preserve_constraints: list[str] reason: str evidence: str confidence: str slot_role: str # Codex 추가: shared_base | shot_specific | rejected_manual_review # Codex Blocking 2: bedroom 등에서 어느 방인지 불분명한 case. multi_node_unresolved: bool = False unresolved_candidates: list[str] = field(default_factory=list) @dataclass class BasePlateSpec: base_plate_id: str covers_nodes: list[str] camera_slot_ids: list[str] must_show: list[str] must_not_show: list[str] shared_consistency_constraints: list[str] overlap_with_other_base_plates: list[str] generation_prompt_payload_draft: str @dataclass class EntityLayoutMarker: entity_id: str entity_name: str kind: str # character | prop | owned | location approximate_zone: str # ZONE_GRID_9 scale_hint: str contact_surface: str # floor|bed|table|wall|none pose_hint: str render_as_placeholder: str # PLACEHOLDER_KINDS @dataclass class ShotBackgroundPayload: shot_id: str camera_slot_id: str base_plate_id: str state_overlay: str visible_entity_short_ids: list[str] entity_layout_markers: list[EntityLayoutMarker] i2i_background_prompt_payload: dict # --------------------------------------------------------------------------- # Loaders # --------------------------------------------------------------------------- def load_bible(source_run: Path) -> dict: path = source_run / "gemini_rooftop_bible.json" if not path.exists(): raise SystemExit(f"ERROR: bible 부재 — {path}") return json.loads(path.read_text(encoding="utf-8")) def load_source_evidence(source_run: Path) -> list[dict]: """Codex Important 1: source_evidence.tsv 로드 + stable id 부여.""" path = source_run / "source_evidence.tsv" if not path.exists(): return [] lines = path.read_text(encoding="utf-8").splitlines() if not lines: return [] header = lines[0].split("\t") out: list[dict] = [] for raw in lines[1:]: cols = raw.split("\t") if len(cols) < len(header): continue row = dict(zip(header, cols)) # stable id: source + char_offset. row["ev_id"] = f"ev_{row.get('source', '?')}_{row.get('char_offset', '?')}" out.append(row) return out def load_existing_base_plates(spatial_run: Path) -> dict: """{plate_stem: {png_path, lvm_card_or_none}}""" out: dict = {} plate_dir = spatial_run / "base_plates" card_dir = spatial_run / "realized_spatial_cards" if not plate_dir.exists(): return out for png in sorted(plate_dir.glob("*.png")): stem = png.stem plate_id = f"base__{stem}" card_path = card_dir / f"{plate_id}.json" card = None if card_path.exists(): try: card = json.loads(card_path.read_text(encoding="utf-8")) except json.JSONDecodeError: card = None out[stem] = {"png_path": png, "lvm_card": card, "plate_id": plate_id} return out def load_l05_shots(session) -> list[bg.ShotMeta]: return bg.load_l05_shots(session) # --------------------------------------------------------------------------- # Stage A — 5 inference cards # --------------------------------------------------------------------------- def build_identity_card(bible: dict) -> IdentityCard: bands = [] for k in ("condition_age", "socioeconomic_tone", "scale_density"): b = bible.get(k, {}) if isinstance(b, dict): bands.append(b.get("confidence_band", "unknown")) # Codex Important 2: Counter mode, tie 시 weakest. from collections import Counter # noqa: PLC0415 if not bands: majority = "unknown" else: counter = Counter(bands) max_count = max(counter.values()) top_bands = [b for b, c in counter.items() if c == max_count] rank = {"trusted": 3, "weak": 2, "unknown": 1} # tie → weakest rank pick. majority = sorted(top_bands, key=lambda b: rank.get(b, 0))[0] evidence: list[str] = [] for k in ("socioeconomic_tone", "scale_density", "condition_age"): b = bible.get(k, {}) if isinstance(b, dict) and b.get("evidence"): evidence.append(b["evidence"]) return IdentityCard( name="L05 옥탑방 내부", tone=bible.get("socioeconomic_tone", {}), scale_density=bible.get("scale_density", {}), condition_age=bible.get("condition_age", {}), evidence_quotes=evidence, confidence_majority=majority, ) def _kind_for_subspace(name: str) -> str: """bible sub_space name → topology kind.""" n = name.lower() if "욕실" in name or "bathroom" in n: return "bathroom" if "주방" in name or "kitchen" in n: return "kitchen" # 거실 겸 주방인 경우 별도 식별 if "거실" in name or "living" in n: return "living_kitchen" if "방" in name or "bedroom" in n or "안방" in name: return "bedroom" if "현관" in name or "entry" in n: return "entry" return "other" def _normalize_node_id(name: str) -> str: """sub_space name → stable node id.""" return name.strip().replace(" ", "_").replace("(", "").replace(")", "") def build_topology_card(bible: dict) -> TopologyCard: nodes: list[TopologyNode] = [] seen: set[str] = set() for sp in bible.get("sub_spaces", []): if not isinstance(sp, dict): continue nid = _normalize_node_id(sp.get("name", "")) if not nid or nid in seen: continue seen.add(nid) nodes.append(TopologyNode( id=nid, label=sp.get("name", ""), kind=_kind_for_subspace(sp.get("name", "")), confidence=sp.get("confidence_band", "unknown"), evidence=list(sp.get("evidence_quotes", []))[:3], )) # 외부 노드 추가 — 현관 도어 너머 / 옥상. if "현관" not in seen: nodes.append(TopologyNode( id="현관", label="현관", kind="entry", confidence="inferred", evidence=["bible.doors_windows 의 현관 철문 from layout_relations"], )) nodes.append(TopologyNode( id="rooftop_outer", label="옥상/옥탑 외부", kind="outer", confidence="inferred", evidence=["entity_canon L05 = 옥탑방 내부 — 외부는 별도 entity (L04 외부+옥상 마당)"], )) edges: list[TopologyEdge] = [] # bible.layout_relations 변환 for r in bible.get("layout_relations", []): if not isinstance(r, dict): continue edges.append(TopologyEdge( from_node=_normalize_node_id(r.get("from_sub_space", "")), to_node=_normalize_node_id(r.get("to_sub_space", "")), kind="door" if "문" in r.get("relation", "") else "opening", confidence=r.get("confidence_band", "unknown"), evidence=r.get("evidence", ""), )) # 외부 edge: 현관 → rooftop_outer edges.append(TopologyEdge( from_node="현관", to_node="rooftop_outer", kind="door", confidence="inferred", evidence="현관 철문은 옥상 외부와 거실 사이 경계 (bible.doors_windows)", )) unknowns = list(bible.get("unknowns", [])) return TopologyCard(nodes=nodes, edges=edges, unknowns=unknowns) def _node_id_for_text(text: str) -> list[str]: """Codex Blocking 1: text 안 sub_space 키워드 → topology node id 매칭. 빈 list 반환 = global (모든 plate 에 적용). """ nodes: list[str] = [] if "거실" in text or "식탁" in text or "TV" in text or "텔레비전" in text: nodes.append("거실") if "수리영" in text: nodes.append("수리영의_방") if "민숙" in text or "안방" in text or "엄마" in text: nodes.append("민숙의_방_안방") if "주방" in text or "싱크대" in text or "냄비" in text: nodes.append("주방_영역") if "욕실" in text or "거울" in text or "화장실" in text: nodes.append("욕실") if "현관" in text or "철문" in text: nodes.append("현관") return sorted(set(nodes)) def _find_evidence_ids(text: str, source_evidence: list[dict], limit: int = 5) -> list[str]: """Codex Important 1: source_evidence 와 cue text 의 keyword 교집합 → evidence id list.""" matches: list[str] = [] for ev in source_evidence: kw = ev.get("keyword", "") if kw and kw in text: ev_id = ev.get("ev_id") or f"ev_{ev['source']}_{ev['char_offset']}" matches.append(ev_id) if len(matches) >= limit: break return matches def build_design_inference_card(bible: dict, source_evidence: Optional[list[dict]] = None ) -> list[DesignInferenceItem]: items: list[DesignInferenceItem] = [] source_evidence = source_evidence or [] seq = 1 def _emit(category: str, description: str, applies_nodes: list[str], confidence: str, derived_from: str) -> None: nonlocal seq items.append(DesignInferenceItem( item_id=f"di_{seq:03d}", category=category, description=description, source_evidence_ids=_find_evidence_ids(description, source_evidence), confidence=confidence, derived_from=derived_from, applies_to_nodes=applies_nodes, )) seq += 1 # 1. required_visual_cues for cue in bible.get("required_visual_cues", []): cat = "fixture" if "창" in cue: cat = "window" elif "문" in cue or "철문" in cue: cat = "door" elif "벽면" in cue or "거울" in cue: cat = "fixture" elif "침대" in cue or "커튼" in cue: cat = "furniture" elif "싱크대" in cue or "냄비" in cue: cat = "fixture" _emit(category=cat, description=cue, applies_nodes=_node_id_for_text(cue), confidence="direct", derived_from="bible.required_visual_cues") # 2. furniture — sub_space 필드 직접 활용. for f in bible.get("furniture", []): if not isinstance(f, dict): continue sub = f.get("sub_space", "") nodes = _node_id_for_text(sub) or _node_id_for_text(f.get("name", "")) _emit(category="furniture", description=f"{f.get('name', '')} ({sub})", applies_nodes=nodes, confidence=f.get("confidence_band", "unknown"), derived_from="bible.furniture") # 3. materials — where 필드. for m in bible.get("materials", []): if not isinstance(m, dict): continue where = m.get("where", "") nodes = _node_id_for_text(where) _emit(category="finish", description=f"{where} — {m.get('material', '')}", applies_nodes=nodes, confidence=m.get("confidence_band", "unknown"), derived_from="bible.materials") # 4. doors_windows — location 필드. for d in bible.get("doors_windows", []): if not isinstance(d, dict): continue cat = "door" if "문" in d.get("kind", "") else "window" loc = d.get("location", "") nodes = _node_id_for_text(loc) or _node_id_for_text(d.get("kind", "")) _emit(category=cat, description=f"{d.get('kind', '')} @ {loc}", applies_nodes=nodes, confidence=d.get("confidence_band", "unknown"), derived_from="bible.doors_windows") return items def build_forbidden_drift_card(bible: dict) -> list[ForbiddenDriftItem]: items: list[ForbiddenDriftItem] = [] socio_evidence = (bible.get("socioeconomic_tone") or {}).get("evidence", "") scale_evidence = (bible.get("scale_density") or {}).get("evidence", "") base_reason_evidence = socio_evidence or scale_evidence or "" for cue in bible.get("forbidden_luxury_cues", []): items.append(ForbiddenDriftItem( cue=cue, reason="source = 서민적·소박, 밀집된 좁고 한정된 옥탑방. 럭셔리 요소 충돌.", evidence=base_reason_evidence, )) # 추가 deterministic luxury list (Phase D 와 동일) for extra in ("marble", "chandelier", "walk-in closet", "luxury bathroom", "high ceiling", "panoramic view"): items.append(ForbiddenDriftItem( cue=extra, reason="deterministic luxury cue list — 옥탑방 source 와 일반적으로 충돌", evidence=base_reason_evidence, )) return items def build_state_overlay_card(bible: dict) -> list[StateOverlay]: out: list[StateOverlay] = [] for s in bible.get("state_variants", []): if not isinstance(s, dict): continue state = s.get("state", "") marks: list[str] = [] if "corpse" in state.lower() or "시신" in state or "피" in state: marks = ["dark floor stain near body location", "disturbed bedding", "scattered objects — NO human figure (final scene phase 에서 추가)"] elif "cleaned" in state.lower() or "정돈" in state or "깨끗" in state: marks = ["furniture restored", "no debris", "no stains"] elif "vandal" in state.lower() or "어지럽" in state: marks = ["knocked-over furniture", "scattered papers", "wall red circular symbol", "NO people"] elif "normal" in state.lower(): marks = ["clean daytime baseline", "no incident traces"] out.append(StateOverlay( state=state, environmental_marks=marks, evidence=s.get("evidence", ""), confidence=s.get("confidence_band", "unknown"), )) return out # --------------------------------------------------------------------------- # Stage B — Camera slot planning (reuse classify_* from bg) # --------------------------------------------------------------------------- def _slot_key_for_plan(plan: bg.ShotPlan) -> tuple[str, str, str, bool]: """Codex 새 Blocking fix: build_camera_slots / build_shot_background_payloads 가 동일한 slot_id 를 생성하도록 단일 helper. returns (sub_space, camera_slot, bedroom_node, multi_unresolved). """ if plan.sub_space == "bedroom": resolved, _candidates = _classify_bedroom_node(plan) return (plan.sub_space, plan.camera_slot, resolved, not resolved) return (plan.sub_space, plan.camera_slot, "", False) def _slot_id_from_parts(sub_space: str, camera_slot: str, bedroom_node: str, multi_unresolved: bool) -> str: parts = [sub_space, camera_slot] if bedroom_node: parts.append(bedroom_node) elif multi_unresolved: parts.append("unresolved") return "slot_" + "__".join(parts) def _classify_bedroom_node(plan: bg.ShotPlan) -> tuple[str, list[str]]: """Codex Blocking 2: bedroom shot 을 수리영의_방 vs 민숙의_방_안방 으로 한 번 더 분류. returns (resolved_node_id, candidates_if_unresolved). resolved 면 candidates=[]. """ text = f"{plan.shot.shot_description}\n{plan.shot.scene_summary}" suriyeong = any(k in text for k in ("수리영", "수리영의 방")) minsook = any(k in text for k in ("민숙", "안방", "엄마")) if suriyeong and not minsook: return ("수리영의_방", []) if minsook and not suriyeong: return ("민숙의_방_안방", []) if suriyeong and minsook: return ("", ["수리영의_방", "민숙의_방_안방"]) # 둘 다 — unresolved return ("", ["수리영의_방", "민숙의_방_안방"]) # 결정 불가 def build_camera_slots( plans: list[bg.ShotPlan], topology: TopologyCard ) -> list[CameraSlot]: # cluster by (sub_space, camera_slot, bedroom 의 경우 분류된 node). cluster: dict[tuple[str, str, str], list[bg.ShotPlan]] = {} unresolved_map: dict[tuple[str, str, str], list[str]] = {} for p in plans: sub_space, camera_slot, bedroom_node, multi_unresolved = _slot_key_for_plan(p) key = (sub_space, camera_slot, bedroom_node) if multi_unresolved: _resolved, candidates = _classify_bedroom_node(p) unresolved_map[key] = candidates cluster.setdefault(key, []).append(p) def _node_for_subspace(sub_space: str, bedroom_node: str = "") -> str: if sub_space == "main_room": return "거실" if sub_space == "bedroom": return bedroom_node or "수리영의_방" if sub_space == "bathroom": return "욕실" return sub_space slots: list[CameraSlot] = [] for (sub_space, slot_label, bedroom_node), items in cluster.items(): used_shots = [it.shot.label for it in items] is_fallback = slot_label == "fallback_unknown" or sub_space == "manual_review_needed" multi_unresolved = (sub_space == "bedroom" and not bedroom_node) if is_fallback or multi_unresolved: slot_role = "rejected_manual_review" elif len(used_shots) >= 2: slot_role = "shared_base" else: slot_role = "shot_specific" slot_desc = next( (d for sp, lbl, d in bg.CAMERA_SLOT_POOL if sp == sub_space and lbl == slot_label), "", ) base_node = _node_for_subspace(sub_space, bedroom_node) visible_nodes: list[str] = [] offscreen_nodes: list[str] = [] if sub_space == "main_room": visible_nodes = ["거실"] offscreen_nodes = ["수리영의_방", "민숙의_방_안방", "욕실"] elif sub_space == "bedroom": if bedroom_node: visible_nodes = [bedroom_node] offscreen_nodes = [ n for n in ("수리영의_방", "민숙의_방_안방") if n != bedroom_node ] + ["거실", "욕실"] else: visible_nodes = [] # multi_unresolved offscreen_nodes = [] elif sub_space == "bathroom": visible_nodes = ["욕실"] offscreen_nodes = ["거실"] preserve: list[str] = [] if slot_label.endswith("eye_level_wide"): preserve = ["거실 전체 좌우 폭", "창문 위치"] elif "table_close" in slot_label: preserve = ["식탁 윗면", "주변 의자"] elif "doorway_wide" in slot_label and bedroom_node: preserve = ["침대 위치 (왼쪽 벽 가정)", "창문/커튼 위치"] elif "mirror_close" in slot_label: preserve = ["거울 정면", "벽면"] if multi_unresolved: reason = ( f"bedroom shot {used_shots} 가 수리영의_방 vs 민숙의_방_안방 분류 불가 — manual review" ) elif is_fallback: reason = f"manual review 필요 — sub_space={sub_space}, slot={slot_label}" else: reason = f"{len(used_shots)} shot 이 같은 (sub_space, framing, node={bedroom_node or 'n/a'}) 로 분류" slot_id = _slot_id_from_parts(sub_space, slot_label, bedroom_node, multi_unresolved) unresolved_candidates = ( unresolved_map.get((sub_space, slot_label, bedroom_node), []) if multi_unresolved else [] ) slots.append(CameraSlot( slot_id=slot_id, used_by_shots=used_shots, base_space_node=base_node, camera_anchor=slot_desc.split(",")[0] if slot_desc else "", looking_toward=", ".join(slot_desc.split(",")[1:]).strip() if slot_desc else "", visible_nodes=visible_nodes, offscreen_nodes=offscreen_nodes, preserve_constraints=preserve, reason=reason, evidence=f"bg.assign_camera_slot deterministic clustering of {used_shots}", confidence="high" if not (is_fallback or multi_unresolved) else "low", slot_role=slot_role, multi_node_unresolved=multi_unresolved, unresolved_candidates=unresolved_candidates, )) return slots # --------------------------------------------------------------------------- # Stage C — Base plate plan # --------------------------------------------------------------------------- def build_base_plate_plan( slots: list[CameraSlot], cards: dict, ) -> tuple[list[BasePlateSpec], dict, list[DesignInferenceItem]]: """Codex Blocking 1: must_show 를 plate covers_nodes 와 scope 매칭. returns (specs, summary, unmatched_design_items). - design_item.applies_to_nodes 가 비면 global → 모든 plate must_show 에. - applies_to_nodes 가 채워지면 plate.covers_nodes 와 교집합 있을 때만 포함. - 매칭 0 plate 인 항목 = unmatched bucket. """ specs: list[BasePlateSpec] = [] forbidden = [c.cue for c in cards["forbidden_drift"]] design_items: list[DesignInferenceItem] = cards["design_inference"] global_items = [d for d in design_items if not d.applies_to_nodes] scoped_items = [d for d in design_items if d.applies_to_nodes] eligible = [s for s in slots if s.slot_role != "rejected_manual_review"] matched_item_ids: set[str] = set() for s in eligible: consistency = list(s.preserve_constraints) plate_nodes = set(s.visible_nodes or [s.base_space_node]) # plate-scoped must_show: scoped item 중 plate_nodes 와 교집합 있는 것 + global. plate_scoped = [ d for d in scoped_items if any(n in plate_nodes for n in d.applies_to_nodes) ] for d in plate_scoped: matched_item_ids.add(d.item_id) must_show_descriptions = ( [d.description for d in plate_scoped] + [d.description for d in global_items] ) # 같은 base_space_node 의 다른 slot 들 overlap = [other.slot_id for other in eligible if other.slot_id != s.slot_id and other.base_space_node == s.base_space_node] bp_id = f"bp_{s.slot_id.removeprefix('slot_')}" prompt_draft = ( f"# BasePlate {bp_id} (draft — production schema 와 다를 수 있음)\n" f"# covers nodes: {', '.join(sorted(plate_nodes))}\n" f"# camera anchor: {s.camera_anchor}\n" f"# looking toward: {s.looking_toward}\n" f"# must show ({len(must_show_descriptions)}): {', '.join(must_show_descriptions[:8])}\n" f"# must NOT show: {', '.join(forbidden[:8])}\n" f"# preserve constraints: {', '.join(consistency)}\n" "# no people, no animals — empty space, daytime baseline." ) specs.append(BasePlateSpec( base_plate_id=bp_id, covers_nodes=sorted(plate_nodes), camera_slot_ids=[s.slot_id], must_show=must_show_descriptions[:12], must_not_show=forbidden[:8], shared_consistency_constraints=consistency, overlap_with_other_base_plates=overlap, generation_prompt_payload_draft=prompt_draft, )) # unmatched bucket — scoped item 중 어느 plate 의 covers_nodes 와도 매칭 안 됨. unmatched = [d for d in scoped_items if d.item_id not in matched_item_ids] summary = { "recommended_base_plate_count": len(specs), "current_base_plate_count": 4, "delta": len(specs) - 4, "rejected_slot_count": sum(1 for s in slots if s.slot_role == "rejected_manual_review"), "global_design_items_count": len(global_items), "unmatched_design_items_count": len(unmatched), "reason": ( f"camera slot clustering 결과 eligible {len(specs)}. " f"rejected (manual_review / multi_node_unresolved) 는 base plate 생성 비대상. " f"design items scoped {len(scoped_items)} + global {len(global_items)}, " f"unmatched {len(unmatched)}." ), } return specs, summary, unmatched # --------------------------------------------------------------------------- # Stage D — Shot background payload # --------------------------------------------------------------------------- def _entity_kind_from_short_id(short_id: str) -> str: if not short_id: return "unknown" return ENTITY_KIND_PREFIX.get(short_id[0], "unknown") def _placeholder_for_entity(kind: str, slot_label: str) -> str: # Codex Q5: 인체=translucent_shape (design reco), 소품=color_box, close-up 은 none. if "close" in slot_label or "mirror_close" in slot_label or "table_close" in slot_label: return "none" if kind == "character": return "translucent_shape" if kind in ("prop", "owned"): return "color_box" return "none" def _zone_for_entity(kind: str, contact_surface: str) -> str: # 단순 heuristic — character 는 midground_center, prop 은 contact surface 기반. if kind == "character": return "midground_center" if contact_surface == "table": return "midground_center" if contact_surface == "bed": return "midground_left" if contact_surface == "wall": return "background_center" if contact_surface == "floor": return "foreground_center" return "midground_center" def _contact_surface_from_shot(shot_desc: str, kind: str) -> str: s = shot_desc if kind == "character": if "주저앉" in s or "바닥" in s: return "floor" if "침대" in s: return "bed" if "식탁" in s: return "table" return "floor" if "식탁" in s: return "table" if "벽면" in s or "거울" in s: return "wall" if "침대" in s: return "bed" return "none" def _pose_hint_from_shot(shot_desc: str, kind: str) -> str: if kind != "character": return "" s = shot_desc if "주저앉" in s: return "collapsed sitting" if "감싸 안" in s or "다정한" in s: return "standing embrace" if "응시" in s or "바라보" in s: return "standing facing camera" if "메시지" in s or "스마트폰" in s or "귀에" in s: return "standing using phone" return "standing" def _scale_hint(kind: str) -> str: if kind == "character": return "adult human standing height" if kind == "prop": return "small hand-held or surface object" return "qualitative" def build_shot_background_payloads( plans: list[bg.ShotPlan], slots: list[CameraSlot], base_plates: list[BasePlateSpec], state_overlays: list[StateOverlay], ) -> list[ShotBackgroundPayload]: slot_index = {s.slot_id: s for s in slots} slot_to_bp = {bp.camera_slot_ids[0]: bp.base_plate_id for bp in base_plates} state_index = {s.state: s for s in state_overlays} out: list[ShotBackgroundPayload] = [] for p in plans: # Codex Blocking: build_camera_slots 와 동일한 helper 사용. sub_space, camera_slot, bedroom_node, multi_unresolved = _slot_key_for_plan(p) slot_id = _slot_id_from_parts(sub_space, camera_slot, bedroom_node, multi_unresolved) slot = slot_index.get(slot_id) bp_id = slot_to_bp.get(slot_id, "") markers: list[EntityLayoutMarker] = [] for sid in p.shot.visible_short_ids: kind = _entity_kind_from_short_id(sid) contact = _contact_surface_from_shot(p.shot.shot_description, kind) markers.append(EntityLayoutMarker( entity_id=sid, entity_name=sid, # name 은 visible_entities_json 의 entity_name 으로 가능하지만 short_id 만 보존 kind=kind, approximate_zone=_zone_for_entity(kind, contact), scale_hint=_scale_hint(kind), contact_surface=contact, pose_hint=_pose_hint_from_shot(p.shot.shot_description, kind), render_as_placeholder=_placeholder_for_entity(kind, p.camera_slot), )) state_match = state_index.get(p.state) payload = { "preserve_layout": ( slot.preserve_constraints if slot else [] ), "may_change_state_overlay": ( state_match.environmental_marks if state_match else [] ), "must_not_change": ( slot.preserve_constraints if slot else [] ), "remove_placeholder_instruction": ( "최종 background plate 에는 placeholder 박스/silhouette 등 design marker 포함 X — 실 rendering 단계에서 인물/소품 합성." ), } out.append(ShotBackgroundPayload( shot_id=p.shot.label, camera_slot_id=slot_id, base_plate_id=bp_id, state_overlay=p.state, visible_entity_short_ids=p.shot.visible_short_ids, entity_layout_markers=markers, i2i_background_prompt_payload=payload, )) return out # --------------------------------------------------------------------------- # Webserver (Phase D 패턴) # --------------------------------------------------------------------------- def _find_free_port(start: int, attempts: int = 6, bind: str = "127.0.0.1") -> Optional[int]: import socket as _socket for offset in range(attempts): port = start + offset s = _socket.socket(_socket.AF_INET, _socket.SOCK_STREAM) try: s.setsockopt(_socket.SOL_SOCKET, _socket.SO_REUSEADDR, 1) s.bind((bind, port)) s.close() return port except OSError: continue return None def start_static_webserver(run_dir: Path, port: int, bind: str = "127.0.0.1") -> dict: import subprocess free = _find_free_port(port, attempts=6, bind=bind) if free is None: return {"status": "no_free_port", "tried_from": port} log_path = run_dir / "_serve.log" log_fh = log_path.open("w", encoding="utf-8") proc = subprocess.Popen( # noqa: S603 [sys.executable, "-m", "http.server", str(free), "--bind", bind], cwd=str(run_dir), stdout=log_fh, stderr=log_fh, ) time.sleep(0.6) rc = proc.poll() if rc is not None: try: log_tail = log_path.read_text(encoding="utf-8", errors="replace")[-400:] except Exception: log_tail = "" return {"status": "died_early", "pid": proc.pid, "port": free, "bind": bind, "exit_code": rc, "log_tail": log_tail} return {"status": "started", "pid": proc.pid, "port": free, "bind": bind, "log": str(log_path)} # --------------------------------------------------------------------------- # Asset copy + HTML # --------------------------------------------------------------------------- def copy_existing_base_plates(run_dir: Path, spatial_run: Path) -> Optional[str]: if not spatial_run.exists(): return None dest = run_dir / "base_plate_assets" / spatial_run.name dest.mkdir(parents=True, exist_ok=True) src_plates = spatial_run / "base_plates" if src_plates.exists(): for png in src_plates.glob("*.png"): shutil.copy2(png, dest / png.name) src_cards = spatial_run / "realized_spatial_cards" if src_cards.exists(): for j in src_cards.glob("*.json"): shutil.copy2(j, dest / j.name) return str(dest.relative_to(run_dir)) def _short(s: str, limit: int = HTML_QUOTE_LIMIT) -> str: s = s or "" return s if len(s) <= limit else s[:limit] + "..." def render_html( run_dir: Path, *, bible: dict, cards: dict, slots: list[CameraSlot], base_plates: list[BasePlateSpec], base_summary: dict, payloads: list[ShotBackgroundPayload], existing_plates: dict, asset_rel: Optional[str], unmatched_design: Optional[list[DesignInferenceItem]] = None, run_meta: dict, ) -> None: # §1 source summary s1 = ( f"

sub_spaces: {len(bible.get('sub_spaces', []))}, " f"layout_relations: {len(bible.get('layout_relations', []))}, " f"doors_windows: {len(bible.get('doors_windows', []))}, " f"furniture: {len(bible.get('furniture', []))}, " f"forbidden_luxury_cues: {len(bible.get('forbidden_luxury_cues', []))}, " f"required_visual_cues: {len(bible.get('required_visual_cues', []))}

" f"

tone: {html.escape(_short((bible.get('socioeconomic_tone') or {}).get('summary', '')))}

" f"

scale: {html.escape(_short((bible.get('scale_density') or {}).get('summary', '')))}

" ) # §2 cards identity = cards["identity"] topo = cards["topology"] design = cards["design_inference"] forbidden = cards["forbidden_drift"] states = cards["state_overlay"] identity_html = ( f"

{html.escape(identity.name)} (confidence majority: " f"{html.escape(identity.confidence_majority)})

" f"" ) topo_nodes_html = "".join( f"{html.escape(n.id)}{html.escape(n.label)}" f"{html.escape(n.kind)}{html.escape(n.confidence)}" f"{html.escape(_short(' | '.join(n.evidence)))}" for n in topo.nodes ) topo_edges_html = "".join( f"{html.escape(e.from_node)}{html.escape(e.to_node)}" f"{html.escape(e.kind)}{html.escape(e.confidence)}" f"{html.escape(_short(e.evidence))}" for e in topo.edges ) design_html = "".join( f"{html.escape(d.item_id)}{html.escape(d.category)}" f"{html.escape(_short(d.description))}" f"{html.escape(', '.join(d.applies_to_nodes) or 'global')}" f"{html.escape(d.confidence)}" f"{len(d.source_evidence_ids)}" f"{html.escape(d.derived_from)}" for d in design ) forbidden_html = "".join( f"{html.escape(_short(f.cue))}{html.escape(_short(f.reason))}" f"{html.escape(_short(f.evidence))}" for f in forbidden ) state_html = "".join( f"{html.escape(s.state)}" f"{html.escape(', '.join(s.environmental_marks))}" f"{html.escape(s.confidence)}" f"{html.escape(_short(s.evidence))}" for s in states ) # §3 camera slot (Codex Blocking 2: multi_node_unresolved 컬럼 추가) slot_rows = "".join( f"{html.escape(s.slot_id)}" f"{html.escape(s.slot_role)}" f"{len(s.used_by_shots)}: {html.escape(', '.join(s.used_by_shots))}" f"{html.escape(s.base_space_node)}" f"{html.escape(_short(s.camera_anchor, 80))}" f"{html.escape(_short(s.looking_toward, 80))}" f"{html.escape(', '.join(s.visible_nodes))}" f"{html.escape(', '.join(s.offscreen_nodes))}" f"{html.escape(_short(' | '.join(s.preserve_constraints)))}" f"{html.escape(s.confidence)}" f"{'⚠ ' + ', '.join(s.unresolved_candidates) if s.multi_node_unresolved else ''}" "" for s in slots ) # §4 base plate plan + existing thumbnails bp_rows = "".join( f"{html.escape(bp.base_plate_id)}" f"{html.escape(', '.join(bp.covers_nodes))}" f"{html.escape(', '.join(bp.camera_slot_ids))}" f"{html.escape(_short(', '.join(bp.must_show)))}" f"{html.escape(_short(', '.join(bp.must_not_show)))}" f"{html.escape(_short(' | '.join(bp.shared_consistency_constraints)))}" f"{html.escape(', '.join(bp.overlap_with_other_base_plates) or '(none)')}" for bp in base_plates ) thumb_html = "" if asset_rel and existing_plates: cards_html = [] for stem, info in existing_plates.items(): rel = f"{asset_rel}/{stem}.png" lvm_summary = "" if info.get("lvm_card"): band = info["lvm_card"].get("overall_confidence_band", "?") fixed = [f["label"] for f in info["lvm_card"].get("fixed_objects", [])[:5]] lvm_summary = f"band={band} | fixed={', '.join(fixed)}" cards_html.append( "
" f'' f'{html.escape(stem)}' f'
' f'
{html.escape(stem)}
' f'
LVM: {html.escape(lvm_summary)}
' "
" ) thumb_html = "

기존 base plate (Phase D run 참고)

" + "".join(cards_html) # §5 shot payloads grouped by base_plate_id group: dict[str, list[ShotBackgroundPayload]] = {} for pl in payloads: group.setdefault(pl.base_plate_id or "(unassigned)", []).append(pl) group_blocks = [] for bp_id in sorted(group): items = group[bp_id] rows = "".join( f"{html.escape(it.shot_id)}" f"{html.escape(it.camera_slot_id)}" f"{html.escape(it.state_overlay)}" f"{html.escape(', '.join(it.visible_entity_short_ids))}" f"{len(it.entity_layout_markers)}" for it in items ) group_blocks.append( f"
{html.escape(bp_id)} ({len(items)} shots)" "" "" f"{rows}
shotcamera_slotstatevisible_entitiesmarker_count
" ) payloads_html = "".join(group_blocks) or "

(payloads 없음)

" # §6 entity layout marker table marker_rows = [] for pl in payloads: for m in pl.entity_layout_markers: marker_rows.append( f"{html.escape(pl.shot_id)}" f"{html.escape(m.entity_id)}" f"{html.escape(m.kind)}" f"{html.escape(m.approximate_zone)}" f"{html.escape(m.scale_hint)}" f"{html.escape(m.contact_surface)}" f"{html.escape(m.pose_hint)}" f"{html.escape(m.render_as_placeholder)}" ) marker_html = ( "" "" "" f"{''.join(marker_rows)}
shotentitykindzonescale_hintcontact_surfacepose_hintplaceholder
" ) # §7 open questions / unknowns / risks risks = [] risks.extend([f"bible.unknown: {u}" for u in bible.get("unknowns", [])[:5]]) risks.extend([f"bible.inference_limit: {l}" for l in bible.get("inference_limits", [])[:5]]) low_slots = [s.slot_id for s in slots if s.confidence == "low"] if low_slots: risks.append(f"low-confidence slots: {', '.join(low_slots)}") unresolved_slots = [s for s in slots if s.multi_node_unresolved] for s in unresolved_slots: risks.append( f"multi_node_unresolved slot {s.slot_id} (candidates: {', '.join(s.unresolved_candidates)}, " f"shots: {', '.join(s.used_by_shots)})" ) if unmatched_design: risks.append( f"unmatched design items: {len(unmatched_design)} (어느 plate 와도 scope 매칭 X — " "topology 누락 또는 sub_space 매핑 누락 가능성)" ) risks_html = "" if risks else "

(없음)

" serve_info = run_meta.get("serve_info") or {} serve_html = "" if serve_info.get("status") == "started": url = f"http://{serve_info['bind']}:{serve_info['port']}/index.html" serve_html = ( f"

webserver: {html.escape(url)} " f"(PID {serve_info['pid']})

" ) body = f""" Rooftop Spatial Pipeline Plan — {html.escape(run_meta['run_id'])}

Rooftop Spatial Pipeline Plan

run_id {html.escape(run_meta['run_id'])} · plan v1 · dry-run only

{serve_html}

1. Source grounding summary

{s1}

2. Inference cards (5)

2-A. set_identity

{identity_html}

2-B. set_topology — nodes ({len(topo.nodes)})

{topo_nodes_html}
idlabelkindconfidenceevidence

2-B. set_topology — edges ({len(topo.edges)})

{topo_edges_html}
fromtokindconfidenceevidence

2-C. production_design_inference ({len(design)})

{design_html}
item_idcategorydescription applies_to_nodesconfidenceev_idsderived_from

2-D. forbidden_drift ({len(forbidden)})

{forbidden_html}
cuereasonevidence

2-E. state_overlay ({len(states)})

{state_html}
stateenvironmental_marksconfidenceevidence

3. Camera slot plan ({len(slots)})

{slot_rows}
slot_idroleshotsbase_node anchorlooking_towardvisibleoffscreen preserveconfidenceunresolved

4. Base plate plan

recommended: {base_summary['recommended_base_plate_count']} vs current 4 (delta={base_summary['delta']:+d}) · rejected_slot_count: {base_summary.get('rejected_slot_count', 0)} · global_design_items: {base_summary.get('global_design_items_count', 0)} · unmatched_design_items: {base_summary.get('unmatched_design_items_count', 0)}

{html.escape(base_summary['reason'])}

{bp_rows}
base_plate_idcovers_nodescamera_slot_ids must_show ({len(base_plates) and len(base_plates[0].must_show) or 0}) must_not_showpreserveoverlap_with
{('

Unmatched design items (어느 plate covers_nodes 와도 매칭 X)

' '' + ''.join(f'' f'' f'' for d in (unmatched_design or [])) + '
item_idcategorydescriptionapplies_to_nodes
{html.escape(d.item_id)}{html.escape(d.category)}{html.escape(_short(d.description))}{html.escape(", ".join(d.applies_to_nodes))}
') if unmatched_design else ''} {thumb_html}

5. Shot background payloads ({len(payloads)})

{payloads_html}

6. Entity layout markers

{marker_html}

7. Open questions / unknowns / risks

{risks_html} """ (run_dir / "index.html").write_text(body, encoding="utf-8") # --------------------------------------------------------------------------- # Writers # --------------------------------------------------------------------------- def _to_dict(obj): if hasattr(obj, "__dataclass_fields__"): d = asdict(obj) return d return obj def _safe_tsv(v) -> str: return str(v).replace("\t", " ").replace("\r", " ").replace("\n", "\\n") def write_outputs( run_dir: Path, *, cards: dict, slots: list[CameraSlot], base_plates: list[BasePlateSpec], base_summary: dict, payloads: list[ShotBackgroundPayload], unmatched_design: Optional[list[DesignInferenceItem]] = None, run_meta: dict, ) -> None: run_dir.mkdir(parents=True, exist_ok=True) (run_dir / "inference_cards.json").write_text(json.dumps({ "identity": asdict(cards["identity"]), "topology": { "nodes": [asdict(n) for n in cards["topology"].nodes], "edges": [asdict(e) for e in cards["topology"].edges], "unknowns": cards["topology"].unknowns, }, "design_inference": [asdict(d) for d in cards["design_inference"]], "forbidden_drift": [asdict(f) for f in cards["forbidden_drift"]], "state_overlay": [asdict(s) for s in cards["state_overlay"]], }, ensure_ascii=False, indent=2), encoding="utf-8") (run_dir / "camera_slot_plan.json").write_text( json.dumps([asdict(s) for s in slots], ensure_ascii=False, indent=2), encoding="utf-8", ) # TSV slot_header = ["slot_id", "slot_role", "used_by_shots", "base_space_node", "camera_anchor", "looking_toward", "visible_nodes", "offscreen_nodes", "preserve_constraints", "confidence"] with (run_dir / "camera_slot_plan.tsv").open("w", encoding="utf-8") as f: f.write("\t".join(slot_header) + "\n") for s in slots: f.write("\t".join(_safe_tsv(x) for x in [ s.slot_id, s.slot_role, ",".join(s.used_by_shots), s.base_space_node, s.camera_anchor, s.looking_toward, ",".join(s.visible_nodes), ",".join(s.offscreen_nodes), " | ".join(s.preserve_constraints), s.confidence, ]) + "\n") (run_dir / "base_plate_plan.json").write_text( json.dumps({ "summary": base_summary, "plates": [asdict(bp) for bp in base_plates], "unmatched_design_items": [asdict(d) for d in (unmatched_design or [])], }, ensure_ascii=False, indent=2), encoding="utf-8", ) (run_dir / "shot_background_payloads.json").write_text( json.dumps([{ **{k: v for k, v in asdict(pl).items() if k != "entity_layout_markers"}, "entity_layout_markers": [asdict(m) for m in pl.entity_layout_markers], } for pl in payloads], ensure_ascii=False, indent=2), encoding="utf-8", ) (run_dir / "run_meta.json").write_text( json.dumps(run_meta, ensure_ascii=False, indent=2), encoding="utf-8" ) # --------------------------------------------------------------------------- # CLI # --------------------------------------------------------------------------- def parse_args() -> argparse.Namespace: ap = argparse.ArgumentParser(description="Rooftop Spatial Pipeline Plan (dry-run)") ap.add_argument("--run-id", default=None) ap.add_argument("--source-grounding-dir", default=str(DEFAULT_SOURCE_RUN)) ap.add_argument("--spatial-bg-dir", default=str(DEFAULT_SPATIAL_RUN)) ap.add_argument("--output-base", default=str(DEFAULT_OUTPUT_DIR)) ap.add_argument("--no-serve", action="store_true") ap.add_argument("--serve-port", type=int, default=8768) ap.add_argument("--bind", default="127.0.0.1") ap.add_argument("--estimate-only", action="store_true") return ap.parse_args() def main() -> int: args = parse_args() if args.bind == "0.0.0.0": print("WARNING: --bind 0.0.0.0 — unauthenticated local network server.", file=sys.stderr) run_id = args.run_id or f"{datetime.now():%Y%m%d_%H%M}_{uuid.uuid4().hex[:6]}" output_base = Path(args.output_base).resolve() run_dir = output_base / run_id source_run = Path(args.source_grounding_dir).resolve() spatial_run = Path(args.spatial_bg_dir).resolve() bible = load_bible(source_run) source_evidence = load_source_evidence(source_run) existing_plates = load_existing_base_plates(spatial_run) with SessionLocal() as session: shots = load_l05_shots(session) plans = bg.build_shot_plans(shots) cards = { "identity": build_identity_card(bible), "topology": build_topology_card(bible), "design_inference": build_design_inference_card(bible, source_evidence), "forbidden_drift": build_forbidden_drift_card(bible), "state_overlay": build_state_overlay_card(bible), } slots = build_camera_slots(plans, cards["topology"]) base_plates, base_summary, unmatched_design = build_base_plate_plan(slots, cards) payloads = build_shot_background_payloads( plans, slots, base_plates, cards["state_overlay"] ) # Codex MINOR: --estimate-only 는 webserver 시작 전 return. serve_info = None if not args.no_serve and not args.estimate_only: run_dir.mkdir(parents=True, exist_ok=True) serve_info = start_static_webserver(run_dir, args.serve_port, bind=args.bind) run_meta = { "run_id": run_id, "created_at": datetime.now().isoformat(timespec="seconds"), "plan_version": "v1", "source_grounding_dir": str(source_run), "spatial_bg_dir": str(spatial_run), "shot_count": len(plans), "card_counts": { "topology_nodes": len(cards["topology"].nodes), "topology_edges": len(cards["topology"].edges), "design_inference": len(cards["design_inference"]), "forbidden_drift": len(cards["forbidden_drift"]), "state_overlay": len(cards["state_overlay"]), }, "camera_slot_count": len(slots), "base_plate_summary": base_summary, "payload_count": len(payloads), "args": { "source_grounding_dir": str(source_run), "spatial_bg_dir": str(spatial_run), "no_serve": args.no_serve, "serve_port": args.serve_port, "bind": args.bind, }, "serve_info": serve_info, } if args.estimate_only: print(json.dumps(run_meta, ensure_ascii=False, indent=2)) return 0 write_outputs(run_dir, cards=cards, slots=slots, base_plates=base_plates, base_summary=base_summary, payloads=payloads, unmatched_design=unmatched_design, run_meta=run_meta) asset_rel = copy_existing_base_plates(run_dir, spatial_run) run_meta["base_plate_assets_dir"] = asset_rel render_html(run_dir, bible=bible, cards=cards, slots=slots, base_plates=base_plates, base_summary=base_summary, payloads=payloads, existing_plates=existing_plates, asset_rel=asset_rel, unmatched_design=unmatched_design, run_meta=run_meta) print(f"\nOK: run 완료 → {run_dir}") print(f" cards: identity/topology({len(cards['topology'].nodes)}n+{len(cards['topology'].edges)}e)/" f"design({len(cards['design_inference'])})/forbidden({len(cards['forbidden_drift'])})/" f"state({len(cards['state_overlay'])})") print(f" slots={len(slots)} (recommended_base_plates={base_summary['recommended_base_plate_count']} " f"vs current 4, delta={base_summary['delta']:+d})") print(f" payloads={len(payloads)}") print(f" index: {run_dir / 'index.html'}") if serve_info and serve_info.get("status") == "started": url = f"http://{serve_info['bind']}:{serve_info['port']}/index.html" print(f" webserver: {url} (PID {serve_info['pid']})") print(f" stop: kill {serve_info['pid']}") elif serve_info and serve_info.get("status") == "died_early": print(f" webserver: DIED_EARLY exit={serve_info.get('exit_code')}", file=sys.stderr) return 0 if __name__ == "__main__": raise SystemExit(main())