Initial commit.

app-configs/rikku/homeassistant/custom_components/localtuya/sensor.py

@@ -0,0 +1,166 @@
+"""Platform to present any Tuya DP as a sensor."""
+
+import logging
+import base64
+from functools import partial
+from .config_flow import col_to_select
+
+import voluptuous as vol
+from homeassistant.components.sensor import (
+    DEVICE_CLASSES_SCHEMA,
+    DOMAIN,
+    STATE_CLASSES_SCHEMA,
+    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
+
+_LOGGER = logging.getLogger(__name__)
+
+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."""
+    return {
+        vol.Optional(CONF_UNIT_OF_MEASUREMENT): str,
+        vol.Optional(CONF_DEVICE_CLASS): DEVICE_CLASSES_SCHEMA,
+        vol.Optional(CONF_STATE_CLASS): col_to_select(
+            [sc.value for sc in SensorStateClass]
+        ),
+        vol.Optional(CONF_SCALING): vol.All(
+            vol.Coerce(float), vol.Range(min=-1000000.0, max=1000000.0)
+        ),
+    }
+
+
+class LocalTuyaSensor(LocalTuyaEntity, SensorEntity):
+    """Representation of a Tuya sensor."""
+
+    def __init__(
+        self,
+        device,
+        config_entry,
+        sensorid,
+        **kwargs,
+    ):
+        """Initialize the Tuya sensor."""
+        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 state_class(self) -> str | None:
+        """Return 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 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)
+
+        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)