Move TAPO smartcamera out of experimental package (#1255)

Co-authored-by: Teemu R. <tpr@iki.fi>

kasa/smartcamera/modules/__init__.py

@@ -0,0 +1,15 @@
+"""Modules for SMARTCAMERA devices."""
+
+from .camera import Camera
+from .childdevice import ChildDevice
+from .device import DeviceModule
+from .led import Led
+from .time import Time
+
+__all__ = [
+    "Camera",
+    "ChildDevice",
+    "DeviceModule",
+    "Led",
+    "Time",
+]

kasa/smartcamera/modules/camera.py

@@ -0,0 +1,71 @@
+"""Implementation of device module."""
+
+from __future__ import annotations
+
+from urllib.parse import quote_plus
+
+from ...credentials import Credentials
+from ...device_type import DeviceType
+from ...feature import Feature
+from ..smartcameramodule import SmartCameraModule
+
+LOCAL_STREAMING_PORT = 554
+
+
+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"] == "off"
+
+    def stream_rtsp_url(self, credentials: Credentials | None = None) -> str | None:
+        """Return the local rtsp streaming url.
+
+        :param credentials: Credentials for camera account.
+            These could be different credentials to tplink cloud credentials.
+            If not provided will use tplink credentials if available
+        :return: rtsp url with escaped credentials or None if no credentials or
+            camera is off.
+        """
+        if not self.is_on:
+            return None
+        dev = self._device
+        if not credentials:
+            credentials = dev.credentials
+        if not credentials or not credentials.username or not credentials.password:
+            return None
+        username = quote_plus(credentials.username)
+        password = quote_plus(credentials.password)
+        return f"rtsp://{username}:{password}@{dev.host}:{LOCAL_STREAMING_PORT}/stream1"
+
+    async def set_state(self, on: bool) -> dict:
+        """Set the device state."""
+        # Turning off enables the privacy mask which is why value is reversed.
+        params = {"enabled": "off" if on else "on"}
+        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

kasa/smartcamera/modules/childdevice.py

@@ -0,0 +1,26 @@
+"""Module for child devices."""
+
+from ...device_type import DeviceType
+from ..smartcameramodule import SmartCameraModule
+
+
+class ChildDevice(SmartCameraModule):
+    """Implementation for child devices."""
+
+    REQUIRED_COMPONENT = "childControl"
+    NAME = "childdevice"
+    QUERY_GETTER_NAME = "getChildDeviceList"
+    # This module is unusual in that QUERY_MODULE_NAME in the response is not
+    # the same one used in the request.
+    QUERY_MODULE_NAME = "child_device_list"
+
+    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: {"childControl": {"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

kasa/smartcamera/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"]

kasa/smartcamera/modules/led.py

@@ -0,0 +1,28 @@
+"""Module for led controls."""
+
+from __future__ import annotations
+
+from ...interfaces.led import Led as LedInterface
+from ..smartcameramodule import SmartCameraModule
+
+
+class Led(SmartCameraModule, LedInterface):
+    """Implementation of led controls."""
+
+    REQUIRED_COMPONENT = "led"
+    QUERY_GETTER_NAME = "getLedStatus"
+    QUERY_MODULE_NAME = "led"
+    QUERY_SECTION_NAMES = "config"
+
+    @property
+    def led(self) -> bool:
+        """Return current led status."""
+        return self.data["config"]["enabled"] == "on"
+
+    async def set_led(self, enable: bool) -> dict:
+        """Set led.
+
+        This should probably be a select with always/never/nightmode.
+        """
+        params = {"enabled": "on"} if enable else {"enabled": "off"}
+        return await self.call("setLedStatus", {"led": {"config": params}})

kasa/smartcamera/modules/time.py

@@ -0,0 +1,91 @@
+"""Implementation of time module."""
+
+from __future__ import annotations
+
+from datetime import datetime, timezone, tzinfo
+from typing import cast
+
+from zoneinfo import ZoneInfo, ZoneInfoNotFoundError
+
+from ...cachedzoneinfo import CachedZoneInfo
+from ...feature import Feature
+from ...interfaces import Time as TimeInterface
+from ..smartcameramodule import SmartCameraModule
+
+
+class Time(SmartCameraModule, TimeInterface):
+    """Implementation of device_local_time."""
+
+    QUERY_GETTER_NAME = "getTimezone"
+    QUERY_MODULE_NAME = "system"
+    QUERY_SECTION_NAMES = "basic"
+
+    _timezone: tzinfo = timezone.utc
+    _time: datetime
+
+    def _initialize_features(self) -> None:
+        """Initialize features after the initial update."""
+        self._add_feature(
+            Feature(
+                device=self._device,
+                id="device_time",
+                name="Device time",
+                attribute_getter="time",
+                container=self,
+                category=Feature.Category.Debug,
+                type=Feature.Type.Sensor,
+            )
+        )
+
+    def query(self) -> dict:
+        """Query to execute during the update cycle."""
+        q = super().query()
+        q["getClockStatus"] = {self.QUERY_MODULE_NAME: {"name": "clock_status"}}
+
+        return q
+
+    async def _post_update_hook(self) -> None:
+        """Perform actions after a device update."""
+        time_data = self.data["getClockStatus"]["system"]["clock_status"]
+        timezone_data = self.data["getTimezone"]["system"]["basic"]
+        zone_id = timezone_data["zone_id"]
+        timestamp = time_data["seconds_from_1970"]
+        try:
+            # Zoneinfo will return a DST aware object
+            tz: tzinfo = await CachedZoneInfo.get_cached_zone_info(zone_id)
+        except ZoneInfoNotFoundError:
+            # timezone string like: UTC+10:00
+            timezone_str = timezone_data["timezone"]
+            tz = cast(tzinfo, datetime.strptime(timezone_str[-6:], "%z").tzinfo)
+
+        self._timezone = tz
+        self._time = datetime.fromtimestamp(
+            cast(float, timestamp),
+            tz=tz,
+        )
+
+    @property
+    def timezone(self) -> tzinfo:
+        """Return current timezone."""
+        return self._timezone
+
+    @property
+    def time(self) -> datetime:
+        """Return device's current datetime."""
+        return self._time
+
+    async def set_time(self, dt: datetime) -> dict:
+        """Set device time."""
+        if not dt.tzinfo:
+            timestamp = dt.replace(tzinfo=self.timezone).timestamp()
+        else:
+            timestamp = dt.timestamp()
+
+        lt = datetime.fromtimestamp(timestamp).isoformat().replace("T", " ")
+        params = {"seconds_from_1970": int(timestamp), "local_time": lt}
+        # Doesn't seem to update the time, perhaps because timing_mode is ntp
+        res = await self.call("setTimezone", {"system": {"clock_status": params}})
+        if (zinfo := dt.tzinfo) and isinstance(zinfo, ZoneInfo):
+            tz_params = {"zone_id": zinfo.key}
+            res = await self.call("setTimezone", {"system": {"basic": tz_params}})
+        return res