From 6a66b4dbffea0c14ad760b224ad8fb0d55f6b8db Mon Sep 17 00:00:00 2001 From: Marc Mueller <30130371+cdce8p@users.noreply.github.com> Date: Tue, 22 Mar 2022 15:15:39 +0100 Subject: [PATCH] Add zha typing [api] (3) (#68353) --- homeassistant/components/zha/api.py | 60 +++++++++++++++-------------- 1 file changed, 31 insertions(+), 29 deletions(-) diff --git a/homeassistant/components/zha/api.py b/homeassistant/components/zha/api.py index ac028597ea8..060e8a5d5eb 100644 --- a/homeassistant/components/zha/api.py +++ b/homeassistant/components/zha/api.py @@ -105,13 +105,15 @@ SERVICE_ZIGBEE_BIND = "service_zigbee_bind" IEEE_SERVICE = "ieee_based_service" 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.Coerce(int), vol.Range(0, 254) ), - vol.Inclusive(ATTR_SOURCE_IEEE, "install_code"): EUI64.convert, - vol.Inclusive(ATTR_INSTALL_CODE, "install_code"): convert_install_code, - vol.Exclusive(ATTR_QR_CODE, "install_code"): vol.All(str, qr_to_install_code), + vol.Inclusive(ATTR_SOURCE_IEEE, "install_code"): vol.All(cv.string, EUI64.convert), + vol.Inclusive(ATTR_INSTALL_CODE, "install_code"): vol.All( + cv.string, convert_install_code + ), + vol.Exclusive(ATTR_QR_CODE, "install_code"): vol.All(cv.string, qr_to_install_code), } SERVICE_SCHEMAS = { @@ -124,12 +126,12 @@ SERVICE_SCHEMAS = { IEEE_SERVICE: vol.Schema( vol.All( 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( { - 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_CLUSTER_ID): cv.positive_int, vol.Optional(ATTR_CLUSTER_TYPE, default=CLUSTER_TYPE_IN): cv.string, @@ -140,7 +142,7 @@ SERVICE_SCHEMAS = { ), SERVICE_WARNING_DEVICE_SQUAWK: vol.Schema( { - vol.Required(ATTR_IEEE): EUI64.convert, + vol.Required(ATTR_IEEE): vol.All(cv.string, EUI64.convert), vol.Optional( ATTR_WARNING_DEVICE_MODE, default=WARNING_DEVICE_SQUAWK_MODE_ARMED ): cv.positive_int, @@ -154,7 +156,7 @@ SERVICE_SCHEMAS = { ), SERVICE_WARNING_DEVICE_WARN: vol.Schema( { - vol.Required(ATTR_IEEE): EUI64.convert, + vol.Required(ATTR_IEEE): vol.All(cv.string, EUI64.convert), vol.Optional( ATTR_WARNING_DEVICE_MODE, default=WARNING_DEVICE_MODE_EMERGENCY ): cv.positive_int, @@ -175,7 +177,7 @@ SERVICE_SCHEMAS = { ), 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_CLUSTER_ID): cv.positive_int, vol.Optional(ATTR_CLUSTER_TYPE, default=CLUSTER_TYPE_IN): cv.string, @@ -320,7 +322,7 @@ async def websocket_get_groups( @websocket_api.websocket_command( { 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 @@ -501,7 +503,7 @@ async def websocket_remove_group_members( @websocket_api.websocket_command( { 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 @@ -551,7 +553,7 @@ async def websocket_update_topology( @websocket_api.websocket_command( { 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 @@ -592,7 +594,7 @@ async def websocket_device_clusters( @websocket_api.websocket_command( { 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_CLUSTER_ID): int, vol.Required(ATTR_CLUSTER_TYPE): str, @@ -639,7 +641,7 @@ async def websocket_device_cluster_attributes( @websocket_api.websocket_command( { 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_CLUSTER_ID): int, vol.Required(ATTR_CLUSTER_TYPE): str, @@ -699,7 +701,7 @@ async def websocket_device_cluster_commands( @websocket_api.websocket_command( { 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_CLUSTER_ID): int, vol.Required(ATTR_CLUSTER_TYPE): str, @@ -754,7 +756,7 @@ async def websocket_read_zigbee_cluster_attributes( @websocket_api.websocket_command( { 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 @@ -787,8 +789,8 @@ async def websocket_get_bindable_devices( @websocket_api.websocket_command( { vol.Required(TYPE): "zha/devices/bind", - vol.Required(ATTR_SOURCE_IEEE): EUI64.convert, - vol.Required(ATTR_TARGET_IEEE): EUI64.convert, + vol.Required(ATTR_SOURCE_IEEE): vol.All(cv.string, EUI64.convert), + vol.Required(ATTR_TARGET_IEEE): vol.All(cv.string, EUI64.convert), } ) @websocket_api.async_response @@ -815,8 +817,8 @@ async def websocket_bind_devices( @websocket_api.websocket_command( { vol.Required(TYPE): "zha/devices/unbind", - vol.Required(ATTR_SOURCE_IEEE): EUI64.convert, - vol.Required(ATTR_TARGET_IEEE): EUI64.convert, + vol.Required(ATTR_SOURCE_IEEE): vol.All(cv.string, EUI64.convert), + vol.Required(ATTR_TARGET_IEEE): vol.All(cv.string, EUI64.convert), } ) @websocket_api.async_response @@ -860,7 +862,7 @@ def is_cluster_binding(value: Any) -> ClusterBinding: @websocket_api.websocket_command( { 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(BINDINGS): vol.All(cv.ensure_list, [is_cluster_binding]), } @@ -882,7 +884,7 @@ async def websocket_bind_group( @websocket_api.websocket_command( { 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(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: """Issue the warning command for an IAS warning device.""" - ieee = service.data[ATTR_IEEE] - mode = service.data.get(ATTR_WARNING_DEVICE_MODE) - strobe = service.data.get(ATTR_WARNING_DEVICE_STROBE) - level = service.data.get(ATTR_LEVEL) - duration = service.data.get(ATTR_WARNING_DEVICE_DURATION) - duty_mode = service.data.get(ATTR_WARNING_DEVICE_STROBE_DUTY_CYCLE) - intensity = service.data.get(ATTR_WARNING_DEVICE_STROBE_INTENSITY) + 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] if (zha_device := zha_gateway.get_device(ieee)) is not None: if channel := _get_ias_wd_channel(zha_device):