Source code for platypush.plugins.assistant.google

import json
import os
from typing import Optional

from platypush.config import Config
from platypush.plugins import RunnablePlugin, action
from platypush.plugins.assistant import AssistantPlugin


[docs] class AssistantGooglePlugin(AssistantPlugin, RunnablePlugin): """ Google Assistant plugin. This plugin allows you to run the Google Assistant _directly_ on your device. It requires you to have an audio microphone and a speaker connected to the device. If you have multiple sound devices, you can specify which one(s) to use for input and output through a ``~/.asoundrc`` configuration file like this: .. code-block:: text pcm.!default { type asym playback.pcm { type plug slave.pcm "hw:0,0" } capture.pcm { type plug slave.pcm "hw:1,0" } } You can use ``aplay -l`` and ``arecord -l`` to respectively list the detected audio output and input devices with their indices. If you are using PulseAudio instead of bare ALSA, then you can: 1. Use the ``pavucontrol`` (GUI) tool to select the audio input and output devices and volumes for the assistant. 2. Use a program like ``pamix`` (ncurses) or ``pamixer`` (CLI). 3. Run the ``pactl list sources`` and ``pactl list sinks`` commands to respectively list the detected audio input and output devices. Take note of their name, and specify which ones the assistant should use by starting the application with the right ``PULSE_SOURCE`` and ``PULSE_SINK`` environment variables. .. warning:: The Google Assistant library used by this backend has been deprecated by Google: https://developers.google.com/assistant/sdk/reference/library/python/. This integration still works on all of my devices, but its future functionality is not guaranteed - Google may decide to turn off the API, the library may no longer be built against new architectures and it's unlikely to be updated. .. note:: Since the Google Assistant library hasn't been updated in several years, some of its dependencies are quite old and may break more recent Python installations. Please refer to the comments in the `manifest file <https://git.platypush.tech/platypush/platypush/src/branch/master/platypush/plugins/assistant/google/manifest.json>`_. for more information on how to install the required dependencies, if the automated ways fail. """ _entity_name = 'Google Assistant' _default_credentials_files = ( os.path.join(Config.get_workdir(), 'credentials', 'google', 'assistant.json'), os.path.join( os.path.expanduser('~/.config'), 'google-oauthlib-tool', 'credentials.json' ), )
[docs] def __init__( self, credentials_file: Optional[str] = None, device_model_id: str = 'Platypush', **kwargs, ): """ :param credentials_file: Path to the Google OAuth credentials file. See https://developers.google.com/assistant/sdk/guides/library/python/embed/install-sample#generate_credentials for instructions to get your own credentials file. By default, it will search for the credentials file under: * ``~/.config/google-oauthlib-tool/credentials.json``: default location supported by the Google Assistant library * ``<PLATYPUSH_WORKDIR>/credentials/google/assistant.json``: recommended location, under the Platypush working directory. :param device_model_id: The device model ID that identifies the device where the assistant is running (default: Platypush). It can be a custom string. """ super().__init__(**kwargs) self._credentials_file = credentials_file self.device_model_id = device_model_id self.credentials = None self._assistant = None self.logger.info('Initialized Google Assistant plugin')
@property def credentials_file(self) -> str: if self._credentials_file: return os.path.abspath(os.path.expanduser(self._credentials_file)) f = None for default_file in self._default_credentials_files: f = default_file if os.path.isfile(default_file): break assert f, 'No credentials_file provided and no default file found' return f @property def assistant(self): if not self._assistant: self.logger.warning('The assistant is not running') return return self._assistant def _process_event(self, event): from google.assistant.library.event import EventType, AlertType self.logger.info('Received assistant event: %s', event) if event.type == EventType.ON_CONVERSATION_TURN_STARTED: self._on_conversation_start() elif event.type == EventType.ON_CONVERSATION_TURN_FINISHED: if not event.args.get('with_follow_on_turn'): self._on_conversation_end() elif event.type == EventType.ON_CONVERSATION_TURN_TIMEOUT: self._on_conversation_timeout() elif event.type == EventType.ON_NO_RESPONSE: self._on_no_response() elif ( hasattr(EventType, 'ON_RENDER_RESPONSE') and event.type == EventType.ON_RENDER_RESPONSE ): self._on_response_render_start(event.args.get('text')) elif ( hasattr(EventType, 'ON_RESPONDING_STARTED') and event.type == EventType.ON_RESPONDING_STARTED and event.args.get('is_error_response') is True ): self.logger.warning('Assistant response error: %s', json.dumps(event.args)) elif event.type == EventType.ON_RECOGNIZING_SPEECH_FINISHED: self._on_speech_recognized(event.args.get('text')) elif event.type == EventType.ON_ALERT_STARTED: if event.args.get('alert_type') == AlertType.ALARM: self._on_alarm_start() elif event.args.get('alert_type') == AlertType.TIMER: self._on_timer_start() else: self._on_alert_start() elif event.type == EventType.ON_ALERT_FINISHED: if event.args.get('alert_type') == AlertType.ALARM: self._on_alarm_end() elif event.args.get('alert_type') == AlertType.TIMER: self._on_timer_end() else: self._on_alert_end() elif event.type == EventType.ON_ASSISTANT_ERROR: if event.args.get('is_fatal'): raise RuntimeError(f'Fatal assistant error: {json.dumps(event.args)}') self.logger.warning('Assistant error: %s', json.dumps(event.args)) elif event.type == EventType.ON_MUTED_CHANGED: self._on_mute_changed(event.args.get('is_muted', False))
[docs] @action def start_conversation(self, *_, **__): """ Programmatically start a conversation with the assistant """ if self.assistant: self.assistant.start_conversation()
def _stop_conversation(self, *_, **__): """ Programmatically stop a running conversation with the assistant """ if self.assistant: self.assistant.stop_conversation()
[docs] @action def mute(self, *_, **__): """ Mute the microphone. Alias for :meth:`.set_mic_mute` with ``muted=True``. """ self.set_mic_mute(muted=True)
[docs] @action def unmute(self, *_, **__): """ Unmute the microphone. Alias for :meth:`.set_mic_mute` with ``muted=False``. """ self.set_mic_mute(muted=False)
[docs] @action def set_mic_mute(self, muted: bool): """ Programmatically mute/unmute the microphone. :param muted: Set to True or False. """ if self.assistant: self.assistant.set_mic_mute(muted) if muted: self._on_mute() else: self._on_unmute()
[docs] @action def toggle_mute(self, *_, **__): """ Toggle the mic mute state. """ self.set_mic_mute(muted=not self._is_muted)
[docs] @action def toggle_mic_mute(self): """ Deprecated alias for :meth:`.toggle_mute`. """ return self.toggle_mute()
[docs] @action def send_text_query(self, *_, query: str, **__): """ Send a text query to the assistant. This is equivalent to saying something to the assistant. :param query: Query to be sent. """ if self.assistant: self.assistant.send_text_query(query)
[docs] def main(self): import google.oauth2.credentials from google.assistant.library import Assistant last_sleep = 0 while not self.should_stop(): try: with open(self.credentials_file, 'r') as f: self.credentials = google.oauth2.credentials.Credentials( token=None, **json.load(f) ) except Exception as e: self.logger.error( 'Error while loading Google Assistant credentials: %s', e ) self.logger.info( 'Please follow the instructions at ' 'https://developers.google.com/assistant/sdk/guides/library/python/embed/install-sample' '#generate_credentials to get your own credentials file' ) self.logger.exception(e) break try: with Assistant( self.credentials, self.device_model_id ) as self._assistant: self.publish_entities([self]) for event in self._assistant.start(): last_sleep = 0 if not self.is_detecting(): self.logger.info( 'Assistant event received but detection is currently paused' ) continue self._process_event(event) except Exception as e: self.logger.exception(e) sleep_secs = min(60, max(5, last_sleep * 2)) self.logger.warning( 'Restarting the assistant in %d seconds after an unrecoverable error', sleep_secs, ) self.wait_stop(sleep_secs) last_sleep = sleep_secs continue
[docs] def stop(self): try: self.stop_conversation() except RuntimeError: pass if self._assistant: del self._assistant self._assistant = None super().stop()
# vim:sw=4:ts=4:et: