feat: LED hot-path perf, tutorials expansion, modal markup polish
Performance (LED hot path, allocation-free per-frame): - Adalight: dedicated single-worker tx executor (avoids asyncio.to_thread overhead), pre-allocated wire buffer + uint8 scratch, header struct precomputed. New tests cover header format, buffer reuse, non-contiguous input, and brightness scaling. - DDP: pre-built struct.Struct for the 10-byte header, allocation-free send buffer + memoryview emit path. New tests cover RGB/RGBW packets, sequence/PUSH semantics, and multi-packet fragmentation. - Calibration: precomputed Phase 3 skip-LED resampling (floor/ceil indices, fractional weights, take/blend scratch buffers) — per-frame work is now np.take + in-place blend, no allocations. - WLED target processor: matching hot-path tightening. Tutorials: - Sub-tab switching, breadcrumb header, and prepare/switchSubTab hooks so a tour can open/close the dashboard customize panel and resolve targets behind sub-tabs. - New steps for integrations tab, dashboard customize panel (presets, global, sections, perf cells), targets, scenes, sync-clocks. - en/ru/zh locales updated with the new tour strings. Dashboard layout: - Structural deep-equal so the "modified" indicator reflects truth after a user edits then reverts, instead of a stale flag. UI polish: - Mod-card / modal markup pass across ~33 modal templates and the tutorial overlay partial. - appearance.css, modal.css, tutorials.css refresh. Tooling: - Add .mcp.json with code-review-graph MCP server config so the graph tools are available to the team out of the box.
This commit is contained in:
@@ -0,0 +1,133 @@
|
||||
"""Tests for AdalightClient frame construction (the LED hot path)."""
|
||||
|
||||
import numpy as np
|
||||
import pytest
|
||||
|
||||
from ledgrab.core.devices.adalight_client import (
|
||||
AdalightClient,
|
||||
_build_adalight_header,
|
||||
)
|
||||
|
||||
|
||||
def _make_client(led_count: int = 10) -> AdalightClient:
|
||||
"""Build a client without opening the serial port."""
|
||||
return AdalightClient(url="COM99", led_count=led_count)
|
||||
|
||||
|
||||
def test_adalight_header_format():
|
||||
# Adalight protocol: 'A' 'd' 'a' <count_hi> <count_lo> <checksum>
|
||||
# where count = led_count - 1, checksum = hi ^ lo ^ 0x55
|
||||
header = _build_adalight_header(10)
|
||||
count = 9
|
||||
hi = (count >> 8) & 0xFF
|
||||
lo = count & 0xFF
|
||||
expected_checksum = hi ^ lo ^ 0x55
|
||||
assert header == bytes([ord("A"), ord("d"), ord("a"), hi, lo, expected_checksum])
|
||||
|
||||
|
||||
def test_build_frame_uint8_fast_path():
|
||||
client = _make_client(led_count=3)
|
||||
pixels = np.array(
|
||||
[[10, 20, 30], [40, 50, 60], [70, 80, 90]],
|
||||
dtype=np.uint8,
|
||||
)
|
||||
|
||||
frame = client._build_frame(pixels, brightness=255)
|
||||
|
||||
# Header (6 bytes) + RGB (9 bytes)
|
||||
assert len(frame) == 6 + 9
|
||||
assert bytes(frame[:6]) == client._header
|
||||
assert bytes(frame[6:]) == bytes([10, 20, 30, 40, 50, 60, 70, 80, 90])
|
||||
|
||||
|
||||
def test_build_frame_reuses_buffer_across_calls():
|
||||
client = _make_client(led_count=3)
|
||||
pixels = np.array([[1, 2, 3]] * 3, dtype=np.uint8)
|
||||
|
||||
frame1 = client._build_frame(pixels, brightness=255)
|
||||
buf_id_1 = id(client._frame_buf)
|
||||
frame2 = client._build_frame(pixels, brightness=255)
|
||||
buf_id_2 = id(client._frame_buf)
|
||||
|
||||
# Same bytearray reused — frame returned IS the internal buffer
|
||||
assert buf_id_1 == buf_id_2
|
||||
assert frame1 is frame2
|
||||
# Header preserved at front, payload identical
|
||||
assert bytes(frame2[:6]) == client._header
|
||||
|
||||
|
||||
def test_build_frame_handles_non_contiguous_input():
|
||||
"""Slicing a wider array yields a non-contiguous view; payload must
|
||||
still serialise to the same bytes as a contiguous copy."""
|
||||
client = _make_client(led_count=4)
|
||||
wide = np.arange(48, dtype=np.uint8).reshape(4, 4, 3)
|
||||
pixels = wide[:, 1, :] # (4, 3) view, possibly non-contiguous
|
||||
|
||||
frame = client._build_frame(pixels, brightness=255)
|
||||
|
||||
expected = np.ascontiguousarray(pixels).tobytes()
|
||||
assert bytes(frame[6:]) == expected
|
||||
|
||||
|
||||
def test_build_frame_clamps_wider_int_dtypes():
|
||||
"""Wider integer inputs (e.g. uint16 from legacy code) clamp to [0, 255]
|
||||
before narrowing, matching historical behaviour."""
|
||||
client = _make_client(led_count=2)
|
||||
pixels = np.array([[100, 200, 300], [400, 50, 25]], dtype=np.uint16)
|
||||
|
||||
frame = client._build_frame(pixels, brightness=255)
|
||||
|
||||
# Values >255 clamp to 255, values <=255 pass through.
|
||||
assert bytes(frame[6:]) == bytes([100, 200, 255, 255, 50, 25])
|
||||
|
||||
|
||||
def test_build_frame_accepts_list_of_tuples():
|
||||
"""Legacy callers pass list[tuple]; should produce same output."""
|
||||
client = _make_client(led_count=2)
|
||||
pixels = [(10, 20, 30), (40, 50, 60)]
|
||||
|
||||
frame = client._build_frame(pixels, brightness=255)
|
||||
|
||||
assert bytes(frame[6:]) == bytes([10, 20, 30, 40, 50, 60])
|
||||
|
||||
|
||||
def test_build_frame_resizes_buffer_on_led_count_change():
|
||||
"""If led count changes between calls, the buffer is reallocated."""
|
||||
client = _make_client(led_count=5)
|
||||
small = np.zeros((3, 3), dtype=np.uint8)
|
||||
big = np.zeros((100, 3), dtype=np.uint8)
|
||||
|
||||
client._build_frame(small, brightness=255)
|
||||
small_len = len(client._frame_buf)
|
||||
|
||||
client._build_frame(big, brightness=255)
|
||||
big_len = len(client._frame_buf)
|
||||
|
||||
# 6-byte header + 3-byte/LED RGB
|
||||
assert small_len == 6 + 3 * 3
|
||||
assert big_len == 6 + 100 * 3
|
||||
|
||||
|
||||
def test_build_frame_no_uint16_round_trip():
|
||||
"""The hot path must NOT promote uint8 → uint16 → uint8.
|
||||
|
||||
Asserts the scratch buffer (used only for non-uint8 input) is never
|
||||
allocated when the input is already uint8.
|
||||
"""
|
||||
client = _make_client(led_count=3)
|
||||
pixels = np.array([[1, 2, 3]] * 3, dtype=np.uint8)
|
||||
|
||||
assert client._u8_scratch is None
|
||||
client._build_frame(pixels, brightness=255)
|
||||
# Scratch never touched on the fast path.
|
||||
assert client._u8_scratch is None
|
||||
|
||||
|
||||
@pytest.mark.parametrize("led_count", [1, 50, 300, 1000])
|
||||
def test_build_frame_total_length(led_count):
|
||||
client = _make_client(led_count=led_count)
|
||||
pixels = np.zeros((led_count, 3), dtype=np.uint8)
|
||||
|
||||
frame = client._build_frame(pixels, brightness=255)
|
||||
|
||||
assert len(frame) == 6 + led_count * 3
|
||||
@@ -0,0 +1,170 @@
|
||||
"""Tests for DDP client packet construction (the LED hot path)."""
|
||||
|
||||
import struct
|
||||
from unittest.mock import MagicMock
|
||||
|
||||
import numpy as np
|
||||
import pytest
|
||||
|
||||
from ledgrab.core.devices.ddp_client import DDPClient
|
||||
|
||||
|
||||
def _make_client(rgbw: bool = False) -> DDPClient:
|
||||
"""Build a DDPClient with a mocked transport that captures sendto bytes."""
|
||||
client = DDPClient(host="127.0.0.1", rgbw=rgbw)
|
||||
transport = MagicMock()
|
||||
captured: list[bytes] = []
|
||||
|
||||
def _sendto(data, addr=None): # noqa: ARG001
|
||||
# asyncio's sendto accepts bytes-like; copy to bytes for assertion stability
|
||||
captured.append(bytes(data))
|
||||
|
||||
transport.sendto.side_effect = _sendto
|
||||
client._transport = transport
|
||||
client._captured = captured # type: ignore[attr-defined]
|
||||
return client
|
||||
|
||||
|
||||
def _parse_header(packet: bytes) -> dict:
|
||||
"""Parse a 10-byte DDP header back into named fields."""
|
||||
flags, seq, dtype, src, offset, dlen = struct.unpack("!BBB B I H", packet[:10])
|
||||
return {
|
||||
"flags": flags,
|
||||
"seq": seq,
|
||||
"type": dtype,
|
||||
"src": src,
|
||||
"offset": offset,
|
||||
"data_len": dlen,
|
||||
"payload": packet[10:],
|
||||
}
|
||||
|
||||
|
||||
def test_send_pixels_numpy_single_packet_rgb():
|
||||
client = _make_client(rgbw=False)
|
||||
pixels = np.array(
|
||||
[[10, 20, 30], [40, 50, 60], [70, 80, 90]],
|
||||
dtype=np.uint8,
|
||||
)
|
||||
|
||||
client.send_pixels_numpy(pixels)
|
||||
|
||||
assert len(client._captured) == 1
|
||||
parsed = _parse_header(client._captured[0])
|
||||
# PUSH on the only packet, VER=1
|
||||
assert parsed["flags"] == 0x40 | 0x01
|
||||
assert parsed["type"] == 0x01
|
||||
assert parsed["offset"] == 0
|
||||
assert parsed["data_len"] == 9
|
||||
assert parsed["payload"] == bytes([10, 20, 30, 40, 50, 60, 70, 80, 90])
|
||||
|
||||
|
||||
def test_send_pixels_numpy_increments_sequence():
|
||||
client = _make_client(rgbw=False)
|
||||
pixels = np.array([[1, 2, 3]], dtype=np.uint8)
|
||||
|
||||
client.send_pixels_numpy(pixels)
|
||||
client.send_pixels_numpy(pixels)
|
||||
client.send_pixels_numpy(pixels)
|
||||
|
||||
seqs = [_parse_header(p)["seq"] for p in client._captured]
|
||||
assert seqs == [1, 2, 3]
|
||||
|
||||
|
||||
def test_send_pixels_numpy_chunks_large_payload():
|
||||
"""A 1000-LED payload (3000 bytes) splits into multiple packets."""
|
||||
client = _make_client(rgbw=False)
|
||||
n = 1000
|
||||
pixels = np.arange(n * 3, dtype=np.uint8).reshape(n, 3)
|
||||
|
||||
client.send_pixels_numpy(pixels, max_packet_size=1400)
|
||||
|
||||
# max_payload = 1390, bytes_per_packet = 1389 (multiple of 3) → 3 packets
|
||||
assert len(client._captured) >= 2
|
||||
headers = [_parse_header(p) for p in client._captured]
|
||||
|
||||
# Only the last packet has PUSH
|
||||
assert (headers[-1]["flags"] & 0x01) == 0x01
|
||||
for h in headers[:-1]:
|
||||
assert (h["flags"] & 0x01) == 0
|
||||
|
||||
# Offsets are contiguous and lengths sum to total
|
||||
total = sum(h["data_len"] for h in headers)
|
||||
assert total == n * 3
|
||||
expected_offset = 0
|
||||
for h in headers:
|
||||
assert h["offset"] == expected_offset
|
||||
expected_offset += h["data_len"]
|
||||
|
||||
|
||||
def test_send_pixels_numpy_rgbw_pads_alpha():
|
||||
client = _make_client(rgbw=True)
|
||||
pixels = np.array([[1, 2, 3], [4, 5, 6]], dtype=np.uint8)
|
||||
|
||||
client.send_pixels_numpy(pixels)
|
||||
|
||||
parsed = _parse_header(client._captured[0])
|
||||
# bpp=4 → payload is 8 bytes, alpha=0
|
||||
assert parsed["data_len"] == 8
|
||||
expected = bytes([1, 2, 3, 0, 4, 5, 6, 0])
|
||||
assert parsed["payload"] == expected
|
||||
|
||||
|
||||
def test_send_pixels_numpy_reuses_send_buffer():
|
||||
"""The internal send_buf must be allocated lazily and reused — no fresh
|
||||
bytearray per call."""
|
||||
client = _make_client(rgbw=False)
|
||||
pixels = np.array([[1, 2, 3]], dtype=np.uint8)
|
||||
|
||||
assert client._send_buf is None
|
||||
client.send_pixels_numpy(pixels)
|
||||
first_buf_id = id(client._send_buf)
|
||||
assert first_buf_id is not None
|
||||
|
||||
client.send_pixels_numpy(pixels)
|
||||
assert id(client._send_buf) == first_buf_id
|
||||
|
||||
|
||||
def test_send_pixels_numpy_growing_payload_grows_buffer():
|
||||
client = _make_client(rgbw=False)
|
||||
small = np.array([[1, 2, 3]], dtype=np.uint8)
|
||||
big = np.zeros((600, 3), dtype=np.uint8)
|
||||
|
||||
client.send_pixels_numpy(small)
|
||||
small_capacity = len(client._send_buf)
|
||||
|
||||
client.send_pixels_numpy(big, max_packet_size=2048)
|
||||
big_capacity = len(client._send_buf)
|
||||
|
||||
assert big_capacity >= small_capacity
|
||||
|
||||
|
||||
def test_send_pixels_numpy_handles_non_contiguous():
|
||||
"""Slicing a wider array yields a non-contiguous view; payload must
|
||||
still serialise to the same bytes as a contiguous copy."""
|
||||
client = _make_client(rgbw=False)
|
||||
wide = np.arange(60, dtype=np.uint8).reshape(5, 4, 3)
|
||||
# take the middle pixel column → (5, 3) but possibly non-contiguous
|
||||
pixels = wide[:, 1, :]
|
||||
|
||||
client.send_pixels_numpy(pixels)
|
||||
payload = _parse_header(client._captured[0])["payload"]
|
||||
assert payload == np.ascontiguousarray(pixels).tobytes()
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_send_pixels_async_delegates_to_numpy():
|
||||
client = _make_client(rgbw=False)
|
||||
pixels = [(10, 20, 30), (40, 50, 60)]
|
||||
|
||||
result = await client.send_pixels(pixels)
|
||||
assert result is True
|
||||
|
||||
payload = _parse_header(client._captured[0])["payload"]
|
||||
assert payload == bytes([10, 20, 30, 40, 50, 60])
|
||||
|
||||
|
||||
def test_send_pixels_numpy_raises_when_disconnected():
|
||||
client = DDPClient(host="127.0.0.1")
|
||||
pixels = np.array([[1, 2, 3]], dtype=np.uint8)
|
||||
with pytest.raises(RuntimeError):
|
||||
client.send_pixels_numpy(pixels)
|
||||
Reference in New Issue
Block a user