diff --git a/lib/logitech_receiver/device.py b/lib/logitech_receiver/device.py index ff3565b7..f5e8a0f4 100644 --- a/lib/logitech_receiver/device.py +++ b/lib/logitech_receiver/device.py @@ -437,6 +437,65 @@ class Device: if self.online and self.protocol >= 2.0: _hidpp20.config_change(self, configuration_, no_reply=no_reply) + def signal_configuration_complete(self, cookie=None): + """SetComplete on ConfigChange to ack end of configuration. + + With no cookie, sends the host's session counter (see + Hidpp20.set_configuration_complete).""" + if self.online and self.protocol >= 2.0: + _hidpp20.set_configuration_complete(self, cookie=cookie) + + def _record_config_cookie(self): + """After a successful apply, SetComplete with the next session cookie + and persist it so the dedup gate in apply_settings_if_needed can + detect drift on follow-up reconfig notifications within this session.""" + if self.protocol < 2.0: + return + if not (self.features and SupportedFeature.CONFIG_CHANGE in self.features): + return + cookie = _hidpp20.next_session_cookie() + self.signal_configuration_complete(cookie=cookie) + if self.persister is not None: + self.persister["_config_cookie"] = [cookie[0], cookie[1]] + + def apply_settings_if_needed(self): + """Cookie-gated dedup helper for repeat WIRELESS_DEVICE_STATUS + reconfig notifications on an already-active device. Skips when the + live ConfigChange cookie matches the value stored by the most + recent apply, otherwise applies and re-records. Must NOT be used as + the initial-activation apply path — across power cycles, devices + whose firmware resets the cookie to a fixed value would falsely + match a stored cookie from a prior session and skip the apply the + device actually needs. + Returns True if apply ran, False if it was skipped.""" + if not self.online: + return False + if self.protocol >= 2.0 and self.features and SupportedFeature.CONFIG_CHANGE in self.features: + live = _hidpp20.get_configuration_cookie(self) + if live and len(live) >= 2: + stored = self.persister.get("_config_cookie") if self.persister else None + live_pair = [live[0], live[1]] + if stored == live_pair: + if logger.isEnabledFor(logging.INFO): + logger.info( + "%s: config cookie %02X%02X matches stored — skip apply_all_settings", + self, + live[0], + live[1], + ) + return False + if logger.isEnabledFor(logging.INFO): + logger.info( + "%s: config cookie live=%02X%02X stored=%s — apply_all_settings", + self, + live[0], + live[1], + "%02X%02X" % (stored[0], stored[1]) if stored else "None", + ) + settings.apply_all_settings(self) + self._record_config_cookie() + return True + def reset(self, no_reply=False): self.set_configuration(0, no_reply) @@ -536,12 +595,17 @@ class Device: ): if logger.isEnabledFor(logging.INFO): logger.info("%s pushing device settings %s", self, self.settings) + # Activation apply must be unconditional — across power + # cycles, the device may have lost state while its cookie + # reset to a value that happens to match what we stored + # last session. Cookie comparison is only a valid dedup + # signal for repeat reconfig notifications within an + # already-active session (see apply_settings_if_needed). settings.apply_all_settings(self) + self._record_config_cookie() if not was_active: if self.protocol < 2.0: # Make sure to set notification flags on the device self.notification_flags = self.enable_connection_notifications() - else: - self.set_configuration(0x11) # signal end of configuration self.read_battery() # battery information may have changed so try to read it now elif was_active and self.receiver and not isinstance(self.receiver, CenturionReceiver): hidpp10.set_configuration_pending_flags(self.receiver, 0xFF) diff --git a/lib/logitech_receiver/hidpp20.py b/lib/logitech_receiver/hidpp20.py index d000b409..992b3433 100644 --- a/lib/logitech_receiver/hidpp20.py +++ b/lib/logitech_receiver/hidpp20.py @@ -24,6 +24,7 @@ import threading from collections import UserDict from enum import Flag from enum import IntEnum +from random import getrandbits from typing import Any from typing import Dict from typing import Generator @@ -1666,6 +1667,11 @@ def feature_request(device, feature, function=0x00, *params, no_reply=False): class Hidpp20: + # Host-side counter for SetComplete cookies (see set_configuration_complete). + # Seeded to a non-zero random 16-bit value at import so successive sessions + # don't trivially collide; we just need to never send 0x0000. + _session_cookie = getrandbits(16) or 1 + def get_firmware(self, device) -> tuple[common.FirmwareInfo] | None: """Reads a device's firmware info. @@ -2051,7 +2057,29 @@ class Hidpp20: return struct.unpack("!B", result[:1])[0] return None + def get_configuration_cookie(self, device: Device): + """ConfigChange (0x0020) GetCookie — read the device's current configuration cookie.""" + response = device.feature_request(SupportedFeature.CONFIG_CHANGE, 0x00) + return response[:2] if response else None + + def next_session_cookie(self): + """Bump and return the host-side counter used as the SetComplete cookie.""" + Hidpp20._session_cookie = (Hidpp20._session_cookie + 1) & 0xFFFF or 1 + return bytes([Hidpp20._session_cookie >> 8, Hidpp20._session_cookie & 0xFF]) + + def set_configuration_complete(self, device: Device, cookie=None, no_reply=False): + """ConfigChange (0x0020) SetComplete — acknowledge host has synced with device configuration. + + Sends a host-side monotonic counter, incremented per call and + always non-zero. Cookie 0x0000 has been observed to release the + SW effect-engine claim on at least the G515 LS TKL; we avoid it.""" + if cookie is None: + cookie = self.next_session_cookie() + if cookie and len(cookie) >= 2: + return device.feature_request(SupportedFeature.CONFIG_CHANGE, 0x10, cookie[0], cookie[1], no_reply=no_reply) + def config_change(self, device: Device, configuration, no_reply=False): + """Deprecated — use set_configuration_complete() instead.""" return device.feature_request(SupportedFeature.CONFIG_CHANGE, 0x10, configuration, no_reply=no_reply) diff --git a/lib/logitech_receiver/notifications.py b/lib/logitech_receiver/notifications.py index 693e3afd..21ede6c5 100644 --- a/lib/logitech_receiver/notifications.py +++ b/lib/logitech_receiver/notifications.py @@ -322,9 +322,14 @@ def _process_feature_notification(device: Device, notification: HIDPPNotificatio if logger.isEnabledFor(logging.DEBUG): logger.debug("wireless status: %s", notification) reason = "powered on" if notification.data[2] == 1 else None - if notification.data[1] == 1: # device is asking for software reconfiguration so need to change status + if notification.data[1] == 1: # device is asking for software reconfiguration alert = Alert.NONE - device.changed(active=True, alert=alert, reason=reason, push=True) + device.changed(active=True, alert=alert, reason=reason) + # changed(active=True) already runs apply_settings_if_needed on + # the first transition; for follow-up reconfig notifications + # on an already-active device, fire the gate here so the + # cookie comparison decides whether to re-push. + device.apply_settings_if_needed() else: logger.warning("%s: unknown WIRELESS %s", device, notification) diff --git a/tests/logitech_receiver/test_hidpp20_simple.py b/tests/logitech_receiver/test_hidpp20_simple.py index f68c0992..b8f88896 100644 --- a/tests/logitech_receiver/test_hidpp20_simple.py +++ b/tests/logitech_receiver/test_hidpp20_simple.py @@ -372,6 +372,47 @@ def test_config_change(): assert result == bytes.fromhex("03FFFF") +def test_get_configuration_cookie(): + responses = [fake_hidpp.Response("12345678", 0x0400)] + device = fake_hidpp.Device(responses=responses, feature=SupportedFeature.CONFIG_CHANGE) + + result = _hidpp20.get_configuration_cookie(device) + + assert result == bytes.fromhex("1234") + + +def test_set_configuration_complete_explicit_cookie(): + # An explicit cookie is sent unchanged. + responses = [fake_hidpp.Response("00", 0x0410, "1234")] + device = fake_hidpp.Device(responses=responses, feature=SupportedFeature.CONFIG_CHANGE) + + result = _hidpp20.set_configuration_complete(device, cookie=bytes.fromhex("1234")) + + assert result == bytes.fromhex("00") + + +def test_set_configuration_complete_monotonic_counter(): + # With no cookie, sends a host-side monotonic counter, +1 per call. + hidpp20.Hidpp20._session_cookie = 0x1233 + responses = [ + fake_hidpp.Response("00", 0x0410, "1234"), + fake_hidpp.Response("00", 0x0410, "1235"), + ] + device = fake_hidpp.Device(responses=responses, feature=SupportedFeature.CONFIG_CHANGE) + + assert _hidpp20.set_configuration_complete(device) == bytes.fromhex("00") + assert _hidpp20.set_configuration_complete(device) == bytes.fromhex("00") + + +def test_set_configuration_complete_skips_zero(): + # Counter at 0xFFFF wraps to 1, never 0. + hidpp20.Hidpp20._session_cookie = 0xFFFF + responses = [fake_hidpp.Response("00", 0x0410, "0001")] + device = fake_hidpp.Device(responses=responses, feature=SupportedFeature.CONFIG_CHANGE) + + assert _hidpp20.set_configuration_complete(device) == bytes.fromhex("00") + + def test_decipher_battery_status(): report = b"\x50\x20\x00\xff\xff"