From 754cde02782c54882a3a8560f6314db267338a2c Mon Sep 17 00:00:00 2001 From: S4NKALP Date: Sun, 5 Jul 2026 23:04:34 +0545 Subject: [PATCH] feat(src/services): add NumLock service and refactor Battery documentation Add new NumLock service to monitor NumLock LED state changes. Refactor Battery service documentation and property cache handling. Update imports to include NumLock. --- src/services/__init__.py | 1 + src/services/battery.py | 28 +++++++++---- src/services/numlock.py | 85 ++++++++++++++++++++++++++++++++++++++++ 3 files changed, 106 insertions(+), 8 deletions(-) create mode 100644 src/services/numlock.py diff --git a/src/services/__init__.py b/src/services/__init__.py index f601b32..b64ed5e 100644 --- a/src/services/__init__.py +++ b/src/services/__init__.py @@ -2,6 +2,7 @@ from fabric.audio import Audio # noqa: F401 from .battery import Battery # noqa: F401 from .brightness import Brightness # noqa: F401 from .capslock import CapsLock # noqa: F401 +from .numlock import NumLock # noqa: F401 from .keyboard_layout import KeyboardLayout # noqa: F401 from .network import NetworkClient, Wifi, Ethernet # noqa: F401 from .mpris import PlayerService, PlayerManager # noqa: F401 diff --git a/src/services/battery.py b/src/services/battery.py index 6d52ead..080285a 100644 --- a/src/services/battery.py +++ b/src/services/battery.py @@ -21,7 +21,18 @@ DEVICE_STATE = { class Battery(Service): - """A service for interacting with the battery's DBus and Power Profiles""" + """A service for interacting with the battery's DBus and Power Profiles. + + Use Battery.get_initial() to get the shared singleton instance. + """ + + _instance = None + + @classmethod + def get_initial(cls) -> "Battery": + if cls._instance is None: + cls._instance = cls() + return cls._instance @Signal def changed(self) -> None: ... @@ -95,7 +106,9 @@ class Battery(Service): self._bus: Gio.DBusConnection | None = None self._proxy = None self._power_profile_proxy = None - self._prop_cache: dict[str, any] = {} + # _prop_cache stores ONLY properties received via PropertiesChanged. + # For fresh reads we always fall through to the GDBus proxy cache. + self._prop_cache: dict[str, object] = {} self.do_register() def do_register(self) -> None: @@ -190,13 +203,12 @@ class Battery(Service): return result.unpack() def do_get_cached_property(self, property_name): - if property_name in self._prop_cache: - return self._prop_cache[property_name] + """Read from the GDBus proxy cache, which DBus keeps up-to-date.""" result = self._proxy.get_cached_property(property_name) - value = result.unpack() if result is not None else None - if value is not None: - self._prop_cache[property_name] = value - return value + if result is not None: + return result.unpack() + # Fallback: value not in proxy cache yet (e.g. very early in startup) + return self._prop_cache.get(property_name) def get_power_profile(self) -> str | None: """Get the current active power profile.""" diff --git a/src/services/numlock.py b/src/services/numlock.py new file mode 100644 index 0000000..bf3c819 --- /dev/null +++ b/src/services/numlock.py @@ -0,0 +1,85 @@ +from pathlib import Path + +from fabric.core.service import Property, Service, Signal +from fabric.utils import GLib, logger + +# Discover NumLock LED device (same pattern as capslock) +numlock_leds = list(Path("/sys/class/leds").glob("input*::numlock")) +numlock_device = numlock_leds[0] if numlock_leds else None + + +class NumLock(Service): + instance = None + + @staticmethod + def get_initial(): + if NumLock.instance is None: + NumLock.instance = NumLock() + return NumLock.instance + + @Signal + def state_changed(self, is_on: bool) -> None: + """Signal emitted when NumLock state changes.""" + + def __init__(self, **kwargs): + super().__init__(**kwargs) + + self.numlock_led_path = numlock_device + self._timeout_id = None + self._last_state = None + self._monitoring_started = False + + if numlock_device is None: + logger.warning("NumLock device not found, NumLock service disabled") + return + + def _start_led_monitoring(self): + """Start efficient polling for NumLock LED state changes.""" + brightness_path = self.numlock_led_path / "brightness" + if not brightness_path.exists(): + logger.warning(f"NumLock brightness file does not exist: {brightness_path}") + return + self._start_efficient_polling() + + def _start_efficient_polling(self): + if self._timeout_id is None: + self._timeout_id = GLib.timeout_add(500, self._efficient_poll) + + def _efficient_poll(self) -> bool: + try: + current_state = self._get_current_state() + if current_state != self._last_state: + self._last_state = current_state + self.emit("state_changed", current_state) + except Exception as e: + logger.error(f"NumLock polling error: {e}") + self._timeout_id = None + return False + return True + + def stop(self): + if self._timeout_id is not None: + GLib.source_remove(self._timeout_id) + self._timeout_id = None + + def _ensure_monitoring_started(self): + if not self._monitoring_started and self.numlock_led_path: + self._monitoring_started = True + self._last_state = self._get_current_state() + self._start_led_monitoring() + + def _get_current_state(self) -> bool: + if not self.numlock_led_path: + return False + brightness_path = self.numlock_led_path / "brightness" + if brightness_path.exists(): + try: + return bool(int(brightness_path.read_text().strip())) + except (ValueError, OSError): + return False + return False + + @Property(bool, "read-write", default_value=False) + def is_on(self) -> bool: + self._ensure_monitoring_started() + return self._get_current_state()