392 lines
14 KiB
Python
392 lines
14 KiB
Python
#!/usr/bin/env python3
|
|
# -*- coding: utf-8 -*-
|
|
"""Speech handling for Storm Games.
|
|
|
|
Provides functionality for:
|
|
- Text-to-speech using different speech providers
|
|
- Speech delay control to prevent stuttering
|
|
- On-screen text display
|
|
"""
|
|
|
|
import pygame
|
|
import textwrap
|
|
import time
|
|
from sys import exit
|
|
|
|
# Keep track of whether dialog instructions have been shown
|
|
dialogUsageInstructions = False
|
|
|
|
class Speech:
|
|
"""Handles text-to-speech functionality."""
|
|
|
|
_instance = None
|
|
|
|
@classmethod
|
|
def get_instance(cls):
|
|
"""Get or create the singleton instance."""
|
|
if cls._instance is None:
|
|
cls._instance = Speech()
|
|
return cls._instance
|
|
|
|
def __init__(self):
|
|
"""Initialize speech system with available provider."""
|
|
# Handle speech delays so we don't get stuttering
|
|
self.lastSpoken = {"text": None, "time": 0}
|
|
self.speechDelay = 250 # ms
|
|
|
|
# Try to initialize a speech provider
|
|
self.provider = None
|
|
self.providerName = None
|
|
|
|
# Try speechd first
|
|
try:
|
|
import speechd
|
|
self.spd = speechd.Client()
|
|
self.provider = self.spd
|
|
self.providerName = "speechd"
|
|
return
|
|
except ImportError:
|
|
pass
|
|
|
|
# Try accessible_output2 next
|
|
try:
|
|
import accessible_output2.outputs.auto
|
|
self.ao2 = accessible_output2.outputs.auto.Auto()
|
|
self.provider = self.ao2
|
|
self.providerName = "accessible_output2"
|
|
return
|
|
except ImportError:
|
|
pass
|
|
|
|
# No speech providers found
|
|
print("No speech providers found.")
|
|
|
|
def speak(self, text, interrupt=True, priority="normal"):
|
|
"""Speak text using the configured speech provider and display on screen.
|
|
|
|
Args:
|
|
text (str): Text to speak and display
|
|
interrupt (bool): Whether to interrupt current speech (default: True)
|
|
priority (str): Speech priority - "important", "normal", or "notification"
|
|
"""
|
|
if not self.provider:
|
|
return
|
|
|
|
currentTime = pygame.time.get_ticks()
|
|
|
|
# Check if this is the same text within the delay window
|
|
if (self.lastSpoken["text"] == text and
|
|
currentTime - self.lastSpoken["time"] < self.speechDelay):
|
|
return
|
|
|
|
# Update last spoken tracking
|
|
self.lastSpoken["text"] = text
|
|
self.lastSpoken["time"] = currentTime
|
|
|
|
# Add to speech history (import here to avoid circular imports)
|
|
try:
|
|
from .services import SpeechHistoryService
|
|
history_service = SpeechHistoryService.get_instance()
|
|
history_service.add_message(text)
|
|
except ImportError:
|
|
pass
|
|
|
|
# Proceed with speech based on provider and priority
|
|
if self.providerName == "speechd":
|
|
if interrupt:
|
|
self.spd.cancel()
|
|
|
|
# Set priority for speechd
|
|
if priority == "important":
|
|
try:
|
|
import speechd
|
|
self.spd.set_priority(speechd.Priority.IMPORTANT)
|
|
except (ImportError, AttributeError):
|
|
pass
|
|
elif priority == "notification":
|
|
try:
|
|
import speechd
|
|
self.spd.set_priority(speechd.Priority.NOTIFICATION)
|
|
except (ImportError, AttributeError):
|
|
pass
|
|
else: # normal
|
|
try:
|
|
import speechd
|
|
self.spd.set_priority(speechd.Priority.TEXT)
|
|
except (ImportError, AttributeError):
|
|
pass
|
|
|
|
self.spd.say(text)
|
|
|
|
elif self.providerName == "accessible_output2":
|
|
# For accessible_output2, use interrupt for important messages
|
|
use_interrupt = interrupt or (priority == "important")
|
|
self.ao2.speak(text, interrupt=use_interrupt)
|
|
|
|
# Display the text on screen
|
|
screen = pygame.display.get_surface()
|
|
if not screen:
|
|
return
|
|
|
|
font = pygame.font.Font(None, 36)
|
|
# Wrap the text
|
|
maxWidth = screen.get_width() - 40 # Leave a 20-pixel margin on each side
|
|
wrappedText = textwrap.wrap(text, width=maxWidth // font.size('A')[0])
|
|
# Render each line
|
|
textSurfaces = [font.render(line, True, (255, 255, 255)) for line in wrappedText]
|
|
screen.fill((0, 0, 0)) # Clear screen with black
|
|
# Calculate total height of text block
|
|
totalHeight = sum(surface.get_height() for surface in textSurfaces)
|
|
# Start y-position (centered vertically)
|
|
currentY = (screen.get_height() - totalHeight) // 2
|
|
# Blit each line of text
|
|
for surface in textSurfaces:
|
|
textRect = surface.get_rect(center=(screen.get_width() // 2, currentY + surface.get_height() // 2))
|
|
screen.blit(surface, textRect)
|
|
currentY += surface.get_height()
|
|
pygame.display.flip()
|
|
|
|
def close(self):
|
|
"""Clean up speech resources."""
|
|
if self.providerName == "speechd":
|
|
self.spd.close()
|
|
|
|
# Global instance for backward compatibility
|
|
_speechInstance = None
|
|
|
|
def speak(text, interrupt=True, priority="normal"):
|
|
"""Speak text using the global speech instance.
|
|
|
|
Args:
|
|
text (str): Text to speak and display
|
|
interrupt (bool): Whether to interrupt current speech (default: True)
|
|
priority (str): Speech priority - "important", "normal", or "notification"
|
|
"""
|
|
global _speechInstance
|
|
if _speechInstance is None:
|
|
_speechInstance = Speech.get_instance()
|
|
_speechInstance.speak(text, interrupt, priority)
|
|
|
|
def messagebox(text, sounds=None):
|
|
"""Enhanced messagebox with dialog support.
|
|
|
|
Args:
|
|
text (str or dict): Simple string message or dialog configuration dict
|
|
sounds (Sound object, optional): Sound system for playing dialog audio
|
|
"""
|
|
speech = Speech.get_instance()
|
|
|
|
# Handle simple string (backward compatibility)
|
|
if isinstance(text, str):
|
|
_show_simple_message(speech, text)
|
|
return
|
|
|
|
# Handle dialog format
|
|
if isinstance(text, dict) and "entries" in text:
|
|
_show_dialog_sequence(speech, text, sounds)
|
|
return
|
|
|
|
# Fallback to simple message if format not recognized
|
|
_show_simple_message(speech, str(text))
|
|
|
|
|
|
def _show_simple_message(speech, text):
|
|
"""Show a simple text message (original messagebox behavior)."""
|
|
speech.speak(text + "\nPress any key to repeat or enter to continue.")
|
|
while True:
|
|
event = pygame.event.wait()
|
|
if event.type == pygame.KEYDOWN:
|
|
if event.key in (pygame.K_ESCAPE, pygame.K_RETURN):
|
|
speech.speak(" ")
|
|
return
|
|
speech.speak(text + "\nPress any key to repeat or enter to continue.")
|
|
|
|
|
|
def _show_dialog_sequence(speech, dialog_config, sounds):
|
|
"""Show a dialog sequence with character speech and optional sounds.
|
|
|
|
Args:
|
|
speech: Speech instance for text-to-speech
|
|
dialog_config (dict): Dialog configuration with entries list and optional settings
|
|
sounds: Sound system for playing audio files
|
|
"""
|
|
entries = dialog_config.get("entries", [])
|
|
allow_skip = dialog_config.get("allow_skip", False)
|
|
dialog_sound = dialog_config.get("sound", None)
|
|
|
|
if not entries:
|
|
return
|
|
|
|
entry_index = 0
|
|
while entry_index < len(entries):
|
|
entry = entries[entry_index]
|
|
|
|
# Play sound before showing dialog
|
|
_play_dialog_sound(entry, dialog_config, sounds)
|
|
|
|
# Format and show the dialog text
|
|
formatted_text = _format_dialog_entry(entry)
|
|
if not formatted_text:
|
|
entry_index += 1
|
|
continue
|
|
|
|
# Show dialog with appropriate controls (only on first dialog of the game)
|
|
global dialogUsageInstructions
|
|
if not dialogUsageInstructions and entry_index == 0:
|
|
if allow_skip:
|
|
control_text = "\nPress any key to repeat, enter for next, or escape to skip all."
|
|
else:
|
|
control_text = "\nPress any key to repeat or enter for next."
|
|
dialogUsageInstructions = True
|
|
else:
|
|
control_text = "" # No instructions after first dialog
|
|
|
|
speech.speak(formatted_text + control_text)
|
|
|
|
# Handle user input
|
|
while True:
|
|
event = pygame.event.wait()
|
|
if event.type == pygame.KEYDOWN:
|
|
if event.key == pygame.K_ESCAPE:
|
|
if allow_skip:
|
|
speech.speak(" ")
|
|
return # Skip entire dialog sequence
|
|
else:
|
|
# Escape acts like enter if skip not allowed
|
|
speech.speak(" ")
|
|
entry_index += 1
|
|
break
|
|
elif event.key == pygame.K_RETURN:
|
|
speech.speak(" ")
|
|
entry_index += 1
|
|
break
|
|
else:
|
|
# Repeat current entry (no instructions when repeating)
|
|
speech.speak(formatted_text)
|
|
|
|
|
|
def _format_dialog_entry(entry):
|
|
"""Format a dialog entry for display.
|
|
|
|
Args:
|
|
entry (dict): Dialog entry with text, optional speaker, and optional narrative flag
|
|
|
|
Returns:
|
|
str: Formatted text for speech
|
|
"""
|
|
text = entry.get("text", "")
|
|
speaker = entry.get("speaker", None)
|
|
is_narrative = entry.get("narrative", False)
|
|
|
|
if not text:
|
|
return ""
|
|
|
|
if is_narrative:
|
|
# Narrative text - no speaker name
|
|
return text
|
|
elif speaker:
|
|
# Character dialog - include speaker name
|
|
return f"{speaker}: \"{text}\""
|
|
else:
|
|
# Plain text - no special formatting
|
|
return text
|
|
|
|
|
|
def _play_dialog_sound(entry, dialog_config, sounds):
|
|
"""Play appropriate sound for a dialog entry and wait for it to complete.
|
|
|
|
Args:
|
|
entry (dict): Dialog entry that may have a sound
|
|
dialog_config (dict): Dialog configuration that may have a default sound
|
|
sounds: Sound system (either Sound class instance or dictionary of sounds)
|
|
"""
|
|
if not sounds:
|
|
return
|
|
|
|
sound_to_play = None
|
|
|
|
# Determine which sound to play (priority order)
|
|
if entry.get("sound"):
|
|
# Entry-specific sound (highest priority)
|
|
sound_to_play = entry["sound"]
|
|
elif dialog_config.get("sound"):
|
|
# Dialog-specific sound (medium priority)
|
|
sound_to_play = dialog_config["sound"]
|
|
else:
|
|
# Default dialogue.ogg (lowest priority)
|
|
sound_to_play = "dialogue" # Will look for dialogue.ogg
|
|
|
|
if sound_to_play:
|
|
try:
|
|
# Handle both Sound class instances and sound dictionaries
|
|
if hasattr(sounds, 'sounds') and sound_to_play in sounds.sounds:
|
|
# Sound class instance (like from libstormgames Sound class)
|
|
sound_obj = sounds.sounds[sound_to_play]
|
|
from .sound import get_available_channel
|
|
channel = get_available_channel()
|
|
channel.play(sound_obj)
|
|
sound_duration = sound_obj.get_length()
|
|
if sound_duration > 0:
|
|
pygame.time.wait(int(sound_duration * 1000))
|
|
pygame.event.clear() # Clear all events queued during sound playback
|
|
elif isinstance(sounds, dict) and sound_to_play in sounds:
|
|
# Dictionary of pygame sound objects (like from initialize_gui)
|
|
sound_obj = sounds[sound_to_play]
|
|
from .sound import get_available_channel
|
|
channel = get_available_channel()
|
|
channel.play(sound_obj)
|
|
sound_duration = sound_obj.get_length()
|
|
if sound_duration > 0:
|
|
pygame.time.wait(int(sound_duration * 1000))
|
|
pygame.event.clear() # Clear all events queued during sound playback
|
|
elif hasattr(sounds, 'play'):
|
|
# Try using a play method if available
|
|
sounds.play(sound_to_play)
|
|
pygame.time.wait(500) # Default delay if can't get duration
|
|
pygame.event.pump() # Clear any events queued during sound playback
|
|
except Exception:
|
|
# Sound missing or error - continue silently without crashing
|
|
pass
|
|
|
|
|
|
def speak_previous():
|
|
"""Navigate to and speak the previous message in speech history."""
|
|
try:
|
|
from .services import SpeechHistoryService
|
|
history_service = SpeechHistoryService.get_instance()
|
|
message = history_service.move_previous()
|
|
if message:
|
|
speak(message, interrupt=True, priority="important")
|
|
else:
|
|
speak("No previous messages", interrupt=True, priority="important")
|
|
except ImportError:
|
|
speak("Speech history not available", interrupt=True, priority="important")
|
|
|
|
|
|
def speak_current():
|
|
"""Repeat the current message in speech history."""
|
|
try:
|
|
from .services import SpeechHistoryService
|
|
history_service = SpeechHistoryService.get_instance()
|
|
message = history_service.get_current()
|
|
if message:
|
|
speak(message, interrupt=True, priority="important")
|
|
else:
|
|
speak("No current message", interrupt=True, priority="important")
|
|
except ImportError:
|
|
speak("Speech history not available", interrupt=True, priority="important")
|
|
|
|
|
|
def speak_next():
|
|
"""Navigate to and speak the next message in speech history."""
|
|
try:
|
|
from .services import SpeechHistoryService
|
|
history_service = SpeechHistoryService.get_instance()
|
|
message = history_service.move_next()
|
|
if message:
|
|
speak(message, interrupt=True, priority="important")
|
|
else:
|
|
speak("No next messages", interrupt=True, priority="important")
|
|
except ImportError:
|
|
speak("Speech history not available", interrupt=True, priority="important")
|