diff --git a/lib/logitech_receiver/base.py b/lib/logitech_receiver/base.py index 837535c2..1580f99a 100644 --- a/lib/logitech_receiver/base.py +++ b/lib/logitech_receiver/base.py @@ -338,6 +338,46 @@ def _centurion_frame_header(state: CenturionHandleState, cpl_length: int, flags: _CENTURION_REPORT_IDS = (CENTURION_REPORT_ID, CENTURION_ADDRESSED_REPORT_ID) +# Per-iteration read timeout (ms) and total iterations for the 0x50 probe below. +_CENTURION_PROBE_READ_TIMEOUT_MS = 500 +_CENTURION_PROBE_READ_ITERATIONS = 3 + + +def probe_centurion_device_addr(handle, state: CenturionHandleState) -> bool: + """Probe the device address byte for a 0x50-variant Centurion handle. + + Sends a 64-byte all-zero frame with the detected report ID and reads back + the first response. The device answers with an error/unsolicited frame + whose byte[1] holds its device address. Without this probe, the very first + real TX ships with device_addr=0x00, which stricter firmware silently + drops — breaking dongle feature discovery before it starts. + + No-op for 0x51 (no device_addr byte) or when an address is already known. + 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) + probe = bytes([state.report_id]) + b"\x00" * (CENTURION_FRAME_SIZE - 1) + try: + hidapi.write(ihandle, probe) + except Exception as reason: + logger.warning("(%s) centurion device_addr probe write failed: %s", handle, reason) + return False + for _ in range(_CENTURION_PROBE_READ_ITERATIONS): + try: + data = hidapi.read(ihandle, CENTURION_FRAME_SIZE, _CENTURION_PROBE_READ_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]) + if logger.isEnabledFor(logging.DEBUG): + logger.debug("(%s) probed centurion device addr 0x%02X", handle, state.device_addr) + return True + logger.warning("(%s) centurion device_addr probe timed out, subsequent TX will use 0x00", handle) + return False + def _unwrap_centurion_frame(data: bytes, ihandle: int, handle) -> bytes: """Unwrap a Centurion CPL frame (0x50 or 0x51) into a standard HID++ long message. diff --git a/lib/logitech_receiver/centurion.py b/lib/logitech_receiver/centurion.py index 63933c34..356310a9 100644 --- a/lib/logitech_receiver/centurion.py +++ b/lib/logitech_receiver/centurion.py @@ -492,7 +492,9 @@ def create_centurion_receiver(low_level, device_info, setting_callback=None): handle = low_level.open_path(device_info.path) if handle: report_id = getattr(device_info, "centurion_report_id", None) or base.CENTURION_REPORT_ID - base._centurion_handles[int(handle)] = base.CenturionHandleState(report_id=report_id) + state = base.CenturionHandleState(report_id=report_id) + base._centurion_handles[int(handle)] = state + base.probe_centurion_device_addr(handle, state) 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 [])) diff --git a/lib/logitech_receiver/device.py b/lib/logitech_receiver/device.py index 4c2676c2..ae8dfdba 100644 --- a/lib/logitech_receiver/device.py +++ b/lib/logitech_receiver/device.py @@ -79,7 +79,9 @@ def create_device(low_level: LowLevelInterface, device_info, setting_callback=No if handle: if getattr(device_info, "centurion", False): report_id = getattr(device_info, "centurion_report_id", None) or base.CENTURION_REPORT_ID - base._centurion_handles[int(handle)] = base.CenturionHandleState(report_id=report_id) + state = base.CenturionHandleState(report_id=report_id) + base._centurion_handles[int(handle)] = state + base.probe_centurion_device_addr(handle, state) # a direct connected device might not be online (as reported by user) return Device( low_level, diff --git a/tests/logitech_receiver/test_base.py b/tests/logitech_receiver/test_base.py index 09549f32..e14164cf 100644 --- a/tests/logitech_receiver/test_base.py +++ b/tests/logitech_receiver/test_base.py @@ -313,3 +313,91 @@ class TestUnwrapCenturionFrame: assert result[0] == 0x11 assert result[1] == 0xFF assert result[2:63] == payload + + +class TestProbeCenturionDeviceAddr: + """Test probe_centurion_device_addr: write-then-read dance to learn device_addr.""" + + HANDLE = 101 + + def setup_method(self): + base._centurion_handles.pop(self.HANDLE, None) + + def teardown_method(self): + base._centurion_handles.pop(self.HANDLE, None) + + def test_learns_addr_from_first_frame(self): + state = CenturionHandleState(report_id=CENTURION_ADDRESSED_REPORT_ID) + reply = bytes([0x50, 0x23, 0x03, 0x00]) + b"\x00" * 60 + with ( + mock.patch.object(base.hidapi, "write") as mock_write, + mock.patch.object(base.hidapi, "read", return_value=reply) as mock_read, + ): + result = base.probe_centurion_device_addr(self.HANDLE, state) + assert result is True + assert state.device_addr == 0x23 + # probe write is a 64-byte all-zero frame with just the report ID + mock_write.assert_called_once() + _, wdata = mock_write.call_args[0] + assert len(wdata) == base.CENTURION_FRAME_SIZE + assert wdata[0] == CENTURION_ADDRESSED_REPORT_ID + assert wdata[1:] == b"\x00" * (base.CENTURION_FRAME_SIZE - 1) + mock_read.assert_called_once() + + def test_skips_non_matching_frames_until_match(self): + state = CenturionHandleState(report_id=CENTURION_ADDRESSED_REPORT_ID) + # first two reads return unrelated frames, third returns our 0x50 frame + noise = b"\x11\xff" + b"\x00" * 62 + match = bytes([0x50, 0x42, 0x03, 0x00]) + b"\x00" * 60 + with ( + mock.patch.object(base.hidapi, "write"), + mock.patch.object(base.hidapi, "read", side_effect=[noise, noise, match]), + ): + result = base.probe_centurion_device_addr(self.HANDLE, state) + assert result is True + assert state.device_addr == 0x42 + + def test_returns_false_on_timeout(self): + state = CenturionHandleState(report_id=CENTURION_ADDRESSED_REPORT_ID) + with ( + mock.patch.object(base.hidapi, "write"), + mock.patch.object(base.hidapi, "read", return_value=None), + ): + result = base.probe_centurion_device_addr(self.HANDLE, state) + assert result is False + assert state.device_addr is None + + def test_noop_for_0x51_variant(self): + state = CenturionHandleState(report_id=CENTURION_REPORT_ID) + with ( + mock.patch.object(base.hidapi, "write") as mock_write, + mock.patch.object(base.hidapi, "read") as mock_read, + ): + result = base.probe_centurion_device_addr(self.HANDLE, state) + assert result is False + assert state.device_addr is None + mock_write.assert_not_called() + mock_read.assert_not_called() + + def test_noop_when_addr_already_known(self): + state = CenturionHandleState(report_id=CENTURION_ADDRESSED_REPORT_ID, device_addr=0x23) + with ( + mock.patch.object(base.hidapi, "write") as mock_write, + mock.patch.object(base.hidapi, "read") as mock_read, + ): + result = base.probe_centurion_device_addr(self.HANDLE, state) + assert result is False + assert state.device_addr == 0x23 + mock_write.assert_not_called() + mock_read.assert_not_called() + + def test_handles_write_failure(self): + state = CenturionHandleState(report_id=CENTURION_ADDRESSED_REPORT_ID) + with ( + mock.patch.object(base.hidapi, "write", side_effect=OSError("no device")), + mock.patch.object(base.hidapi, "read") as mock_read, + ): + result = base.probe_centurion_device_addr(self.HANDLE, state) + assert result is False + assert state.device_addr is None + mock_read.assert_not_called()