Simple trend sensor. (#3073)

* First cut of trend sensor.

* Tidy.

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)

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() == []