From 81097ffa4485b7db931f226755f58179193fafae Mon Sep 17 00:00:00 2001 From: Ken Sanislo Date: Thu, 16 Apr 2026 10:20:49 -0700 Subject: [PATCH] Probe device_addr one at a time instead of blasting all 256 MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Write one candidate, read briefly (20ms), move to next. Stops on first hit. Avoids overwhelming the dongle's RX buffer with 256 back-to-back frames where the response could get lost. Worst case (addr=0xFF): 256 × 20ms ≈ 5s. Typical G522 (addr=0x23): 36 × 20ms ≈ 0.7s. --- lib/logitech_receiver/base.py | 36 +++++++++++++--------------- tests/logitech_receiver/test_base.py | 29 ++++++++++++---------- 2 files changed, 33 insertions(+), 32 deletions(-) diff --git a/lib/logitech_receiver/base.py b/lib/logitech_receiver/base.py index f4447586..48a8ab35 100644 --- a/lib/logitech_receiver/base.py +++ b/lib/logitech_receiver/base.py @@ -338,18 +338,19 @@ def _centurion_frame_header(state: CenturionHandleState, cpl_length: int, flags: _CENTURION_REPORT_IDS = (CENTURION_REPORT_ID, CENTURION_ADDRESSED_REPORT_ID) -# Read timeout (ms) for the brute-force device_addr probe below. -_CENTURION_PROBE_READ_TIMEOUT_MS = 500 -_CENTURION_PROBE_READ_ITERATIONS = 3 +# 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 def probe_centurion_device_addr(handle, state: CenturionHandleState) -> bool: """Brute-force probe the device address byte for a 0x50-variant Centurion handle. - Sends a ROOT.GetProtocolVersion request for every possible device_addr - (0x00–0xFF). The dongle silently ignores wrong addresses and responds - only to the correct one. The response carries the real address at byte[1], - which we extract and store on the handle state. + 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. Stops on first hit. + + Worst case (addr=0xFF): 256 × 20ms = ~5s. Typical G522 (addr=0x23): ~0.7s. No-op for 0x51 (no device_addr byte) or when an address is already known. Returns True if the address was learned. @@ -357,40 +358,35 @@ def probe_centurion_device_addr(handle, state: CenturionHandleState) -> bool: 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: brute-force 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_failed = 0 + write_errors = 0 - # Send a ROOT query for every possible device_addr (256 frames). - # The dongle ignores frames with the wrong address. Only the matching - # one produces a response that we can read back. for addr in range(256): frame = struct.pack("!BBBB", CENTURION_ADDRESSED_REPORT_ID, addr, cpl_length, 0x00) + payload frame = frame + b"\x00" * (CENTURION_FRAME_SIZE - len(frame)) try: hidapi.write(ihandle, frame) except Exception: - write_failed += 1 - if write_failed > 3: + write_errors += 1 + if write_errors > 3: logger.warning("(%s) centurion device_addr probe: too many write failures, aborting", handle) return False - - # Read back the response — dongle only replied to the correct address. - for _attempt in range(_CENTURION_PROBE_READ_ITERATIONS): + continue try: - data = hidapi.read(ihandle, CENTURION_FRAME_SIZE, _CENTURION_PROBE_READ_TIMEOUT_MS) + 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: %s", handle, reason) return False if data and len(data) >= 2 and ord(data[:1]) == state.report_id: state.device_addr = ord(data[1:2]) - logger.info("(%s) probed centurion device addr 0x%02X", handle, state.device_addr) + logger.info("(%s) probed centurion device addr 0x%02X (after %d candidates)", handle, state.device_addr, addr + 1) return True - logger.warning("(%s) centurion device_addr brute-force probe got no response", handle) + logger.warning("(%s) centurion device_addr probe: no response from any of 256 candidates", handle) return False diff --git a/tests/logitech_receiver/test_base.py b/tests/logitech_receiver/test_base.py index 4559fe3d..c43d90bc 100644 --- a/tests/logitech_receiver/test_base.py +++ b/tests/logitech_receiver/test_base.py @@ -326,28 +326,33 @@ class TestProbeCenturionDeviceAddr: def teardown_method(self): base._centurion_handles.pop(self.HANDLE, None) - def test_learns_addr_from_response(self): + 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 + + def read_side_effect(_handle, _size, _timeout): + # Return a response only after the write with addr=0x23 + if mock_write.call_count == 0x24: # 0x23 is the 36th write (1-indexed) + return reply + return None + with ( mock.patch.object(base.hidapi, "write") as mock_write, - mock.patch.object(base.hidapi, "read", return_value=reply), + mock.patch.object(base.hidapi, "read", side_effect=read_side_effect), ): result = base.probe_centurion_device_addr(self.HANDLE, state) assert result is True assert state.device_addr == 0x23 - # Should have written 256 probe frames (one per candidate addr) - assert mock_write.call_count == 256 - # Each frame should be 64 bytes with report_id 0x50 - for call in mock_write.call_args_list: - _, wdata = call[0] - assert len(wdata) == base.CENTURION_FRAME_SIZE - assert wdata[0] == CENTURION_ADDRESSED_REPORT_ID + # Should have stopped at candidate 0x23 (36 writes), not all 256 + assert mock_write.call_count == 0x24 - def test_skips_non_matching_frames_until_match(self): + def test_skips_non_matching_read_until_match(self): + """Non-0x50 frames in the read are ignored; next candidate's read succeeds.""" state = CenturionHandleState(report_id=CENTURION_ADDRESSED_REPORT_ID) noise = b"\x11\xff" + b"\x00" * 62 match = bytes([0x50, 0x42, 0x03, 0x00]) + b"\x00" * 60 + # Reads cycle: noise, noise, match — so addr is found on 3rd candidate with ( mock.patch.object(base.hidapi, "write"), mock.patch.object(base.hidapi, "read", side_effect=[noise, noise, match]), @@ -356,7 +361,7 @@ class TestProbeCenturionDeviceAddr: assert result is True assert state.device_addr == 0x42 - def test_returns_false_on_timeout(self): + def test_returns_false_when_no_response(self): state = CenturionHandleState(report_id=CENTURION_ADDRESSED_REPORT_ID) with ( mock.patch.object(base.hidapi, "write"), @@ -406,7 +411,7 @@ class TestProbeCenturionDeviceAddr: state = CenturionHandleState(report_id=CENTURION_ADDRESSED_REPORT_ID) with ( mock.patch.object(base.hidapi, "write") as mock_write, - mock.patch.object(base.hidapi, "read", return_value=None), + mock.patch.object(base.hidapi, "read", return_value=None), # no response → scans all 256 ): base.probe_centurion_device_addr(self.HANDLE, state) assert mock_write.call_count == 256