agwandClaude Opus 5.5
5de261acd7
Ved et lige antal spillere får kæden et trin mere: i trin 1 tegner ejeren sit eget ord, og rotationen starter et trin senere. Antal trin hedder nu steps og bruges i step-eventet, snapshot og afsløringen; self.n er antal spillere. Snapshot får step.own til trin 1. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
800 lines
36 KiB
Python
800 lines
36 KiB
Python
"""Spilmotoren for Tegn og Gæt (Telestrations-formatet).
|
|
|
|
Ren Python uden Django og uden I/O, i samme stil som `fupogfakta.engine`: alle
|
|
handlinger tager `now` (sekunder siden epoch) som parameter, rummet i
|
|
`party.tegn` kalder `tick()` når en deadline udløber og sender `snapshot()` til
|
|
hver skærm.
|
|
|
|
Med N spillere er der N kæder, én pr. spiller: trin 0 er et ord, ulige trin
|
|
tegnes, lige trin gættes, og kæden slutter altid på et gæt. Ulige N: N trin, og i
|
|
trin t arbejder spiller i på kæde (i - t) mod N. Lige N: N+1 trin; i trin 1
|
|
tegner ejeren sit eget ord, og i trin t >= 1 arbejder spiller i på kæde
|
|
(i - (t-1)) mod N. Ingen gætter på sin egen kæde.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import difflib
|
|
import json
|
|
import math
|
|
import random
|
|
import secrets
|
|
from dataclasses import dataclass
|
|
|
|
from fupogfakta.engine import NAME_PUNCT, GameError, normalize, sanitize
|
|
|
|
WORD_SECONDS = 30
|
|
DRAW_SECONDS = 75
|
|
GUESS_SECONDS = 30
|
|
THEME_SECONDS = 20
|
|
RULES_STEP_MS = 5000
|
|
|
|
# Afsløringen: kæde for kæde, bidrag for bidrag (uden stemme; speakeren kan forlænge et trin)
|
|
CHAIN_INTRO_MS = 2500
|
|
WORD_MS = 3000
|
|
DRAWING_MS = 5500 # tegningen "tegnes frem" (ca. 3,5 s) og står lidt
|
|
GUESS_MS = 3000
|
|
CHAIN_END_MS = 4500 # første ord mod sidste gæt
|
|
|
|
MIN_PLAYERS = 3
|
|
MAX_PLAYERS = 10
|
|
MAX_NAME = 12
|
|
MAX_TEXT = 30 # ord og gæt
|
|
|
|
POINTS_LIKE = 100
|
|
POINTS_SURVIVED = 500 # til ejeren af en kæde, hvor sidste gæt er det første ord
|
|
EMPTY_GUESS = "???"
|
|
|
|
# Tegninger: vektorstrøg i normaliserede koordinater (0..1, lærredet er 4:3)
|
|
PALETTE = ("#1d1b2e", "#e63946", "#2a7de1", "#2bb673")
|
|
MIN_WIDTH, MAX_WIDTH = 0.002, 0.06
|
|
MAX_STROKES = 400
|
|
MAX_POINTS = 20_000
|
|
MAX_BYTES = 64 * 1024
|
|
|
|
REACTIONS = ("😂", "😱", "👏", "🤯", "🙈", "🔥")
|
|
REACT_PER_SECOND = 6
|
|
COLORS = ["#ff4d6d", "#4cc9f0", "#ffd166", "#06d6a0", "#b388ff", "#ff9f1c", "#f15bb5", "#90e0ef",
|
|
"#c0ca33", "#e0aaff"]
|
|
|
|
TIMED_PHASES = {"rules", "theme", "word", "draw", "guess", "reveal"}
|
|
|
|
RULES = [
|
|
{"icon": "✏️", "title": "Skriv et ord", "body": "Alle skriver et ord eller en kort frase inden for temaet.",
|
|
"say": "Sådan her spiller vi. Først skriver alle et ord inden for temaet. Det må gerne være lidt svært at tegne."},
|
|
{"icon": "🎨", "title": "Tegn det", "body": "Telefonen sender ordet videre. Den næste tegner det.",
|
|
"say": "Så sender jeg ordet videre til den næste, som skal tegne det. Ingen bogstaver, kun streger."},
|
|
{"icon": "🤔", "title": "Gæt tegningen", "body": "Den næste ser kun tegningen og skriver, hvad den forestiller.",
|
|
"say": "Den næste ser kun tegningen og skal gætte, hvad det er. Og så tegner den næste igen det gæt."},
|
|
{"icon": "🔁", "title": "Hele vejen rundt", "body": "Skiftevis tegn og gæt, til kæden har været hos alle.",
|
|
"say": "Sådan går det skiftevis, til kæden har været hele vejen rundt. Lad os se, hvad der er tilbage."},
|
|
{"icon": "❤️", "title": "Hjerter og priser", "body": "Giv hjerter til de bedste tegninger og gæt. Point og priser til sidst.",
|
|
"say": "Til sidst viser jeg alle kæderne. Giv hjerter til de bedste tegninger og gæt. Held og lykke!"},
|
|
]
|
|
|
|
|
|
def _num(v) -> bool:
|
|
return isinstance(v, (int, float)) and not isinstance(v, bool) and math.isfinite(v)
|
|
|
|
|
|
def clean_drawing(strokes) -> list[dict]:
|
|
"""Validerer en tegning fra telefonen. Serveren stoler ikke på klienten: ukendte farver, punkter
|
|
uden for lærredet og for store tegninger afvises med `drawing_invalid`."""
|
|
def bad(msg="Tegningen kunne ikke læses."):
|
|
return GameError("drawing_invalid", msg)
|
|
|
|
if not isinstance(strokes, list):
|
|
raise bad()
|
|
if len(strokes) > MAX_STROKES:
|
|
raise bad("Tegningen har for mange streger.")
|
|
out, total = [], 0
|
|
for s in strokes:
|
|
if not isinstance(s, dict) or s.get("color") not in PALETTE:
|
|
raise bad()
|
|
width, points = s.get("width"), s.get("points")
|
|
if not _num(width) or not MIN_WIDTH <= width <= MAX_WIDTH or not isinstance(points, list) or not points:
|
|
raise bad()
|
|
total += len(points)
|
|
if total > MAX_POINTS:
|
|
raise bad("Tegningen er for stor.")
|
|
clean = []
|
|
for p in points:
|
|
if not isinstance(p, list) or len(p) != 2 or not all(_num(v) and 0 <= v <= 1 for v in p):
|
|
raise bad()
|
|
clean.append([round(p[0], 4), round(p[1], 4)])
|
|
out.append({"color": s["color"], "width": round(width, 4), "points": clean})
|
|
if len(json.dumps(out, separators=(",", ":"))) > MAX_BYTES:
|
|
raise bad("Tegningen er for stor.")
|
|
return out
|
|
|
|
|
|
def clean_text(text: str, code: str, what: str) -> str:
|
|
"""Ord og gæt: samme rensning som løgnene i Fup og Fakta, små bogstaver (storskærmen viser versaler)."""
|
|
text = sanitize(str(text or "")).lower()
|
|
if not text or len(text) > MAX_TEXT or not normalize(text):
|
|
raise GameError(code, f"{what} skal være 1-{MAX_TEXT} bogstaver eller tal.")
|
|
return text
|
|
|
|
|
|
def step_kind(t: int) -> str:
|
|
return "word" if t == 0 else "draw" if t % 2 else "guess"
|
|
|
|
|
|
@dataclass
|
|
class Player:
|
|
id: str
|
|
name: str
|
|
color: str
|
|
token: str
|
|
role: str = "player" # player | audience
|
|
score: int = 0
|
|
connected: bool = False
|
|
joined_order: int = 0
|
|
likes: int = 0
|
|
|
|
|
|
class Game:
|
|
game = "tegn"
|
|
show_rules = True # reglerne før rummets første spil (testene slår dem fra)
|
|
fest = False
|
|
|
|
def __init__(self, code: str, themes: list[dict], join_url: str = "", rng: random.Random | None = None):
|
|
self.code = code
|
|
self.join_url = join_url
|
|
self.rng = rng or random.Random()
|
|
# [{"id": "dyr", "name": "Dyr", "words": [...]}]
|
|
self.themes = [t for t in themes if t.get("words")]
|
|
self.players: dict[str, Player] = {}
|
|
self.audience: dict[str, Player] = {}
|
|
self._counter = 0
|
|
self.phase = "lobby"
|
|
self.deadline: float | None = None
|
|
self.hold_until: float | None = None
|
|
self.version = 0
|
|
self.events: list[dict] = []
|
|
self.rules_done = False
|
|
self.rules_started_at: float | None = None
|
|
self.rules_steps_ms: list[int] = []
|
|
self.rules_audio: list | None = None
|
|
self.theme: dict | None = None
|
|
self.chosen_theme: str | None = None
|
|
# Kæderne: order[k] ejer kæde k; chains[k][t] er bidraget i trin t
|
|
self.order: list[str] = []
|
|
self.chains: list[list[dict]] = []
|
|
self.step = 0
|
|
self.submitted: set[str] = set()
|
|
self.drafts: dict[str, object] = {}
|
|
# Afsløringen
|
|
self.reveal_id = 0
|
|
self.plan: list[dict] = [] # [{"chain", "entry", "ms", "audio"}]; entry -1 = intro, N = slut
|
|
self.reveal_started_at: float | None = None
|
|
self.likes: dict[str, set[str]] = {}
|
|
self.survived: list[bool] = []
|
|
self.final_result: dict | None = None
|
|
self._react_bucket: dict[str, list[float]] = {}
|
|
|
|
# ── hjælpere ────────────────────────────────────────────────────────────
|
|
def _next_id(self, prefix: str) -> str:
|
|
self._counter += 1
|
|
return f"{prefix}{self._counter}"
|
|
|
|
def _changed(self):
|
|
self.version += 1
|
|
|
|
def _emit(self, kind: str, **data):
|
|
self.events.append({"type": kind, **data})
|
|
|
|
def drain_events(self) -> list[dict]:
|
|
events, self.events = self.events, []
|
|
return events
|
|
|
|
def held(self, now: float) -> bool:
|
|
return self.hold_until is not None and now < self.hold_until
|
|
|
|
def phase_key(self) -> str:
|
|
return f"{self.reveal_id}.{self.step}.{self.phase}"
|
|
|
|
def vip_id(self) -> str | None:
|
|
return next(iter(self.players), None)
|
|
|
|
def _member(self, member_id: str) -> Player | None:
|
|
return self.players.get(member_id) or self.audience.get(member_id)
|
|
|
|
def _require_phase(self, *phases):
|
|
if self.phase not in phases:
|
|
raise GameError("wrong_phase", "Det kan du ikke lige nu.")
|
|
|
|
@property
|
|
def n(self) -> int:
|
|
"""Antal spillere i kæderne."""
|
|
return len(self.order)
|
|
|
|
@property
|
|
def steps(self) -> int:
|
|
"""Antal trin. Kæden skal slutte på et gæt: ved et lige antal spillere tegner ejeren selv sit ord
|
|
i trin 1, og så bliver der et trin mere."""
|
|
return self.n if self.n % 2 else self.n + 1
|
|
|
|
def own_step(self, t: int | None = None) -> bool:
|
|
"""Trin 1 ved et lige antal spillere: ejeren tegner sit eget ord."""
|
|
return (self.step if t is None else t) == 1 and self.n % 2 == 0
|
|
|
|
@property
|
|
def rules(self) -> list[dict]:
|
|
return RULES
|
|
|
|
@property
|
|
def reaction_set(self) -> tuple[str, ...]:
|
|
return REACTIONS
|
|
|
|
def chain_for(self, member_id: str, t: int | None = None) -> int | None:
|
|
"""Kæden, spilleren arbejder på i trin t (None hvis spilleren ikke er med i runden)."""
|
|
if member_id not in self.order:
|
|
return None
|
|
t = self.step if t is None else t
|
|
if self.n % 2 == 0 and t > 0:
|
|
t -= 1 # trin 1 bliver hos ejeren, rotationen starter et trin senere
|
|
return (self.order.index(member_id) - t) % self.n
|
|
|
|
def _entry(self, entry_id: str) -> dict | None:
|
|
for chain in self.chains:
|
|
for e in chain:
|
|
if e["id"] == entry_id:
|
|
return e
|
|
return None
|
|
|
|
# ── medlemmer ───────────────────────────────────────────────────────────
|
|
def join(self, name: str) -> Player:
|
|
name = sanitize(name, NAME_PUNCT)
|
|
if not name or len(name) > MAX_NAME:
|
|
raise GameError("name_invalid", f"Navnet skal være 1-{MAX_NAME} bogstaver eller tal.")
|
|
taken = {normalize(p.name) for p in list(self.players.values()) + list(self.audience.values())}
|
|
if normalize(name) in taken:
|
|
raise GameError("name_taken", "Det navn er taget. Vælg et andet.")
|
|
as_player = self.phase == "lobby" and len(self.players) < MAX_PLAYERS
|
|
used = {p.color for p in self.players.values()}
|
|
color = next((c for c in COLORS if c not in used), COLORS[0]) if as_player else "#8d99ae"
|
|
p = Player(id=self._next_id("p" if as_player else "u"), name=name, color=color,
|
|
token=secrets.token_urlsafe(16), role="player" if as_player else "audience",
|
|
connected=True, joined_order=self._counter)
|
|
(self.players if as_player else self.audience)[p.id] = p
|
|
self._emit("join", name=p.name, role=p.role)
|
|
self._changed()
|
|
return p
|
|
|
|
def resume(self, token: str) -> Player:
|
|
for p in list(self.players.values()) + list(self.audience.values()):
|
|
if secrets.compare_digest(p.token, token or ""):
|
|
return p
|
|
raise GameError("unknown_token", "Vi kunne ikke finde dig i spillet. Join igen.")
|
|
|
|
def set_connected(self, member_id: str, connected: bool, now: float) -> None:
|
|
p = self._member(member_id)
|
|
if p and p.connected != connected:
|
|
p.connected = connected
|
|
self._changed()
|
|
if not connected:
|
|
self._maybe_finish_early(now) # en der forsvinder, må ikke holde de andre tilbage
|
|
|
|
def kick(self, member_id: str, now: float) -> None:
|
|
p = self.players.pop(member_id, None) or self.audience.pop(member_id, None)
|
|
if not p:
|
|
return
|
|
# Spillerens kæde og pladser i rotationen bliver; de tomme trin fyldes ud, når tiden går.
|
|
self._maybe_finish_early(now)
|
|
if self.phase != "lobby" and not self.players:
|
|
self.phase, self.deadline = "lobby", None
|
|
self._changed()
|
|
|
|
# ── spilforløb ──────────────────────────────────────────────────────────
|
|
def start(self, member_id: str | None, now: float) -> None:
|
|
"""member_id None = værten."""
|
|
self._require_phase("lobby")
|
|
if member_id is not None and member_id != self.vip_id():
|
|
raise GameError("not_allowed", "Det er kun VIP'en, der kan starte spillet.")
|
|
if len(self.players) < MIN_PLAYERS:
|
|
raise GameError("too_few_players", f"Der skal være mindst {MIN_PLAYERS} spillere.")
|
|
for p in self.players.values():
|
|
p.score = p.likes = 0
|
|
self.final_result = None
|
|
self.order = list(self.players)
|
|
self._emit("game_start", players=len(self.players))
|
|
if self.show_rules and not self.rules_done:
|
|
self.rules_done = True
|
|
self.phase = "rules"
|
|
self.rules_audio = None
|
|
self.start_rules_timeline(now)
|
|
self._emit("rules")
|
|
return
|
|
self._begin_theme(now)
|
|
|
|
def start_rules_timeline(self, now: float, steps_ms: list[int] | None = None):
|
|
self.rules_steps_ms = list(steps_ms or [RULES_STEP_MS] * len(self.rules))
|
|
self.rules_started_at = now
|
|
self.deadline = now + sum(self.rules_steps_ms) / 1000 + 1
|
|
self._changed()
|
|
|
|
def hold_rules(self, now: float, max_wait: float = 10.0):
|
|
if self.phase == "rules":
|
|
self.rules_started_at = None
|
|
self.deadline = now + max_wait
|
|
self._changed()
|
|
|
|
def play_again(self, member_id: str | None, now: float) -> None:
|
|
self._require_phase("final")
|
|
if member_id is not None and member_id != self.vip_id():
|
|
raise GameError("not_allowed", "Det er kun VIP'en, der kan starte et nyt spil.")
|
|
# Publikum, der kom til undervejs, rykker op som spillere, hvis der er plads.
|
|
for a in list(self.audience.values()):
|
|
if len(self.players) >= MAX_PLAYERS:
|
|
break
|
|
del self.audience[a.id]
|
|
a.role = "player"
|
|
used = {p.color for p in self.players.values()}
|
|
a.color = next((c for c in COLORS if c not in used), COLORS[0])
|
|
self.players[a.id] = a
|
|
self.phase = "lobby"
|
|
self._emit("play_again")
|
|
self.start(None, now)
|
|
|
|
def _begin_theme(self, now: float):
|
|
if not self.themes:
|
|
raise GameError("no_themes", "Der er ingen temaer.")
|
|
self.phase = "theme"
|
|
self.chosen_theme = None
|
|
self.deadline = now + THEME_SECONDS
|
|
vip = self.players.get(self.vip_id() or "")
|
|
self._emit("theme", vip=vip.name if vip else "")
|
|
self._changed()
|
|
|
|
def choose_theme(self, member_id: str | None, theme_id: str, now: float) -> None:
|
|
"""Værten (member_id None) eller VIP'en vælger temaet."""
|
|
self._require_phase("theme")
|
|
if member_id is not None and member_id != self.vip_id():
|
|
raise GameError("not_allowed", "Det er VIP'en eller værten, der vælger tema.")
|
|
if theme_id not in {t["id"] for t in self.themes}:
|
|
raise GameError("answer_invalid", "Ukendt tema.")
|
|
if self.chosen_theme is not None:
|
|
raise GameError("already_done", "Temaet er valgt.")
|
|
self.chosen_theme = theme_id
|
|
if self.held(now):
|
|
self.deadline = now # skiftet sker, når speakeren er færdig
|
|
self._changed()
|
|
else:
|
|
self._start_chains(now)
|
|
|
|
def _start_chains(self, now: float, timeout: bool = False):
|
|
if self.chosen_theme is None:
|
|
self.chosen_theme = self.rng.choice(self.themes)["id"]
|
|
self.theme = next(t for t in self.themes if t["id"] == self.chosen_theme)
|
|
self._emit("theme_chosen", theme=self.theme["name"], timeout=timeout)
|
|
self.order = [pid for pid in self.order if pid in self.players]
|
|
if len(self.order) < MIN_PLAYERS: # værten smed spillere ud under reglerne/temaet
|
|
self.phase, self.deadline = "lobby", None
|
|
self._changed()
|
|
return
|
|
self.chains = [[] for _ in self.order]
|
|
self.likes = {}
|
|
self.survived = []
|
|
self.plan = []
|
|
self.reveal_started_at = None
|
|
self._begin_step(0, now)
|
|
|
|
def _begin_step(self, t: int, now: float):
|
|
self.step = t
|
|
self.submitted = set()
|
|
self.drafts = {}
|
|
self.phase = step_kind(t)
|
|
self.deadline = now + {"word": WORD_SECONDS, "draw": DRAW_SECONDS, "guess": GUESS_SECONDS}[self.phase]
|
|
self._emit("step", step=self.phase, number=t + 1, of=self.steps, theme=self.theme["name"],
|
|
own=self.own_step(t))
|
|
self._changed()
|
|
|
|
def task(self, member_id: str) -> dict | None:
|
|
"""Spillerens trin lige nu: kun det forrige bidrag i kæden, aldrig resten."""
|
|
if self.phase not in ("word", "draw", "guess"):
|
|
return None
|
|
c = self.chain_for(member_id)
|
|
if c is None:
|
|
return None
|
|
prev = self.chains[c][self.step - 1] if self.step > 0 else None
|
|
mine = self.chains[c][self.step] if len(self.chains[c]) > self.step else None
|
|
return {"kind": self.phase, "done": member_id in self.submitted,
|
|
"from": prev["by"] if prev else None,
|
|
"text": prev["text"] if prev and prev["kind"] != "drawing" else None,
|
|
"drawing": prev["strokes"] if prev and prev["kind"] == "drawing" else None,
|
|
"mine": mine["text"] if mine and mine["kind"] != "drawing" else None,
|
|
"draft": self.drafts.get(member_id)}
|
|
|
|
def _require_task(self, member_id: str, phase: str) -> int:
|
|
self._require_phase(phase)
|
|
if member_id not in self.players:
|
|
raise GameError("not_allowed", "Det kan kun spillere.")
|
|
c = self.chain_for(member_id)
|
|
if c is None:
|
|
raise GameError("not_allowed", "Du er ikke med i denne runde.")
|
|
if member_id in self.submitted:
|
|
raise GameError("already_done", "Du har allerede afleveret.")
|
|
return c
|
|
|
|
def _put(self, c: int, member_id: str, value, auto: bool = False):
|
|
kind = {"word": "word", "draw": "drawing", "guess": "guess"}[self.phase]
|
|
entry = {"id": f"c{c}e{self.step}", "kind": kind, "by": member_id, "auto": auto,
|
|
"text": None if kind == "drawing" else value, "strokes": value if kind == "drawing" else None}
|
|
self.chains[c].append(entry)
|
|
|
|
def submit_word(self, member_id: str, text: str, now: float) -> None:
|
|
c = self._require_task(member_id, "word")
|
|
self._submit(c, member_id, clean_text(text, "word_invalid", "Ordet"), now)
|
|
|
|
def submit_drawing(self, member_id: str, strokes, now: float) -> None:
|
|
c = self._require_task(member_id, "draw")
|
|
self._submit(c, member_id, clean_drawing(strokes), now)
|
|
|
|
def submit_guess(self, member_id: str, text: str, now: float) -> None:
|
|
c = self._require_task(member_id, "guess")
|
|
self._submit(c, member_id, clean_text(text, "guess_invalid", "Gættet"), now)
|
|
|
|
def _submit(self, c: int, member_id: str, value, now: float):
|
|
self._put(c, member_id, value)
|
|
self.submitted.add(member_id)
|
|
self.drafts.pop(member_id, None)
|
|
self._emit("submitted", player=member_id, step=self.phase)
|
|
self._changed()
|
|
self._maybe_finish_early(now)
|
|
|
|
def save_draft(self, member_id: str, value) -> None:
|
|
"""Det, telefonen har nået indtil nu. Bruges, hvis tiden løber ud. Ingen ny tilstand til skærmene."""
|
|
self._require_phase("word", "draw", "guess")
|
|
self._require_task(member_id, self.phase)
|
|
if self.phase == "draw":
|
|
self.drafts[member_id] = clean_drawing(value)
|
|
else:
|
|
text = sanitize(str(value or "")).lower()[:MAX_TEXT]
|
|
if normalize(text):
|
|
self.drafts[member_id] = text
|
|
else:
|
|
self.drafts.pop(member_id, None)
|
|
|
|
def suggest(self, member_id: str) -> str:
|
|
""""Giv mig et forslag": et eksempel-ord fra temaet, som ingen andre har skrevet."""
|
|
self._require_task(member_id, "word")
|
|
return self._example_word()
|
|
|
|
def _example_word(self) -> str:
|
|
taken = {normalize(e["text"]) for chain in self.chains for e in chain if e["kind"] == "word"}
|
|
words = self.theme["words"]
|
|
free = [w for w in words if normalize(w) not in taken] or words
|
|
return sanitize(self.rng.choice(free)).lower()
|
|
|
|
def _all_done(self) -> bool:
|
|
if self.phase not in ("word", "draw", "guess"):
|
|
return False
|
|
active = [pid for pid in self.order if pid in self.players and self.players[pid].connected]
|
|
return bool(active) and all(pid in self.submitted for pid in active)
|
|
|
|
def _maybe_finish_early(self, now: float):
|
|
if not self._all_done():
|
|
return
|
|
if self.held(now):
|
|
self.deadline = now # alle er færdige; skiftet venter på speakeren
|
|
self._changed()
|
|
else:
|
|
self._emit("step_all_done")
|
|
self._end_step(now)
|
|
|
|
def _end_step(self, now: float):
|
|
"""Tiden er gået (eller alle er færdige): det, der mangler, fyldes ud, så spillet aldrig hænger."""
|
|
for pid in self.order:
|
|
c = self.chain_for(pid)
|
|
if len(self.chains[c]) > self.step:
|
|
continue
|
|
draft = self.drafts.get(pid)
|
|
if self.phase == "word":
|
|
value = draft or self._example_word()
|
|
elif self.phase == "draw":
|
|
value = draft or []
|
|
else:
|
|
value = draft or EMPTY_GUESS
|
|
self._put(c, pid, value, auto=True)
|
|
if self.step + 1 < self.steps:
|
|
self._begin_step(self.step + 1, now)
|
|
else:
|
|
self._start_reveal(now)
|
|
|
|
# ── afsløringen ─────────────────────────────────────────────────────────
|
|
def _start_reveal(self, now: float):
|
|
self.reveal_id += 1
|
|
self.likes = {}
|
|
self.survived = []
|
|
for chain in self.chains:
|
|
# Sidste gæt mod første ord
|
|
last = self._last_guess(chain)
|
|
self.survived.append(last is not None and last != EMPTY_GUESS
|
|
and normalize(last) == normalize(chain[0]["text"]))
|
|
self.plan = []
|
|
for k, chain in enumerate(self.chains):
|
|
self.plan.append({"chain": k, "entry": -1, "ms": CHAIN_INTRO_MS, "audio": None})
|
|
for i, e in enumerate(chain):
|
|
ms = {"word": WORD_MS, "drawing": DRAWING_MS, "guess": GUESS_MS}[e["kind"]]
|
|
self.plan.append({"chain": k, "entry": i, "ms": ms, "audio": None})
|
|
self.plan.append({"chain": k, "entry": len(chain), "ms": CHAIN_END_MS, "audio": None})
|
|
self.phase = "reveal"
|
|
self.step = self.steps
|
|
self.start_reveal_timeline(now)
|
|
self._emit("reveal")
|
|
self._changed()
|
|
|
|
@staticmethod
|
|
def _last_guess(chain: list[dict]) -> str | None:
|
|
return next((e["text"] for e in reversed(chain) if e["kind"] == "guess"), None)
|
|
|
|
def start_reveal_timeline(self, now: float, steps_ms: list[int] | None = None):
|
|
if steps_ms:
|
|
for beat, ms in zip(self.plan, steps_ms):
|
|
beat["ms"] = ms
|
|
self.reveal_started_at = now
|
|
self.deadline = now + sum(b["ms"] for b in self.plan) / 1000
|
|
self._changed()
|
|
|
|
def hold_reveal(self, now: float, max_wait: float = 10.0):
|
|
"""Vent med at starte afsløringen (speakeren laver lyd). Starter selv efter max_wait."""
|
|
if self.phase == "reveal":
|
|
self.reveal_started_at = None
|
|
self.deadline = now + max_wait
|
|
self._changed()
|
|
|
|
def beat_at(self, now: float) -> int | None:
|
|
"""Indeks i planen, der er på storskærmen lige nu (None før start)."""
|
|
if self.phase != "reveal" or self.reveal_started_at is None:
|
|
return None
|
|
el = (now - self.reveal_started_at) * 1000
|
|
for i, b in enumerate(self.plan):
|
|
if el < b["ms"]:
|
|
return i
|
|
el -= b["ms"]
|
|
return len(self.plan) - 1
|
|
|
|
def shown_beats(self, now: float) -> list[dict]:
|
|
"""Trinene, storskærmen har vist indtil nu (inklusive det aktuelle)."""
|
|
i = self.beat_at(now)
|
|
return [] if i is None else self.plan[: i + 1]
|
|
|
|
def _revealed(self, entry_id: str, now: float) -> bool:
|
|
return any(0 <= b["entry"] < len(self.chains[b["chain"]]) and self.chains[b["chain"]][b["entry"]]["id"] == entry_id
|
|
for b in self.shown_beats(now))
|
|
|
|
def like(self, member_id: str, entry_id: str, now: float) -> None:
|
|
self._require_phase("reveal")
|
|
if not self._member(member_id):
|
|
raise GameError("not_allowed", "Du er ikke med i spillet.")
|
|
e = self._entry(str(entry_id))
|
|
if not e or e["kind"] not in ("drawing", "guess"):
|
|
raise GameError("answer_invalid", "Hjerter er til tegninger og gæt.")
|
|
if e["by"] == member_id:
|
|
raise GameError("answer_invalid", "Du kan ikke give dig selv et hjerte.")
|
|
if not self._revealed(e["id"], now):
|
|
raise GameError("answer_invalid", "Den er ikke vist endnu.")
|
|
voters = self.likes.setdefault(e["id"], set())
|
|
sign = -1 if member_id in voters else 1
|
|
(voters.discard if sign < 0 else voters.add)(member_id)
|
|
author = self.players.get(e["by"])
|
|
if author:
|
|
author.score += sign * POINTS_LIKE
|
|
author.likes += sign
|
|
self._changed()
|
|
|
|
def featured(self, now: float) -> str | None:
|
|
"""Bidraget i fokus på storskærmen (reaktioner flyver op fra det)."""
|
|
i = self.beat_at(now)
|
|
if i is None:
|
|
return None
|
|
b = self.plan[i]
|
|
chain = self.chains[b["chain"]]
|
|
return chain[b["entry"]]["id"] if 0 <= b["entry"] < len(chain) else None
|
|
|
|
def react(self, member_id: str, emoji: str, now: float) -> dict | None:
|
|
if self.phase != "reveal" or emoji not in self.reaction_set:
|
|
return None
|
|
member = self._member(member_id)
|
|
if not member:
|
|
return None
|
|
bucket = [t for t in self._react_bucket.get(member_id, []) if now - t < 1.0]
|
|
if len(bucket) >= REACT_PER_SECOND:
|
|
self._react_bucket[member_id] = bucket
|
|
return None
|
|
bucket.append(now)
|
|
self._react_bucket[member_id] = bucket
|
|
return {"emoji": emoji, "color": member.color, "answer_id": self.featured(now)}
|
|
|
|
def skip_chain(self, now: float):
|
|
"""Værten springer resten af den aktuelle kæde over."""
|
|
i = self.beat_at(now)
|
|
offset = 0
|
|
for j, b in enumerate(self.plan):
|
|
if j > i and b["entry"] == -1:
|
|
self.reveal_started_at = now - offset / 1000
|
|
self.deadline = self.reveal_started_at + sum(b["ms"] for b in self.plan) / 1000
|
|
self._changed()
|
|
return
|
|
offset += b["ms"]
|
|
self._finish()
|
|
|
|
def _finish(self):
|
|
for k, ok in enumerate(self.survived):
|
|
owner = self.players.get(self.order[k])
|
|
if ok and owner:
|
|
owner.score += POINTS_SURVIVED
|
|
ranking = sorted(self.players.values(), key=lambda p: (-p.score, p.joined_order))
|
|
places, last_score, place = [], None, 0
|
|
for i, p in enumerate(ranking, 1):
|
|
if p.score != last_score:
|
|
place, last_score = i, p.score
|
|
places.append({"player_id": p.id, "score": p.score, "place": place})
|
|
self.final_result = {"ranking": places, "awards": self._awards()}
|
|
winner = ranking[0] if ranking else None
|
|
self._emit("final", winner=winner.name if winner else "", points=winner.score if winner else 0,
|
|
awards=[{"kind": a["kind"], "name": self.players[a["player_id"]].name}
|
|
for a in self.final_result["awards"]])
|
|
self.phase = "final"
|
|
self.deadline = None
|
|
self._changed()
|
|
|
|
def _awards(self) -> list[dict]:
|
|
"""Priserne. Ved lighed vinder den, der blev vist først."""
|
|
awards = []
|
|
entries = [e for chain in self.chains for e in chain]
|
|
|
|
def most_liked(kind, title, detail):
|
|
best = None
|
|
for e in entries:
|
|
n = len(self.likes.get(e["id"], ()))
|
|
if e["kind"] == kind and n and e["by"] in self.players and (best is None or n > best[1]):
|
|
best = (e, n)
|
|
if best:
|
|
e, n = best
|
|
awards.append({"title": title, "player_id": e["by"], "detail": detail(e, n), "kind": kind,
|
|
"entry_id": e["id"], "count": n})
|
|
|
|
hearts = lambda n: f"{n} hjerte{'r' if n != 1 else ''}" # noqa: E731
|
|
most_liked("drawing", "Bedste tegning", lambda e, n: hearts(n))
|
|
most_liked("guess", "Bedste gæt", lambda e, n: f"\"{e['text']}\", {hearts(n)}")
|
|
owners = [(k, self.order[k]) for k in range(len(self.chains)) if self.order[k] in self.players]
|
|
alive = [(k, pid) for k, pid in owners if self.survived[k]]
|
|
if alive:
|
|
k, pid = alive[0]
|
|
more = f" (og {len(alive) - 1} til)" if len(alive) > 1 else ""
|
|
awards.append({"title": "Kæden der overlevede", "player_id": pid, "kind": "survived",
|
|
"detail": f"\"{self.chains[k][0]['text']}\" hele vejen rundt{more}", "count": len(alive)})
|
|
changed = []
|
|
for k, pid in owners:
|
|
first, last = self.chains[k][0]["text"], self._last_guess(self.chains[k])
|
|
if self.survived[k] or last in (None, EMPTY_GUESS):
|
|
continue
|
|
changed.append((difflib.SequenceMatcher(None, normalize(first), normalize(last)).ratio(), k, pid, first, last))
|
|
if changed:
|
|
_, k, pid, first, last = min(changed, key=lambda x: (x[0], x[1]))
|
|
awards.append({"title": "Største forvandling", "player_id": pid, "kind": "transformed",
|
|
"detail": f"\"{first}\" blev til \"{last}\"", "count": 1})
|
|
return awards
|
|
|
|
# ── tid ─────────────────────────────────────────────────────────────────
|
|
def skip(self, now: float) -> None:
|
|
if self.phase not in TIMED_PHASES:
|
|
raise GameError("wrong_phase", "Der er ikke noget at springe over.")
|
|
if self.phase == "reveal" and self.reveal_started_at is not None:
|
|
self.skip_chain(now)
|
|
return
|
|
self._advance(now)
|
|
|
|
def tick(self, now: float) -> bool:
|
|
"""Kaldes når deadline er nået. Returnerer True hvis noget ændrede sig."""
|
|
if self.deadline is None or now < self.deadline or self.held(now):
|
|
return False
|
|
self._advance(now)
|
|
return True
|
|
|
|
def _advance(self, now: float):
|
|
if self.phase == "rules":
|
|
if self.rules_started_at is None:
|
|
self.start_rules_timeline(now) # lyden blev aldrig klar: vis kortene uden stemme
|
|
else:
|
|
self._begin_theme(now)
|
|
elif self.phase == "theme":
|
|
self._start_chains(now, timeout=self.chosen_theme is None)
|
|
elif self.phase in ("word", "draw", "guess"):
|
|
self._end_step(now)
|
|
elif self.phase == "reveal":
|
|
if self.reveal_started_at is None:
|
|
self.start_reveal_timeline(now) # lyden blev aldrig klar: kør uden
|
|
else:
|
|
self._finish()
|
|
|
|
# ── snapshot ────────────────────────────────────────────────────────────
|
|
def snapshot(self, now: float, viewer: Player | None = None, host: bool = False) -> dict:
|
|
vip = self.vip_id()
|
|
stepping = self.phase in ("word", "draw", "guess")
|
|
state = {
|
|
"game": self.game,
|
|
"code": self.code,
|
|
"phase": self.phase,
|
|
"server_time": int(now * 1000),
|
|
"deadline": int(self.deadline * 1000) if self.deadline else None,
|
|
"join_url": self.join_url,
|
|
"min_players": MIN_PLAYERS,
|
|
"max_players": MAX_PLAYERS,
|
|
"round": None,
|
|
"players": [
|
|
{"id": p.id, "name": p.name, "color": p.color, "score": p.score, "connected": p.connected,
|
|
"vip": p.id == vip, "done": stepping and p.id in self.submitted}
|
|
for p in self.players.values()
|
|
],
|
|
"audience_count": len(self.audience),
|
|
"fest": False,
|
|
"reactions": list(self.reaction_set),
|
|
}
|
|
if self.phase == "rules":
|
|
state["rules"] = {"items": [{k: r[k] for k in ("icon", "title", "body")} for r in self.rules],
|
|
"started_at": int(self.rules_started_at * 1000) if self.rules_started_at else None,
|
|
"steps_ms": self.rules_steps_ms, "audio": self.rules_audio}
|
|
if self.phase == "theme":
|
|
state["theme"] = {"options": [{"id": t["id"], "name": t["name"], "icon": t.get("icon", "")}
|
|
for t in self.themes], "chosen_id": self.chosen_theme}
|
|
if stepping or self.phase == "reveal":
|
|
state["theme"] = {"name": self.theme["name"], "icon": self.theme.get("icon", "")}
|
|
if stepping:
|
|
state["step"] = {"index": self.step, "number": self.step + 1, "of": self.steps, "kind": self.phase,
|
|
"own": self.own_step()}
|
|
if self.phase == "reveal":
|
|
state["reveal"] = {
|
|
"id": self.reveal_id,
|
|
"started_at": int(self.reveal_started_at * 1000) if self.reveal_started_at else None,
|
|
# speakerens replik (audio) kun til storskærmen: kædens slutning nævner sidste gæt
|
|
"plan": [{"chain": b["chain"], "entry": b["entry"], "ms": b["ms"], **({"audio": b["audio"]} if host else {})}
|
|
for b in self.plan],
|
|
# Tegningerne følger ikke med her (de er store); storskærmen får dem i beskeden "drawings".
|
|
"chains": self._reveal_chains(now, everything=host),
|
|
}
|
|
if self.phase == "final":
|
|
state["final"] = self.final_result
|
|
if host:
|
|
state["host"] = True
|
|
elif viewer:
|
|
state["me"] = {
|
|
"id": viewer.id, "name": viewer.name, "role": viewer.role, "vip": viewer.id == vip,
|
|
"score": viewer.score,
|
|
"task": self.task(viewer.id) if viewer.role == "player" else None,
|
|
"likes": sorted(eid for eid, voters in self.likes.items() if viewer.id in voters)
|
|
if self.phase == "reveal" else [],
|
|
"can_choose": self.phase == "theme" and viewer.id == vip,
|
|
}
|
|
return state
|
|
|
|
def _reveal_chains(self, now: float, everything: bool) -> list[dict]:
|
|
"""Kæderne i afsløringen. Telefonerne får kun det, storskærmen har vist (ellers kunne man læse
|
|
sidste gæt før tid); storskærmen får det hele og styrer selv tempoet. Rummet sender et nyt
|
|
snapshot ved hvert trin."""
|
|
shown = None if everything else {(b["chain"], b["entry"]) for b in self.shown_beats(now)}
|
|
out = []
|
|
for k, chain in enumerate(self.chains):
|
|
entries = [{"id": e["id"], "kind": e["kind"], "by": e["by"], "text": e["text"], "auto": e["auto"],
|
|
"likes": len(self.likes.get(e["id"], ()))}
|
|
for i, e in enumerate(chain) if shown is None or (k, i) in shown]
|
|
ended = shown is None or (k, len(chain)) in shown
|
|
out.append({"owner": self.order[k], "survived": self.survived[k] if ended else None, "entries": entries})
|
|
return out
|
|
|
|
def beat_starts(self) -> list[float]:
|
|
"""Hvornår hvert trin i afsløringen begynder (sekunder siden epoch)."""
|
|
if self.reveal_started_at is None:
|
|
return []
|
|
t, out = self.reveal_started_at, []
|
|
for b in self.plan:
|
|
out.append(t)
|
|
t += b["ms"] / 1000
|
|
return out
|
|
|
|
def drawings(self) -> dict[str, list]:
|
|
"""Alle tegninger i den aktuelle afsløring (til storskærmen)."""
|
|
return {e["id"]: e["strokes"] for chain in self.chains for e in chain if e["kind"] == "drawing"}
|