Add mashumaro for typing of the devices

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]

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,52 +15,47 @@
 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"
 
 _LOGGER = logging.getLogger(__name__)
 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."""

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"

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