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..e1241cd --- /dev/null +++ b/gp2040ce_bintools/hostlighting.py @@ -0,0 +1,479 @@ +"""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 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 +* 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 HostLightingRejected(HostLightingError): + """The board answered a command with a non-OK status.""" + + +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 HostLightingRejected(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 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. + + :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 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 step {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 + 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: + 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: + send_reboot(device, 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: + 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/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..ee5f51c --- /dev/null +++ b/tests/test_hostlighting.py @@ -0,0 +1,91 @@ +"""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) + + +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')