This is my first try to do a automation in AppDaemon that uses schedules to turn on and off lights with the optional behaviour of having a sensor that when active delays the turning off the lights.
This is a work in progress, any feedback is welcome. I need to add support for multiple sensors, and sensor functionality is mostly written with media players in mind right now.
import appdaemon.appapi as appapi
#
# Appdaemon light schedules App
#
# Schedules lights for on and off switching, also allows off switch to be overridden by sensor
#
# Args:
#
# time: ', ' separated list with one or more time and action
# example: time = 05:00:00 on, sunrise off
# light: ', ' separated list with one or more light entity_id
# example: light = light.lamp1, light.lamp2
# group: ', ' separated list with one or more group containing light
# example: group = group.lightgroup1, group.lightgroup2
# sensor: single sensor that can delay schedule when turning off
#
# Note: takes inspiration from AppDaemon motion detectionlights
class lighting(appapi.AppDaemon):
def initialize(self):
self.lights = []
self.schedules = {}
self.sensors = None
self.sensor_active = False
self.sensor_activity_handle = None
self.schedule_active = False
self.populate_lights()
self.populate_schedules()
self.populate_sensors()
self.schedule_callback()
self.log("Initialized")
def run_daily_callback(self, kwargs):
action = kwargs["action"] if "action" in kwargs else False
trigger = kwargs["trigger"] if "trigger" in kwargs else False
if self.sensors and not self.sensor(action, trigger): return
for light in self.lights:
self.call_service("light/turn_"+action, entity_id = light)
self.log("Turned {} {}".format(action, self.lights))
#########################################################################################
#
# Sensor functions
#
# Sensor is set up to listen when action is on
# Sensor will stop listen on two different situations
# 1. Sensor is low before schedule has triggered
# 2. Sensor is low after
def sensor(self, action, trigger):
if action == "on":
self.sensor_activity_handle = self.listen_state(self.sensor_activity, self.sensors)
self.log("Listening on sensor {}".format(self.sensors))
self.schedule_active = True
elif action == "off":
# Schedule hasn't triggered yet, sensor is not allowed to turn off light
if trigger == "sensor" and self.schedule_active:
self.sensor_active = False
self.log("Sensor tried to turn off lights but it is too early")
return False
# Schedule has triggered and sensor is active, sensor is now responsible for turning off light
elif trigger == "schedule" and self.sensor_active:
self.schedule_active = False
self.log("Sensor is now allowed to turn off lights")
return False
# Everything is ready to turn off
else:
self.cancel_listen_state(self.sensor_activity_handle)
#self.allow_sensor_override = False #self.schedule_active = False
self.schedule_active = False
self.sensor_active = False
return True
# Triggers on sensor activity
def sensor_activity(self, entity, attribute, old, new, kwargs):
self.log("Sensor {} changed state to {}".format(entity, new))
if self.sensor_active:
self.cancel_timer(self.sensor_active)
if new == "idle":
self.sensor_active = self.run_in(self.run_daily_callback, 450, action = "off", trigger = "sensor")
else:
self.sensor_active = True
#########################################################################################
#
# Initalization functions
#
# Splits configured list "light" into a global python list "self.lights"
# Resolves lights from configured list "group" into a global python list "self.lights"
def populate_lights(self):
if "light" in self.args:
self.lights = self.args["light"].split(", ")
if "group" in self.args:
for group in self.args["group"].split(", "):
for light in self.resolve_group(group):
if not light in self.lights:
self.lights.append(light)
self.lights.sort()
self.log("Lights for schedules {}".format(self.lights))
# Splits a configured list of schedules with actions into a dict "self.schedules"
def populate_schedules(self):
if "time" in self.args:
for schedule in self.args["time"].split(", "):
time, action = schedule.split(" ")
self.schedules[time] = action
self.log("Schedules and actions {}".format(self.schedules))
def populate_sensors(self):
if "sensor" in self.args:
self.sensors = self.args["sensor"]
#self.sensors = self.args["sensor"].split(", ")
self.log("Sensors that can override schedule {}".format(self.sensors))
def schedule_callback(self):
for time, action in self.schedules.items():
time = self.parse_time(time)
self.run_daily(self.run_daily_callback, time, action = action, trigger = "schedule")
#########################################################################################
#
# Support functions
#
# Returns a list of entity_ids from the group
def resolve_group(self, group):
group = self.get_state(group, attribute = "all")
return group["attributes"]["entity_id"]
