Add temperature sensors for unifi device (#122518)

* Add temperature sensors for device

* Move to single line

* Use right reference

* Always return a value

* Update tests

* Use slugify for id name

* Return default value if not present

* Make _device_temperature return value

* Add default value if temperatures is None

* Set value to go over all code paths

* Add test for no matching temperatures

* make first part deterministic
pull/124152/head
Kim de Vos 2024-08-16 21:37:24 +02:00 committed by GitHub
parent 9b11aaf1eb
commit a8a7d01a84
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
2 changed files with 234 additions and 2 deletions

View File

@ -21,7 +21,11 @@ from aiounifi.interfaces.ports import Ports
from aiounifi.interfaces.wlans import Wlans
from aiounifi.models.api import ApiItemT
from aiounifi.models.client import Client
from aiounifi.models.device import Device, TypedDeviceUptimeStatsWanMonitor
from aiounifi.models.device import (
Device,
TypedDeviceTemperature,
TypedDeviceUptimeStatsWanMonitor,
)
from aiounifi.models.outlet import Outlet
from aiounifi.models.port import Port
from aiounifi.models.wlan import Wlan
@ -280,6 +284,72 @@ def make_wan_latency_sensors() -> tuple[UnifiSensorEntityDescription, ...]:
)
@callback
def async_device_temperatures_value_fn(
temperature_name: str, hub: UnifiHub, device: Device
) -> float:
"""Retrieve the temperature of the device."""
return_value: float = 0
if device.temperatures:
temperature = _device_temperature(temperature_name, device.temperatures)
return_value = temperature if temperature is not None else 0
return return_value
@callback
def async_device_temperatures_supported_fn(
temperature_name: str, hub: UnifiHub, obj_id: str
) -> bool:
"""Determine if an device have a temperatures."""
if (device := hub.api.devices[obj_id]) and device.temperatures:
return _device_temperature(temperature_name, device.temperatures) is not None
return False
@callback
def _device_temperature(
temperature_name: str, temperatures: list[TypedDeviceTemperature]
) -> float | None:
"""Return the temperature of the device."""
for temperature in temperatures:
if temperature_name in temperature["name"]:
return temperature["value"]
return None
def make_device_temperatur_sensors() -> tuple[UnifiSensorEntityDescription, ...]:
"""Create device temperature sensors."""
def make_device_temperature_entity_description(
name: str,
) -> UnifiSensorEntityDescription:
return UnifiSensorEntityDescription[Devices, Device](
key=f"Device {name} temperature",
device_class=SensorDeviceClass.TEMPERATURE,
entity_category=EntityCategory.DIAGNOSTIC,
state_class=SensorStateClass.MEASUREMENT,
native_unit_of_measurement=UnitOfTemperature.CELSIUS,
entity_registry_enabled_default=False,
api_handler_fn=lambda api: api.devices,
available_fn=async_device_available_fn,
device_info_fn=async_device_device_info_fn,
name_fn=lambda device: f"{device.name} {name} Temperature",
object_fn=lambda api, obj_id: api.devices[obj_id],
supported_fn=partial(async_device_temperatures_supported_fn, name),
unique_id_fn=lambda hub, obj_id: f"temperature-{slugify(name)}-{obj_id}",
value_fn=partial(async_device_temperatures_value_fn, name),
)
return tuple(
make_device_temperature_entity_description(name)
for name in (
"CPU",
"Local",
"PHY",
)
)
@dataclass(frozen=True, kw_only=True)
class UnifiSensorEntityDescription(
SensorEntityDescription, UnifiEntityDescription[HandlerT, ApiItemT]
@ -544,7 +614,7 @@ ENTITY_DESCRIPTIONS: tuple[UnifiSensorEntityDescription, ...] = (
),
)
ENTITY_DESCRIPTIONS += make_wan_latency_sensors()
ENTITY_DESCRIPTIONS += make_wan_latency_sensors() + make_device_temperatur_sensors()
async def async_setup_entry(

View File

@ -1519,3 +1519,165 @@ async def test_wan_monitor_latency_with_no_uptime(
latency_entry = entity_registry.async_get("sensor.mock_name_google_wan_latency")
assert latency_entry is None
@pytest.mark.parametrize(
"device_payload",
[
[
{
"board_rev": 3,
"device_id": "mock-id",
"has_fan": True,
"fan_level": 0,
"ip": "10.0.1.1",
"last_seen": 1562600145,
"mac": "00:00:00:00:01:01",
"model": "US16P150",
"name": "Device",
"next_interval": 20,
"overheating": True,
"state": 1,
"type": "usw",
"upgradable": True,
"uptime": 60,
"version": "4.0.42.10433",
"temperatures": [
{"name": "CPU", "type": "cpu", "value": 66.0},
{"name": "Local", "type": "board", "value": 48.75},
{"name": "PHY", "type": "board", "value": 50.25},
],
}
]
],
)
@pytest.mark.parametrize(
("temperature_id", "state", "updated_state", "index_to_update"),
[
("device_cpu", "66.0", "20", 0),
("device_local", "48.75", "90.64", 1),
("device_phy", "50.25", "80", 2),
],
)
@pytest.mark.usefixtures("config_entry_setup")
async def test_device_temperatures(
hass: HomeAssistant,
entity_registry: er.EntityRegistry,
mock_websocket_message,
device_payload: list[dict[str, Any]],
temperature_id: str,
state: str,
updated_state: str,
index_to_update: int,
) -> None:
"""Verify that device temperatures sensors are working as expected."""
entity_id = f"sensor.device_{temperature_id}_temperature"
assert len(hass.states.async_all()) == 6
assert len(hass.states.async_entity_ids(SENSOR_DOMAIN)) == 2
temperature_entity = entity_registry.async_get(entity_id)
assert temperature_entity.disabled_by == RegistryEntryDisabler.INTEGRATION
# Enable entity
entity_registry.async_update_entity(entity_id=entity_id, disabled_by=None)
await hass.async_block_till_done()
async_fire_time_changed(
hass,
dt_util.utcnow() + timedelta(seconds=RELOAD_AFTER_UPDATE_DELAY + 1),
)
await hass.async_block_till_done()
assert len(hass.states.async_all()) == 7
assert len(hass.states.async_entity_ids(SENSOR_DOMAIN)) == 3
# Verify sensor state
assert hass.states.get(entity_id).state == state
# # Verify state update
device = device_payload[0]
device["temperatures"][index_to_update]["value"] = updated_state
mock_websocket_message(message=MessageKey.DEVICE, data=device)
assert hass.states.get(entity_id).state == updated_state
@pytest.mark.parametrize(
"device_payload",
[
[
{
"board_rev": 2,
"device_id": "mock-id",
"ip": "10.0.1.1",
"mac": "10:00:00:00:01:01",
"last_seen": 1562600145,
"model": "US16P150",
"name": "mock-name",
"port_overrides": [],
"state": 1,
"type": "usw",
"version": "4.0.42.10433",
}
]
],
)
@pytest.mark.usefixtures("config_entry_setup")
async def test_device_with_no_temperature(
hass: HomeAssistant,
entity_registry: er.EntityRegistry,
) -> None:
"""Verify that device temperature sensors is not created if there is no data."""
assert len(hass.states.async_all()) == 6
assert len(hass.states.async_entity_ids(SENSOR_DOMAIN)) == 2
temperature_entity = entity_registry.async_get(
"sensor.device_device_cpu_temperature"
)
assert temperature_entity is None
@pytest.mark.parametrize(
"device_payload",
[
[
{
"board_rev": 2,
"device_id": "mock-id",
"ip": "10.0.1.1",
"mac": "10:00:00:00:01:01",
"last_seen": 1562600145,
"model": "US16P150",
"name": "mock-name",
"port_overrides": [],
"state": 1,
"type": "usw",
"version": "4.0.42.10433",
"temperatures": [
{"name": "MEM", "type": "mem", "value": 66.0},
],
}
]
],
)
@pytest.mark.usefixtures("config_entry_setup")
async def test_device_with_no_matching_temperatures(
hass: HomeAssistant,
entity_registry: er.EntityRegistry,
) -> None:
"""Verify that device temperature sensors is not created if there is no matching data."""
assert len(hass.states.async_all()) == 6
assert len(hass.states.async_entity_ids(SENSOR_DOMAIN)) == 2
temperature_entity = entity_registry.async_get(
"sensor.device_device_cpu_temperature"
)
assert temperature_entity is None