Pull in PR rospogrigio#1256

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"

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):