AppDaemon-based TTS System

config.yaml

living_room_tv:
  module: harmony
  class: HarmonyRemote
  entity: remote.samsung_tv
  activities:
    play_ps4: 27901089
    watch_roku: 39586383
    watch_tv: 27901129
sonos_manager:
  module: sonos
  class: SonosManager
sonos_house_audio:
  module: sonos
  class: SonosSpeaker
  dependencies:
    - sonos_manager
  entity: media_player.house_audio
sonos_living_room:
  module: sonos
  class: SonosSpeaker
  dependencies:
    - sonos_manager
  entity: media_player.living_room
sonos_office_desk:
  module: sonos
  class: SonosSpeaker
  dependencies:
    - sonos_manager
  entity: media_player.office_desk
tts:
  module: tts
  class: TTS
  dependencies:
    - living_room_tv
    - sonos_manager

harmony.py

"""Define an app for working the Living Room TV."""

# pylint: disable=attribute-defined-outside-init,too-few-public-methods

import appdaemon.plugins.hass.hassapi as hass


class HarmonyRemote(hass.Hass):
    """Define a class to represent the Living Room TV."""

    def initialize(self):
        """Initialize."""
        self.activities = self.args['activities']
        self.entity = self.args['entity']

    @property
    def current_activity_id(self):
        """Get the current activity ID (Harmony)."""
        activity = self.get_state(self.entity, attribute='current_activity')
        try:
            return self.activities[activity.replace(' ', '_').lower()]
        except KeyError:
            return None

    def send_command(self, command):
        """Send a command to the Harmony."""
        if self.current_activity_id:
            self.call_service(
                'remote/send_command',
                entity_id=self.entity,
                device=self.current_activity_id,
                command=command)

    def pause(self):
        """Pause the entire thing by pausing the Harmony."""
        self.send_command('Pause')

    def play(self):
        """Play the entire thing by playing the Harmony."""
        self.send_command('Play')

sonos.py

"""Define an app to manage our Sonos players."""

# pylint: disable=too-many-arguments,attribute-defined-outside-init

import appdaemon.plugins.hass.hassapi as hass


class SonosSpeaker(hass.Hass):
    """Define a class to represent a Sonos speaker."""

    def __str__(self):
        """Define a string representation of the speaker."""
        return self.entity

    def initialize(self):
        """Initialize."""
        self._last_snapshot_included_group = False
        self.entity = self.args['entity']

        self.sonos_manager = self.get_app('sonos_manager')
        self.sonos_manager.register_entity(self)

    @property
    def volume(self):
        """Retrieve the audio player's volume."""
        return self.get_state(self.entity, attribute='volume_level')

    @volume.setter
    def volume(self, value):
        """Set the audio player's volume."""
        self.call_service(
            'media_player/volume_set',
            entity_id=self.entity,
            volume_level=value)

    def pause(self):
        """Pause."""
        self.call_service('media_player/media_pause', entity_id=self.entity)

    def play(self):
        """Play."""
        self.call_service('media_player/media_play', entity_id=self.entity)

    def play_file(self, url):
        """Play an audio file at a defined URL."""
        self.call_service(
            'media_player/play_media',
            entity_id=self.entity,
            media_content_id=url,
            media_content_type='MUSIC')

    def restore(self):
        """Restore the previous snapshot of this entity."""
        self.call_service(
            'media_player/sonos_restore',
            entity_id=self.entity,
            with_group=self._last_snapshot_included_group)

    def snapshot(self, include_grouping=True):
        """Snapshot this entity."""
        self._last_snapshot_included_group = include_grouping
        self.call_service(
            'media_player/sonos_snapshot',
            entity_id=self.entity,
            with_group=include_grouping)


class SonosManager(hass.Hass):
    """Define a class to represent the Sono manager."""

    def initialize(self):
        """Initialize."""
        self._last_snapshot_included_group = False
        self.entities = []

    def group(self, entity_list=None):
        """Group a list of entities together (default: all)."""
        entities = entity_list
        if not entity_list:
            entities = [entity for entity in self.entities]

        master = entities.pop(0)

        if not entities:
            self.log(
                'Refusing to group only one entity: {0}'.format(master),
                level='WARNING')

        self.call_service(
            'media_player/sonos_join',
            master=master.entity,
            entity_id=[str(e) for e in entities])

        return master

    def register_entity(self, speaker_object):
        """Register a Sonos entity object."""
        if speaker_object in self.entities:
            self.log('Entity already registered; skipping: {0}'.format(
                speaker_object))
            return

        self.entities.append(speaker_object)

    def restore_all(self):
        """Restore the previous snapshot of all entities."""
        self.call_service(
            'media_player/sonos_restore',
            entity_id=[str(e) for e in self.entities],
            with_group=self._last_snapshot_included_group)

    def snapshot_all(self, include_grouping=True):
        """Snapshot all registered entities simultaneously."""
        self._last_snapshot_included_group = include_grouping
        self.call_service(
            'media_player/sonos_snapshot',
            entity_id=[str(e) for e in self.entities],
            with_group=include_grouping)

    def ungroup_all(self):
        """Return all speakers to "individual" status."""
        self.call_service(
            'media_player/sonos_unjoin',
            entity_id=[str(e) for e in self.entities])

tts.py

"""Define an app for working with TTS (over Sonos)."""

# pylint: disable=attribute-defined-outside-init,too-few-public-methods
# pylint: disable=unused-argument

import appdaemon.plugins.hass.hassapi as hass

OPENER_FILE_URL = '/local/tts_opener.mp3'


class TTS(hass.Hass):
    """Define a class to represent the app."""

    # --- INITIALIZERS --------------------------------------------------------
    def initialize(self):
        """Initialize."""
        self._last_spoken_text = None
        self._last_spoken_volume = None
        self.living_room_tv = self.get_app('living_room_tv')
        self.sonos_manager = self.get_app('sonos_manager')

        self.register_endpoint(self._tts_endpoint, 'tts')

    # --- ENDPOINTS -----------------------------------------------------------
    def _tts_endpoint(self, data):
        """Define an API endpoint to handle incoming TTS requests."""
        self.log('Received TTS data: {}'.format(data), level='DEBUG')
        if 'text' not in data:
            self.error('No TTS data provided')
            return '', 502

        self.speak(data['text'])
        response = {"status": "ok", "message": data['text']}
        return response, 200

    # --- CALLBACKS -----------------------------------------------------------
    def _calculate_ending_duration_cb(self, kwargs):
        """Calculate how long the TTS should play."""
        master_sonos_player = kwargs['master_sonos_player']

        self.run_in(
            self._end_cb,
            self.get_state(
                str(master_sonos_player), attribute='media_duration'),
            master_sonos_player=master_sonos_player)

    def _end_cb(self, kwargs):
        """Restore the Sonos to its previous state after speech is done."""
        master_sonos_player = kwargs['master_sonos_player']

        master_sonos_player.play_file(OPENER_FILE_URL)
        self.run_in(self._restore_cb, 3.25)

    def _restore_cb(self, kwargs):
        """Restore the Sonos to its previous state after speech is done."""
        if self.living_room_tv.current_activity_id:
            self.living_room_tv.play()
        self.sonos_manager.restore_all()

    def _speak_cb(self, kwargs):
        """Restore the Sonos to its previous state after speech is done."""
        master_sonos_player = kwargs['master_sonos_player']
        text = kwargs['text']

        self.call_service(
            'tts/amazon_polly_say',
            entity_id=str(master_sonos_player),
            message=text)

        self.run_in(
            self._calculate_ending_duration_cb,
            1,
            master_sonos_player=master_sonos_player)

    # --- HELPERS -------------------------------------------------------------
    def repeat(self):
        """Repeat the last thing that was spoken."""
        if self._last_spoken_text:
            self.log('Repeating over TTS: {0}'.format(self._last_spoken_text))
            self.speak(self._last_spoken_text, self._last_spoken_volume)

    def speak(self, text, volume=0.5):
        """Speak the provided text through the Sonos (pausing as needed)."""
        if self.get_state('input_boolean.mode_do_not_disturb') == 'off':
            self.log('Speaking over TTS: {0}'.format(text))

            self.sonos_manager.snapshot_all()
            master_sonos_player = self.sonos_manager.group()
            master_sonos_player.volume = volume
            master_sonos_player.play_file(OPENER_FILE_URL)

            if self.living_room_tv.current_activity_id:
                self.living_room_tv.pause()

            self.run_in(
                self._speak_cb,
                3.25,
                master_sonos_player=master_sonos_player,
                text=text,
                volume=volume)

            self._last_spoken_text = text
            self._last_spoken_volume = volume