8000 Use _get_device_info methods for smart and iot devs in devtools (#1265) · ryenitcher/python-kasa@e209d40 · GitHub
[go: up one dir, main page]
Skip to content

Commit e209d40

Browse files
sdb9696sdb9696
authored
Use _get_device_info methods for smart and iot devs in devtools (python-kasa#1265)
1 parent 9d46996 commit e209d40
8000

20 files changed

+386
-168
lines changed

‎SUPPORTED.md

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -179,9 +179,9 @@ All Tapo devices require authentication.<br>Hub-Connected Devices may work acros
179179
### Plugs
180180

181181
- **P100**
182-
- Hardware: 1.0.0 / Firmware: 1.1.3
183-
- Hardware: 1.0.0 / Firmware: 1.3.7
184-
- Hardware: 1.0.0 / Firmware: 1.4.0
182+
- Hardware: 1.0.0 (US) / Firmware: 1.1.3
183+
- Hardware: 1.0.0 (US) / Firmware: 1.3.7
184+
- Hardware: 1.0.0 (US) / Firmware: 1.4.0
185185
- **P110**
186186
- Hardware: 1.0 (EU) / Firmware: 1.0.7
187187
- Hardware: 1.0 (EU) / Firmware: 1.2.3

‎devtools/dump_devinfo.py

Lines changed: 26 additions & 32 deletions
Original file line numberDiff line numberDiff line change
@@ -21,6 +21,7 @@
2121
from collections import defaultdict, namedtuple
2222
from pathlib import Path
2323
from pprint import pprint
24+
from typing import Any
2425

2526
import asyncclick as click
2627

@@ -40,12 +41,13 @@
4041
from kasa.deviceconfig import DeviceEncryptionType, DeviceFamily
4142
from kasa.discover import DiscoveryResult
4243
from kasa.exceptions import SmartErrorCode
44+
from kasa.protocols import IotProtocol
4345
from kasa.protocols.smartcameraprotocol import (
4446
SmartCameraProtocol,
4547
_ChildCameraProtocolWrapper,
4648
)
4749
from kasa.protocols.smartprotocol import SmartProtocol, _ChildProtocolWrapper
48-
from kasa.smart import SmartChildDevice
50+
from kasa.smart import SmartChildDevice, SmartDevice
4951
from kasa.smartcamera import SmartCamera
5052

5153
Call = namedtuple("Call", "module method")
@@ -389,7 +391,9 @@ async def cli(
389391
)
390392

391393

392-
async def get_legacy_fixture(protocol, *, discovery_info):
394+
async def get_legacy_fixture(
395+
protocol: IotProtocol, *, discovery_info: dict[str, Any] | None
396+
) -> FixtureResult:
393397
"""Get fixture for legacy IOT style protocol."""
394398
items = [
395399
Call(module="system", method="get_sysinfo"),
@@ -422,8 +426,8 @@ async def get_legacy_fixture(protocol, *, discovery_info):
422426
finally:
423427
await protocol.close()
424428

425-
final_query = defaultdict(defaultdict)
426-
final = defaultdict(defaultdict)
429+
final_query: dict = defaultdict(defaultdict)
430+
final: dict = defaultdict(defaultdict)
427431
for succ, resp in successes:
428432
final_query[succ.module][succ.method] = {}
429433
final[succ.module][succ.method] = resp
@@ -433,16 +437,14 @@ async def get_legacy_fixture(protocol, *, discovery_info):
433437
try:
434438
final = await protocol.query(final_query)
435439
except Exception as ex:
436-
_echo_error(f"Unable to query all successes at once: {ex}", bold=True, fg="red")
440+
_echo_error(f"Unable to query all successes at once: {ex}")
437441
finally:
438442
await protocol.close()
439443
if discovery_info and not discovery_info.get("system"):
440444
# Need to recreate a DiscoverResult here because we don't want the aliases
441445
# in the fixture, we want the actual field names as returned by the device.
442-
dr = DiscoveryResult.from_dict(protocol._discovery_info)
443-
final["discovery_result"] = dr.dict(
444-
by_alias=False, exclude_unset=True, exclude_none=True, exclude_defaults=True
445-
)
446+
dr = DiscoveryResult.from_dict(discovery_info)
447+
final["discovery_result"] = dr.to_dict()
446448

447449
click.echo("Got %s successes" % len(successes))
448450
click.echo(click.style("## device info file ##", bold=True))
@@ -817,23 +819,21 @@ async def get_smart_test_calls(protocol: SmartProtocol):
817819

818820
def get_smart_child_fixture(response):
819821
"""Get a seperate fixture for the child device."""
820-
info = response["get_device_info"]
821-
hw_version = info["hw_ver"]
822-
sw_version = info["fw_ver"]
823-
sw_version = sw_version.split(" ", maxsplit=1)[0]
824-
model = info["model"]
825-
if region := info.get("specs"):
826-
model += f"({region})"
827-
828-
save_filename = f"{model}_{hw_version}_{sw_version}.json"
822+
model_info = SmartDevice._get_device_info(response, None)
823+
hw_version = model_info.hardware_version
824+
fw_version = model_info.firmware_version
825+
model = model_info.long_name
826+
if model_info.region is not None:
827+
model = f"{model}({model_info.region})"
828+
save_filename = f"{model}_{hw_version}_{fw_version}.json"
829829
return FixtureResult(
830830
filename=save_filename, folder=SMART_CHILD_FOLDER, data=response
831831
)
832832

833833

834834
async def get_smart_fixtures(
835-
protocol: SmartProtocol, *, discovery_info=None, batch_size: int
836-
):
835+
protocol: SmartProtocol, *, discovery_info: dict[str, Any] | None, batch_size: int
836+
) -> list[FixtureResult]:
837837
"""Get fixture for new TAPO style protocol."""
838838
if isinstance(protocol, SmartCameraProtocol):
839839
test_calls, successes = await get_smart_camera_test_calls(protocol)
@@ -964,23 +964,17 @@ async def get_smart_fixtures(
964964

965965
if "get_device_info" in final:
966966
# smart protocol
967-
hw_version = final["get_device_info"]["hw_ver"]
968-
sw_version = final["get_device_info"]["fw_ver"]
969-
if discovery_info:
970-
model = discovery_info["device_model"]
971-
else:
972-
model = final["get_device_info"]["model"] + "(XX)"
973-
sw_version = sw_version.split(" ", maxsplit=1)[0]
967+
model_info = SmartDevice._get_device_info(final, discovery_info)
974968
copy_folder = SMART_FOLDER
975969
else:
976970
# smart camera protocol
977971
model_info = SmartCamera._get_device_info(final, discovery_info)
978-
model = model_info.long_name
979-
hw_version = model_info.hardware_version
980-
sw_version = model_info.firmare_version
981-
if model_info.region is not None:
982-
model = f"{model}({model_info.region})"
983972
copy_folder = SMARTCAMERA_FOLDER
973+
hw_version = model_info.hardware_version
974+
sw_version = model_info.firmware_version
975+
model = model_info.long_name
976+
if model_info.region is not None:
977+
model = f"{model}({model_info.region})"
984978

985979
save_filename = f"{model}_{hw_version}_{sw_version}.json"
986980

‎devtools/generate_supported.py

Lines changed: 18 additions & 65 deletions
Original file line numberDiff line numberDiff line change
@@ -1,23 +1,25 @@
11
#!/usr/bin/env python
22
"""Script that checks supported devices and updates README.md and SUPPORTED.md."""
33

4+
from __future__ import annotations
5+
46
import json
57
import os
68
import sys
79
from pathlib import Path
810
from string import Template
9-
from typing import NamedTuple
11+
from typing import Any, NamedTuple
1012

11-
from kasa.device_factory import _get_device_type_from_sys_info
1213
from kasa.device_type import DeviceType
14+
from kasa.iot import IotDevice
1315
from kasa.smart import SmartDevice
1416
from kasa.smartcamera import SmartCamera
1517

1618

1719
class SupportedVersion(NamedTuple):
1820
"""Supported version."""
1921

20-
region: str
22+
region: str | None
2123
hw: str
2224
fw: str
2325
auth: bool
@@ -45,6 +47,7 @@ class SupportedVersion(NamedTuple):
4547

4648
IOT_FOLDER = "tests/fixtures/"
4749
SMART_FOLDER = "tests/fixtures/smart/"
50+
SMART_CHILD_FOLDER = "tests/fixtures/smart/child"
4851
SMARTCAMERA_FOLDER = "tests/fixtures/smartcamera/"
4952

5053

@@ -59,9 +62,10 @@ def generate_supported(args):
5962

6063
supported = {"kasa": {}, "tapo": {}}
6164

62-
_get_iot_supported(supported)
63-
_get_smart_supported(supported)
64-
_get_smartcamera_supported(supported)
65+
_get_supported_devices(supported, IOT_FOLDER, IotDevice)
66+
_get_supported_devices(supported, SMART_FOLDER, SmartDevice)
67+
_get_supported_devices(supported, SMART_CHILD_FOLDER, SmartDevice)
68+
_get_supported_devices(supported, SMARTCAMERA_FOLDER, SmartCamera)
6569

6670
readme_updated = _update_supported_file(
6771
README_FILENAME, _supported_summary(supported), print_diffs
@@ -201,49 +205,16 @@ def _supported_text(
201205
return brands
202206

203207

204-
def _get_smart_supported(supported):
205-
for file in Path(SMART_FOLDER).glob("**/*.json"):
206-
with file.open() as f:
207-
fixture_data = json.load(f)
208-
209-
if "discovery_result" in fixture_data:
210-
model, _, region = fixture_data["discovery_result"][
211-
"device_model"
212-
].partition("(")
213-
device_type = fixture_data["discovery_result"]["device_type"]
214-
else: # child devices of hubs do not have discovery result
215-
model = fixture_data["get_device_info"]["model"]
216-
region = fixture_data["get_device_info"].get("specs")
217-
device_type = fixture_data["get_device_info"]["type"]
218-
# P100 doesn't have region HW
219-
region = region.replace(")", "") if region else ""
220-
221-
_protocol, devicetype = device_type.split(".")
222-
brand = devicetype[:4].lower()
223-
components = [
224-
component["id"]
225-
for component in fixture_data["component_nego"]["component_list"]
226-
]
227-
dt = SmartDevice._get_device_type_from_components(components, device_type)
228-
supported_type = DEVICE_TYPE_TO_PRODUCT_GROUP[dt]
229-
230-
hw_version = fixture_data["get_device_info"]["hw_ver"]
231-
fw_version = fixture_data["get_device_info"]["fw_ver"]
232-
fw_version = fw_version.split(" ", maxsplit=1)[0]
233-
234-
stype = supported[brand].setdefault(supported_type, {})
235-
smodel = stype.setdefault(model, [])
236-
smodel.append(
237-
SupportedVersion(region=region, hw=hw_version, fw=fw_version, auth=True)
238-
)
239-
240-
241-
def _get_smartcamera_supported(supported):
242-
for file in Path(SMARTCAMERA_FOLDER).glob("**/*.json"):
208+
def _get_supported_devices(
209+
supported: dict[str, Any],
210+
fixture_location: str,
211+
device_cls: type[IotDevice | SmartDevice | SmartCamera],
212+
):
213+
for file in Path(fixture_location).glob("*.json"):
243214
with file.open() as f:
244215
fixture_data = json.load(f)
245216

246-
model_info = SmartCamera._get_device_info(
217+
model_info = device_cls._get_device_info(
247218
fixture_data, fixture_data.get("discovery_result")
248219
)
249220

@@ -255,30 +226,12 @@ def _get_smartcamera_supported(supported):
255226
SupportedVersion(
256227
region=model_info.region,
257228
hw=model_info.hardware_version,
258-
fw=model_info.firmare_version,
229+
fw=model_info.firmware_version,
259230
auth=model_info.requires_auth,
260231
)
261232
)
262233

263234

264-
def _get_iot_supported(supported):
265-
for file in Path(IOT_FOLDER).glob("*.json"):
266-
with file.open() as f:
267-
fixture_data = json.load(f)
268-
sysinfo = fixture_data["system"]["get_sysinfo"]
269-
dt = _get_device_type_from_sys_info(fixture_data)
270-
supported_type = DEVICE_TYPE_TO_PRODUCT_GROUP[dt]
271-
272-
model, _, region = sysinfo["model"][:-1].partition("(")
273-
auth = "discovery_result" in fixture_data
274-
stype = supported["kasa"].setdefault(supported_type, {})
275-
smodel = stype.setdefault(model, [])
276-
fw = sysinfo["sw_ver"].split(" ", maxsplit=1)[0]
277-
smodel.append(
278-
SupportedVersion(region=region, hw=sysinfo["hw_ver"], fw=fw, auth=auth)
279-
)
280-
281-
282235
def main():
283236
"""Entry point to module."""
284237
generate_supported(sys.argv[1:])

‎kasa/device.py

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -162,7 +162,7 @@ class _DeviceInfo:
162162
device_family: str
163163
device_type: DeviceType
164164
hardware_version: str
165-
firmare_version: str
165+
firmware_version: str
166166
firmware_build: str
167167
requires_auth: bool
168168
region: str | None

‎kasa/device_factory.py

Lines changed: 1 addition & 29 deletions
Original file line numberDiff line numberDiff line change
@@ -128,34 +128,6 @@ def _perf_log(has_params: bool, perf_type: str) -> None:
128128
)
129129

130130

131-
def _get_device_type_from_sys_info(info: dict[str, Any]) -> DeviceType:
132-
"""Find SmartDevice subclass for device described by passed data."""
133-
if "system" not in info or "get_sysinfo" not in info["system"]:
134-
raise KasaException("No 'system' or 'get_sysinfo' in response")
135-
136-
sysinfo: dict[str, Any] = info["system"]["get_sysinfo"]
137-
type_: str | None = sysinfo.get("type", sysinfo.get("mic_type"))
138-
if type_ is None:
139-
raise KasaException("Unable to find the device type field!")
140-
141-
if "dev_name" in sysinfo and "Dimmer" in sysinfo["dev_name"]:
142-
return DeviceType.Dimmer
143-
144-
if "smartplug" in type_.lower():
145-
if "children" in sysinfo:
146-
return DeviceType.Strip
147-
if (dev_name := sysinfo.get("dev_name")) and "light" in dev_name.lower():
148-
return DeviceType.WallSwitch
149-
return DeviceType.Plug
150-
151-
if "smartbulb" in type_.lower():
152-
if "length" in sysinfo: # strips have length
153-
return DeviceType.LightStrip
154-
155-
return DeviceType.Bulb
156-
raise UnsupportedDeviceError(f"Unknown device type: {type_}")
157-
158-
159131
def get_device_class_from_sys_info(sysinfo: dict[str, Any]) -> type[IotDevice]:
160132
"""Find SmartDevice subclass for device described by passed data."""
161133
TYPE_TO_CLASS = {
@@ -166,7 +138,7 @@ def get_device_class_from_sys_info(sysinfo: dict[str, Any]) -> type[IotDevice]:
166138
DeviceType.WallSwitch: IotWallSwitch,
167139
DeviceType.LightStrip: IotLightStrip,
168140
}
169-
return TYPE_TO_CLASS[_get_device_type_from_sys_info(sysinfo)]
141+
return TYPE_TO_CLASS[IotDevice._get_device_type_from_sys_info(sysinfo)]
170142

171143

172144
def get_device_class_from_family(

0 commit comments

Comments
 (0)
0