alexei.dolgolyov
8f1140abad
Promotes the existing DDP packet layer (previously WLED-internal) to a
first-class device type so any DDP-speaking receiver (Pixelblaze,
ESPixelStick, xLights/Falcon endpoints, generic firmware) can be driven
directly without WLED in the path.
Backend:
- New DDPLEDClient wraps the DDPClient transport as a proper LEDClient
with supports_fast_send=True (synchronous UDP push on the hot loop).
- New DDPDeviceProvider — no native discovery, manual LED count,
capabilities = {manual_led_count, health_check}.
- DDPConfig joins the typed config union; Device storage gains
ddp_port / ddp_destination_id / ddp_color_order fields with safe
defaults (0/1/1 -> port 4048, destination 1=display, RGB byte order).
- URL scheme: ddp://host[:port] or bare host[:port] (default 4048).
- Health check resolves the host via async DNS; UDP has no reply
channel so reachability is best-effort by design.
- 29 new tests in test_ddp_led_client.py cover URL parsing, packet
hot path (brightness, list/numpy input shapes, fast vs async send),
provider validate/discover/capabilities, config round-trip via
Device.to_config() and to_dict/from_dict.
Frontend:
- 'ddp' in DEVICE_TYPE_KEYS (next to 'dmx'), paper-plane icon.
- isDdpDevice predicate + per-type field show/hide in the create &
settings modals.
- Color-order picker uses IconSelect (project rule bans plain select).
- Locale strings added in en/ru/zh.
Note: this commit also carries two pre-existing in-flight hunks that
were intermixed in the same files and could not be split out
non-interactively:
- api/routes/devices.py: URL-scheme inference for bare WLED hosts,
safer error messages, exception-isolated parallel discovery.
- storage/device_store.py: secret_box helpers + at-rest encryption of
Hue / BLE-Govee / MQTT credentials.
Both are independent of DDP and intentional per the user.
295 lines
9.1 KiB
Python
295 lines
9.1 KiB
Python
"""Tests for the standalone DDP LEDClient wrapper and provider."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from unittest.mock import AsyncMock, MagicMock
|
|
|
|
import numpy as np
|
|
import pytest
|
|
|
|
from ledgrab.core.devices.ddp_led_client import (
|
|
DEFAULT_DDP_PORT,
|
|
DDPLEDClient,
|
|
parse_ddp_url,
|
|
)
|
|
from ledgrab.core.devices.ddp_provider import DDPDeviceProvider
|
|
from ledgrab.core.devices.device_config import DDPConfig
|
|
from ledgrab.core.devices.led_client import ProviderDeps
|
|
|
|
|
|
# ============================================================================
|
|
# parse_ddp_url
|
|
# ============================================================================
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"url,expected",
|
|
[
|
|
("ddp://192.168.1.50", ("192.168.1.50", DEFAULT_DDP_PORT)),
|
|
("ddp://192.168.1.50:4048", ("192.168.1.50", 4048)),
|
|
("ddp://192.168.1.50:5000", ("192.168.1.50", 5000)),
|
|
("192.168.1.50", ("192.168.1.50", DEFAULT_DDP_PORT)),
|
|
("192.168.1.50:4048", ("192.168.1.50", 4048)),
|
|
("pixelblaze.local", ("pixelblaze.local", DEFAULT_DDP_PORT)),
|
|
("ddp://pixelblaze.local:9999", ("pixelblaze.local", 9999)),
|
|
],
|
|
)
|
|
def test_parse_ddp_url_accepts_common_forms(url, expected):
|
|
assert parse_ddp_url(url) == expected
|
|
|
|
|
|
@pytest.mark.parametrize("url", ["", " ", "ddp://", "://192.168.1.1"])
|
|
def test_parse_ddp_url_rejects_empty_or_hostless(url):
|
|
with pytest.raises(ValueError):
|
|
parse_ddp_url(url)
|
|
|
|
|
|
# ============================================================================
|
|
# DDPLEDClient
|
|
# ============================================================================
|
|
|
|
|
|
def _make_connected_client(led_count: int = 3, rgbw: bool = False) -> DDPLEDClient:
|
|
"""Build a DDPLEDClient with its underlying DDPClient transport mocked."""
|
|
client = DDPLEDClient("ddp://127.0.0.1", led_count=led_count, rgbw=rgbw)
|
|
# Avoid real UDP socket creation — stub the DDPClient layer.
|
|
inner = MagicMock()
|
|
inner.send_pixels_numpy = MagicMock()
|
|
inner.close = AsyncMock()
|
|
client._ddp = inner
|
|
client._connected = True
|
|
return client
|
|
|
|
|
|
def test_supports_fast_send_is_true():
|
|
client = DDPLEDClient("ddp://127.0.0.1", led_count=1)
|
|
assert client.supports_fast_send is True
|
|
|
|
|
|
def test_send_pixels_fast_pushes_pixels():
|
|
client = _make_connected_client(led_count=3)
|
|
pixels = np.array([[10, 20, 30], [40, 50, 60], [70, 80, 90]], dtype=np.uint8)
|
|
|
|
client.send_pixels_fast(pixels)
|
|
|
|
client._ddp.send_pixels_numpy.assert_called_once()
|
|
sent = client._ddp.send_pixels_numpy.call_args[0][0]
|
|
np.testing.assert_array_equal(sent, pixels)
|
|
|
|
|
|
def test_send_pixels_fast_applies_brightness():
|
|
client = _make_connected_client(led_count=1)
|
|
pixels = np.array([[200, 100, 50]], dtype=np.uint8)
|
|
|
|
client.send_pixels_fast(pixels, brightness=128)
|
|
|
|
sent = client._ddp.send_pixels_numpy.call_args[0][0]
|
|
# 200 * 128 // 255 = 100, 100 * 128 // 255 = 50, 50 * 128 // 255 = 25
|
|
np.testing.assert_array_equal(sent, np.array([[100, 50, 25]], dtype=np.uint8))
|
|
|
|
|
|
def test_send_pixels_fast_brightness_full_is_passthrough():
|
|
client = _make_connected_client(led_count=1)
|
|
pixels = np.array([[200, 100, 50]], dtype=np.uint8)
|
|
|
|
client.send_pixels_fast(pixels, brightness=255)
|
|
|
|
sent = client._ddp.send_pixels_numpy.call_args[0][0]
|
|
np.testing.assert_array_equal(sent, pixels)
|
|
|
|
|
|
def test_send_pixels_fast_brightness_zero_blanks_all():
|
|
client = _make_connected_client(led_count=2)
|
|
pixels = np.array([[200, 100, 50], [255, 255, 255]], dtype=np.uint8)
|
|
|
|
client.send_pixels_fast(pixels, brightness=0)
|
|
|
|
sent = client._ddp.send_pixels_numpy.call_args[0][0]
|
|
np.testing.assert_array_equal(sent, np.zeros_like(pixels))
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_async_send_pixels_accepts_list():
|
|
client = _make_connected_client(led_count=2)
|
|
|
|
await client.send_pixels([(1, 2, 3), (4, 5, 6)])
|
|
|
|
sent = client._ddp.send_pixels_numpy.call_args[0][0]
|
|
np.testing.assert_array_equal(sent, np.array([[1, 2, 3], [4, 5, 6]], dtype=np.uint8))
|
|
|
|
|
|
def test_send_pixels_fast_when_not_connected_raises():
|
|
client = DDPLEDClient("ddp://127.0.0.1", led_count=1)
|
|
with pytest.raises(RuntimeError, match="not connected"):
|
|
client.send_pixels_fast([(1, 2, 3)])
|
|
|
|
|
|
def test_send_pixels_fast_accepts_flat_array():
|
|
"""A flat 1-D RGB byte array reshapes to (N, 3) before sending."""
|
|
client = _make_connected_client(led_count=2)
|
|
flat = np.array([1, 2, 3, 4, 5, 6], dtype=np.uint8)
|
|
|
|
client.send_pixels_fast(flat)
|
|
|
|
sent = client._ddp.send_pixels_numpy.call_args[0][0]
|
|
np.testing.assert_array_equal(sent, np.array([[1, 2, 3], [4, 5, 6]], dtype=np.uint8))
|
|
|
|
|
|
# ============================================================================
|
|
# DDPDeviceProvider
|
|
# ============================================================================
|
|
|
|
|
|
def test_provider_device_type():
|
|
provider = DDPDeviceProvider()
|
|
assert provider.device_type == "ddp"
|
|
|
|
|
|
def test_provider_capabilities():
|
|
provider = DDPDeviceProvider()
|
|
caps = provider.capabilities
|
|
assert "manual_led_count" in caps
|
|
assert "health_check" in caps
|
|
# No native power / brightness — DDP has no reply channel.
|
|
assert "power_control" not in caps
|
|
assert "brightness_control" not in caps
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_provider_validate_device_accepts_bare_host():
|
|
provider = DDPDeviceProvider()
|
|
result = await provider.validate_device("192.168.1.50")
|
|
# validate_device returns the empty dict (no LED count to report)
|
|
assert result == {}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_provider_validate_device_rejects_empty_url():
|
|
provider = DDPDeviceProvider()
|
|
with pytest.raises(ValueError, match="Invalid DDP URL"):
|
|
await provider.validate_device("")
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_provider_discover_returns_empty():
|
|
"""DDP has no native discovery — provider must return an empty list, not raise."""
|
|
provider = DDPDeviceProvider()
|
|
assert await provider.discover() == []
|
|
|
|
|
|
def test_provider_create_client_threads_config_fields():
|
|
provider = DDPDeviceProvider()
|
|
config = DDPConfig(
|
|
device_id="device_test",
|
|
device_url="ddp://192.168.1.50:9000",
|
|
led_count=144,
|
|
rgbw=True,
|
|
ddp_port=9000,
|
|
ddp_destination_id=2,
|
|
ddp_color_order=4, # BGR
|
|
)
|
|
|
|
client = provider.create_client(config, deps=ProviderDeps())
|
|
|
|
assert isinstance(client, DDPLEDClient)
|
|
assert client.host == "192.168.1.50"
|
|
assert client.port == 9000
|
|
assert client._led_count == 144
|
|
assert client._rgbw is True
|
|
assert client._destination_id == 2
|
|
assert client._color_order == 4
|
|
|
|
|
|
# ============================================================================
|
|
# DDPConfig round-trip via Device.to_config()
|
|
# ============================================================================
|
|
|
|
|
|
def test_device_to_config_round_trip_ddp():
|
|
"""A Device with device_type='ddp' yields a DDPConfig from to_config()."""
|
|
from ledgrab.storage.device_store import Device
|
|
|
|
device = Device(
|
|
device_id="device_abc12345",
|
|
name="My Pixelblaze",
|
|
url="ddp://192.168.1.42",
|
|
led_count=300,
|
|
device_type="ddp",
|
|
rgbw=False,
|
|
ddp_port=4048,
|
|
ddp_destination_id=1,
|
|
ddp_color_order=1,
|
|
)
|
|
|
|
config = device.to_config()
|
|
|
|
assert isinstance(config, DDPConfig)
|
|
assert config.device_type == "ddp"
|
|
assert config.device_url == "ddp://192.168.1.42"
|
|
assert config.led_count == 300
|
|
assert config.ddp_port == 4048
|
|
assert config.ddp_destination_id == 1
|
|
assert config.ddp_color_order == 1
|
|
|
|
|
|
def test_device_to_dict_omits_ddp_defaults():
|
|
"""A DDP device with default DDP fields shouldn't pollute the persisted JSON."""
|
|
from ledgrab.storage.device_store import Device
|
|
|
|
device = Device(
|
|
device_id="device_abc12345",
|
|
name="Defaults",
|
|
url="ddp://192.168.1.42",
|
|
led_count=10,
|
|
device_type="ddp",
|
|
)
|
|
|
|
payload = device.to_dict()
|
|
|
|
# Defaults: ddp_port=0, ddp_destination_id=1, ddp_color_order=1 — all omitted.
|
|
assert "ddp_port" not in payload
|
|
assert "ddp_destination_id" not in payload
|
|
assert "ddp_color_order" not in payload
|
|
|
|
|
|
def test_device_to_dict_preserves_non_default_ddp_fields():
|
|
from ledgrab.storage.device_store import Device
|
|
|
|
device = Device(
|
|
device_id="device_abc12345",
|
|
name="Custom",
|
|
url="ddp://192.168.1.42",
|
|
led_count=10,
|
|
device_type="ddp",
|
|
ddp_port=4048,
|
|
ddp_destination_id=2,
|
|
ddp_color_order=4,
|
|
)
|
|
|
|
payload = device.to_dict()
|
|
|
|
assert payload["ddp_port"] == 4048
|
|
assert payload["ddp_destination_id"] == 2
|
|
assert payload["ddp_color_order"] == 4
|
|
|
|
|
|
def test_device_from_dict_round_trip_ddp_fields():
|
|
from ledgrab.storage.device_store import Device
|
|
|
|
payload = {
|
|
"id": "device_abc12345",
|
|
"name": "Roundtrip",
|
|
"url": "ddp://192.168.1.42",
|
|
"led_count": 50,
|
|
"device_type": "ddp",
|
|
"ddp_port": 9999,
|
|
"ddp_destination_id": 5,
|
|
"ddp_color_order": 2,
|
|
}
|
|
|
|
restored = Device.from_dict(payload)
|
|
|
|
assert restored.ddp_port == 9999
|
|
assert restored.ddp_destination_id == 5
|
|
assert restored.ddp_color_order == 2
|