jeans/core
jeans/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2025-07-28 07:37:34 +00:00
Code Issues Packages Projects Releases Wiki Activity

Add zwave_js.refresh_notifications service (#101370)

Browse Source
This commit is contained in:
Raman Gupta 2023-11-07 09:11:38 -05:00 committed by GitHub
parent 21af563dfe
commit 0fcaa2c581
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
6 changed files with 236 additions and 33 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
homeassistant/components/zwave_js/const.py
View File

@ -72,6 +72,8 @@ ATTR_STATUS = "status"
ATTR_ACKNOWLEDGED_FRAMES = "acknowledged_frames"
ATTR_EVENT_TYPE_LABEL = "event_type_label"
ATTR_DATA_TYPE_LABEL = "data_type_label"
ATTR_NOTIFICATION_TYPE = "notification_type"
ATTR_NOTIFICATION_EVENT = "notification_event"
ATTR_NODE = "node"
ATTR_ZWAVE_VALUE = "zwave_value"
@ -92,6 +94,7 @@ SERVICE_CLEAR_LOCK_USERCODE = "clear_lock_usercode"
SERVICE_INVOKE_CC_API = "invoke_cc_api"
SERVICE_MULTICAST_SET_VALUE = "multicast_set_value"
SERVICE_PING = "ping"
SERVICE_REFRESH_NOTIFICATIONS = "refresh_notifications"
SERVICE_REFRESH_VALUE = "refresh_value"
SERVICE_RESET_METER = "reset_meter"
SERVICE_SET_CONFIG_PARAMETER = "set_config_parameter"

124
homeassistant/components/zwave_js/services.py
View File

@ -4,11 +4,12 @@ from __future__ import annotations
import asyncio
from collections.abc import Generator, Sequence
import logging
from typing import Any
from typing import Any, TypeVar
import voluptuous as vol
from zwave_js_server.client import Client as ZwaveClient
from zwave_js_server.const import SET_VALUE_SUCCESS, CommandClass, CommandStatus
from zwave_js_server.const.command_class.notification import NotificationType
from zwave_js_server.exceptions import FailedZWaveCommand, SetValueFailed
from zwave_js_server.model.endpoint import Endpoint
from zwave_js_server.model.node import Node as ZwaveNode
@ -39,6 +40,8 @@ from .helpers import (
_LOGGER = logging.getLogger(__name__)
T = TypeVar("T", ZwaveNode, Endpoint)
def parameter_name_does_not_need_bitmask(
val: dict[str, int | str | list[str]]
@ -66,8 +69,8 @@ def broadcast_command(val: dict[str, Any]) -> dict[str, Any]:
def get_valid_responses_from_results(
zwave_objects: Sequence[ZwaveNode | Endpoint], results: Sequence[Any]
) -> Generator[tuple[ZwaveNode | Endpoint, Any], None, None]:
zwave_objects: Sequence[T], results: Sequence[Any]
) -> Generator[tuple[T, Any], None, None]:
"""Return valid responses from a list of results."""
for zwave_object, result in zip(zwave_objects, results):
if not isinstance(result, Exception):
@ -93,6 +96,49 @@ def raise_exceptions_from_results(
raise HomeAssistantError("\n".join(lines))
async def _async_invoke_cc_api(
nodes_or_endpoints: set[T],
command_class: CommandClass,
method_name: str,
*args: Any,
) -> None:
"""Invoke the CC API on a node endpoint."""
nodes_or_endpoints_list = list(nodes_or_endpoints)
results = await asyncio.gather(
*(
node_or_endpoint.async_invoke_cc_api(command_class, method_name, *args)
for node_or_endpoint in nodes_or_endpoints_list
),
return_exceptions=True,
)
for node_or_endpoint, result in get_valid_responses_from_results(
nodes_or_endpoints_list, results
):
if isinstance(node_or_endpoint, ZwaveNode):
_LOGGER.info(
(
"Invoked %s CC API method %s on node %s with the following result: "
"%s"
),
command_class.name,
method_name,
node_or_endpoint,
result,
)
else:
_LOGGER.info(
(
"Invoked %s CC API method %s on endpoint %s with the following "
"result: %s"
),
command_class.name,
method_name,
node_or_endpoint,
result,
)
raise_exceptions_from_results(nodes_or_endpoints_list, results)
class ZWaveServices:
"""Class that holds our services (Zwave Commands).
@ -406,6 +452,34 @@ class ZWaveServices:
),
)
self._hass.services.async_register(
const.DOMAIN,
const.SERVICE_REFRESH_NOTIFICATIONS,
self.async_refresh_notifications,
schema=vol.Schema(
vol.All(
{
vol.Optional(ATTR_AREA_ID): vol.All(
cv.ensure_list, [cv.string]
),
vol.Optional(ATTR_DEVICE_ID): vol.All(
cv.ensure_list, [cv.string]
),
vol.Optional(ATTR_ENTITY_ID): cv.entity_ids,
vol.Required(const.ATTR_NOTIFICATION_TYPE): vol.All(
vol.Coerce(int), vol.Coerce(NotificationType)
),
vol.Optional(const.ATTR_NOTIFICATION_EVENT): vol.Coerce(int),
},
cv.has_at_least_one_key(
ATTR_DEVICE_ID, ATTR_ENTITY_ID, ATTR_AREA_ID
),
get_nodes_from_service_data,
has_at_least_one_node,
),
),
)
async def async_set_config_parameter(self, service: ServiceCall) -> None:
"""Set a config value on a node."""
nodes: set[ZwaveNode] = service.data[const.ATTR_NODES]
@ -643,38 +717,14 @@ class ZWaveServices:
method_name: str = service.data[const.ATTR_METHOD_NAME]
parameters: list[Any] = service.data[const.ATTR_PARAMETERS]
async def _async_invoke_cc_api(endpoints: set[Endpoint]) -> None:
"""Invoke the CC API on a node endpoint."""
results = await asyncio.gather(
*(
endpoint.async_invoke_cc_api(
command_class, method_name, *parameters
)
for endpoint in endpoints
),
return_exceptions=True,
)
endpoints_list = list(endpoints)
for endpoint, result in get_valid_responses_from_results(
endpoints_list, results
):
_LOGGER.info(
(
"Invoked %s CC API method %s on endpoint %s with the following "
"result: %s"
),
command_class.name,
method_name,
endpoint,
result,
)
raise_exceptions_from_results(endpoints_list, results)
# If an endpoint is provided, we assume the user wants to call the CC API on
# that endpoint for all target nodes
if (endpoint := service.data.get(const.ATTR_ENDPOINT)) is not None:
await _async_invoke_cc_api(
{node.endpoints[endpoint] for node in service.data[const.ATTR_NODES]}
{node.endpoints[endpoint] for node in service.data[const.ATTR_NODES]},
command_class,
method_name,
*parameters,
)
return
@ -723,4 +773,14 @@ class ZWaveServices:
node.endpoints[endpoint_idx if endpoint_idx is not None else 0]
)
await _async_invoke_cc_api(endpoints)
await _async_invoke_cc_api(endpoints, command_class, method_name, *parameters)
async def async_refresh_notifications(self, service: ServiceCall) -> None:
"""Refresh notifications on a node."""
nodes: set[ZwaveNode] = service.data[const.ATTR_NODES]
notification_type: NotificationType = service.data[const.ATTR_NOTIFICATION_TYPE]
notification_event: int | None = service.data.get(const.ATTR_NOTIFICATION_EVENT)
param: dict[str, int] = {"notificationType": notification_type.value}
if notification_event is not None:
param["notificationEvent"] = notification_event
await _async_invoke_cc_api(nodes, CommandClass.NOTIFICATION, "get", param)

22
homeassistant/components/zwave_js/services.yaml
View File

@ -223,3 +223,25 @@ invoke_cc_api:
required: true
selector:
object:
refresh_notifications:
target:
entity:
integration: zwave_js
fields:
notification_type:
example: 1
required: true
selector:
number:
min: 1
max: 22
mode: box
notification_event:
example: 1
required: false
selector:
number:
min: 1
max: 255
mode: box

14
homeassistant/components/zwave_js/strings.json
View File

@ -363,6 +363,20 @@
"description": "A list of parameters to pass to the API method. Refer to the Z-Wave JS Command Class API documentation (https://zwave-js.github.io/node-zwave-js/#/api/CCs/index) for parameters."
}
}
},
"refresh_notifications": {
"name": "Refresh notifications on a node (advanced)",
"description": "Refreshes notifications on a node based on notification type and optionally notification event.",
"fields": {
"notification_type": {
"name": "Notification Type",
"description": "The Notification Type number as defined in the Z-Wave specs."
},
"notification_event": {
"name": "Notification Event",
"description": "The Notification Event number as defined in the Z-Wave specs."
}
}
}
}
}

9
tests/components/zwave_js/fixtures/multisensor_6_state.json
View File

@ -62,7 +62,14 @@
"index": 0,
"installerIcon": 3079,
"userIcon": 3079,
"commandClasses": []
"commandClasses": [
{
"id": 113,
"name": "Notification",
"version": 8,
"isSecure": false
}
]
}
],
"values": [

97
tests/components/zwave_js/test_services.py
View File

@ -14,6 +14,8 @@ from homeassistant.components.zwave_js.const import (
ATTR_CONFIG_VALUE,
ATTR_ENDPOINT,
ATTR_METHOD_NAME,
ATTR_NOTIFICATION_EVENT,
ATTR_NOTIFICATION_TYPE,
ATTR_OPTIONS,
ATTR_PARAMETERS,
ATTR_PROPERTY,
@ -26,6 +28,7 @@ from homeassistant.components.zwave_js.const import (
SERVICE_INVOKE_CC_API,
SERVICE_MULTICAST_SET_VALUE,
SERVICE_PING,
SERVICE_REFRESH_NOTIFICATIONS,
SERVICE_REFRESH_VALUE,
SERVICE_SET_CONFIG_PARAMETER,
SERVICE_SET_VALUE,
@ -1777,3 +1780,97 @@ async def test_invoke_cc_api(
client.async_send_command.reset_mock()
client.async_send_command_no_wait.reset_mock()
async def test_refresh_notifications(
hass: HomeAssistant, client, zen_31, multisensor_6, integration
) -> None:
"""Test refresh_notifications service."""
dev_reg = async_get_dev_reg(hass)
zen_31_device = dev_reg.async_get_device(
identifiers={get_device_id(client.driver, zen_31)}
)
assert zen_31_device
multisensor_6_device = dev_reg.async_get_device(
identifiers={get_device_id(client.driver, multisensor_6)}
)
assert multisensor_6_device
area_reg = async_get_area_reg(hass)
area = area_reg.async_get_or_create("test")
dev_reg.async_update_device(zen_31_device.id, area_id=area.id)
# Test successful refresh_notifications call
client.async_send_command.return_value = {"response": True}
client.async_send_command_no_wait.return_value = {"response": True}
await hass.services.async_call(
DOMAIN,
SERVICE_REFRESH_NOTIFICATIONS,
{
ATTR_AREA_ID: area.id,
ATTR_DEVICE_ID: [zen_31_device.id, multisensor_6_device.id],
ATTR_NOTIFICATION_TYPE: 1,
ATTR_NOTIFICATION_EVENT: 2,
},
blocking=True,
)
await hass.async_block_till_done()
assert len(client.async_send_command.call_args_list) == 1
args = client.async_send_command.call_args[0][0]
assert args["command"] == "endpoint.invoke_cc_api"
assert args["commandClass"] == 113
assert args["endpoint"] == 0
assert args["methodName"] == "get"
assert args["args"] == [{"notificationType": 1, "notificationEvent": 2}]
assert args["nodeId"] == zen_31.node_id
assert len(client.async_send_command_no_wait.call_args_list) == 1
args = client.async_send_command_no_wait.call_args[0][0]
assert args["command"] == "endpoint.invoke_cc_api"
assert args["commandClass"] == 113
assert args["endpoint"] == 0
assert args["methodName"] == "get"
assert args["args"] == [{"notificationType": 1, "notificationEvent": 2}]
assert args["nodeId"] == multisensor_6.node_id
client.async_send_command.reset_mock()
client.async_send_command_no_wait.reset_mock()
# Test failed refresh_notifications call on one node. We return the error on
# the first node in the call to make sure that gather works as expected
client.async_send_command.return_value = {"response": True}
client.async_send_command_no_wait.side_effect = FailedZWaveCommand(
"test", 12, "test"
)
with pytest.raises(HomeAssistantError):
await hass.services.async_call(
DOMAIN,
SERVICE_REFRESH_NOTIFICATIONS,
{
ATTR_DEVICE_ID: [multisensor_6_device.id, zen_31_device.id],
ATTR_NOTIFICATION_TYPE: 1,
},
blocking=True,
)
assert len(client.async_send_command.call_args_list) == 1
args = client.async_send_command.call_args[0][0]
assert args["command"] == "endpoint.invoke_cc_api"
assert args["commandClass"] == 113
assert args["endpoint"] == 0
assert args["methodName"] == "get"
assert args["args"] == [{"notificationType": 1}]
assert args["nodeId"] == zen_31.node_id
assert len(client.async_send_command_no_wait.call_args_list) == 1
args = client.async_send_command_no_wait.call_args[0][0]
assert args["command"] == "endpoint.invoke_cc_api"
assert args["commandClass"] == 113
assert args["endpoint"] == 0
assert args["methodName"] == "get"
assert args["args"] == [{"notificationType": 1}]
assert args["nodeId"] == multisensor_6.node_id
client.async_send_command.reset_mock()
client.async_send_command_no_wait.reset_mock()