mirror of
https://github.com/Gurkengewuerz/casait-has.git
synced 2025-01-18 07:36:26 +01:00
feat: add rgb led controller
This commit is contained in:
parent
5eb4a9b9f3
commit
482331ccfd
@ -4,8 +4,6 @@ from __future__ import annotations
|
||||
import logging
|
||||
from typing import Any
|
||||
|
||||
import voluptuous as vol
|
||||
|
||||
from homeassistant.config_entries import ConfigEntry
|
||||
from homeassistant.const import Platform, CONF_HOST, CONF_PORT, CONF_NAME
|
||||
from homeassistant.core import HomeAssistant, callback
|
||||
@ -16,7 +14,6 @@ from .const import (
|
||||
DATA_COORDINATOR,
|
||||
DATA_CONFIG,
|
||||
DEFAULT_PORT,
|
||||
DEVICE_TYPE_MAPPING,
|
||||
)
|
||||
from .coordinator import SmartHomeDataUpdateCoordinator
|
||||
|
||||
@ -24,6 +21,7 @@ _LOGGER = logging.getLogger(__name__)
|
||||
|
||||
PLATFORMS: list[Platform] = [
|
||||
Platform.SWITCH,
|
||||
Platform.LIGHT,
|
||||
Platform.BUTTON,
|
||||
Platform.COVER,
|
||||
Platform.SENSOR,
|
||||
|
@ -14,12 +14,3 @@ STEP_USER: Final = "user"
|
||||
|
||||
DATA_COORDINATOR: Final = "coordinator"
|
||||
DATA_CONFIG: Final = "config"
|
||||
|
||||
# Device type mapping
|
||||
DEVICE_TYPE_MAPPING = {
|
||||
"switch": "switch",
|
||||
"pushbutton": "button",
|
||||
"blind": "cover",
|
||||
"sensor": "sensor",
|
||||
"binary_sensor": "binary_sensor"
|
||||
}
|
@ -5,6 +5,7 @@ import logging
|
||||
|
||||
from homeassistant.core import callback
|
||||
from homeassistant.helpers.update_coordinator import CoordinatorEntity
|
||||
from homeassistant.helpers.device_registry import CONNECTION_NETWORK_MAC, DeviceInfo
|
||||
|
||||
from .const import DOMAIN
|
||||
from .coordinator import SmartHomeDataUpdateCoordinator
|
||||
@ -16,6 +17,9 @@ _LOGGER = logging.getLogger(__name__)
|
||||
class SmartHomeEntity(CoordinatorEntity):
|
||||
"""Base class for Smart Home entities."""
|
||||
|
||||
_attr_has_entity_name = True
|
||||
_attr_name = None
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
coordinator: SmartHomeDataUpdateCoordinator,
|
||||
@ -24,13 +28,7 @@ class SmartHomeEntity(CoordinatorEntity):
|
||||
"""Initialize the entity."""
|
||||
super().__init__(coordinator)
|
||||
self._device_id = device_id
|
||||
self._attr_device_info = {
|
||||
"identifiers": {(DOMAIN, device_id)},
|
||||
"name": self.device_data["name"],
|
||||
"manufacturer": "casaIT",
|
||||
"model": self.device_data["device_type"],
|
||||
"via_device": (DOMAIN, coordinator.entry_id),
|
||||
}
|
||||
_LOGGER.debug("Initialized entity %s", self.device_data["name"])
|
||||
|
||||
@property
|
||||
def device_data(self) -> dict:
|
||||
@ -47,17 +45,23 @@ class SmartHomeEntity(CoordinatorEntity):
|
||||
"""Return if entity is available."""
|
||||
return self.coordinator.last_update_success and self._device_id in self.coordinator._devices
|
||||
|
||||
@property
|
||||
def name(self):
|
||||
"""Name of the sensor"""
|
||||
return self.device_data["name"]
|
||||
|
||||
@property
|
||||
def unique_id(self):
|
||||
"""ID of the sensor"""
|
||||
return f'casait_{self.device_data["id"]}_{self.device_data["uuid"]}'
|
||||
|
||||
@callback
|
||||
def _handle_coordinator_update(self) -> None:
|
||||
"""Handle updated data from the coordinator."""
|
||||
self.async_write_ha_state()
|
||||
|
||||
@property
|
||||
def unique_id(self) -> str:
|
||||
"""Return a unique ID to use for this entity."""
|
||||
return self.device_data["uuid"]
|
||||
|
||||
@property
|
||||
def device_info(self) -> DeviceInfo:
|
||||
"""Return device information about this WLED device."""
|
||||
return DeviceInfo(
|
||||
identifiers={(DOMAIN, self.unique_id)},
|
||||
name=self.device_data["name"],
|
||||
manufacturer="casaIT",
|
||||
model=self.device_data["device_type"].replace("_", " ").title(),
|
||||
via_device=(DOMAIN, self.coordinator.entry_id),
|
||||
)
|
||||
|
176
custom_components/smart_home/light.py
Normal file
176
custom_components/smart_home/light.py
Normal file
@ -0,0 +1,176 @@
|
||||
"""Support for Smart Home RGB lights."""
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
import async_timeout
|
||||
import aiohttp
|
||||
from homeassistant.components.light import (
|
||||
ATTR_BRIGHTNESS,
|
||||
ATTR_EFFECT,
|
||||
ATTR_RGB_COLOR,
|
||||
ColorMode,
|
||||
LightEntity,
|
||||
LightEntityFeature,
|
||||
)
|
||||
from homeassistant.config_entries import ConfigEntry
|
||||
from homeassistant.core import HomeAssistant
|
||||
from homeassistant.helpers.entity_platform import AddEntitiesCallback
|
||||
from homeassistant.exceptions import HomeAssistantError
|
||||
|
||||
from .const import DOMAIN, DATA_COORDINATOR
|
||||
from .coordinator import SmartHomeDataUpdateCoordinator
|
||||
from .entity import SmartHomeEntity
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
|
||||
|
||||
async def async_fetch_effects(api_url: str) -> dict:
|
||||
"""Fetch available effects from API."""
|
||||
try:
|
||||
async with aiohttp.ClientSession() as session:
|
||||
async with async_timeout.timeout(10):
|
||||
async with session.get(f"{api_url}/api/effects") as response:
|
||||
if response.status == 200:
|
||||
return await response.json()
|
||||
else:
|
||||
raise HomeAssistantError(f"Failed to fetch effects: {response.status}")
|
||||
except Exception as e:
|
||||
raise HomeAssistantError(f"Error fetching effects: {e}")
|
||||
return {}
|
||||
|
||||
|
||||
async def async_setup_entry(
|
||||
hass: HomeAssistant,
|
||||
entry: ConfigEntry,
|
||||
async_add_entities: AddEntitiesCallback,
|
||||
) -> None:
|
||||
"""Set up Smart Home RGB lights."""
|
||||
coordinator: SmartHomeDataUpdateCoordinator = hass.data[DOMAIN][entry.entry_id][DATA_COORDINATOR]
|
||||
|
||||
try:
|
||||
# Fetch available effects from API
|
||||
effect_map = await async_fetch_effects(coordinator.api_url)
|
||||
_LOGGER.debug("Fetched effects: %s", effect_map)
|
||||
|
||||
entities = []
|
||||
for device_id, device in coordinator._devices.items():
|
||||
if device["device_type"] == "rgb_led":
|
||||
entities.append(SmartHomeLight(coordinator, device_id, effect_map))
|
||||
|
||||
async_add_entities(entities)
|
||||
except HomeAssistantError as e:
|
||||
_LOGGER.error("Failed to set up RGB lights: %s", e)
|
||||
|
||||
|
||||
class SmartHomeLight(SmartHomeEntity, LightEntity):
|
||||
"""Representation of a Smart Home RGB light."""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
coordinator: SmartHomeDataUpdateCoordinator,
|
||||
device_id: str,
|
||||
effect_map: dict[str, str],
|
||||
) -> None:
|
||||
"""Initialize the light."""
|
||||
super().__init__(coordinator, device_id)
|
||||
self._effect_map = effect_map
|
||||
self._attr_supported_features |= LightEntityFeature.EFFECT
|
||||
|
||||
# Set up supported features
|
||||
self._attr_supported_color_modes = {ColorMode.RGB}
|
||||
self._attr_color_mode = ColorMode.RGB
|
||||
|
||||
# No color temperature support
|
||||
self._attr_min_mireds = 0
|
||||
self._attr_max_mireds = 0
|
||||
|
||||
_LOGGER.debug("Initializing RGB light: %s id %s", device_id, self.unique_id)
|
||||
|
||||
@property
|
||||
def is_on(self) -> bool | None:
|
||||
"""Return true if light is on."""
|
||||
if not self.device_state:
|
||||
return None
|
||||
return self.device_state.get("state", False)
|
||||
|
||||
@property
|
||||
def brightness(self) -> int | None:
|
||||
"""Return the brightness of this light between 0..255."""
|
||||
if not self.device_state:
|
||||
return None
|
||||
brightness = self.device_state.get("brightness", 0)
|
||||
return int(brightness)
|
||||
|
||||
@property
|
||||
def rgb_color(self) -> tuple[int, int, int] | None:
|
||||
"""Return the rgb color value [int, int, int]."""
|
||||
if not self.device_state or "colors" not in self.device_state:
|
||||
return None
|
||||
# Use first color from the array
|
||||
if not self.device_state["colors"]:
|
||||
return None
|
||||
color_hex = self.device_state["colors"][0]
|
||||
# Convert hex string to RGB tuple, stripping any leading '0x'
|
||||
color_hex = color_hex.replace('0x', '')
|
||||
return (
|
||||
int(color_hex[0:2], 16),
|
||||
int(color_hex[2:4], 16),
|
||||
int(color_hex[4:6], 16),
|
||||
)
|
||||
|
||||
@property
|
||||
def effect_list(self) -> list[str] | None:
|
||||
"""Return the list of supported effects."""
|
||||
return list(self._effect_map.values())
|
||||
|
||||
@property
|
||||
def effect(self) -> str | None:
|
||||
"""Return the current effect."""
|
||||
if not self.device_state:
|
||||
return None
|
||||
animation = self.device_state.get("animation")
|
||||
# Handle mapping from API animation name to display name
|
||||
if animation in self._effect_map:
|
||||
return self._effect_map[animation]
|
||||
_LOGGER.debug("Unknown animation mode: %s", animation)
|
||||
return None
|
||||
|
||||
async def async_turn_on(self, **kwargs: Any) -> None:
|
||||
"""Turn the light on."""
|
||||
data = {
|
||||
"state": True
|
||||
}
|
||||
_LOGGER.debug("Data given in async_turn_on: %s", kwargs)
|
||||
if ATTR_BRIGHTNESS in kwargs:
|
||||
brightness = kwargs[ATTR_BRIGHTNESS]
|
||||
data["brightness"] = int(brightness)
|
||||
|
||||
if ATTR_RGB_COLOR in kwargs:
|
||||
r, g, b = kwargs[ATTR_RGB_COLOR]
|
||||
# Convert RGB values to hex string
|
||||
color_hex = f"{r:02x}{g:02x}{b:02x}"
|
||||
# Always maintain 5 colors array, set first color and pad with black
|
||||
data["colors"] = [color_hex] + ['000000'] * 4
|
||||
|
||||
if ATTR_EFFECT in kwargs:
|
||||
# Convert HA effect name back to API animation mode
|
||||
effect_name = kwargs[ATTR_EFFECT]
|
||||
if effect_name in self._effect_map:
|
||||
data["animation"] = self._effect_map[effect_name]
|
||||
else:
|
||||
_LOGGER.warning("Unknown effect name: %s. Valid effects are: %s", effect_name, list(self._effect_map.keys()))
|
||||
|
||||
_LOGGER.debug("Sending data to API: %s", data)
|
||||
async with aiohttp.ClientSession() as session:
|
||||
url = f"{self.coordinator.api_url}/api/devices/{self._device_id}/state"
|
||||
async with session.put(url, json=data) as response:
|
||||
if response.status != 200:
|
||||
_LOGGER.error("Failed to turn on light: %s", response.status)
|
||||
|
||||
async def async_turn_off(self, **kwargs: Any) -> None:
|
||||
"""Turn the light off."""
|
||||
async with aiohttp.ClientSession() as session:
|
||||
url = f"{self.coordinator.api_url}/api/devices/{self._device_id}/state"
|
||||
async with session.put(url, json={"state": False}) as response:
|
||||
if response.status != 200:
|
||||
_LOGGER.error("Failed to turn off light: %s", response.status)
|
@ -77,7 +77,6 @@ class SmartHomeLightSensor(SmartHomeEntity, SensorEntity):
|
||||
@property
|
||||
def native_value(self) -> float | None:
|
||||
"""Return the sensor value."""
|
||||
_LOGGER.debug(f"Light sensor value: {self.device_state}")
|
||||
return self.device_state.get("value")
|
||||
|
||||
class SmartHomeGenericSensor(SmartHomeEntity, SensorEntity):
|
||||
|
Loading…
Reference in New Issue
Block a user