evlli/hass-core
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Add NUT device actions (#80986)

Browse source 
Co-authored-by: J. Nick Koston <nick@koston.org>
This commit is contained in:
Pablo Estevez 2023-04-23 15:09:36 -04:00 • committed by GitHub
parent b6f2b29a99
commit 780d0a484d
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
8 changed files with 476 additions and 11 deletions
Download patch file Download diff file Expand all files Collapse all files

4
CODEOWNERS
View file

@ -825,8 +825,8 @@ build.json @home-assistant/supervisor
/tests/components/numato/ @clssn
/homeassistant/components/number/ @home-assistant/core @Shulyaka
/tests/components/number/ @home-assistant/core @Shulyaka
/homeassistant/components/nut/ @bdraco @ollo69
/tests/components/nut/ @bdraco @ollo69
/homeassistant/components/nut/ @bdraco @ollo69 @pestevez
/tests/components/nut/ @bdraco @ollo69 @pestevez
/homeassistant/components/nws/ @MatthewFlamm @kamiyo
/tests/components/nws/ @MatthewFlamm @kamiyo
/homeassistant/components/nzbget/ @chriscla

35
homeassistant/components/nut/__init__.py
View file

@ -4,6 +4,7 @@ from __future__ import annotations
from dataclasses import dataclass
from datetime import timedelta
import logging
from typing import cast
import async_timeout
from pynut2.nut2 import PyNUTClient, PyNUTError
@ -19,6 +20,7 @@ from homeassistant.const import (
CONF_USERNAME,
)
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers import device_registry as dr
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
@ -26,9 +28,11 @@ from .const import (
COORDINATOR,
DEFAULT_SCAN_INTERVAL,
DOMAIN,
INTEGRATION_SUPPORTED_COMMANDS,
PLATFORMS,
PYNUT_DATA,
PYNUT_UNIQUE_ID,
USER_AVAILABLE_COMMANDS,
)
NUT_FAKE_SERIAL = ["unknown", "blank"]
@ -86,11 +90,21 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
if unique_id is None:
unique_id = entry.entry_id
if username is not None and password is not None:
user_available_commands = {
device_supported_command
for device_supported_command in data.list_commands() or {}
if device_supported_command in INTEGRATION_SUPPORTED_COMMANDS
}
else:
user_available_commands = set()
hass.data.setdefault(DOMAIN, {})
hass.data[DOMAIN][entry.entry_id] = {
COORDINATOR: coordinator,
PYNUT_DATA: data,
PYNUT_UNIQUE_ID: unique_id,
USER_AVAILABLE_COMMANDS: user_available_commands,
}
device_registry = dr.async_get(hass)
@ -270,3 +284,24 @@ class PyNUTData:
self._status = self._get_status()
if self._device_info is None:
self._device_info = self._get_device_info()
async def async_run_command(
self, hass: HomeAssistant, command_name: str | None
) -> None:
"""Invoke instant command in UPS."""
try:
await hass.async_add_executor_job(
self._client.run_command, self._alias, command_name
)
except PyNUTError as err:
raise HomeAssistantError(
f"Error running command {command_name}, {err}"
) from err
def list_commands(self) -> dict[str, str] | None:
"""Fetch the list of supported commands."""
try:
return cast(dict[str, str], self._client.list_commands(self._alias))
except PyNUTError as err:
_LOGGER.error("Error retrieving supported commands %s", err)
return None

58
homeassistant/components/nut/const.py
View file

@ -21,6 +21,8 @@ PYNUT_DATA = "data"
PYNUT_UNIQUE_ID = "unique_id"
USER_AVAILABLE_COMMANDS = "user_available_commands"
STATE_TYPES = {
"OL": "Online",
"OB": "On Battery",
@ -38,3 +40,59 @@ STATE_TYPES = {
"FSD": "Forced Shutdown",
"ALARM": "Alarm",
}
COMMAND_BEEPER_DISABLE = "beeper.disable"
COMMAND_BEEPER_ENABLE = "beeper.enable"
COMMAND_BEEPER_MUTE = "beeper.mute"
COMMAND_BEEPER_TOGGLE = "beeper.toggle"
COMMAND_BYPASS_START = "bypass.start"
COMMAND_BYPASS_STOP = "bypass.stop"
COMMAND_CALIBRATE_START = "calibrate.start"
COMMAND_CALIBRATE_STOP = "calibrate.stop"
COMMAND_LOAD_OFF = "load.off"
COMMAND_LOAD_ON = "load.on"
COMMAND_RESET_INPUT_MINMAX = "reset.input.minmax"
COMMAND_RESET_WATCHDOG = "reset.watchdog"
COMMAND_SHUTDOWN_REBOOT = "shutdown.reboot"
COMMAND_SHUTDOWN_REBOOT_GRACEFUL = "shutdown.reboot.graceful"
COMMAND_SHUTDOWN_RETURN = "shutdown.return"
COMMAND_SHUTDOWN_STAYOFF = "shutdown.stayoff"
COMMAND_SHUTDOWN_STOP = "shutdown.stop"
COMMAND_TEST_BATTERY_START = "test.battery.start"
COMMAND_TEST_BATTERY_START_DEEP = "test.battery.start.deep"
COMMAND_TEST_BATTERY_START_QUICK = "test.battery.start.quick"
COMMAND_TEST_BATTERY_STOP = "test.battery.stop"
COMMAND_TEST_FAILURE_START = "test.failure.start"
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"
INTEGRATION_SUPPORTED_COMMANDS = {
COMMAND_BEEPER_DISABLE,
COMMAND_BEEPER_ENABLE,
COMMAND_BEEPER_MUTE,
COMMAND_BEEPER_TOGGLE,
COMMAND_BYPASS_START,
COMMAND_BYPASS_STOP,
COMMAND_CALIBRATE_START,
COMMAND_CALIBRATE_STOP,
COMMAND_LOAD_OFF,
COMMAND_LOAD_ON,
COMMAND_RESET_INPUT_MINMAX,
COMMAND_RESET_WATCHDOG,
COMMAND_SHUTDOWN_REBOOT,
COMMAND_SHUTDOWN_REBOOT_GRACEFUL,
COMMAND_SHUTDOWN_RETURN,
COMMAND_SHUTDOWN_STAYOFF,
COMMAND_SHUTDOWN_STOP,
COMMAND_TEST_BATTERY_START,
COMMAND_TEST_BATTERY_START_DEEP,
COMMAND_TEST_BATTERY_START_QUICK,
COMMAND_TEST_BATTERY_STOP,
COMMAND_TEST_FAILURE_START,
COMMAND_TEST_FAILURE_STOP,
COMMAND_TEST_PANEL_START,
COMMAND_TEST_PANEL_STOP,
COMMAND_TEST_SYSTEM_START,
}

75
homeassistant/components/nut/device_action.py Normal file
View file

@ -0,0 +1,75 @@
"""Provides device actions for Network UPS Tools (NUT)."""
from __future__ import annotations
import voluptuous as vol
from homeassistant.const import CONF_DEVICE_ID, CONF_DOMAIN, CONF_TYPE
from homeassistant.core import Context, HomeAssistant
from homeassistant.helpers import device_registry as dr
import homeassistant.helpers.config_validation as cv
from homeassistant.helpers.typing import ConfigType, TemplateVarsType
from . import PyNUTData
from .const import (
DOMAIN,
INTEGRATION_SUPPORTED_COMMANDS,
PYNUT_DATA,
USER_AVAILABLE_COMMANDS,
)
ACTION_TYPES = {cmd.replace(".", "_") for cmd in INTEGRATION_SUPPORTED_COMMANDS}
ACTION_SCHEMA = cv.DEVICE_ACTION_BASE_SCHEMA.extend(
{
vol.Required(CONF_TYPE): vol.In(ACTION_TYPES),
}
)
async def async_get_actions(
hass: HomeAssistant, device_id: str
) -> list[dict[str, str]]:
"""List device actions for Network UPS Tools (NUT) devices."""
if (entry_id := _get_entry_id_from_device_id(hass, device_id)) is None:
return []
base_action = {
CONF_DEVICE_ID: device_id,
CONF_DOMAIN: DOMAIN,
}
user_available_commands: set[str] = hass.data[DOMAIN][entry_id][
USER_AVAILABLE_COMMANDS
]
return [
{CONF_TYPE: _get_device_action_name(command_name)} | base_action
for command_name in user_available_commands
]
async def async_call_action_from_config(
hass: HomeAssistant,
config: ConfigType,
variables: TemplateVarsType,
context: Context | None,
) -> None:
"""Execute a device action."""
device_action_name: str = config[CONF_TYPE]
command_name = _get_command_name(device_action_name)
device_id: str = config[CONF_DEVICE_ID]
entry_id = _get_entry_id_from_device_id(hass, device_id)
data: PyNUTData = hass.data[DOMAIN][entry_id][PYNUT_DATA]
await data.async_run_command(hass, command_name)
def _get_device_action_name(command_name: str) -> str:
return command_name.replace(".", "_")
def _get_command_name(device_action_name: str) -> str:
return device_action_name.replace("_", ".")
def _get_entry_id_from_device_id(hass: HomeAssistant, device_id: str) -> str | None:
device_registry = dr.async_get(hass)
if (device := device_registry.async_get(device_id)) is None:
return None
return next(entry for entry in device.config_entries)

2
homeassistant/components/nut/manifest.json
View file

@ -1,7 +1,7 @@
{
"domain": "nut",
"name": "Network UPS Tools (NUT)",
"codeowners": ["@bdraco", "@ollo69"],
"codeowners": ["@bdraco", "@ollo69", "@pestevez"],
"config_flow": true,
"documentation": "https://www.home-assistant.io/integrations/nut",
"integration_type": "device",

30
homeassistant/components/nut/strings.json
View file

@ -34,6 +34,36 @@
}
}
},
"device_automation": {
"action_type": {
"beeper_disable": "Disable UPS beeper/buzzer",
"beeper_enable": "Enable UPS beeper/buzzer",
"beeper_mute": "Temporarily mute UPS beeper/buzzer",
"beeper_toggle": "Toggle UPS beeper/buzzer",
"bypass_start": "Put the UPS in bypass mode",
"bypass_stop": "Take the UPS out of bypass mode",
"calibrate_start": "Start runtime calibration",
"calibrate_stop": "Stop runtime calibration",
"load_off": "Turn off the load immediately",
"load_on": "Turn on the load immediately",
"reset_input_minmax": "Reset minimum and maximum input voltage status",
"reset_watchdog": "Reset watchdog timer (forced reboot of load)",
"shutdown_reboot": "Shut down the load briefly while rebooting the UPS",
"shutdown_reboot_graceful": "After a delay, shut down the load briefly while rebooting the UPS",
"shutdown_return": "Turn off the load possibly after a delay and return when power is back",
"shutdown_stayoff": "Turn off the load possibly after a delay and remain off even if power returns",
"shutdown_stop": "Stop a shutdown in progress",
"test_battery_start": "Start a battery test",
"test_battery_start_deep": "Start a deep battery test",
"test_battery_start_quick": "Start a quick battery test",
"test_battery_stop": "Stop the battery test",
"test_failure_start": "Start a simulated power failure",
"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"
}
},
"entity": {
"sensor": {
"ambient_humidity": { "name": "Ambient humidity" },

229
tests/components/nut/test_device_action.py Normal file
View file

@ -0,0 +1,229 @@
"""The tests for Network UPS Tools (NUT) device actions."""
from unittest.mock import MagicMock
from pynut2.nut2 import PyNUTError
import pytest
from homeassistant.components import automation, device_automation
from homeassistant.components.device_automation import DeviceAutomationType
from homeassistant.components.nut import DOMAIN
from homeassistant.components.nut.const import INTEGRATION_SUPPORTED_COMMANDS
from homeassistant.core import HomeAssistant
from homeassistant.helpers import device_registry as dr
from homeassistant.setup import async_setup_component
from .util import async_init_integration
from tests.common import assert_lists_same, async_get_device_automations
async def test_get_all_actions_for_specified_user(
hass: HomeAssistant,
device_registry: dr.DeviceRegistry,
) -> None:
"""Test we get all the expected actions from a nut if user is specified."""
list_commands_return_value = {
supported_command: supported_command
for supported_command in INTEGRATION_SUPPORTED_COMMANDS
}
await async_init_integration(
hass,
username="someuser",
password="somepassword",
list_vars={"ups.status": "OL"},
list_commands_return_value=list_commands_return_value,
)
device_entry = next(device for device in device_registry.devices.values())
expected_actions = [
{
"domain": DOMAIN,
"type": action.replace(".", "_"),
"device_id": device_entry.id,
"metadata": {},
}
for action in INTEGRATION_SUPPORTED_COMMANDS
]
actions = await async_get_device_automations(
hass, DeviceAutomationType.ACTION, device_entry.id
)
assert_lists_same(actions, expected_actions)
async def test_no_actions_for_anonymous_user(
hass: HomeAssistant,
device_registry: dr.DeviceRegistry,
) -> None:
"""Test we get no actions if user is not specified."""
list_commands_return_value = {"some action": "some description"}
await async_init_integration(
hass,
username=None,
password=None,
list_vars={"ups.status": "OL"},
list_commands_return_value=list_commands_return_value,
)
device_entry = next(device for device in device_registry.devices.values())
actions = await async_get_device_automations(
hass, DeviceAutomationType.ACTION, device_entry.id
)
assert len(actions) == 0
async def test_no_actions_invalid_device(
hass: HomeAssistant,
) -> None:
"""Test we get no actions for an invalid device."""
list_commands_return_value = {"beeper.enable": None}
await async_init_integration(
hass,
list_vars={"ups.status": "OL"},
list_commands_return_value=list_commands_return_value,
)
device_id = "invalid_device_id"
platform = await device_automation.async_get_device_automation_platform(
hass, DOMAIN, DeviceAutomationType.ACTION
)
actions = await platform.async_get_actions(hass, device_id)
assert len(actions) == 0
async def test_list_commands_exception(
hass: HomeAssistant, device_registry: dr.DeviceRegistry
) -> None:
"""Test there are no actions if list_commands raises exception."""
await async_init_integration(
hass, list_vars={"ups.status": "OL"}, list_commands_side_effect=PyNUTError
)
device_entry = next(device for device in device_registry.devices.values())
actions = await async_get_device_automations(
hass, DeviceAutomationType.ACTION, device_entry.id
)
assert len(actions) == 0
async def test_unsupported_command(
hass: HomeAssistant, device_registry: dr.DeviceRegistry
) -> None:
"""Test unsupported command is excluded."""
list_commands_return_value = {
"beeper.enable": None,
"device.something": "Does something unsupported",
}
await async_init_integration(
hass,
list_vars={"ups.status": "OL"},
list_commands_return_value=list_commands_return_value,
)
device_entry = next(device for device in device_registry.devices.values())
actions = await async_get_device_automations(
hass, DeviceAutomationType.ACTION, device_entry.id
)
assert len(actions) == 1
async def test_action(hass: HomeAssistant, device_registry: dr.DeviceRegistry) -> None:
"""Test actions are executed."""
list_commands_return_value = {
"beeper.enable": None,
"beeper.disable": None,
}
run_command = MagicMock()
await async_init_integration(
hass,
list_ups={"someUps": "Some UPS"},
list_vars={"ups.status": "OL"},
list_commands_return_value=list_commands_return_value,
run_command=run_command,
)
device_entry = next(device for device in device_registry.devices.values())
assert await async_setup_component(
hass,
automation.DOMAIN,
{
automation.DOMAIN: [
{
"trigger": {
"platform": "event",
"event_type": "test_some_event",
},
"action": {
"domain": DOMAIN,
"device_id": device_entry.id,
"type": "beeper_enable",
},
},
{
"trigger": {
"platform": "event",
"event_type": "test_another_event",
},
"action": {
"domain": DOMAIN,
"device_id": device_entry.id,
"type": "beeper_disable",
},
},
]
},
)
hass.bus.async_fire("test_some_event")
await hass.async_block_till_done()
run_command.assert_called_with("someUps", "beeper.enable")
hass.bus.async_fire("test_another_event")
await hass.async_block_till_done()
run_command.assert_called_with("someUps", "beeper.disable")
async def test_rund_command_exception(
hass: HomeAssistant,
device_registry: dr.DeviceRegistry,
caplog: pytest.LogCaptureFixture,
) -> None:
"""Test logged error if run command raises exception."""
list_commands_return_value = {"beeper.enable": None}
error_message = "Something wrong happened"
run_command = MagicMock(side_effect=PyNUTError(error_message))
await async_init_integration(
hass,
list_vars={"ups.status": "OL"},
list_commands_return_value=list_commands_return_value,
run_command=run_command,
)
device_entry = next(device for device in device_registry.devices.values())
assert await async_setup_component(
hass,
automation.DOMAIN,
{
automation.DOMAIN: [
{
"trigger": {
"platform": "event",
"event_type": "test_some_event",
},
"action": {
"domain": DOMAIN,
"device_id": device_entry.id,
"type": "beeper_enable",
},
},
]
},
)
hass.bus.async_fire("test_some_event")
await hass.async_block_till_done()
assert error_message in caplog.text

54
tests/components/nut/util.py
View file

@ -4,28 +4,61 @@ import json
from unittest.mock import MagicMock, patch
from homeassistant.components.nut.const import DOMAIN
from homeassistant.const import CONF_HOST, CONF_PORT
from homeassistant.const import CONF_HOST, CONF_PASSWORD, CONF_PORT, CONF_USERNAME
from homeassistant.core import HomeAssistant
from tests.common import MockConfigEntry, load_fixture
def _get_mock_pynutclient(list_vars=None, list_ups=None):
def _get_mock_pynutclient(
list_vars=None,
list_ups=None,
list_commands_return_value=None,
list_commands_side_effect=None,
run_command=None,
):
pynutclient = MagicMock()
type(pynutclient).list_ups = MagicMock(return_value=list_ups)
type(pynutclient).list_vars = MagicMock(return_value=list_vars)
if list_commands_return_value is None:
list_commands_return_value = {}
type(pynutclient).list_commands = MagicMock(
return_value=list_commands_return_value, side_effect=list_commands_side_effect
)
if run_command is None:
run_command = MagicMock()
type(pynutclient).run_command = run_command
return pynutclient
async def async_init_integration(
hass: HomeAssistant, ups_fixture: str
hass: HomeAssistant,
ups_fixture: str = None,
username: str = "mock",
password: str = "mock",
list_ups: dict[str, str] = None,
list_vars: dict[str, str] = None,
list_commands_return_value: dict[str, str] = None,
list_commands_side_effect=None,
run_command: MagicMock = None,
) -> MockConfigEntry:
"""Set up the nexia integration in Home Assistant."""
"""Set up the nut integration in Home Assistant."""
ups_fixture = f"nut/{ups_fixture}.json"
list_vars = json.loads(load_fixture(ups_fixture))
if list_ups is None:
list_ups = {"ups1": "UPS 1"}
mock_pynut = _get_mock_pynutclient(list_ups={"ups1": "UPS 1"}, list_vars=list_vars)
if ups_fixture is not None:
ups_fixture = f"nut/{ups_fixture}.json"
if list_vars is None:
list_vars = json.loads(load_fixture(ups_fixture))
mock_pynut = _get_mock_pynutclient(
list_ups=list_ups,
list_vars=list_vars,
list_commands_return_value=list_commands_return_value,
list_commands_side_effect=list_commands_side_effect,
run_command=run_command,
)
with patch(
"homeassistant.components.nut.PyNUTClient",
@ -33,7 +66,12 @@ async def async_init_integration(
):
entry = MockConfigEntry(
domain=DOMAIN,
data={CONF_HOST: "mock", CONF_PORT: "mock"},
data={
CONF_HOST: "mock",
CONF_PASSWORD: password,
CONF_PORT: "mock",
CONF_USERNAME: username,
},
)
entry.add_to_hass(hass)