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.
This commit is contained in:
sankalp committed 2026-07-05 23:04:34 +05:45
1 parent f0c67c0483
commit 754cde0278
3 files changed
+106 -8

No files matched your search

+1
View File
@@ -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
+20 -8
View File
@@ -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."""
+85
View File
@@ -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()