Strict typing threshold (#82786)

This commit is contained in:
G Johansson 2023-03-12 17:10:00 +01:00 committed by GitHub
parent cf7e500a8e
commit e932139721
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
5 changed files with 86 additions and 38 deletions

View file

@ -297,6 +297,7 @@ homeassistant.components.tag.*
homeassistant.components.tailscale.* homeassistant.components.tailscale.*
homeassistant.components.tautulli.* homeassistant.components.tautulli.*
homeassistant.components.tcp.* homeassistant.components.tcp.*
homeassistant.components.threshold.*
homeassistant.components.tibber.* homeassistant.components.tibber.*
homeassistant.components.tile.* homeassistant.components.tile.*
homeassistant.components.tilt_ble.* homeassistant.components.tilt_ble.*

View file

@ -2,12 +2,14 @@
from __future__ import annotations from __future__ import annotations
import logging import logging
from typing import Any
import voluptuous as vol import voluptuous as vol
from homeassistant.components.binary_sensor import ( from homeassistant.components.binary_sensor import (
DEVICE_CLASSES_SCHEMA, DEVICE_CLASSES_SCHEMA,
PLATFORM_SCHEMA, PLATFORM_SCHEMA,
BinarySensorDeviceClass,
BinarySensorEntity, BinarySensorEntity,
) )
from homeassistant.config_entries import ConfigEntry from homeassistant.config_entries import ConfigEntry
@ -19,7 +21,7 @@ from homeassistant.const import (
STATE_UNAVAILABLE, STATE_UNAVAILABLE,
STATE_UNKNOWN, STATE_UNKNOWN,
) )
from homeassistant.core import HomeAssistant, callback from homeassistant.core import Event, HomeAssistant, callback
from homeassistant.helpers import config_validation as cv, entity_registry as er from homeassistant.helpers import config_validation as cv, entity_registry as er
from homeassistant.helpers.entity_platform import AddEntitiesCallback from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.event import async_track_state_change_event from homeassistant.helpers.event import async_track_state_change_event
@ -93,12 +95,15 @@ async def async_setup_platform(
discovery_info: DiscoveryInfoType | None = None, discovery_info: DiscoveryInfoType | None = None,
) -> None: ) -> None:
"""Set up the Threshold sensor.""" """Set up the Threshold sensor."""
entity_id = config.get(CONF_ENTITY_ID) entity_id: str = config[CONF_ENTITY_ID]
name = config.get(CONF_NAME) name: str = config[CONF_NAME]
lower = config.get(CONF_LOWER) lower: float | None = config.get(CONF_LOWER)
upper = config.get(CONF_UPPER) upper: float | None = config.get(CONF_UPPER)
hysteresis = config.get(CONF_HYSTERESIS) hysteresis: float = config[CONF_HYSTERESIS]
device_class = config.get(CONF_DEVICE_CLASS) device_class: BinarySensorDeviceClass | None = config.get(CONF_DEVICE_CLASS)
if lower is None and upper is None:
raise ValueError("Lower or Upper thresholds not provided")
async_add_entities( async_add_entities(
[ [
@ -115,22 +120,29 @@ class ThresholdSensor(BinarySensorEntity):
_attr_should_poll = False _attr_should_poll = False
def __init__( def __init__(
self, hass, entity_id, name, lower, upper, hysteresis, device_class, unique_id self,
): hass: HomeAssistant,
entity_id: str,
name: str,
lower: float | None,
upper: float | None,
hysteresis: float,
device_class: BinarySensorDeviceClass | None,
unique_id: str | None,
) -> None:
"""Initialize the Threshold sensor.""" """Initialize the Threshold sensor."""
self._attr_unique_id = unique_id self._attr_unique_id = unique_id
self._entity_id = entity_id self._entity_id = entity_id
self._name = name self._name = name
self._threshold_lower = lower self._threshold_lower = lower
self._threshold_upper = upper self._threshold_upper = upper
self._hysteresis = hysteresis self._hysteresis: float = hysteresis
self._device_class = device_class self._device_class = device_class
self._state_position = POSITION_UNKNOWN self._state_position = POSITION_UNKNOWN
self._state = None self._state: bool | None = None
self.sensor_value = None self.sensor_value: float | None = None
def _update_sensor_state(): def _update_sensor_state() -> None:
"""Handle sensor state changes.""" """Handle sensor state changes."""
if (new_state := hass.states.get(self._entity_id)) is None: if (new_state := hass.states.get(self._entity_id)) is None:
return return
@ -148,7 +160,7 @@ class ThresholdSensor(BinarySensorEntity):
self._update_state() self._update_state()
@callback @callback
def async_threshold_sensor_state_listener(event): def async_threshold_sensor_state_listener(event: Event) -> None:
"""Handle sensor state changes.""" """Handle sensor state changes."""
_update_sensor_state() _update_sensor_state()
self.async_write_ha_state() self.async_write_ha_state()
@ -161,32 +173,31 @@ class ThresholdSensor(BinarySensorEntity):
_update_sensor_state() _update_sensor_state()
@property @property
def name(self): def name(self) -> str:
"""Return the name of the sensor.""" """Return the name of the sensor."""
return self._name return self._name
@property @property
def is_on(self): def is_on(self) -> bool | None:
"""Return true if sensor is on.""" """Return true if sensor is on."""
return self._state return self._state
@property @property
def device_class(self): def device_class(self) -> BinarySensorDeviceClass | None:
"""Return the sensor class of the sensor.""" """Return the sensor class of the sensor."""
return self._device_class return self._device_class
@property @property
def threshold_type(self): def threshold_type(self) -> str:
"""Return the type of threshold this sensor represents.""" """Return the type of threshold this sensor represents."""
if self._threshold_lower is not None and self._threshold_upper is not None: if self._threshold_lower is not None and self._threshold_upper is not None:
return TYPE_RANGE return TYPE_RANGE
if self._threshold_lower is not None: if self._threshold_lower is not None:
return TYPE_LOWER return TYPE_LOWER
if self._threshold_upper is not None:
return TYPE_UPPER return TYPE_UPPER
@property @property
def extra_state_attributes(self): def extra_state_attributes(self) -> dict[str, Any]:
"""Return the state attributes of the sensor.""" """Return the state attributes of the sensor."""
return { return {
ATTR_ENTITY_ID: self._entity_id, ATTR_ENTITY_ID: self._entity_id,
@ -199,44 +210,51 @@ class ThresholdSensor(BinarySensorEntity):
} }
@callback @callback
def _update_state(self): def _update_state(self) -> None:
"""Update the state.""" """Update the state."""
def below(threshold): def below(sensor_value: float, threshold: float) -> bool:
"""Determine if the sensor value is below a threshold.""" """Determine if the sensor value is below a threshold."""
return self.sensor_value < (threshold - self._hysteresis) return sensor_value < (threshold - self._hysteresis)
def above(threshold): def above(sensor_value: float, threshold: float) -> bool:
"""Determine if the sensor value is above a threshold.""" """Determine if the sensor value is above a threshold."""
return self.sensor_value > (threshold + self._hysteresis) return sensor_value > (threshold + self._hysteresis)
if self.sensor_value is None: if self.sensor_value is None:
self._state_position = POSITION_UNKNOWN self._state_position = POSITION_UNKNOWN
self._state = False self._state = False
return
elif self.threshold_type == TYPE_LOWER: if self.threshold_type == TYPE_LOWER and self._threshold_lower is not None:
if below(self._threshold_lower): if below(self.sensor_value, self._threshold_lower):
self._state_position = POSITION_BELOW self._state_position = POSITION_BELOW
self._state = True self._state = True
elif above(self._threshold_lower): elif above(self.sensor_value, self._threshold_lower):
self._state_position = POSITION_ABOVE self._state_position = POSITION_ABOVE
self._state = False self._state = False
elif self.threshold_type == TYPE_UPPER: if self.threshold_type == TYPE_UPPER and self._threshold_upper is not None:
if above(self._threshold_upper): if above(self.sensor_value, self._threshold_upper):
self._state_position = POSITION_ABOVE self._state_position = POSITION_ABOVE
self._state = True self._state = True
elif below(self._threshold_upper): elif below(self.sensor_value, self._threshold_upper):
self._state_position = POSITION_BELOW self._state_position = POSITION_BELOW
self._state = False self._state = False
elif self.threshold_type == TYPE_RANGE: if (
if below(self._threshold_lower): self.threshold_type == TYPE_RANGE
and self._threshold_lower is not None
and self._threshold_upper is not None
):
if below(self.sensor_value, self._threshold_lower):
self._state_position = POSITION_BELOW self._state_position = POSITION_BELOW
self._state = False self._state = False
if above(self._threshold_upper): if above(self.sensor_value, self._threshold_upper):
self._state_position = POSITION_ABOVE self._state_position = POSITION_ABOVE
self._state = False self._state = False
elif above(self._threshold_lower) and below(self._threshold_upper): elif above(self.sensor_value, self._threshold_lower) and below(
self.sensor_value, self._threshold_upper
):
self._state_position = POSITION_IN_RANGE self._state_position = POSITION_IN_RANGE
self._state = True self._state = True

View file

@ -76,4 +76,5 @@ class ConfigFlowHandler(SchemaConfigFlowHandler, domain=DOMAIN):
def async_config_entry_title(self, options: Mapping[str, Any]) -> str: def async_config_entry_title(self, options: Mapping[str, Any]) -> str:
"""Return config entry title.""" """Return config entry title."""
return options[CONF_NAME] name: str = options[CONF_NAME]
return name

View file

@ -2733,6 +2733,16 @@ disallow_untyped_defs = true
warn_return_any = true warn_return_any = true
warn_unreachable = true warn_unreachable = true
[mypy-homeassistant.components.threshold.*]
check_untyped_defs = true
disallow_incomplete_defs = true
disallow_subclassing_any = true
disallow_untyped_calls = true
disallow_untyped_decorators = true
disallow_untyped_defs = true
warn_return_any = true
warn_unreachable = true
[mypy-homeassistant.components.tibber.*] [mypy-homeassistant.components.tibber.*]
check_untyped_defs = true check_untyped_defs = true
disallow_incomplete_defs = true disallow_incomplete_defs = true

View file

@ -1,4 +1,5 @@
"""The test for the threshold sensor platform.""" """The test for the threshold sensor platform."""
import pytest import pytest
from homeassistant.const import ( from homeassistant.const import (
@ -567,3 +568,20 @@ async def test_sensor_upper_zero_threshold(hass: HomeAssistant) -> None:
await hass.async_block_till_done() await hass.async_block_till_done()
state = hass.states.get("binary_sensor.threshold") state = hass.states.get("binary_sensor.threshold")
assert state.state == "on" assert state.state == "on"
async def test_sensor_no_lower_upper(
hass: HomeAssistant, caplog: pytest.LogCaptureFixture
) -> None:
"""Test if no lower or upper has been provided."""
config = {
"binary_sensor": {
"platform": "threshold",
"entity_id": "sensor.test_monitored",
}
}
await async_setup_component(hass, "binary_sensor", config)
await hass.async_block_till_done()
assert "Lower or Upper thresholds not provided" in caplog.text