Add Switch platform and PDU dynamic outlet switches to NUT (#141159)

This commit is contained in:
tdfountain 2025-03-22 22:27:52 -07:00 committed by GitHub
parent e2e80a850c
commit 9e86ca2e9e
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
6 changed files with 259 additions and 13 deletions

View File

@ -118,6 +118,8 @@ async def async_setup_entry(hass: HomeAssistant, entry: NutConfigEntry) -> bool:
outlet_num_str: str = str(outlet_num)
additional_integration_commands |= {
f"outlet.{outlet_num_str}.load.cycle",
f"outlet.{outlet_num_str}.load.on",
f"outlet.{outlet_num_str}.load.off",
}
valid_integration_commands = (

View File

@ -9,6 +9,7 @@ DOMAIN = "nut"
PLATFORMS = [
Platform.BUTTON,
Platform.SENSOR,
Platform.SWITCH,
]
DEFAULT_NAME = "NUT UPS"
@ -66,10 +67,6 @@ COMMAND_TEST_FAILURE_STOP = "test.failure.stop"
COMMAND_TEST_PANEL_START = "test.panel.start"
COMMAND_TEST_PANEL_STOP = "test.panel.stop"
COMMAND_TEST_SYSTEM_START = "test.system.start"
COMMAND_OUTLET_1_LOAD_OFF = "outlet.1.load.off"
COMMAND_OUTLET_1_LOAD_ON = "outlet.1.load.on"
COMMAND_OUTLET_2_LOAD_OFF = "outlet.2.load.off"
COMMAND_OUTLET_2_LOAD_ON = "outlet.2.load.on"
INTEGRATION_SUPPORTED_COMMANDS = {
COMMAND_BEEPER_DISABLE,
@ -98,8 +95,4 @@ INTEGRATION_SUPPORTED_COMMANDS = {
COMMAND_TEST_PANEL_START,
COMMAND_TEST_PANEL_STOP,
COMMAND_TEST_SYSTEM_START,
COMMAND_OUTLET_1_LOAD_OFF,
COMMAND_OUTLET_1_LOAD_ON,
COMMAND_OUTLET_2_LOAD_OFF,
COMMAND_OUTLET_2_LOAD_ON,
}

View File

@ -156,6 +156,11 @@
"outlet_number_load_cycle": {
"default": "mdi:restart"
}
},
"switch": {
"outlet_number_load_poweronoff": {
"default": "mdi:power"
}
}
}
}

View File

@ -74,11 +74,7 @@
"test_failure_stop": "Stop simulating a power failure",
"test_panel_start": "Start testing the UPS panel",
"test_panel_stop": "Stop a UPS panel test",
"test_system_start": "Start a system test",
"outlet_1_load_on": "Power outlet 1 on",
"outlet_1_load_off": "Power outlet 1 off",
"outlet_2_load_on": "Power outlet 2 on",
"outlet_2_load_off": "Power outlet 2 off"
"test_system_start": "Start a system test"
}
},
"entity": {
@ -224,6 +220,9 @@
},
"button": {
"outlet_number_load_cycle": { "name": "Power cycle outlet {outlet_name}" }
},
"switch": {
"outlet_number_load_poweronoff": { "name": "Power outlet {outlet_name}" }
}
}
}

View File

@ -0,0 +1,88 @@
"""Provides a switch for switchable NUT outlets."""
from __future__ import annotations
import logging
from typing import Any
from homeassistant.components.switch import (
SwitchDeviceClass,
SwitchEntity,
SwitchEntityDescription,
)
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 switches."""
pynut_data = config_entry.runtime_data
coordinator = pynut_data.coordinator
status = coordinator.data
# Dynamically add outlet switch types
if (num_outlets := status.get("outlet.count")) is None:
return
data = pynut_data.data
unique_id = pynut_data.unique_id
user_available_commands = pynut_data.user_available_commands
switch_descriptions = [
SwitchEntityDescription(
key=f"outlet.{outlet_num!s}.load.poweronoff",
translation_key="outlet_number_load_poweronoff",
translation_placeholders={
"outlet_name": status.get(f"outlet.{outlet_num!s}.name")
or str(outlet_num)
},
device_class=SwitchDeviceClass.OUTLET,
entity_registry_enabled_default=True,
)
for outlet_num in range(1, int(num_outlets) + 1)
if (
status.get(f"outlet.{outlet_num!s}.switchable") == "yes"
and f"outlet.{outlet_num!s}.load.on" in user_available_commands
and f"outlet.{outlet_num!s}.load.off" in user_available_commands
)
]
async_add_entities(
NUTSwitch(coordinator, description, data, unique_id)
for description in switch_descriptions
)
class NUTSwitch(NUTBaseEntity, SwitchEntity):
"""Representation of a switch entity for NUT status values."""
@property
def is_on(self) -> bool | None:
"""Return the state of the switch."""
status = self.coordinator.data
outlet, outlet_num_str = self.entity_description.key.split(".", 2)[:2]
if (state := status.get(f"{outlet}.{outlet_num_str}.status")) is None:
return None
return bool(state == "on")
async def async_turn_on(self, **kwargs: Any) -> None:
"""Turn on the device."""
outlet, outlet_num_str = self.entity_description.key.split(".", 2)[:2]
command_name = f"{outlet}.{outlet_num_str}.load.on"
await self.pynut_data.async_run_command(command_name)
async def async_turn_off(self, **kwargs: Any) -> None:
"""Turn off the device."""
outlet, outlet_num_str = self.entity_description.key.split(".", 2)[:2]
command_name = f"{outlet}.{outlet_num_str}.load.off"
await self.pynut_data.async_run_command(command_name)

View File

@ -0,0 +1,159 @@
"""Test the NUT switch platform."""
import json
from unittest.mock import AsyncMock
import pytest
from homeassistant.components.nut.const import INTEGRATION_SUPPORTED_COMMANDS
from homeassistant.components.switch import DOMAIN as SWITCH_DOMAIN
from homeassistant.const import (
ATTR_ENTITY_ID,
SERVICE_TURN_OFF,
SERVICE_TURN_ON,
STATE_ON,
STATE_UNKNOWN,
)
from homeassistant.core import HomeAssistant
from homeassistant.helpers import entity_registry as er
from .util import async_init_integration
from tests.common import load_fixture
@pytest.mark.parametrize(
"model",
[
"CP1350C",
"5E650I",
"5E850I",
"CP1500PFCLCD",
"DL650ELCD",
"EATON5P1550",
"blazer_usb",
],
)
async def test_switch_ups(
hass: HomeAssistant, entity_registry: er.EntityRegistry, model: str
) -> None:
"""Tests that there are no standard switches."""
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,
)
switch = hass.states.get("switch.ups1_power_outlet_1")
assert not switch
@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_switch_pdu_dynamic_outlets(
hass: HomeAssistant,
entity_registry: er.EntityRegistry,
model: str,
unique_id_base: str,
) -> None:
"""Tests that the switch 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.on"
list_commands_return_value[command] = command
command = f"outlet.{num!s}.load.off"
list_commands_return_value[command] = command
ups_fixture = f"nut/{model}.json"
list_vars = json.loads(load_fixture(ups_fixture))
run_command = AsyncMock()
await async_init_integration(
hass,
model,
list_vars=list_vars,
list_commands_return_value=list_commands_return_value,
run_command=run_command,
)
entity_id = "switch.ups1_power_outlet_a1"
entry = entity_registry.async_get(entity_id)
assert entry
assert entry.unique_id == f"{unique_id_base}_outlet.1.load.poweronoff"
switch = hass.states.get(entity_id)
assert switch
assert switch.state == STATE_ON
await hass.services.async_call(
SWITCH_DOMAIN,
SERVICE_TURN_OFF,
{ATTR_ENTITY_ID: entity_id},
blocking=True,
)
run_command.assert_called_with("ups1", "outlet.1.load.off")
await hass.services.async_call(
SWITCH_DOMAIN,
SERVICE_TURN_ON,
{ATTR_ENTITY_ID: entity_id},
blocking=True,
)
run_command.assert_called_with("ups1", "outlet.1.load.on")
switch = hass.states.get("switch.ups1_power_outlet_25")
assert not switch
switch = hass.states.get("switch.ups1_power_outlet_a25")
assert not switch
async def test_switch_pdu_dynamic_outlets_state_unknown(
hass: HomeAssistant,
entity_registry: er.EntityRegistry,
) -> None:
"""Test switch entity with missing status is reported as unknown."""
config_entry = await async_init_integration(
hass,
list_ups={"ups1": "UPS 1"},
list_vars={
"outlet.count": "1",
"outlet.1.switchable": "yes",
"outlet.1.name": "A1",
},
list_commands_return_value={
"outlet.1.load.on": None,
"outlet.1.load.off": None,
},
)
entity_id = "switch.ups1_power_outlet_a1"
entry = entity_registry.async_get(entity_id)
assert entry
assert entry.unique_id == f"{config_entry.entry_id}_outlet.1.load.poweronoff"
switch = hass.states.get(entity_id)
assert switch
assert switch.state == STATE_UNKNOWN