From 09d52820ddd68b2b8fe6c5f838e42b1a4ec96ac2 Mon Sep 17 00:00:00 2001 From: Greg Dowling Date: Mon, 5 Sep 2016 15:32:14 +0100 Subject: [PATCH] Simple trend sensor. (#3073) * First cut of trend sensor. * Tidy. --- .../components/binary_sensor/trend.py | 145 +++++++++++ tests/components/binary_sensor/test_trend.py | 229 ++++++++++++++++++ 2 files changed, 374 insertions(+) create mode 100644 homeassistant/components/binary_sensor/trend.py create mode 100644 tests/components/binary_sensor/test_trend.py diff --git a/homeassistant/components/binary_sensor/trend.py b/homeassistant/components/binary_sensor/trend.py new file mode 100644 index 00000000000..940f80a757b --- /dev/null +++ b/homeassistant/components/binary_sensor/trend.py @@ -0,0 +1,145 @@ +""" +A sensor that monitors trands in other components. + +For more details about this platform, please refer to the documentation at +https://home-assistant.io/components/sensor.template/ +""" +import logging +import voluptuous as vol +import homeassistant.helpers.config_validation as cv + +from homeassistant.components.binary_sensor import ( + BinarySensorDevice, + ENTITY_ID_FORMAT, + PLATFORM_SCHEMA, + SENSOR_CLASSES_SCHEMA) +from homeassistant.const import ( + ATTR_FRIENDLY_NAME, + ATTR_ENTITY_ID, + CONF_SENSOR_CLASS, + STATE_UNKNOWN,) +from homeassistant.helpers.entity import generate_entity_id +from homeassistant.helpers.event import track_state_change + +_LOGGER = logging.getLogger(__name__) +CONF_SENSORS = 'sensors' +CONF_ATTRIBUTE = 'attribute' +CONF_INVERT = 'invert' + +SENSOR_SCHEMA = vol.Schema({ + vol.Required(ATTR_ENTITY_ID): cv.entity_id, + vol.Optional(CONF_ATTRIBUTE): cv.string, + vol.Optional(ATTR_FRIENDLY_NAME): cv.string, + vol.Optional(CONF_INVERT, default=False): cv.boolean, + vol.Optional(CONF_SENSOR_CLASS, default=None): SENSOR_CLASSES_SCHEMA + +}) + +PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend({ + vol.Required(CONF_SENSORS): vol.Schema({cv.slug: SENSOR_SCHEMA}), +}) + + +# pylint: disable=unused-argument +def setup_platform(hass, config, add_devices, discovery_info=None): + """Setup the template sensors.""" + sensors = [] + + for device, device_config in config[CONF_SENSORS].items(): + entity_id = device_config[ATTR_ENTITY_ID] + attribute = device_config.get(CONF_ATTRIBUTE) + friendly_name = device_config.get(ATTR_FRIENDLY_NAME, device) + sensor_class = device_config[CONF_SENSOR_CLASS] + invert = device_config[CONF_INVERT] + + sensors.append( + SensorTrend( + hass, + device, + friendly_name, + entity_id, + attribute, + sensor_class, + invert) + ) + if not sensors: + _LOGGER.error("No sensors added") + return False + add_devices(sensors) + return True + + +class SensorTrend(BinarySensorDevice): + """Representation of a Template Sensor.""" + + # pylint: disable=too-many-arguments, too-many-instance-attributes + def __init__(self, hass, device_id, friendly_name, + target_entity, attribute, sensor_class, invert): + """Initialize the sensor.""" + self._hass = hass + self.entity_id = generate_entity_id(ENTITY_ID_FORMAT, device_id, + hass=hass) + self._name = friendly_name + self._target_entity = target_entity + self._attribute = attribute + self._sensor_class = sensor_class + self._invert = invert + self._state = None + self.from_state = None + self.to_state = None + + self.update() + + def template_sensor_state_listener(entity, old_state, new_state): + """Called when the target device changes state.""" + self.from_state = old_state + self.to_state = new_state + self.update_ha_state(True) + + track_state_change(hass, target_entity, + template_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._state + + @property + def sensor_class(self): + """Return the sensor class of the sensor.""" + return self._sensor_class + + @property + def should_poll(self): + """No polling needed.""" + return False + + def update(self): + """Get the latest data and update the states.""" + if self.from_state is None or self.to_state is None: + return + if (self.from_state.state == STATE_UNKNOWN or + self.to_state.state == STATE_UNKNOWN): + return + try: + if self._attribute: + from_value = float( + self.from_state.attributes.get(self._attribute)) + to_value = float( + self.to_state.attributes.get(self._attribute)) + else: + from_value = float(self.from_state.state) + to_value = float(self.to_state.state) + + self._state = to_value > from_value + if self._invert: + self._state = not self._state + + except (ValueError, TypeError) as ex: + self._state = None + _LOGGER.error(ex) diff --git a/tests/components/binary_sensor/test_trend.py b/tests/components/binary_sensor/test_trend.py new file mode 100644 index 00000000000..beb8683e97f --- /dev/null +++ b/tests/components/binary_sensor/test_trend.py @@ -0,0 +1,229 @@ +"""The test for the Trend sensor platform.""" +import homeassistant.bootstrap as bootstrap + +from tests.common import get_test_home_assistant + + +class TestTrendBinarySensor: + """Test the Trend sensor.""" + + def setup_method(self, method): + """Setup 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_up(self): + """Test up trend.""" + assert bootstrap.setup_component(self.hass, 'binary_sensor', { + 'binary_sensor': { + 'platform': 'trend', + 'sensors': { + 'test_trend_sensor': { + 'entity_id': + "sensor.test_state" + } + } + } + }) + + self.hass.states.set('sensor.test_state', '1') + self.hass.pool.block_till_done() + self.hass.states.set('sensor.test_state', '2') + self.hass.pool.block_till_done() + state = self.hass.states.get('binary_sensor.test_trend_sensor') + assert state.state == 'on' + + def test_down(self): + """Test down trend.""" + assert bootstrap.setup_component(self.hass, 'binary_sensor', { + 'binary_sensor': { + 'platform': 'trend', + 'sensors': { + 'test_trend_sensor': { + 'entity_id': + "sensor.test_state" + } + } + } + }) + + self.hass.states.set('sensor.test_state', '2') + self.hass.pool.block_till_done() + self.hass.states.set('sensor.test_state', '1') + self.hass.pool.block_till_done() + state = self.hass.states.get('binary_sensor.test_trend_sensor') + assert state.state == 'off' + + def test__invert_up(self): + """Test up trend with custom message.""" + assert bootstrap.setup_component(self.hass, 'binary_sensor', { + 'binary_sensor': { + 'platform': 'trend', + 'sensors': { + 'test_trend_sensor': { + 'entity_id': + "sensor.test_state", + 'invert': "Yes" + } + } + } + }) + + self.hass.states.set('sensor.test_state', '1') + self.hass.pool.block_till_done() + self.hass.states.set('sensor.test_state', '2') + self.hass.pool.block_till_done() + state = self.hass.states.get('binary_sensor.test_trend_sensor') + assert state.state == 'off' + + def test_invert_down(self): + """Test down trend with custom message.""" + assert bootstrap.setup_component(self.hass, 'binary_sensor', { + 'binary_sensor': { + 'platform': 'trend', + 'sensors': { + 'test_trend_sensor': { + 'entity_id': + "sensor.test_state", + 'invert': "Yes" + } + } + } + }) + + self.hass.states.set('sensor.test_state', '2') + self.hass.pool.block_till_done() + self.hass.states.set('sensor.test_state', '1') + self.hass.pool.block_till_done() + state = self.hass.states.get('binary_sensor.test_trend_sensor') + assert state.state == 'on' + + def test_attribute_up(self): + """Test attribute up trend.""" + assert bootstrap.setup_component(self.hass, 'binary_sensor', { + 'binary_sensor': { + 'platform': 'trend', + 'sensors': { + 'test_trend_sensor': { + 'entity_id': + "sensor.test_state", + 'attribute': 'attr' + } + } + } + }) + self.hass.states.set('sensor.test_state', 'State', {'attr': '1'}) + self.hass.pool.block_till_done() + self.hass.states.set('sensor.test_state', 'State', {'attr': '2'}) + self.hass.pool.block_till_done() + state = self.hass.states.get('binary_sensor.test_trend_sensor') + assert state.state == 'on' + + def test_attribute_down(self): + """Test attribute down trend.""" + assert bootstrap.setup_component(self.hass, 'binary_sensor', { + 'binary_sensor': { + 'platform': 'trend', + 'sensors': { + 'test_trend_sensor': { + 'entity_id': + "sensor.test_state", + 'attribute': 'attr' + } + } + } + }) + + self.hass.states.set('sensor.test_state', 'State', {'attr': '2'}) + self.hass.pool.block_till_done() + self.hass.states.set('sensor.test_state', 'State', {'attr': '1'}) + + self.hass.pool.block_till_done() + state = self.hass.states.get('binary_sensor.test_trend_sensor') + assert state.state == 'off' + + def test_non_numeric(self): + """Test up trend.""" + assert bootstrap.setup_component(self.hass, 'binary_sensor', { + 'binary_sensor': { + 'platform': 'trend', + 'sensors': { + 'test_trend_sensor': { + 'entity_id': + "sensor.test_state" + } + } + } + }) + + self.hass.states.set('sensor.test_state', 'Non') + self.hass.pool.block_till_done() + self.hass.states.set('sensor.test_state', 'Numeric') + self.hass.pool.block_till_done() + state = self.hass.states.get('binary_sensor.test_trend_sensor') + assert state.state == 'off' + + def test_missing_attribute(self): + """Test attribute down trend.""" + assert bootstrap.setup_component(self.hass, 'binary_sensor', { + 'binary_sensor': { + 'platform': 'trend', + 'sensors': { + 'test_trend_sensor': { + 'entity_id': + "sensor.test_state", + 'attribute': 'missing' + } + } + } + }) + + self.hass.states.set('sensor.test_state', 'State', {'attr': '2'}) + self.hass.pool.block_till_done() + self.hass.states.set('sensor.test_state', 'State', {'attr': '1'}) + + self.hass.pool.block_till_done() + state = self.hass.states.get('binary_sensor.test_trend_sensor') + assert state.state == 'off' + + def test_invalid_name_does_not_create(self): + """Test invalid name.""" + assert not bootstrap.setup_component(self.hass, 'binary_sensor', { + 'binary_sensor': { + 'platform': 'template', + 'sensors': { + 'test INVALID sensor': { + 'entity_id': + "sensor.test_state" + } + } + } + }) + assert self.hass.states.all() == [] + + def test_invalid_sensor_does_not_create(self): + """Test invalid sensor.""" + assert not bootstrap.setup_component(self.hass, 'binary_sensor', { + 'binary_sensor': { + 'platform': 'template', + 'sensors': { + 'test_trend_sensor': { + 'not_entity_id': + "sensor.test_state" + } + } + } + }) + assert self.hass.states.all() == [] + + def test_no_sensors_does_not_create(self): + """Test no sensors.""" + assert not bootstrap.setup_component(self.hass, 'binary_sensor', { + 'binary_sensor': { + 'platform': 'trend' + } + }) + assert self.hass.states.all() == []