Tighten Centurion probe and silence FeaturesArray query spam

Three cleanup fixes after field testing confirmed the brute-force probe
finds device_addr=0x23 on the G522 (no special addresses at 0x00/0xFF):

1. Short-circuit the probe on first hit instead of sweeping all 256. The
   diagnostic full-sweep served its purpose — only 0x23 responds.
2. Drop per-candidate read timeout from 20ms to 5ms. USB round-trip is
   <1ms, so 5ms is 5x margin. Worst case: 1.3s. Typical: 180ms.
3. Short-circuit FeaturesArray.__getitem__ for Centurion devices. All
   parent + sub-device features are enumerated upfront by
   _check_centurion(). If a feature isn't in the dict after _check(),
   it genuinely doesn't exist — skip the raw ROOT.GetFeature query that
   the dongle rejects with LOGITECH_ERROR. Eliminates cycling
   {0002}..{000F} error log spam during settings init and speeds up
   check_feature_settings() by ~225 round trips per device.
This commit is contained in:
Ken Sanislo 2026-04-16 22:46:10 -07:00
parent 4dad052d36
commit ca500f80bb
3 changed files with 28 additions and 30 deletions

View File

@ -339,8 +339,8 @@ def _centurion_frame_header(state: CenturionHandleState, cpl_length: int, flags:
_CENTURION_REPORT_IDS = (CENTURION_REPORT_ID, CENTURION_ADDRESSED_REPORT_ID)
# Per-candidate read timeout (ms) for the device_addr probe.
# USB round-trip is <1ms; 20ms gives plenty of margin.
_CENTURION_PROBE_PER_ADDR_TIMEOUT_MS = 20
# USB round-trip is <1ms; 5ms gives 5x margin.
_CENTURION_PROBE_PER_ADDR_TIMEOUT_MS = 5
def probe_centurion_device_addr(handle, state: CenturionHandleState) -> bool:
@ -348,25 +348,23 @@ def probe_centurion_device_addr(handle, state: CenturionHandleState) -> bool:
Sends a ROOT.GetProtocolVersion request for each candidate device_addr
(0x000xFF), reading briefly after each write. The dongle silently ignores
wrong addresses and responds only to the correct one.
wrong addresses and responds only to the correct one. Stops on first hit.
Sweeps ALL 256 candidates and logs every address that responds, so we can
discover special addresses (broadcast, etc.) during initial field testing.
Uses the first responding address as the device_addr.
Worst case (no response): 256 × 5ms = ~1.3s.
Typical G522 (addr=0x23): 36 × 5ms = ~180ms.
No-op for 0x51 (no device_addr byte) or when an address is already known.
Returns True if at least one address responded.
Returns True if the address was learned.
"""
if state.report_id != CENTURION_ADDRESSED_REPORT_ID or state.device_addr is not None:
return False
ihandle = int(handle)
logger.info("(%s) probing centurion device_addr: full sweep 0x00-0xFF", handle)
logger.info("(%s) probing centurion device_addr: scanning 0x00-0xFF", handle)
# ROOT.GetProtocolVersion: feat_idx=0x00, func=0x10, 3 zero param bytes
payload = bytes([0x00, 0x10, 0x00, 0x00, 0x00])
cpl_length = len(payload) + 1 # +1 for flags byte
write_errors = 0
responding_addrs = []
for addr in range(256):
frame = struct.pack("!BBBB", CENTURION_ADDRESSED_REPORT_ID, addr, cpl_length, 0x00) + payload
@ -377,30 +375,24 @@ def probe_centurion_device_addr(handle, state: CenturionHandleState) -> bool:
write_errors += 1
if write_errors > 3:
logger.warning("(%s) centurion device_addr probe: too many write failures, aborting", handle)
break
return False
continue
try:
data = hidapi.read(ihandle, CENTURION_FRAME_SIZE, _CENTURION_PROBE_PER_ADDR_TIMEOUT_MS)
except Exception as reason:
logger.warning("(%s) centurion device_addr probe read failed at addr 0x%02X: %s", handle, addr, reason)
break
return False
if data and len(data) >= 2 and ord(data[:1]) == state.report_id:
resp_addr = ord(data[1:2])
responding_addrs.append((addr, resp_addr, common.strhex(data[:8])))
if state.device_addr is None:
state.device_addr = resp_addr
state.device_addr = ord(data[1:2])
logger.info(
"(%s) probed centurion device addr 0x%02X (after %d candidates)",
handle,
state.device_addr,
addr + 1,
)
return True
logger.info(
"(%s) centurion device_addr probe complete: %d responding, results=%s",
handle,
len(responding_addrs),
responding_addrs,
)
if state.device_addr is not None:
logger.info("(%s) using centurion device addr 0x%02X", handle, state.device_addr)
return True
logger.warning("(%s) centurion device_addr probe: no response from any candidate", handle)
logger.warning("(%s) centurion device_addr probe: no response from any of 256 candidates", handle)
return False

View File

@ -333,6 +333,12 @@ class FeaturesArray(dict):
index = super().get(feature)
if index is not None:
return index
# Centurion devices enumerate all features upfront in _check_centurion().
# If the feature isn't in the dict after _check(), it genuinely doesn't
# exist — skip the raw ROOT.GetFeature query that the dongle rejects
# with LOGITECH_ERROR and that creates cycling log spam during settings init.
if getattr(self.device, "centurion", False):
return None
try:
response = self.device.request(0x0000, struct.pack("!H", feature))
except exceptions.FeatureCallError:

View File

@ -326,8 +326,8 @@ class TestProbeCenturionDeviceAddr:
def teardown_method(self):
base._centurion_handles.pop(self.HANDLE, None)
def test_learns_addr_and_sweeps_all(self):
"""Full sweep finds addr=0x23 and continues to 0xFF, logging all responders."""
def test_learns_addr_on_first_hit(self):
"""Probe finds addr=0x23 on candidate #36 (0-indexed 0x23=35) and stops."""
state = CenturionHandleState(report_id=CENTURION_ADDRESSED_REPORT_ID)
reply = bytes([0x50, 0x23, 0x03, 0x00]) + b"\x00" * 60
@ -344,8 +344,8 @@ class TestProbeCenturionDeviceAddr:
result = base.probe_centurion_device_addr(self.HANDLE, state)
assert result is True
assert state.device_addr == 0x23
# Full sweep — should have written all 256 candidates
assert mock_write.call_count == 256
# Short-circuit: stopped at candidate 0x23 (36 writes), not all 256
assert mock_write.call_count == 0x24
def test_skips_non_matching_read_until_match(self):
"""Non-0x50 frames in the read are ignored; next candidate's read succeeds."""