Home-Assistant
/
core
mirror of https://github.com/home-assistant/core.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Make the minimum number of samples used by the trend sensor configurable (#101102) 

* Make the minimum of samples configurable & raise issue when min_samples > max_samples

* Wording

* Remove issue creation and use a custom schema validator

* Remove issue from strings.json

* Add test for validator and fix error message
pull/103660/head^2
Jan-Philipp Benecke 2023-11-30 15:41:58 +01:00 committed by GitHub
parent aa4382e091
commit f59588b413
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
3 changed files with 115 additions and 18 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

45
homeassistant/components/trend/binary_sensor.py
View File

@ -52,23 +52,39 @@ from .const import (
CONF_INVERT,
CONF_MAX_SAMPLES,
CONF_MIN_GRADIENT,
CONF_MIN_SAMPLES,
CONF_SAMPLE_DURATION,
DOMAIN,
)
_LOGGER = logging.getLogger(__name__)
SENSOR_SCHEMA = vol.Schema(
{
vol.Required(CONF_ENTITY_ID): cv.entity_id,
vol.Optional(CONF_ATTRIBUTE): cv.string,
vol.Optional(CONF_DEVICE_CLASS): DEVICE_CLASSES_SCHEMA,
vol.Optional(CONF_FRIENDLY_NAME): cv.string,
vol.Optional(CONF_INVERT, default=False): cv.boolean,
vol.Optional(CONF_MAX_SAMPLES, default=2): cv.positive_int,
vol.Optional(CONF_MIN_GRADIENT, default=0.0): vol.Coerce(float),
vol.Optional(CONF_SAMPLE_DURATION, default=0): cv.positive_int,
}
def _validate_min_max(data: dict[str, Any]) -> dict[str, Any]:
if (
CONF_MIN_SAMPLES in data
and CONF_MAX_SAMPLES in data
and data[CONF_MAX_SAMPLES] < data[CONF_MIN_SAMPLES]
):
raise vol.Invalid("min_samples must be smaller than or equal to max_samples")
return data
SENSOR_SCHEMA = vol.All(
vol.Schema(
{
vol.Required(CONF_ENTITY_ID): cv.entity_id,
vol.Optional(CONF_ATTRIBUTE): cv.string,
vol.Optional(CONF_DEVICE_CLASS): DEVICE_CLASSES_SCHEMA,
vol.Optional(CONF_FRIENDLY_NAME): cv.string,
vol.Optional(CONF_INVERT, default=False): cv.boolean,
vol.Optional(CONF_MAX_SAMPLES, default=2): cv.positive_int,
vol.Optional(CONF_MIN_GRADIENT, default=0.0): vol.Coerce(float),
vol.Optional(CONF_SAMPLE_DURATION, default=0): cv.positive_int,
vol.Optional(CONF_MIN_SAMPLES, default=2): cv.positive_int,
}
),
_validate_min_max,
)
PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend(
@ -96,6 +112,7 @@ async def async_setup_platform(
max_samples = device_config[CONF_MAX_SAMPLES]
min_gradient = device_config[CONF_MIN_GRADIENT]
sample_duration = device_config[CONF_SAMPLE_DURATION]
min_samples = device_config[CONF_MIN_SAMPLES]
sensors.append(
SensorTrend(
@ -109,8 +126,10 @@ async def async_setup_platform(
max_samples,
min_gradient,
sample_duration,
min_samples,
)
)
if not sensors:
_LOGGER.error("No sensors added")
return
@ -137,6 +156,7 @@ class SensorTrend(BinarySensorEntity, RestoreEntity):
max_samples: int,
min_gradient: float,
sample_duration: int,
min_samples: int,
) -> None:
"""Initialize the sensor."""
self._hass = hass
@ -148,6 +168,7 @@ class SensorTrend(BinarySensorEntity, RestoreEntity):
self._invert = invert
self._sample_duration = sample_duration
self._min_gradient = min_gradient
self._min_samples = min_samples
self.samples: deque = deque(maxlen=max_samples)
@property
@ -210,7 +231,7 @@ class SensorTrend(BinarySensorEntity, RestoreEntity):
while self.samples and self.samples[0][0] < cutoff:
self.samples.popleft()
if len(self.samples) < 2:
if len(self.samples) < self._min_samples:
return
# Calculate gradient of linear trend

1
homeassistant/components/trend/const.py
View File

@ -12,3 +12,4 @@ CONF_INVERT = "invert"
CONF_MAX_SAMPLES = "max_samples"
CONF_MIN_GRADIENT = "min_gradient"
CONF_SAMPLE_DURATION = "sample_duration"
CONF_MIN_SAMPLES = "min_samples"

87
tests/components/trend/test_binary_sensor.py
View File

@ -1,5 +1,6 @@
"""The test for the Trend sensor platform."""
from datetime import timedelta
import logging
from unittest.mock import patch
import pytest
@ -68,6 +69,7 @@ class TestTrendBinarySensor:
"sample_duration": 10000,
"min_gradient": 1,
"max_samples": 25,
"min_samples": 5,
}
},
}
@ -76,24 +78,35 @@ class TestTrendBinarySensor:
self.hass.block_till_done()
now = dt_util.utcnow()
# add not enough states to trigger calculation
for val in [10, 0, 20, 30]:
with patch("homeassistant.util.dt.utcnow", return_value=now):
self.hass.states.set("sensor.test_state", val)
self.hass.block_till_done()
now += timedelta(seconds=2)
state = self.hass.states.get("binary_sensor.test_trend_sensor")
assert state.state == "on"
assert (
self.hass.states.get("binary_sensor.test_trend_sensor").state == "unknown"
)
# have to change state value, otherwise sample will lost
# add one more state to trigger gradient calculation
for val in [100]:
with patch("homeassistant.util.dt.utcnow", return_value=now):
self.hass.states.set("sensor.test_state", val)
self.hass.block_till_done()
now += timedelta(seconds=2)
assert self.hass.states.get("binary_sensor.test_trend_sensor").state == "on"
# add more states to trigger a downtrend
for val in [0, 30, 1, 0]:
with patch("homeassistant.util.dt.utcnow", return_value=now):
self.hass.states.set("sensor.test_state", val)
self.hass.block_till_done()
now += timedelta(seconds=2)
state = self.hass.states.get("binary_sensor.test_trend_sensor")
assert state.state == "off"
assert self.hass.states.get("binary_sensor.test_trend_sensor").state == "off"
def test_down_using_trendline(self):
"""Test down trend using multiple samples and trendline calculation."""
@ -434,10 +447,72 @@ async def test_restore_state(
{
"binary_sensor": {
"platform": "trend",
"sensors": {"test_trend_sensor": {"entity_id": "sensor.test_state"}},
"sensors": {
"test_trend_sensor": {
"entity_id": "sensor.test_state",
"sample_duration": 10000,
"min_gradient": 1,
"max_samples": 25,
"min_samples": 5,
}
},
}
},
)
await hass.async_block_till_done()
# restored sensor should match saved one
assert hass.states.get("binary_sensor.test_trend_sensor").state == restored_state
now = dt_util.utcnow()
# add not enough samples to trigger calculation
for val in [10, 20, 30, 40]:
with patch("homeassistant.util.dt.utcnow", return_value=now):
hass.states.async_set("sensor.test_state", val)
await hass.async_block_till_done()
now += timedelta(seconds=2)
# state should match restored state as no calculation happened
assert hass.states.get("binary_sensor.test_trend_sensor").state == restored_state
# add more samples to trigger calculation
for val in [50, 60, 70, 80]:
with patch("homeassistant.util.dt.utcnow", return_value=now):
hass.states.async_set("sensor.test_state", val)
await hass.async_block_till_done()
now += timedelta(seconds=2)
# sensor should detect an upwards trend and turn on
assert hass.states.get("binary_sensor.test_trend_sensor").state == "on"
async def test_invalid_min_sample(
hass: HomeAssistant, caplog: pytest.LogCaptureFixture
) -> None:
"""Test if error is logged when min_sample is larger than max_samples."""
with caplog.at_level(logging.ERROR):
assert await setup.async_setup_component(
hass,
"binary_sensor",
{
"binary_sensor": {
"platform": "trend",
"sensors": {
"test_trend_sensor": {
"entity_id": "sensor.test_state",
"max_samples": 25,
"min_samples": 30,
}
},
}
},
)
await hass.async_block_till_done()
record = caplog.records[0]
assert record.levelname == "ERROR"
assert (
"Invalid config for 'binary_sensor.trend': min_samples must be smaller than or equal to max_samples"
in record.message
)