Improve msg type hint in websocket commands (#80530)

This commit is contained in:
epenet 2022-10-18 16:41:17 +02:00 committed by GitHub
parent 76bd8c431b
commit 5442d6af01
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
21 changed files with 128 additions and 104 deletions

View File

@ -1,4 +1,6 @@
"""Send instance and usage analytics."""
from typing import Any
import voluptuous as vol
from homeassistant.components import websocket_api
@ -41,7 +43,7 @@ async def async_setup(hass: HomeAssistant, _: ConfigType) -> bool:
async def websocket_analytics(
hass: HomeAssistant,
connection: websocket_api.connection.ActiveConnection,
msg: dict,
msg: dict[str, Any],
) -> None:
"""Return analytics preferences."""
analytics: Analytics = hass.data[DOMAIN]
@ -62,7 +64,7 @@ async def websocket_analytics(
async def websocket_analytics_preferences(
hass: HomeAssistant,
connection: websocket_api.connection.ActiveConnection,
msg: dict,
msg: dict[str, Any],
) -> None:
"""Update analytics preferences."""
preferences = msg[ATTR_PREFERENCES]

View File

@ -1,4 +1,6 @@
"""Websocket commands for the Backup integration."""
from typing import Any
import voluptuous as vol
from homeassistant.components import websocket_api
@ -22,7 +24,7 @@ def async_register_websocket_handlers(hass: HomeAssistant) -> None:
async def handle_info(
hass: HomeAssistant,
connection: websocket_api.ActiveConnection,
msg: dict,
msg: dict[str, Any],
) -> None:
"""List all stored backups."""
manager: BackupManager = hass.data[DOMAIN]
@ -47,7 +49,7 @@ async def handle_info(
async def handle_remove(
hass: HomeAssistant,
connection: websocket_api.ActiveConnection,
msg: dict,
msg: dict[str, Any],
) -> None:
"""Remove a backup."""
manager: BackupManager = hass.data[DOMAIN]
@ -61,7 +63,7 @@ async def handle_remove(
async def handle_create(
hass: HomeAssistant,
connection: websocket_api.ActiveConnection,
msg: dict,
msg: dict[str, Any],
) -> None:
"""Generate a backup."""
manager: BackupManager = hass.data[DOMAIN]

View File

@ -12,7 +12,7 @@ from functools import partial
import logging
import os
from random import SystemRandom
from typing import Final, Optional, cast, final
from typing import Any, Final, Optional, cast, final
from aiohttp import hdrs, web
import async_timeout
@ -786,7 +786,7 @@ class CameraMjpegStream(CameraView):
)
@websocket_api.async_response
async def ws_camera_stream(
hass: HomeAssistant, connection: ActiveConnection, msg: dict
hass: HomeAssistant, connection: ActiveConnection, msg: dict[str, Any]
) -> None:
"""Handle get camera stream websocket command.
@ -816,7 +816,7 @@ async def ws_camera_stream(
)
@websocket_api.async_response
async def ws_camera_web_rtc_offer(
hass: HomeAssistant, connection: ActiveConnection, msg: dict
hass: HomeAssistant, connection: ActiveConnection, msg: dict[str, Any]
) -> None:
"""Handle the signal path for a WebRTC stream.
@ -856,7 +856,7 @@ async def ws_camera_web_rtc_offer(
)
@websocket_api.async_response
async def websocket_get_prefs(
hass: HomeAssistant, connection: ActiveConnection, msg: dict
hass: HomeAssistant, connection: ActiveConnection, msg: dict[str, Any]
) -> None:
"""Handle request for account info."""
prefs = hass.data[DATA_CAMERA_PREFS].get(msg["entity_id"])
@ -873,7 +873,7 @@ async def websocket_get_prefs(
)
@websocket_api.async_response
async def websocket_update_prefs(
hass: HomeAssistant, connection: ActiveConnection, msg: dict
hass: HomeAssistant, connection: ActiveConnection, msg: dict[str, Any]
) -> None:
"""Handle request for account info."""
prefs = hass.data[DATA_CAMERA_PREFS]

View File

@ -1,6 +1,8 @@
"""HTTP views to interact with the device registry."""
from __future__ import annotations
from typing import Any
import voluptuous as vol
from homeassistant import loader
@ -109,7 +111,9 @@ def websocket_update_device(hass, connection, msg):
)
@websocket_api.async_response
async def websocket_remove_config_entry_from_device(
hass: HomeAssistant, connection: websocket_api.ActiveConnection, msg: dict
hass: HomeAssistant,
connection: websocket_api.ActiveConnection,
msg: dict[str, Any],
) -> None:
"""Remove config entry from a device."""
registry = async_get(hass)

View File

@ -82,7 +82,9 @@ def _ws_with_manager(
@websocket_api.async_response
@functools.wraps(func)
async def with_manager(
hass: HomeAssistant, connection: websocket_api.ActiveConnection, msg: dict
hass: HomeAssistant,
connection: websocket_api.ActiveConnection,
msg: dict[str, Any],
) -> None:
manager = await async_get_manager(hass)
@ -146,7 +148,7 @@ async def ws_save_prefs(
async def ws_info(
hass: HomeAssistant,
connection: websocket_api.ActiveConnection,
msg: dict,
msg: dict[str, Any],
) -> None:
"""Handle get info command."""
forecast_platforms = await async_get_energy_platforms(hass)
@ -168,7 +170,7 @@ async def ws_info(
async def ws_validate(
hass: HomeAssistant,
connection: websocket_api.ActiveConnection,
msg: dict,
msg: dict[str, Any],
) -> None:
"""Handle validate command."""
connection.send_result(msg["id"], (await async_validate(hass)).as_dict())
@ -239,7 +241,9 @@ async def ws_solar_forecast(
)
@websocket_api.async_response
async def ws_get_fossil_energy_consumption(
hass: HomeAssistant, connection: websocket_api.ActiveConnection, msg: dict
hass: HomeAssistant,
connection: websocket_api.ActiveConnection,
msg: dict[str, Any],
) -> None:
"""Calculate amount of fossil based energy."""
start_time_str = msg["start_time"]

View File

@ -673,7 +673,7 @@ def websocket_get_themes(
)
@websocket_api.async_response
async def websocket_get_translations(
hass: HomeAssistant, connection: ActiveConnection, msg: dict
hass: HomeAssistant, connection: ActiveConnection, msg: dict[str, Any]
) -> None:
"""Handle get translations command."""
resources = await async_get_translations(
@ -691,7 +691,7 @@ async def websocket_get_translations(
@websocket_api.websocket_command({"type": "frontend/get_version"})
@websocket_api.async_response
async def websocket_get_version(
hass: HomeAssistant, connection: ActiveConnection, msg: dict
hass: HomeAssistant, connection: ActiveConnection, msg: dict[str, Any]
) -> None:
"""Handle get version command."""
integration = await async_get_integration(hass, "frontend")

View File

@ -61,7 +61,7 @@ def with_store(orig_func: Callable) -> Callable:
async def websocket_set_user_data(
hass: HomeAssistant,
connection: ActiveConnection,
msg: dict,
msg: dict[str, Any],
store: Store,
data: dict[str, Any],
) -> None:
@ -82,7 +82,7 @@ async def websocket_set_user_data(
async def websocket_get_user_data(
hass: HomeAssistant,
connection: ActiveConnection,
msg: dict,
msg: dict[str, Any],
store: Store,
data: dict[str, Any],
) -> None:

View File

@ -4,6 +4,7 @@ from __future__ import annotations
import contextlib
from dataclasses import asdict, dataclass
from datetime import datetime, timedelta
from typing import Any
import psutil_home_assistant as ha_psutil
import voluptuous as vol
@ -46,7 +47,7 @@ async def async_setup(hass: HomeAssistant) -> None:
)
@websocket_api.async_response
async def ws_info(
hass: HomeAssistant, connection: websocket_api.ActiveConnection, msg: dict
hass: HomeAssistant, connection: websocket_api.ActiveConnection, msg: dict[str, Any]
) -> None:
"""Return hardware info."""
hardware_info = []
@ -72,7 +73,7 @@ async def ws_info(
)
@websocket_api.async_response
async def ws_subscribe_system_status(
hass: HomeAssistant, connection: websocket_api.ActiveConnection, msg: dict
hass: HomeAssistant, connection: websocket_api.ActiveConnection, msg: dict[str, Any]
):
"""Subscribe to system status updates."""

View File

@ -2,6 +2,7 @@
import logging
from numbers import Number
import re
from typing import Any
import voluptuous as vol
@ -60,7 +61,7 @@ def async_load_websocket_api(hass: HomeAssistant):
@websocket_api.websocket_command({vol.Required(WS_TYPE): WS_TYPE_SUBSCRIBE})
@websocket_api.async_response
async def websocket_subscribe(
hass: HomeAssistant, connection: ActiveConnection, msg: dict
hass: HomeAssistant, connection: ActiveConnection, msg: dict[str, Any]
):
"""Subscribe to supervisor events."""
@ -83,7 +84,7 @@ async def websocket_subscribe(
)
@websocket_api.async_response
async def websocket_supervisor_event(
hass: HomeAssistant, connection: ActiveConnection, msg: dict
hass: HomeAssistant, connection: ActiveConnection, msg: dict[str, Any]
):
"""Publish events from the Supervisor."""
connection.send_result(msg[WS_ID])
@ -101,7 +102,7 @@ async def websocket_supervisor_event(
)
@websocket_api.async_response
async def websocket_supervisor_api(
hass: HomeAssistant, connection: ActiveConnection, msg: dict
hass: HomeAssistant, connection: ActiveConnection, msg: dict[str, Any]
):
"""Websocket handler to call Supervisor API."""
if not connection.user.is_admin and not WS_NO_ADMIN_ENDPOINTS.match(

View File

@ -6,7 +6,7 @@ from datetime import datetime as dt, timedelta
from http import HTTPStatus
import logging
import time
from typing import cast
from typing import Any, cast
from aiohttp import web
import voluptuous as vol
@ -79,7 +79,7 @@ async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
)
@websocket_api.async_response
async def ws_get_statistics_during_period(
hass: HomeAssistant, connection: websocket_api.ActiveConnection, msg: dict
hass: HomeAssistant, connection: websocket_api.ActiveConnection, msg: dict[str, Any]
) -> None:
"""Handle statistics websocket command."""
_LOGGER.warning(
@ -97,7 +97,7 @@ async def ws_get_statistics_during_period(
)
@websocket_api.async_response
async def ws_get_list_statistic_ids(
hass: HomeAssistant, connection: websocket_api.ActiveConnection, msg: dict
hass: HomeAssistant, connection: websocket_api.ActiveConnection, msg: dict[str, Any]
) -> None:
"""Fetch a list of available statistic_id."""
_LOGGER.warning(
@ -164,7 +164,7 @@ def _ws_get_significant_states(
)
@websocket_api.async_response
async def ws_get_history_during_period(
hass: HomeAssistant, connection: websocket_api.ActiveConnection, msg: dict
hass: HomeAssistant, connection: websocket_api.ActiveConnection, msg: dict[str, Any]
) -> None:
"""Handle history during period websocket command."""
start_time_str = msg["start_time"]

View File

@ -1,5 +1,7 @@
"""Web socket API for Insteon devices."""
from typing import Any
from pyinsteon import devices
from pyinsteon.constants import ALDBStatus
from pyinsteon.topics import (
@ -71,7 +73,7 @@ async def async_reload_and_save_aldb(hass, device):
async def websocket_get_aldb(
hass: HomeAssistant,
connection: websocket_api.connection.ActiveConnection,
msg: dict,
msg: dict[str, Any],
) -> None:
"""Get the All-Link Database for an Insteon device."""
if not (device := devices[msg[DEVICE_ADDRESS]]):
@ -107,7 +109,7 @@ async def websocket_get_aldb(
async def websocket_change_aldb_record(
hass: HomeAssistant,
connection: websocket_api.connection.ActiveConnection,
msg: dict,
msg: dict[str, Any],
) -> None:
"""Change an All-Link Database record for an Insteon device."""
if not (device := devices[msg[DEVICE_ADDRESS]]):
@ -140,7 +142,7 @@ async def websocket_change_aldb_record(
async def websocket_create_aldb_record(
hass: HomeAssistant,
connection: websocket_api.connection.ActiveConnection,
msg: dict,
msg: dict[str, Any],
) -> None:
"""Create an All-Link Database record for an Insteon device."""
if not (device := devices[msg[DEVICE_ADDRESS]]):
@ -170,7 +172,7 @@ async def websocket_create_aldb_record(
async def websocket_write_aldb(
hass: HomeAssistant,
connection: websocket_api.connection.ActiveConnection,
msg: dict,
msg: dict[str, Any],
) -> None:
"""Create an All-Link Database record for an Insteon device."""
if not (device := devices[msg[DEVICE_ADDRESS]]):
@ -193,7 +195,7 @@ async def websocket_write_aldb(
async def websocket_load_aldb(
hass: HomeAssistant,
connection: websocket_api.connection.ActiveConnection,
msg: dict,
msg: dict[str, Any],
) -> None:
"""Create an All-Link Database record for an Insteon device."""
if not (device := devices[msg[DEVICE_ADDRESS]]):
@ -215,7 +217,7 @@ async def websocket_load_aldb(
async def websocket_reset_aldb(
hass: HomeAssistant,
connection: websocket_api.connection.ActiveConnection,
msg: dict,
msg: dict[str, Any],
) -> None:
"""Create an All-Link Database record for an Insteon device."""
if not (device := devices[msg[DEVICE_ADDRESS]]):
@ -237,7 +239,7 @@ async def websocket_reset_aldb(
async def websocket_add_default_links(
hass: HomeAssistant,
connection: websocket_api.connection.ActiveConnection,
msg: dict,
msg: dict[str, Any],
) -> None:
"""Add the default All-Link Database records for an Insteon device."""
if not (device := devices[msg[DEVICE_ADDRESS]]):
@ -261,7 +263,7 @@ async def websocket_add_default_links(
async def websocket_notify_on_aldb_status(
hass: HomeAssistant,
connection: websocket_api.connection.ActiveConnection,
msg: dict,
msg: dict[str, Any],
) -> None:
"""Tell Insteon a new ALDB record was added."""
if not (device := devices[msg[DEVICE_ADDRESS]]):

View File

@ -1,5 +1,7 @@
"""API interface to get an Insteon device."""
from typing import Any
from pyinsteon import devices
from pyinsteon.constants import DeviceAction
import voluptuous as vol
@ -66,7 +68,7 @@ def notify_device_not_found(connection, msg, text):
async def websocket_get_device(
hass: HomeAssistant,
connection: websocket_api.connection.ActiveConnection,
msg: dict,
msg: dict[str, Any],
) -> None:
"""Get an Insteon device."""
dev_registry = dr.async_get(hass)
@ -98,7 +100,7 @@ async def websocket_get_device(
async def websocket_add_device(
hass: HomeAssistant,
connection: websocket_api.connection.ActiveConnection,
msg: dict,
msg: dict[str, Any],
) -> None:
"""Add one or more Insteon devices."""
@ -134,7 +136,7 @@ async def websocket_add_device(
async def websocket_cancel_add_device(
hass: HomeAssistant,
connection: websocket_api.connection.ActiveConnection,
msg: dict,
msg: dict[str, Any],
) -> None:
"""Cancel the Insteon all-linking process."""
await devices.async_cancel_all_linking()

View File

@ -1,5 +1,7 @@
"""Property update methods and schemas."""
from typing import Any
from pyinsteon import devices
from pyinsteon.config import RADIO_BUTTON_GROUPS, RAMP_RATE_IN_SEC, get_usable_value
from pyinsteon.constants import (
@ -158,7 +160,7 @@ def update_property(device, prop_name, value):
async def websocket_get_properties(
hass: HomeAssistant,
connection: websocket_api.connection.ActiveConnection,
msg: dict,
msg: dict[str, Any],
) -> None:
"""Add the default All-Link Database records for an Insteon device."""
if not (device := devices[msg[DEVICE_ADDRESS]]):
@ -183,7 +185,7 @@ async def websocket_get_properties(
async def websocket_change_properties_record(
hass: HomeAssistant,
connection: websocket_api.connection.ActiveConnection,
msg: dict,
msg: dict[str, Any],
) -> None:
"""Add the default All-Link Database records for an Insteon device."""
if not (device := devices[msg[DEVICE_ADDRESS]]):
@ -205,7 +207,7 @@ async def websocket_change_properties_record(
async def websocket_write_properties(
hass: HomeAssistant,
connection: websocket_api.connection.ActiveConnection,
msg: dict,
msg: dict[str, Any],
) -> None:
"""Add the default All-Link Database records for an Insteon device."""
if not (device := devices[msg[DEVICE_ADDRESS]]):
@ -235,7 +237,7 @@ async def websocket_write_properties(
async def websocket_load_properties(
hass: HomeAssistant,
connection: websocket_api.connection.ActiveConnection,
msg: dict,
msg: dict[str, Any],
) -> None:
"""Add the default All-Link Database records for an Insteon device."""
if not (device := devices[msg[DEVICE_ADDRESS]]):
@ -266,7 +268,7 @@ async def websocket_load_properties(
async def websocket_reset_properties(
hass: HomeAssistant,
connection: websocket_api.connection.ActiveConnection,
msg: dict,
msg: dict[str, Any],
) -> None:
"""Add the default All-Link Database records for an Insteon device."""
if not (device := devices[msg[DEVICE_ADDRESS]]):

View File

@ -260,7 +260,7 @@ async def _async_events_consumer(
)
@websocket_api.async_response
async def ws_event_stream(
hass: HomeAssistant, connection: websocket_api.ActiveConnection, msg: dict
hass: HomeAssistant, connection: websocket_api.ActiveConnection, msg: dict[str, Any]
) -> None:
"""Handle logbook stream events websocket command."""
start_time_str = msg["start_time"]
@ -451,7 +451,7 @@ def _ws_formatted_get_events(
)
@websocket_api.async_response
async def ws_get_events(
hass: HomeAssistant, connection: websocket_api.ActiveConnection, msg: dict
hass: HomeAssistant, connection: websocket_api.ActiveConnection, msg: dict[str, Any]
) -> None:
"""Handle logbook get events websocket command."""
start_time_str = msg["start_time"]

View File

@ -163,7 +163,7 @@ async def async_resolve_media(
)
@websocket_api.async_response
async def websocket_browse_media(
hass: HomeAssistant, connection: ActiveConnection, msg: dict
hass: HomeAssistant, connection: ActiveConnection, msg: dict[str, Any]
) -> None:
"""Browse available media."""
try:
@ -185,7 +185,7 @@ async def websocket_browse_media(
)
@websocket_api.async_response
async def websocket_resolve_media(
hass: HomeAssistant, connection: ActiveConnection, msg: dict
hass: HomeAssistant, connection: ActiveConnection, msg: dict[str, Any]
) -> None:
"""Resolve media."""
try:

View File

@ -5,6 +5,7 @@ import logging
import mimetypes
from pathlib import Path
import shutil
from typing import Any
from aiohttp import web
from aiohttp.web_request import FileField
@ -323,7 +324,7 @@ class UploadMediaView(http.HomeAssistantView):
@websocket_api.require_admin
@websocket_api.async_response
async def websocket_remove_media(
hass: HomeAssistant, connection: websocket_api.ActiveConnection, msg: dict
hass: HomeAssistant, connection: websocket_api.ActiveConnection, msg: dict[str, Any]
) -> None:
"""Remove media."""
try:

View File

@ -1,6 +1,8 @@
"""The Network Configuration integration websocket commands."""
from __future__ import annotations
from typing import Any
import voluptuous as vol
from homeassistant.components import websocket_api
@ -24,7 +26,7 @@ def async_register_websocket_commands(hass: HomeAssistant) -> None:
async def websocket_network_adapters(
hass: HomeAssistant,
connection: ActiveConnection,
msg: dict,
msg: dict[str, Any],
) -> None:
"""Return network preferences."""
network = await async_get_network(hass)
@ -48,7 +50,7 @@ async def websocket_network_adapters(
async def websocket_network_adapters_configure(
hass: HomeAssistant,
connection: ActiveConnection,
msg: dict,
msg: dict[str, Any],
) -> None:
"""Update network config."""
network = await async_get_network(hass)

View File

@ -3,7 +3,7 @@ from __future__ import annotations
from datetime import datetime as dt
import logging
from typing import Literal
from typing import Any, Literal
import voluptuous as vol
@ -135,7 +135,7 @@ async def ws_handle_get_statistics_during_period(
)
@websocket_api.async_response
async def ws_get_statistics_during_period(
hass: HomeAssistant, connection: websocket_api.ActiveConnection, msg: dict
hass: HomeAssistant, connection: websocket_api.ActiveConnection, msg: dict[str, Any]
) -> None:
"""Handle statistics websocket command."""
await ws_handle_get_statistics_during_period(hass, connection, msg)
@ -174,7 +174,7 @@ async def ws_handle_list_statistic_ids(
)
@websocket_api.async_response
async def ws_list_statistic_ids(
hass: HomeAssistant, connection: websocket_api.ActiveConnection, msg: dict
hass: HomeAssistant, connection: websocket_api.ActiveConnection, msg: dict[str, Any]
) -> None:
"""Fetch a list of available statistic_id."""
await ws_handle_list_statistic_ids(hass, connection, msg)
@ -187,7 +187,7 @@ async def ws_list_statistic_ids(
)
@websocket_api.async_response
async def ws_validate_statistics(
hass: HomeAssistant, connection: websocket_api.ActiveConnection, msg: dict
hass: HomeAssistant, connection: websocket_api.ActiveConnection, msg: dict[str, Any]
) -> None:
"""Fetch a list of available statistic_id."""
instance = get_instance(hass)
@ -226,7 +226,7 @@ def ws_clear_statistics(
)
@websocket_api.async_response
async def ws_get_statistics_metadata(
hass: HomeAssistant, connection: websocket_api.ActiveConnection, msg: dict
hass: HomeAssistant, connection: websocket_api.ActiveConnection, msg: dict[str, Any]
) -> None:
"""Get metadata for a list of statistic_ids."""
instance = get_instance(hass)
@ -296,7 +296,7 @@ def ws_change_statistics_unit(
)
@websocket_api.async_response
async def ws_adjust_sum_statistics(
hass: HomeAssistant, connection: websocket_api.ActiveConnection, msg: dict
hass: HomeAssistant, connection: websocket_api.ActiveConnection, msg: dict[str, Any]
) -> None:
"""Adjust sum statistics.
@ -417,7 +417,7 @@ def ws_info(
@websocket_api.websocket_command({vol.Required("type"): "backup/start"})
@websocket_api.async_response
async def ws_backup_start(
hass: HomeAssistant, connection: websocket_api.ActiveConnection, msg: dict
hass: HomeAssistant, connection: websocket_api.ActiveConnection, msg: dict[str, Any]
) -> None:
"""Backup start notification."""
@ -435,7 +435,7 @@ async def ws_backup_start(
@websocket_api.websocket_command({vol.Required("type"): "backup/end"})
@websocket_api.async_response
async def ws_backup_end(
hass: HomeAssistant, connection: websocket_api.ActiveConnection, msg: dict
hass: HomeAssistant, connection: websocket_api.ActiveConnection, msg: dict[str, Any]
) -> None:
"""Backup end notification."""

View File

@ -1,6 +1,7 @@
"""Support to manage a shopping list."""
from http import HTTPStatus
import logging
from typing import Any
import uuid
import voluptuous as vol
@ -372,7 +373,7 @@ def websocket_handle_items(
async def websocket_handle_add(
hass: HomeAssistant,
connection: websocket_api.connection.ActiveConnection,
msg: dict,
msg: dict[str, Any],
) -> None:
"""Handle add item to shopping_list."""
item = await hass.data[DOMAIN].async_add(msg["name"], connection.context(msg))
@ -383,7 +384,7 @@ async def websocket_handle_add(
async def websocket_handle_update(
hass: HomeAssistant,
connection: websocket_api.connection.ActiveConnection,
msg: dict,
msg: dict[str, Any],
) -> None:
"""Handle update shopping_list item."""
msg_id = msg.pop("id")
@ -406,7 +407,7 @@ async def websocket_handle_update(
async def websocket_handle_clear(
hass: HomeAssistant,
connection: websocket_api.connection.ActiveConnection,
msg: dict,
msg: dict[str, Any],
) -> None:
"""Handle clearing shopping_list items."""
await hass.data[DOMAIN].async_clear_completed(connection.context(msg))

View File

@ -315,7 +315,7 @@ class USBDiscovery:
async def websocket_usb_scan(
hass: HomeAssistant,
connection: ActiveConnection,
msg: dict,
msg: dict[str, Any],
) -> None:
"""Scan for new usb devices."""
usb_discovery: USBDiscovery = hass.data[DOMAIN]

View File

@ -448,7 +448,7 @@ def async_register_api(hass: HomeAssistant) -> None:
)
@websocket_api.async_response
async def websocket_network_status(
hass: HomeAssistant, connection: ActiveConnection, msg: dict
hass: HomeAssistant, connection: ActiveConnection, msg: dict[str, Any]
) -> None:
"""Get the status of the Z-Wave JS network."""
if ENTRY_ID in msg:
@ -518,7 +518,7 @@ async def websocket_network_status(
async def websocket_subscribe_node_status(
hass: HomeAssistant,
connection: ActiveConnection,
msg: dict,
msg: dict[str, Any],
node: Node,
) -> None:
"""Subscribe to node status update events of a Z-Wave JS node."""
@ -559,7 +559,7 @@ async def websocket_subscribe_node_status(
async def websocket_node_status(
hass: HomeAssistant,
connection: ActiveConnection,
msg: dict,
msg: dict[str, Any],
node: Node,
) -> None:
"""Get the status of a Z-Wave JS node."""
@ -577,7 +577,7 @@ async def websocket_node_status(
async def websocket_node_metadata(
hass: HomeAssistant,
connection: ActiveConnection,
msg: dict,
msg: dict[str, Any],
node: Node,
) -> None:
"""Get the metadata of a Z-Wave JS node."""
@ -607,7 +607,7 @@ async def websocket_node_metadata(
async def websocket_node_comments(
hass: HomeAssistant,
connection: ActiveConnection,
msg: dict,
msg: dict[str, Any],
node: Node,
) -> None:
"""Get the comments of a Z-Wave JS node."""
@ -648,7 +648,7 @@ async def websocket_node_comments(
async def websocket_add_node(
hass: HomeAssistant,
connection: ActiveConnection,
msg: dict,
msg: dict[str, Any],
entry: ConfigEntry,
client: Client,
driver: Driver,
@ -791,7 +791,7 @@ async def websocket_add_node(
async def websocket_grant_security_classes(
hass: HomeAssistant,
connection: ActiveConnection,
msg: dict,
msg: dict[str, Any],
entry: ConfigEntry,
client: Client,
driver: Driver,
@ -819,7 +819,7 @@ async def websocket_grant_security_classes(
async def websocket_validate_dsk_and_enter_pin(
hass: HomeAssistant,
connection: ActiveConnection,
msg: dict,
msg: dict[str, Any],
entry: ConfigEntry,
client: Client,
driver: Driver,
@ -849,7 +849,7 @@ async def websocket_validate_dsk_and_enter_pin(
async def websocket_provision_smart_start_node(
hass: HomeAssistant,
connection: ActiveConnection,
msg: dict,
msg: dict[str, Any],
entry: ConfigEntry,
client: Client,
driver: Driver,
@ -902,7 +902,7 @@ async def websocket_provision_smart_start_node(
async def websocket_unprovision_smart_start_node(
hass: HomeAssistant,
connection: ActiveConnection,
msg: dict,
msg: dict[str, Any],
entry: ConfigEntry,
client: Client,
driver: Driver,
@ -935,7 +935,7 @@ async def websocket_unprovision_smart_start_node(
async def websocket_get_provisioning_entries(
hass: HomeAssistant,
connection: ActiveConnection,
msg: dict,
msg: dict[str, Any],
entry: ConfigEntry,
client: Client,
driver: Driver,
@ -961,7 +961,7 @@ async def websocket_get_provisioning_entries(
async def websocket_parse_qr_code_string(
hass: HomeAssistant,
connection: ActiveConnection,
msg: dict,
msg: dict[str, Any],
entry: ConfigEntry,
client: Client,
driver: Driver,
@ -987,7 +987,7 @@ async def websocket_parse_qr_code_string(
async def websocket_supports_feature(
hass: HomeAssistant,
connection: ActiveConnection,
msg: dict,
msg: dict[str, Any],
entry: ConfigEntry,
client: Client,
driver: Driver,
@ -1013,7 +1013,7 @@ async def websocket_supports_feature(
async def websocket_stop_inclusion(
hass: HomeAssistant,
connection: ActiveConnection,
msg: dict,
msg: dict[str, Any],
entry: ConfigEntry,
client: Client,
driver: Driver,
@ -1040,7 +1040,7 @@ async def websocket_stop_inclusion(
async def websocket_stop_exclusion(
hass: HomeAssistant,
connection: ActiveConnection,
msg: dict,
msg: dict[str, Any],
entry: ConfigEntry,
client: Client,
driver: Driver,
@ -1068,7 +1068,7 @@ async def websocket_stop_exclusion(
async def websocket_remove_node(
hass: HomeAssistant,
connection: ActiveConnection,
msg: dict,
msg: dict[str, Any],
entry: ConfigEntry,
client: Client,
driver: Driver,
@ -1147,7 +1147,7 @@ async def websocket_remove_node(
async def websocket_replace_failed_node(
hass: HomeAssistant,
connection: ActiveConnection,
msg: dict,
msg: dict[str, Any],
node: Node,
) -> None:
"""Replace a failed node with a new node."""
@ -1298,7 +1298,7 @@ async def websocket_replace_failed_node(
async def websocket_remove_failed_node(
hass: HomeAssistant,
connection: ActiveConnection,
msg: dict,
msg: dict[str, Any],
node: Node,
) -> None:
"""Remove a failed node from the Z-Wave network."""
@ -1342,7 +1342,7 @@ async def websocket_remove_failed_node(
async def websocket_begin_healing_network(
hass: HomeAssistant,
connection: ActiveConnection,
msg: dict,
msg: dict[str, Any],
entry: ConfigEntry,
client: Client,
driver: Driver,
@ -1369,7 +1369,7 @@ async def websocket_begin_healing_network(
async def websocket_subscribe_heal_network_progress(
hass: HomeAssistant,
connection: ActiveConnection,
msg: dict,
msg: dict[str, Any],
entry: ConfigEntry,
client: Client,
driver: Driver,
@ -1413,7 +1413,7 @@ async def websocket_subscribe_heal_network_progress(
async def websocket_stop_healing_network(
hass: HomeAssistant,
connection: ActiveConnection,
msg: dict,
msg: dict[str, Any],
entry: ConfigEntry,
client: Client,
driver: Driver,
@ -1440,7 +1440,7 @@ async def websocket_stop_healing_network(
async def websocket_heal_node(
hass: HomeAssistant,
connection: ActiveConnection,
msg: dict,
msg: dict[str, Any],
node: Node,
) -> None:
"""Heal a node on the Z-Wave network."""
@ -1468,7 +1468,7 @@ async def websocket_heal_node(
async def websocket_refresh_node_info(
hass: HomeAssistant,
connection: ActiveConnection,
msg: dict,
msg: dict[str, Any],
node: Node,
) -> None:
"""Re-interview a node."""
@ -1518,7 +1518,7 @@ async def websocket_refresh_node_info(
async def websocket_refresh_node_values(
hass: HomeAssistant,
connection: ActiveConnection,
msg: dict,
msg: dict[str, Any],
node: Node,
) -> None:
"""Refresh node values."""
@ -1540,7 +1540,7 @@ async def websocket_refresh_node_values(
async def websocket_refresh_node_cc_values(
hass: HomeAssistant,
connection: ActiveConnection,
msg: dict,
msg: dict[str, Any],
node: Node,
) -> None:
"""Refresh node values for a particular CommandClass."""
@ -1574,7 +1574,7 @@ async def websocket_refresh_node_cc_values(
async def websocket_set_config_parameter(
hass: HomeAssistant,
connection: ActiveConnection,
msg: dict,
msg: dict[str, Any],
node: Node,
) -> None:
"""Set a config parameter value for a Z-Wave node."""
@ -1619,7 +1619,7 @@ async def websocket_set_config_parameter(
@websocket_api.async_response
@async_get_node
async def websocket_get_config_parameters(
hass: HomeAssistant, connection: ActiveConnection, msg: dict, node: Node
hass: HomeAssistant, connection: ActiveConnection, msg: dict[str, Any], node: Node
) -> None:
"""Get a list of configuration parameters for a Z-Wave node."""
values = node.get_configuration_values()
@ -1671,7 +1671,7 @@ def filename_is_present_if_logging_to_file(obj: dict) -> dict:
async def websocket_subscribe_log_updates(
hass: HomeAssistant,
connection: ActiveConnection,
msg: dict,
msg: dict[str, Any],
entry: ConfigEntry,
client: Client,
driver: Driver,
@ -1758,7 +1758,7 @@ async def websocket_subscribe_log_updates(
async def websocket_update_log_config(
hass: HomeAssistant,
connection: ActiveConnection,
msg: dict,
msg: dict[str, Any],
entry: ConfigEntry,
client: Client,
driver: Driver,
@ -1782,7 +1782,7 @@ async def websocket_update_log_config(
async def websocket_get_log_config(
hass: HomeAssistant,
connection: ActiveConnection,
msg: dict,
msg: dict[str, Any],
entry: ConfigEntry,
client: Client,
driver: Driver,
@ -1809,7 +1809,7 @@ async def websocket_get_log_config(
async def websocket_update_data_collection_preference(
hass: HomeAssistant,
connection: ActiveConnection,
msg: dict,
msg: dict[str, Any],
entry: ConfigEntry,
client: Client,
driver: Driver,
@ -1841,7 +1841,7 @@ async def websocket_update_data_collection_preference(
async def websocket_data_collection_status(
hass: HomeAssistant,
connection: ActiveConnection,
msg: dict,
msg: dict[str, Any],
entry: ConfigEntry,
client: Client,
driver: Driver,
@ -1868,7 +1868,7 @@ async def websocket_data_collection_status(
async def websocket_abort_firmware_update(
hass: HomeAssistant,
connection: ActiveConnection,
msg: dict,
msg: dict[str, Any],
node: Node,
) -> None:
"""Abort a firmware update."""
@ -1889,7 +1889,7 @@ async def websocket_abort_firmware_update(
async def websocket_is_node_firmware_update_in_progress(
hass: HomeAssistant,
connection: ActiveConnection,
msg: dict,
msg: dict[str, Any],
node: Node,
) -> None:
"""Get whether firmware update is in progress for given node."""
@ -1921,7 +1921,7 @@ def _get_firmware_update_progress_dict(
async def websocket_subscribe_firmware_update_status(
hass: HomeAssistant,
connection: ActiveConnection,
msg: dict,
msg: dict[str, Any],
node: Node,
) -> None:
"""Subscribe to the status of a firmware update."""
@ -1994,7 +1994,7 @@ async def websocket_subscribe_firmware_update_status(
async def websocket_get_firmware_update_capabilities(
hass: HomeAssistant,
connection: ActiveConnection,
msg: dict,
msg: dict[str, Any],
node: Node,
) -> None:
"""Abort a firmware update."""
@ -2015,7 +2015,7 @@ async def websocket_get_firmware_update_capabilities(
async def websocket_is_any_ota_firmware_update_in_progress(
hass: HomeAssistant,
connection: ActiveConnection,
msg: dict,
msg: dict[str, Any],
entry: ConfigEntry,
client: Client,
driver: Driver,
@ -2092,7 +2092,7 @@ class FirmwareUploadView(HomeAssistantView):
async def websocket_check_for_config_updates(
hass: HomeAssistant,
connection: ActiveConnection,
msg: dict,
msg: dict[str, Any],
entry: ConfigEntry,
client: Client,
driver: Driver,
@ -2121,7 +2121,7 @@ async def websocket_check_for_config_updates(
async def websocket_install_config_update(
hass: HomeAssistant,
connection: ActiveConnection,
msg: dict,
msg: dict[str, Any],
entry: ConfigEntry,
client: Client,
driver: Driver,
@ -2160,7 +2160,7 @@ def _get_controller_statistics_dict(
async def websocket_subscribe_controller_statistics(
hass: HomeAssistant,
connection: ActiveConnection,
msg: dict,
msg: dict[str, Any],
entry: ConfigEntry,
client: Client,
driver: Driver,
@ -2257,7 +2257,7 @@ def _get_node_statistics_dict(
async def websocket_subscribe_node_statistics(
hass: HomeAssistant,
connection: ActiveConnection,
msg: dict,
msg: dict[str, Any],
node: Node,
) -> None:
"""Subsribe to the statistics updates for a node."""