Add PDU dynamic outlet buttons to NUT (#140317)

homeassistant/components/nut/__init__.py

@@ -103,7 +103,7 @@ async def async_setup_entry(hass: HomeAssistant, entry: NutConfigEntry) -> bool:
     )
     status = coordinator.data
 
-    _LOGGER.debug("NUT Sensors Available: %s", status)
+    _LOGGER.debug("NUT Sensors Available: %s", status if status else None)
 
     entry.async_on_unload(entry.add_update_listener(_async_update_listener))
     unique_id = _unique_id_from_status(status)
@@ -111,14 +111,32 @@ async def async_setup_entry(hass: HomeAssistant, entry: NutConfigEntry) -> bool:
         unique_id = entry.entry_id
 
     if username is not None and password is not None:
+        # Dynamically add outlet integration commands
+        additional_integration_commands = set()
+        if (num_outlets := status.get("outlet.count")) is not None:
+            for outlet_num in range(1, int(num_outlets) + 1):
+                outlet_num_str: str = str(outlet_num)
+                additional_integration_commands |= {
+                    f"outlet.{outlet_num_str}.load.cycle",
+                }
+
+        valid_integration_commands = (
+            INTEGRATION_SUPPORTED_COMMANDS | additional_integration_commands
+        )
+
         user_available_commands = {
-            device_supported_command
+            device_command
-            for device_supported_command in await data.async_list_commands() or {}
+            for device_command in await data.async_list_commands() or {}
-            if device_supported_command in INTEGRATION_SUPPORTED_COMMANDS
+            if device_command in valid_integration_commands
         }
     else:
         user_available_commands = set()
 
+    _LOGGER.debug(
+        "NUT Commands Available: %s",
+        user_available_commands if user_available_commands else None,
+    )
+
     entry.runtime_data = NutRuntimeData(
         coordinator, data, unique_id, user_available_commands
     )

homeassistant/components/nut/button.py

@@ -0,0 +1,65 @@
+"""Provides a switch for switchable NUT outlets."""
+
+from __future__ import annotations
+
+import logging
+
+from homeassistant.components.button import (
+    ButtonDeviceClass,
+    ButtonEntity,
+    ButtonEntityDescription,
+)
+from homeassistant.core import HomeAssistant
+from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
+
+from . import NutConfigEntry
+from .entity import NUTBaseEntity
+
+_LOGGER = logging.getLogger(__name__)
+
+
+async def async_setup_entry(
+    hass: HomeAssistant,
+    config_entry: NutConfigEntry,
+    async_add_entities: AddConfigEntryEntitiesCallback,
+) -> None:
+    """Set up the NUT buttons."""
+    pynut_data = config_entry.runtime_data
+    coordinator = pynut_data.coordinator
+    status = coordinator.data
+
+    # Dynamically add outlet button types
+    if (num_outlets := status.get("outlet.count")) is None:
+        return
+
+    data = pynut_data.data
+    unique_id = pynut_data.unique_id
+    valid_button_types: dict[str, ButtonEntityDescription] = {}
+    for outlet_num in range(1, int(num_outlets) + 1):
+        outlet_num_str = str(outlet_num)
+        outlet_name: str = status.get(f"outlet.{outlet_num_str}.name") or outlet_num_str
+        valid_button_types |= {
+            f"outlet.{outlet_num_str}.load.cycle": ButtonEntityDescription(
+                key=f"outlet.{outlet_num_str}.load.cycle",
+                translation_key="outlet_number_load_cycle",
+                translation_placeholders={"outlet_name": outlet_name},
+                device_class=ButtonDeviceClass.RESTART,
+                entity_registry_enabled_default=True,
+            ),
+        }
+
+    async_add_entities(
+        NUTButton(coordinator, description, data, unique_id)
+        for button_id, description in valid_button_types.items()
+        if button_id in pynut_data.user_available_commands
+    )
+
+
+class NUTButton(NUTBaseEntity, ButtonEntity):
+    """Representation of a button entity for NUT."""
+
+    async def async_press(self) -> None:
+        """Press the button."""
+        name_list = self.entity_description.key.split(".")
+        command_name = f"{name_list[0]}.{name_list[1]}.load.cycle"
+        await self.pynut_data.async_run_command(command_name)

homeassistant/components/nut/const.py

@@ -6,7 +6,10 @@ from homeassistant.const import Platform
 
 DOMAIN = "nut"
 
-PLATFORMS = [Platform.SENSOR]
+PLATFORMS = [
+    Platform.BUTTON,
+    Platform.SENSOR,
+]
 
 DEFAULT_NAME = "NUT UPS"
 DEFAULT_HOST = "localhost"

homeassistant/components/nut/entity.py

@@ -12,6 +12,7 @@ from homeassistant.const import (
     ATTR_SW_VERSION,
 )
 from homeassistant.helpers.device_registry import DeviceInfo
+from homeassistant.helpers.entity import EntityDescription
 from homeassistant.helpers.update_coordinator import (
     CoordinatorEntity,
     DataUpdateCoordinator,
@@ -36,12 +37,16 @@ class NUTBaseEntity(CoordinatorEntity[DataUpdateCoordinator]):
     def __init__(
         self,
         coordinator: DataUpdateCoordinator,
+        entity_description: EntityDescription,
         data: PyNUTData,
         unique_id: str,
     ) -> None:
         """Initialize the entity."""
         super().__init__(coordinator)
 
+        self.entity_description = entity_description
+        self._attr_unique_id = f"{unique_id}_{entity_description.key}"
+
         self.pynut_data = data
         self._attr_device_info = DeviceInfo(
             identifiers={(DOMAIN, unique_id)},

homeassistant/components/nut/icons.json

@@ -151,6 +151,11 @@
       "ups_watchdog_status": {
         "default": "mdi:information-outline"
       }
+    },
+    "button": {
+      "outlet_number_load_cycle": {
+        "default": "mdi:restart"
+      }
     }
   }
 }

homeassistant/components/nut/sensor.py

@@ -25,9 +25,8 @@ from homeassistant.const import (
 )
 from homeassistant.core import HomeAssistant
 from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
-from homeassistant.helpers.update_coordinator import DataUpdateCoordinator
 
-from . import NutConfigEntry, PyNUTData
+from . import NutConfigEntry
 from .const import KEY_STATUS, KEY_STATUS_DISPLAY, STATE_TYPES
 from .entity import NUTBaseEntity
 
@@ -1089,20 +1088,6 @@ async def async_setup_entry(
 class NUTSensor(NUTBaseEntity, SensorEntity):
     """Representation of a sensor entity for NUT status values."""
 
-    _attr_has_entity_name = True
-
-    def __init__(
-        self,
-        coordinator: DataUpdateCoordinator[dict[str, str]],
-        sensor_description: SensorEntityDescription,
-        data: PyNUTData,
-        unique_id: str,
-    ) -> None:
-        """Initialize the sensor."""
-        super().__init__(coordinator, data, unique_id)
-        self.entity_description = sensor_description
-        self._attr_unique_id = f"{unique_id}_{sensor_description.key}"
-
     @property
     def native_value(self) -> str | None:
         """Return entity state from NUT device."""

homeassistant/components/nut/strings.json

@@ -221,6 +221,9 @@
       "ups_type": { "name": "UPS type" },
       "ups_watchdog_status": { "name": "Watchdog status" },
       "watts": { "name": "Watts" }
+    },
+    "button": {
+      "outlet_number_load_cycle": { "name": "Power cycle outlet {outlet_name}" }
     }
   }
 }

tests/components/nut/test_button.py

@@ -0,0 +1,102 @@
+"""Test the NUT button platform."""
+
+import pytest
+
+from homeassistant.components.button import DOMAIN as BUTTON_DOMAIN, SERVICE_PRESS
+from homeassistant.components.nut.const import INTEGRATION_SUPPORTED_COMMANDS
+from homeassistant.const import ATTR_ENTITY_ID, STATE_UNKNOWN
+from homeassistant.core import HomeAssistant
+from homeassistant.helpers import entity_registry as er
+
+from .util import async_init_integration
+
+
+@pytest.mark.parametrize(
+    "model",
+    [
+        "CP1350C",
+        "5E650I",
+        "5E850I",
+        "CP1500PFCLCD",
+        "DL650ELCD",
+        "EATON5P1550",
+        "blazer_usb",
+    ],
+)
+async def test_buttons_ups(
+    hass: HomeAssistant, entity_registry: er.EntityRegistry, model: str
+) -> None:
+    """Tests that there are no standard buttons."""
+
+    list_commands_return_value = {
+        supported_command: supported_command
+        for supported_command in INTEGRATION_SUPPORTED_COMMANDS
+    }
+
+    await async_init_integration(
+        hass,
+        model,
+        list_commands_return_value=list_commands_return_value,
+    )
+
+    button = hass.states.get("button.ups1_power_cycle_outlet_1")
+    assert not button
+
+
+@pytest.mark.parametrize(
+    ("model", "unique_id_base"),
+    [
+        (
+            "EATON-EPDU-G3",
+            "EATON_ePDU MA 00U-C IN: TYPE 00A 0P OUT: 00xTYPE_A000A00000_",
+        ),
+    ],
+)
+async def test_buttons_pdu_dynamic_outlets(
+    hass: HomeAssistant,
+    entity_registry: er.EntityRegistry,
+    model: str,
+    unique_id_base: str,
+) -> None:
+    """Tests that the button entities are correct."""
+
+    list_commands_return_value = {
+        supported_command: supported_command
+        for supported_command in INTEGRATION_SUPPORTED_COMMANDS
+    }
+
+    for num in range(1, 25):
+        command = f"outlet.{num!s}.load.cycle"
+        list_commands_return_value[command] = command
+
+    await async_init_integration(
+        hass,
+        model,
+        list_commands_return_value=list_commands_return_value,
+    )
+
+    entity_id = "button.ups1_power_cycle_outlet_a1"
+    entry = entity_registry.async_get(entity_id)
+    assert entry
+    assert entry.unique_id == f"{unique_id_base}outlet.1.load.cycle"
+
+    button = hass.states.get(entity_id)
+    assert button
+    assert button.state == STATE_UNKNOWN
+
+    await hass.services.async_call(
+        BUTTON_DOMAIN,
+        SERVICE_PRESS,
+        {ATTR_ENTITY_ID: entity_id},
+        blocking=True,
+    )
+    await hass.async_block_till_done()
+
+    button = hass.states.get(entity_id)
+    assert button.state != STATE_UNKNOWN
+
+    button = hass.states.get("button.ups1_power_cycle_outlet_25")
+    assert not button
+
+    button = hass.states.get("button.ups1_power_cycle_outlet_a25")
+    assert not button