diff --git a/CHANGELOG.md b/CHANGELOG.md index ff0f971..a8bfe8e 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -3,6 +3,28 @@ 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`. +* `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. +* `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. +* 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 ### Miscellaneous diff --git a/README.md b/README.md index 3ee7ce6..98ec5da 100644 --- a/README.md +++ b/README.md @@ -119,6 +119,67 @@ 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, 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-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. 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. + +Sample usage: + +``` +% hlp-ping +Haute42 COSMOX (v0.7.12), board ID 433031343539302E +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 + +% 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. On v1.2 firmware +`hlp-set-light` addresses those lights one at a time by their position in the table. + ### 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..b5e2145 --- /dev/null +++ b/gp2040ce_bintools/hostlighting.py @@ -0,0 +1,955 @@ +"""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, 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`). + +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_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 +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 +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 + +# SET_LIGHT (v1.2) stages lights by their page 5 record index; one report +# 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 +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. 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'} + +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 + + +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.""" + + +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 pages + :return: the reply report ([3..] is the page's payload) + """ + 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 _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 + 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, plus the + per-report outcome masks (meaningful from protocol v1.3) + :raises HostLightingRejected: on firmware without SET_LIGHT (pre-v1.2) + """ + 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. + + 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.""" + 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) + _print_lights(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, [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 + 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 + """ + 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] 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 + fingerprint (little endian, matches page 1's for a coherent snapshot). + + :param device: an opened HostLightingDevice + """ + 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 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 (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) + 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: + """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 + """ + 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(): + """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 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[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) + 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) + 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])) + 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") + 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..634de97 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,13 @@ 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" +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 new file mode 100644 index 0000000..6f47cce --- /dev/null +++ b/tests/test_hostlighting.py @@ -0,0 +1,724 @@ +"""Test the Host Lighting protocol helpers. + +SPDX-FileCopyrightText: © 2026 Jacob Simpson +SPDX-License-Identifier: GPL-3.0-or-later +""" +import contextlib +import io + +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') + + +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])] + + +# --- SET_LIGHT (protocol v1.2) ---------------------------------------------- + +class _CountingDevice(hostlighting.HostLightingDevice): + """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 scripted reply.""" + self.requests.append((command, bytes(payload))) + applied, skipped = self.counts.pop(0) + 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)]) + 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) + + +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): + """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')