tts_warm: forvarm også Tegn og Gæt (--only, --dry-run)
Tegn og Gæts faste replikker, reglerne, {tema}-replikkerne for alle
temaer og spillets start for 3-10 spillere laves først, med den
almindelige speaker; derefter Fup og Fakta som før. Teksterne kommer
fra samme Narrator-kode, som rummet bruger, så cache-nøglen er den
samme. --only fof|tegn vælger ét spil, og --dry-run tæller klip og
tegn uden at kalde ElevenLabs (også uden nøgle).
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
@@ -8,6 +8,7 @@ from fupogfakta.engine import RULES, RULES_FEST
|
||||
from fupogfakta.models import Category, Question
|
||||
from party import tts
|
||||
from party.narrator import LINES, TEASES, TEASES_FEST, Narrator, speakable_prompt
|
||||
from party.tegn import warm_texts as tegn_texts
|
||||
|
||||
|
||||
def _expand_categories(lines: list[str], categories: list[str]) -> list[str]:
|
||||
@@ -22,32 +23,50 @@ def _expand_categories(lines: list[str], categories: list[str]) -> list[str]:
|
||||
class Command(BaseCommand):
|
||||
help = ("Genererer speakerens replikker på forhånd. Uden --budget: kun de faste replikker (gratis, når de findes). "
|
||||
"Med --budget N: derefter op til N tegn af spørgsmålenes oplæsning og forklaring (de mindst brugte først, "
|
||||
"dvs. dem der kommer næste gang) og kategori-replikkerne. Køres om natten, så kvoten fordeles.")
|
||||
"dvs. dem der kommer næste gang) og kategori-replikkerne. Køres om natten, så kvoten fordeles. "
|
||||
"Tegn og Gæts faste replikker (inkl. ét sæt pr. tema) laves først.")
|
||||
|
||||
def add_arguments(self, parser):
|
||||
parser.add_argument("--budget", type=int, default=0, help="højst så mange nye tegn ud over de faste replikker")
|
||||
parser.add_argument("--reserve", type=int, default=40000,
|
||||
help="rør ikke de sidste så mange credits i perioden (de skal bruges til spil)")
|
||||
parser.add_argument("--only", choices=["fof", "tegn"], help="kun ét spil (fof = Fup og Fakta, tegn = Tegn og Gæt)")
|
||||
parser.add_argument("--dry-run", action="store_true",
|
||||
help="tæl klip og tegn uden at kalde ElevenLabs (virker også uden nøgle)")
|
||||
|
||||
def handle(self, *args, budget, reserve, **opts):
|
||||
if not tts.enabled():
|
||||
def handle(self, *args, budget, reserve, only=None, dry_run=False, **opts):
|
||||
if not dry_run and not tts.enabled():
|
||||
self.stdout.write("ElevenLabs er ikke sat op (ELEVENLABS_API_KEY/ELEVENLABS_VOICE_ID). Springer over.")
|
||||
return
|
||||
static = [line for k, variants in LINES.items() if not k.endswith("_fest")
|
||||
for line in variants if not re.search(r"\{\w+\}", line)]
|
||||
static += TEASES
|
||||
static += [r["say"] for r in RULES]
|
||||
ok = sum(1 for line in static if tts.synth_sync(line))
|
||||
self.stdout.write(self.style.SUCCESS(f"{ok} af {len(static)} faste replikker ligger i cachen."))
|
||||
fest = settings.FEST_VOICE_ID or None
|
||||
if fest:
|
||||
# fest-speakeren siger "<begivenhed>_fest", og den almindelige replik, hvor der ikke er en fest-udgave
|
||||
keys = [k for k in LINES if k.endswith("_fest") or k + "_fest" not in LINES]
|
||||
fest_static = [line for k in keys for line in LINES[k] if not re.search(r"\{\w+\}", line)]
|
||||
fest_static += TEASES_FEST + [r["say"] for r in RULES_FEST]
|
||||
ok = sum(1 for line in fest_static if tts.synth_sync(line, fest))
|
||||
self.stdout.write(self.style.SUCCESS(f"{ok} af {len(fest_static)} faste fest-replikker ligger i cachen."))
|
||||
if budget <= 0:
|
||||
# (navn, tekster, stemme); Tegn og Gæt først, da de er få. Tegn og Gæt bruger den almindelige speaker.
|
||||
groups = []
|
||||
if only != "fof":
|
||||
groups.append(("faste Tegn og Gæt-replikker", tegn_texts(), None))
|
||||
if only != "tegn":
|
||||
static = [line for k, variants in LINES.items() if not k.endswith("_fest")
|
||||
for line in variants if not re.search(r"\{\w+\}", line)]
|
||||
static += TEASES
|
||||
static += [r["say"] for r in RULES]
|
||||
groups.append(("faste replikker", static, None))
|
||||
if fest:
|
||||
# fest-speakeren siger "<begivenhed>_fest", og den almindelige replik, hvor der ikke er en fest-udgave
|
||||
keys = [k for k in LINES if k.endswith("_fest") or k + "_fest" not in LINES]
|
||||
fest_static = [line for k in keys for line in LINES[k] if not re.search(r"\{\w+\}", line)]
|
||||
fest_static += TEASES_FEST + [r["say"] for r in RULES_FEST]
|
||||
groups.append(("faste fest-replikker", fest_static, fest))
|
||||
if dry_run:
|
||||
for name, texts, voice in groups:
|
||||
missing = [t for t in texts if not tts.is_cached(t, voice)]
|
||||
self.stdout.write(f"{name}: {len(texts)} klip, {sum(map(len, texts))} tegn; "
|
||||
f"mangler i cachen: {len(missing)} klip, {sum(map(len, missing))} tegn.")
|
||||
if budget > 0 and only != "tegn":
|
||||
self.stdout.write("Spørgsmål og kategori-replikker (--budget) tælles ikke i --dry-run.")
|
||||
return
|
||||
for name, texts, voice in groups:
|
||||
ok = sum(1 for line in texts if tts.synth_sync(line, voice))
|
||||
self.stdout.write(self.style.SUCCESS(f"{ok} af {len(texts)} {name} ligger i cachen."))
|
||||
if budget <= 0 or only == "tegn":
|
||||
return
|
||||
left = tts.credits_left()
|
||||
if left is not None:
|
||||
|
||||
+14
-4
@@ -97,9 +97,16 @@ class Narrator:
|
||||
self.fest = fest # fest: replikken "<begivenhed>_fest" bruges, når den findes
|
||||
self._recent: dict[str, list[int]] = {}
|
||||
|
||||
def _event(self, event: str) -> str:
|
||||
return event + "_fest" if self.fest and self.lines.get(event + "_fest") else event
|
||||
|
||||
@staticmethod
|
||||
def fill(variant: str, values: dict) -> str:
|
||||
filled = {k: (number_words(v) if isinstance(v, int) else safe(v)) for k, v in values.items()}
|
||||
return capitalize(re.sub(r"\{(\w+)\}", lambda m: filled.get(m.group(1), ""), variant))
|
||||
|
||||
def line(self, event: str, **values) -> str | None:
|
||||
if self.fest and self.lines.get(event + "_fest"):
|
||||
event += "_fest"
|
||||
event = self._event(event)
|
||||
variants = self.lines.get(event)
|
||||
if not variants:
|
||||
return None
|
||||
@@ -108,8 +115,11 @@ class Narrator:
|
||||
i = self.rng.choice(fresh)
|
||||
recent.append(i)
|
||||
del recent[:-max(1, len(variants) // 2)] # undgå gentagelser i den seneste halvdel
|
||||
filled = {k: (number_words(v) if isinstance(v, int) else safe(v)) for k, v in values.items()}
|
||||
return capitalize(re.sub(r"\{(\w+)\}", lambda m: filled.get(m.group(1), ""), variants[i]))
|
||||
return self.fill(variants[i], values)
|
||||
|
||||
def variants(self, event: str, **values) -> list[str]:
|
||||
"""Alle udgaver af replikken, som line() ville sige dem (til forvarmning af lyden)."""
|
||||
return [self.fill(v, values) for v in self.lines.get(self._event(event), [])]
|
||||
|
||||
# ── afsløringen ──────────────────────────────────────────────────────────
|
||||
def reveal_step(self, step: dict, names: dict[str, str], n_players: int, comment: str | None = None) -> str | None:
|
||||
|
||||
+18
-1
@@ -15,7 +15,7 @@ import time
|
||||
from pathlib import Path
|
||||
|
||||
from fupogfakta.engine import GameError
|
||||
from tegnoggaet.engine import EMPTY_GUESS, RULES_STEP_MS, Game
|
||||
from tegnoggaet.engine import EMPTY_GUESS, MAX_PLAYERS, MIN_PLAYERS, RULES, RULES_STEP_MS, Game
|
||||
from tegnoggaet.themes import load_themes
|
||||
|
||||
from . import rooms, tts
|
||||
@@ -27,6 +27,11 @@ LINES = json.loads((Path(__file__).resolve().parents[1] / "tegnoggaet" / "narrat
|
||||
.read_text(encoding="utf-8"))
|
||||
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"}
|
||||
|
||||
@@ -252,6 +257,18 @@ class TegnRoom(rooms.BaseRoom):
|
||||
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 create_tegn_room(host_username: str, public_url: str) -> TegnRoom:
|
||||
rooms.cleanup()
|
||||
code = rooms.new_code()
|
||||
|
||||
@@ -0,0 +1,178 @@
|
||||
"""tts_warm forvarmer Tegn og Gæt: de forvarmede tekster skal være præcis dem, rummet sender til tts.synth
|
||||
(samme tekst og stemme giver samme cache-nøgle i party/tts.py). ElevenLabs kaldes aldrig."""
|
||||
|
||||
import asyncio
|
||||
import io
|
||||
import random
|
||||
import re
|
||||
import time
|
||||
from unittest import mock
|
||||
|
||||
from django.core.management import call_command
|
||||
from django.test import SimpleTestCase, override_settings
|
||||
|
||||
from fupogfakta.engine import RULES as FOF_RULES
|
||||
from party import tts
|
||||
from tegnoggaet.engine import RULES as TEGN_RULES
|
||||
from party.narrator import LINES as FOF_LINES, TEASES, Narrator
|
||||
from party.tegn import COUNT_LINES, FIXED_LINES, LINES, THEME_LINES, TegnRoom, warm_texts
|
||||
from tegnoggaet.engine import Game
|
||||
from tegnoggaet.themes import load_themes
|
||||
|
||||
STROKE = {"color": "#1d1b2e", "width": 0.01, "points": [[0.1, 0.1], [0.5, 0.5]]}
|
||||
NAMES = ("Qxa", "Zyb", "Wvc") # spillernavne og -ord, der ikke findes i replikkerne
|
||||
WORDS = ("qqord", "zzgaet")
|
||||
|
||||
|
||||
def fof_static():
|
||||
"""Fup og Fakta's faste replikker, som tts_warm har lavet dem hidtil."""
|
||||
out = [line for k, variants in FOF_LINES.items() if not k.endswith("_fest")
|
||||
for line in variants if not re.search(r"\{\w+\}", line)]
|
||||
return out + TEASES + [r["say"] for r in FOF_RULES]
|
||||
|
||||
|
||||
class TegnWarmTextTests(SimpleTestCase):
|
||||
def test_variants_are_exactly_what_line_says(self):
|
||||
n = Narrator(lines=LINES, rng=random.Random(1))
|
||||
for event in FIXED_LINES + THEME_LINES + COUNT_LINES:
|
||||
values = {"tema": "Ting i køkkenet"} if event in THEME_LINES else {"antal_spillere": 7} if event in COUNT_LINES else {}
|
||||
with self.subTest(event=event):
|
||||
said = {n.line(event, **values) for _ in range(300)}
|
||||
self.assertEqual(said, set(n.variants(event, **values)))
|
||||
|
||||
def test_no_names_or_placeholders_and_every_theme(self):
|
||||
texts = warm_texts()
|
||||
self.assertEqual(len(texts), len(set(texts)))
|
||||
self.assertFalse([t for t in texts if "{" in t or "}" in t])
|
||||
for theme in load_themes():
|
||||
self.assertTrue(any(theme["name"] in t for t in texts), theme["name"])
|
||||
self.assertFalse(set(texts) & set(Narrator(lines=LINES).variants("reveal_intro")),
|
||||
"reveal_intro siges kun sammen med et navn")
|
||||
|
||||
def test_the_room_says_warmed_texts_with_the_default_voice(self):
|
||||
"""Et helt spil (to gange, så både valgt tema og timeout kommer med) med en falsk stemme: alt, rummet
|
||||
siger uden navne eller spillertekst, ligger blandt de forvarmede tekster, med samme stemme (None)."""
|
||||
said: list[tuple[str, str | None]] = []
|
||||
|
||||
async def fake_synth(text, voice=None):
|
||||
said.append((text, voice))
|
||||
return {"url": "/tts/x.mp3", "ms": 10}
|
||||
|
||||
async def play(seed):
|
||||
g = Game("TEST", load_themes(), rng=random.Random(seed))
|
||||
g.show_rules = True
|
||||
room = TegnRoom("TEST", g, "agw")
|
||||
room.narrator.rng = random.Random(seed)
|
||||
|
||||
async def step(action=None):
|
||||
if action:
|
||||
action()
|
||||
room.handle_events()
|
||||
await asyncio.sleep(0.02) # lad stemme-opgaverne køre
|
||||
|
||||
now = time.time()
|
||||
await step(room.host_opened)
|
||||
ps = [g.join(n) for n in NAMES]
|
||||
await step()
|
||||
for game_no in range(2):
|
||||
if game_no == 0:
|
||||
await step(lambda: g.start(None, now))
|
||||
for _ in range(100): # reglerne: lyden laves først
|
||||
if g.rules_audio is not None:
|
||||
break
|
||||
await asyncio.sleep(0.01)
|
||||
await step(lambda: g.tick(time.time() + 3600))
|
||||
self.assertEqual(g.phase, "theme")
|
||||
await step(lambda: g.choose_theme(None, g.themes[seed % len(g.themes)]["id"], time.time()))
|
||||
if g.phase == "theme": # speakeren holdt skiftet; rummets timer skifter bagefter
|
||||
await step(lambda: g.tick(time.time() + 3600))
|
||||
else:
|
||||
await step(lambda: g.play_again(None, now))
|
||||
await step(lambda: g.tick(time.time() + 600)) # temaet: tiden løber ud
|
||||
self.assertEqual(g.phase, "word")
|
||||
while g.phase in ("word", "draw", "guess"):
|
||||
t, step_no = time.time(), g.step
|
||||
for i, p in enumerate(ps):
|
||||
if g.phase == "word":
|
||||
g.submit_word(p.id, WORDS[i % 2], t)
|
||||
elif g.phase == "draw":
|
||||
g.submit_drawing(p.id, [STROKE], t)
|
||||
else:
|
||||
g.submit_guess(p.id, WORDS[(i + 1) % 2], t)
|
||||
await step()
|
||||
if g.step == step_no and g.phase in ("word", "draw", "guess"):
|
||||
await step(lambda: g.tick(time.time() + 3600)) # alle er færdige; speakeren holdt skiftet
|
||||
for _ in range(100): # afsløringens lyd laves
|
||||
if g.reveal_started_at:
|
||||
break
|
||||
await asyncio.sleep(0.01)
|
||||
await step(lambda: g.tick(time.time() + 3600)) # podiet
|
||||
self.assertEqual(g.phase, "final")
|
||||
room.close()
|
||||
|
||||
# MAX_HOLD = 0: speakeren holder ikke skærmen tilbage, så testen ikke skal vente på lydens længde
|
||||
with mock.patch.object(tts, "enabled", return_value=True), mock.patch.object(tts, "synth", side_effect=fake_synth), \
|
||||
mock.patch("party.rooms.MAX_HOLD", 0):
|
||||
for seed in range(6):
|
||||
asyncio.run(play(seed))
|
||||
|
||||
warm = set(warm_texts())
|
||||
self.assertEqual({v for _, v in said}, {None}) # samme stemme som forvarmningen
|
||||
plain = [t for t, _ in said if not any(x.lower() in t.lower() for x in NAMES + WORDS)]
|
||||
self.assertGreater(len(plain), 40)
|
||||
self.assertEqual(sorted({t for t in plain if t not in warm}), [])
|
||||
events = {ev for ev in FIXED_LINES + THEME_LINES
|
||||
if any(t in plain for t in Narrator(lines=LINES).variants(ev, tema="x") + [
|
||||
v for th in load_themes() for v in Narrator(lines=LINES).variants(ev, tema=th["name"])])}
|
||||
self.assertLessEqual({"lobby_welcome", "draw_start", "guess_start", "step_all_done", "play_again",
|
||||
"theme_chosen", "theme_timeout", "word_start"}, events)
|
||||
self.assertTrue(any(t in plain for t in Narrator(lines=LINES).variants("game_start", antal_spillere=3)))
|
||||
self.assertLessEqual({r["say"] for r in TEGN_RULES}, set(plain))
|
||||
|
||||
|
||||
@override_settings(ELEVENLABS_API_KEY="nøgle", ELEVENLABS_VOICE_ID="stemme", FEST_VOICE_ID="")
|
||||
class TtsWarmCommandTests(SimpleTestCase):
|
||||
def run_cmd(self, *args):
|
||||
calls = []
|
||||
|
||||
def fake_sync(text, voice=None):
|
||||
calls.append((text, voice))
|
||||
return {"url": "/tts/x.mp3", "ms": 10}
|
||||
|
||||
out = io.StringIO()
|
||||
with mock.patch.object(tts, "synth_sync", side_effect=fake_sync), \
|
||||
mock.patch.object(tts, "_fetch", side_effect=AssertionError("netværk")), \
|
||||
mock.patch.object(tts, "credits_left", side_effect=AssertionError("netværk")):
|
||||
call_command("tts_warm", *args, stdout=out)
|
||||
return calls, out.getvalue()
|
||||
|
||||
def test_tegn_first_then_fup_og_fakta_as_before(self):
|
||||
calls, out = self.run_cmd()
|
||||
tegn = warm_texts()
|
||||
self.assertEqual([t for t, _ in calls], tegn + fof_static())
|
||||
self.assertEqual({v for _, v in calls}, {None})
|
||||
self.assertIn(f"{len(tegn)} af {len(tegn)} faste Tegn og Gæt-replikker ligger i cachen.", out)
|
||||
self.assertIn(f"{len(fof_static())} af {len(fof_static())} faste replikker ligger i cachen.", out)
|
||||
|
||||
def test_only(self):
|
||||
calls, _ = self.run_cmd("--only", "tegn", "--budget", "1000") # Tegn og Gæt har ingen budget-kø
|
||||
self.assertEqual([t for t, _ in calls], warm_texts())
|
||||
calls, _ = self.run_cmd("--only", "fof")
|
||||
self.assertEqual([t for t, _ in calls], fof_static())
|
||||
|
||||
@override_settings(ELEVENLABS_API_KEY="", ELEVENLABS_VOICE_ID="")
|
||||
def test_dry_run_counts_without_key_or_network(self):
|
||||
calls, out = self.run_cmd("--dry-run", "--only", "tegn")
|
||||
self.assertEqual(calls, [])
|
||||
tegn = warm_texts()
|
||||
self.assertIn(f"faste Tegn og Gæt-replikker: {len(tegn)} klip, {sum(map(len, tegn))} tegn", out)
|
||||
calls, out = self.run_cmd("--dry-run", "--budget", "500")
|
||||
self.assertEqual(calls, [])
|
||||
self.assertIn(f"faste replikker: {len(fof_static())} klip", out)
|
||||
self.assertIn("tælles ikke i --dry-run", out)
|
||||
|
||||
@override_settings(ELEVENLABS_API_KEY="")
|
||||
def test_without_key_and_without_dry_run_it_still_skips(self):
|
||||
calls, out = self.run_cmd()
|
||||
self.assertEqual(calls, [])
|
||||
self.assertIn("Springer over", out)
|
||||
Reference in New Issue
Block a user