feat(root): add keyboard layouts, OSDWindow, and update services & utilities

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
This commit is contained in:
sankalp committed 2026-03-23 23:33:53 +05:45
1 parent 129906cb9c
commit 34cfb2f49e
12 files changed
+504 -96

No files matched your search

+6 -1
View File
@@ -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
}
+2 -2
View File
@@ -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())
+12 -17
View File
@@ -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
+10 -4
View File
@@ -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
+18 -3
View File
@@ -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, *_):
+100
View File
@@ -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()
+154
View File
@@ -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"
+9 -2
View File
@@ -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
+142 -59
View File
@@ -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):
+47 -4
View File
@@ -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))
+3 -3
View File
@@ -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):
+1 -1
View File
@@ -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):