diff --git a/homeassistant/components/binary_sensor/threshold.py b/homeassistant/components/binary_sensor/threshold.py new file mode 100644 index 00000000000..4dc11a3c5c7 --- /dev/null +++ b/homeassistant/components/binary_sensor/threshold.py @@ -0,0 +1,128 @@ +""" +Support for monitoring if a sensor value is below/above a threshold. + +For more details about this platform, please refer to the documentation at +https://home-assistant.io/components/sensor.threshold/ +""" +import asyncio +import logging + +import voluptuous as vol + +import homeassistant.helpers.config_validation as cv +from homeassistant.components.binary_sensor import ( + BinarySensorDevice, PLATFORM_SCHEMA, SENSOR_CLASSES_SCHEMA) +from homeassistant.const import ( + CONF_NAME, CONF_ENTITY_ID, CONF_TYPE, STATE_UNKNOWN, CONF_SENSOR_CLASS, + ATTR_ENTITY_ID) +from homeassistant.core import callback +from homeassistant.helpers.event import async_track_state_change + +_LOGGER = logging.getLogger(__name__) + +ATTR_SENSOR_VALUE = 'sensor_value' +ATTR_THRESHOLD = 'threshold' +ATTR_TYPE = 'type' + +CONF_LOWER = 'lower' +CONF_THRESHOLD = 'threshold' +CONF_UPPER = 'upper' + +DEFAULT_NAME = 'Threshold' + +SENSOR_TYPES = [CONF_LOWER, CONF_UPPER] + +PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend({ + vol.Required(CONF_ENTITY_ID): cv.entity_id, + vol.Required(CONF_THRESHOLD): vol.Coerce(float), + vol.Required(CONF_TYPE): vol.In(SENSOR_TYPES), + vol.Optional(CONF_NAME, default=DEFAULT_NAME): cv.string, + vol.Optional(CONF_SENSOR_CLASS, default=None): SENSOR_CLASSES_SCHEMA, +}) + + +@asyncio.coroutine +def async_setup_platform(hass, config, async_add_devices, discovery_info=None): + """Set up the Threshold sensor.""" + entity_id = config.get(CONF_ENTITY_ID) + name = config.get(CONF_NAME) + threshold = config.get(CONF_THRESHOLD) + limit_type = config.get(CONF_TYPE) + sensor_class = config.get(CONF_SENSOR_CLASS) + + yield from async_add_devices( + [ThresholdSensor(hass, entity_id, name, threshold, limit_type, + sensor_class)], True) + return True + + +class ThresholdSensor(BinarySensorDevice): + """Representation of a Threshold sensor.""" + + def __init__(self, hass, entity_id, name, threshold, limit_type, + sensor_class): + """Initialize the Threshold sensor.""" + self._hass = hass + self._entity_id = entity_id + self.is_upper = limit_type == 'upper' + self._name = name + self._threshold = threshold + self._sensor_class = sensor_class + self._deviation = False + self.sensor_value = 0 + + @callback + # pylint: disable=invalid-name + def async_threshold_sensor_state_listener( + entity, old_state, new_state): + """Called when the sensor changes state.""" + if new_state.state == STATE_UNKNOWN: + return + + try: + self.sensor_value = float(new_state.state) + except ValueError: + _LOGGER.error("State is not numerical") + + hass.async_add_job(self.async_update_ha_state, True) + + async_track_state_change( + hass, entity_id, async_threshold_sensor_state_listener) + + @property + def name(self): + """Return the name of the sensor.""" + return self._name + + @property + def is_on(self): + """Return true if sensor is on.""" + return self._deviation + + @property + def should_poll(self): + """No polling needed.""" + return False + + @property + def sensor_class(self): + """Return the sensor class of the sensor.""" + return self._sensor_class + + @property + def state_attributes(self): + """Return the state attributes of the sensor.""" + return { + ATTR_ENTITY_ID: self._entity_id, + ATTR_SENSOR_VALUE: self.sensor_value, + ATTR_THRESHOLD: self._threshold, + ATTR_TYPE: CONF_UPPER if self.is_upper else CONF_LOWER, + } + + @asyncio.coroutine + def async_update(self): + """Get the latest data and updates the states.""" + if self.is_upper: + self._deviation = bool(self.sensor_value > self._threshold) + else: + self._deviation = bool(self.sensor_value < self._threshold) diff --git a/tests/components/binary_sensor/test_threshold.py b/tests/components/binary_sensor/test_threshold.py new file mode 100644 index 00000000000..6af2bbe5b39 --- /dev/null +++ b/tests/components/binary_sensor/test_threshold.py @@ -0,0 +1,98 @@ +"""The test for the threshold sensor platform.""" +import unittest + +from homeassistant.bootstrap import setup_component +from homeassistant.const import (ATTR_UNIT_OF_MEASUREMENT, TEMP_CELSIUS) + +from tests.common import get_test_home_assistant + + +class TestThresholdSensor(unittest.TestCase): + """Test the threshold sensor.""" + + def setup_method(self, method): + """Set up things to be run when tests are started.""" + self.hass = get_test_home_assistant() + + def teardown_method(self, method): + """Stop everything that was started.""" + self.hass.stop() + + def test_sensor_upper(self): + """Test if source is above threshold.""" + config = { + 'binary_sensor': { + 'platform': 'threshold', + 'threshold': '15', + 'type': 'upper', + 'entity_id': 'sensor.test_monitored', + } + } + + assert setup_component(self.hass, 'binary_sensor', config) + + self.hass.states.set('sensor.test_monitored', 16, + {ATTR_UNIT_OF_MEASUREMENT: TEMP_CELSIUS}) + self.hass.block_till_done() + + state = self.hass.states.get('binary_sensor.threshold') + + self.assertEqual('upper', state.attributes.get('type')) + self.assertEqual('sensor.test_monitored', + state.attributes.get('entity_id')) + self.assertEqual(16, state.attributes.get('sensor_value')) + self.assertEqual(float(config['binary_sensor']['threshold']), + state.attributes.get('threshold')) + + assert state.state == 'on' + + self.hass.states.set('sensor.test_monitored', 14) + self.hass.block_till_done() + + state = self.hass.states.get('binary_sensor.threshold') + + assert state.state == 'off' + + self.hass.states.set('sensor.test_monitored', 15) + self.hass.block_till_done() + + state = self.hass.states.get('binary_sensor.threshold') + + assert state.state == 'off' + + def test_sensor_lower(self): + """Test if source is below threshold.""" + config = { + 'binary_sensor': { + 'platform': 'threshold', + 'threshold': '15', + 'name': 'Test_threshold', + 'type': 'lower', + 'entity_id': 'sensor.test_monitored', + } + } + + assert setup_component(self.hass, 'binary_sensor', config) + + self.hass.states.set('sensor.test_monitored', 16) + self.hass.block_till_done() + + state = self.hass.states.get('binary_sensor.test_threshold') + + self.assertEqual('lower', state.attributes.get('type')) + + assert state.state == 'off' + + self.hass.states.set('sensor.test_monitored', 14) + self.hass.block_till_done() + + state = self.hass.states.get('binary_sensor.test_threshold') + + assert state.state == 'on' + + self.hass.states.set('sensor.test_monitored', 15) + self.hass.block_till_done() + + state = self.hass.states.get('binary_sensor.test_threshold') + + assert state.state == 'off'