Add LedSC integration

.coveragerc

@@ -700,6 +700,10 @@ omit =
     homeassistant/components/ld2410_ble/sensor.py
     homeassistant/components/led_ble/__init__.py
     homeassistant/components/led_ble/light.py
+    homeassistant/components/ledsc/__init__.py
+    homeassistant/components/ledsc/light.py
+    homeassistant/components/ledsc/config_flow.py
+    homeassistant/components/ledsc/consts.py
     homeassistant/components/lg_netcast/media_player.py
     homeassistant/components/lg_soundbar/__init__.py
     homeassistant/components/lg_soundbar/media_player.py

CODEOWNERS

@@ -768,6 +768,7 @@ build.json @home-assistant/supervisor
 /tests/components/leaone/ @bdraco
 /homeassistant/components/led_ble/ @bdraco
 /tests/components/led_ble/ @bdraco
+/homeassistant/components/ledsc/ @PatrikKr010
 /homeassistant/components/lg_netcast/ @Drafteed @splinter98
 /tests/components/lg_netcast/ @Drafteed @splinter98
 /homeassistant/components/lidarr/ @tkdrob

homeassistant/components/ledsc/__init__.py

@@ -0,0 +1,35 @@
+"""Platform for LedSC integration."""
+
+from __future__ import annotations
+
+import logging
+
+import voluptuous as vol
+
+from homeassistant.components.light import PLATFORM_SCHEMA
+from homeassistant.config_entries import ConfigEntry
+from homeassistant.const import CONF_HOST, CONF_PORT, Platform
+from homeassistant.core import HomeAssistant
+import homeassistant.helpers.config_validation as cv
+
+PLATFORMS: list[str] = [Platform.LIGHT]
+_LOGGER = logging.getLogger(__name__)
+
+# Validation of the user's configuration
+PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend(
+    {
+        vol.Required(CONF_HOST): cv.string,
+        vol.Required(CONF_PORT): cv.port,
+    }
+)
+
+
+async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
+    """Set up LedSC from a config entry."""
+    await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
+    return True
+
+
+async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
+    """Unload a config entry."""
+    return await hass.config_entries.async_unload_platforms(entry, PLATFORMS)

homeassistant/components/ledsc/config_flow.py

@@ -0,0 +1,61 @@
+"""Config flow for LedSC integration."""
+
+from __future__ import annotations
+
+import logging
+from typing import Any
+import voluptuous as vol
+from websc_client import WebSClientAsync as WebSClient
+from websc_client.exceptions import WebSClientError
+
+from homeassistant.config_entries import ConfigFlow, ConfigFlowResult
+from homeassistant.const import CONF_HOST, CONF_PORT
+from homeassistant.core import HomeAssistant
+import homeassistant.helpers.config_validation as cv
+
+from .consts import DEFAULT_PORT, DOMAIN
+
+_LOGGER = logging.getLogger(__name__)
+
+DATA_SCHEMA = vol.Schema(
+    {
+        vol.Required(CONF_HOST): str,
+        vol.Required(CONF_PORT, default=DEFAULT_PORT): cv.port,
+    }
+)
+
+
+async def validate_input(hass: HomeAssistant, data: dict) -> dict[str, Any]:
+    """Validate the user input allows us to connect."""
+    host = data[CONF_HOST]
+    port = data[CONF_PORT]
+
+    client = WebSClient(host, port)
+    await client.connect()
+    await client.disconnect()
+
+    return {"title": f"LedSC server {host}:{port}"}
+
+
+class LedSCConfigFlow(ConfigFlow, domain=DOMAIN):
+    """Handle a config flow for LedSC."""
+
+    VERSION = 1
+
+    async def async_step_user(
+            self, user_input: dict[str, Any] | None = None
+    ) -> ConfigFlowResult:
+        """Handle the initial step."""
+        errors: dict[str, str] = {}
+        if user_input:
+            self._async_abort_entries_match(user_input)
+            try:
+                info = await validate_input(self.hass, user_input)
+            except WebSClientError:
+                errors["base"] = "cannot_connect"
+            else:
+                return self.async_create_entry(title=info["title"], data=user_input)
+
+        return self.async_show_form(
+            step_id="user", data_schema=DATA_SCHEMA, errors=errors
+        )

homeassistant/components/ledsc/consts.py

@@ -0,0 +1,4 @@
+"""Constants for the LedSC integration."""
+
+DOMAIN = "ledsc"
+DEFAULT_PORT = 8443

homeassistant/components/ledsc/light.py

@@ -0,0 +1,151 @@
+"""LedSC light."""
+
+import logging
+from typing import Any
+
+from websc_client import WebSCAsync, WebSClientAsync as WebSClient
+
+from homeassistant.config_entries import ConfigEntry
+from homeassistant.const import CONF_HOST, CONF_PORT
+from homeassistant.helpers.entity_platform import AddEntitiesCallback
+from homeassistant.components.light import ColorMode, LightEntity
+from homeassistant.core import HomeAssistant
+
+_LOGGER = logging.getLogger(__name__)
+
+
+async def async_setup_entry(
+    hass: HomeAssistant, config: ConfigEntry, add_entities: AddEntitiesCallback
+):
+    """Connect to WebSC.
+
+    load the configured devices from WebSC Server and add them to hass.
+    """
+    client = WebSClient(host=config.data[CONF_HOST], port=config.data[CONF_PORT])
+    await client.connect()
+    hass.async_create_background_task(client.observer(), name="ledsc-observer")
+
+    devices: list[LedSCLightEntity] = []
+    for websc in client.devices.values():
+        ledsc = LedSCLightEntity(
+            client_id=f"{config.data[CONF_HOST]}:{config.data[CONF_PORT]}",
+            websc=websc,
+            hass=hass,
+        )
+        websc.set_callback(__generate_callback(ledsc))
+        devices.append(ledsc)
+    add_entities(devices, True)
+
+
+class LedSCLightEntity(LightEntity):
+    """Representation of an LedSC Light."""
+
+    def __init__(
+        self,
+        client_id: str,
+        websc: WebSCAsync,
+        hass: HomeAssistant,
+    ) -> None:
+        """Initialize an LedSC Light."""
+        self._hass: HomeAssistant = hass
+        self._websc: WebSCAsync = websc
+        self._attr_unique_id = f"{client_id}-{websc.name}"
+        _LOGGER.debug(f"LedSC '%s' initialized!", self.name)
+
+    @property
+    def supported_color_modes(self) -> set[ColorMode] | set[str] | None:
+        """List of supported color modes."""
+        return {ColorMode.RGBW}
+
+    @property
+    def color_mode(self) -> ColorMode | str | None:
+        """Return the current color mode (static)."""
+        return ColorMode.RGBW
+
+    @property
+    def available(self) -> bool:
+        """Check if light is available."""
+        return not self._websc.is_lost
+
+    @property
+    def name(self) -> str:
+        """Return the display name of this light."""
+        return self._websc.name
+
+    @property
+    def brightness(self) -> int | None:
+        """Return the brightness of the light."""
+        return max(self._websc.rgbw)
+
+    @brightness.setter
+    def brightness(self, value: int) -> None:
+        """Set brightness of the light."""
+        actual = self.brightness
+        if actual is None or actual == 0:
+            self.hass.async_create_task(
+                self._websc.set_rgbw(red=value, green=value, blue=value, white=value)
+            )
+        else:
+            diff = value - actual
+            ratio = diff / actual
+            self.hass.async_create_task(
+                self._websc.set_rgbw(
+                    red=round(self._websc.red * (1 + ratio)),
+                    green=round(self._websc.green * (1 + ratio)),
+                    blue=round(self._websc.blue * (1 + ratio)),
+                    white=round(self._websc.white * (1 + ratio)),
+                )
+            )
+
+    @property
+    def rgbw_color(self) -> tuple[int, int, int, int] | None:
+        """Get color."""
+        return self._websc.rgbw
+
+    @rgbw_color.setter
+    def rgbw_color(self, value: tuple[int, int, int, int]) -> None:
+        """Set color to WebSC."""
+        self.hass.async_create_task(
+            self._websc.set_rgbw(
+                red=value[0],
+                green=value[1],
+                blue=value[2],
+                white=value[3],
+            )
+        )
+
+    @property
+    def is_on(self) -> bool | None:
+        """Return true if light is on."""
+        return bool(
+            self._websc.red
+            or self._websc.green
+            or self._websc.blue
+            or self._websc.white
+        )
+
+    async def async_turn_on(self, **kwargs: Any) -> None:
+        """Instruct the light to turn on."""
+        if "brightness" in kwargs:
+            self.brightness = kwargs["brightness"]
+        elif "rgbw_color" in kwargs:
+            self.rgbw_color = kwargs["rgbw_color"]
+        elif not self.is_on:
+            await self.switch()
+
+    async def async_turn_off(self, **kwargs: Any) -> None:
+        """Instruct the light to turn off."""
+        if self.is_on:
+            await self.switch()
+
+    async def switch(self) -> None:
+        """Send switch event to WebSC."""
+        await self._websc.do_px_trigger()
+
+
+def __generate_callback(ledsc: LedSCLightEntity):
+    """Generates a callback to respond to a LedSC state change."""
+    async def on_device_change(data: dict[str, int]):
+        await ledsc.async_update_ha_state(force_refresh=True)
+
+    return on_device_change

homeassistant/components/ledsc/manifest.json

@@ -0,0 +1,10 @@
+{
+  "domain": "ledsc",
+  "name": "LedSC",
+  "codeowners": ["@PatrikKr010"],
+  "config_flow": true,
+  "documentation": "https://www.home-assistant.io/integrations/ledsc/",
+  "integration_type": "hub",
+  "iot_class": "local_polling",
+  "requirements": ["websc-client==1.0.2"]
+}

homeassistant/generated/config_flows.py

@@ -294,6 +294,7 @@
         "ld2410_ble",
         "leaone",
         "led_ble",
+        "ledsc",
         "lg_netcast",
         "lg_soundbar",
         "lidarr",

homeassistant/generated/integrations.json

@@ -3193,6 +3193,12 @@
       "config_flow": true,
       "iot_class": "local_polling"
     },
+    "ledsc": {
+      "name": "LedSC",
+      "integration_type": "hub",
+      "config_flow": true,
+      "iot_class": "local_polling"
+    },
     "legrand": {
       "name": "Legrand",
       "integration_type": "virtual",

requirements_all.txt

@@ -2875,6 +2875,9 @@ webmin-xmlrpc==0.0.2
 # homeassistant.components.assist_pipeline
 webrtc-noise-gain==1.2.3
 
+# homeassistant.components.ledsc
+websc-client==1.0.2
+
 # homeassistant.components.whirlpool
 whirlpool-sixth-sense==0.18.8
 

tests/components/ledsc/test_config_flow.py

@@ -0,0 +1,76 @@
+from unittest.mock import patch
+
+from homeassistant import config_entries, data_entry_flow
+from homeassistant.components.ledsc.consts import DOMAIN
+
+USER_INPUT = {"host": "127.0.0.1", "port": 8080}
+IMPORT_CONFIG = {"host": "127.0.0.1", "port": 8080}
+RESULT = {
+    "type": "create_entry",
+    "title": f"LedSC server {USER_INPUT['host']}:{USER_INPUT['port']}",
+    "data": USER_INPUT,
+}
+ENTRY_CONFIG = {"host": "127.0.0.1", "port": 8080}
+
+
+async def test_form(hass):
+    """Test that the form is served with no input."""
+    result = await hass.config_entries.flow.async_init(
+        DOMAIN,
+        context={"source": config_entries.SOURCE_USER},
+    )
+
+    assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
+    assert result["errors"] == {}
+
+    with patch(
+            "homeassistant.components.ledsc.config_flow.validate_input",
+            return_value=RESULT,
+    ), patch(
+        "homeassistant.components.ledsc.async_setup_entry",
+        return_value=True,
+    ):
+        result2 = await hass.config_entries.flow.async_configure(
+            result["flow_id"],
+            USER_INPUT,
+        )
+    assert result2["type"] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY
+    assert result2["title"] == RESULT["title"]
+    assert result2["data"] == USER_INPUT
+
+
+async def test_form_cannot_connect(hass):
+    """Test we handle errors."""
+    result = await hass.config_entries.flow.async_init(
+        DOMAIN,
+        context={"source": config_entries.SOURCE_USER},
+    )
+
+    with patch(
+            "homeassistant.components.ledsc.config_flow.validate_input",
+            side_effect=ConnectionError,
+    ):
+        result2 = await hass.config_entries.flow.async_configure(
+            result["flow_id"],
+            USER_INPUT,
+        )
+
+    assert result2["type"] == "form"
+    assert result2["errors"] == {"base": "cannot_connect"}
+
+
+async def test_form_unique_id_check(hass):
+    """Test we handle unique id check"""
+    await hass.config_entries.flow.async_init(
+        DOMAIN,
+        context={"source": config_entries.SOURCE_USER},
+        data=USER_INPUT,
+    )
+
+    result = await hass.config_entries.flow.async_init(
+        DOMAIN,
+        context={"source": config_entries.SOURCE_USER},
+        data=USER_INPUT,
+    )
+    assert result["type"] == data_entry_flow.RESULT_TYPE_ABORT
+    assert result["reason"] == "already_configured"