"""Placeholder backgrounds for M1 (no image model). Deterministic solid colors. Real generated backgrounds arrive in Plan 3 via the OpenRouter provider; this module is the swappable stand-in behind the same 'label -> image' idea. """ from __future__ import annotations import io import logging from PIL import Image from ..config import ( AD_SIZE, BACKGROUND_PROMPTS, GEN_PARAMS, OPENROUTER_IMAGE_URL, RENDERER_VERSION, ) from .cache import CacheError, cache_key logger = logging.getLogger(__name__) # One distinct, deterministic color per known background label. _LABEL_COLORS: dict[str, tuple[int, int, int]] = { "hotel_exterior": (38, 70, 83), "city_skyline": (42, 54, 110), "room_interior": (120, 66, 52), "generic": (90, 90, 90), } _FALLBACK_COLOR = (60, 60, 60) def make_placeholder_background(label: str, size: tuple[int, int] = AD_SIZE) -> Image.Image: """Return a deterministic RGB placeholder background for ``label``.""" color = _LABEL_COLORS.get(label, _FALLBACK_COLOR) return Image.new("RGB", size, color) def background_cache_key(label: str, *, model: str) -> str: """The cache key for a generated background. Shared by generate + tooling.""" return cache_key( { "provider": "openrouter", "model_slug": model, "endpoint": OPENROUTER_IMAGE_URL, "prompt": BACKGROUND_PROMPTS[label], "gen_params": GEN_PARAMS, "renderer_version": RENDERER_VERSION, } ) def generate_background( label: str, *, cache, provider, model: str, api_key: str | None = None, size: tuple[int, int] = AD_SIZE, replay: bool = False, ) -> Image.Image: """Provider-mode background: eval hit -> dev re-check -> provider fill. In replay/eval mode an eval miss is a hard fail (never touches the network). `provider(*, prompt, model, api_key, params, endpoint) -> (bytes, meta)`. """ prompt = BACKGROUND_PROMPTS[label] key = background_cache_key(label, model=model) if cache.has(key, "eval"): logger.debug("bg eval hit", extra={"label": label, "model": model}) return cache.get(key, "eval", expect_slug=model)[0] if replay: raise CacheError( f"replay/eval miss for background {label!r} (key {key[:12]}…) — " "no network in replay mode; promote a dev entry first" ) with cache.lock(key): if cache.has(key, "dev"): # re-check: a peer worker may have just filled it return cache.get(key, "dev")[0] image_bytes, meta = provider( prompt=prompt, model=model, api_key=api_key, params=GEN_PARAMS, endpoint=OPENROUTER_IMAGE_URL, ) image = Image.open(io.BytesIO(image_bytes)).convert("RGB") if image.size != size: image = image.resize(size) # safety net if the provider ignored size manifest = { "model_slug": model, "prompt": prompt, "gen_params": GEN_PARAMS, "endpoint": OPENROUTER_IMAGE_URL, "response_meta": meta, } cache.put(key, image, manifest) # public put = DEV tier (held under lock above) logger.debug("bg dev fill", extra={"label": label, "model": model}) return image