countdown_timer.py

"""
Countdown Timer component for Home Assistant https://home-assistant.io/
Documentation: https://gist.github.com/JerryWorkman/d669d2da69d5861af227a4812b0928ba
Jerry Workman <jerry.workman@gmail.com>

Place this file in custom_components/sensor/countdown_timer.py

Combines one or more binary_sensors (typically PIR sensors) 
and a switch to provide a switch that automatically
turns off after a specified delay time. 

Creates a sensor that displays the time remaining before the
switch is turned off. 

If the sensors detect activity while the switch is on then the timer is
restarted.

The can be accomplished otherwise in Home Assistant. However this component reduces
the user configuration from dozens of lines of automation and scripting
to just a few lines in the configuration.yaml file.

# Example configuration.yaml entry
sensor countdown_timer:
  platform: countdown_timer
  timers:
    lr_light:
      sensors: binary_sensor.lr_pir1
      switch:  switch.lr_light_switch
"""

import string
import logging
from datetime import timedelta
import voluptuous as vol
import homeassistant.helpers.config_validation as cv
from homeassistant.components.sensor import PLATFORM_SCHEMA
from homeassistant.components import switch as switch
from homeassistant.helpers.entity import Entity
from homeassistant.const import (STATE_ON, STATE_UNKNOWN)
import homeassistant.util.dt as dt_util
from homeassistant.helpers.event import track_state_change, \
    track_point_in_time  # , track_time_interval - not yet

_LOGGER = logging.getLogger(__name__)

DOMAIN = "countdown_timer"
ICON = "mdi:timer"
UNIT_OF_MEASUREMENT = "minutes"

CONF_SENSORS = 'sensors'
CONF_SWITCH = 'switch'
CONF_DELAY = 'delay'
CONF_RESTART = 'restart'
DEFAULT_DELAY = 20
DEFAULT_RESTART = False

PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend({
    vol.Required(CONF_SENSORS, default=None): cv.string,
    vol.Required(CONF_SWITCH, default=None): cv.string,
    vol.Optional(CONF_DELAY,
                 default=DEFAULT_DELAY): cv.positive_int,
    vol.Optional(CONF_RESTART,
                 default=DEFAULT_RESTART): cv.boolean,
})


# pylint: disable=unused-argument
def setup_platform(hass, config, add_devices_callback, discovery_info=None):
    """Setup countdown_timer."""
    value_template = config.get('timers', {})
    timers = []

    for dev_name, properties in value_template.items():
        _LOGGER.debug("Adding timed switch %s", dev_name)
        timers.append(
            CountdownTimer(
                hass,
                dev_name,
                properties[CONF_SENSORS],
                properties[CONF_SWITCH],
                properties.get(CONF_DELAY, DEFAULT_DELAY),
                properties.get(CONF_RESTART, DEFAULT_RESTART),
            ))
    if not timers:
        _LOGGER.info("No countdown timers added. Check configuration.")
        return False
    add_devices_callback(timers)
    return True


class CountdownTimer(Entity):
    """
    Binary sensor(s) and a switch(light) once turned on by a sensor
    turns off after a delay.
    """

    # pylint: disable=too-many-instance-attributes,too-many-arguments
    def __init__(self, hass, dev_name, sensors, switch_id, delay, restart):
        self._name = dev_name
        full_name = DOMAIN + "." + dev_name
        self._full_name = full_name
        self._hass = hass
        self._sensors = sensors
        self._switch = switch_id
        self._timer = None
        self._delay = delay  # minutes
        self._count_down = 0  # minutes
        self._state = 0
        self.expire_time = dt_util.utcnow()
        if restart:
            # restart the countdown timers
            self._count_down = delay
            self._start_timer(dt_util.utcnow())
        # async_track_state_change is used in the core
        track_state_change(self._hass, self._sensors.split(','),
                           self._sensor_changed)

    def _debug(self, msg):
        _LOGGER.debug('%s: %s', self._name, str(msg))

    # Thanks to John Mihalic https://github.com/mezz64
    # see: https://github.com/home-assistant/home-assistant/blob/
    # 5e1e5992af38a5ad47aa44f989db568ead2cd600/homeassistant/
    # components/binary_sensor/hikvision.py

    # pylint: disable=unused-argument
    def _sensor_changed(self, entity, old_state, new_state):
        """Binary_sensor callback."""
        self._debug('new state: %s' % new_state.state)
        if new_state.state == STATE_ON:
            switch.turn_on(self._hass, self._switch)
            if self._count_down > self._delay:
                # leave the timer alone if > _delay (max time)
                self._state = self._count_down
            else:
                # reset all
                self._count_down = self._delay + 1
                self.count_down(dt_util.utcnow())

    def _start_timer(self, t_time):
        """Start HA track_point_in_time"""
        self._debug("Time now: " + str(t_time))
        self.expire_time = t_time + timedelta(seconds=60)
        self._debug("Tic - Starting timer, expires at: " + str(self.expire_time))
        # kill existing timer
        if self._timer is not None:
            self._timer()
            self._timer = None
        # start new timer
        self._timer = track_point_in_time(
            self._hass, self.count_down, self.expire_time)

    def count_down(self, t_time):
        """Timed out so turn switch off."""
        self._count_down -= 1  # minute
        self._state = int(self._count_down)  # countdowm timer
        if self._count_down <= 0:
            self._state = 0
            self._debug('Timed out')
            switch.turn_off(self._hass, self._switch)
        else:
            self._start_timer(t_time)
        # self.schedule_update_ha_state() # too slow
        self.update_ha_state()

    @property
    def name(self):
        """Return sensor name."""
        return self._name

    @property
    def should_poll(self):
        """No polling needed."""
        return False

    @property
    def friendly_name(self):
        """Return friendly sensor name - has no effect."""
        return string.capwords(self._name.replace('_', ' '))

    @property
    def unit_of_measurement(self):
        """Return the unit the value is expressed in."""
        return UNIT_OF_MEASUREMENT

    @property
    def icon(self):
        """Return the icon to use in the frontend, if any."""
        return ICON

    @property
    def state(self):
        """Get state."""
        return self._state

    def is_on(self):
        """Am I on?."""
        switch_state = self._hass.states.get(self._switch)
        if not switch_state:
            return STATE_UNKNOWN
        else:
            return switch_state == STATE_ON

    def update(self):
        """Update state, async not needed here."""
        self._state = self._count_down