A Puppy Named Toby, with Alexa Voice Control

#!/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 ev3dev2.led import Leds
from ev3dev2.sound import Sound
from ev3dev2.motor import LargeMotor, OUTPUT_A, OUTPUT_B, OUTPUT_C, MoveTank, SpeedPercent, MediumMotor
from ev3dev2.sensor.lego import InfraredSensor

# 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 Direction(Enum):
    """
    The list of directional commands and their variations.
    These variations correspond to the skill slot values.
    """
    FORWARD = ['forward', 'forwards', 'go forward']
    BACKWARD = ['back', 'backward', 'backwards', 'go backward']
    LEFT = ['left', 'go left']
    RIGHT = ['right', 'go right']
    STOP = ['stop', 'brake', 'halt']


class Command(Enum):
    """
    The list of preset commands and their invocation variation.
    These variations correspond to the skill slot values.
    """
    BARK = ['bark', 'speak']    
    SENTRY = ['sentry']
    SIT = ['sit', 'set']
    STAY = ['stay']
    DOWN = ['down']
    DANCE = ['dance', 'dance mode']


class EventName(Enum):
    """
    The list of custom event name sent from this gadget
    """
    SENTRY = "Sentry"
    PROXIMITY = "Proximity"
    SPEECH = "Speech"


class MindstormsGadget(AlexaGadget):
    """
    A Mindstorms gadget that can perform bi-directional interaction with an Alexa skill.
    """

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

        # Connect two large motors on output ports B and C
        self.front = LargeMotor(OUTPUT_B)
        self.back = LargeMotor(OUTPUT_C)
        self.muzzle = MediumMotor(OUTPUT_A)
        self.sound = Sound()
        self.leds = Leds()
        self.ir = InfraredSensor()
        
        # Robot state
        self.sentry_mode = False
        self.dance_mode = False   
        self.front.position = 0
        self.back.position = 0
        self.muzzle.position = 0   
        self.bpm = 0
        self.trigger_bpm = "off"  

        # Start threads
        threading.Thread(target=self._proximity_thread, daemon=True).start()        
        
    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_gadget_control(self, directive):
        """
        Handles the Custom.Mindstorms.Gadget control directive.
        :param directive: the custom directive with the matching namespace and name
        """
        try:
            payload = json.loads(directive.payload.decode("utf-8"))
            print("Control payload: {}".format(payload), file=sys.stderr)
            control_type = payload["type"]
            if control_type == "move":

                # Expected params: [direction, duration, speed]
                self._move(payload["direction"], int(payload["duration"]), int(payload["speed"]))

            if control_type == "command":
                # Expected params: [command]
                self._activate(payload["command"])

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

    def _move(self, direction, duration: int, speed: int, is_blocking=False):
        """
        Handles move commands from the directive.
        Right and left movement can under or over turn depending on the surface type.
        :param direction: the move direction
        :param duration: the duration in seconds
        :param speed: the speed percentage as an integer
        :param is_blocking: if set, motor run until duration expired before accepting another command
        """
        print("Move command: ({}, {}, {}, {})".format(direction, speed, duration, is_blocking), file=sys.stderr)
        
        if direction in Direction.STOP.value:
            self.front.run_to_abs_pos(position_sp = 0, speed_sp = 100, stop_action = "coast")
            self.back.run_to_abs_pos(position_sp = 0, speed_sp = 150, stop_action = "coast")
            self.dance_mode = False

    def _activate(self, command, speed=50):
        """
        Handles preset commands.
        :param command: the preset command
        :param speed: the speed if applicable
        """
        print("Activate command: ({}, {})".format(command, speed), file=sys.stderr)
        if command in Command.SIT.value:
            print("Puppy SIT", file=sys.stderr)
            self.front.run_to_abs_pos(position_sp = 100, speed_sp = 500, stop_action = "hold")
            self.back.run_to_abs_pos(position_sp = 0, speed_sp = 150, stop_action = "brake")

        if command in Command.STAY.value:
            print("Puppy STAY", file=sys.stderr)
            self.front.run_to_abs_pos(position_sp = 100, speed_sp = 400, stop_action = "hold")
            self.back.run_to_abs_pos(position_sp = -100, speed_sp = 400, stop_action = "brake")   

        if command in Command.DOWN.value:
            print("Puppy DOWN", file=sys.stderr)
            self.front.run_to_abs_pos(position_sp = 0, speed_sp = 100, stop_action = "coast")
            self.back.run_to_abs_pos(position_sp = 0, speed_sp = 150, stop_action = "coast")

        if command in Command.DANCE.value:
            # Set dance mode to resume dance thread processing
            self.dance_mode = True

        if command in Command.SENTRY.value:
            self.sentry_mode = True
            self._send_event(EventName.SPEECH, {'speechOut': "Sentry mode activated"})
            
            print("Puppy SENTRY", file=sys.stderr)
            self.front.run_to_abs_pos(position_sp = 100, speed_sp = 400, stop_action = "hold")
            self.back.run_to_abs_pos(position_sp = -100, speed_sp = 400, stop_action = "hold") 

            self.leds.set_color("LEFT", "YELLOW", 1)
            self.leds.set_color("RIGHT", "YELLOW", 1)
        
        if command in Command.BARK.value:
            print("BARK", file=sys.stderr)
            self.sound.play_file('bark.wav', play_type=1)
            time.sleep(1.5)
            self.muzzle.on_for_degrees(50, -90, brake=True, block=True)
            self.muzzle.on_for_degrees(50, 90, brake=True, block=True)
            self.sound.play_file('bark.wav', play_type=1)
            time.sleep(1.5)
            self.muzzle.on_for_degrees(50, -90, brake=True, block=True)            
            self.muzzle.on_for_degrees(50, 90, brake=True, block=True)
            self._send_event(EventName.SENTRY, {'fire': 3})
            self.sentry_mode = False
            print("sent sentry event - alarm reset", file=sys.stderr)
            self.leds.set_color("LEFT", "GREEN", 1)
            self.leds.set_color("RIGHT", "GREEN", 1)
    
    def _send_event(self, name: EventName, payload):
        """
        Sends a custom event to trigger a sentry action.
        :param name: the name of the custom event
        :param payload: the sentry JSON payload
        """
        self.send_custom_event('Custom.Mindstorms.Gadget', name.value, payload)

    def _proximity_thread(self):
        """
        Monitors the distance between the robot and an obstacle when sentry mode is activated.
        If the minimum distance is breached, send a custom event to trigger action on
        the Alexa skill.
        """
        count = 0
        while True:
            while self.sentry_mode:
                distance = self.ir.proximity
                print("Proximity: {}".format(distance), file=sys.stderr)
                count = count + 1 if distance < 20 else 0
                if count > 3:
                    print("Proximity breached. Sending event to skill", file=sys.stderr)
                    self.leds.set_color("LEFT", "RED", 1)
                    self.leds.set_color("RIGHT", "RED", 1)

                    self._send_event(EventName.PROXIMITY, {'distance': distance})
                    self.sentry_mode = False

                time.sleep(0.2)
            time.sleep(1)     
        
    def on_alexa_gadget_musicdata_tempo(self, directive):
        """
        Provides the music tempo of the song currently playing on the Echo device.
        :param directive: the music data directive containing the beat per minute value
        """
        tempo_data = directive.payload.tempoData
        for tempo in tempo_data:

            print("tempo value: {}".format(tempo.value), file=sys.stderr)
            if tempo.value > 0:
                # dance pose
                self.front.run_to_abs_pos(position_sp = 100, speed_sp = 400, stop_action = "hold")
                self.back.run_to_abs_pos(position_sp = -100, speed_sp = 400, stop_action = "brake")
                self.leds.set_color("LEFT", "GREEN")
                self.leds.set_color("RIGHT", "GREEN")
                time.sleep(3)
                # starts the dance loop
                self.trigger_bpm = "on"
                threading.Thread(target=self._dance_loop, args=(tempo.value,)).start()

            elif tempo.value == 0:
                # stops the dance loop
                self.trigger_bpm = "off"
                self.leds.set_color("LEFT", "BLACK")
                self.leds.set_color("RIGHT", "BLACK")

    def _dance_loop(self, bpm):
        """
        Perform motor movement in sync with the beat per minute value from tempo data.
        :param bpm: beat per minute from AGT
        """
        color_list = ["GREEN", "RED", "AMBER", "YELLOW"]
        led_color = random.choice(color_list)
        milli_per_beat = min(1000, (round(60000 / bpm)) * 0.65)
        print("Adjusted milli_per_beat: {}".format(milli_per_beat), file=sys.stderr)
        while self.trigger_bpm == "on":

            # Alternate led color and motor direction
            led_color = "BLACK" if led_color != "BLACK" else random.choice(color_list)            

            self.leds.set_color("LEFT", led_color)
            self.leds.set_color("RIGHT", led_color)
            self.muzzle.on_for_degrees(50, -90, brake=True, block=True)
            self.muzzle.on_for_degrees(50, 90, brake=True, block=True)            
            time.sleep(milli_per_beat / 1000)

        print("Exiting BPM process.", 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")