diff --git a/kasa/experimental/modules/__init__.py b/kasa/experimental/modules/__init__.py new file mode 100644 index 00000000..9f168384 --- /dev/null +++ b/kasa/experimental/modules/__init__.py @@ -0,0 +1,11 @@ +"""Modules for SMARTCAMERA devices.""" + +from .camera import Camera +from .childdevice import ChildDevice +from .device import DeviceModule + +__all__ = [ + "Camera", + "ChildDevice", + "DeviceModule", +] diff --git a/kasa/experimental/modules/camera.py b/kasa/experimental/modules/camera.py new file mode 100644 index 00000000..76701b52 --- /dev/null +++ b/kasa/experimental/modules/camera.py @@ -0,0 +1,45 @@ +"""Implementation of device module.""" + +from __future__ import annotations + +from ...device_type import DeviceType +from ...feature import Feature +from ..smartcameramodule import SmartCameraModule + + +class Camera(SmartCameraModule): + """Implementation of device module.""" + + QUERY_GETTER_NAME = "getLensMaskConfig" + QUERY_MODULE_NAME = "lens_mask" + QUERY_SECTION_NAMES = "lens_mask_info" + + def _initialize_features(self) -> None: + """Initialize features after the initial update.""" + self._add_feature( + Feature( + self._device, + id="state", + name="State", + attribute_getter="is_on", + attribute_setter="set_state", + type=Feature.Type.Switch, + category=Feature.Category.Primary, + ) + ) + + @property + def is_on(self) -> bool: + """Return the device id.""" + return self.data["lens_mask_info"]["enabled"] == "on" + + async def set_state(self, on: bool) -> dict: + """Set the device state.""" + params = {"enabled": "on" if on else "off"} + return await self._device._query_setter_helper( + "setLensMaskConfig", self.QUERY_MODULE_NAME, "lens_mask_info", params + ) + + async def _check_supported(self) -> bool: + """Additional check to see if the module is supported by the device.""" + return self._device.device_type is DeviceType.Camera diff --git a/kasa/experimental/modules/childdevice.py b/kasa/experimental/modules/childdevice.py new file mode 100644 index 00000000..837793f1 --- /dev/null +++ b/kasa/experimental/modules/childdevice.py @@ -0,0 +1,23 @@ +"""Module for child devices.""" + +from ...device_type import DeviceType +from ..smartcameramodule import SmartCameraModule + + +class ChildDevice(SmartCameraModule): + """Implementation for child devices.""" + + NAME = "childdevice" + QUERY_GETTER_NAME = "getChildDeviceList" + QUERY_MODULE_NAME = "childControl" + + def query(self) -> dict: + """Query to execute during the update cycle. + + Default implementation uses the raw query getter w/o parameters. + """ + return {self.QUERY_GETTER_NAME: {self.QUERY_MODULE_NAME: {"start_index": 0}}} + + async def _check_supported(self) -> bool: + """Additional check to see if the module is supported by the device.""" + return self._device.device_type is DeviceType.Hub diff --git a/kasa/experimental/modules/device.py b/kasa/experimental/modules/device.py new file mode 100644 index 00000000..34474ef2 --- /dev/null +++ b/kasa/experimental/modules/device.py @@ -0,0 +1,40 @@ +"""Implementation of device module.""" + +from __future__ import annotations + +from ...feature import Feature +from ..smartcameramodule import SmartCameraModule + + +class DeviceModule(SmartCameraModule): + """Implementation of device module.""" + + NAME = "devicemodule" + QUERY_GETTER_NAME = "getDeviceInfo" + QUERY_MODULE_NAME = "device_info" + QUERY_SECTION_NAMES = ["basic_info", "info"] + + def _initialize_features(self) -> None: + """Initialize features after the initial update.""" + self._add_feature( + Feature( + self._device, + id="device_id", + name="Device ID", + attribute_getter="device_id", + category=Feature.Category.Debug, + type=Feature.Type.Sensor, + ) + ) + + async def _post_update_hook(self) -> None: + """Overriden to prevent module disabling. + + Overrides the default behaviour to disable a module if the query returns + an error because this module is critical. + """ + + @property + def device_id(self) -> str: + """Return the device id.""" + return self.data["basic_info"]["dev_id"] diff --git a/kasa/experimental/smartcamera.py b/kasa/experimental/smartcamera.py index 3224c003..52a6acdf 100644 --- a/kasa/experimental/smartcamera.py +++ b/kasa/experimental/smartcamera.py @@ -2,16 +2,26 @@ from __future__ import annotations +import logging from typing import Any from ..device_type import DeviceType -from ..exceptions import SmartErrorCode -from ..smart import SmartDevice +from ..module import Module +from ..smart import SmartChildDevice, SmartDevice +from .modules.childdevice import ChildDevice +from .modules.device import DeviceModule +from .smartcameramodule import SmartCameraModule +from .smartcameraprotocol import _ChildCameraProtocolWrapper + +_LOGGER = logging.getLogger(__name__) class SmartCamera(SmartDevice): """Class for smart cameras.""" + # Modules that are called as part of the init procedure on first update + FIRST_UPDATE_MODULES = {DeviceModule, ChildDevice} + @staticmethod def _get_device_type_from_sysinfo(sysinfo: dict[str, Any]) -> DeviceType: """Find type to be displayed as a supported device category.""" @@ -20,17 +30,108 @@ class SmartCamera(SmartDevice): return DeviceType.Hub return DeviceType.Camera - async def update(self, update_children: bool = False): - """Update the device.""" + def _update_internal_info(self, info_resp: dict) -> None: + """Update the internal device info.""" + info = self._try_get_response(info_resp, "getDeviceInfo") + self._info = self._map_info(info["device_info"]) + + def _update_children_info(self) -> None: + """Update the internal child device info from the parent info.""" + if child_info := self._try_get_response( + self._last_update, "getChildDeviceList", {} + ): + for info in child_info["child_device_list"]: + self._children[info["device_id"]]._update_internal_state(info) + + async def _initialize_smart_child(self, info: dict) -> SmartDevice: + """Initialize a smart child device attached to a smartcamera.""" + child_id = info["device_id"] + child_protocol = _ChildCameraProtocolWrapper(child_id, self.protocol) + try: + initial_response = await child_protocol.query( + {"component_nego": None, "get_connect_cloud_state": None} + ) + child_components = { + item["id"]: item["ver_code"] + for item in initial_response["component_nego"]["component_list"] + } + except Exception as ex: + _LOGGER.exception("Error initialising child %s: %s", child_id, ex) + + return await SmartChildDevice.create( + parent=self, + child_info=info, + child_components=child_components, + protocol=child_protocol, + last_update=initial_response, + ) + + async def _initialize_children(self) -> None: + """Initialize children for hubs.""" + if not ( + child_info := self._try_get_response( + self._last_update, "getChildDeviceList", {} + ) + ): + return + + children = {} + for info in child_info["child_device_list"]: + if ( + category := info.get("category") + ) and category in SmartChildDevice.CHILD_DEVICE_TYPE_MAP: + child_id = info["device_id"] + children[child_id] = await self._initialize_smart_child(info) + else: + _LOGGER.debug("Child device type not supported: %s", info) + + self._children = children + + async def _initialize_modules(self) -> None: + """Initialize modules based on component negotiation response.""" + for mod in SmartCameraModule.REGISTERED_MODULES.values(): + module = mod(self, mod._module_name()) + if await module._check_supported(): + self._modules[module.name] = module + + async def _initialize_features(self) -> None: + """Initialize device features.""" + for module in self.modules.values(): + module._initialize_features() + for feat in module._module_features.values(): + self._add_feature(feat) + + for child in self._children.values(): + await child._initialize_features() + + async def _query_setter_helper( + self, method: str, module: str, section: str, params: dict | None = None + ) -> dict: + res = await self.protocol.query({method: {module: {section: params}}}) + + return res + + async def _query_getter_helper( + self, method: str, module: str, sections: str | list[str] + ) -> Any: + res = await self.protocol.query({method: {module: {"name": sections}}}) + + return res + + async def _negotiate(self) -> None: + """Perform initialization. + + We fetch the device info and the available components as early as possible. + If the device reports supporting child devices, they are also initialized. + """ initial_query = { "getDeviceInfo": {"device_info": {"name": ["basic_info", "info"]}}, - "getLensMaskConfig": {"lens_mask": {"name": ["lens_mask_info"]}}, + "getChildDeviceList": {"childControl": {"start_index": 0}}, } resp = await self.protocol.query(initial_query) self._last_update.update(resp) - info = self._try_get_response(resp, "getDeviceInfo") - self._info = self._map_info(info["device_info"]) - self._last_update = resp + self._update_internal_info(resp) + await self._initialize_children() def _map_info(self, device_info: dict) -> dict: basic_info = device_info["basic_info"] @@ -48,25 +149,17 @@ class SmartCamera(SmartDevice): @property def is_on(self) -> bool: """Return true if the device is on.""" - if isinstance(self._last_update["getLensMaskConfig"], SmartErrorCode): - return True - return ( - self._last_update["getLensMaskConfig"]["lens_mask"]["lens_mask_info"][ - "enabled" - ] - == "on" - ) + if (camera := self.modules.get(Module.Camera)) and not camera.disabled: + return camera.is_on - async def set_state(self, on: bool): + return True + + async def set_state(self, on: bool) -> dict: """Set the device state.""" - if isinstance(self._last_update["getLensMaskConfig"], SmartErrorCode): - return - query = { - "setLensMaskConfig": { - "lens_mask": {"lens_mask_info": {"enabled": "on" if on else "off"}} - }, - } - return await self.protocol.query(query) + if (camera := self.modules.get(Module.Camera)) and not camera.disabled: + return await camera.set_state(on) + + return {} @property def device_type(self) -> DeviceType: @@ -82,6 +175,14 @@ class SmartCamera(SmartDevice): return self._info.get("alias") return None + async def set_alias(self, alias: str) -> dict: + """Set the device name (alias).""" + return await self.protocol.query( + { + "setDeviceAlias": {"system": {"sys": {"dev_alias": alias}}}, + } + ) + @property def hw_info(self) -> dict: """Return hardware info for the device.""" diff --git a/kasa/experimental/smartcameramodule.py b/kasa/experimental/smartcameramodule.py new file mode 100644 index 00000000..fed97cb3 --- /dev/null +++ b/kasa/experimental/smartcameramodule.py @@ -0,0 +1,96 @@ +"""Base implementation for SMART modules.""" + +from __future__ import annotations + +import logging +from typing import TYPE_CHECKING + +from ..exceptions import DeviceError, KasaException, SmartErrorCode +from ..smart.smartmodule import SmartModule + +if TYPE_CHECKING: + from .smartcamera import SmartCamera + +_LOGGER = logging.getLogger(__name__) + + +class SmartCameraModule(SmartModule): + """Base class for SMARTCAMERA modules.""" + + #: Query to execute during the main update cycle + QUERY_GETTER_NAME: str + #: Module name to be queried + QUERY_MODULE_NAME: str + #: Section name or names to be queried + QUERY_SECTION_NAMES: str | list[str] + + REGISTERED_MODULES = {} + + _device: SmartCamera + + def query(self) -> dict: + """Query to execute during the update cycle. + + Default implementation uses the raw query getter w/o parameters. + """ + return { + self.QUERY_GETTER_NAME: { + self.QUERY_MODULE_NAME: {"name": self.QUERY_SECTION_NAMES} + } + } + + async def call(self, method: str, params: dict | None = None) -> dict: + """Call a method. + + Just a helper method. + """ + if params: + module = next(iter(params)) + section = next(iter(params[module])) + else: + module = "system" + section = "null" + + if method[:3] == "get": + return await self._device._query_getter_helper(method, module, section) + + return await self._device._query_setter_helper(method, module, section, params) + + @property + def data(self) -> dict: + """Return response data for the module.""" + dev = self._device + q = self.query() + + if not q: + return dev.sys_info + + if len(q) == 1: + query_resp = dev._last_update.get(self.QUERY_GETTER_NAME, {}) + if isinstance(query_resp, SmartErrorCode): + raise DeviceError( + f"Error accessing module data in {self._module}", + error_code=SmartErrorCode, + ) + + if not query_resp: + raise KasaException( + f"You need to call update() prior accessing module data" + f" for '{self._module}'" + ) + + return query_resp.get(self.QUERY_MODULE_NAME) + else: + found = {key: val for key, val in dev._last_update.items() if key in q} + for key in q: + if key not in found: + raise KasaException( + f"{key} not found, you need to call update() prior accessing" + f" module data for '{self._module}'" + ) + if isinstance(found[key], SmartErrorCode): + raise DeviceError( + f"Error accessing module data {key} in {self._module}", + error_code=SmartErrorCode, + ) + return found diff --git a/kasa/module.py b/kasa/module.py index 2c6014e5..e10b2d63 100644 --- a/kasa/module.py +++ b/kasa/module.py @@ -55,6 +55,7 @@ from .modulemapping import ModuleName if TYPE_CHECKING: from . import interfaces from .device import Device + from .experimental import modules as experimental from .iot import modules as iot from .smart import modules as smart @@ -127,6 +128,9 @@ class Module(ABC): "WaterleakSensor" ) + # SMARTCAMERA only modules + Camera: Final[ModuleName[experimental.Camera]] = ModuleName("Camera") + def __init__(self, device: Device, module: str): self._device = device self._module = module diff --git a/kasa/smart/smartchilddevice.py b/kasa/smart/smartchilddevice.py index 1fe0014e..f3e39ce9 100644 --- a/kasa/smart/smartchilddevice.py +++ b/kasa/smart/smartchilddevice.py @@ -36,17 +36,19 @@ class SmartChildDevice(SmartDevice): def __init__( self, parent: SmartDevice, - info, - component_info, + info: dict, + component_info: dict, + *, config: DeviceConfig | None = None, protocol: SmartProtocol | None = None, ) -> None: - super().__init__(parent.host, config=parent.config, protocol=parent.protocol) + super().__init__(parent.host, config=parent.config, protocol=protocol) self._parent = parent self._update_internal_state(info) self._components = component_info self._id = info["device_id"] - self.protocol = _ChildProtocolWrapper(self._id, parent.protocol) + # wrap device protocol if no protocol is given + self.protocol = protocol or _ChildProtocolWrapper(self._id, parent.protocol) async def update(self, update_children: bool = True): """Update child module info. @@ -79,9 +81,27 @@ class SmartChildDevice(SmartDevice): self._last_update_time = now @classmethod - async def create(cls, parent: SmartDevice, child_info, child_components): - """Create a child device based on device info and component listing.""" - child: SmartChildDevice = cls(parent, child_info, child_components) + async def create( + cls, + parent: SmartDevice, + child_info: dict, + child_components: dict, + protocol: SmartProtocol | None = None, + *, + last_update: dict | None = None, + ) -> SmartDevice: + """Create a child device based on device info and component listing. + + If creating a smart child from a different protocol, i.e. a camera hub, + protocol: SmartProtocol and last_update should be provided as per the + FIRST_UPDATE_MODULES expected by the update cycle as these cannot be + derived from the parent. + """ + child: SmartChildDevice = cls( + parent, child_info, child_components, protocol=protocol + ) + if last_update: + child._last_update = last_update await child._initialize_modules() return child diff --git a/kasa/smart/smartdevice.py b/kasa/smart/smartdevice.py index 0a8c136c..f4012b68 100644 --- a/kasa/smart/smartdevice.py +++ b/kasa/smart/smartdevice.py @@ -37,15 +37,15 @@ _LOGGER = logging.getLogger(__name__) # same issue, homekit perhaps? NON_HUB_PARENT_ONLY_MODULES = [DeviceModule, Time, Firmware, Cloud] -# Modules that are called as part of the init procedure on first update -FIRST_UPDATE_MODULES = {DeviceModule, ChildDevice, Cloud} - # Device must go last as the other interfaces also inherit Device # and python needs a consistent method resolution order. class SmartDevice(Device): """Base class to represent a SMART protocol based device.""" + # Modules that are called as part of the init procedure on first update + FIRST_UPDATE_MODULES = {DeviceModule, ChildDevice, Cloud} + def __init__( self, host: str, @@ -67,6 +67,7 @@ class SmartDevice(Device): self._last_update = {} self._last_update_time: float | None = None self._on_since: datetime | None = None + self._info: dict[str, Any] = {} async def _initialize_children(self): """Initialize children for power strips.""" @@ -154,6 +155,18 @@ class SmartDevice(Device): if "child_device" in self._components and not self.children: await self._initialize_children() + def _update_children_info(self) -> None: + """Update the internal child device info from the parent info.""" + if child_info := self._try_get_response( + self._last_update, "get_child_device_list", {} + ): + for info in child_info["child_device_list"]: + self._children[info["device_id"]]._update_internal_state(info) + + def _update_internal_info(self, info_resp: dict) -> None: + """Update the internal device info.""" + self._info = self._try_get_response(info_resp, "get_device_info") + async def update(self, update_children: bool = False): """Update the device.""" if self.credentials is None and self.credentials_hash is None: @@ -172,11 +185,7 @@ class SmartDevice(Device): resp = await self._modular_update(first_update, now) - if child_info := self._try_get_response( - self._last_update, "get_child_device_list", {} - ): - for info in child_info["child_device_list"]: - self._children[info["device_id"]]._update_internal_state(info) + self._update_children_info() # Call child update which will only update module calls, info is updated # from get_child_device_list. update_children only affects hub devices, other # devices will always update children to prevent errors on module access. @@ -227,10 +236,10 @@ class SmartDevice(Device): mq = { module: query for module in self._modules.values() - if module.disabled is False and (query := module.query()) + if (first_update or module.disabled is False) and (query := module.query()) } for module, query in mq.items(): - if first_update and module.__class__ in FIRST_UPDATE_MODULES: + if first_update and module.__class__ in self.FIRST_UPDATE_MODULES: module._last_update_time = update_time continue if ( @@ -256,7 +265,7 @@ class SmartDevice(Device): info_resp = self._last_update if first_update else resp self._last_update.update(**resp) - self._info = self._try_get_response(info_resp, "get_device_info") + self._update_internal_info(info_resp) # Call handle update for modules that want to update internal data for module in self._modules.values(): @@ -570,7 +579,7 @@ class SmartDevice(Device): """Return all the internal state data.""" return self._last_update - def _update_internal_state(self, info): + def _update_internal_state(self, info: dict) -> None: """Update the internal info state. This is used by the parent to push updates to its children. diff --git a/kasa/smart/smartmodule.py b/kasa/smart/smartmodule.py index 8fea1d9f..f20186ec 100644 --- a/kasa/smart/smartmodule.py +++ b/kasa/smart/smartmodule.py @@ -80,7 +80,7 @@ class SmartModule(Module): # other classes can inherit from smartmodule and not be registered if cls.__module__.split(".")[-2] == "modules": _LOGGER.debug("Registering %s", cls) - cls.REGISTERED_MODULES[cls.__name__] = cls + cls.REGISTERED_MODULES[cls._module_name()] = cls def _set_error(self, err: Exception | None): if err is None: @@ -118,10 +118,14 @@ class SmartModule(Module): """Return true if the module is disabled due to errors.""" return self._error_count >= self.DISABLE_AFTER_ERROR_COUNT + @classmethod + def _module_name(cls): + return getattr(cls, "NAME", cls.__name__) + @property def name(self) -> str: """Name of the module.""" - return getattr(self, "NAME", self.__class__.__name__) + return self._module_name() async def _post_update_hook(self): # noqa: B027 """Perform actions after a device update. diff --git a/kasa/tests/smartcamera/test_smartcamera.py b/kasa/tests/smartcamera/test_smartcamera.py index 9c8893c0..50a1a136 100644 --- a/kasa/tests/smartcamera/test_smartcamera.py +++ b/kasa/tests/smartcamera/test_smartcamera.py @@ -6,7 +6,7 @@ import pytest from kasa import Device, DeviceType -from ..conftest import device_smartcamera +from ..conftest import device_smartcamera, hub_smartcamera @device_smartcamera @@ -18,3 +18,30 @@ async def test_state(dev: Device): await dev.set_state(not state) await dev.update() assert dev.is_on is not state + + +@device_smartcamera +async def test_alias(dev): + test_alias = "TEST1234" + original = dev.alias + + assert isinstance(original, str) + await dev.set_alias(test_alias) + await dev.update() + assert dev.alias == test_alias + + await dev.set_alias(original) + await dev.update() + assert dev.alias == original + + +@hub_smartcamera +async def test_hub(dev): + assert dev.children + for child in dev.children: + assert "Cloud" in child.modules + assert child.modules["Cloud"].data + assert child.alias + await child.update() + assert "Time" not in child.modules + assert child.time