# wcs/games/millionaire_tts.py """ TTS helper for Millionaire game. Handles text composition, option formatting, and TTS playback for all game states. Eliminates duplicated TTS code. """ import random import re import threading import xbmc import xbmcaddon # Module-level TTS activity tracking _tts_active = threading.Event() # Set when TTS is generating/playing _tts_serial = 0 _tts_owner = None _tts_path = '' _tts_lock = threading.RLock() # Regex for stripping ElevenLabs v3 audio tags from UI text _AUDIO_TAG_RE = re.compile(r'\[(?:excited|nervous|frustrated|sorrowful|calm|sad|angry|happily|tired|curious|mischievously|sigh|sighs|laughs|laughing|gasps|gulps|whispers|crying|clears throat|snorts|exhales|panting|hesitates|stammers|pauses|rushed|cheerfully|flatly|deadpan|playfully|dramatically|sarcastically|matter-of-fact|whiny|whispering|shouting|quietly|loudly|gunshot|clapping|explosion|door creaks|pirate voice|evil scientist voice|childlike tone|British accent|Australian accent|Southern US accent|French accent)\]') def strip_audio_tags(text): """Remove ElevenLabs v3 audio tags from text for UI display. Tags like [excited], [laughs], [pauses] are stripped. Cleans up resulting double spaces. """ if not text: return text cleaned = _AUDIO_TAG_RE.sub('', text) # Collapse multiple spaces cleaned = re.sub(r' +', ' ', cleaned).strip() return cleaned def is_elevenlabs_v3_active(): """Check if ElevenLabs v3 model is currently selected.""" addon = xbmcaddon.Addon() provider = addon.getSetting("milionar_tts_provider") or "OpenAI" if provider == "1": provider = "ElevenLabs" elif provider == "0" or provider not in ("OpenAI", "ElevenLabs"): return False if provider != "ElevenLabs": return False model = addon.getSetting("milionar_tts_model_elevenlabs") or "" from wcs.ai.model_catalog import ELEVENLABS_AUDIO_TAG_MODELS return model in ELEVENLABS_AUDIO_TAG_MODELS # --------------------------------------------------------------------------- # Option reading styles (randomly selected per question) # --------------------------------------------------------------------------- OPTION_STYLES = [ # "Za A: text, za B: text, ..." lambda opts: ", ".join( "za {}: {}".format(letter, text) for letter, text in zip(["A", "B", "C", "D"], opts) if text ), # "A: text, B: text, ..." lambda opts: ", ".join( "{}: {}".format(letter, text) for letter, text in zip(["A", "B", "C", "D"], opts) if text ), # "text, text, text, text" (bez pismen) lambda opts: ", ".join(text for text in opts if text), ] def format_options(options): """Format answer options using a randomly selected style. Args: options: List of 4 option strings [A, B, C, D]. Returns: Formatted string with all options. """ style = random.choice(OPTION_STYLES) return style(options) # --------------------------------------------------------------------------- # TTS settings loader # --------------------------------------------------------------------------- def _load_tts_settings(): """Load all TTS settings from addon. Returns dict.""" addon = xbmcaddon.Addon() provider = addon.getSetting("milionar_tts_provider") or "OpenAI" # Normalize: settings may return index "0"/"1" or value if provider == "1": provider = "ElevenLabs" elif provider == "0" or provider not in ("OpenAI", "ElevenLabs"): provider = "OpenAI" if provider == "ElevenLabs": voice = addon.getSetting("milionar_tts_voice_elevenlabs") or "Adam" model = addon.getSetting("milionar_tts_model_elevenlabs") or "eleven_flash_v2_5" else: voice = addon.getSetting("milionar_tts_voice") or "alloy" model = addon.getSetting("milionar_tts_model") or "gpt-4o-mini-tts" return { "tts_welcome": addon.getSetting("milionar_tts_welcome") == "true", "tts_intro": addon.getSetting("milionar_tts_intro") == "true", "tts_question": addon.getSetting("milionar_tts_question") == "true", "tts_options": addon.getSetting("milionar_tts_options") == "true", "tts_result": addon.getSetting("milionar_tts_result") == "true", "tts_transition": addon.getSetting("milionar_tts_transition") == "true", "tts_explanation": addon.getSetting("milionar_tts_explanation") == "true", "provider": provider, "voice": voice, "model": model, } def _any_enabled(settings, keys): """Check if any of the given keys are enabled.""" return any(settings.get(k, False) for k in keys) # --------------------------------------------------------------------------- # Text composers (build TTS string with pauses) # --------------------------------------------------------------------------- def compose_question_tts(settings, moderator_intro, question_text, options): """Compose TTS text for question display. Combines enabled parts with natural pauses (periods, commas). Args: settings: TTS settings dict from _load_tts_settings(). moderator_intro: Moderator intro text (str or empty). question_text: Question text. options: List of 4 option strings. Returns: Combined TTS string, or empty if nothing enabled. """ parts = [] if settings["tts_intro"] and moderator_intro: parts.append(moderator_intro.rstrip(".") + ".") if settings["tts_question"] and question_text: parts.append(question_text.rstrip("?") + "?") if settings["tts_options"] and options: parts.append(format_options(options)) return " ... ".join(parts) def compose_feedback_tts(settings, result_text, transition_text, explanation=""): """Compose TTS text for feedback/game over display. Args: settings: TTS settings dict. result_text: Result announcement from AI. transition_text: Transition text from AI. explanation: Explanation text (fallback). Returns: Combined TTS string. """ parts = [] if settings["tts_explanation"] and explanation: parts.append(explanation.rstrip(".") + ".") if settings["tts_result"] and result_text: parts.append(result_text.rstrip(".") + ".") if settings["tts_transition"] and transition_text: parts.append(transition_text) return " ... ".join(parts) def compose_welcome_tts(settings, welcome_text, rules_text=""): """Compose TTS text for welcome screen. Args: settings: TTS settings dict. welcome_text: Moderator welcome text. rules_text: Rules summary text. Returns: Combined TTS string. """ if not settings["tts_welcome"]: return "" parts = [] if welcome_text: parts.append(welcome_text.rstrip(".") + ".") if rules_text: parts.append(rules_text) result = " ... ".join(parts) return result # --------------------------------------------------------------------------- # TTS playback (background thread) # --------------------------------------------------------------------------- def play_tts(text, owner=None): """Play TTS in background thread using TTSClient. Loads settings, composes nothing -- expects ready text. Runs in daemon thread, never blocks caller. Sets _tts_active event before launch, clears after playback. Args: text: Text to speak. If empty, does nothing. """ if not text or not text.strip(): return global _tts_serial, _tts_owner, _tts_path settings = _load_tts_settings() voice = settings["voice"] # Mark TTS as active BEFORE launching thread with _tts_lock: _tts_serial += 1 serial = _tts_serial _tts_owner = owner _tts_path = '' _tts_active.set() provider = settings.get("provider", "OpenAI") model = settings.get("model", "") def _tts_worker(): global _tts_path try: if provider == "ElevenLabs": from wcs.ai import ElevenLabsTTSClient as el_tts audio_path = el_tts.generate_tts( text.strip(), voice_name=voice, model_id=model, ) else: from wcs.ai import TTSClient as openai_tts audio_path = openai_tts.generate_tts( text.strip(), voice=voice, ) with _tts_lock: if serial != _tts_serial: return _tts_path = audio_path xbmc.executebuiltin('PlayMedia("{}", noresume)'.format(audio_path)) # Wait for playback to actually finish import time time.sleep(0.5) # Give Kodi time to start playing while xbmc.getCondVisibility("Player.Playing"): time.sleep(0.3) except Exception as e: xbmc.log( "[plugin.video.milionar] TTS chyba: {}".format(e), level=xbmc.LOGWARNING, ) finally: with _tts_lock: if serial == _tts_serial: _tts_active.clear() threading.Thread(target=_tts_worker, daemon=True).start() def stop_tts(owner=None): """Stop any currently playing TTS audio.""" global _tts_serial, _tts_path, _tts_owner with _tts_lock: if owner is not None and owner is not _tts_owner: return _tts_serial += 1 # A Story with no active owned audio must never stop a video underneath its card. playing_owned = _tts_path and xbmc.getInfoLabel('Player.Filenameandpath') == _tts_path if owner is None or playing_owned: xbmc.executebuiltin('PlayerControl(Stop)') _tts_path = '' _tts_owner = None _tts_active.clear() def is_tts_playing(): """Check if Kodi is currently playing TTS audio.""" return xbmc.getCondVisibility("Player.Playing") def wait_for_tts_completion(timeout=30): """Block until TTS playback finishes. For autoplay sync. Uses _tts_active event (set by play_tts, cleared after playback). Falls back to Player.Playing check. Args: timeout: Maximum seconds to wait. """ import time start = time.time() # Wait until _tts_active is cleared (worker finished) while _tts_active.is_set(): if time.time() - start > timeout: break time.sleep(0.3) # --------------------------------------------------------------------------- # High-level TTS triggers (called from MillionaireGameController) # --------------------------------------------------------------------------- def tts_welcome(welcome_text, rules_text="", owner=None): """Trigger TTS for welcome screen if enabled.""" settings = _load_tts_settings() text = compose_welcome_tts(settings, welcome_text, rules_text) if text: play_tts(text, owner=owner) def tts_question(moderator_intro, question_text, options, owner=None): """Trigger TTS for question display if any question TTS enabled.""" settings = _load_tts_settings() keys = ["tts_intro", "tts_question", "tts_options"] if not _any_enabled(settings, keys): return text = compose_question_tts(settings, moderator_intro, question_text, options) if text: play_tts(text, owner=owner) FEEDBACK_KEYS = ["tts_result", "tts_transition", "tts_explanation"] def feedback_enabled(): """True when any feedback part (result, transition, explanation) is read aloud.""" return _any_enabled(_load_tts_settings(), FEEDBACK_KEYS) def tts_feedback(result_text, transition_text, explanation="", owner=None): """Trigger TTS for answer feedback if any feedback TTS enabled.""" settings = _load_tts_settings() if not _any_enabled(settings, FEEDBACK_KEYS): return text = compose_feedback_tts(settings, result_text, transition_text, explanation) if text: play_tts(text, owner=owner)