Token HP and conditions with DM visibility toggles, fog of war, live sight range

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:21:39 +00:00
parent a2103566a0
commit 6aac03f112
7 changed files with 248 additions and 21 deletions
+154 -13
View File
@@ -13,7 +13,7 @@ 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, ImageOps
from PIL import Image, ImageChops, ImageDraw, ImageOps
from . import vision
from .portrait import image as portrait_image, tag as portrait_tag
@@ -38,9 +38,16 @@ def schema(con):
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 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", "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")):
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}")
@@ -84,9 +91,37 @@ def clean_token(t: dict, w: int, h: int, keep_id=True) -> dict:
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)
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 ""}
"npc_id": npc if npc.isalnum() and len(npc) <= 16 else "", "hp": hp, "hp_max": hp_max}
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:
@@ -109,7 +144,7 @@ def clean_walls(raw, w: int, h: int) -> list:
return out
def install(api, db, current_user, lobby_for, dm_only, hub, event, store_dir: Path, can_read):
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:
@@ -302,10 +337,44 @@ def install(api, db, current_user, lobby_for, dm_only, hub, event, store_dir: Pa
have = {r["user_id"] for r in con.execute("SELECT user_id FROM lobby_tokens WHERE lobby_id=? AND user_id IS NOT NULL", (lid,))}
ids = {m["user_id"] for m in members}
for uid in ids - have:
con.execute("INSERT INTO lobby_tokens (id, lobby_id, user_id, name, size) VALUES (?,?,?,?,1)", (uuid.uuid4().hex[:12], lid, uid, ""))
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 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, poly, 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)
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
@@ -337,10 +406,13 @@ def install(api, db, current_user, lobby_for, dm_only, hub, event, store_dir: Pa
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)
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])
if lobby["fog"]:
ctx["fog"] = explore(con, lobby, viewer, m, poly, key)
return ctx
def tokens_for(con, lobby, user, ctx) -> list:
@@ -383,7 +455,14 @@ def install(api, db, current_user, lobby_for, dm_only, hub, event, store_dir: Pa
pt = portrait_tag(json.loads(src).get("portrait"))
except ValueError:
pass
out.append({"id": r["id"], "kind": "player" if player else "npc", "username": r["username"], "name": name, "portrait": pt,
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)
out.append({"id": r["id"], "kind": "player" if player else "npc", "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)),
@@ -400,9 +479,12 @@ def install(api, db, current_user, lobby_for, dm_only, hub, event, store_dir: Pa
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"]}}
"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"])
if ctx["viewer"] is None:
res["walls"] = [{**w, "open": bool(ctx["doors"].get(w["id"]))} for w in ctx["walls"]]
return res
@@ -425,6 +507,40 @@ def install(api, db, current_user, lobby_for, dm_only, hub, event, store_dir: Pa
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.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:
@@ -447,7 +563,13 @@ def install(api, db, current_user, lobby_for, dm_only, hub, event, store_dir: Pa
"""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))
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}
@@ -458,12 +580,15 @@ def install(api, db, current_user, lobby_for, dm_only, hub, event, store_dir: Pa
dm_only(lobby_for(con, lid, user), user)
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("UPDATE lobbies SET sight=NULL WHERE id=?", (lid,))
if mid:
r = usable_map(con, str(mid), user)
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
con.execute("INSERT INTO lobby_tokens (id, lobby_id, name, x, y, color, size, hidden, npc_id) VALUES (?,?,?,?,?,?,?,?,?)",
(uuid.uuid4().hex[:12], lid, t["name"], t["x"], t["y"], t["color"], t["size"], int(t["hidden"]), npc))
con.execute("INSERT INTO lobby_tokens (id, lobby_id, name, x, y, color, size, hidden, npc_id, hp, hp_max) VALUES (?,?,?,?,?,?,?,?,?,?,?)",
(uuid.uuid4().hex[:12], lid, t["name"], t["x"], t["y"], t["color"], t["size"], int(t["hidden"]), npc,
t.get("hp") if t.get("hp") is not None else t.get("hp_max"), t.get("hp_max")))
sync_players(con, lid)
event(con, lid, f"The DM put the map \"{r['name']}\" on the table", user["id"])
else:
@@ -488,8 +613,9 @@ def install(api, db, current_user, lobby_for, dm_only, hub, event, store_dir: Pa
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
con.execute("INSERT INTO lobby_tokens (id, lobby_id, name, x, y, color, size, hidden, npc_id) VALUES (?,?,?,?,?,?,?,?,?)",
(t["id"], lid, t["name"], t["x"], t["y"], t["color"], t["size"], int(t["hidden"]), npc))
con.execute("INSERT INTO lobby_tokens (id, lobby_id, name, x, y, color, size, hidden, npc_id, hp, hp_max) VALUES (?,?,?,?,?,?,?,?,?,?,?)",
(t["id"], lid, t["name"], t["x"], t["y"], t["color"], t["size"], int(t["hidden"]), npc,
t["hp"] if t["hp"] is not None else t["hp_max"], t["hp_max"]))
hub.notify(lid, "map")
return {"id": t["id"]}
@@ -517,6 +643,21 @@ def install(api, db, current_user, lobby_for, dm_only, hub, event, store_dir: Pa
c = clean_token(merged, m["w"], m["h"])
con.execute("UPDATE lobby_tokens SET name=?, x=?, y=?, color=?, size=?, hidden=? WHERE id=?",
(c["name"] if t["user_id"] is None else t["name"], c["x"], c["y"], c["color"], c["size"], int(c["hidden"]), tid))
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}