From d066dfbca06aab26715512ca185908919d99837e Mon Sep 17 00:00:00 2001 From: Zachary Wander Date: Fri, 30 Jan 2026 17:13:05 -0500 Subject: [PATCH] Persistently cache brightness, color temp, and RGB --- custom_components/sengled_udp/conversion.py | 198 ++++++++++++++++++++ custom_components/sengled_udp/light.py | 139 ++++++++++---- 2 files changed, 296 insertions(+), 41 deletions(-) create mode 100644 custom_components/sengled_udp/conversion.py diff --git a/custom_components/sengled_udp/conversion.py b/custom_components/sengled_udp/conversion.py new file mode 100644 index 0000000..5dd4cbc --- /dev/null +++ b/custom_components/sengled_udp/conversion.py @@ -0,0 +1,198 @@ +import math +from typing import Tuple + +def unscale_sengled_value_by_brightness( + sengled_value: Tuple[int, int, int], + brightness_fraction: float, +) -> Tuple[int, int, int]: + """ + Use to unscale Sengled's returned UDP RGB value using the current + brightness fraction for reversal and comparison calculations. + The returned value is already scaled by the current brightness, + so we need to reverse that scaling since we track color and + brightness separately. + :param sengled_value: the scaled RGB value returned by the light. + :param brightness_fraction: the current brightness of the light + :return: the unscaled RGB UDP value. + """ + + res = list(sengled_value) + + for i in range(len(sengled_value)): + channel = sengled_value[i] + + # Don't unscale 0 or 1; we can't know if the weakest + # channel value would change based on brightness. + if channel == (0 or 1): + continue + + res[i] = round(channel / brightness_fraction) + + max_val = max(res) + + # Unscaling isn't perfect because of rounding, but we need + # the strongest channel to be 99. In some cases it can unscale + # to 98, so always force it to be 99. + if max_val < 99: + max_idx = res.index(max_val) + res[max_idx] = 99 + + return tuple[int, int, int](res) + +def is_likely_match( + calculated_sengled_value: Tuple[int, int, int], + api_sengled_value: Tuple[int, int, int], +) -> bool: + f""" + Compare two full-brightness Sengled-encoded colors + to see if they're likely the same. + + :param calculated_sengled_value: value calculated using {calculate}. + :param api_sengled_value: value returned by light and passed through + {calculated_sengled_value}. + :return: whether the two values likely represent the same color. + """ + + calc_r, calc_g, calc_b = calculated_sengled_value + api_r, api_g, api_b = api_sengled_value + + # If everything is equal, don't bother trying heuristics + if calc_r == api_r and calc_g == api_g and calc_b == api_b: + return True + + # If weakest or strongest channel is different, + # colors aren't the same + for i in calculated_sengled_value: + # Our RGB -> Sengled function sometimes returns + # 1 or 0 where Sengled has 0 or 1, so treat them + # interchangeably for comparison purposes. + if calculated_sengled_value[i] == (0 or 1): + if api_sengled_value[i] != (0 or 1): + return False + + if calculated_sengled_value[i] == 99: + if api_sengled_value[i] != 99: + return False + + if calculated_sengled_value[i] != (0 or 1 or 99): + calc_mid = calculated_sengled_value[i] + api_mid = api_sengled_value[i] + + abs_diff = abs(calc_mid - api_mid) + + # 5 is a random error threshold choice. + if abs_diff > 5: + return False + + return True + +def calculate(r_in: int, g_in: int, b_in: int) -> tuple[int, int, int]: + """ + Approximate whatever Sengled is doing to convert RGB + to what it returns in its UDP response. This was mostly + written by AI, so it's not exact, but it does get close. + We can use this to compare our cached RGB value to what + the UDP response has to determine if our cached value is + likely accurate. This function expects full-brightness RGB + values, i.e, any shade of gray would be 255,255,255 with a + fractional brightness value. + + :param r_in: the cached red value. + :param g_in: the cached green value. + :param b_in: the cached blue value. + :return: the approximated UDP representation as a tuple. + """ + + # Full white equals 19, 19, 19 for some reason. + if r_in == g_in == b_in: + return 19, 19, 19 + + in_vals = [r_in, g_in, b_in] + max_val, min_val = max(in_vals), min(in_vals) + max_idx, min_idx = in_vals.index(max_val), in_vals.index(min_val) + mid_idx = 3 - (max_idx + min_idx) + + mid_val = in_vals[mid_idx] + + gammas = { + 0: {1: 2.55, 2: 3.00}, # Red Max + 1: {0: 2.00, 2: 3.00}, # Green Max + 2: {0: 2.15, 1: 1.05} # Blue Max + } + + exponent = gammas[max_idx][mid_idx] + + res = [0, 0, 0] + res[max_idx] = 99 + res[mid_idx] = round(math.pow(mid_val / 255, exponent) * 99) + res[min_idx] = 1 if (min_val > (mid_val / 2) or min_idx == 1) and min_val > 50 else 0 + + return tuple[int, int, int](res) + +def smart_reverse(r_out: int, g_out: int, b_out: int) -> tuple[int, int, int]: + f""" + Best-effort reversal of the values Sengled returns in its + UDP response for the current bulb color. This function was + mostly made using AI, so it isn't exact, but it can never + be exact since Sengled's "encoding" method for converting + RGB to the values it returns through UDP is inherently + lossy. This function returns an estimated RGB value based on + Sengled's values but the weakest/dimmest channel will always be + pretty inaccurate. + + Note that Sengled's UDP API returns values linearly scaled by + brightness except for the weakest channel, and this function + assumes 100% brightness. Use {unscale_sengled_value_by_brightness} + on values returned by lights before passing to this function. + + :param r_out: the encoded red value. + :param g_out: the encoded green value. + :param b_out: the encoded blue value. + :return: the estimated RGB value as a tuple. + """ + + # 1. Grayscale Exception + if r_out == g_out == b_out: + return tuple[int, int, int]([round(r_out * (255/19))] * 3) + + out_vals = [r_out, g_out, b_out] + max_val, min_val = max(out_vals), min(out_vals) + max_idx, min_idx = out_vals.index(max_val), out_vals.index(min_val) + mid_idx = 3 - (max_idx + min_idx) + + # 2. Re-map the exact Gammas used in forward calculation + gammas = { + 0: {1: 2.55, 2: 3.00}, # Red Max + 1: {0: 2.00, 2: 3.00}, # Green Max + 2: {0: 2.15, 1: 1.05} # Blue Max + } + + exponent = gammas[max_idx][mid_idx] + + # 3. The Reverse Formula + res = [0, 0, 0] + res[max_idx] = 255 # Assume ceiling was 255 + + # Reverse the Mid Channel + mid_ratio = math.pow(out_vals[mid_idx] / 99, 1 / exponent) + res[mid_idx] = round(mid_ratio * 255) + + # 4. Smart Floor Estimation + # If out is 1, the input was likely between 50-80. + # If out is 0, it was likely between 0-40. + if min_val == 1: + if min_idx != 1: + # If not green channel, likely at least + # half of mid channel's strength + res[min_idx] = round(res[mid_idx] * 0.65) + else: + # Estimate based on the Mid channel's strength + res[min_idx] = round(45 + (res[mid_idx] * 0.15)) + else: + # Estimate a "dark" floor + res[min_idx] = min( + round(res[mid_idx] * res[max_idx] * 0.001), + round(res[mid_idx] / 2) + ) + + return tuple[int, int, int](res) diff --git a/custom_components/sengled_udp/light.py b/custom_components/sengled_udp/light.py index 530fc72..cf292c0 100644 --- a/custom_components/sengled_udp/light.py +++ b/custom_components/sengled_udp/light.py @@ -1,4 +1,5 @@ import asyncio +import dataclasses import json import logging import socket @@ -15,6 +16,9 @@ from homeassistant.config_entries import ConfigEntry from homeassistant.core import HomeAssistant from homeassistant.helpers.entity_platform import AddEntitiesCallback +from homeassistant.helpers.storage import Store + +from .conversion import calculate, is_likely_match, smart_reverse, unscale_sengled_value_by_brightness _LOGGER = logging.getLogger(__name__) @@ -56,11 +60,18 @@ async def async_setup_entry( async_add_entities(entities) +@dataclasses.dataclass +class LightCache: + brightness: int | None + rgb: tuple[int, int, int] | None + color_temp_kelvin: int | None class SengledLight(LightEntity): """Representation of a Sengled UDP Light.""" _attr_force_update = True + _storage: Store[dict[str, Any]] | None + _cache: LightCache | None def __init__( self, @@ -81,11 +92,6 @@ def __init__( self._attr_rgb_color = (255, 255, 255) self._attr_color_temp_kelvin = None - # --- CACHES FOR STABILITY --- - self._req_kelvin = None - self._req_rgb = None - # ---------------------------- - # Debounce timer to prevent reading stale state immediately after a command self._last_req_time = 0.0 @@ -117,6 +123,14 @@ def __init__( self._attr_min_color_temp_kelvin = 2000 self._attr_max_color_temp_kelvin = 6500 + async def async_added_to_hass(self) -> None: + self._storage = Store( + hass = self.hass, + version=1, + key=self.unique_id, + ) + await self._load_cache() + @property def is_on(self) -> bool: return self._attr_is_on @@ -142,12 +156,20 @@ async def async_update(self) -> None: if (time.time() - self._last_req_time) < 2.0: return + # Probably not necessary since we're currently only saving cache from + # this function as well. + await self._load_cache() + status = await self._get_device_status() if status: if self._is_rgb is None: self._detect_capabilities(status) brightness_info = await self._get_device_brightness() - self._update_state_from_status(status, brightness_info) + await self._update_state_from_status(status, brightness_info) + else: + self._attr_brightness = self._cache.brightness + self._attr_color_temp_kelvin = self._cache.color_temp_kelvin + self._attr_rgb_color = self._cache.rgb def _detect_capabilities(self, status: Dict[str, Any]) -> None: has_rgb = "R" in status and "G" in status and "B" in status @@ -182,28 +204,28 @@ async def async_turn_on(self, **kwargs: Any) -> None: if ATTR_COLOR_TEMP_KELVIN in kwargs: color_temp_kelvin = kwargs[ATTR_COLOR_TEMP_KELVIN] device_temp = self._kelvin_to_device_temp(color_temp_kelvin) - + # Update attributes self._attr_color_temp_kelvin = color_temp_kelvin self._attr_color_mode = ColorMode.COLOR_TEMP self._attr_rgb_color = None - + # Update caches - self._req_kelvin = color_temp_kelvin - self._req_rgb = None + self._cache.color_temp_kelvin = color_temp_kelvin + self._cache.rgb = None # Handle RGB color elif ATTR_RGB_COLOR in kwargs: rgb = kwargs[ATTR_RGB_COLOR] - + # Update attributes self._attr_rgb_color = rgb self._attr_color_mode = ColorMode.RGB self._attr_color_temp_kelvin = None # Update caches - self._req_rgb = rgb - self._req_kelvin = None + self._cache.rgb = rgb + self._cache.color_temp_kelvin = None # Update local state before sending commands self._attr_is_on = True @@ -229,6 +251,7 @@ async def async_turn_on(self, **kwargs: Any) -> None: await self._send_command("set_device_switch", {"switch": 1}) self.async_write_ha_state() + await self._save_cache() async def async_turn_off(self, **kwargs: Any) -> None: """Turn off the light.""" @@ -255,8 +278,10 @@ async def _get_device_brightness(self) -> Optional[Dict[str, Any]]: return result return None - def _update_state_from_status( - self, status: Dict[str, Any], brightness_info: Optional[Dict[str, Any]] = None + async def _update_state_from_status( + self, + status: Dict[str, Any], + brightness_info: Optional[Dict[str, Any]] = None, ) -> None: """Update internal state from device status.""" try: @@ -305,8 +330,11 @@ def _update_state_from_status( w_percent2 = None if w_percent2 is not None and 0 <= w_percent2 <= 100: self._attr_brightness = min(255, max(0, int((w_percent2 / 100) * 255))) - else: - self._attr_brightness = self._attr_brightness if self._attr_is_on else 0 + + if self._cache.brightness is not None: + # Probably within calculation error, use cached value + if abs(self._cache.brightness - self._attr_brightness) < 3: + self._attr_brightness = self._cache.brightness self._attr_color_mode = ColorMode.BRIGHTNESS self._available = True @@ -334,8 +362,8 @@ def _update_state_from_status( self._attr_rgb_color = None # Use cached request if available to prevent math drift/jitter - if self._req_kelvin is not None: - self._attr_color_temp_kelvin = self._req_kelvin + if self._cache.color_temp_kelvin is not None: + self._attr_color_temp_kelvin = self._cache.color_temp_kelvin else: # Fallback Math max_raw = max(r_raw, g_raw, b_raw, w_raw, 1) @@ -368,34 +396,53 @@ def _update_state_from_status( # RGB Mode (W is strictly 0) self._attr_color_mode = ColorMode.RGB self._attr_color_temp_kelvin = None - self._req_kelvin = None - # Use cached request if values are zeroed out by dimness - if self._req_rgb is not None: - self._attr_rgb_color = self._req_rgb + # Brightness + if brightness_info and "brightness" in brightness_info: + device_brightness = brightness_info["brightness"] + self._attr_brightness = min(255, int((device_brightness / 100) * 255)) else: - max_rgb = max(r_raw, g_raw, b_raw, 1) - self._attr_rgb_color = ( - int((r_raw / max_rgb) * 255), - int((g_raw / max_rgb) * 255), - int((b_raw / max_rgb) * 255), + max_channel_value = max(r_raw, g_raw, b_raw, w_raw) + if max_channel_value > 100: + self._attr_brightness = max_channel_value + else: + self._attr_brightness = int((max_channel_value / 100) * 255) + + if self._cache.brightness is not None: + # Probably within calculation error, use cached value + if abs(self._cache.brightness - self._attr_brightness) < 3: + self._attr_brightness = self._cache.brightness + + brightness_factor = self._attr_brightness / 255 + + # Use cached request if values are zeroed out by dimness + if self._cache.rgb is not None: + cached_as_udp = calculate(*self._cache.rgb) + likely_match = is_likely_match( + calculated_sengled_value=cached_as_udp, + api_sengled_value=unscale_sengled_value_by_brightness( + (r_raw, g_raw, b_raw), + brightness_factor, + ), ) - # Brightness - if brightness_info and "brightness" in brightness_info: - device_brightness = brightness_info["brightness"] - self._attr_brightness = min(255, int((device_brightness / 100) * 255)) - else: - max_channel_value = max(r_raw, g_raw, b_raw, w_raw) - if max_channel_value > 100: - self._attr_brightness = max_channel_value + if likely_match: + self._attr_rgb_color = self._cache.rgb + else: + self._attr_rgb_color = smart_reverse(*unscale_sengled_value_by_brightness( + (r_raw, g_raw, b_raw), + brightness_factor, + )) else: - self._attr_brightness = int((max_channel_value / 100) * 255) - else: - self._attr_brightness = 0 + self._attr_rgb_color = smart_reverse(*unscale_sengled_value_by_brightness( + (r_raw, g_raw, b_raw), + brightness_factor, + )) self._available = True + await self._save_cache() + except Exception as e: _LOGGER.error(f"Error updating state from status: {e}") self._available = False @@ -430,6 +477,16 @@ def send_udp(): except Exception: return None - def _kelvin_to_device_temp(self, kelvin: int) -> int: + @staticmethod + def _kelvin_to_device_temp(kelvin: int) -> int: device_temp = int(1 + ((kelvin - 2000) / (6500 - 2000)) * 99) - return max(1, min(100, device_temp)) \ No newline at end of file + return max(1, min(100, device_temp)) + + async def _load_cache(self): + self._cache = LightCache(**await self._storage.async_load()) + + async def _save_cache(self): + self._cache.brightness = self._attr_brightness + self._cache.rgb = self._attr_rgb_color + self._cache.color_temp_kelvin = self._attr_color_temp_kelvin + await self._storage.async_save(dataclasses.asdict(self._cache))