diff --git a/plugp100/__init__.py b/plugp100/__init__.py index deb7548..e3aabba 100644 --- a/plugp100/__init__.py +++ b/plugp100/__init__.py @@ -16,4 +16,4 @@ from plugp100.api import * from plugp100.common import * -__version__ = "3.11.0" +__version__ = "3.12.0" diff --git a/plugp100/api/hub/hub_child_device.py b/plugp100/api/hub/hub_child_device.py new file mode 100644 index 0000000..917df65 --- /dev/null +++ b/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 diff --git a/plugp100/api/hub/switch_child_device.py b/plugp100/api/hub/switch_child_device.py new file mode 100644 index 0000000..da0c4fd --- /dev/null +++ b/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 + ) diff --git a/plugp100/responses/child_device_list.py b/plugp100/responses/child_device_list.py index 2731989..ae923da 100644 --- a/plugp100/responses/child_device_list.py +++ b/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: diff --git a/plugp100/responses/hub_childs/hub_child_base_info.py b/plugp100/responses/hub_childs/hub_child_base_info.py new file mode 100644 index 0000000..4a0a1ab --- /dev/null +++ b/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") diff --git a/plugp100/responses/hub_childs/s200b_device_state.py b/plugp100/responses/hub_childs/s200b_device_state.py index 85d10a1..c1d9f2d 100644 --- a/plugp100/responses/hub_childs/s200b_device_state.py +++ b/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: diff --git a/plugp100/responses/hub_childs/switch_child_device_state.py b/plugp100/responses/hub_childs/switch_child_device_state.py new file mode 100644 index 0000000..6094061 --- /dev/null +++ b/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), + ) + ) + ) diff --git a/plugp100/responses/hub_childs/t100_device_state.py b/plugp100/responses/hub_childs/t100_device_state.py index b163362..4112dff 100644 --- a/plugp100/responses/hub_childs/t100_device_state.py +++ b/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: diff --git a/plugp100/responses/hub_childs/t110_device_state.py b/plugp100/responses/hub_childs/t110_device_state.py index b0cece8..dedb15b 100644 --- a/plugp100/responses/hub_childs/t110_device_state.py +++ b/plugp100/responses/hub_childs/t110_device_state.py @@ -1,62 +1,28 @@ -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 T110SmartDoorState: - 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 is_open: bool @staticmethod def try_from_json(kwargs: dict[str, Any]) -> Try["T110SmartDoorState"]: - return Try.of( - lambda: T110SmartDoorState( - 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), - is_open=kwargs.get("open"), + return HubChildBaseInfo.from_json(kwargs).flat_map( + lambda base_info: Try.of( + lambda: T110SmartDoorState( + base_info=base_info, + report_interval_seconds=kwargs.get("report_interval", 0), + is_open=kwargs.get("open"), + ) ) ) - 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 OpenEvent: diff --git a/plugp100/responses/hub_childs/t31x_device_state.py b/plugp100/responses/hub_childs/t31x_device_state.py index d65434b..4f923e7 100644 --- a/plugp100/responses/hub_childs/t31x_device_state.py +++ b/plugp100/responses/hub_childs/t31x_device_state.py @@ -1,12 +1,10 @@ -import base64 import enum from dataclasses import dataclass from datetime import datetime 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 class TemperatureUnit(enum.Enum): @@ -16,19 +14,7 @@ class TemperatureUnit(enum.Enum): @dataclass class T31DeviceState: - 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 current_humidity: int @@ -41,22 +27,14 @@ class T31DeviceState: @staticmethod def from_json(kwargs: dict[str, Any]) -> Try["T31DeviceState"]: - return Try.of(lambda: T31DeviceState(**kwargs)) + return HubChildBaseInfo.from_json(kwargs).flat_map( + lambda base_info: Try.of( + lambda: T31DeviceState(**kwargs, base_info=base_info) + ) + ) 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) + self.base_info = kwargs["base_info"] self.report_interval_seconds = kwargs.get("report_interval", 0) self.current_humidity = kwargs.get("current_humidity") self.current_humidity_exception = kwargs.get("current_humidity_exception") @@ -71,16 +49,6 @@ def __init__(self, **kwargs): TemperatureUnit.CELSIUS, ) - 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 TemperatureHumidityRecordsRaw: diff --git a/tests/test_button_t310.py b/tests/test_button_t310.py index 536c1b9..75ee9d0 100644 --- a/tests/test_button_t310.py +++ b/tests/test_button_t310.py @@ -1,8 +1,7 @@ import unittest +from plugp100.api.hub.hub_child_device import create_hub_child_device from plugp100.api.hub.hub_device import HubDevice -from plugp100.api.hub.t31x_device import T31Device -from plugp100.api.tapo_client import TapoClient from plugp100.responses.hub_childs.t31x_device_state import TemperatureUnit from tests.tapo_test_helper import ( get_test_config, @@ -21,9 +20,8 @@ async def asyncSetUp(self) -> None: credential, ip = await get_test_config(device_type="hub") self._api = await get_initialized_client(credential, ip) self._hub = HubDevice(self._api) - self._device = T31Device( - self._hub, - (await self._hub.get_children()).get_or_raise().find_device("T310").pop(), + self._device = create_hub_child_device( + self._hub, (await self._hub.get_children()).get_or_raise().find_device("T310") ) async def asyncTearDown(self): @@ -31,13 +29,13 @@ async def asyncTearDown(self): async def test_should_get_state(self): state = (await self._device.get_device_state()).get_or_raise() - self.assertIsNotNone(state.parent_device_id) - self.assertIsNotNone(state.device_id) - self.assertIsNotNone(state.mac) - self.assertIsNotNone(state.rssi) - self.assertIsNotNone(state.model) - self.assertIsNotNone(state.get_semantic_firmware_version()) - self.assertIsNotNone(state.nickname) + self.assertIsNotNone(state.base_info.parent_device_id) + self.assertIsNotNone(state.base_info.device_id) + self.assertIsNotNone(state.base_info.mac) + self.assertIsNotNone(state.base_info.rssi) + self.assertIsNotNone(state.base_info.model) + self.assertIsNotNone(state.base_info.get_semantic_firmware_version()) + self.assertIsNotNone(state.base_info.nickname) self.assertIsNotNone(state.current_humidity) self.assertIsNotNone(state.current_temperature) self.assertIsNotNone(state.current_humidity_exception) @@ -45,7 +43,7 @@ async def test_should_get_state(self): self.assertIsNotNone(state.current_temperature_exception) self.assertIsNotNone(state.report_interval_seconds) self.assertEqual(state.temperature_unit, TemperatureUnit.CELSIUS) - self.assertEqual(state.at_low_battery, False) + self.assertEqual(state.base_info.at_low_battery, False) async def test_should_get_temperature_humidity_records(self): state = (await self._device.get_temperature_humidity_records()).get_or_raise() diff --git a/tests/test_hub.py b/tests/test_hub.py index 7ab77fe..426e571 100644 --- a/tests/test_hub.py +++ b/tests/test_hub.py @@ -1,7 +1,6 @@ import unittest from plugp100.api.hub.hub_device import HubDevice -from plugp100.api.tapo_client import TapoClient from tests.tapo_test_helper import ( _test_expose_device_info, get_test_config, @@ -50,3 +49,9 @@ async def test_should_get_supported_alarm_tones(self): async def test_should_get_children(self): state = (await self._device.get_children()).get_or_raise() self.assertTrue(len(state.get_device_ids()) > 0) + + async def test_should_get_base_children_info(self): + children = ( + (await self._device.get_children()).get_or_raise().get_children_base_info() + ) + self.assertTrue(len(children) > 0) diff --git a/tests/test_sensor_s200b.py b/tests/test_sensor_s200b.py index e715fe8..e1ad66f 100644 --- a/tests/test_sensor_s200b.py +++ b/tests/test_sensor_s200b.py @@ -1,8 +1,7 @@ import unittest +from plugp100.api.hub.hub_child_device import create_hub_child_device from plugp100.api.hub.hub_device import HubDevice -from plugp100.api.hub.s200b_device import S200ButtonDevice -from plugp100.api.tapo_client import TapoClient from plugp100.responses.hub_childs.s200b_device_state import ( SingleClickEvent, RotationEvent, @@ -24,9 +23,9 @@ async def asyncSetUp(self) -> None: credential, ip = await get_test_config(device_type="hub") self._api = await get_initialized_client(credential, ip) self._hub = HubDevice(self._api) - self._device = S200ButtonDevice( + self._device = create_hub_child_device( self._hub, - (await self._hub.get_children()).get_or_raise().find_device("S200B").pop(), + (await self._hub.get_children()).get_or_raise().find_device("S200B"), ) async def asyncTearDown(self): @@ -34,16 +33,16 @@ async def asyncTearDown(self): async def test_should_get_state(self): state = (await self._device.get_device_info()).get_or_raise() - self.assertIsNotNone(state.parent_device_id) - self.assertIsNotNone(state.device_id) - self.assertIsNotNone(state.mac) - self.assertIsNotNone(state.rssi) - self.assertIsNotNone(state.model) - self.assertIsNotNone(state.get_semantic_firmware_version()) - self.assertIsNotNone(state.nickname) + self.assertIsNotNone(state.base_info.parent_device_id) + self.assertIsNotNone(state.base_info.device_id) + self.assertIsNotNone(state.base_info.mac) + self.assertIsNotNone(state.base_info.rssi) + self.assertIsNotNone(state.base_info.model) + self.assertIsNotNone(state.base_info.get_semantic_firmware_version()) + self.assertIsNotNone(state.base_info.nickname) self.assertIsNotNone(state.report_interval_seconds) - self.assertEqual(state.at_low_battery, False) - self.assertEqual(state.status, "online") + self.assertEqual(state.base_info.at_low_battery, False) + self.assertEqual(state.base_info.status, "online") async def test_should_get_button_events(self): logs = (await self._device.get_event_logs(100)).get_or_raise()