starred/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2026-09-28 02:18:10 -04:00
Code Issues Packages Projects Releases Wiki Activity

Move service registration to async_setup in zha (#181186)

Browse Source
This commit is contained in:
epenet
2026-09-14 11:44:49 +02:00
committed by GitHub
parent c8035dc829
commit 152ca3f515
9 changed files with 604 additions and 440 deletions
Download Patch File Download Diff File Expand all files Collapse all files
homeassistant/components/zha/__init__.py
+3 -2
View File
@@ -61,6 +61,7 @@ from .repairs.wrong_silabs_firmware import (
AlreadyRunningEZSP,
warn_on_wrong_silabs_firmware,
)
from .services import async_setup_services
DEVICE_CONFIG_SCHEMA_ENTRY = probatio.Schema({probatio.Optional(CONF_TYPE): cv.string})
ZHA_CONFIG_SCHEMA = {
@@ -122,6 +123,8 @@ async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
async_register_firmware_info_provider(hass, DOMAIN, homeassistant_hardware)
async_setup_services(hass)
return True
@@ -299,8 +302,6 @@ async def async_unload_entry(hass: HomeAssistant, config_entry: ConfigEntry) ->
for platform in PLATFORMS:
del ha_zha_data.platforms[platform]
websocket_api.async_unload_api(hass)
return True
homeassistant/components/zha/const.py
+6
View File
@@ -5,6 +5,10 @@ EZSP_OVERWRITE_EUI64 = (
)
ATTR_ACTIVE_COORDINATOR = "active_coordinator"
ATTR_DURATION = "duration"
ATTR_INSTALL_CODE = "install_code"
ATTR_QR_CODE = "qr_code"
ATTR_SOURCE_IEEE = "source_ieee"
ATTR_ATTRIBUTES = "attributes"
ATTR_AVAILABLE = "available"
ATTR_DEVICE_TYPE = "device_type"
@@ -79,3 +83,5 @@ ZHA_OPTIONS = "zha_options"
# Dispatcher signal carrying device reconfigure progress events (bind result,
# attribute reporting result, configure complete) to the websocket subscriber.
SIGNAL_DEVICE_RECONFIGURE_EVENT = "zha_device_reconfigure_event"
RESPONSE = "response"
homeassistant/components/zha/device_action.py
+1 -1
View File
@@ -14,7 +14,7 @@ from homeassistant.helpers.typing import ConfigType, TemplateVarsType
from .const import DOMAIN
from .helpers import async_get_zha_device_proxy, convert_zha_error_to_ha_error
from .websocket_api import SERVICE_WARNING_DEVICE_SQUAWK, SERVICE_WARNING_DEVICE_WARN
from .services import SERVICE_WARNING_DEVICE_SQUAWK, SERVICE_WARNING_DEVICE_WARN
# mypy: disallow-any-generics
homeassistant/components/zha/helpers.py
+24 -1
View File
@@ -60,6 +60,8 @@ from zha.application.helpers import (
QuirksConfiguration,
ZHAConfiguration,
ZHAData,
convert_install_code,
qr_to_install_code,
)
from zha.application.platforms import GroupEntity, PlatformEntity
from zha.event import EventBase
@@ -118,27 +120,31 @@ from homeassistant.helpers import (
)
from homeassistant.helpers.dispatcher import async_dispatcher_send, dispatcher_send
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.typing import ConfigType
from homeassistant.helpers.typing import ConfigType, VolDictType
from homeassistant.util.logging import HomeAssistantQueueHandler
from .const import (
ATTR_ACTIVE_COORDINATOR,
ATTR_AVAILABLE,
ATTR_DEVICE_TYPE,
ATTR_DURATION,
ATTR_ENDPOINT_NAMES,
ATTR_EXPOSES_FEATURES,
ATTR_IEEE,
ATTR_INSTALL_CODE,
ATTR_LAST_SEEN,
ATTR_LQI,
ATTR_MANUFACTURER_CODE,
ATTR_NEIGHBORS,
ATTR_NWK,
ATTR_POWER_SOURCE,
ATTR_QR_CODE,
ATTR_QUIRK_APPLIED,
ATTR_QUIRK_CLASS,
ATTR_ROUTES,
ATTR_RSSI,
ATTR_SIGNATURE,
ATTR_SOURCE_IEEE,
CONF_ALARM_ARM_REQUIRES_CODE,
CONF_ALARM_FAILED_TRIES,
CONF_ALARM_MASTER_CODE,
@@ -1473,3 +1479,20 @@ def exclude_none_values(obj: Mapping[str, Any]) -> dict[str, Any]:
def get_config_entry_unique_id(network_info: NetworkInfo) -> str:
"""Generate a unique id for a config entry based on the network info."""
return f"epid={network_info.extended_pan_id}".lower()
IEEE_SCHEMA = probatio.All(cv.string, EUI64.convert)
SERVICE_PERMIT_PARAMS: VolDictType = {
probatio.Optional(ATTR_IEEE): IEEE_SCHEMA,
probatio.Optional(ATTR_DURATION, default=60): probatio.All(
probatio.Coerce(int), probatio.Range(0, 254)
),
probatio.Inclusive(ATTR_SOURCE_IEEE, "install_code"): IEEE_SCHEMA,
probatio.Inclusive(ATTR_INSTALL_CODE, "install_code"): probatio.All(
cv.string, convert_install_code
),
probatio.Exclusive(ATTR_QR_CODE, "install_code"): probatio.All(
cv.string, qr_to_install_code
),
}
homeassistant/components/zha/services.py
+452
View File
@@ -0,0 +1,452 @@
"""Support for Zigbee Home Automation services."""
import logging
from typing import Any, cast
import probatio
from zha.application.const import (
ATTR_ARGS,
ATTR_ATTRIBUTE,
ATTR_CLUSTER_ID,
ATTR_CLUSTER_TYPE,
ATTR_COMMAND_TYPE,
ATTR_ENDPOINT_ID,
ATTR_IEEE,
ATTR_LEVEL,
ATTR_MANUFACTURER,
ATTR_PARAMS,
ATTR_VALUE,
ATTR_WARNING_DEVICE_DURATION,
ATTR_WARNING_DEVICE_MODE,
ATTR_WARNING_DEVICE_STROBE,
ATTR_WARNING_DEVICE_STROBE_DUTY_CYCLE,
ATTR_WARNING_DEVICE_STROBE_INTENSITY,
CLUSTER_TYPE_IN,
)
from zha.application.gateway import Gateway
from zha.application.platforms.siren import (
BaseSiren,
SirenLevel,
SquawkMode,
Strobe,
StrobeLevel,
WarningMode,
)
from zigpy.types.named import EUI64, KeyData
from zigpy.typing import (
UNDEFINED as ZIGPY_UNDEFINED,
UndefinedType as ZigpyUndefinedType,
)
from homeassistant.const import ATTR_COMMAND, Platform
from homeassistant.core import HomeAssistant, ServiceCall, callback
from homeassistant.exceptions import ServiceValidationError
from homeassistant.helpers import config_validation as cv
from homeassistant.helpers.service import async_register_admin_service
from homeassistant.helpers.typing import VolSchemaType
from .const import (
ATTR_DURATION,
ATTR_INSTALL_CODE,
ATTR_QR_CODE,
ATTR_SOURCE_IEEE,
DOMAIN,
MFG_CLUSTER_ID_START,
RESPONSE,
)
from .helpers import IEEE_SCHEMA, SERVICE_PERMIT_PARAMS, get_zha_data
_LOGGER = logging.getLogger(__name__)
ATTR_GROUP = "group"
ATTR_IEEE_ADDRESS = "ieee_address"
SERVICE_PERMIT = "permit"
SERVICE_REMOVE = "remove"
SERVICE_SET_ZIGBEE_CLUSTER_ATTRIBUTE = "set_zigbee_cluster_attribute"
SERVICE_ISSUE_ZIGBEE_CLUSTER_COMMAND = "issue_zigbee_cluster_command"
SERVICE_ISSUE_ZIGBEE_GROUP_COMMAND = "issue_zigbee_group_command"
SERVICE_WARNING_DEVICE_SQUAWK = "warning_device_squawk"
SERVICE_WARNING_DEVICE_WARN = "warning_device_warn"
IEEE_SERVICE = "ieee_based_service"
def _ensure_list_if_present[_T](value: _T | None) -> list[_T] | list[Any] | None:
"""Wrap value in list if it is provided and not one."""
if value is None:
return None
return cast("list[_T]", value) if isinstance(value, list) else [value]
SERVICE_SCHEMAS: dict[str, VolSchemaType] = {
SERVICE_PERMIT: probatio.Schema(
probatio.All(
cv.deprecated(ATTR_IEEE_ADDRESS, replacement_key=ATTR_IEEE),
SERVICE_PERMIT_PARAMS,
)
),
IEEE_SERVICE: probatio.Schema(
probatio.All(
cv.deprecated(ATTR_IEEE_ADDRESS, replacement_key=ATTR_IEEE),
{probatio.Required(ATTR_IEEE): IEEE_SCHEMA},
)
),
SERVICE_SET_ZIGBEE_CLUSTER_ATTRIBUTE: probatio.Schema(
{
probatio.Required(ATTR_IEEE): IEEE_SCHEMA,
probatio.Required(ATTR_ENDPOINT_ID): cv.positive_int,
probatio.Required(ATTR_CLUSTER_ID): cv.positive_int,
probatio.Optional(ATTR_CLUSTER_TYPE, default=CLUSTER_TYPE_IN): cv.string,
probatio.Required(ATTR_ATTRIBUTE): probatio.Any(cv.positive_int, str),
probatio.Required(ATTR_VALUE): probatio.Any(int, cv.boolean, cv.string),
probatio.Optional(ATTR_MANUFACTURER): probatio.All(
probatio.Coerce(int), probatio.Range(min=-1)
),
}
),
SERVICE_WARNING_DEVICE_SQUAWK: probatio.Schema(
{
probatio.Required(ATTR_IEEE): IEEE_SCHEMA,
probatio.Optional(
ATTR_WARNING_DEVICE_MODE, default=SquawkMode.Armed
): cv.positive_int,
probatio.Optional(
ATTR_WARNING_DEVICE_STROBE, default=Strobe.Strobe
): cv.positive_int,
probatio.Optional(
ATTR_LEVEL, default=SirenLevel.High_level_sound
): cv.positive_int,
}
),
SERVICE_WARNING_DEVICE_WARN: probatio.Schema(
{
probatio.Required(ATTR_IEEE): IEEE_SCHEMA,
probatio.Optional(
ATTR_WARNING_DEVICE_MODE, default=WarningMode.Emergency
): cv.positive_int,
probatio.Optional(
ATTR_WARNING_DEVICE_STROBE, default=Strobe.Strobe
): cv.positive_int,
probatio.Optional(
ATTR_LEVEL, default=SirenLevel.High_level_sound
): cv.positive_int,
probatio.Optional(ATTR_WARNING_DEVICE_DURATION, default=5): cv.positive_int,
probatio.Optional(
ATTR_WARNING_DEVICE_STROBE_DUTY_CYCLE, default=0x00
): cv.positive_int,
probatio.Optional(
ATTR_WARNING_DEVICE_STROBE_INTENSITY,
default=StrobeLevel.High_level_strobe,
): cv.positive_int,
}
),
SERVICE_ISSUE_ZIGBEE_CLUSTER_COMMAND: probatio.All(
probatio.Schema(
{
probatio.Required(ATTR_IEEE): IEEE_SCHEMA,
probatio.Required(ATTR_ENDPOINT_ID): cv.positive_int,
probatio.Required(ATTR_CLUSTER_ID): cv.positive_int,
probatio.Optional(
ATTR_CLUSTER_TYPE, default=CLUSTER_TYPE_IN
): cv.string,
probatio.Required(ATTR_COMMAND): cv.positive_int,
probatio.Required(ATTR_COMMAND_TYPE): cv.string,
probatio.Exclusive(ATTR_ARGS, "attrs_params"): _ensure_list_if_present,
probatio.Exclusive(ATTR_PARAMS, "attrs_params"): dict,
probatio.Optional(ATTR_MANUFACTURER): probatio.All(
probatio.Coerce(int), probatio.Range(min=-1)
),
}
),
cv.deprecated(ATTR_ARGS),
cv.has_at_least_one_key(ATTR_ARGS, ATTR_PARAMS),
),
SERVICE_ISSUE_ZIGBEE_GROUP_COMMAND: probatio.Schema(
{
probatio.Required(ATTR_GROUP): cv.positive_int,
probatio.Required(ATTR_CLUSTER_ID): cv.positive_int,
probatio.Optional(ATTR_CLUSTER_TYPE, default=CLUSTER_TYPE_IN): cv.string,
probatio.Required(ATTR_COMMAND): cv.positive_int,
probatio.Optional(ATTR_ARGS, default=[]): cv.ensure_list,
probatio.Optional(ATTR_MANUFACTURER): probatio.All(
probatio.Coerce(int), probatio.Range(min=-1)
),
}
),
}
def _get_gateway(service: ServiceCall) -> Gateway:
"""Return the ZHA gateway, raising a user-facing error if it is unavailable."""
if (gateway_proxy := get_zha_data(service.hass).gateway_proxy) is None:
raise ServiceValidationError(
translation_domain=DOMAIN,
translation_key="no_gateway",
)
return gateway_proxy.gateway
async def _permit(service: ServiceCall) -> None:
"""Allow devices to join this network."""
application_controller = _get_gateway(service).application_controller
duration: int = service.data[ATTR_DURATION]
ieee: EUI64 | None = service.data.get(ATTR_IEEE)
src_ieee: EUI64
link_key: KeyData
if ATTR_SOURCE_IEEE in service.data:
src_ieee = service.data[ATTR_SOURCE_IEEE]
link_key = service.data[ATTR_INSTALL_CODE]
_LOGGER.info("Allowing join for %s device with link key", src_ieee)
await application_controller.permit_with_link_key(
time_s=duration, node=src_ieee, link_key=link_key
)
return
if ATTR_QR_CODE in service.data:
src_ieee, link_key = service.data[ATTR_QR_CODE]
_LOGGER.info("Allowing join for %s device with link key", src_ieee)
await application_controller.permit_with_link_key(
time_s=duration, node=src_ieee, link_key=link_key
)
return
if ieee:
_LOGGER.info("Permitting joins for %ss on %s device", duration, ieee)
else:
_LOGGER.info("Permitting joins for %ss", duration)
await application_controller.permit(time_s=duration, node=ieee)
async def _remove(service: ServiceCall) -> None:
"""Remove a node from the network."""
zha_gateway = _get_gateway(service)
ieee: EUI64 = service.data[ATTR_IEEE]
_LOGGER.info("Removing node %s", ieee)
await zha_gateway.async_remove_device(ieee)
async def _set_zigbee_cluster_attributes(service: ServiceCall) -> None:
"""Set zigbee attribute for cluster on zha entity."""
zha_gateway = _get_gateway(service)
ieee: EUI64 = service.data[ATTR_IEEE]
endpoint_id: int = service.data[ATTR_ENDPOINT_ID]
cluster_id: int = service.data[ATTR_CLUSTER_ID]
cluster_type: str = service.data[ATTR_CLUSTER_TYPE]
attribute: int | str = service.data[ATTR_ATTRIBUTE]
value: int | bool | str = service.data[ATTR_VALUE]
manufacturer: int | ZigpyUndefinedType = service.data.get(
ATTR_MANUFACTURER, ZIGPY_UNDEFINED
)
zha_device = zha_gateway.get_device(ieee)
response = None
if zha_device is not None:
response = await zha_device.write_zigbee_attribute(
endpoint_id,
cluster_id,
attribute,
value,
cluster_type=cluster_type,
manufacturer=manufacturer,
)
else:
raise ValueError(f"Device with IEEE {ieee!s} not found")
_LOGGER.debug(
(
"Set attribute for: %s: [%s] %s: [%s] %s: [%s] %s: [%s] %s: [%s] %s:"
" [%s] %s: [%s]"
),
ATTR_CLUSTER_ID,
cluster_id,
ATTR_CLUSTER_TYPE,
cluster_type,
ATTR_ENDPOINT_ID,
endpoint_id,
ATTR_ATTRIBUTE,
attribute,
ATTR_VALUE,
value,
ATTR_MANUFACTURER,
manufacturer,
RESPONSE,
response,
)
async def _issue_zigbee_cluster_command(service: ServiceCall) -> None:
"""Issue command on zigbee cluster on ZHA entity."""
zha_gateway = _get_gateway(service)
ieee: EUI64 = service.data[ATTR_IEEE]
endpoint_id: int = service.data[ATTR_ENDPOINT_ID]
cluster_id: int = service.data[ATTR_CLUSTER_ID]
cluster_type: str = service.data[ATTR_CLUSTER_TYPE]
command: int = service.data[ATTR_COMMAND]
command_type: str = service.data[ATTR_COMMAND_TYPE]
args: list | None = service.data.get(ATTR_ARGS)
params: dict | None = service.data.get(ATTR_PARAMS)
manufacturer: int | ZigpyUndefinedType = service.data.get(
ATTR_MANUFACTURER, ZIGPY_UNDEFINED
)
zha_device = zha_gateway.get_device(ieee)
if zha_device is not None:
if cluster_id >= MFG_CLUSTER_ID_START and manufacturer is None:
manufacturer = zha_device.manufacturer_code
await zha_device.issue_cluster_command(
endpoint_id,
cluster_id,
command,
command_type,
args,
params,
cluster_type=cluster_type,
manufacturer=manufacturer,
)
_LOGGER.debug(
(
"Issued command for: %s: [%s] %s: [%s] %s: [%s] %s: [%s] %s: [%s]"
" %s: [%s] %s: [%s] %s: [%s]"
),
ATTR_CLUSTER_ID,
cluster_id,
ATTR_CLUSTER_TYPE,
cluster_type,
ATTR_ENDPOINT_ID,
endpoint_id,
ATTR_COMMAND,
command,
ATTR_COMMAND_TYPE,
command_type,
ATTR_ARGS,
args,
ATTR_PARAMS,
params,
ATTR_MANUFACTURER,
manufacturer,
)
else:
raise ValueError(f"Device with IEEE {ieee!s} not found")
async def _issue_zigbee_group_command(service: ServiceCall) -> None:
"""Issue command on zigbee cluster on a zigbee group."""
zha_gateway = _get_gateway(service)
group_id: int = service.data[ATTR_GROUP]
cluster_id: int = service.data[ATTR_CLUSTER_ID]
command: int = service.data[ATTR_COMMAND]
args: list = service.data[ATTR_ARGS]
manufacturer: int | ZigpyUndefinedType = service.data.get(
ATTR_MANUFACTURER, ZIGPY_UNDEFINED
)
group = zha_gateway.get_group(group_id)
if cluster_id >= MFG_CLUSTER_ID_START and manufacturer is None:
_LOGGER.error("Missing manufacturer attribute for cluster: %d", cluster_id)
response = None
if group is not None:
cluster = group.endpoint[cluster_id]
response = await cluster.command(
command, *args, manufacturer=manufacturer, expect_reply=True
)
_LOGGER.debug(
"Issued group command for: %s: [%s] %s: [%s] %s: %s %s: [%s] %s: %s",
ATTR_CLUSTER_ID,
cluster_id,
ATTR_COMMAND,
command,
ATTR_ARGS,
args,
ATTR_MANUFACTURER,
manufacturer,
RESPONSE,
response,
)
async def _warning_device_squawk(service: ServiceCall) -> None:
"""Issue the squawk command for an IAS warning device."""
zha_gateway = _get_gateway(service)
ieee: EUI64 = service.data[ATTR_IEEE]
mode: int = service.data[ATTR_WARNING_DEVICE_MODE]
strobe: int = service.data[ATTR_WARNING_DEVICE_STROBE]
level: int = service.data[ATTR_LEVEL]
device = zha_gateway.get_device(ieee)
siren: BaseSiren = device.get_entity(Platform.SIREN, pick_first=True)
await siren.async_squawk(mode=mode, strobe=strobe, squawk_level=level)
async def _warning_device_warn(service: ServiceCall) -> None:
"""Issue the warning command for an IAS warning device."""
zha_gateway = _get_gateway(service)
ieee: EUI64 = service.data[ATTR_IEEE]
mode: int = service.data[ATTR_WARNING_DEVICE_MODE]
strobe: int = service.data[ATTR_WARNING_DEVICE_STROBE]
level: int = service.data[ATTR_LEVEL]
duration: int = service.data[ATTR_WARNING_DEVICE_DURATION]
duty_mode: int = service.data[ATTR_WARNING_DEVICE_STROBE_DUTY_CYCLE]
intensity: int = service.data[ATTR_WARNING_DEVICE_STROBE_INTENSITY]
device = zha_gateway.get_device(ieee)
siren: BaseSiren = device.get_entity(Platform.SIREN, pick_first=True)
await siren.async_turn_on(
tone=mode,
volume_level=level,
duration=duration,
strobe=strobe,
strobe_duty_cycle=duty_mode,
strobe_intensity=intensity,
)
@callback
def async_setup_services(hass: HomeAssistant) -> None:
"""Register the ZHA services."""
async_register_admin_service(
hass,
DOMAIN,
SERVICE_PERMIT,
_permit,
schema=SERVICE_SCHEMAS[SERVICE_PERMIT],
)
async_register_admin_service(
hass, DOMAIN, SERVICE_REMOVE, _remove, schema=SERVICE_SCHEMAS[IEEE_SERVICE]
)
async_register_admin_service(
hass,
DOMAIN,
SERVICE_SET_ZIGBEE_CLUSTER_ATTRIBUTE,
_set_zigbee_cluster_attributes,
schema=SERVICE_SCHEMAS[SERVICE_SET_ZIGBEE_CLUSTER_ATTRIBUTE],
)
async_register_admin_service(
hass,
DOMAIN,
SERVICE_ISSUE_ZIGBEE_CLUSTER_COMMAND,
_issue_zigbee_cluster_command,
schema=SERVICE_SCHEMAS[SERVICE_ISSUE_ZIGBEE_CLUSTER_COMMAND],
)
async_register_admin_service(
hass,
DOMAIN,
SERVICE_ISSUE_ZIGBEE_GROUP_COMMAND,
_issue_zigbee_group_command,
schema=SERVICE_SCHEMAS[SERVICE_ISSUE_ZIGBEE_GROUP_COMMAND],
)
async_register_admin_service(
hass,
DOMAIN,
SERVICE_WARNING_DEVICE_SQUAWK,
_warning_device_squawk,
schema=SERVICE_SCHEMAS[SERVICE_WARNING_DEVICE_SQUAWK],
)
async_register_admin_service(
hass,
DOMAIN,
SERVICE_WARNING_DEVICE_WARN,
_warning_device_warn,
schema=SERVICE_SCHEMAS[SERVICE_WARNING_DEVICE_WARN],
)
homeassistant/components/zha/strings.json
+5
View File
@@ -2291,6 +2291,11 @@
}
}
},
"exceptions": {
"no_gateway": {
"message": "ZHA is not loaded, so this action cannot be performed."
}
},
"issues": {
"inconsistent_network_settings": {
"fix_flow": {
homeassistant/components/zha/websocket_api.py
+10 -430
View File
@@ -6,24 +6,14 @@ from typing import TYPE_CHECKING, Any, Literal, NamedTuple, cast
import probatio
from zha.application.const import (
ATTR_ARGS,
ATTR_ATTRIBUTE,
ATTR_CLUSTER_ID,
ATTR_CLUSTER_TYPE,
ATTR_COMMAND_TYPE,
ATTR_ENDPOINT_ID,
ATTR_IEEE,
ATTR_LEVEL,
ATTR_MANUFACTURER,
ATTR_MEMBERS,
ATTR_PARAMS,
ATTR_TYPE,
ATTR_VALUE,
ATTR_WARNING_DEVICE_DURATION,
ATTR_WARNING_DEVICE_MODE,
ATTR_WARNING_DEVICE_STROBE,
ATTR_WARNING_DEVICE_STROBE_DUTY_CYCLE,
ATTR_WARNING_DEVICE_STROBE_INTENSITY,
CLUSTER_COMMAND_SERVER,
CLUSTER_COMMANDS_CLIENT,
CLUSTER_COMMANDS_SERVER,
@@ -32,20 +22,7 @@ from zha.application.const import (
ZHA_GW_MSG,
)
from zha.application.gateway import Gateway
from zha.application.helpers import (
async_is_bindable_target,
convert_install_code,
get_matched_clusters,
qr_to_install_code,
)
from zha.application.platforms.siren import (
BaseSiren,
SirenLevel,
SquawkMode,
Strobe,
StrobeLevel,
WarningMode,
)
from zha.application.helpers import async_is_bindable_target, get_matched_clusters
from zha.zigbee.group import GroupMemberReference
import zigpy.backups
from zigpy.config import CONF_DEVICE
@@ -60,12 +37,10 @@ import zigpy.zdo.types as zdo_types
from homeassistant.components import websocket_api
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import ATTR_COMMAND, ATTR_ID, ATTR_NAME, Platform
from homeassistant.core import HomeAssistant, ServiceCall, callback
from homeassistant.const import ATTR_ID, ATTR_NAME
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers import config_validation as cv, entity_registry as er
from homeassistant.helpers.dispatcher import async_dispatcher_connect
from homeassistant.helpers.service import async_register_admin_service
from homeassistant.helpers.typing import VolDictType, VolSchemaType
from .api import (
async_change_channel,
@@ -73,13 +48,16 @@ from .api import (
async_get_radio_type,
)
from .const import (
ATTR_DURATION,
ATTR_INSTALL_CODE,
ATTR_QR_CODE,
ATTR_SOURCE_IEEE,
CUSTOM_CONFIGURATION,
DOMAIN,
EZSP_OVERWRITE_EUI64,
GROUP_ID,
GROUP_IDS,
GROUP_NAME,
MFG_CLUSTER_ID_START,
RESPONSE,
SIGNAL_DEVICE_RECONFIGURE_EVENT,
ZHA_ALARM_OPTIONS,
ZHA_OPTIONS,
@@ -87,6 +65,8 @@ from .const import (
from .helpers import (
CONF_ZHA_ALARM_SCHEMA,
CONF_ZHA_OPTIONS_SCHEMA,
IEEE_SCHEMA,
SERVICE_PERMIT_PARAMS,
EntityReference,
ZHAGatewayProxy,
async_cluster_exists,
@@ -104,152 +84,16 @@ _LOGGER = logging.getLogger(__name__)
TYPE = "type"
CLIENT = "client"
ID = "id"
RESPONSE = "response"
DEVICE_INFO = "device_info"
ATTR_DURATION = "duration"
ATTR_GROUP = "group"
ATTR_IEEE_ADDRESS = "ieee_address"
ATTR_INSTALL_CODE = "install_code"
ATTR_NEW_CHANNEL = "new_channel"
ATTR_SOURCE_IEEE = "source_ieee"
ATTR_TARGET_IEEE = "target_ieee"
ATTR_QR_CODE = "qr_code"
BINDINGS = "bindings"
SERVICE_PERMIT = "permit"
SERVICE_REMOVE = "remove"
SERVICE_SET_ZIGBEE_CLUSTER_ATTRIBUTE = "set_zigbee_cluster_attribute"
SERVICE_ISSUE_ZIGBEE_CLUSTER_COMMAND = "issue_zigbee_cluster_command"
SERVICE_ISSUE_ZIGBEE_GROUP_COMMAND = "issue_zigbee_group_command"
SERVICE_DIRECT_ZIGBEE_BIND = "issue_direct_zigbee_bind"
SERVICE_DIRECT_ZIGBEE_UNBIND = "issue_direct_zigbee_unbind"
SERVICE_WARNING_DEVICE_SQUAWK = "warning_device_squawk"
SERVICE_WARNING_DEVICE_WARN = "warning_device_warn"
SERVICE_ZIGBEE_BIND = "service_zigbee_bind"
IEEE_SERVICE = "ieee_based_service"
IEEE_SCHEMA = probatio.All(cv.string, EUI64.convert)
def _ensure_list_if_present[_T](value: _T | None) -> list[_T] | list[Any] | None:
"""Wrap value in list if it is provided and not one."""
if value is None:
return None
return cast("list[_T]", value) if isinstance(value, list) else [value]
SERVICE_PERMIT_PARAMS: VolDictType = {
probatio.Optional(ATTR_IEEE): IEEE_SCHEMA,
probatio.Optional(ATTR_DURATION, default=60): probatio.All(
probatio.Coerce(int), probatio.Range(0, 254)
),
probatio.Inclusive(ATTR_SOURCE_IEEE, "install_code"): IEEE_SCHEMA,
probatio.Inclusive(ATTR_INSTALL_CODE, "install_code"): probatio.All(
cv.string, convert_install_code
),
probatio.Exclusive(ATTR_QR_CODE, "install_code"): probatio.All(
cv.string, qr_to_install_code
),
}
SERVICE_SCHEMAS: dict[str, VolSchemaType] = {
SERVICE_PERMIT: probatio.Schema(
probatio.All(
cv.deprecated(ATTR_IEEE_ADDRESS, replacement_key=ATTR_IEEE),
SERVICE_PERMIT_PARAMS,
)
),
IEEE_SERVICE: probatio.Schema(
probatio.All(
cv.deprecated(ATTR_IEEE_ADDRESS, replacement_key=ATTR_IEEE),
{probatio.Required(ATTR_IEEE): IEEE_SCHEMA},
)
),
SERVICE_SET_ZIGBEE_CLUSTER_ATTRIBUTE: probatio.Schema(
{
probatio.Required(ATTR_IEEE): IEEE_SCHEMA,
probatio.Required(ATTR_ENDPOINT_ID): cv.positive_int,
probatio.Required(ATTR_CLUSTER_ID): cv.positive_int,
probatio.Optional(ATTR_CLUSTER_TYPE, default=CLUSTER_TYPE_IN): cv.string,
probatio.Required(ATTR_ATTRIBUTE): probatio.Any(cv.positive_int, str),
probatio.Required(ATTR_VALUE): probatio.Any(int, cv.boolean, cv.string),
probatio.Optional(ATTR_MANUFACTURER): probatio.All(
probatio.Coerce(int), probatio.Range(min=-1)
),
}
),
SERVICE_WARNING_DEVICE_SQUAWK: probatio.Schema(
{
probatio.Required(ATTR_IEEE): IEEE_SCHEMA,
probatio.Optional(
ATTR_WARNING_DEVICE_MODE, default=SquawkMode.Armed
): cv.positive_int,
probatio.Optional(
ATTR_WARNING_DEVICE_STROBE, default=Strobe.Strobe
): cv.positive_int,
probatio.Optional(
ATTR_LEVEL, default=SirenLevel.High_level_sound
): cv.positive_int,
}
),
SERVICE_WARNING_DEVICE_WARN: probatio.Schema(
{
probatio.Required(ATTR_IEEE): IEEE_SCHEMA,
probatio.Optional(
ATTR_WARNING_DEVICE_MODE, default=WarningMode.Emergency
): cv.positive_int,
probatio.Optional(
ATTR_WARNING_DEVICE_STROBE, default=Strobe.Strobe
): cv.positive_int,
probatio.Optional(
ATTR_LEVEL, default=SirenLevel.High_level_sound
): cv.positive_int,
probatio.Optional(ATTR_WARNING_DEVICE_DURATION, default=5): cv.positive_int,
probatio.Optional(
ATTR_WARNING_DEVICE_STROBE_DUTY_CYCLE, default=0x00
): cv.positive_int,
probatio.Optional(
ATTR_WARNING_DEVICE_STROBE_INTENSITY,
default=StrobeLevel.High_level_strobe,
): cv.positive_int,
}
),
SERVICE_ISSUE_ZIGBEE_CLUSTER_COMMAND: probatio.All(
probatio.Schema(
{
probatio.Required(ATTR_IEEE): IEEE_SCHEMA,
probatio.Required(ATTR_ENDPOINT_ID): cv.positive_int,
probatio.Required(ATTR_CLUSTER_ID): cv.positive_int,
probatio.Optional(
ATTR_CLUSTER_TYPE, default=CLUSTER_TYPE_IN
): cv.string,
probatio.Required(ATTR_COMMAND): cv.positive_int,
probatio.Required(ATTR_COMMAND_TYPE): cv.string,
probatio.Exclusive(ATTR_ARGS, "attrs_params"): _ensure_list_if_present,
probatio.Exclusive(ATTR_PARAMS, "attrs_params"): dict,
probatio.Optional(ATTR_MANUFACTURER): probatio.All(
probatio.Coerce(int), probatio.Range(min=-1)
),
}
),
cv.deprecated(ATTR_ARGS),
cv.has_at_least_one_key(ATTR_ARGS, ATTR_PARAMS),
),
SERVICE_ISSUE_ZIGBEE_GROUP_COMMAND: probatio.Schema(
{
probatio.Required(ATTR_GROUP): cv.positive_int,
probatio.Required(ATTR_CLUSTER_ID): cv.positive_int,
probatio.Optional(ATTR_CLUSTER_TYPE, default=CLUSTER_TYPE_IN): cv.string,
probatio.Required(ATTR_COMMAND): cv.positive_int,
probatio.Optional(ATTR_ARGS, default=[]): cv.ensure_list,
probatio.Optional(ATTR_MANUFACTURER): probatio.All(
probatio.Coerce(int), probatio.Range(min=-1)
),
}
),
}
ZHA_CONFIG_SCHEMAS = {
@@ -1294,258 +1138,6 @@ async def websocket_change_channel(
@callback
def async_load_api(hass: HomeAssistant) -> None:
"""Set up the web socket API."""
zha_gateway = get_zha_gateway(hass)
application_controller = zha_gateway.application_controller
async def permit(service: ServiceCall) -> None:
"""Allow devices to join this network."""
duration: int = service.data[ATTR_DURATION]
ieee: EUI64 | None = service.data.get(ATTR_IEEE)
src_ieee: EUI64
link_key: KeyData
if ATTR_SOURCE_IEEE in service.data:
src_ieee = service.data[ATTR_SOURCE_IEEE]
link_key = service.data[ATTR_INSTALL_CODE]
_LOGGER.info("Allowing join for %s device with link key", src_ieee)
await application_controller.permit_with_link_key(
time_s=duration, node=src_ieee, link_key=link_key
)
return
if ATTR_QR_CODE in service.data:
src_ieee, link_key = service.data[ATTR_QR_CODE]
_LOGGER.info("Allowing join for %s device with link key", src_ieee)
await application_controller.permit_with_link_key(
time_s=duration, node=src_ieee, link_key=link_key
)
return
if ieee:
_LOGGER.info("Permitting joins for %ss on %s device", duration, ieee)
else:
_LOGGER.info("Permitting joins for %ss", duration)
await application_controller.permit(time_s=duration, node=ieee)
async_register_admin_service(
hass, DOMAIN, SERVICE_PERMIT, permit, schema=SERVICE_SCHEMAS[SERVICE_PERMIT]
)
async def remove(service: ServiceCall) -> None:
"""Remove a node from the network."""
zha_gateway = get_zha_gateway(hass)
ieee: EUI64 = service.data[ATTR_IEEE]
_LOGGER.info("Removing node %s", ieee)
await zha_gateway.async_remove_device(ieee)
async_register_admin_service(
hass, DOMAIN, SERVICE_REMOVE, remove, schema=SERVICE_SCHEMAS[IEEE_SERVICE]
)
async def set_zigbee_cluster_attributes(service: ServiceCall) -> None:
"""Set zigbee attribute for cluster on zha entity."""
ieee: EUI64 = service.data[ATTR_IEEE]
endpoint_id: int = service.data[ATTR_ENDPOINT_ID]
cluster_id: int = service.data[ATTR_CLUSTER_ID]
cluster_type: str = service.data[ATTR_CLUSTER_TYPE]
attribute: int | str = service.data[ATTR_ATTRIBUTE]
value: int | bool | str = service.data[ATTR_VALUE]
manufacturer: int | ZigpyUndefinedType = service.data.get(
ATTR_MANUFACTURER, ZIGPY_UNDEFINED
)
zha_device = zha_gateway.get_device(ieee)
response = None
if zha_device is not None:
response = await zha_device.write_zigbee_attribute(
endpoint_id,
cluster_id,
attribute,
value,
cluster_type=cluster_type,
manufacturer=manufacturer,
)
else:
raise ValueError(f"Device with IEEE {ieee!s} not found")
_LOGGER.debug(
(
"Set attribute for: %s: [%s] %s: [%s] %s: [%s] %s: [%s] %s: [%s] %s:"
" [%s] %s: [%s]"
),
ATTR_CLUSTER_ID,
cluster_id,
ATTR_CLUSTER_TYPE,
cluster_type,
ATTR_ENDPOINT_ID,
endpoint_id,
ATTR_ATTRIBUTE,
attribute,
ATTR_VALUE,
value,
ATTR_MANUFACTURER,
manufacturer,
RESPONSE,
response,
)
async_register_admin_service(
hass,
DOMAIN,
SERVICE_SET_ZIGBEE_CLUSTER_ATTRIBUTE,
set_zigbee_cluster_attributes,
schema=SERVICE_SCHEMAS[SERVICE_SET_ZIGBEE_CLUSTER_ATTRIBUTE],
)
async def issue_zigbee_cluster_command(service: ServiceCall) -> None:
"""Issue command on zigbee cluster on ZHA entity."""
ieee: EUI64 = service.data[ATTR_IEEE]
endpoint_id: int = service.data[ATTR_ENDPOINT_ID]
cluster_id: int = service.data[ATTR_CLUSTER_ID]
cluster_type: str = service.data[ATTR_CLUSTER_TYPE]
command: int = service.data[ATTR_COMMAND]
command_type: str = service.data[ATTR_COMMAND_TYPE]
args: list | None = service.data.get(ATTR_ARGS)
params: dict | None = service.data.get(ATTR_PARAMS)
manufacturer: int | ZigpyUndefinedType = service.data.get(
ATTR_MANUFACTURER, ZIGPY_UNDEFINED
)
zha_device = zha_gateway.get_device(ieee)
if zha_device is not None:
if cluster_id >= MFG_CLUSTER_ID_START and manufacturer is None:
manufacturer = zha_device.manufacturer_code
await zha_device.issue_cluster_command(
endpoint_id,
cluster_id,
command,
command_type,
args,
params,
cluster_type=cluster_type,
manufacturer=manufacturer,
)
_LOGGER.debug(
(
"Issued command for: %s: [%s] %s: [%s] %s: [%s] %s: [%s] %s: [%s]"
" %s: [%s] %s: [%s] %s: [%s]"
),
ATTR_CLUSTER_ID,
cluster_id,
ATTR_CLUSTER_TYPE,
cluster_type,
ATTR_ENDPOINT_ID,
endpoint_id,
ATTR_COMMAND,
command,
ATTR_COMMAND_TYPE,
command_type,
ATTR_ARGS,
args,
ATTR_PARAMS,
params,
ATTR_MANUFACTURER,
manufacturer,
)
else:
raise ValueError(f"Device with IEEE {ieee!s} not found")
async_register_admin_service(
hass,
DOMAIN,
SERVICE_ISSUE_ZIGBEE_CLUSTER_COMMAND,
issue_zigbee_cluster_command,
schema=SERVICE_SCHEMAS[SERVICE_ISSUE_ZIGBEE_CLUSTER_COMMAND],
)
async def issue_zigbee_group_command(service: ServiceCall) -> None:
"""Issue command on zigbee cluster on a zigbee group."""
group_id: int = service.data[ATTR_GROUP]
cluster_id: int = service.data[ATTR_CLUSTER_ID]
command: int = service.data[ATTR_COMMAND]
args: list = service.data[ATTR_ARGS]
manufacturer: int | ZigpyUndefinedType = service.data.get(
ATTR_MANUFACTURER, ZIGPY_UNDEFINED
)
group = zha_gateway.get_group(group_id)
if cluster_id >= MFG_CLUSTER_ID_START and manufacturer is None:
_LOGGER.error("Missing manufacturer attribute for cluster: %d", cluster_id)
response = None
if group is not None:
cluster = group.endpoint[cluster_id]
response = await cluster.command(
command, *args, manufacturer=manufacturer, expect_reply=True
)
_LOGGER.debug(
"Issued group command for: %s: [%s] %s: [%s] %s: %s %s: [%s] %s: %s",
ATTR_CLUSTER_ID,
cluster_id,
ATTR_COMMAND,
command,
ATTR_ARGS,
args,
ATTR_MANUFACTURER,
manufacturer,
RESPONSE,
response,
)
async_register_admin_service(
hass,
DOMAIN,
SERVICE_ISSUE_ZIGBEE_GROUP_COMMAND,
issue_zigbee_group_command,
schema=SERVICE_SCHEMAS[SERVICE_ISSUE_ZIGBEE_GROUP_COMMAND],
)
async def warning_device_squawk(service: ServiceCall) -> None:
"""Issue the squawk command for an IAS warning device."""
ieee: EUI64 = service.data[ATTR_IEEE]
mode: int = service.data[ATTR_WARNING_DEVICE_MODE]
strobe: int = service.data[ATTR_WARNING_DEVICE_STROBE]
level: int = service.data[ATTR_LEVEL]
device = zha_gateway.get_device(ieee)
siren: BaseSiren = device.get_entity(Platform.SIREN, pick_first=True)
await siren.async_squawk(mode=mode, strobe=strobe, squawk_level=level)
async_register_admin_service(
hass,
DOMAIN,
SERVICE_WARNING_DEVICE_SQUAWK,
warning_device_squawk,
schema=SERVICE_SCHEMAS[SERVICE_WARNING_DEVICE_SQUAWK],
)
async def warning_device_warn(service: ServiceCall) -> None:
"""Issue the warning command for an IAS warning device."""
ieee: EUI64 = service.data[ATTR_IEEE]
mode: int = service.data[ATTR_WARNING_DEVICE_MODE]
strobe: int = service.data[ATTR_WARNING_DEVICE_STROBE]
level: int = service.data[ATTR_LEVEL]
duration: int = service.data[ATTR_WARNING_DEVICE_DURATION]
duty_mode: int = service.data[ATTR_WARNING_DEVICE_STROBE_DUTY_CYCLE]
intensity: int = service.data[ATTR_WARNING_DEVICE_STROBE_INTENSITY]
device = zha_gateway.get_device(ieee)
siren: BaseSiren = device.get_entity(Platform.SIREN, pick_first=True)
await siren.async_turn_on(
tone=mode,
volume_level=level,
duration=duration,
strobe=strobe,
strobe_duty_cycle=duty_mode,
strobe_intensity=intensity,
)
async_register_admin_service(
hass,
DOMAIN,
SERVICE_WARNING_DEVICE_WARN,
warning_device_warn,
schema=SERVICE_SCHEMAS[SERVICE_WARNING_DEVICE_WARN],
)
websocket_api.async_register_command(hass, websocket_permit_devices)
websocket_api.async_register_command(hass, websocket_get_devices)
websocket_api.async_register_command(hass, websocket_get_groupable_devices)
@@ -1574,15 +1166,3 @@ def async_load_api(hass: HomeAssistant) -> None:
websocket_api.async_register_command(hass, websocket_create_network_backup)
websocket_api.async_register_command(hass, websocket_restore_network_backup)
websocket_api.async_register_command(hass, websocket_change_channel)
@callback
def async_unload_api(hass: HomeAssistant) -> None:
"""Unload the ZHA API."""
hass.services.async_remove(DOMAIN, SERVICE_PERMIT)
hass.services.async_remove(DOMAIN, SERVICE_REMOVE)
hass.services.async_remove(DOMAIN, SERVICE_SET_ZIGBEE_CLUSTER_ATTRIBUTE)
hass.services.async_remove(DOMAIN, SERVICE_ISSUE_ZIGBEE_CLUSTER_COMMAND)
hass.services.async_remove(DOMAIN, SERVICE_ISSUE_ZIGBEE_GROUP_COMMAND)
hass.services.async_remove(DOMAIN, SERVICE_WARNING_DEVICE_SQUAWK)
hass.services.async_remove(DOMAIN, SERVICE_WARNING_DEVICE_WARN)
tests/components/zha/test_services.py
+95
View File
@@ -0,0 +1,95 @@
"""Tests for the ZHA services."""
import pytest
from homeassistant.components.zha.const import DOMAIN
from homeassistant.components.zha.services import (
SERVICE_ISSUE_ZIGBEE_CLUSTER_COMMAND,
SERVICE_ISSUE_ZIGBEE_GROUP_COMMAND,
SERVICE_PERMIT,
SERVICE_REMOVE,
SERVICE_SET_ZIGBEE_CLUSTER_ATTRIBUTE,
SERVICE_WARNING_DEVICE_SQUAWK,
SERVICE_WARNING_DEVICE_WARN,
)
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import ServiceValidationError
from homeassistant.setup import async_setup_component
IEEE_SWITCH_DEVICE = "01:2d:6f:00:0a:90:69:e7"
async def test_services_registered_without_gateway(hass: HomeAssistant) -> None:
"""Test the actions are registered during component setup."""
assert await async_setup_component(hass, DOMAIN, {})
await hass.async_block_till_done()
for service in (
SERVICE_PERMIT,
SERVICE_REMOVE,
SERVICE_SET_ZIGBEE_CLUSTER_ATTRIBUTE,
SERVICE_ISSUE_ZIGBEE_CLUSTER_COMMAND,
SERVICE_ISSUE_ZIGBEE_GROUP_COMMAND,
SERVICE_WARNING_DEVICE_SQUAWK,
SERVICE_WARNING_DEVICE_WARN,
):
assert hass.services.has_service(DOMAIN, service)
@pytest.mark.parametrize(
("service", "data"),
[
pytest.param(SERVICE_PERMIT, {}, id="permit"),
pytest.param(SERVICE_REMOVE, {"ieee": IEEE_SWITCH_DEVICE}, id="remove"),
pytest.param(
SERVICE_SET_ZIGBEE_CLUSTER_ATTRIBUTE,
{
"ieee": IEEE_SWITCH_DEVICE,
"endpoint_id": 1,
"cluster_id": 6,
"attribute": 0,
"value": 1,
},
id="set_zigbee_cluster_attribute",
),
pytest.param(
SERVICE_ISSUE_ZIGBEE_CLUSTER_COMMAND,
{
"ieee": IEEE_SWITCH_DEVICE,
"endpoint_id": 1,
"cluster_id": 6,
"command": 0,
"command_type": "server",
"params": {},
},
id="issue_zigbee_cluster_command",
),
pytest.param(
SERVICE_ISSUE_ZIGBEE_GROUP_COMMAND,
{"group": 1, "cluster_id": 6, "command": 0},
id="issue_zigbee_group_command",
),
pytest.param(
SERVICE_WARNING_DEVICE_SQUAWK,
{"ieee": IEEE_SWITCH_DEVICE},
id="warning_device_squawk",
),
pytest.param(
SERVICE_WARNING_DEVICE_WARN,
{"ieee": IEEE_SWITCH_DEVICE},
id="warning_device_warn",
),
],
)
async def test_service_without_gateway_raises(
hass: HomeAssistant, service: str, data: dict
) -> None:
"""Test the actions raise a user-facing error when no gateway is loaded."""
assert await async_setup_component(hass, DOMAIN, {})
await hass.async_block_till_done()
with pytest.raises(ServiceValidationError) as err:
await hass.services.async_call(DOMAIN, service, data, blocking=True)
assert err.value.translation_domain == DOMAIN
assert err.value.translation_key == "no_gateway"
tests/components/zha/test_websocket_api.py
+8 -6
View File
@@ -42,25 +42,27 @@ from homeassistant.components.websocket_api import (
TYPE_RESULT,
)
from homeassistant.components.zha import DOMAIN
from homeassistant.components.zha.const import EZSP_OVERWRITE_EUI64
from homeassistant.components.zha.const import (
ATTR_DURATION,
ATTR_INSTALL_CODE,
ATTR_QR_CODE,
ATTR_SOURCE_IEEE,
EZSP_OVERWRITE_EUI64,
)
from homeassistant.components.zha.helpers import (
ZHADeviceProxy,
ZHAGatewayProxy,
get_zha_gateway,
get_zha_gateway_proxy,
)
from homeassistant.components.zha.services import SERVICE_PERMIT
from homeassistant.components.zha.websocket_api import (
ATTR_DURATION,
ATTR_INSTALL_CODE,
ATTR_QR_CODE,
ATTR_SOURCE_IEEE,
ATTR_TARGET_IEEE,
BINDINGS,
GROUP_ID,
GROUP_IDS,
GROUP_NAME,
ID,
SERVICE_PERMIT,
TYPE,
async_load_api,
)