#!/usr/bin/env python3 """마커 맵을 **어느 모델이 제대로 그리는가** — 같은 재료로 나란히 잰다. 왜 있나 (2026-08-25) — `S2sh1` 의 카메라 시야각 부채꼴을 `gpt-image-2` 가 **두 주행에 걸쳐 6회 연속 반대 방향**으로 그렸다. 모델을 grok 으로 바꾸자 3회차에 통과했다. 그러면 남는 물음은 하나다 — **nb2 는 그리는가?** ★프로덕션 경로를 그대로 쓴다. 문안(`marker_prompt`)·바탕 지도·검사기 (`lane_marker_map_check`)를 **완주 판 체크포인트에서 그대로 꺼내** 쓰고, 모델만 갈아 끼운다. 재는 자리와 도는 자리가 갈리면 무엇을 쟀는지 알 수 없다(2026-08-10 카나리아: 측정 도구에만 없던 sanitizer 로 첫 실측이 죽었다). ★판정도 프로덕션 것을 그대로 부른다 — 「잘 그렸나」를 내가 눈으로 정하지 않는다. `markers_ok` 는 프로덕션 검사기가 낸 값이다. ★한 모델당 여러 번 돌린다 — 한 번은 우연일 수 있다. 통과/실패를 세어 **비율**로 말한다. usage: ab_marker_map_engine.py # nb2 3회 ab_marker_map_engine.py --models nb2,gpt-image-2 --rounds 2 ab_marker_map_engine.py --tag S2sh1 --out artifact/20260825_marker_ab """ import argparse import datetime as dt import json import sys from pathlib import Path sys.path.insert(0, "/Users/manta/Documents/Projects/TheRoad-I1/scratchpad") import _opik_env # noqa: E402,F401 ★cwd 를 backend 로 (설정을 읽는 유일한 조건) ROOT = Path("/Users/manta/Documents/Projects/TheRoad-I1") STATE = json.loads( (ROOT / "scratchpad" / "minimal_e2e_state.json").read_text()) PID, EID = STATE["project_id"], STATE["episode_id"] CP = (ROOT / "projects" / PID / "checkpoints" / "episodes" / EID / "shot_conti_light" / "manifest.json") def load_case(tag: str): """완주 판이 실제로 쓴 재료 — 문안·바탕 지도·기하.""" d = json.loads(CP.read_text()) lane = (d.get("data") or {}).get("lane_conti") or {} e = lane.get(tag) if not e: raise SystemExit(f"★{tag} 가 체크포인트에 없다 (있는 것: " f"{sorted(lane)})") for k in ("marker_prompt", "base_map_path", "geometry"): if not e.get(k): raise SystemExit(f"★{tag} 에 {k} 가 없다 — 잴 재료가 모자란다") return e def render(model: str, prompt: str, base_map: Path) -> bytes: """모델 하나로 마커 맵을 그린다 — 경로는 **모델 종류가 정한다**.""" from app.modules.pipeline.marker_map_engine import ( is_openrouter_model, render_via_openrouter, ) if is_openrouter_model(model): return render_via_openrouter( model=model, prompt=prompt, ref_paths=[base_map], role="marker_map_probe") if model.startswith("gemini") or model == "nb2": # nb2 = 이 저장소가 부르는 이름. 물리 모델은 설정이 준다. from app.core.config import settings from app.modules.llm.gemini_image_client import GeminiImageClient physical = (settings.gemini_image_model if model == "nb2" else model) client = GeminiImageClient(model=physical) png, _ms = client.generate_image( prompt, labeled_references=[("SITE PLAN:", base_map.read_bytes())]) return png # OpenAI 직행 (gpt-image-2 계열) from app.core.openai_keys import openai_client from app.core.config import settings from app.modules.llm.gpt_image_primitive import call_gpt_image_bytes client = openai_client(api_key=settings.openai_api_key) return call_gpt_image_bytes( client, mode="edit", prompt=prompt, ref_paths=[str(base_map)], call_kwargs={"model": model, "size": "1024x1024", "quality": "high", "n": 1}) def check(base_map: Path, marker_png: bytes, geometry: dict) -> dict: """★프로덕션 검사기를 그대로 부른다 — 판정을 내가 하지 않는다.""" from app.core.steps.shot_conti_light_step import LANE_SKETCH_PACK_VERSION from app.modules.llm.llm_client import call_structured from app.modules.pipeline.multiroll_gemini import png_part from app.modules.pipeline.outdoor_marker_map import ( _ANCHOR_SKETCH_PACKS, _VIEW_SKETCH_PACKS, build_marker_check_schema, build_marker_check_user_lines, load_marker_check_sys, ) user = ("Expected staging (each marker described relative to landmarks " "visible on the plan; fractions are coarse estimates, x left→" "right 0..1, y top→bottom 0..1):\n" + "\n".join(build_marker_check_user_lines( geometry, require_anchors=( LANE_SKETCH_PACK_VERSION in _ANCHOR_SKETCH_PACKS), require_view=( LANE_SKETCH_PACK_VERSION in _VIEW_SKETCH_PACKS)))) return call_structured( "marker_map_probe_check", load_marker_check_sys(LANE_SKETCH_PACK_VERSION), [{"type": "text", "text": user}, {"type": "text", "text": "ORIGINAL CLEAN SITE PLAN:"}, png_part(base_map.read_bytes()), {"type": "text", "text": "ANNOTATED SITE PLAN (to check):"}, png_part(marker_png)], build_marker_check_schema(), schema_name="marker_map_probe_check") def main() -> int: ap = argparse.ArgumentParser(description=__doc__) ap.add_argument("--models", default="nb2", help="쉼표 구분. nb2 | gpt-image-2 | provider/slug") ap.add_argument("--rounds", type=int, default=3) ap.add_argument("--tag", default="S2sh1") ap.add_argument("--out", default="") args = ap.parse_args() case = load_case(args.tag) base_map = Path(case["base_map_path"]) prompt = case["marker_prompt"] geometry = case["geometry"] out_dir = Path(args.out) if args.out else ( ROOT / "artifact" / "20260825_marker_map_engine_ab") if not out_dir.is_absolute(): out_dir = ROOT / out_dir out_dir.mkdir(parents=True, exist_ok=True) models = [m.strip() for m in args.models.split(",") if m.strip()] print(f"샷 {args.tag} · 바탕 {base_map.name} · 문안 {len(prompt):,}자") print(f"모델 {models} × {args.rounds}회 — 판정은 프로덕션 검사기\n") rows = [] for model in models: ok = 0 for r in range(1, args.rounds + 1): stamp = f"{model.replace('/', '_')}_r{r}" try: png = render(model, prompt, base_map) except Exception as exc: # noqa: BLE001 print(f" {stamp:44s} ✘ 생성 실패 {type(exc).__name__}: " f"{str(exc)[:120]}") rows.append({"model": model, "round": r, "ok": False, "error": f"{type(exc).__name__}: {exc}"[:300]}) continue png_path = out_dir / f"{args.tag}_{stamp}.png" png_path.write_bytes(png) try: v = check(base_map, png, geometry) except Exception as exc: # noqa: BLE001 print(f" {stamp:44s} ✘ 판정 실패 {type(exc).__name__}") rows.append({"model": model, "round": r, "ok": None, "error": f"판정 {type(exc).__name__}"}) continue passed = bool(v.get("markers_ok")) ok += passed print(f" {stamp:44s} {'✔ 통과' if passed else '✘ 반려'} " f"{str(v.get('details_ko') or '')[:90]}") rows.append({"model": model, "round": r, "ok": passed, "details_ko": v.get("details_ko"), "file": png_path.name}) print(f" → {model}: {ok}/{args.rounds} 통과\n") (out_dir / "result.json").write_text( json.dumps({"tag": args.tag, "rounds": args.rounds, "rows": rows}, ensure_ascii=False, indent=1), encoding="utf-8") print(f"산출 → {out_dir}") print(f"({dt.datetime.now().strftime('%m-%d %H:%M:%S')})") return 0 if __name__ == "__main__": raise SystemExit(main())