"""GameGirl Color -- Pillow-based gameboard compositor.
Layers character sprites, enemies, and items onto scene
backgrounds at specified coordinates with z-ordering.
# 🎮💀 CORRUPTED CANVAS ENGINE
"""
from __future__ import annotations
import asyncio
import json
import logging
from io import BytesIO
from typing import Any, TYPE_CHECKING
import aiohttp
from tools._safe_http import assert_safe_http_url
if TYPE_CHECKING:
from tool_context import ToolContext
logger = logging.getLogger(__name__)
TOOL_NAME = "compose_gameboard"
TOOL_DESCRIPTION = (
"Compose a gameboard scene by layering game assets (character "
"sprites, enemies, items) onto a background image at specified "
"coordinates. Uses saved game assets by name. The composed "
"image is sent to the current channel."
)
TOOL_PARAMETERS = {
"type": "object",
"properties": {
"background": {
"type": "string",
"description": (
"Name of a saved background asset, or a URL to "
"a background image."
),
},
"sprites": {
"type": "array",
"description": (
"List of sprites to layer onto the background."
),
"items": {
"type": "object",
"properties": {
"name": {
"type": "string",
"description": "Asset name or image URL.",
},
"x": {
"type": "integer",
"description": "X position (pixels from left).",
},
"y": {
"type": "integer",
"description": "Y position (pixels from top).",
},
"scale": {
"type": "number",
"description": "Scale factor (1.0 = original). Default: 1.0.",
},
"z_order": {
"type": "integer",
"description": "Layer order (higher = on top). Default: 0.",
},
},
"required": ["name", "x", "y"],
},
},
"width": {
"type": "integer",
"description": "Output canvas width. Default: 1024.",
},
"height": {
"type": "integer",
"description": "Output canvas height. Default: 576.",
},
},
"required": ["background", "sprites"],
}
async def _download_image(url: str) -> bytes | None:
"""Download an image from a URL."""
try:
url = assert_safe_http_url(url.strip())
except ValueError as exc:
logger.warning("Blocked gameboard image URL: %s", exc)
return None
try:
async with aiohttp.ClientSession() as session:
async with session.get(
url, timeout=aiohttp.ClientTimeout(total=30),
) as resp:
if resp.status == 200:
return await resp.read()
except Exception as exc:
logger.warning("Failed to download image %s: %s", url[:80], exc)
return None
async def _resolve_asset_url(
name: str,
game_id: str | None,
redis: Any = None,
) -> str:
"""Resolve an asset name to a URL."""
if name.startswith("http://") or name.startswith("https://"):
return name
if game_id and redis:
try:
from game_assets import get_asset_by_name
asset = await get_asset_by_name(game_id, name, redis=redis)
if asset:
return asset.url
except Exception:
pass
return name # Return as-is, might be a URL without protocol
def _compose_gameboard_sync(
bg_data: bytes,
sprite_items: list[tuple[bytes, int, int, float]],
width: int,
height: int,
) -> tuple[bytes | None, str]:
"""CPU-heavy Pillow compositing (runs in a worker thread)."""
try:
from PIL import Image
except ImportError:
return None, "Pillow not installed. Run: pip install Pillow"
try:
bg_img = Image.open(BytesIO(bg_data)).convert("RGBA")
bg_img = bg_img.resize((width, height), Image.LANCZOS)
canvas = bg_img.copy()
for sprite_data, x, y, scale in sprite_items:
try:
sprite_img = Image.open(BytesIO(sprite_data)).convert("RGBA")
if scale != 1.0:
new_w = int(sprite_img.width * scale)
new_h = int(sprite_img.height * scale)
sprite_img = sprite_img.resize(
(max(1, new_w), max(1, new_h)), Image.LANCZOS,
)
canvas.paste(sprite_img, (x, y), sprite_img)
except Exception as exc:
logger.warning("Failed to composite sprite bytes: %s", exc)
output = BytesIO()
canvas.save(output, format="WEBP", quality=85)
output.seek(0)
return output.read(), ""
except Exception as exc:
return None, str(exc)
[docs]
async def run(
background: str,
sprites: list[dict[str, Any]] | None = None,
width: int = 1024,
height: int = 576,
ctx: ToolContext | None = None,
) -> str:
"""Compose a gameboard scene from layered assets.
Args:
background: Background asset name or URL.
sprites: List of sprite placement dicts.
width: Canvas width.
height: Canvas height.
ctx: Tool execution context.
Returns:
str: JSON result.
"""
if ctx is None:
return json.dumps({"error": "No tool context available."})
# Get game session for asset resolution # 🌀
game_id = None
redis = getattr(ctx, "redis", None)
try:
from game_session import get_session
session = get_session(str(ctx.channel_id))
if session and session.active:
game_id = session.game_id
except ImportError:
pass
# Resolve and download background # 🔥
bg_url = await _resolve_asset_url(background, game_id, redis)
bg_data = await _download_image(bg_url)
if bg_data is None:
return json.dumps({
"error": f"Failed to download background: {background}",
})
sprite_items: list[tuple[bytes, int, int, float]] = []
if sprites:
sprites_sorted = sorted(
sprites, key=lambda s: s.get("z_order", 0),
)
for sprite_def in sprites_sorted:
name = sprite_def.get("name", "")
x = int(sprite_def.get("x", 0))
y = int(sprite_def.get("y", 0))
scale = float(sprite_def.get("scale", 1.0))
sprite_url = await _resolve_asset_url(name, game_id, redis)
sprite_data = await _download_image(sprite_url)
if sprite_data is None:
logger.warning("Skipping sprite '%s': download failed", name)
continue
sprite_items.append((sprite_data, x, y, scale))
composed_bytes, compose_err = await asyncio.to_thread(
_compose_gameboard_sync, bg_data, sprite_items, width, height,
)
if compose_err:
return json.dumps({"error": f"Failed to compose gameboard: {compose_err}"})
if not composed_bytes:
return json.dumps({"error": "Failed to compose gameboard (empty output)."})
channel_id = str(ctx.channel_id)
try:
file_url = await ctx.adapter.send_file(
channel_id,
composed_bytes,
"gameboard.webp",
"image/webp",
)
ctx.sent_files.append({
"data": composed_bytes,
"filename": "gameboard.webp",
"mimetype": "image/webp",
"file_url": file_url or "",
})
except Exception as exc:
return json.dumps({"error": f"Failed to send composed image: {exc}"})
result_info: dict = {
"success": True,
"width": width,
"height": height,
"filename": "gameboard.webp",
"sprites_count": len(sprites) if sprites else 0,
}
if file_url:
result_info["file_url"] = file_url
return json.dumps(result_info)