Statistics component typing (#60997)
* Implement optional manually defined uniqueid * Fix test case via mocked environment * Add typing to statistics component * Fix minor inconsistency * Fix linter issues * Execute hassfest * Fix stricter mypy warnings * Fix maxsplit warning * Make binary value range explicit check * Add basic typing to statistics tests * Add empty config testcase * Minor improvements * Improve after comments * Remove unnecessary test case * Fix changed type * Remove dict.get default
This commit is contained in:
Thomas Dietrich
GitHub
parent
2f0b73c4ad
commit
28af0b4092
4 changed files with 214 additions and 169 deletions
|
|
@ -1,11 +1,17 @@
|
|||
"""Support for statistics for sensor values."""
|
||||
from __future__ import annotations
|
||||
|
||||
from collections import deque
|
||||
from collections.abc import Callable
|
||||
import contextlib
|
||||
from datetime import datetime, timedelta
|
||||
import logging
|
||||
import statistics
|
||||
from typing import Any, Literal, cast
|
||||
|
||||
import voluptuous as vol
|
||||
|
||||
from homeassistant.components.binary_sensor import DOMAIN as BINARY_SENSOR_DOMAIN
|
||||
from homeassistant.components.recorder.models import States
|
||||
from homeassistant.components.recorder.util import execute, session_scope
|
||||
from homeassistant.components.sensor import (
|
||||
|
|
@ -21,14 +27,23 @@ from homeassistant.const import (
|
|||
STATE_UNAVAILABLE,
|
||||
STATE_UNKNOWN,
|
||||
)
|
||||
from homeassistant.core import callback
|
||||
from homeassistant.core import (
|
||||
CALLBACK_TYPE,
|
||||
Event,
|
||||
HomeAssistant,
|
||||
State,
|
||||
callback,
|
||||
split_entity_id,
|
||||
)
|
||||
from homeassistant.helpers import config_validation as cv
|
||||
from homeassistant.helpers.entity_platform import AddEntitiesCallback
|
||||
from homeassistant.helpers.event import (
|
||||
async_track_point_in_utc_time,
|
||||
async_track_state_change_event,
|
||||
)
|
||||
from homeassistant.helpers.reload import async_setup_reload_service
|
||||
from homeassistant.helpers.start import async_at_start
|
||||
from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType, StateType
|
||||
from homeassistant.util import dt as dt_util
|
||||
|
||||
from . import DOMAIN, PLATFORMS
|
||||
|
|
@ -100,13 +115,13 @@ DEFAULT_QUANTILE_METHOD = "exclusive"
|
|||
ICON = "mdi:calculator"
|
||||
|
||||
|
||||
def valid_binary_characteristic_configuration(config):
|
||||
def valid_binary_characteristic_configuration(config: dict[str, Any]) -> dict[str, Any]:
|
||||
"""Validate that the characteristic selected is valid for the source sensor type, throw if it isn't."""
|
||||
if config.get(CONF_ENTITY_ID).split(".")[0] == "binary_sensor":
|
||||
if split_entity_id(str(config.get(CONF_ENTITY_ID)))[0] == BINARY_SENSOR_DOMAIN:
|
||||
if config.get(CONF_STATE_CHARACTERISTIC) not in STATS_BINARY_SUPPORT:
|
||||
raise ValueError(
|
||||
"The configured characteristic '"
|
||||
+ config.get(CONF_STATE_CHARACTERISTIC)
|
||||
+ str(config.get(CONF_STATE_CHARACTERISTIC))
|
||||
+ "' is not supported for a binary source sensor."
|
||||
)
|
||||
return config
|
||||
|
|
@ -162,28 +177,32 @@ PLATFORM_SCHEMA = vol.All(
|
|||
)
|
||||
|
||||
|
||||
async def async_setup_platform(hass, config, async_add_entities, discovery_info=None):
|
||||
async def async_setup_platform(
|
||||
hass: HomeAssistant,
|
||||
config: ConfigType,
|
||||
async_add_entities: AddEntitiesCallback,
|
||||
discovery_info: DiscoveryInfoType | None = None,
|
||||
) -> None:
|
||||
"""Set up the Statistics sensor."""
|
||||
|
||||
await async_setup_reload_service(hass, DOMAIN, PLATFORMS)
|
||||
|
||||
async_add_entities(
|
||||
[
|
||||
new_entities=[
|
||||
StatisticsSensor(
|
||||
source_entity_id=config.get(CONF_ENTITY_ID),
|
||||
name=config.get(CONF_NAME),
|
||||
source_entity_id=config[CONF_ENTITY_ID],
|
||||
name=config[CONF_NAME],
|
||||
unique_id=config.get(CONF_UNIQUE_ID),
|
||||
state_characteristic=config.get(CONF_STATE_CHARACTERISTIC),
|
||||
samples_max_buffer_size=config.get(CONF_SAMPLES_MAX_BUFFER_SIZE),
|
||||
state_characteristic=config[CONF_STATE_CHARACTERISTIC],
|
||||
samples_max_buffer_size=config[CONF_SAMPLES_MAX_BUFFER_SIZE],
|
||||
samples_max_age=config.get(CONF_MAX_AGE),
|
||||
precision=config.get(CONF_PRECISION),
|
||||
quantile_intervals=config.get(CONF_QUANTILE_INTERVALS),
|
||||
quantile_method=config.get(CONF_QUANTILE_METHOD),
|
||||
precision=config[CONF_PRECISION],
|
||||
quantile_intervals=config[CONF_QUANTILE_INTERVALS],
|
||||
quantile_method=config[CONF_QUANTILE_METHOD],
|
||||
)
|
||||
],
|
||||
True,
|
||||
update_before_add=True,
|
||||
)
|
||||
return True
|
||||
|
||||
|
||||
class StatisticsSensor(SensorEntity):
|
||||
|
|
@ -191,41 +210,46 @@ class StatisticsSensor(SensorEntity):
|
|||
|
||||
def __init__(
|
||||
self,
|
||||
source_entity_id,
|
||||
name,
|
||||
unique_id,
|
||||
state_characteristic,
|
||||
samples_max_buffer_size,
|
||||
samples_max_age,
|
||||
precision,
|
||||
quantile_intervals,
|
||||
quantile_method,
|
||||
):
|
||||
source_entity_id: str,
|
||||
name: str,
|
||||
unique_id: str | None,
|
||||
state_characteristic: str,
|
||||
samples_max_buffer_size: int,
|
||||
samples_max_age: timedelta | None,
|
||||
precision: int,
|
||||
quantile_intervals: int,
|
||||
quantile_method: str,
|
||||
) -> None:
|
||||
"""Initialize the Statistics sensor."""
|
||||
self._source_entity_id = source_entity_id
|
||||
self.is_binary = self._source_entity_id.split(".")[0] == "binary_sensor"
|
||||
self._name = name
|
||||
self._unique_id = unique_id
|
||||
self._state_characteristic = state_characteristic
|
||||
self._attr_icon: str = ICON
|
||||
self._attr_name: str = name
|
||||
self._attr_should_poll: bool = False
|
||||
self._attr_unique_id: str | None = unique_id
|
||||
self._source_entity_id: str = source_entity_id
|
||||
self.is_binary: bool = (
|
||||
split_entity_id(self._source_entity_id)[0] == BINARY_SENSOR_DOMAIN
|
||||
)
|
||||
self._state_characteristic: str = state_characteristic
|
||||
if self._state_characteristic == STAT_DEFAULT:
|
||||
self._state_characteristic = STAT_COUNT if self.is_binary else STAT_MEAN
|
||||
_LOGGER.warning(DEPRECATION_WARNING, self._state_characteristic, name)
|
||||
self._samples_max_buffer_size = samples_max_buffer_size
|
||||
self._samples_max_age = samples_max_age
|
||||
self._precision = precision
|
||||
self._quantile_intervals = quantile_intervals
|
||||
self._quantile_method = quantile_method
|
||||
self._value = None
|
||||
self._unit_of_measurement = None
|
||||
self._available = False
|
||||
self.states = deque(maxlen=self._samples_max_buffer_size)
|
||||
self.ages = deque(maxlen=self._samples_max_buffer_size)
|
||||
self.attributes = {
|
||||
self._samples_max_buffer_size: int = samples_max_buffer_size
|
||||
self._samples_max_age: timedelta | None = samples_max_age
|
||||
self._precision: int = precision
|
||||
self._quantile_intervals: int = quantile_intervals
|
||||
self._quantile_method: str = quantile_method
|
||||
self._value: StateType | datetime = None
|
||||
self._unit_of_measurement: str | None = None
|
||||
self._available: bool = False
|
||||
self.states: deque[float | bool] = deque(maxlen=self._samples_max_buffer_size)
|
||||
self.ages: deque[datetime] = deque(maxlen=self._samples_max_buffer_size)
|
||||
self.attributes: dict[str, StateType] = {
|
||||
STAT_AGE_COVERAGE_RATIO: None,
|
||||
STAT_BUFFER_USAGE_RATIO: None,
|
||||
STAT_SOURCE_VALUE_VALID: None,
|
||||
}
|
||||
|
||||
self._state_characteristic_fn: Callable[[], StateType | datetime]
|
||||
if self.is_binary:
|
||||
self._state_characteristic_fn = getattr(
|
||||
self, f"_stat_binary_{self._state_characteristic}"
|
||||
|
|
@ -235,20 +259,20 @@ class StatisticsSensor(SensorEntity):
|
|||
self, f"_stat_{self._state_characteristic}"
|
||||
)
|
||||
|
||||
self._update_listener = None
|
||||
self._update_listener: CALLBACK_TYPE | None = None
|
||||
|
||||
async def async_added_to_hass(self):
|
||||
async def async_added_to_hass(self) -> None:
|
||||
"""Register callbacks."""
|
||||
|
||||
@callback
|
||||
def async_stats_sensor_state_listener(event):
|
||||
def async_stats_sensor_state_listener(event: Event) -> None:
|
||||
"""Handle the sensor state changes."""
|
||||
if (new_state := event.data.get("new_state")) is None:
|
||||
return
|
||||
self._add_state_to_queue(new_state)
|
||||
self.async_schedule_update_ha_state(True)
|
||||
|
||||
async def async_stats_sensor_startup(_):
|
||||
async def async_stats_sensor_startup(_: HomeAssistant) -> None:
|
||||
"""Add listener and get recorded state."""
|
||||
_LOGGER.debug("Startup for %s", self.entity_id)
|
||||
|
||||
|
|
@ -265,7 +289,7 @@ class StatisticsSensor(SensorEntity):
|
|||
|
||||
async_at_start(self.hass, async_stats_sensor_startup)
|
||||
|
||||
def _add_state_to_queue(self, new_state):
|
||||
def _add_state_to_queue(self, new_state: State) -> None:
|
||||
"""Add the state to the queue."""
|
||||
self._available = new_state.state != STATE_UNAVAILABLE
|
||||
if new_state.state == STATE_UNAVAILABLE:
|
||||
|
|
@ -277,7 +301,8 @@ class StatisticsSensor(SensorEntity):
|
|||
|
||||
try:
|
||||
if self.is_binary:
|
||||
self.states.append(new_state.state)
|
||||
assert new_state.state in ("on", "off")
|
||||
self.states.append(new_state.state == "on")
|
||||
else:
|
||||
self.states.append(float(new_state.state))
|
||||
self.ages.append(new_state.last_updated)
|
||||
|
|
@ -293,8 +318,9 @@ class StatisticsSensor(SensorEntity):
|
|||
|
||||
self._unit_of_measurement = self._derive_unit_of_measurement(new_state)
|
||||
|
||||
def _derive_unit_of_measurement(self, new_state):
|
||||
base_unit = new_state.attributes.get(ATTR_UNIT_OF_MEASUREMENT)
|
||||
def _derive_unit_of_measurement(self, new_state: State) -> str | None:
|
||||
base_unit: str | None = new_state.attributes.get(ATTR_UNIT_OF_MEASUREMENT)
|
||||
unit: str | None
|
||||
if self.is_binary and self._state_characteristic in (
|
||||
STAT_AVERAGE_STEP,
|
||||
STAT_AVERAGE_TIMELESS,
|
||||
|
|
@ -336,66 +362,46 @@ class StatisticsSensor(SensorEntity):
|
|||
return unit
|
||||
|
||||
@property
|
||||
def name(self):
|
||||
"""Return the name of the sensor."""
|
||||
return self._name
|
||||
|
||||
@property
|
||||
def unique_id(self):
|
||||
"""Return the unique id of the sensor."""
|
||||
return self._unique_id
|
||||
|
||||
@property
|
||||
def state_class(self):
|
||||
def state_class(self) -> Literal[SensorStateClass.MEASUREMENT] | None:
|
||||
"""Return the state class of this entity."""
|
||||
if self._state_characteristic in STATS_NOT_A_NUMBER:
|
||||
return None
|
||||
return SensorStateClass.MEASUREMENT
|
||||
|
||||
@property
|
||||
def native_value(self):
|
||||
def native_value(self) -> StateType | datetime:
|
||||
"""Return the state of the sensor."""
|
||||
return self._value
|
||||
|
||||
@property
|
||||
def native_unit_of_measurement(self):
|
||||
def native_unit_of_measurement(self) -> str | None:
|
||||
"""Return the unit the value is expressed in."""
|
||||
return self._unit_of_measurement
|
||||
|
||||
@property
|
||||
def available(self):
|
||||
def available(self) -> bool:
|
||||
"""Return the availability of the sensor linked to the source sensor."""
|
||||
return self._available
|
||||
|
||||
@property
|
||||
def should_poll(self):
|
||||
"""No polling needed."""
|
||||
return False
|
||||
|
||||
@property
|
||||
def extra_state_attributes(self):
|
||||
def extra_state_attributes(self) -> dict[str, StateType] | None:
|
||||
"""Return the state attributes of the sensor."""
|
||||
return {
|
||||
key: value for key, value in self.attributes.items() if value is not None
|
||||
}
|
||||
|
||||
@property
|
||||
def icon(self):
|
||||
"""Return the icon to use in the frontend, if any."""
|
||||
return ICON
|
||||
|
||||
def _purge_old(self):
|
||||
"""Remove states which are older than self._samples_max_age."""
|
||||
def _purge_old_states(self, max_age: timedelta) -> None:
|
||||
"""Remove states which are older than a given age."""
|
||||
now = dt_util.utcnow()
|
||||
|
||||
_LOGGER.debug(
|
||||
"%s: purging records older then %s(%s)",
|
||||
self.entity_id,
|
||||
dt_util.as_local(now - self._samples_max_age),
|
||||
dt_util.as_local(now - max_age),
|
||||
self._samples_max_age,
|
||||
)
|
||||
|
||||
while self.ages and (now - self.ages[0]) > self._samples_max_age:
|
||||
while self.ages and (now - self.ages[0]) > max_age:
|
||||
_LOGGER.debug(
|
||||
"%s: purging record with datetime %s(%s)",
|
||||
self.entity_id,
|
||||
|
|
@ -405,7 +411,7 @@ class StatisticsSensor(SensorEntity):
|
|||
self.ages.popleft()
|
||||
self.states.popleft()
|
||||
|
||||
def _next_to_purge_timestamp(self):
|
||||
def _next_to_purge_timestamp(self) -> datetime | None:
|
||||
"""Find the timestamp when the next purge would occur."""
|
||||
if self.ages and self._samples_max_age:
|
||||
# Take the oldest entry from the ages list and add the configured max_age.
|
||||
|
|
@ -414,11 +420,11 @@ class StatisticsSensor(SensorEntity):
|
|||
return self.ages[0] + self._samples_max_age
|
||||
return None
|
||||
|
||||
async def async_update(self):
|
||||
async def async_update(self) -> None:
|
||||
"""Get the latest data and updates the states."""
|
||||
_LOGGER.debug("%s: updating statistics", self.entity_id)
|
||||
if self._samples_max_age is not None:
|
||||
self._purge_old()
|
||||
self._purge_old_states(self._samples_max_age)
|
||||
|
||||
self._update_attributes()
|
||||
self._update_value()
|
||||
|
|
@ -434,7 +440,7 @@ class StatisticsSensor(SensorEntity):
|
|||
self._update_listener = None
|
||||
|
||||
@callback
|
||||
def _scheduled_update(now):
|
||||
def _scheduled_update(now: datetime) -> None:
|
||||
"""Timer callback for sensor update."""
|
||||
_LOGGER.debug("%s: executing scheduled update", self.entity_id)
|
||||
self.async_schedule_update_ha_state(True)
|
||||
|
|
@ -444,7 +450,7 @@ class StatisticsSensor(SensorEntity):
|
|||
self.hass, _scheduled_update, next_to_purge_timestamp
|
||||
)
|
||||
|
||||
async def _initialize_from_database(self):
|
||||
async def _initialize_from_database(self) -> None:
|
||||
"""Initialize the list of states from the database.
|
||||
|
||||
The query will get the list of states in DESCENDING order so that we
|
||||
|
|
@ -478,14 +484,15 @@ class StatisticsSensor(SensorEntity):
|
|||
)
|
||||
states = execute(query, to_native=True, validate_entity_ids=False)
|
||||
|
||||
for state in reversed(states):
|
||||
self._add_state_to_queue(state)
|
||||
if states:
|
||||
for state in reversed(states):
|
||||
self._add_state_to_queue(state)
|
||||
|
||||
self.async_schedule_update_ha_state(True)
|
||||
|
||||
_LOGGER.debug("%s: initializing from database completed", self.entity_id)
|
||||
|
||||
def _update_attributes(self):
|
||||
def _update_attributes(self) -> None:
|
||||
"""Calculate and update the various attributes."""
|
||||
self.attributes[STAT_BUFFER_USAGE_RATIO] = round(
|
||||
len(self.states) / self._samples_max_buffer_size, 2
|
||||
|
|
@ -500,7 +507,7 @@ class StatisticsSensor(SensorEntity):
|
|||
else:
|
||||
self.attributes[STAT_AGE_COVERAGE_RATIO] = None
|
||||
|
||||
def _update_value(self):
|
||||
def _update_value(self) -> None:
|
||||
"""Front to call the right statistical characteristics functions.
|
||||
|
||||
One of the _stat_*() functions is represented by self._state_characteristic_fn().
|
||||
|
|
@ -510,16 +517,16 @@ class StatisticsSensor(SensorEntity):
|
|||
|
||||
if self._state_characteristic not in STATS_NOT_A_NUMBER:
|
||||
with contextlib.suppress(TypeError):
|
||||
value = round(value, self._precision)
|
||||
value = round(cast(float, value), self._precision)
|
||||
if self._precision == 0:
|
||||
value = int(value)
|
||||
self._value = value
|
||||
|
||||
# Statistics for numeric sensor
|
||||
|
||||
def _stat_average_linear(self):
|
||||
def _stat_average_linear(self) -> StateType:
|
||||
if len(self.states) >= 2:
|
||||
area = 0
|
||||
area: float = 0
|
||||
for i in range(1, len(self.states)):
|
||||
area += (
|
||||
0.5
|
||||
|
|
@ -530,9 +537,9 @@ class StatisticsSensor(SensorEntity):
|
|||
return area / age_range_seconds
|
||||
return None
|
||||
|
||||
def _stat_average_step(self):
|
||||
def _stat_average_step(self) -> StateType:
|
||||
if len(self.states) >= 2:
|
||||
area = 0
|
||||
area: float = 0
|
||||
for i in range(1, len(self.states)):
|
||||
area += (
|
||||
self.states[i - 1]
|
||||
|
|
@ -542,65 +549,65 @@ class StatisticsSensor(SensorEntity):
|
|||
return area / age_range_seconds
|
||||
return None
|
||||
|
||||
def _stat_average_timeless(self):
|
||||
def _stat_average_timeless(self) -> StateType:
|
||||
return self._stat_mean()
|
||||
|
||||
def _stat_change(self):
|
||||
def _stat_change(self) -> StateType:
|
||||
if len(self.states) > 0:
|
||||
return self.states[-1] - self.states[0]
|
||||
return None
|
||||
|
||||
def _stat_change_sample(self):
|
||||
def _stat_change_sample(self) -> StateType:
|
||||
if len(self.states) > 1:
|
||||
return (self.states[-1] - self.states[0]) / (len(self.states) - 1)
|
||||
return None
|
||||
|
||||
def _stat_change_second(self):
|
||||
def _stat_change_second(self) -> StateType:
|
||||
if len(self.states) > 1:
|
||||
age_range_seconds = (self.ages[-1] - self.ages[0]).total_seconds()
|
||||
if age_range_seconds > 0:
|
||||
return (self.states[-1] - self.states[0]) / age_range_seconds
|
||||
return None
|
||||
|
||||
def _stat_count(self):
|
||||
def _stat_count(self) -> StateType:
|
||||
return len(self.states)
|
||||
|
||||
def _stat_datetime_newest(self):
|
||||
def _stat_datetime_newest(self) -> datetime | None:
|
||||
if len(self.states) > 0:
|
||||
return self.ages[-1]
|
||||
return None
|
||||
|
||||
def _stat_datetime_oldest(self):
|
||||
def _stat_datetime_oldest(self) -> datetime | None:
|
||||
if len(self.states) > 0:
|
||||
return self.ages[0]
|
||||
return None
|
||||
|
||||
def _stat_distance_95_percent_of_values(self):
|
||||
def _stat_distance_95_percent_of_values(self) -> StateType:
|
||||
if len(self.states) >= 2:
|
||||
return 2 * 1.96 * self._stat_standard_deviation()
|
||||
return 2 * 1.96 * cast(float, self._stat_standard_deviation())
|
||||
return None
|
||||
|
||||
def _stat_distance_99_percent_of_values(self):
|
||||
def _stat_distance_99_percent_of_values(self) -> StateType:
|
||||
if len(self.states) >= 2:
|
||||
return 2 * 2.58 * self._stat_standard_deviation()
|
||||
return 2 * 2.58 * cast(float, self._stat_standard_deviation())
|
||||
return None
|
||||
|
||||
def _stat_distance_absolute(self):
|
||||
def _stat_distance_absolute(self) -> StateType:
|
||||
if len(self.states) > 0:
|
||||
return max(self.states) - min(self.states)
|
||||
return None
|
||||
|
||||
def _stat_mean(self):
|
||||
def _stat_mean(self) -> StateType:
|
||||
if len(self.states) > 0:
|
||||
return statistics.mean(self.states)
|
||||
return None
|
||||
|
||||
def _stat_median(self):
|
||||
def _stat_median(self) -> StateType:
|
||||
if len(self.states) > 0:
|
||||
return statistics.median(self.states)
|
||||
return None
|
||||
|
||||
def _stat_noisiness(self):
|
||||
def _stat_noisiness(self) -> StateType:
|
||||
if len(self.states) >= 2:
|
||||
diff_sum = sum(
|
||||
abs(j - i) for i, j in zip(list(self.states), list(self.states)[1:])
|
||||
|
|
@ -608,62 +615,64 @@ class StatisticsSensor(SensorEntity):
|
|||
return diff_sum / (len(self.states) - 1)
|
||||
return None
|
||||
|
||||
def _stat_quantiles(self):
|
||||
def _stat_quantiles(self) -> StateType:
|
||||
if len(self.states) > self._quantile_intervals:
|
||||
return [
|
||||
round(quantile, self._precision)
|
||||
for quantile in statistics.quantiles(
|
||||
self.states,
|
||||
n=self._quantile_intervals,
|
||||
method=self._quantile_method,
|
||||
)
|
||||
]
|
||||
return str(
|
||||
[
|
||||
round(quantile, self._precision)
|
||||
for quantile in statistics.quantiles(
|
||||
self.states,
|
||||
n=self._quantile_intervals,
|
||||
method=self._quantile_method,
|
||||
)
|
||||
]
|
||||
)
|
||||
return None
|
||||
|
||||
def _stat_standard_deviation(self):
|
||||
def _stat_standard_deviation(self) -> StateType:
|
||||
if len(self.states) >= 2:
|
||||
return statistics.stdev(self.states)
|
||||
return None
|
||||
|
||||
def _stat_total(self):
|
||||
def _stat_total(self) -> StateType:
|
||||
if len(self.states) > 0:
|
||||
return sum(self.states)
|
||||
return None
|
||||
|
||||
def _stat_value_max(self):
|
||||
def _stat_value_max(self) -> StateType:
|
||||
if len(self.states) > 0:
|
||||
return max(self.states)
|
||||
return None
|
||||
|
||||
def _stat_value_min(self):
|
||||
def _stat_value_min(self) -> StateType:
|
||||
if len(self.states) > 0:
|
||||
return min(self.states)
|
||||
return None
|
||||
|
||||
def _stat_variance(self):
|
||||
def _stat_variance(self) -> StateType:
|
||||
if len(self.states) >= 2:
|
||||
return statistics.variance(self.states)
|
||||
return None
|
||||
|
||||
# Statistics for binary sensor
|
||||
|
||||
def _stat_binary_average_step(self):
|
||||
def _stat_binary_average_step(self) -> StateType:
|
||||
if len(self.states) >= 2:
|
||||
on_seconds = 0
|
||||
on_seconds: float = 0
|
||||
for i in range(1, len(self.states)):
|
||||
if self.states[i - 1] == "on":
|
||||
if self.states[i - 1] is True:
|
||||
on_seconds += (self.ages[i] - self.ages[i - 1]).total_seconds()
|
||||
age_range_seconds = (self.ages[-1] - self.ages[0]).total_seconds()
|
||||
return 100 / age_range_seconds * on_seconds
|
||||
return None
|
||||
|
||||
def _stat_binary_average_timeless(self):
|
||||
def _stat_binary_average_timeless(self) -> StateType:
|
||||
return self._stat_binary_mean()
|
||||
|
||||
def _stat_binary_count(self):
|
||||
def _stat_binary_count(self) -> StateType:
|
||||
return len(self.states)
|
||||
|
||||
def _stat_binary_mean(self):
|
||||
def _stat_binary_mean(self) -> StateType:
|
||||
if len(self.states) > 0:
|
||||
return 100.0 / len(self.states) * self.states.count("on")
|
||||
return 100.0 / len(self.states) * self.states.count(True)
|
||||
return None
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue