Initial revision

1 files changed, 89 insertions, 0 deletions

diff --git a/google_assistant.py b/google_assistant.py
new file mode 100644
index 0000000..500a909
--- /dev/null
+++ b/google_assistant.py
@@ -0,0 +1,89 @@
+#!/usr/bin/env python3
+
+import logging
+from typing import NamedTuple
+
+import requests
+import speech_recognition as sr  # type: ignore
+
+import config
+
+logger = logging.getLogger(__name__)
+
+parser = config.add_commandline_args(
+    f"Google Assistant ({__file__})",
+    "Args related to contacting the Google Assistant",
+)
+parser.add_argument(
+    "--google_assistant_bridge",
+    type=str,
+    default="http://kiosk.house:3000",
+    metavar="URL",
+    help="How to contact the Google Assistant bridge"
+)
+parser.add_argument(
+    "--google_assistant_username",
+    type=str,
+    metavar="GOOGLE_ACCOUNT",
+    default="scott.gasch",
+    help="The user account for talking to Google Assistant"
+)
+
+
+class GoogleResponse(NamedTuple):
+    success: bool
+    response: str
+    audio_url: str
+    audio_transcription: str
+
+    def __repr__(self):
+        return f"""
+success: {self.success}
+response: {self.response}
+audio_transcription: {self.audio_transcription}
+audio_url: {self.audio_url}"""
+
+
+def tell_google(cmd: str, *, recognize_speech=True) -> GoogleResponse:
+    return ask_google(cmd, recognize_speech=recognize_speech)
+
+
+def ask_google(cmd: str, *, recognize_speech=True) -> GoogleResponse:
+    payload = {
+        "command": cmd,
+        "user": config.config['google_assistant_username'],
+    }
+    url = f"{config.config['google_assistant_bridge']}/assistant"
+    r = requests.post(url, json=payload)
+    success = False
+    response = ""
+    audio = ""
+    audio_transcription = ""
+    if r.status_code == 200:
+        j = r.json()
+        success = bool(j["success"])
+        response = j["response"] if success else j["error"]
+        audio = f"{config.config['google_assistant_bridge']}{j['audio']}"
+        if recognize_speech:
+            recognizer = sr.Recognizer()
+            r = requests.get(audio)
+            if r.status_code == 200:
+                raw = r.content
+                speech = sr.AudioData(
+                    frame_data=raw,
+                    sample_rate=24000,
+                    sample_width=2,
+                )
+                audio_transcription = recognizer.recognize_google(
+                    speech,
+                )
+    else:
+        logger.error(
+            f'HTTP request to {url} with {payload} failed; code {r.status_code}'
+        )
+    return GoogleResponse(
+        success=success,
+        response=response,
+        audio_url=audio,
+        audio_transcription=audio_transcription,
+    )