google_assistant.py

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