app.py

from datetime import datetime, timedelta
import json
import time

import ephem
from flask import Flask
import requests
from werkzeug.contrib.fixers import ProxyFix

from constants import MAIN_LIGHTS_GROUP_ID
from utils import get_light_states, set_light_states, set_group_state


app = Flask(__name__)
app.config.from_pyfile('config.py')


@app.route('/incoming-text')
def incoming_text():
    light_states = get_light_states()
    data = {
        "hue": 56100,
        "on": True,
        "bri": 255,
        "sat": 255,
        "alert": "select"
    }
    set_group_state(json.dumps(data))

    time.sleep(5)

    set_light_states(prior_states=light_states)

    return "Success"


@app.route('/incoming-call')
def incoming_call():
    light_states = get_light_states()
    data = {
        "hue": 46920,
        "on": True,
        "bri": 255,
        "sat": 255,
        "alert": "lselect"
    }
    set_group_state(json.dumps(data))

    time.sleep(29)

    set_light_states(prior_states=light_states)

    return "Success"


app.wsgi_app = ProxyFix(app.wsgi_app)


if __name__ == '__main__':
    app.run()

utils.py

from functools import partial
import json
import multiprocessing

import requests

from constants import (ALL_LIGHTS_GROUP_ID, GROUP_STATE_URL, LIGHT_IDS,
                       LIGHT_URL, LIGHT_STATE_URL)


POOL_SIZE = multiprocessing.cpu_count() * 2


def get_light_state(light_id):
    """
    Returns the current state for the provided light ID
    """
    response = requests.get(LIGHT_URL.format(light_id))
    light_state = response.json()['state']

    # If the light is off, don't change any other color settings
    if not light_state['on']:
        light_state = {'on': False}

    # Remove any alert and add a transition time
    light_state['transitiontime'] = 60
    light_state['alert'] = 'none'
    return (light_id, light_state)


def get_light_states():
    """
    Returns all lights current state as a dict
    """
    pool = multiprocessing.Pool(processes=POOL_SIZE)
    light_states = pool.map(get_light_state, LIGHT_IDS)
    pool.close() # no more tasks to process
    pool.join() # wrap up current tasks
    light_states = {id: state for id, state in light_states}
    return light_states


def set_light_state(light_id, json_data=None, prior_states=None):
    """
    Changes the state for the provided light id
    """
    if prior_states:
        prior_state = prior_states[light_id]
        json_data = json.dumps(prior_state)

    requests.put(LIGHT_STATE_URL.format(light_id), data=json_data)


def set_light_states(json_data='', prior_states=None, light_ids=LIGHT_IDS):
    """
    Changes all light states using the provided data
    """
    pool = multiprocessing.Pool(processes=POOL_SIZE)
    if prior_states:
        partial_set_light_state = partial(set_light_state,
                                          prior_states=prior_states)
    else:
        partial_set_light_state = partial(set_light_state, json_data=json_data)
    pool.map(partial_set_light_state, light_ids)
    pool.close() # no more tasks to process
    pool.join() # wrap up current tasks


def set_group_state(json_data, group_id=ALL_LIGHTS_GROUP_ID):
    """
    Sets the state of a group
    """
    r = requests.put(GROUP_STATE_URL.format(group_id), data=json_data)