PEP 604: Use union typing (#609)

* PEP 604: Use union typing

* formatting

* additional formatting

* additional formatting

* line-too-long

* line-too-long

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"',

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)

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

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:

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
 

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"

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
 
 

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
 
 

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"

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

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"

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"

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"

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"

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
 
 

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"

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"

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"

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"

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)

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

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__(

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.

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

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