Probe Centurion 0x50 device address on handle open

The 0x50 variant requires a device-specific address byte at frame[1]
on every TX frame. Until now we left state.device_addr=None and sent
0x00 as a placeholder, relying on the device to either be permissive
enough to respond, or to send an unsolicited frame early enough for
_unwrap_centurion_frame to learn the address passively.

Strict firmware silently drops device_addr=0x00 requests, which breaks
dongle feature discovery before it can start: _discover_dongle_features
times out, has_bridge is False, create_centurion_receiver falls through
to create_device, Device.__init__ leaves _protocol=None, and a later
read_battery() dispatches HID++ 1.0 read_register(BATTERY_CHARGE) that
the dongle rejects with INVALID_SUB_ID_COMMAND.

Port strain08's fix from LGSTrayEx (commits 1439b27a + c6d21972): right
after registering a 0x50 handle, write a 64-byte all-zero frame with
just the report ID set. That elicits an error/unsolicited response
whose byte[1] is the real device address. Read up to 3 x 500ms until
a matching frame arrives, then store the address on the handle state
so subsequent TX frames carry it correctly.

On timeout the probe logs a warning and leaves device_addr=None, so
behavior falls back to the current 0x00-placeholder path (no regression
for devices where the probe isn't needed). The passive learn-on-first-RX
in _unwrap_centurion_frame is preserved as a second line of defense.

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.

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 []))

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,

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()