feat(src/services): rename BatteryService to Battery and add availability properties

Rename service to Battery, expose availability, charging, discharging properties and simplify interface for clarity

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

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

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

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