refactor: improve brightness polling and caching via pathlib

This commit is contained in:
sankalp committed 2026-07-14 02:35:25 +05:45
1 parent 112f4b1039
commit e734c4bbf4
3 files changed
+67 -98

No files matched your search

+15 -41
View File
@@ -1,7 +1,9 @@
from pathlib import Path
from fabric.core.service import Property, Service, Signal
from fabric.utils import GLib, logger
from fabric.utils import logger
from utils.utils import EvdevLEDMonitor
# Discover CapsLock LED device
capslock_leds = list(Path("/sys/class/leds").glob("input*::capslock"))
@@ -26,7 +28,6 @@ class CapsLock(Service):
super().__init__(**kwargs)
self.capslock_led_path = capslock_device
self._timeout_id = None
self._last_state = None
self._monitoring_started = False
@@ -34,54 +35,27 @@ class CapsLock(Service):
logger.warning("CapsLock device not found, CapsLock service disabled")
return
# Don't access is_on property during init to avoid starting monitoring
# The last_state will be set when monitoring actually starts
# LED_CAPSL = 1
self._monitor = EvdevLEDMonitor(
self.capslock_led_path, 1, self._on_state_changed
)
def _start_led_monitoring(self):
"""Start efficient polling for CapsLock LED state changes."""
brightness_path = self.capslock_led_path / "brightness"
if not brightness_path.exists():
logger.warning(
f"CapsLock brightness file does not exist: {brightness_path}"
)
return
self._start_efficient_polling()
def _start_efficient_polling(self):
"""Start efficient polling with 500ms intervals."""
if self._timeout_id is None:
self._timeout_id = GLib.timeout_add(500, self._efficient_poll)
def _efficient_poll(self) -> bool:
"""Monitor CapsLock state with 500ms polling intervals.
Returns:
bool: True to continue polling, False to stop.
"""
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"CapsLock polling error: {e}")
self._timeout_id = None
return False # Stop polling on error
return True # Continue polling
def _on_state_changed(self, is_on: bool):
if is_on != self._last_state:
self._last_state = is_on
self.emit("state_changed", is_on)
def stop(self):
if self._timeout_id is not None:
GLib.source_remove(self._timeout_id)
self._timeout_id = None
if hasattr(self, "_monitor"):
self._monitor.stop()
def _ensure_monitoring_started(self):
if not self._monitoring_started and self.capslock_led_path:
self._monitoring_started = True
# Set initial state before starting monitoring
self._last_state = self._get_current_state()
self._start_led_monitoring()
self._monitor._last_state = self._last_state
self._monitor.start()
def _get_current_state(self) -> bool:
if not self.capslock_led_path: