Add G522 LIGHTSPEED headset support (Centurion 0x50 transport variant)
The G522 uses a Centurion protocol variant with report ID 0x50 that adds a device address byte at frame position [1], shifting all CPL fields by +1 compared to the PRO X 2's 0x51 variant. This commit adds transport- layer support for both variants while consolidating per-handle centurion state into a single CenturionHandleState dataclass. Key changes: - Consolidate _centurion_handles (set) and _centurion_protocol_versions (dict) into a single dict[int, CenturionHandleState] keyed by handle - Add _unwrap_centurion_frame() helper that auto-detects 0x50 vs 0x51 from raw frame bytes and learns the device address on first RX - Add _centurion_frame_header() to build the correct TX header per variant - Detect both report IDs in udev report descriptor parsing - Adjust bridge fragment chunk sizes for 0x50's extra header byte - Propagate full CenturionHandleState when opening per-thread handles Wire format verified against G522 diagnostic logs from LGSTrayEx#15.
This commit is contained in:
Ken Sanislo
parent
0a6421ef82
commit
3c6770a680
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
||||
|
|
|
|||
|
|
@ -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:
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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.
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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(
|
||||
|
|
|
|||
Loading…
Reference in New Issue