agwandClaude Opus 5.5
2980a7d7b0
Tegn og Gæts _narrate_rules (et regelkort pr. replik, i replikkens længde) flyttes uændret til BaseRoom, så Spionen kan bruge den samme. Fup og Fakta har sin egen og er uberørt. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
284 lines
13 KiB
Python
284 lines
13 KiB
Python
"""Rum til Tegn og Gæt.
|
|
|
|
Deler forbindelser, timer, udsendelse og speakerens kø med Fup og Fakta
|
|
(`rooms.BaseRoom`) og ligger i samme kodeliste som de andre rum. Speakeren
|
|
siger kun noget, når der er stemme (ElevenLabs); ellers kører alt på motorens
|
|
egne timere.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import json
|
|
import logging
|
|
import time
|
|
from pathlib import Path
|
|
|
|
from fupogfakta.engine import GameError
|
|
from tegnoggaet.engine import EMPTY_GUESS, MAX_PLAYERS, MIN_PLAYERS, RULES, RULES_STEP_MS, Game
|
|
from party.themes import load_themes
|
|
|
|
from . import rooms, tts
|
|
from .narrator import Narrator, generic_texts, one_of, untag
|
|
|
|
log = logging.getLogger(__name__)
|
|
|
|
NARRATOR_DIR = Path(__file__).resolve().parents[1] / "tegnoggaet" / "narrator"
|
|
LINES = json.loads((NARRATOR_DIR / "lines.json").read_text(encoding="utf-8"))
|
|
GENERIC = json.loads((NARRATOR_DIR / "generic.json").read_text(encoding="utf-8")) # se party.narrator
|
|
BEAT_PAD_MS = 700 # luft efter speakerens replik i et trin af afsløringen
|
|
HURRY_BEFORE = 10 # "ti sekunder tilbage"
|
|
# Replikker uden navne eller spillertekst, som rummet siger alene: dem kan tts_warm lave på forhånd.
|
|
# reveal_intro er udeladt med vilje: den siges kun sammen med første kædes intro, som har et navn.
|
|
FIXED_LINES = ("lobby_welcome", "draw_start", "guess_start", "step_hurry", "step_all_done", "play_again")
|
|
THEME_LINES = ("theme_chosen", "theme_timeout", "word_start") # {tema}: ét sæt pr. tema
|
|
COUNT_LINES = ("game_start",) # {antal_spillere}: ét sæt pr. antal spillere
|
|
AWARD_LINES = {"drawing": "award_best_drawing", "guess": "award_best_guess", "survived": "award_survived",
|
|
"transformed": "award_transformed"}
|
|
|
|
|
|
class TegnRoom(rooms.BaseRoom):
|
|
kind = "tegn"
|
|
|
|
def __init__(self, code: str, game: Game, host_username: str):
|
|
super().__init__(code, game, host_username)
|
|
self.narrator = Narrator(lines=LINES, generic=GENERIC)
|
|
self._last_join_line = 0.0
|
|
self._lobby_waiting_said = False
|
|
self._beat_timers: list[asyncio.TimerHandle] = []
|
|
|
|
def changed(self):
|
|
super().changed()
|
|
self._schedule_beats()
|
|
|
|
def _schedule_beats(self):
|
|
"""Telefonerne får kun de trin, storskærmen har vist: et nyt snapshot hver gang et trin begynder."""
|
|
for h in self._beat_timers:
|
|
h.cancel()
|
|
self._beat_timers = []
|
|
if self.closed:
|
|
return
|
|
g = self.game
|
|
if g.phase != "reveal":
|
|
return
|
|
loop, now = asyncio.get_running_loop(), time.time()
|
|
for t in g.beat_starts():
|
|
if t > now:
|
|
self._beat_timers.append(loop.call_later(t - now + 0.02, self.broadcast))
|
|
|
|
def close(self):
|
|
super().close()
|
|
for h in self._beat_timers:
|
|
h.cancel()
|
|
|
|
# ── handlinger ───────────────────────────────────────────────────────────
|
|
async def host_action(self, conn, action: str, content: dict, now: float) -> bool:
|
|
if action != "choose_theme":
|
|
return False
|
|
self.game.choose_theme(None, str(content["theme_id"]), now)
|
|
return True
|
|
|
|
async def play_action(self, conn, member_id: str, action: str, content: dict, now: float):
|
|
g = self.game
|
|
if action == "start":
|
|
g.start(member_id, now)
|
|
elif action == "choose_theme":
|
|
g.choose_theme(member_id, str(content["theme_id"]), now)
|
|
elif action == "submit_word":
|
|
g.submit_word(member_id, str(content.get("text", "")), now)
|
|
elif action == "submit_drawing":
|
|
g.submit_drawing(member_id, content.get("strokes"), now)
|
|
elif action == "submit_guess":
|
|
g.submit_guess(member_id, str(content.get("text", "")), now)
|
|
elif action == "draft":
|
|
# Kladder sendes automatisk undervejs; en forældet eller ugyldig kladde ignoreres stille.
|
|
try:
|
|
g.save_draft(member_id, content.get("strokes") if g.phase == "draw" else str(content.get("text", "")))
|
|
except GameError:
|
|
pass
|
|
elif action == "suggest":
|
|
await conn.send_json({"type": "suggestion", "text": g.suggest(member_id)})
|
|
elif action == "like":
|
|
g.like(member_id, str(content["entry_id"]), now)
|
|
elif action == "react":
|
|
reaction = g.react(member_id, str(content.get("emoji", "")), now)
|
|
if reaction:
|
|
await self.forward_reaction(reaction)
|
|
elif action == "play_again":
|
|
g.play_again(member_id, now)
|
|
else:
|
|
await conn.send_error("bad_request", "Ukendt handling.")
|
|
|
|
async def before_host_state(self, conn):
|
|
"""Tegningerne fylder for meget til at følge med hvert snapshot: storskærmen får dem én gang pr. afsløring."""
|
|
g = self.game
|
|
if g.phase == "reveal" and getattr(conn, "tegn_drawings", None) != g.reveal_id:
|
|
conn.tegn_drawings = g.reveal_id
|
|
await conn.send_json({"type": "drawings", "reveal_id": g.reveal_id, "drawings": g.drawings()})
|
|
|
|
# ── speakeren ────────────────────────────────────────────────────────────
|
|
def host_opened(self):
|
|
if tts.enabled() and not self._welcomed and self.game.phase == "lobby" and not self.game.players:
|
|
self._welcomed = True
|
|
self.say(self.narrator.line("lobby_welcome"))
|
|
|
|
def handle_events(self):
|
|
events = self.game.drain_events()
|
|
if not tts.enabled():
|
|
return
|
|
skip_start = any(e["type"] == "play_again" for e in events)
|
|
for ev in events:
|
|
t = ev["type"]
|
|
if t == "game_start" and skip_start:
|
|
continue
|
|
if t == "reveal":
|
|
asyncio.get_running_loop().create_task(self._narrate_reveal())
|
|
continue
|
|
if t == "rules":
|
|
asyncio.get_running_loop().create_task(self._narrate_rules())
|
|
continue
|
|
if t == "join":
|
|
now = time.time()
|
|
if now - self._last_join_line < 2.5:
|
|
continue
|
|
self._last_join_line = now
|
|
self._maybe_schedule_lobby_waiting()
|
|
for text in self.lines_for(ev):
|
|
self.say(text)
|
|
if t == "step":
|
|
self._hurry()
|
|
|
|
def lines_for(self, ev: dict) -> list[str]:
|
|
g, t, line = self.game, ev["type"], self.narrator.line
|
|
if t == "join" and ev["role"] == "player" and g.phase == "lobby":
|
|
return [line("lobby_join", navn=ev["name"])]
|
|
if t == "game_start":
|
|
return [line("game_start", antal_spillere=ev["players"])]
|
|
if t == "theme":
|
|
return [line("theme_choose", vip=ev["vip"])]
|
|
if t == "theme_chosen":
|
|
return [line("theme_timeout" if ev["timeout"] else "theme_chosen", tema=ev["theme"])]
|
|
if t == "step":
|
|
if ev["step"] == "word":
|
|
return [line("word_start", tema=ev["theme"])]
|
|
return [line(f"{ev['step']}_start")]
|
|
if t in ("step_all_done", "play_again"):
|
|
return [line(t)]
|
|
if t == "final":
|
|
out = [line("winner", vinder=ev["winner"], point=ev["points"])]
|
|
return out + [line(AWARD_LINES[a["kind"]], navn=a["name"]) for a in ev["awards"]]
|
|
return []
|
|
|
|
def _hurry(self):
|
|
g = self.game
|
|
ident = g.phase_key()
|
|
delay = (g.deadline or 0) - HURRY_BEFORE - time.time()
|
|
if delay <= 1:
|
|
return
|
|
|
|
def fire():
|
|
if g.phase_key() == ident and not g._all_done():
|
|
self.say(self.narrator.line("step_hurry"))
|
|
self._side_timers.append(asyncio.get_running_loop().call_later(delay, fire))
|
|
self._side_timers = [h for h in self._side_timers if not h.cancelled()][-20:]
|
|
|
|
def _maybe_schedule_lobby_waiting(self):
|
|
g = self.game
|
|
if self._lobby_waiting_said or g.phase != "lobby" or len(g.players) < 3:
|
|
return
|
|
self._lobby_waiting_said = True
|
|
|
|
def fire():
|
|
vip = g.players.get(g.vip_id() or "")
|
|
if g.phase == "lobby" and vip:
|
|
self.say(self.narrator.line("lobby_waiting", vip=vip.name))
|
|
self._side_timers.append(asyncio.get_running_loop().call_later(rooms.LOBBY_WAITING_AFTER, fire))
|
|
|
|
def reveal_texts(self) -> list[str | None]:
|
|
"""Speakerens replik pr. trin i afsløringen: kædens intro og slutningen (første ord mod sidste gæt)."""
|
|
g = self.game
|
|
names = {p.id: p.name for p in g.players.values()}
|
|
texts = []
|
|
for i, b in enumerate(g.plan):
|
|
chain = g.chains[b["chain"]]
|
|
if b["entry"] == -1:
|
|
intro = self.narrator.line("chain_intro", navn=names.get(g.order[b["chain"]], ""))
|
|
if i == 0: # afsløringens intro og første kæde siges i ét; afløseren er de to uden navn
|
|
ri = self.narrator.line("reveal_intro")
|
|
intro = one_of(f"{ri} {intro}", first_chain_generics(self.narrator, ri), self.narrator.rng)
|
|
texts.append(intro)
|
|
elif b["entry"] == len(chain):
|
|
first, last = chain[0]["text"], g._last_guess(chain) or EMPTY_GUESS
|
|
if g.survived[b["chain"]]:
|
|
texts.append(self.narrator.line("chain_survived", ord=first))
|
|
else:
|
|
texts.append(self.narrator.line("chain_changed", ord=first,
|
|
gaet="ingenting" if last == EMPTY_GUESS else last))
|
|
else:
|
|
texts.append(None)
|
|
return texts
|
|
|
|
async def _narrate_reveal(self):
|
|
"""Lav lyd til kædernes intro og slutning, og lad afsløringens tempo følge replikkerne."""
|
|
g = self.game
|
|
rid = g.reveal_id
|
|
g.hold_reveal(time.time(), max_wait=rooms.REVEAL_AUDIO_WAIT + 2 + max(0.0, self.speech_end - time.time()))
|
|
self.changed()
|
|
texts = self.reveal_texts()
|
|
jobs = [self._synth(t) if t else asyncio.sleep(0, None) for t in texts]
|
|
try:
|
|
results = await asyncio.wait_for(asyncio.gather(*jobs), timeout=rooms.REVEAL_AUDIO_WAIT)
|
|
except asyncio.TimeoutError:
|
|
results = [None] * len(jobs)
|
|
wait = min(self.speech_end - time.time(), rooms.MAX_HOLD)
|
|
if wait > 0:
|
|
await asyncio.sleep(wait)
|
|
if g.phase != "reveal" or g.reveal_id != rid or g.reveal_started_at is not None:
|
|
return # værten sprang over, eller tiden løb ud
|
|
steps_ms = []
|
|
for b, text, audio in zip(g.plan, texts, results):
|
|
b["audio"] = {"url": audio["url"], "ms": audio["ms"], "text": untag(text)} if audio else None
|
|
steps_ms.append(max(b["ms"], audio["ms"] + BEAT_PAD_MS) if audio else b["ms"])
|
|
g.start_reveal_timeline(time.time(), steps_ms)
|
|
self.changed()
|
|
|
|
|
|
|
|
def warm_texts(themes: list[dict] | None = None) -> list[str]:
|
|
"""Teksterne præcis som rummet sender dem til tts.synth (samme stemme, samme cache-nøgle): de faste
|
|
replikker, reglerne, {tema}-replikkerne for alle temaer og spillets start for 3-10 spillere."""
|
|
themes = load_themes() if themes is None else themes
|
|
n = Narrator(lines=LINES)
|
|
out = [t for ev in FIXED_LINES for t in n.variants(ev)]
|
|
out += [r["say"] for r in RULES]
|
|
out += [t for ev in THEME_LINES for th in themes for t in n.variants(ev, tema=th["name"])]
|
|
out += [t for ev in COUNT_LINES for k in range(MIN_PLAYERS, MAX_PLAYERS + 1) for t in n.variants(ev, antal_spillere=k)]
|
|
return list(dict.fromkeys(out))
|
|
|
|
|
|
def warm_generic() -> list[str]:
|
|
"""Tegn og Gæts generiske replikker (standardstemmen), også afsløringens intro med første kædes."""
|
|
n = Narrator(lines=LINES, generic=GENERIC)
|
|
out = [t for t, _ in generic_texts(GENERIC, None)]
|
|
out += [t for ri in n.variants("reveal_intro") for t in first_chain_generics(n, ri)]
|
|
return list(dict.fromkeys(out))
|
|
|
|
|
|
def first_chain_generics(narrator: Narrator, reveal_intro: str) -> list[str]:
|
|
"""Afsløringens intro efterfulgt af en generisk intro til første kæde, som rummet siger dem."""
|
|
return [f"{reveal_intro} {v}" for v in narrator.generic_variants("chain_intro_first")]
|
|
|
|
|
|
def create_tegn_room(host_username: str, public_url: str, code: str | None = None,
|
|
show_rules: bool = True) -> TegnRoom:
|
|
"""code: et kørende rums kode = skift spil i det rum (se rooms.add_room)."""
|
|
rooms.cleanup()
|
|
code = code or rooms.new_code()
|
|
game = Game(code, load_themes(), join_url=f"{public_url}/?code={code}")
|
|
if not show_rules:
|
|
game.show_rules = False
|
|
room = TegnRoom(code, game, host_username)
|
|
rooms.add_room(room)
|
|
log.info("Tegn og Gæt-rum %s oprettet af %s", code, host_username)
|
|
return room
|