App #2: Smart Light

Hey,
I used this code example as a base to add more functionality and also fixed a few things. To summarize all the features:

  • turn on/off light after motion is detected and timeout occurred
  • timeout can be specified (used AppDaemon internal timer) or a HA countdown entity can be used (easier to show remaining time in HA)
  • entity to measure illuminance and threshold can be specified to keep the lights off if there is already enough light in the room
  • support for multiple motion sensors
  • entity to disable smartlights can be passed
  • validation checks
  • re-start countdown at startup if lights are already on (can happen if a motion is detected, light turns on and then you restart HA before lights go off)

Here is the code:

import appdaemon.plugins.hass.hassapi as hass
import appdaemon.plugins.hass
import threading

"""
App to automatically turn lights on/off for a certain time depending on the input of one or multiple motions sensors.

Args:
    light_entity - the light/switch that shall be controlled
    motion_entities - a list of motion sensors
    lux_entity (optional) - the sensor that returns the current illuminance 
    lux_limit (optional) - the light will not be turned on if the lux_entity returns a value smaller than the lux_limit
    light_timeout (optional) - the time in seconds after which the lights shall be turned off. Each motion event restarts the timer
    countdown_entity (optional) - a timer entity that can be used for the timeout handling. If this entity is set, the light_timeout argument will be ignored, because the entity already has its own timeout
    disable_motion_sensor_entity - a switch entity that can enable/disable the automatic light on/off handling
"""

class SmartLight(hass.Hass):

    def initialize(self):
        self.__single_threading_lock = threading.Lock()

        try:
            self._light_entity = self.args.get('light_entity', None)
            self._motion_entities = self.args.get('motion_entities', None)
            self._lux_entity = self.args.get('lux_entity', None)
            self._lux_limit = int(self.args.get('lux_limit', 0))
            self._light_timeout = int(self.args.get('light_timeout', 0))
            self._countdown_entity = self.args.get('countdown_entity', None)
            self._disable_motion_sensor_entity = self.args.get('disable_motion_sensor_entity', None)
        except (TypeError, ValueError):
            self.log("Invalid Configuration", level="ERROR")
            return

        if type(self._motion_entities) is not list:
            self.log('motion_entities must be a list of entities', level="ERROR")

        if self._light_entity is None:
            self.log('light entity missing.', level="ERROR")
            return

        if self._lux_limit > 0 and self._lux_entity is None:
            self.log('lux_limit given but no lux_entity specified', level="ERROR")
            return

        if self._countdown_entity is not None and self._light_timeout > 0:
            self.log('light_timeout will be ignored, because a countdown_entity is given. The timeout of the entity '+self._countdown_entity+ " will be used instead.", level="WARNING")

        if self._countdown_entity is None and self._light_timeout == 0:
            self.log('No timeout given. 5s will be used by default.', level="WARNING")
            self._light_timeout = 5

        self._timer = None
        if self._countdown_entity is not None:
            self.listen_event(self.turn_off_light, "timer.finished", entity_id = self._countdown_entity)
        if self.get_state(self._light_entity) == "on":
            self.log('light ' + self._light_entity + " is already on.")
            self.restart_timer()
        for motion_entity in self._motion_entities:
            self.listen_state(self.motion_detected_callback, motion_entity, new="on")
        if self._disable_motion_sensor_entity is not None:
            self.listen_state(self.motion_sensor_disabled_callback, self._disable_motion_sensor_entity)
        self.log('initialized')

    def is_motion_sensor_disabled(self):
        if self._disable_motion_sensor_entity is not None:
            return self.get_state(self._disable_motion_sensor_entity) == "on"
        return False 

    def motion_sensor_disabled_callback(self, entity, attribute, old, new, kwargs):
        self.log("Smartlights disabled: {}".format(self.is_motion_sensor_disabled()))
        if self.is_motion_sensor_disabled():
            self.stop_timer()
        else:
            self.restart_timer()

    def motion_detected_callback(self, entity, attribute, old, new, kwargs):
        self.motion_detected()

    def motion_detected(self):
        with self.__single_threading_lock: # prevents code from parallel execution e.g. due to multiple consecutive events
            self.log("Motion detected")
            if self.is_motion_sensor_disabled():
                self.log("... but motion sensor disabled")
                return
            if self._lux_limit > 0:
                try:
                    ambient_light = int(self.get_state(self._lux_entity))
                    self.log("Ambient light received: {} (threshold: {})".format(ambient_light, self._lux_limit))
                except (ValueError, TypeError):
                    self.log("Could not get Ambient Light Level", level="WARNING")
                    return

            light_state = self.get_state(self._light_entity)
            if light_state not in ['off', 'on']:
                self.log("Invalid Light State: {}. Expecting either on or off.".format(light_state), level="WARNING")
                return
        
            if (self._lux_limit == 0 or ambient_light <= self._lux_limit) and light_state == "off":
                self.turn_on_light()
                self.restart_timer()
            elif light_state == "on":
                self.log('Light already ON') 
                self.restart_timer()

    def stop_timer(self):
        if self._countdown_entity is not None:
            self.call_service("timer/cancel", entity_id = self._countdown_entity)
        else:
            self.cancel_timer(self._timer)

    def start_timer(self):
        if self._countdown_entity is not None:
            self.call_service("timer/start", entity_id  = self._countdown_entity)
        else:
            self._timer = self.run_in(self.turn_off_light, self._light_timeout)

    def restart_timer(self):
        if self.is_motion_sensor_disabled():
            self.log('Timer not restarted because smartlight disabled')
            return
        self.log('Restart timer')
        self.stop_timer()
        self.start_timer()

    def turn_on_light(self):
        self.log('Turning light on')
        self.turn_on(self._light_entity)

    def turn_off_light(self, event_name = None, data = None, kwargs = None):
        self.log('Timer expired')
        self._timer = None
        self.turn_off(self._light_entity)

Input configuration can be as simple as:

KitchenMotionLight:
  module: smartlight
  class: SmartLight
  light_entity: light.kitchen_lightstripe
  motion_entities:
    - binary_sensor.kitchen_movement_occupancy
  light_timeout: 300 #uses AppDaemon internal timer

or a bit more complex:

DiningRoomMotionLight:
  module: smartlight
  class: SmartLight
  light_entity: light.dining_room
  motion_entities: 
    - binary_sensor.dining_room_movement1_occupancy
    - binary_sensor.dining_room_movement2_occupancy
  lux_limit: 17000
  countdown_entity: timer.turn_off_dining_room_lights #uses HA countdown entity
  disable_motion_sensor_entity: input_boolean.disable_dining_room_motion