textinput.py

   1
   2 import os
   3 import logging
   4 import json
   5
   6 import click
   7 import google.auth.transport.grpc
   8 import google.auth.transport.requests
   9 import google.oauth2.credentials
  10
  11 from google.assistant.embedded.v1alpha2 import (
  12     embedded_assistant_pb2,
  13     embedded_assistant_pb2_grpc
  14 )
  15
  16 try:
  17     from . import (
  18         assistant_helpers,
  19         browser_helpers,
  20     )
  21 except (SystemError, ImportError):
  22     import assistant_helpers
  23     import browser_helpers
  24
  25
  26 ASSISTANT_API_ENDPOINT = 'embeddedassistant.googleapis.com'
  27 DEFAULT_GRPC_DEADLINE = 60 * 3 + 5
  28 PLAYING = embedded_assistant_pb2.ScreenOutConfig.PLAYING
  29
  30
  31 class SampleTextAssistant(object):
  32     """Sample Assistant that supports text based conversations.
  33     Args:
  34       language_code: language for the conversation.
  35       device_model_id: identifier of the device model.
  36       device_id: identifier of the registered device instance.
  37       display: enable visual display of assistant response.
  38       channel: authorized gRPC channel for connection to the
  39         Google Assistant API.
  40       deadline_sec: gRPC deadline in seconds for Google Assistant API call.
  41     """
  42
  43     def __init__(self, language_code, device_model_id, device_id,
  44                  display, channel, deadline_sec):
  45         self.language_code = language_code
  46         self.device_model_id = device_model_id
  47         self.device_id = device_id
  48         self.conversation_state = None
  49         # Force reset of first conversation.
  50         self.is_new_conversation = True
  51         self.display = display
  52         self.assistant = embedded_assistant_pb2_grpc.EmbeddedAssistantStub(
  53             channel
  54         )
  55         self.deadline = deadline_sec
  56
  57     def __enter__(self):
  58         return self
  59
  60     def __exit__(self, etype, e, traceback):
  61         if e:
  62             return False
  63
  64     def assist(self, text_query):
  65         """Send a text request to the Assistant and playback the response.
  66         """
  67         def iter_assist_requests():
  68             config = embedded_assistant_pb2.AssistConfig(
  69                 audio_out_config=embedded_assistant_pb2.AudioOutConfig(
  70                     encoding='LINEAR16',
  71                     sample_rate_hertz=16000,
  72                     volume_percentage=0,
  73                 ),
  74                 dialog_state_in=embedded_assistant_pb2.DialogStateIn(
  75                     language_code=self.language_code,
  76                     conversation_state=self.conversation_state,
  77                     is_new_conversation=self.is_new_conversation,
  78                 ),
  79                 device_config=embedded_assistant_pb2.DeviceConfig(
  80                     device_id=self.device_id,
  81                     device_model_id=self.device_model_id,
  82                 ),
  83                 text_query=text_query,
  84             )
  85             # Continue current conversation with later requests.
  86             self.is_new_conversation = False
  87             if self.display:
  88                 config.screen_out_config.screen_mode = PLAYING
  89             req = embedded_assistant_pb2.AssistRequest(config=config)
  90             assistant_helpers.log_assist_request_without_audio(req)
  91             yield req
  92
  93         text_response = None
  94         html_response = None
  95         for resp in self.assistant.Assist(iter_assist_requests(),
  96                                           self.deadline):
  97             assistant_helpers.log_assist_response_without_audio(resp)
  98             if resp.screen_out.data:
  99                 html_response = resp.screen_out.data
 100             if resp.dialog_state_out.conversation_state:
 101                 conversation_state = resp.dialog_state_out.conversation_state
 102                 self.conversation_state = conversation_state
 103             if resp.dialog_state_out.supplemental_display_text:
 104                 text_response = resp.dialog_state_out.supplemental_display_text
 105         return text_response, html_response
 106
 107
 108 @click.command()
 109 @click.option('--api-endpoint', default=ASSISTANT_API_ENDPOINT,
 110               metavar='<api endpoint>', show_default=True,
 111               help='Address of Google Assistant API service.')
 112 @click.option('--credentials',
 113               metavar='<credentials>', show_default=True,
 114               default=os.path.join(click.get_app_dir('google-oauthlib-tool'),
 115                                    'credentials.json'),
 116               help='Path to read OAuth2 credentials.')
 117 @click.option('--device-model-id',
 118               metavar='<device model id>',
 119               required=True,
 120               help=(('Unique device model identifier, '
 121                      'if not specifed, it is read from --device-config')))
 122 @click.option('--device-id',
 123               metavar='<device id>',
 124               required=True,
 125               help=(('Unique registered device instance identifier, '
 126                      'if not specified, it is read from --device-config, '
 127                      'if no device_config found: a new device is registered '
 128                      'using a unique id and a new device config is saved')))
 129 @click.option('--lang', show_default=True,
 130               metavar='<language code>',
 131               default='en-US',
 132               help='Language code of the Assistant')
 133 @click.option('--display', is_flag=True, default=False,
 134               help='Enable visual display of Assistant responses in HTML.')
 135 @click.option('--verbose', '-v', is_flag=True, default=False,
 136               help='Verbose logging.')
 137 @click.option('--grpc-deadline', default=DEFAULT_GRPC_DEADLINE,
 138               metavar='<grpc deadline>', show_default=True,
 139               help='gRPC deadline in seconds')
 140 def main(api_endpoint, credentials,
 141          device_model_id, device_id, lang, display, verbose,
 142          grpc_deadline, *args, **kwargs):
 143     # Setup logging.
 144     logging.basicConfig(level=logging.DEBUG if verbose else logging.INFO)
 145
 146     # Load OAuth 2.0 credentials.
 147     try:
 148         with open(credentials, 'r') as f:
 149             credentials = google.oauth2.credentials.Credentials(token=None,
 150                                                                 **json.load(f))
 151             http_request = google.auth.transport.requests.Request()
 152             credentials.refresh(http_request)
 153     except Exception as e:
 154         logging.error('Error loading credentials: %s', e)
 155         logging.error('Run google-oauthlib-tool to initialize '
 156                       'new OAuth 2.0 credentials.')
 157         return
 158
 159     # Create an authorized gRPC channel.
 160     grpc_channel = google.auth.transport.grpc.secure_authorized_channel(
 161         credentials, http_request, api_endpoint)
 162     logging.info('Connecting to %s', api_endpoint)
 163
 164     with SampleTextAssistant(lang, device_model_id, device_id, display,
 165                              grpc_channel, grpc_deadline) as assistant:
 166         while True:
 167             query = click.prompt('')
 168             click.echo('<you> %s' % query)
 169             response_text, response_html = assistant.assist(text_query=query)
 170             if display and response_html:
 171                 system_browser = browser_helpers.system_browser
 172                 system_browser.display(response_html)
 173             if response_text:
 174                 click.echo('<@assistant> %s' % response_text)
 175
 176
 177 if __name__ == '__main__':
 178     main()