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

Probe device_addr one at a time instead of blasting all 256 

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.
This commit is contained in:
Ken Sanislo 2026-04-16 10:20:49 -07:00
parent b25d0e009e
commit 81097ffa44
2 changed files with 33 additions and 32 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

36
lib/logitech_receiver/base.py
View File

@ -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

29
tests/logitech_receiver/test_base.py
View File

@ -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