Basic support for Light categories

README.md

@@ -48,6 +48,9 @@ The integration works locally, but connection to Tuya BLE device requires device
 * Irrigation computer (category_id 'ggq')
   + Irrigation computer (product_id '6pahkcau')
 
+* Lights
+  + Most light products should be supported as the Light class tries to get device description from the cloud when there are added but only Strip Lights (category_id 'dd') Magiacous RGB light bar (product_id 'nvfrtxlq') has has been tested
+
 ## Support project
 
 I am working on this integration in Ukraine. Our country was subjected to brutal aggression by Russia. The war still continues. The capital of Ukraine - Kyiv, where I live, and many other cities and villages are constantly under threat of rocket attacks. Our air defense forces are doing wonders, but they also need support. So if you want to help the development of this integration, donate some money and I will spend it to support our air defense.

custom_components/tuya_ble/__init__.py

@@ -24,6 +24,7 @@
     Platform.NUMBER,
     Platform.SENSOR,
     Platform.BINARY_SENSOR,
+    Platform.LIGHT,
     Platform.SELECT,
     Platform.SWITCH,
     Platform.TEXT,

custom_components/tuya_ble/base.py

@@ -0,0 +1,130 @@
+"""Tuya Home Assistant Base Device Model."""
+from __future__ import annotations
+
+import base64
+from dataclasses import dataclass
+import json
+import struct
+from typing import Any, Literal, Self, overload
+
+from tuya_iot import TuyaDevice, TuyaDeviceManager
+
+from homeassistant.helpers.device_registry import DeviceInfo
+from homeassistant.helpers.dispatcher import async_dispatcher_connect
+from homeassistant.helpers.entity import Entity
+
+from homeassistant.components.tuya.const import (
+    DPCode,
+)
+
+from .util import remap_value
+
+
+@dataclass
+class IntegerTypeData:
+    """Integer Type Data."""
+
+    dpcode: DPCode
+    min: int
+    max: int
+    scale: float
+    step: float
+    unit: str | None = None
+    type: str | None = None
+
+    @property
+    def max_scaled(self) -> float:
+        """Return the max scaled."""
+        return self.scale_value(self.max)
+
+    @property
+    def min_scaled(self) -> float:
+        """Return the min scaled."""
+        return self.scale_value(self.min)
+
+    @property
+    def step_scaled(self) -> float:
+        """Return the step scaled."""
+        return self.step / (10**self.scale)
+
+    def scale_value(self, value: float | int) -> float:
+        """Scale a value."""
+        return value / (10**self.scale)
+
+    def scale_value_back(self, value: float | int) -> int:
+        """Return raw value for scaled."""
+        return int(value * (10**self.scale))
+
+    def remap_value_to(
+        self,
+        value: float,
+        to_min: float | int = 0,
+        to_max: float | int = 255,
+        reverse: bool = False,
+    ) -> float:
+        """Remap a value from this range to a new range."""
+        return remap_value(value, self.min, self.max, to_min, to_max, reverse)
+
+    def remap_value_from(
+        self,
+        value: float,
+        from_min: float | int = 0,
+        from_max: float | int = 255,
+        reverse: bool = False,
+    ) -> float:
+        """Remap a value from its current range to this range."""
+        return remap_value(value, from_min, from_max, self.min, self.max, reverse)
+
+    @classmethod
+    def from_json(cls, dpcode: DPCode, data: str | dict) -> IntegerTypeData | None:
+        """Load JSON string and return a IntegerTypeData object."""
+
+        if isinstance(data, str):
+            parsed = json.loads(data)
+        else:
+            parsed = data
+
+        if parsed is None:
+            return
+
+        return cls(
+            dpcode,
+            min=int(parsed["min"]),
+            max=int(parsed["max"]),
+            scale=float(parsed["scale"]),
+            step=max(float(parsed["step"]), 1),
+            unit=parsed.get("unit"),
+            type=parsed.get("type"),
+        )
+
+    @classmethod
+    def from_dict(cls, dpcode: DPCode, data: dict | None) -> IntegerTypeData | None:
+        """Load Dict and return a IntegerTypeData object."""
+
+        if not dict:
+            return None
+
+        return cls(
+            dpcode,
+            min=int(dict.get("min", 0)),
+            max=int(dict.get("max", 0)),
+            scale=float(dict.get("scale", 0)),
+            step=max(float(dict.get("step", 0)), 1),
+            unit=dict.get("unit"),
+            type=dict.get("type"),
+        )
+
+@dataclass
+class EnumTypeData:
+    """Enum Type Data."""
+
+    dpcode: DPCode
+    range: list[str]
+
+    @classmethod
+    def from_json(cls, dpcode: DPCode, data: str) -> EnumTypeData | None:
+        """Load JSON string and return a EnumTypeData object."""
+        if not (parsed := json.loads(data)):
+            return None
+        return cls(dpcode, **parsed)
+

custom_components/tuya_ble/cloud.py

@@ -32,7 +32,6 @@
     TuyaOpenAPI,
     AuthType,
     TuyaOpenMQ,
-    TuyaDeviceManager,
 )
 
 from .tuya_ble import (
@@ -49,9 +48,12 @@
     CONF_PRODUCT_ID,
     CONF_DEVICE_NAME,
     CONF_PRODUCT_NAME,
+    CONF_FUNCTIONS,
+    CONF_STATUS_RANGE,
     DOMAIN,
     TUYA_API_DEVICES_URL,
     TUYA_API_FACTORY_INFO_URL,
+    TUYA_API_DEVICE_SPECIFICATION,
     TUYA_FACTORY_INFO_MAC,
 )
 
@@ -172,14 +174,15 @@ async def _fill_cache_item(self, item: TuyaCloudCacheItem) -> None:
             item.api.get,
             TUYA_API_DEVICES_URL % (item.api.token_info.uid),
         )
-        if devices_response.get(TUYA_RESPONSE_SUCCESS):
+        if devices_response.get(TUYA_RESPONSE_RESULT):
             devices = devices_response.get(TUYA_RESPONSE_RESULT)
             if isinstance(devices, Iterable):
                 for device in devices:
                     fi_response = await self._hass.async_add_executor_job(
                         item.api.get,
                         TUYA_API_FACTORY_INFO_URL % (device.get("id")),
                     )
+
                     fi_response_result = fi_response.get(TUYA_RESPONSE_RESULT)
                     if fi_response_result and len(fi_response_result) > 0:
                         factory_info = fi_response_result[0]
@@ -200,6 +203,20 @@ async def _fill_cache_item(self, item: TuyaCloudCacheItem) -> None:
                                 CONF_PRODUCT_NAME: device.get("product_name"),
                             }
 
+                            spec_response = await self._hass.async_add_executor_job(
+                                    item.api.get,
+                                    TUYA_API_DEVICE_SPECIFICATION % device.get("id")
+                            )
+
+                            spec_response_result = spec_response.get(TUYA_RESPONSE_RESULT)
+                            if spec_response_result:
+                                functions = spec_response_result.get("functions")
+                                if functions:
+                                    item.credentials[mac][CONF_FUNCTIONS] = functions
+                                status = spec_response_result.get("status")
+                                if status:
+                                    item.credentials[mac][CONF_STATUS_RANGE] = status
+
     async def build_cache(self) -> None:
         global _cache
         data = {}
@@ -258,6 +275,7 @@ async def get_device_credentials(
                         break
             if cache_key:
                 item = _cache.get(cache_key)
+
             if item is None or force_update:
                 if self._is_login_success(await self.login(True)):
                     item = _cache.get(cache_key)
@@ -277,6 +295,8 @@ async def get_device_credentials(
                 credentials.get(CONF_DEVICE_NAME, ""),
                 credentials.get(CONF_PRODUCT_MODEL, ""),
                 credentials.get(CONF_PRODUCT_NAME, ""),
+                credentials.get(CONF_FUNCTIONS, []),
+                credentials.get(CONF_STATUS_RANGE, []),
             )
             _LOGGER.debug("Retrieved: %s", result)
             if save_data:

custom_components/tuya_ble/const.py

@@ -18,9 +18,12 @@
 CONF_DEVICE_NAME: Final = "device_name"
 CONF_PRODUCT_MODEL: Final = "product_model"
 CONF_PRODUCT_NAME: Final = "product_name"
+CONF_FUNCTIONS: Final = "functions"
+CONF_STATUS_RANGE: Final = "status_range"
 
 TUYA_API_DEVICES_URL: Final = "/v1.0/users/%s/devices"
 TUYA_API_FACTORY_INFO_URL: Final = "/v1.0/iot-03/devices/factory-infos?device_ids=%s"
+TUYA_API_DEVICE_SPECIFICATION: Final = "/v1.1/devices/%s/specifications"
 TUYA_FACTORY_INFO_MAC: Final = "mac"
 
 BATTERY_STATE_LOW: Final = "low"