Sight groups, DM reveal areas (cones), client-side image shrink for map upload

Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01HJhXuLD2Z6hv3Wqeq9UL9K
This commit is contained in:
Claude
2026-10-24 18:33:08 +00:00
parent 6aac03f112
commit ab3c35aab7
8 changed files with 243 additions and 29 deletions
+2 -2
View File
@@ -674,7 +674,7 @@ def read_lobby(lid: str, user: dict = Depends(current_user)):
lobby = _lobby_for(con, lid, user)
is_dm = lobby["owner_id"] == user["id"]
_ensure_colors(con, lid)
rows = con.execute("""SELECT m.user_id, m.character_id, m.color, m.char_mod, u.username, c.data FROM lobby_members m
rows = con.execute("""SELECT m.user_id, m.character_id, m.color, m.char_mod, m.vgroup, u.username, c.data FROM lobby_members m
JOIN users u ON u.id = m.user_id LEFT JOIN characters c ON c.id = m.character_id
WHERE m.lobby_id=? ORDER BY m.joined""", (lid,)).fetchall()
combat = combat_view(con, lobby, is_dm)
@@ -685,7 +685,7 @@ def read_lobby(lid: str, user: dict = Depends(current_user)):
card = _card(r["data"], rules) if r["data"] else None
# The id is only given out where the viewer may open the full character (the DM sees all, a player only their own)
members.append({"username": r["username"], "me": mine, "character": card, "color": r["color"],
"mod": r["char_mod"] if (is_dm or mine) else None, "online": r["user_id"] in online,
"mod": r["char_mod"] if (is_dm or mine) else None, "online": r["user_id"] in online, "vgroup": r["vgroup"] if is_dm else None,
"character_id": r["character_id"] if (is_dm or mine) and card else None})
hidden = bool(lobby["diff_hidden"])
diff = {"label": lobby["diff_label"], "mod": lobby["diff_mod"], "hidden": hidden}
+110 -15
View File
@@ -5,6 +5,7 @@ Images are re-encoded with Pillow on upload, so only real images get through and
import hashlib
import io
import json
import math
import shutil
import time
import uuid
@@ -41,13 +42,17 @@ def schema(con):
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 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_tokens", "show_cond", "INTEGER NOT NULL DEFAULT 0"),
("lobby_members", "vgroup", "INTEGER 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}")
@@ -345,7 +350,7 @@ def install(api, db, current_user, lobby_for, dm_only, hub, event, store_dir: Pa
f = max(1, -(-max(m["w"], m["h"]) // 512))
return f, -(-m["w"] // f), -(-m["h"] // f)
def explore(con, lobby, viewer, m, poly, key):
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)
@@ -362,7 +367,9 @@ def install(api, db, current_user, lobby_for, dm_only, hub, event, store_dir: Pa
old = None
base = old or Image.new("L", (mw, mh), 0)
cur = Image.new("L", (mw, mh), 0)
ImageDraw.Draw(cur).polygon([(x / f, y / f) for x, y in poly], fill=255)
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:
@@ -392,29 +399,54 @@ def install(api, db, current_user, lobby_for, dm_only, hub, event, store_dir: Pa
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,
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
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"])]
sight = lobby["sight"] if lobby["sight"] is not None else m["sight"]
radius = round(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])
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, poly, key)
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
@@ -438,7 +470,7 @@ def install(api, db, current_user, lobby_for, dm_only, hub, event, store_dir: Pa
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"])):
if ctx["los"] and (ctx["empty"] or not vision.inside_any((r["x"], r["y"]), ctx["polys"])):
continue
name = r["name"]
if player:
@@ -485,6 +517,9 @@ def install(api, db, current_user, lobby_for, dm_only, hub, event, store_dir: Pa
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
@@ -502,7 +537,7 @@ def install(api, db, current_user, lobby_for, dm_only, hub, event, store_dir: Pa
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"])
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"})
@@ -533,6 +568,65 @@ def install(api, db, current_user, lobby_for, dm_only, hub, event, store_dir: Pa
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:
@@ -581,6 +675,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,))
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 WHERE id=?", (lid,))
if mid:
r = usable_map(con, str(mid), user)
+21 -7
View File
@@ -11,8 +11,9 @@ 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."""
def polygon(origin, segs, w, h, radius=0.0, sector=None):
"""Points of the area visible from origin. segs: [(x1, y1, x2, y2)] that block sight; radius 0 means unlimited.
sector = (direction, half angle) in radians limits the view to a cone."""
ox, oy = origin
border = [(0, 0, w, 0), (w, 0, w, h), (w, h, 0, h), (0, h, 0, 0)]
if radius:
@@ -34,7 +35,13 @@ def polygon(origin, segs, w, h, radius=0.0):
angles += (a - EPS, a, a + EPS)
if radius:
angles += [i * (2 * math.pi / 120) for i in range(120)]
angles.sort()
if sector:
d, half = sector
rel = [(a - d + math.pi) % (2 * math.pi) - math.pi for a in angles]
rel = [a for a in rel if abs(a) <= half] + [-half, half] + [-half + i * (2 * half / 48) for i in range(49)]
angles = [d + a for a in sorted(rel)]
else:
angles.sort()
pts = []
for a in angles:
dx, dy = math.cos(a), math.sin(a)
@@ -54,7 +61,7 @@ def polygon(origin, segs, w, h, radius=0.0):
if best == float("inf"):
continue
pts.append((ox + dx * best, oy + dy * best))
return pts
return [(ox, oy)] + pts if sector else pts
def inside(pt, poly) -> bool:
@@ -112,19 +119,26 @@ def source(path):
return im
def cut(path, poly, key):
def inside_any(pt, polys) -> bool:
return any(inside(pt, p) for p in polys)
def cut(path, polys, 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]
xs, ys = [p[0] for poly in polys for p in poly], [p[1] for poly in polys 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)
draw = ImageDraw.Draw(mask)
for poly in polys:
if len(poly) > 2:
draw.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)