#!/usr/bin/env python3 # -*- coding: utf-8 -*- # Fenrir TTY screen reader # By Chrys, Storm Dragon, and contributors. # speech-dispatcher driver from fenrirscreenreader.core import debug from fenrirscreenreader.core.speechDriver import speech_driver class driver(speech_driver): """Speech-dispatcher driver for Fenrir screen reader. This driver provides text-to-speech functionality through speech-dispatcher, which acts as a common interface to various TTS engines. It supports voice selection, speech parameters (rate, pitch, volume), and multiple TTS modules. Features: - Dynamic voice switching and parameter adjustment - Support for multiple speech-dispatcher modules (espeak, festival, etc.) - Real-time speech cancellation and queueing - Language and voice selection - Speech callbacks for synchronization Attributes: _sd: Speech-dispatcher client connection language (str): Current speech language voice (str): Current voice name module (str): Current TTS module name """ def __init__(self): speech_driver.__init__(self) def initialize(self, environment): """Initialize the speech-dispatcher connection. Establishes connection to speech-dispatcher daemon and configures initial speech parameters. Sets up callbacks and prepares the speech subsystem for use. Args: environment: Fenrir environment dictionary with settings and managers Note: Gracefully handles cases where speech-dispatcher is not available. """ self._sd = None self.env = environment self._is_initialized = False # Only set these if they haven't been set yet (preserve existing # values) if not hasattr(self, "language") or self.language is None: self.language = "" if not hasattr(self, "voice") or self.voice is None: self.voice = "" if not hasattr(self, "module") or self.module is None: self.module = "" try: import speechd self._sd = speechd.SSIPClient("fenrir-dev") self._punct = speechd.PunctuationMode() self._is_initialized = True except Exception as e: self.env["runtime"]["DebugManager"].write_debug_out( "SpeechDriver initialize:" + str(e), debug.DebugLevel.ERROR ) def shutdown(self): """Shutdown the speech-dispatcher connection. Cleanly closes the connection to speech-dispatcher and releases any allocated resources. """ if not self._is_initialized: return self.cancel() try: self._sd.close() except Exception as e: pass self._is_initialized = False def speak(self, text, queueable=True, ignore_punctuation=False): """Speak the given text through speech-dispatcher. Args: text (str): Text to speak queueable (bool): Whether speech can be queued with other speech ignore_punctuation (bool): Whether to ignore punctuation settings Note: Handles text preprocessing and manages speech queue based on parameters. """ if not queueable: self.cancel() if not self._is_initialized: self.initialize(self.env) if not queueable: self.cancel() if not self._is_initialized: return try: if self.module != "": self._sd.set_output_module(self.module) except Exception as e: self.env["runtime"]["DebugManager"].write_debug_out( "SpeechDriver set_module:" + str(e), debug.DebugLevel.ERROR ) try: if self.language != "": self._sd.set_language(self.language) except Exception as e: self.env["runtime"]["DebugManager"].write_debug_out( "SpeechDriver set_language:" + str(e), debug.DebugLevel.ERROR ) try: if self.voice != "": self._sd.set_synthesis_voice(self.voice) except Exception as e: self.env["runtime"]["DebugManager"].write_debug_out( "SpeechDriver set_voice:" + str(e), debug.DebugLevel.ERROR ) try: if ignore_punctuation: self._sd.set_punctuation(self._punct.ALL) else: self._sd.set_punctuation(self._punct.NONE) except Exception as e: self.env["runtime"]["DebugManager"].write_debug_out( "SpeechDriver set_punctuation:" + str(e), debug.DebugLevel.ERROR, ) try: # Check if read-all is active and add callback if needed if ("ReadAllManager" in self.env["runtime"] and self.env["runtime"]["ReadAllManager"] and self.env["runtime"]["ReadAllManager"].is_active()): import speechd def read_all_callback(msg_id, client_id, event_type, *args): if (event_type == speechd.CallbackType.END and self.env["runtime"]["ReadAllManager"] and self.env["runtime"]["ReadAllManager"].is_active()): self.env["runtime"]["ReadAllManager"].speech_completed() self._sd.speak(text, callback=read_all_callback, event_types=(speechd.CallbackType.END,)) else: self._sd.speak(text) except Exception as e: self.env["runtime"]["DebugManager"].write_debug_out( "SpeechDriver speak:" + str(e), debug.DebugLevel.ERROR ) self._is_initialized = False def cancel(self): """Cancel all pending and current speech. Immediately stops speech output and clears the speech queue. """ if not self._is_initialized: self.initialize(self.env) if not self._is_initialized: return try: self._sd.cancel() except Exception as e: self.env["runtime"]["DebugManager"].write_debug_out( "SpeechDriver cancel:" + str(e), debug.DebugLevel.ERROR ) self._is_initialized = False def set_pitch(self, pitch): """Set the speech pitch. Args: pitch (float): Speech pitch (0.0 to 1.0, where 0.5 is normal) """ if not self._is_initialized: return try: self._sd.set_pitch(int(-100 + pitch * 200)) except Exception as e: self.env["runtime"]["DebugManager"].write_debug_out( "SpeechDriver set_pitch:" + str(e), debug.DebugLevel.ERROR ) def set_rate(self, rate): """Set the speech rate. Args: rate (float): Speech rate (0.0 to 1.0, where 0.5 is normal) """ if not self._is_initialized: return try: self._sd.set_rate(int(-100 + rate * 200)) except Exception as e: self.env["runtime"]["DebugManager"].write_debug_out( "SpeechDriver set_rate:" + str(e), debug.DebugLevel.ERROR ) def set_volume(self, volume): """Set the speech volume. Args: volume (float): Volume level (0.0 to 1.0) """ if not self._is_initialized: return try: self._sd.set_volume(int(-100 + volume * 200)) except Exception as e: self.env["runtime"]["DebugManager"].write_debug_out( "SpeechDriver set_volume:" + str(e), debug.DebugLevel.ERROR )