software_space_switch.py

#!/usr/bin/env python3

#
# Nova Labs software 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 requests, json, time, math, logging, sys

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

# log file to this script
LOG_FILE = 'software_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_GPIO = 23
PIXEL_GPIO = 18
PIXEL_COUNT = 12

PIXEL_ALL = list(range(PIXEL_COUNT))
PIXEL_HALF = math.floor(PIXEL_COUNT/2)
PIXEL_75_PCT = math.floor(PIXEL_COUNT*.75)
PIXEL_FIRST_HALF = list(range(0, PIXEL_75_PCT))
PIXEL_SECOND_HALF = list(range(PIXEL_75_PCT, PIXEL_COUNT))
PIXEL_CHAR = "*"

RED = '\033[38;5;196m'
RED_DARK = '\033[38;5;88m'
GREEN = '\033[38;5;10m'
GREEN_DARK = '\033[38;5;2m'
YELLOW = '\033[38;5;11m'
GREY = '\033[38;5;243m'
GREY_DARK = '\033[38;5;236m'
OFF = '\033[38;5;233m'
COLOR_RESET = '\033[0m'

STATE_OPEN = "open"
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")


#
# connect to WiFi
#
def novalabs_connect():
    logger.info("WIFI: connecting to wifi...")
    logger.info("WIFI: connected to wifi")

#
#
#
def print_pixels():
    global np
    pixel_str = ""
    for color in np:
        pixel_str += color + PIXEL_CHAR
    pixel_str += COLOR_RESET
    print(pixel_str)

#
# shine all pixels this color
#
def shine_all(color):
    global np
    logger.info("LED: shining all | color %s" % repr(color))
    for i in PIXEL_ALL:
        np[i] = color
    print_pixels()


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


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


# 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 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_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():
    shine_all(GREY_DARK)


# 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_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_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"])
    print(event)


#
# 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 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_state):
    novalabs_connect()
    logger.info("SWITCH: handling switch state change | state %d" % switch_state)
    if switch_state == 0:
        update_closed()
    elif switch_state == 1:
        update_open()
    else:
        logger.info("SWITCH: unknown switch state | state %d" % switch_state)


def _find_getch():
    try:
        import termios
    except ImportError:
        # Non-POSIX. Return msvcrt's (Windows') getch.
        import msvcrt
        return msvcrt.getch

    # POSIX system. Create and return a getch that manipulates the tty.
    import sys, tty
    def _getch():
        fd = sys.stdin.fileno()
        old_settings = termios.tcgetattr(fd)
        try:
            tty.setraw(fd)
            ch = sys.stdin.read(1)
        finally:
            termios.tcsetattr(fd, termios.TCSADRAIN, old_settings)
        return ch

    return _getch

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

# set all pixels to "OFF"
np = [OFF] * PIXEL_COUNT

print("===== Nova Labs Software Space Switch =====")
print("starting up ...\n")

# start with all LEDs dark grey
shine_boot()

# get latest event state from Event Service
event = get_latest_event()
# change LEDs to reflect current state (from Event Service)
shine_new_state(event["value"])
print(event)

if event["value"] == STATE_OPEN:
    switch_state = 1
elif event["value"] == STATE_CLOSED:
    switch_state = 0
else:
    switch_state = -1

getch = _find_getch()

while True:
    pos = "UNKNOWN"
    if switch_state == 1:
        pos = "ON"
    elif switch_state == 0:
        pos = "OFF"

    print("switch: " + pos)
    print("[0: off, 1: on, q: quit, any: toggle]: ")
    char = getch()
    cnum = ord(char)
    if char == 'q' or char == 'Q' or cnum == 3:
        exit()
    elif char == '0':
        switch_state = 0
    elif char == '1':
        switch_state = 1
    else:
        switch_state = (switch_state + 1) % 2

    handle_switch_change(switch_state)