main.py

#!/usr/bin/env python3

#
# Nova Labs space_switch
# This script handles the hardware space switch and updating the state in the Nova Labs Event Service.  Toggling the
# physical switch updates the Event Service with the update state and changes the physical LEDs.  The LEDs indicate the
# current state.
#
# LED indicator
# ----------------------------
# all grey            - startup
# all green           - OPEN
# all red             - CLOSED
# 2nd half grey       - changing state, updating Event Service
# 2nd half dark green - Event Service updated to open, waiting for pull from Event Service to confirm
# 2nd half dark red   - Event Service updated to closed, waiting for pull from Event Service to confirm
# 2nd half yellow     - error updating or pulling from Event Service
#
# The Event Service is a simple RESTful service for adding and retrieving events.  An event has a type, value, time,
# and UUID.  The UUID is generated by the Event Service.
#
#

import json
import logging
import math
import requests
import signal
import sys
import time
import RPi.GPIO as GPIO
import board
import neopixel

# Event Service type - this needs to be unique and not change over time
#EVENT_TYPE = "novalabs_space"
EVENT_TYPE = "test"

# log file to this script
LOG_FILE = 'space_switch.log'

#EVENT_SERVICE_BASE_URL = "http://localhost:8080"
EVENT_SERVICE_BASE_URL = "https://event.nova-labs.org"
EVENT_SERVICE_ADD_URL = EVENT_SERVICE_BASE_URL + "/events"
EVENT_SERVICE_STATUS_URL = EVENT_SERVICE_BASE_URL + "/events/" + EVENT_TYPE + "/latest"

SWITCH_ONE_GPIO = 22
SWITCH_TWO_GPIO = 27

# LED strip configuration:
LED_COUNT = 10  # Number of LED pixels.
ORDER = neopixel.GRB
pixel_pin = board.D18


PIXEL_ALL = list(range(LED_COUNT))
PIXEL_HALF = math.floor(LED_COUNT/2)
PIXEL_75_PCT = math.floor(LED_COUNT*.75)
PIXEL_FIRST_ONE = list(range(0, 0))
PIXEL_FIRST_HALF = list(range(1, PIXEL_75_PCT))
PIXEL_SECOND_HALF = list(range(PIXEL_75_PCT, LED_COUNT))

# red & green are switched from some reason, hence the mapping
RED        = (255, 0, 0)
RED_DARK   = (50, 0, 0)
GREEN      = (0, 255, 0)
GREEN_DARK = (0, 50, 0)
BLUE       = (0, 0, 255)
BLUE_DARK  = (0, 0, 50)
ORANGE      = (110, 60, 0)
ORANGE_DARK = (90, 45, 0)
YELLOW     = (100, 255, 0)
GREY       = (50, 50, 50)
GREY_DARK  = (20, 20, 20)
OFF        = (0, 0, 0)

STATE_OPEN = "open"
STATE_ASSOCIATE = "associate"
STATE_CLOSED = "closed"
STATE_ERROR = "error"
STATE_NONE = "none"
UNKNOWN_UUID = "00000000-0000-0000-0000-000000000000"

ERROR_EVENT = {"type": EVENT_TYPE, "value": STATE_ERROR, "epochMillis": 0, "uuid": UNKNOWN_UUID}
NONE_EVENT = {"type": EVENT_TYPE, "value": STATE_NONE, "epochMillis": 0, "uuid": UNKNOWN_UUID}

HTTP_OK        = 200
HTTP_CREATED   = 201
HTTP_NOT_FOUND = 404
HTTP_ERROR     = 500

LOG_FORMAT = '%(asctime)s.%(msecs)03d %(levelname)8s --- %(message)s'
logging.basicConfig(filename=LOG_FILE, filemode='a', format=LOG_FORMAT, datefmt='%Y-%m-%d %H:%M:%S', level=logging.INFO)
logger = logging.getLogger("space_switch")


def signal_handler(signal, frame):
        colorWipe(strip, OFF)
        sys.exit(0)

# Define functions which animate LEDs in various ways.
def colorWipe(strip, color, wait_ms=50):
        """Wipe color across display a pixel at a time."""
        for i in range(LED_COUNT):
                strip[i] = color
                strip.show()
                time.sleep(wait_ms/1000.0)

#
# shine all pixels this color
#
def shine_all(color):
    logger.info("LED: shining all | color %s" % repr(color))
    colorWipe(strip, color)



#
# shine second half of the LEDs to this color
#
def shine_second_half(color):
    logger.info("LED: shining second half | color %s" % repr(color))
    for i in PIXEL_SECOND_HALF:
        strip[i] = color
    strip.show()


#
# shine every other pixel to this color
#
def shine_alternate(color):
    logger.info("LED: shining alternate | color %s" % repr(color))
    for i in range(LED_COUNT)[::2]:
        strip[i] = color
    strip.show()


# turn all green to indicate Event Service latest event is open
def shine_open():
    shine_all(GREEN)


# turn all red to indicate Event Service latest event is open for associates only
def shine_associate():
    shine_all(ORANGE)


# turn all red to indicate Event Service latest event is closed
def shine_closed():
    shine_all(RED)


# change LEDs to reflect Event Service updated with new state, waiting to confirm by fetching event from Event Service
def shine_updated_open():
    shine_second_half(GREEN_DARK)


# change LEDs to reflect Event Service updated with new state, waiting to confirm by fetching event from Event Service
def shine_updated_associate():
    shine_second_half(ORANGE_DARK)


# change LEDs to reflect Event Service updated with new state, waiting to confirm by fetching event from Event Service
def shine_updated_closed():
    shine_second_half(RED_DARK)


# change LEDs to yellow to indicate error
def shine_error():
    shine_second_half(YELLOW)


# change LEDs to indicate pending update to Event Service
def shine_changing_state():
    shine_second_half(GREY)


# turn all grey on boot
def shine_boot():
    delay = 2
    colorWipe(strip, GREY_DARK)
    time.sleep(delay)
    colorWipe(strip, RED_DARK)
    time.sleep(delay)
    colorWipe(strip, YELLOW)
    time.sleep(delay)
    colorWipe(strip, GREEN_DARK)
    time.sleep(delay)
    colorWipe(strip, GREY_DARK)
    time.sleep(delay)


# turn off all LEDs
def shine_off():
    shine_all(OFF)


# change LEDs to indicate Event Service state updated, waiting on confirmation by pulling event from Event Service
def shine_updated_state(state):
    if state == STATE_OPEN:
        shine_updated_open()
    elif state == STATE_ASSOCIATE:
        shine_updated_associate()
    elif state == STATE_CLOSED:
        shine_updated_closed()
    else:
        shine_error()


# change LEDs to indicate confirmed state from Event Service latest event state
def shine_new_state(state):
    if state == STATE_OPEN:
        shine_open()
    elif state == STATE_ASSOCIATE:
        shine_associate()
    elif state == STATE_CLOSED:
        shine_closed()
    else:
        shine_error()


#
# returns now UTC as epoch milliseconds
#
def epoch_time():
    millis_float = math.floor(time.time() * 1000)
    return int(millis_float)


#
# returns the current event (state)
#
def get_latest_event():
    try:
        response = requests.get(EVENT_SERVICE_STATUS_URL)
        code = response.status_code
    except:
        code = HTTP_ERROR
        event = 0
    if code == HTTP_OK:
        # good response
        event = response.json()
    elif code == HTTP_NOT_FOUND:
        # event type doesn't exist, so return a none event
        event = NONE_EVENT
    else:
        # error getting current state
        event = ERROR_EVENT
    return event


#
# update the state
#
# 1) turn second half LEDs to indicate state is changing
# 2) send new event to Event Service
# 3) turn second half LEDs to indicate success or failure of updating event
#    success - second half is dark color of new state
#    failure - second half color is yellow
# 4) pull latest event from Event Service (to confirm change)
# 5) turn LEDs to latest state from Event Service, or error lights if can't fetch latest event
#
def update_state(state):
    # indicate changing state
    shine_changing_state()

    # create new state to update the Event Service
    event = {"type": EVENT_TYPE, "value": state, "epochMillis": epoch_time()}
    code = 0
    retries = 3
    # send update to Event Service, re-try up to 3 times
    while code != HTTP_CREATED and retries > 0:
        logger.info("UPDATE: sending update to Event Service | event %s" % json.dumps(event))
        try:
            response = requests.post(EVENT_SERVICE_ADD_URL, json=event)
            code = response.status_code
        except:
            code = HTTP_ERROR
        logger.info("UPDATE: Response to update | code %d" % code)
        retries = retries - 1
        time.sleep(1)

    if code == HTTP_CREATED:
        logger.info("UPDATE: event added to Event Service | event %s" % json.dumps(event))
        # change LEDs to indicate Event Service updated
        shine_updated_state(state)
    else:
        logger.warning("UPDATE: failed adding event to Event Service | code %d | event %s" % (code, json.dumps(event)))
        # change LEDs to indicate error
        shine_error()
        # sleep for 2 second to indicate the error
        time.sleep(2)

    # get latest event state from Event Service - normally should be the just updated event (unless update failed)
    event = get_latest_event()
    # change LEDs to reflect current state (from Event Service)
    shine_new_state(event["value"])


#
# update the new state to open
#
def update_open():
    logger.info("STATE: updating new state | state " + STATE_OPEN)
    update_state(STATE_OPEN)


#
# update the new state to associate
#
def update_associate():
    logger.info("STATE: updating new state | state " + STATE_ASSOCIATE)
    update_state(STATE_ASSOCIATE)


#
# update the new state to closed
#
def update_closed():
    logger.info("STATE: updating new state | state " + STATE_CLOSED)
    update_state(STATE_CLOSED)


#
# handle switch change
#
def handle_switch_change(switch_one_state, switch_two_state):
    logger.info("SWITCH: handling switch state change | state %d - %d" % (switch_one_state, switch_two_state))
    if switch_one_state == 1:
        if switch_two_state == 0:
            update_closed()
        else:
            logger.info("SWITCH: unknown switch state | state %d - %d" % (switch_one_state, switch_two_state))
    elif switch_two_state == 1:
        update_open()
    else:
       update_associate()



# ----------------------------------------------------------------------------
# start up
# ----------------------------------------------------------------------------
logger.info("STARTUP: starting")

# setup NeoPixel
strip = neopixel.NeoPixel(pixel_pin, LED_COUNT, brightness=0.2, auto_write=False, pixel_order=ORDER)
logger.info("STARTUP: %d neopixels on GPIO %d" % (LED_COUNT, 18))

# setup switch
GPIO.setmode(GPIO.BCM)
GPIO.setwarnings(False)
GPIO.setup(SWITCH_ONE_GPIO, GPIO.IN, pull_up_down=GPIO.PUD_DOWN)
GPIO.setup(SWITCH_TWO_GPIO, GPIO.IN, pull_up_down=GPIO.PUD_DOWN)
logger.info("STARTUP: switch on GPIOs %d & %d" % (SWITCH_ONE_GPIO, SWITCH_TWO_GPIO))

# start with all LEDs dark grey
shine_boot()

# get current switch value, update state with Event Service
#old_switch_state = switch.value()
old_switch_one_state = GPIO.input(SWITCH_ONE_GPIO)
old_switch_two_state = GPIO.input(SWITCH_TWO_GPIO)
logger.info("STARTUP: initial values %d - %d" % (old_switch_one_state, old_switch_two_state))
handle_switch_change(old_switch_one_state, old_switch_two_state)

signal.signal(signal.SIGINT, signal_handler)
signal.signal(signal.SIGABRT, signal_handler)
signal.signal(signal.SIGHUP, signal_handler)
signal.signal(signal.SIGTERM, signal_handler)

while True:
    current_switch_one_state = GPIO.input(SWITCH_ONE_GPIO)
    current_switch_two_state = GPIO.input(SWITCH_TWO_GPIO)
    if old_switch_one_state == current_switch_one_state and old_switch_two_state == current_switch_two_state:
        time.sleep(.5)
        continue
    old_switch_one_state = current_switch_one_state
    old_switch_two_state = current_switch_two_state
    logger.info("SWITCH: new values %d - %d" % (current_switch_one_state, current_switch_two_state))
    handle_switch_change(current_switch_one_state, current_switch_two_state)