diff --git a/src/services/__init__.py b/src/services/__init__.py index 7d00053..c9de1ad 100644 --- a/src/services/__init__.py +++ b/src/services/__init__.py @@ -1,5 +1,5 @@ from fabric.audio import Audio # noqa: F401 -from .battery import BatteryService # noqa: F401 +from .battery import Battery # noqa: F401 from .brightness import Brightness # noqa: F401 from .capslock import CapsLock # noqa: F401 from .keyboard_layout import KeyboardLayout # noqa: F401 diff --git a/src/services/battery.py b/src/services/battery.py index 554c8c7..21432cb 100644 --- a/src/services/battery.py +++ b/src/services/battery.py @@ -1,11 +1,19 @@ -from typing import Literal, Optional - +import gi from fabric.core.service import Property, Service, Signal -from fabric.utils import Gio, GLib, logger +from gi.repository import Gio, GLib +from loguru import logger -from utils.dbus_helper import GioDBusHelper +gi.require_version("Gtk", "3.0") -DeviceState = { +BATTERY_BUS_NAME = "org.freedesktop.UPower" +BATTERY_BUS_PATH = "/org/freedesktop/UPower/devices/DisplayDevice" +BATTERY_INTERFACE = "org.freedesktop.UPower.Device" + +POWER_PROFILE_BUS_NAME = "net.hadess.PowerProfiles" +POWER_PROFILE_BUS_PATH = "/net/hadess/PowerProfiles" +POWER_PROFILE_INTERFACE = "net.hadess.PowerProfiles" + +DEVICE_STATE = { 0: "UNKNOWN", 1: "CHARGING", 2: "DISCHARGING", @@ -15,188 +23,207 @@ DeviceState = { 6: "PENDING_DISCHARGE", } -PowerProfile = { - "power-saver": "Power Saver", - "balanced": "Balanced", - "performance": "Performance", -} - -class BatteryService(Service): - """Service to interact with UPower and Power Profiles via GIO D-Bus""" +class Battery(Service): + """A service for interacting with the battery's DBus and Power Profiles""" @Signal - def changed(self) -> None: - """Signal emitted when battery changes.""" + def changed(self) -> None: ... @Signal - def power_profile_changed(self) -> None: - """Signal emitted when power profile changes.""" + def power_profile_changed(self) -> None: ... - _instance = None + @Property(bool, "readable", default_value=False) + def available(self) -> bool: + return self.do_get_cached_property("IsPresent") or False - def __new__(cls): - if cls._instance is None: - cls._instance = super().__new__(cls) - return cls._instance + @Property(str, "readable", default_value="") + def vendor(self) -> str: + return self.do_get_cached_property("Vendor") or "" + + @Property(int, "readable", default_value=0) + def percent(self) -> int: + return self.do_get_cached_property("Percentage") or 0 + + @Property(bool, "readable", default_value=False) + def charging(self) -> bool: + val = self.do_get_cached_property("State") + if val is None: + return False + return val == DEVICE_STATE.get("CHARGING", 1) + + @Property(bool, "readable", default_value=False) + def discharging(self) -> bool: + val = self.do_get_cached_property("State") + if val is None: + return False + return val == DEVICE_STATE.get("DISCHARGING", 2) + + @Property(bool, "readable", default_value=False) + def charged(self) -> bool: + val = self.do_get_cached_property("State") + if val is None: + return False + return val == DEVICE_STATE.get("FULLY_CHARGED", 4) + + @Property(str, "readable", default_value="") + def icon_name(self) -> str: + return self.do_get_cached_property("IconName") or "" + + @Property(int, "readable", default_value=0) + def time_remaining(self) -> int: + return self.do_get_cached_property("TimeToEmpty") or 0 + + @Property(int, "readable", default_value=0) + def time_to_full(self) -> int: + return self.do_get_cached_property("TimeToFull") or 0 + + @Property(float, "readable", default_value=0.0) + def energy(self) -> float: + return self.do_get_cached_property("Energy") or 0.0 + + @Property(float, "readable", default_value=0.0) + def energy_full(self) -> float: + return self.do_get_cached_property("EnergyFull") or 0.0 + + @Property(float, "readable", default_value=0.0) + def energy_rate(self) -> float: + return self.do_get_cached_property("EnergyRate") or 0.0 + + @Property(float, "readable", default_value=0.0) + def temperature(self) -> float: + return self.do_get_cached_property("Temperature") or 0.0 def __init__(self, **kwargs): super().__init__(**kwargs) + self._bus: Gio.DBusConnection | None = None + self._proxy = None + self._power_profile_proxy = None + self.do_register() - # UPower D-Bus configuration - self.bus_name = "org.freedesktop.UPower" - self.object_path = "/org/freedesktop/UPower/devices/DisplayDevice" - self.interface_name = "org.freedesktop.UPower.Device" - - self.dbus_helper = GioDBusHelper( - bus_type=Gio.BusType.SYSTEM, - bus_name=self.bus_name, - object_path=self.object_path, - interface_name=self.interface_name, + def do_register(self) -> None: + self._bus = Gio.bus_get_sync(Gio.BusType.SYSTEM) + self._proxy = Gio.DBusProxy.new_sync( + self._bus, + Gio.DBusProxyFlags.NONE, + None, + BATTERY_BUS_NAME, + BATTERY_BUS_PATH, + BATTERY_INTERFACE, + None, ) - self.proxy = self.dbus_helper.proxy + logger.info("[Battery] Proxy initialized") - # Listen for PropertiesChanged signals from UPower - self.dbus_helper.listen_signal( - member="PropertiesChanged", - callback=self.handle_property_change, + self._bus.signal_subscribe( + BATTERY_BUS_NAME, + "org.freedesktop.DBus.Properties", + "PropertiesChanged", + BATTERY_BUS_PATH, + None, + Gio.DBusSignalFlags.NONE, + self.do_handle_property_change, ) - # Power Profiles D-Bus configuration - self.power_profile_bus_name = "net.hadess.PowerProfiles" - self.power_profile_object_path = "/net/hadess/PowerProfiles" - self.power_profile_interface_name = "net.hadess.PowerProfiles" - try: - self.power_profile_helper = GioDBusHelper( - bus_type=Gio.BusType.SYSTEM, - bus_name=self.power_profile_bus_name, - object_path=self.power_profile_object_path, - interface_name=self.power_profile_interface_name, + self._power_profile_proxy = Gio.DBusProxy.new_sync( + self._bus, + Gio.DBusProxyFlags.NONE, + None, + POWER_PROFILE_BUS_NAME, + POWER_PROFILE_BUS_PATH, + POWER_PROFILE_INTERFACE, + None, ) - - self.power_profile_proxy = self.power_profile_helper.proxy - - # Listen for PropertiesChanged signals from Power Profiles - self.power_profile_helper.listen_signal( - member="PropertiesChanged", - callback=self.handle_power_profile_change, + self._bus.signal_subscribe( + POWER_PROFILE_BUS_NAME, + "org.freedesktop.DBus.Properties", + "PropertiesChanged", + POWER_PROFILE_BUS_PATH, + None, + Gio.DBusSignalFlags.NONE, + self.do_handle_power_profile_change, ) - - self._power_profiles_available = True + logger.info("[Battery] Power Profiles Proxy initialized") except Exception as e: logger.warning(f"[Battery] Power Profiles daemon not available: {e}") - self.power_profile_helper = None - self.power_profile_proxy = None - self._power_profiles_available = False - def get_property( + def do_handle_property_change(self, *_): + self.emit("changed") + + def do_handle_power_profile_change(self, *_): + self.emit("power_profile_changed") + + def do_call_proxy_method( self, - property: Literal[ - "Percentage", - "Temperature", - "TimeToEmpty", - "TimeToFull", - "IconName", - "State", - "Capacity", - "IsPresent", - "Vendor", - ], + bus_name, + object_path, + interface_name, + method_name, + parameters=None, + timeout=-1, ): - try: - result = self.proxy.get_cached_property(property) - return result.unpack() if result is not None else None - except Exception as e: - logger.exception(f"[Battery] Error retrieving '{property}': {e}") - return None + if parameters is None: + parameters = GLib.Variant("()", ()) + result = self._bus.call_sync( + bus_name, + object_path, + interface_name, + method_name, + parameters, + None, + Gio.DBusCallFlags.NONE, + timeout, + None, + ) + return result.unpack() - def get_power_profile(self) -> Optional[str]: + def do_get_cached_property(self, property_name): + result = self._proxy.get_cached_property(property_name) + return result.unpack() if result is not None else None + + def get_power_profile(self) -> str | None: """Get the current active power profile.""" - if not self._power_profiles_available: + if not self._power_profile_proxy: return None + result = self._power_profile_proxy.get_cached_property("ActiveProfile") + return result.unpack() if result is not None else None - try: - result = self.power_profile_proxy.get_cached_property("ActiveProfile") - return result.unpack() if result is not None else None - except Exception as e: - logger.exception(f"[Battery] Error retrieving active power profile: {e}") - return None - - def set_power_profile( - self, profile: Literal["power-saver", "balanced", "performance"] - ) -> bool: + def set_power_profile(self, profile: str) -> bool: """Set the active power profile.""" - if not self._power_profiles_available: - logger.warning("[Battery] Power Profiles daemon not available") + if not self._power_profile_proxy: return False - - if profile not in PowerProfile: - return False - try: - self.power_profile_helper.set_property( - interface_name=self.power_profile_interface_name, - property_name="ActiveProfile", - value_variant=GLib.Variant("s", profile), + parameters = GLib.Variant( + "(ssv)", + (POWER_PROFILE_INTERFACE, "ActiveProfile", GLib.Variant("s", profile)), + ) + self._bus.call_sync( + POWER_PROFILE_BUS_NAME, + POWER_PROFILE_BUS_PATH, + "org.freedesktop.DBus.Properties", + "Set", + parameters, + None, + Gio.DBusCallFlags.NONE, + -1, + None, ) return True except Exception as e: - logger.exception( - f"[Battery] Error setting power profile to '{profile}': {e}" - ) + logger.error(f"[Battery] Failed to set power profile: {e}") return False - def get_available_power_profiles(self) -> Optional[list]: + def get_available_power_profiles(self) -> list | None: """Get list of available power profiles.""" - if not self._power_profiles_available: + if not self._power_profile_proxy: return None - - try: - result = self.power_profile_proxy.get_cached_property("Profiles") - if result is not None: - profiles_data = result.unpack() - # Extract profile names from the array of dictionaries - profiles = [] - for profile_dict in profiles_data: - if "Profile" in profile_dict: - profiles.append(profile_dict["Profile"]) - return profiles - return None - except Exception as e: - logger.exception( - f"[Battery] Error retrieving available power profiles: {e}" - ) - return None - - def is_power_profiles_available(self) -> bool: - """Check if power profiles daemon is available.""" - return self._power_profiles_available - - def get_power_profile_display_name(self, profile: str) -> str: - """Get display name for a power profile.""" - return PowerProfile.get(profile, profile.title()) - - @Property(int, "readable", default_value=0) - def percentage(self) -> int: - return self.get_property("Percentage") or 0 - - @Property(int, "readable", default_value=0) - def state(self) -> int: - return self.get_property("State") or 0 - - @Property(bool, "readable", default_value=False) - def is_present(self) -> bool: - return self.get_property("IsPresent") or False - - def handle_property_change(self, *args): - # Notify about property changes for OSD and other listeners - self.notify("percentage") - self.notify("state") - self.notify("is-present") - self.emit("changed") - - def handle_power_profile_change(self, *_): - """Handle power profile property changes.""" - self.emit("power_profile_changed") + result = self._power_profile_proxy.get_cached_property("Profiles") + if result is not None: + profiles_data = result.unpack() + profiles = [] + for profile_dict in profiles_data: + if "Profile" in profile_dict: + profiles.append(profile_dict["Profile"]) + return profiles + return None diff --git a/src/services/brightness.py b/src/services/brightness.py index 0723e2f..f29a21f 100644 --- a/src/services/brightness.py +++ b/src/services/brightness.py @@ -1,91 +1,338 @@ +import subprocess +import time +import shutil + from fabric.core.service import Property, Service, Signal -from fabric.utils import GLib, exec_shell_command_async, logger, monitor_file, os - - -def exec_brightnessctl_async(args: str): - exec_shell_command_async(f"brightnessctl {args}", lambda _: None) - - -# Discover screen backlight device -try: - screen_device = os.listdir("/sys/class/backlight") - screen_device = screen_device[0] if screen_device else "" -except FileNotFoundError: - logger.error("No backlight devices found, brightness control disabled") - screen_device = "" +from fabric.utils import GLib, exec_shell_command_async, logger, os, re class Brightness(Service): - """Service to manage screen brightness levels.""" + """Service for controlling screen brightness using ddcutil or brightnessctl backends. - instance = None + The service works with RAW values (0 to max_screen) for both backends: + - brightnessctl: raw values are device-specific (e.g., 0-96000) + - ddcutil: raw values are percentages (0-100) - @staticmethod - def get_initial(): - if Brightness.instance is None: - Brightness.instance = Brightness() + The 'screen' signal emits percentage values (0-100) for UI display. + """ - return Brightness.instance + _instance = None + DDCUTIL_PARAMS = "--disable-dynamic-sleep --sleep-multiplier=0.05" + MIN_CHANGE_THRESHOLD = 2 + CACHE_INTERVAL = 3 + POLL_INTERVAL = 500 + + def __new__(cls, *args, **kwargs): + if cls._instance is None: + cls._instance = super().__new__(cls) + return cls._instance @Signal def screen(self, value: int) -> None: - """Signal emitted when screen brightness changes.""" - # Implement as needed for your application + """Signal emitted when screen brightness changes (value: percentage from 0 to 100).""" - def __init__(self, **kwargs): - super().__init__(**kwargs) - - # Path for screen backlight control - self.screen_backlight_path = f"/sys/class/backlight/{screen_device}" - - # Initialize maximum brightness level - self.max_screen = self.do_read_max_brightness(self.screen_backlight_path) - - if screen_device == "": + def __init__(self, backend=None, **kwargs): + """Initialize service with automatic backend detection.""" + if getattr(self, "_initialized", False): return + self._initialized = True + super().__init__(**kwargs) + self._pending_raw = None + self._timer_id = None + self._poll_timer_id = None + self._lock = GLib.Mutex() + self._last_percent = -1 + self._last_raw = -1 + self._last_update_time = 0 + self._last_file_mtime = 0 - # Monitor screen brightness file - self.screen_monitor = monitor_file(f"{self.screen_backlight_path}/brightness") + self.backend = self._detect_backend(backend) - self.screen_monitor.connect( - "changed", - lambda _, file, *args: self.emit( - "screen", - round(int(file.load_bytes()[0].get_data())), - ), + self.max_screen = self._read_max_brightness() or 100 + + if self.backend: + if self.backend == "ddcutil": + GLib.timeout_add(100, lambda: self._update_brightness_cache()) + else: + self._setup_polling() + + def _setup_polling(self): + """Setup periodic polling of brightness file.""" + try: + file_path = f"/sys/class/backlight/{self._get_screen_device()}/brightness" + if os.path.exists(file_path): + with open(file_path) as f: + self._last_raw = int(f.readline().strip()) + self._last_percent = ( + int((self._last_raw / self.max_screen) * 100) + if self.max_screen > 0 + else 0 + ) + + self._last_file_mtime = os.path.getmtime(file_path) + self._poll_timer_id = GLib.timeout_add( + self.POLL_INTERVAL, self._check_brightness_file + ) + except Exception as e: + logger.error(f"Error setting up brightness polling: {e}") + + def _check_brightness_file(self): + """Periodically check brightness file for changes.""" + try: + file_path = f"/sys/class/backlight/{self._get_screen_device()}/brightness" + if os.path.exists(file_path): + current_mtime = os.path.getmtime(file_path) + if current_mtime > self._last_file_mtime: + self._last_file_mtime = current_mtime + with open(file_path) as f: + raw = int(f.readline().strip()) + + if raw != self._last_raw: + self._last_raw = raw + percent = ( + int((raw / self.max_screen) * 100) + if self.max_screen > 0 + else 0 + ) + if ( + abs(percent - self._last_percent) + >= self.MIN_CHANGE_THRESHOLD + ): + self._last_percent = percent + self.emit("screen", percent) + return True + except Exception as e: + logger.error(f"Error checking brightness file: {e}") + return True + + def _detect_backend(self, backend): + """Detect appropriate backend for brightness control.""" + if backend: + logger.info(f"Using forced backend: {backend}") + return backend + + if shutil.which("brightnessctl"): + device = self._get_screen_device() + if device: + logger.info(f"Using brightnessctl backend with device: {device}") + return "brightnessctl" + else: + logger.debug( + "brightnessctl is available but no backlight devices found in /sys/class/backlight/" + ) + + if shutil.which("ddcutil"): + bus = self._detect_ddcutil_bus() + if bus != -1: + self.ddcutil_bus = bus + logger.info(f"Using ddcutil backend with I2C bus: {bus}") + return "ddcutil" + else: + logger.debug( + "ddcutil is available but no DDC/CI capable monitors detected" + ) + + logger.warning( + "No available backend for brightness control - no backlight devices or DDC/CI monitors found" ) + return None - # Log the initialization of the service - logger.info(f"Brightness service initialized for device: {screen_device}") + def _get_screen_device(self): + """Return first backlight device from sysfs.""" + try: + return os.listdir("/sys/class/backlight")[0] + except Exception: + return "" - def do_read_max_brightness(self, path: str) -> int: - # Reads the maximum brightness value from the specified path. - max_brightness_path = os.path.join(path, "max_brightness") - if os.path.exists(max_brightness_path): - with open(max_brightness_path) as f: - return int(f.readline()) - return -1 # Return -1 if file doesn't exist, indicating an error. + def _detect_ddcutil_bus(self): + """Detect I2C bus number for ddcutil.""" + try: + process = subprocess.run( + ["ddcutil", "detect"], text=True, capture_output=True, timeout=2 + ) + if process.returncode == 0: + match = re.search(r"I2C bus:\s*/dev/i2c-(\d+)", process.stdout) + return int(match.group(1)) if match else -1 + return -1 + except Exception: + return -1 + + def _read_max_brightness(self): + """Read maximum brightness value""" + if self.backend: + if self.backend == "ddcutil": + try: + process = subprocess.run( + [ + "ddcutil", + "--bus", + str(self.ddcutil_bus), + *self.DDCUTIL_PARAMS.split(), + "getvcp", + "10", + ], + text=True, + capture_output=True, + timeout=2, + ) + + if process.returncode == 0: + match = re.search( + r"current value\s*=\s*(\d+)\s*,\s*max value\s*=\s*(\d+)", + process.stdout, + ) + if match: + return int(match.group(2)) + except Exception as e: + logger.error(f"Error executing ddcutil: {e}") + else: + try: + with open( + f"/sys/class/backlight/{self._get_screen_device()}/max_brightness" + ) as f: + return int(f.readline().strip()) + except Exception: + return None + + def _update_brightness_cache(self): + """Update brightness cache with current value.""" + if self.backend == "ddcutil": + self.screen_brightness + return False @Property(int, "read-write") - def screen_brightness(self) -> int: - # Property to get or set the screen brightness. - brightness_path = os.path.join(self.screen_backlight_path, "brightness") - if os.path.exists(brightness_path): - with open(brightness_path) as f: - return int(f.readline()) - logger.warning(f"Brightness file does not exist: {brightness_path}") - return -1 # Return -1 if file doesn't exist, indicating error. + def screen_brightness(self): + """Getter returns current brightness in RAW value (0 to max_screen).""" + if not self.backend: + return -1 + + if self.backend == "brightnessctl": + if self._last_raw != -1: + return self._last_raw + + try: + with open( + f"/sys/class/backlight/{self._get_screen_device()}/brightness" + ) as f: + raw = int(f.readline().strip()) + self._last_raw = raw + return raw + except Exception as e: + logger.error(f"Error reading brightness file: {e}") + return -1 + elif self.backend == "ddcutil": + if ( + time.time() - self._last_update_time < self.CACHE_INTERVAL + and self._last_raw != -1 + ): + return self._last_raw + + try: + process = subprocess.run( + [ + "ddcutil", + "--bus", + str(self.ddcutil_bus), + *self.DDCUTIL_PARAMS.split(), + "getvcp", + "10", + ], + text=True, + capture_output=True, + timeout=2, + ) + + if process.returncode == 0: + match = re.search( + r"current value\s*=\s*(\d+)\s*,\s*max value\s*=\s*(\d+)", + process.stdout, + ) + if match: + current = int(match.group(1)) + + self._last_raw = current + self._last_update_time = time.time() + return current + except Exception as e: + logger.error(f"Error executing ddcutil: {e}") + + return self._last_raw if self._last_raw != -1 else -1 @screen_brightness.setter def screen_brightness(self, value: int): - # Setter for screen brightness property. - if not (0 <= value <= self.max_screen): + """Setter accepts brightness value in RAW (0 to max_screen).""" + self._lock.lock() + try: value = max(0, min(value, self.max_screen)) + current_percent = ( + int((self._last_raw / self.max_screen) * 100) + if self._last_raw != -1 and self.max_screen > 0 + else -1 + ) + new_percent = ( + int((value / self.max_screen) * 100) if self.max_screen > 0 else 0 + ) + + if ( + abs(new_percent - current_percent) < self.MIN_CHANGE_THRESHOLD + and self._last_raw != -1 + ): + return + + self._pending_raw = value + + if self._timer_id: + GLib.source_remove(self._timer_id) + self._timer_id = GLib.timeout_add(50, self._apply_brightness) + finally: + self._lock.unlock() + + def _apply_brightness(self): + """Apply pending brightness change with optimized debouncing.""" + self._lock.lock() try: - exec_brightnessctl_async(f"--device '{screen_device}' set {value}") - self.emit("screen", int((value / self.max_screen) * 100)) - except GLib.Error as e: - logger.error(f"Error setting screen brightness: {e.message}") + if self._pending_raw is None: + self._timer_id = None + return False + + raw = self._pending_raw + self._pending_raw = None + self._timer_id = None + finally: + self._lock.unlock() + + try: + self._last_raw = raw + + percent = int((raw / self.max_screen) * 100) if self.max_screen > 0 else 0 + + if self.backend == "brightnessctl": + self.emit("screen", percent) + exec_shell_command_async( + f"brightnessctl --device '{self._get_screen_device()}' set {raw}" + ) + elif self.backend == "ddcutil": + self._last_update_time = time.time() + self.emit("screen", percent) + exec_shell_command_async( + f"ddcutil --bus {self.ddcutil_bus} {self.DDCUTIL_PARAMS} --terse setvcp 10 {raw}", + lambda exit_code, stdout, stderr: ( + logger.error(f"ddcutil error (code {exit_code}): {stderr}") + if exit_code != 0 + else None + ), + ) except Exception as e: - logger.exception(f"Unexpected error setting screen brightness: {e}") + logger.error(f"Error setting brightness: {e}") + return False + + def cleanup(self): + """Clean up resources when service is stopped.""" + if self._timer_id: + GLib.source_remove(self._timer_id) + self._timer_id = None + + if self._poll_timer_id: + GLib.source_remove(self._poll_timer_id) + self._poll_timer_id = None diff --git a/src/services/custom_notification.py b/src/services/custom_notification.py index c7cf3c3..2729e8d 100644 --- a/src/services/custom_notification.py +++ b/src/services/custom_notification.py @@ -271,17 +271,24 @@ class CachedNotifications(Notifications): max_cache_id = 0 for notification in data: - cached_notification = CachedNotification.create_from_dict(notification) - cache_id = cached_notification.cache_id - max_cache_id = max(max_cache_id, cache_id) + try: + cached_notification = CachedNotification.create_from_dict(notification) + cache_id = cached_notification.cache_id + max_cache_id = max(max_cache_id, cache_id) - handler_id = cached_notification.connect( - "removed-from-cache", - lambda *args: self.remove_cached_notification(notification_id=cache_id), - ) - self._signal_handlers[cache_id] = handler_id - self._cached_notifications[cache_id] = cached_notification - self._count += 1 + handler_id = cached_notification.connect( + "removed-from-cache", + lambda *args: self.remove_cached_notification( + notification_id=cache_id + ), + ) + self._signal_handlers[cache_id] = handler_id + self._cached_notifications[cache_id] = cached_notification + self._count += 1 + except Exception as e: + logger.warning( + f"Failed to load a cached notification (might be from an old schema): {e}" + ) # Set next cache ID to be higher than any existing ID self._next_cache_id = max_cache_id + 1