Add childsetup module to smartcam hubs (#1469)

Add the `childsetup` module for `smartcam` hubs to allow pairing and unpairing child devices.

kasa/smartcam/modules/__init__.py

@@ -5,6 +5,7 @@ from .babycrydetection import BabyCryDetection
 from .battery import Battery
 from .camera import Camera
 from .childdevice import ChildDevice
+from .childsetup import ChildSetup
 from .device import DeviceModule
 from .homekit import HomeKit
 from .led import Led
@@ -23,6 +24,7 @@ __all__ = [
     "Battery",
     "Camera",
     "ChildDevice",
+    "ChildSetup",
     "DeviceModule",
     "Led",
     "PanTilt",

kasa/smartcam/modules/childsetup.py

@@ -0,0 +1,107 @@
+"""Implementation for child device setup.
+
+This module allows pairing and disconnecting child devices.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import logging
+
+from ...feature import Feature
+from ..smartcammodule import SmartCamModule
+
+_LOGGER = logging.getLogger(__name__)
+
+
+class ChildSetup(SmartCamModule):
+    """Implementation for child device setup."""
+
+    REQUIRED_COMPONENT = "childQuickSetup"
+    QUERY_GETTER_NAME = "getSupportChildDeviceCategory"
+    QUERY_MODULE_NAME = "childControl"
+    _categories: list[str] = []
+
+    def _initialize_features(self) -> None:
+        """Initialize features."""
+        self._add_feature(
+            Feature(
+                self._device,
+                id="pair",
+                name="Pair",
+                container=self,
+                attribute_setter="pair",
+                category=Feature.Category.Config,
+                type=Feature.Type.Action,
+            )
+        )
+
+    async def _post_update_hook(self) -> None:
+        if not self._categories:
+            self._categories = [
+                cat["category"].replace("ipcamera", "camera")
+                for cat in self.data["device_category_list"]
+            ]
+
+    @property
+    def supported_child_device_categories(self) -> list[str]:
+        """Supported child device categories."""
+        return self._categories
+
+    async def pair(self, *, timeout: int = 10) -> list[dict]:
+        """Scan for new devices and pair after discovering first new device."""
+        await self.call(
+            "startScanChildDevice", {"childControl": {"category": self._categories}}
+        )
+
+        _LOGGER.info("Waiting %s seconds for discovering new devices", timeout)
+
+        await asyncio.sleep(timeout)
+        res = await self.call(
+            "getScanChildDeviceList", {"childControl": {"category": self._categories}}
+        )
+
+        detected_list = res["getScanChildDeviceList"]["child_device_list"]
+        if not detected_list:
+            _LOGGER.warning(
+                "No devices found, make sure to activate pairing "
+                "mode on the devices to be added."
+            )
+            return []
+
+        _LOGGER.info(
+            "Discovery done, found %s devices: %s",
+            len(detected_list),
+            detected_list,
+        )
+        return await self._add_devices(detected_list)
+
+    async def _add_devices(self, detected_list: list[dict]) -> list:
+        """Add devices based on getScanChildDeviceList response."""
+        await self.call(
+            "addScanChildDeviceList",
+            {"childControl": {"child_device_list": detected_list}},
+        )
+
+        await self._device.update()
+
+        successes = []
+        for detected in detected_list:
+            device_id = detected["device_id"]
+
+            result = "not added"
+            if device_id in self._device._children:
+                result = "added"
+                successes.append(detected)
+
+            msg = f"{detected['device_model']} - {device_id} - {result}"
+            _LOGGER.info("Adding child to %s: %s", self._device.host, msg)
+
+        return successes
+
+    async def unpair(self, device_id: str) -> dict:
+        """Remove device from the hub."""
+        _LOGGER.info("Going to unpair %s from %s", device_id, self)
+
+        payload = {"childControl": {"child_device_list": [{"device_id": device_id}]}}
+        return await self.call("removeChildDeviceList", payload)