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 — Add smartcam child device support for smartcam hubs (#1413) · rytilahti/python-kasa@2b917bb

File tree

Expand file treeCollapse file tree
Open diff view settings
Filter options
  • devtools
    • dump_devinfo.py
    • generate_supported.py
  • kasa/smartcam
    • __init__.py
    • smartcamchild.py
    • smartcamdevice.py
  • tests
    • device_fixtures.py
    • fakeprotocol_smart.py
    • fakeprotocol_smartcam.py
    • fixtureinfo.py
    • smartcam
      • modules
        • test_camera.py
      • test_smartcamdevice.py
    • test_device.py
    • test_devtools.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
+134-51Lines changed: 134 additions & 51 deletions
Original file line numberDiff line numberDiff line change
@@ -54,19 +54,22 @@
5454
from kasa.protocols.smartprotocol import REDACTORS as SMART_REDACTORS
5555
from kasa.protocols.smartprotocol import SmartProtocol, _ChildProtocolWrapper
5656
from kasa.smart import SmartChildDevice, SmartDevice
57-
from kasa.smartcam import SmartCamDevice
57+
from kasa.smartcam import SmartCamChild, SmartCamDevice
58+
from kasa.smartcam.smartcamchild import CHILD_INFO_FROM_PARENT
5859

5960
Call = namedtuple("Call", "module method")
6061
FixtureResult = namedtuple("FixtureResult", "filename, folder, data, protocol_suffix")
6162

6263
SMART_FOLDER = "tests/fixtures/smart/"
6364
SMARTCAM_FOLDER = "tests/fixtures/smartcam/"
6465
SMART_CHILD_FOLDER = "tests/fixtures/smart/child/"
66+
SMARTCAM_CHILD_FOLDER = "tests/fixtures/smartcam/child/"
6567
IOT_FOLDER = "tests/fixtures/iot/"
6668

6769
SMART_PROTOCOL_SUFFIX = "SMART"
6870
SMARTCAM_SUFFIX = "SMARTCAM"
6971
SMART_CHILD_SUFFIX = "SMART.CHILD"
72+
SMARTCAM_CHILD_SUFFIX = "SMARTCAM.CHILD"
7073
IOT_SUFFIX = "IOT"
7174

7275
NO_GIT_FIXTURE_FOLDER = "kasa-fixtures"
@@ -844,9 +847,8 @@ async def get_smart_test_calls(protocol: SmartProtocol):
844847
return test_calls, successes
845848

846849

847-
def get_smart_child_fixture(response):
850+
def get_smart_child_fixture(response, model_info, folder, suffix):
848851
"""Get a seperate fixture for the child device."""
849-
model_info = SmartDevice._get_device_info(response, None)
850852
hw_version = model_info.hardware_version
851853
fw_version = model_info.firmware_version
852854
model = model_info.long_name
@@ -855,12 +857,68 @@ def get_smart_child_fixture(response):
855857
save_filename = f"{model}_{hw_version}_{fw_version}"
856858
return FixtureResult(
857859
filename=save_filename,
858-
folder=SMART_CHILD_FOLDER,
860+
folder=folder,
859861
data=response,
860-
protocol_suffix=SMART_CHILD_SUFFIX,
862+
protocol_suffix=suffix,
861863
)
862864

863865

866+
def scrub_child_device_ids(
867+
main_response: dict, child_responses: dict
868+
) -> dict[str, str]:
869+
"""Scrub all the child device ids in the responses."""
870+
# Make the scrubbed id map
871+
scrubbed_child_id_map = {
872+
device_id: f"SCRUBBED_CHILD_DEVICE_ID_{index + 1}"
873+
for index, device_id in enumerate(child_responses.keys())
874+
if device_id != ""
875+
}
876+
877+
for child_id, response in child_responses.items():
878+
scrubbed_child_id = scrubbed_child_id_map[child_id]
879+
# scrub the device id in the child's get info response
880+
# The checks for the device_id will ensure we can get a fixture
881+
# even if the data is unexpectedly not available although it should
882+
# always be there
883+
if "get_device_info" in response and "device_id" in response["get_device_info"]:
884+
response["get_device_info"]["device_id"] = scrubbed_child_id
885+
elif (
886+
basic_info := response.get("getDeviceInfo", {})
887+
.get("device_info", {})
888+
.get("basic_info")
889+
) and "dev_id" in basic_info:
890+
basic_info["dev_id"] = scrubbed_child_id
891+
else:
892+
_LOGGER.error(
893+
"Cannot find device id in child get device info: %s", child_id
894+
)
895+
896+
# Scrub the device ids in the parent for smart protocol
897+
if gc := main_response.get("get_child_device_component_list"):
898+
for child in gc["child_component_list"]:
899+
device_id = child["device_id"]
900+
child["device_id"] = scrubbed_child_id_map[device_id]
901+
for child in main_response["get_child_device_list"]["child_device_list"]:
902+
device_id = child["device_id"]
903+
child["device_id"] = scrubbed_child_id_map[device_id]
904+
905+
# Scrub the device ids in the parent for the smart camera protocol
906+
if gc := main_response.get("getChildDeviceComponentList"):
907+
for child in gc["child_component_list"]:
908+
device_id = child["device_id"]
909+
child["device_id"] = scrubbed_child_id_map[device_id]
910+
for child in main_response["getChildDeviceList"]["child_device_list"]:
911+
if device_id := child.get("device_id"):
912+
child["device_id"] = scrubbed_child_id_map[device_id]
913+
continue
914+
elif dev_id := child.get("dev_id"):
915+
child["dev_id"] = scrubbed_child_id_map[dev_id]
916+
continue
917+
_LOGGER.error("Could not find a device id for the child device: %s", child)
918+
919+
return scrubbed_child_id_map
920+
921+
864922
async def get_smart_fixtures(
865923
protocol: SmartProtocol,
866924
*,
@@ -917,21 +975,19 @@ async def get_smart_fixtures(
917975
finally:
918976
await protocol.close()
919977

978+
# Put all the successes into a dict[child_device_id or "", successes[]]
920979
device_requests: dict[str, list[SmartCall]] = {}
921980
for success in successes:
922981
device_request = device_requests.setdefault(success.child_device_id, [])
923982
device_request.append(success)
924983

925-
scrubbed_device_ids = {
926-
device_id: f"SCRUBBED_CHILD_DEVICE_ID_{index}"
927-
for index, device_id in enumerate(device_requests.keys())
928-
if device_id != ""
929-
}
930-
931984
final = await _make_final_calls(
932985
protocol, device_requests[""], "All successes", batch_size, child_device_id=""
933986
)
934987
fixture_results = []
988+
989+
# Make the final child calls
990+
child_responses = {}
935991
for child_device_id, requests in device_requests.items():
936992
if child_device_id == "":
937993
continue
@@ -942,55 +998,82 @@ async def get_smart_fixtures(
942998
batch_size,
943999
child_device_id=child_device_id,
9441000
)
1001+
child_responses[child_device_id] = response
9451002

946-
scrubbed = scrubbed_device_ids[child_device_id]
947-
if "get_device_info" in response and "device_id" in response["get_device_info"]:
948-
response["get_device_info"]["device_id"] = scrubbed
949-
# If the child is a different model to the parent create a seperate fixture
950-
if "get_device_info" in final:
951-
parent_model = final["get_device_info"]["model"]
952-
elif "getDeviceInfo" in final:
953-
parent_model = final["getDeviceInfo"]["device_info"]["basic_info"][
954-
"device_model"
955-
]
1003+
# scrub the child ids
1004+
scrubbed_child_id_map = scrub_child_device_ids(final, child_responses)
1005+
1006+
# Redact data from the main device response. _wrap_redactors ensure we do
1007+
# not redact the scrubbed child device ids and replaces REDACTED_partial_id
1008+
# with zeros
1009+
final = redact_data(final, _wrap_redactors(SMART_REDACTORS))
1010+
1011+
# smart cam child devices provide more information in getChildDeviceList on the
1012+
# parent than they return when queried directly for getDeviceInfo so we will store
1013+
# it in the child fixture.
1014+
if smart_cam_child_list := final.get("getChildDeviceList"):
1015+
child_infos_on_parent = {
1016+
info["device_id"]: info
1017+
for info in smart_cam_child_list["child_device_list"]
1018+
}
1019+
1020+
for child_id, response in child_responses.items():
1021+
scrubbed_child_id = scrubbed_child_id_map[child_id]
1022+
1023+
# Get the parent model for checking whether to create a seperate child fixture
1024+
if model := final.get("get_device_info", {}).get("model"):
1025+
parent_model = model
1026+
elif (
1027+
device_model := final.get("getDeviceInfo", {})
1028+
.get("device_info", {})
1029+
.get("basic_info", {})
1030+
.get("device_model")
1031+
):
1032+
parent_model = device_model
9561033
else:
957-
raise KasaException("Cannot determine parent device model.")
1034+
parent_model = None
1035+
_LOGGER.error("Cannot determine parent device model.")
1036+
1037+
# different model smart child device
9581038
if (
959-
"component_nego" in response
960-
and "get_device_info" in response
961-
and (child_model := response["get_device_info"].get("model"))
1039+
(child_model := response.get("get_device_info", {}).get("model"))
1040+
and parent_model
1041+
and child_model != parent_model
1042+
):
1043+
response = redact_data(response, _wrap_redactors(SMART_REDACTORS))
1044+
model_info = SmartDevice._get_device_info(response, None)
1045+
fixture_results.append(
1046+
get_smart_child_fixture(
1047+
response, model_info, SMART_CHILD_FOLDER, SMART_CHILD_SUFFIX
1048+
)
1049+
)
1050+
# different model smartcam child device
1051+
elif (
1052+
(
1053+
child_model := response.get("getDeviceInfo", {})
1054+
.get("device_info", {})
1055+
.get("basic_info", {})
1056+
.get("device_model")
1057+
)
1058+
and parent_model
9621059
and child_model != parent_model
9631060
):
9641061
response = redact_data(response, _wrap_redactors(SMART_REDACTORS))
965-
fixture_results.append(get_smart_child_fixture(response))
1062+
# There is more info in the childDeviceList on the parent
1063+
# particularly the region is needed here.
1064+
child_info_from_parent = child_infos_on_parent[scrubbed_child_id]
1065+
response[CHILD_INFO_FROM_PARENT] = child_info_from_parent
1066+
model_info = SmartCamChild._get_device_info(response, None)
1067+
fixture_results.append(
1068+
get_smart_child_fixture(
1069+
response, model_info, SMARTCAM_CHILD_FOLDER, SMARTCAM_CHILD_SUFFIX
1070+
)
1071+
)
1072+
# same model child device
9661073
else:
9671074
cd = final.setdefault("child_devices", {})
968-
cd[scrubbed] = response
1075+
cd[scrubbed_child_id] = response
9691076

970-
# Scrub the device ids in the parent for smart protocol
971-
if gc := final.get("get_child_device_component_list"):
972-
for child in gc["child_component_list"]:
973-
device_id = child["device_id"]
974-
child["device_id"] = scrubbed_device_ids[device_id]
975-
for child in final["get_child_device_list"]["child_device_list"]:
976-
device_id = child["device_id"]
977-
child["device_id"] = scrubbed_device_ids[device_id]
978-
979-
# Scrub the device ids in the parent for the smart camera protocol
980-
if gc := final.get("getChildDeviceComponentList"):
981-
for child in gc["child_component_list"]:
982-
device_id = child["device_id"]
983-
child["device_id"] = scrubbed_device_ids[device_id]
984-
for child in final["getChildDeviceList"]["child_device_list"]:
985-
if device_id := child.get("device_id"):
986-
child["device_id"] = scrubbed_device_ids[device_id]
987-
continue
988-
elif dev_id := child.get("dev_id"):
989-
child["dev_id"] = scrubbed_device_ids[dev_id]
990-
continue
991-
_LOGGER.error("Could not find a device for the child device: %s", child)
992-
993-
final = redact_data(final, _wrap_redactors(SMART_REDACTORS))
9941077
discovery_result = None
9951078
if discovery_info:
9961079
final["discovery_result"] = redact_data(
Collapse file

‎devtools/generate_supported.py‎

Copy file name to clipboardExpand all lines: devtools/generate_supported.py
+3-1Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -13,7 +13,7 @@
1313
from kasa.device_type import DeviceType
1414
from kasa.iot import IotDevice
1515
from kasa.smart import SmartDevice
16-
from kasa.smartcam import SmartCamDevice
16+
from kasa.smartcam import SmartCamChild, SmartCamDevice
1717

1818

1919
class SupportedVersion(NamedTuple):
@@ -49,6 +49,7 @@ class SupportedVersion(NamedTuple):
4949
SMART_FOLDER = "tests/fixtures/smart/"
5050
SMART_CHILD_FOLDER = "tests/fixtures/smart/child"
5151
SMARTCAM_FOLDER = "tests/fixtures/smartcam/"
52+
SMARTCAM_CHILD_FOLDER = "tests/fixtures/smartcam/child"
5253

5354

5455
def generate_supported(args):
@@ -66,6 +67,7 @@ def generate_supported(args):
6667
_get_supported_devices(supported, SMART_FOLDER, SmartDevice)
6768
_get_supported_devices(supported, SMART_CHILD_FOLDER, SmartDevice)
6869
_get_supported_devices(supported, SMARTCAM_FOLDER, SmartCamDevice)
70+
_get_supported_devices(supported, SMARTCAM_CHILD_FOLDER, SmartCamChild)
6971

7072
readme_updated = _update_supported_file(
7173
README_FILENAME, _supported_summary(supported), print_diffs
Collapse file

‎kasa/smartcam/__init__.py‎

Copy file name to clipboard
+2-1Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,6 @@
11
"""Package for supporting tapo-branded cameras."""
22

3+
from .smartcamchild import SmartCamChild
34
from .smartcamdevice import SmartCamDevice
45

5-
__all__ = ["SmartCamDevice"]
6+
__all__ = ["SmartCamDevice", "SmartCamChild"]

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.