parent
b7b82b1e3f
commit
7aa79415c1
|
@ -137,7 +137,7 @@ async def async_setup_entry(
|
|||
|
||||
def calc_min(
|
||||
sensor_values: list[tuple[str, float, State]]
|
||||
) -> tuple[dict[str, str | None], float]:
|
||||
) -> tuple[dict[str, str | None], float | None]:
|
||||
"""Calculate min value."""
|
||||
val: float | None = None
|
||||
entity_id: str | None = None
|
||||
|
@ -153,7 +153,7 @@ def calc_min(
|
|||
|
||||
def calc_max(
|
||||
sensor_values: list[tuple[str, float, State]]
|
||||
) -> tuple[dict[str, str | None], float]:
|
||||
) -> tuple[dict[str, str | None], float | None]:
|
||||
"""Calculate max value."""
|
||||
val: float | None = None
|
||||
entity_id: str | None = None
|
||||
|
@ -169,7 +169,7 @@ def calc_max(
|
|||
|
||||
def calc_mean(
|
||||
sensor_values: list[tuple[str, float, State]]
|
||||
) -> tuple[dict[str, str | None], float]:
|
||||
) -> tuple[dict[str, str | None], float | None]:
|
||||
"""Calculate mean value."""
|
||||
result = (sensor_value for _, sensor_value, _ in sensor_values)
|
||||
|
||||
|
@ -179,7 +179,7 @@ def calc_mean(
|
|||
|
||||
def calc_median(
|
||||
sensor_values: list[tuple[str, float, State]]
|
||||
) -> tuple[dict[str, str | None], float]:
|
||||
) -> tuple[dict[str, str | None], float | None]:
|
||||
"""Calculate median value."""
|
||||
result = (sensor_value for _, sensor_value, _ in sensor_values)
|
||||
|
||||
|
@ -189,10 +189,11 @@ def calc_median(
|
|||
|
||||
def calc_last(
|
||||
sensor_values: list[tuple[str, float, State]]
|
||||
) -> tuple[dict[str, str | None], float]:
|
||||
) -> tuple[dict[str, str | None], float | None]:
|
||||
"""Calculate last value."""
|
||||
last_updated: datetime | None = None
|
||||
last_entity_id: str | None = None
|
||||
last: float | None = None
|
||||
for entity_id, state_f, state in sensor_values:
|
||||
if last_updated is None or state.last_updated > last_updated:
|
||||
last_updated = state.last_updated
|
||||
|
@ -227,7 +228,9 @@ def calc_sum(
|
|||
|
||||
CALC_TYPES: dict[
|
||||
str,
|
||||
Callable[[list[tuple[str, float, State]]], tuple[dict[str, str | None], float]],
|
||||
Callable[
|
||||
[list[tuple[str, float, State]]], tuple[dict[str, str | None], float | None]
|
||||
],
|
||||
] = {
|
||||
"min": calc_min,
|
||||
"max": calc_max,
|
||||
|
|
|
@ -368,3 +368,28 @@ async def test_sensor_calculated_properties(hass: HomeAssistant) -> None:
|
|||
assert state.attributes.get("device_class") is None
|
||||
assert state.attributes.get("state_class") is None
|
||||
assert state.attributes.get("unit_of_measurement") is None
|
||||
|
||||
|
||||
async def test_last_sensor(hass: HomeAssistant) -> None:
|
||||
"""Test the last sensor."""
|
||||
config = {
|
||||
SENSOR_DOMAIN: {
|
||||
"platform": GROUP_DOMAIN,
|
||||
"name": "test_last",
|
||||
"type": "last",
|
||||
"entities": ["sensor.test_1", "sensor.test_2", "sensor.test_3"],
|
||||
"unique_id": "very_unique_id_last_sensor",
|
||||
}
|
||||
}
|
||||
|
||||
assert await async_setup_component(hass, "sensor", config)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
entity_ids = config["sensor"]["entities"]
|
||||
|
||||
for entity_id, value in dict(zip(entity_ids, VALUES)).items():
|
||||
hass.states.async_set(entity_id, value)
|
||||
await hass.async_block_till_done()
|
||||
state = hass.states.get("sensor.test_last")
|
||||
assert str(float(value)) == state.state
|
||||
assert entity_id == state.attributes.get("last_entity_id")
|
||||
|
|
Loading…
Reference in New Issue