[WIP] Kasa Cam support

kasa/__init__.py

@@ -23,6 +23,7 @@
 )
 from kasa.protocol import TPLinkSmartHomeProtocol
 from kasa.smartbulb import SmartBulb, SmartBulbPreset, TurnOnBehavior, TurnOnBehaviors
+from kasa.smartcamera import SmartCamera
 from kasa.smartdevice import DeviceType, SmartDevice
 from kasa.smartdimmer import SmartDimmer
 from kasa.smartlightstrip import SmartLightStrip
@@ -34,6 +35,7 @@
 
 __all__ = [
     "Discover",
+    "SmartCamera",
     "TPLinkSmartHomeProtocol",
     "SmartBulb",
     "SmartBulbPreset",

kasa/cli.py

@@ -14,6 +14,7 @@
     Credentials,
     Discover,
     SmartBulb,
+    SmartCamera,
     SmartDevice,
     SmartDimmer,
     SmartLightStrip,
@@ -47,6 +48,7 @@ def wrapper(message=None, *args, **kwargs):
     "bulb": SmartBulb,
     "dimmer": SmartDimmer,
     "strip": SmartStrip,
+    "kasacam": SmartCamera,
     "lightstrip": SmartLightStrip,
 }
 

kasa/discover.py

@@ -15,6 +15,7 @@
 from kasa.json import loads as json_loads
 from kasa.protocol import TPLinkSmartHomeProtocol
 from kasa.smartbulb import SmartBulb
+from kasa.smartcamera import SmartCamera
 from kasa.smartdevice import SmartDevice, SmartDeviceException
 from kasa.smartdimmer import SmartDimmer
 from kasa.smartlightstrip import SmartLightStrip
@@ -354,9 +355,13 @@ def _get_device_class(info: dict) -> Type[SmartDevice]:
             raise SmartDeviceException("No 'system' or 'get_sysinfo' in response")
 
         sysinfo = info["system"]["get_sysinfo"]
+        if "system" in sysinfo:
+            sysinfo = sysinfo["system"]
         type_ = sysinfo.get("type", sysinfo.get("mic_type"))
         if type_ is None:
-            raise SmartDeviceException("Unable to find the device type field!")
+            raise SmartDeviceException(
+                f"Unable to find the device type field in {sysinfo}"
+            )
 
         if "dev_name" in sysinfo and "Dimmer" in sysinfo["dev_name"]:
             return SmartDimmer
@@ -373,4 +378,7 @@ def _get_device_class(info: dict) -> Type[SmartDevice]:
 
             return SmartBulb
 
+        if "ipcamera" in type_.lower():
+            return SmartCamera
+
         raise SmartDeviceException("Unknown device type: %s" % type_)

kasa/modules/__init__.py

@@ -6,6 +6,7 @@
 from .emeter import Emeter
 from .module import Module
 from .motion import Motion
+from .ptz import PTZ
 from .rulemodule import Rule, RuleModule
 from .schedule import Schedule
 from .time import Time
@@ -19,6 +20,7 @@
     "Emeter",
     "Module",
     "Motion",
+    "PTZ",
     "Rule",
     "RuleModule",
     "Schedule",

kasa/modules/ptz.py

@@ -0,0 +1,124 @@
+"""Pan/Tilt/Zoom Module."""
+from .module import Module, merge
+
+try:
+    from pydantic.v1 import BaseModel
+except ImportError:
+    from pydantic import BaseModel
+
+from typing import Literal, Optional
+
+from ..exceptions import SmartDeviceException
+
+
+class Position(BaseModel):
+    """Camera Position Schema."""
+
+    x: int
+    y: int
+
+
+Direction = Literal["up", "down", "left", "right"]
+
+
+class PTZ(Module):
+    """Module implementing support for pan/tilt/zoom cameras."""
+
+    def query(self):
+        """Request PTZ info."""
+        q = self.query_for_command("get_position")
+        merge(q, self.query_for_command("get_patrol_is_enable"))
+        return q
+
+    @property
+    def position(self) -> Position:
+        """Return the camera's position coordinates."""
+        return Position.parse_obj(self.data["get_position"])
+
+    @property
+    def is_patrol_enabled(self) -> bool:
+        """Whether or not the camera is patrolling."""
+        return self.data["get_patrol_is_enable"]["value"] == "on"
+
+    async def set_enable_patrol(self, enabled: bool):
+        """Enable or disable camera patrolling."""
+        return await self.call(
+            "set_patrol_is_enable", {"value": "on" if enabled else "off"}
+        )
+
+    async def go_to(
+        self,
+        position: Optional[Position] = None,
+        x: Optional[int] = None,
+        y: Optional[int] = None,
+    ):
+        """Move the camera to the given Position's x,y coordinates."""
+        if position is None and x is not None and y is not None:
+            position = Position(x=x, y=y)
+        if not position:
+            raise SmartDeviceException(
+                "Either a Position object or both x and y are required."
+            )
+        return await self.call("set_move", {"x": position.x, "y": position.y})
+
+    async def stop(self):
+        """Stop the camera where it is."""
+        return await self.call("set_stop")
+
+    async def move(self, direction: Direction, speed: int):
+        """Move the camera in a relative direction."""
+        return await self.call("set_target", {"direction": direction, "speed": speed})
+
+    # async def add_preset(self):
+    #     # public String api_srv_url;
+    #     # public Integer index;
+    #     # public String name;
+    #     # public String path;
+    #     # public Integer wait_time;
+    #     pass
+
+    # async def delete_preset(self, index: int):
+    #     pass
+
+    # async def edit_preset(self, index: int, name: str, wait_time: int):
+    #     pass
+
+    # async def get_all_preset(self):
+    #     # public Integer maximum;
+    #     # public List preset_attr;
+    #     pass
+
+    # async def get_patrol_is_enable(self):
+    #     pass
+
+    # async def get_position(self):
+    #     # public Integer x;
+    #     # public Integer y;
+    #     pass
+
+    # async def get_ptz_rectify_state(self):
+    #     pass
+
+    # async def get_ptz_tracking_is_enable(self):
+    #     pass
+
+    # async def set_motor_rectify(self):
+    #     pass
+
+    # async def set_move(self, x: int, y: int):
+    #     pass
+
+    # async def set_patrol_is_enable(self):
+    #     pass
+
+    # async def set_ptz_tracking_is_enable(self):
+    #     pass
+
+    # async def set_run_to_preset(self, index: int):
+    #     pass
+
+    # async def set_stop(self):
+    #     pass
+
+    # async def set_target(self, direction: str, speed: int):
+    #     pass

kasa/modules/time.py

@@ -13,12 +13,16 @@ def query(self):
         q = self.query_for_command("get_time")
 
         merge(q, self.query_for_command("get_timezone"))
+        merge(q, self.query_for_command("get_time_zone"))
         return q
 
     @property
     def time(self) -> datetime:
         """Return current device time."""
         res = self.data["get_time"]
+        if "epoch_sec" in res:
+            return datetime.fromtimestamp(res["epoch_sec"])
+
         return datetime(
             res["year"],
             res["month"],
@@ -38,6 +42,8 @@ async def get_time(self):
         """Return current device time."""
         try:
             res = await self.call("get_time")
+            if "epoch_sec" in res:
+                return datetime.fromtimestamp(res["epoch_sec"])
             return datetime(
                 res["year"],
                 res["month"],

kasa/protocol.py

@@ -14,6 +14,7 @@
 import errno
 import logging
 import struct
+from abc import ABC, abstractmethod
 from pprint import pformat as pf
 from typing import Dict, Generator, Optional, Union
 
@@ -29,7 +30,24 @@
 _NO_RETRY_ERRORS = {errno.EHOSTDOWN, errno.EHOSTUNREACH, errno.ECONNREFUSED}
 
 
-class TPLinkSmartHomeProtocol:
+class TPLinkProtocol(ABC):
+    """Base class for all TP-Link Smart Home communication."""
+
+    def __init__(self, host: str, *, port: Optional[int] = None) -> None:
+        """Create a protocol object."""
+        self.host = host
+        self.port = port or TPLinkSmartHomeProtocol.DEFAULT_PORT
+
+    @abstractmethod
+    async def query(self, request: Union[str, Dict], retry_count: int = 3) -> Dict:
+        """Query the device associated with the protocol."""
+
+    @abstractmethod
+    async def close(self) -> None:
+        """Close the protocol.  Abstract method to be overriden."""
+
+
+class TPLinkSmartHomeProtocol(TPLinkProtocol):
     """Implementation of the TP-Link Smart Home protocol."""
 
     INITIALIZATION_VECTOR = 171
@@ -42,7 +60,9 @@ def __init__(
     ) -> None:
         """Create a protocol object."""
         self.host = host
-        self.port = port or TPLinkSmartHomeProtocol.DEFAULT_PORT
+        self.port = port or self.DEFAULT_PORT
+        super().__init__(host=host, port=port or self.DEFAULT_PORT)
+
         self.reader: Optional[asyncio.StreamReader] = None
         self.writer: Optional[asyncio.StreamWriter] = None
         self.query_lock = asyncio.Lock()
@@ -133,17 +153,15 @@ async def _query(self, request: str, retry_count: int, timeout: int) -> Dict:
                 await self.close()
                 if ex.errno in _NO_RETRY_ERRORS or retry >= retry_count:
                     raise SmartDeviceException(
-                        f"Unable to connect to the device:"
-                        f" {self.host}:{self.port}: {ex}"
+                        f"Unable to connect to the device:" f" {self.host}:{self.port}"
                     ) from ex
                 continue
             except Exception as ex:
                 await self.close()
                 if retry >= retry_count:
                     _LOGGER.debug("Giving up on %s after %s retries", self.host, retry)
                     raise SmartDeviceException(
-                        f"Unable to connect to the device:"
-                        f" {self.host}:{self.port}: {ex}"
+                        f"Unable to connect to the device:" f" {self.host}:{self.port}"
                     ) from ex
                 continue