diff --git a/custom_components/localtuya/cloud_api.py b/custom_components/localtuya/cloud_api.py index 680217187..19901f5aa 100644 --- a/custom_components/localtuya/cloud_api.py +++ b/custom_components/localtuya/cloud_api.py @@ -134,3 +134,18 @@ async def async_get_devices_list(self): # _LOGGER.debug("DEV_LIST: %s", self.device_list) return "ok" + +async def async_get_device_specifications(self, device_id): + """Obtain the DP ID mappings for a device.""" + resp = await self.async_make_request( + "GET", url=f"/v1.1/devices/{device_id}/specifications" + ) + + if not resp.ok: + return {}, "Request failed, status " + str(resp.status) + + r_json = resp.json() + if not r_json["success"]: + return {}, f"Error {r_json['code']}: {r_json['msg']}" + + return r_json["result"], "ok" \ No newline at end of file diff --git a/custom_components/localtuya/config_flow.py b/custom_components/localtuya/config_flow.py index 5c87e2547..b1c53a2b4 100644 --- a/custom_components/localtuya/config_flow.py +++ b/custom_components/localtuya/config_flow.py @@ -172,9 +172,15 @@ def schema_defaults(schema, dps_list=None, **defaults): return copy -def dps_string_list(dps_data): +def dps_string_list(dps_data, cloud_dp_codes): """Return list of friendly DPS values.""" - return [f"{id} (value: {value})" for id, value in dps_data.items()] + strs = [] + for dp, value in dps_data.items(): + if dp in cloud_dp_codes: + strs.append(f"{dp} (code: {cloud_dp_codes[dp]}, value: {value})") + else: + strs.append(f"{dp} (value: {value})") + return strs def gen_dps_strings(): @@ -299,7 +305,20 @@ async def validate_input(hass: core.HomeAssistant, data): _LOGGER.debug("Total DPS: %s", detected_dps) - return dps_string_list(detected_dps) + # Get DP descriptions from the cloud, if the device is there. + cloud_dp_codes = {} + if data[CONF_DEVICE_ID] in hass.data[DOMAIN][DATA_CLOUD].device_list: + cloud_device_specs, res = await hass.data[DOMAIN][ + DATA_CLOUD + ].async_get_device_specifications(data[CONF_DEVICE_ID]) + if res != "ok": + _LOGGER.error("Cloud DP specification request failed: %s", res) + else: + for category in ("functions", "status"): + cloud_dp_codes.update( + {str(e["dp_id"]): e["code"] for e in cloud_device_specs[category]} + ) + return dps_string_list(detected_dps, cloud_dp_codes) async def attempt_cloud_connection(hass, user_input):