feat(sensors): handle raw data for "phase_x"

custom_components/localtuya/const.py

@@ -228,3 +228,6 @@ def __post_init__(self) -> None:
         self.reset_dps: str = self.device_config.get(CONF_RESET_DPIDS, "")
         self.manual_dps: str = self.device_config.get(CONF_MANUAL_DPS, "")
         self.dps_strings: list = self.device_config.get(CONF_DPS_STRINGS, [])
+
+    def as_dict(self):
+        return self._device_config

custom_components/localtuya/core/ha_entities/sensors.py

@@ -1045,73 +1045,16 @@ def localtuya_sensor(unit_of_measurement=None, scale_factor: float = 1) -> dict:
         LocalTuyaEntity(
             id=DPCode.PHASE_A,
             name="Phase C Current",
-            device_class=SensorDeviceClass.CURRENT,
-            state_class=SensorStateClass.MEASUREMENT,
-            custom_configs=localtuya_sensor(UnitOfElectricCurrent.MILLIAMPERE),
-            entity_category=EntityCategory.DIAGNOSTIC,
-        ),
-        LocalTuyaEntity(
-            id=DPCode.PHASE_A,
-            name="Phase C Power",
-            device_class=SensorDeviceClass.POWER,
-            state_class=SensorStateClass.MEASUREMENT,
-            custom_configs=localtuya_sensor(UnitOfPower.WATT, 0.1),
-            entity_category=EntityCategory.DIAGNOSTIC,
-        ),
-        LocalTuyaEntity(
-            id=DPCode.PHASE_A,
-            name="Phase A Voltage",
-            device_class=SensorDeviceClass.VOLTAGE,
-            state_class=SensorStateClass.MEASUREMENT,
-            custom_configs=localtuya_sensor(UnitOfElectricPotential.VOLT, 0.1),
-            entity_category=EntityCategory.DIAGNOSTIC,
-        ),
-        LocalTuyaEntity(
-            id=DPCode.PHASE_B,
-            name="Phase B Current",
-            device_class=SensorDeviceClass.CURRENT,
-            state_class=SensorStateClass.MEASUREMENT,
-            custom_configs=localtuya_sensor(UnitOfElectricCurrent.MILLIAMPERE),
-            entity_category=EntityCategory.DIAGNOSTIC,
-        ),
-        LocalTuyaEntity(
-            id=DPCode.PHASE_B,
-            name="Phase B Power",
-            device_class=SensorDeviceClass.POWER,
-            state_class=SensorStateClass.MEASUREMENT,
-            custom_configs=localtuya_sensor(UnitOfPower.WATT, 0.1),
             entity_category=EntityCategory.DIAGNOSTIC,
         ),
         LocalTuyaEntity(
             id=DPCode.PHASE_B,
-            name="Phase B Voltage",
-            device_class=SensorDeviceClass.VOLTAGE,
-            state_class=SensorStateClass.MEASUREMENT,
-            custom_configs=localtuya_sensor(UnitOfElectricPotential.VOLT, 0.1),
-            entity_category=EntityCategory.DIAGNOSTIC,
-        ),
-        LocalTuyaEntity(
-            id=DPCode.PHASE_C,
-            name="Phase C Current",
-            device_class=SensorDeviceClass.CURRENT,
-            state_class=SensorStateClass.MEASUREMENT,
-            custom_configs=localtuya_sensor(UnitOfElectricCurrent.MILLIAMPERE),
+            name="Phase B",
             entity_category=EntityCategory.DIAGNOSTIC,
         ),
         LocalTuyaEntity(
             id=DPCode.PHASE_C,
-            name="Phase C Power",
-            device_class=SensorDeviceClass.POWER,
-            state_class=SensorStateClass.MEASUREMENT,
-            custom_configs=localtuya_sensor(UnitOfPower.WATT, 0.1),
-            entity_category=EntityCategory.DIAGNOSTIC,
-        ),
-        LocalTuyaEntity(
-            id=DPCode.PHASE_C,
-            name="Phase C Voltage",
-            device_class=SensorDeviceClass.VOLTAGE,
-            state_class=SensorStateClass.MEASUREMENT,
-            custom_configs=localtuya_sensor(UnitOfElectricPotential.VOLT, 0.1),
+            name="Phase C",
             entity_category=EntityCategory.DIAGNOSTIC,
         ),
         ## PHASE X Are probably encrypted values. since it duplicated it probably raw dict data.

custom_components/localtuya/entity.py

@@ -95,6 +95,8 @@ async def async_setup_entry(
                         device,
                         dev_entry,
                         entity_config[CONF_ID],
+                        # we need add_entites_callback in-case we want to add sub-entites, such as electric sensor "phase_a"
+                        add_entites_callback=async_add_entities,
                     )
                 )
     # Once the entities have been created, add to the TuyaDevice instance
@@ -139,6 +141,9 @@ def __init__(
         self._last_state = None
         self._stored_states: State | None = None
         self._hass = device._hass
+        self._componet_add_entities: AddEntitiesCallback = kwargs.get(
+            "add_entites_callback"
+        )
         self._loaded = False
 
         # Default value is available to be provided by Platform entities if required
@@ -221,7 +226,7 @@ def device_info(self) -> DeviceInfo:
     @property
     def name(self) -> str:
         """Get name of Tuya entity."""
-        return self._config.get(CONF_FRIENDLY_NAME)
+        return getattr(self, "_attr_name", self._config.get(CONF_FRIENDLY_NAME))
 
     @property
     def icon(self) -> str | None:
@@ -231,6 +236,9 @@ def icon(self) -> str | None:
     @property
     def unique_id(self) -> str:
         """Return unique device identifier."""
+        if getattr(self, "_attr_unique_id") is not None:
+            return self._attr_unique_id
+
         return f"local_{self._device_config.id}_{self._dp_id}"
 
     @property
@@ -257,7 +265,7 @@ def entity_category(self) -> str:
     @property
     def device_class(self):
         """Return the class of this device."""
-        return self._config.get(CONF_DEVICE_CLASS, self._attr_device_class)
+        return getattr(self, "_attr_device_class", self._config.get(CONF_DEVICE_CLASS))
 
     def has_config(self, attr) -> bool:
         """Return if a config parameter has a valid value."""

custom_components/localtuya/sensor.py

@@ -1,6 +1,7 @@
 """Platform to present any Tuya DP as a sensor."""
 
 import logging
+import base64
 from functools import partial
 from .config_flow import col_to_select
 
@@ -9,14 +10,20 @@
     DEVICE_CLASSES_SCHEMA,
     DOMAIN,
     STATE_CLASSES_SCHEMA,
-    SensorStateClass,
+    SensorDeviceClass,
     SensorEntity,
+    SensorStateClass,
 )
 from homeassistant.const import (
     CONF_DEVICE_CLASS,
     CONF_UNIT_OF_MEASUREMENT,
+    Platform,
     STATE_UNKNOWN,
+    UnitOfElectricCurrent,
+    UnitOfElectricPotential,
+    UnitOfPower,
 )
+from homeassistant.helpers import entity_registry as er
 
 from .entity import LocalTuyaEntity, async_setup_entry
 from .const import CONF_SCALING, CONF_STATE_CLASS
@@ -25,6 +32,15 @@
 
 DEFAULT_PRECISION = 2
 
+ATTR_POWER = "power"
+ATTR_VOLTAGE = "voltage"
+ATTR_CURRENT = "current"
+MAP_UOM = {
+    ATTR_CURRENT: UnitOfElectricCurrent.AMPERE,
+    ATTR_VOLTAGE: UnitOfElectricPotential.VOLT,
+    ATTR_POWER: UnitOfPower.KILO_WATT,
+}
+
 
 def flow_schema(dps):
     """Return schema used in config flow."""
@@ -54,36 +70,97 @@ def __init__(
         super().__init__(device, config_entry, sensorid, _LOGGER, **kwargs)
         self._state = None
 
+        self._has_sub_entities = False
+        self._attr_device_class = self._config.get(CONF_DEVICE_CLASS)
+
     @property
     def native_value(self):
         """Return sensor state."""
         return self._state
 
-    @property
-    def device_class(self):
-        """Return the class of this device."""
-        return self._config.get(CONF_DEVICE_CLASS)
-
     @property
     def state_class(self) -> str | None:
         """Return state class."""
-        return self._config.get(CONF_STATE_CLASS)
+        return getattr(self, "_attr_state_class", self._config.get(CONF_STATE_CLASS))
 
     @property
     def native_unit_of_measurement(self):
         """Return the unit of measurement of this entity, if any."""
-        return self._config.get(CONF_UNIT_OF_MEASUREMENT)
+        return getattr(
+            self,
+            "_attr_native_unit_of_measurement",
+            self._config.get(CONF_UNIT_OF_MEASUREMENT),
+        )
 
     def status_updated(self):
         """Device status was updated."""
+
         state = self.dp_value(self._dp_id)
 
-        self._state = self.scale(state)
+        if self.is_base64(state):
+            if not self._has_sub_entities:
+                self._hass.add_job(self.__create_sub_sensors())
+
+            if (sub_sensor := getattr(self, "_attr_sub_sensor", None)) and (
+                sub_state := self.decode_base64(state).get(sub_sensor)
+            ):
+                self._state = sub_state
+            else:
+                self._state = state
+        else:
+            self._state = self.scale(state)
+
+    def status_restored(self, stored_state) -> None:
+        super().status_restored(stored_state)
+
+        if (last_state := self._last_state) and self.is_base64(last_state):
+            self._status.update({self._dp_id: last_state})
 
     # No need to restore state for a sensor
     async def restore_state_when_connected(self):
         """Do nothing for a sensor."""
         return
 
+    def is_base64(self, data):
+        """Return if the data is valid Tuya raw Base64 encoded data."""
+        return (
+            (data and isinstance(data, str))
+            and len(data) >= 12
+            and len(data) % 2 == 0
+            and data.endswith("=")
+        )
+
+    def decode_base64(self, data):
+        """Decode data base64 such as DPS phase_a."""
+        buf = base64.b64decode(data)
+        voltage = (buf[1] | buf[0] << 8) / 10
+        current = (buf[4] | buf[3] << 8) / 1000
+        power = (buf[7] | buf[6] << 8) / 1000
+        return {ATTR_VOLTAGE: voltage, ATTR_CURRENT: current, ATTR_POWER: power}
+
+    async def __create_sub_sensors(self):
+        """Create sub entities for voltage, current and power and hide this parent sensor."""
+        sub_entities = []
+
+        for sensor in (ATTR_CURRENT, ATTR_POWER, ATTR_VOLTAGE):
+            sub_entity = LocalTuyaSensor(
+                self._device, self._device_config.as_dict(), self._dp_id
+            )
+            setattr(sub_entity, "_attr_sub_sensor", sensor)
+            setattr(sub_entity, "_attr_unique_id", f"{self.unique_id}_{sensor}")
+            setattr(sub_entity, "_attr_name", f"{self.name} {sensor.capitalize()}")
+            setattr(sub_entity, "_attr_device_class", SensorDeviceClass(sensor))
+            setattr(sub_entity, "_attr_state_class", SensorStateClass.MEASUREMENT)
+            setattr(sub_entity, "_attr_native_unit_of_measurement", MAP_UOM[sensor])
+            sub_entities.append(sub_entity)
+
+        # Sub entities shouldn't have add entities attr.
+        if sub_entities and self._componet_add_entities:
+            self._has_sub_entities = True
+            self._componet_add_entities(sub_entities)
+            er.async_get(self._hass).async_update_entity(
+                self.entity_id, hidden_by=er.RegistryEntryHider.INTEGRATION
+            )
+
 
 async_setup_entry = partial(async_setup_entry, DOMAIN, LocalTuyaSensor, flow_schema)