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