2018-07-30 22:08:06 +02:00
"""
. . moduleauthor : : Fabio Manganiello < blacklight86 @gmail.com >
. . license : : MIT
"""
import concurrent
2018-03-20 23:34:36 +01:00
import json
2018-06-08 17:10:05 +02:00
import logging
2018-03-20 23:34:36 +01:00
import os
import grpc
import google . auth . transport . grpc
import google . auth . transport . requests
import google . oauth2 . credentials
2018-07-30 22:08:06 +02:00
from google . assistant . embedded . v1alpha2 import embedded_assistant_pb2 , embedded_assistant_pb2_grpc
2018-03-20 23:34:36 +01:00
import googlesamples . assistant . grpc . audio_helpers as audio_helpers
import googlesamples . assistant . grpc . device_helpers as device_helpers
import googlesamples . assistant . grpc . assistant_helpers as assistant_helpers
from tenacity import retry , stop_after_attempt , retry_if_exception
from platypush . backend import Backend
from platypush . message . event . assistant import \
ConversationStartEvent , ConversationEndEvent , SpeechRecognizedEvent
2018-06-06 20:09:18 +02:00
2018-03-20 23:34:36 +01:00
class AssistantGooglePushtotalkBackend ( Backend ) :
2018-06-26 00:16:39 +02:00
"""
Google Assistant pushtotalk backend . Instead of listening for the " OK
Google " hotword like the assistant.google backend, this implementation
programmatically starts a conversation upon start_conversation ( ) method
call . Use this backend on devices that don ' t have an Assistant SDK package
( e . g . arm6 devices like the RaspberryPi Zero or the RaspberryPi 1 ) .
Triggers :
* : class : ` platypush . message . event . assistant . ConversationStartEvent ` when a new conversation starts
* : class : ` platypush . message . event . assistant . SpeechRecognizedEvent ` when a new voice command is recognized
* : class : ` platypush . message . event . assistant . ConversationEndEvent ` when a new conversation ends
Requires :
* * * tenacity * * ( ` ` pip install tenacity ` ` )
* * * grpc * * ( ` ` pip install grpc ` ` )
* * * google - assistant - grpc * * ( ` ` pip install google - assistant - grpc ` ` )
"""
2018-03-20 23:34:36 +01:00
api_endpoint = ' embeddedassistant.googleapis.com '
audio_sample_rate = audio_helpers . DEFAULT_AUDIO_SAMPLE_RATE
audio_sample_width = audio_helpers . DEFAULT_AUDIO_SAMPLE_WIDTH
audio_iter_size = audio_helpers . DEFAULT_AUDIO_ITER_SIZE
audio_block_size = audio_helpers . DEFAULT_AUDIO_DEVICE_BLOCK_SIZE
audio_flush_size = audio_helpers . DEFAULT_AUDIO_DEVICE_FLUSH_SIZE
grpc_deadline = 60 * 3 + 5
2018-07-30 22:08:06 +02:00
def __init__ ( self , * args ,
credentials_file = os . path . join (
os . path . expanduser ( ' ~ ' ) , ' .config ' ,
' google-oauthlib-tool ' , ' credentials.json ' ) ,
device_config = os . path . join (
os . path . expanduser ( ' ~ ' ) , ' .config ' , ' googlesamples-assistant ' ,
' device_config.json ' ) ,
lang = ' en-US ' ,
conversation_start_fifo = os . path . join ( os . path . sep , ' tmp ' , ' pushtotalk.fifo ' ) ,
* * kwargs ) :
2018-06-26 00:16:39 +02:00
"""
: param credentials_file : Path to the Google OAuth credentials file ( default : ~ / . config / google - oauthlib - tool / credentials . json ) . See https : / / developers . google . com / assistant / sdk / guides / library / python / embed / install - sample #generate_credentials for how to get your own credentials file.
: type credentials_file : str
: param device_config : Path to device_config . json . Register your device ( see https : / / developers . google . com / assistant / sdk / guides / library / python / embed / register - device ) and create a project , then run the pushtotalk . py script from googlesamples to create your device_config . json
: type device_config : str
: param lang : Assistant language ( default : en - US )
: type lang : str
2018-03-20 23:34:36 +01:00
"""
super ( ) . __init__ ( * args , * * kwargs )
self . lang = lang
self . credentials_file = credentials_file
self . device_config = device_config
self . conversation_start_fifo = conversation_start_fifo
2018-06-08 17:17:08 +02:00
self . assistant = None
2018-03-20 23:34:36 +01:00
try :
os . mkfifo ( self . conversation_start_fifo )
except FileExistsError :
pass
with open ( self . device_config ) as f :
device = json . load ( f )
self . device_id = device [ ' id ' ]
self . device_model_id = device [ ' model_id ' ]
# Load OAuth 2.0 credentials.
try :
with open ( self . credentials_file , ' r ' ) as f :
credentials = google . oauth2 . credentials . Credentials ( token = None ,
* * json . load ( f ) )
http_request = google . auth . transport . requests . Request ( )
credentials . refresh ( http_request )
2018-07-30 22:08:06 +02:00
except Exception as ex :
self . logger . error ( ' Error loading credentials: %s ' , str ( ex ) )
2018-06-07 09:08:32 +02:00
self . logger . error ( ' Run google-oauthlib-tool to initialize '
2018-07-30 22:08:06 +02:00
' new OAuth 2.0 credentials. ' )
2018-03-20 23:34:36 +01:00
raise
# Create an authorized gRPC channel.
self . grpc_channel = google . auth . transport . grpc . secure_authorized_channel (
credentials , http_request , self . api_endpoint )
2018-06-07 09:08:32 +02:00
self . logger . info ( ' Connecting to %s ' , self . api_endpoint )
2018-03-20 23:34:36 +01:00
# Configure audio source and sink.
audio_device = None
audio_source = audio_device = (
audio_device or audio_helpers . SoundDeviceStream (
sample_rate = self . audio_sample_rate ,
sample_width = self . audio_sample_width ,
block_size = self . audio_block_size ,
flush_size = self . audio_flush_size
)
)
audio_sink = audio_device = (
audio_device or audio_helpers . SoundDeviceStream (
sample_rate = self . audio_sample_rate ,
sample_width = self . audio_sample_width ,
block_size = self . audio_block_size ,
flush_size = self . audio_flush_size
)
)
# Create conversation stream with the given audio source and sink.
self . conversation_stream = audio_helpers . ConversationStream (
source = audio_source ,
sink = audio_sink ,
iter_size = self . audio_iter_size ,
sample_width = self . audio_sample_width ,
)
self . device_handler = device_helpers . DeviceRequestHandler ( self . device_id )
def start_conversation ( self ) :
2018-06-26 00:16:39 +02:00
""" Start a conversation """
2018-03-20 23:34:36 +01:00
if self . assistant :
with open ( self . conversation_start_fifo , ' w ' ) as f :
f . write ( ' 1 ' )
def stop_conversation ( self ) :
2018-06-26 00:16:39 +02:00
""" Stop a conversation """
2018-03-20 23:34:36 +01:00
if self . assistant :
self . conversation_stream . stop_playback ( )
2018-03-24 03:05:46 +01:00
self . bus . post ( ConversationEndEvent ( ) )
2018-03-20 23:34:36 +01:00
2018-03-24 03:05:46 +01:00
def on_conversation_start ( self ) :
2018-07-30 22:08:06 +02:00
""" Conversation start handler """
2018-03-24 03:05:46 +01:00
self . bus . post ( ConversationStartEvent ( ) )
def on_conversation_end ( self ) :
2018-07-30 22:08:06 +02:00
""" Conversation end handler """
2018-03-24 03:05:46 +01:00
self . bus . post ( ConversationEndEvent ( ) )
def on_speech_recognized ( self , speech ) :
2018-07-30 22:08:06 +02:00
""" Speech recognized handler """
2018-03-24 03:05:46 +01:00
self . bus . post ( SpeechRecognizedEvent ( phrase = speech ) )
2018-03-20 23:34:36 +01:00
def run ( self ) :
2018-07-30 22:08:06 +02:00
""" Backend executor """
2018-03-20 23:34:36 +01:00
super ( ) . run ( )
with SampleAssistant ( self . lang , self . device_model_id , self . device_id ,
2018-07-30 22:08:06 +02:00
self . conversation_stream ,
self . grpc_channel , self . grpc_deadline ,
self . device_handler ,
on_conversation_start = self . on_conversation_start ,
on_conversation_end = self . on_conversation_end ,
on_speech_recognized = self . on_speech_recognized ) as self . assistant :
2018-03-20 23:34:36 +01:00
while not self . should_stop ( ) :
with open ( self . conversation_start_fifo , ' r ' ) as f :
2018-07-30 22:08:06 +02:00
f . read ( )
2018-03-20 23:34:36 +01:00
2018-06-07 09:08:32 +02:00
self . logger . info ( ' Received conversation start event ' )
2018-03-20 23:34:36 +01:00
continue_conversation = True
while continue_conversation :
( user_request , continue_conversation ) = self . assistant . assist ( )
2018-07-30 22:08:06 +02:00
self . logger ( ' User request: {} ' . format ( user_request ) )
2018-03-20 23:34:36 +01:00
2018-03-24 03:05:46 +01:00
self . on_conversation_end ( )
2018-03-20 23:34:36 +01:00
2018-07-30 22:08:06 +02:00
class SampleAssistant :
2018-03-20 23:34:36 +01:00
""" Sample Assistant that supports conversations and device actions.
Args :
device_model_id : identifier of the device model .
device_id : identifier of the registered device instance .
2018-07-16 22:12:02 +02:00
conversation_stream ( ConversationStream ) : audio stream for recording query and playing back assistant answer .
channel : authorized gRPC channel for connection to the Google Assistant API .
2018-03-20 23:34:36 +01:00
deadline_sec : gRPC deadline in seconds for Google Assistant API call .
device_handler : callback for device actions .
"""
END_OF_UTTERANCE = embedded_assistant_pb2 . AssistResponse . END_OF_UTTERANCE
DIALOG_FOLLOW_ON = embedded_assistant_pb2 . DialogStateOut . DIALOG_FOLLOW_ON
CLOSE_MICROPHONE = embedded_assistant_pb2 . DialogStateOut . CLOSE_MICROPHONE
def __init__ ( self , language_code , device_model_id , device_id ,
conversation_stream ,
2018-03-24 03:05:46 +01:00
channel , deadline_sec , device_handler ,
on_conversation_start = None ,
on_conversation_end = None ,
on_speech_recognized = None ) :
2018-03-20 23:34:36 +01:00
self . language_code = language_code
self . device_model_id = device_model_id
self . device_id = device_id
self . conversation_stream = conversation_stream
2018-06-08 17:17:08 +02:00
self . logger = logging . getLogger ( __name__ )
2018-03-20 23:34:36 +01:00
2018-03-24 03:05:46 +01:00
self . on_conversation_start = on_conversation_start
self . on_conversation_end = on_conversation_end
self . on_speech_recognized = on_speech_recognized
2018-03-20 23:34:36 +01:00
# Opaque blob provided in AssistResponse that,
# when provided in a follow-up AssistRequest,
# gives the Assistant a context marker within the current state
# of the multi-Assist()-RPC "conversation".
# This value, along with MicrophoneMode, supports a more natural
# "conversation" with the Assistant.
self . conversation_state = None
# Create Google Assistant API gRPC client.
self . assistant = embedded_assistant_pb2_grpc . EmbeddedAssistantStub (
channel
)
self . deadline = deadline_sec
self . device_handler = device_handler
def __enter__ ( self ) :
return self
def __exit__ ( self , etype , e , traceback ) :
if e :
return False
self . conversation_stream . close ( )
2018-07-30 22:08:06 +02:00
return True
2018-03-20 23:34:36 +01:00
2018-07-30 22:08:06 +02:00
@staticmethod
2018-03-20 23:34:36 +01:00
def is_grpc_error_unavailable ( e ) :
2018-07-30 22:08:06 +02:00
""" Returns True if the gRPC is not available """
2018-03-20 23:34:36 +01:00
is_grpc_error = isinstance ( e , grpc . RpcError )
if is_grpc_error and ( e . code ( ) == grpc . StatusCode . UNAVAILABLE ) :
2018-07-30 22:08:06 +02:00
print ( ' grpc unavailable error: {} ' . format ( e ) )
2018-03-20 23:34:36 +01:00
return True
return False
@retry ( reraise = True , stop = stop_after_attempt ( 3 ) ,
retry = retry_if_exception ( is_grpc_error_unavailable ) )
def assist ( self ) :
""" Send a voice request to the Assistant and playback the response.
Returns : True if conversation should continue .
"""
continue_conversation = False
device_actions_futures = [ ]
self . conversation_stream . start_recording ( )
2018-06-07 09:08:32 +02:00
self . logger . info ( ' Recording audio request. ' )
2018-03-20 23:34:36 +01:00
2018-03-24 03:05:46 +01:00
if self . on_conversation_start :
self . on_conversation_start ( )
2018-03-20 23:34:36 +01:00
def iter_assist_requests ( ) :
for c in self . gen_assist_requests ( ) :
assistant_helpers . log_assist_request_without_audio ( c )
yield c
self . conversation_stream . start_playback ( )
2018-03-21 23:21:41 +01:00
user_request = None
2018-03-20 23:34:36 +01:00
# This generator yields AssistResponse proto messages
# received from the gRPC Google Assistant API.
for resp in self . assistant . Assist ( iter_assist_requests ( ) ,
self . deadline ) :
assistant_helpers . log_assist_response_without_audio ( resp )
if resp . event_type == self . END_OF_UTTERANCE :
2018-06-07 09:08:32 +02:00
self . logger . info ( ' End of audio request detected ' )
2018-03-20 23:34:36 +01:00
self . conversation_stream . stop_recording ( )
if resp . speech_results :
user_request = ' ' . join (
r . transcript for r in resp . speech_results )
2018-06-07 09:08:32 +02:00
self . logger . info ( ' Transcript of user request: " %s " . ' , user_request )
self . logger . info ( ' Playing assistant response. ' )
2018-07-30 22:08:06 +02:00
if resp . audio_out . audio_data :
2018-03-20 23:34:36 +01:00
self . conversation_stream . write ( resp . audio_out . audio_data )
if resp . dialog_state_out . conversation_state :
conversation_state = resp . dialog_state_out . conversation_state
2018-06-07 09:08:32 +02:00
self . logger . debug ( ' Updating conversation state. ' )
2018-03-20 23:34:36 +01:00
self . conversation_state = conversation_state
if resp . dialog_state_out . volume_percentage != 0 :
volume_percentage = resp . dialog_state_out . volume_percentage
2018-06-07 09:08:32 +02:00
self . logger . info ( ' Setting volume to %s %% ' , volume_percentage )
2018-03-20 23:34:36 +01:00
self . conversation_stream . volume_percentage = volume_percentage
if resp . dialog_state_out . microphone_mode == self . DIALOG_FOLLOW_ON :
continue_conversation = True
2018-06-07 09:08:32 +02:00
self . logger . info ( ' Expecting follow-on query from user. ' )
2018-03-20 23:34:36 +01:00
elif resp . dialog_state_out . microphone_mode == self . CLOSE_MICROPHONE :
continue_conversation = False
if resp . device_action . device_request_json :
device_request = json . loads (
resp . device_action . device_request_json
)
fs = self . device_handler ( device_request )
if fs :
device_actions_futures . extend ( fs )
2018-07-30 22:08:06 +02:00
if device_actions_futures :
2018-06-07 09:08:32 +02:00
self . logger . info ( ' Waiting for device executions to complete. ' )
2018-03-20 23:34:36 +01:00
concurrent . futures . wait ( device_actions_futures )
2018-06-07 09:08:32 +02:00
self . logger . info ( ' Finished playing assistant response. ' )
2018-03-21 23:21:41 +01:00
2018-03-26 00:32:03 +02:00
try :
2018-03-21 23:21:41 +01:00
self . conversation_stream . stop_playback ( )
2018-07-30 22:08:06 +02:00
except Exception :
2018-03-26 00:32:03 +02:00
pass
if user_request and self . on_speech_recognized :
self . on_speech_recognized ( user_request )
2018-03-24 03:05:46 +01:00
2018-03-20 23:34:36 +01:00
return ( user_request , continue_conversation )
def gen_assist_requests ( self ) :
""" Yields: AssistRequest messages to send to the API. """
dialog_state_in = embedded_assistant_pb2 . DialogStateIn (
2018-07-30 22:08:06 +02:00
language_code = self . language_code ,
conversation_state = b ' '
)
2018-03-20 23:34:36 +01:00
if self . conversation_state :
2018-06-07 09:08:32 +02:00
self . logger . debug ( ' Sending conversation state. ' )
2018-03-20 23:34:36 +01:00
dialog_state_in . conversation_state = self . conversation_state
2018-07-30 22:08:06 +02:00
2018-03-20 23:34:36 +01:00
config = embedded_assistant_pb2 . AssistConfig (
audio_in_config = embedded_assistant_pb2 . AudioInConfig (
encoding = ' LINEAR16 ' ,
sample_rate_hertz = self . conversation_stream . sample_rate ,
) ,
audio_out_config = embedded_assistant_pb2 . AudioOutConfig (
encoding = ' LINEAR16 ' ,
sample_rate_hertz = self . conversation_stream . sample_rate ,
volume_percentage = self . conversation_stream . volume_percentage ,
) ,
dialog_state_in = dialog_state_in ,
device_config = embedded_assistant_pb2 . DeviceConfig (
device_id = self . device_id ,
device_model_id = self . device_model_id ,
)
)
2018-07-30 22:08:06 +02:00
2018-03-20 23:34:36 +01:00
# The first AssistRequest must contain the AssistConfig
# and no audio data.
yield embedded_assistant_pb2 . AssistRequest ( config = config )
for data in self . conversation_stream :
# Subsequent requests need audio data, but not config.
yield embedded_assistant_pb2 . AssistRequest ( audio_in = data )
2018-06-26 00:16:39 +02:00
2018-03-20 23:34:36 +01:00
# vim:sw=4:ts=4:et: