Skip to content

Commit

Permalink
Improve other Tuya typing
Browse files Browse the repository at this point in the history
  • Loading branch information
@TheJulianJES
TheJulianJES committed Feb 26, 2025
1 parent 1f5b0e0 commit d625d1c
Show file tree
Hide file tree
Showing 2 changed files with 57 additions and 66 deletions.
104 changes: 47 additions & 57 deletions zhaquirks/tuya/__init__.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -5,10 +5,11 @@
import datetime
import enum
import logging
from typing import Any, Optional, Union
from typing import Any

from zigpy.quirks import BaseCustomDevice, CustomCluster, CustomDevice
import zigpy.types as t
from zigpy.typing import AddressingMode
from zigpy.zcl import BaseAttributeDefs, foundation
from zigpy.zcl.clusters.closures import WindowCovering
from zigpy.zcl.clusters.general import Basic, LevelControl, OnOff, PowerConfiguration
Expand Down Expand Up @@ -155,16 +156,16 @@ class TuyaData(t.Struct):
@property
def payload(
self,
) -> Union[
t.int32s_be,
t.Bool,
t.CharacterString,
t.enum8,
t.bitmap8,
t.bitmap16,
t.bitmap32,
t.LVBytes,
]:
) -> (
t.int32s_be
| t.Bool
| t.CharacterString
| t.enum8
| t.bitmap8
| t.bitmap16
| t.bitmap32
| t.LVBytes
):
"""Payload accordingly to data point type."""
if self.dp_type == TuyaDPType.VALUE:
return t.int32s_be.deserialize(self.raw)[0]
Expand Down Expand Up @@ -294,11 +295,11 @@ class NoManufacturerCluster(CustomCluster):

async def command(
self,
command_id: Union[foundation.GeneralCommand, int, t.uint8_t],
command_id: foundation.GeneralCommand | int | t.uint8_t,
*args,
manufacturer: Optional[Union[int, t.uint16_t]] = None,
manufacturer: int | t.uint16_t | None = None,
expect_reply: bool = True,
tsn: Optional[Union[int, t.uint8_t]] = None,
tsn: int | t.uint8_t | None = None,
**kwargs: Any,
):
"""Override the default Cluster command."""
Expand Down Expand Up @@ -410,11 +411,9 @@ def tuya_mcu_command(self, command: Command):
def handle_cluster_request(
self,
hdr: foundation.ZCLHeader,
args: tuple,
args: list[Any],
*,
dst_addressing: Optional[
Union[t.Addressing.Group, t.Addressing.IEEE, t.Addressing.NWK]
] = None,
dst_addressing: AddressingMode | None = None,
) -> None:
"""Handle time request."""

Expand Down Expand Up @@ -463,11 +462,9 @@ class TuyaManufClusterAttributes(TuyaManufCluster):
def handle_cluster_request(
self,
hdr: foundation.ZCLHeader,
args: tuple,
args: list[Any],
*,
dst_addressing: Optional[
Union[t.Addressing.Group, t.Addressing.IEEE, t.Addressing.NWK]
] = None,
dst_addressing: AddressingMode | None = None,
) -> None:
"""Handle cluster request."""
if hdr.command_id not in (0x0001, 0x0002):
Expand Down Expand Up @@ -600,11 +597,12 @@ def switch_event(self, channel, state):

async def command(
self,
command_id: Union[foundation.GeneralCommand, int, t.uint8_t],
command_id: foundation.GeneralCommand | int | t.uint8_t,
*args,
manufacturer: Optional[Union[int, t.uint16_t]] = None,
manufacturer: int | t.uint16_t | None = None,
expect_reply: bool = True,
tsn: Optional[Union[int, t.uint8_t]] = None,
tsn: int | t.uint8_t | None = None,
**kwargs: Any,
):
"""Override the default Cluster command."""

Expand Down Expand Up @@ -636,11 +634,9 @@ class TuyaManufacturerClusterOnOff(TuyaManufCluster):
def handle_cluster_request(
self,
hdr: foundation.ZCLHeader,
args: tuple[TuyaManufCluster.Command],
args: list[Any],
*,
dst_addressing: Optional[
Union[t.Addressing.Group, t.Addressing.IEEE, t.Addressing.NWK]
] = None,
dst_addressing: AddressingMode | None = None,
) -> None:
"""Handle cluster request."""

Expand Down Expand Up @@ -757,11 +753,12 @@ async def write_attributes(self, attributes, manufacturer=None):
# pylint: disable=W0236
async def command(
self,
command_id: Union[foundation.GeneralCommand, int, t.uint8_t],
command_id: foundation.GeneralCommand | int | t.uint8_t,
*args,
manufacturer: Optional[Union[int, t.uint16_t]] = None,
manufacturer: int | t.uint16_t | None = None,
expect_reply: bool = True,
tsn: Optional[Union[int, t.uint8_t]] = None,
tsn: int | t.uint8_t | None = None,
**kwargs: Any,
):
"""Implement thermostat commands."""

Expand Down Expand Up @@ -1052,10 +1049,8 @@ def handle_cluster_request(
hdr: foundation.ZCLHeader,
args: list[Any],
*,
dst_addressing: Optional[
Union[t.Addressing.Group, t.Addressing.IEEE, t.Addressing.NWK]
] = None,
):
dst_addressing: AddressingMode | None = None,
) -> None:
"""Handle press_types command."""
# normally if default response sent, TS004x wouldn't send such repeated zclframe (with same sequence number),
# but for stability reasons (e. g. the case the response doesn't arrive the device), we can simply ignore it
Expand Down Expand Up @@ -1152,11 +1147,9 @@ class TuyaManufacturerWindowCover(TuyaManufCluster):
def handle_cluster_request(
self,
hdr: foundation.ZCLHeader,
args: tuple[TuyaManufCluster.Command],
args: list[Any],
*,
dst_addressing: Optional[
Union[t.Addressing.Group, t.Addressing.IEEE, t.Addressing.NWK]
] = None,
dst_addressing: AddressingMode | None = None,
) -> None:
"""Handle cluster request."""
# Tuya Specific Cluster Commands
Expand Down Expand Up @@ -1244,13 +1237,14 @@ def cover_event(self, attribute, value):
self._attr_cache,
)

def command(
async def command(
self,
command_id: Union[foundation.GeneralCommand, int, t.uint8_t],
command_id: foundation.GeneralCommand | int | t.uint8_t,
*args,
manufacturer: Optional[Union[int, t.uint16_t]] = None,
manufacturer: int | t.uint16_t | None = None,
expect_reply: bool = True,
tsn: Optional[Union[int, t.uint8_t]] = None,
tsn: int | t.uint8_t | None = None,
**kwargs: Any,
):
"""Override the default Cluster command."""
if manufacturer is None:
Expand Down Expand Up @@ -1353,11 +1347,9 @@ class TuyaManufacturerLevelControl(TuyaManufCluster):
def handle_cluster_request(
self,
hdr: foundation.ZCLHeader,
args: tuple[TuyaManufCluster.Command],
args: list[Any],
*,
dst_addressing: Optional[
Union[t.Addressing.Group, t.Addressing.IEEE, t.Addressing.NWK]
] = None,
dst_addressing: AddressingMode | None = None,
) -> None:
"""Handle cluster request."""
tuya_payload = args[0]
Expand Down Expand Up @@ -1409,13 +1401,13 @@ def level_event(self, channel, state):
)
self._update_attribute(self.attributes_by_name["current_level"].id, level)

def command(
async def command(
self,
command_id: Union[foundation.GeneralCommand, int, t.uint8_t],
command_id: foundation.GeneralCommand | int | t.uint8_t,
*args,
manufacturer: Optional[Union[int, t.uint16_t]] = None,
manufacturer: int | t.uint16_t | None = None,
expect_reply: bool = True,
tsn: Optional[Union[int, t.uint8_t]] = None,
tsn: int | t.uint8_t | None = None,
**kwargs: Any,
):
"""Override the default Cluster command."""
Expand Down Expand Up @@ -1458,7 +1450,7 @@ class DPToAttributeMapping:
"""Container for datapoint to cluster attribute update mapping."""

ep_attribute: str
attribute_name: Union[str, tuple]
attribute_name: str | tuple[str, ...]
converter: Callable[[Any], Any] | None = None
endpoint_id: int | None = None

Expand Down Expand Up @@ -1567,11 +1559,9 @@ def __init__(self, *args, **kwargs):
def handle_cluster_request(
self,
hdr: foundation.ZCLHeader,
args: tuple,
args: list[Any],
*,
dst_addressing: Optional[
Union[t.Addressing.Group, t.Addressing.IEEE, t.Addressing.NWK]
] = None,
dst_addressing: AddressingMode | None = None,
) -> None:
"""Handle cluster specific request."""

Expand Down
19 changes: 10 additions & 9 deletions zhaquirks/tuya/mcu/__init__.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -3,7 +3,7 @@
from collections.abc import Callable
import dataclasses
import datetime
from typing import Any, Optional, Union
from typing import Any

import zigpy.types as t
from zigpy.zcl import foundation
Expand Down Expand Up @@ -39,7 +39,7 @@ class DPToAttributeMapping:
"""Container for datapoint to cluster attribute update mapping."""

ep_attribute: str
attribute_name: Union[str, tuple]
attribute_name: str | tuple[str, ...]
converter: Callable[[Any], Any] | None = None
dp_converter: Callable[[Any], Any] | None = None
endpoint_id: int | None = None
Expand Down Expand Up @@ -273,7 +273,7 @@ def tuya_mcu_command(self, cluster_data: TuyaClusterData):

def get_dp_mapping(
self, endpoint_id: int, attribute_name: str
) -> Optional[tuple[int, DPToAttributeMapping]]:
) -> dict[int, DPToAttributeMapping]:
"""Search for the DP in _dp_to_attributes."""

result = {}
Expand Down Expand Up @@ -351,11 +351,12 @@ class TuyaOnOff(OnOff, TuyaLocalCluster):

async def command(
self,
command_id: Union[foundation.GeneralCommand, int, t.uint8_t],
command_id: foundation.GeneralCommand | int | t.uint8_t,
*args,
manufacturer: Optional[Union[int, t.uint16_t]] = None,
manufacturer: int | t.uint16_t | None = None,
expect_reply: bool = True,
tsn: Optional[Union[int, t.uint8_t]] = None,
tsn: int | t.uint8_t | None = None,
**kwargs: Any,
):
"""Override the default Cluster command."""

Expand Down Expand Up @@ -541,11 +542,11 @@ class TuyaLevelControl(LevelControl, TuyaLocalCluster):

async def command(
self,
command_id: Union[foundation.GeneralCommand, int, t.uint8_t],
command_id: foundation.GeneralCommand | int | t.uint8_t,
*args,
manufacturer: Optional[Union[int, t.uint16_t]] = None,
manufacturer: int | t.uint16_t | None = None,
expect_reply: bool = True,
tsn: Optional[Union[int, t.uint8_t]] = None,
tsn: int | t.uint8_t | None = None,
**kwargs: Any,
):
"""Override the default Cluster command."""
Expand Down

0 comments on commit d625d1c

Please sign in to comment.