Add effect service to WLED integration (#33026)

* Add effect service to WLED integration

* Inline service schema

homeassistant/components/wled/const.py

@@ -17,6 +17,7 @@ ATTR_ON = "on"
 ATTR_PALETTE = "palette"
 ATTR_PLAYLIST = "playlist"
 ATTR_PRESET = "preset"
+ATTR_REVERSE = "reverse"
 ATTR_SEGMENT_ID = "segment_id"
 ATTR_SOFTWARE_VERSION = "sw_version"
 ATTR_SPEED = "speed"
@@ -26,3 +27,6 @@ ATTR_UDP_PORT = "udp_port"
 # Units of measurement
 CURRENT_MA = "mA"
 SIGNAL_DBM = "dBm"
+
+# Services
+SERVICE_EFFECT = "effect"

homeassistant/components/wled/light.py

@@ -1,6 +1,8 @@
 """Support for LED lights."""
 import logging
-from typing import Any, Callable, Dict, List, Optional, Tuple
+from typing import Any, Callable, Dict, List, Optional, Tuple, Union
+
+import voluptuous as vol
 
 from homeassistant.components.light import (
     ATTR_BRIGHTNESS,
@@ -18,6 +20,7 @@ from homeassistant.components.light import (
     Light,
 )
 from homeassistant.config_entries import ConfigEntry
+from homeassistant.helpers import config_validation as cv, entity_platform
 from homeassistant.helpers.entity import Entity
 from homeassistant.helpers.typing import HomeAssistantType
 import homeassistant.util.color as color_util
@@ -30,9 +33,11 @@ from .const import (
     ATTR_PALETTE,
     ATTR_PLAYLIST,
     ATTR_PRESET,
+    ATTR_REVERSE,
     ATTR_SEGMENT_ID,
     ATTR_SPEED,
     DOMAIN,
+    SERVICE_EFFECT,
 )
 
 _LOGGER = logging.getLogger(__name__)
@@ -48,6 +53,23 @@ async def async_setup_entry(
     """Set up WLED light based on a config entry."""
     coordinator: WLEDDataUpdateCoordinator = hass.data[DOMAIN][entry.entry_id]
 
+    platform = entity_platform.current_platform.get()
+
+    platform.async_register_entity_service(
+        SERVICE_EFFECT,
+        {
+            vol.Optional(ATTR_EFFECT): vol.Any(cv.positive_int, cv.string),
+            vol.Optional(ATTR_INTENSITY): vol.All(
+                vol.Coerce(int), vol.Range(min=0, max=255)
+            ),
+            vol.Optional(ATTR_REVERSE): cv.boolean,
+            vol.Optional(ATTR_SPEED): vol.All(
+                vol.Coerce(int), vol.Range(min=0, max=255)
+            ),
+        },
+        "async_effect",
+    )
+
     lights = [
         WLEDLight(entry.entry_id, coordinator, light.segment_id)
         for light in coordinator.data.state.segments
@@ -94,16 +116,14 @@ class WLEDLight(Light, WLEDDeviceEntity):
         if preset == -1:
             preset = None
 
+        segment = self.coordinator.data.state.segments[self._segment]
         return {
-            ATTR_INTENSITY: self.coordinator.data.state.segments[
+            ATTR_INTENSITY: segment.intensity,
-                self._segment
+            ATTR_PALETTE: segment.palette.name,
-            ].intensity,
-            ATTR_PALETTE: self.coordinator.data.state.segments[
-                self._segment
-            ].palette.name,
             ATTR_PLAYLIST: playlist,
             ATTR_PRESET: preset,
-            ATTR_SPEED: self.coordinator.data.state.segments[self._segment].speed,
+            ATTR_REVERSE: segment.reverse,
+            ATTR_SPEED: segment.speed,
         }
 
     @property
@@ -214,3 +234,28 @@ class WLEDLight(Light, WLEDDeviceEntity):
                 data[ATTR_COLOR_PRIMARY] += (self.white_value,)
 
         await self.coordinator.wled.light(**data)
+
+    @wled_exception_handler
+    async def async_effect(
+        self,
+        effect: Optional[Union[int, str]] = None,
+        intensity: Optional[int] = None,
+        reverse: Optional[bool] = None,
+        speed: Optional[int] = None,
+    ) -> None:
+        """Set the effect of a WLED light."""
+        data = {ATTR_SEGMENT_ID: self._segment}
+
+        if effect is not None:
+            data[ATTR_EFFECT] = effect
+
+        if intensity is not None:
+            data[ATTR_INTENSITY] = intensity
+
+        if reverse is not None:
+            data[ATTR_REVERSE] = reverse
+
+        if speed is not None:
+            data[ATTR_SPEED] = speed
+
+        await self.coordinator.wled.light(**data)

homeassistant/components/wled/services.yaml

@@ -0,0 +1,18 @@
+effect:
+  description: Controls the effect settings of WLED
+  fields:
+    entity_id:
+      description: Name of the WLED light entity.
+      example: "light.wled"
+    effect:
+      description: Name or ID of the WLED light effect.
+      example: "Rainbow"
+    intensity:
+      description: Intensity of the effect
+      example: 100
+    speed:
+      description: Speed of the effect. Number between 0 (slow) and 255 (fast).
+      example: 150
+    reverse:
+      description: Reverse the effect. Either true to reverse or false otherwise.
+      example: false

tests/components/wled/test_light.py

@@ -17,7 +17,10 @@ from homeassistant.components.wled.const import (
     ATTR_PALETTE,
     ATTR_PLAYLIST,
     ATTR_PRESET,
+    ATTR_REVERSE,
     ATTR_SPEED,
+    DOMAIN,
+    SERVICE_EFFECT,
 )
 from homeassistant.const import (
     ATTR_ENTITY_ID,
@@ -52,6 +55,7 @@ async def test_rgb_light_state(
     assert state.attributes.get(ATTR_PALETTE) == "Default"
     assert state.attributes.get(ATTR_PLAYLIST) is None
     assert state.attributes.get(ATTR_PRESET) is None
+    assert state.attributes.get(ATTR_REVERSE) is False
     assert state.attributes.get(ATTR_SPEED) == 32
     assert state.state == STATE_ON
 
@@ -70,6 +74,7 @@ async def test_rgb_light_state(
     assert state.attributes.get(ATTR_PALETTE) == "Random Cycle"
     assert state.attributes.get(ATTR_PLAYLIST) is None
     assert state.attributes.get(ATTR_PRESET) is None
+    assert state.attributes.get(ATTR_REVERSE) is False
     assert state.attributes.get(ATTR_SPEED) == 16
     assert state.state == STATE_ON
 
@@ -223,3 +228,129 @@ async def test_rgbw_light(
         light_mock.assert_called_once_with(
             color_primary=(0, 0, 0, 100), on=True, segment_id=0,
         )
+
+
+async def test_effect_service(
+    hass: HomeAssistant, aioclient_mock: AiohttpClientMocker
+) -> None:
+    """Test the effect service of a WLED light."""
+    await init_integration(hass, aioclient_mock)
+
+    with patch("wled.WLED.light") as light_mock:
+        await hass.services.async_call(
+            DOMAIN,
+            SERVICE_EFFECT,
+            {
+                ATTR_EFFECT: "Rainbow",
+                ATTR_ENTITY_ID: "light.wled_rgb_light",
+                ATTR_INTENSITY: 200,
+                ATTR_REVERSE: True,
+                ATTR_SPEED: 100,
+            },
+            blocking=True,
+        )
+        await hass.async_block_till_done()
+        light_mock.assert_called_once_with(
+            effect="Rainbow", intensity=200, reverse=True, segment_id=0, speed=100,
+        )
+
+    with patch("wled.WLED.light") as light_mock:
+        await hass.services.async_call(
+            DOMAIN,
+            SERVICE_EFFECT,
+            {ATTR_ENTITY_ID: "light.wled_rgb_light", ATTR_EFFECT: 9},
+            blocking=True,
+        )
+        await hass.async_block_till_done()
+        light_mock.assert_called_once_with(
+            segment_id=0, effect=9,
+        )
+
+    with patch("wled.WLED.light") as light_mock:
+        await hass.services.async_call(
+            DOMAIN,
+            SERVICE_EFFECT,
+            {
+                ATTR_ENTITY_ID: "light.wled_rgb_light",
+                ATTR_INTENSITY: 200,
+                ATTR_REVERSE: True,
+                ATTR_SPEED: 100,
+            },
+            blocking=True,
+        )
+        await hass.async_block_till_done()
+        light_mock.assert_called_once_with(
+            intensity=200, reverse=True, segment_id=0, speed=100,
+        )
+
+    with patch("wled.WLED.light") as light_mock:
+        await hass.services.async_call(
+            DOMAIN,
+            SERVICE_EFFECT,
+            {
+                ATTR_EFFECT: "Rainbow",
+                ATTR_ENTITY_ID: "light.wled_rgb_light",
+                ATTR_REVERSE: True,
+                ATTR_SPEED: 100,
+            },
+            blocking=True,
+        )
+        await hass.async_block_till_done()
+        light_mock.assert_called_once_with(
+            effect="Rainbow", reverse=True, segment_id=0, speed=100,
+        )
+
+    with patch("wled.WLED.light") as light_mock:
+        await hass.services.async_call(
+            DOMAIN,
+            SERVICE_EFFECT,
+            {
+                ATTR_EFFECT: "Rainbow",
+                ATTR_ENTITY_ID: "light.wled_rgb_light",
+                ATTR_INTENSITY: 200,
+                ATTR_SPEED: 100,
+            },
+            blocking=True,
+        )
+        await hass.async_block_till_done()
+        light_mock.assert_called_once_with(
+            effect="Rainbow", intensity=200, segment_id=0, speed=100,
+        )
+
+    with patch("wled.WLED.light") as light_mock:
+        await hass.services.async_call(
+            DOMAIN,
+            SERVICE_EFFECT,
+            {
+                ATTR_EFFECT: "Rainbow",
+                ATTR_ENTITY_ID: "light.wled_rgb_light",
+                ATTR_INTENSITY: 200,
+                ATTR_REVERSE: True,
+            },
+            blocking=True,
+        )
+        await hass.async_block_till_done()
+        light_mock.assert_called_once_with(
+            effect="Rainbow", intensity=200, reverse=True, segment_id=0,
+        )
+
+
+async def test_effect_service_error(
+    hass: HomeAssistant, aioclient_mock: AiohttpClientMocker, caplog
+) -> None:
+    """Test error handling of the WLED effect service."""
+    aioclient_mock.post("http://example.local:80/json/state", text="", status=400)
+    await init_integration(hass, aioclient_mock)
+
+    with patch("homeassistant.components.wled.WLEDDataUpdateCoordinator.async_refresh"):
+        await hass.services.async_call(
+            DOMAIN,
+            SERVICE_EFFECT,
+            {ATTR_ENTITY_ID: "light.wled_rgb_light", ATTR_EFFECT: 9},
+            blocking=True,
+        )
+        await hass.async_block_till_done()
+
+        state = hass.states.get("light.wled_rgb_light")
+        assert state.state == STATE_ON
+        assert "Invalid response from API" in caplog.text