"""자동차(좁은 실내) 샷 실험 — A(실사+스케치) vs B(fp 배치도) vs 현행. 설계 SOT: docs/superpowers/specs/2026-08-10-car-interior-realref-vs-fp-experiment-design.md 병목: 현행 판정·계약이 "배경본 유지"를 기준으로 삼는데 배경본 자체가 실제 차 구조와 다르다. 두 안 모두 "구조적으로 옳은 기준 이미지"를 만드는 방법을 겨눈다 — 차이는 기준을 무엇으로 만들었나 뿐이다. A: 검색 실사(그 차종·그 시점) = 구조·표면 기준 → 러프 마네킹 스케치 (실사 참조 i2i 작화) = 인물 배치·시점 → Step2(stage_head_geom). B: 위에서 본 좌석 배치도+카메라 마커(t2i 작화) → 배경본 → 같은 Step2 (스케치 없이 — 배치도는 시점 참조가 아니다). 판정: [A최종, B최종, 현행 sel] 3후보 — 실험 계약(opus_pilot SELECT_SYS, 배경본 유지 축 없음) · G+Q 이중 · 제시 순서 뒤집어 2회(flip) · 두 번 다 이긴 후보만 승. 최종 판단은 육안 갤러리. 프로덕션 코드·recipe·DB 무변경. 모든 호출은 기존 계측 경로로 기록된다. 사용: .venv/bin/python car_interior_ab_experiment.py [--stems S13sh3,...] """ from __future__ import annotations import argparse import json import re import sys import time import urllib.request from pathlib import Path from typing import Any, Dict, List, Optional, Tuple sys.path.insert(0, str(Path(__file__).resolve().parent)) from app.modules.llm.llm_client import call_structured # noqa: E402 from app.modules.llm.qwen_vlm_client import ask_qwen_structured # noqa: E402 from app.modules.pipeline.multiroll_gemini import ( # noqa: E402 make_nb2_gen_fn, ) from app.modules.pipeline.search_grounded_ref import ( # noqa: E402 search_reference_images, ) from app.modules.pipeline.still_recipe import ( # noqa: E402 build_bgfirst_final_prompt, build_bgfirst_refs, ) from app.modules.prompt_sanitizer import PromptSanitizer # noqa: E402 from canary_production import resolve_refs, size_index # noqa: E402 from opus_pilot import SELECT_SYS, select_schema # noqa: E402 PROJ = "e716bafb-24bb-42b7-aea0-fdb383844ee8" EPI = "d6a9aa85-b75e-400c-980c-4ee7e876a15b" ROOT = Path(__file__).resolve().parent.parent RECIPE = ROOT / f"projects/{PROJ}/images/{EPI}/scene/recipe" OUT = ROOT / "artifact/20260810_car_interior_ab" DEFAULT_STEMS = ["S13sh3", "S15sh5", "S18sh1"] ENTITY_LABEL_PREFIXES = ("CHARACTER REFERENCE", "PROP REFERENCE") # ── 실험 전용 계약 문구 (범용 — 차종·구조·언어는 전부 브리프 데이터에서) ── SEARCH_AUTHOR_SYS = """\ You prepare a web image search for ONE photograph need: the real interior of the specific vehicle type named in the LOCATION text, seen from the specific seat/viewpoint the CAMERA text describes. Search ONLY for what is special here (the vehicle type + that seat's viewpoint). Do not add generic words. One need = one directive. Write `directive_native`, `terms_native` (1-2 short queries) and `language_lock_native` (one sentence forbidding any other language in search queries) ALL in the language of the SHOT TEXT — the work's native language. The lock sentence goes first when searching, so make it blunt.""" def search_author_schema() -> Dict[str, Any]: return { "type": "object", "additionalProperties": False, "properties": { "directive_native": {"type": "string"}, "terms_native": {"type": "array", "items": {"type": "string"}, "minItems": 1, "maxItems": 2}, "language_lock_native": {"type": "string"}, }, "required": ["directive_native", "terms_native", "language_lock_native"], } SCENE_SEARCH_AUTHOR_SYS = """\ You prepare a web image search for ONE photograph need: a real photograph of a SIMILAR SCENE to the shot described below — the same kind of vehicle interior, seen from the same seat/viewpoint the CAMERA text describes, WITH ordinary people actually occupying the same seats the SHOT TEXT places people in. A photo where the composition (who sits where, where the camera stands) is already readable. Ordinary, everyday people in the photo are wanted — they will be replaced later; their seating and the cabin around them are what we need. Do not search for empty interiors. Vehicles differ by country — steering wheel side included. Do not name a side. Search in the language of the SHOT TEXT (the work's native language) so the results come from the work's own country, and let the photographs decide such details. Write `directive_native`, `terms_native` (1-2 short queries) and `language_lock_native` (one sentence forbidding any other language in search queries) ALL in that language. The lock sentence goes first.""" C_RESTAGE_HEAD = """\ Restage the FIRST attached photograph as one 16:9 photorealistic film still for the brief below. The first photograph is the ONLY authority for physical layout: the camera's seat, which seat each person occupies, which side the steering wheel is on, and the cabin structure all come from that photograph and nothing else. If any sentence below seems to place a person or the steering wheel elsewhere on screen, the photograph wins — keep its geometry and let the framing fall where the photograph's viewpoint puts it. Replace every person in it with the characters the SHOT TEXT and the attached CHARACTER/PROP references define — same seats as the photograph, the action and expression the SHOT TEXT requires. No person from the original photograph may remain visible. Relight and stage the scene to match the LOCATION, TIME and LIGHTING texts. No text, no watermarks.""" FP_PRECISE_HEAD = """\ Draw ONE precise top-down floor-plan diagram of the enclosed, control-heavy interior the LOCATION text below describes — whatever it is: a road-vehicle cabin, a cockpit, a wheelhouse, an armored crew bay, a bus, an aircraft, a machine's operator station. This diagram is read by a machine and never appears in any final image, so precision beats style. It must clearly show, with distinct shapes and short labels: - every seat / crew station, in the correct layout for this kind of space as the LOCATION text describes it - every primary control the space has (steering wheel, yoke, control stick, helm, levers, consoles — only what this space actually has), each drawn attached at the exact station that operates it. Controls whose side depends on the country (like a road vehicle's driving position) follow what is correct in the work's country — the LOCATION text's language tells you the country. - any mirrors or reflective surfaces the texts name, each with a small arrow showing the direction its reflective face points - the camera position and a clear arrow for its viewing direction, exactly as the CAMERA text describes - one labelled marker on each seat/station a person occupies per the SHOT TEXT Clean diagram lines on white, top-down only, no perspective, no photorealism.""" FP_READBACK_SYS = """\ You are looking at a top-down floor-plan diagram of an enclosed, control-heavy interior (a cabin, cockpit, wheelhouse, crew bay — as the texts below describe). FIRST read it literally: which station each primary control is attached to, where any mirrors/reflective surfaces are and which way their faces point, where the camera stands and which way it looks, and which seats/stations are marked as occupied. Report those in `reads`. THEN compare your reading against the LOCATION / CAMERA / SHOT TEXT below. List every disagreement in `mismatches` (empty if none). FINALLY, write `scene_description_en`: 4-7 sentences describing exactly what this camera sees, derived ONLY from the diagram — for each element (each person, each primary control, each mirror/reflective surface, windows/openings), say WHERE it appears on screen (left / right / center, near / far) and which way it faces. Name who occupies which seat/station. Be explicit about any mirror: where it sits on screen and what its face can physically reflect from this camera. This text will drive image generation, so every left/right must be from the CAMERA's point of view.""" def fp_readback_schema() -> Dict[str, Any]: return { "type": "object", "additionalProperties": False, "properties": { "reads": { "type": "object", "additionalProperties": False, "properties": { "steering_seat": {"type": "string"}, "mirror": {"type": "string"}, "camera": {"type": "string"}, "occupants": {"type": "string"}, }, "required": ["steering_seat", "mirror", "camera", "occupants"], }, "mismatches": {"type": "array", "items": {"type": "string"}}, "scene_description_en": {"type": "string"}, }, "required": ["reads", "mismatches", "scene_description_en"], } # ── v2 (08-11 육안 20샷 후): "정밀"이 세밀 렌더링으로 새서 구도가 # 무너졌다. fp=간략하지만 구도 정확 — 잘못을 한눈에 알아보는 것이 목적. # 같은 공간 샷들은 base 도면 1장을 공유하고 샷별 마커만 얹는다. ── BASE_FP_HEAD_V2 = """\ Draw ONE SIMPLE top-down layout diagram of the interior the LOCATION text below describes — like a stage-blocking chart, NOT a picture. Its only job is to make layout mistakes easy to spot at a glance. Show ONLY, as flat schematic shapes on white paper: - the seat/station arrangement: plain outlined boxes in their true relative positions, sensible proportions, each with a short label - the primary control of this space, drawn as one simple symbol attached to the exact station that operates it. Controls whose side depends on the country (like a road vehicle's driving position) follow what is correct in the work's country — the LOCATION text's language tells you the country. - a mirror symbol with a small facing arrow ONLY if the CAMERA or SHOT texts rely on a mirror NOTHING else. No doors, no dashboard details, no gear levers, no decoration, no textures, no 3D, no perspective view, no photograph, no picture-in-picture, no sketch shading. If an element is obvious and hard to get wrong, leave it out — simplicity beats completeness.""" SHOT_FP_MARK_HEAD_V2 = """\ Edit the attached top-down layout diagram for ONE specific shot. ADD only these markers and change nothing else in the diagram: - one small labelled figure marker on each seat/station a person occupies per the SHOT TEXT below - the camera: a small symbol at its position and a clear arrow for its viewing direction, per the CAMERA text - if the SHOT TEXT aims something directional (a weapon, a pointed object), one small arrow showing where it points Do not redraw or move seats, controls or mirrors. Keep the same flat schematic style — no detail, no perspective, no photo.""" FP_FIX_HEAD = """\ Edit the attached top-down floor-plan diagram. Correct ONLY the problems listed below — move the markers they name to where the text says they belong. Keep every other seat, control, mirror and label exactly where it is. Same clean top-down diagram style, no perspective, no photorealism.""" D_FROM_FP_HEAD = """\ Create ONE 16:9 photorealistic film still for the brief below. The FIRST attached image is a top-down FLOOR PLAN of this interior and the SCENE LAYOUT text below is what a careful reader saw in it. Together they are the ONLY authority for physical arrangement: which seat/station each person occupies, which station every primary control belongs to, where any mirror/reflective surface sits and what it can physically reflect, where the camera stands and what appears on which side of the screen. If any other sentence seems to contradict them, the floor plan wins. The floor plan is a diagram, not scenery — none of its lines, arrows or labels may appear in the photograph. WHO the people are and what they do comes from the SHOT TEXT and the attached CHARACTER/PROP references — never add a person the SHOT TEXT does not place here. No text, no watermarks.""" BG_FROM_PHOTO_HEAD = """\ Create ONE empty 16:9 photorealistic background: the vehicle interior seen exactly from the viewpoint the CAMERA text below requires. The attached reference photograph(s) show the REAL interior of this kind of vehicle. They are the AUTHORITY for structure and surfaces: how many steering wheels exist (one), where seats, mirrors, pillars and controls sit, what the materials look like. Keep that structure true. Do NOT copy the reference's lighting or its exact scenery — relight and stage it to match the LOCATION and LIGHTING texts. NO people, NO text, NO watermarks. An empty cabin only.""" BG_FROM_DIAGRAM_HEAD = """\ Create ONE empty 16:9 photorealistic background: the vehicle interior seen exactly from the camera position marked on the attached seating diagram and described in the CAMERA text below. The attached image is a top-down SEATING DIAGRAM, not a photograph. Use it only to understand where seats and the camera are; do not let any of its lines, labels or diagram style appear in the photograph. Build a physically correct interior: exactly one steering wheel where the layout puts it, seats and mirrors where they belong for this vehicle type. NO people, NO text, NO watermarks. An empty cabin only.""" DIAGRAM_HEAD = """\ Draw ONE simple top-down seating diagram of the vehicle interior the LOCATION text below describes: rough hand-drawn lines on white paper, like a quick production sketch. Show the seat layout of this vehicle type, the steering wheel position, and mark the camera with a small wedge showing where it stands and which way it faces (per the CAMERA text). Keep it rough and schematic — no rendering, no realism, no shading, no readable text labels.""" SKETCH_HEAD = """\ Redraw the attached interior photograph as a ROUGH pencil sketch on white paper: loose gray lines only, no shading, no rendering detail. Keep the interior's structure lines faint. Then add simple gray MANNEQUIN figures for the people the SHOT TEXT below places in this space — right seats, right facing, right scale. Mannequins only — no faces, no clothing detail.""" import re as _re _HEADER_RE = _re.compile(r"^[A-Z][A-Z &/()'-]{4,}") def extract_sections(prompt: str) -> Dict[str, str]: """브리프에서 고정 헤더 절 발췌 — 팩이 만든 헤더 문구의 형식 소비. 없으면 빈 문자열(브리프 구성은 샷마다 다르다 — fail-open 이 맞다: 발췌는 생성 입력을 좁히는 용도이고 전문은 Step2 가 그대로 쓴다). carried/people (08-11): 인물이 어느 자리에 있는지의 원천 절 — fp 샷 마커 입력에 없으면 뒷좌석 인물이 도면에서 빠진다(실측). 두 절은 헤더 줄부터 다음 대문자 헤더 전까지의 블록. """ lines = prompt.split("\n") out = {"shot_text": "", "location": "", "camera": "", "lighting": "", "carried": "", "people": ""} cam: List[str] = [] in_cam = False block_key = "" block: List[str] = [] def flush_block(): nonlocal block_key, block if block_key: out[block_key] = "\n".join(block).strip() block_key, block = "", [] for ln in lines: s = ln.strip() if block_key and _HEADER_RE.match(s): flush_block() if s.startswith("SHOT TEXT"): out["shot_text"] = s elif s.startswith("LOCATION"): out["location"] = s elif s.startswith("CARRIED STATE"): flush_block() block_key, block = "carried", [s] continue elif s.startswith("PEOPLE:"): flush_block() block_key, block = "people", [s] continue elif s.startswith("CAMERA & FRAME"): in_cam = True cam.append(s) continue elif s.startswith("- LIGHTING & MOOD"): out["lighting"] = s in_cam = False elif in_cam: if s.startswith("Compose the frame"): cam.append(s) in_cam = False else: cam.append(s) elif block_key: block.append(ln) flush_block() out["camera"] = "\n".join(cam) return out def strip_camera_block(prompt: str) -> str: """브리프에서 CAMERA & FRAME 블록만 뺀 전문 — C2 재연출·판정용. 2026-08-10 확인: 모든 경로(현행 포함)가 남자를 조수석에 그렸다 — 화면 배치 문자("운전대는 오른쪽")가 핸들 소속 좌석의 결정론 없이 실리면 모델이 우핸들 배치로 수렴한다. 구도 권위는 장면 실사 사진 하나로 좁힌다(핸들 방향은 여전히 어디에도 안 적는다 — 사진이 정함). """ out, in_cam = [], False for ln in prompt.split("\n"): s = ln.strip() if s.startswith("CAMERA & FRAME"): in_cam = True continue if in_cam: if s.startswith("Compose the frame"): in_cam = False continue out.append(ln) return "\n".join(out) def entity_refs(rec: Dict[str, Any], idx) -> List[Tuple[str, Any]]: """record.refs 에서 엔티티(인물·소품 정본)만 — 팩 고정 라벨 접두.""" resolved = resolve_refs(rec.get("refs"), idx) return [(lab, src) for lab, src in resolved if str(lab).startswith(ENTITY_LABEL_PREFIXES)] def download_photos(images: List[Dict[str, Any]], dest_stem: Path, want: int = 2) -> List[Path]: got: List[Path] = [] for i, im in enumerate(images): if len(got) >= want: break for key in ("image_url", "thumbnail_url"): u = im.get(key) if not u: continue try: req = urllib.request.Request( u, headers={"User-Agent": "Mozilla/5.0"}) data = urllib.request.urlopen(req, timeout=30).read() if len(data) < 8_000: # 아이콘·깨진 응답 걸러내기 continue p = dest_stem.with_name( f"{dest_stem.name}_search{len(got) + 1}.png") p.write_bytes(data) got.append(p) break except Exception as exc: # noqa: BLE001 print(f" 사진 다운로드 실패({key}): {exc}") return got def judge_three(tag: str, brief: str, cands: Dict[str, Path], ) -> Dict[str, Any]: """3후보 실험 판정 — G+Q × 제시 순서 flip 2회. 라벨→후보 키 역매핑.""" from app.modules.pipeline.multiroll_gemini import png_part keys = list(cands) # ["A안", "B안", "현행"] orders = [keys, list(reversed(keys))] votes: Dict[str, Dict[str, str]] = {} detail: Dict[str, Any] = {} for oi, order in enumerate(orders): labels = ["A", "B", "C"][: len(order)] parts: List[Dict[str, Any]] = [{ "type": "text", "text": "THE BRIEF (every candidate was made from this):\n" + brief, }] for lab, key in zip(labels, order): parts.append({"type": "text", "text": f"Candidate {lab}:"}) parts.append(png_part(cands[key])) lab2key = dict(zip(labels, order)) for model in ("gemini", "qwen"): jtag = f"car_ab_judge_{tag}_o{oi}_{model}" try: if model == "gemini": data = call_structured( jtag, SELECT_SYS, parts, select_schema(labels), project_config={jtag: {"model": "gemini-pro"}}, schema_name="car_ab_judge", opik_metadata={"operation_type": "car_ab_judge", "shot_tag": tag, "project_id": PROJ, "episode_id": EPI}, enable_fallback=False) else: data = ask_qwen_structured( jtag, SELECT_SYS, parts, select_schema(labels), opik_metadata={"operation_type": "car_ab_judge", "shot_tag": tag, "project_id": PROJ, "episode_id": EPI}) except Exception as exc: # noqa: BLE001 detail[f"o{oi}_{model}"] = {"error": repr(exc)[:300]} continue w = lab2key.get(str(data.get("winner"))) votes.setdefault(model, {})[f"o{oi}"] = w detail[f"o{oi}_{model}"] = { "winner_key": w, "why": data.get("why"), "readings": data.get("readings"), "all_candidates_fail": data.get("all_candidates_fail"), } # 두 번 다 같은 후보를 뽑은 심판만 확정표 — 아니면 무승부(비결정) agreed = {m: v["o0"] for m, v in votes.items() if v.get("o0") and v.get("o0") == v.get("o1")} return {"votes": votes, "agreed": agreed, "detail": detail} def run_c2_only(stems: List[str]) -> None: """C2 전용 흐름 — 장면 검색 → C2 재연출 → C2 vs 현행 판정. 육안 탈락이 확정된 A/B/C1 경로를 만들지 않는다(샷당 이미지 5장·판정 4회 절약). 기존 main 흐름은 무변경 — S13sh3 기록과의 비교를 위해 results.json 의 같은 키(scene_search/judge_c2)를 쓴다. """ OUT.mkdir(parents=True, exist_ok=True) records = json.loads((RECIPE / "records.json").read_text("utf-8")) idx = size_index() gen = make_nb2_gen_fn( project_id=PROJ, episode_id=EPI, operation_type="car_ab_experiment", sanitizer=PromptSanitizer(project_config=None)) res_path = OUT / "results.json" results: Dict[str, Any] = ( json.loads(res_path.read_text("utf-8")) if res_path.exists() else {}) def save() -> None: res_path.write_text( json.dumps(results, ensure_ascii=False, indent=1), "utf-8") for tag in stems: rec = records.get(tag) cur_sel = RECIPE / f"{tag}_sel.png" if not rec or not cur_sel.exists(): print(f"{tag}: record/현행 최종 없음 — 건너뜀") continue print(f"\n=== {tag} (C2 전용) ===") r = results.setdefault(tag, {}) brief = rec.get("prompt", "") sec = extract_sections(brief) ctx = "\n\n".join(x for x in ( sec["shot_text"], sec["location"], sec["camera"], sec["lighting"]) if x) ents = entity_refs(rec, idx) stem_out = OUT / tag try: if "scene_search" not in r: author_c = call_structured( f"car_ab_scene_search_author_{tag}", SCENE_SEARCH_AUTHOR_SYS, f"SHOT TEXT / LOCATION / CAMERA:\n{ctx}", search_author_schema(), project_config={f"car_ab_scene_search_author_{tag}": {"model": "gemini-pro"}}, schema_name="car_ab_scene_search_author", opik_metadata={"operation_type": "car_ab_experiment", "shot_tag": tag, "project_id": PROJ, "episode_id": EPI}, enable_fallback=False) from app.core.openai_keys import openai_client sr_c = search_reference_images( openai_client(), directive_native=author_c["directive_native"], terms_native=author_c["terms_native"], language_lock_native=author_c["language_lock_native"]) r["scene_search"] = {"author": author_c, "queries": sr_c.get("queries"), "n_images": len(sr_c.get("images") or [])} photos_c = download_photos( sr_c.get("images") or [], stem_out.with_name(f"{tag}_C")) r["scene_search"]["downloaded"] = [str(p) for p in photos_c] save() photos_c = [Path(p) for p in r["scene_search"].get("downloaded") or [] if Path(p).exists()] if not photos_c: raise RuntimeError("장면 실사 회수 0 — 중단") brief_nc = strip_camera_block(brief) c2_final = stem_out.with_name(f"{tag}_C2_final.png") if not c2_final.exists(): gen(f"car_ab_{tag}_C2_final", C_RESTAGE_HEAD + "\n\n" + brief_nc, [("PHOTOGRAPH TO RESTAGE — sole physical-layout " "authority (not identity)", photos_c[0]), *ents], c2_final) if "judge_c2" not in r: r["judge_c2"] = judge_three( f"{tag}_c2", brief_nc, {"C2안": c2_final, "현행": cur_sel}) save() print(f" C2 판정 확정표: {r['judge_c2'].get('agreed')}") except Exception as exc: # noqa: BLE001 r["오류"] = f"{type(exc).__name__}: {exc}"[:300] print(f" 실패 — {r['오류']}") save() continue r.pop("오류", None) save() print(f"\n결과: {res_path}") def run_fp_readback(stems: List[str]) -> None: """D안 — 정밀 fp 작화 → VLM readback(검증+장면 설명) → 설명 기반 생성. 2026-08-10 사용자 지시(S15sh5 룸미러 반대 육안 후): fp 를 룸미러· 의자·핸들 방향까지 정밀하게 만들고, **장면 설명을 fp 를 다시 보고 작성**해 그 텍스트를 배치 권위로 생성한다. fp 는 최종 생성에 첨부하지 않는다(설명 텍스트가 운반 — 도면 화풍 오염·도면 추종 실패 둘 다 회피). readback 이 브리프와 어긋남을 보고하면 fp 를 1회 재작화한다(어긋남 목록을 지시에 붙여). """ from app.modules.pipeline.multiroll_gemini import png_part OUT.mkdir(parents=True, exist_ok=True) records = json.loads((RECIPE / "records.json").read_text("utf-8")) idx = size_index() gen = make_nb2_gen_fn( project_id=PROJ, episode_id=EPI, operation_type="car_ab_experiment", sanitizer=PromptSanitizer(project_config=None)) res_path = OUT / "results.json" results: Dict[str, Any] = ( json.loads(res_path.read_text("utf-8")) if res_path.exists() else {}) def save() -> None: res_path.write_text( json.dumps(results, ensure_ascii=False, indent=1), "utf-8") def gpt_fp_gen(prompt: str, out_path: Path, ref: Optional[Path] = None) -> None: """fp 작화·수정 — gpt-image-2 (2026-08-10 사용자 지시: gemini 가 fp 의 핸들 위치를 반복해서 틀림). 프로덕션 fp(location_floor_plan) 도 같은 부품으로 그린다 — 관례 정합. 기록은 부품에 내장.""" from app.core.config import settings from app.core.openai_keys import openai_client from app.modules.llm.gpt_image_primitive import call_gpt_image_bytes client = openai_client(timeout=float(settings.llm_timeout_image_gen)) png = call_gpt_image_bytes( client, mode="edit" if ref else "generate", prompt=prompt, ref_paths=[ref] if ref else None, call_kwargs={"model": settings.openai_image_model, "size": "1536x1024", "quality": "high"}, capture_role="car_ab_fp") if not png: raise RuntimeError("gpt fp 생성 빈 응답") out_path.write_bytes(png) def readback(tag: str, fp_path: Path, ctx: str) -> Dict[str, Any]: return call_structured( f"car_ab_fp_readback_{tag}", FP_READBACK_SYS, [{"type": "text", "text": "LOCATION / CAMERA / SHOT TEXT:\n" + ctx}, {"type": "text", "text": "THE FLOOR-PLAN DIAGRAM:"}, png_part(fp_path)], fp_readback_schema(), project_config={f"car_ab_fp_readback_{tag}": {"model": "gemini-pro"}}, schema_name="car_ab_fp_readback", opik_metadata={"operation_type": "car_ab_experiment", "shot_tag": tag, "project_id": PROJ, "episode_id": EPI}, enable_fallback=False) for tag in stems: rec = records.get(tag) cur_sel = RECIPE / f"{tag}_sel.png" if not rec or not cur_sel.exists(): print(f"{tag}: record/현행 최종 없음 — 건너뜀") continue print(f"\n=== {tag} (D: fp readback) ===") r = results.setdefault(tag, {}) brief = rec.get("prompt", "") sec = extract_sections(brief) ctx = "\n\n".join(x for x in ( sec["shot_text"], sec["location"], sec["camera"], sec["lighting"]) if x) ents = entity_refs(rec, idx) stem_out = OUT / tag try: # D2 (2026-08-10 사용자 정정 "fp 도 보여줘야지"): ①어긋난 # fp 는 t2i 재작화(같은 실수 반복 실측) 대신 readback 지적을 # 지시로 한 i2i 국소 수정 ②최종 생성에 fp 를 **첨부**한다 # (도면+설명이 함께 배치 권위 — 잔류 금지 라벨). fp_path = stem_out.with_name(f"{tag}_D2_fp.png") if "fp2" not in r: base_fp = stem_out.with_name(f"{tag}_D_fp.png") if not base_fp.exists(): gpt_fp_gen(FP_PRECISE_HEAD + "\n\n" + ctx, base_fp) rb = readback(f"{tag}_d2", base_fp, ctx) fixed = False if rb.get("mismatches"): print(f" fp 어긋남 {len(rb['mismatches'])}건 — " "i2i 국소 수정") gpt_fp_gen( FP_FIX_HEAD + "\n- " + "\n- ".join(rb["mismatches"]) + "\n\n" + ctx, fp_path, ref=base_fp) rb = readback(f"{tag}_d2fix", fp_path, ctx) fixed = True else: import shutil as _sh _sh.copy(base_fp, fp_path) r["fp2"] = {**rb, "fixed": fixed} save() scene_desc = r["fp2"]["scene_description_en"] print(f" fp 어긋남(수정 후): " f"{len(r['fp2'].get('mismatches') or [])}건") d_final = stem_out.with_name(f"{tag}_D2_final.png") if not d_final.exists(): gen(f"car_ab_{tag}_D2_final", D_FROM_FP_HEAD + "\n\nSCENE LAYOUT (what a careful reader saw in " "the attached floor plan):\n" + scene_desc + "\n\n" + strip_camera_block(brief), [("FLOOR PLAN — layout authority, a diagram, never " "scenery", fp_path), *ents], d_final) if "judge_d2" not in r: r["judge_d2"] = judge_three( f"{tag}_d2", strip_camera_block(brief), {"D2안": d_final, "현행": cur_sel}) save() print(f" D2 판정 확정표: {r['judge_d2'].get('agreed')}") except Exception as exc: # noqa: BLE001 r["오류"] = f"{type(exc).__name__}: {exc}"[:300] print(f" 실패 — {r['오류']}") save() continue r.pop("오류", None) save() print(f"\n결과: {res_path}") def run_shared_fp(stems: List[str]) -> None: """v2 — 같은 공간 샷들이 base fp 1장을 공유 (08-11 육안 후 재설계). ①base 도면 1장: 간략 구도(좌석 배열+조작 장치+필요시 미러)만 — 같은 공간인데 샷마다 다른 차를 그리던 결함 제거. stems 전부가 같은 공간이라는 것은 호출자가 보장한다(실험 한정 — 프로덕션은 장소 그룹 키로 캐시). ②샷별: base 에 카메라·인물·방향성 소품 마커만 i2i 로 얹는다. ③readback 검증→생성은 기존과 동일. 결과 키 fp3/judge_d3. """ from app.modules.pipeline.multiroll_gemini import png_part # noqa: F401 OUT.mkdir(parents=True, exist_ok=True) records = json.loads((RECIPE / "records.json").read_text("utf-8")) idx = size_index() gen = make_nb2_gen_fn( project_id=PROJ, episode_id=EPI, operation_type="car_ab_experiment", sanitizer=PromptSanitizer(project_config=None)) res_path = OUT / "results.json" results: Dict[str, Any] = ( json.loads(res_path.read_text("utf-8")) if res_path.exists() else {}) def save() -> None: res_path.write_text( json.dumps(results, ensure_ascii=False, indent=1), "utf-8") def gpt_fp_gen(prompt: str, out_path: Path, ref: Optional[Path] = None) -> None: from app.core.config import settings from app.core.openai_keys import openai_client from app.modules.llm.gpt_image_primitive import call_gpt_image_bytes client = openai_client(timeout=float(settings.llm_timeout_image_gen)) png = call_gpt_image_bytes( client, mode="edit" if ref else "generate", prompt=prompt, ref_paths=[ref] if ref else None, call_kwargs={"model": settings.openai_image_model, "size": "1536x1024", "quality": "high"}, capture_role="car_ab_fp_v2") if not png: raise RuntimeError("gpt fp 생성 빈 응답") out_path.write_bytes(png) def readback(tag: str, fp_path: Path, ctx: str) -> Dict[str, Any]: return call_structured( f"car_ab_fp_readback_{tag}", FP_READBACK_SYS, [{"type": "text", "text": "LOCATION / CAMERA / SHOT TEXT:\n" + ctx}, {"type": "text", "text": "THE FLOOR-PLAN DIAGRAM:"}, png_part(fp_path)], fp_readback_schema(), project_config={f"car_ab_fp_readback_{tag}": {"model": "gemini-pro"}}, schema_name="car_ab_fp_readback", opik_metadata={"operation_type": "car_ab_experiment", "shot_tag": tag, "project_id": PROJ, "episode_id": EPI}, enable_fallback=False) # ── base 도면 1장 — 첫 샷의 LOCATION 절이 공간 대표 ── first_rec = records.get(stems[0]) or {} first_sec = extract_sections(first_rec.get("prompt", "")) base_fp = OUT / f"sharedfp_{stems[0]}.png" if not base_fp.exists(): gpt_fp_gen(BASE_FP_HEAD_V2 + "\n\n" + first_sec["location"], base_fp) print(f"base fp 생성: {base_fp.name}") for tag in stems: rec = records.get(tag) cur_sel = RECIPE / f"{tag}_sel.png" if not rec or not cur_sel.exists(): print(f"{tag}: record/현행 최종 없음 — 건너뜀") continue print(f"\n=== {tag} (v2 공유 fp) ===") r = results.setdefault(tag, {}) brief = rec.get("prompt", "") sec = extract_sections(brief) ctx = "\n\n".join(x for x in ( sec["shot_text"], sec["carried"], sec["people"], sec["location"], sec["camera"], sec["lighting"]) if x) ents = entity_refs(rec, idx) stem_out = OUT / tag try: fp_path = stem_out.with_name(f"{tag}_D3_fp.png") if "fp3" not in r: # 인물 배치 절(CARRIED STATE·PEOPLE) 동반 (08-11 지시) — # 없으면 뒷좌석 인물 마커가 도면에서 빠진다(실측). mark_ctx = "\n\n".join(x for x in ( sec["shot_text"], sec["carried"], sec["people"], sec["camera"]) if x) gpt_fp_gen( SHOT_FP_MARK_HEAD_V2 + "\n\n" + mark_ctx, fp_path, ref=base_fp) rb = readback(f"{tag}_d3", fp_path, ctx) fixed = False if rb.get("mismatches"): print(f" fp 어긋남 {len(rb['mismatches'])}건 — " "i2i 국소 수정") gpt_fp_gen( FP_FIX_HEAD + "\n- " + "\n- ".join(rb["mismatches"]) + "\n\n" + ctx, fp_path, ref=fp_path) rb = readback(f"{tag}_d3fix", fp_path, ctx) fixed = True r["fp3"] = {**rb, "fixed": fixed} save() scene_desc = r["fp3"]["scene_description_en"] print(f" fp 어긋남(최종): " f"{len(r['fp3'].get('mismatches') or [])}건") d_final = stem_out.with_name(f"{tag}_D3_final.png") if not d_final.exists(): gen(f"car_ab_{tag}_D3_final", D_FROM_FP_HEAD + "\n\nSCENE LAYOUT (what a careful reader saw in " "the attached floor plan):\n" + scene_desc + "\n\n" + strip_camera_block(brief), [("FLOOR PLAN — layout authority, a diagram, never " "scenery", fp_path), *ents], d_final) if "judge_d3" not in r: r["judge_d3"] = judge_three( f"{tag}_d3", strip_camera_block(brief), {"D3안": d_final, "현행": cur_sel}) save() print(f" D3 판정 확정표: {r['judge_d3'].get('agreed')}") except Exception as exc: # noqa: BLE001 r["오류"] = f"{type(exc).__name__}: {exc}"[:300] print(f" 실패 — {r['오류']}") save() continue r.pop("오류", None) save() print(f"\n결과: {res_path}") def main() -> None: ap = argparse.ArgumentParser() ap.add_argument("--stems", default=",".join(DEFAULT_STEMS)) # C2 확장 실행용 — 육안 탈락이 확정된 A/B/C1 경로를 건너뛰고 # 장면 검색→C2 재연출→판정만 돈다 (2026-08-10 확장 지시). ap.add_argument("--c2-only", action="store_true") # D안 — 정밀 fp → readback → 장면 설명 기반 생성. ap.add_argument("--fp-only", action="store_true") # v2 — 같은 공간 공유 base fp + 샷별 마커 (stems 전부 같은 공간 전제) ap.add_argument("--shared-fp", action="store_true") args = ap.parse_args() if args.shared_fp: run_shared_fp( [s.strip() for s in args.stems.split(",") if s.strip()]) return if args.c2_only: run_c2_only([s.strip() for s in args.stems.split(",") if s.strip()]) return if args.fp_only: run_fp_readback( [s.strip() for s in args.stems.split(",") if s.strip()]) return stems = [s.strip() for s in args.stems.split(",") if s.strip()] OUT.mkdir(parents=True, exist_ok=True) records = json.loads((RECIPE / "records.json").read_text("utf-8")) idx = size_index() gen = make_nb2_gen_fn( project_id=PROJ, episode_id=EPI, operation_type="car_ab_experiment", sanitizer=PromptSanitizer(project_config=None)) results: Dict[str, Any] = {} res_path = OUT / "results.json" if res_path.exists(): results = json.loads(res_path.read_text("utf-8")) def save() -> None: res_path.write_text( json.dumps(results, ensure_ascii=False, indent=1), "utf-8") for tag in stems: rec = records.get(tag) cur_sel = RECIPE / f"{tag}_sel.png" if not rec or not cur_sel.exists(): print(f"{tag}: record/현행 최종 없음 — 건너뜀") continue print(f"\n=== {tag} ===") r = results.setdefault(tag, {}) brief = rec.get("prompt", "") sec = extract_sections(brief) ctx = "\n\n".join(x for x in ( sec["shot_text"], sec["location"], sec["camera"], sec["lighting"]) if x) ents = entity_refs(rec, idx) print(f" 발췌 절 {sum(bool(v) for v in sec.values())}/4 · " f"엔티티 참조 {len(ents)}개") stem_out = OUT / tag try: # ── A안 ────────────────────────────────────────────── if "search" not in r: author = call_structured( f"car_ab_search_author_{tag}", SEARCH_AUTHOR_SYS, f"SHOT TEXT / LOCATION / CAMERA:\n{ctx}", search_author_schema(), project_config={f"car_ab_search_author_{tag}": {"model": "gemini-pro"}}, schema_name="car_ab_search_author", opik_metadata={"operation_type": "car_ab_experiment", "shot_tag": tag, "project_id": PROJ, "episode_id": EPI}, enable_fallback=False) from app.core.openai_keys import openai_client sr = search_reference_images( openai_client(), directive_native=author["directive_native"], terms_native=author["terms_native"], language_lock_native=author["language_lock_native"]) r["search"] = {"author": author, "queries": sr.get("queries"), "n_images": len(sr.get("images") or [])} photos = download_photos( sr.get("images") or [], stem_out) r["search"]["downloaded"] = [str(p) for p in photos] save() photos = [Path(p) for p in r["search"].get("downloaded") or [] if Path(p).exists()] if not photos: raise RuntimeError("A안: 실사 사진 회수 0 — 중단") a_bg = stem_out.with_name(f"{tag}_A_bg.png") if not a_bg.exists(): gen(f"car_ab_{tag}_A_bg", BG_FROM_PHOTO_HEAD + "\n\n" + ctx, [(f"REAL INTERIOR PHOTOGRAPH {i+1} — structure and " "surface authority", p) for i, p in enumerate(photos)], a_bg) a_sketch = stem_out.with_name(f"{tag}_A_sketch.png") if not a_sketch.exists(): gen(f"car_ab_{tag}_A_sketch", SKETCH_HEAD + "\n\n" + sec["shot_text"], [("INTERIOR PHOTOGRAPH to redraw", a_bg)], a_sketch) a_final = stem_out.with_name(f"{tag}_A_final.png") if not a_final.exists(): gen(f"car_ab_{tag}_A_final", build_bgfirst_final_prompt(brief, geom_authority=True), build_bgfirst_refs( bg=a_bg, conti=a_sketch, char_refs=ents, prop_refs=[], geom_authority=True), a_final) # ── B안 ────────────────────────────────────────────── b_diag = stem_out.with_name(f"{tag}_B_diagram.png") if not b_diag.exists(): gen(f"car_ab_{tag}_B_diagram", DIAGRAM_HEAD + "\n\n" + sec["location"] + "\n" + sec["camera"], [], b_diag) b_bg = stem_out.with_name(f"{tag}_B_bg.png") if not b_bg.exists(): gen(f"car_ab_{tag}_B_bg", BG_FROM_DIAGRAM_HEAD + "\n\n" + ctx, [("TOP-DOWN SEATING DIAGRAM — layout reference only, " "not a photograph", b_diag)], b_bg) b_final = stem_out.with_name(f"{tag}_B_final.png") if not b_final.exists(): gen(f"car_ab_{tag}_B_final", build_bgfirst_final_prompt(brief, geom_authority=True), build_bgfirst_refs( bg=b_bg, conti=None, char_refs=ents, prop_refs=[], geom_authority=True), b_final) # ── 판정 ───────────────────────────────────────────── if "judge" not in r: r["judge"] = judge_three( tag, brief, {"A안": a_final, "B안": b_final, "현행": cur_sel}) save() print(f" 판정 확정표: {r['judge'].get('agreed')}") # ── C안 — 장면 유사 실사 기반 재연출 (08-10 육안 후 추가) ── # A·B 육안 탈락 후 사용자 지시: 사람이 타고 있는 "비슷한 # 장면" 실사를 찾아 그 사진을 구도·내부 구조·핸들 측의 # 기반으로 쓰고 탑승자만 작품 인물로 바꾼다. 핸들 방향은 # 어디에도 박지 않는다 — 원어 검색이 그 나라 차를 부르고, # 생성 계약은 "참조 사진이 보여주는 쪽 유지"뿐이다. if "scene_search" not in r: author_c = call_structured( f"car_ab_scene_search_author_{tag}", SCENE_SEARCH_AUTHOR_SYS, f"SHOT TEXT / LOCATION / CAMERA:\n{ctx}", search_author_schema(), project_config={f"car_ab_scene_search_author_{tag}": {"model": "gemini-pro"}}, schema_name="car_ab_scene_search_author", opik_metadata={"operation_type": "car_ab_experiment", "shot_tag": tag, "project_id": PROJ, "episode_id": EPI}, enable_fallback=False) from app.core.openai_keys import openai_client sr_c = search_reference_images( openai_client(), directive_native=author_c["directive_native"], terms_native=author_c["terms_native"], language_lock_native=author_c["language_lock_native"]) r["scene_search"] = {"author": author_c, "queries": sr_c.get("queries"), "n_images": len(sr_c.get("images") or [])} photos_c = download_photos( sr_c.get("images") or [], stem_out.with_name(f"{tag}_C")) r["scene_search"]["downloaded"] = [str(p) for p in photos_c] save() photos_c = [Path(p) for p in r["scene_search"].get("downloaded") or [] if Path(p).exists()] if not photos_c: raise RuntimeError("C안: 장면 실사 회수 0 — 중단") c_final = stem_out.with_name(f"{tag}_C_final.png") if not c_final.exists(): gen(f"car_ab_{tag}_C_final", C_RESTAGE_HEAD + "\n\n" + brief, [("PHOTOGRAPH TO RESTAGE — composition and cabin " "authority (not identity)", photos_c[0]), *ents], c_final) if "judge_c" not in r: r["judge_c"] = judge_three( f"{tag}_c", brief, {"C안": c_final, "현행": cur_sel}) save() print(f" C안 판정 확정표: {r['judge_c'].get('agreed')}") # ── C2 — 프레이밍 문자 제거 + 사진 권위 강화 (08-10 승인) ── # C1 이 참조의 올바른 좌석 물리를 버리고 브리프 배치 문자를 # 따랐다(남자=조수석). 재연출·판정 모두 CAMERA 절을 뺀 # 브리프를 쓴다 — 판정까지 빼는 이유: 그 문자 자체가 함정임이 # 확인됐고, 두 후보에 같은 기준이면 비교는 공정하다. brief_nc = strip_camera_block(brief) c2_final = stem_out.with_name(f"{tag}_C2_final.png") if not c2_final.exists(): gen(f"car_ab_{tag}_C2_final", C_RESTAGE_HEAD + "\n\n" + brief_nc, [("PHOTOGRAPH TO RESTAGE — sole physical-layout " "authority (not identity)", photos_c[0]), *ents], c2_final) if "judge_c2" not in r: r["judge_c2"] = judge_three( f"{tag}_c2", brief_nc, {"C2안": c2_final, "현행": cur_sel}) save() print(f" C2 판정 확정표: {r['judge_c2'].get('agreed')}") except Exception as exc: # noqa: BLE001 r["오류"] = f"{type(exc).__name__}: {exc}"[:300] print(f" 실패 — {r['오류']}") save() continue r.pop("오류", None) save() print(f"\n결과: {res_path}") if __name__ == "__main__": main()