"""Phase C — Task C3: outdoor 3-stage shared-model guide scope (roles + chaining metadata). 3 producer 가 generate_floor_plan_image 수렴점을 공유하므로 default role 'floor_plan_image' 로 오염된다 → stage별 role 교정 + chaining metadata(Decision #2 ⓑ): - generate_location_aerial_base → role=outdoor_aerial_base, producer_stage=aerial_base - inject_shot_blocking → role=outdoor_shot_blocking, parent_stage=aerial_base - camera sketch (Stage C) → role=outdoor_camera_sketch, parent_stage=shot_blocking|aerial_base chaining UUID 불가(flush 시 부여) → group_id/producer_stage/parent_stage metadata 로 사후 그룹 엣지(Phase D structural_inferred). input_image_ids=None(v1 defer). scope 미개방 시 capture 0(byte-identical). gpt-image 는 RecordingClient(실 API 0). 생성물 품질은 실측 canary 영역 — 여기선 결정론(role/metadata/dedup)만. """ import base64 import io import json import uuid from datetime import datetime, timezone from types import SimpleNamespace import app.core.file_paths as fp import app.modules.pipeline.outdoor_site_layout_provider as prov from app.core.database import SessionLocal from app.models.catalog import ProjectRegistry, UserAccount from app.models.project import Episode, ImageAsset from app.services.image_capture.context import generation_context def _now() -> str: return datetime.now(timezone.utc).isoformat() def _seed_base(db) -> tuple[str, str]: uid = str(uuid.uuid4()) db.add(UserAccount( id=uid, username=f"u_{uid[:8]}", display_name="t", password_hash="x", role="admin", created_at=_now(), updated_at=_now(), )) db.commit() pid = str(uuid.uuid4()) db.add(ProjectRegistry( id=pid, name="Phase C3", created_by=uid, created_at=_now(), updated_at=_now(), )) eid = str(uuid.uuid4()) db.add(Episode( id=eid, project_id=pid, episode_number=1, title="Pilot", source_filename="p.pdf", source_path="p.pdf", created_at=_now(), updated_at=_now(), )) db.commit() return pid, eid _BIG = b"\x89PNG\r\n\x1a\n" + b"\x00" * 8192 class _RecordingClient: def __init__(self): b64 = base64.b64encode(_BIG).decode("ascii") class _Images: def edit(self, **kwargs): return SimpleNamespace(data=[SimpleNamespace(b64_json=b64)]) def generate(self, **kwargs): return SimpleNamespace(data=[SimpleNamespace(b64_json=b64)]) self.images = _Images() _LAYOUT = { "landmarks": [ {"id": "L1", "label": "a long road", "kind": "area", "points": [[10, 10], [40, 15], [70, 12]]}, {"id": "L2", "label": "a shelter", "kind": "structure", "points": [[20, 30], [30, 30], [30, 40], [20, 40]]}, ], "cameras": [{"shot_index": 8, "pos": [5, 5], "look_at": [40, 40]}], } def _stub_paths(monkeypatch, tmp_path): from app.core import config as _cfg monkeypatch.setattr(_cfg.settings, "projects_dir", str(tmp_path / "projects")) monkeypatch.setattr(fp, "PROJECT_ROOT", tmp_path) def _rows(pid): db = SessionLocal() try: return db.query(ImageAsset).filter( ImageAsset.project_id == pid, ImageAsset.is_intermediate == True # noqa: E712 ).all() finally: db.close() def test_aerial_base_role_and_metadata(monkeypatch, tmp_path): db = SessionLocal() try: pid, eid = _seed_base(db) finally: db.close() _stub_paths(monkeypatch, tmp_path) # Stage A scope: group 공통 → scene_index None (still_id None). with generation_context(pid, eid, stage="outdoor_site_layout", scene_index=None, shot_index=None): png, meta = prov.generate_location_aerial_base( _LAYOUT, openai_client=_RecordingClient(), model="gpt-image-2.5-sunburst", capture_extra_metadata={"group_id": "osl-g1", "location_id": "L11"}) assert png and meta.get("base_png_hash") rows = _rows(pid) assert len(rows) == 1 row = rows[0] assert row.pipeline_role == "outdoor_aerial_base" assert row.input_image_ids is None # v1 defer pm = json.loads(row.pipeline_metadata_json) assert pm["group_id"] == "osl-g1" assert pm["location_id"] == "L11" assert pm["producer_stage"] == "aerial_base" assert pm.get("parent_stage") is None # root assert "scene_index" not in pm # None scope def test_camera_guide_with_blocking_captures_both_stages(monkeypatch, tmp_path): db = SessionLocal() try: pid, eid = _seed_base(db) finally: db.close() _stub_paths(monkeypatch, tmp_path) # Stage B+C scope: anchor shot → scene_index=12, shot_index=8. with generation_context(pid, eid, stage="outdoor_site_layout", scene_index=12, shot_index=8): png, sm_meta = prov.generate_shared_model_camera_guide( _LAYOUT, 8, base_png=_BIG, use_blocking=True, openai_client=_RecordingClient(), model="gpt-image-2.5-sunburst", capture_extra_metadata={"group_id": "osl-g1", "location_id": "L11"}) assert png rows = _rows(pid) by_role = {r.pipeline_role: r for r in rows} assert set(by_role) == {"outdoor_shot_blocking", "outdoor_camera_sketch"} blk = by_role["outdoor_shot_blocking"] blk_pm = json.loads(blk.pipeline_metadata_json) assert blk_pm["producer_stage"] == "shot_blocking" assert blk_pm["parent_stage"] == "aerial_base" assert blk_pm["group_id"] == "osl-g1" assert blk_pm["scene_index"] == 12 # C0 fold assert blk.input_image_ids is None sk = by_role["outdoor_camera_sketch"] sk_pm = json.loads(sk.pipeline_metadata_json) assert sk_pm["producer_stage"] == "camera_sketch" assert sk_pm["parent_stage"] == "shot_blocking" # use_blocking=True assert sk_pm["group_id"] == "osl-g1" assert sk_pm["scene_index"] == 12 def test_camera_guide_without_blocking_sketch_parent_is_base(monkeypatch, tmp_path): db = SessionLocal() try: pid, eid = _seed_base(db) finally: db.close() _stub_paths(monkeypatch, tmp_path) with generation_context(pid, eid, stage="outdoor_site_layout", scene_index=3, shot_index=1): prov.generate_shared_model_camera_guide( _LAYOUT, 8, base_png=_BIG, use_blocking=False, openai_client=_RecordingClient(), model="gpt-image-2.5-sunburst", capture_extra_metadata={"group_id": "osl-g2", "location_id": "L11"}) rows = _rows(pid) assert len(rows) == 1 # 단순 구도 → 블로킹 생략, sketch 1행만 sk_pm = json.loads(rows[0].pipeline_metadata_json) assert rows[0].pipeline_role == "outdoor_camera_sketch" assert sk_pm["producer_stage"] == "camera_sketch" assert sk_pm["parent_stage"] == "aerial_base" # use_blocking=False → base 직접 ref def test_provider_no_capture_outside_scope_byte_identical(monkeypatch, tmp_path): db = SessionLocal() try: pid, eid = _seed_base(db) finally: db.close() _stub_paths(monkeypatch, tmp_path) # scope 미개방(flag OFF 시 _generate_shared_model_guides 미호출 등가) → capture 0 prov.generate_location_aerial_base( _LAYOUT, openai_client=_RecordingClient(), model="gpt-image-2.5-sunburst") prov.generate_shared_model_camera_guide( _LAYOUT, 8, base_png=_BIG, use_blocking=True, openai_client=_RecordingClient(), model="gpt-image-2.5-sunburst") assert _rows(pid) == [] # production 행 무변