starred/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2026-09-27 01:46:11 -04:00
Code Issues Packages Projects Releases Wiki Activity

Add UniFi UPS battery telemetry (#182423)

Browse Source
This commit is contained in:
Jim Strang
2026-09-18 10:03:21 +02:00
committed by GitHub
parent c00d1c8507
commit f5f0f8ce4f
3 changed files with 295 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files
homeassistant/components/unifi/sensor.py
+153 -2
View File
@@ -9,7 +9,7 @@ from dataclasses import dataclass
from datetime import date, datetime, timedelta
from decimal import Decimal
from functools import partial
from typing import TYPE_CHECKING, Literal, override
from typing import TYPE_CHECKING, Literal, cast, override
from aiounifi.interfaces.api_handlers import APIHandler, ItemEvent
from aiounifi.interfaces.clients import Clients
@@ -39,6 +39,8 @@ from homeassistant.const import (
PERCENTAGE,
EntityCategory,
UnitOfDataRate,
UnitOfElectricCurrent,
UnitOfElectricPotential,
UnitOfPower,
UnitOfTime,
)
@@ -200,6 +202,22 @@ def async_device_outlet_supported_fn(hub: UnifiHub, obj_id: str) -> bool:
return hub.api.devices[obj_id].outlet_ac_power_budget is not None
@callback
def async_device_battery_pool_supported_fn(
field: str, hub: UnifiHub, obj_id: str
) -> bool:
"""Determine if a device provides a battery pool field."""
return field in (hub.api.devices[obj_id].battery_pool or {})
@callback
def async_device_battery_pool_value_fn(
field: str, hub: UnifiHub, device: Device
) -> float | int:
"""Retrieve a battery pool field."""
return cast(dict[str, float | int], device.battery_pool)[field]
@callback
def async_device_uplink_mac_supported_fn(hub: UnifiHub, obj_id: str) -> bool:
"""Determine if a device supports reading uplink MAC address."""
@@ -404,7 +422,7 @@ class UnifiSensorEntityDescription[HandlerT: APIHandler, ApiItemT: ApiItem](
):
"""Class describing UniFi sensor entity."""
value_fn: Callable[[UnifiHub, ApiItemT], datetime | float | str | None]
value_fn: Callable[[UnifiHub, ApiItemT], datetime | float | int | str | None]
# Optional
is_connected_fn: Callable[[UnifiHub, str], bool] | None = None
@@ -625,6 +643,139 @@ ENTITY_DESCRIPTIONS: tuple[UnifiSensorEntityDescription, ...] = (
unique_id_fn=lambda hub, obj_id: f"ac_power_conumption-{obj_id}",
value_fn=lambda hub, device: device.outlet_ac_power_consumption,
),
UnifiSensorEntityDescription[Devices, Device](
key="UPS battery level",
translation_key="ups_battery_level",
device_class=SensorDeviceClass.BATTERY,
state_class=SensorStateClass.MEASUREMENT,
native_unit_of_measurement=PERCENTAGE,
api_handler_fn=lambda api: api.devices,
available_fn=async_device_available_fn,
device_info_fn=async_device_device_info_fn,
object_fn=lambda api, obj_id: api.devices[obj_id],
supported_fn=partial(async_device_battery_pool_supported_fn, "batteryLevel"),
unique_id_fn=lambda hub, obj_id: f"ups_battery_level-{obj_id}",
value_fn=partial(async_device_battery_pool_value_fn, "batteryLevel"),
),
UnifiSensorEntityDescription[Devices, Device](
key="UPS battery runtime",
translation_key="ups_battery_runtime",
device_class=SensorDeviceClass.DURATION,
entity_category=EntityCategory.DIAGNOSTIC,
native_unit_of_measurement=UnitOfTime.SECONDS,
api_handler_fn=lambda api: api.devices,
available_fn=async_device_available_fn,
device_info_fn=async_device_device_info_fn,
object_fn=lambda api, obj_id: api.devices[obj_id],
supported_fn=partial(async_device_battery_pool_supported_fn, "timeToRemain"),
unique_id_fn=lambda hub, obj_id: f"ups_battery_runtime-{obj_id}",
value_fn=partial(async_device_battery_pool_value_fn, "timeToRemain"),
),
UnifiSensorEntityDescription[Devices, Device](
key="UPS output power",
translation_key="ups_output_power",
device_class=SensorDeviceClass.POWER,
entity_category=EntityCategory.DIAGNOSTIC,
state_class=SensorStateClass.MEASUREMENT,
native_unit_of_measurement=UnitOfPower.WATT,
api_handler_fn=lambda api: api.devices,
available_fn=async_device_available_fn,
device_info_fn=async_device_device_info_fn,
object_fn=lambda api, obj_id: api.devices[obj_id],
supported_fn=partial(
async_device_battery_pool_supported_fn, "device_total_power_output"
),
unique_id_fn=lambda hub, obj_id: f"ups_output_power-{obj_id}",
value_fn=partial(
async_device_battery_pool_value_fn, "device_total_power_output"
),
),
UnifiSensorEntityDescription[Devices, Device](
key="UPS output current",
translation_key="ups_output_current",
device_class=SensorDeviceClass.CURRENT,
entity_category=EntityCategory.DIAGNOSTIC,
state_class=SensorStateClass.MEASUREMENT,
native_unit_of_measurement=UnitOfElectricCurrent.AMPERE,
api_handler_fn=lambda api: api.devices,
available_fn=async_device_available_fn,
device_info_fn=async_device_device_info_fn,
object_fn=lambda api, obj_id: api.devices[obj_id],
supported_fn=partial(
async_device_battery_pool_supported_fn, "device_output_current"
),
unique_id_fn=lambda hub, obj_id: f"ups_output_current-{obj_id}",
value_fn=partial(async_device_battery_pool_value_fn, "device_output_current"),
),
UnifiSensorEntityDescription[Devices, Device](
key="UPS output voltage",
translation_key="ups_output_voltage",
device_class=SensorDeviceClass.VOLTAGE,
entity_category=EntityCategory.DIAGNOSTIC,
state_class=SensorStateClass.MEASUREMENT,
native_unit_of_measurement=UnitOfElectricPotential.VOLT,
api_handler_fn=lambda api: api.devices,
available_fn=async_device_available_fn,
device_info_fn=async_device_device_info_fn,
object_fn=lambda api, obj_id: api.devices[obj_id],
supported_fn=partial(
async_device_battery_pool_supported_fn, "device_output_voltage"
),
unique_id_fn=lambda hub, obj_id: f"ups_output_voltage-{obj_id}",
value_fn=partial(async_device_battery_pool_value_fn, "device_output_voltage"),
),
UnifiSensorEntityDescription[Devices, Device](
key="UPS input voltage",
translation_key="ups_input_voltage",
device_class=SensorDeviceClass.VOLTAGE,
entity_category=EntityCategory.DIAGNOSTIC,
state_class=SensorStateClass.MEASUREMENT,
native_unit_of_measurement=UnitOfElectricPotential.VOLT,
api_handler_fn=lambda api: api.devices,
available_fn=async_device_available_fn,
device_info_fn=async_device_device_info_fn,
object_fn=lambda api, obj_id: api.devices[obj_id],
supported_fn=partial(
async_device_battery_pool_supported_fn, "device_input_voltage"
),
unique_id_fn=lambda hub, obj_id: f"ups_input_voltage-{obj_id}",
value_fn=partial(async_device_battery_pool_value_fn, "device_input_voltage"),
),
UnifiSensorEntityDescription[Devices, Device](
key="UPS bypass voltage",
translation_key="ups_bypass_voltage",
device_class=SensorDeviceClass.VOLTAGE,
entity_category=EntityCategory.DIAGNOSTIC,
state_class=SensorStateClass.MEASUREMENT,
native_unit_of_measurement=UnitOfElectricPotential.VOLT,
api_handler_fn=lambda api: api.devices,
available_fn=async_device_available_fn,
device_info_fn=async_device_device_info_fn,
object_fn=lambda api, obj_id: api.devices[obj_id],
supported_fn=partial(
async_device_battery_pool_supported_fn, "device_bypass_voltage"
),
unique_id_fn=lambda hub, obj_id: f"ups_bypass_voltage-{obj_id}",
value_fn=partial(async_device_battery_pool_value_fn, "device_bypass_voltage"),
),
UnifiSensorEntityDescription[Devices, Device](
key="UPS output power factor",
translation_key="ups_output_power_factor",
device_class=SensorDeviceClass.POWER_FACTOR,
entity_category=EntityCategory.DIAGNOSTIC,
state_class=SensorStateClass.MEASUREMENT,
api_handler_fn=lambda api: api.devices,
available_fn=async_device_available_fn,
device_info_fn=async_device_device_info_fn,
object_fn=lambda api, obj_id: api.devices[obj_id],
supported_fn=partial(
async_device_battery_pool_supported_fn, "device_total_power_factor"
),
unique_id_fn=lambda hub, obj_id: f"ups_output_power_factor-{obj_id}",
value_fn=partial(
async_device_battery_pool_value_fn, "device_total_power_factor"
),
),
UnifiSensorEntityDescription[Devices, Device](
key="Device uptime",
device_class=SensorDeviceClass.UPTIME,
homeassistant/components/unifi/strings.json
+24
View File
@@ -121,6 +121,30 @@
"smartpower_ac_power_consumption": {
"name": "AC power consumption"
},
"ups_battery_level": {
"name": "Battery level"
},
"ups_battery_runtime": {
"name": "Battery runtime"
},
"ups_bypass_voltage": {
"name": "Bypass voltage"
},
"ups_input_voltage": {
"name": "Input voltage"
},
"ups_output_current": {
"name": "Output current"
},
"ups_output_power": {
"name": "Output power"
},
"ups_output_power_factor": {
"name": "Output power factor"
},
"ups_output_voltage": {
"name": "Output voltage"
},
"wan_latency": {
"name": "{target} {wan} latency"
},
tests/components/unifi/test_sensor.py
+118
View File
@@ -32,6 +32,7 @@ from homeassistant.components.unifi.const import (
from homeassistant.config_entries import RELOAD_AFTER_UPDATE_DELAY
from homeassistant.const import (
ATTR_DEVICE_CLASS,
ATTR_UNIT_OF_MEASUREMENT,
STATE_UNAVAILABLE,
STATE_UNKNOWN,
EntityCategory,
@@ -402,9 +403,45 @@ UPS_DEVICE_1.update(
"index": 1,
}
],
"vbms_table": {
"battpool": {
"batteryLevel": 100,
"device_input_voltage": 121.9,
"device_output_current": 0.35,
"device_output_voltage": 121.7,
"device_total_power_factor": 0.98,
"device_total_power_output": 42.5,
"timeToRemain": 30600,
}
},
}
)
UPS_DEVICE_2 = deepcopy(UPS_DEVICE_1)
UPS_DEVICE_2.update(
{
"device_id": "mock-ups-2",
"mac": "02:00:00:00:00:02",
"model": "USWDA25",
"name": "Dummy UPS 2U",
"type": "usw",
"outlet_table": [
{
"index": index,
"relay_state": True,
"cycle_enabled": False,
"name": f"Outlet {index}",
"outlet_caps": caps,
}
for index, caps in enumerate(
(65549, 65549, 65549, 65549, 65541, 65541, 65541, 65541), start=1
)
],
}
)
UPS_DEVICE_2["vbms_table"]["battpool"].pop("device_input_voltage")
UPS_DEVICE_2["vbms_table"]["battpool"]["device_bypass_voltage"] = 121.7
@pytest.mark.parametrize(
"config_entry_options",
@@ -1020,6 +1057,87 @@ async def test_outlet_power_reading_extended_caps(
assert hass.states.get(entity_id).state == "43.5"
@pytest.mark.parametrize(
(
"device_payload",
"device_name",
"voltage_sensor",
"voltage_value",
"missing_voltage_sensor",
"metered_outlets",
),
[
pytest.param(
[UPS_DEVICE_1],
"dummy_ups_2u_pro",
"input_voltage",
"121.9",
"bypass_voltage",
(1, 2),
id="input_voltage",
),
pytest.param(
[UPS_DEVICE_2],
"dummy_ups_2u",
"bypass_voltage",
"121.7",
"input_voltage",
(),
id="bypass_voltage",
),
],
)
@pytest.mark.usefixtures("config_entry_setup")
async def test_ups_battery_pool_sensors(
hass: HomeAssistant,
entity_registry: er.EntityRegistry,
mock_websocket_message: WebsocketMessageMock,
device_payload: list[dict[str, Any]],
device_name: str,
voltage_sensor: str,
voltage_value: str,
missing_voltage_sensor: str,
metered_outlets: tuple[int, ...],
) -> None:
"""Test UPS battery pool telemetry sensors."""
assert {
entity_id
for entity_id in entity_registry.entities
if entity_id.startswith(f"sensor.{device_name}_outlet_")
and entity_id.endswith("_outlet_power")
} == {
f"sensor.{device_name}_outlet_{index}_outlet_power" for index in metered_outlets
}
assert hass.states.get(f"sensor.{device_name}_battery_level").state == "100"
assert hass.states.get(f"sensor.{device_name}_battery_runtime").state == "30600"
assert hass.states.get(f"sensor.{device_name}_output_power").state == "42.5"
assert hass.states.get(f"sensor.{device_name}_output_current").state == "0.35"
assert hass.states.get(f"sensor.{device_name}_output_voltage").state == "121.7"
assert (
hass.states.get(f"sensor.{device_name}_{voltage_sensor}").state == voltage_value
)
assert hass.states.get(f"sensor.{device_name}_{missing_voltage_sensor}") is None
power_factor = hass.states.get(f"sensor.{device_name}_output_power_factor")
assert power_factor is not None
assert power_factor.state == "0.98"
assert power_factor.attributes[ATTR_DEVICE_CLASS] == SensorDeviceClass.POWER_FACTOR
assert power_factor.attributes.get(ATTR_UNIT_OF_MEASUREMENT) is None
updated_device_data = deepcopy(device_payload[0])
updated_device_data["vbms_table"]["battpool"]["batteryLevel"] = 95
mock_websocket_message(message=MessageKey.DEVICE, data=updated_device_data)
await hass.async_block_till_done()
assert hass.states.get(f"sensor.{device_name}_battery_level").state == "95"
updated_device_data["vbms_table"].pop("battpool")
mock_websocket_message(message=MessageKey.DEVICE, data=updated_device_data)
await hass.async_block_till_done()
assert hass.states.get(f"sensor.{device_name}_battery_level") is None
assert entity_registry.async_get(f"sensor.{device_name}_battery_level") is None
@pytest.mark.parametrize(
"device_payload",
[