You bet – it’s long, but it relates to the self.register_endpoint(self._eta_endpoint, 'eta')
call. The App
superclass correctly inherits from hass.Hass
:
"""Define an app for working with TTS (over Sonos)."""
# pylint: disable=attribute-defined-outside-init,too-few-public-methods
# pylint: disable=unused-argument,too-many-arguments
from typing import Tuple, Union
from app import App
OPENER_FILE_URL = '/local/tts_opener.mp3'
TTS_HEADING_HOME = ('{0} is headed home and should arrive in '
'approximately {1} minutes.')
class TTS(App):
"""Define a class to represent the app."""
@property
def briefings(self) -> list:
"""Define a property to get saved briefings."""
return self._briefings
# --- INITIALIZERS --------------------------------------------------------
def initialize(self) -> None:
"""Initialize."""
self._briefings = []
self._last_spoken_text = None
self._last_spoken_volume = None
self.living_room_tv = self.get_app('living_room_tv')
self.presence_manager = self.get_app('presence_manager')
self.sonos_manager = self.get_app('sonos_manager')
self.utilities = self.get_app('utilities')
self.register_endpoint(self._emergency_endpoint, 'emergency')
self.register_endpoint(self._eta_endpoint, 'eta')
self.register_endpoint(self._tts_endpoint, 'tts')
# --- ENDPOINTS -----------------------------------------------------------
def _emergency_endpoint(self, data: dict) -> Tuple[str, int]:
"""Define an endpoint to alert us of an emergency."""
if self.presence_manager.anyone(
self.presence_manager.HOME_STATES.Home):
try:
name = data['name'].title()
self.log('Emergency Notification from {0}'.format(name))
statement = 'Please call {0} as soon as possible.'.format(name)
self.speak(statement)
return {"status": "ok", "message": statement}, 200
except KeyError:
self.error("Unknown person: {0}".format(name))
return '', 502
def _eta_endpoint(self, data: dict) -> Tuple[str, int]:
"""Define an endpoint to send Aaron's ETA."""
if self.presence_manager.anyone(
self.presence_manager.HOME_STATES.Home):
try:
key = data['person']
name = key.title()
eta = self.get_state('sensor.{0}_travel_time'.format(key))
statement = TTS_HEADING_HOME.format(name, eta)
self.log("Announcing {0}'s ETA: {1} minutes".format(name, eta))
self.speak(statement)
return {"status": "ok", "message": statement}, 200
except KeyError:
self.error("Can't announce unknown person: {0}".format(name))
return '', 502
def _tts_endpoint(self, data: dict) -> Tuple[str, int]:
"""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'])
return {"status": "ok", "message": data['text']}, 200
# --- CALLBACKS -----------------------------------------------------------
def _calculate_ending_duration_cb(self, kwargs: dict) -> None:
"""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: dict) -> None:
"""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: dict) -> None:
"""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.ungroup_all()
self.sonos_manager.restore_all()
def _speak_cb(self, kwargs: dict) -> None:
"""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)
# --- APP API -------------------------------------------------------------
def repeat(self) -> None:
"""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: str, volume: float = 0.5) -> None:
"""Speak the provided text through the Sonos (pausing as needed)."""
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='Good {0}. {1}'.format(self.utilities.relative_time_of_day(),
text),
volume=volume)
self._last_spoken_text = text
self._last_spoken_volume = volume