ottoandClaude Sonnet 5.5
c1fdb7cc76
Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01HJhXuLD2Z6hv3Wqeq9UL9K
1041 lines
58 KiB
Python
1041 lines
58 KiB
Python
"""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 base64
|
|
import hashlib
|
|
import io
|
|
import json
|
|
import math
|
|
import re
|
|
import shutil
|
|
import time
|
|
import uuid
|
|
from pathlib import Path
|
|
|
|
from fastapi import Body, Depends, HTTPException, Query, Request
|
|
from fastapi.concurrency import run_in_threadpool
|
|
from fastapi.responses import Response
|
|
from PIL import Image, ImageChops, ImageDraw, ImageOps
|
|
|
|
from . import vision
|
|
from .portrait import image as portrait_image, tag as portrait_tag
|
|
|
|
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
|
|
MAX_WALLS = 400 # wall and door segments per map
|
|
MAX_SHARES = 30
|
|
MAX_SCENES = 12 # saved scenes per lobby
|
|
MAX_PROPS = 40 # custom prop images per user
|
|
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 "npc_id" not in [r["name"] for r in con.execute("PRAGMA table_info(lobby_tokens)")]:
|
|
con.execute("ALTER TABLE lobby_tokens ADD COLUMN npc_id TEXT")
|
|
con.execute("""CREATE TABLE IF NOT EXISTS lobby_explored (
|
|
lobby_id TEXT NOT NULL, user_id INTEGER NOT NULL, map_id TEXT NOT NULL, data BLOB NOT NULL, rev TEXT NOT NULL, last TEXT,
|
|
PRIMARY KEY (lobby_id, user_id))""")
|
|
con.execute("""CREATE TABLE IF NOT EXISTS lobby_reveals (
|
|
id TEXT PRIMARY KEY, lobby_id TEXT NOT NULL, x REAL NOT NULL, y REAL NOT NULL, radius REAL NOT NULL, dir REAL NOT NULL,
|
|
fov REAL NOT NULL, targets TEXT, created REAL NOT NULL)""")
|
|
con.execute("""CREATE TABLE IF NOT EXISTS lobby_scenes (
|
|
id TEXT PRIMARY KEY, lobby_id TEXT NOT NULL, name TEXT NOT NULL, map_id TEXT, state TEXT NOT NULL DEFAULT '{}', updated REAL NOT NULL)""")
|
|
con.execute("""CREATE TABLE IF NOT EXISTS user_props (
|
|
id TEXT PRIMARY KEY, owner_id INTEGER NOT NULL, name TEXT NOT NULL, created REAL NOT NULL)""")
|
|
con.execute("CREATE TABLE IF NOT EXISTS map_shares (map_id TEXT NOT NULL, user_id INTEGER NOT NULL, PRIMARY KEY (map_id, user_id))")
|
|
for table, col, ddl in (("lobbies", "map_id", "TEXT"), ("lobbies", "doors", "TEXT NOT NULL DEFAULT '{}'"),
|
|
("lobbies", "los", "INTEGER NOT NULL DEFAULT 1"), ("maps", "sight", "REAL NOT NULL DEFAULT 0"),
|
|
("lobbies", "sight", "REAL"), ("lobbies", "fog", "INTEGER NOT NULL DEFAULT 1"),
|
|
("lobby_tokens", "hp", "INTEGER"), ("lobby_tokens", "hp_max", "INTEGER"),
|
|
("lobby_tokens", "conds", "TEXT NOT NULL DEFAULT '{}'"), ("lobby_tokens", "show_hp", "INTEGER NOT NULL DEFAULT 0"),
|
|
("lobby_tokens", "show_cond", "INTEGER NOT NULL DEFAULT 0"),
|
|
("lobby_members", "vgroup", "INTEGER NOT NULL DEFAULT 0"),
|
|
("lobby_tokens", "prop", "TEXT"), ("lobby_tokens", "rot", "REAL NOT NULL DEFAULT 0"), ("lobby_tokens", "size_h", "REAL"),
|
|
("lobbies", "scene_id", "TEXT")):
|
|
if col not in [r["name"] for r in con.execute(f"PRAGMA table_info({table})")]:
|
|
con.execute(f"ALTER TABLE {table} ADD COLUMN {col} {ddl}")
|
|
|
|
|
|
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 ""
|
|
npc = str(t.get("npc_id") or "")
|
|
hp_max, hp = clean_int(t.get("hp_max"), 1, 9999), clean_int(t.get("hp"), -999, 9999)
|
|
prop = str(t.get("prop") or "")
|
|
prop = prop if re.fullmatch(r"[bc]:[a-z0-9_]{1,24}", prop) else ""
|
|
try:
|
|
rot = float(t.get("rot") or 0) % 360
|
|
size_h = None if t.get("size_h") in (None, "") else min(max(float(t["size_h"]), 0.25), 8)
|
|
except (TypeError, ValueError):
|
|
raise HTTPException(422, "Invalid rotation or size")
|
|
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")),
|
|
"npc_id": npc if npc.isalnum() and len(npc) <= 16 else "", "hp": hp, "hp_max": hp_max,
|
|
"prop": prop, "rot": rot, "size_h": size_h if prop else None}
|
|
|
|
|
|
def clean_int(v, lo, hi):
|
|
if v is None or v == "":
|
|
return None
|
|
try:
|
|
return max(lo, min(hi, int(float(v))))
|
|
except (TypeError, ValueError):
|
|
raise HTTPException(422, "Hit points must be whole numbers")
|
|
|
|
|
|
def clean_conds(raw) -> dict:
|
|
"""{condition id: level 1-3}"""
|
|
if not isinstance(raw, dict) or len(raw) > 20:
|
|
raise HTTPException(422, "Invalid conditions")
|
|
out = {}
|
|
for k, v in raw.items():
|
|
k = str(k)
|
|
if not (k.replace("_", "").isalnum() and len(k) <= 24):
|
|
continue
|
|
try:
|
|
lv = int(v)
|
|
except (TypeError, ValueError):
|
|
continue
|
|
if 1 <= lv <= 3:
|
|
out[k] = lv
|
|
return out
|
|
|
|
|
|
def clean_walls(raw, w: int, h: int) -> list:
|
|
if not isinstance(raw, list) or len(raw) > MAX_WALLS:
|
|
raise HTTPException(422, f"At most {MAX_WALLS} wall segments per map")
|
|
out = []
|
|
for t in raw:
|
|
try:
|
|
x1, y1, x2, y2 = (float(t[k]) for k in ("x1", "y1", "x2", "y2"))
|
|
except (TypeError, ValueError, KeyError, IndexError):
|
|
raise HTTPException(422, "Invalid wall")
|
|
x1, x2 = min(max(x1, 0), w), min(max(x2, 0), w)
|
|
y1, y2 = min(max(y1, 0), h), min(max(y2, 0), h)
|
|
if (x1, y1) == (x2, y2):
|
|
continue
|
|
wid = str(t.get("id", ""))
|
|
out.append({"id": wid if wid.isalnum() and len(wid) <= 16 else uuid.uuid4().hex[:10],
|
|
"x1": round(x1, 1), "y1": round(y1, 1), "x2": round(x2, 1), "y2": round(y2, 1),
|
|
"kind": "door" if t.get("kind") == "door" else "wall"})
|
|
return out
|
|
|
|
|
|
def install(api, db, current_user, lobby_for, dm_only, hub, event, store_dir: Path, can_read, health=lambda data: None):
|
|
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 usable_map(con, mid, user):
|
|
"""A map the user owns or that was shared with them (shared maps can be put on the table but not changed)."""
|
|
row = con.execute("""SELECT * FROM maps WHERE id=? AND (owner_id=? OR EXISTS
|
|
(SELECT 1 FROM map_shares s WHERE s.map_id = maps.id AND s.user_id=?))""", (mid, user["id"], 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"], "sight": r["sight"]}
|
|
|
|
# ---- the DM's library
|
|
|
|
@api.get("/maps")
|
|
def list_maps(user: dict = Depends(current_user)):
|
|
with db() as con:
|
|
rows = con.execute("""SELECT m.*, u.username AS author FROM maps m JOIN users u ON u.id = m.owner_id
|
|
WHERE m.owner_id=? OR EXISTS (SELECT 1 FROM map_shares s WHERE s.map_id = m.id AND s.user_id=?)
|
|
ORDER BY m.created DESC""", (user["id"], user["id"])).fetchall()
|
|
out = []
|
|
for r in rows:
|
|
mine = r["owner_id"] == user["id"]
|
|
shared = [x["username"] for x in con.execute(
|
|
"SELECT u.username FROM map_shares s JOIN users u ON u.id = s.user_id WHERE s.map_id=? ORDER BY u.username", (r["id"],))] if mine else []
|
|
out.append({**meta(r), "tokens": len(json.loads(r["tokens"])), "walls": len(json.loads(r["walls"])),
|
|
"mine": mine, "author": r["author"], "shared_with": shared})
|
|
return out
|
|
|
|
@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)
|
|
tokens = json.loads(r["tokens"])
|
|
for t in tokens: # NPC tokens: tell the editor whether the NPC has a portrait
|
|
t["portrait"] = ""
|
|
if t.get("npc_id") and can_read(con, t["npc_id"], user):
|
|
row = con.execute("SELECT data FROM characters WHERE id=?", (t["npc_id"],)).fetchone()
|
|
try:
|
|
t["portrait"] = portrait_tag(json.loads(row["data"]).get("portrait")) if row else ""
|
|
except ValueError:
|
|
pass
|
|
else:
|
|
t["npc_id"] = ""
|
|
return {**meta(r), "tokens": 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, walls, sight = r["name"], r["grid"], r["tokens"], r["walls"], r["sight"]
|
|
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"]])
|
|
if "walls" in body:
|
|
walls = json.dumps(clean_walls(body["walls"], r["w"], r["h"]))
|
|
if "sight" in body:
|
|
try:
|
|
sight = float(body["sight"])
|
|
except (TypeError, ValueError):
|
|
raise HTTPException(422, "The sight range must be a number")
|
|
if not 0 <= sight <= 1000:
|
|
raise HTTPException(422, "Sight range: 0 (unlimited) to 1000 squares")
|
|
con.execute("UPDATE maps SET name=?, grid=?, tokens=?, walls=?, sight=? WHERE id=?", (name, grid, tokens, walls, sight, 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,))
|
|
con.execute("DELETE FROM map_shares WHERE map_id=?", (mid,))
|
|
for lid in lobbies:
|
|
for tb in ("lobby_tokens", "lobby_explored", "lobby_reveals"):
|
|
con.execute(f"DELETE FROM {tb} WHERE lobby_id=?", (lid,))
|
|
con.execute("UPDATE lobbies SET scene_id=NULL WHERE scene_id IN (SELECT id FROM lobby_scenes WHERE map_id=?)", (mid,))
|
|
con.execute("DELETE FROM lobby_scenes WHERE map_id=?", (mid,))
|
|
con.execute("DELETE FROM maps WHERE id=?", (mid,))
|
|
path(mid).unlink(missing_ok=True)
|
|
(store_dir / f"{mid}.thumb.webp").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))
|
|
|
|
@api.get("/maps/{mid}/thumb")
|
|
def map_thumb(mid: str, user: dict = Depends(current_user)):
|
|
with db() as con:
|
|
usable_map(con, mid, user)
|
|
t = store_dir / f"{mid}.thumb.webp"
|
|
if not t.exists():
|
|
im = Image.open(path(mid)).convert("RGB")
|
|
im.thumbnail((320, 320))
|
|
im.save(t, "WEBP", quality=80)
|
|
return image_response(t)
|
|
|
|
@api.post("/maps/{mid}/shares")
|
|
def share_map(mid: str, body: dict = Body(...), user: dict = Depends(current_user)):
|
|
name = str(body.get("username", "")).strip()
|
|
with db() as con:
|
|
own_map(con, mid, user)
|
|
target = con.execute("SELECT id, username FROM users WHERE username=?", (name,)).fetchone()
|
|
if not target:
|
|
raise HTTPException(404, "No user with this name")
|
|
if target["id"] == user["id"]:
|
|
raise HTTPException(422, "This is your own map")
|
|
if con.execute("SELECT COUNT(*) FROM map_shares WHERE map_id=?", (mid,)).fetchone()[0] >= MAX_SHARES:
|
|
raise HTTPException(409, "Shared with too many users")
|
|
con.execute("INSERT OR IGNORE INTO map_shares VALUES (?,?)", (mid, target["id"]))
|
|
return {"username": target["username"]}
|
|
|
|
@api.delete("/maps/{mid}/shares/{username}")
|
|
def unshare_map(mid: str, username: str, user: dict = Depends(current_user)):
|
|
with db() as con:
|
|
own_map(con, mid, user)
|
|
con.execute("DELETE FROM map_shares WHERE map_id=? AND user_id=(SELECT id FROM users WHERE username=?)", (mid, username))
|
|
return {"ok": True}
|
|
|
|
@api.post("/maps/{mid}/copy")
|
|
def copy_map(mid: str, user: dict = Depends(current_user)):
|
|
"""Your own editable copy (image, grid, tokens, walls) of a map you own or that was shared with you."""
|
|
with db() as con:
|
|
r = usable_map(con, mid, user)
|
|
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")
|
|
nid = uuid.uuid4().hex[:12]
|
|
shutil.copyfile(path(mid), path(nid))
|
|
con.execute("INSERT INTO maps (id, owner_id, name, w, h, grid, tokens, walls, sight, created) VALUES (?,?,?,?,?,?,?,?,?,?)",
|
|
(nid, user["id"], f"{r['name']} (copy)"[:60], r["w"], r["h"], r["grid"], r["tokens"], r["walls"], r["sight"], time.time()))
|
|
return {"id": nid}
|
|
|
|
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"})
|
|
|
|
# ---- custom prop images (decor tokens such as chests or tables); the built-in ones are drawn by the frontend
|
|
|
|
props_dir = store_dir.parent / "props"
|
|
props_dir.mkdir(parents=True, exist_ok=True)
|
|
|
|
def encode_prop(raw: bytes) -> bytes:
|
|
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((512, 512))
|
|
im = im.convert("RGBA")
|
|
out = io.BytesIO()
|
|
im.save(out, "WEBP", quality=88, method=4)
|
|
return out.getvalue()
|
|
|
|
@api.get("/props")
|
|
def list_props(user: dict = Depends(current_user)):
|
|
with db() as con:
|
|
return [{"id": r["id"], "name": r["name"]} for r in
|
|
con.execute("SELECT id, name FROM user_props WHERE owner_id=? ORDER BY created", (user["id"],))]
|
|
|
|
@api.post("/props")
|
|
async def upload_prop(request: Request, name: str = "", user: dict = Depends(current_user)):
|
|
"""Raw image bytes in the body, the name as ?name=. Transparent PNGs work best (top view)."""
|
|
if int(request.headers.get("content-length") or 0) > 3_000_000:
|
|
raise HTTPException(413, "The file is too large (at most 3 MB)")
|
|
chunks, size = [], 0
|
|
async for c in request.stream():
|
|
size += len(c)
|
|
if size > 3_000_000:
|
|
raise HTTPException(413, "The file is too large (at most 3 MB)")
|
|
chunks.append(c)
|
|
with db() as con:
|
|
if con.execute("SELECT COUNT(*) FROM user_props WHERE owner_id=?", (user["id"],)).fetchone()[0] >= MAX_PROPS:
|
|
raise HTTPException(409, "Prop limit reached, delete one first")
|
|
data = await run_in_threadpool(encode_prop, b"".join(chunks))
|
|
pid = uuid.uuid4().hex[:12]
|
|
(props_dir / f"{pid}.webp").write_bytes(data)
|
|
with db() as con:
|
|
con.execute("INSERT INTO user_props (id, owner_id, name, created) VALUES (?,?,?,?)", (pid, user["id"], name.strip()[:30] or "Prop", time.time()))
|
|
return {"id": pid}
|
|
|
|
@api.delete("/props/{pid}")
|
|
def delete_prop(pid: str, user: dict = Depends(current_user)):
|
|
with db() as con:
|
|
if not con.execute("SELECT 1 FROM user_props WHERE id=? AND owner_id=?", (pid, user["id"])).fetchone():
|
|
raise HTTPException(404, "Prop not found")
|
|
con.execute("DELETE FROM user_props WHERE id=?", (pid,))
|
|
(props_dir / f"{pid}.webp").unlink(missing_ok=True)
|
|
return {"ok": True}
|
|
|
|
@api.get("/props/{pid}/image")
|
|
def prop_image(pid: str, user: dict = Depends(current_user)):
|
|
with db() as con:
|
|
ok = con.execute("SELECT 1 FROM user_props WHERE id=? AND owner_id=?", (pid, user["id"])).fetchone()
|
|
if not ok:
|
|
raise HTTPException(404, "Prop not found")
|
|
return image_response(props_dir / f"{pid}.webp")
|
|
|
|
@api.get("/lobbies/{lid}/prop-image/{pid}")
|
|
def lobby_prop_image(lid: str, pid: str, user: dict = Depends(current_user)):
|
|
"""Members of a lobby may load the prop images of the lobby's DM."""
|
|
with db() as con:
|
|
lobby = lobby_for(con, lid, user)
|
|
ok = con.execute("SELECT 1 FROM user_props WHERE id=? AND owner_id=?", (pid, lobby["owner_id"])).fetchone()
|
|
if not ok or not re.fullmatch(r"[0-9a-f]{12}", pid):
|
|
raise HTTPException(404, "Prop not found")
|
|
return image_response(props_dir / f"{pid}.webp")
|
|
|
|
# ---- 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, show_hp, show_cond) VALUES (?,?,?,?,1,1,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 put_token(con, lid, t, npc=None, tid=None):
|
|
"""Inserts a (cleaned or snapshot) token row for the lobby."""
|
|
con.execute("""INSERT INTO lobby_tokens (id, lobby_id, name, x, y, color, size, hidden, npc_id, hp, hp_max, prop, rot, size_h)
|
|
VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?)""",
|
|
(tid or uuid.uuid4().hex[:12], lid, t["name"], t["x"], t["y"], t["color"], t["size"], int(bool(t["hidden"])), npc,
|
|
t.get("hp") if t.get("hp") is not None else t.get("hp_max"), t.get("hp_max"), t.get("prop") or None, t.get("rot") or 0, t.get("size_h")))
|
|
|
|
def mask_dims(m):
|
|
f = max(1, -(-max(m["w"], m["h"]) // 512))
|
|
return f, -(-m["w"] // f), -(-m["h"] // f)
|
|
|
|
def explore(con, lobby, viewer, m, polys, key):
|
|
"""Adds what the player sees right now to the lobby's memory of explored ground (a small mask, 1 cell = f pixels)."""
|
|
lid = lobby["id"]
|
|
f, mw, mh = mask_dims(m)
|
|
row = con.execute("SELECT * FROM lobby_explored WHERE lobby_id=? AND user_id=?", (lid, viewer)).fetchone()
|
|
if row and row["map_id"] != m["id"]:
|
|
row = None
|
|
if row and row["last"] == key:
|
|
return row["rev"]
|
|
old = None
|
|
if row:
|
|
old = Image.open(io.BytesIO(row["data"]))
|
|
old.load()
|
|
if old.size != (mw, mh):
|
|
old = None
|
|
base = old or Image.new("L", (mw, mh), 0)
|
|
cur = Image.new("L", (mw, mh), 0)
|
|
for poly in polys:
|
|
if len(poly) > 2:
|
|
ImageDraw.Draw(cur).polygon([(x / f, y / f) for x, y in poly], fill=255)
|
|
new = ImageChops.lighter(base, cur)
|
|
rev = hashlib.md5(new.tobytes()).hexdigest()[:10]
|
|
if row and row["rev"] == rev:
|
|
con.execute("UPDATE lobby_explored SET last=? WHERE lobby_id=? AND user_id=?", (key, lid, viewer))
|
|
return rev
|
|
out = io.BytesIO()
|
|
new.save(out, "PNG", optimize=True)
|
|
con.execute("""INSERT INTO lobby_explored (lobby_id, user_id, map_id, data, rev, last) VALUES (?,?,?,?,?,?)
|
|
ON CONFLICT(lobby_id, user_id) DO UPDATE SET map_id=excluded.map_id, data=excluded.data, rev=excluded.rev, last=excluded.last""",
|
|
(lid, viewer, m["id"], out.getvalue(), rev, key))
|
|
return rev
|
|
|
|
def lobby_map(con, lobby):
|
|
return con.execute("SELECT * FROM maps WHERE id=?", (lobby["map_id"],)).fetchone() if lobby["map_id"] else None
|
|
|
|
def view_ctx(con, lobby, m, user, as_name=None):
|
|
"""Whose eyes the table is shown through. Players: their own token. DM: everything, or (as=name) one player's view."""
|
|
is_dm = lobby["owner_id"] == user["id"]
|
|
viewer = None
|
|
if is_dm and as_name:
|
|
row = con.execute("""SELECT m.user_id FROM lobby_members m JOIN users u ON u.id = m.user_id
|
|
WHERE m.lobby_id=? AND u.username=?""", (lobby["id"], as_name)).fetchone()
|
|
if not row:
|
|
raise HTTPException(404, "No such player in this lobby")
|
|
viewer = row["user_id"]
|
|
elif not is_dm:
|
|
viewer = user["id"]
|
|
walls = json.loads(m["walls"])
|
|
doors = json.loads(lobby["doors"] or "{}")
|
|
ctx = {"is_dm": is_dm, "viewer": viewer, "walls": walls, "doors": doors, "los": False, "empty": False, "polys": [],
|
|
"rev": "full:" + m["id"], "key": None}
|
|
if viewer is None:
|
|
return ctx
|
|
ctx["rev"] = "open:" + m["id"]
|
|
if not (lobby["los"] and walls):
|
|
return ctx
|
|
ctx["los"] = True
|
|
blocking = [(w["x1"], w["y1"], w["x2"], w["y2"]) for w in walls if w["kind"] == "wall" or not doors.get(w["id"])]
|
|
sight = lobby["sight"] if lobby["sight"] is not None else m["sight"]
|
|
radius = round(sight * m["grid"], 1)
|
|
me = con.execute("""SELECT u.username, mm.vgroup FROM lobby_members mm JOIN users u ON u.id = mm.user_id
|
|
WHERE mm.lobby_id=? AND mm.user_id=?""", (lobby["id"], viewer)).fetchone()
|
|
if me and me["vgroup"]: # a sight group: everybody in it sees what any of them sees
|
|
toks = con.execute("""SELECT t.x, t.y FROM lobby_tokens t JOIN lobby_members mm ON mm.lobby_id = t.lobby_id AND mm.user_id = t.user_id
|
|
WHERE t.lobby_id=? AND mm.vgroup=? AND t.x IS NOT NULL ORDER BY t.rowid""", (lobby["id"], me["vgroup"])).fetchall()
|
|
else:
|
|
toks = con.execute("SELECT x, y FROM lobby_tokens WHERE lobby_id=? AND user_id=? AND x IS NOT NULL", (lobby["id"], viewer)).fetchall()
|
|
origins = [(round(t["x"]), round(t["y"])) for t in toks]
|
|
zones = []
|
|
for z in con.execute("SELECT * FROM lobby_reveals WHERE lobby_id=? ORDER BY created", (lobby["id"],)):
|
|
tg = json.loads(z["targets"]) if z["targets"] else None
|
|
if tg is None or (me and me["username"] in tg):
|
|
zones.append(z)
|
|
if not origins and not zones: # not on the map yet and nothing revealed: sees nothing
|
|
ctx.update(empty=True, rev="empty")
|
|
return ctx
|
|
polys = []
|
|
for (x, y) in origins:
|
|
k = hashlib.sha1(json.dumps([m["id"], blocking, radius, x, y]).encode()).hexdigest()
|
|
polys.append(vision.cached_polygon(k, lambda x=x, y=y: vision.polygon((x, y), blocking, m["w"], m["h"], radius)))
|
|
zkeys = []
|
|
for z in zones:
|
|
ctx_poly, zk = zone_polygon(m, blocking, z)
|
|
polys.append(ctx_poly)
|
|
zkeys.append(zk)
|
|
key = hashlib.sha1(json.dumps([m["id"], blocking, radius, origins, zkeys]).encode()).hexdigest()
|
|
ctx.update(polys=polys, key=key, rev=key[:12])
|
|
if lobby["fog"]:
|
|
ctx["fog"] = explore(con, lobby, viewer, m, polys, key)
|
|
return ctx
|
|
|
|
def zone_polygon(m, blocking, z):
|
|
zr = round(max(0.5, z["radius"]) * m["grid"], 1)
|
|
sector = None if z["fov"] >= 359 else (math.radians(z["dir"]), math.radians(z["fov"]) / 2)
|
|
zk = hashlib.sha1(json.dumps([m["id"], blocking, zr, z["x"], z["y"], z["dir"], z["fov"]]).encode()).hexdigest()
|
|
return vision.cached_polygon(zk, lambda: vision.polygon((z["x"], z["y"]), blocking, m["w"], m["h"], zr, sector)), zk
|
|
|
|
def tokens_for(con, lobby, user, ctx) -> list:
|
|
sync_players(con, lobby["id"])
|
|
rows = con.execute("""SELECT t.*, u.username, m.color AS mcolor, c.data, n.data AS ndata FROM lobby_tokens t
|
|
LEFT JOIN characters n ON n.id = t.npc_id
|
|
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()
|
|
full = ctx["viewer"] is None
|
|
turn_tok = "" # players only learn an NPC's name while it is that NPC's turn
|
|
try:
|
|
cb = json.loads(lobby["combat"]) if lobby["combat"] else None
|
|
if cb:
|
|
turn_tok = next((e.get("token", "") for e in cb["entries"] if e["id"] == cb.get("current")), "")
|
|
except (ValueError, KeyError, TypeError):
|
|
pass
|
|
out = []
|
|
for r in rows:
|
|
player = r["user_id"] is not None
|
|
own = player and r["user_id"] == ctx["viewer"]
|
|
if not full and not own:
|
|
if r["hidden"] or r["x"] is None:
|
|
continue
|
|
if ctx["los"] and (ctx["empty"] or not vision.inside_any((r["x"], r["y"]), ctx["polys"])):
|
|
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"]
|
|
if not player and not full and r["id"] != turn_tok and not r["prop"]:
|
|
name = ""
|
|
pt = ""
|
|
src = r["data"] if player else r["ndata"]
|
|
if src:
|
|
try:
|
|
pt = portrait_tag(json.loads(src).get("portrait"))
|
|
except ValueError:
|
|
pass
|
|
status = {}
|
|
if full or own or r["show_hp"]:
|
|
status.update(hp=r["hp"], hp_max=r["hp_max"])
|
|
if full or own or r["show_cond"]:
|
|
status["conds"] = json.loads(r["conds"] or "{}")
|
|
if full:
|
|
status.update(show_hp=bool(r["show_hp"]), show_cond=bool(r["show_cond"]), hp_hint=health(src) if src else None)
|
|
if r["prop"]:
|
|
status = {}
|
|
out.append({"id": r["id"], "kind": "player" if player else "prop" if r["prop"] else "npc", "prop": r["prop"], "rot": r["rot"], "size_h": r["size_h"], "username": r["username"], "name": name, "portrait": pt, **status,
|
|
"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": ctx["is_dm"] or (r["user_id"] == user["id"] and not (ctx["los"] and r["x"] is None)),
|
|
**({"npc_id": r["npc_id"]} if full and r["npc_id"] else {})})
|
|
return out
|
|
|
|
@api.get("/lobbies/{lid}/map")
|
|
def read_lobby_map(lid: str, as_: str = Query("", alias="as"), user: dict = Depends(current_user)):
|
|
"""Map and tokens as the viewer may see them. Walls are only sent to the DM (in the full view)."""
|
|
with db() as con:
|
|
lobby = lobby_for(con, lid, user)
|
|
m = lobby_map(con, lobby)
|
|
if not m:
|
|
return {"map": None, "tokens": []}
|
|
ctx = view_ctx(con, lobby, m, user, as_)
|
|
res = {"map": meta(m), "tokens": tokens_for(con, lobby, user, ctx),
|
|
"vision": {"los": ctx["los"], "empty": ctx["empty"], "rev": ctx["rev"], "fog": ctx.get("fog")}}
|
|
if ctx["is_dm"]:
|
|
res["los_on"] = bool(lobby["los"])
|
|
res["sight"] = lobby["sight"] if lobby["sight"] is not None else m["sight"]
|
|
res["sight_map"] = m["sight"]
|
|
res["fog_on"] = bool(lobby["fog"])
|
|
res["reveals"] = [{"id": z["id"], "x": z["x"], "y": z["y"], "radius": z["radius"], "dir": z["dir"], "fov": z["fov"],
|
|
"to": json.loads(z["targets"]) if z["targets"] else None}
|
|
for z in con.execute("SELECT * FROM lobby_reveals WHERE lobby_id=? ORDER BY created", (lobby["id"],))]
|
|
if ctx["viewer"] is None:
|
|
res["walls"] = [{**w, "open": bool(ctx["doors"].get(w["id"]))} for w in ctx["walls"]]
|
|
return res
|
|
|
|
@api.get("/lobbies/{lid}/map/view")
|
|
def lobby_map_view(lid: str, as_: str = Query("", alias="as"), user: dict = Depends(current_user)):
|
|
"""The map image for this viewer. With line of sight on, only the part their character sees is sent
|
|
(cut out on the server, the rest is black), together with its position in the X-Crop header (x,y,w,h)."""
|
|
with db() as con:
|
|
lobby = lobby_for(con, lid, user)
|
|
m = lobby_map(con, lobby)
|
|
if not m:
|
|
raise HTTPException(404, "No map")
|
|
ctx = view_ctx(con, lobby, m, user, as_)
|
|
if ctx["empty"]:
|
|
raise HTTPException(404, "Nothing to see yet")
|
|
if ctx["los"]:
|
|
data, (x, y, w, h) = vision.cut(path(m["id"]), ctx["polys"], ctx["key"])
|
|
return Response(data, media_type="image/webp", headers={"X-Crop": f"{x},{y},{w},{h}", "Cache-Control": "no-store"})
|
|
return Response(path(m["id"]).read_bytes(), media_type="image/webp",
|
|
headers={"X-Crop": f"0,0,{m['w']},{m['h']}", "Cache-Control": "private, max-age=3600"})
|
|
|
|
@api.get("/lobbies/{lid}/map/explored")
|
|
def lobby_map_explored(lid: str, as_: str = Query("", alias="as"), user: dict = Depends(current_user)):
|
|
"""Ground the player has already seen, as a (smaller) picture: black where unexplored. The client dims it."""
|
|
with db() as con:
|
|
lobby = lobby_for(con, lid, user)
|
|
m = lobby_map(con, lobby)
|
|
if not m:
|
|
raise HTTPException(404, "No map")
|
|
ctx = view_ctx(con, lobby, m, user, as_)
|
|
row = con.execute("SELECT data, rev FROM lobby_explored WHERE lobby_id=? AND user_id=? AND map_id=?",
|
|
(lid, ctx["viewer"], m["id"])).fetchone() if ctx["viewer"] is not None and lobby["fog"] else None
|
|
if not row:
|
|
raise HTTPException(404, "Nothing explored")
|
|
hit = vision.fog_cache.fetch((m["id"], row["rev"]))
|
|
if not hit:
|
|
im = vision.source(path(m["id"]))
|
|
k = min(1.0, 2048 / max(im.size))
|
|
small = im.resize((max(1, round(im.width * k)), max(1, round(im.height * k)))) if k < 1 else im
|
|
mask = Image.open(io.BytesIO(row["data"])).resize(small.size, Image.BILINEAR)
|
|
piece = Image.composite(small, Image.new("RGB", small.size, (0, 0, 0)), mask)
|
|
out = io.BytesIO()
|
|
piece.save(out, "WEBP", quality=70, method=2)
|
|
hit = out.getvalue()
|
|
vision.fog_cache.put((m["id"], row["rev"]), hit)
|
|
return Response(hit, media_type="image/webp", headers={"Cache-Control": "private, max-age=3600"})
|
|
|
|
@api.put("/lobbies/{lid}/vision-groups")
|
|
def set_vision_groups(lid: str, body: dict = Body(...), user: dict = Depends(current_user)):
|
|
"""DM: players in the same group (1-4) share what they see. 0 = sees only for themselves."""
|
|
with db() as con:
|
|
dm_only(lobby_for(con, lid, user), user)
|
|
if body.get("preset") in ("shared", "separate"):
|
|
con.execute("UPDATE lobby_members SET vgroup=? WHERE lobby_id=?", (1 if body["preset"] == "shared" else 0, lid))
|
|
for name, g in (body.get("groups") or {}).items():
|
|
con.execute("""UPDATE lobby_members SET vgroup=? WHERE lobby_id=? AND user_id=(SELECT id FROM users WHERE username=?)""",
|
|
(clean_int(g, 0, 4) or 0, lid, str(name)))
|
|
hub.notify(lid, "map")
|
|
hub.notify(lid, "lobby")
|
|
return {"ok": True}
|
|
|
|
@api.post("/lobbies/{lid}/reveals")
|
|
def add_reveal(lid: str, body: dict = Body(...), user: dict = Depends(current_user)):
|
|
"""DM: show an area (circle or cone, blocked by walls) to all players or some of them. Stays until removed;
|
|
what was shown is also added to the players' memory of explored ground."""
|
|
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_reveals WHERE lobby_id=?", (lid,)).fetchone()[0] >= 12:
|
|
raise HTTPException(409, "Too many revealed areas, hide some first")
|
|
try:
|
|
x, y = min(max(float(body["x"]), 0), m["w"]), min(max(float(body["y"]), 0), m["h"])
|
|
radius = min(max(float(body.get("radius", 8)), 0.5), 200)
|
|
dr = float(body.get("dir", 0)) % 360
|
|
fov = min(max(float(body.get("fov", 360)), 10), 360)
|
|
except (KeyError, TypeError, ValueError):
|
|
raise HTTPException(422, "Invalid area")
|
|
names = body.get("to")
|
|
targets = None
|
|
members = {r["username"]: r["user_id"] for r in con.execute("""SELECT u.username, mm.user_id FROM lobby_members mm
|
|
JOIN users u ON u.id = mm.user_id WHERE mm.lobby_id=?""", (lid,))}
|
|
if isinstance(names, list):
|
|
targets = [n for n in names if n in members]
|
|
if not targets:
|
|
raise HTTPException(422, "Pick at least one player")
|
|
rid = uuid.uuid4().hex[:12]
|
|
con.execute("INSERT INTO lobby_reveals (id, lobby_id, x, y, radius, dir, fov, targets, created) VALUES (?,?,?,?,?,?,?,?,?)",
|
|
(rid, lid, x, y, radius, dr, fov, json.dumps(targets) if targets is not None else None, time.time()))
|
|
if lobby["fog"]: # the memory keeps it after the area is hidden again
|
|
walls, doors = json.loads(m["walls"]), json.loads(lobby["doors"] or "{}")
|
|
blocking = [(w["x1"], w["y1"], w["x2"], w["y2"]) for w in walls if w["kind"] == "wall" or not doors.get(w["id"])]
|
|
poly, _ = zone_polygon(m, blocking, {"radius": radius, "fov": fov, "dir": dr, "x": x, "y": y})
|
|
for n in (targets if targets is not None else members):
|
|
explore(con, lobby, members[n], m, [poly], "reveal:" + rid)
|
|
hub.notify(lid, "map")
|
|
return {"id": rid}
|
|
|
|
@api.delete("/lobbies/{lid}/reveals/{rid}")
|
|
def remove_reveal(lid: str, rid: str, user: dict = Depends(current_user)):
|
|
with db() as con:
|
|
dm_only(lobby_for(con, lid, user), user)
|
|
con.execute("DELETE FROM lobby_reveals WHERE id=? AND lobby_id=?", (rid, lid))
|
|
hub.notify(lid, "map")
|
|
return {"ok": True}
|
|
|
|
@api.delete("/lobbies/{lid}/explored")
|
|
def forget_explored(lid: str, user: dict = Depends(current_user)):
|
|
with db() as con:
|
|
dm_only(lobby_for(con, lid, user), user)
|
|
con.execute("DELETE FROM lobby_explored WHERE lobby_id=?", (lid,))
|
|
hub.notify(lid, "map")
|
|
return {"ok": True}
|
|
|
|
@api.put("/lobbies/{lid}/doors/{wid}")
|
|
def set_door(lid: str, wid: 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 not any(w["id"] == wid and w["kind"] == "door" for w in json.loads(m["walls"])):
|
|
raise HTTPException(404, "No such door")
|
|
doors = json.loads(lobby["doors"] or "{}")
|
|
if body.get("open"):
|
|
doors[wid] = True
|
|
else:
|
|
doors.pop(wid, None)
|
|
con.execute("UPDATE lobbies SET doors=? WHERE id=?", (json.dumps(doors), lid))
|
|
hub.notify(lid, "map")
|
|
return {"ok": True}
|
|
|
|
@api.put("/lobbies/{lid}/sight")
|
|
def set_sight(lid: str, body: dict = Body(...), user: dict = Depends(current_user)):
|
|
"""DM: switch line of sight on or off for the table (off: players see the whole map)."""
|
|
with db() as con:
|
|
dm_only(lobby_for(con, lid, user), user)
|
|
if "on" in body:
|
|
con.execute("UPDATE lobbies SET los=? WHERE id=?", (1 if body.get("on") else 0, lid))
|
|
if "range" in body: # squares; null = the map's own value
|
|
v = None if body["range"] in (None, "") else clean_int(body["range"], 0, 1000)
|
|
con.execute("UPDATE lobbies SET sight=? WHERE id=?", (v, lid))
|
|
if "fog" in body:
|
|
con.execute("UPDATE lobbies SET fog=? WHERE id=?", (1 if body["fog"] else 0, lid))
|
|
hub.notify(lid, "map")
|
|
return {"ok": True}
|
|
|
|
# ---- scenes: saved states of the table (map + tokens + doors + reveals + explored ground + combat)
|
|
|
|
TOKEN_COLS = ("id", "name", "x", "y", "color", "size", "hidden", "npc_id", "hp", "hp_max", "conds", "show_hp", "show_cond", "prop", "rot", "size_h")
|
|
|
|
def snapshot(con, lid) -> str:
|
|
lobby = con.execute("SELECT * FROM lobbies WHERE id=?", (lid,)).fetchone()
|
|
names = {r["id"]: r["username"] for r in con.execute("SELECT id, username FROM users WHERE id IN (SELECT user_id FROM lobby_members WHERE lobby_id=?)", (lid,))}
|
|
toks = []
|
|
for r in con.execute("SELECT * FROM lobby_tokens WHERE lobby_id=? ORDER BY rowid", (lid,)):
|
|
d = {k: r[k] for k in TOKEN_COLS}
|
|
if r["user_id"] is not None:
|
|
if r["user_id"] not in names:
|
|
continue
|
|
d["username"] = names[r["user_id"]]
|
|
toks.append(d)
|
|
reveals = [{k: z[k] for k in ("id", "x", "y", "radius", "dir", "fov", "targets", "created")}
|
|
for z in con.execute("SELECT * FROM lobby_reveals WHERE lobby_id=?", (lid,))]
|
|
explored = [{"username": names[z["user_id"]], "map_id": z["map_id"], "rev": z["rev"], "data": base64.b64encode(z["data"]).decode()}
|
|
for z in con.execute("SELECT * FROM lobby_explored WHERE lobby_id=?", (lid,)) if z["user_id"] in names]
|
|
return json.dumps({"tokens": toks, "reveals": reveals, "explored": explored, "doors": lobby["doors"], "sight": lobby["sight"],
|
|
"fog": lobby["fog"], "los": lobby["los"], "combat": lobby["combat"]})
|
|
|
|
def restore(con, lid, state: dict):
|
|
con.execute("DELETE FROM lobby_tokens WHERE lobby_id=?", (lid,))
|
|
con.execute("DELETE FROM lobby_explored WHERE lobby_id=?", (lid,))
|
|
con.execute("DELETE FROM lobby_reveals WHERE lobby_id=?", (lid,))
|
|
ids = {r["username"]: r["user_id"] for r in con.execute("SELECT u.username, m.user_id FROM lobby_members m JOIN users u ON u.id = m.user_id WHERE m.lobby_id=?", (lid,))}
|
|
sync_players(con, lid)
|
|
for d in state.get("tokens", []):
|
|
if d.get("username"):
|
|
uid = ids.get(d["username"])
|
|
if uid is not None:
|
|
con.execute("""UPDATE lobby_tokens SET x=?, y=?, hidden=?, hp=?, hp_max=?, conds=?, show_hp=?, show_cond=?, size=? WHERE lobby_id=? AND user_id=?""",
|
|
(d["x"], d["y"], d["hidden"], d["hp"], d["hp_max"], d["conds"], d["show_hp"], d["show_cond"], d["size"], lid, uid))
|
|
else:
|
|
con.execute(f"INSERT INTO lobby_tokens (lobby_id, {', '.join(TOKEN_COLS)}) VALUES (?{',?' * len(TOKEN_COLS)})", (lid, *[d[k] for k in TOKEN_COLS]))
|
|
for z in state.get("reveals", []):
|
|
con.execute("INSERT INTO lobby_reveals (id, lobby_id, x, y, radius, dir, fov, targets, created) VALUES (?,?,?,?,?,?,?,?,?)",
|
|
(z["id"], lid, z["x"], z["y"], z["radius"], z["dir"], z["fov"], z["targets"], z["created"]))
|
|
for z in state.get("explored", []):
|
|
if z["username"] in ids:
|
|
con.execute("INSERT INTO lobby_explored (lobby_id, user_id, map_id, data, rev, last) VALUES (?,?,?,?,?,NULL)",
|
|
(lid, ids[z["username"]], z["map_id"], base64.b64decode(z["data"]), z["rev"]))
|
|
con.execute("UPDATE lobbies SET doors=?, sight=?, fog=?, los=?, combat=? WHERE id=?",
|
|
(state.get("doors", "{}"), state.get("sight"), state.get("fog", 1), state.get("los", 1), state.get("combat"), lid))
|
|
|
|
def fresh_state(con, lid, r, user):
|
|
"""The state of a map that was just put on the table: its prepared tokens, nothing explored."""
|
|
con.execute("DELETE FROM lobby_tokens WHERE lobby_id=?", (lid,))
|
|
con.execute("DELETE FROM lobby_explored WHERE lobby_id=?", (lid,))
|
|
con.execute("DELETE FROM lobby_reveals WHERE lobby_id=?", (lid,))
|
|
con.execute("UPDATE lobbies SET sight=NULL, doors='{}', combat=NULL WHERE id=?", (lid,))
|
|
for t in json.loads(r["tokens"]):
|
|
npc = t.get("npc_id") if t.get("npc_id") and can_read(con, t["npc_id"], user) else None # only NPCs this DM may open
|
|
put_token(con, lid, t, npc)
|
|
sync_players(con, lid)
|
|
|
|
def autosave(con, lid):
|
|
"""Stores the table's current state in its scene (creating the scene for tables that predate scenes)."""
|
|
lobby = con.execute("SELECT * FROM lobbies WHERE id=?", (lid,)).fetchone()
|
|
if not lobby["map_id"]:
|
|
return
|
|
now = time.time()
|
|
if lobby["scene_id"] and con.execute("SELECT 1 FROM lobby_scenes WHERE id=? AND lobby_id=?", (lobby["scene_id"], lid)).fetchone():
|
|
con.execute("UPDATE lobby_scenes SET state=?, updated=? WHERE id=?", (snapshot(con, lid), now, lobby["scene_id"]))
|
|
else:
|
|
new_scene(con, lid, lobby["map_id"], snapshot(con, lid))
|
|
|
|
def new_scene(con, lid, mid, state, name=None) -> str:
|
|
if con.execute("SELECT COUNT(*) FROM lobby_scenes WHERE lobby_id=?", (lid,)).fetchone()[0] >= MAX_SCENES:
|
|
raise HTTPException(409, f"Too many saved scenes (max {MAX_SCENES}) - delete one first")
|
|
m = con.execute("SELECT name FROM maps WHERE id=?", (mid,)).fetchone()
|
|
base = (name or (m["name"] if m else "Scene")).strip()[:50] or "Scene"
|
|
taken = {r["name"] for r in con.execute("SELECT name FROM lobby_scenes WHERE lobby_id=?", (lid,))}
|
|
nm, i = base, 2
|
|
while nm in taken:
|
|
nm, i = f"{base} ({i})", i + 1
|
|
sid = uuid.uuid4().hex[:12]
|
|
con.execute("INSERT INTO lobby_scenes (id, lobby_id, name, map_id, state, updated) VALUES (?,?,?,?,?,?)", (sid, lid, nm, mid, state, time.time()))
|
|
con.execute("UPDATE lobbies SET scene_id=? WHERE id=?", (sid, lid))
|
|
return sid
|
|
|
|
@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 table's current state is kept as a scene;
|
|
the new map starts fresh with its prepared tokens as a scene of its own."""
|
|
with db() as con:
|
|
dm_only(lobby_for(con, lid, user), user)
|
|
mid = body.get("map_id")
|
|
r = usable_map(con, str(mid), user) if mid else None
|
|
autosave(con, lid)
|
|
con.execute("UPDATE lobbies SET scene_id=NULL WHERE id=?", (lid,))
|
|
if r:
|
|
fresh_state(con, lid, r, user)
|
|
con.execute("UPDATE lobbies SET map_id=? WHERE id=?", (r["id"], lid))
|
|
new_scene(con, lid, r["id"], snapshot(con, lid))
|
|
event(con, lid, f"The DM put the map \"{r['name']}\" on the table", user["id"])
|
|
else:
|
|
for tb in ("lobby_tokens", "lobby_explored", "lobby_reveals"):
|
|
con.execute(f"DELETE FROM {tb} WHERE lobby_id=?", (lid,))
|
|
con.execute("UPDATE lobbies SET map_id=NULL, sight=NULL, doors='{}', combat=NULL WHERE id=?", (lid,))
|
|
event(con, lid, "The DM cleared the table", user["id"])
|
|
hub.notify(lid, "map")
|
|
hub.notify(lid, "lobby")
|
|
return {"ok": True}
|
|
|
|
@api.get("/lobbies/{lid}/scenes")
|
|
def list_scenes(lid: str, user: dict = Depends(current_user)):
|
|
with db() as con:
|
|
lobby = lobby_for(con, lid, user)
|
|
dm_only(lobby, user)
|
|
if lobby["map_id"] and not con.execute("SELECT 1 FROM lobby_scenes WHERE id=? AND lobby_id=?", (lobby["scene_id"] or "", lid)).fetchone():
|
|
try:
|
|
autosave(con, lid) # a table that predates scenes becomes one
|
|
except HTTPException:
|
|
pass
|
|
lobby = lobby_for(con, lid, user)
|
|
rows = con.execute("""SELECT s.id, s.name, s.map_id, s.updated, m.name AS map_name, LENGTH(s.state) AS size FROM lobby_scenes s
|
|
LEFT JOIN maps m ON m.id = s.map_id WHERE s.lobby_id=? ORDER BY s.updated DESC""", (lid,)).fetchall()
|
|
return [{"id": r["id"], "name": r["name"], "map_id": r["map_id"], "map_name": r["map_name"] or "", "updated": r["updated"],
|
|
"current": r["id"] == lobby["scene_id"]} for r in rows]
|
|
|
|
@api.put("/lobbies/{lid}/scene")
|
|
def switch_scene(lid: str, body: dict = Body(...), user: dict = Depends(current_user)):
|
|
"""DM: continue another saved scene. The current table is saved first, so nothing is lost when switching back."""
|
|
with db() as con:
|
|
dm_only(lobby_for(con, lid, user), user)
|
|
sc = con.execute("SELECT * FROM lobby_scenes WHERE id=? AND lobby_id=?", (str(body.get("scene_id", "")), lid)).fetchone()
|
|
if not sc:
|
|
raise HTTPException(404, "Scene not found")
|
|
r = usable_map(con, sc["map_id"], user) if sc["map_id"] else None
|
|
if not r:
|
|
raise HTTPException(409, "The map of this scene is no longer available")
|
|
autosave(con, lid)
|
|
restore(con, lid, json.loads(sc["state"]))
|
|
con.execute("UPDATE lobbies SET map_id=?, scene_id=? WHERE id=?", (r["id"], sc["id"], lid))
|
|
event(con, lid, f"The DM switched to the scene \"{sc['name']}\"", user["id"])
|
|
hub.notify(lid, "map")
|
|
hub.notify(lid, "lobby")
|
|
return {"ok": True}
|
|
|
|
@api.post("/lobbies/{lid}/scenes")
|
|
def save_scene_copy(lid: str, body: dict = Body(default={}), user: dict = Depends(current_user)):
|
|
"""DM: store a copy of the current table (a checkpoint to come back to). The table keeps running on its own scene."""
|
|
with db() as con:
|
|
lobby = lobby_for(con, lid, user)
|
|
dm_only(lobby, user)
|
|
if not lobby["map_id"]:
|
|
raise HTTPException(409, "There is no map on the table")
|
|
autosave(con, lid)
|
|
current = con.execute("SELECT scene_id FROM lobbies WHERE id=?", (lid,)).fetchone()["scene_id"]
|
|
sid = new_scene(con, lid, lobby["map_id"], snapshot(con, lid), str(body.get("name", "")) or None)
|
|
con.execute("UPDATE lobbies SET scene_id=? WHERE id=?", (current, lid))
|
|
hub.notify(lid, "lobby")
|
|
return {"id": sid}
|
|
|
|
@api.put("/lobbies/{lid}/scenes/{sid}")
|
|
def edit_scene(lid: str, sid: str, body: dict = Body(...), user: dict = Depends(current_user)):
|
|
with db() as con:
|
|
dm_only(lobby_for(con, lid, user), user)
|
|
if not con.execute("SELECT 1 FROM lobby_scenes WHERE id=? AND lobby_id=?", (sid, lid)).fetchone():
|
|
raise HTTPException(404, "Scene not found")
|
|
name = str(body.get("name", "")).strip()[:50]
|
|
if name:
|
|
if con.execute("SELECT 1 FROM lobby_scenes WHERE lobby_id=? AND name=? AND id<>?", (lid, name, sid)).fetchone():
|
|
raise HTTPException(409, "There is already a scene with this name")
|
|
con.execute("UPDATE lobby_scenes SET name=? WHERE id=?", (name, sid))
|
|
if body.get("save"):
|
|
autosave(con, lid)
|
|
hub.notify(lid, "lobby")
|
|
return {"ok": True}
|
|
|
|
@api.delete("/lobbies/{lid}/scenes/{sid}")
|
|
def delete_scene(lid: str, sid: str, user: dict = Depends(current_user)):
|
|
with db() as con:
|
|
dm_only(lobby_for(con, lid, user), user)
|
|
con.execute("DELETE FROM lobby_scenes WHERE id=? AND lobby_id=?", (sid, lid))
|
|
con.execute("UPDATE lobbies SET scene_id=NULL WHERE id=? AND scene_id=?", (lid, sid))
|
|
hub.notify(lid, "lobby")
|
|
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)
|
|
npc = t["npc_id"] if t["npc_id"] and can_read(con, t["npc_id"], user) else None
|
|
put_token(con, lid, t, npc, t["id"])
|
|
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")
|
|
if not is_dm and t["x"] is None and lobby["los"] and json.loads(m["walls"]):
|
|
raise HTTPException(403, "The DM puts your token on the map") # with line of sight on you could not see where
|
|
merged = {"name": t["name"], "x": t["x"], "y": t["y"], "color": t["color"], "size": t["size"], "hidden": t["hidden"],
|
|
"prop": t["prop"], "rot": t["rot"], "size_h": t["size_h"]}
|
|
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", "rot", "size_h"):
|
|
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=?, rot=?, size_h=? WHERE id=?",
|
|
(c["name"] if t["user_id"] is None else t["name"], c["x"], c["y"], c["color"], c["size"], int(c["hidden"]),
|
|
c["rot"] if t["user_id"] is None else 0, c["size_h"] if t["user_id"] is None else None, tid))
|
|
if is_dm and any(k in body for k in ("hp", "hp_max", "conds", "show_hp", "show_cond")):
|
|
hp_max, hp = t["hp_max"], t["hp"]
|
|
if "hp_max" in body:
|
|
hp_max = clean_int(body["hp_max"], 1, 9999)
|
|
if "hp" in body:
|
|
hp = clean_int(body["hp"], -999, 9999)
|
|
if hp_max is None:
|
|
hp = None
|
|
elif hp is None:
|
|
hp = hp_max
|
|
elif hp > hp_max:
|
|
hp = hp_max
|
|
conds = clean_conds(body["conds"]) if "conds" in body else json.loads(t["conds"] or "{}")
|
|
con.execute("UPDATE lobby_tokens SET hp=?, hp_max=?, conds=?, show_hp=?, show_cond=? WHERE id=?",
|
|
(hp, hp_max, json.dumps(conds), int(bool(body.get("show_hp", t["show_hp"]))), int(bool(body.get("show_cond", t["show_cond"]))), 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}
|
|
|
|
@api.get("/lobbies/{lid}/token-portrait/{tid}")
|
|
def token_portrait(lid: str, tid: str, user: dict = Depends(current_user)):
|
|
"""Portrait of an NPC token for the lobby's members (hidden tokens only for the DM)."""
|
|
with db() as con:
|
|
lobby = lobby_for(con, lid, user)
|
|
t = con.execute("SELECT t.hidden, n.data FROM lobby_tokens t JOIN characters n ON n.id = t.npc_id WHERE t.id=? AND t.lobby_id=?", (tid, lid)).fetchone()
|
|
if not t or (t["hidden"] and lobby["owner_id"] != user["id"]):
|
|
raise HTTPException(404, "No portrait")
|
|
img = portrait_image(json.loads(t["data"]).get("portrait"))
|
|
if not img:
|
|
raise HTTPException(404, "No portrait")
|
|
return Response(img, media_type="image/webp", headers={"Cache-Control": "private, max-age=86400"})
|
|
|
|
return sync_players
|