diff --git a/docs/devices/PRO X 2 LIGHTSPEED 0AF7.text b/docs/devices/PRO X 2 LIGHTSPEED 0AF7.text new file mode 100644 index 00000000..bb908208 --- /dev/null +++ b/docs/devices/PRO X 2 LIGHTSPEED 0AF7.text @@ -0,0 +1,38 @@ +Solaar version 1.1.19 + +PRO X 2 LIGHTSPEED + Device path : /dev/hidraw4 + USB id : 046d:0AF7 + Codename : PRO X 2 LIGHTSPEED + Kind : headset + Protocol : Centurion 2.6 + Serial number: + Model ID: 0508 + Unit ID: + 1: 0.02 + Supports 5 dongle + 10 headset features: + 0: ROOT {0000} V0 + 1: FEATURE SET {0001} V0 + 2: CENTURION DEVICE INFO {0100} V0 + Firmware: 1 0.02 + Hardware: model 26 rev 255 product 0508 + 3: CENTPP BRIDGE {0003} + 4: CENTURION GENERIC DFU {010A} + Headset (via CentPPBridge): + 0: ROOT {0000} V0 + 1: FEATURE SET {0001} V0 + 2: CENTURION DEVICE NAME {0101} + 3: CENTURION DEVICE INFO {0100} V0 + Firmware: 1 0.02 + Serial: + Hardware: model 26 rev 255 product 0508 + 4: CENTURION BATTERY SOC {0104} + Battery: 80%, BatteryStatus.DISCHARGING. + 5: CENTURION GENERIC DFU {010A} + 6: CENTURION UNKNOWN 0108 {0108} + 7: HEADSET AUDIO SIDETONE {0604} + Headset Sidetone: 50 + 8: HEADSET MIC SNR {0602} + Mic SNR: True + 9: HEADSET ONBOARD EQ {0636} + Battery: 80%, BatteryStatus.DISCHARGING. diff --git a/lib/logitech_receiver/base.py b/lib/logitech_receiver/base.py index 1cba0df3..0bbe5f88 100644 --- a/lib/logitech_receiver/base.py +++ b/lib/logitech_receiver/base.py @@ -98,8 +98,8 @@ 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 TX: [0x51, inner_report_id, seq=0x00, feat_idx, func_sw, params..., pad] -# Wire format RX: [0x51, len_byte, seq=0x00, feat_idx, func_sw, data..., pad] +# Wire format (CPL): [0x51, cpl_length, flags=0x00, feat_idx, func_sw, params..., pad] +# 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_FRAME_SIZE = 64 # 1 byte report ID + 63 bytes payload @@ -334,11 +334,12 @@ def write(handle, devnumber, data, long_message=False): ihandle = int(handle) if ihandle in _centurion_handles: - # Centurion framing: [0x51, inner_report_id, seq=0x00, feat_idx, func_sw, params...] - # The device_index is stripped — only the inner report_id + HID++ payload remain. - inner_report_id = ord(wdata[:1]) # 0x10 or 0x11 + # 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. payload = wdata[2:] # skip report_id and devnumber from standard frame - wdata = struct.pack("!BBB", CENTURION_REPORT_ID, inner_report_id, 0x00) + payload + 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 = wdata + b"\x00" * (CENTURION_FRAME_SIZE - len(wdata)) if logger.isEnabledFor(logging.DEBUG): @@ -359,21 +360,20 @@ def write(handle, devnumber, data, long_message=False): raise exceptions.NoReceiver(reason=reason) from reason -def write_centurion_raw(handle, inner_payload): - """Write a raw Centurion command (for headset features like sidetone/inactive time). +def write_centurion_cpl(handle, layer3_payload): + """Send a Centurion CPL frame with the given Layer 3+ payload. - The inner_payload is everything after [0x51, len, seq=0x00]. - The frame is: [0x51, len, 0x00, inner_payload..., zeros to 64 bytes]. - Length byte = len(inner_payload) + 1 (for seq byte). + Builds: [0x51, cpl_length, flags=0x00, layer3_payload..., zero-pad to 64 bytes] + where cpl_length = len(layer3_payload) + 1 (the +1 counts the flags byte). """ ihandle = int(handle) if ihandle not in _centurion_handles: - raise ValueError("write_centurion_raw called on non-Centurion handle") - length = len(inner_payload) + 1 # +1 for seq byte - wdata = struct.pack("!BBB", CENTURION_REPORT_ID, length, 0x00) + inner_payload + raise ValueError("write_centurion_cpl called on non-Centurion handle") + cpl_length = len(layer3_payload) + 1 # +1 for flags byte + wdata = struct.pack("!BBB", CENTURION_REPORT_ID, cpl_length, 0x00) + layer3_payload wdata = wdata + b"\x00" * (CENTURION_FRAME_SIZE - len(wdata)) if logger.isEnabledFor(logging.DEBUG): - logger.debug("(%s) <= centurion_raw[%s]", handle, common.strhex(wdata[:length + 2])) + logger.debug("(%s) <= centurion_cpl[%s]", handle, common.strhex(wdata[: cpl_length + 2])) try: hidapi.write(ihandle, wdata) except Exception as reason: @@ -446,12 +446,13 @@ def _read(handle, timeout) -> tuple[int, int, bytes]: raise exceptions.NoReceiver(reason=reason) from reason if data and is_centurion and ord(data[:1]) == CENTURION_REPORT_ID: - # Unwrap Centurion framing: - # RX: [0x51, len_byte, seq, feat_idx, func_sw, data...] + # 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...] - # Use len_byte to determine actual payload size (includes feat_idx + func_sw + response data). - payload_len = ord(data[1:2]) - inner_payload = data[3:3 + payload_len] # extract exactly payload_len bytes after seq + 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: @@ -629,13 +630,17 @@ def request( if reply_data[:1] == b"\xff" and reply_data[1:3] == request_data[:2]: # a HID++ 2.0 feature call returned with an error error = ord(reply_data[3:4]) + try: + error_name = Hidpp20ErrorCode(error) + except ValueError: + error_name = f"unknown:{error:02X}" logger.error( "(%s) device %d error on feature request {%04X}: %d = %s", handle, devnumber, request_id, error, - Hidpp20ErrorCode(error), + error_name, ) raise exceptions.FeatureCallError( number=devnumber, @@ -756,9 +761,9 @@ def _read_input_buffer(handle, ihandle, notifications_hook): if data: if is_centurion and ord(data[:1]) == CENTURION_REPORT_ID: - # Unwrap Centurion framing same as in _read() - payload_len = ord(data[1:2]) - inner_payload = data[3:3 + payload_len] + # 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)) diff --git a/lib/logitech_receiver/descriptors.py b/lib/logitech_receiver/descriptors.py index 65ffa5e5..41a66670 100644 --- a/lib/logitech_receiver/descriptors.py +++ b/lib/logitech_receiver/descriptors.py @@ -465,11 +465,4 @@ _D( kind=DEVICE_KIND.headset, usbid=0x0ABA, ) -_D( - "PRO X 2 LIGHTSPEED Gaming Headset", - codename="PRO X 2 Headset", - protocol=2.0, - interface=3, - kind=DEVICE_KIND.headset, - usbid=0x0AF7, -) +# PRO X 2 LIGHTSPEED Gaming Headset (0x0AF7) — fully probed via Centurion transport, no static descriptor needed diff --git a/lib/logitech_receiver/device.py b/lib/logitech_receiver/device.py index 3a8755b5..742403c7 100644 --- a/lib/logitech_receiver/device.py +++ b/lib/logitech_receiver/device.py @@ -19,6 +19,7 @@ from __future__ import annotations import errno import logging +import struct import threading import time import typing @@ -68,6 +69,20 @@ class LowLevelInterface(Protocol): ... +def _read_usb_product_string(hidraw_path): + """Read the USB product string from sysfs for a hidraw device path.""" + import pathlib + + try: + # /sys/class/hidraw/hidrawN/device/../../product → USB device product string + hidraw_name = pathlib.Path(hidraw_path).name + product_path = pathlib.Path("/sys/class/hidraw") / hidraw_name / "device" / ".." / ".." / "product" + product = product_path.read_text().strip() + return product if product else None + except (OSError, ValueError): + return None + + def create_device(low_level: LowLevelInterface, device_info, setting_callback=None): """Opens a Logitech Device found attached to the machine, by Linux device path. :returns: An open file handle for the found receiver, or None. @@ -128,8 +143,15 @@ class Device: self.hidpp_short = device_info.hidpp_short if device_info else None self.hidpp_long = device_info.hidpp_long if device_info else None self.centurion = device_info.centurion if device_info else False + self._centurion_usb_name = None if self.centurion: self.hidpp_long = True # Centurion devices always use long HID++ messages + # Read USB product string for device name — avoids slow bridge probe via CENTURION_DEVICE_NAME. + # device_info.product is often None (udev reads USB interface attrs, not device attrs), + # so fall back to reading from sysfs. + self._centurion_usb_name = getattr(device_info, "product", None) if device_info else None + if not self._centurion_usb_name and self.path: + self._centurion_usb_name = _read_usb_product_string(self.path) self.bluetooth = device_info.bus_id == 0x0005 if device_info else False # Bluetooth needs long messages self.hid_serial = device_info.serial if device_info else None self.setting_callback = setting_callback # for changes to settings @@ -236,10 +258,14 @@ class Device: def codename(self): if not self._codename: if self.online and self.protocol >= 2.0: - self._codename = _hidpp20.get_friendly_name(self) + if not self.centurion: + self._codename = _hidpp20.get_friendly_name(self) if not self._codename and self.name: - names = self.name.split(" ") - self._codename = names[1 if len(names) > 1 and names[0] == "Logitech" else 0] + if self.centurion: + self._codename = self.name + else: + names = self.name.split(" ") + self._codename = names[1 if len(names) > 1 and names[0] == "Logitech" else 0] if not self._codename and self.receiver: codename = self.receiver.device_codename(self.number) if codename: @@ -253,17 +279,40 @@ class Device: if not self._name: with self._simple_lock: if self._name is None: - if self.online and self.protocol >= 2.0: + if self.online and self.centurion: + self._name = _hidpp20.get_name_centurion(self) or getattr(self, "_centurion_usb_name", None) + if not self._name: + self._name = f"Unknown device {self.wpid or self.product_id}" + elif self.online and self.protocol >= 2.0: self._name = _hidpp20.get_name(self) return self._name or self._codename or f"Unknown device {self.wpid or self.product_id}" def get_ids(self): + if self.centurion: + self._get_ids_centurion() + return ids = _hidpp20.get_ids(self) if ids: self._unitId, self._modelId, self._tid_map = ids if logger.isEnabledFor(logging.INFO) and self._serial and self._serial != self._unitId: logger.info("%s: unitId %s does not match serial %s", self, self._unitId, self._serial) + def _get_ids_centurion(self): + if getattr(self, "_centurion_ids_done", False): + return + self._centurion_ids_done = True + serial = _hidpp20.get_serial_centurion(self) + if not serial or not serial.strip() or not serial.strip().isprintable(): + serial = _hidpp20.get_serial_centurion_sub(self) + if serial and serial.strip() and serial.strip().isprintable(): + self._serial = serial.strip() + self._unitId = self._serial + hw_info = _hidpp20.get_hardware_info_centurion(self) + if hw_info: + model_id, hw_revision, product_id = hw_info + self._modelId = f"{product_id:04X}" + self._tid_map = {"usbid": f"{product_id:04X}"} + @property def unitId(self): if not self._unitId and self.online and self.protocol >= 2.0: @@ -285,13 +334,32 @@ class Device: @property def kind(self): if not self._kind and self.online and self.protocol >= 2.0: - self._kind = _hidpp20.get_kind(self) + if self.centurion: + self._kind = self._infer_kind_centurion() + else: + self._kind = _hidpp20.get_kind(self) return self._kind or "?" + def _infer_kind_centurion(self): + """Infer device kind from Centurion features (sub-device or top-level).""" + # Check sub-device features (wireless via bridge) + for feature in getattr(self, "_centurion_sub_features", ()): + if isinstance(feature, int) and 0x0600 <= feature <= 0x06FF: + return hidpp10_constants.DEVICE_KIND.headset + # Check top-level features (direct USB connection, no bridge) + if self.features: + for feature, _index in self.features.enumerate(): + feat_int = int(feature) if isinstance(feature, int) else 0 + if 0x0600 <= feat_int <= 0x06FF: + return hidpp10_constants.DEVICE_KIND.headset + return None + @property def firmware(self) -> tuple[common.FirmwareInfo]: if self._firmware is None and self.online: - if self.protocol >= 2.0: + if self.centurion: + self._firmware = _hidpp20.get_firmware_centurion_sub(self) or _hidpp20.get_firmware_centurion(self) + elif self.protocol >= 2.0: self._firmware = _hidpp20.get_firmware(self) else: self._firmware = _hidpp10.get_firmware(self) @@ -299,6 +367,8 @@ class Device: @property def serial(self): + if not self._serial and self.online and self.centurion: + self.get_ids() return self._serial or "" @property @@ -555,20 +625,143 @@ class Device: def feature_request(self, feature, function=0x00, *params, no_reply=False): if self.protocol >= 2.0: + if self.centurion: + # Ensure sub-device features are discovered before routing decision + if self.features is not None: + self.features._check() + if feature in getattr(self, "_centurion_sub_features", ()): + sub_idx = self.features.get(feature) + if sub_idx is not None: + 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) - def centurion_raw_write(self, inner_payload): - """Send a raw Centurion command (for headset features not discoverable via IRoot). + 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. - inner_payload is everything after the [0x51, len, seq] header, e.g.: - sidetone: bytes([0x03, 0x1b, 0x00, 0x05, 0x00, 0x07, 0x1b, 0x01, level]) + Builds the 4-layer nested message: + Layer 1: [0x51] + Layer 2: [cpl_length, flags=0x00] + Layer 3: [bridge_idx, sendFragment_func|swid, bridge_hdr...] + Layer 4: [sub_cpl=0x00, sub_feat_idx, sub_func|swid, params...] + + Returns the sub-device response data (after bridge header), or None. """ if not getattr(self, "centurion", False): - raise ValueError("centurion_raw_write called on non-Centurion device") + raise ValueError("centurion_bridge_request called on non-Centurion device") + bridge_idx = getattr(self, "_centurion_bridge_index", None) + if bridge_idx is None: + raise ValueError("CentPPBridge index not discovered yet") handle = self.handle or (self.receiver.handle if self.receiver else None) - if handle: - from . import base - base.write_centurion_raw(handle, inner_payload) + if not handle: + return None + + sw_id = base._get_next_sw_id() + + # Build sub-device message: [sub_cpl=0x00, sub_feat_idx, sub_func|swid, params...] + # sub_function is in standard HID++ format: func_number << 4 (e.g. 0x10 for function 1) + sub_params = b"".join(struct.pack("B", p) if isinstance(p, int) else p for p in params) if params else b"" + sub_msg = struct.pack("BBB", 0x00, sub_feat_idx, (sub_function & 0xF0) | sw_id) + sub_params + + # Build bridge header: [device_id<<4 | len_hi, len_lo] + # device_id=0 for the headset, len is the sub-message length + sub_len = len(sub_msg) + bridge_hdr = struct.pack("BB", (0x00 << 4) | ((sub_len >> 8) & 0x0F), sub_len & 0xFF) + + # Build Layer 3: [bridge_idx, sendFragment_func(1)<<4|swid, bridge_hdr, sub_msg] + layer3 = struct.pack("BB", bridge_idx, (0x01 << 4) | sw_id) + bridge_hdr + sub_msg + + with base.acquire_timeout(base.handle_lock(handle), handle, base.DEFAULT_TIMEOUT): + base.write_centurion_cpl(handle, layer3) + + if no_reply: + return None + + # Read ACK response (immediate echo of bridge_idx + func|swid) + timeout = base.DEFAULT_TIMEOUT # same timeout as standard device requests + request_started = time.time() + ack_received = False + while time.time() - request_started < timeout: + reply = base._read(handle, timeout) + if not reply: + continue + _report_id, _devnumber, reply_data = reply + # ACK: short response echoing feat_idx and func|swid + if len(reply_data) >= 2 and reply_data[0] == bridge_idx: + func_sw = reply_data[1] + if (func_sw >> 4) == 0x01 and (func_sw & 0x0F) == sw_id: + ack_received = True + break + if (func_sw >> 4) == 0x01 and (func_sw & 0x0F) == 0: + # MessageEvent arrived before ACK — validate it's for our request + if self._is_bridge_response_for(reply_data, sub_feat_idx): + if logger.isEnabledFor(logging.DEBUG): + logger.debug("bridge idx=%d fn=0x%02X -> OK", sub_feat_idx, sub_function) + return self._parse_bridge_response(reply_data) + # Unsolicited notification, skip it + if not ack_received: + logger.warning("centurion_bridge_request: no ACK received") + return None + + # Read MessageEvent response (bridge function 1 with SW ID 0 = event) + while time.time() - request_started < timeout: + reply = base._read(handle, timeout) + if not reply: + continue + _report_id, _devnumber, reply_data = reply + if len(reply_data) >= 2 and reply_data[0] == bridge_idx: + func_sw = reply_data[1] + if (func_sw >> 4) == 0x01 and (func_sw & 0x0F) == 0: + if self._is_bridge_response_for(reply_data, sub_feat_idx): + if logger.isEnabledFor(logging.DEBUG): + logger.debug("bridge idx=%d fn=0x%02X -> OK", sub_feat_idx, sub_function) + return self._parse_bridge_response(reply_data) + # Unsolicited notification for a different feature, skip it + logger.warning("centurion_bridge_request: no MessageEvent received") + return None + + @staticmethod + def _is_bridge_response_for(reply_data, expected_sub_feat_idx): + """Check if a bridge MessageEvent is a response for our specific sub-feature request. + + Accepts both normal responses (sub_feat_idx matches) and error responses + (sub_feat_idx=0xFF with original feat_idx in next byte). + Unsolicited notifications (sub_cpl=0xFF) are rejected. + """ + if len(reply_data) < 6: + return False + sub_cpl = reply_data[4] + sub_feat_idx = reply_data[5] + # Notifications have sub_cpl=0xFF; our responses have sub_cpl=0x00 + if sub_cpl != 0x00: + return False + if sub_feat_idx == expected_sub_feat_idx: + return True + # Error response: sub_feat_idx=0xFF, next byte is the original feat_idx that errored + if sub_feat_idx == 0xFF and len(reply_data) >= 7 and reply_data[6] == expected_sub_feat_idx: + return True + return False + + @staticmethod + def _parse_bridge_response(reply_data): + """Extract sub-device response from a CentPPBridge MessageEvent. + + reply_data layout (after report_id and devnumber have been stripped): + [bridge_idx, func_sw, dev_id<<4|len_hi, len_lo, sub_cpl, sub_feat_idx, sub_func_sw, data...] + Returns the sub-device data starting from sub_feat_idx onward. + + Error responses have sub_feat_idx=0xFF: [... sub_cpl, 0xFF, orig_feat_idx, orig_func_sw, error_code] + These return None. + """ + if len(reply_data) < 7: + return None + sub_feat_idx = reply_data[5] + # Error response from sub-device + if sub_feat_idx == 0xFF: + error_code = reply_data[8] if len(reply_data) > 8 else 0 + orig_feat_idx = reply_data[6] if len(reply_data) > 6 else 0 + logger.debug("bridge sub-device error: feat_idx=%d error=0x%02X", orig_feat_idx, error_code) + return None + return reply_data[7:] # response data after sub_cpl, sub_feat_idx, sub_func_sw def ping(self): """Checks if the device is online and present, returns True of False. diff --git a/lib/logitech_receiver/hidpp20.py b/lib/logitech_receiver/hidpp20.py index 297ea488..58ff7bc6 100644 --- a/lib/logitech_receiver/hidpp20.py +++ b/lib/logitech_receiver/hidpp20.py @@ -173,31 +173,26 @@ class FeaturesArray(dict): self.supported = False return False - # Features accessible via raw Centurion commands on headsets. - # These aren't enumerated by IRoot/FeatureSet but work via hardcoded command bytes. - CENTURION_HEADSET_FEATURES = [ - SupportedFeature.HEADSET_AUDIO_SIDETONE, - ] - def _check_centurion(self, fs_index, count_response): - """Enumerate features on a Centurion device. + """Enumerate features on a Centurion device (parent + sub-device via CentPPBridge). - Centurion FeatureSet.getCount() returns the total feature count INCLUDING ROOT. - Centurion FeatureSet.getFeatureID(index) returns [remaining_count, feat_hi, feat_lo] - where remaining_count decreases with each index. The feature ID is in bytes 1-2. + Phase A: Enumerate parent device features via CenturionFeatureSet. + Find the CentPPBridge index (feature ID 0x0003 on Centurion = CentPPBridge). + Phase B: Route through CentPPBridge to discover sub-device features. + Use CenturionFeatureSet bulk query to get all sub-device features. + Store sub-device features keyed by SupportedFeature enum. """ - feature_count = count_response[0] # includes ROOT, unlike standard HID++ 2.0 + # Phase A: Parent features + feature_count = count_response[0] # includes ROOT on Centurion self.count = feature_count + bridge_index = None for index in range(feature_count): if self.inverse.get(index) is not None: continue # already registered (ROOT=0, FEATURE_SET=fs_index) - # Use device.request() directly to avoid recursion response = self.device.request((fs_index << 8) | 0x10, index) - if response is None: - continue - # Response: [remaining_count, feat_hi, feat_lo, ...] - if len(response) < 3: + if response is None or len(response) < 3: continue + # Centurion FeatureSet response: [remaining_count, feat_hi, feat_lo, type, flags] feat_id = struct.unpack("!H", response[1:3])[0] try: feature = SupportedFeature(feat_id) @@ -205,14 +200,68 @@ class FeaturesArray(dict): feature = f"unknown:{feat_id:04X}" self[feature] = index self.inverse[index] = feature - # Register additional headset features accessible via raw Centurion commands. - # These aren't enumerated by IRoot but are known to work on Centurion headsets. - next_idx = feature_count - for feat in self.CENTURION_HEADSET_FEATURES: - self[feat] = next_idx - self.inverse[next_idx] = feat - next_idx += 1 - self.count = next_idx + # Feature 0x0003 on Centurion = CentPPBridge (not FirmwareInfo) + if feat_id == 0x0003: + bridge_index = index + + if bridge_index is not None: + self.device._centurion_bridge_index = bridge_index + self.device._centurion_sub_features = set() + self.device._centurion_sub_indices = {} + self._discover_sub_device_features(bridge_index) + + def _discover_sub_device_features(self, bridge_index): + """Phase B: Discover sub-device features via CentPPBridge. + + Uses CenturionFeatureSet bulk query (function 1, index 0) routed through + the bridge to get all sub-device features at once. + """ + # First, find the sub-device's FeatureSet index via CenturionRoot (sub_feat_idx=0) + # Query: CenturionRoot.GetFeature(0x0001) to find FeatureSet index on sub-device + fs_id_hi = (SupportedFeature.FEATURE_SET >> 8) & 0xFF + fs_id_lo = SupportedFeature.FEATURE_SET & 0xFF + response = self.device.centurion_bridge_request(0x00, 0x00, fs_id_hi, fs_id_lo) + if response is None or len(response) < 1: + logger.warning("Failed to find FeatureSet on Centurion sub-device") + return + sub_fs_index = response[0] + if sub_fs_index == 0: + logger.warning("Sub-device FeatureSet not found (index=0)") + return + + # Bulk enumerate: CenturionFeatureSet.GetFeatureId(func=1=0x10, start_index=0) + # Response: [count, (feat_hi, feat_lo, type, flags) × count] + response = self.device.centurion_bridge_request(sub_fs_index, 0x10, 0x00) + if response is None or len(response) < 1: + logger.warning("Failed to enumerate sub-device features") + return + + entry_count = response[0] + entries = response[1:] + sub_feat_idx = 0 # sub-device feature indices start at 0 + for i in range(entry_count): + offset = i * 4 + if offset + 2 > len(entries): + break + feat_id = struct.unpack("!H", entries[offset : offset + 2])[0] + try: + feature = SupportedFeature(feat_id) + except ValueError: + feature = f"unknown:{feat_id:04X}" + # Store sub-device index for ALL features (including parent overlaps) + # This enables querying the sub-device's copy of shared features via bridge + self.device._centurion_sub_indices[feature] = sub_feat_idx + # Only store unique sub-device features in dict (skip parent overlaps like ROOT, FEATURE_SET) + # This avoids clobbering parent inverse entries via __setitem__ + if dict.get(self, feature) is None: + dict.__setitem__(self, feature, sub_feat_idx) + self.device._centurion_sub_features.add(feature) + # Always store in offset inverse for sub-device enumerate/display + self.inverse[sub_feat_idx + 0x100] = feature + if logger.isEnabledFor(logging.DEBUG): + logger.debug("Centurion sub-device feature: %s at sub-index %d", feature, sub_feat_idx) + sub_feat_idx += 1 + self._sub_feature_count = sub_feat_idx def get_feature(self, index: int) -> SupportedFeature | None: feature = self.inverse.get(index) @@ -222,28 +271,23 @@ class FeaturesArray(dict): feature = self.inverse.get(index) if feature is not None: return feature + # On Centurion devices, all features are discovered upfront (parent + sub-device) + if getattr(self.device, "centurion", False): + return None try: response = self.device.feature_request(SupportedFeature.FEATURE_SET, 0x10, index) except exceptions.FeatureCallError: logger.warning("failed to retrieve feature at index %d", index) return None if response: - if getattr(self.device, "centurion", False): - # Centurion response: [remaining_count, feat_hi, feat_lo] - if len(response) >= 3: - data = struct.unpack("!H", response[1:3])[0] - else: - return None - else: - data = struct.unpack("!H", response[:2])[0] + data = struct.unpack("!H", response[:2])[0] try: feature = SupportedFeature(data) except ValueError: feature = f"unknown:{data:04X}" self[feature] = index - if not getattr(self.device, "centurion", False): - self.version[feature] = response[3] - self.flags[feature] = response[2] + self.version[feature] = response[3] + self.flags[feature] = response[2] return feature def enumerate(self): # return all features and their index, ordered by index @@ -252,6 +296,12 @@ class FeaturesArray(dict): feature = self.get_feature(index) if feature is not None: yield feature, index + # Also yield sub-device features for Centurion devices + sub_count = getattr(self, "_sub_feature_count", 0) + for sub_idx in range(sub_count): + feature = self.inverse.get(sub_idx + 0x100) + if feature is not None: + yield feature, sub_idx def get_feature_version(self, feature: NamedInt) -> Optional[int]: if self[feature]: @@ -303,7 +353,7 @@ class FeaturesArray(dict): raise ValueError("Don't delete features from FeatureArray") def __len__(self) -> int: - return self.count + return self.count + getattr(self, "_sub_feature_count", 0) __bool__ = __nonzero__ = _check @@ -1635,6 +1685,108 @@ class Hidpp20: fw.append(fw_info) return tuple(fw) + def get_firmware_centurion(self, device): + """Reads firmware info from a Centurion device via DeviceInfo (0x0100) function 1.""" + fw = [] + seen = set() # track response signatures to detect duplicates + for index in range(0, 8): # try up to 8 entities + try: + report = device.feature_request(SupportedFeature.CENTURION_DEVICE_INFO, 0x10, index) + except exceptions.FeatureCallError: + break + if not report or len(report) < 5: + break + # Dedup: parent device returns the same response for every entity index + sig = bytes(report[: 5 + report[4]]) + if sig in seen: + break + seen.add(sig) + fw_type = report[0] + version = struct.unpack("!H", report[2:4])[0] + name_len = report[4] + name = report[5 : 5 + name_len].decode("ascii", errors="replace").rstrip("\x00") if name_len else "" + version_str = f"{version >> 8}.{version & 0xFF:02d}" + kind = FirmwareKind(fw_type) if fw_type <= 3 else FirmwareKind.Other + fw.append(common.FirmwareInfo(kind, name, version_str, None)) + return tuple(fw) if fw else None + + def get_serial_centurion(self, device): + """Reads the serial number from a Centurion device via DeviceInfo (0x0100) function 2.""" + try: + report = device.feature_request(SupportedFeature.CENTURION_DEVICE_INFO, 0x20) + except exceptions.FeatureCallError: + return None + if not report or len(report) < 2: + return None + str_len = report[0] + return report[1 : 1 + str_len].decode("ascii", errors="replace").rstrip("\x00") + + def get_hardware_info_centurion(self, device): + """Reads hardware info from a Centurion device via DeviceInfo (0x0100) function 0. + + Returns (modelId, hardwareRevision, productId) or None. + """ + try: + report = device.feature_request(SupportedFeature.CENTURION_DEVICE_INFO) + except exceptions.FeatureCallError: + return None + if not report or len(report) < 4: + return None + model_id = report[0] + hw_revision = report[1] + product_id = struct.unpack("!H", report[2:4])[0] + return model_id, hw_revision, product_id + + def _centurion_sub_device_info_request(self, device, function=0x00, *params): + """Send a DeviceInfo (0x0100) request to the sub-device via bridge.""" + sub_indices = getattr(device, "_centurion_sub_indices", {}) + sub_idx = sub_indices.get(SupportedFeature.CENTURION_DEVICE_INFO) + if sub_idx is None: + return None + return device.centurion_bridge_request(sub_idx, function, *params) + + def get_firmware_centurion_sub(self, device): + """Reads firmware info from the Centurion sub-device (headset) via bridge.""" + fw = [] + seen = set() + for index in range(0, 8): + report = self._centurion_sub_device_info_request(device, 0x10, index) + if not report or len(report) < 5: + break + sig = bytes(report[: 5 + report[4]]) + if sig in seen: + break + seen.add(sig) + fw_type = report[0] + version = struct.unpack("!H", report[2:4])[0] + name_len = report[4] + name = report[5 : 5 + name_len].decode("ascii", errors="replace").rstrip("\x00") if name_len else "" + version_str = f"{version >> 8}.{version & 0xFF:02d}" + kind = FirmwareKind(fw_type) if fw_type <= 3 else FirmwareKind.Other + fw.append(common.FirmwareInfo(kind, name, version_str, None)) + return tuple(fw) if fw else None + + def get_serial_centurion_sub(self, device): + """Reads the serial number from the Centurion sub-device (headset) via bridge.""" + report = self._centurion_sub_device_info_request(device, 0x20) + if not report or len(report) < 2: + return None + str_len = report[0] + return report[1 : 1 + str_len].decode("ascii", errors="replace").rstrip("\x00") + + def get_hardware_info_centurion_sub(self, device): + """Reads hardware info from the Centurion sub-device (headset) via bridge. + + Returns (modelId, hardwareRevision, productId) or None. + """ + report = self._centurion_sub_device_info_request(device) + if not report or len(report) < 4: + return None + model_id = report[0] + hw_revision = report[1] + product_id = struct.unpack("!H", report[2:4])[0] + return model_id, hw_revision, product_id + def get_ids(self, device): """Reads a device's ids (unit and model numbers)""" ids = device.feature_request(SupportedFeature.DEVICE_FW_VERSION) @@ -1686,6 +1838,38 @@ class Hidpp20: return name.decode("utf-8") + def get_name_centurion(self, device): + """Reads a Centurion device's name via DeviceName (0x0101). + + Tries two response formats: + 1. Inline: function 0 returns [name_len, name_bytes...] (like serial) + 2. Chunked: function 0 returns [name_len], function 1 returns [name_bytes...] (like standard DeviceName) + """ + try: + reply = device.feature_request(SupportedFeature.CENTURION_DEVICE_NAME) + except exceptions.FeatureCallError: + return None + if not reply: + return None + name_length = reply[0] + if name_length == 0: + return None + # If the full name is inline (length + name bytes in one response) + if len(reply) >= 1 + name_length: + return reply[1 : 1 + name_length].decode("utf-8", errors="replace").rstrip("\x00") + # Otherwise, fetch name in chunks via function 1 (like standard DEVICE_NAME) + name = b"" + while len(name) < name_length: + try: + fragment = device.feature_request(SupportedFeature.CENTURION_DEVICE_NAME, 0x10, len(name)) + except exceptions.FeatureCallError: + break + if fragment: + name += fragment[: name_length - len(name)] + else: + break + return name.decode("utf-8", errors="replace").rstrip("\x00") if name else None + def get_friendly_name(self, device: Device): """Reads a device's friendly name. @@ -1730,6 +1914,16 @@ class Hidpp20: except exceptions.FeatureCallError: return SupportedFeature.ADC_MEASUREMENT if SupportedFeature.ADC_MEASUREMENT in device.features else None + def get_battery_centurion(self, device: Device): + try: + report = device.feature_request(SupportedFeature.CENTURION_BATTERY_SOC) + if report is not None: + return decipher_battery_centurion(report) + except exceptions.FeatureCallError: + if SupportedFeature.CENTURION_BATTERY_SOC in device.features: + return SupportedFeature.CENTURION_BATTERY_SOC + return None + def get_battery(self, device, feature): """Return battery information - feature, approximate level, next, charging, voltage or battery feature if there is one but it is not responding or None for no battery feature""" @@ -1961,6 +2155,7 @@ battery_functions = { SupportedFeature.BATTERY_VOLTAGE: Hidpp20.get_battery_voltage, SupportedFeature.UNIFIED_BATTERY: Hidpp20.get_battery_unified, SupportedFeature.ADC_MEASUREMENT: Hidpp20.get_adc_measurement, + SupportedFeature.CENTURION_BATTERY_SOC: Hidpp20.get_battery_centurion, } @@ -2041,6 +2236,19 @@ def decipher_battery_unified(report) -> tuple[SupportedFeature, Battery]: return SupportedFeature.UNIFIED_BATTERY, Battery(discharge if discharge else approx_level, None, status, None) +def decipher_battery_centurion(report) -> tuple[SupportedFeature, Battery]: + """Decipher CENTURION_BATTERY_SOC (0x0104) response.""" + if len(report) < 1: + return SupportedFeature.CENTURION_BATTERY_SOC, Battery(None, None, BatteryStatus.DISCHARGING, None) + soc = report[0] # state of charge percentage (tentative — first byte) + logger.debug("centurion battery SOC raw: %s", report[:8].hex()) + status = BatteryStatus.DISCHARGING + # Check for charging indicator if available + if len(report) >= 2 and report[1] & 0x01: + status = BatteryStatus.RECHARGING + return SupportedFeature.CENTURION_BATTERY_SOC, Battery(soc, None, status, None) + + def decipher_adc_measurement(report) -> tuple[SupportedFeature, Battery]: # partial implementation - needs mapping to levels adc_voltage, flags = struct.unpack("!HB", report[:3]) diff --git a/lib/logitech_receiver/hidpp20_constants.py b/lib/logitech_receiver/hidpp20_constants.py index f9ed6f2c..b54a9fef 100644 --- a/lib/logitech_receiver/hidpp20_constants.py +++ b/lib/logitech_receiver/hidpp20_constants.py @@ -184,6 +184,8 @@ class SupportedFeature(IntEnum): CENTURION_DEVICE_NAME = 0x0101 CENTURION_ROOT = 0x0102 CENTURION_MEMFAULT = 0x0103 + CENTURION_BATTERY_SOC = 0x0104 + CENTURION_UNKNOWN_0108 = 0x0108 CENTURION_GENERIC_DFU = 0x010A # Headsets (Centurion-era) HEADSET_VOLUME = 0x0200 @@ -200,6 +202,13 @@ class SupportedFeature(IntEnum): HEADSET_BATTERY_SAVER = 0x0618 HEADSET_RGB_HOSTMODE = 0x0620 HEADSET_DO_NOT_DISTURB = 0x0631 + CENTURION_ONBOARD_PROFILES = 0x0634 + HEADSET_ONBOARD_EQ = 0x0636 + HEADSET_NOISE_EXPOSURE = 0x060D + HEADSET_USAGE_TRACKING = 0x0617 + HEADSET_RGB_ONBOARD_EFFECTS = 0x0621 + HEADSET_RGB_SIGNATURE_EFFECTS = 0x0622 + HEADSET_RGB_STREAMING = 0x0635 # Fake features for Solaar internal use MOUSE_GESTURE = 0xFE00 diff --git a/lib/logitech_receiver/settings.py b/lib/logitech_receiver/settings.py index 02fac7f4..25257910 100644 --- a/lib/logitech_receiver/settings.py +++ b/lib/logitech_receiver/settings.py @@ -650,38 +650,6 @@ class FeatureRW: return reply if not self.no_reply else True -class CenturionRawRW: - """RW class for raw Centurion headset commands (sidetone, inactive time). - - These features aren't discoverable via IRoot/FeatureSet but can be - controlled by sending raw Centurion frames (reverse-engineered from G Hub). - The read() returns None since these features are write-only; the last-set - value is persisted by the Setting infrastructure. - """ - - kind = NamedInt(0x02, _("feature")) - - def __init__(self, feature, command_template=b"", value_offset=0, **kwargs): - self.feature = feature - self.command_template = command_template - self.value_offset = value_offset - - def read(self, device, data_bytes=b""): - # Write-only: return default (0) so the UI has a valid initial value. - # The persisted value (if any) takes priority via Setting._pre_read(). - return b"\x00" - - def write(self, device, data_bytes): - if isinstance(data_bytes, int): - value = data_bytes - else: - value = int.from_bytes(data_bytes, "big") - cmd = bytearray(self.command_template) - cmd[self.value_offset] = value & 0xFF - device.centurion_raw_write(bytes(cmd)) - return True - - class FeatureRWMap(FeatureRW): kind = NamedInt(0x02, _("feature")) default_read_fnid = 0x00 diff --git a/lib/logitech_receiver/settings_templates.py b/lib/logitech_receiver/settings_templates.py index 41191423..c103c212 100644 --- a/lib/logitech_receiver/settings_templates.py +++ b/lib/logitech_receiver/settings_templates.py @@ -1637,26 +1637,15 @@ class HeadsetAINRLevel(settings.Setting): class HeadsetSidetone(settings.Setting): name = "headset-sidetone" label = _("Headset Sidetone") - description = _("Sidetone level (0 = off, 100 = max). Write-only: value is persisted locally.") + description = _("Sidetone level (0 = off, 100 = max).") feature = _F.HEADSET_AUDIO_SIDETONE - rw_class = settings.CenturionRawRW - # Inner payload: [0x03, 0x1b, 0x00, 0x05, 0x00, 0x07, 0x1b, 0x01, LEVEL] - # Reverse-engineered from G Hub / HeadsetControl for PRO X 2 LIGHTSPEED - rw_options = { - "command_template": bytes([0x03, 0x1b, 0x00, 0x05, 0x00, 0x07, 0x1b, 0x01, 0x00]), - "value_offset": 8, - } + rw_options = {"read_fnid": 0x00, "write_fnid": 0x10} validator_class = settings_validator.RangeValidator min_value = 0 max_value = 100 - - @classmethod - def build(cls, device): - setting = super().build(device) - if setting: - # Write-only: skip current-value comparison so writes always go through - setting._validator.needs_current_value = False - return setting + # GetSidetone returns [mic_id, flag, level] — skip 2 bytes to read level + # SetSidetone takes [mic_id, level] — prefix mic_id=1 before level + validator_options = {"read_skip_byte_count": 2, "write_prefix_bytes": b"\x01"} class HeadsetMicGain(settings.Setting): diff --git a/lib/solaar/cli/show.py b/lib/solaar/cli/show.py index 79f9bfb9..4f1d128f 100644 --- a/lib/solaar/cli/show.py +++ b/lib/solaar/cli/show.py @@ -110,7 +110,8 @@ def _print_device(dev, num=None): print(" Codename :", dev.codename) print(" Kind :", dev.kind) if dev.protocol: - print(f" Protocol : HID++ {dev.protocol:1.1f}") + proto_name = "Centurion" if getattr(dev, "centurion", False) else "HID++" + print(f" Protocol : {proto_name} {dev.protocol:1.1f}") else: print(" Protocol : unknown (device is offline)") if dev.polling_rate: @@ -144,25 +145,43 @@ def _print_device(dev, num=None): print(" Features: (none)") if dev.online and dev.features: - print(f" Supports {len(dev.features)} HID++ 2.0 features:") + is_centurion = getattr(dev, "centurion", False) + parent_count = dev.features.count + sub_count = getattr(dev.features, "_sub_feature_count", 0) + if is_centurion and sub_count > 0: + print(f" Supports {parent_count} dongle + {sub_count} headset features:") + else: + print(f" Supports {len(dev.features)} HID++ 2.0 features:") dev_settings = [] settings_templates.check_feature_settings(dev, dev_settings) + feature_num = 0 + in_sub_device = False for feature, index in dev.features.enumerate(): + if is_centurion and not in_sub_device and feature_num >= parent_count: + in_sub_device = True + print(" Headset (via CentPPBridge):") + feature_num += 1 if isinstance(feature, str): feature_bytes = bytes.fromhex(feature[-4:]) else: feature_bytes = feature.to_bytes(2, byteorder="little") feature_int = int.from_bytes(feature_bytes, byteorder="little") + # On Centurion, parent feature 0x0003 is CentPPBridge, not DEVICE_FW_VERSION + display_name = "CENTPP BRIDGE" if is_centurion and not in_sub_device and feature_int == 0x0003 else feature try: flags = dev.request(0x0000, feature_bytes) except Exception: - print(" %2d: %-22s {%04X} - can't retrieve" % (index, feature, feature_int)) - continue - flags = 0 if flags is None else ord(flags[1:2]) - flags = common.flag_names(hidpp20_constants.FeatureFlag, flags) - version = dev.features.get_feature_version(feature_int) - version = version if version else 0 - print(" %2d: %-22s {%04X} V%s %s " % (index, feature, feature_int, version, ", ".join(flags))) + flags = None + if flags is not None: + flags = ord(flags[1:2]) + flag_names = common.flag_names(hidpp20_constants.FeatureFlag, flags) + version = dev.features.get_feature_version(feature_int) + version = version if version else 0 + print( + " %2d: %-22s {%04X} V%s %s " % (index, display_name, feature_int, version, ", ".join(flag_names)) + ) + else: + print(" %2d: %-22s {%04X}" % (index, display_name, feature_int)) if feature == SupportedFeature.HIRES_WHEEL: wheel = _hidpp20.get_hires_wheel(dev) if wheel: @@ -230,7 +249,25 @@ def _print_device(dev, num=None): print(f" Kind: {_hidpp20.get_kind(dev)}") elif feature == SupportedFeature.DEVICE_FRIENDLY_NAME: print(f" Friendly Name: {_hidpp20.get_friendly_name(dev)}") - elif feature == SupportedFeature.DEVICE_FW_VERSION: + elif feature == SupportedFeature.CENTURION_DEVICE_INFO: + if in_sub_device: + # Use cached device properties to avoid redundant bridge requests + fw_list = dev.firmware + serial = dev.serial + hw_info = _hidpp20.get_hardware_info_centurion_sub(dev) + else: + fw_list = _hidpp20.get_firmware_centurion(dev) + serial = _hidpp20.get_serial_centurion(dev) + hw_info = _hidpp20.get_hardware_info_centurion(dev) + if fw_list: + for fw in fw_list: + print(f" Firmware: {(str(fw.kind) + ' ' + fw.name).strip()} {fw.version}") + if serial and serial.strip() and serial.strip().isprintable(): + print(f" Serial: {serial}") + if hw_info: + model_id, hw_rev, product_id = hw_info + print(f" Hardware: model {model_id}" f" rev {hw_rev} product {product_id:04X}") + elif feature == SupportedFeature.DEVICE_FW_VERSION and not (is_centurion and not in_sub_device): for fw in _hidpp20.get_firmware(dev): extras = strhex(fw.extras) if fw.extras else "" print(f" Firmware: {fw.kind} {fw.name} {fw.version} {extras}") diff --git a/tests/logitech_receiver/fake_hidpp.py b/tests/logitech_receiver/fake_hidpp.py index 01f2d7a6..7758f2f5 100644 --- a/tests/logitech_receiver/fake_hidpp.py +++ b/tests/logitech_receiver/fake_hidpp.py @@ -386,6 +386,7 @@ class Device: version: Optional[int] = 0 wpid: Optional[str] = "0000" setting_callback: Any = None + centurion: bool = False sliding = profiles = _backlight = _keys = _remap_keys = _led_effects = _gestures = None _gestures_lock = threading.Lock() number = "d1" @@ -400,6 +401,7 @@ class Device: gestures = device.Device.gestures __hash__ = device.Device.__hash__ feature_request = device.Device.feature_request + _infer_kind_centurion = device.Device._infer_kind_centurion def __post_init__(self): self._name = self.name @@ -424,6 +426,9 @@ class Device: if self.setting_callback is None: self.setting_callback = lambda x, y, z: None self.add_notification_handler = lambda x, y: None + # Centurion bridge responses: keyed by (sub_feat_idx, sub_function) -> response bytes + if not hasattr(self, "_bridge_responses"): + self._bridge_responses = {} def request(self, id, *params, no_reply=False, long_message=False, protocol=2.0): params = b"".join(pack("B", p) if isinstance(p, int) else p for p in params) @@ -434,6 +439,24 @@ class Device: return bytes.fromhex(r.response) if isinstance(r.response, str) else r.response print("RESPONSE", self._name, None) + def centurion_bridge_request(self, sub_feat_idx, sub_function=0x00, *params, no_reply=False): + """Fake bridge request — looks up (sub_feat_idx, sub_function, params) in _bridge_responses.""" + params_bytes = b"".join(pack("B", p) if isinstance(p, int) else p for p in params) if params else b"" + key = (sub_feat_idx, sub_function, params_bytes.hex().upper()) + print("BRIDGE ", self._name, f"sub_idx={sub_feat_idx} func={sub_function} params={params_bytes.hex().upper()}") + result = self._bridge_responses.get(key) + if result is not None: + print("BRIDGE_R", self._name, result.hex().upper()) + return result + # Try without params for convenience + key_no_params = (sub_feat_idx, sub_function, "") + result = self._bridge_responses.get(key_no_params) + if result is not None: + print("BRIDGE_R", self._name, result.hex().upper()) + return result + print("BRIDGE_R", self._name, None) + return None + def ping(self, handle=None, devnumber=None, long_message=False): print("PING", self._protocol) return self._protocol @@ -451,6 +474,25 @@ class Device: pass +# Centurion headset (PRO X 2 LIGHTSPEED) parent device responses. +# Parent has 5 features: ROOT(0), FeatureSet(1), DeviceInfo(2), CentPPBridge(3), GenericDFU(4) +# Feature 0x0003 on Centurion = CentPPBridge (NOT FirmwareInfo). +r_centurion_headset = [ + Response(2.6, 0x0010), # ping (protocol 2.6) + Response("010001", 0x0000, "0001"), # FeatureSet at index 1 + Response("020001", 0x0000, "0100"), # DeviceInfo at index 2 + Response("030001", 0x0000, "0003"), # CentPPBridge at index 3 + Response("040001", 0x0000, "010A"), # GenericDFU at index 4 + Response("05", 0x0100), # feature count = 5 (includes ROOT on Centurion) + # FeatureSet.getFeatureID responses: [remaining_count, feat_hi, feat_lo, type, flags] + Response("0400000000", 0x0110, "00"), # index 0: ROOT (0x0000) + Response("0300010001", 0x0110, "01"), # index 1: FeatureSet (0x0001) + Response("0201000001", 0x0110, "02"), # index 2: DeviceInfo (0x0100) + Response("0100030001", 0x0110, "03"), # index 3: CentPPBridge (0x0003) + Response("00010A0001", 0x0110, "04"), # index 4: GenericDFU (0x010A) +] + + def match_requests(number, responses, call_args_list): for i in range(0 - number, 0): param = b"".join(pack("B", p) if isinstance(p, int) else p for p in call_args_list[i][0][1:]).hex().upper() diff --git a/tests/logitech_receiver/test_hidpp20_complex.py b/tests/logitech_receiver/test_hidpp20_complex.py index dc7d4b85..a93a94f8 100644 --- a/tests/logitech_receiver/test_hidpp20_complex.py +++ b/tests/logitech_receiver/test_hidpp20_complex.py @@ -101,10 +101,9 @@ def test_FeaturesArray_get_feature(device, expected0, expected1, expected2, expe (hidpp20_constants.SupportedFeature.FEATURE_SET, 1), (hidpp20_constants.SupportedFeature.CONFIG_CHANGE, 2), (hidpp20_constants.SupportedFeature.DEVICE_FW_VERSION, 3), - (common.NamedInt(256, "unknown:0100"), 4), + (hidpp20_constants.SupportedFeature.CENTURION_DEVICE_INFO, 4), (hidpp20_constants.SupportedFeature.REPROG_CONTROLS_V4, 5), - (None, 6), - (None, 7), + # Indices 6 and 7 have no responses — get_feature returns None, enumerate skips them (hidpp20_constants.SupportedFeature.BATTERY_STATUS, 8), ], ), @@ -922,3 +921,210 @@ def test_onboard_profiles_device(responses, name, count, buttons, gbuttons, sect yml_dump = yaml.dump(profiles) assert yaml.safe_load(yml_dump).to_bytes().hex() == profiles.to_bytes().hex() + + +# --- Centurion (PRO X 2 LIGHTSPEED headset) tests --- + +device_centurion = fake_hidpp.Device("CENTURION", True, 2.6, fake_hidpp.r_centurion_headset, centurion=True) + + +def test_centurion_parent_feature_discovery(): + """Parent feature enumeration discovers CentPPBridge at index 3 and stores bridge index.""" + dev = fake_hidpp.Device("CENT_PARENT", True, 2.6, fake_hidpp.r_centurion_headset, centurion=True) + featuresarray = hidpp20.FeaturesArray(dev) + dev.features = featuresarray + + result = featuresarray._check() + + assert result is True + assert featuresarray.count == 5 + # Parent features registered + assert featuresarray[hidpp20_constants.SupportedFeature.ROOT] == 0 + assert featuresarray[hidpp20_constants.SupportedFeature.FEATURE_SET] == 1 + assert featuresarray[hidpp20_constants.SupportedFeature.CENTURION_DEVICE_INFO] == 2 + assert featuresarray[hidpp20_constants.SupportedFeature.CENTURION_GENERIC_DFU] == 4 + # Feature 0x0003 = CentPPBridge on Centurion (stored as DEVICE_FW_VERSION since same ID) + assert featuresarray[hidpp20_constants.SupportedFeature.DEVICE_FW_VERSION] == 3 + # Bridge index stored on device + assert dev._centurion_bridge_index == 3 + assert hasattr(dev, "_centurion_sub_features") + + +def test_centurion_sub_device_feature_discovery(): + """Sub-device feature discovery routes through bridge and populates _centurion_sub_features.""" + dev = fake_hidpp.Device("CENT_SUB", True, 2.6, fake_hidpp.r_centurion_headset, centurion=True) + # Set up bridge responses for sub-device discovery: + # 1. CenturionRoot.GetFeature(0x0001) -> FeatureSet at sub-index 1 + # 2. CenturionFeatureSet.GetFeatureId(index=0) -> bulk feature list + dev._bridge_responses = { + # CenturionRoot(idx=0).GetFeature(func=0) with feature_id=0x0001 -> sub_fs_index=1 + (0x00, 0x00, "0001"): bytes([0x01, 0x00, 0x00]), + # CenturionFeatureSet(idx=1).GetFeatureId(func=0x10, start=0) -> 3 features + # Response: [count, (feat_hi, feat_lo, type, flags) × count] + (0x01, 0x10, "00"): bytes( + [ + 0x03, # 3 features + 0x06, + 0x04, + 0x00, + 0x00, # HEADSET_AUDIO_SIDETONE (0x0604) at sub-idx 0 + 0x06, + 0x01, + 0x00, + 0x00, # HEADSET_MIC_MUTE (0x0601) at sub-idx 1 + 0x06, + 0x11, + 0x00, + 0x00, # HEADSET_MIC_GAIN (0x0611) at sub-idx 2 + ] + ), + } + featuresarray = hidpp20.FeaturesArray(dev) + dev.features = featuresarray + + featuresarray._check() + + # Sub-device features should be discovered + assert hidpp20_constants.SupportedFeature.HEADSET_AUDIO_SIDETONE in dev._centurion_sub_features + assert hidpp20_constants.SupportedFeature.HEADSET_MIC_MUTE in dev._centurion_sub_features + assert hidpp20_constants.SupportedFeature.HEADSET_MIC_GAIN in dev._centurion_sub_features + # Sub-device features should be in features array with their sub-device indices + assert featuresarray[hidpp20_constants.SupportedFeature.HEADSET_AUDIO_SIDETONE] == 0 + assert featuresarray[hidpp20_constants.SupportedFeature.HEADSET_MIC_MUTE] == 1 + assert featuresarray[hidpp20_constants.SupportedFeature.HEADSET_MIC_GAIN] == 2 + # _centurion_sub_indices should map ALL sub-device features to their sub-device indices + assert dev._centurion_sub_indices[hidpp20_constants.SupportedFeature.HEADSET_AUDIO_SIDETONE] == 0 + assert dev._centurion_sub_indices[hidpp20_constants.SupportedFeature.HEADSET_MIC_MUTE] == 1 + assert dev._centurion_sub_indices[hidpp20_constants.SupportedFeature.HEADSET_MIC_GAIN] == 2 + + +def test_centurion_feature_request_routes_sub_device(): + """feature_request() routes sub-device features through centurion_bridge_request().""" + dev = fake_hidpp.Device("CENT_ROUTE", True, 2.6, fake_hidpp.r_centurion_headset, centurion=True) + # Manually set up sub-device state (simulating completed discovery) + dev.features.count = 5 # mark discovery as complete so _check() short-circuits + dev._centurion_bridge_index = 3 + dev._centurion_sub_features = {hidpp20_constants.SupportedFeature.HEADSET_AUDIO_SIDETONE} + dev.features[hidpp20_constants.SupportedFeature.HEADSET_AUDIO_SIDETONE] = 7 # sub-device index + # Set up bridge response for GetSidetoneLevel + dev._bridge_responses = { + (7, 0x00, ""): bytes([0x01, 0x00, 0x32]), # mic_id=1, mute=0, level=50 + } + + result = dev.feature_request(hidpp20_constants.SupportedFeature.HEADSET_AUDIO_SIDETONE, 0x00) + + assert result is not None + assert result == bytes([0x01, 0x00, 0x32]) + + +def test_centurion_feature_request_parent_not_routed(): + """feature_request() does NOT route parent features through bridge.""" + dev = fake_hidpp.Device("CENT_PARENT2", True, 2.6, fake_hidpp.r_centurion_headset, centurion=True) + dev._centurion_bridge_index = 3 + dev._centurion_sub_features = {hidpp20_constants.SupportedFeature.HEADSET_AUDIO_SIDETONE} + + # FeatureSet is a parent feature — should go through normal request(), not bridge + featuresarray = hidpp20.FeaturesArray(dev) + dev.features = featuresarray + featuresarray._check() + + # CENTURION_DEVICE_INFO is a parent feature at index 2 — requesting it should + # NOT go through bridge, it should go through the normal hidpp20.feature_request path + assert hidpp20_constants.SupportedFeature.CENTURION_DEVICE_INFO not in dev._centurion_sub_features + + +def test_centurion_bridge_request_write(): + """centurion_bridge_request with no_reply=True returns None immediately.""" + dev = fake_hidpp.Device("CENT_WRITE", True, 2.6, fake_hidpp.r_centurion_headset, centurion=True) + dev._centurion_bridge_index = 3 + dev._centurion_sub_features = {hidpp20_constants.SupportedFeature.HEADSET_AUDIO_SIDETONE} + dev.features[hidpp20_constants.SupportedFeature.HEADSET_AUDIO_SIDETONE] = 7 + dev._bridge_responses = {} # no responses needed for no_reply + + result = dev.centurion_bridge_request(7, 0x10, 0x32, no_reply=True) + + assert result is None + + +def test_centurion_firmware_dedup(): + """get_firmware_centurion() deduplicates identical firmware entries.""" + # Simulate parent device that returns the same firmware for every entity index + fw_response = "00" + "00" + "0105" + "04" + "44303031" + "00" * 20 # type=0, ver=1.05, name="D001" + responses = fake_hidpp.r_centurion_headset + [fake_hidpp.Response(fw_response, 0x0210, f"{i:02X}") for i in range(8)] + dev = fake_hidpp.Device("CENT_DEDUP", True, 2.6, responses, centurion=True) + + fw = _hidpp20.get_firmware_centurion(dev) + + # Should only get 1 entry, not 8 + assert fw is not None + assert len(fw) == 1 + assert fw[0].name == "D001" + + +def test_centurion_sub_device_firmware(): + """get_firmware_centurion_sub() queries sub-device firmware via bridge.""" + dev = fake_hidpp.Device("CENT_SUBFW", True, 2.6, fake_hidpp.r_centurion_headset, centurion=True) + dev._centurion_bridge_index = 3 + dev._centurion_sub_features = set() + dev._centurion_sub_indices = {hidpp20_constants.SupportedFeature.CENTURION_DEVICE_INFO: 2} + # Sub-device firmware: type=0 (firmware), ver=3.02, name="H001" + dev._bridge_responses = { + (2, 0x10, "00"): bytes([0x00, 0x00, 0x03, 0x02, 0x04]) + b"H001", + (2, 0x10, "01"): bytes([0x00, 0x00, 0x03, 0x02, 0x04]) + b"H001", # duplicate → dedup stops + } + + fw = _hidpp20.get_firmware_centurion_sub(dev) + + assert fw is not None + assert len(fw) == 1 + assert fw[0].name == "H001" + assert fw[0].version == "3.02" + + +def test_centurion_sub_device_serial(): + """get_serial_centurion_sub() queries sub-device serial via bridge.""" + dev = fake_hidpp.Device("CENT_SUBSER", True, 2.6, fake_hidpp.r_centurion_headset, centurion=True) + dev._centurion_bridge_index = 3 + dev._centurion_sub_features = set() + dev._centurion_sub_indices = {hidpp20_constants.SupportedFeature.CENTURION_DEVICE_INFO: 2} + dev._bridge_responses = { + (2, 0x20, ""): bytes([0x0C]) + b"ABC123DEF456", + } + + serial = _hidpp20.get_serial_centurion_sub(dev) + + assert serial == "ABC123DEF456" + + +def test_centurion_sub_device_hardware_info(): + """get_hardware_info_centurion_sub() queries sub-device hardware info via bridge.""" + dev = fake_hidpp.Device("CENT_SUBHW", True, 2.6, fake_hidpp.r_centurion_headset, centurion=True) + dev._centurion_bridge_index = 3 + dev._centurion_sub_features = set() + dev._centurion_sub_indices = {hidpp20_constants.SupportedFeature.CENTURION_DEVICE_INFO: 2} + dev._bridge_responses = { + (2, 0x00, ""): bytes([0x01, 0x03, 0x0A, 0xF7]), + } + + hw_info = _hidpp20.get_hardware_info_centurion_sub(dev) + + assert hw_info is not None + model_id, hw_rev, product_id = hw_info + assert model_id == 1 + assert hw_rev == 3 + assert product_id == 0x0AF7 + + +def test_centurion_kind_inference(): + """Centurion device with 0x06xx audio features infers kind=headset.""" + dev = fake_hidpp.Device("CENT_KIND", True, 2.6, fake_hidpp.r_centurion_headset, centurion=True) + dev._centurion_sub_features = { + hidpp20_constants.SupportedFeature.HEADSET_AUDIO_SIDETONE, + hidpp20_constants.SupportedFeature.HEADSET_MIC_MUTE, + } + + kind = dev._infer_kind_centurion() + + from logitech_receiver import hidpp10_constants + + assert kind == hidpp10_constants.DEVICE_KIND.headset