Implementation

Essentially a copy of https://github.com/googlesamples/assistant-sdk-python/blob/master/google-assistant-sdk/googlesamples/assistant/grpc/textinput.py

README.md

@@ -1,2 +1,34 @@
 # gassist_text
+
 A Python library for interacting with Google Assistant API via text
+
+## Credits
+
+Uses <https://pypi.org/project/google-assistant-grpc/>. See instructions there how to get `credentials.json`.
+
+Code is essentially a copy of <https://github.com/googlesamples/assistant-sdk-python/blob/master/google-assistant-sdk/googlesamples/assistant/grpc/textinput.py> wrapped in a package.
+
+## Example
+
+```python
+import json
+import google.oauth2.credentials
+with open('/path/to/credentials.json', 'r') as f:
+    credentials = google.oauth2.credentials.Credentials(token=None, **json.load(f))
+
+from gassist_text import TextAssistant
+with TextAssistant(credentials) as assistant:
+    print(assistant.assist('tell me a joke')[0])
+    print(assistant.assist('another one')[0])
+```
+
+## How to run
+
+```sh
+$ python3 -m venv venv
+$ source venv/bin/activate
+$ pip3 install -r requirements.txt
+
+# Run command line interactive tool
+$ python gassist_text/textinput.py
+```

gassist_text/__init__.py

@@ -0,0 +1,4 @@
+"""
+.. include:: ../README.md
+"""
+from .textinput import TextAssistant

gassist_text/assistant_helpers.py

@@ -0,0 +1,54 @@
+# Copyright (C) 2017 Google Inc.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""Helper functions for the Google Assistant API."""
+
+import logging
+
+from google.assistant.embedded.v1alpha2 import embedded_assistant_pb2
+
+
+def log_assist_request_without_audio(assist_request):
+    """Log AssistRequest fields without audio data."""
+    if logging.getLogger().isEnabledFor(logging.DEBUG):
+        resp_copy = embedded_assistant_pb2.AssistRequest()
+        resp_copy.CopyFrom(assist_request)
+        if len(resp_copy.audio_in) > 0:
+            size = len(resp_copy.audio_in)
+            resp_copy.ClearField('audio_in')
+            logging.debug('AssistRequest: audio_in (%d bytes)',
+                          size)
+            return
+        logging.debug('AssistRequest: %s', resp_copy)
+
+
+def log_assist_response_without_audio(assist_response):
+    """Log AssistResponse fields without audio data."""
+    if logging.getLogger().isEnabledFor(logging.DEBUG):
+        resp_copy = embedded_assistant_pb2.AssistResponse()
+        resp_copy.CopyFrom(assist_response)
+        has_audio_data = (resp_copy.HasField('audio_out') and
+                          len(resp_copy.audio_out.audio_data) > 0)
+        if has_audio_data:
+            size = len(resp_copy.audio_out.audio_data)
+            resp_copy.audio_out.ClearField('audio_data')
+            if resp_copy.audio_out.ListFields():
+                logging.debug('AssistResponse: %s audio_data (%d bytes)',
+                              resp_copy,
+                              size)
+            else:
+                logging.debug('AssistResponse: audio_data (%d bytes)',
+                              size)
+            return
+        logging.debug('AssistResponse: %s', resp_copy)

gassist_text/browser_helpers.py

@@ -0,0 +1,33 @@
+# Copyright (C) 2018 Google Inc.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import os.path
+import tempfile
+import webbrowser
+
+ASSISTANT_HTML_FILE = 'google-assistant-sdk-screen-out.html'
+
+
+class SystemBrowser(object):
+    def __init__(self):
+        self.tempdir = tempfile.mkdtemp()
+        self.filename = os.path.join(self.tempdir, ASSISTANT_HTML_FILE)
+
+    def display(self, html):
+        with open(self.filename, 'wb') as f:
+            f.write(html)
+        webbrowser.open(self.filename, new=0)
+
+
+system_browser = SystemBrowser()

gassist_text/textinput.py

@@ -0,0 +1,201 @@
+# Copyright (C) 2017 Google Inc.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""Implements a text client for the Google Assistant Service."""
+# Copied from:
+# https://github.com/googlesamples/assistant-sdk-python/blob/master/google-assistant-sdk/googlesamples/assistant/grpc/textinput.py # noqa
+# Changes:
+# - Renamed class
+# - Simplified constructor:
+#   - Added default values
+#   - Moved creation of the authorized gRPC channel in the constructor
+
+import os
+import logging
+import json
+
+import click
+import google.auth.transport.grpc
+import google.auth.transport.requests
+import google.oauth2.credentials
+
+from google.assistant.embedded.v1alpha2 import (
+    embedded_assistant_pb2,
+    embedded_assistant_pb2_grpc
+)
+
+try:
+    from . import (
+        assistant_helpers,
+        browser_helpers,
+    )
+except (SystemError, ImportError):
+    import assistant_helpers
+    import browser_helpers
+
+
+ASSISTANT_API_ENDPOINT = 'embeddedassistant.googleapis.com'
+DEFAULT_GRPC_DEADLINE = 60 * 3 + 5
+PLAYING = embedded_assistant_pb2.ScreenOutConfig.PLAYING
+
+
+class TextAssistant(object):
+    """Assistant that supports text based conversations.
+
+    Args:
+      credentials: OAuth2 credentials.
+      language_code: language for the conversation.
+      device_model_id: identifier of the device model.
+      device_id: identifier of the registered device instance.
+      display: enable visual display of assistant response.
+      deadline_sec: gRPC deadline in seconds for Google Assistant API call.
+      api_endpoint: Address of Google Assistant API service.
+    """
+
+    def __init__(self, credentials, language_code='en-US', device_model_id='default', device_id='default',
+                 display=False, deadline_sec=DEFAULT_GRPC_DEADLINE, api_endpoint=ASSISTANT_API_ENDPOINT):
+        self.language_code = language_code
+        self.device_model_id = device_model_id
+        self.device_id = device_id
+        self.conversation_state = None
+        # Force reset of first conversation.
+        self.is_new_conversation = True
+        self.display = display
+        # Create an authorized gRPC channel.
+        channel = google.auth.transport.grpc.secure_authorized_channel(
+            credentials, google.auth.transport.requests.Request(), api_endpoint)
+        self.assistant = embedded_assistant_pb2_grpc.EmbeddedAssistantStub(
+            channel
+        )
+        self.deadline = deadline_sec
+
+    def __enter__(self):
+        return self
+
+    def __exit__(self, etype, e, traceback):
+        if e:
+            return False
+
+    def assist(self, text_query):
+        """Send a text request to the Assistant and playback the response.
+        """
+        def iter_assist_requests():
+            config = embedded_assistant_pb2.AssistConfig(
+                audio_out_config=embedded_assistant_pb2.AudioOutConfig(
+                    encoding='LINEAR16',
+                    sample_rate_hertz=16000,
+                    volume_percentage=0,
+                ),
+                dialog_state_in=embedded_assistant_pb2.DialogStateIn(
+                    language_code=self.language_code,
+                    conversation_state=self.conversation_state,
+                    is_new_conversation=self.is_new_conversation,
+                ),
+                device_config=embedded_assistant_pb2.DeviceConfig(
+                    device_id=self.device_id,
+                    device_model_id=self.device_model_id,
+                ),
+                text_query=text_query,
+            )
+            # Continue current conversation with later requests.
+            self.is_new_conversation = False
+            if self.display:
+                config.screen_out_config.screen_mode = PLAYING
+            req = embedded_assistant_pb2.AssistRequest(config=config)
+            assistant_helpers.log_assist_request_without_audio(req)
+            yield req
+
+        text_response = None
+        html_response = None
+        for resp in self.assistant.Assist(iter_assist_requests(),
+                                          self.deadline):
+            assistant_helpers.log_assist_response_without_audio(resp)
+            if resp.screen_out.data:
+                html_response = resp.screen_out.data
+            if resp.dialog_state_out.conversation_state:
+                conversation_state = resp.dialog_state_out.conversation_state
+                self.conversation_state = conversation_state
+            if resp.dialog_state_out.supplemental_display_text:
+                text_response = resp.dialog_state_out.supplemental_display_text
+        return text_response, html_response
+
+
+@click.command()
+@click.option('--api-endpoint', default=ASSISTANT_API_ENDPOINT,
+              metavar='<api endpoint>', show_default=True,
+              help='Address of Google Assistant API service.')
+@click.option('--credentials',
+              metavar='<credentials>', show_default=True,
+              default=os.path.join(click.get_app_dir('google-oauthlib-tool'),
+                                   'credentials.json'),
+              help='Path to read OAuth2 credentials.')
+@click.option('--device-model-id',
+              metavar='<device model id>',
+              required=True,
+              default='default',
+              help=(('Unique device model identifier, '
+                     'if not specifed, it is read from --device-config')))
+@click.option('--device-id',
+              metavar='<device id>',
+              required=True,
+              default='default',
+              help=(('Unique registered device instance identifier, '
+                     'if not specified, it is read from --device-config, '
+                     'if no device_config found: a new device is registered '
+                     'using a unique id and a new device config is saved')))
+@click.option('--lang', show_default=True,
+              metavar='<language code>',
+              default='en-US',
+              help='Language code of the Assistant')
+@click.option('--display', is_flag=True, default=False,
+              help='Enable visual display of Assistant responses in HTML.')
+@click.option('--verbose', '-v', is_flag=True, default=False,
+              help='Verbose logging.')
+@click.option('--grpc-deadline', default=DEFAULT_GRPC_DEADLINE,
+              metavar='<grpc deadline>', show_default=True,
+              help='gRPC deadline in seconds')
+def main(api_endpoint, credentials,
+         device_model_id, device_id, lang, display, verbose,
+         grpc_deadline, *args, **kwargs):
+    # Setup logging.
+    logging.basicConfig(level=logging.DEBUG if verbose else logging.INFO)
+
+    # Load OAuth 2.0 credentials.
+    try:
+        with open(credentials, 'r') as f:
+            credentials = google.oauth2.credentials.Credentials(token=None,
+                                                                **json.load(f))
+            http_request = google.auth.transport.requests.Request()
+            credentials.refresh(http_request)
+    except Exception as e:
+        logging.error('Error loading credentials: %s', e)
+        logging.error('Run google-oauthlib-tool to initialize '
+                      'new OAuth 2.0 credentials.')
+        return
+
+    with TextAssistant(credentials, lang, device_model_id, device_id, display,
+                       grpc_deadline, api_endpoint) as assistant:
+        while True:
+            query = click.prompt('')
+            click.echo('<you> %s' % query)
+            response_text, response_html = assistant.assist(text_query=query)
+            if display and response_html:
+                system_browser = browser_helpers.system_browser
+                system_browser.display(response_html)
+            if response_text:
+                click.echo('<@assistant> %s' % response_text)
+
+
+if __name__ == '__main__':
+    main()

requirements.txt

@@ -0,0 +1,5 @@
+click==8.1.3
+google-assistant-grpc==0.3.0
+google-auth==2.14.1
+protobuf==3.20.3
+requests==2.28.1