Add LinkieTransportV2 and basic IOT.IPCAMERA support (#1270)

Add LinkieTransportV2 transport used by kasa cameras and a basic
implementation for IOT.IPCAMERA (kasacam) devices.

---------

Co-authored-by: Zach Price <pricezt@ornl.gov>
Co-authored-by: Steven B <51370195+sdb9696@users.noreply.github.com>
Co-authored-by: Teemu Rytilahti <tpr@iki.fi>
This commit is contained in:
Puxtril
2024-12-06 18:06:58 -05:00
committed by GitHub
parent 6d9b4421fe
commit cb89342be1
13 changed files with 461 additions and 13 deletions

View File

@@ -1,6 +1,7 @@
"""Package for supporting legacy kasa devices."""
from .iotbulb import IotBulb
from .iotcamera import IotCamera
from .iotdevice import IotDevice
from .iotdimmer import IotDimmer
from .iotlightstrip import IotLightStrip
@@ -15,4 +16,5 @@ __all__ = [
"IotDimmer",
"IotLightStrip",
"IotWallSwitch",
"IotCamera",
]

42
kasa/iot/iotcamera.py Normal file
View File

@@ -0,0 +1,42 @@
"""Module for cameras."""
from __future__ import annotations
import logging
from datetime import datetime, tzinfo
from ..device_type import DeviceType
from ..deviceconfig import DeviceConfig
from ..protocols import BaseProtocol
from .iotdevice import IotDevice
_LOGGER = logging.getLogger(__name__)
class IotCamera(IotDevice):
"""Representation of a TP-Link Camera."""
def __init__(
self,
host: str,
*,
config: DeviceConfig | None = None,
protocol: BaseProtocol | None = None,
) -> None:
super().__init__(host=host, config=config, protocol=protocol)
self._device_type = DeviceType.Camera
@property
def time(self) -> datetime:
"""Get the camera's time."""
return datetime.fromtimestamp(self.sys_info["system_time"])
@property
def timezone(self) -> tzinfo:
"""Get the camera's timezone."""
return None # type: ignore
@property # type: ignore
def is_on(self) -> bool:
"""Return whether device is on."""
return True

View File

@@ -70,6 +70,16 @@ def _parse_features(features: str) -> set[str]:
return set(features.split(":"))
def _extract_sys_info(info: dict[str, Any]) -> dict[str, Any]:
"""Return the system info structure."""
sysinfo_default = info.get("system", {}).get("get_sysinfo", {})
sysinfo_nest = sysinfo_default.get("system", {})
if len(sysinfo_nest) > len(sysinfo_default) and isinstance(sysinfo_nest, dict):
return sysinfo_nest
return sysinfo_default
class IotDevice(Device):
"""Base class for all supported device types.
@@ -304,14 +314,14 @@ class IotDevice(Device):
_LOGGER.debug("Performing the initial update to obtain sysinfo")
response = await self.protocol.query(req)
self._last_update = response
self._set_sys_info(response["system"]["get_sysinfo"])
self._set_sys_info(_extract_sys_info(response))
if not self._modules:
await self._initialize_modules()
await self._modular_update(req)
self._set_sys_info(self._last_update["system"]["get_sysinfo"])
self._set_sys_info(_extract_sys_info(self._last_update))
for module in self._modules.values():
await module._post_update_hook()
@@ -705,10 +715,13 @@ class IotDevice(Device):
@staticmethod
def _get_device_type_from_sys_info(info: dict[str, Any]) -> DeviceType:
"""Find SmartDevice subclass for device described by passed data."""
if "system" in info.get("system", {}).get("get_sysinfo", {}):
return DeviceType.Camera
if "system" not in info or "get_sysinfo" not in info["system"]:
raise KasaException("No 'system' or 'get_sysinfo' in response")
sysinfo: dict[str, Any] = info["system"]["get_sysinfo"]
sysinfo: dict[str, Any] = _extract_sys_info(info)
type_: str | None = sysinfo.get("type", sysinfo.get("mic_type"))
if type_ is None:
raise KasaException("Unable to find the device type field!")
@@ -728,6 +741,7 @@ class IotDevice(Device):
return DeviceType.LightStrip
return DeviceType.Bulb
_LOGGER.warning("Unknown device type %s, falling back to plug", type_)
return DeviceType.Plug
@@ -736,7 +750,7 @@ class IotDevice(Device):
info: dict[str, Any], discovery_info: dict[str, Any] | None
) -> _DeviceInfo:
"""Get model information for a device."""
sys_info = info["system"]["get_sysinfo"]
sys_info = _extract_sys_info(info)
# Get model and region info
region = None