diff --git a/lib/logitech_receiver/base.py b/lib/logitech_receiver/base.py index 0317abce..e256ba38 100644 --- a/lib/logitech_receiver/base.py +++ b/lib/logitech_receiver/base.py @@ -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 (0x00–0xFF), 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 diff --git a/lib/logitech_receiver/hidpp20.py b/lib/logitech_receiver/hidpp20.py index f5813836..c384ca07 100644 --- a/lib/logitech_receiver/hidpp20.py +++ b/lib/logitech_receiver/hidpp20.py @@ -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: diff --git a/tests/logitech_receiver/test_base.py b/tests/logitech_receiver/test_base.py index f1edd6d4..a4ae4d1d 100644 --- a/tests/logitech_receiver/test_base.py +++ b/tests/logitech_receiver/test_base.py @@ -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."""