feat(src/utils): add EvdevLEDMonitor helper to monitor EV_LED state changes without polling

src/utils/utils.py

@@ -1,7 +1,9 @@
+import os
+import struct
 from pathlib import Path
 from typing import Literal
 
-from fabric.utils import Gdk, bulk_connect, exec_shell_command_async
+from fabric.utils import Gdk, GLib, bulk_connect, exec_shell_command_async, logger
 from fabric.widgets.svg import Svg
 
 
@@ -83,3 +85,81 @@ def toml_file(relative_path: str) -> str:
 def generate_colors_from_wallpaper(image_path: str) -> bool:
     cmd = f'matugen image "{image_path}" --source-color-index 0'
     return bool(exec_shell_command_async(cmd))
+
+
+class EvdevLEDMonitor:
+    """Helper to monitor EV_LED state changes without polling."""
+
+    def __init__(self, led_path: Path, target_code: int, on_state_changed_cb):
+        self.led_path = led_path
+        self.target_code = target_code
+        self.on_state_changed_cb = on_state_changed_cb
+        self._fd = None
+        self._watch_id = None
+        self._last_state = None
+
+    def _get_evdev_path(self) -> Path | None:
+        if not self.led_path:
+            return None
+        try:
+            device_path = self.led_path / "device"
+            for child in device_path.iterdir():
+                if child.name.startswith("event"):
+                    return Path("/dev/input") / child.name
+        except Exception:
+            pass
+        return None
+
+    def start(self):
+        evdev_path = self._get_evdev_path()
+        if not evdev_path or not evdev_path.exists():
+            logger.warning(f"Evdev path not found for {self.led_path}")
+            return
+
+        try:
+            self._fd = os.open(evdev_path, os.O_RDONLY | os.O_NONBLOCK)
+            self._watch_id = GLib.io_add_watch(
+                self._fd,
+                GLib.PRIORITY_DEFAULT,
+                GLib.IOCondition.IN,
+                self._on_evdev_data,
+            )
+        except PermissionError:
+            logger.warning(
+                f"Permission denied to read {evdev_path}. Add user to 'input' group."
+            )
+        except Exception as e:
+            logger.error(f"Failed to start evdev monitoring: {e}")
+
+    def _on_evdev_data(self, fd, condition):
+        try:
+            event_size = struct.calcsize("llHHi")
+            while True:
+                try:
+                    data = os.read(fd, event_size)
+                except BlockingIOError:
+                    break
+
+                if len(data) == event_size:
+                    _tv_sec, _tv_usec, type_, code, value = struct.unpack("llHHi", data)
+                    if type_ == 17 and code == self.target_code:
+                        current_state = bool(value)
+                        if current_state != self._last_state:
+                            self._last_state = current_state
+                            self.on_state_changed_cb(current_state)
+        except Exception as e:
+            logger.error(f"Evdev error: {e}")
+            self.stop()
+            return False
+        return True
+
+    def stop(self):
+        if self._watch_id is not None:
+            GLib.source_remove(self._watch_id)
+            self._watch_id = None
+        if self._fd is not None:
+            try:
+                os.close(self._fd)
+            except OSError:
+                pass
+            self._fd = None