Add optional photo providers, layout studio and scoped integrations
This commit is contained in:
Pigway
@@ -0,0 +1,227 @@
|
||||
"""Bounded layout plans shared by deterministic and optional AI renderers."""
|
||||
|
||||
import json
|
||||
from io import BytesIO
|
||||
from typing import Literal
|
||||
|
||||
from PIL import Image, ImageDraw, ImageOps
|
||||
from pydantic import Field, model_validator
|
||||
|
||||
from pigway_frame_flow.errors import ErrorCode as E
|
||||
from pigway_frame_flow.errors import FlowError
|
||||
from pigway_frame_flow.models import StrictModel
|
||||
from pigway_frame_flow.settings import secret
|
||||
|
||||
|
||||
class Region(StrictModel):
|
||||
asset_id: str = Field(pattern=r"^[a-f0-9]{32}$")
|
||||
x: int = Field(ge=0)
|
||||
y: int = Field(ge=0)
|
||||
width: int = Field(ge=100)
|
||||
height: int = Field(ge=100)
|
||||
|
||||
|
||||
class LayoutInput(StrictModel):
|
||||
asset_ids: list[str] = Field(min_length=1, max_length=12)
|
||||
title: str = Field(default="", max_length=100)
|
||||
message: str = Field(default="", max_length=500)
|
||||
orientation: Literal["portrait", "landscape"] = "portrait"
|
||||
style: Literal["grid", "hero", "poster"] = "grid"
|
||||
use_ai: bool = False
|
||||
prompt: str = Field(default="", max_length=1500)
|
||||
|
||||
@model_validator(mode="after")
|
||||
def ids(self):
|
||||
import re
|
||||
|
||||
if len(set(self.asset_ids)) != len(self.asset_ids) or any(
|
||||
not re.fullmatch(r"[a-f0-9]{32}", x) for x in self.asset_ids
|
||||
):
|
||||
raise ValueError("Choose distinct local images")
|
||||
return self
|
||||
|
||||
|
||||
class LayoutPlan(StrictModel):
|
||||
width: int = Field(ge=1200, le=1600)
|
||||
height: int = Field(ge=1200, le=1600)
|
||||
title: str = Field(default="", max_length=100)
|
||||
message: str = Field(default="", max_length=500)
|
||||
regions: list[Region] = Field(min_length=1, max_length=12)
|
||||
engine: Literal["local", "ai"] = "local"
|
||||
|
||||
@model_validator(mode="after")
|
||||
def bounds(self):
|
||||
if (self.width, self.height) not in [(1200, 1600), (1600, 1200)]:
|
||||
raise ValueError("Unsupported canvas")
|
||||
for i, region in enumerate(self.regions):
|
||||
if (
|
||||
region.x + region.width > self.width
|
||||
or region.y + region.height > self.height - 120
|
||||
or region.y < 150
|
||||
):
|
||||
raise ValueError("Photo region outside safe photo area")
|
||||
for other in self.regions[:i]:
|
||||
if (
|
||||
region.x < other.x + other.width
|
||||
and other.x < region.x + region.width
|
||||
and region.y < other.y + other.height
|
||||
and other.y < region.y + region.height
|
||||
):
|
||||
raise ValueError("Photo regions overlap")
|
||||
return self
|
||||
|
||||
|
||||
def local_plan(request: LayoutInput):
|
||||
import math
|
||||
|
||||
width, height = (1200, 1600) if request.orientation == "portrait" else (1600, 1200)
|
||||
gap, top, bottom = 24, 170, height - 140
|
||||
n = len(request.asset_ids)
|
||||
regions = []
|
||||
if request.style in ("hero", "poster") and n > 1:
|
||||
hero_height = int((bottom - top) * (0.45 if n > 7 else 0.62))
|
||||
regions.append(
|
||||
Region(
|
||||
asset_id=request.asset_ids[0],
|
||||
x=gap,
|
||||
y=top,
|
||||
width=width - 2 * gap,
|
||||
height=hero_height - gap,
|
||||
)
|
||||
)
|
||||
top += hero_height
|
||||
ids = request.asset_ids[1:]
|
||||
else:
|
||||
ids = request.asset_ids
|
||||
columns = min(3, max(1, math.ceil(math.sqrt(len(ids) * width / (bottom - top)))))
|
||||
rows = math.ceil(len(ids) / columns)
|
||||
cell_w, cell_h = (
|
||||
(width - gap * (columns + 1)) // columns,
|
||||
(bottom - top - gap * (rows - 1)) // rows,
|
||||
)
|
||||
for i, asset_id in enumerate(ids):
|
||||
regions.append(
|
||||
Region(
|
||||
asset_id=asset_id,
|
||||
x=gap + (i % columns) * (cell_w + gap),
|
||||
y=top + (i // columns) * (cell_h + gap),
|
||||
width=cell_w,
|
||||
height=cell_h,
|
||||
)
|
||||
)
|
||||
return LayoutPlan(
|
||||
width=width,
|
||||
height=height,
|
||||
title=request.title,
|
||||
message=request.message,
|
||||
regions=regions,
|
||||
)
|
||||
|
||||
|
||||
class LayoutStudio:
|
||||
def __init__(self, renderer):
|
||||
self.renderer = renderer
|
||||
|
||||
def plan(self, request):
|
||||
plan = local_plan(request)
|
||||
metadata = []
|
||||
for asset_id in request.asset_ids:
|
||||
with Image.open(self.renderer.assets.path(asset_id)) as image:
|
||||
metadata.append(
|
||||
{"asset_id": asset_id, "width": image.width, "height": image.height}
|
||||
)
|
||||
if not request.use_ai:
|
||||
return plan
|
||||
db = self.renderer.db
|
||||
row = db.one("SELECT payload FROM app_config WHERE key='ai_layout'")
|
||||
config = json.loads(row["payload"]) if row else {}
|
||||
key = secret("FRAME_FLOW_AI_KEY")
|
||||
key_path = self.renderer.settings.config_dir / "ai-layout.secret"
|
||||
if not key and key_path.exists():
|
||||
key = key_path.read_text().strip()
|
||||
if not key or not config.get("endpoint") or not config.get("model"):
|
||||
raise FlowError(
|
||||
E.CONFIG_REQUIRED,
|
||||
"Configure the AI layout endpoint, model and key first",
|
||||
409,
|
||||
)
|
||||
self.renderer.settings.additional_secrets.append(key)
|
||||
body = {
|
||||
"model": config["model"],
|
||||
"response_format": {"type": "json_object"},
|
||||
"messages": [
|
||||
{
|
||||
"role": "system",
|
||||
"content": "Return ONLY a JSON layout following this example. Keep canvas size, all asset ids exactly once, non-overlapping photo rectangles y>=150 and bottom<=height-120. Keep title/message unchanged. Never include code, URLs or HTML. Example: "
|
||||
+ plan.model_dump_json(),
|
||||
},
|
||||
{
|
||||
"role": "user",
|
||||
"content": json.dumps(
|
||||
{
|
||||
"images": metadata,
|
||||
"style": request.style,
|
||||
"instruction": request.prompt,
|
||||
},
|
||||
ensure_ascii=False,
|
||||
),
|
||||
},
|
||||
],
|
||||
"max_tokens": 2000,
|
||||
}
|
||||
try:
|
||||
data = self.renderer.assets.fetch_url(
|
||||
config["endpoint"],
|
||||
method="POST",
|
||||
headers={
|
||||
"Authorization": "Bearer " + key,
|
||||
"Content-Type": "application/json",
|
||||
},
|
||||
body=json.dumps(body).encode(),
|
||||
limit=256 * 1024,
|
||||
)
|
||||
result = json.loads(data)["choices"][0]["message"]["content"]
|
||||
candidate = LayoutPlan.model_validate_json(result)
|
||||
if (
|
||||
sorted(x.asset_id for x in candidate.regions)
|
||||
!= sorted(request.asset_ids)
|
||||
or (candidate.width, candidate.height) != (plan.width, plan.height)
|
||||
or candidate.title != request.title
|
||||
or candidate.message != request.message
|
||||
):
|
||||
raise ValueError()
|
||||
candidate.engine = "ai"
|
||||
return candidate
|
||||
except Exception:
|
||||
raise FlowError(
|
||||
E.RENDER_FAILED,
|
||||
"AI layout failed or returned an invalid plan; choose local layout to continue",
|
||||
502,
|
||||
) from None
|
||||
|
||||
def render(self, data, profile):
|
||||
plan = LayoutPlan.model_validate(data["layout"])
|
||||
canvas = Image.new("RGB", (plan.width, plan.height), "white")
|
||||
for region in plan.regions:
|
||||
with Image.open(self.renderer.assets.path(region.asset_id)) as source:
|
||||
image = ImageOps.exif_transpose(source).convert("RGB")
|
||||
image = ImageOps.fit(
|
||||
image,
|
||||
(region.width, region.height),
|
||||
method=Image.Resampling.LANCZOS,
|
||||
)
|
||||
canvas.paste(image, (region.x, region.y))
|
||||
draw = ImageDraw.Draw(canvas)
|
||||
for text, y, size in [
|
||||
(plan.title, 35, 54),
|
||||
(plan.message, plan.height - 100, 30),
|
||||
]:
|
||||
font = self.renderer.font(size)
|
||||
while text and draw.textlength(text, font=font) > plan.width - 48:
|
||||
text = text[:-1]
|
||||
draw.text((24, y), text, font=font, fill="black")
|
||||
if profile == "grayscale":
|
||||
canvas = ImageOps.grayscale(canvas).convert("RGB")
|
||||
out = BytesIO()
|
||||
canvas.save(out, "JPEG", quality=93, optimize=True)
|
||||
return out.getvalue()
|
||||
Reference in New Issue
Block a user