Add WebSocket device type, capability-driven settings, hide filter on collapse

- New WS device type: broadcaster singleton + LEDClient that sends binary
  frames to connected WebSocket clients during processing
- FastAPI WS endpoint at /api/v1/devices/{device_id}/ws with token auth
- Frontend: add/edit WS devices, connection URL with copy button in settings
- Add health_check and auto_restore capabilities to WLED and Serial providers;
  hide health interval and auto-restore toggle for devices without them
- Skip health check loop for virtual devices (Mock, MQTT, WS) — set always-online
- Copy buttons and labels for API CSS push endpoints (REST POST / WebSocket)
- Hide mock:// and ws:// URLs in target device dropdown
- Hide filter textbox when card section is collapsed (cs-collapsed CSS class)

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
2026-02-28 20:55:09 +03:00
parent 175a2c6c10
commit fa81d6a608
21 changed files with 375 additions and 16 deletions

View File

@@ -285,5 +285,8 @@ def _register_builtin_providers():
from wled_controller.core.devices.mqtt_provider import MQTTDeviceProvider
register_provider(MQTTDeviceProvider())
from wled_controller.core.devices.ws_provider import WSDeviceProvider
register_provider(WSDeviceProvider())
_register_builtin_providers()

View File

@@ -28,7 +28,9 @@ class SerialDeviceProvider(LEDDeviceProvider):
# manual_led_count: user must specify LED count (can't auto-detect)
# power_control: can blank LEDs by sending all-black pixels
# brightness_control: software brightness (multiplies pixel values before sending)
return {"manual_led_count", "power_control", "brightness_control"}
# health_check: serial port availability probe
# auto_restore: blank LEDs when targets stop
return {"manual_led_count", "power_control", "brightness_control", "health_check", "auto_restore"}
async def check_health(self, url: str, http_client, prev_health=None) -> DeviceHealth:
# Generic serial port health check — enumerate COM ports

View File

@@ -49,7 +49,7 @@ class WLEDDeviceProvider(LEDDeviceProvider):
@property
def capabilities(self) -> set:
return {"brightness_control", "power_control", "standby_required", "static_color"}
return {"brightness_control", "power_control", "standby_required", "static_color", "health_check", "auto_restore"}
def create_client(self, url: str, **kwargs) -> LEDClient:
from wled_controller.core.devices.wled_client import WLEDClient

View File

@@ -0,0 +1,130 @@
"""WebSocket LED client — broadcasts pixel data to connected WebSocket clients."""
import asyncio
from datetime import datetime
from typing import Dict, List, Optional, Tuple, Union
import numpy as np
from wled_controller.core.devices.led_client import DeviceHealth, LEDClient
from wled_controller.utils import get_logger
logger = get_logger(__name__)
class WSDeviceBroadcaster:
"""Global registry of WebSocket clients subscribed to WS device streams.
Each WS device (identified by device_id) can have zero or more connected
WebSocket clients. The WSLEDClient.send_pixels() method uses this to
broadcast frames.
"""
def __init__(self):
self._clients: Dict[str, List] = {}
def add_client(self, device_id: str, ws) -> None:
self._clients.setdefault(device_id, []).append(ws)
logger.info(
"WS device %s: client connected (%d total)",
device_id, len(self._clients[device_id]),
)
def remove_client(self, device_id: str, ws) -> None:
clients = self._clients.get(device_id)
if clients and ws in clients:
clients.remove(ws)
logger.info(
"WS device %s: client disconnected (%d remaining)",
device_id, len(clients),
)
def get_clients(self, device_id: str) -> List:
return self._clients.get(device_id, [])
_broadcaster = WSDeviceBroadcaster()
def get_ws_broadcaster() -> WSDeviceBroadcaster:
return _broadcaster
def parse_ws_url(url: str) -> str:
"""Extract device_id from a ws:// URL.
Format: ws://device-id
"""
if url.startswith("ws://"):
return url[5:]
return url
class WSLEDClient(LEDClient):
"""Broadcasts binary pixel data to WebSocket clients via the global broadcaster."""
def __init__(self, url: str, led_count: int = 0, **kwargs):
self._device_id = parse_ws_url(url)
self._led_count = led_count
self._connected = False
async def connect(self) -> bool:
self._connected = True
logger.info("WS device client connected for device %s", self._device_id)
return True
async def close(self) -> None:
self._connected = False
logger.info("WS device client closed for device %s", self._device_id)
@property
def is_connected(self) -> bool:
return self._connected
async def send_pixels(
self,
pixels: Union[List[Tuple[int, int, int]], np.ndarray],
brightness: int = 255,
) -> bool:
if not self._connected:
return False
clients = _broadcaster.get_clients(self._device_id)
if not clients:
return True
# Build binary frame: [brightness_byte][R G B R G B ...]
if isinstance(pixels, np.ndarray):
pixel_bytes = pixels.astype(np.uint8).tobytes()
else:
pixel_bytes = bytes(c for rgb in pixels for c in rgb)
data = bytes([brightness]) + pixel_bytes
async def _send_safe(ws):
try:
await ws.send_bytes(data)
return True
except Exception:
return False
results = await asyncio.gather(*[_send_safe(ws) for ws in clients])
disconnected = [ws for ws, ok in zip(clients, results) if not ok]
for ws in disconnected:
_broadcaster.remove_client(self._device_id, ws)
return True
@classmethod
async def check_health(
cls,
url: str,
http_client,
prev_health: Optional[DeviceHealth] = None,
) -> DeviceHealth:
return DeviceHealth(
online=True,
latency_ms=0.0,
last_checked=datetime.utcnow(),
)

View File

@@ -0,0 +1,44 @@
"""WebSocket device provider — factory, validation, health checks."""
from datetime import datetime
from typing import List
from wled_controller.core.devices.led_client import (
DeviceHealth,
DiscoveredDevice,
LEDClient,
LEDDeviceProvider,
)
from wled_controller.core.devices.ws_client import WSLEDClient, parse_ws_url
from wled_controller.utils import get_logger
logger = get_logger(__name__)
class WSDeviceProvider(LEDDeviceProvider):
"""Provider for WebSocket-based virtual LED devices."""
@property
def device_type(self) -> str:
return "ws"
@property
def capabilities(self) -> set:
return {"manual_led_count"}
def create_client(self, url: str, **kwargs) -> LEDClient:
return WSLEDClient(url, **kwargs)
async def check_health(
self, url: str, http_client, prev_health=None,
) -> DeviceHealth:
return DeviceHealth(
online=True, latency_ms=0.0, last_checked=datetime.utcnow(),
)
async def validate_device(self, url: str) -> dict:
"""Validate WS device URL — accepts any URL since it will be auto-set."""
return {}
async def discover(self, timeout: float = 3.0) -> List[DiscoveredDevice]:
return []