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
|
#!/usr/bin/env python3
import logging
import sys
import warnings
from typing import NamedTuple, 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",
)
class GoogleResponse(NamedTuple):
success: bool
response: str
audio_url: str
audio_transcription: Optional[str] # None if not available.
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).
"""
logging.debug(f"Asking google: '{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()
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(f"Google said: '{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(f"Transcription: '{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,
)
sys.exit(-1)
|
