Hackster is hosting Hackster Holidays, Ep. 4: Livestream & Giveaway Drawing. Start streaming on Wednesday!Stream Hackster Holidays, Ep. 4 on Wednesday!
Isabel
Published

Amazing Amazon Alexa's Magic Show (feat. LEGO MINDSTORMS)

Amazing Amazon Alexa teams up with the Legendary LEGO MINDSTORMS EV3 to perform an amazing magic show.

AdvancedFull instructions providedOver 1 day1,828
Finalists
LEGO MINDSTORMS Voice Challenge
First 100 Submissions
LEGO MINDSTORMS Voice Challenge

Things used in this project

Hardware components

Echo Dot
Amazon Alexa Echo Dot
×1
Mindstorms EV3 Programming Brick / Kit
LEGO Mindstorms EV3 Programming Brick / Kit
×1
Pixy2 image sensor
×1

Software apps and online services

Alexa Skills Kit
Amazon Alexa Alexa Skills Kit
Alexa Gadgets Toolkit
Amazon Alexa Alexa Gadgets Toolkit
ev3dev
Studio 2.0

Story

Read more

Schematics

Magical boxes

You need to build 3 of these, 2 to place on top of the trapdoor table and 1 for the mysterious box that the ball magically transports to. Bricklink Studio 2.0 needed to open this file.

Trapdoor robot

Bricklink Studio 2.0 needed to open this file.

Ramp & drawbridge

Bricklink Studio 2.0 needed to open this file.

Seesaw

Bricklink Studio 2.0 needed to open this file.

Sign revealer

Bricklink Studio 2.0 needed to open this file.

Code

Trapdoor.py

Python
Trapdoor EV3 program
#!/usr/bin/env python3
# Copyright 2019 Amazon.com, Inc. or its affiliates.  All Rights Reserved.
# 
# You may not use this file except in compliance with the terms and conditions 
# set forth in the accompanying LICENSE.TXT file.
#
# THESE MATERIALS ARE PROVIDED ON AN "AS IS" BASIS. AMAZON SPECIFICALLY DISCLAIMS, WITH 
# RESPECT TO THESE MATERIALS, ALL WARRANTIES, EXPRESS, IMPLIED, OR STATUTORY, INCLUDING 
# THE IMPLIED WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE, AND NON-INFRINGEMENT.

import os
import sys
import time
import logging
import json
import random
import threading

from enum import Enum
from agt import AlexaGadget
from time import sleep
from smbus import SMBus

from ev3dev2.led import Leds
from ev3dev2.sound import Sound
from ev3dev2.motor import OUTPUT_A, OUTPUT_B, OUTPUT_C, OUTPUT_D, LargeMotor, MediumMotor
from ev3dev2.sensor import INPUT_1, INPUT_3, INPUT_4
from ev3dev2.sensor.lego import TouchSensor
from ev3dev2.port import LegoPort

# Set the logging level to INFO to see messages from AlexaGadget
logging.basicConfig(level=logging.INFO, stream=sys.stdout, format='%(message)s')
logging.getLogger().addHandler(logging.StreamHandler(sys.stderr))
logger = logging.getLogger(__name__)

class MindstormsGadget(AlexaGadget):
    """
    A Mindstorms gadget that performs movement based on voice commands.
    Two types of commands are supported, directional movement and preset.
    """

    def __init__(self):
        """
        Performs Alexa Gadget initialization routines and ev3dev resource allocation.
        """
        super().__init__()

        # Ev3dev initialization
        self.leds = Leds()
        self.sound = Sound()
        self.box1 = MediumMotor(OUTPUT_A)
        self.ts3 = TouchSensor(INPUT_3)
        self.box2 = MediumMotor(OUTPUT_B)
        self.ts4 = TouchSensor(INPUT_4)
        self.catch = LargeMotor(OUTPUT_C)
        self.trapdoor = LargeMotor(OUTPUT_D)
        # initialise pixy2
        self.in1 = LegoPort(INPUT_1)
        self.in1.mode = 'other-i2c'

    def on_connected(self, device_addr):
        """
        Gadget connected to the paired Echo device.
        :param device_addr: the address of the device we connected to
        """
        self.leds.set_color("LEFT", "GREEN")
        self.leds.set_color("RIGHT", "GREEN")
        logger.info("{} connected to Echo device".format(self.friendly_name))

    def on_disconnected(self, device_addr):
        """
        Gadget disconnected from the paired Echo device.
        :param device_addr: the address of the device we disconnected from
        """
        self.leds.set_color("LEFT", "BLACK")
        self.leds.set_color("RIGHT", "BLACK")
        logger.info("{} disconnected from Echo device".format(self.friendly_name))

    def on_custom_mindstorms_trapdoor_boxes(self, directive):
        """
        Handles the Custom.Mindstorms.Trapdoor boxes directive.
        :param directive: the custom directive with the matching namespace and name
        """
        try:
            logger.info("Triggered boxes directive")
            payload = json.loads(directive.payload.decode("utf-8"))
            print("Control payload: {}".format(payload), file=sys.stderr)
            direction = payload["direction"]
            if "box_number" not in payload:
                if direction == "open":
                    # open both boxes
                    self._open_box1()
                    self._open_box2()
                else:
                    self._close_box1()
                    self._close_box2()
            else:
                box_number = payload["box_number"]
                if direction == "open":
                    if box_number == "1":
                        self._open_box1() 
                    if box_number == "2":
                        self._open_box2()


        except KeyError:
            print("Missing expected parameters: {}".format(directive), file=sys.stderr)
            
    def on_custom_mindstorms_trapdoor_ball(self, directive):
        """
        Handles the Custom.Mindstorms.Trapdoor ball directive.
        :param directive: the custom directive with the matching namespace and name
        """
        try:
            logger.info("Triggered ball directive")
            bus = SMBus(3)
            address = 0x54
            # turn on lamp
            data = [174, 193, 22, 2, 1, 1]
            bus.write_i2c_block_data(address, 0, data)
            # move catch and drop trapdoor
            self.catch.on_for_degrees(speed=15,degrees=-25)
            self.trapdoor.on_for_degrees(speed=5,degrees=75)
            sleep(1)
            # return catch and raise trapdoor
            self.trapdoor.on_for_degrees(speed=25,degrees=-100)
            self.catch.on_for_degrees(speed=25,degrees=25)
            self.trapdoor.on_for_degrees(speed=20,degrees=25)
            # find colour
            data = [174, 193, 32, 2, 255, 1]
            bus.write_i2c_block_data(address, 0, data)
            block = bus.read_i2c_block_data(address, 0, 20)
            logger.info("colour: " + str(block[6]))
            # turn off lamp
            data = [174, 193, 22, 2, 0, 0]
            bus.write_i2c_block_data(address, 0, data)
            # send colour to Alexa
            self.send_custom_event('Custom.Mindstorms.Trapdoor', "ColourFound", {'colour': block[6]})
            
        except KeyError:
            print("Missing expected parameters: {}".format(directive), file=sys.stderr)
        
    def on_custom_mindstorms_trapdoor_bow(self, directive):
        """
        Handles the Custom.Mindstorms.Trapdoor bow directive.
        :param directive: the custom directive with the matching namespace and name
        """
        try:
            logger.info("Triggered bow directive")
            # Start threads
            threading.Thread(target=self._box1_thread, daemon=True).start()
            threading.Thread(target=self._box2_thread, daemon=True).start()

            
        except KeyError:
            print("Missing expected parameters: {}".format(directive), file=sys.stderr)

    def _open_box1(self):
        self.box1.on(speed=10)
        while True:
            if self.ts3.is_pressed:
                break
        self.box1.off()

    def _open_box2(self):
        self.box2.on(speed=10)
        while True:
            if self.ts4.is_pressed:
                break
        self.box2.off()

    def _close_box1(self):
        self.box1.on_for_degrees(speed=10,degrees=-90)
        
    def _close_box2(self):
        self.box2.on_for_degrees(speed=10,degrees=-90)

    def _box1_thread(self):
        self.box1.on_for_degrees(speed=10,degrees=-45)
        sleep(2)
        self._open_box1()
    
    def _box2_thread(self):
        self.box2.on_for_degrees(speed=10,degrees=-45)
        sleep(2)
        self._open_box2()

if __name__ == '__main__':

    gadget = MindstormsGadget()

    # Set LCD font and turn off blinking LEDs
    os.system('setfont Lat7-Terminus12x6')
    gadget.leds.set_color("LEFT", "BLACK")
    gadget.leds.set_color("RIGHT", "BLACK")

    # Startup sequence
    gadget.sound.play_song((('C4', 'e'), ('D4', 'e'), ('E5', 'q')))
    gadget.leds.set_color("LEFT", "GREEN")
    gadget.leds.set_color("RIGHT", "GREEN")

    # Gadget main entry point
    gadget.main()

    # Shutdown sequence
    gadget.sound.play_song((('E5', 'e'), ('C4', 'e')))
    gadget.leds.set_color("LEFT", "BLACK")
    gadget.leds.set_color("RIGHT", "BLACK")

Trapdoor.ini

Python
Config file for Trapdoor robot.
[GadgetSettings]
amazonId = [REPLACE WITH YOUR ID]
alexaGadgetSecret = [REPLACE WITH YOUR SECRET]

[GadgetCapabilities]
Custom.Mindstorms.Trapdoor = 1.0

magical_box.py

Python
Magical Box EV3 program.
#!/usr/bin/env python3
# Copyright 2019 Amazon.com, Inc. or its affiliates.  All Rights Reserved.
# 
# You may not use this file except in compliance with the terms and conditions 
# set forth in the accompanying LICENSE.TXT file.
#
# THESE MATERIALS ARE PROVIDED ON AN "AS IS" BASIS. AMAZON SPECIFICALLY DISCLAIMS, WITH 
# RESPECT TO THESE MATERIALS, ALL WARRANTIES, EXPRESS, IMPLIED, OR STATUTORY, INCLUDING 
# THE IMPLIED WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE, AND NON-INFRINGEMENT.

import os
import sys
import time
import logging
import json
import random
import threading

from enum import Enum
from agt import AlexaGadget
from time import sleep

from ev3dev2.led import Leds
from ev3dev2.sound import Sound
from ev3dev2.motor import OUTPUT_A, OUTPUT_B, OUTPUT_C, OUTPUT_D, LargeMotor, MediumMotor
from ev3dev2.sensor import INPUT_3, INPUT_4
from ev3dev2.sensor.lego import TouchSensor

# Set the logging level to INFO to see messages from AlexaGadget
logging.basicConfig(level=logging.INFO, stream=sys.stdout, format='%(message)s')
logging.getLogger().addHandler(logging.StreamHandler(sys.stderr))
logger = logging.getLogger(__name__)

class MindstormsGadget(AlexaGadget):
    """
    A Mindstorms gadget that performs movement based on voice commands.
    Two types of commands are supported, directional movement and preset.
    """

    def __init__(self):
        """
        Performs Alexa Gadget initialization routines and ev3dev resource allocation.
        """
        super().__init__()

        # Ev3dev initialization
        self.leds = Leds()
        self.sound = Sound()
        self.motorA = LargeMotor(OUTPUT_A)
        self.motorB = LargeMotor(OUTPUT_B)
        self.box3 = MediumMotor(OUTPUT_C)
        self.box3ts3 = TouchSensor(INPUT_3)
        self.drawbridge = MediumMotor(OUTPUT_D)
        self.dbts4 = TouchSensor(INPUT_4)

    def on_connected(self, device_addr):
        """
        Gadget connected to the paired Echo device.
        :param device_addr: the address of the device we connected to
        """
        self.leds.set_color("LEFT", "GREEN")
        self.leds.set_color("RIGHT", "GREEN")
        logger.info("{} connected to Echo device".format(self.friendly_name))

    def on_disconnected(self, device_addr):
        """
        Gadget disconnected from the paired Echo device.
        :param device_addr: the address of the device we disconnected from
        """
        self.leds.set_color("LEFT", "BLACK")
        self.leds.set_color("RIGHT", "BLACK")
        logger.info("{} disconnected from Echo device".format(self.friendly_name))

    def on_custom_mindstorms_magicalbox_seesaw(self, directive):
        """
        Handles the Custom.Mindstorms.MagicalBox seesaw directive.
        :param directive: the custom directive with the matching namespace and name
        """
        try:
            logger.info("Triggered seesaw directive")
            payload = json.loads(directive.payload.decode("utf-8"))
            print("Control payload: {}".format(payload), file=sys.stderr)
            self._open_box3()
            self._raise_drawbridge()
            self._lower_drawbridge()
            colour = payload["colour"]
            if colour == 1:
                self.motorA.on_for_degrees(speed=30,degrees=-60)
                sleep(2)
                self.motorA.on_for_degrees(speed=30,degrees=60)
            elif colour == 2:
                self.motorA.on_for_degrees(speed=30,degrees=60)
                sleep(2)
                self.motorA.on_for_degrees(speed=30,degrees=-60)
            elif colour == 3:
                self.motorB.on_for_degrees(speed=30,degrees=60)
                sleep(2)
                self.motorB.on_for_degrees(speed=30,degrees=-60)
            elif colour == 4:
                self.motorB.on_for_degrees(speed=30,degrees=-60)
                sleep(2)
                self.motorB.on_for_degrees(speed=30,degrees=60)
            sleep(5)
            self._raise_drawbridge()
            self._close_box3()

        except KeyError:
            print("Missing expected parameters: {}".format(directive), file=sys.stderr)

    def on_custom_mindstorms_magicalbox_magicalbox(self, directive):
        """
        Handles the Custom.Mindstorms.MagicalBox magicalbox directive.
        :param directive: the custom directive with the matching namespace and name
        """
        try:
            logger.info("Triggered magicalbox directive")
            self._open_box3()
    
        except KeyError:
            print("Missing expected parameters: {}".format(directive), file=sys.stderr)

    def on_custom_mindstorms_magicalbox_bow(self, directive):
        """
        Handles the Custom.Mindstorms.MagicalBox bow directive.
        :param directive: the custom directive with the matching namespace and name
        """
        try:
            logger.info("Triggered bow directive")
            self.box3.on_for_degrees(speed=10,degrees=-45)
            sleep(2)
            self._open_box3()
    
        except KeyError:
            print("Missing expected parameters: {}".format(directive), file=sys.stderr)


    def _open_box3(self):
        self.box3.on(speed=10)
        while True:
            if self.box3ts3.is_pressed:
                break
        self.box3.off()

    def _close_box3(self):
        self.box3.on_for_degrees(speed=10,degrees=-90)

    def _raise_drawbridge(self):
        self.drawbridge.on(speed=10)
        while True:
            if self.dbts4.is_pressed:
                break
        self.drawbridge.off()

    def _lower_drawbridge(self):
        self.drawbridge.on_for_degrees(speed=10,degrees=-90)
    
if __name__ == '__main__':

    gadget = MindstormsGadget()

    # Set LCD font and turn off blinking LEDs
    os.system('setfont Lat7-Terminus12x6')
    gadget.leds.set_color("LEFT", "BLACK")
    gadget.leds.set_color("RIGHT", "BLACK")

    # Startup sequence
    gadget.sound.play_song((('C4', 'e'), ('D4', 'e'), ('E5', 'q')))
    gadget.leds.set_color("LEFT", "GREEN")
    gadget.leds.set_color("RIGHT", "GREEN")

    # Gadget main entry point
    gadget.main()

    # Shutdown sequence
    gadget.sound.play_song((('E5', 'e'), ('C4', 'e')))
    gadget.leds.set_color("LEFT", "BLACK")
    gadget.leds.set_color("RIGHT", "BLACK")

magical_box.ini

Python
Config file for Magical Box robot.
[GadgetSettings]
amazonId = [REPLACE WITH YOUR ID]
alexaGadgetSecret = [REPLACE WITH YOUR SECRET]

[GadgetCapabilities]
Custom.Mindstorms.MagicalBox = 1.0

revealer.py

Python
Revealer EV3 program.
#!/usr/bin/env python3
# Copyright 2019 Amazon.com, Inc. or its affiliates.  All Rights Reserved.
# 
# You may not use this file except in compliance with the terms and conditions 
# set forth in the accompanying LICENSE.TXT file.
#
# THESE MATERIALS ARE PROVIDED ON AN "AS IS" BASIS. AMAZON SPECIFICALLY DISCLAIMS, WITH 
# RESPECT TO THESE MATERIALS, ALL WARRANTIES, EXPRESS, IMPLIED, OR STATUTORY, INCLUDING 
# THE IMPLIED WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE, AND NON-INFRINGEMENT.

import os
import sys
import time
import logging
import json
import random
import threading

from enum import Enum
from agt import AlexaGadget
from time import sleep

from ev3dev2.led import Leds
from ev3dev2.sound import Sound
from ev3dev2.motor import OUTPUT_A, OUTPUT_B, OUTPUT_D, LargeMotor, MoveTank

# Set the logging level to INFO to see messages from AlexaGadget
logging.basicConfig(level=logging.INFO, stream=sys.stdout, format='%(message)s')
logging.getLogger().addHandler(logging.StreamHandler(sys.stderr))
logger = logging.getLogger(__name__)

class MindstormsGadget(AlexaGadget):
    """
    A Mindstorms gadget that performs movement based on voice commands.
    Two types of commands are supported, directional movement and preset.
    """

    def __init__(self):
        """
        Performs Alexa Gadget initialization routines and ev3dev resource allocation.
        """
        super().__init__()

        # Ev3dev initialization
        self.leds = Leds()
        self.sound = Sound()
        self.motorAB = MoveTank(OUTPUT_A, OUTPUT_B)
        self.sign = LargeMotor(OUTPUT_D)

    def on_connected(self, device_addr):
        """
        Gadget connected to the paired Echo device.
        :param device_addr: the address of the device we connected to
        """
        self.leds.set_color("LEFT", "GREEN")
        self.leds.set_color("RIGHT", "GREEN")
        logger.info("{} connected to Echo device".format(self.friendly_name))

    def on_disconnected(self, device_addr):
        """
        Gadget disconnected from the paired Echo device.
        :param device_addr: the address of the device we disconnected from
        """
        self.leds.set_color("LEFT", "BLACK")
        self.leds.set_color("RIGHT", "BLACK")
        logger.info("{} disconnected from Echo device".format(self.friendly_name))

    def on_custom_mindstorms_revealer_table(self, directive):
        """
        Handles the Custom.Mindstorms.Revealer table directive.
        :param directive: the custom directive with the matching namespace and name
        """
        try:
            logger.info("Triggered table directive")
            self.motorAB.on_for_rotations(25, 25, 1.75)

        except KeyError:
            print("Missing expected parameters: {}".format(directive), file=sys.stderr)

    def on_custom_mindstorms_revealer_sign(self, directive):
        """
        Handles the Custom.Mindstorms.Revealer sign directive.
        :param directive: the custom directive with the matching namespace and name
        """
        try:
            logger.info("Triggered sign directive")
            sleep (4)
            self.sign.on_for_rotations(100, 1.5)
            self.sign.on_for_rotations(-100, 1.5)

        except KeyError:
            print("Missing expected parameters: {}".format(directive), file=sys.stderr)
    
if __name__ == '__main__':

    gadget = MindstormsGadget()

    # Set LCD font and turn off blinking LEDs
    os.system('setfont Lat7-Terminus12x6')
    gadget.leds.set_color("LEFT", "BLACK")
    gadget.leds.set_color("RIGHT", "BLACK")

    # Startup sequence
    gadget.sound.play_song((('C4', 'e'), ('D4', 'e'), ('E5', 'q')))
    gadget.leds.set_color("LEFT", "GREEN")
    gadget.leds.set_color("RIGHT", "GREEN")

    # Gadget main entry point
    gadget.main()

    # Shutdown sequence
    gadget.sound.play_song((('E5', 'e'), ('C4', 'e')))
    gadget.leds.set_color("LEFT", "BLACK")
    gadget.leds.set_color("RIGHT", "BLACK")

revealer.ini

Python
Config file for Revealer robot.
[GadgetSettings]
amazonId = [REPLACE WITH YOUR ID]
alexaGadgetSecret = [REPLACE WITH YOUR SECRET]

[GadgetCapabilities]
Custom.Mindstorms.Revealer = 1.0

Alexa lambda_function.py

Python
Alexa skill code
# -*- coding: utf-8 -*-

# This sample demonstrates handling intents from an Alexa skill using the Alexa Skills Kit SDK for Python.
# Please visit https://alexa.design/cookbook for additional examples on implementing slots, dialog management,
# session persistence, api calls, and more.
# This sample is built using the handler classes approach in skill builder.
import logging
import requests
import ask_sdk_core.utils as ask_utils

from ask_sdk_core.skill_builder import SkillBuilder
from ask_sdk_core.dispatch_components import AbstractRequestHandler
from ask_sdk_core.dispatch_components import AbstractExceptionHandler
from ask_sdk_core.handler_input import HandlerInput
from ask_sdk_core.utils import is_request_type

from ask_sdk_model.interfaces.custom_interface_controller import (
    StartEventHandlerDirective, EventFilter, Expiration, FilterMatchAction,
    StopEventHandlerDirective,
    SendDirectiveDirective,
    Header,
    Endpoint,
    EventsReceivedRequest,
    ExpiredRequest
)

from ask_sdk_model import Response

logger = logging.getLogger(__name__)
logger.setLevel(logging.INFO)

# The namespace of the custom directive to be sent by this skill
NAMESPACE_Trapdoor = "Custom.Mindstorms.Trapdoor"
NAMESPACE_MagicalBox = "Custom.Mindstorms.MagicalBox"
NAMESPACE_Revealer = "Custom.Mindstorms.Revealer"

sb = SkillBuilder()

class LaunchRequestHandler(AbstractRequestHandler):
    """Handler for Skill Launch."""
    def can_handle(self, handler_input):
        # type: (HandlerInput) -> bool

        return ask_utils.is_request_type("LaunchRequest")(handler_input)

    def handle(self, handler_input):
        # type: (HandlerInput) -> Response
        speak_output = "<voice name=\"Russell\">Welcome to the Amazing Amazon Alexa's magic show.</voice> <audio src=\"soundbank://soundlibrary/human/amzn_sfx_crowd_applause_01\"></audio> Thank you thank you. Today I will perform a trick to make a ball move from one box to another. Choose one of the 4 coloured balls at random to continue"
        #speak_output = "Welcome to Alexa's magic show."

        system = handler_input.request_envelope.context.system
        api_access_token = system.api_access_token
        api_endpoint = system.api_endpoint

        # Get connected gadget endpoint ID.
        endpoints = get_connected_endpoints(api_endpoint, api_access_token)
        logger.debug("Checking endpoint..")
        if not endpoints:
            logger.debug("No connected gadget endpoints available.")
            return (handler_input.response_builder
                .speak("I couldn't find an EV3 Brick connected to this Echo device. Please check to make sure your EV3 Brick is connected, and try again.")
                    .set_should_end_session(True)
                   .response)
         
        # Store endpoint ID for using it to send custom directives later.          
        session_attr = handler_input.attributes_manager.session_attributes
        for ep in endpoints:
            if ep["capabilities"][0]["interface"] == "Custom.Mindstorms.Trapdoor":
                session_attr["endpoint_trapdoor"] = ep["endpointId"]
            elif ep["capabilities"][0]["interface"] == "Custom.Mindstorms.MagicalBox":
                session_attr["endpoint_magicalbox"] = ep["endpointId"]
            elif ep["capabilities"][0]["interface"] == "Custom.Mindstorms.Revealer":
                session_attr["endpoint_revealer"] = ep["endpointId"]
                
        return (
            handler_input.response_builder
                .speak(speak_output)
                .ask("Now you have chosen a ball, please say open boxes to start.")
                .response
        )

class BoxesIntentHandler(AbstractRequestHandler):
    def can_handle(self, handler_input):
        # type: (HandlerInput) -> bool
        return ask_utils.is_intent_name("BoxesIntent")(handler_input)

    def handle(self, handler_input):
        # type: (HandlerInput) -> Response
        direction = ask_utils.get_slot_value(handler_input,"direction")
        box_number = ask_utils.get_slot_value(handler_input,"box_number")
        # Get data from session attribute
        session_attr = handler_input.attributes_manager.session_attributes
        endpoint_id = session_attr.get("endpoint_trapdoor", [])
        payload = {
            "direction": direction,
            "box_number": box_number
        }
        directive = build_send_directive(NAMESPACE_Trapdoor, "boxes", endpoint_id, payload)

        if box_number is None:
            if direction == "open":
                speak_output = "Ok, place your ball into box 1, the one on the left. Say close boxes when you are ready to move on"
                ask_output = "Say close boxes when you are ready to move on"
            else:
                speak_output = "Now it's time to see my magic work. Let's move the ball"
                ask_output = "I think it's time to move the ball"
        elif box_number == "1":
            speak_output = "The ball is not here!"
            ask_output = "It should have moved to box two"
        elif box_number == "2":
            colour_name = session_attr.get("colour_name", "unknown")
            speak_output = "Oh no! Where is the {} ball? Maybe we have to try another magic word".format(colour_name)
            ask_output = "Maybe we have to try another magic word"
        else:
            speak_output = "unknown box instruction"
            
        return (
            handler_input.response_builder
                .speak(speak_output)
                .ask(ask_output)
                .add_directive(directive)
                .response
        )

class BallIntentHandler(AbstractRequestHandler):
    def can_handle(self, handler_input):
        # type: (HandlerInput) -> bool
        return ask_utils.is_intent_name("BallIntent")(handler_input)

    def handle(self, handler_input):
        # Get data from session attribute
        session_attr = handler_input.attributes_manager.session_attributes
        endpoint_id = session_attr.get("endpoint_trapdoor", [])
        
        speak_output = "I like to move it move it, I like to move it move it, I like to move it move it, I like to - move it. <break time=\"2s\"/> Say the magic word please."
        ask_output = "Say the magic word please"
        directive = build_send_directive(NAMESPACE_Trapdoor, "ball", endpoint_id, {})
        
        # Set the token to track the event handler
        token = handler_input.request_envelope.request.request_id
        session_attr["token"] = token
        
        return (
            handler_input.response_builder
                .speak(speak_output)
                .ask(ask_output)
                .add_directive(directive)
                .add_directive(build_start_event_handler_directive(token, 30000, NAMESPACE_Trapdoor, "ball", 'SEND', {}))
                .response
        )

class MagicWordIntentHandler(AbstractRequestHandler):
    def can_handle(self, handler_input):
        # type: (HandlerInput) -> bool
        return ask_utils.is_intent_name("MagicWordIntent")(handler_input)

    def handle(self, handler_input):
        # type: (HandlerInput) -> Response
        magic_word = ask_utils.get_slot_value(handler_input,"magic_word")
        directive = None
        if magic_word == "the magic word please":
            speak_output = "Haha very funny. Please try a real magic word."
            ask_output = "Please try a real magic word."
        elif magic_word == "abracadabra":
            speak_output = "Let's see if my magic has worked. Tell me to open box one then box two"
            ask_output = "Tell me to open box one then box two"
            session_attr = handler_input.attributes_manager.session_attributes
            colour = session_attr.get("colour", 3)
            endpoint_id = session_attr.get("endpoint_magicalbox", [])
            payload = {
                "colour": colour
            }
            directive = build_send_directive(NAMESPACE_MagicalBox, "seesaw", endpoint_id, payload)
        elif magic_word == "open sesame":
            speak_output = "<break time=\"2s\"/> Wow! What is that. Should we open box three?"
            ask_output = "Should we open box three?"
            session_attr = handler_input.attributes_manager.session_attributes
            endpoint_id = session_attr.get("endpoint_revealer", [])
            directive = build_send_directive(NAMESPACE_Revealer, "table", endpoint_id, {})
        else:
            speak_output = "unknown magic word"
            ask_output = "try another"
        
        if directive is None:
            return (
            handler_input.response_builder
                .speak(speak_output)
                .ask(ask_output)
                .response
            )
            
        else:
            return (
            handler_input.response_builder
                .speak(speak_output)
                .ask(ask_output)
                .add_directive(directive)
                .response
            )

class MagicalBoxIntentHandler(AbstractRequestHandler):
    def can_handle(self, handler_input):
        # type: (HandlerInput) -> bool
        return ask_utils.is_intent_name("MagicalBoxIntent")(handler_input)

    def handle(self, handler_input):
        # type: (HandlerInput) -> Response
        magical_box_open = ask_utils.get_slot_value(handler_input,"magical_box_open")
        session_attr = handler_input.attributes_manager.session_attributes
        directive = None
        if magical_box_open == "no":
            speak_output = "I'm sad. Can I please open the box? Please say yes."
            ask_output = "Can I please open the box? Please say yes."
        elif magical_box_open == "yes" :
            colour_name = session_attr.get("colour_name", "unknown")
            speak_output = "<amazon:emotion name=\"excited\" intensity=\"high\">Ta-da!! </amazon:emotion><audio src=\"soundbank://soundlibrary/ui/gameshow/amzn_ui_sfx_gameshow_positive_response_01\"></audio>The {} ball has moved! What did you think of my show?".format(colour_name)
            ask_output = "What did you think of my show?"
            endpoint_id = session_attr.get("endpoint_magicalbox", [])
            directive = build_send_directive(NAMESPACE_MagicalBox, "magicalbox", endpoint_id, {})
        else:
            speak_output = "unknown magical box open value"
            ask_output = "say yes or no"
            
        if directive is None:
            return (
            handler_input.response_builder
                .speak(speak_output)
                .ask(ask_output)
                .response
            )
            
        else:
            return (
            handler_input.response_builder
                .speak(speak_output)
                .ask(ask_output)
                .add_directive(directive)
                .response
            )
    
class ShowReviewIntentHandler(AbstractRequestHandler):
    def can_handle(self, handler_input):
        # type: (HandlerInput) -> bool
        return ask_utils.is_intent_name("ShowReviewIntent")(handler_input)

    def handle(self, handler_input):
        # type: (HandlerInput) -> Response
        review = ask_utils.get_slot_value(handler_input,"review")
        if review == "amazing":
            speak_output = "Thank you! <voice name=\"Russell\">Let's give it up for the Amazing Amazon Alexa and her assistant the Legendary Lego Mindstorms.</voice> <audio src=\"soundbank://soundlibrary/human/amzn_sfx_crowd_applause_01\"></audio> I'm glad you enjoyed the show"
            session_attr = handler_input.attributes_manager.session_attributes
            endpoint_id = session_attr.get("endpoint_revealer", [])
            directive = build_send_directive(NAMESPACE_Revealer, "sign", endpoint_id, {})
            return (
            handler_input.response_builder
                .speak(speak_output)
                .add_directive(directive)
                .ask("I'm glad you enjoyed the show")
                .response
            )
        else:
            speak_output = "{}? don't you mean amazing?".format(review)
            ask_output = "Say the show was amazing."
            return (
            handler_input.response_builder
                .speak(speak_output)
                .ask(ask_output)
                .response
            )

class BowIntentHandler(AbstractRequestHandler):
    def can_handle(self, handler_input):
        # type: (HandlerInput) -> bool
        return ask_utils.is_intent_name("BowIntent")(handler_input)

    def handle(self, handler_input):
        # type: (HandlerInput) -> Response
        speak_output = "Thank you! You are all so kind <audio src=\"soundbank://soundlibrary/human/amzn_sfx_crowd_applause_01\"></audio>"
        session_attr = handler_input.attributes_manager.session_attributes
        endpoint_id = session_attr.get("endpoint_trapdoor", [])
        directive1 = build_send_directive(NAMESPACE_Trapdoor, "bow", endpoint_id, {})
        endpoint_id = session_attr.get("endpoint_magicalbox", [])
        directive2 = build_send_directive(NAMESPACE_MagicalBox, "bow", endpoint_id, {})
        return (
            handler_input.response_builder
                .speak(speak_output)
                .add_directive(directive1)
                .add_directive(directive2)
                .response
            )

class HelpIntentHandler(AbstractRequestHandler):
    """Handler for Help Intent."""
    def can_handle(self, handler_input):
        # type: (HandlerInput) -> bool
        return ask_utils.is_intent_name("AMAZON.HelpIntent")(handler_input)

    def handle(self, handler_input):
        # type: (HandlerInput) -> Response
        speak_output = "You can say hello to me! How can I help?"

        return (
            handler_input.response_builder
                .speak(speak_output)
                .ask(speak_output)
                .response
        )

def has_valid_token(handler_input):
    if not is_request_type('CustomInterfaceController.EventsReceived')(handler_input):
        return False

    session_attr = handler_input.attributes_manager.session_attributes
    request = handler_input.request_envelope.request

    # Validate event token
    if session_attr.get("token", None) != request.token:
        logger.info("Event token doesn't match. Ignoring this event")
        return False

    return True

def has_valid_endpoint(handler_input):
    if not is_request_type('CustomInterfaceController.EventsReceived')(handler_input):
        return False

    session_attr = handler_input.attributes_manager.session_attributes
    request = handler_input.request_envelope.request
    custom_event = request.events[0]

    # Validate endpoint
    request_endpoint = custom_event.endpoint.endpoint_id
    if request_endpoint != session_attr.get("endpoint_trapdoor", None):
        logger.info("Event endpoint id doesn't match. Ignoring this event")
        return False

    return True

@sb.request_handler(can_handle_func=lambda handler_input:
    is_request_type("CustomInterfaceController.EventsReceived") and
    has_valid_token(handler_input) and 
    has_valid_endpoint(handler_input)
)
def events_received_request_handler(handler_input: HandlerInput):
    logger.info("== Received Custom Event ==")

    custom_event = handler_input.request_envelope.request.events[0]
    payload = custom_event.payload
    name = custom_event.header.name

    speak_output = ""
    if name == "ColourFound":
        session_attr = handler_input.attributes_manager.session_attributes
        colour = int(payload.get("colour"))
        session_attr["colour"] = colour
        if colour == 1:
            colour_name = "blue"
        elif colour == 2:
            colour_name = "green"
        elif colour == 3:
            colour_name = "pink"
        elif colour == 4:
            colour_name = "purple"
        else:
            colour_name = "unknown"
        session_attr["colour_name"] = colour_name
        return (handler_input.response_builder
                .set_should_end_session(False)
                .response)

    return (handler_input.response_builder
        .speak(speak_output + BG_MUSIC, "REPLACE_ALL")
        .response)

@sb.request_handler(can_handle_func=is_request_type("CustomInterfaceController.Expired"))
def custom_interface_expiration_handler(handler_input):
    logger.info("== Custom Event Expiration Input ==")

#    session_attr = handler_input.attributes_manager.session_attributes
    
    # Set the token to track the event handler
#    token = handler_input.request_envelope.request.request_id
#    session_attr["token"] = token

#    duration = session_attr.get("duration", 0)
#    if duration > 0:
#        session_attr["duration"] = duration - 1 
        # extends skill session
#        speak_output = "{} minutes remaining.".format(duration)
#        timeout = 180000
#        directive = build_start_event_handler_directive(token, timeout, NAMESPACE_Trapdoor, "ball", 'SEND', {})
#        return (handler_input.response_builder
#            .add_directive(directive)
#            .speak(speak_output + BG_MUSIC)
#            .response
#        )
#    else:
        # End skill session
#    return (handler_input.response_builder
#        .speak("Skill duration expired. Goodbye.")
#        .set_should_end_session(True)
#        .response)



class CancelOrStopIntentHandler(AbstractRequestHandler):
    """Single handler for Cancel and Stop Intent."""
    def can_handle(self, handler_input):
        # type: (HandlerInput) -> bool
        return (ask_utils.is_intent_name("AMAZON.CancelIntent")(handler_input) or
                ask_utils.is_intent_name("AMAZON.StopIntent")(handler_input))

    def handle(self, handler_input):
        # type: (HandlerInput) -> Response
        speak_output = "Goodbye!"

        return (
            handler_input.response_builder
                .speak(speak_output)
                .response
        )

class SessionEndedRequestHandler(AbstractRequestHandler):
    """Handler for Session End."""
    def can_handle(self, handler_input):
        # type: (HandlerInput) -> bool
        return ask_utils.is_request_type("SessionEndedRequest")(handler_input)

    def handle(self, handler_input):
        # type: (HandlerInput) -> Response

        # Any cleanup logic goes here.

        return handler_input.response_builder.response

class IntentReflectorHandler(AbstractRequestHandler):
    """The intent reflector is used for interaction model testing and debugging.
    It will simply repeat the intent the user said. You can create custom handlers
    for your intents by defining them above, then also adding them to the request
    handler chain below.
    """
    def can_handle(self, handler_input):
        # type: (HandlerInput) -> bool
        return ask_utils.is_request_type("IntentRequest")(handler_input)

    def handle(self, handler_input):
        # type: (HandlerInput) -> Response
        intent_name = ask_utils.get_intent_name(handler_input)
        speak_output = "You just triggered " + intent_name + "."

        return (
            handler_input.response_builder
                .speak(speak_output)
                # .ask("add a reprompt if you want to keep the session open for the user to respond")
                .response
        )

class CatchAllExceptionHandler(AbstractExceptionHandler):
    """Generic error handling to capture any syntax or routing errors. If you receive an error
    stating the request handler chain is not found, you have not implemented a handler for
    the intent being invoked or included it in the skill builder below.
    """
    def can_handle(self, handler_input, exception):
        # type: (HandlerInput, Exception) -> bool
        return True

    def handle(self, handler_input, exception):
        # type: (HandlerInput, Exception) -> Response
        logger.error(exception, exc_info=True)

        speak_output = "Sorry, I had trouble doing what you asked. Please try again."

        return (
            handler_input.response_builder
                .speak(speak_output)
                .ask(speak_output)
                .response
        )

def get_connected_endpoints(api_endpoint, api_access_token):
    headers = {
        'Content-Type': 'application/json',
        'Authorization': 'Bearer {}'.format(api_access_token)
    }

    api_url = api_endpoint + "/v1/endpoints"
    endpoints_response = requests.get(api_url, headers=headers)

    if endpoints_response.status_code == requests.codes.get("ok", ""):
        return endpoints_response.json()["endpoints"]

def build_send_directive(namespace, name, endpoint_id, payload):
    return SendDirectiveDirective(
        header=Header(
            name=name,
            namespace=namespace
        ),
        endpoint=Endpoint(
            endpoint_id=endpoint_id
        ),
        payload=payload
    )

def build_start_event_handler_directive(token, duration_ms, namespace,
                                        name, filter_match_action, expiration_payload):
    return StartEventHandlerDirective(
        token=token,
        expiration=Expiration(
            duration_in_milliseconds=duration_ms,
            expiration_payload=expiration_payload))

def build_stop_event_handler_directive(token):
    return StopEventHandlerDirective(token=token)

# The SkillBuilder object acts as the entry point for your skill, routing all request and response
# payloads to the handlers above. Make sure any new handlers or interceptors you've
# defined are included below. The order matters - they're processed top to bottom.

sb.add_request_handler(LaunchRequestHandler())
sb.add_request_handler(BoxesIntentHandler())
sb.add_request_handler(BallIntentHandler())
sb.add_request_handler(MagicWordIntentHandler())
sb.add_request_handler(MagicalBoxIntentHandler())
sb.add_request_handler(ShowReviewIntentHandler())
sb.add_request_handler(BowIntentHandler())
sb.add_request_handler(HelpIntentHandler())
sb.add_request_handler(CancelOrStopIntentHandler())
sb.add_request_handler(SessionEndedRequestHandler())
sb.add_request_handler(IntentReflectorHandler()) # make sure IntentReflectorHandler is last so it doesn't override your custom intent handlers

sb.add_exception_handler(CatchAllExceptionHandler())

lambda_handler = sb.lambda_handler()

Alexa interaction model

JSON
Alexa interaction model
{
    "interactionModel": {
        "languageModel": {
            "invocationName": "magic show",
            "intents": [
                {
                    "name": "AMAZON.CancelIntent",
                    "samples": []
                },
                {
                    "name": "AMAZON.HelpIntent",
                    "samples": []
                },
                {
                    "name": "AMAZON.StopIntent",
                    "samples": []
                },
                {
                    "name": "BoxesIntent",
                    "slots": [
                        {
                            "name": "direction",
                            "type": "DirectionType"
                        },
                        {
                            "name": "box_number",
                            "type": "AMAZON.NUMBER"
                        }
                    ],
                    "samples": [
                        "{direction} box {box_number}",
                        "{direction} boxes"
                    ]
                },
                {
                    "name": "AMAZON.NavigateHomeIntent",
                    "samples": []
                },
                {
                    "name": "BallIntent",
                    "slots": [],
                    "samples": [
                        "move the ball"
                    ]
                },
                {
                    "name": "MagicWordIntent",
                    "slots": [
                        {
                            "name": "magic_word",
                            "type": "MagicWord"
                        }
                    ],
                    "samples": [
                        "{magic_word}"
                    ]
                },
                {
                    "name": "MagicalBoxIntent",
                    "slots": [
                        {
                            "name": "magical_box_open",
                            "type": "MagicalBoxOpen"
                        }
                    ],
                    "samples": [
                        "{magical_box_open}"
                    ]
                },
                {
                    "name": "ShowReviewIntent",
                    "slots": [
                        {
                            "name": "review",
                            "type": "Review"
                        }
                    ],
                    "samples": [
                        "show was {review}"
                    ]
                },
                {
                    "name": "BowIntent",
                    "slots": [],
                    "samples": [
                        "take a bow"
                    ]
                }
            ],
            "types": [
                {
                    "name": "DirectionType",
                    "values": [
                        {
                            "name": {
                                "value": "close"
                            }
                        },
                        {
                            "name": {
                                "value": "open"
                            }
                        }
                    ]
                },
                {
                    "name": "MagicWord",
                    "values": [
                        {
                            "name": {
                                "value": "open sesame"
                            }
                        },
                        {
                            "name": {
                                "value": "abracadabra"
                            }
                        },
                        {
                            "name": {
                                "value": "the magic word please"
                            }
                        }
                    ]
                },
                {
                    "name": "MagicalBoxOpen",
                    "values": [
                        {
                            "name": {
                                "value": "yes"
                            }
                        },
                        {
                            "name": {
                                "value": "no"
                            }
                        }
                    ]
                },
                {
                    "name": "Review",
                    "values": [
                        {
                            "name": {
                                "value": "bad"
                            }
                        },
                        {
                            "name": {
                                "value": "boring"
                            }
                        },
                        {
                            "name": {
                                "value": "okay"
                            }
                        },
                        {
                            "name": {
                                "value": "pretty good"
                            }
                        },
                        {
                            "name": {
                                "value": "amazing"
                            }
                        }
                    ]
                }
            ]
        },
        "dialog": {
            "intents": [
                {
                    "name": "BoxesIntent",
                    "confirmationRequired": false,
                    "prompts": {},
                    "slots": [
                        {
                            "name": "direction",
                            "type": "DirectionType",
                            "confirmationRequired": false,
                            "elicitationRequired": false,
                            "prompts": {}
                        },
                        {
                            "name": "box_number",
                            "type": "AMAZON.NUMBER",
                            "confirmationRequired": false,
                            "elicitationRequired": false,
                            "prompts": {},
                            "validations": [
                                {
                                    "type": "isInSet",
                                    "prompt": "Slot.Validation.486697849511.468037167142.1343421333231",
                                    "values": [
                                        "1",
                                        "2"
                                    ]
                                }
                            ]
                        }
                    ]
                },
                {
                    "name": "MagicWordIntent",
                    "confirmationRequired": false,
                    "prompts": {},
                    "slots": [
                        {
                            "name": "magic_word",
                            "type": "MagicWord",
                            "confirmationRequired": false,
                            "elicitationRequired": false,
                            "prompts": {},
                            "validations": [
                                {
                                    "type": "hasEntityResolutionMatch",
                                    "prompt": "Slot.Validation.1395753890482.740214684318.420197712174"
                                }
                            ]
                        }
                    ]
                },
                {
                    "name": "ShowReviewIntent",
                    "confirmationRequired": false,
                    "prompts": {},
                    "slots": [
                        {
                            "name": "review",
                            "type": "Review",
                            "confirmationRequired": false,
                            "elicitationRequired": true,
                            "prompts": {
                                "elicitation": "Elicit.Slot.995697757229.6470401521"
                            }
                        }
                    ]
                }
            ],
            "delegationStrategy": "ALWAYS"
        },
        "prompts": [
            {
                "id": "Slot.Validation.486697849511.468037167142.1343421333231",
                "variations": [
                    {
                        "type": "PlainText",
                        "value": "Tell me to open box one then box two"
                    }
                ]
            },
            {
                "id": "Slot.Validation.1395753890482.740214684318.420197712174",
                "variations": [
                    {
                        "type": "PlainText",
                        "value": "say a magic word please"
                    }
                ]
            },
            {
                "id": "Elicit.Slot.995697757229.6470401521",
                "variations": [
                    {
                        "type": "PlainText",
                        "value": "what did you think of the show?"
                    }
                ]
            }
        ]
    }
}

Credits

Isabel

Isabel

1 project • 5 followers
Thanks to Kees Smit and Franklin Lobb.

Comments

Awards
Finalists
LEGO MINDSTORMS Voice Challenge
First 100 Submissions
LEGO MINDSTORMS Voice Challenge
Related channels and tags
  • Alexa Gadgets
  • Alexa Gadgets
  • Amazon Alexa
  • LEGO
  • Robotics
  • Robotics