8000 Initial TapoCamera support (#1165) · msz-coder/python-kasa@dcc36e1 · GitHub
[go: up one dir, main page]
Skip to content

Commit dcc36e1

Browse files
sdb9696sdb9696
authored
Initial TapoCamera support (python-kasa#1165)
Adds experimental support for the Tapo Camera protocol also used by the H200 hub. Creates a new SslAesTransport and a derived SmartCamera and SmartCameraProtocol.
1 parent 380fbb9 commit dcc36e1

13 files changed

+771
-12
lines changed

‎kasa/cli/main.py

Lines changed: 37 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -35,6 +35,7 @@
3535
"strip",
3636
"lightstrip",
3737
"smart",
38+
"camera",
3839
]
3940

4041
ENCRYPT_TYPES = [encrypt_type.value for encrypt_type in DeviceEncryptionType]
@@ -172,6 +173,14 @@ def _legacy_type_to_class(_type):
172173
type=int,
173174
help="The login version for device authentication. Defaults to 2",
174175
)
176+
@click.option(
177+
"--https/--no-https",
178+
envvar="KASA_HTTPS",
179+
default=False,
180+
is_flag=True,
181+
type=bool,
182+
help="Set flag if the device encryption uses https.",
183+
)
175184
@click.option(
176185
"--timeout",
177186
envvar="KASA_TIMEOUT",
@@ -209,6 +218,14 @@ def _legacy_type_to_class(_type):
209218
envvar="KASA_CREDENTIALS_HASH",
210219
help="Hashed credentials used to authenticate to the device.",
211220
)
221+
@click.option(
222+
"--experimental",
223+
default=False,
224+
is_flag=True,
225+
type=bool,
226+
envvar="KASA_EXPERIMENTAL",
227+
help="Enable experimental mode for devices not yet fully supported.",
228+
)
212229
@click.version_option(package_name="python-kasa")
213230
@click.pass_context
214231
async def cli(
@@ -221,6 +238,7 @@ async def cli(
221238
debug,
222239
type,
223240
encrypt_type,
241+
https,
224242
device_family,
225243
login_version,
226244
json,
@@ -229,6 +247,7 @@ async def cli(
229247
username,
230248
password,
231249
credentials_hash,
250+
experimental,
232251
):
233252
"""A tool for controlling TP-Link smart home devices.""" # noqa
234253
# no need to perform any checks if we are just displaying the help
@@ -237,6 +256,11 @@ async def cli(
237256
ctx.obj = object()
238257
return
239258

259+
if experimental:
260+
from kasa.experimental.enabled import Enabled
261+
262+
Enabled.set(True)
263+
240264
logging_config: dict[str, Any] = {
241265
"level": logging.DEBUG if debug > 0 else logging.INFO
242266
}
@@ -295,12 +319,21 @@ async def cli(
295319
return await ctx.invoke(discover)
296320

297321
device_updated = False
298-
if type is not None and type != "smart":
322+
if type is not None and type not in {"smart", "camera"}:
299323
from kasa.deviceconfig import DeviceConfig
300324

301325
config = DeviceConfig(host=host, port_override=port, timeout=timeout)
302326
dev = _legacy_type_to_class(type)(host, config=config)
303-
elif type == "smart" or (device_family and encrypt_type):
327+
elif type in {"smart", "camera"} or (device_family and encrypt_type):
328+
if type == "camera":
329+
if not experimental:
330+
error(
331+
"Camera is an experimental type, please enable with --experimental"
332+
)
333+
encrypt_type = "AES"
334+
https = True
335+
device_family = "SMART.IPCAMERA"
336+
304337
from kasa.device import Device
305338
from kasa.deviceconfig import (
306339
DeviceConfig,
@@ -311,10 +344,12 @@ async def cli(
311344

312345
if not encrypt_type:
313346
encrypt_type = "KLAP"
347+
314348
ctype = DeviceConnectionParameters(
315349
DeviceFamily(device_family),
316350
DeviceEncryptionType(encrypt_type),
317351
login_version,
352+
https,
318353
)
319354
config = DeviceConfig(
320355
host=host,

‎kasa/device_factory.py

Lines changed: 17 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -11,6 +11,9 @@
1111
from .device_type import DeviceType
1212
from .deviceconfig import DeviceConfig
1313
from .exceptions import KasaException, UnsupportedDeviceError
14+
from .experimental.smartcamera import SmartCamera
15+
from .experimental.smartcameraprotocol import SmartCameraProtocol
16+
from .experimental.sslaestransport import SslAesTransport
1417
from .iot import (
1518
IotBulb,
1619
IotDevice,
@@ -171,6 +174,7 @@ def get_device_class_from_family(device_type: str) -> type[Device] | None:
171174
"SMART.TAPOHUB": SmartDevice,
172175
"SMART.KASAHUB": SmartDevice,
173176
"SMART.KASASWITCH": SmartDevice,
177+
"SMART.IPCAMERA": SmartCamera,
174178
"IOT.SMARTPLUGSWITCH": IotPlug,
175179
"IOT.SMARTBULB": IotBulb,
176180
}
@@ -188,8 +192,12 @@ def get_protocol(
188192
) -> BaseProtocol | None:
189193
"""Return the protocol from the connection name."""
190194
protocol_name = config.connection_type.device_family.value.split(".")[0]
195+
ctype = config.connection_type
191196
protocol_transport_key = (
192-
protocol_name + "." + config.connection_type.encryption_type.value
197+
protocol_name
198+
+ "."
199+
+ ctype.encryption_type.value
200+
+ (".HTTPS" if ctype.https else "")
193201
)
194202
supported_device_protocols: dict[
195203
str, tuple[type[BaseProtocol], type[BaseTransport]]
@@ -199,10 +207,11 @@ def get_protocol(
199207
"SMART.AES": (SmartProtocol, AesTransport),
200208
"SMART.KLAP": (SmartProtocol, KlapTransportV2),
201209
}
202-
if protocol_transport_key not in supported_device_protocols:
203-
return None
204-
205-
protocol_class, transport_class = supported_device_protocols.get(
206-
protocol_transport_key
207-
) # type: ignore
208-
return protocol_class(transport=transport_class(config=config))
210+
if not (prot_tran_cls := supported_device_protocols.get(protocol_transport_key)):
211+
from .experimental.enabled import Enabled
212+
213+
if Enabled.value and protocol_transport_key == "SMART.AES.HTTPS":
214+
prot_tran_cls = (SmartCameraProtocol, SslAesTransport)
215+
else:
216+
return None
217+
return prot_tran_cls[0](transport=prot_tran_cls[1](config=config))

‎kasa/device_type.py

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -12,6 +12,7 @@ class DeviceType(Enum):
1212
Plug = "plug"
1313
Bulb = "bulb"
1414
Strip = "strip"
15+
Camera = "camera"
1516
WallSwitch = "wallswitch"
1617
StripSocket = "stripsocket"
1718
Dimmer = "dimmer"

‎kasa/deviceconfig.py

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -72,6 +72,7 @@ class DeviceFamily(Enum):
7272
SmartTapoSwitch = "SMART.TAPOSWITCH"
7373
SmartTapoHub = "SMART.TAPOHUB"
7474
SmartKasaHub = "SMART.KASAHUB"
75+
SmartIpCamera = "SMART.IPCAMERA"
7576

7677

7778
def _dataclass_from_dict(klass, in_val):
@@ -118,19 +119,24 @@ class DeviceConnectionParameters:
118119
device_family: DeviceFamily
119120
encryption_type: DeviceEncryptionType
120121
login_version: Optional[int] = None
122+
https: bool = False
121123

122124
@staticmethod
123125
def from_values(
124126
device_family: str,
125127
encryption_type: str,
126128
login_version: Optional[int] = None,
129+
https: Optional[bool] = None,
127130
) -> "DeviceConnectionParameters":
128131
"""Return connection parameters from string values."""
129132
try:
133+
if https is None:
134+
https = False
130135
return DeviceConnectionParameters(
131136
DeviceFamily(device_family),
132137
DeviceEncryptionType(encryption_type),
133138
login_version,
139+
https,
134140
)
135141
except (ValueError, TypeError) as ex:
136142
raise KasaException(

‎kasa/discover.py

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -637,6 +637,7 @@ def _get_device_instance(
637637
type_,
638638
encrypt_type,
639639
discovery_result.mgt_encrypt_schm.lv,
640+
discovery_result.mgt_encrypt_schm.is_support_https,
640641
)
641642
except KasaException as ex:
642643
raise UnsupportedDeviceError(

‎kasa/experimental/__init__.py

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1 @@
1+
"""Package for experimental."""

‎kasa/experimental/enabled.py

Lines changed: 12 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,12 @@
1+
"""Package for experimental enabled."""
2+
3+
4+
class Enabled:
5+
"""Class for enabling experimental functionality."""
6+
7+
value = False
8+
9+
@classmethod
10+
def set(cls, value):
11+
"""Set the enabled value."""
12+
cls.value = value

‎kasa/experimental/smartcamera.py

Lines changed: 84 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,84 @@
1+
"""Module for smartcamera."""
2+
3+
from __future__ import annotations
4+
5+
from ..device_type import DeviceType
6+
from ..smart import SmartDevice
7+
from .sslaestransport import SmartErrorCode
8+
9+
10+
class SmartCamera(SmartDevice):
11+
"""Class for smart cameras."""
12+
13+
async def update(self, update_children: bool = False):
14+
"""Update the device."""
15+
initial_query = {
16+
"getDeviceInfo": {"device_info": {"name": ["basic_info", "info"]}},
17+
"getLensMaskConfig": {"lens_mask": {"name": ["lens_mask_info"]}},
18+
}
19+
resp = await self.protocol.query(initial_query)
20+
self._last_update.update(resp)
21+
info = self._try_get_response(resp, "getDeviceInfo")
22+
self._info = self._map_info(info["device_info"])
23+
self._last_update = resp
24+
25+
def _map_info(self, device_info: dict) -> dict:
26+
basic_info = device_info["basic_info"]
27+
return {
28+
"model": basic_info["device_model"],
29+
"type": basic_info["device_type"],
30+
"alias": basic_info["device_alias"],
31+
"fw_ver": basic_info["sw_version"],
32+
"hw_ver": basic_info["hw_version"],
33+
"mac": basic_info["mac"],
34+
"hwId": basic_info["hw_id"],
35+
"oem_id": basic_info["oem_id"],
36+
}
37+
38+
@property
39+
def is_on(self) -> bool:
40+
"""Return true if the device is on."""
41+
if isinstance(self._last_update["getLensMaskConfig"], SmartErrorCode):
42+
return True
43+
return (
44+
self._last_update["getLensMaskConfig"]["lens_mask"]["lens_mask_info"][
45+
"enabled"
46+
]
47+
== "on"
48+
)
49+
50+
async def set_state(self, on: bool):
51+
"""Set the device state."""
52+
if isinstance(self._last_update["getLensMaskConfig"], SmartErrorCode):
53+
return
54+
query = {
55+
"setLensMaskConfig": {
56+
"lens_mask": {"lens_mask_info": {"enabled": "on" if on else "off"}}
57+
},
58+
}
59+
return await self.protocol.query(query)
60+
61+
@property
62+
def device_type(self) -> DeviceType:
63+
"""Return the device type."""
64+
return DeviceType.Camera
65+
66+
@property
67+
def alias(self) -> str | None:
68+
"""Returns the device alias or nickname."""
69+
if self._info:
70+
return self._info.get("alias")
71+
return None
72+
73+
@property
74+
def hw_info(self) -> dict:
75+
"""Return hardware info for the device."""
76+
return {
77+
"sw_ver": self._info.get("hw_ver"),
78+
"hw_ver": self._info.get("fw_ver"),
79+
"mac": self._info.get("mac"),
80+
"type": self._info.get("type"),
81+
"hwId": self._info.get("hwId"),
82+
"dev_name": self.alias,
83+
"oemId": self._info.get("oem_id"),
84+
}

0 commit comments

Comments
 (0)
0