#!/usr/bin/env python3 """Rooftop Spatial Resolution Plan W1 — dry-run (plan_v2 APPROVED 2026-05-23). scripts_output/rooftop_spatial_resolution_plan// read-only: 이미지 0 / Gemini·OpenAI API 0 / DB write 0 / production schema 0. Import scope (Codex G6, plan_v2 §4): - app.core.database.SessionLocal — read-only PostgreSQL. - app.models.project.SceneStill — via load_l05_shots. - experiment_rooftop_spatial_bg.ShotMeta — dataclass. - experiment_rooftop_spatial_bg.load_l05_shots — DB read helper. build_shot_plans 절대 import 금지. plan_v2 §9 implementation guards: G1. active_status × plate_policy compatibility matrix (validate_node). G2. evidence_ev_ids strict lookup with lookup_status=missing emit. G3. shot_snapshot_hash = canonical sorted JSON SHA256. G4. Keyword whitelist incl "수리영 방"; "방" 단독 금지; "주방" room-resolve 제외. G5. gap_room_ / gap_camera_ naming. shot scope = single shot_id. G6. W2 분리 — production schema/step/code 무영향. """ from __future__ import annotations import argparse import csv import hashlib import html import json import sys import uuid from dataclasses import dataclass, field, asdict from datetime import datetime, timezone, timedelta from pathlib import Path from typing import Optional # repo root + env -------------------------------------------------------------- _REPO_ROOT = Path(__file__).resolve().parents[2] _BACKEND_ROOT = _REPO_ROOT / "backend" if str(_BACKEND_ROOT) not in sys.path: sys.path.insert(0, str(_BACKEND_ROOT)) _SCRIPTS_DIR = Path(__file__).resolve().parent if str(_SCRIPTS_DIR) not in sys.path: sys.path.insert(0, str(_SCRIPTS_DIR)) def _load_backend_env() -> None: """backend/.env -> os.environ (pydantic-settings env_file cwd-relative 보완).""" import os as _os # noqa: PLC0415 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 key, value = line.split("=", 1) key = key.strip() value = value.strip() if (value.startswith('"') and value.endswith('"')) or ( value.startswith("'") and value.endswith("'") ): value = value[1:-1] if key and key not in _os.environ: _os.environ[key] = value _load_backend_env() from experiment_rooftop_spatial_bg import ShotMeta, load_l05_shots # noqa: E402 # Constants -------------------------------------------------------------------- PROJECT_ID = "6cb862d9-590c-4dce-86e6-d10c2977db19" EPISODE_ID = "08ad2cd3-3e96-4d84-808f-869ee628473c" L05_SHORT_ID = "L05" DEFAULT_SOURCE_RUN = Path( "scripts_output/rooftop_source_grounding/codex_entry_sanity_gemini_ok" ) DEFAULT_OUTPUT_DIR = Path("scripts_output/rooftop_spatial_resolution_plan") PLAN_VERSION = "v2" LOCK_ID = "rooftop_topology_v1" LEDGER_ID = "rooftop_resolution_gaps_v1" # G1: active_status × plate_policy compatibility matrix ------------------------ COMPATIBILITY_MATRIX: dict[str, set[str]] = { "active": {"master_candidate", "contained_region", "boundary_opening"}, "inactive": {"out_of_scope"}, "needs_decision": {"needs_decision"}, } VALID_ACTIVE_STATUS = set(COMPATIBILITY_MATRIX.keys()) VALID_PLATE_POLICY: set[str] = set() for _vs in COMPATIBILITY_MATRIX.values(): VALID_PLATE_POLICY.update(_vs) # G4: room_node keyword whitelist ---------------------------------------------- # Ordered: longer phrases first so substring matching is deterministic. ROOM_KEYWORD_WHITELIST: list[tuple[str, str]] = [ ("수리영의 방", "수리영의_방"), ("수리영의_방", "수리영의_방"), ("수리영 방", "수리영의_방"), ("안방", "민숙의_방_안방"), ("욕실", "욕실"), ("거실", "거실"), ] # weak character/object signals (NOT direct room keywords) WEAK_CHARACTER_SIGNALS = { "수리영": "수리영의_방", # weak — character lives there but not direct room "엄마": "민숙의_방_안방", "민숙": "민숙의_방_안방", } WEAK_BED_SIGNAL = "침대" # Camera framing rules (plan_v2 §5 C4 — direct framing keyword) def _has_all(text: str, needles: list[str]) -> bool: return all(n in text for n in needles) def _has_any(text: str, needles: list[str]) -> bool: return any(n in text for n in needles) # Visibility map (plan_v2 §2-A visibility_from) -------------------------------- VISIBILITY_FROM_TABLE: list[dict] = [ {"from_node": "거실", "camera_family": "eye_level_wide", "default_visible": ["거실", "주방_영역", "현관"], "default_offscreen": ["수리영의_방", "민숙의_방_안방", "욕실"]}, {"from_node": "거실", "camera_family": "eye_level_table_close", "default_visible": ["거실"], "default_offscreen": ["주방_영역", "수리영의_방", "민숙의_방_안방", "욕실", "현관"]}, {"from_node": "수리영의_방", "camera_family": "doorway_wide", "default_visible": ["수리영의_방"], "default_offscreen": ["거실", "민숙의_방_안방", "욕실"]}, {"from_node": "민숙의_방_안방", "camera_family": "doorway_wide", "default_visible": ["민숙의_방_안방"], "default_offscreen": ["거실", "수리영의_방", "욕실"]}, {"from_node": "욕실", "camera_family": "mirror_close", "default_visible": ["욕실"], "default_offscreen": ["거실"]}, ] # Cluster definitions (plan_v2 §2-C) ------------------------------------------- CLUSTER_DEFS: list[dict] = [ { "cluster_id": "cluster_main_room", "topology_anchor_node": "거실", "master_plate_id": "bp_main_room_master", "master_camera_family": "eye_level_wide", "master_camera_anchor": "거실 중앙 standing eye-level", "master_covers_nodes": ["거실", "주방_영역", "현관"], "shared_layout_constraints": [ "거실 전체 좌우 폭", "창문 위치", "주방 코너 위치", ], "derived": [ { "plate_id": "bp_main_room_table_close", "camera_family": "eye_level_table_close", "camera_anchor": "식탁 옆 seated height", "covers_nodes": ["거실"], "inherited_layout_constraints": [ "식탁 윗면 위치 (master 에서 inherit)", "주변 의자 위치 (master 에서 inherit)", ], }, ], }, { "cluster_id": "cluster_suryeong_bedroom", "topology_anchor_node": "수리영의_방", "master_plate_id": "bp_suryeong_bedroom_master", "master_camera_family": "doorway_wide", "master_camera_anchor": "방 입구 문턱 standing eye-level", "master_covers_nodes": ["수리영의_방"], "shared_layout_constraints": ["방 내부 폭", "창문 위치"], "derived": [], }, { "cluster_id": "cluster_minsook_bedroom", "topology_anchor_node": "민숙의_방_안방", "master_plate_id": "bp_minsook_bedroom_master", "master_camera_family": "doorway_wide", "master_camera_anchor": "방 입구 문턱 standing eye-level", "master_covers_nodes": ["민숙의_방_안방"], "shared_layout_constraints": ["방 내부 폭"], "derived": [], }, { "cluster_id": "cluster_bathroom", "topology_anchor_node": "욕실", "master_plate_id": "bp_bathroom_master", "master_camera_family": "mirror_close", "master_camera_anchor": "욕실 standing eye-level, 거울 정면", "master_covers_nodes": ["욕실"], "shared_layout_constraints": ["거울 위치"], "derived": [], }, ] # ----------------------------------------------------------------------------- # Dataclasses (plan_v2 §2 schema) # ----------------------------------------------------------------------------- @dataclass class TopologyNode: id: str label: str kind: str active_status: str # active|inactive|needs_decision plate_policy: str # master_candidate|contained_region|boundary_opening|out_of_scope|needs_decision containment: dict confidence: str evidence_ev_ids: list[dict] = field(default_factory=list) # [{ev_id, lookup_status, lookup_reason?}] evidence_basis: str = "" inference_basis: str = "" needs_decision_protocol: Optional[dict] = None out_of_scope_reason: Optional[str] = None @dataclass class TopologyEdge: from_node: str to_node: str kind: str confidence: str annotation: Optional[str] = None inference_basis: Optional[str] = None @dataclass class VisibilityFrom: from_node: str camera_family: str default_visible: list[str] default_offscreen: list[str] @dataclass class SetTopologyLock: lock_id: str metadata: dict source_hashes: dict nodes: list[TopologyNode] edges: list[TopologyEdge] visibility_from: list[VisibilityFrom] frozen_unknowns: list[str] @dataclass class EvidencePointer: quote_source: str quote: str relation: str # supports|contradicts|ambiguous applies_to_field: str ev_id: Optional[str] = None lookup_status: str = "missing" # resolved|missing|multiple_match lookup_reason: Optional[str] = None @dataclass class CandidateInterpretation: option_id: str candidate_value: str rationale_hint: str strength: str # strong|weak|fallback note_for_decision: str @dataclass class BlockingUnknown: field: str question: str rolled_up_to_gap_id: Optional[str] = None rollup_group_id: Optional[str] = None @dataclass class BasePlateBinding: cluster_id: str plate_id: str role: str @dataclass class ShotSpatialResolution: shot_id: str scene_index: int shot_index: int state_overlay: str visible_entity_short_ids: list[str] room_node: Optional[str] room_node_candidates: list[str] room_node_resolution: dict camera_family: Optional[str] camera_anchor: str looking_toward: str visible_nodes: Optional[list[str]] offscreen_nodes: Optional[list[str]] entity_layout_markers: list[dict] state_overlay_marks: list[str] evidence_pointers: list[EvidencePointer] candidate_interpretations: list[CandidateInterpretation] blocking_unknowns: list[BlockingUnknown] base_plate_binding: Optional[BasePlateBinding] resolution_status: str @dataclass class BasePlateMember: plate_id: str role: str # master_plate|derived_shot_plate camera_family: str camera_anchor: str covers_nodes: list[str] must_show: list[str] = field(default_factory=list) must_not_show: list[str] = field(default_factory=list) shots_using_this_plate: list[str] = field(default_factory=list) shared_layout_constraints: list[str] = field(default_factory=list) derived_from_master: Optional[str] = None covers_nodes_basis: Optional[str] = None inherited_layout_constraints: list[str] = field(default_factory=list) visible_required_items: Optional[dict] = None covers_basis: Optional[str] = None @dataclass class BasePlateCluster: cluster_id: str topology_anchor_node: str active: bool members: list[BasePlateMember] @dataclass class GapEntry: gap_id: str category: str scope: str summary: str proposed_resolution: str blocks: list[str] = field(default_factory=list) shot_id: Optional[str] = None rollup_group_id: Optional[str] = None evidence_pointer_count: Optional[int] = None design_item_ids: Optional[list[str]] = None node_id: Optional[str] = None @dataclass class GapLedger: ledger_id: str metadata: dict entries: list[GapEntry] @dataclass class DesignItem: item_id: str category: str description: str source_evidence_ids: list[dict] confidence: str derived_from: str applies_to_nodes: list[str] # ----------------------------------------------------------------------------- # G1. validate_node # ----------------------------------------------------------------------------- def validate_node(node: TopologyNode) -> None: if node.active_status not in VALID_ACTIVE_STATUS: raise ValueError( f"invalid active_status: {node.active_status} for node {node.id}" ) if node.plate_policy not in VALID_PLATE_POLICY: raise ValueError( f"invalid plate_policy: {node.plate_policy} for node {node.id}" ) if node.plate_policy not in COMPATIBILITY_MATRIX[node.active_status]: raise ValueError( f"incompatible active_status={node.active_status} × " f"plate_policy={node.plate_policy} for node {node.id}" ) # ----------------------------------------------------------------------------- # G2. evidence_ev_ids strict lookup # ----------------------------------------------------------------------------- def lookup_evidence_ev_ids(text: str, source_evidence: list[dict], limit: int = 5) -> list[dict]: """Returns [{ev_id, lookup_status, lookup_reason?}, ...]. G2: strict lookup against source_evidence rows. If no row matches, return a single entry with lookup_status='missing' (fabricate forbidden). """ results: list[dict] = [] for ev in source_evidence: kw = ev.get("keyword", "") if not kw or kw not in text: continue offset = ev.get("char_offset", "") src = ev.get("source", "ep") ev_id = ev.get("ev_id") or f"ev_{src}_{offset}" results.append({"ev_id": ev_id, "lookup_status": "resolved"}) if len(results) >= limit: break if not results: results.append({ "ev_id": None, "lookup_status": "missing", "lookup_reason": f"no keyword match in source_evidence for: {text[:80]}", }) return results # ----------------------------------------------------------------------------- # G3. shot_snapshot_hash + compute_source_hashes # ----------------------------------------------------------------------------- def _collapse_ws(text: str) -> str: return " ".join((text or "").split()) def _extract_snippet(text: str, needle: str, window: int = 40) -> str: """Return a short window of `text` around the first occurrence of `needle`. Used by evidence_pointers to surface shot context instead of bare keyword. Codex IMPORTANT 1. """ if not text or not needle: return "" idx = text.find(needle) if idx < 0: return _collapse_ws(text)[: 2 * window] start = max(0, idx - window) end = min(len(text), idx + len(needle) + window) snippet = text[start:end] if start > 0: snippet = "…" + snippet if end < len(text): snippet = snippet + "…" return _collapse_ws(snippet) def compute_shot_snapshot_hash(shots: list[ShotMeta]) -> str: rows = [] for s in shots: rows.append({ "shot_id": s.label, "shot_description": _collapse_ws(s.shot_description), "scene_summary": _collapse_ws(s.scene_summary), "visible_short_ids": sorted(s.visible_short_ids), }) rows.sort(key=lambda r: r["shot_id"]) payload = json.dumps(rows, ensure_ascii=False, sort_keys=True, separators=(",", ":")) return hashlib.sha256(payload.encode("utf-8")).hexdigest() def _canonical_json_bytes(obj) -> bytes: return json.dumps(obj, ensure_ascii=False, sort_keys=True, separators=(",", ":")).encode("utf-8") def compute_source_hashes(bible: dict, source_evidence: list[dict], shots: list[ShotMeta]) -> dict[str, str]: bible_h = hashlib.sha256(_canonical_json_bytes(bible)).hexdigest() # evidence_sha256 — preserve row order for deterministic reproduction # since source_evidence.tsv has stable row order from grounding step. ev_payload = [] for ev in source_evidence: ev_payload.append({ "source": ev.get("source", ""), "keyword": ev.get("keyword", ""), "category": ev.get("category", ""), "candidate_field": ev.get("candidate_field", ""), "char_offset": ev.get("char_offset", ""), "confidence": ev.get("confidence", ""), "quote": ev.get("quote", ""), }) evidence_h = hashlib.sha256(_canonical_json_bytes(ev_payload)).hexdigest() shot_h = compute_shot_snapshot_hash(shots) return { "bible_sha256": bible_h, "evidence_sha256": evidence_h, "shot_snapshot_hash": shot_h, } # ----------------------------------------------------------------------------- # G4. room_node keyword classifier + conflict policy # ----------------------------------------------------------------------------- def classify_room_node_keyword(shot_description: str, scene_summary: str) -> dict: """Returns dict with: - matched_keywords: list[str] direct whitelist hits - matched_nodes: list[str] deduped node ids from direct hits - candidates: list[str] direct + weak candidates - weak_signals: list[str] weak character/bed evidence - status: 'resolved'|'needs_manual_room'|'needs_keyword' - conflict_reason: optional[str] """ text = f"{shot_description}\n{scene_summary}" matched_pairs: list[tuple[str, str]] = [] seen_phrases: set[str] = set() for keyword, node_id in ROOM_KEYWORD_WHITELIST: if keyword in text and keyword not in seen_phrases: matched_pairs.append((keyword, node_id)) seen_phrases.add(keyword) matched_keywords = [m[0] for m in matched_pairs] matched_nodes: list[str] = [] for _, node_id in matched_pairs: if node_id not in matched_nodes: matched_nodes.append(node_id) weak_signals: list[str] = [] weak_nodes: list[str] = [] has_bed = WEAK_BED_SIGNAL in text for char_name, char_node in WEAK_CHARACTER_SIGNALS.items(): if char_name in text: weak_signals.append(char_name) if char_node not in weak_nodes: weak_nodes.append(char_node) candidates = list(matched_nodes) for n in weak_nodes: if n not in candidates: candidates.append(n) if len(matched_nodes) >= 2: # plan_v2 §5 C4 rule 2: multi direct conflict → needs_manual_room return { "matched_keywords": matched_keywords, "matched_nodes": matched_nodes, "candidates": candidates, "weak_signals": weak_signals, "status": "needs_manual_room", "conflict_reason": ( f"multi direct room keyword conflict: {matched_keywords}" ), } if len(matched_nodes) == 1: return { "matched_keywords": matched_keywords, "matched_nodes": matched_nodes, "candidates": candidates, "weak_signals": weak_signals, "status": "resolved", "conflict_reason": None, } # No direct keyword. # rule 3: 수리영 + 민숙/엄마 동시 → needs_manual_room suryeong = "수리영" in text minsook_or_mom = ("민숙" in text) or ("엄마" in text) if suryeong and minsook_or_mom: return { "matched_keywords": [], "matched_nodes": [], "candidates": ["수리영의_방", "민숙의_방_안방"], "weak_signals": weak_signals, "status": "needs_manual_room", "conflict_reason": ( "수리영 + 민숙/엄마 동시 등장, direct room keyword 부재" ), } # rule 4: 엄마/민숙 + 침대 only → needs_manual_room (weak only) if (("엄마" in text) or ("민숙" in text)) and has_bed: return { "matched_keywords": [], "matched_nodes": [], "candidates": ["민숙의_방_안방"], "weak_signals": weak_signals + [WEAK_BED_SIGNAL], "status": "needs_manual_room", "conflict_reason": "엄마/민숙 + 침대 weak signal only", } # Otherwise: no direct keyword, no actionable weak evidence → needs_keyword return { "matched_keywords": [], "matched_nodes": [], "candidates": candidates, "weak_signals": weak_signals, "status": "needs_keyword", "conflict_reason": None, } # ----------------------------------------------------------------------------- # Camera framing classifier # ----------------------------------------------------------------------------- def classify_camera_family(shot_description: str, scene_summary: str, room_node: Optional[str]) -> Optional[str]: text = f"{shot_description}\n{scene_summary}" # mirror_close — bathroom only if "거울" in text and ("얼굴" in text or "씻" in text): return "mirror_close" # eye_level_table_close — main_room only if "식탁" in text and ("앉" in text or "찻잔" in text or "옆" in text): return "eye_level_table_close" # doorway_wide — bedroom (room_node is suryeong or minsook) if room_node in {"수리영의_방", "민숙의_방_안방"}: if "wide" in text.lower() or "방 안 전체" in text or "문턱" in text: return "doorway_wide" # eye_level_wide — main_room if "거실" in text and ("wide" in text.lower() or "전체" in text or "중앙" in text): return "eye_level_wide" return None CAMERA_FAMILY_TO_ANCHOR: dict[str, tuple[str, str]] = { # camera_family -> (camera_anchor, looking_toward) "eye_level_wide": ("거실 중앙 standing eye-level", "wide framing, 거실 전체"), "eye_level_table_close": ( "식탁 옆 seated height", "식탁 위 small props 중심", ), "doorway_wide": ( "방 입구 문턱 standing eye-level", "wide framing, 방 안 전체", ), "mirror_close": ( "욕실 standing eye-level, 거울 정면", "거울 + 벽 일부", ), } ROOM_CAMERA_OPTIONS: dict[str, list[str]] = { "거실": ["eye_level_wide", "eye_level_table_close"], "수리영의_방": ["doorway_wide"], "민숙의_방_안방": ["doorway_wide"], "욕실": ["mirror_close"], } def _camera_family_options_for_room(room_node: Optional[str]) -> list[str]: if not room_node: return [] return list(ROOM_CAMERA_OPTIONS.get(room_node, [])) # ----------------------------------------------------------------------------- # Topology lock builder # ----------------------------------------------------------------------------- def _normalize_node_id(name: str) -> str: return (name or "").strip().replace(" ", "_").replace("(", "").replace(")", "") def build_set_topology_lock(bible: dict, source_evidence: list[dict], source_hashes: dict) -> SetTopologyLock: nodes: list[TopologyNode] = [] # 거실 — active, master_candidate, parent of 주방_영역. nodes.append(TopologyNode( id="거실", label="거실 (living kitchen)", kind="living_kitchen", active_status="active", plate_policy="master_candidate", containment={"parent_of": ["주방_영역"], "contained_in": None, "visually_part_of": None}, confidence="trusted", evidence_ev_ids=lookup_evidence_ev_ids("거실", source_evidence), evidence_basis="bible.sub_spaces.거실.evidence_quotes", )) # 주방_영역 — active, contained_region, contained_in 거실. nodes.append(TopologyNode( id="주방_영역", label="주방 영역 (sink + stove)", kind="kitchen_region", active_status="active", plate_policy="contained_region", containment={ "parent_of": [], "contained_in": "거실", "containment_confidence": "inferred", "containment_reversibility": ( "reversible_on_evidence — bible.unknowns 의 '거실/주방 가벽 유무'" " 가 사후 확정되어 가벽 존재로 확인되면 contained → adjacent_room" " 으로 강등 + 별도 master_candidate 승격. 그 시점에 main_room" " master/derived 는 invalidated, i2i 재사용 금지." ), }, confidence="inferred", evidence_ev_ids=lookup_evidence_ev_ids("싱크대", source_evidence), inference_basis=( "bible.sub_spaces.주방 영역.layout_notes='거실과 인접하거나 통합된" " 형태의 주방 공간'. bible.unknowns 가벽 유무 미해소." ), )) # 수리영의_방 — active, master_candidate. nodes.append(TopologyNode( id="수리영의_방", label="수리영의 방", kind="bedroom", active_status="active", plate_policy="master_candidate", containment={"parent_of": [], "contained_in": None, "visually_part_of": None}, confidence="trusted", evidence_ev_ids=lookup_evidence_ev_ids("수리영", source_evidence), evidence_basis="bible.sub_spaces.수리영의 방.evidence_quotes", )) # 민숙의_방_안방 — needs_decision. nodes.append(TopologyNode( id="민숙의_방_안방", label="민숙의 방 (안방)", kind="bedroom", active_status="needs_decision", plate_policy="needs_decision", containment={"parent_of": [], "contained_in": None, "visually_part_of": None}, confidence="trusted_for_existence_only", evidence_ev_ids=lookup_evidence_ev_ids("안방", source_evidence), evidence_basis="bible.sub_spaces.민숙의 방 (안방).evidence_quotes", needs_decision_protocol={ "stage": "W1", "engine": "keyword_evidence_collector (LLM 0)", "engine_output": "evidence_table + recommendation (NOT binding)", "decision_owner": "user (after reviewing W1 HTML)", "default_on_ambiguity": ( "active_status='needs_decision' 유지, plate 미생성" ), }, )) # 욕실 — active, master_candidate. nodes.append(TopologyNode( id="욕실", label="욕실", kind="bathroom", active_status="active", plate_policy="master_candidate", containment={"parent_of": [], "contained_in": None, "visually_part_of": None}, confidence="trusted", evidence_ev_ids=lookup_evidence_ev_ids("욕실", source_evidence), evidence_basis="bible.sub_spaces.욕실.evidence_quotes", )) # 현관 — active, boundary_opening, visually_part_of 거실. nodes.append(TopologyNode( id="현관", label="현관", kind="entry", active_status="active", plate_policy="boundary_opening", containment={ "parent_of": [], "contained_in": None, "visually_part_of": "거실", "visually_part_of_basis": ( "di_014 (현관 철문 @ 옥탑방 외부와 거실 사이의 경계)" " applies_to_nodes = [거실]. master plate 별도 승격 X." ), }, confidence="inferred", evidence_ev_ids=lookup_evidence_ev_ids("현관", source_evidence), evidence_basis="bible.doors_windows.현관 철문", )) # rooftop_outer — inactive, out_of_scope. nodes.append(TopologyNode( id="rooftop_outer", label="옥상/옥탑 외부", kind="outer", active_status="inactive", plate_policy="out_of_scope", containment={"parent_of": [], "contained_in": None, "visually_part_of": None}, confidence="trusted_for_separation", evidence_ev_ids=lookup_evidence_ev_ids("옥상", source_evidence), out_of_scope_reason=( "entity_canon L04 (외부+옥상 마당) 가 별도 location entity." " L05 spatial plan 은 내부만 다룬다." ), )) for n in nodes: validate_node(n) # edges ------------------------------------------------------------------- edges: list[TopologyEdge] = [ TopologyEdge(from_node="현관", to_node="거실", kind="opening", confidence="trusted"), TopologyEdge(from_node="거실", to_node="수리영의_방", kind="door", confidence="trusted"), TopologyEdge(from_node="거실", to_node="민숙의_방_안방", kind="door", confidence="trusted"), TopologyEdge(from_node="거실", to_node="욕실", kind="door", confidence="trusted"), TopologyEdge( from_node="거실", to_node="주방_영역", kind="containment", confidence="inferred", inference_basis=( "bible.sub_spaces.주방_영역.layout_notes + bible.unknowns 미해소" " — reversible_on_evidence" ), ), TopologyEdge( from_node="현관", to_node="rooftop_outer", kind="door", confidence="inferred", annotation="out_of_L05_scope — base plate 후보 아님", ), ] visibility = [VisibilityFrom(**row) for row in VISIBILITY_FROM_TABLE] return SetTopologyLock( lock_id=LOCK_ID, metadata={ "generated_at": _now_iso(), "evidence_source": ( "gemini_rooftop_bible.json @ codex_entry_sanity_gemini_ok" ), "plan_version": PLAN_VERSION, }, source_hashes=source_hashes, nodes=nodes, edges=edges, visibility_from=visibility, frozen_unknowns=list(bible.get("unknowns", [])), ) def _now_iso() -> str: return datetime.now(timezone(timedelta(hours=9))).isoformat(timespec="seconds") # ----------------------------------------------------------------------------- # Shot resolver # ----------------------------------------------------------------------------- # Negation patterns (Codex BLOCKING 1) ---------------------------------------- # Re-imported from pipeline_plan.py 의 패턴 (별도 분리 모듈 — share import 금지 정신). import re as _re # noqa: PLC0415 CORPSE_NEGATION_PATTERNS = [ _re.compile(r"시신.{0,5}없"), _re.compile(r"시체.{0,5}없"), _re.compile(r"피.{0,5}없"), _re.compile(r"핏자국.{0,5}없"), _re.compile(r"흔적.{0,5}없"), _re.compile(r"흔적을\s*찾지"), _re.compile(r"감쪽\s*같이\s*사라"), ] VANDALIZED_VISION_HINTS = [ "벽면에 나타난", "벽면에 그려진", "벽에 칠한", "붓으로 칠한", "거친 붉은 원", "어지럽혀", "뒤집힌", "넘어진 가구", ] def _detect_state_overlay(shot: ShotMeta) -> tuple[str, list[str]]: """Returns (state_overlay, state_marks). plan_v2 §2-B 의 state_overlay enum. 우선순위 (Codex BLOCKING 1): 1. cleaned: 'cleaned' 의도 (지나치게 깨끗/깔끔, 흔적 없) — corpse 보다 우선. 2. vandalized: 벽면 붉은 원 / 가구 어지럽힘 / vision 표식 — corpse 어휘와 겹쳐도 우선. 3. corpse_marks: 시신/시체/참혹 + negation guard 없을 때만. 4. empty: 빈 집/혼자 깨어 + 위 3 가지 모두 미해당. 5. normal: default. """ text = f"{shot.shot_description}\n{shot.scene_summary}" shot_text = shot.shot_description or "" # 1. cleaned — corpse keyword 와 동시 등장해도 우선. cleaned_hits = [ k for k in ("깨끗하게 정돈", "지나치게 깔끔", "지나치게 깨끗", "치워", "흔적을 찾지 못해", "흔적이 없", "감쪽같이") if k in text ] if cleaned_hits: return "cleaned", [f"meticulously cleaned ({hit})" for hit in cleaned_hits[:3]] # 2. vandalized vision (shot_description 우선) — corpse 어휘와 겹쳐도 우선. if any(h in shot_text for h in VANDALIZED_VISION_HINTS): return "vandalized", [ "overturned furniture / scratches / vision marks on walls", ] # 3. corpse_marks — negation guard corpse_kw_hit = any(k in text for k in ["시신", "시체", "어깨가 뜯", "쇄골", "참혹"]) if corpse_kw_hit: negated = any(p.search(text) for p in CORPSE_NEGATION_PATTERNS) if not negated: return "corpse_marks", [ "dark floor stain near body location", "disturbed bedding", "scattered objects — NO human figure", ] # Negated: shot 텍스트가 'cleaned' 의도일 가능성 — 위 cleaned 에서 잡혔어야 정상. # 둘 다 놓치면 normal fallback. # 4. vandalized — broader vandalized (전체 텍스트) if any(h in text for h in ["어지럽혀", "뒤집힌", "넘어진 가구", "벽면에 거친", "벽에 칠한", "벽면에 그려진", "붓으로 칠한"]): return "vandalized", ["overturned furniture / scratches / marks on walls"] # 5. empty — shot_description 에서 직접 시그널 (summary 만으론 약함) if any(k in shot_text for k in ["빈 집", "아무도 없", "혼자 깨어"]): return "empty", ["no human figures"] return "normal", [] def resolve_shot_spatial(shot: ShotMeta, topology: SetTopologyLock, source_evidence: list[dict]) -> ShotSpatialResolution: room_decision = classify_room_node_keyword( shot.shot_description, shot.scene_summary, ) # room_node 결정 room_node: Optional[str] = None if room_decision["status"] == "resolved": room_node = room_decision["matched_nodes"][0] camera_family = classify_camera_family( shot.shot_description, shot.scene_summary, room_node, ) state_overlay, state_marks = _detect_state_overlay(shot) # resolution_status room_ok = room_node is not None camera_ok = camera_family is not None if room_ok and camera_ok: status = "resolved" elif room_ok and not camera_ok: status = "needs_camera_anchor" elif (not room_ok) and camera_ok: status = "needs_manual_room" else: # 둘 다 미해소 if room_decision["status"] == "needs_manual_room": status = "needs_manual_room" else: status = "needs_both" # camera_anchor / looking_toward if camera_family and camera_family in CAMERA_FAMILY_TO_ANCHOR: camera_anchor, looking_toward = CAMERA_FAMILY_TO_ANCHOR[camera_family] else: camera_anchor, looking_toward = "", "" # visible / offscreen nodes (lookup from topology.visibility_from) visible_nodes: Optional[list[str]] = None offscreen_nodes: Optional[list[str]] = None if room_node and camera_family: for vf in topology.visibility_from: if vf.from_node == room_node and vf.camera_family == camera_family: visible_nodes = list(vf.default_visible) offscreen_nodes = list(vf.default_offscreen) break # entity_layout_markers — placeholder skeleton (no LLM, deterministic) layout_markers: list[dict] = [] for short_id in shot.visible_short_ids: if short_id == "L05": continue kind = "character" if short_id.startswith("C") else "prop" layout_markers.append({ "entity_id": short_id, "kind": kind, "zone": "midground_center", }) # evidence_pointers — G2 strict lookup + shot snippet quote (IMPORTANT 1) pointers: list[EvidencePointer] = [] for kw in room_decision["matched_keywords"]: for source_name, source_text in ( ("shot_description", shot.shot_description), ("scene_summary", shot.scene_summary), ): if kw not in source_text: continue snippet = _extract_snippet(source_text, kw, window=40) lookups = lookup_evidence_ev_ids(kw, source_evidence) for lk in lookups: pointers.append(EvidencePointer( quote_source=source_name, quote=snippet, relation="supports", applies_to_field="room_node", ev_id=lk.get("ev_id"), lookup_status=lk.get("lookup_status", "missing"), lookup_reason=lk.get("lookup_reason"), )) for weak in room_decision["weak_signals"]: for source_name, source_text in ( ("shot_description", shot.shot_description), ("scene_summary", shot.scene_summary), ): if weak not in source_text: continue snippet = _extract_snippet(source_text, weak, window=40) lookups = lookup_evidence_ev_ids(weak, source_evidence) for lk in lookups: pointers.append(EvidencePointer( quote_source=source_name, quote=snippet, relation="ambiguous", applies_to_field="room_node", ev_id=lk.get("ev_id"), lookup_status=lk.get("lookup_status", "missing"), lookup_reason=lk.get("lookup_reason"), )) # candidate_interpretations — 2+ options for any unresolved shot # (plan_v2 §6 criterion 2: needs_manual_room / needs_camera_anchor 모두 ≥ 2). candidates: list[CandidateInterpretation] = [] if status in {"needs_manual_room", "needs_both"}: cand_nodes = room_decision["candidates"] or [ "수리영의_방", "민숙의_방_안방", ] for idx, cand in enumerate(cand_nodes[:3]): option_id = chr(ord("A") + idx) candidates.append(CandidateInterpretation( option_id=option_id, candidate_value=cand, rationale_hint=( f"weak/direct evidence for {cand}: " f"{room_decision.get('conflict_reason') or 'candidate from text'}" ), strength="weak", note_for_decision="W1 evidence 부족 — 사용자 결정 필요", )) if len(candidates) < 2: candidates.append(CandidateInterpretation( option_id=chr(ord("A") + len(candidates)), candidate_value="both_plates_needed", rationale_hint="두 방 plate 모두 필요 가능성", strength="fallback", note_for_decision="needs_user_decision", )) elif status in {"needs_camera_anchor"}: cam_options = _camera_family_options_for_room(room_node) for idx, cam in enumerate(cam_options[:3]): option_id = chr(ord("A") + idx) candidates.append(CandidateInterpretation( option_id=option_id, candidate_value=cam, rationale_hint=( f"camera_family candidate for room_node={room_node}:" f" framing keyword 미매칭 — 사용자 anchor 결정 필요" ), strength="weak", note_for_decision="shot_description 의 framing 키워드 부족", )) if len(candidates) < 2: candidates.append(CandidateInterpretation( option_id=chr(ord("A") + len(candidates)), candidate_value="manual_camera_anchor", rationale_hint="해당 room 의 기본 framing 외 사용자 직접 anchor 지정", strength="fallback", note_for_decision="needs_user_decision", )) # blocking_unknowns blocking: list[BlockingUnknown] = [] if not room_ok: blocking.append(BlockingUnknown( field="room_node", question=f"{shot.label} 의 room_node 는 어느 노드인가?", rolled_up_to_gap_id=f"gap_room_{shot.label}", rollup_group_id="bedroom_unresolved", )) if not camera_ok: blocking.append(BlockingUnknown( field="camera_family", question=f"{shot.label} 의 camera_family 는 무엇인가?", rolled_up_to_gap_id=f"gap_camera_{shot.label}", rollup_group_id="manual_review_needed", )) return ShotSpatialResolution( shot_id=shot.label, scene_index=shot.scene_index, shot_index=shot.shot_index, state_overlay=state_overlay, visible_entity_short_ids=list(shot.visible_short_ids), room_node=room_node, room_node_candidates=room_decision["candidates"], room_node_resolution={ "rule": "shot_description + scene_summary direct keyword + conflict policy", "matched_direct_keywords": room_decision["matched_keywords"], "matched_nodes": room_decision["matched_nodes"], "weak_signals": room_decision["weak_signals"], "conflict_reason": room_decision.get("conflict_reason"), "confidence": "trusted" if status == "resolved" else "low", }, camera_family=camera_family, camera_anchor=camera_anchor, looking_toward=looking_toward, visible_nodes=visible_nodes, offscreen_nodes=offscreen_nodes, entity_layout_markers=layout_markers, state_overlay_marks=state_marks, evidence_pointers=pointers, candidate_interpretations=candidates, blocking_unknowns=blocking, base_plate_binding=None, # attached in cluster_base_plates / bind step resolution_status=status, ) # ----------------------------------------------------------------------------- # BasePlateCluster builder # ----------------------------------------------------------------------------- def cluster_base_plates(topology: SetTopologyLock, resolutions: list[ShotSpatialResolution] ) -> list[BasePlateCluster]: # node lookup node_by_id = {n.id: n for n in topology.nodes} # must_show / must_not_show derived from bible (passed indirectly via # cluster definitions + resolutions). For W1 keep static must_show seeds # from CLUSTER_DEFS — i2i prompt 합성 W2 단계 책임. clusters: list[BasePlateCluster] = [] for cdef in CLUSTER_DEFS: anchor_id = cdef["topology_anchor_node"] node = node_by_id.get(anchor_id) if node is None: continue active = (node.active_status == "active" and node.plate_policy == "master_candidate") master = BasePlateMember( plate_id=cdef["master_plate_id"], role="master_plate", camera_family=cdef["master_camera_family"], camera_anchor=cdef["master_camera_anchor"], covers_nodes=list(cdef["master_covers_nodes"]), must_show=[], must_not_show=[], shared_layout_constraints=list(cdef.get("shared_layout_constraints", [])), ) members: list[BasePlateMember] = [master] for d in cdef.get("derived", []): members.append(BasePlateMember( plate_id=d["plate_id"], role="derived_shot_plate", camera_family=d["camera_family"], camera_anchor=d["camera_anchor"], covers_nodes=list(d["covers_nodes"]), must_show=[], must_not_show=[], derived_from_master=cdef["master_plate_id"], inherited_layout_constraints=list(d.get("inherited_layout_constraints", [])), visible_required_items={ "emphasize": [], "may_omit": [], "allow": [], }, covers_nodes_basis=( "derived ⊆ master covers_nodes (close framing 으로 일부 노드 시야 밖)" ), )) # bind shots to members using resolution.room_node + camera_family shots_by_member: dict[str, list[str]] = {m.plate_id: [] for m in members} for res in resolutions: if res.resolution_status != "resolved": continue if res.room_node != anchor_id: continue best = None for m in members: if m.camera_family == res.camera_family: best = m break if best is None: # fall back to master if camera_family unrecognised best = master shots_by_member[best.plate_id].append(res.shot_id) res.base_plate_binding = BasePlateBinding( cluster_id=cdef["cluster_id"], plate_id=best.plate_id, role=best.role, ) for m in members: m.shots_using_this_plate = list(shots_by_member[m.plate_id]) clusters.append(BasePlateCluster( cluster_id=cdef["cluster_id"], topology_anchor_node=anchor_id, active=active, members=members, )) return clusters # ----------------------------------------------------------------------------- # Design item inference (from bible → applies_to_nodes) # ----------------------------------------------------------------------------- def _node_id_for_text(text: str) -> list[str]: 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: 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 build_design_items(bible: dict, source_evidence: list[dict] ) -> list[DesignItem]: items: list[DesignItem] = [] seq = 1 def _emit(category: str, description: str, applies_nodes: list[str], confidence: str, derived_from: str) -> None: nonlocal seq items.append(DesignItem( item_id=f"di_{seq:03d}", category=category, description=description, source_evidence_ids=lookup_evidence_ev_ids(description, source_evidence), confidence=confidence, derived_from=derived_from, applies_to_nodes=applies_nodes, )) seq += 1 for cue in bible.get("required_visual_cues", []): cat = "fixture" # Codex IMPORTANT 3: '석창포' has '창' substring. Use '창문' / '작은 창' instead. if ("창문" in cue or "작은 창" in cue) and "석창포" not in cue: cat = "window" elif "문" in cue or "철문" in cue: cat = "door" elif "거울" in cue: cat = "fixture" elif "침대" in cue or "커튼" in cue: cat = "furniture" _emit(category=cat, description=cue, applies_nodes=_node_id_for_text(cue), confidence="direct", derived_from="bible.required_visual_cues") for f in bible.get("furniture", []): if not isinstance(f, dict): continue sub = f.get("sub_space", "") applies = _node_id_for_text(sub) or _node_id_for_text(f.get("name", "")) _emit(category="furniture", description=f"{f.get('name', '')} ({sub})", applies_nodes=applies, confidence=f.get("confidence_band", "unknown"), derived_from="bible.furniture") for m in bible.get("materials", []): if not isinstance(m, dict): continue where = m.get("where", "") applies = _node_id_for_text(where) _emit(category="finish", description=f"{where} — {m.get('material', '')}", applies_nodes=applies, confidence=m.get("confidence_band", "unknown"), derived_from="bible.materials") 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", "") applies = _node_id_for_text(loc) or _node_id_for_text(d.get("kind", "")) _emit(category=cat, description=f"{d.get('kind', '')} @ {loc}", applies_nodes=applies, confidence=d.get("confidence_band", "unknown"), derived_from="bible.doors_windows") return items # ----------------------------------------------------------------------------- # GapLedger builder # ----------------------------------------------------------------------------- def build_gap_ledger(topology: SetTopologyLock, resolutions: list[ShotSpatialResolution], design_items: list[DesignItem], clusters: list[BasePlateCluster]) -> GapLedger: entries: list[GapEntry] = [] # Shot-scope: room + camera gaps (G5 naming, single shot_id) for res in resolutions: if res.resolution_status in {"needs_manual_room", "needs_both"}: entries.append(GapEntry( gap_id=f"gap_room_{res.shot_id}", category="unresolved_room_node", scope="shot", shot_id=res.shot_id, rollup_group_id="bedroom_unresolved", summary=( f"{res.shot_id} — room_node 미결정. matched_keywords=" f"{res.room_node_resolution.get('matched_direct_keywords') or []}" ), evidence_pointer_count=len(res.evidence_pointers), proposed_resolution=( "W1 keyword rule 적용 결과 + evidence_pointers 검토 후" " 사용자 결정 (active=needs_decision 유지 가능)" ), blocks=["BasePlateCluster 매핑"], )) if res.resolution_status in {"needs_camera_anchor", "needs_both"}: entries.append(GapEntry( gap_id=f"gap_camera_{res.shot_id}", category="unresolved_camera_anchor", scope="shot", shot_id=res.shot_id, rollup_group_id="manual_review_needed", summary=( f"{res.shot_id} — camera_family 미결정. room_node=" f"{res.room_node or 'unresolved'}" ), evidence_pointer_count=len(res.evidence_pointers), proposed_resolution=( "W1 keyword rule 미매칭 — shot_description 보강 또는" " 사용자 anchor 결정" ), blocks=["BasePlateCluster 매핑"], )) # Design-scope: unmatched design items grouped. cluster_covers: set[str] = set() for c in clusters: if not c.active: continue for m in c.members: cluster_covers.update(m.covers_nodes) cluster_covers.update({"거실", "주방_영역", "현관"}) # boundary/contained absorbed into 거실 unmatched_ids: list[str] = [] for item in design_items: if not item.applies_to_nodes: continue # global — skip if any(n in cluster_covers for n in item.applies_to_nodes): continue # exception: 민숙 안방 design items are unmatched while cluster inactive unmatched_ids.append(item.item_id) if unmatched_ids: entries.append(GapEntry( gap_id="gap_design_unmatched", category="unmatched_design_item", scope="design", design_item_ids=unmatched_ids, summary=( f"{len(unmatched_ids)} design items 의 applies_to_nodes 가" " 활성 cluster covers_nodes 와 매칭 X" ), proposed_resolution=( "topology containment / boundary_opening 정책 검토. 민숙_방_안방" " 활성화 시 di_011 등 흡수 가능" ), blocks=["cluster_main_room.master.must_show 확정", "cluster_minsook_bedroom 활성화 여부"], )) # Topology-scope: kitchen edge (always present in our model — informational) entries.append(GapEntry( gap_id="gap_topology_edge_kitchen", category="missing_topology_edge", scope="topology", summary=( "bible.layout_relations 에 거실 → 주방_영역 edge 미명시 — inferred" " containment edge 추가됨 (reversible_on_evidence)" ), proposed_resolution=( "bible.unknowns '거실/주방 가벽 유무' 확정 시 edge 재평가." " 가벽 발견 → containment → adjacent_room 강등" ), blocks=["SetTopologyLock.edges 정합성"], )) # Topology-scope: minsook bedroom active decision minsook_node = next( (n for n in topology.nodes if n.id == "민숙의_방_안방"), None, ) if minsook_node is not None and minsook_node.active_status == "needs_decision": entries.append(GapEntry( gap_id="gap_active_minsook_bedroom", category="active_decision_required", scope="topology", node_id="민숙의_방_안방", summary=( "selected 14 shots 안에서 안방 direct keyword evidence 와" " weak (엄마 + 침대) evidence 비교 후 사용자 결정 필요" ), proposed_resolution=( "W1 HTML §7 evidence_table 검토 후 active / inactive 둘 중" " 사용자 결정" ), blocks=["cluster_minsook_bedroom 활성화", "bedroom_unresolved 매핑"], )) return GapLedger( ledger_id=LEDGER_ID, metadata={"plan_version": PLAN_VERSION, "generated_at": _now_iso()}, entries=entries, ) def compute_gap_metrics(ledger: GapLedger) -> dict: """Codex IMPORTANT 2: blocking_gaps_count 의 정확한 breakdown. Definitions: - blocking: entry.blocks 가 non-empty 인 entries. - informational: blocks 가 비어있거나 informational summary 인 entries. """ by_scope = {"shot": 0, "design": 0, "topology": 0} by_category: dict[str, int] = {} blocking_entries: list[GapEntry] = [] informational: list[GapEntry] = [] for e in ledger.entries: by_category[e.category] = by_category.get(e.category, 0) + 1 if e.blocks: blocking_entries.append(e) by_scope[e.scope] = by_scope.get(e.scope, 0) + 1 else: informational.append(e) return { "blocking_decision_gaps_count": len(blocking_entries), "shot_blocking_count": by_scope["shot"], "design_blocking_count": by_scope["design"], "topology_blocking_count": by_scope["topology"], "informational_gap_count": len(informational), "total_gap_entries": len(ledger.entries), "category_breakdown": by_category, } # ----------------------------------------------------------------------------- # I/O helpers # ----------------------------------------------------------------------------- def _load_bible_and_evidence(source_run: Path) -> tuple[dict, list[dict]]: bible_path = source_run / "gemini_rooftop_bible.json" evidence_path = source_run / "source_evidence.tsv" if not bible_path.exists(): raise FileNotFoundError(f"missing bible: {bible_path}") if not evidence_path.exists(): raise FileNotFoundError(f"missing evidence tsv: {evidence_path}") bible = json.loads(bible_path.read_text(encoding="utf-8")) rows: list[dict] = [] with evidence_path.open("r", encoding="utf-8", newline="") as f: reader = csv.DictReader(f, delimiter="\t") for r in reader: rows.append(dict(r)) return bible, rows def _to_jsonable(obj): if isinstance(obj, list): return [_to_jsonable(x) for x in obj] if isinstance(obj, dict): return {k: _to_jsonable(v) for k, v in obj.items()} if hasattr(obj, "__dataclass_fields__"): return _to_jsonable(asdict(obj)) return obj def write_outputs(out_dir: Path, topology: SetTopologyLock, resolutions: list[ShotSpatialResolution], clusters: list[BasePlateCluster], design_items: list[DesignItem], ledger: GapLedger, run_meta: dict) -> None: out_dir.mkdir(parents=True, exist_ok=True) (out_dir / "topology_lock.json").write_text( json.dumps(_to_jsonable(topology), ensure_ascii=False, indent=2), encoding="utf-8", ) (out_dir / "shot_resolutions.json").write_text( json.dumps(_to_jsonable(resolutions), ensure_ascii=False, indent=2), encoding="utf-8", ) _write_shot_resolutions_tsv(out_dir / "shot_resolutions.tsv", resolutions) (out_dir / "base_plate_clusters.json").write_text( json.dumps(_to_jsonable(clusters), ensure_ascii=False, indent=2), encoding="utf-8", ) (out_dir / "design_items.json").write_text( json.dumps(_to_jsonable(design_items), ensure_ascii=False, indent=2), encoding="utf-8", ) (out_dir / "gap_ledger.json").write_text( json.dumps(_to_jsonable(ledger), ensure_ascii=False, indent=2), encoding="utf-8", ) _write_gap_ledger_tsv(out_dir / "gap_ledger.tsv", ledger) (out_dir / "index.html").write_text( render_html(topology, resolutions, clusters, design_items, ledger, run_meta), encoding="utf-8", ) (out_dir / "run_meta.json").write_text( json.dumps(run_meta, ensure_ascii=False, indent=2), encoding="utf-8", ) def _write_shot_resolutions_tsv(path: Path, resolutions: list[ShotSpatialResolution]) -> None: with path.open("w", encoding="utf-8", newline="") as f: writer = csv.writer(f, delimiter="\t") writer.writerow([ "shot_id", "scene_index", "shot_index", "room_node", "camera_family", "resolution_status", "state_overlay", "matched_direct_keywords", "candidates", "base_plate_cluster", "base_plate_id", ]) for r in resolutions: binding = r.base_plate_binding writer.writerow([ r.shot_id, r.scene_index, r.shot_index, r.room_node or "", r.camera_family or "", r.resolution_status, r.state_overlay, ",".join(r.room_node_resolution.get("matched_direct_keywords", []) or []), ",".join(r.room_node_candidates or []), binding.cluster_id if binding else "", binding.plate_id if binding else "", ]) def _write_gap_ledger_tsv(path: Path, ledger: GapLedger) -> None: with path.open("w", encoding="utf-8", newline="") as f: writer = csv.writer(f, delimiter="\t") writer.writerow([ "gap_id", "category", "scope", "shot_id", "rollup_group_id", "summary", "proposed_resolution", "blocks", ]) for e in ledger.entries: writer.writerow([ e.gap_id, e.category, e.scope, e.shot_id or "", e.rollup_group_id or "", e.summary, e.proposed_resolution, ";".join(e.blocks or []), ]) # ----------------------------------------------------------------------------- # HTML renderer — C10 gap-first # ----------------------------------------------------------------------------- def render_html(topology: SetTopologyLock, resolutions: list[ShotSpatialResolution], clusters: list[BasePlateCluster], design_items: list[DesignItem], ledger: GapLedger, run_meta: dict) -> str: metrics = compute_gap_metrics(ledger) blocking_count = metrics["blocking_decision_gaps_count"] informational_count = metrics["informational_gap_count"] resolved_count = sum(1 for r in resolutions if r.resolution_status == "resolved") master_count = sum( 1 for c in clusters if c.active for m in c.members if m.role == "master_plate" ) derived_count = sum( 1 for c in clusters if c.active for m in c.members if m.role == "derived_shot_plate" ) def esc(s: object) -> str: return html.escape(str(s)) if s is not None else "" # Design item → cluster mapping (Codex BLOCKING 2: §5 cluster matching) active_cluster_covers: dict[str, str] = {} # node_id -> cluster_id for c in clusters: if not c.active: continue for m in c.members: for n in m.covers_nodes: active_cluster_covers.setdefault(n, c.cluster_id) # boundary/contained absorption — 거실 cluster 가 흡수. if "거실" in active_cluster_covers: for boundary in ("주방_영역", "현관"): active_cluster_covers.setdefault(boundary, active_cluster_covers["거실"]) design_gap_lookup: dict[str, str] = {} for e in ledger.entries: if e.category == "unmatched_design_item" and e.design_item_ids: for did in e.design_item_ids: design_gap_lookup[did] = e.gap_id def design_cluster_match(item: DesignItem) -> tuple[str, str]: if not item.applies_to_nodes: return ("global", "") matched = [] for n in item.applies_to_nodes: if n in active_cluster_covers: matched.append(active_cluster_covers[n]) if matched: return ("cluster_match: " + ",".join(sorted(set(matched))), "") return ("unmatched", design_gap_lookup.get(item.item_id, "")) parts: list[str] = [] parts.append("") parts.append("Rooftop Spatial Resolution Plan — W1") parts.append("") # Banner ------------------------------------------------------------------ parts.append("") # Breakdown banner ------------------------------------------------------ parts.append("
" f"breakdown: shot_blocking={metrics['shot_blocking_count']}, " f"design_blocking={metrics['design_blocking_count']}, " f"topology_blocking={metrics['topology_blocking_count']} " f" · total_gap_entries={metrics['total_gap_entries']} " f" · category_breakdown=" f"{esc(json.dumps(metrics['category_breakdown'], ensure_ascii=False))}" "
") # §1 Gap ledger ----------------------------------------------------------- parts.append("

§1. Gap ledger

") parts.append("" "" "" "") for e in ledger.entries: parts.append( f"" f"" f"" f"" f"" f"" ) parts.append("
gap_idcategoryscopeshot_idrollup_group_idsummaryproposed_resolutionblocks
{esc(e.gap_id)}{esc(e.category)}{esc(e.scope)}{esc(e.shot_id or '')}{esc(e.rollup_group_id or '')}{esc(e.summary)}{esc(e.proposed_resolution)}{esc('; '.join(e.blocks or []))}
") # §2 Shot resolution status ---------------------------------------------- parts.append("

§2. Shot resolution status

") parts.append("

각 shot row 우측의" " details 를 펼치면 evidence_pointers + candidate_interpretations" " 을 확인할 수 있습니다 (plan_v2 §5 §2).

") parts.append("" "" "" "" "") sorted_res = sorted( resolutions, key=lambda r: (r.resolution_status != "resolved", r.shot_id), ) for r in sorted_res: cls = "status-resolved" if r.resolution_status == "resolved" else "status-needs" pill_cls = "resolved" if r.resolution_status == "resolved" else "needs" binding = r.base_plate_binding bp_text = f"{binding.cluster_id}/{binding.plate_id}" if binding else "" # evidence_pointers expandable ----------------------------------- ev_section = [] if r.evidence_pointers: ev_section.append("
evidence_pointers" f" ({len(r.evidence_pointers)})") for p in r.evidence_pointers: ev_section.append( f"
" f"" f"[{esc(p.relation)}] " f"{esc(p.quote_source)}: " f"“{esc(p.quote)}” " f"ev_id={esc(p.ev_id or 'missing')} " f"({esc(p.lookup_status)})" "
" ) ev_section.append("
") else: ev_section.append("
evidence_pointers: " "evidence_pointer_count=0
") # candidate_interpretations expandable ----------------------------------- cand_section = [] if r.candidate_interpretations: cand_section.append("
candidate_interpretations" f" ({len(r.candidate_interpretations)})") for opt in r.candidate_interpretations: cand_section.append( f"
" f"{esc(opt.option_id)} " f"strength={esc(opt.strength)} " f"→ {esc(opt.candidate_value)}" f"
{esc(opt.rationale_hint)}" f"
" f"{esc(opt.note_for_decision)}" "
" ) cand_section.append("
") else: cand_section.append("
" "candidate_interpretations: (resolved)" "
") details = ("
view details" + "".join(ev_section) + "".join(cand_section) + "
") parts.append( f"" f"" f"" f"" f"" f"" f"" f"" f"" ) parts.append("
shot_idstatusroom_nodecamera_familystate_overlaymatched_keywordscandidatesbase_plateevidence + candidates
{esc(r.shot_id)}{esc(r.resolution_status)}{esc(r.room_node or '')}{esc(r.camera_family or '')}{esc(r.state_overlay)}{esc(','.join(r.room_node_resolution.get('matched_direct_keywords') or []))}{esc(','.join(r.room_node_candidates or []))}{esc(bp_text)}{details}
") # §3 SetTopologyLock ------------------------------------------------------ parts.append("

§3. SetTopologyLock

") parts.append("

3-1. Nodes

") parts.append("" "" "") for n in topology.nodes: parts.append( f"" f"" f"" f"" ) parts.append("
idkindactive_statusplate_policycontainmentconfidence
{esc(n.id)}{esc(n.kind)}{esc(n.active_status)}{esc(n.plate_policy)}{esc(json.dumps(n.containment, ensure_ascii=False))}{esc(n.confidence)}
") parts.append("

3-2. Edges

") parts.append("" "") for e in topology.edges: parts.append( f"" f"" f"" ) parts.append("
fromtokindconfidenceannotation
{esc(e.from_node)}{esc(e.to_node)}{esc(e.kind)}{esc(e.confidence)}{esc(e.annotation or e.inference_basis or '')}
") parts.append("

3-3. Source hashes

") for k, v in topology.source_hashes.items(): parts.append(f"") parts.append("
{esc(k)}{esc(v)}
") parts.append("

3-4. Frozen unknowns

    ") for u in topology.frozen_unknowns: parts.append(f"
  • {esc(u)}
    • ") parts.append("
") # §4 BasePlateCluster ----------------------------------------------------- parts.append("

§4. BasePlateCluster

") for c in clusters: active_str = "active" if c.active else "inactive (needs_decision)" parts.append(f"

{esc(c.cluster_id)} — anchor={esc(c.topology_anchor_node)} ({active_str})

") parts.append("" "" "" "" "") for m in c.members: bucket_html = "" if m.role == "derived_shot_plate": # plan_v2 §2-C C6: inherited_layout_constraints + emphasize/may_omit/allow. bucket_parts = [] if m.inherited_layout_constraints: items = "".join( f"
  • {esc(it)}
    • " for it in m.inherited_layout_constraints ) bucket_parts.append( "
    inherited_layout_constraints" f"
      {items}
    " ) if m.visible_required_items: for bucket_name in ("emphasize", "may_omit", "allow"): bucket_items = m.visible_required_items.get(bucket_name, []) or [] if bucket_items: items = "".join(f"
  • {esc(b)}
    • " for b in bucket_items) bucket_parts.append( f"
    {bucket_name}" f"
      {items}
    " ) else: bucket_parts.append( f"
    {bucket_name}: " "(empty — W2 i2i 합성 단계에서 채워질 예정)
    " ) bucket_html = "".join(bucket_parts) or "(empty)" else: # master plate — shared_layout_constraints if m.shared_layout_constraints: items = "".join( f"
  • {esc(it)}
    • " for it in m.shared_layout_constraints ) bucket_html = ( "
    shared_layout_constraints" f"
      {items}
    " ) parts.append( f"" f"" f"" f"" f"" f"" f"" ) parts.append("
    plate_idrolecamera_familycamera_anchorcovers_nodesshots_using_this_platederived_from_masterinherited / 3 buckets
    {esc(m.plate_id)}{esc(m.role)}{esc(m.camera_family)}{esc(m.camera_anchor)}{esc(','.join(m.covers_nodes))}{esc(','.join(m.shots_using_this_plate))}{esc(m.derived_from_master or '')}{bucket_html}
    ") # §5 Design items --------------------------------------------------------- parts.append("

    §5. Production design inference

    ") parts.append("

    각 design item 의" " applies_to_nodes 가 어떤 active cluster 의 covers_nodes 와" " 매칭되는지 cluster_match 컬럼으로 표시. unmatched 는" " gap_design_unmatched gap 으로 rollup 됩니다.

    ") parts.append("" "" "" "") for d in design_items: match_label, gap_id = design_cluster_match(d) pill_cls = "cluster" if match_label.startswith("cluster_match") else ( "unmatched" if match_label == "unmatched" else "") match_pill = f"{esc(match_label)}" if pill_cls else esc(match_label) parts.append( f"" f"" f"" f"" f"" f"" ) parts.append("
    item_idcategorydescriptionapplies_to_nodesconfidencederived_fromcluster_matchgap_id
    {esc(d.item_id)}{esc(d.category)}{esc(d.description)}{esc(','.join(d.applies_to_nodes))}{esc(d.confidence)}{esc(d.derived_from)}{match_pill}{esc(gap_id)}
    ") # §6 Reference: existing base plate PNG thumbnails ------------------------ parts.append("

    §6. Reference: existing base plate PNGs (read-only)

    ") parts.append("

    plan_v2 §5 §6: rspp/4b552b" " run 의 4 PNG 는 v1 schema 산출물이므로 v2 master/derived 와" " 직접 매핑되지 않습니다. W2 reuse 가능성 검토용 reference" " 입니다 (Codex Q5 — 권장 = 격리 후 재생성).

    ") parts.append("") reference_paths = [ ("scripts_output/rooftop_spatial_pipeline_plan/20260523_2226_4b552b/" "base_plate_assets/20260523_1956_d22e35/main_room__eye_level_wide.png", "main_room wide — master 후보로 사후 라벨링 검토"), ("scripts_output/rooftop_spatial_pipeline_plan/20260523_2226_4b552b/" "base_plate_assets/20260523_1956_d22e35/main_room__eye_level_table_close.png", "main_room table close — derived 후보로 사후 라벨링 검토"), ("scripts_output/rooftop_spatial_pipeline_plan/20260523_2226_4b552b/" "base_plate_assets/20260523_1956_d22e35/bedroom__eye_level_doorway_wide.png", "bedroom doorway wide — 어느 방인지 미구분 (W2 별도 결정)"), ("scripts_output/rooftop_spatial_pipeline_plan/20260523_2226_4b552b/" "base_plate_assets/20260523_1956_d22e35/bathroom__eye_level_mirror_close.png", "bathroom mirror close — cluster_bathroom master 후보"), ] for path, note in reference_paths: parts.append( f"" ) parts.append("
    pathnote
    {esc(path)}{esc(note)}
    ") parts.append("

    thumbnail 자체는 본 HTML" " 에 직접 임베드하지 않습니다 (이미지 생성/복사 0). 위 path 에서" " 직접 확인하세요.

    ") # §7 Open questions ------------------------------------------------------- parts.append("

    §7. Open questions for user decision

      ") parts.append("
    • 민숙의_방_안방 active? " "(gap_active_minsook_bedroom)
      • ") parts.append("
    • bedroom_unresolved 4 shots binding? " "(gap_room_S12_S4 / S12_S6 / S12_S14 / S25_S7)
      • ") parts.append("
    • manual_review_needed shots binding? " "(gap_camera_S14_S5 / S14_S9 / S18_S5 / S25_S3 등)
      • ") parts.append("
    • plan_v2 §5 keyword whitelist 확장 여부 — 현재 " "1/14 resolved. \"전신\" / \"전경\" / \"구도\" framing 키워드" " 추가가 plan_v3 의 범위인지 W1 patch 의 범위인지 결정 필요.
      • ") parts.append("
    • S18_S5 / S18_S9 의 multi-direct \"욕실+거실\" conflict — " "rule 2 strict 적용 결과 needs_manual_room. shot_description" " 우선 weight 도입 여부.
      • ") parts.append("
    ") parts.append("") return "".join(parts) # ----------------------------------------------------------------------------- # main + CLI # ----------------------------------------------------------------------------- def parse_args() -> argparse.Namespace: ap = argparse.ArgumentParser(description="rooftop spatial resolution W1 (dry-run)") ap.add_argument("--source-run", type=Path, default=_REPO_ROOT / DEFAULT_SOURCE_RUN, help="source grounding run directory") ap.add_argument("--output-root", type=Path, default=_REPO_ROOT / DEFAULT_OUTPUT_DIR, help="output root (run_id directory will be created inside)") ap.add_argument("--no-serve", action="store_true", help="skip optional local webserver (default behavior)") ap.add_argument("--limit-shots", type=int, default=0, help="limit number of shots loaded (0=all). debugging only") return ap.parse_args() def _run_id() -> str: stamp = datetime.now(timezone(timedelta(hours=9))).strftime("%Y%m%d_%H%M") return f"{stamp}_{uuid.uuid4().hex[:6]}" def main() -> None: args = parse_args() source_run = args.source_run if not source_run.is_absolute(): source_run = _REPO_ROOT / source_run output_root = args.output_root if not output_root.is_absolute(): output_root = _REPO_ROOT / output_root bible, source_evidence = _load_bible_and_evidence(source_run) # DB read-only ----------------------------------------------------------- from app.core.database import SessionLocal # noqa: PLC0415 with SessionLocal() as session: shots = load_l05_shots(session) if args.limit_shots > 0: shots = shots[: args.limit_shots] source_hashes = compute_source_hashes(bible, source_evidence, shots) topology = build_set_topology_lock(bible, source_evidence, source_hashes) design_items = build_design_items(bible, source_evidence) resolutions = [resolve_shot_spatial(s, topology, source_evidence) for s in shots] clusters = cluster_base_plates(topology, resolutions) ledger = build_gap_ledger(topology, resolutions, design_items, clusters) run_id = _run_id() out_dir = output_root / run_id run_meta = { "run_id": run_id, "plan_version": PLAN_VERSION, "generated_at": _now_iso(), "source_run": str(source_run), "source_hashes": source_hashes, "shot_count": len(shots), "cluster_count": len(clusters), "active_cluster_count": sum(1 for c in clusters if c.active), "master_plate_count": sum( 1 for c in clusters if c.active for m in c.members if m.role == "master_plate" ), "derived_plate_count": sum( 1 for c in clusters if c.active for m in c.members if m.role == "derived_shot_plate" ), "resolved_count": sum( 1 for r in resolutions if r.resolution_status == "resolved" ), "needs_manual_room_count": sum( 1 for r in resolutions if r.resolution_status == "needs_manual_room" ), "needs_camera_anchor_count": sum( 1 for r in resolutions if r.resolution_status == "needs_camera_anchor" ), "needs_both_count": sum( 1 for r in resolutions if r.resolution_status == "needs_both" ), "design_item_count": len(design_items), "gap_metrics": compute_gap_metrics(ledger), } # Banner-friendly aliases (Codex IMPORTANT 2): expose breakdown. run_meta["blocking_decision_gaps_count"] = ( run_meta["gap_metrics"]["blocking_decision_gaps_count"] ) run_meta["informational_gap_count"] = ( run_meta["gap_metrics"]["informational_gap_count"] ) write_outputs(out_dir, topology, resolutions, clusters, design_items, ledger, run_meta) print(f"[w1] run_id={run_id}") print(f"[w1] out_dir={out_dir}") print( f"[w1] blocking_decision_gaps={run_meta['blocking_decision_gaps_count']} " f"(shot={run_meta['gap_metrics']['shot_blocking_count']}, " f"design={run_meta['gap_metrics']['design_blocking_count']}, " f"topology={run_meta['gap_metrics']['topology_blocking_count']}, " f"informational={run_meta['informational_gap_count']})" ) print(f"[w1] shots_resolved={run_meta['resolved_count']}/{run_meta['shot_count']}") print(f"[w1] master={run_meta['master_plate_count']} / " f"derived={run_meta['derived_plate_count']}") if __name__ == "__main__": main()