Add notification reactive color strip source with webhook trigger

New source_type "notification" fires one-shot visual effects (flash, pulse, sweep)
triggered via POST webhook. Designed as a composite layer for overlay on persistent
sources. Includes app color mapping, whitelist/blacklist filtering, and auto-sizing.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

server/src/wled_controller/api/routes/color_strip_sources.py

@@ -19,6 +19,7 @@ from wled_controller.api.schemas.color_strip_sources import (
     ColorStripSourceResponse,
     ColorStripSourceUpdate,
     CSSCalibrationTestRequest,
+    NotifyRequest,
 )
 from wled_controller.api.schemas.devices import (
     Calibration as CalibrationSchema,
@@ -30,7 +31,7 @@ from wled_controller.core.capture.calibration import (
 )
 from wled_controller.core.capture.screen_capture import get_available_displays
 from wled_controller.core.processing.processor_manager import ProcessorManager
-from wled_controller.storage.color_strip_source import ApiInputColorStripSource, PictureColorStripSource
+from wled_controller.storage.color_strip_source import ApiInputColorStripSource, NotificationColorStripSource, PictureColorStripSource
 from wled_controller.storage.color_strip_store import ColorStripStore
 from wled_controller.storage.picture_source import ProcessedPictureSource, ScreenCapturePictureSource
 from wled_controller.storage.picture_source_store import PictureSourceStore
@@ -91,6 +92,13 @@ def _css_to_response(source, overlay_active: bool = False) -> ColorStripSourceRe
         color_peak=getattr(source, "color_peak", None),
         fallback_color=getattr(source, "fallback_color", None),
         timeout=getattr(source, "timeout", None),
+        notification_effect=getattr(source, "notification_effect", None),
+        duration_ms=getattr(source, "duration_ms", None),
+        default_color=getattr(source, "default_color", None),
+        app_colors=getattr(source, "app_colors", None),
+        app_filter_mode=getattr(source, "app_filter_mode", None),
+        app_filter_list=getattr(source, "app_filter_list", None),
+        os_listener=getattr(source, "os_listener", None),
         overlay_active=overlay_active,
         created_at=source.created_at,
         updated_at=source.updated_at,
@@ -175,6 +183,13 @@ async def create_color_strip_source(
             fallback_color=data.fallback_color,
             timeout=data.timeout,
             clock_id=data.clock_id,
+            notification_effect=data.notification_effect,
+            duration_ms=data.duration_ms,
+            default_color=data.default_color,
+            app_colors=data.app_colors,
+            app_filter_mode=data.app_filter_mode,
+            app_filter_list=data.app_filter_list,
+            os_listener=data.os_listener,
         )
         return _css_to_response(source)
 
@@ -251,6 +266,13 @@ async def update_color_strip_source(
             fallback_color=data.fallback_color,
             timeout=data.timeout,
             clock_id=data.clock_id,
+            notification_effect=data.notification_effect,
+            duration_ms=data.duration_ms,
+            default_color=data.default_color,
+            app_colors=data.app_colors,
+            app_filter_mode=data.app_filter_mode,
+            app_filter_list=data.app_filter_list,
+            os_listener=data.os_listener,
         )
 
         # Hot-reload running stream (no restart needed for in-place param changes)
@@ -489,6 +511,45 @@ async def push_colors(
     }
 
 
+@router.post("/api/v1/color-strip-sources/{source_id}/notify", tags=["Color Strip Sources"])
+async def notify_source(
+    source_id: str,
+    _auth: AuthRequired,
+    body: NotifyRequest = None,
+    store: ColorStripStore = Depends(get_color_strip_store),
+    manager: ProcessorManager = Depends(get_processor_manager),
+):
+    """Trigger a notification on a notification color strip source.
+
+    Fires a one-shot visual effect (flash, pulse, sweep) on all running
+    stream instances for this source. Optionally specify an app name for
+    color lookup or a hex color override.
+    """
+    try:
+        source = store.get_source(source_id)
+    except ValueError as e:
+        raise HTTPException(status_code=404, detail=str(e))
+
+    if not isinstance(source, NotificationColorStripSource):
+        raise HTTPException(status_code=400, detail="Source is not a notification type")
+
+    app_name = body.app if body else None
+    color_override = body.color if body else None
+
+    streams = manager._color_strip_stream_manager.get_streams_by_source_id(source_id)
+    accepted = 0
+    for stream in streams:
+        if hasattr(stream, "fire"):
+            if stream.fire(app_name=app_name, color_override=color_override):
+                accepted += 1
+
+    return {
+        "status": "ok",
+        "streams_notified": accepted,
+        "filtered": len(streams) - accepted,
+    }
+
+
 @router.websocket("/api/v1/color-strip-sources/{source_id}/ws")
 async def css_api_input_ws(
     websocket: WebSocket,

server/src/wled_controller/api/schemas/color_strip_sources.py

@@ -1,7 +1,7 @@
 """Color strip source schemas (CRUD)."""
 
 from datetime import datetime
-from typing import List, Literal, Optional
+from typing import Dict, List, Literal, Optional
 
 from pydantic import BaseModel, Field
 
@@ -49,7 +49,7 @@ class ColorStripSourceCreate(BaseModel):
     """Request to create a color strip source."""
 
     name: str = Field(description="Source name", min_length=1, max_length=100)
-    source_type: Literal["picture", "static", "gradient", "color_cycle", "effect", "composite", "mapped", "audio", "api_input"] = Field(default="picture", description="Source type")
+    source_type: Literal["picture", "static", "gradient", "color_cycle", "effect", "composite", "mapped", "audio", "api_input", "notification"] = Field(default="picture", description="Source type")
     # picture-type fields
     picture_source_id: str = Field(default="", description="Picture source ID (for picture type)")
     brightness: float = Field(default=1.0, description="Brightness multiplier (0.0-2.0)", ge=0.0, le=2.0)
@@ -87,6 +87,14 @@ class ColorStripSourceCreate(BaseModel):
     # api_input-type fields
     fallback_color: Optional[List[int]] = Field(None, description="Fallback RGB color [R,G,B] when no data received (api_input type)")
     timeout: Optional[float] = Field(None, description="Seconds before reverting to fallback (api_input type)", ge=0.0, le=300.0)
+    # notification-type fields
+    notification_effect: Optional[str] = Field(None, description="Notification effect: flash|pulse|sweep")
+    duration_ms: Optional[int] = Field(None, ge=100, le=10000, description="Effect duration in milliseconds")
+    default_color: Optional[str] = Field(None, description="Default hex color (#RRGGBB) for notifications")
+    app_colors: Optional[Dict[str, str]] = Field(None, description="Map of app name to hex color (#RRGGBB)")
+    app_filter_mode: Optional[str] = Field(None, description="App filter mode: off|whitelist|blacklist")
+    app_filter_list: Optional[List[str]] = Field(None, description="App names for filter")
+    os_listener: Optional[bool] = Field(None, description="Whether to listen for OS notifications")
     # sync clock
     clock_id: Optional[str] = Field(None, description="Optional sync clock ID for synchronized animation")
 
@@ -132,6 +140,14 @@ class ColorStripSourceUpdate(BaseModel):
     # api_input-type fields
     fallback_color: Optional[List[int]] = Field(None, description="Fallback RGB color [R,G,B] (api_input type)")
     timeout: Optional[float] = Field(None, description="Timeout before fallback (api_input type)", ge=0.0, le=300.0)
+    # notification-type fields
+    notification_effect: Optional[str] = Field(None, description="Notification effect: flash|pulse|sweep")
+    duration_ms: Optional[int] = Field(None, ge=100, le=10000, description="Effect duration in milliseconds")
+    default_color: Optional[str] = Field(None, description="Default hex color (#RRGGBB)")
+    app_colors: Optional[Dict[str, str]] = Field(None, description="Map of app name to hex color")
+    app_filter_mode: Optional[str] = Field(None, description="App filter mode: off|whitelist|blacklist")
+    app_filter_list: Optional[List[str]] = Field(None, description="App names for filter")
+    os_listener: Optional[bool] = Field(None, description="Whether to listen for OS notifications")
     # sync clock
     clock_id: Optional[str] = Field(None, description="Optional sync clock ID for synchronized animation")
 
@@ -179,6 +195,14 @@ class ColorStripSourceResponse(BaseModel):
     # api_input-type fields
     fallback_color: Optional[List[int]] = Field(None, description="Fallback RGB color [R,G,B] (api_input type)")
     timeout: Optional[float] = Field(None, description="Timeout before fallback (api_input type)")
+    # notification-type fields
+    notification_effect: Optional[str] = Field(None, description="Notification effect: flash|pulse|sweep")
+    duration_ms: Optional[int] = Field(None, description="Effect duration in milliseconds")
+    default_color: Optional[str] = Field(None, description="Default hex color (#RRGGBB)")
+    app_colors: Optional[Dict[str, str]] = Field(None, description="Map of app name to hex color")
+    app_filter_mode: Optional[str] = Field(None, description="App filter mode: off|whitelist|blacklist")
+    app_filter_list: Optional[List[str]] = Field(None, description="App names for filter")
+    os_listener: Optional[bool] = Field(None, description="Whether to listen for OS notifications")
     # sync clock
     clock_id: Optional[str] = Field(None, description="Optional sync clock ID for synchronized animation")
     overlay_active: bool = Field(False, description="Whether the screen overlay is currently active")
@@ -199,6 +223,13 @@ class ColorPushRequest(BaseModel):
     colors: List[List[int]] = Field(description="LED color array [[R,G,B], ...] (0-255 each)")
 
 
+class NotifyRequest(BaseModel):
+    """Request to trigger a notification on a notification color strip source."""
+
+    app: Optional[str] = Field(None, description="App name for color lookup")
+    color: Optional[str] = Field(None, description="Hex color override (#RRGGBB)")
+
+
 class CSSCalibrationTestRequest(BaseModel):
     """Request to run a calibration test for a color strip source on a specific device."""