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
This commit is contained in:
Ken Sanislo
parent
8bdb49feec
commit
d98dd3ed3f
|
|
@ -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)
|
||||
)
|
||||
|
|
|
|||
|
|
@ -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)
|
||||
|
||||
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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])
|
||||
|
|
|
|||
|
|
@ -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,
|
||||
]
|
||||
|
||||
|
||||
|
|
|
|||
|
|
@ -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)
|
||||
|
||||
|
|
|
|||
|
|
@ -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)
|
||||
|
|
|
|||
|
|
@ -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"<SolaarListener({self.receiver.path},{self.receiver.handle})>"
|
||||
|
||||
|
|
@ -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_:
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
Loading…
Reference in New Issue