Lobby: maps, map editor and tokens
DM map library (upload re-encoded with Pillow), map editor with prepared enemies/NPCs, map on the lobby table with personal pan/zoom, draggable tokens (players move their own, DM all), full/player view, live sync via the WebSocket. Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01HJhXuLD2Z6hv3Wqeq9UL9K
This commit is contained in:
Claude
@@ -0,0 +1,307 @@
|
||||
"""Maps for the lobby table: the DM's map library, the active map of a lobby and the tokens on it.
|
||||
|
||||
Coordinates are pixels of the stored image. A token's size is in grid squares (the map's `grid` is pixels per square).
|
||||
Images are re-encoded with Pillow on upload, so only real images get through and metadata is gone."""
|
||||
import io
|
||||
import json
|
||||
import time
|
||||
import uuid
|
||||
from pathlib import Path
|
||||
|
||||
from fastapi import Body, Depends, HTTPException, Request
|
||||
from fastapi.concurrency import run_in_threadpool
|
||||
from fastapi.responses import Response
|
||||
from PIL import Image, ImageOps
|
||||
|
||||
MAX_UPLOAD = 12_000_000 # bytes of the uploaded file
|
||||
MAX_SIDE = 4096 # larger images are scaled down
|
||||
MAX_PIXELS = 60_000_000 # decompression bomb guard (before scaling)
|
||||
MAX_MAPS = 30 # per DM
|
||||
MAX_TOKENS = 120 # per map / lobby
|
||||
COLORS = ("#c0392b", "#8e44ad", "#16a085", "#7f8c8d", "#d35400", "#2c3e50", "#27ae60", "#b9770e")
|
||||
|
||||
|
||||
def schema(con):
|
||||
con.execute("""CREATE TABLE IF NOT EXISTS maps (
|
||||
id TEXT PRIMARY KEY, owner_id INTEGER NOT NULL, name TEXT NOT NULL, w INTEGER NOT NULL, h INTEGER NOT NULL,
|
||||
grid REAL NOT NULL, tokens TEXT NOT NULL DEFAULT '[]', walls TEXT NOT NULL DEFAULT '[]', created REAL NOT NULL)""")
|
||||
con.execute("""CREATE TABLE IF NOT EXISTS lobby_tokens (
|
||||
id TEXT PRIMARY KEY, lobby_id TEXT NOT NULL, user_id INTEGER, name TEXT NOT NULL, x REAL, y REAL,
|
||||
color TEXT, size REAL NOT NULL DEFAULT 1, hidden INTEGER NOT NULL DEFAULT 0)""")
|
||||
con.execute("CREATE INDEX IF NOT EXISTS lobby_tokens_lobby ON lobby_tokens (lobby_id)")
|
||||
if "map_id" not in [r["name"] for r in con.execute("PRAGMA table_info(lobbies)")]:
|
||||
con.execute("ALTER TABLE lobbies ADD COLUMN map_id TEXT")
|
||||
|
||||
|
||||
def encode(raw: bytes) -> tuple[bytes, int, int]:
|
||||
"""Validates and re-encodes an uploaded image as WebP, scaled to at most MAX_SIDE."""
|
||||
Image.MAX_IMAGE_PIXELS = MAX_PIXELS
|
||||
try:
|
||||
im = Image.open(io.BytesIO(raw))
|
||||
if im.format not in ("PNG", "JPEG", "WEBP"):
|
||||
raise ValueError("format")
|
||||
im.load()
|
||||
except Exception:
|
||||
raise HTTPException(422, "Please upload a PNG, JPEG or WebP image")
|
||||
im = ImageOps.exif_transpose(im)
|
||||
im.thumbnail((MAX_SIDE, MAX_SIDE))
|
||||
if im.mode not in ("RGB", "RGBA"):
|
||||
im = im.convert("RGBA" if "transparency" in im.info or im.mode in ("LA", "PA") else "RGB")
|
||||
out = io.BytesIO()
|
||||
im.save(out, "WEBP", quality=90, method=4)
|
||||
return out.getvalue(), im.width, im.height
|
||||
|
||||
|
||||
def clean_color(c, default="#c0392b") -> str:
|
||||
c = str(c or "")
|
||||
return c if len(c) == 7 and c[0] == "#" and all(ch in "0123456789abcdefABCDEF" for ch in c[1:]) else default
|
||||
|
||||
|
||||
def clean_token(t: dict, w: int, h: int, keep_id=True) -> dict:
|
||||
if not isinstance(t, dict):
|
||||
raise HTTPException(422, "Invalid token")
|
||||
name = str(t.get("name", "")).strip()[:40] or "Token"
|
||||
try:
|
||||
x, y = t.get("x"), t.get("y")
|
||||
x = None if x is None else min(max(float(x), 0), w)
|
||||
y = None if y is None else min(max(float(y), 0), h)
|
||||
size = min(max(float(t.get("size", 1)), 0.25), 8)
|
||||
except (TypeError, ValueError):
|
||||
raise HTTPException(422, "Invalid token position or size")
|
||||
if (x is None) != (y is None):
|
||||
x = y = None
|
||||
tid = str(t.get("id", "")) if keep_id else ""
|
||||
return {"id": tid if tid.isalnum() and len(tid) <= 16 else uuid.uuid4().hex[:12], "name": name, "x": x, "y": y,
|
||||
"size": size, "color": clean_color(t.get("color")), "hidden": bool(t.get("hidden"))}
|
||||
|
||||
|
||||
def install(api, db, current_user, lobby_for, dm_only, hub, event, store_dir: Path):
|
||||
store_dir.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
def path(mid: str) -> Path:
|
||||
return store_dir / f"{mid}.webp"
|
||||
|
||||
def own_map(con, mid, user):
|
||||
row = con.execute("SELECT * FROM maps WHERE id=? AND owner_id=?", (mid, user["id"])).fetchone()
|
||||
if not row:
|
||||
raise HTTPException(404, "Map not found")
|
||||
return row
|
||||
|
||||
def meta(r) -> dict:
|
||||
return {"id": r["id"], "name": r["name"], "w": r["w"], "h": r["h"], "grid": r["grid"]}
|
||||
|
||||
# ---- the DM's library
|
||||
|
||||
@api.get("/maps")
|
||||
def list_maps(user: dict = Depends(current_user)):
|
||||
with db() as con:
|
||||
rows = con.execute("SELECT * FROM maps WHERE owner_id=? ORDER BY created DESC", (user["id"],)).fetchall()
|
||||
return [{**meta(r), "tokens": len(json.loads(r["tokens"]))} for r in rows]
|
||||
|
||||
@api.post("/maps")
|
||||
async def upload_map(request: Request, name: str = "", user: dict = Depends(current_user)):
|
||||
"""Raw image bytes in the body (no multipart), the name as ?name=."""
|
||||
if int(request.headers.get("content-length") or 0) > MAX_UPLOAD:
|
||||
raise HTTPException(413, f"The file is too large (at most {MAX_UPLOAD // 1_000_000} MB)")
|
||||
chunks, size = [], 0
|
||||
async for c in request.stream():
|
||||
size += len(c)
|
||||
if size > MAX_UPLOAD:
|
||||
raise HTTPException(413, f"The file is too large (at most {MAX_UPLOAD // 1_000_000} MB)")
|
||||
chunks.append(c)
|
||||
with db() as con:
|
||||
if con.execute("SELECT COUNT(*) FROM maps WHERE owner_id=?", (user["id"],)).fetchone()[0] >= MAX_MAPS:
|
||||
raise HTTPException(409, "Map limit reached, delete one first")
|
||||
data, w, h = await run_in_threadpool(encode, b"".join(chunks))
|
||||
mid = uuid.uuid4().hex[:12]
|
||||
path(mid).write_bytes(data)
|
||||
with db() as con:
|
||||
con.execute("INSERT INTO maps (id, owner_id, name, w, h, grid, created) VALUES (?,?,?,?,?,?,?)",
|
||||
(mid, user["id"], name.strip()[:60] or "Map", w, h, round(max(w, h) / 30, 1), time.time()))
|
||||
return {"id": mid}
|
||||
|
||||
@api.get("/maps/{mid}")
|
||||
def read_map(mid: str, user: dict = Depends(current_user)):
|
||||
with db() as con:
|
||||
r = own_map(con, mid, user)
|
||||
return {**meta(r), "tokens": json.loads(r["tokens"]), "walls": json.loads(r["walls"])}
|
||||
|
||||
@api.put("/maps/{mid}")
|
||||
def update_map(mid: str, body: dict = Body(...), user: dict = Depends(current_user)):
|
||||
with db() as con:
|
||||
r = own_map(con, mid, user)
|
||||
name, grid, tokens = r["name"], r["grid"], r["tokens"]
|
||||
if "name" in body:
|
||||
name = str(body["name"]).strip()[:60] or r["name"]
|
||||
if "grid" in body:
|
||||
try:
|
||||
grid = float(body["grid"])
|
||||
except (TypeError, ValueError):
|
||||
raise HTTPException(422, "The grid must be a number")
|
||||
if not 4 <= grid <= max(r["w"], r["h"]) / 2:
|
||||
raise HTTPException(422, "Grid size (pixels per square) is out of range")
|
||||
if "tokens" in body:
|
||||
if not isinstance(body["tokens"], list) or len(body["tokens"]) > MAX_TOKENS:
|
||||
raise HTTPException(422, f"At most {MAX_TOKENS} tokens per map")
|
||||
tokens = json.dumps([clean_token(t, r["w"], r["h"]) for t in body["tokens"]])
|
||||
con.execute("UPDATE maps SET name=?, grid=?, tokens=? WHERE id=?", (name, grid, tokens, mid))
|
||||
lobbies = [x["id"] for x in con.execute("SELECT id FROM lobbies WHERE map_id=?", (mid,))]
|
||||
for lid in lobbies:
|
||||
hub.notify(lid, "map")
|
||||
return {"ok": True}
|
||||
|
||||
@api.delete("/maps/{mid}")
|
||||
def delete_map(mid: str, user: dict = Depends(current_user)):
|
||||
with db() as con:
|
||||
own_map(con, mid, user)
|
||||
lobbies = [x["id"] for x in con.execute("SELECT id FROM lobbies WHERE map_id=?", (mid,))]
|
||||
con.execute("UPDATE lobbies SET map_id=NULL WHERE map_id=?", (mid,))
|
||||
for lid in lobbies:
|
||||
con.execute("DELETE FROM lobby_tokens WHERE lobby_id=?", (lid,))
|
||||
con.execute("DELETE FROM maps WHERE id=?", (mid,))
|
||||
path(mid).unlink(missing_ok=True)
|
||||
for lid in lobbies:
|
||||
hub.notify(lid, "map")
|
||||
return {"ok": True}
|
||||
|
||||
@api.get("/maps/{mid}/image")
|
||||
def map_image(mid: str, user: dict = Depends(current_user)):
|
||||
with db() as con:
|
||||
own_map(con, mid, user)
|
||||
return image_response(path(mid))
|
||||
|
||||
def image_response(p: Path):
|
||||
if not p.exists():
|
||||
raise HTTPException(404, "Image missing")
|
||||
return Response(p.read_bytes(), media_type="image/webp", headers={"Cache-Control": "private, max-age=3600"})
|
||||
|
||||
# ---- the map on a lobby's table
|
||||
|
||||
def sync_players(con, lid: str):
|
||||
"""One token per member (unplaced until the player drags it onto the map); tokens of people who left are removed."""
|
||||
members = con.execute("SELECT user_id FROM lobby_members WHERE lobby_id=?", (lid,)).fetchall()
|
||||
have = {r["user_id"] for r in con.execute("SELECT user_id FROM lobby_tokens WHERE lobby_id=? AND user_id IS NOT NULL", (lid,))}
|
||||
ids = {m["user_id"] for m in members}
|
||||
for uid in ids - have:
|
||||
con.execute("INSERT INTO lobby_tokens (id, lobby_id, user_id, name, size) VALUES (?,?,?,?,1)", (uuid.uuid4().hex[:12], lid, uid, ""))
|
||||
for uid in have - ids:
|
||||
con.execute("DELETE FROM lobby_tokens WHERE lobby_id=? AND user_id=?", (lid, uid))
|
||||
|
||||
def lobby_map(con, lobby):
|
||||
return con.execute("SELECT * FROM maps WHERE id=?", (lobby["map_id"],)).fetchone() if lobby["map_id"] else None
|
||||
|
||||
def tokens_for(con, lobby, user, is_dm: bool) -> list:
|
||||
sync_players(con, lobby["id"])
|
||||
rows = con.execute("""SELECT t.*, u.username, m.color AS mcolor, c.data FROM lobby_tokens t
|
||||
LEFT JOIN users u ON u.id = t.user_id
|
||||
LEFT JOIN lobby_members m ON m.lobby_id = t.lobby_id AND m.user_id = t.user_id
|
||||
LEFT JOIN characters c ON c.id = m.character_id
|
||||
WHERE t.lobby_id=? ORDER BY t.rowid""", (lobby["id"],)).fetchall()
|
||||
out = []
|
||||
for r in rows:
|
||||
player = r["user_id"] is not None
|
||||
if r["hidden"] and not is_dm:
|
||||
continue
|
||||
name = r["name"]
|
||||
if player:
|
||||
try:
|
||||
name = (json.loads(r["data"]).get("name") if r["data"] else "") or r["username"]
|
||||
except ValueError:
|
||||
name = r["username"]
|
||||
out.append({"id": r["id"], "kind": "player" if player else "npc", "username": r["username"], "name": name,
|
||||
"x": r["x"], "y": r["y"], "size": r["size"], "hidden": bool(r["hidden"]),
|
||||
"color": (r["mcolor"] or "#999999") if player else r["color"],
|
||||
"can_move": is_dm or r["user_id"] == user["id"]})
|
||||
return out
|
||||
|
||||
@api.get("/lobbies/{lid}/map")
|
||||
def read_lobby_map(lid: str, user: dict = Depends(current_user)):
|
||||
with db() as con:
|
||||
lobby = lobby_for(con, lid, user)
|
||||
is_dm = lobby["owner_id"] == user["id"]
|
||||
m = lobby_map(con, lobby)
|
||||
return {"map": meta(m) if m else None, "tokens": tokens_for(con, lobby, user, is_dm) if m else []}
|
||||
|
||||
@api.get("/lobbies/{lid}/map/image")
|
||||
def lobby_map_image(lid: str, user: dict = Depends(current_user)):
|
||||
with db() as con:
|
||||
lobby = lobby_for(con, lid, user)
|
||||
if not lobby["map_id"]:
|
||||
raise HTTPException(404, "No map")
|
||||
mid = lobby["map_id"]
|
||||
return image_response(path(mid))
|
||||
|
||||
@api.put("/lobbies/{lid}/map")
|
||||
def set_lobby_map(lid: str, body: dict = Body(...), user: dict = Depends(current_user)):
|
||||
"""DM: show a map from the library (or null to clear the table). The map's prepared tokens are copied to the lobby."""
|
||||
with db() as con:
|
||||
dm_only(lobby_for(con, lid, user), user)
|
||||
mid = body.get("map_id")
|
||||
con.execute("DELETE FROM lobby_tokens WHERE lobby_id=?", (lid,))
|
||||
if mid:
|
||||
r = own_map(con, str(mid), user)
|
||||
for t in json.loads(r["tokens"]):
|
||||
con.execute("INSERT INTO lobby_tokens (id, lobby_id, name, x, y, color, size, hidden) VALUES (?,?,?,?,?,?,?,?)",
|
||||
(uuid.uuid4().hex[:12], lid, t["name"], t["x"], t["y"], t["color"], t["size"], int(t["hidden"])))
|
||||
sync_players(con, lid)
|
||||
event(con, lid, f"The DM put the map \"{r['name']}\" on the table", user["id"])
|
||||
else:
|
||||
event(con, lid, "The DM cleared the table", user["id"])
|
||||
con.execute("UPDATE lobbies SET map_id=? WHERE id=?", (str(mid) if mid else None, lid))
|
||||
hub.notify(lid, "map")
|
||||
return {"ok": True}
|
||||
|
||||
def need_map(con, lobby):
|
||||
m = lobby_map(con, lobby)
|
||||
if not m:
|
||||
raise HTTPException(409, "There is no map on the table")
|
||||
return m
|
||||
|
||||
@api.post("/lobbies/{lid}/tokens")
|
||||
def add_token(lid: str, body: dict = Body(...), user: dict = Depends(current_user)):
|
||||
with db() as con:
|
||||
lobby = lobby_for(con, lid, user)
|
||||
dm_only(lobby, user)
|
||||
m = need_map(con, lobby)
|
||||
if con.execute("SELECT COUNT(*) FROM lobby_tokens WHERE lobby_id=?", (lid,)).fetchone()[0] >= MAX_TOKENS:
|
||||
raise HTTPException(409, "Too many tokens")
|
||||
t = clean_token(body, m["w"], m["h"], keep_id=False)
|
||||
con.execute("INSERT INTO lobby_tokens (id, lobby_id, name, x, y, color, size, hidden) VALUES (?,?,?,?,?,?,?,?)",
|
||||
(t["id"], lid, t["name"], t["x"], t["y"], t["color"], t["size"], int(t["hidden"])))
|
||||
hub.notify(lid, "map")
|
||||
return {"id": t["id"]}
|
||||
|
||||
@api.put("/lobbies/{lid}/tokens/{tid}")
|
||||
def move_token(lid: str, tid: str, body: dict = Body(...), user: dict = Depends(current_user)):
|
||||
"""Players move their own token; the DM moves any and can also rename, recolour, resize or hide NPC tokens."""
|
||||
with db() as con:
|
||||
lobby = lobby_for(con, lid, user)
|
||||
is_dm = lobby["owner_id"] == user["id"]
|
||||
m = need_map(con, lobby)
|
||||
t = con.execute("SELECT * FROM lobby_tokens WHERE id=? AND lobby_id=?", (tid, lid)).fetchone()
|
||||
if not t or (not is_dm and t["user_id"] != user["id"]):
|
||||
raise HTTPException(404, "Token not found")
|
||||
merged = {"name": t["name"], "x": t["x"], "y": t["y"], "color": t["color"], "size": t["size"], "hidden": t["hidden"]}
|
||||
if "x" in body or "y" in body:
|
||||
merged["x"], merged["y"] = body.get("x"), body.get("y")
|
||||
if is_dm and t["user_id"] is None:
|
||||
for k in ("name", "color", "size", "hidden"):
|
||||
if k in body:
|
||||
merged[k] = body[k]
|
||||
elif is_dm and "size" in body:
|
||||
merged["size"] = body["size"]
|
||||
c = clean_token(merged, m["w"], m["h"])
|
||||
con.execute("UPDATE lobby_tokens SET name=?, x=?, y=?, color=?, size=?, hidden=? WHERE id=?",
|
||||
(c["name"] if t["user_id"] is None else t["name"], c["x"], c["y"], c["color"], c["size"], int(c["hidden"]), tid))
|
||||
hub.notify(lid, "map")
|
||||
return {"ok": True}
|
||||
|
||||
@api.delete("/lobbies/{lid}/tokens/{tid}")
|
||||
def delete_token(lid: str, tid: str, user: dict = Depends(current_user)):
|
||||
with db() as con:
|
||||
dm_only(lobby_for(con, lid, user), user)
|
||||
con.execute("DELETE FROM lobby_tokens WHERE id=? AND lobby_id=? AND user_id IS NULL", (tid, lid))
|
||||
hub.notify(lid, "map")
|
||||
return {"ok": True}
|
||||
|
||||
return sync_players
|
||||
Reference in New Issue
Block a user