ColorSchedule

color_swarm_schedule.py

"""
A collection of lighting effects that runs asynchronously on Philips Hue rooms/groups.
Pyscript must be configured to expose the "hass" global variable and allow all imports
so that we can access the Hue bridge configs and entity registry.
"""
import datetime
import heapq
import logging
import random
import time
import calendar

import homeassistant
from homeassistant.helpers import device_registry as dr
from homeassistant.helpers import entity_registry as er

devreg = homeassistant.helpers.device_registry.async_get(hass)
entreg = homeassistant.helpers.entity_registry.async_get(hass)

run_swarm = True
swarm_groups = {}
# Swarm definitions. Add your own here. To favor a particular color, add multiple instances of it to the palette.
# Max hold is the maximum number of seconds a bulb will hold its setting before transitioning to a new random color.
# The other attributes are self-explanatory, I hope.
swarms = {
    "Default": {
        "transition_secs": 0.5,
        "max_hold_secs": 3,
        "type": "random",
        "palette": [
            {
                # Warm Yellow
                "rgb_color": (255, 210, 74),
                "brightness": 100,
            },
            {
                # Soft Orange
                "rgb_color": (240, 109, 43),
                "brightness": 100,
            },
            {
                # Brownish Yellow
                "rgb_color": (158, 126, 0),
                "brightness": 100,
            }
        ],
    },

    "Valentine's Day": {
        "transition_secs": 5,
        "max_hold_secs": 15,
        "type": "all",
        "palette": [
            {
                # Red
                "rgb_color": (255, 0, 0),
                "brightness": 255,
            },
            {
                # Pink
                "rgb_color": (255, 105, 180),
                "brightness": 255,
            },
            {
                # White
                "rgb_color": (255,255,255),
                "brightness": 255,
            }
        ],
    },
    "St. Patrick's Day": {
        "transition_secs": 5,
        "max_hold_secs": 15,
        "type": "all",
        "palette": [
            {
                # Green
                "rgb_color": (0, 255, 0),
                "brightness": 255,
            },
            {
                # White
                "rgb_color": (255, 255, 255),
                "brightness": 255,
            },
            {
                # Gold
                "rgb_color": (255, 215, 0),
                "brightness": 255,
            }
        ],
    },
    "Easter": {
        "transition_secs": 5,
        "max_hold_secs": 15,
        "type": "all",
        "palette": [
            {
                # Pastel Pink
                "rgb_color": (255, 204, 229),
                "brightness": 255,
            },
            {
                # Pastel Blue
                "rgb_color": (153, 204, 255),
                "brightness": 255,
            },
            {
                # Pastel Yellow
                "rgb_color": (255, 255, 153),
                "brightness": 255,
            },
            {
                # Pastel Green
                "rgb_color": (204, 255, 204),
                "brightness": 255,
            }
        ],
    },
    "USA": {
        "transition_secs": 3,
        "max_hold_secs": 60,
        "palette": [
            {
                "rgb_color": (255, 0, 0),
                "brightness": 255,
            },
            {
                "rgb_color": (0, 0, 255),
                "brightness": 255,
            },
            {
                "rgb_color": (255, 255, 255),
                "brightness": 255,
            },
        ],
    },
    "Halloween": {
        "transition_secs": 5,
        "max_hold_secs": 15,
        "type": "all",
        "palette": [
            {
                # Orange
                "rgb_color": (252, 69, 3),
                "brightness": 255,
            },
            {
                # Puurple
                "rgb_color": (102, 0, 166),
                "brightness": 255,
            },
            {
                # Blue
                "rgb_color": (23, 0, 171),
                "brightness": 255,
            }
        ],
    },
    "Spring": {
        "transition_secs": 5,
        "max_hold_secs": 15,
        "type": "all",
        "palette": [
            {
                # Light Green
                "rgb_color": (153, 255, 153),
                "brightness": 255,
            },
            {
                # Bright Green
                "rgb_color": (0, 255, 0),
                "brightness": 255,
            },
            {
                # Dark Green
                "rgb_color": (0, 102, 0),
                "brightness": 255,
            },
            {
                # Gold
                "rgb_color": (255, 215, 0),
                "brightness": 255,
            }
        ],
    },
    "Summer": {
        "transition_secs": 5,
        "max_hold_secs": 15,
        "type": "all",
        "palette": [
            {
                # Golden Yellow
                "rgb_color": (255, 223, 0),
                "brightness": 255,
            },
            {
                # Golden Orange
                "rgb_color": (255, 165, 0),
                "brightness": 255,
            },
            {
                # Golden Brown
                "rgb_color": (222, 184, 135),
                "brightness": 255,
            }
        ],
    },
    "Fall": {
        "transition_secs": 5,
        "max_hold_secs": 10,
        "type": "all",
        "palette": [
            {
                # Red
                "rgb_color": (252, 0, 5),
                "brightness": 128
            },
            {
                # Red-Orange
                "rgb_color": (252, 50, 5),
                "brightness": 128
            },
            {
                # Orange
                "rgb_color": (255, 100, 5),
                "brightness": 128
            },
            {
                # Yellow-Orange
                "rgb_color": (250, 150, 5),
                "brightness": 128
            },
            {
                # Yellow
                "rgb_color": (250, 200, 5),
                "brightness": 128
            },
            {
                # Yellow-Orange
                "rgb_color": (250, 150, 5),
                "brightness": 128
            },
            {
                # Orange
                "rgb_color": (250, 100, 5),
                "brightness": 128
            },
            {
                # Red-Orange
                "rgb_color": (255, 50, 5),
                "brightness": 128
            }
        ],
    },
    "Thanksgiving": {
        "transition_secs": 5,
        "max_hold_secs": 15,
        "type": "all",
        "palette": [
            {
                # Orange
                "rgb_color": (255, 153, 51),
                "brightness": 255,
            },
            {
                # Brown
                "rgb_color": (139, 69, 19),
                "brightness": 255,
            },
            {
                # White
                "rgb_color": (255,255,255),
                "brightness": 255,
            }
        ],
    },
    "Christmas": {
        "transition_secs": 3,
        "max_hold_secs": 5,
        "type": "linear",
        "palette": [
            {
                # Red
                "rgb_color": (220, 0, 0),
                "brightness": 128
            },
            {
                # Green
                "rgb_color": (0, 255, 0),
                "brightness": 128
            },
            {
                # Red
                "rgb_color": (220, 0, 0),
                "brightness": 128
            },
            {
                # Orange
                "rgb_color": (225, 127, 0),
                "brightness": 128
            },
            {
                # Blue
                "rgb_color": (0, 0, 227),
                "brightness": 128
            },
        ],
    }
}



# Set our schedules here, so we just let it run, and don't have to change things seasonally.
current_year = datetime.datetime.now().year
next_year = current_year + 1

_, last_week = calendar.monthrange(current_year, 5)
last_monday = (last_week - calendar.SUNDAY) - 6
mem_start_date = datetime.date(current_year, 5, last_monday - 7)
mem_end_date = datetime.date(current_year, 5, last_week)

_, _, _, fourth_thursday, _, _, _ = calendar.monthcalendar(current_year, 11)[3]

thanks_start_date = datetime.date(current_year, 11, fourth_thursday) - datetime.timedelta(days=7)

thanks_end_date = datetime.date(current_year, 12, 1)

month = 4
day = 1
while datetime.date(current_year, month, day).weekday() != calendar.SUNDAY:
    day += 1
easter_start = datetime.date(current_year, month, day) - datetime.timedelta(days=7)
easter_end = easter_start + datetime.timedelta(days=14)

swarm_schedules = [
    (datetime.date(current_year, 2, 1), datetime.date(current_year, 2, 21), "Valentine's Day"),
    (datetime.date(current_year, 3, 10), datetime.date(current_year, 3, 21), "St. Patrick's Day"),
    (easter_start, easter_end, "Easter"),
    (mem_start_date, mem_end_date, "Memorial Day"),
    (datetime.date(current_year, 7, 3), datetime.date(current_year, 7, 7), "USA"),
    (datetime.date(current_year, 10, 1), datetime.date(current_year, 11, 7), "Halloween"),
    (thanks_start_date, thanks_end_date, "Thanksgiving"),
    (datetime.date(current_year, 12, 1), datetime.date(next_year, 1, 7), "Christmas"),
    (datetime.date(current_year, 3, 20), datetime.date(current_year, 6, 13), "Spring"),
    (datetime.date(current_year, 6, 14), datetime.date(current_year, 8, 31), "Summer"),
    (datetime.date(current_year, 9, 1), datetime.date(current_year, 9, 30), "Fall")
]

def light_entities_for_area(tgt_area_name):
    """Find light entity IDs for a specified area. Assumes all lights are color-changing.

    :param tgt_area_name: The HA Area containing the lights.
    :return: List of light entity IDs for the group name or empty set if no matching group or entities are found.
    """
    log.info(f"Searching for entities in {tgt_area_name}")
    entity_ids = []
    entities = er.async_entries_for_area(entreg, tgt_area_name)
    if entities:
        entities.extend([e for x in dr.async_entries_for_area(devreg, tgt_area_name) for e in
                         homeassistant.helpers.entity_registry.async_entries_for_device(entreg, x.id)])

    for entity in entities:
        if "light" in entity.entity_id:
            if entity is not None:
                if entity.capabilities is not None:
                    modes = entity.capabilities.get("supported_color_modes")
                    if "hs" in modes:
                        entity_ids.append(entity.entity_id)

    log.info(f"Returning id list: {sorted(entity_ids)}")
    return sorted(entity_ids)


@service
def color_swarm_turn_on(area_id="Office", swarm_name="Christmas"):
    """Start the color swarm effect on the specified Philips Hue light group.

    The color swarm continues running on the group until it is turned off or turned on with different parameters.

    :param area_id: ID Of the HA Area to control. Case-sensitive.
    :param swarm_name: The predefined swarm definition including color palette and transitions.
    """
    global run_swarm
    default_swarm = "Default"
    # Pick our swarm name from the schedule
    swarm_name = default_swarm
    current_date = datetime.datetime.now().date()
    for schedule in swarm_schedules:
        start_date, end_date, name = schedule
        if start_date <= current_date <= end_date:
            swarm_name = name
            break
    if swarm_name not in swarms:
        raise ValueError(f"Swarm '{swarm_name}' does not exist.")
    print(f"Using swarm: {swarm_name}")
    task.unique(f"color-swarm-{area_id}")
    entity_ids = light_entities_for_area(area_id)
    if entity_ids:
        log.info(
            f"Started '{swarm_name}' color swarm for area '{area_id}' consisting of {len(entity_ids)} light(s)."
        )
    else:
        log.error(f"No light entities found for area '{area_id}'.")
    swarm_groups[area_id] = entity_ids
    # Create a priority queue of the next transition per light, sorted by random future transition times.
    swarm = swarms[swarm_name]
    anim_type = "random"
    color_idx = 0
    if 'type' in swarm:
        anim_type = swarm["type"]
    ent_times = {}
    run_swarm = True
    # This will loop forever as long as the task isn't killed.
    now = time.monotonic()
    trans_time = swarm["transition_secs"]
    max_time = swarm["max_hold_secs"]
    next_time = max_time + trans_time + now
    for entity_id in entity_ids:
        if anim_type == "random":
            next_time = trans_time + random.uniform(now, now + max_time)
        ent_times[entity_id] = next_time

    start_idx = 0
    while run_swarm:
        ent_idx = 0
        color_idx = start_idx
        for entity_id in entity_ids:
            if anim_type == "random":
                sleep_time = trans_time + random.uniform(0, max_time)
                next_color = random.choice(swarm["palette"])
            else:
                sleep_time = max_time + trans_time
                if anim_type == "linear":
                    color_idx += 1
                    if ent_idx == 0:
                        start_idx += 1
                        if start_idx >= len(swarm["palette"]):
                            start_idx = 0
                else:
                    if ent_idx == 0:
                        color_idx += 1
                if color_idx >= len(swarm["palette"]):
                    color_idx = 0

                next_color = swarm["palette"][color_idx]

            ent_times[entity_id] = next_time
            light_args = {
                "entity_id": entity_id,
                "transition": trans_time,
                **next_color,
            }

            light.turn_on(**light_args)

            if anim_type is "random":
                task.sleep(sleep_time)
            else:
                if ent_idx == len(entity_ids) - 1:
                    task.sleep(sleep_time)
            ent_idx += 1
    log.info("Swarm stopped.")


@service
def color_swarm_turn_off(area_id="Office"):
    global run_swarm
    foo = False
    run_swarm = False
    """Stop any running color swarm effect on the specified area."""
    log.info(f"Stopping swarm: {area_id}")
    task.unique(f"color-swarm-{area_id}")
    if area_id in swarm_groups:
        entities = swarm_groups[area_id]
        for entity in entities:
            light_args = {
                "entity_id": entity,
                "transition": 0
            }
            light.turn_off(**light_args)
        log.info(f"Stopped lights for {len(entities)} lights.")

Like the last color swarm script, but this time, with a bunch of predefined swarms, and a schedule function that automatically swaps the swarm depending on holiday, with a fallback to a default if nothing else is found.