google_assistant.py

   1 #!/usr/bin/env python3
   2
   3 import logging
   4 import sys
   5 import warnings
   6 from typing import NamedTuple, Optional
   7
   8 import requests
   9 import speech_recognition as sr  # type: ignore
  10
  11 import config
  12
  13 logger = logging.getLogger(__name__)
  14
  15 parser = config.add_commandline_args(
  16     f"Google Assistant ({__file__})",
  17     "Args related to contacting the Google Assistant",
  18 )
  19 parser.add_argument(
  20     "--google_assistant_bridge",
  21     type=str,
  22     default="http://kiosk.house:3000",
  23     metavar="URL",
  24     help="How to contact the Google Assistant bridge",
  25 )
  26 parser.add_argument(
  27     "--google_assistant_username",
  28     type=str,
  29     metavar="GOOGLE_ACCOUNT",
  30     default="scott.gasch",
  31     help="The user account for talking to Google Assistant",
  32 )
  33
  34
  35 class GoogleResponse(NamedTuple):
  36     success: bool
  37     response: str
  38     audio_url: str
  39     audio_transcription: Optional[str]  # None if not available.
  40
  41     def __repr__(self):
  42         return f"""
  43 success: {self.success}
  44 response: {self.response}
  45 audio_transcription: {self.audio_transcription}
  46 audio_url: {self.audio_url}"""
  47
  48
  49 def tell_google(cmd: str, *, recognize_speech=True) -> GoogleResponse:
  50     """Alias for ask_google."""
  51     return ask_google(cmd, recognize_speech=recognize_speech)
  52
  53
  54 def ask_google(cmd: str, *, recognize_speech=True) -> GoogleResponse:
  55     """Send a command string to Google via the google_assistant_bridge as the
  56     user google_assistant_username and return the response.  If recognize_speech
  57     is True, perform speech recognition on the audio response from Google so as
  58     to translate it into text (best effort, YMMV).
  59     """
  60     logging.debug(f"Asking google: '{cmd}'")
  61     payload = {
  62         "command": cmd,
  63         "user": config.config['google_assistant_username'],
  64     }
  65     url = f"{config.config['google_assistant_bridge']}/assistant"
  66     r = requests.post(url, json=payload)
  67     success = False
  68     response = ""
  69     audio = ""
  70     audio_transcription: Optional[str] = ""
  71     if r.status_code == 200:
  72         j = r.json()
  73         logger.debug(j)
  74         success = bool(j["success"])
  75         response = j["response"] if success else j["error"]
  76         if success:
  77             logger.debug('Google request succeeded.')
  78             if len(response) > 0:
  79                 logger.debug(f"Google said: '{response}'")
  80         audio = f"{config.config['google_assistant_bridge']}{j['audio']}"
  81         if recognize_speech:
  82             recognizer = sr.Recognizer()
  83             r = requests.get(audio)
  84             if r.status_code == 200:
  85                 raw = r.content
  86                 speech = sr.AudioData(
  87                     frame_data=raw,
  88                     sample_rate=24000,
  89                     sample_width=2,
  90                 )
  91                 try:
  92                     audio_transcription = recognizer.recognize_google(
  93                         speech,
  94                     )
  95                     logger.debug(f"Transcription: '{audio_transcription}'")
  96                 except sr.UnknownValueError as e:
  97                     logger.exception(e)
  98                     msg = 'Unable to parse Google assistant\'s response.'
  99                     logger.warning(msg)
 100                     warnings.warn(msg, stacklevel=3)
 101                     audio_transcription = None
 102         return GoogleResponse(
 103             success=success,
 104             response=response,
 105             audio_url=audio,
 106             audio_transcription=audio_transcription,
 107         )
 108     else:
 109         message = f'HTTP request to {url} with {payload} failed; code {r.status_code}'
 110         logger.error(message)
 111         return GoogleResponse(
 112             success=False,
 113             response=message,
 114             audio_url=audio,
 115             audio_transcription=audio_transcription,
 116         )
 117         sys.exit(-1)