From e068fa6764fc97ddbe70c654252b756f8bb6d37a Mon Sep 17 00:00:00 2001 From: Jacob Simpson <28767380+djGLiTCH@users.noreply.github.com> Date: Tue, 11 Aug 2026 13:51:46 +1000 Subject: [PATCH 1/6] Add tools for the GP2040-CE Host Lighting add-on New hlp-* console tools talking to the add-on's vendor HID interface (usage page 0xFF47) over hidapi: hlp-ping (handshake and round-trip), hlp-caps (decode the board's self-reported LED capabilities), hlp-fill (visual test), hlp-input-mode, hlp-reboot-webconfig and hlp-reboot-bootsel. Multiple connected boards are selected between with --board-id. Adds the hidapi dependency. Signed-off-by: Jacob Simpson <28767380+djGLiTCH@users.noreply.github.com> --- CHANGELOG.md | 9 + README.md | 29 ++ gp2040ce_bintools/hostlighting.py | 444 ++++++++++++++++++++++++++++++ pyproject.toml | 8 +- tests/test_hostlighting.py | 54 ++++ 5 files changed, 543 insertions(+), 1 deletion(-) create mode 100644 gp2040ce_bintools/hostlighting.py create mode 100644 tests/test_hostlighting.py diff --git a/CHANGELOG.md b/CHANGELOG.md index ff0f971..3a659f0 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -3,6 +3,15 @@ Included is a summary of changes to the project. For full details, especially on behind-the-scenes code changes and development tools, see the commit history. +## Unreleased + +### Features + +* New `hlp-*` tools for the GP2040-CE Host Lighting add-on: `hlp-ping`, `hlp-caps`, `hlp-fill`, + `hlp-input-mode`, `hlp-reboot-webconfig`, and `hlp-reboot-bootsel` talk to the add-on's vendor HID + interface to verify a board, decode its self-reported LED capabilities, run a quick visual test, and + manage the board. Adds a dependency on `hidapi`. + ## v0.11.1 ### Miscellaneous diff --git a/README.md b/README.md index 3ee7ce6..4ad3ff8 100644 --- a/README.md +++ b/README.md @@ -119,6 +119,35 @@ Sample usage: % dump-gp2040ce `date +%Y%m%d`-backup.bin ``` +### hlp-* (Host Lighting tools) + +The `hlp-*` tools talk to a board running the GP2040-CE Host Lighting add-on, which exposes a vendor HID +interface for driving the board's RGB LEDs from host software. They require the `hidapi` package and a board +with the add-on enabled (Configuration -> Add-Ons -> Host Lighting in the web configurator). If more than one +board is connected, select one with `--board-id `. + +* `hlp-ping` verifies the protocol handshake (magic, version) and measures the command round-trip. +* `hlp-caps` decodes the board's self-reported capabilities: identity, runtime state, the LED map (buttons, + case, player LEDs), animation selection, and per-light positions where supported. +* `hlp-fill` fills the LEDs with a colour (`--scope all|buttons|case|pleds`) as a quick visual test, then + restores the board's own animations. +* `hlp-input-mode` sets the board's input mode and reboots into it. +* `hlp-reboot-webconfig` / `hlp-reboot-bootsel` reboot the board into the web configurator or the BOOTSEL + bootloader for flashing. + +Sample usage: + +``` +% hlp-ping +Haute42 COSMOX (v0.7.12), board ID 433031343539302E +magic GPHL, protocol 1.0 +100 pings in 601 ms (6.0 ms average) + +% hlp-fill --scope case 00FF00 +filled case with #00FF00 for 3.0s +released - on-board animations restored +``` + ### summarize-gp2040ce `summarize-gp2040ce` prints information regarding the provided USB device or file. It attempts to detect the firmware diff --git a/gp2040ce_bintools/hostlighting.py b/gp2040ce_bintools/hostlighting.py new file mode 100644 index 0000000..c45f8ff --- /dev/null +++ b/gp2040ce_bintools/hostlighting.py @@ -0,0 +1,444 @@ +"""Talk to a GP2040-CE board's Host Lighting interface over HID. + +The Host Lighting add-on exposes a vendor HID interface (usage page 0xFF47, +usage 0x4C) carrying the Host Lighting Protocol (HLP): fixed 64-byte reports +that let host software drive the board's RGB LEDs live and read the board's +LED layout from its own configuration. + +Every request is `[0]=command, [1]=sequence, [2..]=payload`; the reply echoes +the command with bit 7 set: `[0]=command|0x80, [1]=sequence, [2]=status, +[3..]=payload`. The board describes itself through GET_CAPS pages: + +* page 0 (identity): factory-unique board ID, board label, firmware version +* page 1 (runtime state): input mode, profile, brightness, host-assigned + player, LED-map fingerprint, current animation index +* page 2 (LED map): totals, colour format, per-button/case/player LED ranges +* page 3 (animations): current on-board animation index and how many exist +* page 4 (positions): per-light grid positions, where the render pipeline + provides them + +See docs/host-lighting.md in the GP2040-CE repository for the full protocol +reference. These tools require the `hidapi` package (`pip install hidapi`). + +SPDX-FileCopyrightText: © 2026 Jacob Simpson +SPDX-License-Identifier: GPL-3.0-or-later +""" +import argparse +import logging +import time + +from gp2040ce_bintools import core_parser + +logger = logging.getLogger(__name__) + +# discovery: match the interface by these, never by VID:PID (which varies by input mode) +USAGE_PAGE = 0xFF47 +USAGE = 0x4C +REPORT_SIZE = 64 +REQUIRED_VERSION = (1, 0) + +# command IDs, grouped by function range (see the protocol's compatibility contract) +CMD_PING = 0x01 # session and discovery, 0x01-0x0F +CMD_GET_CAPS = 0x02 +CMD_SET_MODE = 0x03 +CMD_SET_BUTTONS = 0x10 # frame staging, 0x10-0x2F +CMD_SET_RANGE = 0x11 +CMD_SET_RANGE_RGBW = 0x12 +CMD_FILL = 0x13 +CMD_CLEAR = 0x14 +CMD_COMMIT = 0x30 # frame lifecycle, 0x30-0x3F +CMD_RELEASE = 0x31 +CMD_SET_ANIMATION = 0x40 # board features, 0x40-0x4F +CMD_SET_INPUT_MODE = 0x7B # privileged (magic-guarded), 0x70-0x7F +CMD_REBOOT_WEBCONFIG = 0x7C +CMD_REBOOT_BOOTSEL = 0x7F + +RESPONSE_FLAG = 0x80 +STATUS_NAMES = {0: 'OK', 1: 'UNSUPPORTED', 2: 'INVALID_ARG'} + +# GET_CAPS pages (payload byte [2] of the request) +CAPS_PAGE_IDENTITY = 0 +CAPS_PAGE_STATE = 1 +CAPS_PAGE_LED_MAP = 2 +CAPS_PAGE_ANIMATIONS = 3 +CAPS_PAGE_POSITIONS = 4 + +# FILL scopes (payload byte [2] of a FILL request) +FILL_SCOPE_ALL = 0x00 +FILL_SCOPE_BUTTONS = 0x01 +FILL_SCOPE_CASE = 0x02 +FILL_SCOPE_PLEDS = 0x03 + +# magic payloads guarding the privileged commands against stray reports +MAGIC_INPUT_MODE = b'MODE' +MAGIC_REBOOT_WEBCONFIG = b'WEBC' +MAGIC_REBOOT_BOOTSEL = b'BOOT' + +# button IDs 0-17 as indexed in the page 2 LED map +BUTTON_NAMES = ['Up', 'Down', 'Left', 'Right', 'B1', 'B2', 'B3', 'B4', 'L1', 'R1', 'L2', 'R2', + 'S1', 'S2', 'L3', 'R3', 'A1', 'A2'] + +LED_FORMAT_NAMES = {0: 'GRB', 1: 'RGB', 2: 'GRBW', 3: 'RGBW'} + +INPUT_MODE_NAMES = {0: 'XINPUT', 3: 'KEYBOARD', 14: 'GENERIC'} + +UNMAPPED = 0xFF + + +def build_request(command: int, sequence: int, payload: bytes = b'') -> bytes: + """Frame an HLP request as a 64-byte report. + + :param command: HLP command byte (0x01-0x7F) + :param sequence: sequence byte echoed by the board in its reply + :param payload: command payload, at most 62 bytes + :return: the request framed to exactly REPORT_SIZE bytes + """ + if len(payload) > REPORT_SIZE - 2: + raise ValueError(f"payload too long ({len(payload)} > {REPORT_SIZE - 2})") + return bytes([command, sequence]) + payload + bytes(REPORT_SIZE - 2 - len(payload)) + + +def match_reply(reply: bytes, command: int, sequence: int) -> bool: + """Check whether a reply report answers the given request. + + :param reply: a reply report as read from the interface + :param command: the command byte of the original request + :param sequence: the sequence byte of the original request + :return: True if the reply's command echo and sequence match + """ + return len(reply) >= 3 and reply[0] == (command | RESPONSE_FLAG) and reply[1] == sequence + + +class HostLightingError(RuntimeError): + """Errors talking to a Host Lighting interface.""" + + +class HostLightingDevice: + """One GP2040-CE board's Host Lighting interface.""" + + def __init__(self, path: bytes): + """Open the HID device at the given hidapi path. + + :param path: platform-specific hidapi device path from enumeration + """ + hid = _import_hid() + self.device = hid.device() + self.device.open_path(path) + self.device.set_nonblocking(True) + self.sequence = 0 + + def close(self) -> None: + """Close the HID device.""" + self.device.close() + + def request(self, command: int, payload: bytes = b'', timeout: float = 0.5) -> bytes: + """Send one HLP command and wait for its matching reply. + + Because commands can be pipelined, replies may arrive interleaved; + each incoming report is matched against this request by its command + echo and sequence number rather than assuming strict ordering. + + :param command: HLP command byte + :param payload: command payload bytes + :param timeout: seconds to wait for the matching reply + :return: the reply report ([2] is the status byte, [3..] the payload) + """ + self.sequence = (self.sequence % 127) + 1 + request = build_request(command, self.sequence, payload) + # the interface uses unnumbered reports; hidapi wants a leading 0x00 report ID on write + self.device.write(b'\x00' + request) + deadline = time.monotonic() + timeout + while time.monotonic() < deadline: + report = bytes(self.device.read(REPORT_SIZE)) + if match_reply(report, command, self.sequence): + return report + time.sleep(0.001) + raise HostLightingError(f"no reply to command 0x{command:02X} within {timeout}s") + + def request_ok(self, command: int, payload: bytes = b'', timeout: float = 0.5) -> bytes: + """Send one HLP command and require an OK status in the reply. + + :param command: HLP command byte + :param payload: command payload bytes + :param timeout: seconds to wait for the matching reply + :return: the reply report + """ + reply = self.request(command, payload, timeout) + if reply[2] != 0: + status = STATUS_NAMES.get(reply[2], hex(reply[2])) + raise HostLightingError(f"command 0x{command:02X} rejected: {status}") + return reply + + def get_caps_page(self, page: int, start_entry: int = 0) -> bytes: + """Read one GET_CAPS page. + + :param page: which capability page to read (CAPS_PAGE_* constant) + :param start_entry: first entry to return, for the paged positions page + :return: the reply report ([3..] is the page's payload) + """ + if page == CAPS_PAGE_POSITIONS: + return self.request_ok(CMD_GET_CAPS, bytes([page, start_entry])) + return self.request_ok(CMD_GET_CAPS, bytes([page])) + + +def _import_hid(): + """Import the hidapi module, with a helpful error if it is missing.""" + try: + import hid + except ImportError as error: + raise HostLightingError("these tools require the hidapi package: pip install hidapi") from error + return hid + + +def find_devices() -> list: + """Enumerate all Host Lighting interfaces on the system. + + :return: list of hidapi enumeration dicts for matching interfaces + """ + hid = _import_hid() + return [info for info in hid.enumerate() + if info.get('usage_page') == USAGE_PAGE and info.get('usage') == USAGE] + + +def open_device(board_id_prefix: str = '') -> HostLightingDevice: + """Open a Host Lighting device, disambiguating by board ID if needed. + + :param board_id_prefix: optional hex prefix of the page 0 factory board ID + :return: an opened HostLightingDevice + """ + infos = find_devices() + if not infos: + raise HostLightingError("no Host Lighting interface found - is a board connected " + "with the add-on enabled?") + candidates = [] + for info in infos: + device = HostLightingDevice(info['path']) + try: + board_id, _, _ = read_identity(device) + except HostLightingError: + device.close() + continue + if board_id.startswith(board_id_prefix.upper()): + candidates.append((board_id, device)) + else: + device.close() + if not candidates: + raise HostLightingError(f"no board matches ID prefix '{board_id_prefix}'") + if len(candidates) > 1: + ids = ', '.join(board_id for board_id, _ in candidates) + for _, device in candidates: + device.close() + raise HostLightingError(f"multiple boards found ({ids}) - select one with --board-id") + return candidates[0][1] + + +def read_identity(device: HostLightingDevice) -> tuple[str, str, str]: + """Read the board's identity from GET_CAPS page 0. + + Page 0 reply layout: [3] caps format, [4..11] factory-unique board ID, + then two NUL-terminated strings (board label, firmware version). + + :param device: an opened HostLightingDevice + :return: (factory board ID as hex, board label, firmware version) + """ + reply = device.get_caps_page(CAPS_PAGE_IDENTITY) + board_id = reply[4:12].hex().upper() + strings = reply[12:].split(b'\x00') + label = strings[0].decode('ascii', 'replace') + firmware = strings[1].decode('ascii', 'replace') if len(strings) > 1 else '' + return board_id, label, firmware + + +def _device_parser(description: str) -> argparse.ArgumentParser: + """Build an argument parser with the common device-selection flag. + + :param description: help text for the tool + :return: an ArgumentParser with core and device-selection arguments + """ + parser = argparse.ArgumentParser(description=description, parents=[core_parser]) + parser.add_argument('--board-id', default='', + help="hex prefix of the factory board ID, to select one of several connected boards") + return parser + + +############ +# COMMANDS # +############ + + +def ping(): + """Check a board's Host Lighting interface and protocol version.""" + parser = _device_parser("Ping a GP2040-CE Host Lighting interface and verify the protocol handshake.") + args, _ = parser.parse_known_args() + device = open_device(args.board_id) + try: + reply = device.request_ok(CMD_PING) + # PING reply layout: [3..6] the ASCII magic "GPHL", [7] major version, [8] minor version + magic = reply[3:7].decode('ascii', 'replace') + major, minor = reply[7], reply[8] + board_id, label, firmware = read_identity(device) + print(f"{label} ({firmware}), board ID {board_id}") + print(f"magic {magic}, protocol {major}.{minor}") + if magic != 'GPHL' or (major, minor) < REQUIRED_VERSION: + raise SystemExit("handshake failed: expected GPHL >= " + f"{REQUIRED_VERSION[0]}.{REQUIRED_VERSION[1]}") + count = 100 + start = time.monotonic() + for _ in range(count): + device.request_ok(CMD_PING) + elapsed = time.monotonic() - start + print(f"{count} pings in {elapsed * 1000:.0f} ms ({elapsed * 1000 / count:.1f} ms average)") + finally: + device.close() + + +def caps(): + """Print a board's Host Lighting capabilities, page by page.""" + parser = _device_parser("Read and decode all Host Lighting capability pages from a GP2040-CE board.") + args, _ = parser.parse_known_args() + device = open_device(args.board_id) + try: + board_id, label, firmware = read_identity(device) + print(f"page 0 (identity): {label} ({firmware}), board ID {board_id}") + _print_state(device) + _print_led_map(device) + _print_animations(device) + _print_positions(device) + finally: + device.close() + + +def _print_state(device: HostLightingDevice) -> None: + """Read and print GET_CAPS page 1, the board's runtime state. + + Page 1 reply layout: [3] input mode, [4] profile, [5] brightness, + [6] host-assigned player (0 = none), [7..10] LED-map fingerprint (little + endian), [11] current animation index. + + :param device: an opened HostLightingDevice + """ + reply = device.get_caps_page(CAPS_PAGE_STATE) + fingerprint = int.from_bytes(reply[7:11], 'little') + mode = INPUT_MODE_NAMES.get(reply[3], str(reply[3])) + print(f"page 1 (runtime state): input mode {mode}, profile {reply[4]}, brightness {reply[5]}, " + f"host player {reply[6]}, map fingerprint 0x{fingerprint:08X}") + + +def _print_led_map(device: HostLightingDevice) -> None: + """Read and print GET_CAPS page 2, the LED map. + + Page 2 reply layout: [3] LEDs per button, [4] LED colour format, + [5] button layout enum, [6] total LED count, [7] brightness maximum, + [8..43] per-button {first LED, count} pairs for button IDs 0-17 + (first = 0xFF means unmapped), [44..47] player LED indexes, [48] turbo + LED index, [49..50] case strip {first LED, count}, [51..54] LED-map + fingerprint (little endian, matches page 1's for a coherent snapshot). + + :param device: an opened HostLightingDevice + """ + reply = device.get_caps_page(CAPS_PAGE_LED_MAP) + colour = LED_FORMAT_NAMES.get(reply[4], str(reply[4])) + print(f"page 2 (LED map): {reply[6]} LEDs total, {reply[3]} per button, colour format {colour}, " + f"brightness maximum {reply[7]}, layout enum {reply[5]}") + entries = [] + for index, name in enumerate(BUTTON_NAMES): + first, count = reply[8 + index * 2], reply[9 + index * 2] + if first != UNMAPPED and count: + entries.append(f"{name}={first}" + (f"+{count}" if count > 1 else '')) + print(" buttons: " + (', '.join(entries) if entries else 'none mapped')) + if reply[50]: + print(f" case strip: LEDs {reply[49]}..{reply[49] + reply[50] - 1}") + + +def _print_animations(device: HostLightingDevice) -> None: + """Read and print GET_CAPS page 3, the on-board animation selection. + + Page 3 reply layout: [3] current animation index, [4] number of + animations the board offers (board-specific). + + :param device: an opened HostLightingDevice + """ + reply = device.get_caps_page(CAPS_PAGE_ANIMATIONS) + print(f"page 3 (animations): index {reply[3]} of {reply[4]} available") + + +def _print_positions(device: HostLightingDevice) -> None: + """Read and print GET_CAPS page 4, the per-light grid positions. + + Page 4 reply layout: [3] total position entries, [4] entries in this + reply, then that many {first LED, x, y} triples. Boards whose render + pipeline has no per-light positions report zero entries; hosts fall back + to the layout enum from page 2. + + :param device: an opened HostLightingDevice + """ + reply = device.get_caps_page(CAPS_PAGE_POSITIONS) + total = reply[3] + print(f"page 4 (positions): {total if total else 'none (this render pipeline has no per-light positions)'}") + + +def fill(): + """Fill a board's LEDs with one colour as a quick visual test.""" + parser = _device_parser("Fill a GP2040-CE board's LEDs with a colour, hold, then restore animations.") + parser.add_argument('--scope', choices=['all', 'buttons', 'case', 'pleds'], default='all', + help="which lights to fill (default: all)") + parser.add_argument('--seconds', type=float, default=3.0, help="how long to hold the fill (default: 3)") + parser.add_argument('colour', help="colour as RRGGBB hex, e.g. FF0000 for red") + args, _ = parser.parse_known_args() + value = int(args.colour, 16) + scope = {'all': FILL_SCOPE_ALL, 'buttons': FILL_SCOPE_BUTTONS, + 'case': FILL_SCOPE_CASE, 'pleds': FILL_SCOPE_PLEDS}[args.scope] + device = open_device(args.board_id) + try: + # SET_MODE payload: [2] takeover mode (0 = whole frame), [3..4] keepalive + # timeout in ms (little endian, 10000 here), [5] apply board brightness + device.request_ok(CMD_SET_MODE, bytes([0x00, 0x10, 0x27, 0x01])) + # FILL payload: [2] scope, [3..5] colour as R, G, B + device.request_ok(CMD_FILL, bytes([scope, (value >> 16) & 0xFF, (value >> 8) & 0xFF, value & 0xFF])) + device.request_ok(CMD_COMMIT) + print(f"filled {args.scope} with #{value:06X} for {args.seconds}s") + time.sleep(args.seconds) + device.request_ok(CMD_RELEASE) + print("released - on-board animations restored") + finally: + device.close() + + +def input_mode(): + """Set a board's input mode over Host Lighting.""" + parser = _device_parser("Set a GP2040-CE board's input mode; the board saves it and reboots into it.") + parser.add_argument('mode', type=int, help="input mode number (0 XINPUT, 3 KEYBOARD, 14 GENERIC " + "carry the lighting interface)") + args, _ = parser.parse_known_args() + device = open_device(args.board_id) + try: + # SET_INPUT_MODE payload: [2] mode, [3..6] guard magic + device.request_ok(CMD_SET_INPUT_MODE, bytes([args.mode]) + MAGIC_INPUT_MODE) + name = INPUT_MODE_NAMES.get(args.mode, str(args.mode)) + print(f"input mode {name} set - board is rebooting") + finally: + device.close() + + +def reboot_webconfig(): + """Reboot a board into web configurator mode.""" + parser = _device_parser("Reboot a GP2040-CE board into its web configurator (usually at 192.168.7.1).") + args, _ = parser.parse_known_args() + device = open_device(args.board_id) + try: + device.request_ok(CMD_REBOOT_WEBCONFIG, MAGIC_REBOOT_WEBCONFIG) + print("rebooting to web configurator") + finally: + device.close() + + +def reboot_bootsel(): + """Reboot a board into the RP2040/RP2350 bootloader for flashing.""" + parser = _device_parser("Reboot a GP2040-CE board into BOOTSEL mode (the RPI-RP2 drive) for flashing.") + args, _ = parser.parse_known_args() + device = open_device(args.board_id) + try: + device.request_ok(CMD_REBOOT_BOOTSEL, MAGIC_REBOOT_BOOTSEL) + print("rebooting to BOOTSEL - watch for the RPI-RP2 drive") + finally: + device.close() diff --git a/pyproject.toml b/pyproject.toml index fda0f8d..da5a9b4 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -11,7 +11,7 @@ authors = [ {name = "Brian S. Stephan", email = "bss@incorporeal.org"}, ] requires-python = ">=3.9" -dependencies = ["grpcio-tools", "pyusb", "textual"] +dependencies = ["grpcio-tools", "hidapi", "pyusb", "textual"] dynamic = ["version"] classifiers = [ "Environment :: Console", @@ -38,6 +38,12 @@ concatenate = "gp2040ce_bintools.builder:concatenate" dump-config = "gp2040ce_bintools.storage:dump_config" dump-gp2040ce = "gp2040ce_bintools.builder:dump_gp2040ce" edit-config = "gp2040ce_bintools.gui:edit_config" +hlp-caps = "gp2040ce_bintools.hostlighting:caps" +hlp-fill = "gp2040ce_bintools.hostlighting:fill" +hlp-input-mode = "gp2040ce_bintools.hostlighting:input_mode" +hlp-ping = "gp2040ce_bintools.hostlighting:ping" +hlp-reboot-bootsel = "gp2040ce_bintools.hostlighting:reboot_bootsel" +hlp-reboot-webconfig = "gp2040ce_bintools.hostlighting:reboot_webconfig" summarize-gp2040ce = "gp2040ce_bintools.builder:summarize_gp2040ce" visualize-config = "gp2040ce_bintools.storage:visualize" diff --git a/tests/test_hostlighting.py b/tests/test_hostlighting.py new file mode 100644 index 0000000..4aceb94 --- /dev/null +++ b/tests/test_hostlighting.py @@ -0,0 +1,54 @@ +"""Test the Host Lighting protocol helpers. + +SPDX-FileCopyrightText: © 2026 Jacob Simpson +SPDX-License-Identifier: GPL-3.0-or-later +""" +import pytest + +from gp2040ce_bintools import hostlighting + + +def test_build_request_frames_to_report_size(): + """Test that a request is framed to exactly one 64-byte report.""" + report = hostlighting.build_request(hostlighting.CMD_PING, 0x42) + assert len(report) == hostlighting.REPORT_SIZE + assert report[0] == hostlighting.CMD_PING + assert report[1] == 0x42 + assert all(byte == 0 for byte in report[2:]) + + +def test_build_request_places_payload(): + """Test that the payload lands at offset 2 and the rest is zero padding.""" + report = hostlighting.build_request(hostlighting.CMD_FILL, 1, bytes([0x02, 0xAB, 0xCD, 0xEF])) + assert report[2:6] == bytes([0x02, 0xAB, 0xCD, 0xEF]) + assert all(byte == 0 for byte in report[6:]) + + +def test_build_request_rejects_oversized_payload(): + """Test that a payload larger than the report is rejected.""" + with pytest.raises(ValueError): + hostlighting.build_request(hostlighting.CMD_FILL, 1, bytes(hostlighting.REPORT_SIZE - 1)) + + +def test_match_reply_accepts_matching_echo(): + """Test that a reply matches on the command echo plus sequence.""" + reply = bytes([hostlighting.CMD_COMMIT | hostlighting.RESPONSE_FLAG, 0x17, 0x00]) + bytes(61) + assert hostlighting.match_reply(reply, hostlighting.CMD_COMMIT, 0x17) + + +def test_match_reply_rejects_wrong_sequence(): + """Test that a reply for a different request is not matched.""" + reply = bytes([hostlighting.CMD_COMMIT | hostlighting.RESPONSE_FLAG, 0x18, 0x00]) + bytes(61) + assert not hostlighting.match_reply(reply, hostlighting.CMD_COMMIT, 0x17) + + +def test_match_reply_rejects_wrong_command(): + """Test that an interleaved reply to another command is not matched.""" + reply = bytes([hostlighting.CMD_SET_RANGE | hostlighting.RESPONSE_FLAG, 0x17, 0x00]) + bytes(61) + assert not hostlighting.match_reply(reply, hostlighting.CMD_COMMIT, 0x17) + + +def test_match_reply_rejects_short_report(): + """Test that a truncated report is not matched.""" + assert not hostlighting.match_reply(b'', hostlighting.CMD_PING, 1) + assert not hostlighting.match_reply(bytes([hostlighting.CMD_PING | 0x80]), hostlighting.CMD_PING, 1) From 7238685c9af0758ea70a76b4318f19a3d337377f Mon Sep 17 00:00:00 2001 From: Jacob Simpson <28767380+djGLiTCH@users.noreply.github.com> Date: Tue, 11 Aug 2026 15:51:27 +1000 Subject: [PATCH 2/6] Tolerate the board disconnecting during reboot commands Reboot-style commands execute before their reply is sent, so the board often drops off the bus before the acknowledgement can be read; the reboot tools previously surfaced that as a raw read error even though the reboot had succeeded. A vanished device or missing reply is now treated as the reboot proceeding, while a rejection reply (wrong guard magic) still raises, via a distinct HostLightingRejected error. Found by flashing a board with hlp-reboot-bootsel; the fix is verified against a live reboot cycle and covered by unit tests. Signed-off-by: Jacob Simpson <28767380+djGLiTCH@users.noreply.github.com> --- gp2040ce_bintools/hostlighting.py | 36 ++++++++++++++++++++++++++---- tests/test_hostlighting.py | 37 +++++++++++++++++++++++++++++++ 2 files changed, 69 insertions(+), 4 deletions(-) diff --git a/gp2040ce_bintools/hostlighting.py b/gp2040ce_bintools/hostlighting.py index c45f8ff..c53a9cc 100644 --- a/gp2040ce_bintools/hostlighting.py +++ b/gp2040ce_bintools/hostlighting.py @@ -113,6 +113,10 @@ class HostLightingError(RuntimeError): """Errors talking to a Host Lighting interface.""" +class HostLightingRejected(HostLightingError): + """The board answered a command with a non-OK status.""" + + class HostLightingDevice: """One GP2040-CE board's Host Lighting interface.""" @@ -166,7 +170,7 @@ def request_ok(self, command: int, payload: bytes = b'', timeout: float = 0.5) - reply = self.request(command, payload, timeout) if reply[2] != 0: status = STATUS_NAMES.get(reply[2], hex(reply[2])) - raise HostLightingError(f"command 0x{command:02X} rejected: {status}") + raise HostLightingRejected(f"command 0x{command:02X} rejected: {status}") return reply def get_caps_page(self, page: int, start_entry: int = 0) -> bytes: @@ -249,6 +253,30 @@ def read_identity(device: HostLightingDevice) -> tuple[str, str, str]: return board_id, label, firmware +def send_reboot(device: HostLightingDevice, command: int, payload: bytes) -> bool: + """Send a reboot-style command, tolerating the board disconnecting. + + Reboot commands execute before their reply is sent, so the board often + drops off the bus before the acknowledgement can be read; a vanished + device or a missing reply means the reboot is under way, not a failure. + A rejection reply (for example a wrong guard magic) still raises. + + :param device: an opened HostLightingDevice + :param command: the reboot-style HLP command byte + :param payload: the command's guard-magic payload + :return: True if the board acknowledged before rebooting, False if it + went away without a readable acknowledgement + """ + try: + device.request_ok(command, payload) + return True + except HostLightingRejected: + raise + except (HostLightingError, OSError): + logger.debug("no acknowledgement before disconnect; reboot is proceeding") + return False + + def _device_parser(description: str) -> argparse.ArgumentParser: """Build an argument parser with the common device-selection flag. @@ -413,7 +441,7 @@ def input_mode(): device = open_device(args.board_id) try: # SET_INPUT_MODE payload: [2] mode, [3..6] guard magic - device.request_ok(CMD_SET_INPUT_MODE, bytes([args.mode]) + MAGIC_INPUT_MODE) + send_reboot(device, CMD_SET_INPUT_MODE, bytes([args.mode]) + MAGIC_INPUT_MODE) name = INPUT_MODE_NAMES.get(args.mode, str(args.mode)) print(f"input mode {name} set - board is rebooting") finally: @@ -426,7 +454,7 @@ def reboot_webconfig(): args, _ = parser.parse_known_args() device = open_device(args.board_id) try: - device.request_ok(CMD_REBOOT_WEBCONFIG, MAGIC_REBOOT_WEBCONFIG) + send_reboot(device, CMD_REBOOT_WEBCONFIG, MAGIC_REBOOT_WEBCONFIG) print("rebooting to web configurator") finally: device.close() @@ -438,7 +466,7 @@ def reboot_bootsel(): args, _ = parser.parse_known_args() device = open_device(args.board_id) try: - device.request_ok(CMD_REBOOT_BOOTSEL, MAGIC_REBOOT_BOOTSEL) + send_reboot(device, CMD_REBOOT_BOOTSEL, MAGIC_REBOOT_BOOTSEL) print("rebooting to BOOTSEL - watch for the RPI-RP2 drive") finally: device.close() diff --git a/tests/test_hostlighting.py b/tests/test_hostlighting.py index 4aceb94..ee5f51c 100644 --- a/tests/test_hostlighting.py +++ b/tests/test_hostlighting.py @@ -52,3 +52,40 @@ def test_match_reply_rejects_short_report(): """Test that a truncated report is not matched.""" assert not hostlighting.match_reply(b'', hostlighting.CMD_PING, 1) assert not hostlighting.match_reply(bytes([hostlighting.CMD_PING | 0x80]), hostlighting.CMD_PING, 1) + + +class _StubDevice: + """Stand-in device whose request_ok raises a scripted exception.""" + + def __init__(self, error=None): + self.error = error + + def request_ok(self, command, payload=b'', timeout=0.5): + """Raise the scripted error, or return a fake OK reply.""" + if self.error is not None: + raise self.error + return bytes(64) + + +def test_send_reboot_acknowledged(): + """Test that an acknowledged reboot reports True.""" + assert hostlighting.send_reboot(_StubDevice(), hostlighting.CMD_REBOOT_BOOTSEL, b'BOOT') is True + + +def test_send_reboot_tolerates_disconnect(): + """Test that the board vanishing mid-reply is treated as rebooting.""" + device = _StubDevice(OSError('read error')) + assert hostlighting.send_reboot(device, hostlighting.CMD_REBOOT_BOOTSEL, b'BOOT') is False + + +def test_send_reboot_tolerates_missing_reply(): + """Test that a reply timeout is treated as rebooting.""" + device = _StubDevice(hostlighting.HostLightingError('no reply')) + assert hostlighting.send_reboot(device, hostlighting.CMD_REBOOT_WEBCONFIG, b'WEBC') is False + + +def test_send_reboot_still_raises_on_rejection(): + """Test that a board rejection (e.g. wrong magic) is not swallowed.""" + device = _StubDevice(hostlighting.HostLightingRejected('command 0x7F rejected: INVALID_ARG')) + with pytest.raises(hostlighting.HostLightingRejected): + hostlighting.send_reboot(device, hostlighting.CMD_REBOOT_BOOTSEL, b'XXXX') From c42dccce6bcbd976e1fd45b4b2a8fc60c52352c2 Mon Sep 17 00:00:00 2001 From: Jacob Simpson <28767380+djGLiTCH@users.noreply.github.com> Date: Wed, 12 Aug 2026 13:14:04 +1000 Subject: [PATCH 3/6] Label page 1's brightness field as a step index Page 1 byte [5] is a step index into the board's brightness steps, not a 0-255 level. Page 2's brightness maximum is a separate 0-255 ceiling, so printing byte [5] as plain "brightness" beside it implies one is a fraction of the other: "brightness 5" against "brightness maximum 200" reads as nearly off. It is neither. HLP does not report the step count, and that count varies by firmware - mainline defaults to 5 and the web configurator can set 1 to 10, while the LED refactor fixes it at 10 - so a step number means nothing without the board's total. On a refactor board, step 5 is half brightness. The firmware and the protocol reference both call it a step; only these tools were vague. Wording only, no behaviour change. Signed-off-by: Jacob Simpson <28767380+djGLiTCH@users.noreply.github.com> --- gp2040ce_bintools/hostlighting.py | 13 ++++++++++--- 1 file changed, 10 insertions(+), 3 deletions(-) diff --git a/gp2040ce_bintools/hostlighting.py b/gp2040ce_bintools/hostlighting.py index c53a9cc..e1241cd 100644 --- a/gp2040ce_bintools/hostlighting.py +++ b/gp2040ce_bintools/hostlighting.py @@ -10,7 +10,7 @@ [3..]=payload`. The board describes itself through GET_CAPS pages: * page 0 (identity): factory-unique board ID, board label, firmware version -* page 1 (runtime state): input mode, profile, brightness, host-assigned +* page 1 (runtime state): input mode, profile, brightness step, host-assigned player, LED-map fingerprint, current animation index * page 2 (LED map): totals, colour format, per-button/case/player LED ranges * page 3 (animations): current on-board animation index and how many exist @@ -339,16 +339,23 @@ def caps(): def _print_state(device: HostLightingDevice) -> None: """Read and print GET_CAPS page 1, the board's runtime state. - Page 1 reply layout: [3] input mode, [4] profile, [5] brightness, + Page 1 reply layout: [3] input mode, [4] profile, [5] brightness step, [6] host-assigned player (0 = none), [7..10] LED-map fingerprint (little endian), [11] current animation index. + Byte [5] is a step index into the board's brightness steps, not a 0-255 + level, and HLP does not report how many steps there are. That count varies + by firmware - mainline defaults to 5 and the web configurator can set 1 to + 10, while the LED refactor fixes it at 10 - so a step number cannot be + turned into a level on its own. It is unrelated to page 2's brightness + maximum, which is a 0-255 ceiling. + :param device: an opened HostLightingDevice """ reply = device.get_caps_page(CAPS_PAGE_STATE) fingerprint = int.from_bytes(reply[7:11], 'little') mode = INPUT_MODE_NAMES.get(reply[3], str(reply[3])) - print(f"page 1 (runtime state): input mode {mode}, profile {reply[4]}, brightness {reply[5]}, " + print(f"page 1 (runtime state): input mode {mode}, profile {reply[4]}, brightness step {reply[5]}, " f"host player {reply[6]}, map fingerprint 0x{fingerprint:08X}") From 1fd3c40d1e0a3fd6b10f945e934a6aba59499045 Mon Sep 17 00:00:00 2001 From: Jacob Simpson <28767380+djGLiTCH@users.noreply.github.com> Date: Mon, 17 Aug 2026 17:23:00 +1000 Subject: [PATCH 4/6] Support Host Lighting Protocol v1.1 hlp-caps decodes everything v1.1 added: the light table (capability page 5), which names every individual light and the control that owns it, including its grid position; and the feature bitmask, LED framework, animation namespace and render rate appended to the runtime state page. Paged reads re-check the page fingerprint and guard against a board that stops making progress, so a table read across several reports is known coherent rather than assumed. Boards speaking v1.0 decode unchanged: fields they never sent report as absent rather than guessed, and the light table reads as unsupported rather than as an error. The suite grows to 60 tests covering the new decoders, the stall guards and the sentinel values, and the README samples are real output from the reference boards. --- CHANGELOG.md | 5 + README.md | 27 +- gp2040ce_bintools/hostlighting.py | 419 ++++++++++++++++++++-- tests/test_hostlighting.py | 563 ++++++++++++++++++++++++++++++ 4 files changed, 983 insertions(+), 31 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 3a659f0..b2a521a 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -11,6 +11,11 @@ development tools, see the commit history. `hlp-input-mode`, `hlp-reboot-webconfig`, and `hlp-reboot-bootsel` talk to the add-on's vendor HID interface to verify a board, decode its self-reported LED capabilities, run a quick visual test, and manage the board. Adds a dependency on `hidapi`. +* `hlp-caps` decodes everything Host Lighting Protocol v1.1 added: the light table, which names every + individual light and the control that owns it; the per-light grid positions; and the feature bitmask, + LED framework, animation namespace and render rate on the runtime state page. Boards speaking v1.0 are + handled unchanged - fields they never sent are reported as absent rather than guessed, and the light + table reads as unsupported rather than as an error. ## v0.11.1 diff --git a/README.md b/README.md index 4ad3ff8..42a2c3a 100644 --- a/README.md +++ b/README.md @@ -128,7 +128,9 @@ board is connected, select one with `--board-id `. * `hlp-ping` verifies the protocol handshake (magic, version) and measures the command round-trip. * `hlp-caps` decodes the board's self-reported capabilities: identity, runtime state, the LED map (buttons, - case, player LEDs), animation selection, and per-light positions where supported. + case, player LEDs), animation selection, per-light positions where supported, and the light table, which + names every individual light and the control that owns it. A board that keeps no per-light table rebuilds + those records from its per-control configuration and says so, because duplicates cannot be seen that way. * `hlp-fill` fills the LEDs with a colour (`--scope all|buttons|case|pleds`) as a quick visual test, then restores the board's own animations. * `hlp-input-mode` sets the board's input mode and reboots into it. @@ -140,14 +142,33 @@ Sample usage: ``` % hlp-ping Haute42 COSMOX (v0.7.12), board ID 433031343539302E -magic GPHL, protocol 1.0 -100 pings in 601 ms (6.0 ms average) +magic GPHL, protocol 1.1 +100 pings in 207 ms (2.1 ms average) + +% hlp-caps +page 0 (identity): Haute42 COSMOX (v0.7.12), board ID 433031343539302E +page 1 (runtime state): input mode XINPUT, profile 1, brightness step 0, host player 2, map fingerprint 0x92207B00 + offers positions, light table; renders at 40 Hz; animations are stored profiles; LED-refactor LED framework +page 2 (LED map): addresses 16 LEDs, 1 per button, colour format GRB, brightness maximum 100, layout enum 27 + buttons: Up=3, Down=1, Left=0, Right=2, B1=8, B2=9, B3=4, B4=5, L1=7, R1=6, L2=11, R2=10, L3=13, R3=14 +page 3 (animations): index 0 of 1 available +page 4 (positions): 16 lights + LED 0@(0,2), LED 1@(2,2), ... +page 5 (light table): 16 lights, fingerprint 0x92207B00 + LED 0 button Left GP5 at (0,2) + LED 3 button Up GP2 at (5,7) (duplicate) + ... + Up owns 2 lights: LEDs 3, 12 + L3 owns 2 lights: LEDs 13, 15 % hlp-fill --scope case 00FF00 filled case with #00FF00 for 3.0s released - on-board animations restored ``` +The light table is what reveals that this board wires two physical buttons to `Up` and two to `L3`; the LED +map on page 2 can only report one range per control, so it cannot express that. + ### summarize-gp2040ce `summarize-gp2040ce` prints information regarding the provided USB device or file. It attempts to detect the firmware diff --git a/gp2040ce_bintools/hostlighting.py b/gp2040ce_bintools/hostlighting.py index e1241cd..ba3d6ba 100644 --- a/gp2040ce_bintools/hostlighting.py +++ b/gp2040ce_bintools/hostlighting.py @@ -11,11 +11,19 @@ * page 0 (identity): factory-unique board ID, board label, firmware version * page 1 (runtime state): input mode, profile, brightness step, host-assigned - player, LED-map fingerprint, current animation index + player, LED-map fingerprint, current animation index, and from protocol + v1.1 the feature bitmask, LED framework, animation namespace and render rate * page 2 (LED map): totals, colour format, per-button/case/player LED ranges * page 3 (animations): current on-board animation index and how many exist * page 4 (positions): per-light grid positions, where the render pipeline provides them +* page 5 (light table, protocol v1.1): every light the board has, naming the + control that owns each one. This is the page that can say a control owns + more than one light, which page 2's one-range-per-control table cannot. + +Page 2 and page 5 answer different questions. Page 2 is "where do I write this +control"; page 5 is the board's inventory of lights. A host that only colours +canonical controls needs page 2 alone. See docs/host-lighting.md in the GP2040-CE repository for the full protocol reference. These tools require the `hidapi` package (`pip install hidapi`). @@ -62,6 +70,29 @@ CAPS_PAGE_LED_MAP = 2 CAPS_PAGE_ANIMATIONS = 3 CAPS_PAGE_POSITIONS = 4 +CAPS_PAGE_LIGHTS = 5 # protocol v1.1 + +# pages that answer a paged request, taking a start entry in payload byte [3] +PAGED_CAPS_PAGES = (CAPS_PAGE_POSITIONS, CAPS_PAGE_LIGHTS) + +# page 1 fields appended by protocol v1.1. Firmware predating them zero-fills its +# replies, so reading zero here means "not reported" rather than a real value. +FEATURE_POSITIONS = 1 << 0 +FEATURE_LIGHT_TABLE = 1 << 1 +FRAMEWORK_NAMES = {0: 'unreported', 1: 'classic', 2: 'LED-refactor'} +ANIMATION_NAMESPACE_NAMES = {0: 'unreported', 1: 'built-in effects', 2: 'stored profiles'} + +# page 5 geometry. Read the stride from the reply rather than assuming this one: +# it is on the wire precisely so a later record width does not break a decoder. +LIGHT_STRIDE = 12 +LIGHTS_PER_PAGE = 4 +LIGHT_KIND_NAMES = {0: 'button', 1: 'case', 2: 'turbo', + 3: 'player 1', 4: 'player 2', 5: 'player 3', 6: 'player 4'} +# Both flags are positive assertions: a set bit is the board vouching for +# something, so a record left at zero claims nothing. +LIGHT_FLAG_POSITION = 0x01 # the grid coordinates in this record are real +LIGHT_FLAG_PER_LIGHT = 0x02 # read from a per-light table, describes one light +ACTION_NONE = -32768 # no owning GPIO action, distinct from an action of NONE # FILL scopes (payload byte [2] of a FILL request) FILL_SCOPE_ALL = 0x00 @@ -74,10 +105,30 @@ MAGIC_REBOOT_WEBCONFIG = b'WEBC' MAGIC_REBOOT_BOOTSEL = b'BOOT' -# button IDs 0-17 as indexed in the page 2 LED map +# button IDs 0-17 as indexed in the page 2 LED map. Page 2's table is exactly +# these eighteen and can never grow, so this stays a plain list of that length. BUTTON_NAMES = ['Up', 'Down', 'Left', 'Right', 'B1', 'B2', 'B3', 'B4', 'L1', 'R1', 'L2', 'R2', 'S1', 'S2', 'L3', 'R3', 'A1', 'A2'] +# The wider namespace page 5 can report. Protocol v1.1 names A3 and A4 at 18-19 +# and E1-E12 at 30-41; page 2 has no slot for those, so they appear only in the +# light table. 20-23 are permanently unassigned - those gamepad bits are the +# dpad in a second encoding rather than four more controls - so they are absent +# here on purpose and resolve to the same "unknown" as any other stray value. +EXTENDED_BUTTON_NAMES = {18: 'A3', 19: 'A4', + 24: 'Player 1', 25: 'Player 2', 26: 'Player 3', 27: 'Player 4', + 28: 'Turbo', 29: 'Case'} +EXTENDED_BUTTON_NAMES.update({30 + n: f'E{n + 1}' for n in range(12)}) + +BUTTON_NONE = 0xFF # no button ID names this light + +# Controls that own a whole strip rather than one lamp. Several lights sharing +# one of these is the normal shape of the board, not something a reader needs +# flagging: a case strip is routinely dozens of lights. The duplicate marker is +# reserved for a control that should own exactly one and does not, which is the +# case worth looking at. +MULTI_LIGHT_BUTTONS = frozenset({29}) # case + LED_FORMAT_NAMES = {0: 'GRB', 1: 'RGB', 2: 'GRBW', 3: 'RGBW'} INPUT_MODE_NAMES = {0: 'XINPUT', 3: 'KEYBOARD', 14: 'GENERIC'} @@ -109,6 +160,149 @@ def match_reply(reply: bytes, command: int, sequence: int) -> bool: return len(reply) >= 3 and reply[0] == (command | RESPONSE_FLAG) and reply[1] == sequence +def button_name(button_id: int) -> str: + """Name a protocol button ID, across the whole namespace the light table can report. + + :param button_id: a button ID as reported on page 2 or page 5 + :return: the control's name, 'none' where no ID names the light, or + 'unknown (N)' for a value this version of the protocol does not name + """ + if button_id == BUTTON_NONE: + return 'none' + if button_id < len(BUTTON_NAMES): + return BUTTON_NAMES[button_id] + return EXTENDED_BUTTON_NAMES.get(button_id, f'unknown ({button_id})') + + +def light_kind_name(kind: int) -> str: + """Name a light kind from a page 5 record. + + :param kind: the light kind byte + :return: the kind's name, or 'unknown (N)' for anything unrecognised + """ + return LIGHT_KIND_NAMES.get(kind, f'unknown ({kind})') + + +def decode_state(reply: bytes) -> dict: + """Decode GET_CAPS page 1, the board's runtime state. + + Fields from `features` onward were appended by protocol v1.1. Replies are + zero-filled before the board builds them, so firmware predating those fields + reports them as zero, which reads correctly as nothing supported and nothing + stated rather than as a real value. + + :param reply: a page 1 reply report + :return: the decoded fields, with `render_hz` None when the board did not say + """ + return { + 'input_mode': reply[3], + 'profile': reply[4], + 'brightness_step': reply[5], + 'host_player': reply[6], + 'fingerprint': int.from_bytes(reply[7:11], 'little'), + # 0xFF is 'none selected' - the lights-off state some pipelines persist + 'animation_index': None if reply[11] == UNMAPPED else reply[11], + 'features': int.from_bytes(reply[12:16], 'little'), + 'framework': reply[16], + 'animation_namespace': reply[17], + 'render_hz': reply[18] or None, + } + + +def decode_led_map(reply: bytes) -> dict: + """Decode GET_CAPS page 2, the per-control LED map. + + Page 2 answers "where do I write this control" for the eighteen canonical + controls. It is not an inventory of the board's lights: a control may own + several lights and only the first appears here, and a board may carry lights + on controls this page cannot name. Read page 5 for the full picture. + + `led_extent` is the highest LED index in use plus one - the number to size a + frame buffer from. It is not a light count: it is deliberately not the sum of + the ranges below it, because that sum omits the lights page 2 cannot name. + + :param reply: a page 2 reply report + :return: the decoded fields, `buttons` mapping button ID to (first LED, count) + and `players` mapping player number to LED index + """ + buttons = {} + for index in range(len(BUTTON_NAMES)): + first, count = reply[8 + index * 2], reply[9 + index * 2] + if first != UNMAPPED: + buttons[index] = (first, count) + # The four player slots are positional, so an unmapped one has to drop out + # by key rather than by position: filtering them into a list would slide + # player 2's index into player 1's place whenever player 1 is absent. + players = {slot + 1: reply[44 + slot] for slot in range(4) + if reply[44 + slot] != UNMAPPED} + return { + 'leds_per_button': reply[3], + 'colour_format': reply[4], + 'layout': reply[5], + 'led_extent': reply[6], + 'brightness_maximum': reply[7], + 'buttons': buttons, + 'players': players, + 'turbo': None if reply[48] == UNMAPPED else reply[48], + 'case': None if reply[49] == UNMAPPED or not reply[50] else (reply[49], reply[50]), + 'fingerprint': int.from_bytes(reply[51:55], 'little'), + } + + +def decode_positions(reply: bytes) -> tuple[int, list]: + """Decode one page of GET_CAPS page 4, the per-light grid positions. + + :param reply: a page 4 reply report + :return: (total entries the board has, the entries in this reply as + (first LED, x, y) triples) + """ + total, count = reply[3], reply[4] + entries = [(reply[5 + n * 3], reply[6 + n * 3], reply[7 + n * 3]) for n in range(count)] + return total, entries + + +def decode_lights(reply: bytes) -> dict: + """Decode one page of GET_CAPS page 5, the light table. + + The stride is read from the reply rather than assumed: it is on the wire so + that a later, wider record does not silently misalign an older decoder. A + record is identified by its ordinal, not by its first LED, which is not a + unique key - boards exist with two lights starting at the same index. + + :param reply: a page 5 reply report + :return: the page header plus its decoded records + """ + total, start, count, stride = reply[3], reply[4], reply[5], reply[6] + records = [] + for n in range(count): + base = 7 + n * stride + record = reply[base:base + stride] + flags = record[11] + records.append({ + 'ordinal': start + n, + 'first_led': record[0], + 'led_count': record[1], + 'kind': record[2], + 'button_id': record[3], + 'gpio_pin': None if record[4] == UNMAPPED else record[4], + 'gpio_action': int.from_bytes(record[5:7], 'little', signed=True), + 'player_index': None if record[7] == UNMAPPED else record[7], + 'case_group': None if record[8] == UNMAPPED else record[8], + 'position': (record[9], record[10]) if flags & LIGHT_FLAG_POSITION else None, + # Reported as the caveat rather than the assertion: a caller wants + # to know when a record cannot show duplicates, and that is the + # absence of the board's per-light claim + 'synthesised': not (flags & LIGHT_FLAG_PER_LIGHT), + }) + return { + 'total': total, + 'start': start, + 'stride': stride, + 'records': records, + 'fingerprint': int.from_bytes(reply[60:64], 'little'), + } + + class HostLightingError(RuntimeError): """Errors talking to a Host Lighting interface.""" @@ -177,13 +371,55 @@ def get_caps_page(self, page: int, start_entry: int = 0) -> bytes: """Read one GET_CAPS page. :param page: which capability page to read (CAPS_PAGE_* constant) - :param start_entry: first entry to return, for the paged positions page + :param start_entry: first entry to return, for the paged pages :return: the reply report ([3..] is the page's payload) """ - if page == CAPS_PAGE_POSITIONS: + if page in PAGED_CAPS_PAGES: return self.request_ok(CMD_GET_CAPS, bytes([page, start_entry])) return self.request_ok(CMD_GET_CAPS, bytes([page])) + def read_positions(self) -> list: + """Read every page 4 entry, walking the pages. + + :return: the board's (first LED, x, y) triples, empty when it has none + :raises HostLightingError: if the board stops making progress + """ + entries: list = [] + while True: + total, page = decode_positions(self.get_caps_page(CAPS_PAGE_POSITIONS, len(entries))) + if not page: + if len(entries) < total: + raise HostLightingError( + f"positions stalled at {len(entries)} of {total} entries") + return entries + entries.extend(page) + if len(entries) >= total: + return entries + + def read_lights(self) -> tuple[list, int]: + """Read every page 5 record, walking the pages. + + The fingerprint is returned alongside so a caller can tell whether the + map changed underneath a walk that took several reads; if it moves, the + partial table is not coherent and the walk should be repeated. + + :return: (the board's light records, the fingerprint of the last reply) + :raises HostLightingError: if the board stops making progress + """ + records: list = [] + fingerprint = 0 + while True: + page = decode_lights(self.get_caps_page(CAPS_PAGE_LIGHTS, len(records))) + fingerprint = page['fingerprint'] + if not page['records']: + if len(records) < page['total']: + raise HostLightingError( + f"light table stalled at {len(records)} of {page['total']} records") + return records, fingerprint + records.extend(page['records']) + if len(records) >= page['total']: + return records, fingerprint + def _import_hid(): """Import the hidapi module, with a helpful error if it is missing.""" @@ -332,6 +568,7 @@ def caps(): _print_led_map(device) _print_animations(device) _print_positions(device) + _print_lights(device) finally: device.close() @@ -341,7 +578,8 @@ def _print_state(device: HostLightingDevice) -> None: Page 1 reply layout: [3] input mode, [4] profile, [5] brightness step, [6] host-assigned player (0 = none), [7..10] LED-map fingerprint (little - endian), [11] current animation index. + endian), [11] current animation index, [12..15] feature bitmask, + [16] LED framework, [17] animation namespace, [18] render rate in Hz. Byte [5] is a step index into the board's brightness steps, not a 0-255 level, and HLP does not report how many steps there are. That count varies @@ -352,18 +590,55 @@ def _print_state(device: HostLightingDevice) -> None: :param device: an opened HostLightingDevice """ - reply = device.get_caps_page(CAPS_PAGE_STATE) - fingerprint = int.from_bytes(reply[7:11], 'little') - mode = INPUT_MODE_NAMES.get(reply[3], str(reply[3])) - print(f"page 1 (runtime state): input mode {mode}, profile {reply[4]}, brightness step {reply[5]}, " - f"host player {reply[6]}, map fingerprint 0x{fingerprint:08X}") + state = decode_state(device.get_caps_page(CAPS_PAGE_STATE)) + mode = INPUT_MODE_NAMES.get(state['input_mode'], str(state['input_mode'])) + print(f"page 1 (runtime state): input mode {mode}, profile {state['profile']}, " + f"brightness step {state['brightness_step']}, host player {state['host_player']}, " + f"map fingerprint 0x{state['fingerprint']:08X}") + + # Everything below arrived with protocol v1.1 and reads as zero on older + # firmware, so say nothing rather than reporting a value the board never sent. + # + # The framework byte and animation namespace both reserve 0 for "not + # reported", so either is a witness for "this board speaks v1.1". The + # feature bitmask cannot serve that purpose, however tempting - a v1.1 board + # legitimately reports no features when its light registry has not populated + # yet. + # A value this build does not recognise is reported as unknown rather than + # printed as None or dropped: a later firmware naming a third namespace + # should read as a board saying something new, not as a decoder fault. + speaks_v11 = bool(state['animation_namespace']) or bool(state['framework']) + framework = FRAMEWORK_NAMES.get(state['framework'], f"unknown ({state['framework']})") + namespace = ANIMATION_NAMESPACE_NAMES.get(state['animation_namespace'], + f"unknown ({state['animation_namespace']})") + details = [] + available = [name for bit, name in ((FEATURE_POSITIONS, 'positions'), + (FEATURE_LIGHT_TABLE, 'light table')) + if state['features'] & bit] + unnamed = state['features'] & ~(FEATURE_POSITIONS | FEATURE_LIGHT_TABLE) + if unnamed: + available.append(f"unknown bits 0x{unnamed:02X}") + # Unnamed bits are folded into the list above, so this is empty exactly + # when the mask is zero and the line vanishes rather than printing "offers" + # over nothing + if available: + details.append('offers ' + ', '.join(available)) + if state['render_hz']: + details.append(f"renders at {state['render_hz']} Hz") + if speaks_v11: + details.append(f"animations are {namespace}") + if state['framework']: + details.append(f"{framework} LED framework") + if details: + print(" " + '; '.join(details)) def _print_led_map(device: HostLightingDevice) -> None: """Read and print GET_CAPS page 2, the LED map. Page 2 reply layout: [3] LEDs per button, [4] LED colour format, - [5] button layout enum, [6] total LED count, [7] brightness maximum, + [5] button layout enum, [6] LED extent (highest index in use plus one, + not a light count), [7] brightness maximum, [8..43] per-button {first LED, count} pairs for button IDs 0-17 (first = 0xFF means unmapped), [44..47] player LED indexes, [48] turbo LED index, [49..50] case strip {first LED, count}, [51..54] LED-map @@ -371,30 +646,35 @@ def _print_led_map(device: HostLightingDevice) -> None: :param device: an opened HostLightingDevice """ - reply = device.get_caps_page(CAPS_PAGE_LED_MAP) - colour = LED_FORMAT_NAMES.get(reply[4], str(reply[4])) - print(f"page 2 (LED map): {reply[6]} LEDs total, {reply[3]} per button, colour format {colour}, " - f"brightness maximum {reply[7]}, layout enum {reply[5]}") - entries = [] - for index, name in enumerate(BUTTON_NAMES): - first, count = reply[8 + index * 2], reply[9 + index * 2] - if first != UNMAPPED and count: - entries.append(f"{name}={first}" + (f"+{count}" if count > 1 else '')) + led_map = decode_led_map(device.get_caps_page(CAPS_PAGE_LED_MAP)) + colour = LED_FORMAT_NAMES.get(led_map['colour_format'], str(led_map['colour_format'])) + print(f"page 2 (LED map): addresses {led_map['led_extent']} LEDs, {led_map['leds_per_button']} per button, " + f"colour format {colour}, brightness maximum {led_map['brightness_maximum']}, " + f"layout enum {led_map['layout']}") + entries = [f"{button_name(button)}={first}" + (f"+{count}" if count > 1 else '') + for button, (first, count) in sorted(led_map['buttons'].items()) if count] print(" buttons: " + (', '.join(entries) if entries else 'none mapped')) - if reply[50]: - print(f" case strip: LEDs {reply[49]}..{reply[49] + reply[50] - 1}") + if led_map['players']: + print(" player LEDs: " + ', '.join(f"P{player}={index}" + for player, index in sorted(led_map['players'].items()))) + if led_map['turbo'] is not None: + print(f" turbo LED: {led_map['turbo']}") + if led_map['case']: + first, count = led_map['case'] + print(f" case strip: LEDs {first}..{first + count - 1}") def _print_animations(device: HostLightingDevice) -> None: """Read and print GET_CAPS page 3, the on-board animation selection. - Page 3 reply layout: [3] current animation index, [4] number of - animations the board offers (board-specific). + Page 3 reply layout: [3] current animation index (0xFF = none selected), + [4] number of animations the board offers (board-specific). :param device: an opened HostLightingDevice """ reply = device.get_caps_page(CAPS_PAGE_ANIMATIONS) - print(f"page 3 (animations): index {reply[3]} of {reply[4]} available") + current = 'none selected' if reply[3] == UNMAPPED else f"index {reply[3]}" + print(f"page 3 (animations): {current} of {reply[4]} available") def _print_positions(device: HostLightingDevice) -> None: @@ -407,9 +687,92 @@ def _print_positions(device: HostLightingDevice) -> None: :param device: an opened HostLightingDevice """ - reply = device.get_caps_page(CAPS_PAGE_POSITIONS) - total = reply[3] - print(f"page 4 (positions): {total if total else 'none (this render pipeline has no per-light positions)'}") + entries = device.read_positions() + if not entries: + print("page 4 (positions): none (this render pipeline has no per-light positions)") + return + print(f"page 4 (positions): {len(entries)} lights") + print(" " + ', '.join(f"LED {first}@({x},{y})" for first, x, y in entries)) + + +def _is_unexpected_duplicate(button_id: int, owners: dict) -> bool: + """Say whether a control owning several lights is worth flagging. + + :param button_id: the record's button ID + :param owners: records grouped by button ID + :return: True when a one-lamp control owns more than one light + """ + if button_id == BUTTON_NONE or button_id in MULTI_LIGHT_BUTTONS: + return False + return len(owners[button_id]) > 1 + + +def _describe_light(record: dict, owners: dict) -> str: + """Render one light-table record as a single human-readable line. + + Every field the board declines to supply is simply left out, so a row shows + what is known rather than a column of sentinels. + + :param record: one decoded page 5 record + :param owners: records grouped by button ID, used to flag duplicates + :return: the formatted line, without indentation + """ + pieces = [f"LED {record['first_led']}"] + if record['led_count'] > 1: + pieces.append(f"x{record['led_count']}") + pieces.append(light_kind_name(record['kind'])) + pieces.append(button_name(record['button_id'])) + if record['gpio_pin'] is not None: + pieces.append(f"GP{record['gpio_pin']}") + if record['position'] is not None: + pieces.append("at (%d,%d)" % record['position']) + if _is_unexpected_duplicate(record['button_id'], owners): + pieces.append('(duplicate)') + return ' '.join(pieces) + + +def _print_lights(device: HostLightingDevice) -> None: + """Read and print GET_CAPS page 5, the light table. + + Page 5 lists every light the board has, naming the control that owns each + one. Where a control owns several lights, several records carry the same + button ID - that is the page's purpose, and it is what page 2 cannot say. + + Boards that keep no per-light table rebuild these records from their + per-control configuration and mark them synthesised: the rows are then + per-control, so duplicates cannot appear however many the board really has. + + Older firmware answers an unknown page with INVALID_ARG, which is how a host + tells "this board predates the light table" from "this board has none". + + :param device: an opened HostLightingDevice + """ + try: + records, fingerprint = device.read_lights() + except HostLightingRejected: + print("page 5 (light table): not supported by this firmware") + return + + if not records: + print("page 5 (light table): no lights reported") + return + + synthesised = any(record['synthesised'] for record in records) + note = ', rebuilt from per-control config' if synthesised else '' + print(f"page 5 (light table): {len(records)} lights, fingerprint 0x{fingerprint:08X}{note}") + + owners: dict = {} + for record in records: + owners.setdefault(record['button_id'], []).append(record) + + for record in records: + print(" " + _describe_light(record, owners)) + + duplicated = {button: found for button, found in owners.items() + if _is_unexpected_duplicate(button, owners)} + for button, found in sorted(duplicated.items()): + leds = ', '.join(str(record['first_led']) for record in found) + print(f" {button_name(button)} owns {len(found)} lights: LEDs {leds}") def fill(): diff --git a/tests/test_hostlighting.py b/tests/test_hostlighting.py index ee5f51c..f834693 100644 --- a/tests/test_hostlighting.py +++ b/tests/test_hostlighting.py @@ -3,6 +3,9 @@ SPDX-FileCopyrightText: © 2026 Jacob Simpson SPDX-License-Identifier: GPL-3.0-or-later """ +import contextlib +import io + import pytest from gp2040ce_bintools import hostlighting @@ -89,3 +92,563 @@ def test_send_reboot_still_raises_on_rejection(): device = _StubDevice(hostlighting.HostLightingRejected('command 0x7F rejected: INVALID_ARG')) with pytest.raises(hostlighting.HostLightingRejected): hostlighting.send_reboot(device, hostlighting.CMD_REBOOT_BOOTSEL, b'XXXX') + + +def _reply(payload: bytes = b'', tail: bytes = b'', command: int = hostlighting.CMD_GET_CAPS) -> bytes: + """Build a 64-byte reply: header, payload from [3], optional tail at the end.""" + report = bytearray(hostlighting.REPORT_SIZE) + report[0] = command | hostlighting.RESPONSE_FLAG + report[2] = 0x00 + report[3:3 + len(payload)] = payload + if tail: + report[hostlighting.REPORT_SIZE - len(tail):] = tail + return bytes(report) + + +def _light_record(first_led=0, led_count=1, kind=0, button_id=0, gpio_pin=0xFF, + gpio_action=hostlighting.ACTION_NONE, player=0xFF, case_group=0xFF, + x=0, y=0, flags=0) -> bytes: + """Build one 12-byte page 5 record.""" + return (bytes([first_led, led_count, kind, button_id, gpio_pin]) + + int(gpio_action).to_bytes(2, 'little', signed=True) + + bytes([player, case_group, x, y, flags])) + + +def _lights_reply(total, records, start=0, fingerprint=0, stride=hostlighting.LIGHT_STRIDE) -> bytes: + """Build a page 5 reply carrying the given records.""" + header = bytes([total, start, len(records), stride]) + return _reply(header + b''.join(records), fingerprint.to_bytes(4, 'little')) + + +# --- naming across the whole button namespace ------------------------------- + +def test_button_name_covers_the_canonical_eighteen(): + """Test that the page 2 controls are named from their table.""" + assert hostlighting.button_name(0) == 'Up' + assert hostlighting.button_name(14) == 'L3' + assert hostlighting.button_name(17) == 'A2' + + +def test_button_name_covers_the_extended_controls(): + """Test the IDs protocol v1.1 added, which only page 5 can report.""" + assert hostlighting.button_name(18) == 'A3' + assert hostlighting.button_name(19) == 'A4' + assert hostlighting.button_name(30) == 'E1' + assert hostlighting.button_name(41) == 'E12' + + +def test_button_name_covers_the_specials(): + """Test the player, turbo and case IDs.""" + assert hostlighting.button_name(24) == 'Player 1' + assert hostlighting.button_name(27) == 'Player 4' + assert hostlighting.button_name(28) == 'Turbo' + assert hostlighting.button_name(29) == 'Case' + + +def test_button_name_leaves_the_dpad_alias_range_unnamed(): + """Test that 20-23 stay unnamed. + + Those gamepad bits are the dpad in a second encoding rather than four more + controls, so the protocol never assigns them. Naming one would give Up a + second identity. + """ + for button_id in range(20, 24): + assert hostlighting.button_name(button_id).startswith('unknown') + + +def test_button_name_reports_unnameable_and_unknown_distinctly(): + """Test that "no owner" reads differently from "an ID we do not know".""" + assert hostlighting.button_name(hostlighting.BUTTON_NONE) == 'none' + assert hostlighting.button_name(200) == 'unknown (200)' + + +def test_light_kind_name_covers_every_kind(): + """Test that each light kind is named, and anything else is flagged.""" + assert hostlighting.light_kind_name(0) == 'button' + assert hostlighting.light_kind_name(1) == 'case' + assert hostlighting.light_kind_name(2) == 'turbo' + assert hostlighting.light_kind_name(6) == 'player 4' + assert hostlighting.light_kind_name(0xFF) == 'unknown (255)' + + +# --- page 1, runtime state -------------------------------------------------- + +def test_decode_state_reads_the_original_fields(): + """Test the fields page 1 has carried since 1.0.""" + payload = bytes([14, 2, 3, 1]) + (0xDEADBEEF).to_bytes(4, 'little') + bytes([5]) + state = hostlighting.decode_state(_reply(payload)) + assert state['input_mode'] == 14 + assert state['profile'] == 2 + assert state['brightness_step'] == 3 + assert state['host_player'] == 1 + assert state['fingerprint'] == 0xDEADBEEF + assert state['animation_index'] == 5 + + +def test_decode_state_reads_the_fields_added_in_1_1(): + """Test the feature bitmask, framework, animation namespace and render rate.""" + payload = (bytes([0, 0, 0, 0]) + bytes(4) + bytes([0]) + + (hostlighting.FEATURE_POSITIONS | hostlighting.FEATURE_LIGHT_TABLE).to_bytes(4, 'little') + + bytes([2, 2, 40])) + state = hostlighting.decode_state(_reply(payload)) + assert state['features'] == 3 + assert state['framework'] == 2 + assert state['animation_namespace'] == 2 + assert state['render_hz'] == 40 + + +def test_decode_state_reports_no_selected_animation_as_absent(): + """Test that the 0xFF animation index reads as none rather than as 255. + + Some pipelines persist a lights-off state with no profile selected; a host + indexing its animation list by 255 would read far out of bounds. + """ + payload = bytes([0, 0, 0, 0]) + bytes(4) + bytes([0xFF]) + assert hostlighting.decode_state(_reply(payload))['animation_index'] is None + + +def test_decode_state_treats_a_1_0_board_as_reporting_nothing(): + """Test that zero-filled trailing bytes read as absent, not as real values. + + Firmware predating 1.1 zeroes its whole reply, so the appended fields arrive + as zero. That has to mean "not stated" rather than "no features, classic + framework, zero hertz", or a host would act on values the board never sent. + """ + state = hostlighting.decode_state(_reply(bytes([14, 0, 0, 0]) + bytes(4) + bytes([0]))) + assert state['features'] == 0 + assert state['render_hz'] is None + + +# --- page 2, the LED map ---------------------------------------------------- + +def _led_map_reply(buttons, case=None, players=(), turbo=None, fingerprint=0): + """Build a page 2 reply from {button_id: (first, count)}.""" + body = bytearray([1, 0, 0, 46, 255]) + table = bytearray() + for button_id in range(len(hostlighting.BUTTON_NAMES)): + first, count = buttons.get(button_id, (0xFF, 0)) + table += bytes([first, count]) + body += table + body += bytes(list(players) + [0xFF] * (4 - len(players))) + body += bytes([0xFF if turbo is None else turbo]) + body += bytes(case if case else [0xFF, 0]) + body += fingerprint.to_bytes(4, 'little') + return _reply(bytes(body)) + + +def test_decode_led_map_reads_mapped_controls_and_skips_unmapped(): + """Test that only controls with a light appear, keyed by button ID.""" + led_map = hostlighting.decode_led_map(_led_map_reply({0: (3, 1), 14: (13, 1)})) + assert led_map['buttons'] == {0: (3, 1), 14: (13, 1)} + assert led_map['led_extent'] == 46 + + +def test_decode_led_map_reads_the_case_range_and_fingerprint(): + """Test the case strip and the fingerprint that pairs page 2 with page 1.""" + led_map = hostlighting.decode_led_map(_led_map_reply({}, case=[16, 30], fingerprint=0x11223344)) + assert led_map['case'] == (16, 30) + assert led_map['fingerprint'] == 0x11223344 + + +def test_decode_led_map_reports_absent_specials_as_none(): + """Test that an unmapped turbo or empty case reads as absent rather than 255.""" + led_map = hostlighting.decode_led_map(_led_map_reply({})) + assert led_map['turbo'] is None + assert led_map['case'] is None + assert led_map['players'] == {} + + +def test_decode_led_map_keeps_player_slots_positional(): + """Test that a gap in the player slots does not slide later players down.""" + led_map = hostlighting.decode_led_map(_led_map_reply({}, players=(0xFF, 7, 0xFF, 9))) + assert led_map['players'] == {2: 7, 4: 9} + + +# --- page 4, positions ------------------------------------------------------ + +def test_decode_positions_reads_its_entries(): + """Test that a positions page yields its triples.""" + total, entries = hostlighting.decode_positions( + _reply(bytes([2, 2]) + bytes([0, 4, 4]) + bytes([1, 6, 4]))) + assert total == 2 + assert entries == [(0, 4, 4), (1, 6, 4)] + + +def test_decode_positions_handles_a_board_with_none(): + """Test the empty answer a pipeline without per-light positions gives.""" + total, entries = hostlighting.decode_positions(_reply(bytes([0, 0]))) + assert (total, entries) == (0, []) + + +# --- page 5, the light table ------------------------------------------------ + +def test_decode_lights_reads_a_record(): + """Test that every field of a record is decoded.""" + record = _light_record(first_led=12, led_count=2, kind=0, button_id=0, gpio_pin=27, + gpio_action=1, x=7, y=2, + flags=hostlighting.LIGHT_FLAG_POSITION | hostlighting.LIGHT_FLAG_PER_LIGHT) + page = hostlighting.decode_lights(_lights_reply(1, [record], fingerprint=0xABCD1234)) + decoded = page['records'][0] + assert decoded['first_led'] == 12 + assert decoded['led_count'] == 2 + assert decoded['button_id'] == 0 + assert decoded['gpio_pin'] == 27 + assert decoded['gpio_action'] == 1 + assert decoded['position'] == (7, 2) + assert decoded['synthesised'] is False + assert page['fingerprint'] == 0xABCD1234 + + +def test_decode_lights_reports_sentinels_as_absent(): + """Test that the 0xFF sentinels decode to None rather than to 255.""" + page = hostlighting.decode_lights(_lights_reply(1, [_light_record()])) + decoded = page['records'][0] + assert decoded['gpio_pin'] is None + assert decoded['player_index'] is None + assert decoded['case_group'] is None + + +def test_decode_lights_treats_a_clear_position_flag_as_no_position(): + """Test that (0,0) is not mistaken for a real coordinate. + + Positions are origin-normalised, so some light always sits at (0,0) on a + board that has them. Only the flag can say whether the pair means anything. + """ + page = hostlighting.decode_lights(_lights_reply(1, [_light_record(x=0, y=0, flags=0)])) + assert page['records'][0]['position'] is None + + +def test_decode_lights_reads_a_negative_gpio_action(): + """Test that the action travels as a signed value. + + GpioAction has negative members, and the protocol sends it verbatim so it + never needs a new protocol allocation when GP2040-CE adds one. + """ + page = hostlighting.decode_lights(_lights_reply(1, [_light_record(gpio_action=-10)])) + assert page['records'][0]['gpio_action'] == -10 + + +def test_decode_lights_marks_a_record_without_the_per_light_flag_synthesised(): + """Test that a record which does not assert a per-light table reads as rebuilt.""" + record = _light_record(flags=hostlighting.LIGHT_FLAG_PER_LIGHT) + page = hostlighting.decode_lights(_lights_reply(1, [record])) + assert page['records'][0]['synthesised'] is False + + record = _light_record(flags=0) + page = hostlighting.decode_lights(_lights_reply(1, [record])) + assert page['records'][0]['synthesised'] is True + + +def test_decode_lights_treats_an_unasserted_record_as_the_weaker_case(): + """Test that a record claiming nothing is not read as a genuine light table. + + The flags are positive assertions precisely so an all-zero record - which a + partial implementation could easily emit - degrades to "rebuilt, no + positions" rather than passing as a real per-light table. + """ + page = hostlighting.decode_lights(_lights_reply(1, [_light_record(flags=0)])) + decoded = page['records'][0] + assert decoded['synthesised'] is True + assert decoded['position'] is None + + +def test_decode_lights_honours_the_stride_from_the_wire(): + """Test that a wider record is walked by its stated stride, not an assumed one. + + The stride is on the wire so that a later, wider record does not silently + misalign a decoder written against today's width. + """ + wide = [_light_record(first_led=n) + bytes([0, 0]) for n in (5, 9)] + page = hostlighting.decode_lights(_lights_reply(2, wide, stride=hostlighting.LIGHT_STRIDE + 2)) + assert [record['first_led'] for record in page['records']] == [5, 9] + + +def test_decode_lights_numbers_records_from_the_echoed_start(): + """Test that ordinals continue across pages. + + A record's identity is its ordinal: the first LED index is not unique, since + boards exist with two lights starting at the same index. + """ + page = hostlighting.decode_lights( + _lights_reply(6, [_light_record(first_led=0), _light_record(first_led=0)], start=4)) + assert [record['ordinal'] for record in page['records']] == [4, 5] + + +def test_decode_lights_expresses_a_duplicated_control(): + """Test the case the page exists for: one control owning two lights. + + On a Haute42 COSMOX M Ultra Gen 2 the second Up sits on LED 12 while the + first is on LED 3. Page 2 can only name one of them; here both appear with + the same button ID, which needs no special case in a host. + """ + records = [_light_record(first_led=3, button_id=0, gpio_pin=2), + _light_record(first_led=12, button_id=0, gpio_pin=27)] + page = hostlighting.decode_lights(_lights_reply(2, records)) + assert [record['button_id'] for record in page['records']] == [0, 0] + assert [record['first_led'] for record in page['records']] == [3, 12] + + +# --- paged reads ------------------------------------------------------------ + +class _PagingDevice: + """Stand-in device serving a scripted sequence of capability pages.""" + + def __init__(self, pages): + self.pages = list(pages) + self.requests = [] + + def get_caps_page(self, page, start_entry=0): + """Hand back the next scripted page, recording what was asked for.""" + self.requests.append((page, start_entry)) + return self.pages.pop(0) + + +def test_read_lights_walks_every_page(): + """Test that a table spanning several reads is assembled in order.""" + first = _lights_reply(6, [_light_record(first_led=n) for n in range(4)], start=0) + second = _lights_reply(6, [_light_record(first_led=n) for n in (4, 5)], start=4) + device = _PagingDevice([first, second]) + records, _ = hostlighting.HostLightingDevice.read_lights(device) + assert [record['first_led'] for record in records] == [0, 1, 2, 3, 4, 5] + assert device.requests == [(hostlighting.CAPS_PAGE_LIGHTS, 0), (hostlighting.CAPS_PAGE_LIGHTS, 4)] + + +def test_read_lights_stops_on_an_empty_page(): + """Test that a board reporting no lights terminates the walk immediately.""" + device = _PagingDevice([_lights_reply(0, [])]) + records, _ = hostlighting.HostLightingDevice.read_lights(device) + assert records == [] + assert len(device.requests) == 1 + + +def test_read_positions_walks_every_page(): + """Test that page 4 is paged the same way.""" + first = _reply(bytes([3, 2]) + bytes([0, 1, 1]) + bytes([1, 2, 2])) + second = _reply(bytes([3, 1]) + bytes([2, 3, 3])) + device = _PagingDevice([first, second]) + entries = hostlighting.HostLightingDevice.read_positions(device) + assert entries == [(0, 1, 1), (1, 2, 2), (2, 3, 3)] + + +# --- request framing for the paged pages ------------------------------------ + +class _StallingDevice(hostlighting.HostLightingDevice): + """A device whose light table claims more records than it ever serves.""" + + def __init__(self, pages): + self.pages = list(pages) + + def request_ok(self, command, payload=b'', timeout=0.5): + """Serve the scripted page 5 replies in order.""" + return self.pages.pop(0) + + +def test_read_lights_raises_when_the_walk_stalls(): + """Test that a truncated table raises rather than passing as complete. + + A board that reports a total but answers a mid-walk request with no + records would otherwise hand the caller a quarter of the inventory as if + it were all of it. + """ + first = _lights_reply(16, [_light_record(n) for n in range(4)]) + stalled = _lights_reply(16, [], start=4) + with pytest.raises(hostlighting.HostLightingError): + _StallingDevice([first, stalled]).read_lights() + + +def test_read_lights_accepts_an_empty_table(): + """Test that a board with no lights returns cleanly rather than raising.""" + records, _ = _StallingDevice([_lights_reply(0, [])]).read_lights() + assert records == [] + + +def test_read_lights_raises_when_the_first_page_is_already_empty(): + """Test that a nonzero total with no records at all raises too. + + Zero collected records is still a stall, not a boards-got-no-lights case: + the board itself claims a total it then never serves. + """ + with pytest.raises(hostlighting.HostLightingError): + _StallingDevice([_lights_reply(16, [])]).read_lights() + + +def _positions_reply(total, triples, start=0): + """Build a page 4 reply carrying the given (led, x, y) triples.""" + body = bytes([total, len(triples)]) + for led, x, y in triples: + body += bytes([led, x, y]) + return _reply(body) + + +def test_read_positions_raises_when_the_walk_stalls(): + """Test that a truncated position walk raises like the light table's.""" + first = _positions_reply(8, [(n, 0, 0) for n in range(4)]) + stalled = _positions_reply(8, [], start=4) + with pytest.raises(hostlighting.HostLightingError): + _StallingDevice([first, stalled]).read_positions() + + +def test_read_positions_accepts_a_board_with_none(): + """Test that no positions returns an empty list rather than raising.""" + assert _StallingDevice([_positions_reply(0, [])]).read_positions() == [] + + +def test_print_animations_renders_the_none_selected_sentinel(): + """Test that 0xFF prints as none selected rather than as index 255.""" + buffer = io.StringIO() + with contextlib.redirect_stdout(buffer): + hostlighting._print_animations(_StallingDevice([_reply(bytes([0xFF, 2]))])) + assert 'none selected of 2 available' in buffer.getvalue() + buffer = io.StringIO() + with contextlib.redirect_stdout(buffer): + hostlighting._print_animations(_StallingDevice([_reply(bytes([1, 4]))])) + assert 'index 1 of 4 available' in buffer.getvalue() + + +class _RecordingDevice(hostlighting.HostLightingDevice): + """A device that records the payloads it would have sent.""" + + def __init__(self): + self.payloads = [] + + def request_ok(self, command, payload=b'', timeout=0.5): + """Record the payload and return an empty page.""" + self.payloads.append(payload) + return bytes(hostlighting.REPORT_SIZE) + + +@pytest.mark.parametrize('page', hostlighting.PAGED_CAPS_PAGES) +def test_get_caps_page_sends_a_start_entry_for_paged_pages(page): + """Test that the paged pages carry their start entry in the request.""" + device = _RecordingDevice() + device.get_caps_page(page, 7) + assert device.payloads == [bytes([page, 7])] + + +@pytest.mark.parametrize('page', [hostlighting.CAPS_PAGE_IDENTITY, hostlighting.CAPS_PAGE_STATE, + hostlighting.CAPS_PAGE_LED_MAP, hostlighting.CAPS_PAGE_ANIMATIONS]) +def test_get_caps_page_sends_no_start_entry_for_single_pages(page): + """Test that a page answered in one report is asked for by number alone.""" + device = _RecordingDevice() + device.get_caps_page(page) + assert device.payloads == [bytes([page])] + + +# --- light table rendering -------------------------------------------------- + +def _decoded_light(first, count=1, kind=0, button=0xFF, gpio=None, position=None): + """Build one decoded page 5 record, as decode_lights would return it.""" + return {'ordinal': first, 'first_led': first, 'led_count': count, 'kind': kind, + 'button_id': button, 'gpio_pin': gpio, 'gpio_action': 0, + 'player_index': None, 'case_group': None, + 'position': position, 'synthesised': False} + + +def test_describe_light_flags_a_one_lamp_control_owning_two(): + """Test that a button with a second light is called out as a duplicate.""" + records = [_decoded_light(3, button=0), _decoded_light(12, button=0)] + owners = {0: records} + assert '(duplicate)' in hostlighting._describe_light(records[0], owners) + + +def test_describe_light_does_not_flag_a_case_strip(): + """Test that a case strip's many lights are normal rather than duplicates. + + A case strip is routinely dozens of lights; marking every one would bury the + button duplicates the marker exists to surface. + """ + records = [_decoded_light(led, kind=1, button=29) for led in range(16, 46)] + owners = {29: records} + rendered = hostlighting._describe_light(records[0], owners) + assert '(duplicate)' not in rendered + assert 'Case' in rendered + + +def test_describe_light_omits_fields_the_board_did_not_supply(): + """Test that absent GPIO and position produce no placeholder text.""" + record = _decoded_light(4, button=1) + rendered = hostlighting._describe_light(record, {1: [record]}) + assert 'GP' not in rendered + assert 'at (' not in rendered + + +def test_describe_light_renders_position_when_present(): + """Test that a real grid position is shown.""" + record = _decoded_light(4, button=1, gpio=13, position=(8, 9)) + rendered = hostlighting._describe_light(record, {1: [record]}) + assert 'GP13' in rendered + assert 'at (8,9)' in rendered + + +def test_unnamed_lights_are_never_duplicates(): + """Test that lights owned by no control are not grouped against each other.""" + records = [_decoded_light(1), _decoded_light(2)] + owners = {hostlighting.BUTTON_NONE: records} + assert not hostlighting._is_unexpected_duplicate(hostlighting.BUTTON_NONE, owners) + + +# --- page 1 rendering across firmware generations --------------------------- + +def _state_reply(features, framework, namespace, hz): + """Build a page 1 reply with the v1.1 appended fields set as given.""" + report = bytearray(hostlighting.REPORT_SIZE) + report[0] = hostlighting.CMD_GET_CAPS | hostlighting.RESPONSE_FLAG + report[3:7] = bytes([0, 0, 5, 1]) + report[7:11] = (0x1234).to_bytes(4, 'little') + report[12:16] = features.to_bytes(4, 'little') + report[16], report[17], report[18] = framework, namespace, hz + return bytes(report) + + +class _CannedDevice: + """Device stub that answers every capability page with one canned reply.""" + + def __init__(self, reply): + self.reply = reply + + def get_caps_page(self, page, start=None): + """Return the canned reply regardless of which page was asked for.""" + return self.reply + + +def _state_line(features, framework, namespace, hz): + """Return the v1.1 detail line _print_state emits, or '' if it emits none.""" + buffer = io.StringIO() + with contextlib.redirect_stdout(buffer): + hostlighting._print_state(_CannedDevice(_state_reply(features, framework, namespace, hz))) + lines = [line.strip() for line in buffer.getvalue().splitlines()[1:] if line.strip()] + return lines[0] if lines else '' + + +def test_print_state_says_nothing_extra_for_v1_0_firmware(): + """Test that a board predating the appended fields has none reported for it. + + v1.0 firmware zero-fills those bytes, and every one reserves 0 for "not + reported", so nothing is claimed on its behalf. + """ + assert _state_line(0, 0, 0, 0) == '' + + +def test_print_state_reports_the_framework_when_no_features_are_offered(): + """Test that an empty feature mask does not hide a stated framework byte. + + A v1.1 board clears both feature bits until its light registry populates, + so the mask cannot stand in for "does this board speak v1.1". + """ + line = _state_line(0, 2, 2, 40) + assert 'LED-refactor LED framework' in line + assert 'classic LED framework' in _state_line(0, 1, 1, 100) + + +def test_print_state_names_values_it_does_not_recognise(): + """Test that later firmware reads as unknown rather than as a decoder fault.""" + line = _state_line(0x04, 7, 3, 60) + assert 'unknown bits 0x04' in line + assert 'animations are unknown (3)' in line + assert 'unknown (7) LED framework' in line + assert 'None' not in line + + +def test_print_state_omits_the_offers_clause_for_a_zero_mask(): + """Test that a board offering nothing gets no offers clause at all.""" + assert not _state_line(0, 1, 2, 40).startswith('offers') From 1707ef60e949e10765bb6fe2a51fea2aaff6e69d Mon Sep 17 00:00:00 2001 From: Jacob Simpson <28767380+djGLiTCH@users.noreply.github.com> Date: Mon, 17 Aug 2026 17:23:01 +1000 Subject: [PATCH 5/6] Add hlp-set-light for Host Lighting Protocol v1.2 SET_LIGHT stages one light per entry by its light-table ordinal, the only address that can name a single light of a control that owns several. The tool takes ORDINAL:RRGGBB pairs, chunks them at the 15-entry report capacity, sums the board's applied and skipped counts across reports, and refuses cleanly on firmware whose PING predates v1.2. Ordinals a board does not have are counted as skipped rather than erroring, matching the protocol's SET_BUTTONS semantics. --- CHANGELOG.md | 4 +++ README.md | 9 ++++- gp2040ce_bintools/hostlighting.py | 56 +++++++++++++++++++++++++++++++ pyproject.toml | 1 + tests/test_hostlighting.py | 41 ++++++++++++++++++++++ 5 files changed, 110 insertions(+), 1 deletion(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index b2a521a..c814112 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -16,6 +16,10 @@ development tools, see the commit history. LED framework, animation namespace and render rate on the runtime state page. Boards speaking v1.0 are handled unchanged - fields they never sent are reported as absent rather than guessed, and the light table reads as unsupported rather than as an error. +* `hlp-set-light` stages per-light colours with Host Lighting Protocol v1.2's SET_LIGHT command, which + addresses one light of a control that owns several by its light-table ordinal. Ordinals a board does + not have are counted as skipped rather than erroring, matching the protocol's SET_BUTTONS semantics, + and the tool refuses cleanly on firmware that predates v1.2. ## v0.11.1 diff --git a/README.md b/README.md index 42a2c3a..e5312a3 100644 --- a/README.md +++ b/README.md @@ -133,6 +133,8 @@ board is connected, select one with `--board-id `. those records from its per-control configuration and says so, because duplicates cannot be seen that way. * `hlp-fill` fills the LEDs with a colour (`--scope all|buttons|case|pleds`) as a quick visual test, then restores the board's own animations. +* `hlp-set-light` colours individual lights by their light-table ordinal (protocol v1.2) - the way to + address one light of a control that owns several - then restores the board's own animations. * `hlp-input-mode` sets the board's input mode and reboots into it. * `hlp-reboot-webconfig` / `hlp-reboot-bootsel` reboot the board into the web configurator or the BOOTSEL bootloader for flashing. @@ -164,10 +166,15 @@ page 5 (light table): 16 lights, fingerprint 0x92207B00 % hlp-fill --scope case 00FF00 filled case with #00FF00 for 3.0s released - on-board animations restored + +% hlp-set-light 3:FF0000 12:0000FF +2 applied, 0 skipped for 3.0s +released - on-board animations restored ``` The light table is what reveals that this board wires two physical buttons to `Up` and two to `L3`; the LED -map on page 2 can only report one range per control, so it cannot express that. +map on page 2 can only report one range per control, so it cannot express that. On v1.2 firmware +`hlp-set-light` addresses those lights one at a time by their position in the table. ### summarize-gp2040ce diff --git a/gp2040ce_bintools/hostlighting.py b/gp2040ce_bintools/hostlighting.py index ba3d6ba..38d8a81 100644 --- a/gp2040ce_bintools/hostlighting.py +++ b/gp2040ce_bintools/hostlighting.py @@ -54,6 +54,7 @@ CMD_SET_RANGE_RGBW = 0x12 CMD_FILL = 0x13 CMD_CLEAR = 0x14 +CMD_SET_LIGHT = 0x15 # protocol v1.2 CMD_COMMIT = 0x30 # frame lifecycle, 0x30-0x3F CMD_RELEASE = 0x31 CMD_SET_ANIMATION = 0x40 # board features, 0x40-0x4F @@ -94,6 +95,10 @@ LIGHT_FLAG_PER_LIGHT = 0x02 # read from a per-light table, describes one light ACTION_NONE = -32768 # no owning GPIO action, distinct from an action of NONE +# SET_LIGHT (v1.2) stages lights by their page 5 record index; one report +# carries at most this many (ordinal, R, G, B) entries. +SET_LIGHT_MAX_ENTRIES = 15 + # FILL scopes (payload byte [2] of a FILL request) FILL_SCOPE_ALL = 0x00 FILL_SCOPE_BUTTONS = 0x01 @@ -396,6 +401,26 @@ def read_positions(self) -> list: if len(entries) >= total: return entries + def set_lights(self, entries: list) -> tuple[int, int]: + """Stage per-light colours by page 5 ordinal (protocol v1.2). + + Entries beyond one report's capacity are sent as further reports. An + ordinal at or past the board's light total is counted as skipped by + the board, not treated as an error. + + :param entries: (ordinal, red, green, blue) tuples + :return: totals of (applied, skipped) across all reports + :raises HostLightingRejected: on firmware without SET_LIGHT (pre-v1.2) + """ + applied = skipped = 0 + for start in range(0, len(entries), SET_LIGHT_MAX_ENTRIES): + chunk = entries[start:start + SET_LIGHT_MAX_ENTRIES] + payload = bytes([len(chunk)]) + b''.join(bytes(entry) for entry in chunk) + reply = self.request_ok(CMD_SET_LIGHT, payload) + applied += reply[3] + skipped += reply[4] + return applied, skipped + def read_lights(self) -> tuple[list, int]: """Read every page 5 record, walking the pages. @@ -802,6 +827,37 @@ def fill(): device.close() +def set_light(): + """Colour individual lights by their page 5 ordinal (protocol v1.2).""" + parser = _device_parser("Colour individual lights on a GP2040-CE board by their light-table ordinal, " + "hold, then restore animations.") + parser.add_argument('--seconds', type=float, default=3.0, help="how long to hold the colours (default: 3)") + parser.add_argument('entries', nargs='+', metavar='ORDINAL:RRGGBB', + help="a light's page 5 ordinal and its colour, e.g. 3:FF0000") + args, _ = parser.parse_known_args() + staged = [] + for token in args.entries: + ordinal, _, colour = token.partition(':') + value = int(colour, 16) + staged.append((int(ordinal), (value >> 16) & 0xFF, (value >> 8) & 0xFF, value & 0xFF)) + device = open_device(args.board_id) + try: + reply = device.request_ok(CMD_PING) + if (reply[7], reply[8]) < (1, 2): + raise SystemExit(f"board reports protocol {reply[7]}.{reply[8]}; SET_LIGHT needs v1.2") + # SET_MODE payload: [2] takeover mode (0 = whole frame), [3..4] keepalive + # timeout in ms (little endian, 10000 here), [5] apply board brightness + device.request_ok(CMD_SET_MODE, bytes([0x00, 0x10, 0x27, 0x01])) + applied, skipped = device.set_lights(staged) + device.request_ok(CMD_COMMIT) + print(f"{applied} applied, {skipped} skipped for {args.seconds}s") + time.sleep(args.seconds) + device.request_ok(CMD_RELEASE) + print("released - on-board animations restored") + finally: + device.close() + + def input_mode(): """Set a board's input mode over Host Lighting.""" parser = _device_parser("Set a GP2040-CE board's input mode; the board saves it and reboots into it.") diff --git a/pyproject.toml b/pyproject.toml index da5a9b4..634de97 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -44,6 +44,7 @@ hlp-input-mode = "gp2040ce_bintools.hostlighting:input_mode" hlp-ping = "gp2040ce_bintools.hostlighting:ping" hlp-reboot-bootsel = "gp2040ce_bintools.hostlighting:reboot_bootsel" hlp-reboot-webconfig = "gp2040ce_bintools.hostlighting:reboot_webconfig" +hlp-set-light = "gp2040ce_bintools.hostlighting:set_light" summarize-gp2040ce = "gp2040ce_bintools.builder:summarize_gp2040ce" visualize-config = "gp2040ce_bintools.storage:visualize" diff --git a/tests/test_hostlighting.py b/tests/test_hostlighting.py index f834693..ee3ca77 100644 --- a/tests/test_hostlighting.py +++ b/tests/test_hostlighting.py @@ -534,6 +534,47 @@ def test_get_caps_page_sends_no_start_entry_for_single_pages(page): assert device.payloads == [bytes([page])] +# --- SET_LIGHT (protocol v1.2) ---------------------------------------------- + +class _CountingDevice(hostlighting.HostLightingDevice): + """A device that records SET_LIGHT requests and returns scripted counts.""" + + def __init__(self, counts): + self.requests = [] + self.counts = list(counts) + + def request_ok(self, command, payload=b'', timeout=0.5): + """Record the request and answer with the next (applied, skipped) pair.""" + self.requests.append((command, bytes(payload))) + applied, skipped = self.counts.pop(0) + return _reply(bytes([applied, skipped]), command=hostlighting.CMD_SET_LIGHT) + + +def test_set_lights_places_each_entry_on_the_wire(): + """Test the payload layout: entry count, then ordinal/R/G/B per entry.""" + device = _CountingDevice([(2, 0)]) + applied, skipped = device.set_lights([(3, 0xFF, 0x00, 0x00), (12, 0x00, 0x00, 0xFF)]) + assert (applied, skipped) == (2, 0) + command, payload = device.requests[0] + assert command == hostlighting.CMD_SET_LIGHT + assert payload == bytes([2, 3, 0xFF, 0x00, 0x00, 12, 0x00, 0x00, 0xFF]) + + +def test_set_lights_chunks_at_the_report_capacity(): + """Test that a batch past one report's capacity is sent as several.""" + device = _CountingDevice([(15, 0), (5, 0)]) + applied, skipped = device.set_lights([(o, 1, 2, 3) for o in range(20)]) + assert (applied, skipped) == (20, 0) + assert [payload[0] for _, payload in device.requests] == [15, 5] + + +def test_set_lights_sums_applied_and_skipped_across_reports(): + """Test that the reply counters accumulate over the whole batch.""" + device = _CountingDevice([(14, 1), (3, 2)]) + applied, skipped = device.set_lights([(o, 9, 9, 9) for o in range(20)]) + assert (applied, skipped) == (17, 3) + + # --- light table rendering -------------------------------------------------- def _decoded_light(first, count=1, kind=0, button=0xFF, gpio=None, position=None): From 8b3d87a1ebeeee0f3c88b4858b1cb7852055f263 Mon Sep 17 00:00:00 2001 From: Jacob Simpson <28767380+djGLiTCH@users.noreply.github.com> Date: Mon, 17 Aug 2026 17:23:08 +1000 Subject: [PATCH 6/6] Decode Host Lighting Protocol v1.3 in hlp-set-light An eight-digit RRGGBBWW colour stages through the new SET_LIGHT_RGBW command, twelve entries per report; boards without a white channel ignore W. On v1.3 firmware the reply's per-entry outcome mask is decoded to name any skipped ordinals instead of only counting them, and set_lights grows a set_lights_rgbw sibling, both returning the raw masks alongside the counts. The README states the white contract: send the subtractive conversion (W = min(R,G,B), RGB reduced) to v1.3 firmware, and plain six-digit colours to older boards, whose achromatic mapping never reads a supplied W. The suite grows to 66 tests, covering the mask pass-through, the five-byte entry layout and the twelve-entry chunking. --- CHANGELOG.md | 4 ++ README.md | 6 +- gp2040ce_bintools/hostlighting.py | 91 +++++++++++++++++++++++++------ tests/test_hostlighting.py | 41 ++++++++++++-- 4 files changed, 118 insertions(+), 24 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index c814112..a8bfe8e 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -20,6 +20,10 @@ development tools, see the commit history. addresses one light of a control that owns several by its light-table ordinal. Ordinals a board does not have are counted as skipped rather than erroring, matching the protocol's SET_BUTTONS semantics, and the tool refuses cleanly on firmware that predates v1.2. +* Host Lighting Protocol v1.3 support: an eight-digit `RRGGBBWW` colour stages through the new + SET_LIGHT_RGBW command (boards without a white channel ignore W), and on v1.3 firmware the reply's + per-entry outcome mask is decoded to name any skipped ordinals instead of only counting them. + `set_lights` and the new `set_lights_rgbw` return the raw masks alongside the counts. ## v0.11.1 diff --git a/README.md b/README.md index e5312a3..98ec5da 100644 --- a/README.md +++ b/README.md @@ -134,7 +134,11 @@ board is connected, select one with `--board-id `. * `hlp-fill` fills the LEDs with a colour (`--scope all|buttons|case|pleds`) as a quick visual test, then restores the board's own animations. * `hlp-set-light` colours individual lights by their light-table ordinal (protocol v1.2) - the way to - address one light of a control that owns several - then restores the board's own animations. + address one light of a control that owns several - then restores the board's own animations. An + eight-digit colour (`RRGGBBWW`) stages through the v1.3 RGBW variant, and on v1.3 firmware any + skipped ordinals are named from the reply's outcome mask. For white-channel chains send the + subtractive conversion (W = min(R,G,B), RGB reduced by W) - v1.3 firmware renders it; earlier + firmware maps plain-RGB whites to the white emitter itself, so send six-digit colours there. * `hlp-input-mode` sets the board's input mode and reboots into it. * `hlp-reboot-webconfig` / `hlp-reboot-bootsel` reboot the board into the web configurator or the BOOTSEL bootloader for flashing. diff --git a/gp2040ce_bintools/hostlighting.py b/gp2040ce_bintools/hostlighting.py index 38d8a81..b5e2145 100644 --- a/gp2040ce_bintools/hostlighting.py +++ b/gp2040ce_bintools/hostlighting.py @@ -55,6 +55,7 @@ CMD_FILL = 0x13 CMD_CLEAR = 0x14 CMD_SET_LIGHT = 0x15 # protocol v1.2 +CMD_SET_LIGHT_RGBW = 0x16 # protocol v1.3 CMD_COMMIT = 0x30 # frame lifecycle, 0x30-0x3F CMD_RELEASE = 0x31 CMD_SET_ANIMATION = 0x40 # board features, 0x40-0x4F @@ -96,8 +97,12 @@ ACTION_NONE = -32768 # no owning GPIO action, distinct from an action of NONE # SET_LIGHT (v1.2) stages lights by their page 5 record index; one report -# carries at most this many (ordinal, R, G, B) entries. +# carries at most this many (ordinal, R, G, B) entries. The RGBW variant +# (v1.3) carries five-byte entries, so fewer fit. From v1.3 the reply also +# carries a per-entry outcome mask (bit n set = entry n applied); v1.2 +# boards zero-fill it, so only read it when PING reports minor >= 3. SET_LIGHT_MAX_ENTRIES = 15 +SET_LIGHT_RGBW_MAX_ENTRIES = 12 # FILL scopes (payload byte [2] of a FILL request) FILL_SCOPE_ALL = 0x00 @@ -401,7 +406,28 @@ def read_positions(self) -> list: if len(entries) >= total: return entries - def set_lights(self, entries: list) -> tuple[int, int]: + def _stage_light_entries(self, command: int, entries: list, capacity: int) -> tuple[int, int, list]: + """Send one per-light staging command, chunked to the report capacity. + + :param command: CMD_SET_LIGHT or CMD_SET_LIGHT_RGBW + :param entries: per-light tuples, one report entry each + :param capacity: entries per report for this command + :return: (applied, skipped, masks) - masks is one raw outcome mask per + report sent; boards before v1.3 zero-fill it, so only interpret + the masks when PING reports minor >= 3 + """ + applied = skipped = 0 + masks = [] + for start in range(0, len(entries), capacity): + chunk = entries[start:start + capacity] + payload = bytes([len(chunk)]) + b''.join(bytes(entry) for entry in chunk) + reply = self.request_ok(command, payload) + applied += reply[3] + skipped += reply[4] + masks.append(reply[5] | (reply[6] << 8)) + return applied, skipped, masks + + def set_lights(self, entries: list) -> tuple[int, int, list]: """Stage per-light colours by page 5 ordinal (protocol v1.2). Entries beyond one report's capacity are sent as further reports. An @@ -409,17 +435,24 @@ def set_lights(self, entries: list) -> tuple[int, int]: the board, not treated as an error. :param entries: (ordinal, red, green, blue) tuples - :return: totals of (applied, skipped) across all reports + :return: totals of (applied, skipped) across all reports, plus the + per-report outcome masks (meaningful from protocol v1.3) :raises HostLightingRejected: on firmware without SET_LIGHT (pre-v1.2) """ - applied = skipped = 0 - for start in range(0, len(entries), SET_LIGHT_MAX_ENTRIES): - chunk = entries[start:start + SET_LIGHT_MAX_ENTRIES] - payload = bytes([len(chunk)]) + b''.join(bytes(entry) for entry in chunk) - reply = self.request_ok(CMD_SET_LIGHT, payload) - applied += reply[3] - skipped += reply[4] - return applied, skipped + return self._stage_light_entries(CMD_SET_LIGHT, entries, SET_LIGHT_MAX_ENTRIES) + + def set_lights_rgbw(self, entries: list) -> tuple[int, int, list]: + """Stage per-light colours with a white component (protocol v1.3). + + Boards whose chain has no white channel ignore W, the same rule as + SET_RANGE_RGBW, so a host may always send it. + + :param entries: (ordinal, red, green, blue, white) tuples + :return: totals of (applied, skipped) across all reports, plus the + per-report outcome masks + :raises HostLightingRejected: on firmware without SET_LIGHT_RGBW (pre-v1.3) + """ + return self._stage_light_entries(CMD_SET_LIGHT_RGBW, entries, SET_LIGHT_RGBW_MAX_ENTRIES) def read_lights(self) -> tuple[list, int]: """Read every page 5 record, walking the pages. @@ -832,25 +865,49 @@ def set_light(): parser = _device_parser("Colour individual lights on a GP2040-CE board by their light-table ordinal, " "hold, then restore animations.") parser.add_argument('--seconds', type=float, default=3.0, help="how long to hold the colours (default: 3)") - parser.add_argument('entries', nargs='+', metavar='ORDINAL:RRGGBB', - help="a light's page 5 ordinal and its colour, e.g. 3:FF0000") + parser.add_argument('entries', nargs='+', metavar='ORDINAL:RRGGBB[WW]', + help="a light's page 5 ordinal and its colour, e.g. 3:FF0000; " + "an eight-digit colour adds a white byte (v1.3 firmware)") args, _ = parser.parse_known_args() staged = [] + hexdigits = set('0123456789abcdefABCDEF') + for token in args.entries: + colour = token.partition(':')[2] + if len(colour) not in (6, 8) or not set(colour) <= hexdigits: + raise SystemExit(f"colour must be six or eight hex digits: {token}") + rgbw = any(len(token.partition(':')[2]) == 8 for token in args.entries) for token in args.entries: ordinal, _, colour = token.partition(':') value = int(colour, 16) - staged.append((int(ordinal), (value >> 16) & 0xFF, (value >> 8) & 0xFF, value & 0xFF)) + white = value & 0xFF if len(colour) == 8 else 0 + if len(colour) == 8: + value >>= 8 + entry = (int(ordinal), (value >> 16) & 0xFF, (value >> 8) & 0xFF, value & 0xFF) + staged.append(entry + (white,) if rgbw else entry) device = open_device(args.board_id) try: reply = device.request_ok(CMD_PING) - if (reply[7], reply[8]) < (1, 2): - raise SystemExit(f"board reports protocol {reply[7]}.{reply[8]}; SET_LIGHT needs v1.2") + needed = (1, 3) if rgbw else (1, 2) + if (reply[7], reply[8]) < needed: + raise SystemExit(f"board reports protocol {reply[7]}.{reply[8]}; this needs " + f"v{needed[0]}.{needed[1]}") # SET_MODE payload: [2] takeover mode (0 = whole frame), [3..4] keepalive # timeout in ms (little endian, 10000 here), [5] apply board brightness device.request_ok(CMD_SET_MODE, bytes([0x00, 0x10, 0x27, 0x01])) - applied, skipped = device.set_lights(staged) + if rgbw: + applied, skipped, masks = device.set_lights_rgbw(staged) + capacity = SET_LIGHT_RGBW_MAX_ENTRIES + else: + applied, skipped, masks = device.set_lights(staged) + capacity = SET_LIGHT_MAX_ENTRIES device.request_ok(CMD_COMMIT) print(f"{applied} applied, {skipped} skipped for {args.seconds}s") + if skipped and (reply[7], reply[8]) >= (1, 3): + stale = [staged[report * capacity + bit][0] + for report, mask in enumerate(masks) + for bit in range(min(capacity, len(staged) - report * capacity)) + if not (mask >> bit) & 1] + print(f"skipped ordinals (not on this board): {stale}") time.sleep(args.seconds) device.request_ok(CMD_RELEASE) print("released - on-board animations restored") diff --git a/tests/test_hostlighting.py b/tests/test_hostlighting.py index ee3ca77..6f47cce 100644 --- a/tests/test_hostlighting.py +++ b/tests/test_hostlighting.py @@ -537,23 +537,26 @@ def test_get_caps_page_sends_no_start_entry_for_single_pages(page): # --- SET_LIGHT (protocol v1.2) ---------------------------------------------- class _CountingDevice(hostlighting.HostLightingDevice): - """A device that records SET_LIGHT requests and returns scripted counts.""" + """A device that records staging requests and returns scripted counts.""" def __init__(self, counts): self.requests = [] self.counts = list(counts) + self.masks = [] def request_ok(self, command, payload=b'', timeout=0.5): - """Record the request and answer with the next (applied, skipped) pair.""" + """Record the request and answer with the next scripted reply.""" self.requests.append((command, bytes(payload))) applied, skipped = self.counts.pop(0) - return _reply(bytes([applied, skipped]), command=hostlighting.CMD_SET_LIGHT) + mask = self.masks.pop(0) if self.masks else 0 + return _reply(bytes([applied, skipped, mask & 0xFF, mask >> 8]), + command=command) def test_set_lights_places_each_entry_on_the_wire(): """Test the payload layout: entry count, then ordinal/R/G/B per entry.""" device = _CountingDevice([(2, 0)]) - applied, skipped = device.set_lights([(3, 0xFF, 0x00, 0x00), (12, 0x00, 0x00, 0xFF)]) + applied, skipped, _ = device.set_lights([(3, 0xFF, 0x00, 0x00), (12, 0x00, 0x00, 0xFF)]) assert (applied, skipped) == (2, 0) command, payload = device.requests[0] assert command == hostlighting.CMD_SET_LIGHT @@ -563,7 +566,7 @@ def test_set_lights_places_each_entry_on_the_wire(): def test_set_lights_chunks_at_the_report_capacity(): """Test that a batch past one report's capacity is sent as several.""" device = _CountingDevice([(15, 0), (5, 0)]) - applied, skipped = device.set_lights([(o, 1, 2, 3) for o in range(20)]) + applied, skipped, _ = device.set_lights([(o, 1, 2, 3) for o in range(20)]) assert (applied, skipped) == (20, 0) assert [payload[0] for _, payload in device.requests] == [15, 5] @@ -571,10 +574,36 @@ def test_set_lights_chunks_at_the_report_capacity(): def test_set_lights_sums_applied_and_skipped_across_reports(): """Test that the reply counters accumulate over the whole batch.""" device = _CountingDevice([(14, 1), (3, 2)]) - applied, skipped = device.set_lights([(o, 9, 9, 9) for o in range(20)]) + applied, skipped, _ = device.set_lights([(o, 9, 9, 9) for o in range(20)]) assert (applied, skipped) == (17, 3) +def test_set_lights_returns_one_outcome_mask_per_report(): + """Test that each report's raw outcome mask is passed through in order.""" + device = _CountingDevice([(15, 0), (4, 1)]) + device.masks = [0x7FFF, 0x000B] + _, _, masks = device.set_lights([(o, 9, 9, 9) for o in range(20)]) + assert masks == [0x7FFF, 0x000B] + + +def test_set_lights_rgbw_places_the_white_byte_on_the_wire(): + """Test the five-byte RGBW entry layout and command ID.""" + device = _CountingDevice([(1, 0)]) + applied, skipped, _ = device.set_lights_rgbw([(3, 0x10, 0x20, 0x30, 0x40)]) + assert (applied, skipped) == (1, 0) + command, payload = device.requests[0] + assert command == hostlighting.CMD_SET_LIGHT_RGBW + assert payload == bytes([1, 3, 0x10, 0x20, 0x30, 0x40]) + + +def test_set_lights_rgbw_chunks_at_twelve_entries(): + """Test that the RGBW variant chunks at its smaller report capacity.""" + device = _CountingDevice([(12, 0), (8, 0)]) + applied, skipped, _ = device.set_lights_rgbw([(o, 1, 2, 3, 4) for o in range(20)]) + assert (applied, skipped) == (20, 0) + assert [payload[0] for _, payload in device.requests] == [12, 8] + + # --- light table rendering -------------------------------------------------- def _decoded_light(first, count=1, kind=0, button=0xFF, gpio=None, position=None):