"""Light platform for Ninlight bulbs.""" from __future__ import annotations import logging from typing import Any from homeassistant.components.light import ( ATTR_BRIGHTNESS, ATTR_COLOR_TEMP_KELVIN, ATTR_RGB_COLOR, ATTR_TRANSITION, ColorMode, LightEntity, LightEntityFeature, ) from homeassistant.core import HomeAssistant from homeassistant.exceptions import HomeAssistantError from homeassistant.helpers.entity_platform import AddEntitiesCallback from .client import NinlightBadRequest, NinlightError, NinlightState from .const import MAX_TRANSITION_MS from .coordinator import NinlightConfigEntry, NinlightCoordinator from .entity import NinlightEntity _LOGGER = logging.getLogger(__name__) async def async_setup_entry( hass: HomeAssistant, entry: NinlightConfigEntry, async_add_entities: AddEntitiesCallback, ) -> None: """Set up the light for a Ninlight bulb config entry.""" async_add_entities([NinlightLight(entry.runtime_data, entry)]) def _to_ha_brightness(pct: int) -> int: """Bulb brightness is 1..100; HA brightness is 0..255.""" return max(1, min(255, round(pct * 255 / 100))) def _to_bulb_brightness(level: int) -> int: """The bulb rejects brightness 0 and anything above 100 (invalid brightness).""" return max(1, min(100, round(level * 100 / 255))) class NinlightLight(NinlightEntity, LightEntity): """A Klyqa bulb running the Ninlight firmware.""" # No translation_key: with has_entity_name and a single entity per device, HA names the # light after the device, giving light.schreibtisch_links rather than ..._bulb. _attr_name = None def __init__(self, coordinator: NinlightCoordinator, entry: NinlightConfigEntry) -> None: super().__init__(coordinator, entry) self._attr_unique_id = f"{(entry.unique_id or '').lower()}-light" self._attr_supported_features = LightEntityFeature.TRANSITION # Capabilities come from the polled state, not from a one-time __init__ snapshot: an # entry set up while the bulb was powered off starts from defaults and must pick up # the real RGB/CT range as soon as the bulb answers again. @property def available(self) -> bool: # An unreachable (powered-off) bulb is an expected state: the entity goes # unavailable instead of the config entry erroring. return self.coordinator.last_update_success and self.coordinator.data.available @property def supported_color_modes(self) -> set[ColorMode]: state = self.coordinator.data # Both modes for RGB bulbs: the firmware keeps the last color and the last color # temperature and picks between them with `mode`, exactly how HA models a # color_temp + rgb light. return {ColorMode.COLOR_TEMP, ColorMode.RGB} if state.has_rgb else {ColorMode.COLOR_TEMP} @property def min_color_temp_kelvin(self) -> int: return self.coordinator.data.ct_min @property def max_color_temp_kelvin(self) -> int: return self.coordinator.data.ct_max @property def is_on(self) -> bool: return self.coordinator.data.on @property def brightness(self) -> int | None: return _to_ha_brightness(self.coordinator.data.brightness) @property def color_mode(self) -> ColorMode: state = self.coordinator.data if not state.on: # Nothing is lit; reporting a mode would contradict is_on=False. return ColorMode.UNKNOWN if state.has_rgb and state.mode == "color": return ColorMode.RGB return ColorMode.COLOR_TEMP @property def color_temp_kelvin(self) -> int | None: state = self.coordinator.data return state.color_temp if self.color_mode == ColorMode.COLOR_TEMP else None @property def rgb_color(self) -> tuple[int, int, int] | None: state = self.coordinator.data if self.color_mode == ColorMode.RGB: return (state.r, state.g, state.b) return None async def async_turn_on(self, **kwargs: Any) -> None: state = self.coordinator.data payload: dict[str, Any] = {"on": True} if (level := kwargs.get(ATTR_BRIGHTNESS)) is not None: payload["brightness"] = _to_bulb_brightness(int(level)) elif not state.on: # HA expects turn_on without a brightness to restore the previous level. The # firmware already does that (it keeps `brightness` while off), so sending it # would only re-assert a value the bulb may have clamped itself - but a bulb # that has never been on reports brightness 0, which its API rejects. Mirror # the last known value only when the bulb actually has one. if state.brightness < 1: payload["brightness"] = 100 # The firmware rejects color and color_temp in one request with # "color and color_temp are exclusive", so pick one. HA only sends both when a # caller passes them together; color is the more specific intent and wins. kelvin = kwargs.get(ATTR_COLOR_TEMP_KELVIN) rgb = kwargs.get(ATTR_RGB_COLOR) seg: dict[str, Any] | None = None if rgb is not None and state.has_rgb: r, g, b = (int(max(0, min(255, v))) for v in rgb) if state.strip: seg = {"mode": "solid", "color": {"r": r, "g": g, "b": b}} else: payload["color"] = {"r": r, "g": g, "b": b} elif kelvin is not None: ct = max(state.ct_min, min(state.ct_max, int(kelvin))) if state.strip: seg = {"mode": "white", "color_temp": ct} else: payload["color_temp"] = ct if (transition := kwargs.get(ATTR_TRANSITION)) is not None: payload["transition"] = max(0, min(MAX_TRANSITION_MS, int(float(transition) * 1000))) await self._apply(payload, seg) async def async_turn_off(self, **kwargs: Any) -> None: payload: dict[str, Any] = {"on": False} if (transition := kwargs.get(ATTR_TRANSITION)) is not None: payload["transition"] = max(0, min(MAX_TRANSITION_MS, int(float(transition) * 1000))) await self._apply(payload, None) async def _apply(self, payload: dict[str, Any], seg_patch: dict[str, Any] | None) -> None: """Send one command and update immediately. POST /device/state answers with the full new state, so the coordinator does not need a refresh round-trip and the UI settles on the real bulb state at once. On the segment-model firmware the color goes into segment 0 as a separate PATCH (`PUT /device/segments/0`, fields omitted keep their value); the master on and brightness stay in the state POST, because the firmware's fade rides on them. Segment brightness stays untouched so master scaling remains the single knob. """ try: data = await self.coordinator.client.set_state(payload) if seg_patch is not None: segs = self.coordinator.data.segments start = segs[0]["start"] if segs else 0 end = segs[0]["end"] if segs else max(0, int(data.get("pixel_count", 1)) - 1) await self.coordinator.client.patch_segment( 0, {**seg_patch, "start": start, "end": end} ) except NinlightBadRequest as err: # A rejected command is a bug or a firmware mismatch, not a network fault: # surface it to the caller and leave the entity's state alone. raise HomeAssistantError(str(err)) from err except NinlightError as err: self.coordinator.async_set_update_error(err) raise HomeAssistantError(str(err)) from err # Push the device's own reply through the coordinator so every entity of this # device (light, sensors) settles on the real state at once. segments = None if self.coordinator.data.strip: try: segments = (await self.coordinator.client.get_segments()).get("segments") except NinlightError: pass # the next poll recovers it; the state reply above is still right self.coordinator.async_set_updated_data( NinlightState.from_api(data, self.coordinator.info, segments) )