RomanNum3ral
/
Solaar
mirror of https://github.com/pwr-Solaar/Solaar.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Fix G522 0x50 Centurion init failure and protocol version crash 

Three interrelated fixes for G522 LIGHTSPEED headset support:

1. Deferred init for silent 0x50 dongles: When the probe fails and
   feature discovery returns nothing (dongle silently drops all frames
   with device_addr=0x00), return a "pending" CenturionReceiver instead
   of falling through to the broken create_device direct-device path.
   The listener thread starts reading; when the first unsolicited frame
   arrives, _unwrap_centurion_frame learns device_addr, the notification
   handler detects the pending state, and re-runs feature discovery with
   the correct address — finding the bridge and creating the child device.

2. Centurion protocol version floor: The G522 dongle reports protocol
   1.1 (major=1, minor=1), which routes all protocol < 2.0 gates into
   HID++ 1.0 code paths (battery register reads, etc.) that crash with
   INVALID_SUB_ID_COMMAND. Centurion devices always use HID++ 2.0
   features, so the protocol property now returns 2.0 as a floor for
   any device with centurion=True.

3. CI segfault fix: Mock probe_centurion_device_addr in the two
   Centurion device tests so fake handles never reach the real hidapi
   C extension on macOS.
This commit is contained in:
Ken Sanislo 2026-04-15 23:00:45 -07:00
parent c27e4c1317
commit e91cbff361
4 changed files with 100 additions and 12 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

87
lib/logitech_receiver/centurion.py
View File

@ -265,6 +265,7 @@ class CenturionReceiver:
self._devices = {}
self._firmware = None
self._dongle_features = None # independently probed dongle features
self._pending = False # True when device_addr unknown; deferred init completes on first RX
self.cleanups = []
# Receiver identity
@ -364,11 +365,63 @@ class CenturionReceiver:
self._firmware = get_firmware_centurion(self)
return self._firmware or ()
def _complete_deferred_init(self):
"""Re-run feature discovery after device_addr has been learned.
Called once from the notification handler when the first 0x50 frame
arrives on a pending CenturionReceiver.
"""
if not self._pending:
return False
self._pending = False
ihandle = int(self.handle)
state = base._centurion_handles.get(ihandle)
learned_addr = state.device_addr if state else None
logger.info(
"CenturionReceiver %s: completing deferred init (device_addr=0x%02X)",
self.path,
learned_addr or 0,
)
self._dongle_features = None
self._discover_dongle_features()
logger.info(
"CenturionReceiver %s: deferred discovery found %d feature(s): %s",
self.path,
len(self._dongle_features or []),
[(f"{feat_id:#06x}", idx) for _, feat_id, idx in (self._dongle_features or [])],
)
if self.serial is None:
try:
s = get_serial_centurion(self)
if s and s.strip() and s.strip().isprintable():
self.serial = s.strip()
except Exception:
pass
has_bridge = any(feat_id == CenturionCoreFeature.CENT_PP_BRIDGE for _, feat_id, _ in (self._dongle_features or []))
if has_bridge:
self.notify_devices()
return True
logger.warning(
"CenturionReceiver %s: deferred init completed but no bridge found " "(features: %s)",
self.path,
[f"{feat_id:#06x}" for _, feat_id, _ in (self._dongle_features or [])],
)
return False
def notify_devices(self):
"""Create child Device for the headset and trigger its initialization."""
# Import Device locally to avoid circular import (centurion.py ↔ device.py)
from .device import Device
if self._pending:
# Don't create children yet — feature discovery hasn't succeeded.
# Signal receiver to UI so the tray entry exists.
self.changed(alert=Alert.NONE)
return
# Signal receiver to UI first — tray/window need the receiver entry
# before a child device can be added under it.
self.changed(alert=Alert.NONE)
@ -407,6 +460,13 @@ class CenturionReceiver:
# Ping to determine online status.
# Notify UI either way — offline devices show as greyed out (matching receiver behavior).
online = dev.ping()
logger.info(
"CenturionReceiver %s: child device created, bridge_idx=%s, online=%s, protocol=%s",
self.path,
getattr(dev, "_centurion_bridge_index", None),
online,
dev._protocol,
)
dev.changed(active=online)
if self.status_callback is not None:
self.status_callback(dev)
@ -498,13 +558,26 @@ def create_centurion_receiver(low_level, device_info, setting_callback=None):
cr = CenturionReceiver(low_level, handle, device_info, setting_callback)
# Check if any discovered feature is CentPPBridge (0x0003)
has_bridge = any(feat_id == CenturionCoreFeature.CENT_PP_BRIDGE for _, feat_id, _ in (cr.dongle_features or []))
if not has_bridge:
logger.info("Centurion device %s has no bridge, treating as direct device", device_info.path)
base._centurion_handles.pop(int(handle), None)
cr.handle = None # prevent __del__ from double-closing
low_level.close(handle)
return None
return cr
if has_bridge:
return cr
# No bridge found. Distinguish "silent 0x50 dongle" (device_addr
# unknown, headset not yet powered on) from "wired 0x50 device"
# (responded to probe, features found, but no bridge).
is_0x50 = state.report_id == base.CENTURION_ADDRESSED_REPORT_ID
if is_0x50 and state.device_addr is None and not cr.dongle_features:
logger.info(
"Centurion 0x50 device %s: probe and discovery failed, " "deferring init until first RX frame",
device_info.path,
)
cr._pending = True
return cr
logger.info("Centurion device %s has no bridge, treating as direct device", device_info.path)
base._centurion_handles.pop(int(handle), None)
cr.handle = None # prevent __del__ from double-closing
low_level.close(handle)
return None
except OSError as e:
logger.exception("open %s", device_info)
if e.errno == errno.EACCES:

8
lib/logitech_receiver/device.py
View File

@ -243,7 +243,13 @@ class Device:
self.ping()
except exceptions.NoSuchDevice:
logger.warning("device %s inaccessible - no protocol set", self)
return self._protocol or 0
result = self._protocol or 0
# Centurion devices always use HID++ 2.0 features regardless of the
# protocol version the dongle reports (e.g. G522 reports 1.1).
# Ensure all `protocol < 2.0` gates route through the 2.0 code path.
if self.centurion and result < 2.0:
return 2.0
return result
@property
def codename(self):

6
lib/solaar/listener.py
View File

@ -152,6 +152,12 @@ class SolaarListener(listener.EventsListener):
from logitech_receiver.device import CenturionReceiver
if isinstance(self.receiver, CenturionReceiver):
if self.receiver._pending:
ihandle = int(self.receiver.handle)
state = base._centurion_handles.get(ihandle)
if state and state.device_addr is not None:
self.receiver._complete_deferred_init()
self._status_changed(self.receiver)
self._handle_centurion_notification(n)
return
# a receiver notification

11
tests/logitech_receiver/test_device.py
View File

@ -17,6 +17,7 @@
from dataclasses import dataclass
from functools import partial
from typing import Optional
from unittest import mock
import pytest
@ -101,8 +102,9 @@ def test_create_centurion_device():
"""Test that a centurion device gets hidpp_long forced to True and centurion flag set."""
from logitech_receiver import base
low_level_mock = LowLevelInterfaceFake(fake_hidpp.r_empty)
test_device = device.create_device(low_level_mock, di_0AF7)
with mock.patch.object(base, "probe_centurion_device_addr", return_value=False):
low_level_mock = LowLevelInterfaceFake(fake_hidpp.r_empty)
test_device = device.create_device(low_level_mock, di_0AF7)
assert test_device is not None
assert test_device.centurion is True
@ -122,8 +124,9 @@ def test_create_centurion_0x50_device():
"""Test that a 0x50 centurion device gets the correct report ID registered."""
from logitech_receiver import base
low_level_mock = LowLevelInterfaceFake(fake_hidpp.r_empty)
test_device = device.create_device(low_level_mock, di_0B18)
with mock.patch.object(base, "probe_centurion_device_addr", return_value=False):
low_level_mock = LowLevelInterfaceFake(fake_hidpp.r_empty)
test_device = device.create_device(low_level_mock, di_0B18)
assert test_device is not None
assert test_device.centurion is True