mirror of
https://github.com/home-assistant/core.git
synced 2025-07-25 14:17:45 +00:00
Threshold sensor (#4216)
* Add threshold sensor * New config requirement, update async, other changes, and update tests * Update threshold.py
This commit is contained in:
parent
44a508e86c
commit
1e6c660f59
128
homeassistant/components/binary_sensor/threshold.py
Normal file
128
homeassistant/components/binary_sensor/threshold.py
Normal file
@ -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)
|
98
tests/components/binary_sensor/test_threshold.py
Normal file
98
tests/components/binary_sensor/test_threshold.py
Normal file
@ -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'
|
Loading…
x
Reference in New Issue
Block a user