From 7a755f6bce3faab2ca94b1c4a665ee276318af25 Mon Sep 17 00:00:00 2001 From: Joostlek Date: Tue, 21 Jan 2025 14:11:31 +0100 Subject: [PATCH] Add mashumaro for typing of the devices --- poetry.lock | 34 +++++-- pyproject.toml | 1 + switchbot_api/__init__.py | 82 +++++++--------- switchbot_api/const.py | 11 +++ switchbot_api/models.py | 138 +++++++++++++++++++++++++++ tests/__snapshots__/test_client.ambr | 110 ++++++++++++--------- 6 files changed, 277 insertions(+), 99 deletions(-) create mode 100644 switchbot_api/const.py create mode 100644 switchbot_api/models.py diff --git a/poetry.lock b/poetry.lock index 89b2233..adfdfff 100644 --- a/poetry.lock +++ b/poetry.lock @@ -456,13 +456,13 @@ files = [ [[package]] name = "identify" -version = "2.6.5" +version = "2.6.6" description = "File identification library for Python" optional = false python-versions = ">=3.9" files = [ - {file = "identify-2.6.5-py2.py3-none-any.whl", hash = "sha256:14181a47091eb75b337af4c23078c9d09225cd4c48929f521f3bf16b09d02566"}, - {file = "identify-2.6.5.tar.gz", hash = "sha256:c10b33f250e5bba374fae86fb57f3adcebf1161bce7cdf92031915fd480c13bc"}, + {file = "identify-2.6.6-py2.py3-none-any.whl", hash = "sha256:cbd1810bce79f8b671ecb20f53ee0ae8e86ae84b557de31d89709dc2a48ba881"}, + {file = "identify-2.6.6.tar.gz", hash = "sha256:7bec12768ed44ea4761efb47806f0a41f86e7c0a5fdf5950d4648c90eca7e251"}, ] [package.extras] @@ -493,6 +493,26 @@ files = [ {file = "iniconfig-2.0.0.tar.gz", hash = "sha256:2d91e135bf72d31a410b17c16da610a82cb55f6b0477d1a902134b24a455b8b3"}, ] +[[package]] +name = "mashumaro" +version = "3.15" +description = "Fast and well tested serialization library" +optional = false +python-versions = ">=3.9" +files = [ + {file = "mashumaro-3.15-py3-none-any.whl", hash = "sha256:cdd45ef5a4d09860846a3ee37a4c2f5f4bc70eb158caa55648c4c99451ca6c4c"}, + {file = "mashumaro-3.15.tar.gz", hash = "sha256:32a2a38a1e942a07f2cbf9c3061cb2a247714ee53e36a5958548b66bd116d0a9"}, +] + +[package.dependencies] +typing-extensions = ">=4.1.0" + +[package.extras] +msgpack = ["msgpack (>=0.5.6)"] +orjson = ["orjson"] +toml = ["tomli (>=1.1.0)", "tomli-w (>=1.0)"] +yaml = ["pyyaml (>=3.13)"] + [[package]] name = "multidict" version = "6.1.0" @@ -733,13 +753,13 @@ testing = ["pytest", "pytest-benchmark"] [[package]] name = "pre-commit" -version = "4.0.1" +version = "4.1.0" description = "A framework for managing and maintaining multi-language pre-commit hooks." optional = false python-versions = ">=3.9" files = [ - {file = "pre_commit-4.0.1-py2.py3-none-any.whl", hash = "sha256:efde913840816312445dc98787724647c65473daefe420785f885e8ed9a06878"}, - {file = "pre_commit-4.0.1.tar.gz", hash = "sha256:80905ac375958c0444c65e9cebebd948b3cdb518f335a091a670a89d652139d2"}, + {file = "pre_commit-4.1.0-py2.py3-none-any.whl", hash = "sha256:d29e7cb346295bcc1cc75fc3e92e343495e3ea0196c9ec6ba53f49f10ab6ae7b"}, + {file = "pre_commit-4.1.0.tar.gz", hash = "sha256:ae3f018575a588e30dfddfab9a05448bfbd6b73d78709617b5a2b853549716d4"}, ] [package.dependencies] @@ -1313,4 +1333,4 @@ propcache = ">=0.2.0" [metadata] lock-version = "2.0" python-versions = "^3.10" -content-hash = "93ef926194c8644aac78c5ee192554921c161fc098bc007a915d7cc2c391c114" +content-hash = "52432e4eb781f04e93cb0dd51a9f3365c811d6bb7dffbb00a70ad6b01ae55a49" diff --git a/pyproject.toml b/pyproject.toml index b844926..22a9c83 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -11,6 +11,7 @@ repository = "https://github.com/SeraphicCorp/py-switchbot-api" [tool.poetry.dependencies] python = "^3.10" aiohttp = ">=3.0.0" +mashumaro = "^3.15" [tool.poetry.group.dev.dependencies] diff --git a/switchbot_api/__init__.py b/switchbot_api/__init__.py index daae7ce..cc66980 100644 --- a/switchbot_api/__init__.py +++ b/switchbot_api/__init__.py @@ -3,7 +3,6 @@ from __future__ import annotations import base64 -from dataclasses import dataclass from enum import Enum import hashlib import hmac @@ -16,11 +15,40 @@ from aiohttp import ClientError, ClientResponseError, ClientSession from aiohttp.hdrs import METH_GET, METH_POST +from switchbot_api.const import Model from switchbot_api.exceptions import ( SwitchBotAuthenticationError, SwitchBotConnectionError, SwitchBotDeviceOfflineError, ) +from switchbot_api.models import ( + Curtain, + Curtain3, + Curtain3Status, + CurtainStatus, + Device, + DeviceList, + DeviceStatus, + Hub2, + Hub2Status, + OpenDirection, + Remote, +) + +__all__ = [ + "Curtain", + "Curtain3", + "Curtain3Status", + "CurtainStatus", + "Device", + "DeviceList", + "DeviceStatus", + "Hub2", + "Hub2Status", + "Model", + "OpenDirection", + "Remote", +] _API_HOST = "https://api.switch-bot.com" @@ -28,40 +56,6 @@ NON_OBSERVED_REMOTE_TYPES = ["Others"] -@dataclass -class Device: - """Device.""" - - device_id: str - device_name: str - device_type: str - hub_device_id: str - - def __init__(self, **kwargs: Any) -> None: - """Initialize.""" - self.device_id = kwargs["deviceId"] - self.device_name = kwargs["deviceName"] - self.device_type = kwargs.get("deviceType", "-") - self.hub_device_id = kwargs["hubDeviceId"] - - -@dataclass -class Remote: - """Remote device.""" - - device_id: str - device_name: str - device_type: str - hub_device_id: str - - def __init__(self, **kwargs: Any) -> None: - """Initialize.""" - self.device_id = kwargs["deviceId"] - self.device_name = kwargs["deviceName"] - self.device_type = kwargs.get("remoteType", "-") - self.hub_device_id = kwargs["hubDeviceId"] - - class PowerState(Enum): """Power state.""" @@ -260,21 +254,15 @@ async def _request( _LOGGER.error("Error %d: %s", response.status, body) raise SwitchBotConnectionError - async def list_devices(self) -> list[Device | Remote]: + async def list_devices(self) -> DeviceList: """List devices.""" body = await self._request(METH_GET, "devices") - _LOGGER.debug("Devices: %s", body) - devices = [Device(**device) for device in body.get("deviceList")] # type: ignore[union-attr] - remotes = [ - Remote(**remote) - for remote in body.get("infraredRemoteList") # type: ignore[union-attr] - if remote.get("remoteType") not in NON_OBSERVED_REMOTE_TYPES - ] - return [*devices, *remotes] - - async def get_status(self, device_id: str) -> dict[str, Any]: + return DeviceList.from_dict(body) + + async def get_status(self, device_id: str) -> DeviceStatus: """No status for IR devices.""" - return await self._request(METH_GET, f"devices/{device_id}/status") + body = await self._request(METH_GET, f"devices/{device_id}/status") + return DeviceStatus.from_dict(body) async def get_webook_configuration(self) -> dict[str, Any]: """List webhooks.""" diff --git a/switchbot_api/const.py b/switchbot_api/const.py new file mode 100644 index 0000000..3277e2a --- /dev/null +++ b/switchbot_api/const.py @@ -0,0 +1,11 @@ +"""Constants for SwitchBot API.""" + +from enum import StrEnum + + +class Model(StrEnum): + """Model of SwitchBot device.""" + + HUB_2 = "Hub 2" + CURTAIN = "Curtain" + CURTAIN3 = "Curtain3" diff --git a/switchbot_api/models.py b/switchbot_api/models.py new file mode 100644 index 0000000..dc22b80 --- /dev/null +++ b/switchbot_api/models.py @@ -0,0 +1,138 @@ +"""Models for SwitchBot API.""" + +from __future__ import annotations + +from dataclasses import dataclass, field +from enum import StrEnum +from typing import Annotated, Any + +from mashumaro import field_options +from mashumaro.mixins.json import DataClassJSONMixin +from mashumaro.types import Discriminator + +from switchbot_api.const import Model + + +class OpenDirection(StrEnum): + """Open direction.""" + + LEFT = "left" + RIGHT = "right" + + +@dataclass +class DeviceList(DataClassJSONMixin): + """Device list.""" + + devices: list[ + Annotated[Device, Discriminator(field="deviceType", include_subtypes=True)] + ] = field(metadata=field_options(alias="deviceList")) + remotes: list[Remote] = field(metadata=field_options(alias="infraredRemoteList")) + + +@dataclass +class Device: + """Device.""" + + deviceType: Model # noqa: N815 + device_id: str = field(metadata=field_options(alias="deviceId")) + name: str = field(metadata=field_options(alias="deviceName")) + type: Model = field(metadata=field_options(alias="deviceType")) + hub_device_id: str | None = field(metadata=field_options(alias="hubDeviceId")) + + @classmethod + def __pre_deserialize__(cls, d: dict[str, Any]) -> dict[str, Any]: + """Pre deserialize hook.""" + if d["hubDeviceId"] == "": + d["hubDeviceId"] = None + return d + + +@dataclass +class Hub2(Device): + """Hub2 device.""" + + deviceType = Model.HUB_2 # noqa: N815 + + +@dataclass +class Curtain3(Device): + """Curtain3 device.""" + + deviceType = Model.CURTAIN3 # noqa: N815 + calibrate: bool + open_direction: OpenDirection = field(metadata=field_options(alias="openDirection")) + main: bool = field(metadata=field_options(alias="master")) + + +@dataclass +class Curtain(Device): + """Curtain device.""" + + deviceType = Model.CURTAIN # noqa: N815 + calibrate: bool + group: bool + curtain_devices_ids: list[str] = field( + metadata=field_options(alias="curtainDevicesIds") + ) + + +@dataclass +class Remote: + """Remote device.""" + + device_id: str = field(metadata=field_options(alias="deviceId")) + name: str = field(metadata=field_options(alias="deviceName")) + type: str = field(metadata=field_options(alias="remoteType")) + hub_device_id: str = field(metadata=field_options(alias="hubDeviceId")) + + +@dataclass +class DeviceStatus(DataClassJSONMixin): + """Device status.""" + + class Config: + """Config.""" + + discriminator = Discriminator(field="deviceType", include_subtypes=True) + + deviceType: Model # noqa: N815 + version: str + device_id: str = field(metadata=field_options(alias="deviceId")) + type: Model = field(metadata=field_options(alias="deviceType")) + hub_device_id: str | None = field(metadata=field_options(alias="hubDeviceId")) + + @classmethod + def __pre_deserialize__(cls, d: dict[str, Any]) -> dict[str, Any]: + """Pre deserialize hook.""" + if d["hubDeviceId"] in ["000000000000", d["deviceId"]]: + d["hubDeviceId"] = None + return d + + +@dataclass +class Hub2Status(DeviceStatus): + """Hub2 status.""" + + deviceType = Model.HUB_2 # noqa: N815 + temperature: float + humidity: float + + +@dataclass +class CurtainStatus(DeviceStatus): + """Curtain status.""" + + deviceType = Model.CURTAIN # noqa: N815 + calibrate: bool + group: bool + moving: bool + battery: int + slide_position: int = field(metadata=field_options(alias="slidePosition")) + + +@dataclass +class Curtain3Status(CurtainStatus): + """Hub2 status.""" + + deviceType = Model.CURTAIN3 # noqa: N815 diff --git a/tests/__snapshots__/test_client.ambr b/tests/__snapshots__/test_client.ambr index 22c6776..289cb63 100644 --- a/tests/__snapshots__/test_client.ambr +++ b/tests/__snapshots__/test_client.ambr @@ -1,48 +1,67 @@ # serializer version: 1 # name: test_device_list - list([ - dict({ - 'device_id': 'D82812821F08', - 'device_name': 'Hub 2 08', - 'device_type': 'Hub 2', - 'hub_device_id': '', - }), - dict({ - 'device_id': 'E2562C1DCC1A', - 'device_name': 'Curtain 1A', - 'device_type': 'Curtain3', - 'hub_device_id': 'D82812821F08', - }), - dict({ - 'device_id': 'FCE84BB8B04F', - 'device_name': 'Curtain 4F', - 'device_type': 'Curtain', - 'hub_device_id': '', - }), - dict({ - 'device_id': 'FF711F51601E', - 'device_name': 'Curtain 1E', - 'device_type': 'Curtain3', - 'hub_device_id': 'D82812821F08', - }), - dict({ - 'device_id': '01-202501201703-63754985', - 'device_name': 'TV', - 'device_type': 'DIY TV', - 'hub_device_id': 'D82812821F08', - }), - ]) + dict({ + 'devices': list([ + dict({ + 'deviceType': , + 'device_id': 'D82812821F08', + 'hub_device_id': None, + 'name': 'Hub 2 08', + 'type': , + }), + dict({ + 'calibrate': True, + 'deviceType': , + 'device_id': 'E2562C1DCC1A', + 'hub_device_id': 'D82812821F08', + 'main': True, + 'name': 'Curtain 1A', + 'open_direction': , + 'type': , + }), + dict({ + 'calibrate': True, + 'curtain_devices_ids': list([ + ]), + 'deviceType': , + 'device_id': 'FCE84BB8B04F', + 'group': False, + 'hub_device_id': None, + 'name': 'Curtain 4F', + 'type': , + }), + dict({ + 'calibrate': True, + 'deviceType': , + 'device_id': 'FF711F51601E', + 'hub_device_id': 'D82812821F08', + 'main': True, + 'name': 'Curtain 1E', + 'open_direction': , + 'type': , + }), + ]), + 'remotes': list([ + dict({ + 'device_id': '01-202501201703-63754985', + 'hub_device_id': 'D82812821F08', + 'name': 'TV', + 'type': 'DIY TV', + }), + ]), + }) # --- # name: test_device_status[curtain3] dict({ 'battery': 44, 'calibrate': True, - 'deviceId': 'FF711F51601E', - 'deviceType': 'Curtain3', + 'deviceType': , + 'device_id': 'FF711F51601E', 'group': False, - 'hubDeviceId': 'D82812821F08', + 'hub_device_id': 'D82812821F08', 'moving': False, - 'slidePosition': 0, + 'slide_position': 0, + 'type': , 'version': 'V1.2', }) # --- @@ -50,23 +69,24 @@ dict({ 'battery': 0, 'calibrate': False, - 'deviceId': 'FCE84BB8B04F', - 'deviceType': 'Curtain', + 'deviceType': , + 'device_id': 'FCE84BB8B04F', 'group': False, - 'hubDeviceId': '000000000000', + 'hub_device_id': None, 'moving': False, - 'slidePosition': 0, + 'slide_position': 0, + 'type': , 'version': 'V6.3', }) # --- # name: test_device_status[hub_2] dict({ - 'deviceId': 'D82812821F08', - 'deviceType': 'Hub 2', - 'hubDeviceId': 'D82812821F08', - 'humidity': 52, - 'lightLevel': 1, + 'deviceType': , + 'device_id': 'D82812821F08', + 'hub_device_id': None, + 'humidity': 52.0, 'temperature': 17.8, + 'type': , 'version': 'V2.3-1.4', }) # ---