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