refactor(services): migrate run_command to Gio.Subprocess in KeyboardLayout and Modus

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):

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}]])'"

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):