appdeomon party lights

apps.yaml

party:
  module: party
  class: Party
  trigger_input: "input_boolean.party"
  brightness_input: "input_number.party_brightness"
  interval: 0.5
  light_group: "light.party_lights"
  restore_state_delay: 2

party.py

import appdaemon.plugins.hass.hassapi as hass
import random


class Party(hass.Hass):

    def initialize(self):
        self.previous_states = {}
        self.listen_state(self.update_mode, self.args["trigger_input"])

        self.brightness = self.get_state(self.args["brightness_input"])
        self.listen_state(self.update_brightness, self.args["brightness_input"])

        self.run_every(self.loop, "now", self.args["interval"])

        self.log("Party mode initialized")

    def _get_entities_of_group(self, entity_id, result=[]):
        group = self.get_state(entity_id, attribute="all")
        if group is not None:
            entity_ids = group["attributes"].get("entity_id")
            if entity_ids is not None:
                for entity_id in entity_ids:
                    if entity_id not in result:
                        self._get_entities_of_group(entity_id, result)
            else:
                result.append(entity_id)

    def update_mode(self, entity, attribute, old, new, kwargs):

        entity_ids = []
        self._get_entities_of_group(self.args["light_group"], entity_ids)

        if new == "on":
            for entity_id in entity_ids:
                # if entity was already turned on
                if self.previous_states.get(entity_id) != None:
                    continue

                ha_state = self.get_state(entity_id, attribute="all")

                # fix when light was not triggered by HA and has no state
                on_state = ha_state["state"]
                if on_state == None:
                    on_state = "on"

                self.previous_states[entity_id] = {
                    "state": on_state,
                    "attributes": ha_state["attributes"]
                }
        else:
            restore_list = {}
            for entity_id in entity_ids:
                state = self.previous_states.get(entity_id)

                # check if entity was on before
                if state != None:
                    del self.previous_states[entity_id]
                    restore_list[entity_id] = state

            if len(restore_list) > 0:
                # delay resetoration, because some color change events might still be in the queue
                self.run_in(lambda kwargs: self.restore_state(restore_list), self.args["restore_state_delay"])

        self.log(f"Party mode changed to {new}")

    def restore_state(self, restore_list):
        for entity_id, state in restore_list.items():
            self.set_state(entity_id, state=state["state"], attributes=state["attributes"])
            # for some reason setting the state is not enough
            self.call_service(
                'light/turn_on',
                entity_id = entity_id,
                rgb_color = state["attributes"]["rgb_color"],
                brightness = state["attributes"]["brightness"],
                transition = 0
            )
            self.log(f"Restored state for entity {entity_id} with color {state['attributes']['rgb_color']} and brightness {state['attributes']['brightness']}")

    def update_brightness(self, entity, attribute, old, new, kwargs):
        self.brightness = new
        self.log(f"Party mode brightness changed to {self.brightness}")

    def loop(self, kwargs):
        for entity_id in self.previous_states.keys():
            self.log(self.brightness)
            self.call_service(
                'light/turn_on',
                entity_id = entity_id,
                hs_color = [random.randint(0, 360), 100],
                brightness = float(self.brightness) * 2.54,
                transition = 0
            )
            # self.log(f"Party mode: {entity_id} changed color to {self.get_state(entity_id, attribute='rgb_color')}")