From f5f0f8ce4f9b85a6885e7c790851307943f0522f Mon Sep 17 00:00:00 2001 From: Jim Strang Date: Fri, 18 Sep 2026 04:03:21 -0400 Subject: [PATCH] Add UniFi UPS battery telemetry (#182423) --- homeassistant/components/unifi/sensor.py | 155 +++++++++++++++++++- homeassistant/components/unifi/strings.json | 24 +++ tests/components/unifi/test_sensor.py | 118 +++++++++++++++ 3 files changed, 295 insertions(+), 2 deletions(-) diff --git a/homeassistant/components/unifi/sensor.py b/homeassistant/components/unifi/sensor.py index 61698aab9c07..ebbc75d7432e 100644 --- a/homeassistant/components/unifi/sensor.py +++ b/homeassistant/components/unifi/sensor.py @@ -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, diff --git a/homeassistant/components/unifi/strings.json b/homeassistant/components/unifi/strings.json index c44b2d9c5260..3fbc43e82be9 100644 --- a/homeassistant/components/unifi/strings.json +++ b/homeassistant/components/unifi/strings.json @@ -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" }, diff --git a/tests/components/unifi/test_sensor.py b/tests/components/unifi/test_sensor.py index 80e65b8651f9..158bce8f455e 100644 --- a/tests/components/unifi/test_sensor.py +++ b/tests/components/unifi/test_sensor.py @@ -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", [