Claude
72d8113ece
Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01HJhXuLD2Z6hv3Wqeq9UL9K
171 lines
8.2 KiB
Python
171 lines
8.2 KiB
Python
"""Initiative tracker. One fight per lobby, stored as JSON on the lobby row.
|
|
|
|
{"round": 1, "current": "<entry id>", "entries": [{"id", "kind": "player"|"npc", "user": username, "token": token id,
|
|
"name", "init": number, "base": initiative modifier, "roll": d20}]}
|
|
The DM starts and ends the fight; players are added automatically at the start. Initiative is 1d20 + the character's
|
|
Initiative value (Agility Mod + Perception Mod), the DM can overwrite every number."""
|
|
import json
|
|
import secrets
|
|
import uuid
|
|
|
|
from fastapi import Body, Depends, HTTPException
|
|
|
|
from .derive import compute
|
|
|
|
MAX_ENTRIES = 60
|
|
|
|
|
|
def schema(con):
|
|
if "combat" not in [r["name"] for r in con.execute("PRAGMA table_info(lobbies)")]:
|
|
con.execute("ALTER TABLE lobbies ADD COLUMN combat TEXT")
|
|
|
|
|
|
def _sorted(c: dict):
|
|
c["entries"].sort(key=lambda e: (-(e["init"] if e["init"] is not None else -999), e.get("seq", 0)))
|
|
|
|
|
|
def install(api, db, current_user, lobby_for, dm_only, hub, event, load_rules):
|
|
def base_of(data, rules) -> int:
|
|
try:
|
|
return int(compute(json.loads(data), rules)["derived"].get("initiative", 0))
|
|
except Exception:
|
|
return 0
|
|
|
|
def new_entry(c, kind, name, base, **kw):
|
|
roll = secrets.randbelow(20) + 1
|
|
c["seq"] = c.get("seq", 0) + 1
|
|
return {"id": uuid.uuid4().hex[:10], "kind": kind, "name": name, "base": base, "roll": roll, "init": roll + base, "seq": c["seq"], **kw}
|
|
|
|
def load(lobby):
|
|
return json.loads(lobby["combat"]) if lobby["combat"] else None
|
|
|
|
def save(con, lid, c, log=None, uid=None):
|
|
con.execute("UPDATE lobbies SET combat=? WHERE id=?", (json.dumps(c) if c else None, lid))
|
|
if log:
|
|
event(con, lid, log, uid)
|
|
hub.notify(lid, "lobby")
|
|
|
|
def view(con, lobby, is_dm):
|
|
"""What the viewer gets: hidden NPCs are left out for players; NPCs whose token is gone disappear."""
|
|
c = load(lobby)
|
|
if not c:
|
|
return None
|
|
toks = {r["id"]: r["hidden"] for r in con.execute("SELECT id, hidden FROM lobby_tokens WHERE lobby_id=?", (lobby["id"],))}
|
|
out = []
|
|
for e in c["entries"]:
|
|
if e["kind"] == "npc":
|
|
if e["token"] not in toks:
|
|
continue
|
|
if toks[e["token"]] and not is_dm:
|
|
continue
|
|
out.append({k: e[k] for k in ("id", "kind", "name", "init", "base", "roll") if k in e} | {k: e[k] for k in ("user", "token") if k in e})
|
|
cur = c.get("current")
|
|
return {"round": c["round"], "current": cur if any(e["id"] == cur for e in out) else None, "entries": out}
|
|
|
|
@api.post("/lobbies/{lid}/combat")
|
|
def start(lid: str, user: dict = Depends(current_user)):
|
|
rules = load_rules()
|
|
with db() as con:
|
|
lobby = lobby_for(con, lid, user)
|
|
dm_only(lobby, user)
|
|
c = {"round": 1, "current": None, "seq": 0, "entries": []}
|
|
for r in con.execute("""SELECT u.username, ch.data FROM lobby_members m JOIN users u ON u.id = m.user_id
|
|
JOIN characters ch ON ch.id = m.character_id WHERE m.lobby_id=? ORDER BY m.joined""", (lid,)):
|
|
nm = json.loads(r["data"]).get("name") or r["username"]
|
|
c["entries"].append(new_entry(c, "player", nm, base_of(r["data"], rules), user=r["username"]))
|
|
_sorted(c)
|
|
c["current"] = c["entries"][0]["id"] if c["entries"] else None
|
|
save(con, lid, c, "⚔ A fight begins. Initiative: " + ", ".join(f"{e['name']} {e['init']}" for e in c["entries"]), user["id"])
|
|
return {"ok": True}
|
|
|
|
@api.put("/lobbies/{lid}/combat")
|
|
def change(lid: str, body: dict = Body(...), user: dict = Depends(current_user)):
|
|
rules = load_rules()
|
|
with db() as con:
|
|
lobby = lobby_for(con, lid, user)
|
|
dm_only(lobby, user)
|
|
c = load(lobby)
|
|
if not c:
|
|
raise HTTPException(409, "There is no fight")
|
|
op, eid = body.get("op"), str(body.get("id") or "")
|
|
ents = c["entries"]
|
|
ent = next((e for e in ents if e["id"] == eid), None)
|
|
log = None
|
|
if op == "add":
|
|
if len(ents) >= MAX_ENTRIES:
|
|
raise HTTPException(409, "Too many entries")
|
|
if body.get("token"):
|
|
t = con.execute("""SELECT t.id, t.name, t.user_id, n.data FROM lobby_tokens t LEFT JOIN characters n ON n.id = t.npc_id
|
|
WHERE t.id=? AND t.lobby_id=?""", (str(body["token"]), lid)).fetchone()
|
|
if not t or t["user_id"]:
|
|
raise HTTPException(404, "No NPC token")
|
|
if any(e.get("token") == t["id"] for e in ents):
|
|
raise HTTPException(409, "Already in the fight")
|
|
e = new_entry(c, "npc", t["name"], base_of(t["data"], rules) if t["data"] else 0, token=t["id"])
|
|
elif body.get("user"):
|
|
r = con.execute("""SELECT u.username, ch.data FROM lobby_members m JOIN users u ON u.id = m.user_id
|
|
JOIN characters ch ON ch.id = m.character_id WHERE m.lobby_id=? AND u.username=?""", (lid, str(body["user"]))).fetchone()
|
|
if not r:
|
|
raise HTTPException(404, "No such player with a character")
|
|
if any(e.get("user") == r["username"] for e in ents):
|
|
raise HTTPException(409, "Already in the fight")
|
|
e = new_entry(c, "player", json.loads(r["data"]).get("name") or r["username"], base_of(r["data"], rules), user=r["username"])
|
|
else:
|
|
raise HTTPException(422, "Say which token or player to add")
|
|
ents.append(e)
|
|
log = f"{e['name']} joins the fight with initiative {e['init']}"
|
|
elif op == "remove":
|
|
if not ent:
|
|
raise HTTPException(404, "No such entry")
|
|
if c["current"] == eid:
|
|
i = ents.index(ent)
|
|
nxt = ents[(i + 1) % len(ents)] if len(ents) > 1 else None
|
|
c["current"] = nxt["id"] if nxt else None
|
|
ents.remove(ent)
|
|
elif op == "init":
|
|
if not ent:
|
|
raise HTTPException(404, "No such entry")
|
|
try:
|
|
ent["init"] = max(-99, min(999, int(body.get("value"))))
|
|
except (TypeError, ValueError):
|
|
raise HTTPException(422, "Initiative must be a whole number")
|
|
elif op == "reroll":
|
|
if not ent:
|
|
raise HTTPException(404, "No such entry")
|
|
ent["roll"] = secrets.randbelow(20) + 1
|
|
ent["init"] = ent["roll"] + ent["base"]
|
|
log = f"{ent['name']} rerolls initiative: {ent['init']}"
|
|
elif op in ("next", "prev"):
|
|
_sorted(c)
|
|
if ents:
|
|
i = next((k for k, e in enumerate(ents) if e["id"] == c.get("current")), -1)
|
|
if op == "next":
|
|
if i + 1 >= len(ents):
|
|
c["round"] += 1
|
|
i = (i + 1) % len(ents)
|
|
else:
|
|
if i <= 0:
|
|
c["round"] = max(1, c["round"] - 1)
|
|
i = (i - 1) % len(ents)
|
|
c["current"] = ents[i]["id"]
|
|
else:
|
|
raise HTTPException(422, "Unknown action")
|
|
cur = c.get("current")
|
|
_sorted(c)
|
|
if cur and not any(e["id"] == cur for e in c["entries"]):
|
|
c["current"] = c["entries"][0]["id"] if c["entries"] else None
|
|
save(con, lid, c, log, user["id"])
|
|
return {"ok": True}
|
|
|
|
@api.delete("/lobbies/{lid}/combat")
|
|
def end(lid: str, user: dict = Depends(current_user)):
|
|
with db() as con:
|
|
lobby = lobby_for(con, lid, user)
|
|
dm_only(lobby, user)
|
|
c = load(lobby)
|
|
if c:
|
|
save(con, lid, None, f"The fight is over after {c['round']} round{'s' if c['round'] != 1 else ''}.", user["id"])
|
|
return {"ok": True}
|
|
|
|
return view
|