| FazBrowse GitHub Viewer | Trending | | Home |
| Tools: [Download Repo ZIP] [Original HTTPS Page] |
1 parent a211cc0 commit 589d150
13 files changed
| Original file line number | Diff line number | Diff line change | |
|---|---|---|---|
@@ -54,19 +54,22 @@ | |||
| 54 | 54 | from kasa.protocols.smartprotocol import REDACTORS as SMART_REDACTORS | |
| 55 | 55 | from kasa.protocols.smartprotocol import SmartProtocol, _ChildProtocolWrapper | |
| 56 | 56 | 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 | ||
| 58 | 59 | ||
| 59 | 60 | Call = namedtuple("Call", "module method") | |
| 60 | 61 | FixtureResult = namedtuple("FixtureResult", "filename, folder, data, protocol_suffix") | |
| 61 | 62 | ||
| 62 | 63 | SMART_FOLDER = "tests/fixtures/smart/" | |
| 63 | 64 | SMARTCAM_FOLDER = "tests/fixtures/smartcam/" | |
| 64 | 65 | SMART_CHILD_FOLDER = "tests/fixtures/smart/child/" | |
| 66 | + SMARTCAM_CHILD_FOLDER = "tests/fixtures/smartcam/child/" | ||
| 65 | 67 | IOT_FOLDER = "tests/fixtures/iot/" | |
| 66 | 68 | ||
| 67 | 69 | SMART_PROTOCOL_SUFFIX = "SMART" | |
| 68 | 70 | SMARTCAM_SUFFIX = "SMARTCAM" | |
| 69 | 71 | SMART_CHILD_SUFFIX = "SMART.CHILD" | |
| 72 | + SMARTCAM_CHILD_SUFFIX = "SMARTCAM.CHILD" | ||
| 70 | 73 | IOT_SUFFIX = "IOT" | |
| 71 | 74 | ||
| 72 | 75 | NO_GIT_FIXTURE_FOLDER = "kasa-fixtures" | |
@@ -844,9 +847,8 @@ async def get_smart_test_calls(protocol: SmartProtocol): | |||
| 844 | 847 | return test_calls, successes | |
| 845 | 848 | ||
| 846 | 849 | ||
| 847 | - def get_smart_child_fixture(response): | ||
| 850 | + def get_smart_child_fixture(response, model_info, folder, suffix): | ||
| 848 | 851 | """Get a seperate fixture for the child device.""" | |
| 849 | - model_info = SmartDevice._get_device_info(response, None) | ||
| 850 | 852 | hw_version = model_info.hardware_version | |
| 851 | 853 | fw_version = model_info.firmware_version | |
| 852 | 854 | model = model_info.long_name | |
@@ -855,12 +857,68 @@ def get_smart_child_fixture(response): | |||
| 855 | 857 | save_filename = f"{model}_{hw_version}_{fw_version}" | |
| 856 | 858 | return FixtureResult( | |
| 857 | 859 | filename=save_filename, | |
| 858 | - folder=SMART_CHILD_FOLDER, | ||
| 860 | + folder=folder, | ||
| 859 | 861 | data=response, | |
| 860 | - protocol_suffix=SMART_CHILD_SUFFIX, | ||
| 862 | + protocol_suffix=suffix, | ||
| 861 | 863 | ) | |
| 862 | 864 | ||
| 863 | 865 | ||
| 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 | + | ||
| 864 | 922 | async def get_smart_fixtures( | |
| 865 | 923 | protocol: SmartProtocol, | |
| 866 | 924 | *, | |
@@ -917,21 +975,19 @@ async def get_smart_fixtures( | |||
| 917 | 975 | finally: | |
| 918 | 976 | await protocol.close() | |
| 919 | 977 | ||
| 978 | + # Put all the successes into a dict[child_device_id or "", successes[]] | ||
| 920 | 979 | device_requests: dict[str, list[SmartCall]] = {} | |
| 921 | 980 | for success in successes: | |
| 922 | 981 | device_request = device_requests.setdefault(success.child_device_id, []) | |
| 923 | 982 | device_request.append(success) | |
| 924 | 983 | ||
| 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 | - | ||
| 931 | 984 | final = await _make_final_calls( | |
| 932 | 985 | protocol, device_requests[""], "All successes", batch_size, child_device_id="" | |
| 933 | 986 | ) | |
| 934 | 987 | fixture_results = [] | |
| 988 | + | ||
| 989 | + # Make the final child calls | ||
| 990 | + child_responses = {} | ||
| 935 | 991 | for child_device_id, requests in device_requests.items(): | |
| 936 | 992 | if child_device_id == "": | |
| 937 | 993 | continue | |
@@ -942,55 +998,82 @@ async def get_smart_fixtures( | |||
| 942 | 998 | batch_size, | |
| 943 | 999 | child_device_id=child_device_id, | |
| 944 | 1000 | ) | |
| 1001 | + child_responses[child_device_id] = response | ||
| 945 | 1002 | ||
| 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 | ||
| 956 | 1033 | 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 | ||
| 958 | 1038 | 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 | ||
| 962 | 1059 | and child_model != parent_model | |
| 963 | 1060 | ): | |
| 964 | 1061 | 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 | ||
| 966 | 1073 | else: | |
| 967 | 1074 | cd = final.setdefault("child_devices", {}) | |
| 968 | - cd[scrubbed] = response | ||
| 1075 | + cd[scrubbed_child_id] = response | ||
| 969 | 1076 | ||
| 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)) | ||
| 994 | 1077 | discovery_result = None | |
| 995 | 1078 | if discovery_info: | |
| 996 | 1079 | final["discovery_result"] = redact_data( | |
| Original file line number | Diff line number | Diff line change | |
|---|---|---|---|
@@ -13,7 +13,7 @@ | |||
| 13 | 13 | from kasa.device_type import DeviceType | |
| 14 | 14 | from kasa.iot import IotDevice | |
| 15 | 15 | from kasa.smart import SmartDevice | |
| 16 | - from kasa.smartcam import SmartCamDevice | ||
| 16 | + from kasa.smartcam import SmartCamChild, SmartCamDevice | ||
| 17 | 17 | ||
| 18 | 18 | ||
| 19 | 19 | class SupportedVersion(NamedTuple): | |
@@ -49,6 +49,7 @@ class SupportedVersion(NamedTuple): | |||
| 49 | 49 | SMART_FOLDER = "tests/fixtures/smart/" | |
| 50 | 50 | SMART_CHILD_FOLDER = "tests/fixtures/smart/child" | |
| 51 | 51 | SMARTCAM_FOLDER = "tests/fixtures/smartcam/" | |
| 52 | + SMARTCAM_CHILD_FOLDER = "tests/fixtures/smartcam/child" | ||
| 52 | 53 | ||
| 53 | 54 | ||
| 54 | 55 | def generate_supported(args): | |
@@ -66,6 +67,7 @@ def generate_supported(args): | |||
| 66 | 67 | _get_supported_devices(supported, SMART_FOLDER, SmartDevice) | |
| 67 | 68 | _get_supported_devices(supported, SMART_CHILD_FOLDER, SmartDevice) | |
| 68 | 69 | _get_supported_devices(supported, SMARTCAM_FOLDER, SmartCamDevice) | |
| 70 | + _get_supported_devices(supported, SMARTCAM_CHILD_FOLDER, SmartCamChild) | ||
| 69 | 71 | ||
| 70 | 72 | readme_updated = _update_supported_file( | |
| 71 | 73 | README_FILENAME, _supported_summary(supported), print_diffs | |
| Original file line number | Diff line number | Diff line change | |
|---|---|---|---|
@@ -1,5 +1,6 @@ | |||
| 1 | 1 | """Package for supporting tapo-branded cameras.""" | |
| 2 | 2 | ||
| 3 | + from .smartcamchild import SmartCamChild | ||
| 3 | 4 | from .smartcamdevice import SmartCamDevice | |
| 4 | 5 | ||
| 5 | - __all__ = ["SmartCamDevice"] | ||
| 6 | + __all__ = ["SmartCamDevice", "SmartCamChild"] | ||
| Back | FazBrowse Home | New Git URL |
0 commit comments