Lobby: MP3 dice sound with pitch per die, DM sound settings
Default dice sound is public/dice.mp3 with a small random pitch and a base pitch per die size (d4 lighter ... d100 heavier). The DM can upload an own dice sound and sounds for critical success / failure per lobby, reset them, and switch result and critical sounds on or off. Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01HJhXuLD2Z6hv3Wqeq9UL9K
This commit is contained in:
+5
-1
@@ -16,6 +16,7 @@ from . import auth
|
||||
from .derive import compute
|
||||
from . import rules_edit
|
||||
from . import maps as maps_mod
|
||||
from . import sounds as sounds_mod
|
||||
from .realtime import hub
|
||||
from .export import to_docx, to_pdf
|
||||
|
||||
@@ -102,6 +103,7 @@ def db() -> sqlite3.Connection:
|
||||
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}")
|
||||
maps_mod.schema(con)
|
||||
sounds_mod.schema(con)
|
||||
for table in ("characters", "presets"): # older databases: add the owner column
|
||||
if "owner_id" not in [r["name"] for r in con.execute(f"PRAGMA table_info({table})")]:
|
||||
con.execute(f"ALTER TABLE {table} ADD COLUMN owner_id INTEGER")
|
||||
@@ -612,7 +614,7 @@ def read_lobby(lid: str, user: dict = Depends(current_user)):
|
||||
diff = {"label": "", "mod": None, "hidden": True}
|
||||
return {"id": lobby["id"], "name": lobby["name"], "dm": lobby["dm"], "is_dm": is_dm,
|
||||
"code": lobby["code"] if is_dm else None, "members": members, "difficulty": diff,
|
||||
"levels": _levels(rules), "palette": [{"id": i, "hex": h} for i, h in PALETTE], "dm_color": DM_COLOR}
|
||||
"sounds": sounds_mod.settings(lobby["sounds"]), "levels": _levels(rules), "palette": [{"id": i, "hex": h} for i, h in PALETTE], "dm_color": DM_COLOR}
|
||||
|
||||
|
||||
@api.put("/lobbies/{lid}/difficulty")
|
||||
@@ -841,6 +843,7 @@ def delete_lobby(lid: str, user: dict = Depends(current_user)):
|
||||
con.execute("DELETE FROM lobby_messages WHERE lobby_id=?", (lid,))
|
||||
con.execute("DELETE FROM lobby_tokens WHERE lobby_id=?", (lid,))
|
||||
con.execute("DELETE FROM lobbies WHERE id=?", (lid,))
|
||||
purge_sounds(lid)
|
||||
hub.close_lobby(lid)
|
||||
return {"ok": True}
|
||||
|
||||
@@ -1018,6 +1021,7 @@ def editor_reset(section: str, confirm: bool = False, user: dict = Depends(admin
|
||||
|
||||
|
||||
maps_mod.install(api, db, current_user, _lobby_for, _dm_only, hub, _event, Path(DB_PATH).parent / "maps")
|
||||
purge_sounds = sounds_mod.install(api, db, current_user, _lobby_for, _dm_only, hub, Path(DB_PATH).parent / "sounds")
|
||||
|
||||
|
||||
@app.websocket("/api/lobbies/{lid}/ws")
|
||||
|
||||
@@ -0,0 +1,104 @@
|
||||
"""Lobby sounds: the DM can replace the dice sound and add sounds for critical successes / failures (MP3, per lobby)."""
|
||||
import json
|
||||
from pathlib import Path
|
||||
|
||||
from fastapi import Body, Depends, HTTPException, Request
|
||||
from fastapi.responses import Response
|
||||
|
||||
MAX_SOUND = 1_500_000 # bytes per file
|
||||
SLOTS = ("roll", "crit_success", "crit_failure")
|
||||
FLAGS = ("result", "crit_success", "crit_failure") # result = the short tones for success / failure of a check
|
||||
|
||||
|
||||
def schema(con):
|
||||
if "sounds" not in [r["name"] for r in con.execute("PRAGMA table_info(lobbies)")]:
|
||||
con.execute("ALTER TABLE lobbies ADD COLUMN sounds TEXT NOT NULL DEFAULT '{}'")
|
||||
|
||||
|
||||
def settings(raw) -> dict:
|
||||
try:
|
||||
s = json.loads(raw or "{}")
|
||||
except ValueError:
|
||||
s = {}
|
||||
return {"result": s.get("result", True), "crit_success": s.get("crit_success", True), "crit_failure": s.get("crit_failure", True),
|
||||
"custom": {k: bool(s.get("custom", {}).get(k)) for k in SLOTS}, "v": int(s.get("v", 0))}
|
||||
|
||||
|
||||
def install(api, db, current_user, lobby_for, dm_only, hub, store_dir: Path):
|
||||
store_dir.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
def path(lid: str, slot: str) -> Path:
|
||||
if slot not in SLOTS:
|
||||
raise HTTPException(404, "Unknown sound")
|
||||
return store_dir / f"{lid}_{slot}.mp3"
|
||||
|
||||
def save(con, lid, s):
|
||||
s["v"] += 1
|
||||
con.execute("UPDATE lobbies SET sounds=? WHERE id=?", (json.dumps(s), lid))
|
||||
|
||||
@api.put("/lobbies/{lid}/sounds")
|
||||
def set_flags(lid: str, body: dict = Body(...), user: dict = Depends(current_user)):
|
||||
with db() as con:
|
||||
lobby = lobby_for(con, lid, user)
|
||||
dm_only(lobby, user)
|
||||
s = settings(lobby["sounds"])
|
||||
for k in FLAGS:
|
||||
if k in body:
|
||||
s[k] = bool(body[k])
|
||||
save(con, lid, s)
|
||||
hub.notify(lid, "lobby")
|
||||
return {"ok": True}
|
||||
|
||||
@api.post("/lobbies/{lid}/sounds/{slot}")
|
||||
async def upload_sound(lid: str, slot: str, request: Request, user: dict = Depends(current_user)):
|
||||
"""Raw MP3 bytes in the body."""
|
||||
p = path(lid, slot)
|
||||
with db() as con:
|
||||
dm_only(lobby_for(con, lid, user), user)
|
||||
if int(request.headers.get("content-length") or 0) > MAX_SOUND:
|
||||
raise HTTPException(413, f"The file is too large (at most {MAX_SOUND // 1_000_000} MB)")
|
||||
chunks, size = [], 0
|
||||
async for c in request.stream():
|
||||
size += len(c)
|
||||
if size > MAX_SOUND:
|
||||
raise HTTPException(413, f"The file is too large (at most {MAX_SOUND // 1_000_000} MB)")
|
||||
chunks.append(c)
|
||||
data = b"".join(chunks)
|
||||
if not (data[:3] == b"ID3" or (len(data) > 4 and data[0] == 0xFF and data[1] & 0xE0 == 0xE0)):
|
||||
raise HTTPException(422, "Please upload an MP3 file")
|
||||
p.write_bytes(data)
|
||||
with db() as con:
|
||||
lobby = lobby_for(con, lid, user)
|
||||
s = settings(lobby["sounds"])
|
||||
s["custom"][slot] = True
|
||||
save(con, lid, s)
|
||||
hub.notify(lid, "lobby")
|
||||
return {"ok": True}
|
||||
|
||||
@api.delete("/lobbies/{lid}/sounds/{slot}")
|
||||
def reset_sound(lid: str, slot: str, user: dict = Depends(current_user)):
|
||||
p = path(lid, slot)
|
||||
with db() as con:
|
||||
lobby = lobby_for(con, lid, user)
|
||||
dm_only(lobby, user)
|
||||
s = settings(lobby["sounds"])
|
||||
s["custom"][slot] = False
|
||||
save(con, lid, s)
|
||||
p.unlink(missing_ok=True)
|
||||
hub.notify(lid, "lobby")
|
||||
return {"ok": True}
|
||||
|
||||
@api.get("/lobbies/{lid}/sounds/{slot}")
|
||||
def get_sound(lid: str, slot: str, v: int = 0, user: dict = Depends(current_user)):
|
||||
p = path(lid, slot)
|
||||
with db() as con:
|
||||
lobby_for(con, lid, user)
|
||||
if not p.exists():
|
||||
raise HTTPException(404, "No custom sound")
|
||||
return Response(p.read_bytes(), media_type="audio/mpeg", headers={"Cache-Control": "private, max-age=86400"})
|
||||
|
||||
def purge(lid: str):
|
||||
for slot in SLOTS:
|
||||
path(lid, slot).unlink(missing_ok=True)
|
||||
|
||||
return purge
|
||||
Reference in New Issue
Block a user