From 6a5f85a10d82cb97338d419ff9fe127388af0062 Mon Sep 17 00:00:00 2001 From: S4NKALP Date: Fri, 10 Jul 2026 10:31:09 +0545 Subject: [PATCH] refactor(services): migrate run_command to Gio.Subprocess in KeyboardLayout and Modus --- src/services/keyboard_layout.py | 9 +- src/services/modus.py | 30 ++---- src/services/screencapture.py | 165 +++++++++++++++++++------------- 3 files changed, 110 insertions(+), 94 deletions(-) diff --git a/src/services/keyboard_layout.py b/src/services/keyboard_layout.py index efafa53..931a146 100644 --- a/src/services/keyboard_layout.py +++ b/src/services/keyboard_layout.py @@ -6,7 +6,6 @@ from fabric.utils import Gio, GLib, logger import shared.data as data from services.config import on_config_change -from utils.functions import run_command HYPRCTL_BIN = "hyprctl" @@ -137,15 +136,15 @@ class KeyboardLayout(Service): return layouts_str = ",".join(self.layouts) lua_eval_cmd = f"hl.config({{ input = {{ kb_layout = '{layouts_str}' }} }})" - run_command( + Gio.Subprocess.new( [HYPRCTL_BIN, "eval", lua_eval_cmd], - timeout=2, + Gio.SubprocessFlags.STDOUT_SILENCE | Gio.SubprocessFlags.STDERR_SILENCE, ) def _apply_current_layout_index(self): - run_command( + Gio.Subprocess.new( [HYPRCTL_BIN, "switchxkblayout", "all", str(self.current_index)], - timeout=2, + Gio.SubprocessFlags.STDOUT_SILENCE | Gio.SubprocessFlags.STDERR_SILENCE, ) def switch_to_next(self): diff --git a/src/services/modus.py b/src/services/modus.py index 3d609e4..483ba89 100644 --- a/src/services/modus.py +++ b/src/services/modus.py @@ -1,26 +1,28 @@ import json import os +import re import subprocess import sys from fabric.core.service import Property, Service, Signal from fabric.hyprland.service import Hyprland -from fabric.utils import logger +from fabric.utils import exec_shell_command_async, get_relative_path, logger +from services.custom_notification import CachedNotifications +from utils.app_name_resolver import app_name_resolver from utils.functions import ( find_binary, find_process_pid, - kill_process as _kill_process, - run_command, spawn_detached, ) +from utils.functions import ( + kill_process as _kill_process, +) def __getattr__(name): mod = sys.modules[__name__] if name == "notification_service": - from services.custom_notification import CachedNotifications - _inst = CachedNotifications() setattr(mod, name, _inst) return _inst @@ -46,8 +48,6 @@ def __getattr__(name): def get_notification_service(): - from services.custom_notification import CachedNotifications - return CachedNotifications() @@ -365,7 +365,6 @@ def _update_active_window(): if not title and not wmclass: service.current_active_app_name = "Finder" return - from utils.app_name_resolver import app_name_resolver name = app_name_resolver.format_app_name(title, wmclass) service.current_active_wm_class = wmclass @@ -447,17 +446,13 @@ def toggle_caffeine() -> bool: global _caffeine_process try: if is_caffeine_active(): - from fabric.utils import get_relative_path - inhibit_script = get_relative_path("utils/inhibit.py") - run_command(["python3", inhibit_script, "off"]) + spawn_detached(["python3", inhibit_script, "off"]) if _caffeine_process: _caffeine_process.terminate() _caffeine_process = None return False else: - from fabric.utils import get_relative_path - inhibit_script = get_relative_path("utils/inhibit.py") _caffeine_process = spawn_detached(["python3", inhibit_script, "on"]) return True @@ -540,15 +535,12 @@ def get_screen_dimensions() -> tuple[int, int]: def focus_window(address: str): - run_command( - ["hyprctl", "dispatch", f'hl.dsp.focus({{ window = "address:{address}" }})'], - timeout=2, + exec_shell_command_async( + f"hyprctl dispatch 'hl.dsp.focus({{ window = \"address:{address}\" }})'" ) def close_window(address: str): - from fabric.utils import exec_shell_command_async - exec_shell_command_async( f"hyprctl dispatch 'hl.dsp.window.close({{ window = \"address:{address}\" }})'" ) @@ -557,10 +549,8 @@ def close_window(address: str): def launch_app(command_line: str): if not command_line: return - import re cleaned = re.sub(r"%\w+", "", command_line).strip() - from fabric.utils import exec_shell_command_async exec_shell_command_async( f"hyprctl dispatch 'hl.dsp.exec_cmd([[uwsm app -- {cleaned}]])'" diff --git a/src/services/screencapture.py b/src/services/screencapture.py index d38fc9c..c99497e 100644 --- a/src/services/screencapture.py +++ b/src/services/screencapture.py @@ -237,41 +237,50 @@ class ScreenCapture(Service): out_file = save_dir / f"{timestamp}.png" if target == "region": - geo_proc = Gio.Subprocess.new( + slurp_proc = Gio.Subprocess.new( ["slurp"], Gio.SubprocessFlags.STDOUT_PIPE | Gio.SubprocessFlags.STDERR_SILENCE, ) - try: - geo_proc.wait(None) - except Exception: - pass - if geo_proc.get_exit_status() != 0: - self.notify_send( - "Screenshot cancelled", - "Selection was cancelled", - icon="camera-photo-symbolic", - ) - self.screenshot_taken(None) - return True - _, geo_bytes, _ = geo_proc.communicate(None) - geo = geo_bytes.decode().strip() if geo_bytes else "" - if not geo: - self.notify_send( - "Screenshot cancelled", - "Selection was cancelled", - icon="camera-photo-symbolic", - ) - self.screenshot_taken(None) - return True - cmd = ["grim", "-c", "-g", geo, str(out_file)] - else: - monitor = self._get_active_monitor() if target == "active" else None - cmd = ["grim", "-c"] - if monitor: - cmd.extend(["-o", monitor]) - cmd.append(str(out_file)) - expected = out_file + def on_slurp_done(proc, task, *_): + try: + _, stdout, _ = proc.communicate_utf8_finish(task) + except Exception: + self.notify_send( + "Screenshot cancelled", + "Selection was cancelled", + icon="camera-photo-symbolic", + ) + self.screenshot_taken(None) + return + + geo = stdout.strip() if stdout else "" + if not geo: + self.notify_send( + "Screenshot cancelled", + "Selection was cancelled", + icon="camera-photo-symbolic", + ) + self.screenshot_taken(None) + return + self._capture_grim_with_cursor(geo, str(out_file)) + + slurp_proc.communicate_utf8_async(None, None, on_slurp_done) + return True + + monitor = self._get_active_monitor() if target == "active" else None + self._capture_grim_with_cursor(None, str(out_file), monitor=monitor) + return True + + def _capture_grim_with_cursor(self, geometry, out_file, monitor=None): + cmd = ["grim", "-c"] + if geometry: + cmd.extend(["-g", geometry]) + elif monitor: + cmd.extend(["-o", monitor]) + cmd.append(out_file) + + expected = Path(out_file) def on_grim_done(proc, task, *_): try: @@ -291,7 +300,31 @@ class ScreenCapture(Service): proc = Gio.Subprocess.new(cmd, Gio.SubprocessFlags.NONE) proc.wait_async(None, on_grim_done) - return True + + def _start_wf_recorder( + self, output_file, use_audio, show_cursor, geometry=None, monitor=None + ): + cmd = [ + "wf-recorder", + "--file", + str(output_file), + "--pixel-format", + "yuv420p", + ] + + if not use_audio: + cmd.append("--no-audio") + + if show_cursor: + cmd.append("--show-cursor") + + if geometry: + cmd.extend(["-g", geometry]) + elif monitor: + cmd.extend(["-o", monitor]) + + Gio.Subprocess.new(cmd, Gio.SubprocessFlags.NONE) + self.recording_started(str(output_file)) def record(self, target="selection", use_audio=False, show_cursor=False): """ @@ -313,46 +346,40 @@ class ScreenCapture(Service): self.recording_file.write_text(str(output_file)) self.recording_start_time_file.write_text(str(int(time.time()))) - cmd = [ - "wf-recorder", - "--file", - str(output_file), - "--pixel-format", - "yuv420p", - ] - - if not use_audio: - cmd.append("--no-audio") - - if show_cursor: - cmd.append("--show-cursor") - if target == "selection": - try: - geometry = exec_shell_command("slurp 2>/dev/null") - 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", - "Selection was cancelled", - icon="camera-video-symbolic", - ) - return False - cmd.extend(["-g", geometry]) - elif target == "active": - cmd.extend(["-o", self._get_active_monitor()]) - else: - cmd.extend(["-o", target]) + slurp_proc = Gio.Subprocess.new( + ["slurp"], + Gio.SubprocessFlags.STDOUT_PIPE | Gio.SubprocessFlags.STDERR_SILENCE, + ) - Gio.Subprocess.new(cmd, Gio.SubprocessFlags.NONE) + def on_slurp_done(proc, task, *_): + try: + _, stdout, _ = proc.communicate_utf8_finish(task) + except Exception: + self.notify_send( + "Recording cancelled", + "Selection was cancelled", + icon="camera-video-symbolic", + ) + return - self.recording_started(str(output_file)) + geometry = stdout.strip() if stdout else "" + if not geometry: + self.notify_send( + "Recording cancelled", + "Selection was cancelled", + icon="camera-video-symbolic", + ) + return + self._start_wf_recorder( + output_file, use_audio, show_cursor, geometry=geometry + ) + + slurp_proc.communicate_utf8_async(None, None, on_slurp_done) + return True + + monitor = self._get_active_monitor() if target == "active" else target + self._start_wf_recorder(output_file, use_audio, show_cursor, monitor=monitor) return True def stop_recording(self):