Remove static PRO X 2 descriptor; fully probe Centurion devices at runtime
Replace the hardcoded descriptor entry with dynamic discovery of all device properties via the Centurion protocol. The headset name, kind, serial, firmware, and battery are now probed at runtime — matching how the device actually presents itself rather than relying on static data. Key changes: - Discover sub-device features via CentPPBridge and route requests through the bridge automatically - Infer device kind from feature IDs (0x06xx = headset) for both wireless and direct USB connections - Read device name from USB product string with protocol probe fallback - Parse bridge error responses (sub_feat_idx=0xFF) instead of timing out - Handle unknown HID++ error codes gracefully in base.py - Fix firmware deduplication for Centurion parent devices - Prefer sub-device serial/firmware over parent (non-printable) values - Add Centurion-aware display in solaar show with parent/sub-device sections - Support both wireless (0AF7 dongle) and direct USB (0AF8) connections
This commit is contained in:
Ken Sanislo
parent
693741f216
commit
8bdb49feec
|
|
@ -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: <redacted>
|
||||
Model ID: 0508
|
||||
Unit ID: <redacted>
|
||||
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: <redacted>
|
||||
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.
|
||||
|
|
@ -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))
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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.
|
||||
|
|
|
|||
|
|
@ -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])
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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):
|
||||
|
|
|
|||
|
|
@ -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}")
|
||||
|
|
|
|||
|
|
@ -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()
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
Loading…
Reference in New Issue