Fabian Affolter
Paulus Schoutsen
parent
6fcb1b548e
commit
9743e17d62
|
|
@ -0,0 +1,146 @@
|
|||
"""
|
||||
Support for displaying the minimal and the maximal value.
|
||||
|
||||
For more details about this platform, please refer to the documentation at
|
||||
https://home-assistant.io/components/sensor.min_max/
|
||||
"""
|
||||
import logging
|
||||
|
||||
import voluptuous as vol
|
||||
|
||||
from homeassistant.components.sensor import PLATFORM_SCHEMA
|
||||
from homeassistant.const import (
|
||||
CONF_NAME, STATE_UNKNOWN, CONF_TYPE, ATTR_UNIT_OF_MEASUREMENT)
|
||||
import homeassistant.helpers.config_validation as cv
|
||||
from homeassistant.helpers.entity import Entity
|
||||
from homeassistant.helpers.event import track_state_change
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
|
||||
ATTR_MIN_VALUE = 'min_value'
|
||||
ATTR_MAX_VALUE = 'max_value'
|
||||
ATTR_COUNT_SENSORS = 'count_sensors'
|
||||
ATTR_MEAN = 'mean'
|
||||
|
||||
ATTR_TO_PROPERTY = [
|
||||
ATTR_COUNT_SENSORS,
|
||||
ATTR_MAX_VALUE,
|
||||
ATTR_MEAN,
|
||||
ATTR_MIN_VALUE,
|
||||
]
|
||||
|
||||
CONF_ENTITY_IDS = 'entity_ids'
|
||||
|
||||
DEFAULT_NAME = 'Min/Max Sensor'
|
||||
|
||||
ICON = 'mdi:calculator'
|
||||
|
||||
SENSOR_TYPES = {
|
||||
ATTR_MIN_VALUE: 'min',
|
||||
ATTR_MAX_VALUE: 'max',
|
||||
ATTR_MEAN: 'mean',
|
||||
}
|
||||
|
||||
PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend({
|
||||
vol.Optional(CONF_TYPE, default=SENSOR_TYPES[ATTR_MAX_VALUE]):
|
||||
vol.All(cv.string, vol.In(SENSOR_TYPES.values())),
|
||||
vol.Optional(CONF_NAME, default=DEFAULT_NAME): cv.string,
|
||||
vol.Required(CONF_ENTITY_IDS): cv.entity_ids,
|
||||
})
|
||||
|
||||
|
||||
def setup_platform(hass, config, add_devices, discovery_info=None):
|
||||
"""Set up the min/max sensor."""
|
||||
entity_ids = config.get(CONF_ENTITY_IDS)
|
||||
name = config.get(CONF_NAME)
|
||||
sensor_type = config.get(CONF_TYPE)
|
||||
|
||||
add_devices([MinMaxSensor(hass, entity_ids, name, sensor_type)])
|
||||
|
||||
|
||||
# pylint: disable=too-many-instance-attributes
|
||||
class MinMaxSensor(Entity):
|
||||
"""Representation of a min/max sensor."""
|
||||
|
||||
def __init__(self, hass, entity_ids, name, sensor_type):
|
||||
"""Initialize the min/max sensor."""
|
||||
self._hass = hass
|
||||
self._entity_ids = entity_ids
|
||||
self._sensor_type = sensor_type
|
||||
self._name = '{} {}'.format(
|
||||
name, next(v for k, v in SENSOR_TYPES.items()
|
||||
if self._sensor_type == v))
|
||||
self._unit_of_measurement = None
|
||||
self.min_value = self.max_value = self.mean = STATE_UNKNOWN
|
||||
self.count_sensors = len(self._entity_ids)
|
||||
self.states = {}
|
||||
self.update()
|
||||
|
||||
def min_max_sensor_state_listener(entity, old_state, new_state):
|
||||
"""Called when the sensor changes state."""
|
||||
if new_state.state is None or new_state.state in STATE_UNKNOWN:
|
||||
return
|
||||
|
||||
if self._unit_of_measurement is None:
|
||||
self._unit_of_measurement = new_state.attributes.get(
|
||||
ATTR_UNIT_OF_MEASUREMENT)
|
||||
|
||||
if self._unit_of_measurement != new_state.attributes.get(
|
||||
ATTR_UNIT_OF_MEASUREMENT):
|
||||
_LOGGER.warning("Units of measurement do not match")
|
||||
return
|
||||
try:
|
||||
self.states[entity] = float(new_state.state)
|
||||
except ValueError:
|
||||
_LOGGER.warning("Unable to store state. "
|
||||
"Only numerical states are supported")
|
||||
|
||||
self.update_ha_state(True)
|
||||
|
||||
track_state_change(hass, entity_ids, min_max_sensor_state_listener)
|
||||
|
||||
@property
|
||||
def name(self):
|
||||
"""Return the name of the sensor."""
|
||||
return self._name
|
||||
|
||||
@property
|
||||
def state(self):
|
||||
"""Return the state of the sensor."""
|
||||
return getattr(self, next(
|
||||
k for k, v in SENSOR_TYPES.items() if self._sensor_type == v))
|
||||
|
||||
@property
|
||||
def unit_of_measurement(self):
|
||||
"""Return the unit the value is expressed in."""
|
||||
return self._unit_of_measurement
|
||||
|
||||
@property
|
||||
def should_poll(self):
|
||||
"""No polling needed."""
|
||||
return False
|
||||
|
||||
@property
|
||||
def device_state_attributes(self):
|
||||
"""Return the state attributes of the sensor."""
|
||||
state_attr = {
|
||||
attr: getattr(self, attr) for attr
|
||||
in ATTR_TO_PROPERTY if getattr(self, attr) is not None
|
||||
}
|
||||
return state_attr
|
||||
|
||||
@property
|
||||
def icon(self):
|
||||
"""Return the icon to use in the frontend, if any."""
|
||||
return ICON
|
||||
|
||||
def update(self):
|
||||
"""Get the latest data and updates the states."""
|
||||
sensor_values = [self.states[k] for k in self._entity_ids
|
||||
if k in self.states]
|
||||
if len(sensor_values) == self.count_sensors:
|
||||
self.min_value = min(sensor_values)
|
||||
self.max_value = max(sensor_values)
|
||||
self.mean = round(sum(sensor_values) / self.count_sensors, 2)
|
||||
else:
|
||||
self.min_value = self.max_value = self.mean = STATE_UNKNOWN
|
||||
|
|
@ -0,0 +1,161 @@
|
|||
"""The test for the min/max sensor platform."""
|
||||
import unittest
|
||||
|
||||
from homeassistant.bootstrap import setup_component
|
||||
from homeassistant.const import (
|
||||
STATE_UNKNOWN, ATTR_UNIT_OF_MEASUREMENT, TEMP_CELSIUS, TEMP_FAHRENHEIT)
|
||||
from tests.common import get_test_home_assistant
|
||||
|
||||
|
||||
class TestMinMaxSensor(unittest.TestCase):
|
||||
"""Test the min/max sensor."""
|
||||
|
||||
def setup_method(self, method):
|
||||
"""Set up things to be run when tests are started."""
|
||||
self.hass = get_test_home_assistant()
|
||||
self.values = [17, 20, 15.2]
|
||||
self.count = len(self.values)
|
||||
self.min = min(self.values)
|
||||
self.max = max(self.values)
|
||||
self.mean = round(sum(self.values) / self.count, 2)
|
||||
|
||||
def teardown_method(self, method):
|
||||
"""Stop everything that was started."""
|
||||
self.hass.stop()
|
||||
|
||||
def test_min_sensor(self):
|
||||
"""Test the min sensor."""
|
||||
config = {
|
||||
'sensor': {
|
||||
'platform': 'min_max',
|
||||
'name': 'test',
|
||||
'type': 'min',
|
||||
'entity_ids': [
|
||||
'sensor.test_1',
|
||||
'sensor.test_2',
|
||||
'sensor.test_3',
|
||||
]
|
||||
}
|
||||
}
|
||||
|
||||
assert setup_component(self.hass, 'sensor', config)
|
||||
|
||||
entity_ids = config['sensor']['entity_ids']
|
||||
|
||||
for entity_id, value in dict(zip(entity_ids, self.values)).items():
|
||||
self.hass.states.set(entity_id, value)
|
||||
self.hass.block_till_done()
|
||||
|
||||
state = self.hass.states.get('sensor.test_min')
|
||||
|
||||
self.assertEqual(str(float(self.min)), state.state)
|
||||
self.assertEqual(self.max, state.attributes.get('max_value'))
|
||||
self.assertEqual(self.mean, state.attributes.get('mean'))
|
||||
|
||||
def test_max_sensor(self):
|
||||
"""Test the max sensor."""
|
||||
config = {
|
||||
'sensor': {
|
||||
'platform': 'min_max',
|
||||
'name': 'test',
|
||||
'type': 'max',
|
||||
'entity_ids': [
|
||||
'sensor.test_1',
|
||||
'sensor.test_2',
|
||||
'sensor.test_3',
|
||||
]
|
||||
}
|
||||
}
|
||||
|
||||
assert setup_component(self.hass, 'sensor', config)
|
||||
|
||||
entity_ids = config['sensor']['entity_ids']
|
||||
|
||||
for entity_id, value in dict(zip(entity_ids, self.values)).items():
|
||||
self.hass.states.set(entity_id, value)
|
||||
self.hass.block_till_done()
|
||||
|
||||
state = self.hass.states.get('sensor.test_max')
|
||||
|
||||
self.assertEqual(str(float(self.max)), state.state)
|
||||
self.assertEqual(self.min, state.attributes.get('min_value'))
|
||||
self.assertEqual(self.mean, state.attributes.get('mean'))
|
||||
|
||||
def test_not_enough_sensor_value(self):
|
||||
"""Test that there is nothing done if not enough values available."""
|
||||
config = {
|
||||
'sensor': {
|
||||
'platform': 'min_max',
|
||||
'name': 'test',
|
||||
'type': 'max',
|
||||
'entity_ids': [
|
||||
'sensor.test_1',
|
||||
'sensor.test_2',
|
||||
'sensor.test_3',
|
||||
]
|
||||
}
|
||||
}
|
||||
|
||||
assert setup_component(self.hass, 'sensor', config)
|
||||
|
||||
entity_ids = config['sensor']['entity_ids']
|
||||
|
||||
self.hass.states.set(entity_ids[0], self.values[0])
|
||||
self.hass.block_till_done()
|
||||
|
||||
state = self.hass.states.get('sensor.test_max')
|
||||
self.assertEqual(STATE_UNKNOWN, state.state)
|
||||
|
||||
self.hass.states.set(entity_ids[1], self.values[1])
|
||||
self.hass.block_till_done()
|
||||
|
||||
state = self.hass.states.get('sensor.test_max')
|
||||
self.assertEqual(STATE_UNKNOWN, state.state)
|
||||
|
||||
self.hass.states.set(entity_ids[2], self.values[2])
|
||||
self.hass.block_till_done()
|
||||
|
||||
state = self.hass.states.get('sensor.test_max')
|
||||
self.assertNotEqual(STATE_UNKNOWN, state.state)
|
||||
|
||||
def test_different_unit_of_measurement(self):
|
||||
"""Test for different unit of measurement."""
|
||||
config = {
|
||||
'sensor': {
|
||||
'platform': 'min_max',
|
||||
'name': 'test',
|
||||
'type': 'mean',
|
||||
'entity_ids': [
|
||||
'sensor.test_1',
|
||||
'sensor.test_2',
|
||||
'sensor.test_3',
|
||||
]
|
||||
}
|
||||
}
|
||||
|
||||
assert setup_component(self.hass, 'sensor', config)
|
||||
|
||||
entity_ids = config['sensor']['entity_ids']
|
||||
|
||||
self.hass.states.set(entity_ids[0], self.values[0],
|
||||
{ATTR_UNIT_OF_MEASUREMENT: TEMP_CELSIUS})
|
||||
self.hass.block_till_done()
|
||||
|
||||
state = self.hass.states.get('sensor.test_mean')
|
||||
|
||||
self.assertEqual(STATE_UNKNOWN, state.state)
|
||||
self.assertEqual('°C', state.attributes.get('unit_of_measurement'))
|
||||
|
||||
self.hass.states.set(entity_ids[1], self.values[1],
|
||||
{ATTR_UNIT_OF_MEASUREMENT: TEMP_FAHRENHEIT})
|
||||
self.hass.block_till_done()
|
||||
|
||||
self.assertEqual(STATE_UNKNOWN, state.state)
|
||||
self.assertEqual('°C', state.attributes.get('unit_of_measurement'))
|
||||
|
||||
self.hass.states.set(entity_ids[2], self.values[2],
|
||||
{ATTR_UNIT_OF_MEASUREMENT: '%'})
|
||||
self.hass.block_till_done()
|
||||
|
||||
self.assertEqual(STATE_UNKNOWN, state.state)
|
||||
self.assertEqual('°C', state.attributes.get('unit_of_measurement'))
|
||||
Loading…
Reference in New Issue