qconnex-listen/ (OTA-Falle: AP + DHCP + Range-Server, Onboarding-Client) und ha-integration/ (Home-Assistant-custom_component ninlight) waren seit der Erfindung nur auf diesem Rechner. Bilder, Request-Log, hostapd-Log und PIDs ueber .gitignore draussen - die Images sind generierbar, sobald die Signierschluessel erreichbar sind. otaserve.py bildet jetzt auch die Montana-Black-Produkt-ID auf ledshield_rgb ab.
183 lines
7.6 KiB
Python
183 lines
7.6 KiB
Python
"""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)
|
|
state = coordinator.data
|
|
self._attr_unique_id = f"{(entry.unique_id or '').lower()}-light"
|
|
# Both modes are advertised for RGB bulbs: the firmware keeps the last color and
|
|
# the last color temperature and picks between them with `mode`, which is exactly
|
|
# how HA models a color_temp + rgb light.
|
|
self._attr_supported_color_modes = {ColorMode.COLOR_TEMP, ColorMode.RGB} if state.has_rgb else {
|
|
ColorMode.COLOR_TEMP
|
|
}
|
|
self._attr_supported_features = LightEntityFeature.TRANSITION
|
|
self._attr_min_color_temp_kelvin = state.ct_min
|
|
self._attr_max_color_temp_kelvin = state.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)
|
|
)
|