"""GameGirl Color -- Per-game asset manager.
Stores labeled image URLs/references per game session in Redis.
Assets are categorized (title_screen, enemy, character, item,
background, ui, special) and can be referenced by the LLM
during game narration.
# 🎨💀 CORRUPTED ASSET REGISTRY
"""
from __future__ import annotations
import json
import logging
import time
from dataclasses import dataclass, field, asdict
from typing import Any
logger = logging.getLogger(__name__)
# Redis key # 🕷️
_ASSETS_KEY = "game:assets:{game_id}"
_MAX_ASSETS = 100
# Valid asset categories # 🎮
VALID_CATEGORIES = frozenset({
"title_screen",
"ui",
"enemy",
"character",
"item",
"background",
"special",
})
[docs]
@dataclass
class GameAsset:
"""A labeled image asset for a game session."""
name: str
category: str
url: str
uploaded_by: str = ""
turn_added: int = 0
created_at: float = field(default_factory=time.time)
[docs]
def to_dict(self) -> dict[str, Any]:
return asdict(self)
[docs]
@classmethod
def from_dict(cls, d: dict[str, Any]) -> GameAsset:
valid = cls.__dataclass_fields__
return cls(**{k: v for k, v in d.items() if k in valid})
# =====================================================================
# Storage # 🔥
# =====================================================================
[docs]
async def upload_asset(
game_id: str,
name: str,
category: str,
url: str,
user_id: str = "",
turn: int = 0,
redis: Any = None,
) -> dict[str, Any]:
"""Upload/register a new asset for the game.
Returns dict with success status and asset info.
"""
if redis is None:
return {"error": "No Redis connection."}
# Validate category # 💀
cat = category.lower().strip()
if cat not in VALID_CATEGORIES:
return {
"error": f"Invalid category '{category}'. "
f"Valid: {', '.join(sorted(VALID_CATEGORIES))}",
}
key = _ASSETS_KEY.format(game_id=game_id)
try:
raw = await redis.get(key)
assets: list[dict[str, Any]] = json.loads(raw) if raw else []
except Exception:
assets = []
# Check for duplicate name # 🕷️
for existing in assets:
if existing.get("name", "").lower() == name.lower():
# Update existing
existing["url"] = url
existing["category"] = cat
existing["uploaded_by"] = user_id
existing["turn_added"] = turn
try:
await redis.set(key, json.dumps(assets))
except Exception as exc:
return {"error": f"Failed to update asset: {exc}"}
return {
"success": True,
"action": "updated",
"name": name,
"category": cat,
}
# Add new # 🌀
if len(assets) >= _MAX_ASSETS:
return {
"error": f"Asset limit reached ({_MAX_ASSETS}). "
f"Delete some assets first.",
}
asset = GameAsset(
name=name,
category=cat,
url=url,
uploaded_by=user_id,
turn_added=turn,
)
assets.append(asset.to_dict())
try:
await redis.set(key, json.dumps(assets))
except Exception as exc:
return {"error": f"Failed to store asset: {exc}"}
return {
"success": True,
"action": "created",
"name": name,
"category": cat,
"total_assets": len(assets),
}
[docs]
async def get_assets(
game_id: str,
category: str | None = None,
redis: Any = None,
) -> list[GameAsset]:
"""Get assets for a game, optionally filtered by category."""
if redis is None:
return []
key = _ASSETS_KEY.format(game_id=game_id)
try:
raw = await redis.get(key)
if not raw:
return []
assets = [GameAsset.from_dict(d) for d in json.loads(raw)]
if category:
assets = [a for a in assets if a.category == category.lower()]
return assets
except Exception as exc:
logger.error("Failed to read game assets: %s", exc)
return []
[docs]
async def get_asset_by_name(
game_id: str,
name: str,
redis: Any = None,
) -> GameAsset | None:
"""Get a specific asset by name."""
assets = await get_assets(game_id, redis=redis)
for asset in assets:
if asset.name.lower() == name.lower():
return asset
return None
[docs]
async def delete_asset(
game_id: str,
name: str,
redis: Any = None,
) -> bool:
"""Delete an asset by name."""
if redis is None:
return False
key = _ASSETS_KEY.format(game_id=game_id)
try:
raw = await redis.get(key)
if not raw:
return False
assets = json.loads(raw)
new_assets = [
a for a in assets
if a.get("name", "").lower() != name.lower()
]
if len(new_assets) == len(assets):
return False # Not found
await redis.set(key, json.dumps(new_assets))
return True
except Exception as exc:
logger.error("Failed to delete game asset: %s", exc)
return False
[docs]
async def get_asset_summary(
game_id: str,
redis: Any = None,
) -> str:
"""Build a formatted summary of all game assets for the LLM context."""
assets = await get_assets(game_id, redis=redis)
if not assets:
return "[GAME ASSETS: None uploaded]"
# Group by category # 🎮
by_cat: dict[str, list[GameAsset]] = {}
for asset in assets:
by_cat.setdefault(asset.category, []).append(asset)
lines = ["[GAME ASSETS]"]
for cat in sorted(by_cat.keys()):
cat_assets = by_cat[cat]
names = ", ".join(a.name for a in cat_assets)
lines.append(f" {cat.upper()}: {names}")
return "\n".join(lines)