From 34cfb2f49e11a1706707011c70f1193c163789ee Mon Sep 17 00:00:00 2001 From: S4NKALP Date: Mon, 23 Mar 2026 23:33:53 +0545 Subject: [PATCH] feat(root): add keyboard layouts, OSDWindow, and update services & utilities MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Introduce a new OSDWindow implementation, expose keyboard layout configuration, and modernize dependency handling and helper utilities. - **New Features** - Added `keyboard_layouts` list (`us`, `np`) to `config.json` - Created `OSDWindow` class and updated `main.py` to use it - Introduced new service modules (`battery`, `brightness`, `capslock`, `network`, `mpris`, `todo`, `modus`) and expanded `services/__init__.py` - Added `CommandResult` and `run_command` helper in `utils/functions.py` - **Enhancements** - Upgraded numerous dependencies to newer versions and switched `fabric` to a git source - Replaced bare `except:` with `except Exception:` in `WiFiPasswordDialog` - Improved type hints (`NamedTuple`, `TypeVar`) and refactored JSON file helpers - Added property notifications for battery percentage, state, and presence - **Bug Fixes** - Fixed incorrect return type annotation in `read_json_file` - Corrected tick‑handler logic to avoid double registration - Properly notify listeners on battery property changes - Ensured proper exception handling and cleanup in utility functions --- config.json | 7 +- main.py | 4 +- requirements.txt | 29 ++--- services/__init__.py | 14 ++- services/battery.py | 21 +++- services/capslock.py | 100 ++++++++++++++++ services/keyboard_layout.py | 154 ++++++++++++++++++++++++ services/todo.py | 11 +- utils/animator.py | 201 ++++++++++++++++++++++---------- utils/functions.py | 51 +++++++- utils/icon_resolver.py | 6 +- widgets/wifi_password_dialog.py | 2 +- 12 files changed, 504 insertions(+), 96 deletions(-) create mode 100644 services/capslock.py create mode 100644 services/keyboard_layout.py diff --git a/config.json b/config.json index 872860b..8ce87ad 100644 --- a/config.json +++ b/config.json @@ -23,7 +23,12 @@ "workspace_indicator": true, "bluetooth": true, "date_time": true, + "keyboard_layouts": [ + "us", + "np" + ], "dock": true, - "window_switcher": true + "window_switcher": true, + "osd": true } diff --git a/main.py b/main.py index 990868f..cf16cec 100644 --- a/main.py +++ b/main.py @@ -5,7 +5,7 @@ from config.data import APP_NAME from modules.desktop.widget import Deskwidgets from modules.launcher.main import Launcher from modules.notification.notification import ModusNoti -from modules.osd import OSD +from modules.osd.main import OSDWindow from modules.panel.main import Panel from modules.switcher import ApplicationSwitcher from utils.functions import set_process_name @@ -32,7 +32,7 @@ if __name__ == "__main__": modusnoti = ModusNoti() launcher = Launcher() deskwidget = Deskwidgets() - osd = OSD() + osd = OSDWindow() # Monitor CSS files for changes css_file = monitor_file(get_relative_path("styles")) _ = css_file.connect("changed", lambda *_: set_css()) diff --git a/requirements.txt b/requirements.txt index 276d0c2..f126fee 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,17 +1,12 @@ -certifi==2025.8.3 -charset-normalizer==3.4.2 -click==8.2.1 -idna==3.10 -loguru==0.7.3 -pillow==11.3.0 -psutil==7.0.0 -pycairo==1.28.0 -pydbus==0.6.0 -PyGObject==3.52.3 -pyotp==2.9.0 -pyzbar==0.1.9 -RapidFuzz==3.13.0 -requests==2.32.4 -setproctitle==1.3.6 -thefuzz==0.22.1 -urllib3==2.6.3 +fabric @ git+https://github.com/Fabric-Development/fabric.git@fd2aabbd7e1859aa7c11c626a6c36a937aca736a +ijson>=3.5.0 +pam>=0.2.0 +pillow>=12.1.1 +psutil>=7.2.2 +pycairo>=1.29.0 +pydbus>=0.6.0 +pyotp>=2.9.0 +pyzbar>=0.1.9 +requests>=2.32.5 +six>=1.17.0 +thefuzz>=0.22.1 diff --git a/services/__init__.py b/services/__init__.py index 3f03f12..6999ed4 100644 --- a/services/__init__.py +++ b/services/__init__.py @@ -1,4 +1,10 @@ -""" -Modus services package. -Contains background services and utilities for the shell. -""" +from fabric.audio import Audio # noqa: F401 +from .battery import BatteryService # noqa: F401 +from .brightness import Brightness # noqa: F401 +from .capslock import CapsLock # noqa: F401 +from .keyboard_layout import KeyboardLayout # noqa: F401 +from .network import NetworkClient, Wifi, Ethernet, AccessPoint # noqa: F401 +from .mpris import MprisPlayer, MprisPlayerManager # noqa: F401 +from .todo import TodoService # noqa: F401 +from .modus import ModusService # noqa: F401 +from .config import on_config_change, start_config_service, ConfigServic # noqa: F401 diff --git a/services/battery.py b/services/battery.py index 92200a6..554c8c7 100644 --- a/services/battery.py +++ b/services/battery.py @@ -1,6 +1,6 @@ from typing import Literal, Optional -from fabric import Service, Signal +from fabric.core.service import Property, Service, Signal from fabric.utils import Gio, GLib, logger from utils.dbus_helper import GioDBusHelper @@ -178,8 +178,23 @@ class BatteryService(Service): """Get display name for a power profile.""" return PowerProfile.get(profile, profile.title()) - def handle_property_change(self, *_): - # You may filter which property changed by checking parameters[1] + @Property(int, "readable", default_value=0) + def percentage(self) -> int: + return self.get_property("Percentage") or 0 + + @Property(int, "readable", default_value=0) + def state(self) -> int: + return self.get_property("State") or 0 + + @Property(bool, "readable", default_value=False) + def is_present(self) -> bool: + return self.get_property("IsPresent") or False + + def handle_property_change(self, *args): + # Notify about property changes for OSD and other listeners + self.notify("percentage") + self.notify("state") + self.notify("is-present") self.emit("changed") def handle_power_profile_change(self, *_): diff --git a/services/capslock.py b/services/capslock.py new file mode 100644 index 0000000..931c420 --- /dev/null +++ b/services/capslock.py @@ -0,0 +1,100 @@ +from pathlib import Path + +from fabric.core.service import Property, Service, Signal +from fabric.utils import GLib, logger + +# Discover CapsLock LED device +capslock_leds = list(Path("/sys/class/leds").glob("input*::capslock")) +capslock_device = capslock_leds[0] if capslock_leds else None + + +class CapsLock(Service): + instance = None + + @staticmethod + def get_initial(): + if CapsLock.instance is None: + CapsLock.instance = CapsLock() + + return CapsLock.instance + + @Signal + def state_changed(self, is_on: bool) -> None: + """Signal emitted when CapsLock state changes.""" + + def __init__(self, **kwargs): + super().__init__(**kwargs) + + self.capslock_led_path = capslock_device + self._timeout_id = None + self._last_state = None + self._monitoring_started = False + + if capslock_device is None: + 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 + + 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 100ms intervals.""" + if self._timeout_id is None: + self._timeout_id = GLib.timeout_add(100, self._efficient_poll) + + def _efficient_poll(self) -> bool: + """Monitor CapsLock state with 100ms 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 stop(self): + if self._timeout_id is not None: + GLib.source_remove(self._timeout_id) + self._timeout_id = None + + 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() + + def _get_current_state(self) -> bool: + if not self.capslock_led_path: + return False + brightness_path = self.capslock_led_path / "brightness" + if brightness_path.exists(): + try: + return bool(int(brightness_path.read_text().strip())) + except (ValueError, OSError): + return False + return False + + @Property(bool, "read-write", default_value=False) + def is_on(self) -> bool: + self._ensure_monitoring_started() + return self._get_current_state() diff --git a/services/keyboard_layout.py b/services/keyboard_layout.py new file mode 100644 index 0000000..70aa140 --- /dev/null +++ b/services/keyboard_layout.py @@ -0,0 +1,154 @@ +import json +import socket +import threading +from pathlib import Path + +from fabric.core.service import Property, Service, Signal +from fabric.utils import GLib, logger, os + +import config.data as data +from services.config import on_config_change +from utils.functions import read_json_file, run_command, write_json_file + +HYPRCTL_BIN = "hyprctl" + + +class KeyboardLayout(Service): + """Service to manage keyboard layout switching and monitoring via Hyprland.""" + + _instance = None + + def __new__(cls): + if cls._instance is None: + cls._instance = super().__new__(cls) + return cls._instance + + @staticmethod + def get_initial(): + if KeyboardLayout._instance is None: + KeyboardLayout._instance = KeyboardLayout() + return KeyboardLayout._instance + + @Signal + def layout_changed(self, layout: str) -> None: + """Signal emitted when keyboard layout changes.""" + + def __init__(self, **kwargs): + if getattr(self, "_initialized", False): + return + super().__init__(**kwargs) + self._initialized = True + + self.layout_json_file = Path(data.CACHE_DIR) / "kb_layout.json" + self._last_layout = None + self.layouts = [] + self.current_index = 0 + + self._init_layout_config() + self._start_event_listener() + + 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) + + if json_data: + self.layouts = json_data.get("layouts", config_layouts) + self.current_index = json_data.get("current_index", 0) + else: + self.layouts = config_layouts + self.current_index = 0 + self._save_layout_json() + + # Sync with actual Hyprland state + self._sync_with_hyprland() + + def _sync_with_hyprland(self): + try: + result = run_command([HYPRCTL_BIN, "devices", "-j"]) + devices = json.loads(result) + 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 + except Exception as e: + logger.error(f"[KeyboardLayout] Sync error: {e}") + + def _save_layout_json(self): + write_json_file( + { + "layouts": self.layouts, + "current_index": self.current_index, + }, + self.layout_json_file, + ) + + 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() + 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}") + 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 + ) + except Exception as e: + logger.error(f"[KeyboardLayout] Event listener error: {e}") + + def switch_to_next(self) -> bool: + if not self.layouts: + return False + + self.current_index = (self.current_index + 1) % len(self.layouts) + 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._save_layout_json() + return True + + @Property(str, "readable") + def current_layout(self) -> str: + return self._last_layout or "Unknown" diff --git a/services/todo.py b/services/todo.py index 233a23b..c440381 100644 --- a/services/todo.py +++ b/services/todo.py @@ -164,5 +164,12 @@ class TodoService(Service): } -# Global service instance -todo_service = TodoService() +# Global service instance getter +_todo_service = None + + +def get_todo_service() -> TodoService: + global _todo_service + if _todo_service is None: + _todo_service = TodoService() + return _todo_service diff --git a/utils/animator.py b/utils/animator.py index 7dd063d..96b8a94 100644 --- a/utils/animator.py +++ b/utils/animator.py @@ -1,21 +1,108 @@ -from typing import cast +# Author: Yousef EL-Darsh +# License (SPDX): AGPL-3.0-or-later -from gi.repository import GLib, Gtk +from functools import cache +from typing import Protocol, cast -from fabric import Property, Service, Signal +from fabric.core.service import Property, Service, Signal +from fabric.utils import GLib, Gtk, clamp + + +@cache +def lerp(start: float, end: float, progress: float) -> float: + return start + (end - start) * progress + + +@cache +def steps(n: int, progress: float, start_jump: bool = False) -> float: + if start_jump: + return min(int(progress * n), n - 1) / (n - 1) if n > 1 else 0.0 + return min(int(progress * n + 1e-10), n) / n + + +@cache +def cubic_bezier( + x1: float, y1: float, x2: float, y2: float, progress: float, epsilon=1e-6 +) -> float: + # implementation yanked off of the internet, don't blame me about anything. + # Fast-path boundaries to avoid overshoot and unnecessary work + if progress <= 0.0 or progress >= 1.0: + return clamp(progress, 0.0, 1.0) + + t_guess = progress + for _ in range(8): + t = t_guess + t_sq = t * t + omt = 1.0 - t + omt_sq = omt * omt + + x = 3 * x1 * omt_sq * t + 3 * x2 * omt * t_sq + t * t_sq + dx = 3 * x1 * omt_sq + 6 * (x2 - x1) * omt * t + 3 * (1 - x2) * t_sq + + if abs(dx) < epsilon: + break + + delta = (x - progress) / dx + t_guess -= delta + t_guess = clamp(t_guess, 0.0, 1.0) + + if abs(delta) < epsilon: + break + + t = clamp(t_guess, 0.0, 1.0) + t_sq = t * t + omt = 1.0 - t + return 3 * y1 * omt * omt * t + 3 * y2 * omt * t_sq + t * t_sq + + +def ease_linear(progress: float) -> float: + return cubic_bezier(1, 1, 0, 0, progress) + + +def ease_in(progress: float) -> float: + return cubic_bezier(0.4, 0, 1, 1, progress) + + +def ease_out(progress: float) -> float: + return cubic_bezier(0, 0, 0.2, 1, progress) + + +def ease_in_out(progress: float) -> float: + return cubic_bezier(0.4, 0, 0.2, 1, progress) + + +class TimingFunctionCallback(Protocol): + def __call__(self, progress: float, *args, **kwargs) -> float: ... class Animator(Service): + """ + An animator is a simple way for animating a value on + a set timeline based on a given timing function + """ + @Signal def finished(self) -> None: ... - @Property(tuple[float, float, float, float], "read-write") - def bezier_curve(self) -> tuple[float, float, float, float]: - return self._bezier_curve + @Property(TimingFunctionCallback, "read-write") + def timing_function(self) -> TimingFunctionCallback: + return self._timing_function - @bezier_curve.setter - def bezier_curve(self, value: tuple[float, float, float, float]): - self._bezier_curve = value + @timing_function.setter + def timing_function(self, value: TimingFunctionCallback): + self._timing_function = value + return + + @Property(float, "read-write") + def duration(self): + return self._duration + + @duration.setter + def duration(self, value: float): + if value <= 0.0: + raise ValueError("duration can't be smaller than or equal to 0.0") + + self._duration = value return @Property(float, "read-write") @@ -50,7 +137,7 @@ class Animator(Service): return self._playing @playing.setter - def playing(self, value: bool): + def playing(self, value: bool): # this setter is intended for internal usage only self._playing = value return @@ -65,76 +152,68 @@ class Animator(Service): def __init__( self, - bezier_curve: tuple[float, float, float, float], - duration: float, + duration: float = 0.8, + timing_function: TimingFunctionCallback = ease_linear, + value: float = 0.0, min_value: float = 0.0, max_value: float = 1.0, repeat: bool = False, tick_widget: Gtk.Widget | None = None, + tick_interval: int = 16, **kwargs, ): super().__init__(**kwargs) - self._bezier_curve = (1, 0, 1, 1) - self._duration = 5 - self._value = 0.0 + self._playing = False + self._value = value self._min_value = 0.0 self._max_value = 1.0 self._repeat = False + self._duration = 0.8 + self._timing_function = timing_function + self._tick_widget = tick_widget + self._tick_interval = tick_interval - self.bezier_curve = bezier_curve + self.timing_function = timing_function + self.repeat = repeat self.duration = duration - self.value = min_value self.min_value = min_value self.max_value = max_value - self.repeat = repeat - + self.value = value self.playing = False + self._start_time = None self._tick_handler = None - self._timeline_pos = 0 - self._tick_widget = tick_widget + self._timeline_pos = 0.0 def do_get_time_now(self): return GLib.get_monotonic_time() / 1_000_000 - def do_lerp(self, start: float, end: float, time: float) -> float: - return start + (end - start) * time - - def do_interpolate_cubic_bezier(self, time: float) -> float: - y_points = (0, self.bezier_curve[1], self.bezier_curve[3], 1) - return ( - (1 - time) ** 3 * y_points[0] - + 3 * (1 - time) ** 2 * time * y_points[1] - + 3 * (1 - time) * time**2 * y_points[2] - + time**3 * y_points[3] - ) - - def do_ease(self, time: float) -> float: - return self.do_lerp( - self.min_value, self.max_value, self.do_interpolate_cubic_bezier(time) - ) - def do_update_value(self, delta_time: float): - if not self.playing: + if not self._playing: return elapsed_time = delta_time - cast(float, self._start_time) - self._timeline_pos = min(1, elapsed_time / self.duration) + self._timeline_pos = min(1.0, elapsed_time / self._duration) - self.value = self.do_ease(self._timeline_pos) + self.value = lerp( + self._min_value, + self._max_value, + self._timing_function(progress=self._timeline_pos), + ) - if not self._timeline_pos >= 1: + if not self._timeline_pos >= 1.0: return - if not self.repeat: - self.value = self.max_value + if not self._repeat: + # all done.. + self.value = self._max_value self.finished() self.pause() return self._start_time = delta_time - self._timeline_pos = 0 + self._timeline_pos = 0.0 return def do_handle_tick(self, *_): @@ -143,29 +222,33 @@ class Animator(Service): return True def do_remove_tick_handlers(self): - if self._tick_handler: - if self._tick_widget: - self._tick_widget.remove_tick_callback(self._tick_handler) - else: - GLib.source_remove(self._tick_handler) + if not self._tick_handler: + return + + if self._tick_widget: + self._tick_widget.remove_tick_callback(self._tick_handler) + else: + GLib.source_remove(self._tick_handler) self._tick_handler = None return def play(self): - if self.playing: + if self._playing: return + self.playing = True self._start_time = self.do_get_time_now() - if not self._tick_handler: - if self._tick_widget: - self._tick_handler = self._tick_widget.add_tick_callback( - self.do_handle_tick - ) - else: - self._tick_handler = GLib.timeout_add(16, self.do_handle_tick) + if self._tick_handler: + return - self.playing = True + if self._tick_widget: + self._tick_handler = self._tick_widget.add_tick_callback( + self.do_handle_tick + ) + return + + self._tick_handler = GLib.timeout_add(self._tick_interval, self.do_handle_tick) return def pause(self): diff --git a/utils/functions.py b/utils/functions.py index c322fea..877ca0c 100644 --- a/utils/functions.py +++ b/utils/functions.py @@ -4,9 +4,15 @@ import json import os import subprocess import threading -from typing import Dict, List, Optional +from typing import Dict, NamedTuple, Optional, TypeVar -from fabric.utils import exec_shell_command, exec_shell_command_async, logger +from fabric.utils import ( + exec_shell_command, + exec_shell_command_async, + logger, +) + +T = TypeVar("T") def set_process_name(name: str): @@ -66,7 +72,6 @@ def run_in_thread(func): return wrapper -# @run_in_thread def write_json_file(data: Dict, path: str): try: with open(path, "w") as f: @@ -75,7 +80,7 @@ def write_json_file(data: Dict, path: str): logger.warning(f"Failed to write json: {e}") -def read_json_file(file_path: str) -> Optional[List]: +def read_json_file(file_path: str) -> Optional[Dict]: if not os.path.exists(file_path): logger.error(f"JSON file {file_path} does not exist.") return None @@ -253,3 +258,41 @@ def clear_children(container) -> None: child.destroy() except Exception as e: logger.warning(f"clear_children failed: {e}") + + +class CommandResult(NamedTuple): + returncode: int + stdout: bytes | str + stderr: bytes | str + + +def run_command( + args: list[str], + timeout: float | None = None, + cwd: Optional[str] = None, + env: Optional[Dict[str, str]] = None, + *, + input: bytes | str | None = None, + text: bool = True, +) -> CommandResult: + try: + result = subprocess.run( + args, + capture_output=True, + text=text, + timeout=timeout, + cwd=cwd, + env=env, + input=input, + ) + return CommandResult(result.returncode, result.stdout, result.stderr) + except subprocess.TimeoutExpired as e: + return CommandResult( + -1, + (e.stdout or b"" if not text else e.stdout or ""), + (e.stderr or (b"timeout" if not text else "timeout")), + ) + except FileNotFoundError as e: + return CommandResult(127, "", str(e)) + except Exception as e: + return CommandResult(1, "", str(e)) diff --git a/utils/icon_resolver.py b/utils/icon_resolver.py index 58ac1e4..5f0f7dc 100644 --- a/utils/icon_resolver.py +++ b/utils/icon_resolver.py @@ -5,10 +5,10 @@ import re import gi gi.require_version("Gtk", "3.0") -from gi.repository import GLib, Gtk -from loguru import logger +from gi.repository import GLib, Gtk # noqa: E402 +from loguru import logger # noqa: E402 -import config.data as data +import config.data as data # noqa: E402 ICON_CACHE_FILE = data.CACHE_DIR + "/icons.json" if not os.path.exists(data.CACHE_DIR): diff --git a/widgets/wifi_password_dialog.py b/widgets/wifi_password_dialog.py index 9d53586..3b33577 100644 --- a/widgets/wifi_password_dialog.py +++ b/widgets/wifi_password_dialog.py @@ -280,7 +280,7 @@ class WiFiPasswordDialog(Window): self.password_entry.grab_focus() self.password_entry.select_region(0, -1) return False - except: + except Exception: return False def get_password(self):