diff --git a/scripts/generate_multilevel_sensor_constants.py b/scripts/generate_multilevel_sensor_constants.py index fb048a43e..91c946c5e 100755 --- a/scripts/generate_multilevel_sensor_constants.py +++ b/scripts/generate_multilevel_sensor_constants.py @@ -157,6 +157,8 @@ def generate_int_enum_base_class(class_name: str, docstring: str) -> list[str]: '# THE "END OF AUTOGENERATED CONTENT" COMMENT BLOCK AND ADD YOUR CODE BELOW IT) #', "# ----------------------------------------------------------------------------------- #", "", + "from __future__ import annotations", + "", "from enum import IntEnum", 'CC_SPECIFIC_SCALE = "scale"', 'CC_SPECIFIC_SENSOR_TYPE = "sensorType"', diff --git a/scripts/run_mock_server.py b/scripts/run_mock_server.py index 89b31a23e..2b12facb8 100644 --- a/scripts/run_mock_server.py +++ b/scripts/run_mock_server.py @@ -1,11 +1,13 @@ """Run a mock zwave-js-server instance off of a network state dump.""" +from __future__ import annotations + import argparse import asyncio import json import logging from collections import defaultdict from collections.abc import Hashable -from typing import Any, Optional, Union +from typing import Any from aiohttp import WSMsgType, web, web_request @@ -58,7 +60,7 @@ def __init__( web.post("/replay", self.replay_handler), ] ) - self.primary_ws_resp: Optional[web.WebSocketResponse] = None + self.primary_ws_resp: web.WebSocketResponse | None = None self.events_to_replay = events_to_replay self.command_results = command_results @@ -77,9 +79,7 @@ async def send_command_result( await self.send_json({**data, "messageId": message_id}) async def send_success_command_result( - self, - result: Optional[dict], - message_id: str, + self, result: dict | None, message_id: str ) -> None: """Send success message.""" if result is None: @@ -194,7 +194,7 @@ async def replay_handler(self, request: web_request.Request) -> web.Response: return web.Response(status=200) -def _hashable_value(item: Union[dict, list, Hashable]) -> Union[tuple, list, Hashable]: +def _hashable_value(item: dict | list | Hashable) -> tuple | list | Hashable: """Return hashable value from item.""" if isinstance(item, dict): return make_dict_hashable(item) diff --git a/zwave_js_server/__main__.py b/zwave_js_server/__main__.py index 74098f5da..52852ee40 100644 --- a/zwave_js_server/__main__.py +++ b/zwave_js_server/__main__.py @@ -1,4 +1,6 @@ """Basic CLI to test Z-Wave JS server.""" +from __future__ import annotations + import argparse import asyncio import logging diff --git a/zwave_js_server/client.py b/zwave_js_server/client.py index d9b52a6fc..f011465e0 100644 --- a/zwave_js_server/client.py +++ b/zwave_js_server/client.py @@ -1,4 +1,6 @@ """Client.""" +from __future__ import annotations + import asyncio import logging import pprint @@ -8,7 +10,7 @@ from datetime import datetime from operator import itemgetter from types import TracebackType -from typing import Any, Optional, cast +from typing import Any, cast from aiohttp import ClientSession, ClientWebSocketResponse, WSMsgType, client_exceptions @@ -55,17 +57,17 @@ def __init__( ws_server_url: str, aiohttp_session: ClientSession, schema_version: int = MAX_SERVER_SCHEMA_VERSION, - additional_user_agent_components: Optional[dict[str, str]] = None, + additional_user_agent_components: dict[str, str] | None = None, record_messages: bool = False, ): """Initialize the Client class.""" self.ws_server_url = ws_server_url self.aiohttp_session = aiohttp_session - self.driver: Optional[Driver] = None + self.driver: Driver | None = None # The WebSocket client - self._client: Optional[ClientWebSocketResponse] = None + self._client: ClientWebSocketResponse | None = None # Version of the connected server - self.version: Optional[VersionInfo] = None + self.version: VersionInfo | None = None self.schema_version: int = schema_version self.additional_user_agent_components = { PACKAGE_NAME: __version__, @@ -74,7 +76,7 @@ def __init__( self._logger = logging.getLogger(__package__) self._loop = asyncio.get_running_loop() self._result_futures: dict[str, asyncio.Future] = {} - self._shutdown_complete_event: Optional[asyncio.Event] = None + self._shutdown_complete_event: asyncio.Event | None = None self._record_messages = record_messages self._recorded_commands: defaultdict[str, dict] = defaultdict(dict) self._recorded_events: list[dict] = [] @@ -95,9 +97,7 @@ def recording_messages(self) -> bool: return self._record_messages async def async_send_command( - self, - message: dict[str, Any], - require_schema: Optional[int] = None, + self, message: dict[str, Any], require_schema: int | None = None ) -> dict: """Send a command and get a response.""" if require_schema is not None and require_schema > self.schema_version: @@ -118,7 +118,7 @@ async def async_send_command( self._result_futures.pop(message_id) async def async_send_command_no_wait( - self, message: dict[str, Any], require_schema: Optional[int] = None + self, message: dict[str, Any], require_schema: int | None = None ) -> None: """Send a command without waiting for the response.""" if require_schema is not None and require_schema > self.schema_version: diff --git a/zwave_js_server/const/__init__.py b/zwave_js_server/const/__init__.py index e894c02b2..8e29461df 100644 --- a/zwave_js_server/const/__init__.py +++ b/zwave_js_server/const/__init__.py @@ -1,4 +1,6 @@ """Constants for the Z-Wave JS python library.""" +from __future__ import annotations + from enum import Enum, IntEnum from importlib import metadata diff --git a/zwave_js_server/const/command_class/barrier_operator.py b/zwave_js_server/const/command_class/barrier_operator.py index d19d18a65..ca232df59 100644 --- a/zwave_js_server/const/command_class/barrier_operator.py +++ b/zwave_js_server/const/command_class/barrier_operator.py @@ -1,4 +1,6 @@ """Constants for the Barrier Operator CC.""" +from __future__ import annotations + from enum import IntEnum SIGNALING_STATE_PROPERTY = "signalingState" diff --git a/zwave_js_server/const/command_class/color_switch.py b/zwave_js_server/const/command_class/color_switch.py index f3b92db21..85b707bed 100644 --- a/zwave_js_server/const/command_class/color_switch.py +++ b/zwave_js_server/const/command_class/color_switch.py @@ -1,4 +1,6 @@ """Constants for the Color Switch CC.""" +from __future__ import annotations + from enum import IntEnum diff --git a/zwave_js_server/const/command_class/entry_control.py b/zwave_js_server/const/command_class/entry_control.py index 6eb433369..18c17d3f4 100644 --- a/zwave_js_server/const/command_class/entry_control.py +++ b/zwave_js_server/const/command_class/entry_control.py @@ -1,4 +1,6 @@ """Constants for the Entry Control CC.""" +from __future__ import annotations + from enum import IntEnum diff --git a/zwave_js_server/const/command_class/humidity_control.py b/zwave_js_server/const/command_class/humidity_control.py index acfa83c83..48815d410 100644 --- a/zwave_js_server/const/command_class/humidity_control.py +++ b/zwave_js_server/const/command_class/humidity_control.py @@ -4,6 +4,8 @@ Includes Humidity Control Mode, Humidity Control Operating State, and Humidity Control Setpoint CCs. """ +from __future__ import annotations + from enum import IntEnum HUMIDITY_CONTROL_MODE_PROPERTY = "mode" diff --git a/zwave_js_server/const/command_class/lock.py b/zwave_js_server/const/command_class/lock.py index 7fd062c76..c9c66cf61 100644 --- a/zwave_js_server/const/command_class/lock.py +++ b/zwave_js_server/const/command_class/lock.py @@ -3,6 +3,8 @@ Includes Door Lock and Lock CCs. """ +from __future__ import annotations + from enum import IntEnum from .. import CommandClass diff --git a/zwave_js_server/const/command_class/meter.py b/zwave_js_server/const/command_class/meter.py index cd2cd0d78..7146970ef 100644 --- a/zwave_js_server/const/command_class/meter.py +++ b/zwave_js_server/const/command_class/meter.py @@ -1,4 +1,6 @@ """Constants for Meter CC.""" +from __future__ import annotations + from enum import IntEnum VALUE_PROPERTY = "value" diff --git a/zwave_js_server/const/command_class/multilevel_sensor.py b/zwave_js_server/const/command_class/multilevel_sensor.py index d91ae6328..22c228b0a 100644 --- a/zwave_js_server/const/command_class/multilevel_sensor.py +++ b/zwave_js_server/const/command_class/multilevel_sensor.py @@ -5,6 +5,8 @@ # THE "END OF AUTOGENERATED CONTENT" COMMENT BLOCK AND ADD YOUR CODE BELOW IT) # # ----------------------------------------------------------------------------------- # +from __future__ import annotations + from enum import IntEnum CC_SPECIFIC_SCALE = "scale" diff --git a/zwave_js_server/const/command_class/multilevel_switch.py b/zwave_js_server/const/command_class/multilevel_switch.py index 8d9690273..066b6b23f 100644 --- a/zwave_js_server/const/command_class/multilevel_switch.py +++ b/zwave_js_server/const/command_class/multilevel_switch.py @@ -1,4 +1,6 @@ """Constants for the Multilevel Switch CC.""" +from __future__ import annotations + from enum import IntEnum COVER_OPEN_PROPERTY = "Open" diff --git a/zwave_js_server/const/command_class/notification.py b/zwave_js_server/const/command_class/notification.py index 176feb41e..964e256c1 100644 --- a/zwave_js_server/const/command_class/notification.py +++ b/zwave_js_server/const/command_class/notification.py @@ -1,3 +1,4 @@ """Constants for the Notification CC.""" +from __future__ import annotations CC_SPECIFIC_NOTIFICATION_TYPE = "notificationType" diff --git a/zwave_js_server/const/command_class/power_level.py b/zwave_js_server/const/command_class/power_level.py index e947eb4a6..234e8f02d 100644 --- a/zwave_js_server/const/command_class/power_level.py +++ b/zwave_js_server/const/command_class/power_level.py @@ -1,4 +1,6 @@ """Constants for the Power Level Command Class.""" +from __future__ import annotations + from enum import IntEnum diff --git a/zwave_js_server/const/command_class/protection.py b/zwave_js_server/const/command_class/protection.py index 148b624be..4fb808550 100644 --- a/zwave_js_server/const/command_class/protection.py +++ b/zwave_js_server/const/command_class/protection.py @@ -1,4 +1,5 @@ """Constants for the Protection CC.""" +from __future__ import annotations LOCAL_PROPERTY = "local" RF_PROPERTY = "rf" diff --git a/zwave_js_server/const/command_class/sound_switch.py b/zwave_js_server/const/command_class/sound_switch.py index 0af65980b..792ae77dd 100644 --- a/zwave_js_server/const/command_class/sound_switch.py +++ b/zwave_js_server/const/command_class/sound_switch.py @@ -1,4 +1,6 @@ """Constants for the Sound Switch CC.""" +from __future__ import annotations + from enum import IntEnum TONE_ID_PROPERTY = "toneId" diff --git a/zwave_js_server/const/command_class/thermostat.py b/zwave_js_server/const/command_class/thermostat.py index 6639e25a0..dbe8f11cd 100644 --- a/zwave_js_server/const/command_class/thermostat.py +++ b/zwave_js_server/const/command_class/thermostat.py @@ -4,6 +4,8 @@ Includes Thermostat Fan Mode, Thermostat Fan State, Thermostat Mode, Thermostat Operating State, Thermostat Setback, and Thermostat Setpoint CCs. """ +from __future__ import annotations + from enum import IntEnum THERMOSTAT_MODE_PROPERTY = "mode" diff --git a/zwave_js_server/const/command_class/wake_up.py b/zwave_js_server/const/command_class/wake_up.py index 11798b2f2..5fdbc618c 100644 --- a/zwave_js_server/const/command_class/wake_up.py +++ b/zwave_js_server/const/command_class/wake_up.py @@ -1,3 +1,5 @@ """Constants for the Wake Up CC.""" +from __future__ import annotations + WAKE_UP_INTERVAL_PROPERTY = "wakeUpInterval" WAKE_UP_CONTROLLER_NODE_ID_PROPERTY = "controllerNodeId" diff --git a/zwave_js_server/dump.py b/zwave_js_server/dump.py index 652c41570..06dcdca1d 100644 --- a/zwave_js_server/dump.py +++ b/zwave_js_server/dump.py @@ -1,6 +1,7 @@ """Dump helper.""" +from __future__ import annotations + import asyncio -from typing import Optional import aiohttp @@ -11,8 +12,8 @@ async def dump_msgs( url: str, session: aiohttp.ClientSession, - additional_user_agent_components: Optional[dict[str, str]] = None, - timeout: Optional[float] = None, + additional_user_agent_components: dict[str, str] | None = None, + timeout: float | None = None, ) -> list[dict]: """Dump server state.""" client = await session.ws_connect(url, compress=15, max_msg_size=0) diff --git a/zwave_js_server/event.py b/zwave_js_server/event.py index b25e729d2..b6db117de 100644 --- a/zwave_js_server/event.py +++ b/zwave_js_server/event.py @@ -1,4 +1,6 @@ """Provide Event base classes for Z-Wave JS.""" +from __future__ import annotations + import logging from dataclasses import dataclass, field from typing import Callable, Literal diff --git a/zwave_js_server/exceptions.py b/zwave_js_server/exceptions.py index ebf68b7eb..f345e6c12 100644 --- a/zwave_js_server/exceptions.py +++ b/zwave_js_server/exceptions.py @@ -1,5 +1,7 @@ """Exceptions for zwave-js-server.""" -from typing import TYPE_CHECKING, Optional +from __future__ import annotations + +from typing import TYPE_CHECKING from .const import RssiError @@ -16,7 +18,7 @@ class BaseZwaveJSServerError(Exception): class TransportError(BaseZwaveJSServerError): """Exception raised to represent transport errors.""" - def __init__(self, message: str, error: Optional[Exception] = None) -> None: + def __init__(self, message: str, error: Exception | None = None) -> None: """Initialize a transport error.""" super().__init__(message) self.error = error @@ -37,7 +39,7 @@ def __init__(self, error: Exception) -> None: class ConnectionFailed(TransportError): """Exception raised when an established connection fails.""" - def __init__(self, error: Optional[Exception] = None) -> None: + def __init__(self, error: Exception | None = None) -> None: """Initialize a connection failed error.""" if error is None: super().__init__("Connection failed.") @@ -80,7 +82,9 @@ def __init__( class FailedCommand(BaseZwaveJSServerError): """When a command has failed.""" - def __init__(self, message_id: str, error_code: str, msg: Optional[str] = None): + def __init__( + self, message_id: str, error_code: str, msg: str | None = None + ) -> None: """Initialize a failed command error.""" super().__init__(msg or f"Command failed: {error_code}") self.message_id = message_id @@ -91,10 +95,7 @@ class FailedZWaveCommand(FailedCommand): """When a command has failed because of Z-Wave JS error.""" def __init__( - self, - message_id: str, - zwave_error_code: int, - zwave_error_message: str, + self, message_id: str, zwave_error_code: int, zwave_error_message: str ): """Initialize a failed command error.""" super().__init__( diff --git a/zwave_js_server/firmware.py b/zwave_js_server/firmware.py index c8b835c2a..c7f443cb6 100644 --- a/zwave_js_server/firmware.py +++ b/zwave_js_server/firmware.py @@ -1,6 +1,8 @@ """Firmware update helper.""" +from __future__ import annotations + import asyncio -from typing import Optional, cast +from typing import cast import aiohttp @@ -15,7 +17,7 @@ async def update_firmware( node: Node, updates: list[NodeFirmwareUpdateData], session: aiohttp.ClientSession, - additional_user_agent_components: Optional[dict[str, str]] = None, + additional_user_agent_components: dict[str, str] | None = None, ) -> bool: """Send updateFirmware command to Node.""" client = Client( @@ -44,7 +46,7 @@ async def controller_firmware_update_otw( url: str, firmware_file: ControllerFirmwareUpdateData, session: aiohttp.ClientSession, - additional_user_agent_components: Optional[dict[str, str]] = None, + additional_user_agent_components: dict[str, str] | None = None, ) -> bool: """ Send firmwareUpdateOTW command to Controller. diff --git a/zwave_js_server/model/association.py b/zwave_js_server/model/association.py index c46fe2e47..96de491b3 100644 --- a/zwave_js_server/model/association.py +++ b/zwave_js_server/model/association.py @@ -1,6 +1,7 @@ """Provide a model for the association.""" +from __future__ import annotations + from dataclasses import dataclass, field -from typing import Optional @dataclass @@ -11,7 +12,7 @@ class AssociationGroup: is_lifeline: bool multi_channel: bool label: str - profile: Optional[int] = None + profile: int | None = None issued_commands: dict[int, list[int]] = field(default_factory=dict) @@ -20,4 +21,4 @@ class AssociationAddress: """Represent a association dict type.""" node_id: int - endpoint: Optional[int] = None + endpoint: int | None = None diff --git a/zwave_js_server/model/command_class.py b/zwave_js_server/model/command_class.py index 95d670fbe..70444a110 100644 --- a/zwave_js_server/model/command_class.py +++ b/zwave_js_server/model/command_class.py @@ -3,6 +3,8 @@ https://zwave-js.github.io/node-zwave-js/#/api/endpoint?id=commandclasses """ +from __future__ import annotations + from typing import TypedDict from ..const import CommandClass diff --git a/zwave_js_server/model/controller/__init__.py b/zwave_js_server/model/controller/__init__.py index a38d17faa..273d1acab 100644 --- a/zwave_js_server/model/controller/__init__.py +++ b/zwave_js_server/model/controller/__init__.py @@ -1,6 +1,8 @@ """Provide a model for the Z-Wave JS controller.""" +from __future__ import annotations + from dataclasses import dataclass -from typing import TYPE_CHECKING, Any, Literal, Optional, Union, cast +from typing import TYPE_CHECKING, Any, Literal, cast from zwave_js_server.model.node.firmware import ( NodeFirmwareUpdateFileInfo, @@ -51,11 +53,9 @@ def __init__(self, client: "Client", state: dict) -> None: super().__init__() self.client = client self.nodes: dict[int, Node] = {} - self._heal_network_progress: Optional[dict[int, str]] = None + self._heal_network_progress: dict[int, str] | None = None self._statistics = ControllerStatistics() - self._firmware_update_progress: Optional[ - ControllerFirmwareUpdateProgress - ] = None + self._firmware_update_progress: ControllerFirmwareUpdateProgress | None = None for node_state in state["nodes"]: node = Node(client, node_state) self.nodes[node.node_id] = node @@ -76,81 +76,81 @@ def __eq__(self, other: object) -> bool: return self.home_id == other.home_id @property - def sdk_version(self) -> Optional[str]: + def sdk_version(self) -> str | None: """Return sdk_version.""" return self.data.get("sdkVersion") @property - def controller_type(self) -> Optional[int]: + def controller_type(self) -> int | None: """Return controller_type.""" return self.data.get("type") @property - def home_id(self) -> Optional[int]: + def home_id(self) -> int | None: """Return home_id.""" return self.data.get("homeId") @property - def own_node_id(self) -> Optional[int]: + def own_node_id(self) -> int | None: """Return own_node_id.""" return self.data.get("ownNodeId") @property - def own_node(self) -> Optional[Node]: + def own_node(self) -> Node | None: """Return own_node.""" if self.own_node_id is None: return None return self.nodes.get(self.own_node_id) @property - def is_primary(self) -> Optional[bool]: + def is_primary(self) -> bool | None: """Return is_primary.""" return self.data.get("isPrimary") @property - def is_using_home_id_from_other_network(self) -> Optional[bool]: + def is_using_home_id_from_other_network(self) -> bool | None: """Return is_using_home_id_from_other_network.""" return self.data.get("isUsingHomeIdFromOtherNetwork") @property - def is_SIS_present(self) -> Optional[bool]: # pylint: disable=invalid-name + def is_SIS_present(self) -> bool | None: # pylint: disable=invalid-name """Return is_SIS_present.""" return self.data.get("isSISPresent") @property - def was_real_primary(self) -> Optional[bool]: + def was_real_primary(self) -> bool | None: """Return was_real_primary.""" return self.data.get("wasRealPrimary") @property - def is_suc(self) -> Optional[bool]: + def is_suc(self) -> bool | None: """Return is_suc.""" return self.data.get("isSUC") @property - def node_type(self) -> Optional[NodeType]: + def node_type(self) -> NodeType | None: """Return node_type.""" if (node_type := self.data.get("nodeType")) is not None: return NodeType(node_type) return None @property - def firmware_version(self) -> Optional[str]: + def firmware_version(self) -> str | None: """Return firmware_version.""" return self.data.get("firmwareVersion") @property - def manufacturer_id(self) -> Optional[int]: + def manufacturer_id(self) -> int | None: """Return manufacturer_id.""" return self.data.get("manufacturerId") @property - def product_type(self) -> Optional[int]: + def product_type(self) -> int | None: """Return product_type.""" return self.data.get("productType") @property - def product_id(self) -> Optional[int]: + def product_id(self) -> int | None: """Return product_id.""" return self.data.get("productId") @@ -160,17 +160,17 @@ def supported_function_types(self) -> list[int]: return self.data.get("supportedFunctionTypes", []) @property - def suc_node_id(self) -> Optional[int]: + def suc_node_id(self) -> int | None: """Return suc_node_id.""" return self.data.get("sucNodeId") @property - def supports_timers(self) -> Optional[bool]: + def supports_timers(self) -> bool | None: """Return supports_timers.""" return self.data.get("supportsTimers") @property - def is_heal_network_active(self) -> Optional[bool]: + def is_heal_network_active(self) -> bool | None: """Return is_heal_network_active.""" return self.data.get("isHealNetworkActive") @@ -180,7 +180,7 @@ def statistics(self) -> ControllerStatistics: return self._statistics @property - def heal_network_progress(self) -> Optional[dict[int, str]]: + def heal_network_progress(self) -> dict[int, str] | None: """Return heal network progress state.""" return self._heal_network_progress @@ -190,14 +190,14 @@ def inclusion_state(self) -> InclusionState: return InclusionState(self.data["inclusionState"]) @property - def rf_region(self) -> Optional[RFRegion]: + def rf_region(self) -> RFRegion | None: """Return RF region of controller.""" if (rf_region := self.data.get("rfRegion")) is None: return None return RFRegion(rf_region) @property - def firmware_update_progress(self) -> Optional[ControllerFirmwareUpdateProgress]: + def firmware_update_progress(self) -> ControllerFirmwareUpdateProgress | None: """Return firmware update progress.""" return self._firmware_update_progress @@ -214,11 +214,9 @@ async def async_begin_inclusion( InclusionStrategy.SECURITY_S2, InclusionStrategy.INSECURE, ], - force_security: Optional[bool] = None, - provisioning: Optional[ - Union[str, ProvisioningEntry, QRProvisioningInformation] - ] = None, - dsk: Optional[str] = None, + force_security: bool | None = None, + provisioning: str | ProvisioningEntry | QRProvisioningInformation | None = None, + dsk: str | None = None, ) -> bool: """Send beginInclusion command to Controller.""" # Most functionality was introduced in Schema 8 @@ -289,7 +287,7 @@ async def async_begin_inclusion( async def async_provision_smart_start_node( self, - provisioning_info: Union[ProvisioningEntry, QRProvisioningInformation, str], + provisioning_info: ProvisioningEntry | QRProvisioningInformation | str, ) -> None: """Send provisionSmartStartNode command to Controller.""" if ( @@ -310,8 +308,7 @@ async def async_provision_smart_start_node( ) async def async_unprovision_smart_start_node( - self, - dsk_or_node_id: Union[str, int], + self, dsk_or_node_id: int | str ) -> None: """Send unprovisionSmartStartNode command to Controller.""" await self.client.async_send_command( @@ -323,8 +320,8 @@ async def async_unprovision_smart_start_node( ) async def async_get_provisioning_entry( - self, dsk_or_node_id: Union[str, int] - ) -> Optional[ProvisioningEntry]: + self, dsk_or_node_id: int | str + ) -> ProvisioningEntry | None: """Send getProvisioningEntry command to Controller.""" data = await self.client.async_send_command( { @@ -355,10 +352,10 @@ async def async_stop_inclusion(self) -> bool: return cast(bool, data["success"]) async def async_begin_exclusion( - self, strategy: Optional[ExclusionStrategy] = None + self, strategy: ExclusionStrategy | None = None ) -> bool: """Send beginExclusion command to Controller.""" - payload: dict[str, Union[str, ExclusionStrategy]] = { + payload: dict[str, str | ExclusionStrategy] = { "command": "controller.begin_exclusion" } if strategy is not None: @@ -388,10 +385,8 @@ async def async_replace_failed_node( InclusionStrategy.SECURITY_S2, InclusionStrategy.INSECURE, ], - force_security: Optional[bool] = None, - provisioning: Optional[ - Union[str, ProvisioningEntry, QRProvisioningInformation] - ] = None, + force_security: bool | None = None, + provisioning: str | ProvisioningEntry | QRProvisioningInformation | None = None, ) -> bool: """Send replaceFailedNode command to Controller.""" # Most functionality was introduced in Schema 8 @@ -646,7 +641,7 @@ async def async_validate_dsk_and_enter_pin(self, pin: str) -> None: } ) - async def async_supports_feature(self, feature: ZwaveFeature) -> Optional[bool]: + async def async_supports_feature(self, feature: ZwaveFeature) -> bool | None: """ Send supportsFeature command to Controller. @@ -657,7 +652,7 @@ async def async_supports_feature(self, feature: ZwaveFeature) -> Optional[bool]: {"command": "controller.supports_feature", "feature": feature.value}, require_schema=12, ) - return cast(Optional[bool], data.get("supported")) + return cast(bool | None, data.get("supported")) async def async_get_state(self) -> ControllerDataType: """Get controller state.""" diff --git a/zwave_js_server/model/controller/data_model.py b/zwave_js_server/model/controller/data_model.py index fcffc5118..cfb2ca00f 100644 --- a/zwave_js_server/model/controller/data_model.py +++ b/zwave_js_server/model/controller/data_model.py @@ -1,4 +1,6 @@ """Data model for a Z-Wave JS controller.""" +from __future__ import annotations + from typing import TypedDict from .statistics import ControllerStatisticsDataType diff --git a/zwave_js_server/model/controller/event_model.py b/zwave_js_server/model/controller/event_model.py index ea03525da..67cce0c49 100644 --- a/zwave_js_server/model/controller/event_model.py +++ b/zwave_js_server/model/controller/event_model.py @@ -1,4 +1,6 @@ """Provide a model for the Z-Wave JS controller's events.""" +from __future__ import annotations + from typing import Literal, TypedDict from ...event import BaseEventModel diff --git a/zwave_js_server/model/controller/firmware.py b/zwave_js_server/model/controller/firmware.py index d46834b5b..1387e7b10 100644 --- a/zwave_js_server/model/controller/firmware.py +++ b/zwave_js_server/model/controller/firmware.py @@ -1,7 +1,9 @@ """Provide a model for Z-Wave controller firmware.""" +from __future__ import annotations + from dataclasses import dataclass from enum import IntEnum -from typing import Optional, TypedDict +from typing import TypedDict from ...util.helpers import convert_bytes_to_base64 @@ -20,7 +22,7 @@ class ControllerFirmwareUpdateData: filename: str file: bytes - file_format: Optional[str] = None + file_format: str | None = None def to_dict(self) -> ControllerFirmwareUpdateDataDataType: """Convert firmware update data to dict.""" diff --git a/zwave_js_server/model/controller/inclusion_and_provisioning.py b/zwave_js_server/model/controller/inclusion_and_provisioning.py index 0a7c2b144..0ab733519 100644 --- a/zwave_js_server/model/controller/inclusion_and_provisioning.py +++ b/zwave_js_server/model/controller/inclusion_and_provisioning.py @@ -1,6 +1,8 @@ """Provide a model for the Z-Wave JS controller's inclusion/provisioning data structures.""" +from __future__ import annotations + from dataclasses import dataclass -from typing import Any, Optional, TypedDict +from typing import Any, TypedDict from ...const import Protocols, ProvisioningEntryStatus, QRCodeVersion, SecurityClass @@ -44,9 +46,9 @@ class ProvisioningEntry: dsk: str security_classes: list[SecurityClass] - requested_security_classes: Optional[list[SecurityClass]] = None + requested_security_classes: list[SecurityClass] | None = None status: ProvisioningEntryStatus = ProvisioningEntryStatus.ACTIVE - additional_properties: Optional[dict[str, Any]] = None + additional_properties: dict[str, Any] | None = None def to_dict(self) -> dict[str, Any]: """Return PlannedProvisioning data dict from self.""" @@ -98,9 +100,9 @@ class QRProvisioningInformationMixin: product_type: int product_id: int application_version: str - max_inclusion_request_interval: Optional[int] - uuid: Optional[str] - supported_protocols: Optional[list[Protocols]] + max_inclusion_request_interval: int | None + uuid: str | None + supported_protocols: list[Protocols] | None @dataclass diff --git a/zwave_js_server/model/controller/statistics.py b/zwave_js_server/model/controller/statistics.py index da7c3ebd0..77f4bb006 100644 --- a/zwave_js_server/model/controller/statistics.py +++ b/zwave_js_server/model/controller/statistics.py @@ -1,6 +1,8 @@ """Provide a model for the Z-Wave JS controller's statistics.""" +from __future__ import annotations + from dataclasses import dataclass -from typing import TYPE_CHECKING, Optional, TypedDict +from typing import TYPE_CHECKING, TypedDict from ..statistics import RouteStatistics, RouteStatisticsDataType @@ -32,12 +34,12 @@ def __init__( self._nlwr = RouteStatistics(client, nlwr) @property - def lwr(self) -> Optional[RouteStatistics]: + def lwr(self) -> RouteStatistics | None: """Return the last working route from the controller to this node.""" return self._lwr @property - def nlwr(self) -> Optional[RouteStatistics]: + def nlwr(self) -> RouteStatistics | None: """Return the next to last working route from the controller to this node.""" return self._nlwr @@ -60,7 +62,7 @@ class ControllerStatisticsDataType(TypedDict): class ControllerStatistics: """Represent a controller statistics update.""" - def __init__(self, data: Optional[ControllerStatisticsDataType] = None) -> None: + def __init__(self, data: ControllerStatisticsDataType | None = None) -> None: """Initialize controller statistics.""" self.data = data or ControllerStatisticsDataType( CAN=0, diff --git a/zwave_js_server/model/device_class.py b/zwave_js_server/model/device_class.py index 4e5f45b7e..7ce1b144b 100644 --- a/zwave_js_server/model/device_class.py +++ b/zwave_js_server/model/device_class.py @@ -3,6 +3,7 @@ https://zwave-js.github.io/node-zwave-js/#/api/node?id=deviceclass """ +from __future__ import annotations from dataclasses import dataclass from typing import TypedDict diff --git a/zwave_js_server/model/device_config.py b/zwave_js_server/model/device_config.py index 80d6e0c0b..bed1c2178 100644 --- a/zwave_js_server/model/device_config.py +++ b/zwave_js_server/model/device_config.py @@ -3,7 +3,9 @@ https://zwave-js.github.io/node-zwave-js/#/api/node?id=deviceconfig """ -from typing import Any, Literal, Optional, TypedDict, Union +from __future__ import annotations + +from typing import Any, Literal, TypedDict class DeviceDeviceDataType(TypedDict, total=False): @@ -21,12 +23,12 @@ def __init__(self, data: DeviceDeviceDataType) -> None: self.data = data @property - def product_type(self) -> Optional[str]: + def product_type(self) -> str | None: """Return product type.""" return self.data.get("productType") @property - def product_id(self) -> Optional[str]: + def product_id(self) -> str | None: """Return product id.""" return self.data.get("productId") @@ -46,12 +48,12 @@ def __init__(self, data: DeviceFirmwareVersionRangeDataType) -> None: self.data = data @property - def min(self) -> Optional[str]: + def min(self) -> str | None: """Return min version.""" return self.data.get("min") @property - def max(self) -> Optional[str]: + def max(self) -> str | None: """Return max version.""" return self.data.get("max") @@ -73,7 +75,7 @@ class DeviceMetadataDataType(TypedDict, total=False): exclusion: str reset: str manual: str - comments: Union[CommentDataType, list[CommentDataType]] + comments: CommentDataType | list[CommentDataType] class DeviceMetadata: @@ -84,27 +86,27 @@ def __init__(self, data: DeviceMetadataDataType) -> None: self.data = data @property - def wakeup(self) -> Optional[str]: + def wakeup(self) -> str | None: """Return wakeup instructions.""" return self.data.get("wakeup") @property - def inclusion(self) -> Optional[str]: + def inclusion(self) -> str | None: """Return inclusion instructions.""" return self.data.get("inclusion") @property - def exclusion(self) -> Optional[str]: + def exclusion(self) -> str | None: """Return exclusion instructions.""" return self.data.get("exclusion") @property - def reset(self) -> Optional[str]: + def reset(self) -> str | None: """Return reset instructions.""" return self.data.get("reset") @property - def manual(self) -> Optional[str]: + def manual(self) -> str | None: """Return manual instructions.""" return self.data.get("manual") @@ -151,27 +153,27 @@ def __init__(self, data: DeviceConfigDataType) -> None: self._metadata = DeviceMetadata(self.data.get("metadata", {})) @property - def filename(self) -> Optional[str]: + def filename(self) -> str | None: """Return config filename.""" return self.data.get("filename") @property - def manufacturer(self) -> Optional[str]: + def manufacturer(self) -> str | None: """Return name of the manufacturer.""" return self.data.get("manufacturer") @property - def manufacturer_id(self) -> Optional[str]: # TODO: In the dump this is an int. + def manufacturer_id(self) -> str | None: # TODO: In the dump this is an int. """Return manufacturer id (as defined in the specs) as a 4-digit hexadecimal string.""" return self.data.get("manufacturerId") @property - def label(self) -> Optional[str]: + def label(self) -> str | None: """Return short label for the device.""" return self.data.get("label") @property - def description(self) -> Optional[str]: + def description(self) -> str | None: """Return longer description of the device, usually the full name.""" return self.data.get("description") @@ -196,7 +198,7 @@ def param_information(self) -> dict[str, dict]: return self.data.get("paramInformation", {}) @property - def supports_zwave_plus(self) -> Optional[bool]: + def supports_zwave_plus(self) -> bool | None: """Return if the device complies with the Z-Wave+ standard.""" return self.data.get("supportsZWavePlus") @@ -216,6 +218,6 @@ def metadata(self) -> DeviceMetadata: return self._metadata @property - def is_embedded(self) -> Optional[bool]: + def is_embedded(self) -> bool | None: """Return whether device config is embedded in zwave-js-server.""" return self.data.get("isEmbedded") diff --git a/zwave_js_server/model/driver.py b/zwave_js_server/model/driver.py index 98985e7b6..1884ae1b5 100644 --- a/zwave_js_server/model/driver.py +++ b/zwave_js_server/model/driver.py @@ -1,5 +1,7 @@ """Provide a model for the Z-Wave JS Driver.""" -from typing import TYPE_CHECKING, Any, Literal, Optional, Union, cast +from __future__ import annotations + +from typing import TYPE_CHECKING, Any, Literal, cast from pydantic import create_model_from_typeddict @@ -50,7 +52,7 @@ def __init__(self, data: dict) -> None: """Initialize class.""" self.installed_version: str = data["installedVersion"] self.update_available: bool = data["updateAvailable"] - self.new_version: Optional[str] = data.get("newVersion") + self.new_version: str | None = data.get("newVersion") class Driver(EventBase): @@ -88,7 +90,7 @@ def receive_event(self, event: Event) -> None: self.emit(event.type, event.data) async def _async_send_command( - self, command: str, require_schema: Optional[int] = None, **kwargs: Any + self, command: str, require_schema: int | None = None, **kwargs: Any ) -> dict: """Send a driver command. For internal use only.""" return await self.client.async_send_command( @@ -155,7 +157,7 @@ async def async_install_config_update(self) -> bool: return cast(bool, result["success"]) async def async_set_preferred_scales( - self, scales: dict[Union[str, int], Union[str, int]] + self, scales: dict[str | int, str | int] ) -> None: """Send command to set preferred sensor scales.""" await self._async_send_command( diff --git a/zwave_js_server/model/endpoint.py b/zwave_js_server/model/endpoint.py index 4ecbbdcc7..f45fa0ddf 100644 --- a/zwave_js_server/model/endpoint.py +++ b/zwave_js_server/model/endpoint.py @@ -3,8 +3,9 @@ https://zwave-js.github.io/node-zwave-js/#/api/endpoint?id=endpoint-properties """ +from __future__ import annotations -from typing import TYPE_CHECKING, Any, Optional, TypedDict, Union, cast +from typing import TYPE_CHECKING, Any, TypedDict, cast from ..const import NodeStatus from ..event import EventBase @@ -37,13 +38,13 @@ def __init__( self, client: "Client", data: EndpointDataType, - values: dict[str, Union[ConfigurationValue, Value]], + values: dict[str, ConfigurationValue | Value], ) -> None: """Initialize.""" super().__init__() self.client = client self.data: EndpointDataType = {} - self.values: dict[str, Union[ConfigurationValue, Value]] = {} + self.values: dict[str, ConfigurationValue | Value] = {} self.update(data, values) def __repr__(self) -> str: @@ -80,12 +81,12 @@ def device_class(self) -> DeviceClass: return DeviceClass(self.data["deviceClass"]) @property - def installer_icon(self) -> Optional[int]: + def installer_icon(self) -> int | None: """Return installer icon property.""" return self.data.get("installerIcon") @property - def user_icon(self) -> Optional[int]: + def user_icon(self) -> int | None: """Return user icon property.""" return self.data.get("userIcon") @@ -95,14 +96,12 @@ def command_classes(self) -> list[CommandClassInfo]: return [CommandClassInfo(cc) for cc in self.data["commandClasses"]] @property - def endpoint_label(self) -> Optional[str]: + def endpoint_label(self) -> str | None: """Return endpoint label property.""" return self.data.get("endpointLabel") def update( - self, - data: EndpointDataType, - values: dict[str, Union[ConfigurationValue, Value]], + self, data: EndpointDataType, values: dict[str, ConfigurationValue | Value] ) -> None: """Update the endpoint data.""" self.data = data @@ -120,10 +119,10 @@ def update( async def async_send_command( self, cmd: str, - require_schema: Optional[int] = None, - wait_for_result: Optional[bool] = None, + require_schema: int | None = None, + wait_for_result: bool | None = None, **cmd_kwargs: Any, - ) -> Optional[dict[str, Any]]: + ) -> dict[str, Any] | None: """ Send an endpoint command. For internal use only. @@ -159,7 +158,7 @@ async def async_invoke_cc_api( command_class: CommandClass, method_name: str, *args: Any, - wait_for_result: Optional[bool] = None, + wait_for_result: bool | None = None, ) -> Any: """Call endpoint.invoke_cc_api command.""" if not any(cc.id == command_class.value for cc in self.command_classes): diff --git a/zwave_js_server/model/log_config.py b/zwave_js_server/model/log_config.py index d09e557a3..d43a04fb3 100644 --- a/zwave_js_server/model/log_config.py +++ b/zwave_js_server/model/log_config.py @@ -1,6 +1,8 @@ """Provide a model for the log config.""" +from __future__ import annotations + from dataclasses import dataclass -from typing import Optional, TypedDict, cast +from typing import TypedDict, cast from ..const import LogLevel @@ -21,11 +23,11 @@ class LogConfig: # https://github.com/zwave-js/node-zwave-js/blob/master/packages/core/src/log/shared.ts#L85 # Must include at least one key - enabled: Optional[bool] = None - level: Optional[LogLevel] = None - log_to_file: Optional[bool] = None - filename: Optional[str] = None - force_console: Optional[bool] = None + enabled: bool | None = None + level: LogLevel | None = None + log_to_file: bool | None = None + filename: str | None = None + force_console: bool | None = None def to_dict(self) -> LogConfigDataType: """Return LogConfigDataType dict from self.""" diff --git a/zwave_js_server/model/log_message.py b/zwave_js_server/model/log_message.py index dbd5f3a69..6a6f40c3d 100644 --- a/zwave_js_server/model/log_message.py +++ b/zwave_js_server/model/log_message.py @@ -1,5 +1,7 @@ """Provide a model for a log message event.""" -from typing import Literal, Optional, TypedDict, Union +from __future__ import annotations + +from typing import Literal, TypedDict from ..const import CommandClass @@ -16,8 +18,8 @@ class LogMessageContextDataType(TypedDict, total=False): internal: bool endpoint: int commandClass: int - property: Union[int, str] - propertyKey: Union[int, str] + property: int | str + propertyKey: int | str class LogMessageContext: @@ -33,56 +35,56 @@ def source(self) -> Literal["config", "serial", "controller", "driver"]: return self.data["source"] @property - def type(self) -> Optional[Literal["controller", "value", "node"]]: + def type(self) -> Literal["controller", "value", "node"] | None: """Return the object type for the log message if applicable.""" return self.data.get("type") @property - def node_id(self) -> Optional[int]: + def node_id(self) -> int | None: """Return the Node ID for the log message if applicable.""" return self.data.get("nodeId") @property - def header(self) -> Optional[str]: + def header(self) -> str | None: """Return the header for the log message if applicable.""" return self.data.get("header") @property - def direction(self) -> Optional[Literal["inbound", "outbound", "none"]]: + def direction(self) -> Literal["inbound", "outbound", "none"] | None: """Return the direction for the log message if applicable.""" return self.data.get("direction") @property def change( self, - ) -> Optional[Literal["added", "removed", "updated", "notification"]]: + ) -> Literal["added", "removed", "updated", "notification"] | None: """Return the change type for the log message if applicable.""" return self.data.get("change") @property - def internal(self) -> Optional[bool]: + def internal(self) -> bool | None: """Return the internal flag for the log message if applicable.""" return self.data.get("internal") @property - def endpoint(self) -> Optional[int]: + def endpoint(self) -> int | None: """Return the Node/Value endpoint for the log message if applicable.""" return self.data.get("endpoint") @property - def command_class(self) -> Optional[CommandClass]: + def command_class(self) -> CommandClass | None: """Return the Value command class for the log message if applicable.""" if command_class := self.data.get("commandClass"): return CommandClass(command_class) return None @property - def property_(self) -> Optional[Union[int, str]]: + def property_(self) -> int | str | None: """Return the Value property for the log message if applicable.""" return self.data.get("property") @property - def property_key(self) -> Optional[Union[int, str]]: + def property_key(self) -> int | str | None: """Return the Value property key for the log message if applicable.""" return self.data.get("propertyKey") @@ -92,8 +94,8 @@ class LogMessageDataType(TypedDict, total=False): source: Literal["driver"] # required event: Literal["logging"] # required - message: Union[str, list[str]] # required - formattedMessage: Union[str, list[str]] # required + message: str | list[str] # required + formattedMessage: str | list[str] # required direction: str # required level: str # required context: LogMessageContextDataType # required @@ -113,7 +115,7 @@ def __init__(self, data: LogMessageDataType): self.data = data def _process_message( - self, field_name: Union[Literal["message"], Literal["formattedMessage"]] + self, field_name: Literal["message", "formattedMessage"] ) -> list[str]: """Process a message and always return a list.""" if isinstance(self.data[field_name], str): @@ -144,32 +146,32 @@ def level(self) -> str: return self.data["level"] @property - def primary_tags(self) -> Optional[str]: + def primary_tags(self) -> str | None: """Return primary tags.""" return self.data.get("primaryTags") @property - def secondary_tags(self) -> Optional[str]: + def secondary_tags(self) -> str | None: """Return secondary tags.""" return self.data.get("secondaryTags") @property - def secondary_tag_padding(self) -> Optional[int]: + def secondary_tag_padding(self) -> int | None: """Return secondary tag padding.""" return self.data.get("secondaryTagPadding") @property - def multiline(self) -> Optional[bool]: + def multiline(self) -> bool | None: """Return whether message is multiline.""" return self.data.get("multiline") @property - def timestamp(self) -> Optional[str]: + def timestamp(self) -> str | None: """Return timestamp.""" return self.data.get("timestamp") @property - def label(self) -> Optional[str]: + def label(self) -> str | None: """Return label.""" return self.data.get("label") diff --git a/zwave_js_server/model/node/__init__.py b/zwave_js_server/model/node/__init__.py index b4bc8078e..6dded860b 100644 --- a/zwave_js_server/model/node/__init__.py +++ b/zwave_js_server/model/node/__init__.py @@ -1,6 +1,8 @@ """Provide a model for the Z-Wave JS node.""" +from __future__ import annotations + import logging -from typing import TYPE_CHECKING, Any, Optional, Union, cast +from typing import TYPE_CHECKING, Any, cast from ...const import ( INTERVIEW_FAILED, @@ -91,8 +93,8 @@ def __init__(self, client: "Client", data: NodeDataType) -> None: self.data: NodeDataType = {} self._device_config = DeviceConfig({}) self._statistics = NodeStatistics(client, data.get("statistics")) - self._firmware_update_progress: Optional[NodeFirmwareUpdateProgress] = None - self.values: dict[str, Union[ConfigurationValue, Value]] = {} + self._firmware_update_progress: NodeFirmwareUpdateProgress | None = None + self.values: dict[str, ConfigurationValue | Value] = {} self.endpoints: dict[int, Endpoint] = {} self.update(data) @@ -128,12 +130,12 @@ def device_class(self) -> DeviceClass: return DeviceClass(self.data["deviceClass"]) @property - def installer_icon(self) -> Optional[int]: + def installer_icon(self) -> int | None: """Return installer icon property.""" return self.data.get("installerIcon") @property - def user_icon(self) -> Optional[int]: + def user_icon(self) -> int | None: """Return user icon property.""" return self.data.get("userIcon") @@ -143,27 +145,27 @@ def status(self) -> NodeStatus: return NodeStatus(self.data["status"]) @property - def ready(self) -> Optional[bool]: + def ready(self) -> bool | None: """Return the ready.""" return self.data.get("ready") @property - def is_listening(self) -> Optional[bool]: + def is_listening(self) -> bool | None: """Return the is_listening.""" return self.data.get("isListening") @property - def is_frequent_listening(self) -> Optional[Union[bool, str]]: + def is_frequent_listening(self) -> bool | str | None: """Return the is_frequent_listening.""" return self.data.get("isFrequentListening") @property - def is_routing(self) -> Optional[bool]: + def is_routing(self) -> bool | None: """Return the is_routing.""" return self.data.get("isRouting") @property - def max_data_rate(self) -> Optional[int]: + def max_data_rate(self) -> int | None: """Return the max_data_rate.""" return self.data.get("maxDataRate") @@ -173,69 +175,69 @@ def supported_data_rates(self) -> list[int]: return self.data.get("supportedDataRates", []) @property - def is_secure(self) -> Optional[bool]: + def is_secure(self) -> bool | None: """Return the is_secure.""" if (is_secure := self.data.get("isSecure")) == "unknown": return None return is_secure @property - def protocol_version(self) -> Optional[int]: + def protocol_version(self) -> int | None: """Return the protocol_version.""" return self.data.get("protocolVersion") @property - def supports_beaming(self) -> Optional[bool]: + def supports_beaming(self) -> bool | None: """Return the supports_beaming.""" return self.data.get("supportsBeaming") @property - def supports_security(self) -> Optional[bool]: + def supports_security(self) -> bool | None: """Return the supports_security.""" return self.data.get("supportsSecurity") @property - def manufacturer_id(self) -> Optional[int]: + def manufacturer_id(self) -> int | None: """Return the manufacturer_id.""" return self.data.get("manufacturerId") @property - def product_id(self) -> Optional[int]: + def product_id(self) -> int | None: """Return the product_id.""" return self.data.get("productId") @property - def product_type(self) -> Optional[int]: + def product_type(self) -> int | None: """Return the product_type.""" return self.data.get("productType") @property - def firmware_version(self) -> Optional[str]: + def firmware_version(self) -> str | None: """Return the firmware_version.""" return self.data.get("firmwareVersion") @property - def zwave_plus_version(self) -> Optional[int]: + def zwave_plus_version(self) -> int | None: """Return the zwave_plus_version.""" return self.data.get("zwavePlusVersion") @property - def zwave_plus_node_type(self) -> Optional[int]: + def zwave_plus_node_type(self) -> int | None: """Return the zwave_plus_node_type.""" return self.data.get("zwavePlusNodeType") @property - def zwave_plus_role_type(self) -> Optional[int]: + def zwave_plus_role_type(self) -> int | None: """Return the zwave_plus_role_type.""" return self.data.get("zwavePlusRoleType") @property - def name(self) -> Optional[str]: + def name(self) -> str | None: """Return the name.""" return self.data.get("name") @property - def location(self) -> Optional[str]: + def location(self) -> str | None: """Return the location.""" return self.data.get("location") @@ -245,42 +247,42 @@ def device_config(self) -> DeviceConfig: return self._device_config @property - def label(self) -> Optional[str]: + def label(self) -> str | None: """Return the label.""" return self.data.get("label") @property - def device_database_url(self) -> Optional[str]: + def device_database_url(self) -> str | None: """Return the device database URL.""" return self.data.get("deviceDatabaseUrl") @property - def endpoint_count_is_dynamic(self) -> Optional[bool]: + def endpoint_count_is_dynamic(self) -> bool | None: """Return the endpoint_count_is_dynamic.""" return self.data.get("endpointCountIsDynamic") @property - def endpoints_have_identical_capabilities(self) -> Optional[bool]: + def endpoints_have_identical_capabilities(self) -> bool | None: """Return the endpoints_have_identical_capabilities.""" return self.data.get("endpointsHaveIdenticalCapabilities") @property - def individual_endpoint_count(self) -> Optional[int]: + def individual_endpoint_count(self) -> int | None: """Return the individual_endpoint_count.""" return self.data.get("individualEndpointCount") @property - def aggregated_endpoint_count(self) -> Optional[int]: + def aggregated_endpoint_count(self) -> int | None: """Return the aggregated_endpoint_count.""" return self.data.get("aggregatedEndpointCount") @property - def interview_attempts(self) -> Optional[int]: + def interview_attempts(self) -> int | None: """Return the interview_attempts.""" return self.data.get("interviewAttempts") @property - def interview_stage(self) -> Optional[Union[int, str]]: + def interview_stage(self) -> int | str | None: """Return the interview_stage.""" return self.data.get("interviewStage") @@ -309,12 +311,12 @@ def statistics(self) -> NodeStatistics: return self._statistics @property - def firmware_update_progress(self) -> Optional[NodeFirmwareUpdateProgress]: + def firmware_update_progress(self) -> NodeFirmwareUpdateProgress | None: """Return firmware update progress.""" return self._firmware_update_progress @property - def highest_security_class(self) -> Optional[SecurityClass]: + def highest_security_class(self) -> SecurityClass | None: """Return highest security class configured on the node.""" if (security_class := self.data.get("highestSecurityClass")) is None: return None @@ -380,8 +382,8 @@ def update(self, data: NodeDataType) -> None: ) def get_command_class_values( - self, command_class: CommandClass, endpoint: Optional[int] = None - ) -> dict[str, Union[ConfigurationValue, Value]]: + self, command_class: CommandClass, endpoint: int | None = None + ) -> dict[str, ConfigurationValue | Value]: """Return all values for a given command class.""" return { value_id: value @@ -409,10 +411,10 @@ def receive_event(self, event: Event) -> None: async def async_send_command( self, cmd: str, - require_schema: Optional[int] = None, - wait_for_result: Optional[bool] = None, + require_schema: int | None = None, + wait_for_result: bool | None = None, **cmd_kwargs: Any, - ) -> Optional[dict[str, Any]]: + ) -> dict[str, Any] | None: """ Send a node command. For internal use only. @@ -435,11 +437,11 @@ async def async_send_command( async def async_set_value( self, - val: Union[Value, str], + val: Value | str, new_value: Any, - options: Optional[dict] = None, - wait_for_result: Optional[bool] = None, - ) -> Optional[bool]: + options: dict | None = None, + wait_for_result: bool | None = None, + ) -> bool | None: """Send setValue command to Node for given value (or value_id).""" # a value may be specified as value_id or the value itself if not isinstance(val, Value): @@ -512,7 +514,7 @@ async def async_get_defined_value_ids(self) -> list[Value]: for value_id in data["valueIds"] ] - async def async_get_value_metadata(self, val: Union[Value, str]) -> ValueMetadata: + async def async_get_value_metadata(self, val: Value | str) -> ValueMetadata: """Send getValueMetadata command to Node.""" # a value may be specified as value_id or the value itself if not isinstance(val, Value): @@ -557,7 +559,7 @@ async def async_abort_firmware_update(self) -> None: """Send abortFirmwareUpdate command to Node.""" await self.async_send_command("abort_firmware_update", wait_for_result=True) - async def async_poll_value(self, val: Union[Value, str]) -> None: + async def async_poll_value(self, val: Value | str) -> None: """Send pollValue command to Node for given value (or value_id).""" # a value may be specified as value_id or the value itself if not isinstance(val, Value): @@ -583,7 +585,7 @@ async def async_invoke_cc_api( command_class: CommandClass, method_name: str, *args: Any, - wait_for_result: Optional[bool] = None, + wait_for_result: bool | None = None, ) -> Any: """Call endpoint.invoke_cc_api command.""" return await self.endpoints[0].async_invoke_cc_api( @@ -649,7 +651,7 @@ async def async_test_power_level( return cast(int, data["framesAcked"]) async def async_check_lifeline_health( - self, rounds: Optional[int] = None + self, rounds: int | None = None ) -> LifelineHealthCheckSummary: """Send checkLifelineHealth command to Node.""" kwargs = {} @@ -665,7 +667,7 @@ async def async_check_lifeline_health( return LifelineHealthCheckSummary(data["summary"]) async def async_check_route_health( - self, target_node: "Node", rounds: Optional[int] = None + self, target_node: "Node", rounds: int | None = None ) -> RouteHealthCheckSummary: """Send checkRouteHealth command to Node.""" kwargs = {"targetNodeId": target_node.node_id} @@ -689,7 +691,7 @@ async def async_get_state(self) -> NodeDataType: return cast(NodeDataType, data["state"]) async def async_set_name( - self, name: str, update_cc: bool = True, wait_for_result: Optional[bool] = None + self, name: str, update_cc: bool = True, wait_for_result: bool | None = None ) -> None: """Set node name.""" # If we may not potentially update the name CC, we should just wait for the @@ -709,7 +711,7 @@ async def async_set_location( self, location: str, update_cc: bool = True, - wait_for_result: Optional[bool] = None, + wait_for_result: bool | None = None, ) -> None: """Set node location.""" # If we may not potentially update the location CC, we should just wait for the diff --git a/zwave_js_server/model/node/data_model.py b/zwave_js_server/model/node/data_model.py index f9cadaa12..cad8ddffe 100644 --- a/zwave_js_server/model/node/data_model.py +++ b/zwave_js_server/model/node/data_model.py @@ -1,5 +1,7 @@ """Data model for a Z-Wave JS node.""" -from typing import Literal, Optional, TypedDict, Union +from __future__ import annotations + +from typing import Literal, TypedDict from ..device_class import DeviceClassDataType from ..device_config import DeviceConfigDataType @@ -32,11 +34,11 @@ class NodeDataType(TypedDict, total=False): zwavePlusNodeType: int zwavePlusRoleType: int isListening: bool - isFrequentListening: Union[bool, str] + isFrequentListening: bool | str isRouting: bool maxDataRate: int supportedDataRates: list[int] - isSecure: Union[bool, Literal["unknown"]] + isSecure: bool | Literal["unknown"] supportsBeaming: bool supportsSecurity: bool protocolVersion: int @@ -55,7 +57,7 @@ class NodeDataType(TypedDict, total=False): individualEndpointCount: int aggregatedEndpointCount: int interviewAttempts: int - interviewStage: Optional[Union[int, str]] + interviewStage: int | str | None values: list[ValueDataType] statistics: NodeStatisticsDataType highestSecurityClass: int diff --git a/zwave_js_server/model/node/event_model.py b/zwave_js_server/model/node/event_model.py index 0b8717661..f5e63a755 100644 --- a/zwave_js_server/model/node/event_model.py +++ b/zwave_js_server/model/node/event_model.py @@ -1,5 +1,7 @@ """Provide a model for the Z-Wave JS node's events.""" -from typing import Literal, Optional, Union +from __future__ import annotations + +from typing import Literal from pydantic import BaseModel @@ -74,8 +76,8 @@ class InterviewFailedEventArgsModel(BaseModel): errorMessage: str isFinal: bool - attempt: Optional[int] - maxAttempts: Optional[int] + attempt: int | None + maxAttempts: int | None class InterviewFailedEventModel(BaseNodeEventModel): @@ -103,11 +105,11 @@ class NotificationEventModel(BaseNodeEventModel): event: Literal["notification"] ccId: CommandClass - args: Union[ - NotificationNotificationArgsDataType, - EntryControlNotificationArgsDataType, - PowerLevelNotificationArgsDataType, - ] + args: ( + NotificationNotificationArgsDataType + | EntryControlNotificationArgsDataType + | PowerLevelNotificationArgsDataType + ) class ReadyEventModel(BaseNodeEventModel): diff --git a/zwave_js_server/model/node/firmware.py b/zwave_js_server/model/node/firmware.py index df98db4d9..c6dd6b13e 100644 --- a/zwave_js_server/model/node/firmware.py +++ b/zwave_js_server/model/node/firmware.py @@ -1,7 +1,9 @@ """Provide a model for Z-Wave firmware.""" +from __future__ import annotations + from dataclasses import asdict, dataclass from enum import IntEnum -from typing import TYPE_CHECKING, Optional, TypedDict, Union, cast +from typing import TYPE_CHECKING, TypedDict, cast from ...const import VALUE_UNKNOWN from ...util.helpers import convert_bytes_to_base64 @@ -25,8 +27,8 @@ class NodeFirmwareUpdateData: filename: str file: bytes - file_format: Optional[str] = None - firmware_target: Optional[int] = None + file_format: str | None = None + firmware_target: int | None = None def to_dict(self) -> NodeFirmwareUpdateDataDataType: """Convert firmware update data to dict.""" @@ -46,8 +48,8 @@ class NodeFirmwareUpdateCapabilitiesDataType(TypedDict, total=False): firmwareUpgradable: bool # required firmwareTargets: list[int] - continuesToFunction: Union[bool, str] - supportsActivation: Union[bool, str] + continuesToFunction: bool | str + supportsActivation: bool | str class NodeFirmwareUpdateCapabilitiesDict(TypedDict, total=False): @@ -55,8 +57,8 @@ class NodeFirmwareUpdateCapabilitiesDict(TypedDict, total=False): firmware_upgradable: bool # required firmware_targets: list[int] - continues_to_function: Optional[bool] - supports_activation: Optional[bool] + continues_to_function: bool | None + supports_activation: bool | None class NodeFirmwareUpdateCapabilities: @@ -79,7 +81,7 @@ def firmware_targets(self) -> list[int]: return self.data["firmwareTargets"] @property - def continues_to_function(self) -> Optional[bool]: + def continues_to_function(self) -> bool | None: """Return whether node continues to function during update.""" if not self.firmware_upgradable: raise TypeError("Firmware is not upgradeable.") @@ -89,7 +91,7 @@ def continues_to_function(self) -> Optional[bool]: return val @property - def supports_activation(self) -> Optional[bool]: + def supports_activation(self) -> bool | None: """Return whether node supports delayed activation of the new firmware.""" if not self.firmware_upgradable: raise TypeError("Firmware is not upgradeable.") @@ -203,7 +205,7 @@ def success(self) -> bool: return self.data["success"] @property - def wait_time(self) -> Optional[int]: + def wait_time(self) -> int | None: """Return the wait time in seconds before the device is functional again.""" return self.data.get("waitTime") diff --git a/zwave_js_server/model/node/health_check.py b/zwave_js_server/model/node/health_check.py index 4ad11ff42..5dbc2f375 100644 --- a/zwave_js_server/model/node/health_check.py +++ b/zwave_js_server/model/node/health_check.py @@ -1,6 +1,8 @@ """Provide a model for the Z-Wave JS node's health checks and power tests.""" +from __future__ import annotations + from dataclasses import dataclass -from typing import Optional, TypedDict +from typing import TypedDict from ...const import PowerLevel @@ -49,12 +51,12 @@ def failed_pings_node(self) -> int: return self.data["failedPingsNode"] @property - def route_changes(self) -> Optional[int]: + def route_changes(self) -> int | None: """Return number of route changes.""" return self.data.get("routeChanges") @property - def min_power_level(self) -> Optional[PowerLevel]: + def min_power_level(self) -> PowerLevel | None: """Return minimum power level.""" power_level = self.data.get("minPowerlevel") if power_level is not None: @@ -62,12 +64,12 @@ def min_power_level(self) -> Optional[PowerLevel]: return None @property - def failed_pings_controller(self) -> Optional[int]: + def failed_pings_controller(self) -> int | None: """Return number of failed pings to controller.""" return self.data.get("failedPingsController") @property - def snr_margin(self) -> Optional[int]: + def snr_margin(self) -> int | None: """Return SNR margin.""" return self.data.get("snrMargin") @@ -129,17 +131,17 @@ def rating(self) -> int: return self.data["rating"] @property - def failed_pings_to_target(self) -> Optional[int]: + def failed_pings_to_target(self) -> int | None: """Return number of failed pings to target.""" return self.data.get("failedPingsToTarget") @property - def failed_pings_to_source(self) -> Optional[int]: + def failed_pings_to_source(self) -> int | None: """Return number of failed pings to source.""" return self.data.get("failedPingsToSource") @property - def min_power_level_source(self) -> Optional[PowerLevel]: + def min_power_level_source(self) -> PowerLevel | None: """Return minimum power level source.""" power_level = self.data.get("minPowerlevelSource") if power_level is not None: @@ -147,7 +149,7 @@ def min_power_level_source(self) -> Optional[PowerLevel]: return None @property - def min_power_level_target(self) -> Optional[PowerLevel]: + def min_power_level_target(self) -> PowerLevel | None: """Return minimum power level target.""" power_level = self.data.get("minPowerlevelTarget") if power_level is not None: diff --git a/zwave_js_server/model/node/statistics.py b/zwave_js_server/model/node/statistics.py index 983557bff..97ffc63ee 100644 --- a/zwave_js_server/model/node/statistics.py +++ b/zwave_js_server/model/node/statistics.py @@ -1,5 +1,7 @@ """Provide a model for the Z-Wave JS node's statistics.""" -from typing import TYPE_CHECKING, Optional, TypedDict +from __future__ import annotations + +from typing import TYPE_CHECKING, TypedDict from zwave_js_server.exceptions import RssiErrorReceived @@ -28,9 +30,7 @@ class NodeStatisticsDataType(TypedDict, total=False): class NodeStatistics: """Represent a node statistics update.""" - def __init__( - self, client: "Client", data: Optional[NodeStatisticsDataType] - ) -> None: + def __init__(self, client: "Client", data: NodeStatisticsDataType | None) -> None: """Initialize node statistics.""" self.data = data or NodeStatisticsDataType( commandsDroppedRX=0, @@ -81,7 +81,7 @@ def timeout_response(self) -> int: return self.data["timeoutResponse"] @property - def rtt(self) -> Optional[int]: + def rtt(self) -> int | None: """ Return average round trip time (RTT) to this node in milliseconds. @@ -90,7 +90,7 @@ def rtt(self) -> Optional[int]: return self.data.get("rtt") @property - def rssi(self) -> Optional[int]: + def rssi(self) -> int | None: """ Return average RSSI of frames received by this node. @@ -104,11 +104,11 @@ def rssi(self) -> Optional[int]: return rssi_ @property - def lwr(self) -> Optional[RouteStatistics]: + def lwr(self) -> RouteStatistics | None: """Return last working route from the controller to this node.""" return self._lwr @property - def nlwr(self) -> Optional[RouteStatistics]: + def nlwr(self) -> RouteStatistics | None: """Return next to last working route from the controller to this node.""" return self._nlwr diff --git a/zwave_js_server/model/notification.py b/zwave_js_server/model/notification.py index c59d607fb..85031df02 100644 --- a/zwave_js_server/model/notification.py +++ b/zwave_js_server/model/notification.py @@ -3,8 +3,9 @@ https://zwave-js.github.io/node-zwave-js/#/api/node?id=quotnotificationquot """ +from __future__ import annotations -from typing import TYPE_CHECKING, Any, Literal, Optional, TypedDict, Union +from typing import TYPE_CHECKING, Any, Literal, TypedDict from ..const.command_class.multilevel_switch import MultilevelSwitchCommand from ..const.command_class.power_level import PowerLevelTestStatus @@ -30,7 +31,7 @@ class EntryControlNotificationArgsDataType(TypedDict, total=False): eventTypeLabel: str # required dataType: int # required dataTypeLabel: str # required - eventData: Union[str, dict[str, Any]] + eventData: str | dict[str, Any] class EntryControlNotificationDataType(BaseNotificationDataType): @@ -78,7 +79,7 @@ def data_type_label(self) -> str: return self.data["args"]["dataTypeLabel"] @property - def event_data(self) -> Optional[str]: + def event_data(self) -> str | None: """Return event data property.""" if event_data := self.data["args"].get("eventData"): return parse_buffer(event_data) @@ -238,7 +239,7 @@ def event_type_label(self) -> str: return self.data["args"]["eventTypeLabel"] @property - def direction(self) -> Optional[str]: + def direction(self) -> str | None: """Return direction property.""" if direction := self.data["args"].get("direction"): return direction diff --git a/zwave_js_server/model/statistics.py b/zwave_js_server/model/statistics.py index c5e8d957d..9e1e4c938 100644 --- a/zwave_js_server/model/statistics.py +++ b/zwave_js_server/model/statistics.py @@ -1,6 +1,8 @@ """Common models for statistics.""" +from __future__ import annotations + from dataclasses import dataclass -from typing import TYPE_CHECKING, Optional, TypedDict +from typing import TYPE_CHECKING, TypedDict from zwave_js_server.exceptions import RepeaterRssiErrorReceived, RssiErrorReceived @@ -26,9 +28,9 @@ class RouteStatisticsDict(TypedDict): protocol_data_rate: int repeaters: list["Node"] - rssi: Optional[int] + rssi: int | None repeater_rssi: list[int] - route_failed_between: Optional[tuple["Node", "Node"]] + route_failed_between: tuple["Node", "Node"] | None @dataclass @@ -55,7 +57,7 @@ def repeaters(self) -> list["Node"]: ] @property - def rssi(self) -> Optional[int]: + def rssi(self) -> int | None: """Return RSSI.""" if (rssi := self.data.get("rssi")) is None: return None @@ -74,7 +76,7 @@ def repeater_rssi(self) -> list[int]: return repeater_rssi @property - def route_failed_between(self) -> Optional[tuple["Node", "Node"]]: + def route_failed_between(self) -> tuple["Node", "Node"] | None: """Return route failed between.""" if (node_ids := self.data.get("routeFailedBetween")) is None: return None diff --git a/zwave_js_server/model/utils.py b/zwave_js_server/model/utils.py index 028bf78c3..139092b7e 100644 --- a/zwave_js_server/model/utils.py +++ b/zwave_js_server/model/utils.py @@ -1,5 +1,5 @@ """Model for utils commands.""" -from typing import Optional +from __future__ import annotations from ..client import Client from ..const import MINIMUM_QR_STRING_LENGTH @@ -25,7 +25,7 @@ async def async_parse_qr_code_string( async def async_try_parse_dsk_from_qr_code_string( client: Client, qr_code_string: str -) -> Optional[str]: +) -> str | None: """Try to get DSK QR code.""" data = await client.async_send_command( {"command": "utils.try_parse_dsk_from_qr_code_string", "qr": qr_code_string} diff --git a/zwave_js_server/model/value.py b/zwave_js_server/model/value.py index d95e23a47..3bf80c8fb 100644 --- a/zwave_js_server/model/value.py +++ b/zwave_js_server/model/value.py @@ -1,5 +1,7 @@ """Provide a model for the Z-Wave JS value.""" -from typing import TYPE_CHECKING, Any, Optional, TypedDict, Union +from __future__ import annotations + +from typing import TYPE_CHECKING, Any, TypedDict from ..const import VALUE_UNKNOWN, CommandClass, ConfigurationValueType from ..event import Event @@ -33,9 +35,9 @@ class ValueDataType(TypedDict, total=False): commandClass: int # required commandClassName: str # required endpoint: int - property: Union[str, int] # required + property: int | str # required propertyName: str - propertyKey: Union[str, int] + propertyKey: int | str propertyKeyName: str value: Any newValue: Any @@ -44,9 +46,7 @@ class ValueDataType(TypedDict, total=False): ccVersion: int -def _init_value( - node: "Node", val: ValueDataType -) -> Union["Value", "ConfigurationValue"]: +def _init_value(node: "Node", val: ValueDataType) -> "Value" | "ConfigurationValue": """Initialize a Value object from ValueDataType.""" if val["commandClass"] == CommandClass.CONFIGURATION: return ConfigurationValue(node, val) @@ -67,9 +67,9 @@ def _get_value_id_str_from_dict(node: "Node", val: ValueDataType) -> str: def get_value_id_str( node: "Node", command_class: int, - property_: Union[str, int], - endpoint: Optional[int] = None, - property_key: Optional[Union[str, int]] = None, + property_: int | str, + endpoint: int | None = None, + property_key: int | str | None = None, ) -> str: """Return string ID of value.""" # If endpoint is not provided, assume root endpoint @@ -94,37 +94,37 @@ def type(self) -> str: return self.data["type"] @property - def readable(self) -> Optional[bool]: + def readable(self) -> bool | None: """Return readable.""" return self.data.get("readable") @property - def writeable(self) -> Optional[bool]: + def writeable(self) -> bool | None: """Return writeable.""" return self.data.get("writeable") @property - def label(self) -> Optional[str]: + def label(self) -> str | None: """Return label.""" return self.data.get("label") @property - def description(self) -> Optional[str]: + def description(self) -> str | None: """Return description.""" return self.data.get("description") @property - def min(self) -> Optional[int]: + def min(self) -> int | None: """Return min.""" return self.data.get("min") @property - def max(self) -> Optional[int]: + def max(self) -> int | None: """Return max.""" return self.data.get("max") @property - def unit(self) -> Optional[str]: + def unit(self) -> str | None: """Return unit.""" return self.data.get("unit") @@ -144,12 +144,12 @@ def value_change_options(self) -> list[str]: return self.data.get("valueChangeOptions", []) @property - def allow_manual_entry(self) -> Optional[bool]: + def allow_manual_entry(self) -> bool | None: """Return allowManualEntry.""" return self.data.get("allowManualEntry") @property - def value_size(self) -> Optional[int]: + def value_size(self) -> int | None: """Return valueSize.""" return self.data.get("valueSize") @@ -194,7 +194,7 @@ def metadata(self) -> ValueMetadata: return self._metadata @property - def value(self) -> Optional[Any]: + def value(self) -> Any | None: """Return value.""" # Treat unknown values like they are None if self._value == VALUE_UNKNOWN: @@ -217,12 +217,12 @@ def cc_version(self) -> int: return self.data["ccVersion"] @property - def endpoint(self) -> Optional[int]: + def endpoint(self) -> int | None: """Return endpoint.""" return self.data.get("endpoint") @property - def property_(self) -> Union[str, int]: + def property_(self) -> int | str: """Return property. Note the underscore in the end of this property name. @@ -232,17 +232,17 @@ def property_(self) -> Union[str, int]: return self.data["property"] @property - def property_key(self) -> Optional[Union[str, int]]: + def property_key(self) -> int | str | None: """Return propertyKey.""" return self.data.get("propertyKey") @property - def property_name(self) -> Optional[str]: + def property_name(self) -> str | None: """Return propertyName.""" return self.data.get("propertyName") @property - def property_key_name(self) -> Optional[str]: + def property_key_name(self) -> str | None: """Return propertyKeyName.""" return self.data.get("propertyKeyName") diff --git a/zwave_js_server/model/version.py b/zwave_js_server/model/version.py index e6905dc67..7a411c534 100644 --- a/zwave_js_server/model/version.py +++ b/zwave_js_server/model/version.py @@ -1,4 +1,5 @@ """Represents the version from the server.""" +from __future__ import annotations from dataclasses import dataclass from typing import TypedDict diff --git a/zwave_js_server/util/command_class/meter.py b/zwave_js_server/util/command_class/meter.py index aa55f53cf..75e03a396 100644 --- a/zwave_js_server/util/command_class/meter.py +++ b/zwave_js_server/util/command_class/meter.py @@ -1,4 +1,6 @@ """Meter Command Class specific utility functions for values.""" +from __future__ import annotations + from ...const import CommandClass from ...const.command_class.meter import ( CC_SPECIFIC_METER_TYPE, diff --git a/zwave_js_server/util/command_class/multilevel_sensor.py b/zwave_js_server/util/command_class/multilevel_sensor.py index ca4363223..3e5a98fc6 100644 --- a/zwave_js_server/util/command_class/multilevel_sensor.py +++ b/zwave_js_server/util/command_class/multilevel_sensor.py @@ -1,4 +1,6 @@ """Multilevel Sensor Command Class specific utility functions for values.""" +from __future__ import annotations + from ...const import CommandClass from ...const.command_class.multilevel_sensor import ( CC_SPECIFIC_SCALE, diff --git a/zwave_js_server/util/helpers.py b/zwave_js_server/util/helpers.py index 1b11ce8b1..3cb523f05 100644 --- a/zwave_js_server/util/helpers.py +++ b/zwave_js_server/util/helpers.py @@ -1,7 +1,9 @@ """Generic Utility helper functions.""" +from __future__ import annotations + import base64 import json -from typing import Any, Union +from typing import Any from ..exceptions import UnparseableValue @@ -22,7 +24,7 @@ def convert_base64_to_bytes(data: str) -> bytes: return base64.b64decode(data) -def parse_buffer(value: Union[dict[str, Any], str]) -> str: +def parse_buffer(value: dict[str, Any] | str) -> str: """Parse value from a buffer data type.""" if isinstance(value, dict): return parse_buffer_from_dict(value) diff --git a/zwave_js_server/util/multicast.py b/zwave_js_server/util/multicast.py index 3dae85fcd..d502e329b 100644 --- a/zwave_js_server/util/multicast.py +++ b/zwave_js_server/util/multicast.py @@ -1,5 +1,7 @@ """Support for multicast commands.""" -from typing import Any, Optional, cast +from __future__ import annotations + +from typing import Any, cast from ..client import Client from ..const import TARGET_VALUE_PROPERTY, CommandClass @@ -11,8 +13,8 @@ async def _async_send_command( client: Client, command: str, - nodes: Optional[list[Node]] = None, - require_schema: Optional[int] = None, + nodes: list[Node] | None = None, + require_schema: int | None = None, **kwargs: Any, ) -> dict: """Send a multicast command.""" @@ -32,8 +34,8 @@ async def async_multicast_set_value( client: Client, new_value: Any, value_data: ValueDataType, - nodes: Optional[list[Node]] = None, - options: Optional[dict] = None, + nodes: list[Node] | None = None, + options: dict | None = None, ) -> bool: """Send a multicast set_value command.""" assert client.driver @@ -69,7 +71,7 @@ async def async_multicast_set_value( async def async_multicast_get_endpoint_count( - client: Client, nodes: Optional[list[Node]] = None + client: Client, nodes: list[Node] | None = None ) -> int: """Send a multicast get_endpoint_count command.""" result = await _async_send_command( @@ -82,7 +84,7 @@ async def async_multicast_endpoint_supports_cc( client: Client, endpoint: int, command_class: CommandClass, - nodes: Optional[list[Node]] = None, + nodes: list[Node] | None = None, ) -> bool: """Send a supports_cc command to a multicast endpoint.""" result = await _async_send_command( @@ -100,7 +102,7 @@ async def async_multicast_endpoint_get_cc_version( client: Client, endpoint: int, command_class: CommandClass, - nodes: Optional[list[Node]] = None, + nodes: list[Node] | None = None, ) -> int: """Send a get_cc_version command to a multicast endpoint.""" result = await _async_send_command( @@ -119,8 +121,8 @@ async def async_multicast_endpoint_invoke_cc_api( endpoint: int, command_class: CommandClass, method_name: str, - args: Optional[list[Any]] = None, - nodes: Optional[list[Node]] = None, + args: list[Any] | None = None, + nodes: list[Node] | None = None, ) -> Any: """Send a invoke_cc_api command to a multicast endpoint.""" result = await _async_send_command( @@ -140,7 +142,7 @@ async def async_multicast_endpoint_supports_cc_api( client: Client, endpoint: int, command_class: CommandClass, - nodes: Optional[list[Node]] = None, + nodes: list[Node] | None = None, ) -> bool: """Send a supports_cc_api command to a multicast endpoint.""" result = await _async_send_command( diff --git a/zwave_js_server/util/node.py b/zwave_js_server/util/node.py index 9ab23ab70..d098eb7bd 100644 --- a/zwave_js_server/util/node.py +++ b/zwave_js_server/util/node.py @@ -1,7 +1,9 @@ """Utility functions for Z-Wave JS nodes.""" +from __future__ import annotations + import json import logging -from typing import Optional, Union, cast +from typing import cast from ..const import CommandClass, CommandStatus, ConfigurationValueType from ..exceptions import ( @@ -26,9 +28,9 @@ def partial_param_bit_shift(property_key: int) -> int: async def async_set_config_parameter( node: Node, - new_value: Union[int, str], - property_or_property_name: Union[int, str], - property_key: Optional[Union[int, str]] = None, + new_value: int | str, + property_or_property_name: int | str, + property_key: int | str | None = None, ) -> tuple[ConfigurationValue, CommandStatus]: """ Set a value for a config parameter on this node. @@ -85,9 +87,7 @@ async def async_set_config_parameter( async def async_bulk_set_partial_config_parameters( - node: Node, - property_: int, - new_value: Union[int, dict[Union[int, str], Union[int, str]]], + node: Node, property_: int, new_value: int | dict[int | str, int | str] ) -> CommandStatus: """Bulk set partial configuration values on this node.""" config_values = node.get_configuration_values() @@ -159,7 +159,7 @@ async def async_bulk_set_partial_config_parameters( def _validate_and_transform_new_value( - zwave_value: ConfigurationValue, new_value: Union[int, str] + zwave_value: ConfigurationValue, new_value: int | str ) -> int: """Validate a new value and return the integer value to set.""" # Validate that new value for enumerated configuration parameter is a valid state @@ -228,9 +228,7 @@ def _validate_and_transform_new_value( def _bulk_set_validate_and_transform_new_value( - zwave_value: ConfigurationValue, - property_key: int, - new_partial_value: Union[int, str], + zwave_value: ConfigurationValue, property_key: int, new_partial_value: int | str ) -> int: """ Validate and transform new value for a bulk set function call. @@ -250,7 +248,7 @@ def _get_int_from_partials_dict( node: Node, partial_param_values: dict[str, ConfigurationValue], property_: int, - new_value: dict[Union[int, str], Union[int, str]], + new_value: dict[int | str, int | str], ) -> int: """Take an input dict for a set of partial values and compute the raw int value.""" int_value = 0 diff --git a/zwave_js_server/version.py b/zwave_js_server/version.py index 815be3c2a..cc6a68888 100644 --- a/zwave_js_server/version.py +++ b/zwave_js_server/version.py @@ -1,4 +1,6 @@ """Version helper.""" +from __future__ import annotations + import aiohttp from .model.version import VersionInfo