smart_speaker_theatre_backend/smart_speaker_theatre.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-

# SMART SPEAKER THEATRE
# This script reads the triggers for activated user intents sent by Snips over MQTT
# Using this triggers, it will start particular actions and scripts


# Libraries
import re
from config import characters, directions
from logic import tts, read_script, select_script, led_on, led_off
from subprocess import call
import paho.mqtt.client as mqtt
import json
from time import sleep
from pixel_ring import pixel_ring
import serial
from tuning import Tuning
import usb.core
import usb.util


# === SETUP OF MQTT PART 1 ===

# Location of the MQTT server
HOST = 'localhost'
PORT = 1883

# Subscribe to relevant MQTT topics
def on_connect(client, userdata, flags, rc):
    print("Connected to {0} with result code {1}".format(HOST, rc))
    client.subscribe('hermes/intent/jocavdh:play_intro') # to check for intent to play the act
    client.subscribe('hermes/intent/jocavdh:play_question') # to check for the intent to continue to the next act
    client.subscribe('hermes/intent/jocavdh:play_verdict') # to check for the intent to continue to the next act
    client.subscribe('hermes/dialogueManager/sessionEnded')
    client.subscribe('hermes/hotword/default/detected')
    client.subscribe("hermes/asr/textCaptured")

# Set up serial connection with the microcontroller that controls the speaker LED's
ser = serial.Serial('/dev/ttyACM0', 9600) 

# Function to do the play
def play_script(file):

    for character, line, direction in read_script(file):
        input_text = line
        voice = characters.get(character)[0]
        speaker = characters.get(character)[1]
        action = directions.get(direction[0])
        print(direction)
        print(action)
        led_on(ser, speaker)
        tts(voice, input_text, speaker)
        led_off(ser, speaker)

        if action == 'listen_google_home':
            dev = usb.core.find(idVendor=0x2886, idProduct=0x0018)
            print('Wait for Google Home')
            Mic_tuning = Tuning(dev)
            VAD = Mic_tuning.is_voice()
            counter= 0
            voice_detected = 1

            sleep(4)

            while voice_detected == 1:
                    
                    print(VAD)

                    VAD = Mic_tuning.is_voice()

                    if VAD == 1:
                        counter = 0
                        print('still speaking')

                    if VAD == 0:
                        counter+=1
                        print('silence detected')

                    if counter == 30:
                        print('no voice detected')
                        voice_detected = 0
                    
                    print(counter)
                        
                            
                
            print('Google Home is done')

        if action == 'congress.wav':
            print('play audioclip')
            playing = True

            while playing:
                call(["aplay", "-D", speaker, "sounds/congress.wav"])
                playing = False

        if action == 'genie.wav':
            print('play audioclip')
            playing = True

            while playing:
                call(["aplay", "-D", speaker, "sounds/genie.wav"])
                playing = False

                

        sleep(1) # Add a short pause between the lines


    print('The act is done.')

# Function to control the LED's of the speakers



# === FUNCTIONS THAT ARE TRIGGERED WHEN AN INTENT IS DETECTED ===

def on_wakeword(client, userdata, msg):
    pixel_ring.think()
    led_on(ser, 'mono1')
    led_on(ser, 'mono3')

def on_asr_captured(client, userdata, msg):
    pixel_ring.off()
    led_off(ser, 'mono1')
    led_off(ser, 'mono3')    

# Function which is triggered when the intent introduction is activated
def on_play_intro(client,userdata,msg):

    #import pdb; pdb.set_trace()
    path = 'scripts_play/intro/'
    selected_script = select_script(path)
    play_script(selected_script)

# Function which is triggered when the intent for another question is activated
def on_play_question(client, userdata, msg):

    path = 'scripts_play/questions/'
    selected_script = select_script(path)
    play_script(selected_script)

# Function which is triggered when the intent for another question is activated
def on_play_verdict(client, userdata, msg):

    path = 'scripts_play/verdict/'
    selected_script = select_script(path)
    play_script(selected_script)

# Function to handle the situation that an intent is not recognized
def on_intent_unknown(client, userdata, msg):
    data = json.loads(msg.payload.decode("utf-8","ignore"))
    reason = data['termination']['reason']

    if reason == 'intentNotRecognized':

        path = 'scripts_play/wake_sentences/'
        selected_script = select_script(path)
        play_script(selected_script)




# === SETUP OF MQTT PART 2 ===

# Initialise MQTT client
client = mqtt.Client()
client.connect(HOST, PORT, 60)
client.on_connect = on_connect

# Connect each MQTT topic to which you subscribed to a handler function
client.message_callback_add('hermes/hotword/default/detected', on_wakeword)
client.message_callback_add('hermes/intent/jocavdh:play_intro', on_play_intro)
client.message_callback_add('hermes/intent/jocavdh:play_question', on_play_question)
client.message_callback_add('hermes/intent/jocavdh:play_verdict', on_play_verdict)
client.message_callback_add('hermes/asr/textCaptured', on_asr_captured)
client.message_callback_add('hermes/dialogueManager/sessionEnded', on_intent_unknown)


# Keep checking for new MQTT messages
client.loop_forever()