"""실험 — Gemini Pro + Nano Banana(1.5 Pro 이미지 모델) + 45° 회전 4 view + 직전 image chain. 흐름: Step A. run_dir + base_plan_id로 도면 + base_photo + 관련 shots 수집 Step B. Gemini Pro vision(`gemini-3.1-pro-preview`): 도면 PNG + scene context → 4 view JSON (0°, 45°, 90°, 135°) 각 view: { description, t2i_prompt } + atmosphere_canon Step C. Gemini Nano Banana(`gemini-3.1-flash-image-preview`): - view_0 : 도면 ref + view_0 prompt → photo - view_45 : 도면 + view_0 image + view_45 prompt - view_90 : 도면 + view_45 image + view_90 prompt - view_135 : 도면 + view_90 image + view_135 prompt """ from __future__ import annotations import argparse import base64 import glob import json import logging import os import sys import time import urllib.error import urllib.request from pathlib import Path BACKEND = Path(__file__).resolve().parent.parent sys.path.insert(0, str(BACKEND)) from dotenv import load_dotenv # noqa: E402 load_dotenv(BACKEND / ".env") from app.core.config import settings # noqa: E402 from app.modules.llm.gemini_image_client import GeminiImageClient # noqa: E402 from app.modules.llm.gemini_key_pool import get_next_key # noqa: E402 logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s") logger = logging.getLogger("gemini_45deg") GEMINI_TEXT_URL = ( "https://generativelanguage.googleapis.com/v1beta/models/{model}:generateContent?key={api_key}" ) VIEW_KEYS = ["view_0_deg", "view_45_deg", "view_90_deg", "view_135_deg"] VIEW_DEGREES = [0, 45, 90, 135] SYSTEM_PROMPT = """You are an architectural visualization specialist. You will receive: 1. A top-down architectural FLOOR PLAN of an interior room (image attached). 2. Existing wall-by-wall scene description (N/E/S/W info). 3. SHOT context — narrative scenes set in this room with camera positions, doors, lighting, props. Your task: produce FOUR photorealistic eye-level view prompts, each at a different camera rotation around the room center, covering a 135° arc: - view_0_deg : camera facing 0° (toward the wall labeled NORTH on the floor plan) - view_45_deg : camera facing 45° clockwise from view_0 - view_90_deg : camera facing 90° clockwise from view_0 (toward EAST wall) - view_135_deg : camera facing 135° clockwise from view_0 For EACH view: - description: 2-3 English sentences describing what is visible in this camera frustum (which wall(s), which doors/windows/fixtures/furniture, which corner is in foreground). - t2i_prompt: a complete English photorealistic image generation prompt that includes: * camera position: at the room's geometric center, eye-level ~1.6m, slight wide angle ~28mm, facing the specified rotation * the visible content from the description, with door/window/furniture silhouettes consistent with the floor plan * 35mm cinematic still aesthetic, soft natural daylight + dim domestic practicals, realistic shadows, lived-in texture (worn wallpaper, dust, scuff) * NO people, empty room * NO overlaid text, NO diagram lines * Atmospheric and material identity (wall finish / floor / ceiling / lighting / palette) MUST be consistent with the other views STRICT RULES: - ENGLISH ONLY in all prompts and any rendered text. No Korean, no Chinese, no other scripts. - COMMON NOUNS only. NEVER use proper names of characters, places, or productions. - Each view's t2i_prompt should reference "the same interior room from the floor plan" or "the dwelling shown in the floor plan reference" so the image model anchors to the plan. - Reflect scene atmosphere (lighting tone, wear) but NO narrative events — no people, no blood, no broken glass. - Each of the 4 views must show DIFFERENT content. As the camera rotates 45° each step, the wall/furniture in frame must change accordingly. - Mention rotation explicitly in each t2i_prompt (e.g., "camera facing 45 degrees clockwise from the north reference, looking toward the corner between the north and east walls"). Also produce: - atmosphere_canon: ONE paragraph summarizing the wall finish, floor, ceiling, lighting fixtures, color palette, and overall ambience that MUST appear identically in all 4 views — the image model uses this as a consistency anchor. Return strict JSON. No markdown fence, no commentary outside JSON. """ RESPONSE_SCHEMA = { "type": "object", "properties": { "view_0_deg": { "type": "object", "properties": { "description": {"type": "string"}, "t2i_prompt": {"type": "string"}, }, "required": ["description", "t2i_prompt"], }, "view_45_deg": { "type": "object", "properties": { "description": {"type": "string"}, "t2i_prompt": {"type": "string"}, }, "required": ["description", "t2i_prompt"], }, "view_90_deg": { "type": "object", "properties": { "description": {"type": "string"}, "t2i_prompt": {"type": "string"}, }, "required": ["description", "t2i_prompt"], }, "view_135_deg": { "type": "object", "properties": { "description": {"type": "string"}, "t2i_prompt": {"type": "string"}, }, "required": ["description", "t2i_prompt"], }, "atmosphere_canon": {"type": "string"}, }, "required": ["view_0_deg", "view_45_deg", "view_90_deg", "view_135_deg", "atmosphere_canon"], } def collect_scene_context(run_dir: Path, base_plan_id: str) -> dict: ctx: dict = {"base_plan_id": base_plan_id, "base_plan": None, "base_photo": None, "shots": [], "spatial": None} sp2 = run_dir / "step2_plan_specs.json" if sp2.exists(): d = json.loads(sp2.read_text(encoding="utf-8")) for p in d.get("base_plans", []): if p.get("id") == base_plan_id: ctx["base_plan"] = p; break sp7 = run_dir / "step7_photo_specs.json" if sp7.exists(): d = json.loads(sp7.read_text(encoding="utf-8")) for ph in d.get("base_photos", []): if ph.get("source_plan_id") == base_plan_id: ctx["base_photo"] = ph; break for f in sorted(glob.glob(str(run_dir / "step3_shot_*.json"))): s = json.loads(Path(f).read_text(encoding="utf-8")) if s.get("base_plan_id") == base_plan_id: ctx["shots"].append(s) sp1 = run_dir / "step1_spatial.json" if sp1.exists(): ctx["spatial"] = json.loads(sp1.read_text(encoding="utf-8")) return ctx def build_user_prompt(ctx: dict) -> str: parts: list[str] = [] parts.append("== BASE PLAN SPEC ==") bp = ctx.get("base_plan") or {} parts.append(f"id: {bp.get('id')}") parts.append(f"label: {bp.get('label')}") parts.append(f"visual_domain: {bp.get('visual_domain')}") if bp.get("t2i_prompt"): parts.append(f"\nplan_drawing_prompt:\n{bp['t2i_prompt']}") if bp.get("legend"): parts.append(f"\nlegend: {json.dumps(bp['legend'], ensure_ascii=False)}") if bp.get("elements_meta"): parts.append(f"\nelements_meta: {json.dumps(bp['elements_meta'], ensure_ascii=False)}") parts.append("\n== BASE PHOTO SPEC (existing wall-by-wall annotation) ==") bph = ctx.get("base_photo") or {} if bph.get("camera_note"): parts.append(f"camera_note: {bph['camera_note']}") if bph.get("lighting"): parts.append(f"lighting: {bph['lighting']}") if bph.get("t2i_prompt"): parts.append(f"\nbase_photo_prompt (wall-by-wall sentences):\n{bph['t2i_prompt']}") parts.append("\n== RELATED SHOTS ==") for s in ctx.get("shots", []): si = s.get("scene_index"); sx = s.get("shot_index") parts.append(f"\n--- Shot S{si:02d}_Shot{sx} ---") cam = s.get("camera") or {} parts.append(f"camera position: {cam.get('position', '(none)')}") parts.append(f"camera heading: {cam.get('heading', '(none)')}") parts.append(f"height/fov/lens: {cam.get('height')} / {cam.get('fov')} / {cam.get('lens_note')}") for a in s.get("additions", []): parts.append(f" + [{a.get('type')}] @ {a.get('position', '?')}: {a.get('note', '')}") parts.append("\n== INTERIOR CANON ==") sp = ctx.get("spatial") or {} interior = (sp.get("environment_canon") or {}).get("interior") or {} if interior: parts.append(json.dumps(interior, ensure_ascii=False, indent=2)) parts.append("\n== TASK ==") parts.append("Use the floor plan IMAGE (attached) and ALL information above to produce 4 view prompts (0°/45°/90°/135°) + atmosphere_canon as defined in the system instruction.") parts.append("Return JSON only.") return "\n".join(parts) def call_gemini_pro_vision(model: str, plan_png_bytes: bytes, system_prompt: str, user_prompt: str, response_schema: dict) -> dict: """Gemini Pro REST vision call with JSON schema response.""" body = { "systemInstruction": {"parts": [{"text": system_prompt}]}, "contents": [{ "role": "user", "parts": [ {"text": user_prompt}, {"inline_data": { "mime_type": "image/png", "data": base64.b64encode(plan_png_bytes).decode("ascii"), }}, ], }], "generationConfig": { "temperature": 0.2, "responseMimeType": "application/json", "responseJsonSchema": response_schema, "maxOutputTokens": settings.llm_max_output_tokens, }, } last_err: Exception | None = None for attempt in range(1, 4): api_key = get_next_key() url = GEMINI_TEXT_URL.format(model=model, api_key=api_key) req = urllib.request.Request( url, data=json.dumps(body).encode("utf-8"), headers={"Content-Type": "application/json"}, method="POST", ) try: with urllib.request.urlopen(req, timeout=180) as resp: payload = json.loads(resp.read().decode("utf-8")) break except urllib.error.HTTPError as e: err = e.read().decode("utf-8", errors="replace") last_err = RuntimeError(f"Gemini Pro {e.code}: {err[:400]}") if e.code in (429, 500, 502, 503, 504) and attempt < 3: time.sleep(3 * attempt); continue raise last_err cand = (payload.get("candidates") or [{}])[0] parts = (cand.get("content") or {}).get("parts") or [] text = "".join(p.get("text", "") for p in parts).strip() if not text: raise RuntimeError(f"Gemini Pro empty response: {json.dumps(payload)[:600]}") return json.loads(text) def main() -> int: p = argparse.ArgumentParser() p.add_argument("--plan-png", required=True) p.add_argument("--run-dir", required=True) p.add_argument("--base-plan-id", required=True) p.add_argument("--out-dir", required=True) p.add_argument("--text-model", default=settings.gemini_text_model) p.add_argument("--image-model", default=settings.gemini_image_model) p.add_argument("--aspect-ratio", default="16:9") p.add_argument("--skip-text", action="store_true", help="기존 prompts.json 재사용 (이미지만 재생성)") p.add_argument("--text-only", action="store_true", help="LLM 텍스트 단계만 실행") args = p.parse_args() plan_png = Path(args.plan_png).resolve() run_dir = Path(args.run_dir).resolve() out_dir = Path(args.out_dir).resolve() out_dir.mkdir(parents=True, exist_ok=True) if not os.getenv("GEMINI_API_KEY"): logger.error("GEMINI_API_KEY not set"); return 1 plan_bytes = plan_png.read_bytes() prompts_p = out_dir / "prompts.json" # Step A + B if args.skip_text and prompts_p.exists(): prompts = json.loads(prompts_p.read_text(encoding="utf-8")) logger.info("[skip text] reusing prompts.json") else: ctx = collect_scene_context(run_dir, args.base_plan_id) if not ctx["base_plan"]: logger.error("base_plan_id 매칭 실패: %s", args.base_plan_id); return 1 logger.info("collected: base_plan=ok, base_photo=%s, %d shots", bool(ctx["base_photo"]), len(ctx["shots"])) user_prompt = build_user_prompt(ctx) logger.info("user prompt: %d chars; calling Gemini Pro %s", len(user_prompt), args.text_model) prompts = call_gemini_pro_vision(args.text_model, plan_bytes, SYSTEM_PROMPT, user_prompt, RESPONSE_SCHEMA) prompts["_meta"] = { "base_plan_id": args.base_plan_id, "shot_count": len(ctx["shots"]), "text_model": args.text_model, "image_model": args.image_model, "aspect_ratio": args.aspect_ratio, "user_prompt_chars": len(user_prompt), } prompts_p.write_text(json.dumps(prompts, ensure_ascii=False, indent=2), encoding="utf-8") logger.info("saved prompts.json") # Print for review print("\n" + "=" * 60) print("=== ATMOSPHERE CANON ===") print("=" * 60) print(prompts.get("atmosphere_canon", "(missing)")) for k, deg in zip(VIEW_KEYS, VIEW_DEGREES): v = prompts.get(k) or {} print(f"\n--- {k} ({deg}°) ---") print(f"description: {v.get('description', '(missing)')}") print(f"t2i_prompt:\n{v.get('t2i_prompt', '(missing)')}") print() if args.text_only: logger.info("text-only mode → skip image generation") return 0 # Step C: chained image generation image_client = GeminiImageClient(model=args.image_model) atmosphere = prompts.get("atmosphere_canon", "") prev_image_bytes: bytes | None = None for k, deg in zip(VIEW_KEYS, VIEW_DEGREES): v = prompts.get(k) or {} view_prompt = (v.get("t2i_prompt") or "").strip() if not view_prompt: logger.error("missing t2i_prompt for %s — skip", k); continue # consistency anchor: atmosphere canon + prev-view note full_prompt = view_prompt if atmosphere: full_prompt += f"\n\nAtmosphere consistency anchor (must match across all views): {atmosphere}" if prev_image_bytes is not None: full_prompt += ( "\n\nThe second reference image is the photorealistic view from the previous " "rotation step. Match its wall finish, floor finish, ceiling, lighting tone, " "color palette, and overall material aging — same room, just rotated. The new " "view should show DIFFERENT walls/objects than the previous image." ) labeled_refs = [("Floor plan reference (top-down):", plan_bytes)] if prev_image_bytes is not None: labeled_refs.append(("Previous rotation step photo (consistency reference):", prev_image_bytes)) out_p = out_dir / f"photo_{k}.png" logger.info("[%s @ %d°] generating with %d refs (plan%s)", k, deg, len(labeled_refs), " + prev" if prev_image_bytes is not None else "") img_bytes, ms = image_client.generate_image( prompt=full_prompt, labeled_references=labeled_refs, aspect_ratio=args.aspect_ratio, ) out_p.write_bytes(img_bytes) logger.info(" saved %s (%d KB, %d ms)", out_p.name, len(img_bytes) // 1024, ms) prev_image_bytes = img_bytes logger.info("=== DONE === %s", out_dir) return 0 if __name__ == "__main__": sys.exit(main())