GV

GeezerVison

The World Wide Web, restored to 800 pixels of glory.

Home|About|Gaming cabinet|Archives|How it works|Starter sites

On this site

  • Sign in
  • ActionsCI/CD and automation
  • CodespacesCloud development environments
  • All Features
  • Documentation
  • Training
  • Certifications
  • Build with Copilot
  • Customize Copilot
  • Integrations
  • Trending repos
  • GitHub Sponsors
  • Developer articles
  • Enterprises

This view

  • Original page
  • New address

GeezerVison is showing github.com — Update dump_devinfo to collect child device info · python-kasa/python-kasa@f33f099

File tree

Expand file treeCollapse file tree
Open diff view settings
Filter options
  • devtools
    • dump_devinfo.py
    • helpers
      • smartrequests.py
Expand file treeCollapse file tree
Open diff view settings
Collapse file

‎devtools/dump_devinfo.py‎

Copy file name to clipboardExpand all lines: devtools/dump_devinfo.py
+154-35Lines changed: 154 additions & 35 deletions
Original file line numberDiff line numberDiff line change
@@ -32,9 +32,10 @@
3232
from kasa.discover import DiscoveryResult
3333
from kasa.exceptions import SmartErrorCode
3434
from kasa.smart import SmartDevice
35+
from kasa.smartprotocol import _ChildProtocolWrapper
3536

3637
Call = namedtuple("Call", "module method")
37-
SmartCall = namedtuple("SmartCall", "module request should_succeed")
38+
SmartCall = namedtuple("SmartCall", "module request should_succeed child_device_id")
3839

3940
_LOGGER = logging.getLogger(__name__)
4041

@@ -69,6 +70,10 @@ def scrub(res):
6970
"parent_device_id", # for hub children
7071
"setup_code", # matter
7172
"setup_payload", # matter
73+
"mfi_setup_code", # mfi_ for homekit
74+
"mfi_setup_id",
75+
"mfi_token_token",
76+
"mfi_token_uuid",
7277
]
7378

7479
for k, v in res.items():
@@ -105,6 +110,8 @@ def scrub(res):
105110
v = "#MASKED_NAME#"
106111
elif isinstance(res[k], int):
107112
v = 0
113+
elif k == "device_id" and "SCRUBBED" in v:
114+
pass # already scrubbed
108115
elif k == "device_id" and len(v) > 40:
109116
# retain the last two chars when scrubbing child ids
110117
end = v[-2:]
@@ -289,8 +296,15 @@ async def _make_requests_or_exit(
289296
requests: List[SmartRequest],
290297
name: str,
291298
batch_size: int,
299+
*,
300+
child_device_id: str,
292301
) -> Dict[str, Dict]:
293302
final = {}
303+
protocol = (
304+
device.protocol
305+
if child_device_id == 0
306+
else _ChildProtocolWrapper(child_device_id, device.protocol)
307+
)
294308
try:
295309
end = len(requests)
296310
step = batch_size # Break the requests down as there seems to be a size limit
@@ -300,9 +314,7 @@ async def _make_requests_or_exit(
300314
request: Union[List[SmartRequest], SmartRequest] = (
301315
requests_step[0] if len(requests_step) == 1 else requests_step
302316
)
303-
responses = await device.protocol.query(
304-
SmartRequest._create_request_dict(request)
305-
)
317+
responses = await protocol.query(SmartRequest._create_request_dict(request))
306318
for method, result in responses.items():
307319
final[method] = result
308320
return final
@@ -331,38 +343,36 @@ async def _make_requests_or_exit(
331343
await device.protocol.close()
332344

333345

334-
async def get_smart_fixture(device: SmartDevice, batch_size: int):
335-
"""Get fixture for new TAPO style protocol."""
346+
async def get_smart_test_calls(device: SmartDevice):
347+
"""Get the list of test calls to make."""
348+
test_calls = []
349+
successes = []
350+
child_device_components = {}
351+
336352
extra_test_calls = [
337353
SmartCall(
338354
module="temp_humidity_records",
339355
request=SmartRequest.get_raw_request("get_temp_humidity_records"),
340356
should_succeed=False,
341-
),
342-
SmartCall(
343-
module="child_device_list",
344-
request=SmartRequest.get_raw_request("get_child_device_list"),
345-
should_succeed=False,
346-
),
347-
SmartCall(
348-
module="child_device_component_list",
349-
request=SmartRequest.get_raw_request("get_child_device_component_list"),
350-
should_succeed=False,
357+
child_device_id="",
351358
),
352359
SmartCall(
353360
module="trigger_logs",
354361
request=SmartRequest.get_raw_request(
355362
"get_trigger_logs", SmartRequest.GetTriggerLogsParams(5, 0)
356363
),
357364
should_succeed=False,
365+
child_device_id="",
358366
),
359367
]
360368

361-
successes = []
362-
363369
click.echo("Testing component_nego call ..", nl=False)
364370
responses = await _make_requests_or_exit(
365-
device, [SmartRequest.component_nego()], "component_nego call", batch_size
371+
device,
372+
[SmartRequest.component_nego()],
373+
"component_nego call",
374+
batch_size=1,
375+
child_device_id="",
366376
)
367377
component_info_response = responses["component_nego"]
368378
click.echo(click.style("OK", fg="green"))
@@ -371,35 +381,109 @@ async def get_smart_fixture(device: SmartDevice, batch_size: int):
371381
module="component_nego",
372382
request=SmartRequest("component_nego"),
373383
should_succeed=True,
384+
child_device_id="",
374385
)
375386
)
376-
377-
test_calls = []
378-
should_succeed = []
379-
380-
for item in component_info_response["component_list"]:
381-
component_id = item["id"]
382-
ver_code = item["ver_code"]
387+
components = {
388+
item["id"]: item["ver_code"]
389+
for item in component_info_response["component_list"]
390+
}
391+
392+
if "child_device" in components:
393+
child_components = await _make_requests_or_exit(
394+
device,
395+
[SmartRequest.get_child_device_component_list()],
396+
"child device component list",
397+
batch_size=1,
398+
child_device_id="",
399+
)
400+
successes.append(
401+
SmartCall(
402+
module="child_component_list",
403+
request=SmartRequest.get_child_device_component_list(),
404+
should_succeed=True,
405+
child_device_id="",
406+
)
407+
)
408+
test_calls.append(
409+
SmartCall(
410+
module="child_device_list",
411+
request=SmartRequest.get_child_device_list(),
412+
should_succeed=True,
413+
child_device_id="",
414+
)
415+
)
416+
# Get list of child components to call
417+
if "control_child" in components:
418+
child_device_components = {
419+
child_component_list["device_id"]: {
420+
item["id"]: item["ver_code"]
421+
for item in child_component_list["component_list"]
422+
}
423+
for child_component_list in child_components[
424+
"get_child_device_component_list"
425+
]["child_component_list"]
426+
}
427+
428+
# Get component calls
429+
for component_id, ver_code in components.items():
430+
if component_id == "child_device":
431+
continue
383432
if (requests := get_component_requests(component_id, ver_code)) is not None:
384433
component_test_calls = [
385-
SmartCall(module=component_id, request=request, should_succeed=True)
434+
SmartCall(
435+
module=component_id,
436+
request=request,
437+
should_succeed=True,
438+
child_device_id="",
439+
)
386440
for request in requests
387441
]
388442
test_calls.extend(component_test_calls)
389-
should_succeed.extend(component_test_calls)
390443
else:
391444
click.echo(f"Skipping {component_id}..", nl=False)
392445
click.echo(click.style("UNSUPPORTED", fg="yellow"))
393446

447+
# Child component calls
448+
for child_device_id, child_components in child_device_components.items():
449+
for component_id, ver_code in child_components.items():
450+
if (requests := get_component_requests(component_id, ver_code)) is not None:
451+
component_test_calls = [
452+
SmartCall(
453+
module=component_id,
454+
request=request,
455+
should_succeed=True,
456+
child_device_id=child_device_id,
457+
)
458+
for request in requests
459+
]
460+
test_calls.extend(component_test_calls)
461+
else:
462+
click.echo(f"Skipping {component_id}..", nl=False)
463+
click.echo(click.style("UNSUPPORTED", fg="yellow"))
464+
394465
test_calls.extend(extra_test_calls)
395466

467+
return test_calls, successes
468+
469+
470+
async def get_smart_fixture(device: SmartDevice, batch_size: int):
471+
"""Get fixture for new TAPO style protocol."""
472+
test_calls, successes = await get_smart_test_calls(device)
473+
396474
for test_call in test_calls:
397475
click.echo(f"Testing {test_call.module}..", nl=False)
398476
try:
399477
click.echo(f"Testing {test_call}..", nl=False)
400-
response = await device.protocol.query(
401-
SmartRequest._create_request_dict(test_call.request)
402-
)
478+
if test_call.child_device_id == 0:
479+
response = await device.protocol.query(
480+
SmartRequest._create_request_dict(test_call.request)
481+
)
482+
else:
483+
cp = _ChildProtocolWrapper(test_call.child_device_id, device.protocol)
484+
response = await cp.query(
485+
SmartRequest._create_request_dict(test_call.request)
486+
)
403487
except AuthenticationError as ex:
404488
_echo_error(
405489
f"Unable to query the device due to an authentication error: {ex}",
@@ -430,13 +514,48 @@ async def get_smart_fixture(device: SmartDevice, batch_size: int):
430514
finally:
431515
await device.protocol.close()
432516

433-
requests = []
434-
for succ in successes:
435-
requests.append(succ.request)
517+
device_requests: Dict[str, List[SmartRequest]] = {}
518+
for success in successes:
519+
device_request = device_requests.setdefault(success.child_device_id, [])
520+
device_request.append(success.request)
521+
522+
scrubbed_device_ids = {
523+
device_id: f"SCRUBBED_CHILD_DEVICE_ID_{index}"
524+
for index, device_id in enumerate(device_requests.keys())
525+
if device_id != ""
526+
}
436527

437528
final = await _make_requests_or_exit(
438-
device, requests, "all successes at once", batch_size
529+
device,
530+
device_requests[""],
531+
"all successes at once",
532+
batch_size,
533+
child_device_id="",
439534
)
535+
for child_device_id, requests in device_requests.items():
536+
if child_device_id == "":
537+
continue
538+
response = await _make_requests_or_exit(
539+
device,
540+
requests,
541+
"all child successes at once",
542+
batch_size,
543+
child_device_id=child_device_id,
544+
)
545+
scrubbed = scrubbed_device_ids[child_device_id]
546+
if "get_device_info" in response and "device_id" in response["get_device_info"]:
547+
response["get_device_info"]["device_id"] = scrubbed
548+
cd = final.setdefault("child_devices", {})
549+
cd[scrubbed] = response
550+
551+
# Scrub the device ids
552+
if gc := final.get("get_child_device_component_list"):
553+
for child in gc["child_component_list"]:
554+
device_id = child["device_id"]
555+
child["device_id"] = scrubbed_device_ids[device_id]
556+
for child in final["get_child_device_list"]["child_device_list"]:
557+
device_id = child["device_id"]
558+
child["device_id"] = scrubbed_device_ids[device_id]
440559

441560
# Need to recreate a DiscoverResult here because we don't want the aliases
442561
# in the fixture, we want the actual field names as returned by the device.
Collapse file

‎devtools/helpers/smartrequests.py‎

Copy file name to clipboardExpand all lines: devtools/helpers/smartrequests.py
+27-1Lines changed: 27 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -75,6 +75,13 @@ class GetRulesParams(SmartRequestParams):
7575

7676
start_index: int = 0
7777

78+
@dataclass
79+
class GetScheduleRulesParams(SmartRequestParams):
80+
"""Get Rules Params."""
81+
82+
start_index: int = 0
83+
schedule_mode: str = ""
84+
7885
@dataclass
7986
class GetTriggerLogsParams(SmartRequestParams):
8087
"""Trigger Logs params."""
@@ -166,6 +173,16 @@ def get_device_time() -> "SmartRequest":
166173
"""Get device time."""
167174
return SmartRequest("get_device_time")
168175

176+
@staticmethod
177+
def get_child_device_list() -> "SmartRequest":
178+
"""Get child device list."""
179+
return SmartRequest("get_child_device_list")
180+
181+
@staticmethod
182+
def get_child_device_component_list() -> "SmartRequest":
183+
"""Get child device component list."""
184+
return SmartRequest("get_child_device_component_list")
185+
169186
@staticmethod
170187
def get_wireless_scan_info(
171188
params: Optional[GetRulesParams] = None,
@@ -179,7 +196,7 @@ def get_wireless_scan_info(
179196
def get_schedule_rules(params: Optional[GetRulesParams] = None) -> "SmartRequest":
180197
"""Get schedule rules."""
181198
return SmartRequest(
182-
"get_schedule_rules", params or SmartRequest.GetRulesParams()
199+
"get_schedule_rules", params or SmartRequest.GetScheduleRulesParams()
183200
)
184201

185202
@staticmethod
@@ -381,4 +398,13 @@ def get_component_requests(component_id, ver_code):
381398
SmartRequest.get_raw_request("get_alarm_configure"),
382399
],
383400
"alarm_logs": [SmartRequest.get_raw_request("get_alarm_triggers")],
401+
"child_device": [
402+
SmartRequest.get_raw_request("get_child_device_list"),
403+
SmartRequest.get_raw_request("get_child_device_component_list"),
404+
],
405+
"control_child": [],
406+
"homekit": [SmartRequest.get_raw_request("get_homekit_info")],
407+
"dimmer_calibration": [],
408+
"fan_control": [],
409+
"overheat_protection": [],
384410
}

0 commit comments

Comments
0 (0)
You are visitor 01402520 — GeezerVison does not store the pages you look up.
Best viewed with Netscape Navigator 4.7 or Internet Explorer 6 at 800×600. No cookies required.