diff --git a/src/services/brightness.py b/src/services/brightness.py index 3293373..a7c5a2d 100644 --- a/src/services/brightness.py +++ b/src/services/brightness.py @@ -1,4 +1,5 @@ import time +from pathlib import Path from fabric.core.service import Property, Service, Signal from fabric.utils import GLib, exec_shell_command_async, logger, os, re @@ -44,6 +45,9 @@ class Brightness(Service): self._last_percent = -1 self._last_raw = -1 self._last_update_time = 0 + self._screen_device = None + self._brightness_path = None + self._max_brightness_path = None self.backend = self._detect_backend(backend) @@ -58,15 +62,13 @@ class Brightness(Service): 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 - ) + if self._brightness_path and self._brightness_path.exists(): + self._last_raw = int(self._brightness_path.read_text().strip()) + self._last_percent = ( + int((self._last_raw / self.max_screen) * 100) + if self.max_screen > 0 + else 0 + ) self._poll_timer_id = GLib.timeout_add( self.POLL_INTERVAL, self._check_brightness_file @@ -77,10 +79,8 @@ class Brightness(Service): 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): - with open(file_path) as f: - raw = int(f.readline().strip()) + if self._brightness_path and self._brightness_path.exists(): + raw = int(self._brightness_path.read_text().strip()) if raw != self._last_raw: self._last_raw = raw @@ -128,11 +128,21 @@ class Brightness(Service): return None def _get_screen_device(self): - """Return first backlight device from sysfs.""" + """Return first backlight device from sysfs, cached.""" + if getattr(self, "_screen_device", None): + return self._screen_device + try: - return os.listdir("/sys/class/backlight")[0] + devices = os.listdir("/sys/class/backlight") + if devices: + self._screen_device = devices[0] + base_path = Path(f"/sys/class/backlight/{self._screen_device}") + self._brightness_path = base_path / "brightness" + self._max_brightness_path = base_path / "max_brightness" + return self._screen_device except Exception: - return "" + pass + return "" def _detect_ddcutil_bus(self): """Detect I2C bus number for ddcutil.""" @@ -173,10 +183,8 @@ class Brightness(Service): 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()) + if self._max_brightness_path and self._max_brightness_path.exists(): + return int(self._max_brightness_path.read_text().strip()) except Exception: return None return None @@ -198,15 +206,13 @@ class Brightness(Service): 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 + if self._brightness_path and self._brightness_path.exists(): + raw = int(self._brightness_path.read_text().strip()) + self._last_raw = raw + return raw except Exception as e: logger.error(f"Error reading brightness file: {e}") - return -1 + return -1 elif self.backend == "ddcutil": if ( time.time() - self._last_update_time < self.CACHE_INTERVAL @@ -300,7 +306,7 @@ class Brightness(Service): if self.backend == "brightnessctl": self.emit("screen", percent) exec_shell_command_async( - f"brightnessctl --device '{self._get_screen_device()}' set {raw}" + f"brightnessctl --device '{self._screen_device}' set {raw}" ) elif self.backend == "ddcutil": self._last_update_time = time.time() diff --git a/src/services/capslock.py b/src/services/capslock.py index f0788b5..da7f08e 100644 --- a/src/services/capslock.py +++ b/src/services/capslock.py @@ -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: diff --git a/src/services/numlock.py b/src/services/numlock.py index bf3c819..7434e9b 100644 --- a/src/services/numlock.py +++ b/src/services/numlock.py @@ -1,9 +1,11 @@ from pathlib import Path from fabric.core.service import Property, Service, Signal -from fabric.utils import GLib, logger +from fabric.utils import logger -# Discover NumLock LED device (same pattern as capslock) +from utils.utils import EvdevLEDMonitor + +# Discover NumLock LED device numlock_leds = list(Path("/sys/class/leds").glob("input*::numlock")) numlock_device = numlock_leds[0] if numlock_leds else None @@ -15,6 +17,7 @@ class NumLock(Service): def get_initial(): if NumLock.instance is None: NumLock.instance = NumLock() + return NumLock.instance @Signal @@ -25,7 +28,6 @@ class NumLock(Service): super().__init__(**kwargs) self.numlock_led_path = numlock_device - self._timeout_id = None self._last_state = None self._monitoring_started = False @@ -33,40 +35,27 @@ class NumLock(Service): 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() + # LED_NUML = 0 + self._monitor = EvdevLEDMonitor( + self.numlock_led_path, 0, self._on_state_changed + ) - 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 _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.numlock_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.numlock_led_path: