ef22a6e18d
* Fix statistics sensor mean and median when only one sample is available. With only one data point stddev and variance throw an exception. This would clear the (valid) mean and median calculations. Separate the try..catch blocks for one-or-more and two-or-more stats so that this doesn't happen. Test this with a new sampling_size_1 test. * test_statistics trivial whitespace fix
197 lines
7.8 KiB
Python
197 lines
7.8 KiB
Python
"""The test for the statistics sensor platform."""
|
|
import unittest
|
|
import statistics
|
|
|
|
from homeassistant.setup import setup_component
|
|
from homeassistant.const import (
|
|
ATTR_UNIT_OF_MEASUREMENT, TEMP_CELSIUS, STATE_UNKNOWN)
|
|
from homeassistant.util import dt as dt_util
|
|
from tests.common import get_test_home_assistant
|
|
from unittest.mock import patch
|
|
from datetime import datetime, timedelta
|
|
from tests.common import init_recorder_component
|
|
from homeassistant.components import recorder
|
|
|
|
|
|
class TestStatisticsSensor(unittest.TestCase):
|
|
"""Test the Statistics sensor."""
|
|
|
|
def setup_method(self, method):
|
|
"""Setup things to be run when tests are started."""
|
|
self.hass = get_test_home_assistant()
|
|
self.values = [17, 20, 15.2, 5, 3.8, 9.2, 6.7, 14, 6]
|
|
self.count = len(self.values)
|
|
self.min = min(self.values)
|
|
self.max = max(self.values)
|
|
self.total = sum(self.values)
|
|
self.mean = round(sum(self.values) / len(self.values), 2)
|
|
self.median = round(statistics.median(self.values), 2)
|
|
self.deviation = round(statistics.stdev(self.values), 2)
|
|
self.variance = round(statistics.variance(self.values), 2)
|
|
self.change = self.values[-1] - self.values[0]
|
|
self.average_change = self.change / (len(self.values) - 1)
|
|
|
|
def teardown_method(self, method):
|
|
"""Stop everything that was started."""
|
|
self.hass.stop()
|
|
|
|
def test_binary_sensor_source(self):
|
|
"""Test if source is a sensor."""
|
|
values = [1, 0, 1, 0, 1, 0, 1]
|
|
assert setup_component(self.hass, 'sensor', {
|
|
'sensor': {
|
|
'platform': 'statistics',
|
|
'name': 'test',
|
|
'entity_id': 'binary_sensor.test_monitored',
|
|
}
|
|
})
|
|
|
|
for value in values:
|
|
self.hass.states.set('binary_sensor.test_monitored', value)
|
|
self.hass.block_till_done()
|
|
|
|
state = self.hass.states.get('sensor.test_count')
|
|
|
|
self.assertEqual(str(len(values)), state.state)
|
|
|
|
def test_sensor_source(self):
|
|
"""Test if source is a sensor."""
|
|
assert setup_component(self.hass, 'sensor', {
|
|
'sensor': {
|
|
'platform': 'statistics',
|
|
'name': 'test',
|
|
'entity_id': 'sensor.test_monitored',
|
|
}
|
|
})
|
|
|
|
for value in self.values:
|
|
self.hass.states.set('sensor.test_monitored', value,
|
|
{ATTR_UNIT_OF_MEASUREMENT: TEMP_CELSIUS})
|
|
self.hass.block_till_done()
|
|
|
|
state = self.hass.states.get('sensor.test_mean')
|
|
|
|
self.assertEqual(str(self.mean), state.state)
|
|
self.assertEqual(self.min, state.attributes.get('min_value'))
|
|
self.assertEqual(self.max, state.attributes.get('max_value'))
|
|
self.assertEqual(self.variance, state.attributes.get('variance'))
|
|
self.assertEqual(self.median, state.attributes.get('median'))
|
|
self.assertEqual(self.deviation,
|
|
state.attributes.get('standard_deviation'))
|
|
self.assertEqual(self.mean, state.attributes.get('mean'))
|
|
self.assertEqual(self.count, state.attributes.get('count'))
|
|
self.assertEqual(self.total, state.attributes.get('total'))
|
|
self.assertEqual('°C', state.attributes.get('unit_of_measurement'))
|
|
self.assertEqual(self.change, state.attributes.get('change'))
|
|
self.assertEqual(self.average_change,
|
|
state.attributes.get('average_change'))
|
|
|
|
def test_sampling_size(self):
|
|
"""Test rotation."""
|
|
assert setup_component(self.hass, 'sensor', {
|
|
'sensor': {
|
|
'platform': 'statistics',
|
|
'name': 'test',
|
|
'entity_id': 'sensor.test_monitored',
|
|
'sampling_size': 5,
|
|
}
|
|
})
|
|
|
|
for value in self.values:
|
|
self.hass.states.set('sensor.test_monitored', value,
|
|
{ATTR_UNIT_OF_MEASUREMENT: TEMP_CELSIUS})
|
|
self.hass.block_till_done()
|
|
|
|
state = self.hass.states.get('sensor.test_mean')
|
|
|
|
self.assertEqual(3.8, state.attributes.get('min_value'))
|
|
self.assertEqual(14, state.attributes.get('max_value'))
|
|
|
|
def test_sampling_size_1(self):
|
|
"""Test validity of stats requiring only one sample."""
|
|
assert setup_component(self.hass, 'sensor', {
|
|
'sensor': {
|
|
'platform': 'statistics',
|
|
'name': 'test',
|
|
'entity_id': 'sensor.test_monitored',
|
|
'sampling_size': 1,
|
|
}
|
|
})
|
|
|
|
for value in self.values[-3:]: # just the last 3 will do
|
|
self.hass.states.set('sensor.test_monitored', value,
|
|
{ATTR_UNIT_OF_MEASUREMENT: TEMP_CELSIUS})
|
|
self.hass.block_till_done()
|
|
|
|
state = self.hass.states.get('sensor.test_mean')
|
|
|
|
# require only one data point
|
|
self.assertEqual(self.values[-1], state.attributes.get('min_value'))
|
|
self.assertEqual(self.values[-1], state.attributes.get('max_value'))
|
|
self.assertEqual(self.values[-1], state.attributes.get('mean'))
|
|
self.assertEqual(self.values[-1], state.attributes.get('median'))
|
|
self.assertEqual(self.values[-1], state.attributes.get('total'))
|
|
self.assertEqual(0, state.attributes.get('change'))
|
|
self.assertEqual(0, state.attributes.get('average_change'))
|
|
|
|
# require at least two data points
|
|
self.assertEqual(STATE_UNKNOWN, state.attributes.get('variance'))
|
|
self.assertEqual(STATE_UNKNOWN,
|
|
state.attributes.get('standard_deviation'))
|
|
|
|
def test_max_age(self):
|
|
"""Test value deprecation."""
|
|
mock_data = {
|
|
'return_time': datetime(2017, 8, 2, 12, 23, tzinfo=dt_util.UTC),
|
|
}
|
|
|
|
def mock_now():
|
|
return mock_data['return_time']
|
|
|
|
with patch('homeassistant.components.sensor.statistics.dt_util.utcnow',
|
|
new=mock_now):
|
|
assert setup_component(self.hass, 'sensor', {
|
|
'sensor': {
|
|
'platform': 'statistics',
|
|
'name': 'test',
|
|
'entity_id': 'sensor.test_monitored',
|
|
'max_age': {'minutes': 3}
|
|
}
|
|
})
|
|
|
|
for value in self.values:
|
|
self.hass.states.set('sensor.test_monitored', value,
|
|
{ATTR_UNIT_OF_MEASUREMENT: TEMP_CELSIUS})
|
|
self.hass.block_till_done()
|
|
# insert the next value one minute later
|
|
mock_data['return_time'] += timedelta(minutes=1)
|
|
|
|
state = self.hass.states.get('sensor.test_mean')
|
|
|
|
self.assertEqual(6, state.attributes.get('min_value'))
|
|
self.assertEqual(14, state.attributes.get('max_value'))
|
|
|
|
def test_initialize_from_database(self):
|
|
"""Test initializing the statistics from the database."""
|
|
# enable the recorder
|
|
init_recorder_component(self.hass)
|
|
# store some values
|
|
for value in self.values:
|
|
self.hass.states.set('sensor.test_monitored', value,
|
|
{ATTR_UNIT_OF_MEASUREMENT: TEMP_CELSIUS})
|
|
self.hass.block_till_done()
|
|
# wait for the recorder to really store the data
|
|
self.hass.data[recorder.DATA_INSTANCE].block_till_done()
|
|
# only now create the statistics component, so that it must read the
|
|
# data from the database
|
|
assert setup_component(self.hass, 'sensor', {
|
|
'sensor': {
|
|
'platform': 'statistics',
|
|
'name': 'test',
|
|
'entity_id': 'sensor.test_monitored',
|
|
'sampling_size': 100,
|
|
}
|
|
})
|
|
# check if the result is as in test_sensor_source()
|
|
state = self.hass.states.get('sensor.test_mean')
|
|
self.assertEqual(str(self.mean), state.state)
|