jeans/core
jeans/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2025-07-28 07:37:34 +00:00
Code Issues Packages Projects Releases Wiki Activity

Fix missing device_class and state_class on compensation entities (#146115)

Browse Source 
Co-authored-by: Robert Resch <robert@resch.dev>
This commit is contained in:
Petro31 2025-07-02 07:39:19 -04:00 committed by GitHub
parent 73251fbb1c
commit cb8e076703
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
3 changed files with 361 additions and 198 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

24
homeassistant/components/compensation/__init__.py
View File

@ -6,11 +6,18 @@ from operator import itemgetter
import numpy as np
import voluptuous as vol
from homeassistant.components.sensor import DOMAIN as SENSOR_DOMAIN
from homeassistant.components.sensor import (
CONF_STATE_CLASS,
DEVICE_CLASSES_SCHEMA as SENSOR_DEVICE_CLASSES_SCHEMA,
DOMAIN as SENSOR_DOMAIN,
STATE_CLASSES_SCHEMA as SENSOR_STATE_CLASSES_SCHEMA,
)
from homeassistant.const import (
CONF_ATTRIBUTE,
CONF_DEVICE_CLASS,
CONF_MAXIMUM,
CONF_MINIMUM,
CONF_NAME,
CONF_SOURCE,
CONF_UNIQUE_ID,
CONF_UNIT_OF_MEASUREMENT,
@ -50,20 +57,23 @@ def datapoints_greater_than_degree(value: dict) -> dict:
COMPENSATION_SCHEMA = vol.Schema(
{
vol.Required(CONF_SOURCE): cv.entity_id,
vol.Optional(CONF_ATTRIBUTE): cv.string,
vol.Required(CONF_DATAPOINTS): [
vol.ExactSequence([vol.Coerce(float), vol.Coerce(float)])
],
vol.Optional(CONF_UNIQUE_ID): cv.string,
vol.Optional(CONF_ATTRIBUTE): cv.string,
vol.Optional(CONF_UPPER_LIMIT, default=False): cv.boolean,
vol.Optional(CONF_LOWER_LIMIT, default=False): cv.boolean,
vol.Optional(CONF_PRECISION, default=DEFAULT_PRECISION): cv.positive_int,
vol.Optional(CONF_DEGREE, default=DEFAULT_DEGREE): vol.All(
vol.Coerce(int),
vol.Range(min=1, max=7),
),
vol.Optional(CONF_DEVICE_CLASS): SENSOR_DEVICE_CLASSES_SCHEMA,
vol.Optional(CONF_LOWER_LIMIT, default=False): cv.boolean,
vol.Optional(CONF_NAME): cv.string,
vol.Optional(CONF_PRECISION, default=DEFAULT_PRECISION): cv.positive_int,
vol.Required(CONF_SOURCE): cv.entity_id,
vol.Optional(CONF_STATE_CLASS): SENSOR_STATE_CLASSES_SCHEMA,
vol.Optional(CONF_UNIQUE_ID): cv.string,
vol.Optional(CONF_UNIT_OF_MEASUREMENT): cv.string,
vol.Optional(CONF_UPPER_LIMIT, default=False): cv.boolean,
}
)

76
homeassistant/components/compensation/sensor.py
View File

@ -7,15 +7,23 @@ from typing import Any
import numpy as np
from homeassistant.components.sensor import SensorEntity
from homeassistant.components.sensor import (
ATTR_STATE_CLASS,
CONF_STATE_CLASS,
SensorEntity,
)
from homeassistant.const import (
ATTR_DEVICE_CLASS,
ATTR_UNIT_OF_MEASUREMENT,
CONF_ATTRIBUTE,
CONF_DEVICE_CLASS,
CONF_MAXIMUM,
CONF_MINIMUM,
CONF_NAME,
CONF_SOURCE,
CONF_UNIQUE_ID,
CONF_UNIT_OF_MEASUREMENT,
STATE_UNAVAILABLE,
STATE_UNKNOWN,
)
from homeassistant.core import (
@ -59,24 +67,13 @@ async def async_setup_platform(
source: str = conf[CONF_SOURCE]
attribute: str | None = conf.get(CONF_ATTRIBUTE)
if not (name := conf.get(CONF_NAME)):
name = f"{DEFAULT_NAME} {source}"
if attribute is not None:
name = f"{name} {attribute}"
async_add_entities(
[
CompensationSensor(
conf.get(CONF_UNIQUE_ID),
name,
source,
attribute,
conf[CONF_PRECISION],
conf[CONF_POLYNOMIAL],
conf.get(CONF_UNIT_OF_MEASUREMENT),
conf[CONF_MINIMUM],
conf[CONF_MAXIMUM],
)
]
[CompensationSensor(conf.get(CONF_UNIQUE_ID), name, source, attribute, conf)]
)
@ -91,23 +88,27 @@ class CompensationSensor(SensorEntity):
name: str,
source: str,
attribute: str | None,
precision: int,
polynomial: np.poly1d,
unit_of_measurement: str | None,
minimum: tuple[float, float] | None,
maximum: tuple[float, float] | None,
config: dict[str, Any],
) -> None:
"""Initialize the Compensation sensor."""
self._attr_name = name
self._source_entity_id = source
self._precision = precision
self._source_attribute = attribute
self._attr_native_unit_of_measurement = unit_of_measurement
self._precision = config[CONF_PRECISION]
self._attr_native_unit_of_measurement = config.get(CONF_UNIT_OF_MEASUREMENT)
polynomial: np.poly1d = config[CONF_POLYNOMIAL]
self._poly = polynomial
self._coefficients = polynomial.coefficients.tolist()
self._attr_unique_id = unique_id
self._attr_name = name
self._minimum = minimum
self._maximum = maximum
self._minimum = config[CONF_MINIMUM]
self._maximum = config[CONF_MAXIMUM]
self._attr_device_class = config.get(CONF_DEVICE_CLASS)
self._attr_state_class = config.get(CONF_STATE_CLASS)
async def async_added_to_hass(self) -> None:
"""Handle added to Hass."""
@ -137,13 +138,40 @@ class CompensationSensor(SensorEntity):
"""Handle sensor state changes."""
new_state: State | None
if (new_state := event.data["new_state"]) is None:
_LOGGER.warning(
"While updating compensation %s, the new_state is None", self.name
)
self._attr_native_value = None
self.async_write_ha_state()
return
if new_state.state == STATE_UNKNOWN:
self._attr_native_value = None
self.async_write_ha_state()
return
if new_state.state == STATE_UNAVAILABLE:
self._attr_available = False
self.async_write_ha_state()
return
self._attr_available = True
if self.native_unit_of_measurement is None and self._source_attribute is None:
self._attr_native_unit_of_measurement = new_state.attributes.get(
ATTR_UNIT_OF_MEASUREMENT
)
if self._attr_device_class is None and (
device_class := new_state.attributes.get(ATTR_DEVICE_CLASS)
):
self._attr_device_class = device_class
if self._attr_state_class is None and (
state_class := new_state.attributes.get(ATTR_STATE_CLASS)
):
self._attr_state_class = state_class
if self._source_attribute:
value = new_state.attributes.get(self._source_attribute)
else:

381
tests/components/compensation/test_sensor.py
View File

@ -1,115 +1,181 @@
"""The tests for the integration sensor platform."""
from typing import Any
from unittest.mock import patch
import pytest
from homeassistant import config as hass_config
from homeassistant.components.compensation.const import CONF_PRECISION, DOMAIN
from homeassistant.components.compensation.sensor import ATTR_COEFFICIENTS
from homeassistant.components.sensor import DOMAIN as SENSOR_DOMAIN
from homeassistant.components.sensor import (
ATTR_STATE_CLASS,
SensorDeviceClass,
SensorStateClass,
)
from homeassistant.const import (
ATTR_DEVICE_CLASS,
ATTR_UNIT_OF_MEASUREMENT,
EVENT_HOMEASSISTANT_START,
EVENT_STATE_CHANGED,
SERVICE_RELOAD,
STATE_UNAVAILABLE,
STATE_UNKNOWN,
UnitOfTemperature,
)
from homeassistant.core import HomeAssistant
from homeassistant.setup import async_setup_component
from tests.common import assert_setup_component, get_fixture_path
async def test_linear_state(hass: HomeAssistant) -> None:
"""Test compensation sensor state."""
config = {
"compensation": {
"test": {
"source": "sensor.uncompensated",
TEST_OBJECT_ID = "test_compensation"
TEST_ENTITY_ID = "sensor.test_compensation"
TEST_SOURCE = "sensor.uncompensated"
TEST_BASE_CONFIG = {
"source": TEST_SOURCE,
"data_points": [
[1.0, 2.0],
[2.0, 3.0],
],
"precision": 2,
}
TEST_CONFIG = {
"name": TEST_OBJECT_ID,
"unit_of_measurement": "a",
**TEST_BASE_CONFIG,
}
}
}
expected_entity_id = "sensor.compensation_sensor_uncompensated"
assert await async_setup_component(hass, DOMAIN, config)
assert await async_setup_component(hass, SENSOR_DOMAIN, config)
async def async_setup_compensation(hass: HomeAssistant, config: dict[str, Any]) -> None:
"""Do setup of a compensation integration sensor."""
with assert_setup_component(1, DOMAIN):
assert await async_setup_component(
hass,
DOMAIN,
{DOMAIN: {"test": config}},
)
await hass.async_block_till_done()
await hass.async_start()
await hass.async_block_till_done()
@pytest.fixture
async def setup_compensation(hass: HomeAssistant, config: dict[str, Any]) -> None:
"""Do setup of a compensation integration sensor."""
await async_setup_compensation(hass, config)
@pytest.fixture
async def setup_compensation_with_limits(
hass: HomeAssistant,
config: dict[str, Any],
upper: bool,
lower: bool,
):
"""Do setup of a compensation integration sensor with extra config."""
await async_setup_compensation(
hass,
{
**config,
"lower_limit": lower,
"upper_limit": upper,
},
)
@pytest.fixture
async def caplog_setup_text(caplog: pytest.LogCaptureFixture) -> str:
"""Return setup log of integration."""
return caplog.text
@pytest.mark.parametrize("config", [TEST_CONFIG])
@pytest.mark.usefixtures("setup_compensation")
async def test_linear_state(hass: HomeAssistant, config: dict[str, Any]) -> None:
"""Test compensation sensor state."""
hass.bus.async_fire(EVENT_HOMEASSISTANT_START)
entity_id = config[DOMAIN]["test"]["source"]
hass.states.async_set(entity_id, 4, {})
hass.states.async_set(TEST_SOURCE, 4, {})
await hass.async_block_till_done()
state = hass.states.get(expected_entity_id)
state = hass.states.get(TEST_ENTITY_ID)
assert state is not None
assert round(float(state.state), config[DOMAIN]["test"][CONF_PRECISION]) == 5.0
assert round(float(state.state), config[CONF_PRECISION]) == 5.0
assert state.attributes.get(ATTR_UNIT_OF_MEASUREMENT) == "a"
coefs = [round(v, 1) for v in state.attributes.get(ATTR_COEFFICIENTS)]
assert coefs == [1.0, 1.0]
hass.states.async_set(entity_id, "foo", {})
hass.states.async_set(TEST_SOURCE, "foo", {})
await hass.async_block_till_done()
state = hass.states.get(expected_entity_id)
state = hass.states.get(TEST_ENTITY_ID)
assert state is not None
assert state.state == STATE_UNKNOWN
async def test_linear_state_from_attribute(hass: HomeAssistant) -> None:
"""Test compensation sensor state that pulls from attribute."""
config = {
"compensation": {
"test": {
"source": "sensor.uncompensated",
"attribute": "value",
"data_points": [
[1.0, 2.0],
[2.0, 3.0],
],
"precision": 2,
}
}
}
expected_entity_id = "sensor.compensation_sensor_uncompensated_value"
assert await async_setup_component(hass, DOMAIN, config)
assert await async_setup_component(hass, SENSOR_DOMAIN, config)
@pytest.mark.parametrize("config", [{"name": TEST_OBJECT_ID, **TEST_BASE_CONFIG}])
@pytest.mark.usefixtures("setup_compensation")
async def test_attributes_come_from_source(hass: HomeAssistant) -> None:
"""Test compensation sensor state."""
hass.bus.async_fire(EVENT_HOMEASSISTANT_START)
hass.states.async_set(
TEST_SOURCE,
4,
{
ATTR_DEVICE_CLASS: SensorDeviceClass.TEMPERATURE,
ATTR_UNIT_OF_MEASUREMENT: UnitOfTemperature.CELSIUS,
ATTR_STATE_CLASS: SensorStateClass.MEASUREMENT,
},
)
await hass.async_block_till_done()
state = hass.states.get(TEST_ENTITY_ID)
assert state is not None
assert state.state == "5.0"
assert state.attributes[ATTR_DEVICE_CLASS] == SensorDeviceClass.TEMPERATURE
assert state.attributes[ATTR_UNIT_OF_MEASUREMENT] == UnitOfTemperature.CELSIUS
assert state.attributes[ATTR_STATE_CLASS] == SensorStateClass.MEASUREMENT
@pytest.mark.parametrize("config", [{"attribute": "value", **TEST_CONFIG}])
@pytest.mark.usefixtures("setup_compensation")
async def test_linear_state_from_attribute(
hass: HomeAssistant, config: dict[str, Any]
) -> None:
"""Test compensation sensor state that pulls from attribute."""
hass.bus.async_fire(EVENT_HOMEASSISTANT_START)
entity_id = config[DOMAIN]["test"]["source"]
hass.states.async_set(entity_id, 3, {"value": 4})
hass.states.async_set(TEST_SOURCE, 3, {"value": 4})
await hass.async_block_till_done()
state = hass.states.get(expected_entity_id)
state = hass.states.get(TEST_ENTITY_ID)
assert state is not None
assert round(float(state.state), config[DOMAIN]["test"][CONF_PRECISION]) == 5.0
assert round(float(state.state), config[CONF_PRECISION]) == 5.0
coefs = [round(v, 1) for v in state.attributes.get(ATTR_COEFFICIENTS)]
assert coefs == [1.0, 1.0]
hass.states.async_set(entity_id, 3, {"value": "bar"})
hass.states.async_set(TEST_SOURCE, 3, {"value": "bar"})
await hass.async_block_till_done()
state = hass.states.get(expected_entity_id)
state = hass.states.get(TEST_ENTITY_ID)
assert state is not None
assert state.state == STATE_UNKNOWN
async def test_quadratic_state(hass: HomeAssistant) -> None:
"""Test 3 degree polynominial compensation sensor."""
config = {
"compensation": {
"test": {
"source": "sensor.temperature",
@pytest.mark.parametrize(
"config",
[
{
"name": TEST_OBJECT_ID,
"source": TEST_SOURCE,
"data_points": [
[50, 3.3],
[50, 2.8],
@ -129,46 +195,38 @@ async def test_quadratic_state(hass: HomeAssistant) -> None:
],
"degree": 2,
"precision": 3,
}
}
}
assert await async_setup_component(hass, DOMAIN, config)
await hass.async_block_till_done()
await hass.async_start()
},
],
)
@pytest.mark.usefixtures("setup_compensation")
async def test_quadratic_state(hass: HomeAssistant, config: dict[str, Any]) -> None:
"""Test 3 degree polynominial compensation sensor."""
hass.states.async_set(TEST_SOURCE, 43.2, {})
await hass.async_block_till_done()
entity_id = config[DOMAIN]["test"]["source"]
hass.states.async_set(entity_id, 43.2, {})
await hass.async_block_till_done()
state = hass.states.get("sensor.compensation_sensor_temperature")
state = hass.states.get(TEST_ENTITY_ID)
assert state is not None
assert round(float(state.state), config[DOMAIN]["test"][CONF_PRECISION]) == 3.327
assert round(float(state.state), config[CONF_PRECISION]) == 3.327
async def test_numpy_errors(
hass: HomeAssistant, caplog: pytest.LogCaptureFixture
) -> None:
"""Tests bad polyfits."""
config = {
"compensation": {
"test": {
"source": "sensor.uncompensated",
@pytest.mark.parametrize(
"config",
[
{
"source": TEST_SOURCE,
"data_points": [
[0.0, 1.0],
[0.0, 1.0],
],
},
}
}
await async_setup_component(hass, DOMAIN, config)
await hass.async_block_till_done()
await hass.async_start()
await hass.async_block_till_done()
assert "invalid value encountered in divide" in caplog.text
],
)
@pytest.mark.usefixtures("setup_compensation")
async def test_numpy_errors(hass: HomeAssistant, caplog_setup_text) -> None:
"""Tests bad polyfits."""
assert "invalid value encountered in divide" in caplog_setup_text
async def test_datapoints_greater_than_degree(
@ -178,7 +236,7 @@ async def test_datapoints_greater_than_degree(
config = {
"compensation": {
"test": {
"source": "sensor.uncompensated",
"source": TEST_SOURCE,
"data_points": [
[1.0, 2.0],
[2.0, 3.0],
@ -195,35 +253,13 @@ async def test_datapoints_greater_than_degree(
assert "data_points must have at least 3 data_points" in caplog.text
@pytest.mark.parametrize("config", [TEST_CONFIG])
@pytest.mark.usefixtures("setup_compensation")
async def test_new_state_is_none(hass: HomeAssistant) -> None:
"""Tests catch for empty new states."""
config = {
"compensation": {
"test": {
"source": "sensor.uncompensated",
"data_points": [
[1.0, 2.0],
[2.0, 3.0],
],
"precision": 2,
"unit_of_measurement": "a",
}
}
}
expected_entity_id = "sensor.compensation_sensor_uncompensated"
await async_setup_component(hass, DOMAIN, config)
await hass.async_block_till_done()
await hass.async_start()
await hass.async_block_till_done()
last_changed = hass.states.get(expected_entity_id).last_changed
hass.bus.async_fire(
EVENT_STATE_CHANGED, event_data={"entity_id": "sensor.uncompensated"}
)
assert last_changed == hass.states.get(expected_entity_id).last_changed
last_changed = hass.states.get(TEST_ENTITY_ID).last_changed
hass.bus.async_fire(EVENT_STATE_CHANGED, event_data={"entity_id": TEST_SOURCE})
assert last_changed == hass.states.get(TEST_ENTITY_ID).last_changed
@pytest.mark.parametrize(
@ -234,40 +270,129 @@ async def test_new_state_is_none(hass: HomeAssistant) -> None:
(True, True),
],
)
async def test_limits(hass: HomeAssistant, lower: bool, upper: bool) -> None:
"""Test compensation sensor state."""
source = "sensor.test"
config = {
"compensation": {
"test": {
"source": source,
@pytest.mark.parametrize(
"config",
[
{
"name": TEST_OBJECT_ID,
"source": TEST_SOURCE,
"data_points": [
[1.0, 0.0],
[3.0, 2.0],
[2.0, 1.0],
],
"precision": 2,
"lower_limit": lower,
"upper_limit": upper,
"unit_of_measurement": "a",
}
}
}
await async_setup_component(hass, DOMAIN, config)
},
],
)
@pytest.mark.usefixtures("setup_compensation_with_limits")
async def test_limits(hass: HomeAssistant, lower: bool, upper: bool) -> None:
"""Test compensation sensor state."""
hass.states.async_set(TEST_SOURCE, 0, {})
await hass.async_block_till_done()
await hass.async_start()
await hass.async_block_till_done()
entity_id = "sensor.compensation_sensor_test"
hass.states.async_set(source, 0, {})
await hass.async_block_till_done()
state = hass.states.get(entity_id)
state = hass.states.get(TEST_ENTITY_ID)
value = 0.0 if lower else -1.0
assert float(state.state) == value
hass.states.async_set(source, 5, {})
hass.states.async_set(TEST_SOURCE, 5, {})
await hass.async_block_till_done()
state = hass.states.get(entity_id)
state = hass.states.get(TEST_ENTITY_ID)
value = 2.0 if upper else 4.0
assert float(state.state) == value
@pytest.mark.parametrize(
("config", "expected"),
[
(TEST_BASE_CONFIG, "sensor.compensation_sensor_uncompensated"),
(
{"attribute": "value", **TEST_BASE_CONFIG},
"sensor.compensation_sensor_uncompensated_value",
),
],
)
@pytest.mark.usefixtures("setup_compensation")
async def test_default_name(hass: HomeAssistant, expected: str) -> None:
"""Test default configuration name."""
assert hass.states.get(expected) is not None
@pytest.mark.parametrize("config", [TEST_CONFIG])
@pytest.mark.parametrize(
("source_state", "expected"),
[(STATE_UNKNOWN, STATE_UNKNOWN), (STATE_UNAVAILABLE, STATE_UNAVAILABLE)],
)
@pytest.mark.usefixtures("setup_compensation")
async def test_non_numerical_states_from_source_entity(
hass: HomeAssistant, config: dict[str, Any], source_state: str, expected: str
) -> None:
"""Test non-numerical states from source entity."""
hass.bus.async_fire(EVENT_HOMEASSISTANT_START)
hass.states.async_set(TEST_SOURCE, source_state)
await hass.async_block_till_done()
state = hass.states.get(TEST_ENTITY_ID)
assert state is not None
assert state.state == expected
hass.states.async_set(TEST_SOURCE, 4)
await hass.async_block_till_done()
state = hass.states.get(TEST_ENTITY_ID)
assert state is not None
assert round(float(state.state), config[CONF_PRECISION]) == 5.0
hass.states.async_set(TEST_SOURCE, source_state)
await hass.async_block_till_done()
state = hass.states.get(TEST_ENTITY_ID)
assert state is not None
assert state.state == expected
async def test_source_state_none(hass: HomeAssistant) -> None:
"""Test is source sensor state is null and sets state to STATE_UNKNOWN."""
config = {
"sensor": [
{
"platform": "template",
"sensors": {
"uncompensated": {
"value_template": "{{ states.sensor.test_state.state }}"
}
},
},
]
}
await async_setup_component(hass, "sensor", config)
await async_setup_compensation(hass, TEST_CONFIG)
hass.states.async_set("sensor.test_state", 4)
await hass.async_block_till_done()
state = hass.states.get(TEST_SOURCE)
assert state.state == "4"
await hass.async_block_till_done()
state = hass.states.get(TEST_ENTITY_ID)
assert state.state == "5.0"
# Force Template Reload
yaml_path = get_fixture_path("sensor_configuration.yaml", "template")
with patch.object(hass_config, "YAML_CONFIG_FILE", yaml_path):
await hass.services.async_call(
"template",
SERVICE_RELOAD,
{},
blocking=True,
)
await hass.async_block_till_done()
# Template state gets to None
state = hass.states.get(TEST_SOURCE)
assert state is None
# Filter sensor ignores None state setting state to STATE_UNKNOWN
state = hass.states.get(TEST_ENTITY_ID)
assert state.state == STATE_UNKNOWN