diff --git a/lib/logitech_receiver/rgb_power.py b/lib/logitech_receiver/rgb_power.py index 96ae6e2f..464d0ca3 100644 --- a/lib/logitech_receiver/rgb_power.py +++ b/lib/logitech_receiver/rgb_power.py @@ -802,6 +802,17 @@ class RGBPowerManager: if logger.isEnabledFor(logging.WARNING): logger.warning("%s: failed to wake RGB LEDs: %s", self._device, e) + # --- Suspend / Restore (used e.g. to toggle LEDs on screen lock) --- + + def suspend_leds(self): + """Force enter sleep state (e.g. on screen lock).""" + if self._state != self.SLEEPING: + self._start_sleep() + + def restore_leds(self): + """Force wake from sleep state (e.g. on screen unlock).""" + self._wake() + def _cancel_dim_timer(self): if self._dim_timer_id is not None: GLib.source_remove(self._dim_timer_id) diff --git a/lib/logitech_receiver/settings.py b/lib/logitech_receiver/settings.py index 0f32f72b..c5ff4806 100644 --- a/lib/logitech_receiver/settings.py +++ b/lib/logitech_receiver/settings.py @@ -156,7 +156,7 @@ class Setting: self._value = value self._pre_write(save) - def write(self, value, save=True): + def write(self, value, save=True, volatile=False): assert hasattr(self, "_value") assert hasattr(self, "_device") assert value is not None @@ -165,7 +165,7 @@ class Setting: logger.debug("%s: write %r to %s", self.name, value, self._device) if self._device.online: - if self._value != value: + if self._value != value and not volatile: self.update(value, save) current_value = None diff --git a/lib/solaar/dbus.py b/lib/solaar/dbus.py index 142b5904..121b221a 100644 --- a/lib/solaar/dbus.py +++ b/lib/solaar/dbus.py @@ -22,23 +22,36 @@ from typing import Callable logger = logging.getLogger(__name__) +system_bus: dbus.SystemBus | None = None +session_bus: dbus.SessionBus | None = None + try: import dbus from dbus.mainloop.glib import DBusGMainLoop # integration into the main GLib loop DBusGMainLoop(set_as_default=True) - bus = dbus.SystemBus() - assert bus -except Exception: - # Either the dbus library is not available or the system dbus is not running - logger.warning("failed to set up dbus") - bus = None + try: + system_bus = dbus.SystemBus() + except dbus.exceptions.DBusException as e: + logger.warning(f"Failed to connect to system dbus: {e}") + + try: + session_bus = dbus.SessionBus() + except dbus.exceptions.DBusException as e: + logger.warning(f"Failed to connect to session dbus: {e}") + +except ImportError: + logger.warning("dbus library not available, dbus setup failed.") +except Exception as e: + logger.error(f"Unexpected error during D-Bus initialization: {e}") _suspend_callback = None _resume_callback = None +_lock_callback = None +_unlock_callback = None def _suspend_or_resume(suspend): @@ -48,8 +61,17 @@ def _suspend_or_resume(suspend): _resume_callback() +def _lock_or_unlock(locked): + if locked and _lock_callback: + _lock_callback() + if not locked and _unlock_callback: + _unlock_callback() + + _LOGIND_PATH = "/org/freedesktop/login1" _LOGIND_INTERFACE = "org.freedesktop.login1.Manager" +_SCREEN_SAVER_PATH = "/ScreenSaver" +_SCREEN_SAVER_INTERFACE = "org.freedesktop.ScreenSaver" def watch_suspend_resume( @@ -61,8 +83,8 @@ def watch_suspend_resume( global _resume_callback, _suspend_callback _suspend_callback = on_suspend_callback _resume_callback = on_resume_callback - if bus is not None and on_resume_callback is not None or on_suspend_callback is not None: - bus.add_signal_receiver( + if system_bus is not None and on_resume_callback is not None or on_suspend_callback is not None: + system_bus.add_signal_receiver( _suspend_or_resume, "PrepareForSleep", dbus_interface=_LOGIND_INTERFACE, @@ -71,6 +93,22 @@ def watch_suspend_resume( logger.info("connected to system dbus, watching for suspend/resume events") +def watch_screen_lock_state( + on_lock_callback: Callable[[], None], + on_unlock_callback: Callable[[], None], +): + """Register callbacks for screen lock/unlock events. + They are called only if the session DBus is running and the Desktop environments sends + events on the org.freedesktop.ScreenSaver interface.""" + global _lock_callback, _unlock_callback + _lock_callback = on_lock_callback + _unlock_callback = on_unlock_callback + session_bus.add_signal_receiver( + _lock_or_unlock, "ActiveChanged", dbus_interface=_SCREEN_SAVER_INTERFACE, path=_SCREEN_SAVER_PATH + ) + logger.info("connected to session dbus watching for screen lock/unlock events") + + _BLUETOOTH_PATH_PREFIX = "/org/bluez/hci0/dev_" _BLUETOOTH_INTERFACE = "org.freedesktop.DBus.Properties" @@ -81,7 +119,7 @@ def watch_bluez_connect(serial, callback=None): if _bluetooth_callbacks.get(serial): _bluetooth_callbacks.get(serial).remove() path = _BLUETOOTH_PATH_PREFIX + serial.replace(":", "_").upper() - if bus is not None and callback is not None: - _bluetooth_callbacks[serial] = bus.add_signal_receiver( + if system_bus is not None and callback is not None: + _bluetooth_callbacks[serial] = system_bus.add_signal_receiver( callback, "PropertiesChanged", path=path, dbus_interface=_BLUETOOTH_INTERFACE ) diff --git a/lib/solaar/gtk.py b/lib/solaar/gtk.py index e0680820..84a7b978 100755 --- a/lib/solaar/gtk.py +++ b/lib/solaar/gtk.py @@ -198,6 +198,8 @@ def main(): else: dbus.watch_suspend_resume(lambda: listener.ping_all(True)) + dbus.watch_screen_lock_state(listener.suspend_leds, listener.restore_leds) + configuration.defer_saves = True # allow configuration saves to be deferred # main UI event loop diff --git a/lib/solaar/listener.py b/lib/solaar/listener.py index f09a9f18..d5830379 100644 --- a/lib/solaar/listener.py +++ b/lib/solaar/listener.py @@ -35,6 +35,9 @@ from logitech_receiver import exceptions from logitech_receiver import hidpp10_constants from logitech_receiver import listener from logitech_receiver import notifications +from logitech_receiver import rgb_power +from logitech_receiver import settings_templates as st +from logitech_receiver.hidpp20 import LEDEffectSetting from . import configuration from . import dbus @@ -452,6 +455,52 @@ def ping_all(resuming=False): break +def suspend_leds(): + """Turn off LEDs for all connected devices.""" + logger.info("screen locked, suspending LED lighting on all devices") + + for listener_thread in _all_listeners.values(): + devices = [listener_thread.receiver] if listener_thread.receiver.isDevice else list(listener_thread.receiver) + for dev in devices: + if not dev.online: + continue + + if mgr := rgb_power.get_manager(dev): + mgr.suspend_leds() + return + + for s in dev.settings: + if s.name in (st.Backlight.name, st.Backlight2.name, st.Backlight2Level.name): + s.write(0, save=False, volatile=True) + elif s.name.startswith(st.LEDZoneSetting.name) or s.name.startswith(st.RGBEffectSetting.name): + s.write(LEDEffectSetting(ID=0), save=False, volatile=True) + elif s.name == st.PerKeyLighting.name and s.read(): + s.write({k: 0 for k in s._value.keys()}, save=False, volatile=True) + + +def restore_leds(): + """Restore LED lighting for all connected devices to their saved states.""" + logger.info("screen unlocked, restoring LED lighting on all devices") + + for listener_thread in _all_listeners.values(): + devices = [listener_thread.receiver] if listener_thread.receiver.isDevice else list(listener_thread.receiver) + for dev in devices: + if not dev.online: + continue + + if mgr := rgb_power.get_manager(dev): + mgr.restore_leds() + return + + for s in dev.settings: + if ( + s.name in (st.Backlight.name, st.Backlight2.name, st.Backlight2Level.name, st.PerKeyLighting.name) + or s.name.startswith(st.LEDZoneSetting.name) + or s.name.startswith(st.RGBEffectSetting.name) + ): + s.apply() + + _status_callback = None # GUI callback to change UI in response to changes to receiver or device status _setting_callback = None # GUI callback to change UI in response to changes to status _error_callback = None # GUI callback to report errors