I notice that I am using my LoggedApp class as the base here, so you will need to do some interpretation, but I think the principle is what you are looking for. There are also several extras that are probably not relevant to you.
Basically, it starts a thread to run the pyttx engine and adds sentences to speak to its queue with the say method. It can also detect a appd_tts event and speak the data in that, but I think that is mainly for testing.
import logged_app
import pyttsx3
import time
import threading
import datetime
#
# App to do Text to Speech entirely locally (no cloud involved).
# Uses pyttsx3 module. In your appdaemon VENV do
# pip3 install pyttsx3
#
# On linux this needs espeak
# sudo apt-get install espeak libespeak1
# also, on a raspberry pi, you may need to add the user that runs appdaemon
# to the audio group
# sudo usermod -aG audio <your appdaemon user>
# you will need to restart appdaemon after doing this.
# On Windows
# TBD uses sapi5?
# On Mac
# TBD uses nsss NSSpeechSynthesizer
#
# Args:
# voicename: the espeak voicename to use - see espeak --voices
# volume: Initial volume (0-1.0). Optional, default 1.0
# rate: the speed the voice speaks. Optional, default 200
# prefix: Something to say before the msg if nothing else is being said. Optional
# tts_topic: an MQTT topic to send message on. Optional
# toast_topic: (optional) another MQTT topic to send message on. Optional
class Tts(logged_app.LoggedApp):
def initialize(self):
self.log("initialize", level="DEBUG")
self.lock = threading.Lock()
self.connect_list = []
self._engine = pyttsx3.init()
self.count = 0
try:
self._engine.setProperty('voice', self.args["voicename"])
except KeyError as e:
self.log("Argument not found : {}".format(e), level="ERROR")
return
try:
self._engine.setProperty('volume', float(self.args["volume"]))
except KeyError as e:
self.log("Using default volume", level="DEBUG")
except ValueError as e:
self.log("volume must be a float", level="ERROR")
self.error("volume must be a float", level="ERROR")
return
try:
self._engine.setProperty('rate', int(self.args["rate"]))
except KeyError as e:
self.log("Using default rate", level="DEBUG")
except ValueError as e:
self.log("rate must be a int", level="ERROR")
self.error("rate must be a int", level="ERROR")
return
try:
self._prefix = self.args["prefix"]
except KeyError as e:
self._prefix = ""
# Set callbacks and store ids on connect_list
# self.connect_list.append(
# self._engine.connect('started-utterance', self.onStart))
# self.connect_list.append(
# self._engine.connect('started-word', self.onWord))
self.connect_list.append(
self._engine.connect('finished-utterance', self.onEnd))
self.connect_list.append(
self._engine.connect('error', self.onError))
# Run the engine in a separate thread
self._thread = threading.Thread(target=self.runEngine,
args=(self._engine, 1))
self._thread.start()
# Enable this to run a periodic test
#self.run_every(self.period, self.datetime(), 30)
# Set callback to receive appd_tts events
self.listen_event(self.say_event, "appd_tts")
def period(self, kwargs):
self.log("period", level="DEBUG")
self.say("The spare window is open and it is raining")
self.say("The not so spare window is open and it is raining")
def terminate(self):
self.log("terminate", level="DEBUG")
if not hasattr(self, "lock"):
return
with self.lock:
while len(self.connect_list) > 0:
self._engine.disconnect(self.connect_list.pop())
try:
if hasattr(self, "_engine"):
self.log("stopping engine", level="DEBUG")
self._engine.endLoop()
if hasattr(self, "_thread"):
self.log("Ending Engine thread", level="DEBUG")
self._thread.join(0.5)
if self._thread.is_alive():
self.log("Engine Thread didn't end", level="ERROR")
self.error("Engine Thread didn't end", level="ERROR")
except RuntimeError as e:
self.log("RuntimeError: {}".format(e), level="ERROR")
self.error("RuntimeError: {}".format(e), level="ERROR")
# Note that callbacks from the engine are running in the Engine thread
def onStart(self, name):
self.log ("starting {}".format(name), level="DEBUG")
def onWord(self, name, location, length):
self.log ("word {} {} {}".format(name, location, length), level="DEBUG")
def onEnd(self, name, completed):
self.log ("finishing {} {}".format(name, completed), level="DEBUG")
if int(name) == self.count:
self._turn_off_speakers()
def onError(self, name, exception):
self.log("Error from engine: {}".format(exception), level="ERROR")
self.error("Error from engine: {}".format(exception), level="ERROR")
# Runs in its own thread
def runEngine(self, engine, delay):
self.log("engine starting", level="DEBUG")
try:
engine.startLoop(True)
except RuntimeError:
# loop is already started
self.log("Loop already started, using existing", level="WARNING")
def say(self, orig_text):
self.log("Saying '{}'".format(orig_text), level="DEBUG")
if self._engine.isBusy():
text = "...and " + orig_text
else:
text = self._prefix + orig_text
self._turn_on_speakers()
with self.lock:
self.count+=1
self._engine.say(text, self.count)
try:
self.call_service("mqtt/publish", topic=self.args["tts_topic"],
payload=orig_text)
except ValueError:
pass
try:
self.call_service("mqtt/publish", topic=self.args["toast_topic"],
payload=orig_text)
except ValueError:
pass
def say_event(self, event_name, data, kwargs):
try:
self.log("say_event: {}".format(data['msg']), level="DEBUG")
self.say(data['msg'])
except KeyError:
self.log("say_event: msg not found in data)", level="WARNING")
def set_volume(self, volume):
self.log("set_volume: {}".format(volume), level="DEBUG")
try:
self._engine.setProperty('volume', float(volume))
except ValueError:
self.log("volume must be a float", level="WARNING")
def set_rate(self, rate):
self.log("set_rate: {}".format(rate), level="DEBUG")
try:
self._engine.setProperty('rate', int(rate))
except ValueError:
self.log("rate must be a int", level="WARNING")
def set_voice(self, voice):
self.log("set_voice: {}".format(voice), level="DEBUG")
self._engine.setProperty('voice', voice)
def _turn_on_speakers(self):
try:
speakers = self.get_app("speakers")
speakers.turn_on()
except AttributeError:
self.log("Speakers app not found")
def _turn_off_speakers(self):
try:
speakers = self.get_app("speakers")
speakers.turn_off()
except AttributeError:
self.log("Speakers app not found")