From d98dd3ed3f35eac54ef55e17cd483b318b4439d9 Mon Sep 17 00:00:00 2001 From: Ken Sanislo Date: Mon, 2 Mar 2026 18:29:55 -0700 Subject: [PATCH] Display Centurion dongle as receiver with headset as child device - Add CenturionReceiver class that provides the Receiver UI interface so the dongle appears as a parent with the headset indented underneath, matching how Lightspeed/Unifying receivers display - Independently probe dongle features via feature_request() on the CenturionReceiver, separate from headset features via bridge - Fix bridge notification dispatch: remove incorrect sub_cpl=0xFF filter that was silently dropping all battery and other notifications - Fix battery status decoding: charging status is at byte 2 (not byte 1) of the CENTURION_BATTERY_SOC response - Detect wired vs wireless by checking for CentPPBridge in discovered features; wired headsets fall back to standalone Device - Name the dongle "Centurion Receiver" to distinguish from the headset - Filter unprintable dongle serial (control characters 0x14-0x1F) - Update CLI show output with proper receiver/child hierarchy and spacing --- lib/logitech_receiver/device.py | 287 +++++++++++++++++- lib/logitech_receiver/hidpp20.py | 19 +- lib/logitech_receiver/hidpp20_constants.py | 2 +- lib/logitech_receiver/notifications.py | 3 + lib/logitech_receiver/settings_templates.py | 13 + lib/solaar/cli/__init__.py | 9 +- lib/solaar/cli/show.py | 146 ++++++--- lib/solaar/listener.py | 62 ++++ .../logitech_receiver/test_hidpp20_complex.py | 152 ++++++++++ 9 files changed, 644 insertions(+), 49 deletions(-) diff --git a/lib/logitech_receiver/device.py b/lib/logitech_receiver/device.py index 742403c7..988a7f00 100644 --- a/lib/logitech_receiver/device.py +++ b/lib/logitech_receiver/device.py @@ -83,6 +83,276 @@ def _read_usb_product_string(hidraw_path): return None +class CenturionReceiver: + """A lightweight receiver-like container for Centurion (PRO X 2 LIGHTSPEED) dongles. + + Provides the Receiver interface to the UI so the dongle appears as a parent + with the headset as an indented child device. NOT a subclass of Receiver — + Receiver's __init__ does HID++ 1.0 register reads and pairing setup that + don't apply to Centurion. + + All centurion communication (bridge, features, settings, battery) lives in + the child Device; this class is just a UI container + handle owner. + """ + + read_register: Callable = hidpp10.read_register + write_register: Callable = hidpp10.write_register + number = 0xFF + kind = None + isDevice = False + may_unpair = False + re_pairs = False + max_devices = 1 + + def __init__(self, low_level, handle, device_info, setting_callback=None): + assert handle + self.low_level = low_level + self.handle = handle + self.path = device_info.path + self.product_id = device_info.product_id + self.setting_callback = setting_callback + self.status_callback = None + self.notification_flags = None + self._devices = {} + self._firmware = None + self._dongle_features = None # independently probed dongle features + self.cleanups = [] + + # Receiver identity + self.serial = None + self._usb_name = getattr(device_info, "product", None) + if not self._usb_name and self.path: + self._usb_name = _read_usb_product_string(self.path) + self.name = "Centurion Receiver" + + # Dummy pairing object — lock_open stays False + from .receiver import Pairing + + self.pairing = Pairing() + + # Discover dongle features independently + self._discover_dongle_features() + + # Read serial from dongle's CENTURION_DEVICE_INFO if available + if self.serial is None: + try: + s = _hidpp20.get_serial_centurion(self) + if s and s.strip() and s.strip().isprintable(): + self.serial = s.strip() + except Exception: + pass + + def enable_connection_notifications(self, enable=True): + return False + + def remaining_pairings(self, cache=True): + return None + + def device_codename(self, n): + return self._usb_name + + def request(self, request_id, *params, no_reply=False): + """Send an HID++ request directly to the dongle (not through bridge).""" + if self.handle: + return self.low_level.request( + self.handle, 0xFF, request_id, *params, no_reply=no_reply, long_message=True, protocol=2.0 + ) + + def feature_request(self, feature, function=0x00, *params, no_reply=False): + """Send a feature request to the dongle using discovered feature indices.""" + if self._dongle_features is None: + self._discover_dongle_features() + feature_int = int(feature) + for _feat, feat_id, index in (self._dongle_features or []): + if feat_id == feature_int: + request_id = (index << 8) | (function & 0xFF) + return self.request(request_id, *params, no_reply=no_reply) + raise exceptions.FeatureNotSupported(feature) + + def _discover_dongle_features(self): + """Independently discover features on the dongle hardware.""" + self._dongle_features = [] + try: + # Query ROOT for FEATURE_SET index + response = self.request(0x0000, 0x00, 0x01) + if response is None or response[0] == 0: + return + fs_index = response[0] + # Get feature count + count_resp = self.request(fs_index << 8) + if count_resp is None: + return + feature_count = count_resp[0] + # Enumerate features via CenturionFeatureSet (func 1 = 0x10, per-index query) + for idx in range(feature_count): + resp = self.request((fs_index << 8) | 0x10, idx) + if resp is None or len(resp) < 3: + continue + feat_id = struct.unpack("!H", resp[1:3])[0] + try: + feature = SupportedFeature(feat_id) + except ValueError: + feature = f"unknown:{feat_id:04X}" + self._dongle_features.append((feature, feat_id, idx)) + if logger.isEnabledFor(logging.DEBUG): + logger.debug("Centurion dongle features: %s", self._dongle_features) + except Exception: + logger.debug("Centurion dongle feature discovery failed", exc_info=True) + + @property + def dongle_features(self): + """Return list of (feature, feat_id, index) tuples for dongle features.""" + if self._dongle_features is None: + self._discover_dongle_features() + return self._dongle_features + + def count(self): + return len([d for d in self._devices.values() if d is not None]) + + @property + def firmware(self): + if self._firmware is None and self.handle: + self._firmware = _hidpp20.get_firmware_centurion(self) + return self._firmware or () + + def notify_devices(self): + """Create child Device for the headset and trigger its initialization.""" + # Signal receiver to UI first — tray/window need the receiver entry + # before a child device can be added under it. + self.changed() + + # Create child Device with receiver=self, number=1 + pairing_info = { + "wpid": self.product_id, + "kind": None, + "serial": None, + "polling": None, + "power_switch": None, + } + dev = Device( + self.low_level, + self, + 1, + None, + pairing_info=pairing_info, + setting_callback=self.setting_callback, + ) + # Set centurion attributes on the child + dev.centurion = True + dev.product_id = self.product_id + dev.hidpp_long = True + dev._centurion_usb_name = self._usb_name + + self._devices[1] = dev + configuration.attach_to(dev) + dev.status_callback = self.status_callback + + # Ping to determine online status + if dev.ping(): + dev.changed(active=True) + if self.status_callback is not None: + self.status_callback(dev) + + def changed(self, alert=Alert.NOTIFICATION, reason=None): + if self.status_callback is not None: + self.status_callback(self, alert=alert, reason=reason) + + def status_string(self): + count = self.count() + if count == 0: + return "No devices." + return f"{count} device connected." + + def close(self): + handle, self.handle = self.handle, None + for _n, d in self._devices.items(): + if d: + d.close() + self._devices.clear() + for cleanup in self.cleanups: + cleanup(self) + return handle and self.low_level.close(handle) + + def __del__(self): + self.close() + + def __iter__(self): + for dev in self._devices.values(): + if dev is not None: + yield dev + + def __getitem__(self, key): + dev = self._devices.get(key) + if dev is not None: + return dev + raise IndexError(key) + + def __len__(self): + return len([d for d in self._devices.values() if d is not None]) + + def __contains__(self, dev): + if isinstance(dev, int): + return self._devices.get(dev) is not None + return self.__contains__(dev.number) + + def __bool__(self): + return self.handle is not None + + __nonzero__ = __bool__ + + def __eq__(self, other): + return other is not None and self.kind == other.kind and self.path == other.path + + def __ne__(self, other): + return other is None or self.kind != other.kind or self.path != other.path + + def __hash__(self): + return self.path.__hash__() + + def __str__(self): + return "<%s(%s,%s%s)>" % ( + self.name.replace(" ", "") if self.name else "CenturionReceiver", + self.path, + "" if isinstance(self.handle, int) else "T", + self.handle, + ) + + __repr__ = __str__ + + +def create_centurion_receiver(low_level: LowLevelInterface, device_info, setting_callback=None): + """Opens a Centurion dongle and wraps it as a receiver-like container. + + Creates a CenturionReceiver, discovers its features, then checks if + CentPPBridge (0x0003) is among them. If not, this is a direct-connected + device (wired headset) — close and return None so the caller can fall + back to create_device(). + + :returns: A CenturionReceiver, or None. + """ + try: + handle = low_level.open_path(device_info.path) + if handle: + base._centurion_handles.add(int(handle)) + cr = CenturionReceiver(low_level, handle, device_info, setting_callback) + # Check if any discovered feature is CentPPBridge (0x0003) + has_bridge = any(feat_id == 0x0003 for _, feat_id, _ in (cr.dongle_features or [])) + if not has_bridge: + logger.info("Centurion device %s has no bridge, treating as direct device", device_info.path) + base._centurion_handles.discard(int(handle)) + cr.handle = None # prevent __del__ from double-closing + low_level.close(handle) + return None + return cr + except OSError as e: + logger.exception("open %s", device_info) + if e.errno == errno.EACCES: + raise e + except Exception as e: + logger.exception("open %s", device_info) + raise e + + 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. @@ -613,9 +883,12 @@ class Device: long = self.hidpp_long is True or ( self.hidpp_long is None and (self.bluetooth or self._protocol is not None and self._protocol >= 2.0) ) + # Centurion child: CPL framing strips devnumber and responses always + # have devnumber=0xFF, so we must send 0xFF to match responses. + devnumber = 0xFF if (self.centurion and self.receiver and not self.handle) else self.number return self.low_level.request( self.handle or (self.receiver.handle if self.receiver else None), - self.number, + devnumber, request_id, *params, no_reply=no_reply, @@ -766,6 +1039,18 @@ class Device: def ping(self): """Checks if the device is online and present, returns True of False. Some devices are integral with their receiver but may not be present even if the receiver responds to ping.""" + if self.centurion and self.receiver and not self.handle: + # Centurion child: ping the dongle with 0xFF (CPL framing has no device number, + # and responses always have devnumber=0xFF) + handle = self.receiver.handle + try: + protocol = self.low_level.ping(handle, 0xFF, long_message=True) + except exceptions.NoReceiver: + protocol = None + self.online = protocol is not None and self.present + if protocol: + self._protocol = protocol + return self.online long = self.hidpp_long is True or ( self.hidpp_long is None and (self.bluetooth or self._protocol is not None and self._protocol >= 2.0) ) diff --git a/lib/logitech_receiver/hidpp20.py b/lib/logitech_receiver/hidpp20.py index 58ff7bc6..495357b9 100644 --- a/lib/logitech_receiver/hidpp20.py +++ b/lib/logitech_receiver/hidpp20.py @@ -2237,15 +2237,24 @@ def decipher_battery_unified(report) -> tuple[SupportedFeature, Battery]: def decipher_battery_centurion(report) -> tuple[SupportedFeature, Battery]: - """Decipher CENTURION_BATTERY_SOC (0x0104) response.""" + """Decipher CENTURION_BATTERY_SOC (0x0104) response. + + Response format (3 bytes): + Byte 0: Battery Percentage (0-100) + Byte 1: Battery Percentage (duplicate) + Byte 2: Charging Status (0=discharging, 1=charging, 2=charging via USB, 3=charge complete) + """ 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) + soc = report[0] 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: + charging_status = report[2] if len(report) >= 3 else 0 + if charging_status in (1, 2): status = BatteryStatus.RECHARGING + elif charging_status == 3: + status = BatteryStatus.FULL + else: + status = BatteryStatus.DISCHARGING return SupportedFeature.CENTURION_BATTERY_SOC, Battery(soc, None, status, None) diff --git a/lib/logitech_receiver/hidpp20_constants.py b/lib/logitech_receiver/hidpp20_constants.py index b54a9fef..edd7567f 100644 --- a/lib/logitech_receiver/hidpp20_constants.py +++ b/lib/logitech_receiver/hidpp20_constants.py @@ -185,7 +185,7 @@ class SupportedFeature(IntEnum): CENTURION_ROOT = 0x0102 CENTURION_MEMFAULT = 0x0103 CENTURION_BATTERY_SOC = 0x0104 - CENTURION_UNKNOWN_0108 = 0x0108 + CENTURION_AUTO_SLEEP = 0x0108 CENTURION_GENERIC_DFU = 0x010A # Headsets (Centurion-era) HEADSET_VOLUME = 0x0200 diff --git a/lib/logitech_receiver/notifications.py b/lib/logitech_receiver/notifications.py index 1cc46248..693e3afd 100644 --- a/lib/logitech_receiver/notifications.py +++ b/lib/logitech_receiver/notifications.py @@ -287,6 +287,9 @@ def _process_feature_notification(device: Device, notification: HIDPPNotificatio else: logger.warning("%s: unknown ADC MEASUREMENT %s", device, notification) + elif feature == SupportedFeature.CENTURION_BATTERY_SOC: + device.set_battery_info(hidpp20.decipher_battery_centurion(notification.data)[1]) + elif feature == SupportedFeature.SOLAR_DASHBOARD: if notification.data[5:9] == b"GOOD": charge, lux, adc = struct.unpack("!BHH", notification.data[:5]) diff --git a/lib/logitech_receiver/settings_templates.py b/lib/logitech_receiver/settings_templates.py index c103c212..4883df77 100644 --- a/lib/logitech_receiver/settings_templates.py +++ b/lib/logitech_receiver/settings_templates.py @@ -1671,6 +1671,18 @@ class HeadsetMixBalance(settings.Setting): validator_options = {"byte_count": 1} +class HeadsetAutoSleep(settings.Setting): + name = "headset-auto-sleep" + label = _("Auto Sleep Timeout") + description = _("Idle time in minutes before the headset enters sleep mode (0 = disabled).") + feature = _F.CENTURION_AUTO_SLEEP + rw_options = {"read_fnid": 0x00, "write_fnid": 0x10} + validator_class = settings_validator.RangeValidator + min_value = 0 + max_value = 255 + validator_options = {"byte_count": 1} + + class BrightnessControl(settings.Setting): name = "brightness_control" label = _("Brightness Control") @@ -2159,6 +2171,7 @@ SETTINGS: list[settings.Setting] = [ HeadsetSidetone, HeadsetMicGain, HeadsetMixBalance, + HeadsetAutoSleep, ] diff --git a/lib/solaar/cli/__init__.py b/lib/solaar/cli/__init__.py index a6b6818c..99312a9b 100644 --- a/lib/solaar/cli/__init__.py +++ b/lib/solaar/cli/__init__.py @@ -128,7 +128,14 @@ def _receivers_and_devices(dev_path=None): continue try: if dev_info.isDevice: - d = device.create_device(base, dev_info) + if getattr(dev_info, "centurion", False): + d = device.create_centurion_receiver(base, dev_info) + if d is not None: + d.notify_devices() + else: + d = device.create_device(base, dev_info) + else: + d = device.create_device(base, dev_info) else: d = receiver.create_receiver(base, dev_info) diff --git a/lib/solaar/cli/show.py b/lib/solaar/cli/show.py index 4f1d128f..5ea8d230 100644 --- a/lib/solaar/cli/show.py +++ b/lib/solaar/cli/show.py @@ -25,6 +25,7 @@ from logitech_receiver import settings_templates from logitech_receiver.common import LOGITECH_VENDOR_ID from logitech_receiver.common import NamedInt from logitech_receiver.common import strhex +from logitech_receiver.device import CenturionReceiver from logitech_receiver.hidpp20_constants import SupportedFeature from solaar import NAME @@ -35,36 +36,80 @@ _hidpp20 = hidpp20.Hidpp20() def _print_receiver(receiver): + is_centurion = isinstance(receiver, CenturionReceiver) paired_count = receiver.count() print(receiver.name) print(" Device path :", receiver.path) print(f" USB id : {LOGITECH_VENDOR_ID:04x}:{receiver.product_id}") - print(" Serial :", receiver.serial) - pending = hidpp10.get_configuration_pending_flags(receiver) - if pending: - print(f" C Pending : {pending:02x}") + if is_centurion: + print(" Protocol : Centurion") + if receiver.serial: + print(" Serial :", receiver.serial) + if not is_centurion: + pending = hidpp10.get_configuration_pending_flags(receiver) + if pending: + print(f" C Pending : {pending:02x}") if receiver.firmware: for f in receiver.firmware: print(" %-11s: %s" % (f.kind, f.version)) - print(" Has", paired_count, f"paired device(s) out of a maximum of {int(receiver.max_devices)}.") - if receiver.remaining_pairings() and receiver.remaining_pairings() >= 0: - print(f" Has {int(receiver.remaining_pairings())} successful pairing(s) remaining.") + if is_centurion: + print(" Has", paired_count, f"device(s) out of a maximum of {int(receiver.max_devices)}.") + else: + print(" Has", paired_count, f"paired device(s) out of a maximum of {int(receiver.max_devices)}.") + if receiver.remaining_pairings() and receiver.remaining_pairings() >= 0: + print(f" Has {int(receiver.remaining_pairings())} successful pairing(s) remaining.") - notification_flags = _hidpp10.get_notification_flags(receiver) - if notification_flags is not None: - if notification_flags: - notification_names = hidpp10_constants.NotificationFlag.flag_names(notification_flags) - print(f" Notifications: {', '.join(notification_names)} (0x{notification_flags.value:06X})") + if is_centurion: + _print_centurion_dongle_features(receiver) + else: + notification_flags = _hidpp10.get_notification_flags(receiver) + if notification_flags is not None: + if notification_flags: + notification_names = hidpp10_constants.NotificationFlag.flag_names(notification_flags) + print(f" Notifications: {', '.join(notification_names)} (0x{notification_flags.value:06X})") + else: + print(" Notifications: (none)") + + activity = receiver.read_register(hidpp10_constants.Registers.DEVICES_ACTIVITY) + if activity: + activity = [(d, ord(activity[d - 1 : d])) for d in range(1, receiver.max_devices)] + activity_text = ", ".join(f"{int(d)}={int(a)}" for d, a in activity if a > 0) + print(" Device activity counters:", activity_text or "(empty)") + + +def _print_centurion_dongle_features(receiver): + """Print dongle-level features, probed independently on the dongle hardware.""" + features = receiver.dongle_features + if not features: + return + print(f" Supports {len(features)} dongle features:") + for feature, feat_id, index in features: + display_name = "CENTPP BRIDGE" if feat_id == 0x0003 else feature + feat_bytes = feat_id.to_bytes(2, byteorder="big") + try: + flags_resp = receiver.request(0x0000, feat_bytes[0], feat_bytes[1]) + except Exception: + flags_resp = None + if flags_resp is not None and len(flags_resp) >= 2: + flags = flags_resp[1] + flag_names = common.flag_names(hidpp20_constants.FeatureFlag, flags) + print(" %2d: %-22s {%04X} %s " % (index, display_name, feat_id, ", ".join(flag_names))) else: - print(" Notifications: (none)") - - activity = receiver.read_register(hidpp10_constants.Registers.DEVICES_ACTIVITY) - if activity: - activity = [(d, ord(activity[d - 1 : d])) for d in range(1, receiver.max_devices)] - activity_text = ", ".join(f"{int(d)}={int(a)}" for d, a in activity if a > 0) - print(" Device activity counters:", activity_text or "(empty)") + print(" %2d: %-22s {%04X}" % (index, display_name, feat_id)) + if feature == SupportedFeature.CENTURION_DEVICE_INFO: + fw_list = _hidpp20.get_firmware_centurion(receiver) + serial = _hidpp20.get_serial_centurion(receiver) + hw_info = _hidpp20.get_hardware_info_centurion(receiver) + 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}") def _battery_text(level) -> str: @@ -91,6 +136,8 @@ def _battery_line(dev): def _print_device(dev, num=None): assert dev is not None + is_centurion = getattr(dev, "centurion", False) + is_centurion_child = is_centurion and isinstance(getattr(dev, "receiver", None), CenturionReceiver) # try to ping the device to see if it actually exists and to wake it up try: dev.ping() @@ -102,19 +149,21 @@ def _print_device(dev, num=None): print(f" {int(num or dev.number)}: {dev.name}") else: print(f"{dev.name}") - print(" Device path :", dev.path) - if dev.wpid: + # Centurion child has no separate hidraw path — show receiver's path + device_path = dev.path or (dev.receiver.path if is_centurion_child else None) + print(" Device path :", device_path) + if dev.wpid and not is_centurion_child: print(f" WPID : {dev.wpid}") if dev.product_id: print(f" USB id : {LOGITECH_VENDOR_ID:04x}:{dev.product_id}") print(" Codename :", dev.codename) print(" Kind :", dev.kind) if dev.protocol: - proto_name = "Centurion" if getattr(dev, "centurion", False) else "HID++" + proto_name = "Centurion" if is_centurion else "HID++" print(f" Protocol : {proto_name} {dev.protocol:1.1f}") else: print(" Protocol : unknown (device is offline)") - if dev.polling_rate: + if not is_centurion and dev.polling_rate: print(" Report Rate :", dev.polling_rate) print(" Serial number:", dev.serial) if dev.modelId: @@ -128,7 +177,8 @@ def _print_device(dev, num=None): if dev.power_switch_location: print(f" The power switch is located on the {dev.power_switch_location}.") - if dev.online: + # Skip HID++ 1.0 register reads for centurion devices — they don't support these + if dev.online and not is_centurion: notification_flags = _hidpp10.get_notification_flags(dev) if notification_flags is not None: if notification_flags: @@ -148,7 +198,11 @@ def _print_device(dev, num=None): 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: + # For centurion child devices, dongle features are shown on the receiver — + # only show sub-device (headset) features here. + if is_centurion_child and sub_count > 0: + print(f" Supports {sub_count} HID++ 2.0 features:") + elif 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:") @@ -159,8 +213,12 @@ def _print_device(dev, num=None): 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):") + if not is_centurion_child: + print(" Headset (via CentPPBridge):") feature_num += 1 + # For centurion child, skip dongle features (already shown on the receiver) + if is_centurion_child and not in_sub_device: + continue if isinstance(feature, str): feature_bytes = bytes.fromhex(feature[-4:]) else: @@ -168,20 +226,26 @@ def _print_device(dev, num=None): 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: - 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)) - ) + if is_centurion_child and in_sub_device: + # Use cached version — skip slow bridge ROOT queries + version = dev.features.get_feature_version(feature_int) or 0 + print(" %2d: %-22s {%04X} V%s" % (index, display_name, feature_int, version)) else: - print(" %2d: %-22s {%04X}" % (index, display_name, feature_int)) + try: + flags = dev.request(0x0000, feature_bytes) + except Exception: + 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: @@ -361,7 +425,7 @@ def run(devices, args, find_receiver, find_device): if device_name == "all": for d in devices: - if isinstance(d, receiver.Receiver): + if isinstance(d, (receiver.Receiver, CenturionReceiver)): _print_receiver(d) count = d.count() if count: @@ -373,8 +437,8 @@ def run(devices, args, find_receiver, find_device): break print("") else: - print("") _print_device(d) + print("") return dev = find_receiver(devices, device_name) diff --git a/lib/solaar/listener.py b/lib/solaar/listener.py index 74d2cd89..9cf747b1 100644 --- a/lib/solaar/listener.py +++ b/lib/solaar/listener.py @@ -149,6 +149,12 @@ class SolaarListener(listener.EventsListener): def _notifications_handler(self, n): assert self.receiver if n.devnumber == 0xFF: + # For CenturionReceiver, intercept bridge notifications and dispatch to child device + from logitech_receiver.device import CenturionReceiver + + if isinstance(self.receiver, CenturionReceiver): + self._handle_centurion_notification(n) + return # a receiver notification notifications.process(self.receiver, n) return @@ -228,6 +234,55 @@ class SolaarListener(listener.EventsListener): elif dev.online is None: dev.ping() + def _handle_centurion_notification(self, n): + """Handle notifications from a CenturionReceiver dongle. + + Bridge MessageEvents (sub_id=bridge_index, sw_id=0) carry wrapped sub-device + notifications. Unwrap them and dispatch to the child device so that battery + updates and other feature events are processed. + + Bridge data layout after make_notification: + n.data[0:2] = dev_id<<4|len_hi, len_lo + n.data[2] = sub_cpl (0x00 for both responses and notifications) + n.data[3] = sub_feat_idx + n.data[4] = sub_func_sw (sw_id=0 for unsolicited notifications) + n.data[5:] = payload + """ + child = self.receiver._devices.get(1) + if not child or not child.online: + if logger.isEnabledFor(logging.DEBUG): + logger.debug("CenturionReceiver: notification ignored (no online child): %s", n) + return + + bridge_idx = getattr(child, "_centurion_bridge_index", None) + if bridge_idx is not None and n.sub_id == bridge_idx: + # Bridge MessageEvent — unwrap sub-device notification + # n.data layout: [dev_id<<4|len_hi, len_lo, sub_cpl, sub_feat_idx, sub_func_sw, payload...] + if len(n.data) < 5: + return + sub_feat_idx = n.data[3] + sub_func_sw = n.data[4] + payload = n.data[5:] + if logger.isEnabledFor(logging.DEBUG): + logger.debug( + "CenturionReceiver: bridge notification sub_feat=%d func=0x%02X payload=%s -> child %s", + sub_feat_idx, + sub_func_sw, + payload[:8].hex() if payload else "", + child, + ) + # Create synthetic notification and dispatch directly to feature processing. + # Sub-device features use 0x100 offset in FeaturesArray.inverse. + synthetic = base.HIDPPNotification(n.report_id, child.number, sub_feat_idx + 0x100, sub_func_sw, payload) + child.online = True + if child.features: + notifications._process_feature_notification(child, synthetic) + return + + if logger.isEnabledFor(logging.DEBUG): + logger.debug("CenturionReceiver: unhandled notification sub_id=%d addr=0x%02X data=%s", + n.sub_id, n.address, n.data[:12].hex() if n.data else "") + def __str__(self): return f"" @@ -260,6 +315,13 @@ def _start(device_info: DeviceInfo): if not device_info.isDevice: receiver_ = logitech_receiver.receiver.create_receiver(base, device_info, _setting_callback) + elif getattr(device_info, "centurion", False): + receiver_ = logitech_receiver.device.create_centurion_receiver(base, device_info, _setting_callback) + if receiver_ is None: + # No bridge found — treat as a direct-connected centurion device (e.g., wired headset) + receiver_ = logitech_receiver.device.create_device(base, device_info, _setting_callback) + if receiver_: + configuration.attach_to(receiver_) else: receiver_ = logitech_receiver.device.create_device(base, device_info, _setting_callback) if receiver_: diff --git a/tests/logitech_receiver/test_hidpp20_complex.py b/tests/logitech_receiver/test_hidpp20_complex.py index a93a94f8..f049c145 100644 --- a/tests/logitech_receiver/test_hidpp20_complex.py +++ b/tests/logitech_receiver/test_hidpp20_complex.py @@ -22,6 +22,7 @@ from logitech_receiver import exceptions from logitech_receiver import hidpp20 from logitech_receiver import hidpp20_constants from logitech_receiver import special_keys +from logitech_receiver.device import CenturionReceiver from logitech_receiver.hidpp20 import KeyFlag from logitech_receiver.hidpp20 import MappingFlag from logitech_receiver.hidpp20_constants import GestureId @@ -1128,3 +1129,154 @@ def test_centurion_kind_inference(): from logitech_receiver import hidpp10_constants assert kind == hidpp10_constants.DEVICE_KIND.headset + + +# --- CenturionReceiver tests --- + + +class FakeCenturionDeviceInfo: + """Minimal device_info for CenturionReceiver tests.""" + + def __init__(self, path="/dev/hidraw99", product_id="0AF0", product=None, centurion=True): + self.path = path + self.product_id = product_id + self.product = product + self.centurion = centurion + self.isDevice = True + + +class FakeLowLevel: + """Minimal low_level for CenturionReceiver tests.""" + + def __init__(self, ping_protocol=2.6): + self.ping_protocol = ping_protocol + self.opened_paths = [] + self.closed_handles = [] + + def open_path(self, path): + self.opened_paths.append(path) + return 0x99 + + def ping(self, handle, number, long_message=False): + return self.ping_protocol + + def request(self, handle, devnumber, request_id, *params, **kwargs): + return None + + def close(self, handle, *args, **kwargs): + self.closed_handles.append(handle) + return True + + def find_paired_node(self, receiver_path, index, timeout): + return None + + +def test_centurion_receiver_attributes(): + """CenturionReceiver has correct receiver-like attributes.""" + info = FakeCenturionDeviceInfo(product="PRO X 2 LIGHTSPEED") + recv = CenturionReceiver(FakeLowLevel(), 0x99, info) + + assert recv.kind is None + assert recv.isDevice is False + assert recv.number == 0xFF + assert recv.max_devices == 1 + assert recv.may_unpair is False + assert recv.re_pairs is False + assert recv.handle == 0x99 + assert recv.path == "/dev/hidraw99" + assert recv.product_id == "0AF0" + assert recv.name == "PRO X 2 LIGHTSPEED" + assert recv.serial is None + assert recv.pairing is not None + assert recv.pairing.lock_open is False + assert bool(recv) is True + + +def test_centurion_receiver_container_empty(): + """Empty CenturionReceiver has correct container behavior.""" + info = FakeCenturionDeviceInfo() + recv = CenturionReceiver(FakeLowLevel(), 0x99, info) + + assert len(recv) == 0 + assert recv.count() == 0 + assert 1 not in recv + assert list(recv) == [] + assert recv.status_string() == "No devices." + + +def test_centurion_receiver_container_with_device(): + """CenturionReceiver with a child device has correct container behavior.""" + info = FakeCenturionDeviceInfo() + recv = CenturionReceiver(FakeLowLevel(), 0x99, info) + + # Simulate adding a child device (a simple mock) + class FakeChild: + number = 1 + + def close(self): + pass + + recv._devices[1] = FakeChild() + + assert len(recv) == 1 + assert recv.count() == 1 + assert 1 in recv + assert 2 not in recv + assert recv[1] is not None + assert recv.status_string() == "1 device connected." + with pytest.raises(IndexError): + recv[2] + + +def test_centurion_receiver_enable_connection_notifications(): + """CenturionReceiver.enable_connection_notifications() returns False.""" + info = FakeCenturionDeviceInfo() + recv = CenturionReceiver(FakeLowLevel(), 0x99, info) + + assert recv.enable_connection_notifications() is False + assert recv.remaining_pairings() is None + + +def test_centurion_receiver_device_codename(): + """CenturionReceiver.device_codename() returns USB product name.""" + info = FakeCenturionDeviceInfo(product="PRO X 2 LIGHTSPEED") + recv = CenturionReceiver(FakeLowLevel(), 0x99, info) + + assert recv.device_codename(1) == "PRO X 2 LIGHTSPEED" + + +def test_centurion_receiver_close(): + """CenturionReceiver.close() closes handle and clears devices.""" + low_level = FakeLowLevel() + info = FakeCenturionDeviceInfo() + recv = CenturionReceiver(low_level, 0x99, info) + + class FakeChild: + closed = False + + def close(self): + self.closed = True + + child = FakeChild() + recv._devices[1] = child + + recv.close() + + assert recv.handle is None + assert len(recv._devices) == 0 + assert child.closed is True + assert 0x99 in low_level.closed_handles + assert bool(recv) is False + + +def test_centurion_receiver_changed_callback(): + """CenturionReceiver.changed() invokes status_callback.""" + info = FakeCenturionDeviceInfo() + recv = CenturionReceiver(FakeLowLevel(), 0x99, info) + calls = [] + recv.status_callback = lambda *args, **kwargs: calls.append((args, kwargs)) + + recv.changed() + + assert len(calls) == 1 + assert calls[0][0][0] is recv