diff --git a/src/utils/utils.py b/src/utils/utils.py index d0cd9f9..01b2434 100644 --- a/src/utils/utils.py +++ b/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