diff --git a/lib/hidapi/common.py b/lib/hidapi/common.py index 6817511a..3946d98f 100644 --- a/lib/hidapi/common.py +++ b/lib/hidapi/common.py @@ -19,3 +19,4 @@ class DeviceInfo: hidpp_short: str | None hidpp_long: str | None centurion: bool = False + centurion_report_id: int | None = None # 0x50 or 0x51 when centurion=True diff --git a/lib/hidapi/udev_impl.py b/lib/hidapi/udev_impl.py index 4b5a6168..a5b2cf3f 100644 --- a/lib/hidapi/udev_impl.py +++ b/lib/hidapi/udev_impl.py @@ -102,6 +102,7 @@ def _match(action: str, device, filter_func: typing.Callable[[int, int, int, boo from hid_parser import ReportDescriptor hidpp_short = hidpp_long = centurion = False + centurion_report_id = None devfile = "/sys" + hid_device.properties.get("DEVPATH") + "/report_descriptor" with fileopen(devfile, "rb") as fd: with warnings.catch_warnings(): @@ -111,16 +112,22 @@ def _match(action: str, device, filter_func: typing.Callable[[int, int, int, boo # and _Usage(0xFF00, 0x0001) in rd.get_input_items(0x10)[0].usages # be more permissive hidpp_long = 0x11 in rd.input_report_ids and 19 * 8 == int(rd.get_input_report_size(0x11)) # and _Usage(0xFF00, 0x0002) in rd.get_input_items(0x11)[0].usages # be more permissive - # Centurion transport: report ID 0x51, 63-byte reports (usage page 0xFFA0) - centurion = ( - 0x51 in rd.input_report_ids and 63 * 8 == int(rd.get_input_report_size(0x51)) and 0x51 in rd.output_report_ids - ) + # Centurion transport: 63-byte reports on usage page 0xFFA0 (both input and output) + # 0x51 = PRO X 2 LIGHTSPEED variant, 0x50 = G522 LIGHTSPEED variant (with device address byte) + if 0x51 in rd.input_report_ids and 63 * 8 == int(rd.get_input_report_size(0x51)) and 0x51 in rd.output_report_ids: + centurion_report_id = 0x51 + elif ( + 0x50 in rd.input_report_ids and 63 * 8 == int(rd.get_input_report_size(0x50)) and 0x50 in rd.output_report_ids + ): + centurion_report_id = 0x50 + centurion = centurion_report_id is not None if not hidpp_short and not hidpp_long and not centurion: return except Exception as e: # if can't process report descriptor fall back to old scheme hidpp_short = None hidpp_long = None centurion = False + centurion_report_id = None logger.info( "Report Descriptor not processed for DEVICE %s BID %s VID %s PID %s: %s", device.device_node, @@ -171,6 +178,7 @@ def _match(action: str, device, filter_func: typing.Callable[[int, int, int, boo hidpp_short=hidpp_short, hidpp_long=hidpp_long, centurion=centurion if centurion else False, + centurion_report_id=centurion_report_id, ) return d_info diff --git a/lib/logitech_receiver/base.py b/lib/logitech_receiver/base.py index b66e2a1e..837535c2 100644 --- a/lib/logitech_receiver/base.py +++ b/lib/logitech_receiver/base.py @@ -97,18 +97,30 @@ HIDPP_LONG_MESSAGE_ID = 0x11 DJ_MESSAGE_ID = 0x20 # Centurion transport (used by PRO X 2 LIGHTSPEED headset and similar) -# Uses report ID 0x51 on usage page 0xFFA0, 64-byte frames. -# Wire format (CPL): [0x51, cpl_length, flags=0x00, feat_idx, func_sw, params..., pad] +# Two variants exist, distinguished by report ID: +# 0x51 (PRO X 2): [0x51, cpl_length, flags, feat_idx, func_sw, params..., pad] +# 0x50 (G522): [0x50, device_addr, cpl_length, flags, feat_idx, func_sw, params..., pad] +# The 0x50 variant adds a device_addr byte at position [1], shifting all CPL fields by +1. # cpl_length = number of bytes from flags to end of meaningful data (includes flags byte). # The device_index byte from standard HID++ is NOT present in Centurion framing. CENTURION_REPORT_ID = 0x51 +CENTURION_ADDRESSED_REPORT_ID = 0x50 # addressed variant with device_addr byte at frame[1] (G522 etc.) CENTURION_FRAME_SIZE = 64 # 1 byte report ID + 63 bytes payload _CENTURION_MSG_SIZE = 63 # max reconstructed message size after unwrapping (2 + 61 payload bytes) -# Set of handles that use Centurion framing -_centurion_handles: set[int] = set() -# Raw Centurion protocol version (major, minor) by handle, from ping response -_centurion_protocol_versions: dict[int, tuple[int, int]] = {} + +@dataclasses.dataclass +class CenturionHandleState: + """Per-handle state for Centurion devices.""" + + report_id: int = CENTURION_REPORT_ID # 0x50 or 0x51 + device_addr: int | None = None # learned from first RX (0x50 only) + protocol_version: tuple[int, int] | None = None # from ping response + + +# All centurion per-handle state in a single dict. +# Membership test (ihandle in _centurion_handles) gates centurion-specific code paths. +_centurion_handles: dict[int, CenturionHandleState] = {} """Default timeout on read (in seconds).""" @@ -301,8 +313,7 @@ def close(handle): if handle: try: if isinstance(handle, int): - _centurion_handles.discard(handle) - _centurion_protocol_versions.pop(handle, None) + _centurion_handles.pop(handle, None) hidapi.close(handle) else: handle.close() @@ -313,6 +324,58 @@ def close(handle): return False +def _centurion_frame_header(state: CenturionHandleState, cpl_length: int, flags: int) -> bytes: + """Build the fixed prefix of a centurion frame. + + 0x51: [0x51, cpl_length, flags] (3 bytes) + 0x50: [0x50, device_addr, cpl_length, flags] (4 bytes) + """ + if state.report_id == CENTURION_ADDRESSED_REPORT_ID: + device_addr = state.device_addr if state.device_addr is not None else 0x00 + return struct.pack("!BBBB", CENTURION_ADDRESSED_REPORT_ID, device_addr, cpl_length, flags) + return struct.pack("!BBB", CENTURION_REPORT_ID, cpl_length, flags) + + +_CENTURION_REPORT_IDS = (CENTURION_REPORT_ID, CENTURION_ADDRESSED_REPORT_ID) + + +def _unwrap_centurion_frame(data: bytes, ihandle: int, handle) -> bytes: + """Unwrap a Centurion CPL frame (0x50 or 0x51) into a standard HID++ long message. + + Auto-detects the variant from the raw report ID byte (self-describing), + matching how _read() handles 0x10 vs 0x11. + + For 0x50, learns the device address from byte[1] on first receive. + """ + raw_report_id = ord(data[:1]) + if raw_report_id == CENTURION_ADDRESSED_REPORT_ID: + # 0x50: [report_id, device_addr, cpl_length, flags, feat_idx, func_sw, data...] + device_addr = ord(data[1:2]) + state = _centurion_handles.get(ihandle) + if state is not None and state.device_addr is None: + state.device_addr = device_addr + if logger.isEnabledFor(logging.DEBUG): + logger.debug("(%s) learned centurion device addr 0x%02X", handle, device_addr) + cpl_length = ord(data[2:3]) + inner_payload = data[4 : 3 + cpl_length] # cpl_length - 1 bytes (skip flags) + elif raw_report_id == CENTURION_REPORT_ID: + # 0x51: [report_id, cpl_length, flags, feat_idx, func_sw, data...] + cpl_length = ord(data[1:2]) + inner_payload = data[3 : 2 + cpl_length] # cpl_length - 1 bytes (skip flags) + else: + return data # not a centurion frame + + data = bytes([HIDPP_LONG_MESSAGE_ID, 0xFF]) + inner_payload + # Pad to a valid message size: standard long (20) or Centurion extended (63) + if len(data) <= _LONG_MESSAGE_SIZE: + data = data + b"\x00" * (_LONG_MESSAGE_SIZE - len(data)) + elif len(data) <= _CENTURION_MSG_SIZE: + data = data + b"\x00" * (_CENTURION_MSG_SIZE - len(data)) + else: + data = data[:_CENTURION_MSG_SIZE] + return data + + def write(handle, devnumber, data, long_message=False): """Writes some data to the receiver, addressed to a certain device. @@ -337,12 +400,12 @@ def write(handle, devnumber, data, long_message=False): ihandle = int(handle) if ihandle in _centurion_handles: - # Centurion CPL framing: [0x51, cpl_length, flags=0x00, feat_idx, func_sw, params...] - # cpl_length = len(meaningful_payload) + 1 (the +1 counts the flags byte) - # The device_index is stripped — only the HID++ payload (feat_idx + func_sw + params) remains. + # Centurion CPL framing — strip device_index from HID++ and wrap in CPL header. + # cpl_length = len(meaningful_payload) + 1 (the +1 counts the flags byte). + state = _centurion_handles[ihandle] payload = wdata[2:] # skip report_id and devnumber from standard frame cpl_length = len(data) + 1 # data is the unpadded payload; +1 for flags byte - wdata = struct.pack("!BBB", CENTURION_REPORT_ID, cpl_length, 0x00) + payload + wdata = _centurion_frame_header(state, cpl_length, 0x00) + payload wdata = wdata + b"\x00" * (CENTURION_FRAME_SIZE - len(wdata)) if logger.isEnabledFor(logging.DEBUG): @@ -366,7 +429,9 @@ def write(handle, devnumber, data, long_message=False): def write_centurion_cpl(handle, layer3_payload, flags=0x00): """Send a Centurion CPL frame with the given Layer 3+ payload. - Builds: [0x51, cpl_length, flags, layer3_payload..., zero-pad to 64 bytes] + Builds the appropriate header for the handle's report ID variant: + 0x51: [0x51, cpl_length, flags, layer3_payload..., pad to 64] + 0x50: [0x50, device_addr, cpl_length, flags, layer3_payload..., pad to 64] where cpl_length = len(layer3_payload) + 1 (the +1 counts the flags byte). For multi-fragment sends, flags encodes fragment index and continuation: @@ -376,11 +441,13 @@ def write_centurion_cpl(handle, layer3_payload, flags=0x00): ihandle = int(handle) if ihandle not in _centurion_handles: raise ValueError("write_centurion_cpl called on non-Centurion handle") + state = _centurion_handles[ihandle] cpl_length = len(layer3_payload) + 1 # +1 for flags byte - wdata = struct.pack("!BBB", CENTURION_REPORT_ID, cpl_length, flags) + layer3_payload + header = _centurion_frame_header(state, cpl_length, flags) + wdata = header + layer3_payload wdata = wdata + b"\x00" * (CENTURION_FRAME_SIZE - len(wdata)) if logger.isEnabledFor(logging.DEBUG): - logger.debug("(%s) <= centurion_cpl[%s]", handle, common.strhex(wdata[: cpl_length + 2])) + logger.debug("(%s) <= centurion_cpl[%s]", handle, common.strhex(wdata[: len(header) + cpl_length - 1])) try: hidapi.write(ihandle, wdata) except Exception as reason: @@ -452,22 +519,8 @@ def _read(handle, timeout) -> tuple[int, int, bytes]: close(handle) raise exceptions.NoReceiver(reason=reason) from reason - if data and is_centurion and ord(data[:1]) == CENTURION_REPORT_ID: - # Unwrap Centurion CPL framing: - # RX: [0x51, cpl_length, flags=0x00, feat_idx, func_sw, data...] - # cpl_length includes the flags byte, so meaningful payload starts at byte 3 - # and has (cpl_length - 1) bytes. - # Reconstruct as HID++ long message: [0x11, devnumber=0xFF, feat_idx, func_sw, data...] - cpl_length = ord(data[1:2]) - inner_payload = data[3 : 2 + cpl_length] # bytes 3..2+cpl_length-1 = cpl_length-1 bytes - data = bytes([HIDPP_LONG_MESSAGE_ID, 0xFF]) + inner_payload - # Pad to a valid message size: standard long (20) or Centurion extended (63) - if len(data) <= _LONG_MESSAGE_SIZE: - data = data + b"\x00" * (_LONG_MESSAGE_SIZE - len(data)) - elif len(data) <= _CENTURION_MSG_SIZE: - data = data + b"\x00" * (_CENTURION_MSG_SIZE - len(data)) - else: - data = data[:_CENTURION_MSG_SIZE] + if data and is_centurion and ord(data[:1]) in _CENTURION_REPORT_IDS: + data = _unwrap_centurion_frame(data, ihandle, handle) if data and _is_relevant_message(data): # ignore messages that fail check report_id = ord(data[:1]) @@ -725,7 +778,7 @@ def ping(handle, devnumber, long_message: bool = False): major = ord(reply_data[2:3]) minor = ord(reply_data[3:4]) if is_centurion: - _centurion_protocol_versions[int(handle)] = (major, minor) + _centurion_handles[int(handle)].protocol_version = (major, minor) return major + minor / 10.0 if ( @@ -771,17 +824,8 @@ def _read_input_buffer(handle, ihandle, notifications_hook): raise exceptions.NoReceiver(reason=reason) from reason if data: - if is_centurion and ord(data[:1]) == CENTURION_REPORT_ID: - # Unwrap Centurion CPL framing same as in _read() - cpl_length = ord(data[1:2]) - inner_payload = data[3 : 2 + cpl_length] - data = bytes([HIDPP_LONG_MESSAGE_ID, 0xFF]) + inner_payload - if len(data) <= _LONG_MESSAGE_SIZE: - data = data + b"\x00" * (_LONG_MESSAGE_SIZE - len(data)) - elif len(data) <= _CENTURION_MSG_SIZE: - data = data + b"\x00" * (_CENTURION_MSG_SIZE - len(data)) - else: - data = data[:_CENTURION_MSG_SIZE] + if is_centurion and ord(data[:1]) in _CENTURION_REPORT_IDS: + data = _unwrap_centurion_frame(data, ihandle, handle) if _is_relevant_message(data): # only process messages that pass check # report_id = ord(data[:1]) if notifications_hook: diff --git a/lib/logitech_receiver/centurion.py b/lib/logitech_receiver/centurion.py index f1222869..63933c34 100644 --- a/lib/logitech_receiver/centurion.py +++ b/lib/logitech_receiver/centurion.py @@ -491,13 +491,14 @@ def create_centurion_receiver(low_level, device_info, setting_callback=None): try: handle = low_level.open_path(device_info.path) if handle: - base._centurion_handles.add(int(handle)) + report_id = getattr(device_info, "centurion_report_id", None) or base.CENTURION_REPORT_ID + base._centurion_handles[int(handle)] = base.CenturionHandleState(report_id=report_id) cr = CenturionReceiver(low_level, handle, device_info, setting_callback) # Check if any discovered feature is CentPPBridge (0x0003) has_bridge = any(feat_id == CenturionCoreFeature.CENT_PP_BRIDGE for _, feat_id, _ in (cr.dongle_features or [])) if not has_bridge: logger.info("Centurion device %s has no bridge, treating as direct device", device_info.path) - base._centurion_handles.discard(int(handle)) + base._centurion_handles.pop(int(handle), None) cr.handle = None # prevent __del__ from double-closing low_level.close(handle) return None diff --git a/lib/logitech_receiver/descriptors.py b/lib/logitech_receiver/descriptors.py index f2c9bbd3..4948b201 100644 --- a/lib/logitech_receiver/descriptors.py +++ b/lib/logitech_receiver/descriptors.py @@ -466,3 +466,4 @@ _D( usbid=0x0ABA, ) # PRO X 2 LIGHTSPEED Gaming Headset (0x0AF7) — fully probed via Centurion transport, no static descriptor needed +# G522 LIGHTSPEED Gaming Headset (0x0B18 dongle, 0x0B19 wired) — Centurion 0x50 variant, no static descriptor needed diff --git a/lib/logitech_receiver/device.py b/lib/logitech_receiver/device.py index 2f23c5d1..f04ca426 100644 --- a/lib/logitech_receiver/device.py +++ b/lib/logitech_receiver/device.py @@ -78,7 +78,8 @@ def create_device(low_level: LowLevelInterface, device_info, setting_callback=No handle = low_level.open_path(device_info.path) if handle: if getattr(device_info, "centurion", False): - base._centurion_handles.add(int(handle)) + report_id = getattr(device_info, "centurion_report_id", None) or base.CENTURION_REPORT_ID + base._centurion_handles[int(handle)] = base.CenturionHandleState(report_id=report_id) # a direct connected device might not be online (as reported by user) return Device( low_level, @@ -625,20 +626,22 @@ class Device: return self.centurion_bridge_request(sub_idx, function, *params, no_reply=no_reply) return hidpp20.feature_request(self, feature, function, *params, no_reply=no_reply) - # Max sub-message bytes in the first bridge fragment: + # Max sub-message bytes in the first bridge fragment (for 0x51): # 64 - 1 (report ID) - 1 (cpl_len) - 1 (flags) - 2 (bridge prefix) - 2 (bridge hdr) = 57 # LGHUB uses 56 for first fragment (60 byte payload - 4 bridge overhead) + # For 0x50, subtract 1 more for the device_addr byte. _BRIDGE_FIRST_CHUNK = 56 # Continuation fragments carry raw sub_msg data (no bridge prefix/hdr): # 64 - 1 (report ID) - 1 (cpl_len) - 1 (flags) = 61, but LGHUB uses 60 + # For 0x50, subtract 1 more for the device_addr byte. _BRIDGE_CONT_CHUNK = 60 def centurion_bridge_request(self, sub_feat_idx, sub_function=0x00, *params, no_reply=False): """Send a request to a Centurion sub-device via CentPPBridge. Builds the 4-layer nested message: - Layer 1: [0x51] - Layer 2: [cpl_length, flags] + Layer 1: [report_id] (0x51 or 0x50) + Layer 2: [device_addr (0x50 only),] cpl_length, flags Layer 3: [bridge_idx, sendFragment_func|swid, bridge_hdr...] Layer 4: [sub_cpl=0x00, sub_feat_idx, sub_func|swid, params...] @@ -659,6 +662,12 @@ class Device: if not handle: return None + # Adjust bridge chunk sizes for 0x50 variant (device_addr byte takes 1 frame byte) + cent_state = base._centurion_handles.get(int(handle)) + addr_overhead = 1 if cent_state and cent_state.report_id == base.CENTURION_ADDRESSED_REPORT_ID else 0 + first_chunk = self._BRIDGE_FIRST_CHUNK - addr_overhead + cont_chunk = self._BRIDGE_CONT_CHUNK - addr_overhead + sw_id = base._get_next_sw_id() # Build sub-device message: [sub_cpl=0x00, sub_feat_idx, sub_func|swid, params...] @@ -674,7 +683,7 @@ class Device: timeout = base.DEFAULT_TIMEOUT with base.acquire_timeout(base.handle_lock(handle), handle, timeout): - if sub_len <= self._BRIDGE_FIRST_CHUNK: + if sub_len <= first_chunk: # Single-frame path layer3 = bridge_prefix + bridge_hdr + sub_msg base.write_centurion_cpl(handle, layer3) @@ -691,11 +700,11 @@ class Device: offset = 0 while offset < sub_len: if frag_index == 0: - chunk_size = self._BRIDGE_FIRST_CHUNK + chunk_size = first_chunk chunk = sub_msg[offset : offset + chunk_size] layer3 = bridge_prefix + bridge_hdr + chunk else: - chunk_size = self._BRIDGE_CONT_CHUNK + chunk_size = cont_chunk chunk = sub_msg[offset : offset + chunk_size] layer3 = chunk has_more = (offset + chunk_size) < sub_len @@ -811,9 +820,9 @@ class Device: def _record_ping_protocol(self, handle, protocol): """Record a successful ping's protocol version, including raw Centurion (major, minor).""" self._protocol = protocol - cent_ver = base._centurion_protocol_versions.get(int(handle)) - if cent_ver: - self._centurion_protocol = cent_ver + cent_state = base._centurion_handles.get(int(handle)) + if cent_state and cent_state.protocol_version: + self._centurion_protocol = cent_state.protocol_version def ping(self): """Checks if the device is online and present, returns True of False. diff --git a/lib/logitech_receiver/listener.py b/lib/logitech_receiver/listener.py index 4137afd4..1ca4a778 100644 --- a/lib/logitech_receiver/listener.py +++ b/lib/logitech_receiver/listener.py @@ -15,6 +15,7 @@ ## with this program; if not, write to the Free Software Foundation, Inc., ## 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA. +import dataclasses import logging import queue import threading @@ -52,9 +53,11 @@ class _ThreadedHandle: else: # if logger.isEnabledFor(logging.DEBUG): # logger.debug("%r opened new handle %d", self, handle) - # If original handle was centurion, register new per-thread handle too - if any(h in base._centurion_handles for h in self._handles): - base._centurion_handles.add(handle) + # If original handle was centurion, copy state to new per-thread handle + for h in self._handles: + if h in base._centurion_handles: + base._centurion_handles[handle] = dataclasses.replace(base._centurion_handles[h]) + break self._local.handle = handle self._handles.append(handle) return handle diff --git a/tests/logitech_receiver/test_base.py b/tests/logitech_receiver/test_base.py index a02696fd..09549f32 100644 --- a/tests/logitech_receiver/test_base.py +++ b/tests/logitech_receiver/test_base.py @@ -8,7 +8,10 @@ import pytest from logitech_receiver import base from logitech_receiver import exceptions +from logitech_receiver.base import CENTURION_ADDRESSED_REPORT_ID +from logitech_receiver.base import CENTURION_REPORT_ID from logitech_receiver.base import HIDPP_SHORT_MESSAGE_ID +from logitech_receiver.base import CenturionHandleState from logitech_receiver.common import LOGITECH_VENDOR_ID from logitech_receiver.common import BusID from logitech_receiver.hidpp10_constants import ErrorCode as Hidpp10Error @@ -200,3 +203,113 @@ def test_ping_errors(simulated_error: Hidpp10Error, expected_result): else: result = base.ping(handle=handle, devnumber=device_number) assert result == expected_result + + +# --- Centurion transport tests --- + + +class TestCenturionFrameHeader: + """Test _centurion_frame_header builds correct headers for both variants.""" + + def test_0x51_header(self): + state = CenturionHandleState(report_id=CENTURION_REPORT_ID) + header = base._centurion_frame_header(state, cpl_length=5, flags=0x00) + assert header == bytes([0x51, 5, 0x00]) + + def test_0x51_header_with_flags(self): + state = CenturionHandleState(report_id=CENTURION_REPORT_ID) + header = base._centurion_frame_header(state, cpl_length=10, flags=0x03) + assert header == bytes([0x51, 10, 0x03]) + + def test_0x50_header_unknown_addr(self): + state = CenturionHandleState(report_id=CENTURION_ADDRESSED_REPORT_ID, device_addr=None) + header = base._centurion_frame_header(state, cpl_length=5, flags=0x00) + # device_addr defaults to 0x00 when unknown + assert header == bytes([0x50, 0x00, 5, 0x00]) + + def test_0x50_header_known_addr(self): + state = CenturionHandleState(report_id=CENTURION_ADDRESSED_REPORT_ID, device_addr=0x23) + header = base._centurion_frame_header(state, cpl_length=5, flags=0x00) + assert header == bytes([0x50, 0x23, 5, 0x00]) + + def test_0x50_header_with_flags(self): + state = CenturionHandleState(report_id=CENTURION_ADDRESSED_REPORT_ID, device_addr=0x23) + header = base._centurion_frame_header(state, cpl_length=10, flags=0x07) + assert header == bytes([0x50, 0x23, 10, 0x07]) + + +class TestUnwrapCenturionFrame: + """Test _unwrap_centurion_frame for both 0x51 and 0x50 variants.""" + + HANDLE = 99 + + def setup_method(self): + """Ensure no leftover centurion state between tests.""" + base._centurion_handles.pop(self.HANDLE, None) + + def teardown_method(self): + base._centurion_handles.pop(self.HANDLE, None) + + def test_unwrap_0x51_frame(self): + """0x51 frame with feat_idx=0x02, func_sw=0x1A, 2 data bytes.""" + # cpl_length = 1(flags) + 1(feat_idx) + 1(func_sw) + 2(data) = 5 + raw = bytes([0x51, 5, 0x00, 0x02, 0x1A, 0xAA, 0xBB]) + b"\x00" * 57 + result = base._unwrap_centurion_frame(raw, self.HANDLE, self.HANDLE) + # Should reconstruct as [0x11, 0xFF, feat_idx, func_sw, data..., pad to 20] + assert result[0] == 0x11 + assert result[1] == 0xFF + assert result[2] == 0x02 # feat_idx + assert result[3] == 0x1A # func_sw + assert result[4] == 0xAA + assert result[5] == 0xBB + assert len(result) == 20 # padded to standard long + + def test_unwrap_0x50_frame(self): + """0x50 frame with device_addr=0x23, same payload as above.""" + # Frame: [0x50, device_addr, cpl_length, flags, feat_idx, func_sw, data...] + raw = bytes([0x50, 0x23, 5, 0x00, 0x02, 0x1A, 0xAA, 0xBB]) + b"\x00" * 56 + base._centurion_handles[self.HANDLE] = CenturionHandleState(report_id=CENTURION_ADDRESSED_REPORT_ID) + result = base._unwrap_centurion_frame(raw, self.HANDLE, self.HANDLE) + assert result[0] == 0x11 + assert result[1] == 0xFF + assert result[2] == 0x02 # feat_idx + assert result[3] == 0x1A # func_sw + assert result[4] == 0xAA + assert result[5] == 0xBB + assert len(result) == 20 + + def test_0x50_learns_device_addr(self): + """First RX on a 0x50 handle should learn the device address.""" + base._centurion_handles[self.HANDLE] = CenturionHandleState(report_id=CENTURION_ADDRESSED_REPORT_ID) + assert base._centurion_handles[self.HANDLE].device_addr is None + + raw = bytes([0x50, 0x23, 3, 0x00, 0x02, 0x1A]) + b"\x00" * 58 + base._unwrap_centurion_frame(raw, self.HANDLE, self.HANDLE) + + assert base._centurion_handles[self.HANDLE].device_addr == 0x23 + + def test_0x50_does_not_overwrite_addr(self): + """Once learned, device address should not be overwritten.""" + base._centurion_handles[self.HANDLE] = CenturionHandleState(report_id=CENTURION_ADDRESSED_REPORT_ID, device_addr=0x23) + raw = bytes([0x50, 0xFF, 3, 0x00, 0x02, 0x1A]) + b"\x00" * 58 + base._unwrap_centurion_frame(raw, self.HANDLE, self.HANDLE) + + # Should keep the original address, not overwrite with 0xFF + assert base._centurion_handles[self.HANDLE].device_addr == 0x23 + + def test_non_centurion_frame_passthrough(self): + """Non-centurion report IDs should be returned unchanged.""" + raw = bytes([0x11, 0x01, 0x02, 0x1A]) + b"\x00" * 16 + result = base._unwrap_centurion_frame(raw, self.HANDLE, self.HANDLE) + assert result == raw + + def test_unwrap_0x51_large_payload(self): + """0x51 frame with payload large enough to need 63-byte padding.""" + # cpl_length covers all 61 payload bytes + flags = 62 + payload = bytes(range(61)) + raw = bytes([0x51, 62, 0x00]) + payload + result = base._unwrap_centurion_frame(raw, self.HANDLE, self.HANDLE) + assert len(result) == 63 # padded to centurion extended + assert result[0] == 0x11 + assert result[1] == 0xFF + assert result[2:63] == payload diff --git a/tests/logitech_receiver/test_device.py b/tests/logitech_receiver/test_device.py index b291dba9..63e632bc 100644 --- a/tests/logitech_receiver/test_device.py +++ b/tests/logitech_receiver/test_device.py @@ -61,6 +61,7 @@ class DeviceInfoStub: bus_id: int = 0x0003 # USB serial: str = "aa:aa:aa;aa" centurion: bool = False + centurion_report_id: int | None = None di_bad_handle = DeviceInfoStub(None, product_id="CCCC") @@ -107,9 +108,33 @@ def test_create_centurion_device(): assert test_device.centurion is True assert test_device.hidpp_long is True assert int(test_device.handle) in base._centurion_handles + state = base._centurion_handles[int(test_device.handle)] + assert state.report_id == base.CENTURION_REPORT_ID # 0x51 default # Clean up - base._centurion_handles.discard(int(test_device.handle)) + base._centurion_handles.pop(int(test_device.handle), None) + + +di_0B18 = DeviceInfoStub("11", product_id="0B18", centurion=True, centurion_report_id=0x50) + + +def test_create_centurion_0x50_device(): + """Test that a 0x50 centurion device gets the correct report ID registered.""" + from logitech_receiver import base + + low_level_mock = LowLevelInterfaceFake(fake_hidpp.r_empty) + test_device = device.create_device(low_level_mock, di_0B18) + + assert test_device is not None + assert test_device.centurion is True + assert test_device.hidpp_long is True + assert int(test_device.handle) in base._centurion_handles + state = base._centurion_handles[int(test_device.handle)] + assert state.report_id == base.CENTURION_ADDRESSED_REPORT_ID # 0x50 + assert state.device_addr is None # not yet learned + + # Clean up + base._centurion_handles.pop(int(test_device.handle), None) @pytest.mark.parametrize(