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

Add zha typing [api] (3) (#68353)

Browse Source
This commit is contained in:
Marc Mueller 2022-03-22 15:15:39 +01:00 committed by GitHub
parent df05e8b950
commit 6a66b4dbff
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
1 changed files with 31 additions and 29 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

60
homeassistant/components/zha/api.py
View File

@ -105,13 +105,15 @@ SERVICE_ZIGBEE_BIND = "service_zigbee_bind"
IEEE_SERVICE = "ieee_based_service" IEEE_SERVICE = "ieee_based_service"
SERVICE_PERMIT_PARAMS = { SERVICE_PERMIT_PARAMS = {
vol.Optional(ATTR_IEEE, default=None): EUI64.convert, vol.Optional(ATTR_IEEE): vol.All(cv.string, EUI64.convert),
vol.Optional(ATTR_DURATION, default=60): vol.All( vol.Optional(ATTR_DURATION, default=60): vol.All(
vol.Coerce(int), vol.Range(0, 254) vol.Coerce(int), vol.Range(0, 254)
), ),
vol.Inclusive(ATTR_SOURCE_IEEE, "install_code"): EUI64.convert, vol.Inclusive(ATTR_SOURCE_IEEE, "install_code"): vol.All(cv.string, EUI64.convert),
vol.Inclusive(ATTR_INSTALL_CODE, "install_code"): convert_install_code, vol.Inclusive(ATTR_INSTALL_CODE, "install_code"): vol.All(
vol.Exclusive(ATTR_QR_CODE, "install_code"): vol.All(str, qr_to_install_code), cv.string, convert_install_code
),
vol.Exclusive(ATTR_QR_CODE, "install_code"): vol.All(cv.string, qr_to_install_code),
} }
SERVICE_SCHEMAS = { SERVICE_SCHEMAS = {
@ -124,12 +126,12 @@ SERVICE_SCHEMAS = {
IEEE_SERVICE: vol.Schema( IEEE_SERVICE: vol.Schema(
vol.All( vol.All(
cv.deprecated(ATTR_IEEE_ADDRESS, replacement_key=ATTR_IEEE), cv.deprecated(ATTR_IEEE_ADDRESS, replacement_key=ATTR_IEEE),
{vol.Required(ATTR_IEEE): EUI64.convert}, {vol.Required(ATTR_IEEE): vol.All(cv.string, EUI64.convert)},
) )
), ),
SERVICE_SET_ZIGBEE_CLUSTER_ATTRIBUTE: vol.Schema( SERVICE_SET_ZIGBEE_CLUSTER_ATTRIBUTE: vol.Schema(
{ {
vol.Required(ATTR_IEEE): EUI64.convert, vol.Required(ATTR_IEEE): vol.All(cv.string, EUI64.convert),
vol.Required(ATTR_ENDPOINT_ID): cv.positive_int, vol.Required(ATTR_ENDPOINT_ID): cv.positive_int,
vol.Required(ATTR_CLUSTER_ID): cv.positive_int, vol.Required(ATTR_CLUSTER_ID): cv.positive_int,
vol.Optional(ATTR_CLUSTER_TYPE, default=CLUSTER_TYPE_IN): cv.string, vol.Optional(ATTR_CLUSTER_TYPE, default=CLUSTER_TYPE_IN): cv.string,
@ -140,7 +142,7 @@ SERVICE_SCHEMAS = {
), ),
SERVICE_WARNING_DEVICE_SQUAWK: vol.Schema( SERVICE_WARNING_DEVICE_SQUAWK: vol.Schema(
{ {
vol.Required(ATTR_IEEE): EUI64.convert, vol.Required(ATTR_IEEE): vol.All(cv.string, EUI64.convert),
vol.Optional( vol.Optional(
ATTR_WARNING_DEVICE_MODE, default=WARNING_DEVICE_SQUAWK_MODE_ARMED ATTR_WARNING_DEVICE_MODE, default=WARNING_DEVICE_SQUAWK_MODE_ARMED
): cv.positive_int, ): cv.positive_int,
@ -154,7 +156,7 @@ SERVICE_SCHEMAS = {
), ),
SERVICE_WARNING_DEVICE_WARN: vol.Schema( SERVICE_WARNING_DEVICE_WARN: vol.Schema(
{ {
vol.Required(ATTR_IEEE): EUI64.convert, vol.Required(ATTR_IEEE): vol.All(cv.string, EUI64.convert),
vol.Optional( vol.Optional(
ATTR_WARNING_DEVICE_MODE, default=WARNING_DEVICE_MODE_EMERGENCY ATTR_WARNING_DEVICE_MODE, default=WARNING_DEVICE_MODE_EMERGENCY
): cv.positive_int, ): cv.positive_int,
@ -175,7 +177,7 @@ SERVICE_SCHEMAS = {
), ),
SERVICE_ISSUE_ZIGBEE_CLUSTER_COMMAND: vol.Schema( SERVICE_ISSUE_ZIGBEE_CLUSTER_COMMAND: vol.Schema(
{ {
vol.Required(ATTR_IEEE): EUI64.convert, vol.Required(ATTR_IEEE): vol.All(cv.string, EUI64.convert),
vol.Required(ATTR_ENDPOINT_ID): cv.positive_int, vol.Required(ATTR_ENDPOINT_ID): cv.positive_int,
vol.Required(ATTR_CLUSTER_ID): cv.positive_int, vol.Required(ATTR_CLUSTER_ID): cv.positive_int,
vol.Optional(ATTR_CLUSTER_TYPE, default=CLUSTER_TYPE_IN): cv.string, vol.Optional(ATTR_CLUSTER_TYPE, default=CLUSTER_TYPE_IN): cv.string,
@ -320,7 +322,7 @@ async def websocket_get_groups(
@websocket_api.websocket_command( @websocket_api.websocket_command(
{ {
vol.Required(TYPE): "zha/device", vol.Required(TYPE): "zha/device",
vol.Required(ATTR_IEEE): EUI64.convert, vol.Required(ATTR_IEEE): vol.All(cv.string, EUI64.convert),
} }
) )
@websocket_api.async_response @websocket_api.async_response
@ -501,7 +503,7 @@ async def websocket_remove_group_members(
@websocket_api.websocket_command( @websocket_api.websocket_command(
{ {
vol.Required(TYPE): "zha/devices/reconfigure", vol.Required(TYPE): "zha/devices/reconfigure",
vol.Required(ATTR_IEEE): EUI64.convert, vol.Required(ATTR_IEEE): vol.All(cv.string, EUI64.convert),
} }
) )
@websocket_api.async_response @websocket_api.async_response
@ -551,7 +553,7 @@ async def websocket_update_topology(
@websocket_api.websocket_command( @websocket_api.websocket_command(
{ {
vol.Required(TYPE): "zha/devices/clusters", vol.Required(TYPE): "zha/devices/clusters",
vol.Required(ATTR_IEEE): EUI64.convert, vol.Required(ATTR_IEEE): vol.All(cv.string, EUI64.convert),
} }
) )
@websocket_api.async_response @websocket_api.async_response
@ -592,7 +594,7 @@ async def websocket_device_clusters(
@websocket_api.websocket_command( @websocket_api.websocket_command(
{ {
vol.Required(TYPE): "zha/devices/clusters/attributes", vol.Required(TYPE): "zha/devices/clusters/attributes",
vol.Required(ATTR_IEEE): EUI64.convert, vol.Required(ATTR_IEEE): vol.All(cv.string, EUI64.convert),
vol.Required(ATTR_ENDPOINT_ID): int, vol.Required(ATTR_ENDPOINT_ID): int,
vol.Required(ATTR_CLUSTER_ID): int, vol.Required(ATTR_CLUSTER_ID): int,
vol.Required(ATTR_CLUSTER_TYPE): str, vol.Required(ATTR_CLUSTER_TYPE): str,
@ -639,7 +641,7 @@ async def websocket_device_cluster_attributes(
@websocket_api.websocket_command( @websocket_api.websocket_command(
{ {
vol.Required(TYPE): "zha/devices/clusters/commands", vol.Required(TYPE): "zha/devices/clusters/commands",
vol.Required(ATTR_IEEE): EUI64.convert, vol.Required(ATTR_IEEE): vol.All(cv.string, EUI64.convert),
vol.Required(ATTR_ENDPOINT_ID): int, vol.Required(ATTR_ENDPOINT_ID): int,
vol.Required(ATTR_CLUSTER_ID): int, vol.Required(ATTR_CLUSTER_ID): int,
vol.Required(ATTR_CLUSTER_TYPE): str, vol.Required(ATTR_CLUSTER_TYPE): str,
@ -699,7 +701,7 @@ async def websocket_device_cluster_commands(
@websocket_api.websocket_command( @websocket_api.websocket_command(
{ {
vol.Required(TYPE): "zha/devices/clusters/attributes/value", vol.Required(TYPE): "zha/devices/clusters/attributes/value",
vol.Required(ATTR_IEEE): EUI64.convert, vol.Required(ATTR_IEEE): vol.All(cv.string, EUI64.convert),
vol.Required(ATTR_ENDPOINT_ID): int, vol.Required(ATTR_ENDPOINT_ID): int,
vol.Required(ATTR_CLUSTER_ID): int, vol.Required(ATTR_CLUSTER_ID): int,
vol.Required(ATTR_CLUSTER_TYPE): str, vol.Required(ATTR_CLUSTER_TYPE): str,
@ -754,7 +756,7 @@ async def websocket_read_zigbee_cluster_attributes(
@websocket_api.websocket_command( @websocket_api.websocket_command(
{ {
vol.Required(TYPE): "zha/devices/bindable", vol.Required(TYPE): "zha/devices/bindable",
vol.Required(ATTR_IEEE): EUI64.convert, vol.Required(ATTR_IEEE): vol.All(cv.string, EUI64.convert),
} }
) )
@websocket_api.async_response @websocket_api.async_response
@ -787,8 +789,8 @@ async def websocket_get_bindable_devices(
@websocket_api.websocket_command( @websocket_api.websocket_command(
{ {
vol.Required(TYPE): "zha/devices/bind", vol.Required(TYPE): "zha/devices/bind",
vol.Required(ATTR_SOURCE_IEEE): EUI64.convert, vol.Required(ATTR_SOURCE_IEEE): vol.All(cv.string, EUI64.convert),
vol.Required(ATTR_TARGET_IEEE): EUI64.convert, vol.Required(ATTR_TARGET_IEEE): vol.All(cv.string, EUI64.convert),
} }
) )
@websocket_api.async_response @websocket_api.async_response
@ -815,8 +817,8 @@ async def websocket_bind_devices(
@websocket_api.websocket_command( @websocket_api.websocket_command(
{ {
vol.Required(TYPE): "zha/devices/unbind", vol.Required(TYPE): "zha/devices/unbind",
vol.Required(ATTR_SOURCE_IEEE): EUI64.convert, vol.Required(ATTR_SOURCE_IEEE): vol.All(cv.string, EUI64.convert),
vol.Required(ATTR_TARGET_IEEE): EUI64.convert, vol.Required(ATTR_TARGET_IEEE): vol.All(cv.string, EUI64.convert),
} }
) )
@websocket_api.async_response @websocket_api.async_response
@ -860,7 +862,7 @@ def is_cluster_binding(value: Any) -> ClusterBinding:
@websocket_api.websocket_command( @websocket_api.websocket_command(
{ {
vol.Required(TYPE): "zha/groups/bind", vol.Required(TYPE): "zha/groups/bind",
vol.Required(ATTR_SOURCE_IEEE): EUI64.convert, vol.Required(ATTR_SOURCE_IEEE): vol.All(cv.string, EUI64.convert),
vol.Required(GROUP_ID): cv.positive_int, vol.Required(GROUP_ID): cv.positive_int,
vol.Required(BINDINGS): vol.All(cv.ensure_list, [is_cluster_binding]), vol.Required(BINDINGS): vol.All(cv.ensure_list, [is_cluster_binding]),
} }
@ -882,7 +884,7 @@ async def websocket_bind_group(
@websocket_api.websocket_command( @websocket_api.websocket_command(
{ {
vol.Required(TYPE): "zha/groups/unbind", vol.Required(TYPE): "zha/groups/unbind",
vol.Required(ATTR_SOURCE_IEEE): EUI64.convert, vol.Required(ATTR_SOURCE_IEEE): vol.All(cv.string, EUI64.convert),
vol.Required(GROUP_ID): cv.positive_int, vol.Required(GROUP_ID): cv.positive_int,
vol.Required(BINDINGS): vol.All(cv.ensure_list, [is_cluster_binding]), vol.Required(BINDINGS): vol.All(cv.ensure_list, [is_cluster_binding]),
} }
@ -1257,13 +1259,13 @@ def async_load_api(hass: HomeAssistant) -> None:
async def warning_device_warn(service: ServiceCall) -> None: async def warning_device_warn(service: ServiceCall) -> None:
"""Issue the warning command for an IAS warning device.""" """Issue the warning command for an IAS warning device."""
ieee = service.data[ATTR_IEEE] ieee: EUI64 = service.data[ATTR_IEEE]
mode = service.data.get(ATTR_WARNING_DEVICE_MODE) mode: int = service.data[ATTR_WARNING_DEVICE_MODE]
strobe = service.data.get(ATTR_WARNING_DEVICE_STROBE) strobe: int = service.data[ATTR_WARNING_DEVICE_STROBE]
level = service.data.get(ATTR_LEVEL) level: int = service.data[ATTR_LEVEL]
duration = service.data.get(ATTR_WARNING_DEVICE_DURATION) duration: int = service.data[ATTR_WARNING_DEVICE_DURATION]
duty_mode = service.data.get(ATTR_WARNING_DEVICE_STROBE_DUTY_CYCLE) duty_mode: int = service.data[ATTR_WARNING_DEVICE_STROBE_DUTY_CYCLE]
intensity = service.data.get(ATTR_WARNING_DEVICE_STROBE_INTENSITY) intensity: int = service.data[ATTR_WARNING_DEVICE_STROBE_INTENSITY]
if (zha_device := zha_gateway.get_device(ieee)) is not None: if (zha_device := zha_gateway.get_device(ieee)) is not None:
if channel := _get_ias_wd_channel(zha_device): if channel := _get_ias_wd_channel(zha_device):