Add common childsetup interface (#1470)

Add a common interface for the `childsetup` module across `smart` and `smartcam` hubs.
Co-authored-by: Teemu R. <tpr@iki.fi>
This commit is contained in:
Steven B.
2025-01-24 08:08:04 +00:00
committed by GitHub
parent b701441215
commit 09fce3f426
15 changed files with 185 additions and 61 deletions

View File

@@ -9,16 +9,21 @@ import asyncio
import logging
from ...feature import Feature
from ...interfaces.childsetup import ChildSetup as ChildSetupInterface
from ..smartmodule import SmartModule
_LOGGER = logging.getLogger(__name__)
class ChildSetup(SmartModule):
class ChildSetup(SmartModule, ChildSetupInterface):
"""Implementation for child device setup."""
REQUIRED_COMPONENT = "child_quick_setup"
QUERY_GETTER_NAME = "get_support_child_device_category"
_categories: list[str] = []
# Supported child device categories will hardly ever change
MINIMUM_UPDATE_INTERVAL_SECS = 60 * 60 * 24
def _initialize_features(self) -> None:
"""Initialize features."""
@@ -34,13 +39,18 @@ class ChildSetup(SmartModule):
)
)
async def get_supported_device_categories(self) -> list[dict]:
"""Get supported device categories."""
categories = await self.call("get_support_child_device_category")
return categories["get_support_child_device_category"]["device_category_list"]
async def _post_update_hook(self) -> None:
self._categories = [
cat["category"] for cat in self.data["device_category_list"]
]
@property
def supported_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."""
"""Scan for new devices and pair them."""
await self.call("begin_scanning_child_device")
_LOGGER.info("Waiting %s seconds for discovering new devices", timeout)
@@ -60,28 +70,43 @@ class ChildSetup(SmartModule):
detected,
)
await self._add_devices(detected)
return detected["child_device_list"]
return await self._add_devices(detected)
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 = {"child_device_list": [{"device_id": device_id}]}
return await self.call("remove_child_device_list", payload)
res = await self.call("remove_child_device_list", payload)
await self._device.update()
return res
async def _add_devices(self, devices: dict) -> dict:
async def _add_devices(self, devices: dict) -> list[dict]:
"""Add devices based on get_detected_device response.
Pass the output from :ref:_get_detected_devices: as a parameter.
"""
res = await self.call("add_child_device_list", devices)
return res
await self.call("add_child_device_list", devices)
await self._device.update()
successes = []
for detected in devices["child_device_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("Added child to %s: %s", self._device.host, msg)
return successes
async def _get_detected_devices(self) -> dict:
"""Return list of devices detected during scanning."""
param = {"scan_list": await self.get_supported_device_categories()}
param = {"scan_list": self.data["device_category_list"]}
res = await self.call("get_scan_child_device_list", param)
_LOGGER.debug("Scan status: %s", res)
return res["get_scan_child_device_list"]