feat: support for s200 buttons (#139)

* support for s200 buttons

* bump to 3.12.0

plugp100/__init__.py

@@ -16,4 +16,4 @@
 from plugp100.api import *
 from plugp100.common import *
 
-__version__ = "3.11.0"
+__version__ = "3.12.0"

plugp100/api/hub/hub_child_device.py

@@ -0,0 +1,31 @@
+from typing import Union, Optional, Any
+
+from plugp100.api.hub.hub_device import HubDevice
+from plugp100.api.hub.s200b_device import S200ButtonDevice
+from plugp100.api.hub.switch_child_device import SwitchChildDevice
+from plugp100.api.hub.t100_device import T100MotionSensor
+from plugp100.api.hub.t110_device import T110SmartDoor
+from plugp100.api.hub.t31x_device import T31Device
+
+HubChildDevice = Union[
+    T100MotionSensor, T110SmartDoor, T31Device, S200ButtonDevice, SwitchChildDevice
+]
+
+
+def create_hub_child_device(
+    hub: HubDevice, child_state: dict[str, Any]
+) -> Optional[HubChildDevice]:
+    model = child_state.get("model").lower()
+    device_id = child_state.get("device_id")
+    if "t31" in model:
+        return T31Device(hub, device_id)
+    elif "t110" in model:
+        return T110SmartDoor(hub, device_id)
+    elif "s200" in model:
+        return S200ButtonDevice(hub, device_id)
+    elif "t100" in model:
+        return T100MotionSensor(hub, device_id)
+    elif any(supported in model for supported in ["s200", "s210"]):
+        return SwitchChildDevice(hub, device_id)
+    else:
+        return None

plugp100/api/hub/switch_child_device.py

@@ -0,0 +1,40 @@
+from plugp100.api.hub.hub_device import HubDevice
+
+from plugp100.common.functional.tri import Try
+from plugp100.common.utils.json_utils import dataclass_encode_json
+from plugp100.requests.set_device_info.set_plug_info_params import SetPlugInfoParams
+from plugp100.requests.tapo_request import TapoRequest
+from plugp100.responses.hub_childs.switch_child_device_state import SwitchChildDeviceState
+
+
+class SwitchChildDevice:
+    def __init__(self, hub: HubDevice, device_id: str):
+        self._hub = hub
+        self._device_id = device_id
+
+    async def get_device_info(self) -> Try[SwitchChildDeviceState]:
+        """
+        The function `get_device_info` sends a request to retrieve device information and returns either the device state or
+        an exception.
+        @  return: an instance of the `Either` class, which can hold either an instance of `S200BDeviceState` or an instance
+        of `Exception`.
+        """
+        return (
+            await self._hub.control_child(self._device_id, TapoRequest.get_device_info())
+        ).flat_map(SwitchChildDeviceState.try_from_json)
+
+    async def on(self) -> Try[bool]:
+        request = TapoRequest.set_device_info(
+            dataclass_encode_json(SetPlugInfoParams(device_on=True))
+        )
+        return (await self._hub.control_child(self._device_id, request)).map(
+            lambda _: True
+        )
+
+    async def off(self) -> Try[bool]:
+        request = TapoRequest.set_device_info(
+            dataclass_encode_json(SetPlugInfoParams(device_on=False))
+        )
+        return (await self._hub.control_child(self._device_id, request)).map(
+            lambda _: True
+        )

plugp100/responses/child_device_list.py

@@ -2,6 +2,8 @@
 from dataclasses import dataclass
 from typing import Any, Set, Callable, List, TypeVar
 
+from plugp100.responses.hub_childs.hub_child_base_info import HubChildBaseInfo
+
 Child = TypeVar("Child")
 
 
@@ -26,17 +28,29 @@ def get_device_ids(self) -> Set[str]:
             if child.get("device_id", None) is not None
         }
 
-    def find_device(self, model_like: str) -> Set[str]:
-        return {
-            child.get("device_id")
-            for child in self.child_device_list
-            if child.get("device_id", None) is not None
-            and model_like.lower() in child.get("model", "").lower()
-        }
+    def find_device(self, model_like: str) -> dict[str, Any]:
+        return next(
+            (
+                child
+                for child in self.child_device_list
+                if child.get("device_id", None) is not None
+                and model_like.lower() in child.get("model", "").lower()
+            ),
+        )
 
     def get_children(self, parse: Callable[[dict[str, Any]], Child]) -> List[Child]:
         return list(map(lambda x: parse(x), self.child_device_list))
 
+    def get_children_base_info(self) -> List[HubChildBaseInfo]:
+        return list(
+            filter(
+                lambda x: x is not None,
+                self.get_children(
+                    lambda x: HubChildBaseInfo.from_json(x).get_or_else(None)
+                ),
+            )
+        )
+
 
 @dataclass
 class PowerStripChild:

plugp100/responses/hub_childs/hub_child_base_info.py

@@ -0,0 +1,51 @@
+import base64
+from typing import Any
+
+import semantic_version
+
+from plugp100.common.functional.tri import Try
+
+
+class HubChildBaseInfo:
+    hardware_version: str
+    firmware_version: str
+    device_id: str
+    parent_device_id: str
+    mac: str
+    type: str
+    model: str
+    status: str
+    rssi: int
+    signal_level: int
+    at_low_battery: bool
+    nickname: str
+    last_onboarding_timestamp: int
+
+    @staticmethod
+    def from_json(kwargs: dict[str, Any]) -> Try["HubChildBaseInfo"]:
+        return Try.of(lambda: HubChildBaseInfo(**kwargs))
+
+    def __init__(self, **kwargs):
+        self.firmware_version = kwargs["fw_ver"]
+        self.hardware_version = kwargs["hw_ver"]
+        self.device_id = kwargs["device_id"]
+        self.parent_device_id = kwargs["parent_device_id"]
+        self.mac = kwargs["mac"]
+        self.type = kwargs["type"]
+        self.model = kwargs["model"]
+        self.status = kwargs.get("status", False)
+        self.rssi = kwargs.get("rssi", 0)
+        self.signal_level = kwargs.get("signal_level", 0)
+        self.at_low_battery = kwargs.get("at_low_battery", False)
+        self.nickname = base64.b64decode(kwargs["nickname"]).decode("UTF-8")
+        self.last_onboarding_timestamp = kwargs.get("lastOnboardingTimestamp", 0)
+
+    def get_semantic_firmware_version(self) -> semantic_version.Version:
+        pieces = self.firmware_version.split("Build")
+        try:
+            if len(pieces) > 0:
+                return semantic_version.Version(pieces[0].strip())
+            else:
+                return semantic_version.Version("0.0.0")
+        except ValueError:
+            return semantic_version.Version("0.0.0")

plugp100/responses/hub_childs/s200b_device_state.py

@@ -1,60 +1,26 @@
-import base64
 from dataclasses import dataclass
 from typing import Any, Union
 
-import semantic_version
-
 from plugp100.common.functional.tri import Try
+from plugp100.responses.hub_childs.hub_child_base_info import HubChildBaseInfo
 
 
 @dataclass
 class S200BDeviceState:
-    hardware_version: str
-    firmware_version: str
-    device_id: str
-    parent_device_id: str
-    mac: str
-    type: str
-    model: str
-    status: str
-    rssi: int
-    signal_level: int
-    at_low_battery: bool
-    nickname: str
-    last_onboarding_timestamp: int
+    base_info: HubChildBaseInfo
     report_interval_seconds: int  # Seconds between each report
 
     @staticmethod
     def try_from_json(kwargs: dict[str, Any]) -> Try["S200BDeviceState"]:
-        return Try.of(
-            lambda: S200BDeviceState(
-                firmware_version=kwargs["fw_ver"],
-                hardware_version=kwargs["hw_ver"],
-                device_id=kwargs["device_id"],
-                parent_device_id=kwargs["parent_device_id"],
-                mac=kwargs["mac"],
-                type=kwargs["type"],
-                model=kwargs["model"],
-                status=kwargs.get("status", False),
-                rssi=kwargs.get("rssi", 0),
-                signal_level=kwargs.get("signal_level", 0),
-                at_low_battery=kwargs.get("at_low_battery", False),
-                nickname=base64.b64decode(kwargs["nickname"]).decode("UTF-8"),
-                last_onboarding_timestamp=kwargs.get("lastOnboardingTimestamp", 0),
-                report_interval_seconds=kwargs.get("report_interval", 0),
+        return HubChildBaseInfo.from_json(kwargs).flat_map(
+            lambda base_info: Try.of(
+                lambda: S200BDeviceState(
+                    base_info=base_info,
+                    report_interval_seconds=kwargs.get("report_interval", 0),
+                )
             )
         )
 
-    def get_semantic_firmware_version(self) -> semantic_version.Version:
-        pieces = self.firmware_version.split("Build")
-        try:
-            if len(pieces) > 0:
-                return semantic_version.Version(pieces[0].strip())
-            else:
-                return semantic_version.Version("0.0.0")
-        except ValueError:
-            return semantic_version.Version("0.0.0")
-
 
 @dataclass
 class RotationEvent:

plugp100/responses/hub_childs/switch_child_device_state.py

@@ -0,0 +1,24 @@
+from dataclasses import dataclass
+from typing import Any
+
+from plugp100.common.functional.tri import Try
+from plugp100.responses.hub_childs.hub_child_base_info import HubChildBaseInfo
+
+
+@dataclass
+class SwitchChildDeviceState:
+    base_info: HubChildBaseInfo
+    device_on: bool
+    led_off: int
+
+    @staticmethod
+    def try_from_json(kwargs: dict[str, Any]) -> Try["SwitchChildDeviceState"]:
+        return HubChildBaseInfo.from_json(kwargs).flat_map(
+            lambda base_info: Try.of(
+                lambda: SwitchChildDeviceState(
+                    base_info=base_info,
+                    device_on=kwargs["device_on"],
+                    led_off=kwargs.get("led_off", 0),
+                )
+            )
+        )

plugp100/responses/hub_childs/t100_device_state.py

@@ -1,62 +1,28 @@
-import base64
 from dataclasses import dataclass
 from typing import Any
 
-import semantic_version
-
 from plugp100.common.functional.tri import Try
+from plugp100.responses.hub_childs.hub_child_base_info import HubChildBaseInfo
 
 
 @dataclass
 class T100MotionSensorState:
-    hardware_version: str
-    firmware_version: str
-    device_id: str
-    parent_device_id: str
-    mac: str
-    type: str
-    model: str
-    status: str
-    rssi: int
-    signal_level: int
-    at_low_battery: bool
-    nickname: str
-    last_onboarding_timestamp: int
+    base_info: HubChildBaseInfo
     report_interval_seconds: int  # Seconds between each report
     detected: bool
 
     @staticmethod
     def from_json(kwargs: dict[str, Any]) -> Try["T100MotionSensorState"]:
-        return Try.of(
-            lambda: T100MotionSensorState(
-                firmware_version=kwargs["fw_ver"],
-                hardware_version=kwargs["hw_ver"],
-                device_id=kwargs["device_id"],
-                parent_device_id=kwargs["parent_device_id"],
-                mac=kwargs["mac"],
-                type=kwargs["type"],
-                model=kwargs["model"],
-                status=kwargs.get("status", False),
-                rssi=kwargs.get("rssi", 0),
-                signal_level=kwargs.get("signal_level", 0),
-                at_low_battery=kwargs.get("at_low_battery", False),
-                nickname=base64.b64decode(kwargs["nickname"]).decode("UTF-8"),
-                last_onboarding_timestamp=kwargs.get("lastOnboardingTimestamp", 0),
-                report_interval_seconds=kwargs.get("report_interval", 0),
-                detected=kwargs.get("detected"),
+        return HubChildBaseInfo.from_json(kwargs).flat_map(
+            lambda base_info: Try.of(
+                lambda: T100MotionSensorState(
+                    base_info=base_info,
+                    report_interval_seconds=kwargs.get("report_interval", 0),
+                    detected=kwargs.get("detected"),
+                )
             )
         )
 
-    def get_semantic_firmware_version(self) -> semantic_version.Version:
-        pieces = self.firmware_version.split("Build")
-        try:
-            if len(pieces) > 0:
-                return semantic_version.Version(pieces[0].strip())
-            else:
-                return semantic_version.Version("0.0.0")
-        except ValueError:
-            return semantic_version.Version("0.0.0")
-
 
 @dataclass
 class MotionDetectedEvent: