binary_sensor.py

Go to the documentation of this file.

"""Support for Pilight binary sensors."""
 
from __future__ import annotations
 
import datetime
 
import voluptuous as vol
 
from homeassistant.components.binary_sensor import (
    PLATFORM_SCHEMA as BINARY_SENSOR_PLATFORM_SCHEMA,
    BinarySensorEntity,
)
from homeassistant.const import (
    CONF_DISARM_AFTER_TRIGGER,
    CONF_NAME,
    CONF_PAYLOAD,
    CONF_PAYLOAD_OFF,
    CONF_PAYLOAD_ON,
)
from homeassistant.core import HomeAssistant
from homeassistant.helpers import config_validation as cv
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.event import track_point_in_time
from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType
from homeassistant.util import dt as dt_util
 
from . import EVENT
 
CONF_VARIABLE = "variable"
CONF_RESET_DELAY_SEC = "reset_delay_sec"
 
DEFAULT_NAME = "Pilight Binary Sensor"
PLATFORM_SCHEMA = BINARY_SENSOR_PLATFORM_SCHEMA.extend(
    {
        vol.Required(CONF_VARIABLE): cv.string,
        vol.Required(CONF_PAYLOAD): vol.Schema(dict),
        vol.Optional(CONF_NAME, default=DEFAULT_NAME): cv.string,
        vol.Optional(CONF_PAYLOAD_ON, default="on"): vol.Any(
            cv.positive_int, cv.small_float, cv.string
        ),
        vol.Optional(CONF_PAYLOAD_OFF, default="off"): vol.Any(
            cv.positive_int, cv.small_float, cv.string
        ),
        vol.Optional(CONF_DISARM_AFTER_TRIGGER, default=False): cv.boolean,
        vol.Optional(CONF_RESET_DELAY_SEC, default=30): cv.positive_int,
    }
)
 
 
def setup_platform(
    hass: HomeAssistant,
    config: ConfigType,
    add_entities: AddEntitiesCallback,
    discovery_info: DiscoveryInfoType | None = None,
) -> None:
    """Set up Pilight Binary Sensor."""
    if config.get(CONF_DISARM_AFTER_TRIGGER):
        add_entities(
            [
                PilightTriggerSensor(
                    hass=hass,
                    name=config.get(CONF_NAME),
                    variable=config.get(CONF_VARIABLE),
                    payload=config.get(CONF_PAYLOAD),
                    on_value=config.get(CONF_PAYLOAD_ON),
                    off_value=config.get(CONF_PAYLOAD_OFF),
                    rst_dly_sec=config.get(CONF_RESET_DELAY_SEC),
                )
            ]
        )
    else:
        add_entities(
            [
                PilightBinarySensor(
                    hass=hass,
                    name=config.get(CONF_NAME),
                    variable=config.get(CONF_VARIABLE),
                    payload=config.get(CONF_PAYLOAD),
                    on_value=config.get(CONF_PAYLOAD_ON),
                    off_value=config.get(CONF_PAYLOAD_OFF),
                )
            ]
        )
 
 
class PilightBinarySensor(BinarySensorEntity):
    """Representation of a binary sensor that can be updated using Pilight."""
 
    def __init__(self, hass, name, variable, payload, on_value, off_value):
        """Initialize the sensor."""
        self._state_state = False
        self._hass_hass = hass
        self._name_name = name
        self._variable_variable = variable
        self._payload_payload = payload
        self._on_value_on_value = on_value
        self._off_value_off_value = off_value
 
        hass.bus.listen(EVENT, self._handle_code_handle_code)
 
    @property
    def name(self):
        """Return the name of the sensor."""
        return self._name_name
 
    @property
    def is_on(self):
        """Return True if the binary sensor is on."""
        return self._state_state
 
    def _handle_code(self, call):
        """Handle received code by the pilight-daemon.
 
        If the code matches the defined payload
        of this sensor the sensor state is changed accordingly.
        """
        # Check if received code matches defined playoad
        # True if payload is contained in received code dict
        payload_ok = True
        for key in self._payload_payload:
            if key not in call.data:
                payload_ok = False
                continue
            if self._payload_payload[key] != call.data[key]:
                payload_ok = False
        # Read out variable if payload ok
        if payload_ok:
            if self._variable_variable not in call.data:
                return
            value = call.data[self._variable_variable]
            self._state_state = value == self._on_value_on_value
            self.schedule_update_ha_stateschedule_update_ha_state()
 
 
class PilightTriggerSensor(BinarySensorEntity):
    """Representation of a binary sensor that can be updated using Pilight."""
 
    def __init__(
        self, hass, name, variable, payload, on_value, off_value, rst_dly_sec=30
    ):
        """Initialize the sensor."""
        self._state_state = False
        self._hass_hass = hass
        self._name_name = name
        self._variable_variable = variable
        self._payload_payload = payload
        self._on_value_on_value = on_value
        self._off_value_off_value = off_value
        self._reset_delay_sec_reset_delay_sec = rst_dly_sec
        self._delay_after_delay_after = None
        self._hass_hass = hass
 
        hass.bus.listen(EVENT, self._handle_code_handle_code)
 
    @property
    def name(self):
        """Return the name of the sensor."""
        return self._name_name
 
    @property
    def is_on(self):
        """Return True if the binary sensor is on."""
        return self._state_state
 
    def _reset_state(self, call):
        self._state_state = False
        self._delay_after_delay_after = None
        self.schedule_update_ha_stateschedule_update_ha_state()
 
    def _handle_code(self, call):
        """Handle received code by the pilight-daemon.
 
        If the code matches the defined payload
        of this sensor the sensor state is changed accordingly.
        """
        # Check if received code matches defined payload
        # True if payload is contained in received code dict
        payload_ok = True
        for key in self._payload_payload:
            if key not in call.data:
                payload_ok = False
                continue
            if self._payload_payload[key] != call.data[key]:
                payload_ok = False
        # Read out variable if payload ok
        if payload_ok:
            if self._variable_variable not in call.data:
                return
            value = call.data[self._variable_variable]
            self._state_state = value == self._on_value_on_value
            if self._delay_after_delay_after is None:
                self._delay_after_delay_after = dt_util.utcnow() + datetime.timedelta(
                    seconds=self._reset_delay_sec_reset_delay_sec
                )
                track_point_in_time(self._hass_hass, self._reset_state_reset_state, self._delay_after_delay_after)
            self.schedule_update_ha_stateschedule_update_ha_state()