Skip to content

Commit b87ac89

Browse files
TheJulianJESTheJulianJES
authored
Improve typing in Tuya classes (#3904)
1 parent b7db677 commit b87ac89

File tree

3 files changed

+82
-108
lines changed

3 files changed

+82
-108
lines changed

‎zhaquirks/tuya/__init__.py

+49-64
Original file line numberDiff line numberDiff line change
@@ -1,14 +1,17 @@
11
"""Tuya devices."""
22

3+
from __future__ import annotations
4+
35
from collections.abc import Callable
46
import dataclasses
57
import datetime
68
import enum
79
import logging
8-
from typing import Any, Optional, Union
10+
from typing import Any
911

1012
from zigpy.quirks import BaseCustomDevice, CustomCluster, CustomDevice
1113
import zigpy.types as t
14+
from zigpy.typing import AddressingMode
1215
from zigpy.zcl import BaseAttributeDefs, foundation
1316
from zigpy.zcl.clusters.closures import WindowCovering
1417
from zigpy.zcl.clusters.general import Basic, LevelControl, OnOff, PowerConfiguration
@@ -155,16 +158,16 @@ class TuyaData(t.Struct):
155158
@property
156159
def payload(
157160
self,
158-
) -> Union[
159-
t.int32s_be,
160-
t.Bool,
161-
t.CharacterString,
162-
t.enum8,
163-
t.bitmap8,
164-
t.bitmap16,
165-
t.bitmap32,
166-
t.LVBytes,
167-
]:
161+
) -> (
162+
t.int32s_be
163+
| t.Bool
164+
| t.CharacterString
165+
| t.enum8
166+
| t.bitmap8
167+
| t.bitmap16
168+
| t.bitmap32
169+
| t.LVBytes
170+
):
168171
"""Payload accordingly to data point type."""
169172
if self.dp_type == TuyaDPType.VALUE:
170173
return t.int32s_be.deserialize(self.raw)[0]
@@ -294,11 +297,11 @@ class NoManufacturerCluster(CustomCluster):
294297

295298
async def command(
296299
self,
297-
command_id: Union[foundation.GeneralCommand, int, t.uint8_t],
300+
command_id: foundation.GeneralCommand | int | t.uint8_t,
298301
*args,
299-
manufacturer: Optional[Union[int, t.uint16_t]] = None,
302+
manufacturer: int | t.uint16_t | None = None,
300303
expect_reply: bool = True,
301-
tsn: Optional[Union[int, t.uint8_t]] = None,
304+
tsn: int | t.uint8_t | None = None,
302305
**kwargs: Any,
303306
):
304307
"""Override the default Cluster command."""
@@ -410,11 +413,9 @@ def tuya_mcu_command(self, command: Command):
410413
def handle_cluster_request(
411414
self,
412415
hdr: foundation.ZCLHeader,
413-
args: tuple,
416+
args: list[Any],
414417
*,
415-
dst_addressing: Optional[
416-
Union[t.Addressing.Group, t.Addressing.IEEE, t.Addressing.NWK]
417-
] = None,
418+
dst_addressing: AddressingMode | None = None,
418419
) -> None:
419420
"""Handle time request."""
420421

@@ -463,11 +464,9 @@ class TuyaManufClusterAttributes(TuyaManufCluster):
463464
def handle_cluster_request(
464465
self,
465466
hdr: foundation.ZCLHeader,
466-
args: tuple,
467+
args: list[Any],
467468
*,
468-
dst_addressing: Optional[
469-
Union[t.Addressing.Group, t.Addressing.IEEE, t.Addressing.NWK]
470-
] = None,
469+
dst_addressing: AddressingMode | None = None,
471470
) -> None:
472471
"""Handle cluster request."""
473472
if hdr.command_id not in (0x0001, 0x0002):
@@ -600,11 +599,12 @@ def switch_event(self, channel, state):
600599

601600
async def command(
602601
self,
603-
command_id: Union[foundation.GeneralCommand, int, t.uint8_t],
602+
command_id: foundation.GeneralCommand | int | t.uint8_t,
604603
*args,
605-
manufacturer: Optional[Union[int, t.uint16_t]] = None,
604+
manufacturer: int | t.uint16_t | None = None,
606605
expect_reply: bool = True,
607-
tsn: Optional[Union[int, t.uint8_t]] = None,
606+
tsn: int | t.uint8_t | None = None,
607+
**kwargs: Any,
608608
):
609609
"""Override the default Cluster command."""
610610

@@ -636,11 +636,9 @@ class TuyaManufacturerClusterOnOff(TuyaManufCluster):
636636
def handle_cluster_request(
637637
self,
638638
hdr: foundation.ZCLHeader,
639-
args: tuple[TuyaManufCluster.Command],
639+
args: list[Any],
640640
*,
641-
dst_addressing: Optional[
642-
Union[t.Addressing.Group, t.Addressing.IEEE, t.Addressing.NWK]
643-
] = None,
641+
dst_addressing: AddressingMode | None = None,
644642
) -> None:
645643
"""Handle cluster request."""
646644

@@ -757,11 +755,12 @@ async def write_attributes(self, attributes, manufacturer=None):
757755
# pylint: disable=W0236
758756
async def command(
759757
self,
760-
command_id: Union[foundation.GeneralCommand, int, t.uint8_t],
758+
command_id: foundation.GeneralCommand | int | t.uint8_t,
761759
*args,
762-
manufacturer: Optional[Union[int, t.uint16_t]] = None,
760+
manufacturer: int | t.uint16_t | None = None,
763761
expect_reply: bool = True,
764-
tsn: Optional[Union[int, t.uint8_t]] = None,
762+
tsn: int | t.uint8_t | None = None,
763+
**kwargs: Any,
765764
):
766765
"""Implement thermostat commands."""
767766

@@ -1052,10 +1051,8 @@ def handle_cluster_request(
10521051
hdr: foundation.ZCLHeader,
10531052
args: list[Any],
10541053
*,
1055-
dst_addressing: Optional[
1056-
Union[t.Addressing.Group, t.Addressing.IEEE, t.Addressing.NWK]
1057-
] = None,
1058-
):
1054+
dst_addressing: AddressingMode | None = None,
1055+
) -> None:
10591056
"""Handle press_types command."""
10601057
# normally if default response sent, TS004x wouldn't send such repeated zclframe (with same sequence number),
10611058
# but for stability reasons (e. g. the case the response doesn't arrive the device), we can simply ignore it
@@ -1152,11 +1149,9 @@ class TuyaManufacturerWindowCover(TuyaManufCluster):
11521149
def handle_cluster_request(
11531150
self,
11541151
hdr: foundation.ZCLHeader,
1155-
args: tuple[TuyaManufCluster.Command],
1152+
args: list[Any],
11561153
*,
1157-
dst_addressing: Optional[
1158-
Union[t.Addressing.Group, t.Addressing.IEEE, t.Addressing.NWK]
1159-
] = None,
1154+
dst_addressing: AddressingMode | None = None,
11601155
) -> None:
11611156
"""Handle cluster request."""
11621157
# Tuya Specific Cluster Commands
@@ -1246,11 +1241,12 @@ def cover_event(self, attribute, value):
12461241

12471242
def command(
12481243
self,
1249-
command_id: Union[foundation.GeneralCommand, int, t.uint8_t],
1244+
command_id: foundation.GeneralCommand | int | t.uint8_t,
12501245
*args,
1251-
manufacturer: Optional[Union[int, t.uint16_t]] = None,
1246+
manufacturer: int | t.uint16_t | None = None,
12521247
expect_reply: bool = True,
1253-
tsn: Optional[Union[int, t.uint8_t]] = None,
1248+
tsn: int | t.uint8_t | None = None,
1249+
**kwargs: Any,
12541250
):
12551251
"""Override the default Cluster command."""
12561252
if manufacturer is None:
@@ -1353,11 +1349,9 @@ class TuyaManufacturerLevelControl(TuyaManufCluster):
13531349
def handle_cluster_request(
13541350
self,
13551351
hdr: foundation.ZCLHeader,
1356-
args: tuple[TuyaManufCluster.Command],
1352+
args: list[Any],
13571353
*,
1358-
dst_addressing: Optional[
1359-
Union[t.Addressing.Group, t.Addressing.IEEE, t.Addressing.NWK]
1360-
] = None,
1354+
dst_addressing: AddressingMode | None = None,
13611355
) -> None:
13621356
"""Handle cluster request."""
13631357
tuya_payload = args[0]
@@ -1411,11 +1405,11 @@ def level_event(self, channel, state):
14111405

14121406
def command(
14131407
self,
1414-
command_id: Union[foundation.GeneralCommand, int, t.uint8_t],
1408+
command_id: foundation.GeneralCommand | int | t.uint8_t,
14151409
*args,
1416-
manufacturer: Optional[Union[int, t.uint16_t]] = None,
1410+
manufacturer: int | t.uint16_t | None = None,
14171411
expect_reply: bool = True,
1418-
tsn: Optional[Union[int, t.uint8_t]] = None,
1412+
tsn: int | t.uint8_t | None = None,
14191413
**kwargs: Any,
14201414
):
14211415
"""Override the default Cluster command."""
@@ -1458,16 +1452,9 @@ class DPToAttributeMapping:
14581452
"""Container for datapoint to cluster attribute update mapping."""
14591453

14601454
ep_attribute: str
1461-
attribute_name: Union[str, tuple]
1462-
converter: Optional[
1463-
Callable[
1464-
[
1465-
Any,
1466-
],
1467-
Any,
1468-
]
1469-
] = None
1470-
endpoint_id: Optional[int] = None
1455+
attribute_name: str | tuple[str, ...]
1456+
converter: Callable[[Any], Any] | None = None
1457+
endpoint_id: int | None = None
14711458

14721459

14731460
@dataclasses.dataclass
@@ -1574,11 +1561,9 @@ def __init__(self, *args, **kwargs):
15741561
def handle_cluster_request(
15751562
self,
15761563
hdr: foundation.ZCLHeader,
1577-
args: tuple,
1564+
args: list[Any],
15781565
*,
1579-
dst_addressing: Optional[
1580-
Union[t.Addressing.Group, t.Addressing.IEEE, t.Addressing.NWK]
1581-
] = None,
1566+
dst_addressing: AddressingMode | None = None,
15821567
) -> None:
15831568
"""Handle cluster specific request."""
15841569

‎zhaquirks/tuya/builder/__init__.py

+14-14
Original file line numberDiff line numberDiff line change
@@ -6,7 +6,7 @@
66
import math
77
import pathlib
88
from types import FrameType
9-
from typing import Any, Optional
9+
from typing import Any
1010

1111
from zigpy.quirks import _DEVICE_REGISTRY
1212
from zigpy.quirks.registry import DeviceRegistry
@@ -256,7 +256,7 @@ def tuya_illuminance(
256256
self,
257257
dp_id: int,
258258
illuminance_cfg: TuyaLocalCluster = TuyaIlluminance,
259-
converter: Optional[Callable[[Any], Any]] = (
259+
converter: Callable[[Any], Any] | None = (
260260
lambda x: 10000 * math.log10(x) + 1 if x != 0 else 0
261261
),
262262
) -> QuirkBuilder:
@@ -353,7 +353,7 @@ def tuya_ias(
353353
self,
354354
dp_id: int,
355355
ias_cfg: TuyaLocalCluster,
356-
converter: Optional[Callable[[Any], Any]] = None,
356+
converter: Callable[[Any], Any] | None = None,
357357
) -> QuirkBuilder:
358358
"""Add a Tuya IAS Configuration."""
359359
self.tuya_dp(
@@ -496,11 +496,11 @@ def tuya_dp(
496496
dp_id: int,
497497
ep_attribute: str,
498498
attribute_name: str,
499-
converter: Optional[Callable[[Any], Any]] = None,
500-
dp_converter: Optional[Callable[[Any], Any]] = None,
501-
endpoint_id: Optional[int] = None,
499+
converter: Callable[[Any], Any] | None = None,
500+
dp_converter: Callable[[Any], Any] | None = None,
501+
endpoint_id: int | None = None,
502502
dp_handler: str = "_dp_2_attr_update",
503-
) -> QuirkBuilder: # fmt: skip
503+
) -> QuirkBuilder:
504504
"""Add Tuya DP Converter."""
505505

506506
self.tuya_dp_multi(
@@ -538,14 +538,14 @@ def tuya_dp_attribute(
538538
dp_id: int,
539539
attribute_name: str,
540540
ep_attribute: str = TuyaMCUCluster.ep_attribute,
541-
converter: Optional[Callable[[Any], Any]] = None,
542-
dp_converter: Optional[Callable[[Any], Any]] = None,
543-
endpoint_id: Optional[int] = None,
541+
converter: Callable[[Any], Any] | None = None,
542+
dp_converter: Callable[[Any], Any] | None = None,
543+
endpoint_id: int | None = None,
544544
dp_handler: str = "_dp_2_attr_update",
545545
type: type = t.uint16_t,
546546
access: foundation.ZCLAttributeAccess = foundation.ZCLAttributeAccess.NONE,
547547
is_manufacturer_specific=True,
548-
) -> QuirkBuilder: # fmt: skip
548+
) -> QuirkBuilder:
549549
"""Add an Tuya DataPoint and corresponding AttributeDef."""
550550
self.tuya_attribute(
551551
dp_id=dp_id,
@@ -742,8 +742,8 @@ def tuya_sensor(
742742
dp_id: int,
743743
attribute_name: str,
744744
type: type,
745-
converter: Optional[Callable[[Any], Any]] = None,
746-
dp_converter: Optional[Callable[[Any], Any]] = None,
745+
converter: Callable[[Any], Any] | None = None,
746+
dp_converter: Callable[[Any], Any] | None = None,
747747
endpoint_id: int = 1,
748748
divisor: int = 1,
749749
multiplier: int = 1,
@@ -755,7 +755,7 @@ def tuya_sensor(
755755
attribute_initialized_from_cache: bool = True,
756756
translation_key: str | None = None,
757757
fallback_name: str | None = None,
758-
) -> QuirkBuilder: # fmt: skip
758+
) -> QuirkBuilder:
759759
"""Add an EntityMetadata containing ZCLSensorMetadata and return self.
760760
761761
This method allows exposing a sensor entity in Home Assistant.

0 commit comments

Comments
 (0)