Electric-Special/ha-core
1
0
Fork 0
mirror of https://github.com/Electric-Special/ha-core.git synced 2026-03-21 03:03:17 +01:00
Code Issues Packages Projects Releases Wiki Activity

Improve derivative units and auto-device_class (#157369)

Browse Source
This commit is contained in:
karwosts
2026-02-17 08:08:59 -08:00
committed by GitHub
parent 0b8312d942
commit 9f551f3d5b
2 changed files with 223 additions and 19 deletions
Download Patch File Download Diff File Expand all files Collapse all files

171
tests/components/derivative/test_sensor.py
View File

@@ -11,13 +11,22 @@ import pytest
from homeassistant import config as hass_config, core as ha
from homeassistant.components.derivative.const import DOMAIN
from homeassistant.components.sensor import ATTR_STATE_CLASS, SensorStateClass
from homeassistant.components.sensor import (
ATTR_STATE_CLASS,
SensorDeviceClass,
SensorStateClass,
)
from homeassistant.const import (
ATTR_DEVICE_CLASS,
SERVICE_RELOAD,
STATE_UNAVAILABLE,
STATE_UNKNOWN,
UnitOfDataRate,
UnitOfEnergy,
UnitOfPower,
UnitOfSpeed,
UnitOfTime,
UnitOfVolumeFlowRate,
)
from homeassistant.core import (
Event,
@@ -642,6 +651,137 @@ async def test_sub_intervals_with_time_window(hass: HomeAssistant) -> None:
assert expect_min <= derivative <= expect_max, f"Failed at time {time}"
@pytest.mark.parametrize(
("extra_config", "source_unit", "source_class", "derived_unit", "derived_class"),
[
(
{},
UnitOfEnergy.KILO_WATT_HOUR,
SensorDeviceClass.ENERGY,
UnitOfPower.KILO_WATT,
SensorDeviceClass.POWER,
),
(
{},
UnitOfEnergy.TERA_WATT_HOUR,
SensorDeviceClass.ENERGY,
UnitOfPower.TERA_WATT,
SensorDeviceClass.POWER,
),
(
{"unit_prefix": "m"},
UnitOfEnergy.WATT_HOUR,
SensorDeviceClass.ENERGY_STORAGE,
UnitOfPower.MILLIWATT,
SensorDeviceClass.POWER,
),
(
{"unit_prefix": "k"},
UnitOfEnergy.WATT_HOUR,
SensorDeviceClass.ENERGY,
UnitOfPower.KILO_WATT,
SensorDeviceClass.POWER,
),
(
{"unit_prefix": "n"},
UnitOfEnergy.WATT_HOUR,
SensorDeviceClass.ENERGY,
"nW",
None,
),
(
{},
"GB",
SensorDeviceClass.DATA_SIZE,
"GB/h",
None,
),
(
{"unit_time": "s"},
"GB",
SensorDeviceClass.DATA_SIZE,
UnitOfDataRate.GIGABYTES_PER_SECOND,
SensorDeviceClass.DATA_RATE,
),
(
{},
"km",
SensorDeviceClass.DISTANCE,
UnitOfSpeed.KILOMETERS_PER_HOUR,
SensorDeviceClass.SPEED,
),
(
{},
"m³",
SensorDeviceClass.GAS,
UnitOfVolumeFlowRate.CUBIC_METERS_PER_HOUR,
SensorDeviceClass.VOLUME_FLOW_RATE,
),
(
{"unit_time": "min"},
"gal",
SensorDeviceClass.WATER,
UnitOfVolumeFlowRate.GALLONS_PER_MINUTE,
SensorDeviceClass.VOLUME_FLOW_RATE,
),
(
{},
UnitOfEnergy.KILO_WATT_HOUR,
"not_a_real_device_class",
"kWh/h",
None,
),
],
)
async def test_device_classes(
extra_config: dict[str, Any],
source_unit: str,
source_class: str,
derived_unit: str,
derived_class: str,
hass: HomeAssistant,
) -> None:
"""Test derivative sensor handles unit conversions and device classes."""
config = {
"sensor": {
"platform": "derivative",
"name": "derivative",
"source": "sensor.source",
"round": 2,
"unit_time": "h",
**extra_config,
}
}
assert await async_setup_component(hass, "sensor", config)
entity_id = config["sensor"]["source"]
base = dt_util.utcnow()
with freeze_time(base) as freezer:
hass.states.async_set(
entity_id,
1000,
{
"unit_of_measurement": source_unit,
"device_class": source_class,
},
)
await hass.async_block_till_done()
freezer.move_to(dt_util.utcnow() + timedelta(seconds=3600))
hass.states.async_set(
entity_id,
2000,
{
"unit_of_measurement": source_unit,
"device_class": source_class,
},
)
await hass.async_block_till_done()
state = hass.states.get("sensor.derivative")
assert state is not None
assert state.attributes.get("unit_of_measurement") == derived_unit
assert state.attributes.get("device_class") == derived_class
async def test_prefix(hass: HomeAssistant) -> None:
"""Test derivative sensor state using a power source."""
config = {
@@ -885,13 +1025,11 @@ async def test_unavailable_boot(
State(
"sensor.power",
restore_state,
{
"unit_of_measurement": "kWh/s",
},
{"unit_of_measurement": "kW", "device_class": "power"},
),
{
"native_value": restore_state,
"native_unit_of_measurement": "kWh/s",
"native_unit_of_measurement": "kW",
},
),
],
@@ -902,12 +1040,16 @@ async def test_unavailable_boot(
"name": "power",
"source": "sensor.energy",
"round": 2,
"unit_time": "s",
"unit_time": "h",
}
config = {"sensor": config}
entity_id = config["sensor"]["source"]
hass.states.async_set(entity_id, STATE_UNAVAILABLE, {"unit_of_measurement": "kWh"})
hass.states.async_set(
entity_id,
STATE_UNAVAILABLE,
{"unit_of_measurement": "kWh", "device_class": "energy"},
)
await hass.async_block_till_done()
assert await async_setup_component(hass, "sensor", config)
@@ -917,11 +1059,14 @@ async def test_unavailable_boot(
assert state is not None
# Sensor is unavailable as source is unavailable
assert state.state == STATE_UNAVAILABLE
assert state.attributes.get(ATTR_DEVICE_CLASS) == "power"
base = dt_util.utcnow()
with freeze_time(base) as freezer:
freezer.move_to(base + timedelta(seconds=1))
hass.states.async_set(entity_id, 10, {"unit_of_measurement": "kWh"})
freezer.move_to(base + timedelta(hours=1))
hass.states.async_set(
entity_id, 10, {"unit_of_measurement": "kWh", "device_class": "energy"}
)
await hass.async_block_till_done()
state = hass.states.get("sensor.power")
@@ -930,15 +1075,17 @@ async def test_unavailable_boot(
# so just hold until the next tick
assert state.state == restore_state
freezer.move_to(base + timedelta(seconds=2))
hass.states.async_set(entity_id, 15, {"unit_of_measurement": "kWh"})
freezer.move_to(base + timedelta(hours=2))
hass.states.async_set(
entity_id, 15, {"unit_of_measurement": "kWh", "device_class": "energy"}
)
await hass.async_block_till_done()
state = hass.states.get("sensor.power")
assert state is not None
# Now that the source sensor has two valid datapoints, we can calculate derivative
assert state.state == "5.00"
assert state.attributes.get("unit_of_measurement") == "kWh/s"
assert state.attributes.get("unit_of_measurement") == "kW"
async def test_source_unit_change(