Maps tab with sharing, walls/doors and strict line of sight
Maps page (list, upload, edit, share, copy), wall/room/door tools in the map editor, server-side visibility polygon: the image is cut per player and other tokens are filtered, DM can view as a player, open doors and switch sight on/off. Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01HJhXuLD2Z6hv3Wqeq9UL9K
This commit is contained in:
Claude
+210
-22
@@ -2,22 +2,28 @@
|
||||
|
||||
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 hashlib
|
||||
import io
|
||||
import json
|
||||
import shutil
|
||||
import time
|
||||
import uuid
|
||||
from pathlib import Path
|
||||
|
||||
from fastapi import Body, Depends, HTTPException, Request
|
||||
from fastapi import Body, Depends, HTTPException, Query, Request
|
||||
from fastapi.concurrency import run_in_threadpool
|
||||
from fastapi.responses import Response
|
||||
from PIL import Image, ImageOps
|
||||
|
||||
from . import vision
|
||||
|
||||
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
|
||||
COLORS = ("#c0392b", "#8e44ad", "#16a085", "#7f8c8d", "#d35400", "#2c3e50", "#27ae60", "#b9770e")
|
||||
|
||||
|
||||
@@ -29,8 +35,11 @@ def schema(con):
|
||||
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")
|
||||
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")):
|
||||
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]:
|
||||
@@ -75,6 +84,26 @@ def clean_token(t: dict, w: int, h: int, keep_id=True) -> dict:
|
||||
"size": size, "color": clean_color(t.get("color")), "hidden": bool(t.get("hidden"))}
|
||||
|
||||
|
||||
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):
|
||||
store_dir.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
@@ -87,16 +116,33 @@ def install(api, db, current_user, lobby_for, dm_only, hub, event, store_dir: Pa
|
||||
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"]}
|
||||
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 * 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]
|
||||
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)):
|
||||
@@ -130,7 +176,7 @@ def install(api, db, current_user, lobby_for, dm_only, hub, event, store_dir: Pa
|
||||
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"]
|
||||
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:
|
||||
@@ -144,7 +190,16 @@ def install(api, db, current_user, lobby_for, dm_only, hub, event, store_dir: Pa
|
||||
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))
|
||||
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")
|
||||
@@ -156,10 +211,12 @@ def install(api, db, current_user, lobby_for, dm_only, hub, event, store_dir: Pa
|
||||
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:
|
||||
con.execute("DELETE FROM lobby_tokens WHERE lobby_id=?", (lid,))
|
||||
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}
|
||||
@@ -170,6 +227,52 @@ def install(api, db, current_user, lobby_for, dm_only, hub, event, store_dir: Pa
|
||||
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")
|
||||
@@ -190,18 +293,57 @@ def install(api, db, current_user, lobby_for, dm_only, hub, event, store_dir: Pa
|
||||
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:
|
||||
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, "poly": None,
|
||||
"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
|
||||
tok = con.execute("SELECT x, y FROM lobby_tokens WHERE lobby_id=? AND user_id=?", (lobby["id"], viewer)).fetchone()
|
||||
ctx["los"] = True
|
||||
if not tok or tok["x"] is None: # not on the map yet: sees nothing
|
||||
ctx.update(empty=True, rev="empty")
|
||||
return ctx
|
||||
x, y = round(tok["x"]), round(tok["y"])
|
||||
blocking = [(w["x1"], w["y1"], w["x2"], w["y2"]) for w in walls if w["kind"] == "wall" or not doors.get(w["id"])]
|
||||
radius = round(m["sight"] * m["grid"], 1)
|
||||
key = hashlib.sha1(json.dumps([m["id"], blocking, radius, x, y]).encode()).hexdigest()
|
||||
poly = vision.cached_polygon(key, lambda: vision.polygon((x, y), blocking, m["w"], m["h"], radius))
|
||||
ctx.update(poly=poly, key=key, rev=key[:12])
|
||||
return ctx
|
||||
|
||||
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 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()
|
||||
full = ctx["viewer"] is None
|
||||
out = []
|
||||
for r in rows:
|
||||
player = r["user_id"] is not None
|
||||
if r["hidden"] and not is_dm:
|
||||
continue
|
||||
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((r["x"], r["y"]), ctx["poly"])):
|
||||
continue
|
||||
name = r["name"]
|
||||
if player:
|
||||
try:
|
||||
@@ -211,25 +353,69 @@ def install(api, db, current_user, lobby_for, dm_only, hub, event, store_dir: Pa
|
||||
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"]})
|
||||
"can_move": ctx["is_dm"] or (r["user_id"] == user["id"] and not (ctx["los"] and r["x"] is None))})
|
||||
return out
|
||||
|
||||
@api.get("/lobbies/{lid}/map")
|
||||
def read_lobby_map(lid: str, user: dict = Depends(current_user)):
|
||||
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)
|
||||
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 []}
|
||||
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"]}}
|
||||
if ctx["is_dm"]:
|
||||
res["los_on"] = bool(lobby["los"])
|
||||
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/image")
|
||||
def lobby_map_image(lid: str, user: dict = Depends(current_user)):
|
||||
@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)
|
||||
if not lobby["map_id"]:
|
||||
m = lobby_map(con, lobby)
|
||||
if not m:
|
||||
raise HTTPException(404, "No map")
|
||||
mid = lobby["map_id"]
|
||||
return image_response(path(mid))
|
||||
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["poly"], 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.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)
|
||||
con.execute("UPDATE lobbies SET los=? WHERE id=?", (1 if body.get("on") else 0, lid))
|
||||
hub.notify(lid, "map")
|
||||
return {"ok": True}
|
||||
|
||||
@api.put("/lobbies/{lid}/map")
|
||||
def set_lobby_map(lid: str, body: dict = Body(...), user: dict = Depends(current_user)):
|
||||
@@ -239,7 +425,7 @@ def install(api, db, current_user, lobby_for, dm_only, hub, event, store_dir: Pa
|
||||
mid = body.get("map_id")
|
||||
con.execute("DELETE FROM lobby_tokens WHERE lobby_id=?", (lid,))
|
||||
if mid:
|
||||
r = own_map(con, str(mid), user)
|
||||
r = usable_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"])))
|
||||
@@ -247,7 +433,7 @@ def install(api, db, current_user, lobby_for, dm_only, hub, event, store_dir: Pa
|
||||
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))
|
||||
con.execute("UPDATE lobbies SET map_id=?, doors='{}' WHERE id=?", (str(mid) if mid else None, lid))
|
||||
hub.notify(lid, "map")
|
||||
return {"ok": True}
|
||||
|
||||
@@ -281,6 +467,8 @@ def install(api, db, current_user, lobby_for, dm_only, hub, event, store_dir: Pa
|
||||
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"]}
|
||||
if "x" in body or "y" in body:
|
||||
merged["x"], merged["y"] = body.get("x"), body.get("y")
|
||||
|
||||
@@ -0,0 +1,132 @@
|
||||
"""Line of sight: visibility polygon from a point around wall segments, and cutting the map image to it.
|
||||
|
||||
The server cuts the image, so a player's browser only ever receives the pixels their character can see (no peeking through
|
||||
dev tools). Pure Python on purpose (no numpy): a few hundred wall segments are fast enough, and results are cached."""
|
||||
import io
|
||||
import math
|
||||
from collections import OrderedDict
|
||||
|
||||
from PIL import Image, ImageDraw
|
||||
|
||||
EPS = 1e-5
|
||||
|
||||
|
||||
def polygon(origin, segs, w, h, radius=0.0):
|
||||
"""Points of the area visible from origin. segs: [(x1, y1, x2, y2)] that block sight; radius 0 means unlimited."""
|
||||
ox, oy = origin
|
||||
border = [(0, 0, w, 0), (w, 0, w, h), (w, h, 0, h), (0, h, 0, 0)]
|
||||
if radius:
|
||||
r2 = (radius + 1) ** 2
|
||||
def near(s):
|
||||
x1, y1, x2, y2 = s
|
||||
dx, dy = x2 - x1, y2 - y1
|
||||
L = dx * dx + dy * dy
|
||||
t = 0 if L == 0 else max(0, min(1, ((ox - x1) * dx + (oy - y1) * dy) / L))
|
||||
return (x1 + t * dx - ox) ** 2 + (y1 + t * dy - oy) ** 2 <= r2
|
||||
segs = [s for s in segs if near(s)]
|
||||
allsegs = segs + border
|
||||
angles = []
|
||||
for x1, y1, x2, y2 in allsegs:
|
||||
for x, y in ((x1, y1), (x2, y2)):
|
||||
if radius and (x - ox) ** 2 + (y - oy) ** 2 > (radius * 3) ** 2:
|
||||
continue
|
||||
a = math.atan2(y - oy, x - ox)
|
||||
angles += (a - EPS, a, a + EPS)
|
||||
if radius:
|
||||
angles += [i * (2 * math.pi / 120) for i in range(120)]
|
||||
angles.sort()
|
||||
pts = []
|
||||
for a in angles:
|
||||
dx, dy = math.cos(a), math.sin(a)
|
||||
best = radius if radius else float("inf")
|
||||
for x1, y1, x2, y2 in allsegs:
|
||||
ex, ey = x2 - x1, y2 - y1
|
||||
den = dx * ey - dy * ex
|
||||
if abs(den) < 1e-12:
|
||||
continue
|
||||
px, py = x1 - ox, y1 - oy
|
||||
t = (px * ey - py * ex) / den
|
||||
if t < 0 or t >= best:
|
||||
continue
|
||||
u = (px * dy - py * dx) / den
|
||||
if -1e-9 <= u <= 1 + 1e-9:
|
||||
best = t
|
||||
if best == float("inf"):
|
||||
continue
|
||||
pts.append((ox + dx * best, oy + dy * best))
|
||||
return pts
|
||||
|
||||
|
||||
def inside(pt, poly) -> bool:
|
||||
x, y = pt
|
||||
c = False
|
||||
j = len(poly) - 1
|
||||
for i in range(len(poly)):
|
||||
xi, yi = poly[i]
|
||||
xj, yj = poly[j]
|
||||
if (yi > y) != (yj > y) and x < (xj - xi) * (y - yi) / (yj - yi) + xi:
|
||||
c = not c
|
||||
j = i
|
||||
return c
|
||||
|
||||
|
||||
class Lru(OrderedDict):
|
||||
def __init__(self, size):
|
||||
super().__init__()
|
||||
self.size = size
|
||||
|
||||
def put(self, k, v):
|
||||
self[k] = v
|
||||
self.move_to_end(k)
|
||||
while len(self) > self.size:
|
||||
self.popitem(last=False)
|
||||
|
||||
def fetch(self, k):
|
||||
if k in self:
|
||||
self.move_to_end(k)
|
||||
return self[k]
|
||||
|
||||
|
||||
_images = Lru(2)
|
||||
_polys = Lru(128)
|
||||
_cuts = Lru(24)
|
||||
|
||||
|
||||
def cached_polygon(key, make):
|
||||
v = _polys.fetch(key)
|
||||
if v is None:
|
||||
v = make()
|
||||
_polys.put(key, v)
|
||||
return v
|
||||
|
||||
|
||||
def source(path):
|
||||
key = (str(path), path.stat().st_mtime_ns)
|
||||
im = _images.fetch(key)
|
||||
if im is None:
|
||||
im = Image.open(path)
|
||||
im.load()
|
||||
im = im.convert("RGB")
|
||||
_images.put(key, im)
|
||||
return im
|
||||
|
||||
|
||||
def cut(path, poly, key):
|
||||
"""The part of the image inside the polygon (everything else black), cropped to its bounding box.
|
||||
Returns (webp bytes, (x, y, w, h))."""
|
||||
hit = _cuts.fetch(key)
|
||||
if hit:
|
||||
return hit
|
||||
im = source(path)
|
||||
xs, ys = [p[0] for p in poly], [p[1] for p in poly]
|
||||
x0, y0 = max(0, int(math.floor(min(xs)))), max(0, int(math.floor(min(ys))))
|
||||
x1, y1 = min(im.width, int(math.ceil(max(xs)))), min(im.height, int(math.ceil(max(ys))))
|
||||
w, h = max(1, x1 - x0), max(1, y1 - y0)
|
||||
mask = Image.new("L", (w, h), 0)
|
||||
ImageDraw.Draw(mask).polygon([(x - x0, y - y0) for x, y in poly], fill=255)
|
||||
piece = Image.composite(im.crop((x0, y0, x0 + w, y0 + h)), Image.new("RGB", (w, h), (0, 0, 0)), mask)
|
||||
out = io.BytesIO()
|
||||
piece.save(out, "WEBP", quality=82, method=1)
|
||||
res = (out.getvalue(), (x0, y0, w, h))
|
||||
_cuts.put(key, res)
|
||||
return res
|
||||
Reference in New Issue
Block a user