Skip to content

Add zha typing [api] (3) #68353

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
dmulcahey merged 3 commits into from
Mar 22, 2022
Merged

Add zha typing [api] (3) #68353

Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
f3e7568
Add zha typing [api] (3)
cdce8p Mar 19, 2022
6b79d4c
Merge remote-tracking branch 'upstream/dev' into zha-api_3
cdce8p Mar 19, 2022
5e7fdab
Merge branch 'dev' into zha-api_3
cdce8p Mar 20, 2022
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
60 changes: 31 additions & 29 deletions homeassistant/components/zha/api.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -105,13 +105,15 @@
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 = {
Expand All @@ -124,12 +126,12 @@
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,
Expand All @@ -140,7 +142,7 @@
),
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,
Expand All @@ -154,7 +156,7 @@
),
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,
Expand All @@ -175,7 +177,7 @@
),
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,
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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,
Expand Down Expand Up @@ -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,
Expand Down Expand Up @@ -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,
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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
Expand All @@ -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
Expand Down Expand Up @@ -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]),
}
Expand All @@ -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]),
}
Expand Down Expand Up @@ -1257,13 +1259,13 @@ async def warning_device_squawk(service: ServiceCall) -> 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):
Expand Down