[WIP] Text to speech mixin

debian/control

@@ -82,6 +82,9 @@ Replaces:
  pt-device-manager (<< 4.0.0),
 # pt-oled
  python3-pt-oled (<< 3.0.0),
+# Festival speech engine
+ festival,
+ festvox-us-slt-hts,
 Description: pi-top Python 3 Library - Main
  General purpose Python library for controlling a pi-top.
  .

debian/py3dist-overrides

@@ -21,3 +21,4 @@ spidev python3-spidev; PEP386
 systemd_python python3-systemd; PEP386
 wget python3-wget; PEP386
 pyzmq python3-zmq; PEP386
+pyfestival python3-pt-pyfestival

examples/system/pitop_speech.py

@@ -0,0 +1,17 @@
+from pitop import Pitop
+
+pitop = Pitop()
+
+pitop.speak.print_voices()
+
+voices = pitop.speak.available_voices
+
+for language, voices in voices.items():
+    for voice in voices:
+        pitop.speak.set_voice(language, voice)
+        print(f"LANGUAGE: {language} | VOICE: {voice}", flush=True)
+        pitop.speak("Hello")
+
+while True:
+    text = input("Enter text: ")
+    pitop.speak(text)

examples/system/tts_service.py

@@ -0,0 +1,9 @@
+from pitop.processing import tts
+
+config = {
+    "language": "us",
+}
+
+speech = tts.services.get(service_id="FESTIVAL", **config)
+
+speech.say("This is the festival tts service")

pitop/core/mixins/__init__.py

@@ -3,3 +3,4 @@
 from .stateful import Stateful
 from .supports_battery import SupportsBattery
 from .supports_miniscreen import SupportsMiniscreen
+from .supports_speech import SupportsSpeech

pitop/core/mixins/supports_speech.py

@@ -0,0 +1,22 @@
+from pitop.processing import tts
+from pitop.processing.tts.services.tts_service import TTSService
+
+
+class SupportsSpeech:
+    def __init__(self, service_id: str = "DEFAULT"):
+        self._tts = tts.services.get(service_id=service_id)
+
+    @property
+    def speak(self):
+        return self._tts
+
+    @speak.setter
+    def speak(self, service: TTSService):
+        self._tts = service
+
+    @classmethod
+    def using_speech_service(cls, service_id):
+        obj = cls()
+        speech_service = tts.services.get(service_id=service_id)
+        obj.speak = speech_service
+        return obj

pitop/processing/tts/__init__.py

@@ -0,0 +1 @@
+from .tts_service_provider import services

pitop/processing/tts/object_factory.py

@@ -0,0 +1,12 @@
+class ObjectFactory:
+    def __init__(self):
+        self._builders = {}
+
+    def register_builder(self, key, builder):
+        self._builders[key] = builder
+
+    def create(self, key, **kwargs):
+        builder = self._builders.get(key)
+        if not builder:
+            raise ValueError(key)
+        return builder(**kwargs)

pitop/processing/tts/services/festival_service.py

@@ -0,0 +1,111 @@
+import os
+from .tts_service import TTSService
+from threading import Thread
+from typing import Optional
+
+
+class FestivalBuilder:
+    def __init__(self):
+        self._instance = None
+
+    def __call__(self, **kwargs):
+        if self._instance is None:
+            self._instance = FestivalService(**kwargs)
+        return self._instance
+
+
+class FestivalService(TTSService):
+
+    __VOICE_DIR = os.path.join(os.sep, "usr", "share", "festival", "voices")
+
+    def __init__(self, language="us", **_ignored):
+        self._speed = 1.0
+        self._available_voices = self.__get_available_voices()
+        self._language = language
+        self._voice = self._available_voices.get(self.language)[0]
+        self.set_voice(self._language, self._voice)
+        self._say_subprocess = None
+        self._say_thread = Thread()
+
+    def __call__(self, text: str, blocking: bool = True):
+        self.say(text=text, blocking=blocking)
+
+    def say(self, text: str, blocking: bool = True) -> None:
+        if not self.__validate_request(text):
+            return
+
+        self._say_thread = Thread(target=self.__festival_say_commands, args=(text,), daemon=True)
+        self._say_thread.start()
+
+        if blocking:
+            self._say_thread.join()
+
+    def __festival_say_commands(self, text):
+        # Have to import festival within the thread since it doesn't work with multi-threading
+        # As a result, have to set voice and stretch factor with every call
+        import festival
+        festival.execCommand(f"(voice_{self.voice})")
+        festival.setStretchFactor(1 / self.speed)
+        festival.sayText(text)
+
+    def __validate_request(self, text):
+        if text == "" or type(text) != str:
+            raise ValueError("Text must be a string and cannot be empty.")
+
+        if self._say_thread.is_alive():
+            print("Speech already in progress, request cancelled.")
+            return False
+
+        return True
+
+    def __get_available_voices(self):
+        languages = os.listdir(self.__VOICE_DIR)
+        language_dirs = (os.path.join(self.__VOICE_DIR, lang) for lang in languages)
+
+        voice_dict = {}
+        for lang, lang_dir in zip(languages, language_dirs):
+            voice_dict[lang] = os.listdir(lang_dir)
+
+        return voice_dict
+
+    @property
+    def available_voices(self) -> dict:
+        return self._available_voices
+
+    def set_voice(self, language: str, voice: Optional[str] = None) -> None:
+        available_languages = list(self._available_voices.keys())
+        if language not in available_languages:
+            raise ValueError("Invalid language choice. Please choose from:\n"
+                             f"{available_languages}")
+
+        available_voices = self._available_voices.get(language)
+
+        voice = available_voices[0] if voice is None else voice
+
+        if voice not in available_voices:
+            raise ValueError(f"Invalid voice choice. Please choose from:\n"
+                             f"{available_voices}\n"
+                             f"Or choose a different language. "
+                             f"Run display_voices() method to see what is available.")
+
+        self._voice = voice
+        self._language = language
+
+    @property
+    def voice(self):
+        return self._voice
+
+    @property
+    def language(self):
+        return self._language
+
+    @property
+    def speed(self):
+        return self._speed
+
+    @speed.setter
+    def speed(self, value: float):
+        if value < 0.2:
+            raise ValueError("Speed value must be greater than or equal to 0.2.")
+
+        self._speed = value

pitop/processing/tts/services/tts_service.py

@@ -0,0 +1,65 @@
+from abc import (
+    ABCMeta,
+    abstractmethod,
+)
+
+
+class AttributeMeta(ABCMeta):
+    """Metaclass for ensuring children of abstract base class define specified
+    attributes."""
+    required_attributes = []
+
+    def __call__(cls, *args, **kwargs):
+        obj = super(AttributeMeta, cls).__call__(*args, **kwargs)
+        for attr_name in obj.required_attributes:
+            if not getattr(obj, attr_name):
+                raise ValueError('required attribute (%s) not set' % attr_name)
+        return obj
+
+
+class TTSService(metaclass=AttributeMeta):
+    required_attributes = ['_voice', '_available_voices', '_language', '_speed']
+
+    @abstractmethod
+    def __call__(self, text: str, blocking: bool = True) -> None:
+        pass
+
+    @abstractmethod
+    def say(self, text: str, blocking: bool = True) -> None:
+        pass
+
+    @property
+    @abstractmethod
+    def available_voices(self) -> dict:
+        pass
+
+    @abstractmethod
+    def set_voice(self, language: str, name: str) -> None:
+        pass
+
+    @property
+    @abstractmethod
+    def voice(self) -> str:
+        pass
+
+    @property
+    @abstractmethod
+    def language(self) -> str:
+        pass
+
+    @property
+    @abstractmethod
+    def speed(self):
+        pass
+
+    @speed.setter
+    @abstractmethod
+    def speed(self, speed: float):
+        pass
+
+    def print_voices(self):
+        print("LANGUAGE   VOICES")
+        print("--------   ------")
+        for language, voices in self.available_voices.items():
+            print(f"{language:<10} {', '.join(voices)}")
+        print()

pitop/processing/tts/tts_service_provider.py

@@ -0,0 +1,12 @@
+from .object_factory import ObjectFactory
+from .services.festival_service import FestivalBuilder
+
+
+class TTSServiceProvider(ObjectFactory):
+    def get(self, service_id="DEFAULT", **kwargs):
+        return self.create(service_id, **kwargs)
+
+
+services = TTSServiceProvider()
+services.register_builder("DEFAULT", FestivalBuilder())
+services.register_builder("FESTIVAL", FestivalBuilder())

pitop/system/pitop.py

@@ -3,10 +3,11 @@
     Componentable,
     SupportsBattery,
     SupportsMiniscreen,
+    SupportsSpeech,
 )
 
 
-class Pitop(SupportsMiniscreen, SupportsBattery, Componentable, metaclass=Singleton):
+class Pitop(SupportsMiniscreen, SupportsBattery, SupportsSpeech, Componentable, metaclass=Singleton):
     """Represents a pi-top Device.
 
     When creating a `Pitop` object, multiple properties will be set,
@@ -33,4 +34,5 @@ class in 2 different files, they will share the internal state.
     def __init__(self):
         SupportsMiniscreen.__init__(self)
         SupportsBattery.__init__(self)
+        SupportsSpeech.__init__(self)
         Componentable.__init__(self)

setup.py

@@ -137,6 +137,11 @@
     # Download Model Files #
     ########################
     "wget>=3.2,<4.0",
+
+    ##################
+    # Text to Speech #
+    ##################
+    "pt-pyfestival",
 ]
 
 __extra_requires__ = {