diff --git a/partyhub/settings.py b/partyhub/settings.py index a61368c..3d959b9 100644 --- a/partyhub/settings.py +++ b/partyhub/settings.py @@ -75,6 +75,7 @@ INSTALLED_APPS = [ "channels", "fupogfakta", "tegnoggaet", + "spionen", "party", ] diff --git a/spionen/__init__.py b/spionen/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/spionen/apps.py b/spionen/apps.py new file mode 100644 index 0000000..fe15dac --- /dev/null +++ b/spionen/apps.py @@ -0,0 +1,6 @@ +from django.apps import AppConfig + + +class SpionenConfig(AppConfig): + name = "spionen" + verbose_name = "Spionen" diff --git a/spionen/engine.py b/spionen/engine.py new file mode 100644 index 0000000..79638a7 --- /dev/null +++ b/spionen/engine.py @@ -0,0 +1,649 @@ +"""Spilmotoren for Spionen (social deduktion: skriv et stikord, snak, stem). + +Ren Python uden Django og uden I/O, i samme stil som `tegnoggaet.engine`: alle handlinger tager `now` (sekunder +siden epoch), rummet i `party.spion` kalder `tick()` når en deadline udløber og sender `snapshot()` til hver skærm. + +En runde: alle undtagen spionen (to ved otte eller flere spillere) ser et hemmeligt ord; spionen ser kun temaet. +Alle skriver ét stikord, stikordene vises på storskærmen, I snakker, I stemmer. Flest stemmer bliver anklaget +(lighed: ingen anklages, og spionerne slipper). Er den anklagede spion, får den et sidste gæt på ordet. + +Hemmeligheden er spillet: ordet og hvem der er spion står kun i det snapshot, der må se det (se snapshot()). +""" + +from __future__ import annotations + +import random +import re +import secrets +import unicodedata +from dataclasses import dataclass + +from fupogfakta.engine import NAME_PUNCT, GameError, normalize, sanitize + +MIN_PLAYERS = 4 +MAX_PLAYERS = 10 +TWO_SPIES_FROM = 8 # så mange spillere eller flere: to spioner +ROUNDS = 4 # standard; værten kan vælge 3-5 +ROUND_CHOICES = (3, 4, 5) +MAX_NAME = 12 +MAX_CLUE = 20 +MAX_GUESS = 30 + +CLUE_SECONDS = 40 +TALK_SECONDS = 60 +VOTE_SECONDS = 30 +GUESS_SECONDS = 30 +VERDICT_SECONDS = 7 # stemmerne og den anklagede vises; til sidst afsløres, om det var spionen +RESULT_SECONDS = 10 +RULES_STEP_MS = 5000 +CLUES_INTRO_MS = 2000 # afsløringen af stikordene: kort intro, så ét stikord ad gangen +CLUE_MS = 2600 + +POINTS_VOTE = 100 # en, der ikke er spion, og som stemte på en spion +POINTS_SPY_FREE = 300 # spionen blev ikke fanget +POINTS_SPY_GUESS = 200 # spionen blev fanget, men gættede ordet + +REACTIONS = ("😂", "😱", "👏", "🤯", "🙈", "🔥") +REACT_PER_SECOND = 6 +COLORS = ["#ff4d6d", "#4cc9f0", "#ffd166", "#06d6a0", "#b388ff", "#ff9f1c", "#f15bb5", "#90e0ef", + "#c0ca33", "#e0aaff"] + +TIMED_PHASES = {"rules", "clue", "clues", "talk", "vote", "verdict", "guess", "result"} + +RULES = [ + {"icon": "🕵️", "title": "Et hemmeligt ord", "body": "Alle ser det samme ord på telefonen. Spionen ser kun temaet.", + "say": "Sådan her spiller vi. Alle får det samme hemmelige ord på telefonen. Alle undtagen spionen, " + "som kun får temaet at vide."}, + {"icon": "✏️", "title": "Ét stikord", "body": "Skriv ét ord, der viser, at du kender ordet. Uden at afsløre det.", + "say": "Alle skriver ét stikord. Det skal vise, at I kender ordet, men uden at spionen kan gætte det."}, + {"icon": "🗣️", "title": "Snak sammen", "body": "Stikordene kommer på skærmen. Så har I et minut til at snakke.", + "say": "Så viser jeg stikordene på skærmen, og I får et minut til at snakke sammen. Hvem lyder mistænkelig?"}, + {"icon": "🗳️", "title": "Stem", "body": "Stem på den, I tror er spionen. Lige mange stemmer: spionen slipper.", + "say": "Til sidst stemmer I på telefonen. Får to lige mange stemmer, slipper spionen."}, + {"icon": "🎯", "title": "Sidste chance", "body": "Bliver spionen anklaget, kan den stadig redde sig ved at gætte ordet.", + "say": "Og bliver spionen anklaget, får den en sidste chance. Gætter spionen ordet, er det alligevel en sejr."}, +] + +# Bøjninger: ordet med en af de endelser (også med fordoblet slutkonsonant, "kat" -> "katten", og uden slut-e, +# "pande" -> "pander"). Det er groft, men dækker navneordenes ental, flertal og bestemt form plus genitiv. +ENDINGS = ("", "e", "n", "en", "et", "t", "r", "er", "ne", "ene", "rne", "erne", "s", "es", "ns", "ens", "ets", + "rs", "ers", "nes", "enes", "rnes", "ernes") + + +@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 + spy_points: int = 0 # til prisen "Bedste løgner" + right_votes: int = 0 # "Skarpeste øje" + suspected: int = 0 # "Mest mistænkt": stemmer fået i alt + + +# ── ord ────────────────────────────────────────────────────────────────────── +def squash(text: str) -> str: + """Til sammenligning: små bogstaver, kun bogstaver og tal (mellemrum og bindestreger væk), æ, ø og å bevares.""" + text = unicodedata.normalize("NFC", str(text or "")).casefold() + return "".join(ch for ch in text if ch.isalnum()) + + +def _inflection(form: str, base: str) -> bool: + """Er form en bøjning af base?""" + if not base: + return False + stems = [base] + if len(base) >= 3 and base[-1] not in "aeiouyæøå": + stems.append(base + base[-1]) # kat -> katt(en) + if len(base) >= 4 and base.endswith("e"): + stems.append(base[:-1]) # pande -> pand(er) (og "pander" er også base + "r") + return any(form.startswith(s) and form[len(s):] in ENDINGS for s in stems) + + +def levenshtein(a: str, b: str) -> int: + prev = list(range(len(b) + 1)) + for i, ca in enumerate(a, 1): + cur = [i] + for j, cb in enumerate(b, 1): + cur.append(min(prev[j] + 1, cur[j - 1] + 1, prev[j - 1] + (ca != cb))) + prev = cur + return prev[-1] + + +def typo_allowance(word: str) -> int: + """Små stavefejl: ingen i ord under fire bogstaver, én op til syv, to i længere ord.""" + n = len(word) + return 0 if n < 4 else 1 if n < 8 else 2 + + +def same_word(a: str, b: str, typos: bool = True) -> bool: + """Samme ord, uanset store/små bogstaver, mellemrum og bindestreger, bøjning og (med typos) små stavefejl.""" + x, y = squash(a), squash(b) + if not x or not y: + return False + if x == y or _inflection(x, y) or _inflection(y, x): + return True + return typos and levenshtein(x, y) <= typo_allowance(y) + + +def clean_text(text: str, code: str, what: str, max_len: int) -> str: + text = sanitize(str(text or "")).lower() + if not text or len(text) > max_len or not normalize(text): + raise GameError(code, f"{what} skal være 1-{max_len} bogstaver eller tal.") + return text + + +class Game: + game = "spion" + 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, + rounds: int = ROUNDS): + self.code = code + self.join_url = join_url + self.rng = rng or random.Random() + self.themes = [t for t in themes if t.get("words")] # [{"id", "name", "icon", "words": [...]}] + self.rounds = rounds if rounds in ROUND_CHOICES else ROUNDS + 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.round = 0 + self.order: list[str] = [] # spillerne i runden (dem der var med, da den startede) + self.spies: list[str] = [] + self.last_spies: set[str] = set() + self.used_words: set[str] = set() + self.used_themes: list[str] = [] + self.theme: dict | None = None + self.word: str | None = None + self.clues: dict[str, str] = {} # spiller -> stikord + self.clue_order: list[str] = [] # afsløringens rækkefølge (tilfældig) + self.clues_started_at: float | None = None + self.votes: dict[str, str] = {} # spiller -> den, spilleren stemte på + self.audience_votes: dict[str, str] = {} + self.accused: str | None = None + self.guess: dict | None = None # {"text", "correct"} + self.result: dict | None = None + 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.round}.{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.") + + def _in_round(self, member_id: str) -> bool: + return member_id in self.order and member_id in self.players + + @property + def rules(self) -> list[dict]: + return RULES + + @property + def reaction_set(self) -> tuple[str, ...]: + return REACTIONS + + def _name(self, pid: str | None) -> str: + p = self.players.get(pid) if pid else None + return p.name if p else "" + + # ── 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 + self.audience_votes.pop(member_id, None) + if self.phase != "lobby" and not self.players: + self.phase, self.deadline = "lobby", None + elif self.phase == "guess" and member_id == self.accused: + self._end_round(now) # den anklagede spion gik: intet gæt + else: + self._maybe_finish_early(now) + 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.") + if not self.themes: + raise GameError("no_themes", "Der er ingen temaer.") + for p in self.players.values(): + p.score = p.spy_points = p.right_votes = p.suspected = 0 + self.round, self.final_result, self.last_spies = 0, None, set() + self.used_words, self.used_themes = set(), [] + 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_round(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.") + for a in list(self.audience.values()): # publikum rykker op, hvis der er plads + 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 _pick_theme_and_word(self): + themes = [t for t in self.themes if t["id"] not in self.used_themes[-len(self.themes) + 1:]] or self.themes + self.theme = self.rng.choice(themes) + self.used_themes.append(self.theme["id"]) + words = [w for w in self.theme["words"] if w not in self.used_words] or list(self.theme["words"]) + self.word = self.rng.choice(words) + self.used_words.add(self.word) + + def _pick_spies(self) -> list[str]: + """Tilfældige spioner, men ingen er spion to runder i træk, hvis det kan undgås.""" + n = 2 if len(self.order) >= TWO_SPIES_FROM else 1 + fresh = [pid for pid in self.order if pid not in self.last_spies] + pool = fresh if len(fresh) >= n else list(self.order) + return self.rng.sample(pool, n) + + def _begin_round(self, now: float): + if len(self.players) < MIN_PLAYERS: # for få tilbage til en runde til + self._finish() if self.round else self._to_lobby() + return + self.round += 1 + self.order = list(self.players) + self._pick_theme_and_word() + self.spies = self._pick_spies() + self.last_spies = set(self.spies) + self.clues, self.clue_order, self.clues_started_at = {}, [], None + self.votes, self.audience_votes = {}, {} + self.accused, self.guess, self.result = None, None, None + self.phase = "clue" + self.deadline = now + CLUE_SECONDS + self._emit("round_start", round=self.round, of=self.rounds, theme=self.theme["name"], + last=self.round == self.rounds) + self._changed() + + def _to_lobby(self): + self.phase, self.deadline = "lobby", None + self._changed() + + # ── stikord ───────────────────────────────────────────────────────────── + def check_clue(self, member_id: str, text: str) -> str: + """Ét ord, højst MAX_CLUE tegn; ikke temaet og ikke et stikord, der er brugt i runden. For dem, der kender + ordet, heller ikke ordet selv (eller en bøjning eller en lille stavefejl af det). Spionens stikord holdes + ikke op mod ordet: en afvisning ville røbe det.""" + clue = clean_text(text, "clue_invalid", "Stikordet", MAX_CLUE) + if " " in clue: + raise GameError("clue_invalid", "Skriv kun ét ord.") + if member_id not in self.spies and same_word(clue, self.word): + raise GameError("clue_is_word", "Det er jo ordet selv. Find på noget andet.") + theme_words = [w for w in re.split(r"[\s-]+", self.theme["name"]) if len(squash(w)) >= 3] + if any(same_word(clue, w, typos=False) for w in theme_words) or same_word(clue, self.theme["name"], typos=False): + raise GameError("clue_is_theme", "Temaet må ikke være dit stikord.") + if any(same_word(clue, c, typos=False) for pid, c in self.clues.items() if pid != member_id): + raise GameError("clue_taken", "Det stikord har en anden allerede skrevet.") + return clue + + def submit_clue(self, member_id: str, text: str, now: float) -> None: + self._require_phase("clue") + if not self._in_round(member_id): + raise GameError("not_in_round", "Du er ikke med i denne runde.") + self.clues[member_id] = self.check_clue(member_id, text) + self._changed() + self._maybe_finish_early(now) + + def _start_clues(self, now: float): + self.clue_order = [pid for pid in self.order if pid in self.players] + self.rng.shuffle(self.clue_order) # tilfældig rækkefølge: den siger intet om, hvem der er spion + self.phase = "clues" + self.clues_started_at = now + self.deadline = now + (CLUES_INTRO_MS + CLUE_MS * len(self.clue_order)) / 1000 + self._emit("clues", count=len(self.clue_order), missing=sum(pid not in self.clues for pid in self.clue_order)) + self._changed() + + # ── afstemning ────────────────────────────────────────────────────────── + def vote(self, member_id: str, target: str, now: float) -> None: + self._require_phase("vote") + if target not in self.order or target not in self.players: + raise GameError("bad_vote", "Den spiller er ikke med i runden.") + if member_id in self.audience: # publikum stemmer for sjov; tæller ikke + self.audience_votes[member_id] = target + self._changed() + return + if not self._in_round(member_id): + raise GameError("not_in_round", "Du er ikke med i denne runde.") + if target == member_id: + raise GameError("bad_vote", "Du kan ikke stemme på dig selv.") + self.votes[member_id] = target + self._changed() + self._maybe_finish_early(now) + + def _tally(self, votes: dict[str, str]) -> dict[str, int]: + counts: dict[str, int] = {} + for voter, target in votes.items(): + if target in self.players and (voter in self.players or voter in self.audience): + counts[target] = counts.get(target, 0) + 1 + return counts + + def _verdict(self, now: float): + counts = self._tally(self.votes) + top = max(counts.values(), default=0) + leaders = [pid for pid, n in counts.items() if n == top] + self.accused = leaders[0] if top and len(leaders) == 1 else None # lighed: ingen anklages + for pid, n in counts.items(): + self.players[pid].suspected += n + self.phase = "verdict" + self.deadline = now + VERDICT_SECONDS + self._emit("verdict", accused=self._name(self.accused), tie=self.accused is None and top > 0, + no_votes=top == 0, spy=self.accused in self.spies) + self._changed() + + # ── spionens sidste chance ───────────────────────────────────────────── + def spy_guess(self, member_id: str, text: str, now: float) -> None: + self._require_phase("guess") + if member_id != self.accused: + raise GameError("not_allowed", "Det er kun den anklagede spion, der gætter.") + guess = clean_text(text, "guess_invalid", "Gættet", MAX_GUESS) + self.guess = {"text": guess, "correct": same_word(guess, self.word)} + self._end_round(now) + + # ── resultat ──────────────────────────────────────────────────────────── + def _end_round(self, now: float): + caught = self.accused if self.accused in self.spies else None + guessed = bool(caught and self.guess and self.guess["correct"]) + deltas: dict[str, int] = {} + for pid in self.order: + p = self.players.get(pid) + if not p: + continue + if pid in self.spies: + d = POINTS_SPY_FREE if pid != caught else POINTS_SPY_GUESS if guessed else 0 + p.spy_points += d + else: + d = POINTS_VOTE if self.votes.get(pid) in self.spies else 0 + p.right_votes += d > 0 + p.score += d + deltas[pid] = d + counts = self._tally(self.votes) + self.result = {"spies": list(self.spies), "word": self.word, "caught": caught, "guess": self.guess, + "spy_won": caught is None or guessed, "deltas": deltas} + self.phase = "result" + self.deadline = now + RESULT_SECONDS + self._emit("result", spies=[self._name(s) for s in self.spies if s in self.players], word=self.word, + caught=bool(caught), guessed=guessed, accused_innocent=bool(self.accused and not caught), + tie=self.accused is None and bool(counts), last=self.round >= self.rounds) + self._changed() + + def _finish(self): + 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 kom først i lobbyen.""" + awards = [] + players = sorted(self.players.values(), key=lambda p: p.joined_order) + + def best(attr, kind, title, detail): + top = max(players, key=lambda p: getattr(p, attr), default=None) + if top and getattr(top, attr) > 0: + n = getattr(top, attr) + awards.append({"title": title, "player_id": top.id, "kind": kind, "detail": detail(n), "count": n}) + + best("spy_points", "liar", "Bedste løgner", lambda n: f"{n} point som spion") + best("right_votes", "sharp", "Skarpeste øje", lambda n: f"stemte rigtigt {n} gang{'e' if n != 1 else ''}") + best("suspected", "suspected", "Mest mistænkt", lambda n: f"{n} stemme{'r' if n != 1 else ''} i alt") + return awards + + # ── reaktioner ────────────────────────────────────────────────────────── + def react(self, member_id: str, emoji: str, now: float) -> dict | None: + p = self._member(member_id) + if not p or emoji not in self.reaction_set: + return None + bucket = [t for t in self._react_bucket.get(member_id, []) if now - t < 1] + if len(bucket) >= REACT_PER_SECOND: + return None + bucket.append(now) + self._react_bucket[member_id] = bucket + return {"emoji": emoji, "color": p.color, "name": p.name} + + # ── tid ───────────────────────────────────────────────────────────────── + def _all_done(self) -> bool: + if self.phase not in ("clue", "vote"): + return False + done = self.clues if self.phase == "clue" else self.votes + active = [pid for pid in self.order if pid in self.players and self.players[pid].connected] + return bool(active) and all(pid in done 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(f"{self.phase}_all_done") + self._advance(now) + + def skip(self, now: float) -> None: + if self.phase not in TIMED_PHASES: + raise GameError("wrong_phase", "Der er ikke noget at springe over.") + 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_round(now) + elif self.phase == "clue": + self._start_clues(now) + elif self.phase == "clues": + self.phase, self.deadline = "talk", now + TALK_SECONDS + self._emit("talk") + self._changed() + elif self.phase == "talk": + self.phase, self.deadline = "vote", now + VOTE_SECONDS + self._emit("vote") + self._changed() + elif self.phase == "vote": + self._verdict(now) + elif self.phase == "verdict": + if self.accused in self.spies and self.accused in self.players: + self.phase, self.deadline = "guess", now + GUESS_SECONDS + self._emit("guess", name=self._name(self.accused)) + self._changed() + else: + self._end_round(now) + elif self.phase == "guess": + self._end_round(now) # tiden løb ud: intet gæt + elif self.phase == "result": + if self.round >= self.rounds: + self._finish() + else: + self._begin_round(now) + + # ── snapshot ──────────────────────────────────────────────────────────── + def snapshot(self, now: float, viewer: Player | None = None, host: bool = False) -> dict: + """Tilstanden til én skærm. Ordet står kun i kortet til dem, der kender det (aldrig til spionen, publikum + eller storskærmen før resultatet), og hvem der er spion står intet sted før resultatet: ikke i spillerlisten, + ikke i rækkefølger eller tider.""" + vip = self.vip_id() + p_ph = self.phase + in_round = p_ph in ("clue", "clues", "talk", "vote", "verdict", "guess", "result") + done = self.clues if p_ph == "clue" else self.votes if p_ph == "vote" else {} + state = { + "game": self.game, + "code": self.code, + "phase": p_ph, + "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": {"number": self.round, "of": self.rounds} if in_round else None, + "players": [ + {"id": p.id, "name": p.name, "color": p.color, "score": p.score, "connected": p.connected, + "vip": p.id == vip, "done": p.id in done, "in_round": p.id in self.order} + for p in self.players.values() + ], + "audience_count": len(self.audience), + "fest": False, + "reactions": list(self.reaction_set), + } + if p_ph == "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 in_round: + state["theme"] = {"name": self.theme["name"], "icon": self.theme.get("icon", "")} + state["spies"] = 2 if len(self.order) >= TWO_SPIES_FROM else 1 # hvor mange, ikke hvem + if p_ph == "clues": # telefonerne kigger på storskærmen; den får stikordene og styrer tempoet + state["clues"] = {"started_at": int(self.clues_started_at * 1000), "intro_ms": CLUES_INTRO_MS, + "clue_ms": CLUE_MS, "items": self._clue_items() if host else []} + elif p_ph in ("talk", "vote", "verdict", "guess", "result"): + state["clues"] = {"items": self._clue_items()} + if p_ph in ("verdict", "guess", "result"): + counts, aud = self._tally(self.votes), self._tally(self.audience_votes) + state["votes"] = {"counts": [{"player_id": pid, "votes": counts.get(pid, 0)} for pid in self.order + if pid in self.players], + "audience": [{"player_id": pid, "votes": n} for pid, n in aud.items()], + "by": [{"voter": v, "target": t} for v, t in self.votes.items()] if p_ph == "result" else [], + "accused": self.accused, + # om den anklagede var spion, afsløres i slutningen af "verdict" (og ses i "guess") + "accused_is_spy": self.accused in self.spies if self.accused else False} + if p_ph == "result": + state["result"] = self.result + if p_ph == "final": + state["final"] = self.final_result + if host: + state["host"] = True + elif viewer: + card = None + if in_round and viewer.role == "player" and viewer.id in self.order: + spy = viewer.id in self.spies + card = {"spy": spy, "theme": self.theme["name"], "word": None if spy else self.word} + state["me"] = { + "id": viewer.id, "name": viewer.name, "role": viewer.role, "vip": viewer.id == vip, + "score": viewer.score, "card": card, + "clue": self.clues.get(viewer.id) if in_round else None, + "vote": (self.audience_votes if viewer.role == "audience" else self.votes).get(viewer.id) + if in_round else None, + "can_guess": p_ph == "guess" and viewer.id == self.accused, + } + return state + + def _clue_items(self) -> list[dict]: + return [{"player_id": pid, "text": self.clues.get(pid)} for pid in self.clue_order] diff --git a/spionen/tests/__init__.py b/spionen/tests/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/spionen/tests/test_engine.py b/spionen/tests/test_engine.py new file mode 100644 index 0000000..f077828 --- /dev/null +++ b/spionen/tests/test_engine.py @@ -0,0 +1,348 @@ +"""Spionens motor: roller, stikord, afstemning, spionens gæt, point og at hemmeligheden holder.""" + +import json +import random +from unittest import TestCase + +from fupogfakta.engine import GameError +from spionen import engine +from spionen.engine import Game, same_word + +THEMES = [ + {"id": "dyr", "name": "Dyr", "icon": "🐘", "words": ["kat", "pingvin", "kænguru", "isbjørn", "pindsvin"]}, + {"id": "koekken", "name": "Ting i køkkenet", "icon": "🍳", "words": ["pande", "gaffel", "ovn"]}, +] +NAMES = ["Asger", "Sus", "Bo", "Mette", "Kasper", "Line", "Jens", "Ida", "Ole", "Pia"] + + +def game(n=4, seed=1, rounds=4, themes=THEMES): + g = Game("TEST", themes, rng=random.Random(seed), rounds=rounds) + g.show_rules = False + ps = [g.join(NAMES[i]) for i in range(n)] + g.start(None, 0) + return g, ps + + +def play_clues(g, now=1): + """Alle skriver et stikord, der ikke røber ordet (spionerne et tilfældigt).""" + for i, pid in enumerate(g.order): + g.submit_clue(pid, f"stikord{chr(97 + i)}", now) + + +def to_vote(g, now=2): + if g.phase == "clue": + play_clues(g, now) + while g.phase in ("clues", "talk"): + g.skip(now) + assert g.phase == "vote" + + +class WordTests(TestCase): + def test_inflections_case_spaces_and_hyphens(self): + for form in ("KAT", "katten", "katte", "kattene", "kattens", "kat-ten", " Kat "): + self.assertTrue(same_word(form, "kat"), form) + for form in ("panden", "pander", "panderne", "pandens"): + self.assertTrue(same_word(form, "pande"), form) + self.assertTrue(same_word("is bjørn", "isbjørn")) + self.assertTrue(same_word("isbjørnen", "isbjørn")) + self.assertTrue(same_word("kænguruen", "kænguru")) + self.assertFalse(same_word("katteskål", "kat")) # sammensatte ord er nye ord + self.assertFalse(same_word("hund", "kat")) + + def test_small_typos_scaled_by_length(self): + self.assertTrue(same_word("pingvn", "pingvin")) # en fejl i et ord på syv + self.assertTrue(same_word("pindsvn", "pindsvin")) + self.assertTrue(same_word("pndsvn", "pindsvin")) # to fejl i et ord på otte + self.assertTrue(same_word("kroodile", "krokodille")) + self.assertFalse(same_word("pinsvni", "pingvin")) # tre fejl + self.assertFalse(same_word("kas", "kat")) # ingen stavefejl i korte ord + self.assertFalse(same_word("katten", "kat", typos=False) is False) # bøjning gælder uden stavefejl + self.assertFalse(same_word("pingvn", "pingvin", typos=False)) + + +class RoleTests(TestCase): + def test_one_spy_and_two_from_eight(self): + for n, spies in ((4, 1), (7, 1), (8, 2), (10, 2)): + g, _ = game(n) + self.assertEqual(len(g.spies), spies, n) + self.assertEqual(len(set(g.spies)), spies) + + def test_nobody_is_spy_twice_in_a_row(self): + for seed in range(30): + g, _ = game(4, seed=seed, rounds=5) + seen = [] + while g.phase != "final": + if g.phase == "clue": + seen.append(list(g.spies)) + g.skip(10) + for a, b in zip(seen, seen[1:]): + self.assertFalse(set(a) & set(b), (seed, seen)) + + def test_words_and_themes_vary(self): + g, _ = game(4, rounds=5) + words, themes = [], [] + while g.phase != "final": + if g.phase == "clue": + words.append(g.word) + themes.append(g.theme["id"]) + g.skip(10) + self.assertEqual(len(words), len(set(words))) + self.assertTrue(all(a != b for a, b in zip(themes, themes[1:]))) # to temaer: skiftevis + + +class ClueTests(TestCase): + def setUp(self): + self.g, self.ps = game(4) + self.g.word, self.g.theme = "kat", THEMES[0] + self.agent = next(pid for pid in self.g.order if pid not in self.g.spies) + self.spy = self.g.spies[0] + + def bad(self, pid, text, code): + with self.assertRaises(GameError) as e: + self.g.submit_clue(pid, text, 1) + self.assertEqual(e.exception.code, code) + + def test_word_inflections_and_typos_are_refused(self): + for text in ("kat", "Katten", "kattene"): + self.bad(self.agent, text, "clue_is_word") + self.g.word = "pingvin" + self.bad(self.agent, "pingvn", "clue_is_word") + + def test_theme_spaces_length_and_duplicates(self): + self.bad(self.agent, "dyr", "clue_is_theme") + self.bad(self.agent, "dyrene", "clue_is_theme") + self.bad(self.agent, "to ord", "clue_invalid") + self.bad(self.agent, "x" * 21, "clue_invalid") + self.bad(self.agent, "", "clue_invalid") + self.g.submit_clue(self.agent, "Mjav", 1) + other = next(pid for pid in self.g.order if pid not in (self.agent, self.spy)) + self.bad(other, "mjav", "clue_taken") + self.g.submit_clue(self.agent, "spinder", 1) # sit eget må man skifte + self.g.submit_clue(other, "mjav", 1) + + def test_the_spy_is_never_told_the_word(self): + self.g.submit_clue(self.spy, "kat", 1) # spionen ramte ordet: ingen afvisning, ingen lækage + self.assertEqual(self.g.clues[self.spy], "kat") + self.bad(self.spy, "dyr", "clue_is_theme") + + def test_audience_and_wrong_phase(self): + u = self.g.join("Sene") # kom efter start: publikum + self.assertEqual(u.role, "audience") + self.bad(u.id, "mjav", "not_in_round") + play_clues(self.g) + self.bad(self.agent, "mjav", "wrong_phase") + + +class VoteAndPointsTests(TestCase): + def round_to(self, n=4, seed=1): + g, ps = game(n, seed=seed) + to_vote(g) + return g, ps + + def test_no_self_vote_and_audience_counts_separately(self): + g, ps = self.round_to() + a = g.order[0] + with self.assertRaises(GameError): + g.vote(a, a, 3) + u = g.join("Sene") + g.vote(u.id, g.spies[0], 3) + self.assertEqual(g.audience_votes, {u.id: g.spies[0]}) + self.assertNotIn(u.id, g.votes) + + def test_majority_catches_the_spy_who_then_guesses(self): + g, ps = self.round_to() + spy = g.spies[0] + agents = [pid for pid in g.order if pid != spy] + for pid in agents: + g.vote(pid, spy, 3) + g.vote(spy, agents[0], 3) # alle har stemt: videre med det samme + self.assertEqual((g.phase, g.accused), ("verdict", spy)) + g.skip(4) + self.assertEqual(g.phase, "guess") + with self.assertRaises(GameError): + g.spy_guess(agents[0], g.word, 5) # kun spionen gætter + g.spy_guess(spy, g.word.upper() + "en", 5) + self.assertEqual(g.phase, "result") + self.assertTrue(g.result["guess"]["correct"] and g.result["spy_won"]) + self.assertEqual(g.players[spy].score, engine.POINTS_SPY_GUESS) + self.assertTrue(all(g.players[pid].score == engine.POINTS_VOTE for pid in agents)) + + def test_caught_spy_who_misses_gets_nothing(self): + g, ps = self.round_to() + spy = g.spies[0] + for pid in g.order: + g.vote(pid, spy if pid != spy else next(x for x in g.order if x != spy), 3) + g.skip(4) + g.spy_guess(spy, "noget helt andet", 5) + self.assertFalse(g.result["spy_won"]) + self.assertEqual(g.players[spy].score, 0) + + def test_tie_lets_the_spy_go(self): + g, ps = self.round_to() + spy = g.spies[0] + agents = [pid for pid in g.order if pid != spy] + g.vote(agents[0], spy, 3) + g.vote(agents[1], agents[2], 3) # 1-1: lighed + g.skip(3) + self.assertIsNone(g.accused) + g.skip(4) # ingen anklaget: intet gæt + self.assertEqual(g.phase, "result") + self.assertEqual(g.players[spy].score, engine.POINTS_SPY_FREE) + self.assertEqual(g.players[agents[0]].score, engine.POINTS_VOTE) # stemte stadig rigtigt + self.assertEqual(g.players[agents[1]].score, 0) + + def test_no_votes_at_all(self): + g, ps = self.round_to() + g.skip(3) + g.skip(4) + self.assertEqual((g.phase, g.accused), ("result", None)) + self.assertEqual(g.players[g.spies[0]].score, engine.POINTS_SPY_FREE) + + def test_innocent_accused(self): + g, ps = self.round_to() + spy = g.spies[0] + agents = [pid for pid in g.order if pid != spy] + for pid in g.order: + g.vote(pid, agents[0] if pid != agents[0] else agents[1], 3) + g.skip(4) + self.assertEqual(g.phase, "result") # ikke spion: intet gæt + self.assertEqual(g.players[spy].score, engine.POINTS_SPY_FREE) + self.assertEqual(g.result["caught"], None) + + def test_two_spies(self): + """Den anklagede spion er fanget og gætter; den anden slap. En stemme på en af dem er rigtig.""" + g, ps = self.round_to(8) + s1, s2 = g.spies + agents = [pid for pid in g.order if pid not in g.spies] + for i, pid in enumerate(agents): + g.vote(pid, s1 if i < 4 else s2, 3) + g.vote(s1, agents[0], 3) + g.vote(s2, agents[0], 3) + self.assertEqual(g.accused, s1) + g.skip(4) + self.assertEqual(g.phase, "guess") + g.skip(5) # tiden løb ud + self.assertEqual(g.players[s1].score, 0) + self.assertEqual(g.players[s2].score, engine.POINTS_SPY_FREE) + self.assertTrue(all(g.players[pid].score == engine.POINTS_VOTE for pid in agents)) + + def test_whole_game_ends_with_podium_and_awards(self): + g, ps = game(5, rounds=3) + rounds = 0 + while g.phase != "final": + if g.phase == "clue": + rounds += 1 + if g.phase == "vote": + spy = g.spies[0] + for pid in g.order: + g.vote(pid, spy if pid != spy else next(x for x in g.order if x != spy), 3) + continue + g.skip(10) + self.assertEqual(rounds, 3) + fin = g.final_result + self.assertEqual(len(fin["ranking"]), 5) + self.assertIn("sharp", [a["kind"] for a in fin["awards"]]) + self.assertIn("suspected", [a["kind"] for a in fin["awards"]]) + + +class NeverHangTests(TestCase): + def test_missing_clues_and_votes_just_time_out(self): + g, _ = game(4) + for phase in ("clue", "clues", "talk", "vote", "verdict"): + self.assertEqual(g.phase, phase) + self.assertTrue(g.tick(g.deadline)) + self.assertEqual(g.phase, "result") + self.assertIn(None, [c["text"] for c in g._clue_items()]) # ingen stikord: står tomt + + def test_disconnected_players_do_not_hold_the_round(self): + g2, _ = game(4) + g2.set_connected(g2.order[3], False, 1) + for pid in g2.order[:3]: + g2.submit_clue(pid, f"ord{pid}", 1) + self.assertEqual(g2.phase, "clues") # de forbundne er færdige + + def test_accused_spy_leaves_during_guess(self): + g, _ = game(4) + to_vote(g) + spy = g.spies[0] + for pid in g.order: + g.vote(pid, spy if pid != spy else next(x for x in g.order if x != spy), 3) + g.skip(4) + self.assertEqual(g.phase, "guess") + g.kick(spy, 5) + self.assertEqual(g.phase, "result") + + def test_too_few_players_ends_the_game(self): + g, _ = game(4) + g.kick(g.order[0], 1) + while g.phase not in ("final", "lobby"): + g.skip(10) + self.assertEqual(g.phase, "final") # runden blev spillet færdig; ingen ny med tre + + def test_everyone_leaves(self): + g, _ = game(4) + for pid in list(g.players): + g.kick(pid, 1) + self.assertEqual(g.phase, "lobby") + + +class SecrecyTests(TestCase): + """Ordet kun til dem, der kender det; spionen ingen steder før resultatet. Hvert snapshot i hver fase læses.""" + + def snaps(self, g): + out = {"host": g.snapshot(5, host=True)} + for p in list(g.players.values()) + list(g.audience.values()): + out[p.id] = g.snapshot(5, viewer=p) + return out + + def test_every_snapshot_in_every_phase(self): + g, ps = game(8, seed=4) + u = g.join("Sene") + checked = set() + while g.phase != "final": + snaps = self.snaps(g) + word, spies = g.word, set(g.spies) + for who, st in snaps.items(): + text = json.dumps(st, ensure_ascii=False).lower() + with self.subTest(phase=g.phase, who=who): + if g.phase == "result": + self.assertIn(word, text) + continue + self.assertNotIn("spies\": [", text) # aldrig en liste over spionerne + knows = who in g.order and who not in spies + if not knows: # spionen, publikum og storskærmen + self.assertNotIn(f"\"{word}\"", text) + self.assertNotIn(word, json.dumps(st.get("me", {}), ensure_ascii=False).lower()) + else: + self.assertEqual(st["me"]["card"]["word"], word) + card = (st.get("me") or {}).get("card") + if card: + self.assertEqual(card["spy"], who in spies) # kun ens egen rolle + accused_is_spy = (st.get("votes") or {}).get("accused_is_spy") + if accused_is_spy: + self.assertIn(g.phase, ("verdict", "guess")) + self.assertIn(st["votes"]["accused"], spies) + checked.add(g.phase) + if g.phase == "clue": + play_clues(g) + continue + if g.phase == "vote": + g.vote(u.id, g.spies[0], 3) # publikum stemmer for sjov + for pid in g.order: + g.vote(pid, g.spies[0] if pid != g.spies[0] else g.spies[1], 3) + continue + g.skip(10) + self.assertLessEqual({"clue", "clues", "talk", "vote", "verdict", "guess", "result"}, checked) + + def test_clue_order_is_shuffled_and_phones_wait_for_the_tv(self): + g, ps = game(6, seed=2) + play_clues(g) + self.assertEqual(g.phase, "clues") + orders = set() + for seed in range(8): + h, _ = game(6, seed=seed) + play_clues(h) + orders.add(tuple(h.clue_order.index(pid) for pid in h.order)) + self.assertGreater(len(orders), 1) + self.assertEqual(g.snapshot(5, viewer=ps[0])["clues"]["items"], []) # telefonerne kigger på TV'et + self.assertEqual(len(g.snapshot(5, host=True)["clues"]["items"]), 6)