From 01dc64498f0db7035f3b810d87d269974a80353d Mon Sep 17 00:00:00 2001 From: S4NKALP Date: Thu, 9 Jul 2026 15:32:29 +0545 Subject: [PATCH] feat(src/services): refactor KeyboardLayout to use TOML state persistence and improve layout switching Switch layouts are now stored in TOML, added proper error handling and a static method to trigger switching, replaced JSON handling, updated imports and fixed related services. --- src/services/keyboard_layout.py | 179 ++++++++++++++++++-------------- src/services/modus.py | 7 +- src/services/screencapture.py | 13 ++- 3 files changed, 115 insertions(+), 84 deletions(-) diff --git a/src/services/keyboard_layout.py b/src/services/keyboard_layout.py index e642587..efafa53 100644 --- a/src/services/keyboard_layout.py +++ b/src/services/keyboard_layout.py @@ -1,14 +1,12 @@ -import json -import socket -import threading from pathlib import Path +import tomlkit from fabric.core.service import Property, Service, Signal -from fabric.utils import GLib, logger, os +from fabric.utils import Gio, GLib, logger import shared.data as data from services.config import on_config_change -from utils.functions import read_json_file, run_command, write_json_file +from utils.functions import run_command HYPRCTL_BIN = "hyprctl" @@ -39,116 +37,139 @@ class KeyboardLayout(Service): super().__init__(**kwargs) self._initialized = True - self.layout_json_file = Path(data.CACHE_DIR) / "kb_layout.json" + self.layout_state_file = Path(data.CACHE_DIR) / "kb_layout.toml" + self.layout_file = Path(data.CACHE_DIR) / "kb_layout_current.txt" self._last_layout = None self.layouts = [] self.current_index = 0 self._init_layout_config() - self._start_event_listener() + self._start_file_monitor() on_config_change(self._on_config_change) def _init_layout_config(self): config_layouts = data.load_config().get("keyboard_layouts", ["us", "np"]) - json_data = read_json_file(self.layout_json_file) + state = self._read_state() - if json_data: - self.layouts = json_data.get("layouts", config_layouts) - self.current_index = json_data.get("current_index", 0) + if state: + self.layouts = state.get("layouts", config_layouts) + self.current_index = state.get("current_index", 0) else: self.layouts = config_layouts self.current_index = 0 - self._save_layout_json() + self._save_state() - # Sync with actual Hyprland state - self._sync_with_hyprland() + self._apply_layouts_to_hyprland() + self._apply_current_layout_index() - def _sync_with_hyprland(self): + def _read_state(self): + if not self.layout_state_file.exists(): + return None try: - result = run_command([HYPRCTL_BIN, "devices", "-j"]) - devices = json.loads(result.stdout) - keyboards = devices.get("keyboards", []) - for k in keyboards: - if k.get("main"): - layout_name = k.get("active_keymap") - if layout_name: - self._last_layout = layout_name - # Try to match with our list to update index - for i, layout_item in enumerate(self.layouts): - if layout_item.lower() in layout_name.lower(): - self.current_index = i - break - break + with open(self.layout_state_file, "r") as f: + data = tomlkit.load(f) + return dict(data) if data else None except Exception as e: - logger.error(f"[KeyboardLayout] Sync error: {e}") + logger.error(f"[KeyboardLayout] Failed to read state: {e}") + return None - def _save_layout_json(self): - write_json_file( - { - "layouts": self.layouts, - "current_index": self.current_index, - }, - self.layout_json_file, - ) + def _save_state(self): + try: + self.layout_state_file.parent.mkdir(parents=True, exist_ok=True) + doc = tomlkit.document() + arr = tomlkit.array() + for layout in self.layouts: + arr.append(layout) + doc["layouts"] = arr + doc["current_index"] = self.current_index + with open(self.layout_state_file, "w") as f: + tomlkit.dump(doc, f) + except Exception as e: + logger.error(f"[KeyboardLayout] Failed to save state: {e}") def _on_config_change(self, new_config, old_config): new_layouts = new_config.get("keyboard_layouts") if new_layouts and new_layouts != self.layouts: self.layouts = new_layouts - self._save_layout_json() + self._save_state() + self._apply_layouts_to_hyprland() logger.info(f"[KeyboardLayout] Layouts updated from config: {self.layouts}") - def _start_event_listener(self): - """Start a thread to listen for Hyprland layout events.""" - thread = threading.Thread(target=self._event_loop, daemon=True) - thread.start() - - def _event_loop(self): - his = os.getenv("HYPRLAND_INSTANCE_SIGNATURE") - xdg_runtime_dir = os.getenv("XDG_RUNTIME_DIR", f"/run/user/{os.getuid()}") - socket_path = f"{xdg_runtime_dir}/hypr/{his}/.socket2.sock" - - if not os.path.exists(socket_path): - logger.error(f"[KeyboardLayout] Hyprland socket not found: {socket_path}") + def _start_file_monitor(self): + if not self.layout_state_file.exists(): return - try: - with socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) as s: - s.connect(socket_path) - logger.info("[KeyboardLayout] Connected to Hyprland event socket.") - while True: - data = s.recv(1024).decode("utf-8") - if not data: - break - for line in data.split("\n"): - if line.startswith("activelayout>>"): - # Format: activelayout>>keyboardname,layoutname - parts = line.split(">>")[1].split(",") - if len(parts) >= 2: - layout_name = parts[1] - if layout_name != self._last_layout: - self._last_layout = layout_name - GLib.idle_add( - self.emit, "layout_changed", layout_name - ) + gio_file = Gio.File.new_for_path(str(self.layout_state_file)) + self._file_monitor = gio_file.monitor_file(Gio.FileMonitorFlags.NONE, None) + self._file_monitor.connect("changed", self._on_file_changed) except Exception as e: - logger.error(f"[KeyboardLayout] Event listener error: {e}") + logger.error(f"[KeyboardLayout] Failed to start file monitor: {e}") - def switch_to_next(self) -> bool: + def _on_file_changed(self, monitor, file, *args): + GLib.timeout_add(50, self._reload_layout) + + def _reload_layout(self): + new_layout = self._read_layout() + if new_layout != self._last_layout: + self._last_layout = new_layout + self.emit("layout_changed", new_layout) + return False + + def _read_layout(self): + state = self._read_state() + if state: + layouts = state.get("layouts", []) + current_index = state.get("current_index", 0) + if layouts and 0 <= current_index < len(layouts): + return layouts[current_index] + return "us" + + def _write_layout(self, layout: str): + try: + self.layout_file.parent.mkdir(parents=True, exist_ok=True) + self.layout_file.write_text(layout, encoding="utf-8") + except Exception as e: + logger.error(f"[KeyboardLayout] Failed writing layout state file: {e}") + + def _apply_layouts_to_hyprland(self): if not self.layouts: + return + layouts_str = ",".join(self.layouts) + lua_eval_cmd = f"hl.config({{ input = {{ kb_layout = '{layouts_str}' }} }})" + run_command( + [HYPRCTL_BIN, "eval", lua_eval_cmd], + timeout=2, + ) + + def _apply_current_layout_index(self): + run_command( + [HYPRCTL_BIN, "switchxkblayout", "all", str(self.current_index)], + timeout=2, + ) + + def switch_to_next(self): + if not self.layouts: + logger.warning("[KeyboardLayout] No layouts configured, cannot cycle.") return False self.current_index = (self.current_index + 1) % len(self.layouts) - self.layouts[self.current_index] + new_layout = self.layouts[self.current_index] - # Use hyprctl to switch layout for all devices - # We assume the layouts in Hyprland config match self.layouts in order - run_command([HYPRCTL_BIN, "switchxkblayout", "all", "next"]) + self._apply_current_layout_index() + self._save_state() + self._write_layout(new_layout) - self._save_layout_json() + logger.info( + f"[KeyboardLayout] Switched layout successfully to '{new_layout}' " + f"(index {self.current_index})" + ) return True + @staticmethod + def switch_keyboard_layout(): + KeyboardLayout.get_initial().switch_to_next() + @Property(str, "readable") - def current_layout(self) -> str: - return self._last_layout or "Unknown" + def current_layout(self): + return self._read_layout() diff --git a/src/services/modus.py b/src/services/modus.py index 54649c8..6dbe882 100644 --- a/src/services/modus.py +++ b/src/services/modus.py @@ -580,9 +580,10 @@ def get_screen_dimensions() -> tuple[int, int]: def focus_window(address: str): - conn = _ensure_hyprland() - if conn: - conn.send_command(f"dispatch focuswindow address:{address}") + run_command( + ["hyprctl", "dispatch", f'hl.dsp.focus({{ window = "address:{address}" }})'], + timeout=2, + ) def close_window(address: str): diff --git a/src/services/screencapture.py b/src/services/screencapture.py index 6fa98e5..9f2fd89 100644 --- a/src/services/screencapture.py +++ b/src/services/screencapture.py @@ -64,7 +64,7 @@ class ScreenCapture(Service): cmd.extend(["-A", f"{action}={action}"]) cmd.extend([title, message]) - return run_command(cmd) + run_command(cmd, timeout=5) def send_screenshot_notification(self, file_path=None): cmd = ["notify-send"] @@ -289,6 +289,7 @@ class ScreenCapture(Service): self.screenshot_taken(None) except Exception as e: logger.error(f"Screenshot notification failed: {e}") + self.screenshot_taken(None) def record(self, target="selection", use_audio=False, show_cursor=False): """ @@ -325,7 +326,15 @@ class ScreenCapture(Service): cmd.append("--show-cursor") if target == "selection": - geometry = exec_shell_command("slurp") + try: + geometry = exec_shell_command("slurp") + except Exception: + self.notify_send( + "Recording cancelled", + "Selection was cancelled", + icon="camera-video-symbolic", + ) + return False if not geometry or not str(geometry).strip(): self.notify_send( "Recording cancelled",