self-hosted/home-assistant-core
self-hosted/home-assistant-core
1
0
Fork 0
mirror of https://github.com/home-assistant/core synced 2024-10-01 18:09:26 +00:00
Code Issues Packages Projects Releases Wiki Activity

Fully type min_max (#79496)

Browse source
This commit is contained in:
G Johansson 2022-11-23 19:54:12 +01:00 committed by GitHub
parent 97b40b5f49
commit 8704f1aa47
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
5 changed files with 86 additions and 47 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

1
.strict-typing
View file

@ -182,6 +182,7 @@ homeassistant.components.media_player.*
homeassistant.components.media_source.*
homeassistant.components.metoffice.*
homeassistant.components.mikrotik.*
homeassistant.components.min_max.*
homeassistant.components.mjpeg.*
homeassistant.components.modbus.*
homeassistant.components.modem_callerid.*

92
homeassistant/components/min_max/sensor.py
View file

@ -1,8 +1,10 @@
"""Support for displaying minimal, maximal, mean or median values."""
from __future__ import annotations
from datetime import datetime
import logging
import statistics
from typing import Any
import voluptuous as vol
@ -20,12 +22,17 @@ from homeassistant.const import (
STATE_UNAVAILABLE,
STATE_UNKNOWN,
)
from homeassistant.core import Event, HomeAssistant, callback
from homeassistant.core import Event, HomeAssistant, State, callback
from homeassistant.helpers import config_validation as cv, entity_registry as er
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.event import async_track_state_change_event
from homeassistant.helpers.reload import async_setup_reload_service
from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType
from homeassistant.helpers.typing import (
ConfigType,
DiscoveryInfoType,
EventType,
StateType,
)
from . import PLATFORMS
from .const import CONF_ENTITY_IDS, CONF_ROUND_DIGITS, DOMAIN
@ -62,7 +69,7 @@ PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend(
vol.Optional(CONF_NAME): cv.string,
vol.Required(CONF_ENTITY_IDS): cv.entity_ids,
vol.Optional(CONF_ROUND_DIGITS, default=2): vol.Coerce(int),
vol.Optional(CONF_UNIQUE_ID): str,
vol.Optional(CONF_UNIQUE_ID): cv.string,
}
)
@ -100,10 +107,10 @@ async def async_setup_platform(
discovery_info: DiscoveryInfoType | None = None,
) -> None:
"""Set up the min/max/mean sensor."""
entity_ids = config.get(CONF_ENTITY_IDS)
name = config.get(CONF_NAME)
sensor_type = config.get(CONF_TYPE)
round_digits = config.get(CONF_ROUND_DIGITS)
entity_ids: list[str] = config[CONF_ENTITY_IDS]
name: str | None = config.get(CONF_NAME)
sensor_type: str = config[CONF_TYPE]
round_digits: int = config[CONF_ROUND_DIGITS]
unique_id = config.get(CONF_UNIQUE_ID)
await async_setup_reload_service(hass, DOMAIN, PLATFORMS)
@ -113,10 +120,10 @@ async def async_setup_platform(
)
def calc_min(sensor_values):
def calc_min(sensor_values: list[tuple[str, Any]]) -> tuple[str | None, float | None]:
"""Calculate min value, honoring unknown states."""
val = None
entity_id = None
val: float | None = None
entity_id: str | None = None
for sensor_id, sensor_value in sensor_values:
if sensor_value not in [STATE_UNKNOWN, STATE_UNAVAILABLE] and (
val is None or val > sensor_value
@ -125,10 +132,10 @@ def calc_min(sensor_values):
return entity_id, val
def calc_max(sensor_values):
def calc_max(sensor_values: list[tuple[str, Any]]) -> tuple[str | None, float | None]:
"""Calculate max value, honoring unknown states."""
val = None
entity_id = None
val: float | None = None
entity_id: str | None = None
for sensor_id, sensor_value in sensor_values:
if sensor_value not in [STATE_UNKNOWN, STATE_UNAVAILABLE] and (
val is None or val < sensor_value
@ -137,7 +144,7 @@ def calc_max(sensor_values):
return entity_id, val
def calc_mean(sensor_values, round_digits):
def calc_mean(sensor_values: list[tuple[str, Any]], round_digits: int) -> float | None:
"""Calculate mean value, honoring unknown states."""
result = [
sensor_value
@ -147,10 +154,13 @@ def calc_mean(sensor_values, round_digits):
if not result:
return None
return round(statistics.mean(result), round_digits)
value: float = round(statistics.mean(result), round_digits)
return value
def calc_median(sensor_values, round_digits):
def calc_median(
sensor_values: list[tuple[str, Any]], round_digits: int
) -> float | None:
"""Calculate median value, honoring unknown states."""
result = [
sensor_value
@ -160,10 +170,11 @@ def calc_median(sensor_values, round_digits):
if not result:
return None
return round(statistics.median(result), round_digits)
value: float = round(statistics.median(result), round_digits)
return value
def calc_range(sensor_values, round_digits):
def calc_range(sensor_values: list[tuple[str, Any]], round_digits: int) -> float | None:
"""Calculate range value, honoring unknown states."""
result = [
sensor_value
@ -173,7 +184,8 @@ def calc_range(sensor_values, round_digits):
if not result:
return None
return round(max(result) - min(result), round_digits)
value: float = round(max(result) - min(result), round_digits)
return value
class MinMaxSensor(SensorEntity):
@ -183,7 +195,14 @@ class MinMaxSensor(SensorEntity):
_attr_should_poll = False
_attr_state_class = SensorStateClass.MEASUREMENT
def __init__(self, entity_ids, name, sensor_type, round_digits, unique_id):
def __init__(
self,
entity_ids: list[str],
name: str | None,
sensor_type: str,
round_digits: int,
unique_id: str | None,
) -> None:
"""Initialize the min/max sensor."""
self._attr_unique_id = unique_id
self._entity_ids = entity_ids
@ -197,11 +216,17 @@ class MinMaxSensor(SensorEntity):
self._sensor_attr = SENSOR_TYPE_TO_ATTR[self._sensor_type]
self._unit_of_measurement = None
self._unit_of_measurement_mismatch = False
self.min_value = self.max_value = self.mean = self.last = self.median = None
self.range = None
self.min_entity_id = self.max_entity_id = self.last_entity_id = None
self.min_value: float | None = None
self.max_value: float | None = None
self.mean: float | None = None
self.last: float | None = None
self.median: float | None = None
self.range: float | None = None
self.min_entity_id: str | None = None
self.max_entity_id: str | None = None
self.last_entity_id: str | None = None
self.count_sensors = len(self._entity_ids)
self.states = {}
self.states: dict[str, Any] = {}
async def async_added_to_hass(self) -> None:
"""Handle added to Hass."""
@ -220,21 +245,22 @@ class MinMaxSensor(SensorEntity):
self._calc_values()
@property
def native_value(self):
def native_value(self) -> StateType | datetime:
"""Return the state of the sensor."""
if self._unit_of_measurement_mismatch:
return None
return getattr(self, self._sensor_attr)
value: StateType | datetime = getattr(self, self._sensor_attr)
return value
@property
def native_unit_of_measurement(self):
def native_unit_of_measurement(self) -> str | None:
"""Return the unit the value is expressed in."""
if self._unit_of_measurement_mismatch:
return "ERR"
return self._unit_of_measurement
@property
def extra_state_attributes(self):
def extra_state_attributes(self) -> dict[str, Any] | None:
"""Return the state attributes of the sensor."""
if self._sensor_type == "min":
return {ATTR_MIN_ENTITY_ID: self.min_entity_id}
@ -245,10 +271,12 @@ class MinMaxSensor(SensorEntity):
return None
@callback
def _async_min_max_sensor_state_listener(self, event, update_state=True):
def _async_min_max_sensor_state_listener(
self, event: EventType, update_state: bool = True
) -> None:
"""Handle the sensor state changes."""
new_state = event.data.get("new_state")
entity = event.data.get("entity_id")
new_state: State | None = event.data.get("new_state")
entity: str = event.data["entity_id"]
if (
new_state is None
@ -296,7 +324,7 @@ class MinMaxSensor(SensorEntity):
self.async_write_ha_state()
@callback
def _calc_values(self):
def _calc_values(self) -> None:
"""Calculate the values."""
sensor_values = [
(entity_id, self.states[entity_id])

10
mypy.ini
View file

@ -1573,6 +1573,16 @@ disallow_untyped_defs = true
warn_return_any = true
warn_unreachable = true
[mypy-homeassistant.components.min_max.*]
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.mjpeg.*]
check_untyped_defs = true
disallow_incomplete_defs = true

4
tests/components/min_max/test_config_flow.py
View file

@ -12,7 +12,7 @@ from tests.common import MockConfigEntry
@pytest.mark.parametrize("platform", ("sensor",))
async def test_config_flow(hass: HomeAssistant, platform) -> None:
async def test_config_flow(hass: HomeAssistant, platform: str) -> None:
"""Test the config flow."""
input_sensors = ["sensor.input_one", "sensor.input_two"]
@ -66,7 +66,7 @@ def get_suggested(schema, key):
@pytest.mark.parametrize("platform", ("sensor",))
async def test_options(hass: HomeAssistant, platform) -> None:
async def test_options(hass: HomeAssistant, platform: str) -> None:
"""Test reconfiguring."""
hass.states.async_set("sensor.input_one", "10")
hass.states.async_set("sensor.input_two", "20")

26
tests/components/min_max/test_sensor.py
View file

@ -35,7 +35,7 @@ RANGE_1_DIGIT = round(max(VALUES) - min(VALUES), 1)
RANGE_4_DIGITS = round(max(VALUES) - min(VALUES), 4)
async def test_default_name_sensor(hass):
async def test_default_name_sensor(hass: HomeAssistant) -> None:
"""Test the min sensor with a default name."""
config = {
"sensor": {
@ -60,7 +60,7 @@ async def test_default_name_sensor(hass):
assert entity_ids[2] == state.attributes.get("min_entity_id")
async def test_min_sensor(hass):
async def test_min_sensor(hass: HomeAssistant) -> None:
"""Test the min sensor."""
config = {
"sensor": {
@ -92,7 +92,7 @@ async def test_min_sensor(hass):
assert entity.unique_id == "very_unique_id"
async def test_max_sensor(hass):
async def test_max_sensor(hass: HomeAssistant) -> None:
"""Test the max sensor."""
config = {
"sensor": {
@ -119,7 +119,7 @@ async def test_max_sensor(hass):
assert state.attributes.get(ATTR_STATE_CLASS) == SensorStateClass.MEASUREMENT
async def test_mean_sensor(hass):
async def test_mean_sensor(hass: HomeAssistant) -> None:
"""Test the mean sensor."""
config = {
"sensor": {
@ -145,7 +145,7 @@ async def test_mean_sensor(hass):
assert state.attributes.get(ATTR_STATE_CLASS) == SensorStateClass.MEASUREMENT
async def test_mean_1_digit_sensor(hass):
async def test_mean_1_digit_sensor(hass: HomeAssistant) -> None:
"""Test the mean with 1-digit precision sensor."""
config = {
"sensor": {
@ -171,7 +171,7 @@ async def test_mean_1_digit_sensor(hass):
assert str(float(MEAN_1_DIGIT)) == state.state
async def test_mean_4_digit_sensor(hass):
async def test_mean_4_digit_sensor(hass: HomeAssistant) -> None:
"""Test the mean with 4-digit precision sensor."""
config = {
"sensor": {
@ -197,7 +197,7 @@ async def test_mean_4_digit_sensor(hass):
assert str(float(MEAN_4_DIGITS)) == state.state
async def test_median_sensor(hass):
async def test_median_sensor(hass: HomeAssistant) -> None:
"""Test the median sensor."""
config = {
"sensor": {
@ -223,7 +223,7 @@ async def test_median_sensor(hass):
assert state.attributes.get(ATTR_STATE_CLASS) == SensorStateClass.MEASUREMENT
async def test_range_4_digit_sensor(hass):
async def test_range_4_digit_sensor(hass: HomeAssistant) -> None:
"""Test the range with 4-digit precision sensor."""
config = {
"sensor": {
@ -249,7 +249,7 @@ async def test_range_4_digit_sensor(hass):
assert str(float(RANGE_4_DIGITS)) == state.state
async def test_range_1_digit_sensor(hass):
async def test_range_1_digit_sensor(hass: HomeAssistant) -> None:
"""Test the range with 1-digit precision sensor."""
config = {
"sensor": {
@ -275,7 +275,7 @@ async def test_range_1_digit_sensor(hass):
assert str(float(RANGE_1_DIGIT)) == state.state
async def test_not_enough_sensor_value(hass):
async def test_not_enough_sensor_value(hass: HomeAssistant) -> None:
"""Test that there is nothing done if not enough values available."""
config = {
"sensor": {
@ -327,7 +327,7 @@ async def test_not_enough_sensor_value(hass):
assert state.attributes.get("max_value") is None
async def test_different_unit_of_measurement(hass):
async def test_different_unit_of_measurement(hass: HomeAssistant) -> None:
"""Test for different unit of measurement."""
config = {
"sensor": {
@ -374,7 +374,7 @@ async def test_different_unit_of_measurement(hass):
assert state.attributes.get("unit_of_measurement") == "ERR"
async def test_last_sensor(hass):
async def test_last_sensor(hass: HomeAssistant) -> None:
"""Test the last sensor."""
config = {
"sensor": {
@ -399,7 +399,7 @@ async def test_last_sensor(hass):
assert state.attributes.get(ATTR_STATE_CLASS) == SensorStateClass.MEASUREMENT
async def test_reload(hass):
async def test_reload(hass: HomeAssistant) -> None:
"""Verify we can reload filter sensors."""
hass.states.async_set("sensor.test_1", 12345)
hass.states.async_set("sensor.test_2", 45678)