Add reference-anchored concept generation

tools/generate_textures.py

@@ -21,8 +21,10 @@ import os
 import random
 import stat
 import sys
+import re
 import urllib.error
 import urllib.request
+from pathlib import Path
 from PIL import Image, ImageDraw, ImageFilter, ImageFont
 
 ROOT = os.path.join(os.path.dirname(os.path.abspath(__file__)), "..")
@@ -373,19 +375,122 @@ def _api_key() -> str:
     return key
 
 
+CHAT_COMPLETIONS_URL = "https://openrouter.ai/api/v1/chat/completions"
+_MIME_BY_SUFFIX = {".jpg": "image/jpeg", ".jpeg": "image/jpeg",
+                   ".png": "image/png", ".webp": "image/webp"}
+
+
+def _openrouter_headers() -> dict:
+    return {
+        "Authorization": f"Bearer {_api_key()}",
+        "Content-Type": "application/json",
+        "HTTP-Referer": "https://git.allucanget.biz/zwitschi/security-game",
+        "X-Title": "Gig Line",
+    }
+
+
+def _openrouter_post(req: urllib.request.Request) -> dict:
+    try:
+        with urllib.request.urlopen(req, timeout=300) as resp:
+            return json.loads(resp.read().decode("utf-8"))
+    except urllib.error.HTTPError as exc:
+        detail = exc.read().decode("utf-8", "replace")
+        sys.exit(f"error: OpenRouter returned HTTP {exc.code}: {detail}")
+    except urllib.error.URLError as exc:
+        sys.exit(f"error: could not reach OpenRouter: {exc.reason}")
+
+
+def _data_url(path) -> str:
+    """Encode a local image file as a base64 `data:` URL for multimodal input."""
+    path = Path(path)
+    if not path.is_file():
+        sys.exit(f"error: reference image not found: {path}")
+    mime = _MIME_BY_SUFFIX.get(path.suffix.lower())
+    if not mime:
+        sys.exit(f"error: unsupported reference image type: {path.suffix}")
+    encoded = base64.b64encode(path.read_bytes()).decode("ascii")
+    return f"data:{mime};base64,{encoded}"
+
+
+def _decode_chat_image(payload: dict) -> Image.Image:
+    """Pull the returned image out of a chat-completions response.
+
+    Models return either `message.images[].image_url.url` (a `data:` URL or an
+    https URL) or a `data:` URL embedded in the message content.
+    """
+    message = (payload.get("choices") or [{}])[0].get("message", {})
+    urls = []
+    for item in message.get("images") or []:
+        if isinstance(item, dict):
+            urls.append(item.get("image_url", {}).get("url", ""))
+    if not urls:
+        content = message.get("content") or ""
+        match = re.search(r"data:image/[a-z]+;base64,[A-Za-z0-9+/=]+", str(content))
+        if match:
+            urls.append(match.group(0))
+    if not urls:
+        sys.exit(f"error: OpenRouter returned no image data: {payload}")
+    url = urls[0]
+    if url.startswith("data:"):
+        raw = base64.b64decode(url.split(",", 1)[1])
+    else:
+        try:
+            with urllib.request.urlopen(urllib.request.Request(url), timeout=120) as resp:
+                raw = resp.read()
+        except urllib.error.URLError as exc:
+            sys.exit(f"error: could not download returned image: {exc.reason}")
+    try:
+        img = Image.open(io.BytesIO(raw))
+        img.load()
+    except Exception as exc:  # noqa: BLE001 - surface any decode failure clearly
+        sys.exit(f"error: could not decode returned image: {exc}")
+    return img
+
+
+def _generate_image_with_reference(prompt: str, model: str, reference_path) -> Image.Image:
+    """Image-to-image request: the reference photo anchors the composition."""
+    body = {
+        "model": model,
+        "messages": [{
+            "role": "user",
+            "content": [
+                {"type": "text", "text": prompt},
+                {"type": "image_url", "image_url": {"url": _data_url(reference_path)}},
+            ],
+        }],
+    }
+    req = urllib.request.Request(
+        CHAT_COMPLETIONS_URL,
+        data=json.dumps(body).encode("utf-8"),
+        headers=_openrouter_headers(),
+        method="POST",
+    )
+    return _decode_chat_image(_openrouter_post(req))
+
+
 def generate_image(prompt: str, model: str, size: int,
                    transparent: bool = False,
-                   aspect_ratio: str = "1:1") -> Image.Image:
+                   aspect_ratio: str = "1:1",
+                   reference_path=None) -> Image.Image:
     """Call the OpenRouter image API and return the decoded PIL image.
 
     Requests a square image at `size` px or a role-shaped image at the matching
     resolution tier and decodes `data[].b64_json`. Seedream's square requests
     retain the existing GEN_SIZE floor. When `transparent` is true, the API is
     asked for a transparent PNG background.
+
+    When `reference_path` points at an image file, the request switches to the
+    multimodal chat endpoint and sends the file as a base64 data URL next to the
+    prompt (image-to-image guidance). Size/aspect parameters do not apply on
+    that endpoint; the caller crops or rescales the returned frame as usual.
+    Only licence-cleared reference photos may be passed (see
+    `scene_concepts._reference_source`).
     """
     size = max(1, int(size))
-    if model == DEFAULT_IMAGE_MODEL:
+    if model == DEFAULT_IMAGE_MODEL and reference_path is None:
         size = max(GEN_SIZE, size)
+    if reference_path is not None:
+        return _generate_image_with_reference(prompt, model, reference_path)
     body = {
         "model": model,
         "prompt": prompt,
@@ -405,22 +510,10 @@ def generate_image(prompt: str, model: str, size: int,
     req = urllib.request.Request(
         OPENROUTER_IMAGES_URL,
         data=json.dumps(body).encode("utf-8"),
-        headers={
-            "Authorization": f"Bearer {_api_key()}",
-            "Content-Type": "application/json",
-            "HTTP-Referer": "https://git.allucanget.biz/zwitschi/security-game",
-            "X-Title": "Gig Line",
-        },
+        headers=_openrouter_headers(),
         method="POST",
     )
-    try:
-        with urllib.request.urlopen(req, timeout=300) as resp:
-            payload = json.loads(resp.read().decode("utf-8"))
-    except urllib.error.HTTPError as exc:
-        detail = exc.read().decode("utf-8", "replace")
-        sys.exit(f"error: OpenRouter returned HTTP {exc.code}: {detail}")
-    except urllib.error.URLError as exc:
-        sys.exit(f"error: could not reach OpenRouter: {exc.reason}")
+    payload = _openrouter_post(req)
 
     data = payload.get("data") or []
     if not data or not data[0].get("b64_json"):

tools/scene_concepts.py

@@ -323,30 +323,104 @@ def compose_prompt(name: str) -> str:
     return f"{prompt}. {style}."
 
 
-def generate_one(name: str, model: str, request_size: int) -> Path:
+REFERENCE_PROMPT_SUFFIX = (
+    "Use the attached reference photograph as the fixed source of geometry and "
+    "composition: keep the building's proportions, column rhythm, band heights, "
+    "signage placement and camera angle exactly as photographed. Only restage "
+    "lighting, weather, people and surface condition where the prompt asks for it. "
+    "Do not invent extra storeys, wings or ornaments that the photo does not show."
+)
+
+REFS_DIR = ROOT / "docs" / "screens" / "refs"
+
+
+def _reference_source(path_str: str, allow_unchecked: bool = False) -> dict:
+    """Resolve and licence-check a reference photo for anchored generation.
+
+    A file under `docs/screens/refs/` must already be listed in that venue's
+    `SIDECAR.md` (the licence gate); its licence and Commons page are returned
+    for provenance. Anything else requires the explicit `--reference-unchecked`
+    override, which keeps unreviewed material out of the committed pipeline by
+    default.
+    """
+    path = Path(path_str).expanduser().resolve()
+    if not path.is_file():
+        sys.exit(f"error: reference photo not found: {path}")
+    if REFS_DIR in path.parents:
+        sidecar = path.parent / "SIDECAR.md"
+        text = sidecar.read_text(encoding="utf-8") if sidecar.is_file() else ""
+        row = next((line for line in text.splitlines()
+                    if f"`{path.name}`" in line), None)
+        if not row:
+            sys.exit(f"error: {path.name} is not listed in {sidecar}; add the "
+                     "sidecar row (licence gate) before anchoring generation on it")
+        cells = [c.strip() for c in row.split("|")]
+        licence = next((c for c in cells if c and c.lower().startswith(("cc", "public"))), "?")
+        page = next((c.strip("<> ") for c in cells if c.startswith("<http")), "")
+        return {"path": path, "licence": licence, "page": page, "checked": True}
+    if not allow_unchecked:
+        sys.exit(f"error: {path} is outside docs/screens/refs/ and not licence-checked; "
+                 "move it into a refs board with a SIDECAR row, or pass "
+                 "--reference-unchecked to accept the licence responsibility")
+    return {"path": path, "licence": "unchecked", "page": "", "checked": False}
+
+
+def _write_reference_provenance(out_path: Path, reference: dict, name: str,
+                                model: str) -> Path:
+    """Record which photo anchored a generated frame (`.refgen.json` sidecar)."""
+    sidecar = out_path.with_suffix(".refgen.json")
+    sidecar.write_text(json.dumps({
+        "concept": name,
+        "generated": out_path.name,
+        "model": model,
+        "reference": {"file": str(reference["path"].relative_to(ROOT))
+                      if reference["path"].is_relative_to(ROOT) else str(reference["path"]),
+                      "licence": reference["licence"],
+                      "page": reference["page"],
+                      "licence_checked": reference["checked"]},
+        "suffix": REFERENCE_PROMPT_SUFFIX,
+    }, indent=2) + "\n", encoding="utf-8")
+    return sidecar
+
+
+def generate_one(name: str, model: str, request_size: int,
+                 reference: dict | None = None) -> Path:
     group = CATALOGUE[name][0]
     prompt = compose_prompt(name)
+    if reference:
+        prompt = f"{prompt}\n\n{REFERENCE_PROMPT_SUFFIX}"
     if group == "scene":
         image = generate_textures.generate_image(prompt, model, request_size,
-                                                 transparent=False, aspect_ratio="3:2")
+                                                 transparent=False, aspect_ratio="3:2",
+                                                 reference_path=reference["path"]
+                                                 if reference else None)
         image = ImageOps.fit(image.convert("RGB"), (SCENE_W, SCENE_H),
                              method=Image.Resampling.LANCZOS)
     elif group == "backdrop":
         image = generate_textures.generate_image(prompt, model, request_size,
-                                                 transparent=True, aspect_ratio="3:2")
+                                                 transparent=True, aspect_ratio="3:2",
+                                                 reference_path=reference["path"]
+                                                 if reference else None)
     else:
         image = generate_textures.generate_image(prompt, model, request_size,
-                                                 transparent=False, aspect_ratio="1:1")
+                                                 transparent=False, aspect_ratio="1:1",
+                                                 reference_path=reference["path"]
+                                                 if reference else None)
         image = image.convert("RGB")
     out = next_free_path(raw_path(name))
     out.parent.mkdir(parents=True, exist_ok=True)
     image.save(out)
+    if reference:
+        provenance = _write_reference_provenance(out, reference, name, model)
+        print(f"provenance {_rel(provenance)} <- {reference['licence']}")
     print(f"wrote {_rel(out)} {image.size}")
     return out
 
 
 def cmd_generate(args) -> int:
     _load_env_file()
+    reference = _reference_source(args.reference, args.reference_unchecked) \
+        if args.reference else None
     names = _select(args.group, args.only)
     if not names:
         print("nothing selected", file=sys.stderr)
@@ -356,7 +430,7 @@ def cmd_generate(args) -> int:
         for _ in range(max(1, args.variants)):
             print(f"generating {name} ...", flush=True)
             try:
-                generate_one(name, args.model, args.request_size)
+                generate_one(name, args.model, args.request_size, reference=reference)
             except SystemExit as exc:  # generate_textures exits on API errors
                 failed.append(name)
                 print(f"  FAILED {name}: {exc}", file=sys.stderr)
@@ -1032,6 +1106,12 @@ def build_parser():
     gen.add_argument("--variants", type=int, default=1)
     gen.add_argument("--model", default=DEFAULT_MODEL)
     gen.add_argument("--request-size", type=int, default=1536)
+    gen.add_argument("--reference", default="",
+                     help="anchor generation on a licence-cleared reference photo "
+                          "(must be listed in its refs-board SIDECAR.md)")
+    gen.add_argument("--reference-unchecked", action="store_true",
+                     help="allow a reference photo from outside docs/screens/refs/ "
+                          "(you take responsibility for its licence)")
     gen.set_defaults(handler=cmd_generate)
     proc = sub.add_parser("process")
     proc.add_argument("--group", choices=("backdrop", "texture"))

tools/test_scene_concepts.py

@@ -1,5 +1,7 @@
+import json
 import tempfile
 import unittest
+from unittest import mock
 from pathlib import Path
 
 from PIL import Image
@@ -31,6 +33,62 @@ class CatalogueTests(unittest.TestCase):
         self.assertTrue(sc.raw_path("asphalt").is_relative_to(sc.RAW_DIR))
 
 
+class ReferenceGateTests(unittest.TestCase):
+    ROOT = Path(sc.__file__).resolve().parent.parent
+
+    def test_refs_photo_listed_in_sidecar_passes_with_provenance(self):
+        result = sc._reference_source(
+            self.ROOT / "docs/screens/refs/hollywood/"
+            "el-capitan-facade-day-wikimedia-2009-10-27.jpg")
+        self.assertTrue(result["checked"])
+        self.assertEqual(result["licence"], "CC BY 2.0")
+        self.assertIn("commons.wikimedia.org", result["page"])
+
+    def test_refs_photo_missing_from_sidecar_is_refused(self):
+        with tempfile.TemporaryDirectory(dir=self.ROOT / "docs/screens/refs") as tmp:
+            path = Path(tmp) / "unlisted.jpg"
+            path.write_bytes(b"x")
+            with self.assertRaises(SystemExit):
+                sc._reference_source(path)
+
+    def test_outside_photo_requires_the_explicit_override(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            path = Path(tmp) / "photo.jpg"
+            path.write_bytes(b"x")
+            with self.assertRaises(SystemExit):
+                sc._reference_source(path)
+            result = sc._reference_source(path, allow_unchecked=True)
+            self.assertFalse(result["checked"])
+            self.assertEqual(result["licence"], "unchecked")
+
+    def test_missing_file_is_refused(self):
+        with self.assertRaises(SystemExit):
+            sc._reference_source(self.ROOT / "docs/screens/refs/nope.jpg")
+
+    def test_generate_one_writes_reference_provenance(self):
+        reference = {"path": self.ROOT / "docs/screens/refs/hollywood/"
+                     "el-capitan-facade-day-wikimedia-2009-10-27.jpg",
+                     "licence": "CC BY 2.0", "page": "https://commons/x",
+                     "checked": True}
+        class FakeImage:
+            def convert(self, mode):
+                return Image.new("RGB", (1600, 900))
+        with tempfile.TemporaryDirectory() as tmp:
+            out = Path(tmp) / "hollywood_boulevard_entrance_day.png"
+            with mock.patch.object(sc.generate_textures, "generate_image",
+                                   return_value=FakeImage()) as mocked, \
+                    mock.patch.object(sc, "next_free_path", return_value=out), \
+                    mock.patch.object(sc, "_rel", side_effect=str):
+                sc.generate_one("hollywood_boulevard_entrance_day", "m", 1536,
+                                reference=reference)
+            call = mocked.call_args
+            self.assertIn("keep the building", call.args[0].lower())
+            self.assertEqual(call.kwargs["reference_path"], reference["path"])
+            provenance = json.loads(out.with_suffix(".refgen.json").read_text())
+            self.assertEqual(provenance["reference"]["licence"], "CC BY 2.0")
+            self.assertTrue(provenance["reference"]["licence_checked"])
+
+
 class NoOverwriteTests(unittest.TestCase):
     def test_next_free_path_never_returns_an_existing_file(self):
         with tempfile.TemporaryDirectory() as tmp: