From f3f17bb5fc744a34dd895416bbdbbfd836b17cd6 Mon Sep 17 00:00:00 2001 From: S4NKALP Date: Mon, 13 Jul 2026 23:52:32 +0545 Subject: [PATCH] refactor(services): split screen capture into screenshot + recorder services Split the monolithic ScreenCapture service into focused Screenshot and ScreenRecorder services behind a pluggable capture backend. - Add shared/capture.py: CaptureBackend abstraction + HyprlandCaptureBackend (hyprshot/grim/slurp/wf-recorder) and reusable CaptureNotifier - Add services/screenshot.py and services/screenrecorder.py with Fabric properties/signals; recording state kept in memory (no /tmp files) - Keep services/screencapture.py as a backwards-compatible facade so existing widgets and signals keep working - Resolve monitors via Hyprland IPC helpers in services/modus.py instead of spawning `hyprctl monitors -j` per capture - Implement real window recording (slurp window boxes) and split the record-window / record-region UI mappings - Register screenshot_service / screen_recorder_service singletons - Remove dead src/utils/inhibit.py duplicate; minor modus.lua quoting fix --- config/hypr/modus.lua | 2 +- src/services/modus.py | 31 ++ src/services/screencapture.py | 443 +++++----------------------- src/services/screenrecorder.py | 281 ++++++++++++++++++ src/services/screenshot.py | 187 ++++++++++++ src/shared/capture.py | 482 +++++++++++++++++++++++++++++++ src/utils/inhibit.py | 193 ------------- src/utils/roam.py | 2 + src/window/screencapture/main.py | 8 +- 9 files changed, 1063 insertions(+), 566 deletions(-) create mode 100644 src/services/screenrecorder.py create mode 100644 src/services/screenshot.py create mode 100644 src/shared/capture.py delete mode 100644 src/utils/inhibit.py diff --git a/config/hypr/modus.lua b/config/hypr/modus.lua index f2c571e..b245154 100644 --- a/config/hypr/modus.lua +++ b/config/hypr/modus.lua @@ -30,7 +30,7 @@ for key, method in pairs({ ["SUPER + SHIFT + Y"] = "app.set_css()", -- Reload CSS ["ALT + TAB"] = "switcher.show_switcher()", -- Application Switcher ["SUPER + Z"] = "screencapture.toggle()", -- ScreenCapture - ["SUPER + S"] = 'screencapture.toggle(ss="region")', -- Screenshot Region + ["SUPER + S"] = "screencapture.toggle(ss='region')", -- Screenshot Region ["ALT + SPACE"] = "switch_keyboard_layout()", -- KB_Layout Switcher ["ALT + SHIFT + W"] = "spotlight.toggle('wr', external=True)", -- Random Wallpaper }) do diff --git a/src/services/modus.py b/src/services/modus.py index 760549d..8c80bea 100644 --- a/src/services/modus.py +++ b/src/services/modus.py @@ -43,6 +43,18 @@ def __getattr__(name): _inst = ScreenCapture() setattr(mod, name, _inst) return _inst + if name == "screenshot_service": + from services.screenshot import Screenshot + + _inst = Screenshot() + setattr(mod, name, _inst) + return _inst + if name == "screen_recorder_service": + from services.screenrecorder import ScreenRecorder + + _inst = ScreenRecorder() + setattr(mod, name, _inst) + return _inst raise AttributeError(f"module {__name__!r} has no attribute {name!r}") @@ -490,6 +502,25 @@ def get_monitors() -> list: return data if isinstance(data, list) else [] +def get_focused_monitor() -> dict | None: + monitors = get_monitors() + for monitor in monitors: + if monitor.get("focused"): + return monitor + return monitors[0] if monitors else None + + +def get_monitor_names() -> list[str]: + return [name for m in get_monitors() if (name := m.get("name"))] + + +def get_active_monitor_name(fallback: str = "eDP-1") -> str: + monitor = get_focused_monitor() + if monitor and (name := monitor.get("name")): + return name + return fallback + + def get_active_workspace_id() -> int: data = _hyprctl_json("j/activeworkspace") if isinstance(data, dict): diff --git a/src/services/screencapture.py b/src/services/screencapture.py index c99497e..a4b1a14 100644 --- a/src/services/screencapture.py +++ b/src/services/screencapture.py @@ -1,411 +1,114 @@ -import json -from datetime import datetime +"""Backwards-compatible facade over the split capture services. + +The screenshot and recording logic now live in :mod:`services.screenshot` +and :mod:`services.screenrecorder`. This module keeps the historical +``ScreenCapture`` API and its ``screenshot-taken`` / ``recording-started`` +/ ``recording-stopped`` signals so existing widgets keep working while new +code can depend on the focused services directly. +""" + from pathlib import Path -from fabric import Service, Signal -from fabric.core.service import Property -from fabric.utils import ( - Gio, - exec_shell_command, - exec_shell_command_async, - logger, - time, -) +from fabric.core.service import Property, Service, Signal -from utils.functions import is_app_running, kill_process +from services.screenrecorder import ScreenRecorder +from services.screenshot import Screenshot class ScreenCapture(Service): - """Service for screen capture and recording functionality""" + """Facade delegating to :class:`Screenshot` and :class:`ScreenRecorder`.""" @Signal def screenshot_taken(self, path: str) -> None: - """Signal emitted when screenshot is taken.""" + """Emitted after a screenshot attempt (``None`` path on cancel/fail).""" @Signal def recording_started(self, path: str) -> None: - """Signal emitted when recording starts.""" + """Emitted when a recording starts.""" @Signal def recording_stopped(self, path: str) -> None: - """Signal emitted when recording stops.""" + """Emitted when a recording stops.""" - _instance = None + _instance: "ScreenCapture | None" = None - def __new__(cls): + def __new__(cls) -> "ScreenCapture": if cls._instance is None: cls._instance = super().__new__(cls) return cls._instance - def __init__(self, **kwargs): + def __init__(self, **kwargs) -> None: if getattr(self, "_initialized", False): return - super().__init__(**kwargs) self._initialized = True - self.home = Path.home() - self.screenshots_dir = self.home / "Pictures" / "Screenshots" - self.recordings_dir = self.home / "Videos" / "Recordings" - self.recording_file = Path("/tmp/recording.txt") - self.recording_start_time_file = Path("/tmp/recording_start_time.txt") - self.screenshots_dir.mkdir(parents=True, exist_ok=True) - self.recordings_dir.mkdir(parents=True, exist_ok=True) + self._screenshot = Screenshot() + self._recorder = ScreenRecorder() - def notify_send(self, title, message, icon=None, actions=None): - cmd = ["notify-send", "-a", "Modus"] + self.screenshots_dir = self._screenshot.screenshots_dir + self.recordings_dir = self._recorder.recordings_dir - if icon: - cmd.extend(["-i", str(icon)]) + self._screenshot.connect("captured", self._on_captured) + self._screenshot.connect("failed", self._on_screenshot_failed) + self._recorder.connect("started", self._on_recording_started) + self._recorder.connect("stopped", self._on_recording_stopped) - if actions: - for action in actions: - cmd.extend(["-A", f"{action}={action}"]) + @property + def screenshot_service(self) -> Screenshot: + """The underlying screenshot service.""" + return self._screenshot - cmd.extend([title, message]) - try: - Gio.Subprocess.new( - cmd, - Gio.SubprocessFlags.STDOUT_SILENCE | Gio.SubprocessFlags.STDERR_SILENCE, - ) - except Exception: - pass + @property + def recorder_service(self) -> ScreenRecorder: + """The underlying recorder service.""" + return self._recorder - def send_screenshot_notification(self, file_path=None): - cmd = ["notify-send"] - cmd.extend( - [ - "-A", - "files=Show in Files", - "-A", - "view=View", - "-A", - "edit=Edit", - "-i", - "camera-photo-symbolic", - "-a", - "Fabric Screenshot Utility", - "-h", - f"STRING:image-path:{file_path}", - "Screenshot Saved", - f"Saved Screenshot at {file_path}", - ] - if file_path - else ["Screenshot Sent to Clipboard"] - ) - - proc: Gio.Subprocess = Gio.Subprocess.new(cmd, Gio.SubprocessFlags.STDOUT_PIPE) - - def do_callback(process: Gio.Subprocess, task: Gio.Task): - try: - _, stdout, stderr = process.communicate_utf8_finish(task) - except Exception: - logger.error( - f"[SCREENSHOT] Failed read notification action with error {stderr}" - ) - return - - match stdout.strip("\n"): - case "files": - exec_shell_command_async(f"xdg-open {self.screenshots_dir}") - case "view": - exec_shell_command_async(f"xdg-open {file_path}") - case "edit": - exec_shell_command_async(f"satty -f {file_path}") - - proc.communicate_utf8_async(None, None, do_callback) - self.screenshot_taken(file_path) - - def send_recording_notification(self, file_path): - cmd = ["notify-send"] - cmd.extend( - [ - "-A", - "files=Show in Files", - "-A", - "view=View", - "-i", - "camera-video-symbolic", - "-a", - "Fabric Screenshot Utility", - "Screencast Saved", - f"Saved Screencast at {file_path}", - ] - ) - - proc: Gio.Subprocess = Gio.Subprocess.new(cmd, Gio.SubprocessFlags.STDOUT_PIPE) - - def do_callback(process: Gio.Subprocess, task: Gio.Task): - try: - _, stdout, stderr = process.communicate_utf8_finish(task) - except Exception: - logger.error( - f"[SCREENCAST] Failed read notification action with error {stderr}" - ) - return - - match stdout.strip("\n"): - case "files": - exec_shell_command_async(f"xdg-open {self.recordings_dir}") - case "view": - exec_shell_command_async(f"xdg-open {file_path}") - - proc.communicate_utf8_async(None, None, do_callback) - self.recording_stopped(file_path) - - def _get_active_monitor(self): - """Get the name of the currently focused monitor.""" - try: - monitors_json = exec_shell_command("hyprctl monitors -j") - if monitors_json: - monitors = json.loads(monitors_json) - for monitor in monitors: - if monitor.get("focused"): - return monitor.get("name") - except Exception as e: - logger.error(f"[SCREENSHOT] Failed to get active monitor: {e}") - return "eDP-1" # Fallback - - def screenshot(self, target="region", output_dir=None, show_cursor=False): - """ - Take a screenshot. - :param target: 'region', 'window', 'active', 'output', or a display name - :param output_dir: where to save; defaults to self.screenshots_dir - :param show_cursor: include the mouse cursor in the capture - """ - timestamp = datetime.now().strftime("%Y-%m-%d_%H-%M-%S") - save_dir = Path(output_dir).expanduser() if output_dir else self.screenshots_dir - save_dir.mkdir(parents=True, exist_ok=True) - - # When cursor is needed we call grim directly (hyprshot has no cursor flag). - # This covers region and fullscreen; window mode still falls back to hyprshot. - if show_cursor and target in ("region", "active", "output"): - return self._screenshot_with_cursor(target, save_dir, timestamp) - - command = [ - "hyprshot", - "-s", - "-o", - str(save_dir), - "-f", - f"{timestamp}.png", - ] - - if target == "region": - command.extend(["-m", "region"]) - elif target == "window": - command.extend(["-m", "window"]) - elif target == "active": - active_monitor = self._get_active_monitor() - command.extend(["-m", "output", "-m", active_monitor]) - elif target == "output": - command.extend(["-m", "output"]) - else: - command.extend(["-m", "output", "-m", target]) - - expected = save_dir / f"{timestamp}.png" - - def on_capture(proc, task, *_): - try: - proc.wait_finish(task) - if expected.exists(): - self.send_screenshot_notification(file_path=str(expected)) - else: - self.notify_send( - "Screenshot cancelled", - "Selection was cancelled", - icon="camera-photo-symbolic", - ) - self.screenshot_taken(None) - except Exception as e: - logger.error(f"Screenshot notification failed: {e}") - self.screenshot_taken(None) - - try: - proc = Gio.Subprocess.new(command, Gio.SubprocessFlags.STDERR_SILENCE) - proc.wait_async(None, on_capture) - except Exception as e: - logger.error(f"Screenshot failed: {e}") - return False - - return True - - def _screenshot_with_cursor( - self, target: str, save_dir: Path, timestamp: str + def screenshot( + self, + target: str = "region", + output_dir: str | Path | None = None, + show_cursor: bool = False, ) -> bool: - """Use grim -c directly so the cursor is included in the capture.""" - out_file = save_dir / f"{timestamp}.png" + """Take a screenshot. See :meth:`Screenshot.screenshot`.""" + return self._screenshot.screenshot( + target, output_dir=output_dir, show_cursor=show_cursor + ) - if target == "region": - slurp_proc = Gio.Subprocess.new( - ["slurp"], - Gio.SubprocessFlags.STDOUT_PIPE | Gio.SubprocessFlags.STDERR_SILENCE, - ) + def record( + self, + target: str = "selection", + use_audio: bool = False, + show_cursor: bool = False, + ) -> bool: + """Start a recording. See :meth:`ScreenRecorder.start`.""" + return self._recorder.start( + target, use_audio=use_audio, show_cursor=show_cursor + ) - 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: - proc.wait_finish(task) - if expected.exists(): - self.send_screenshot_notification(file_path=str(expected)) - else: - self.notify_send( - "Screenshot cancelled", - "Selection was cancelled", - icon="camera-photo-symbolic", - ) - self.screenshot_taken(None) - except Exception as e: - logger.error(f"Screenshot notification failed: {e}") - self.screenshot_taken(None) - - proc = Gio.Subprocess.new(cmd, Gio.SubprocessFlags.NONE) - proc.wait_async(None, on_grim_done) - - 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): - """ - Start recording. - :param target: 'selection', 'active', a monitor name like 'eDP-1', etc. - :param use_audio: capture system/mic audio (removes --no-audio) - :param show_cursor: show mouse cursor in the recording - """ - if self.is_recording: - logger.error( - "[SCREENRECORD] Another instance of wf-recorder is already running" - ) - return False - - timestamp = datetime.now().strftime("%Y-%m-%d_%H-%M-%S") - output_file = self.recordings_dir / f"{timestamp}.mkv" - - self._current_recording_path = str(output_file) - self.recording_file.write_text(str(output_file)) - self.recording_start_time_file.write_text(str(int(time.time()))) - - if target == "selection": - slurp_proc = Gio.Subprocess.new( - ["slurp"], - Gio.SubprocessFlags.STDOUT_PIPE | Gio.SubprocessFlags.STDERR_SILENCE, - ) - - 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 - - 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): - kill_process("wf-recorder") - - path = getattr(self, "_current_recording_path", None) - if not path and self.recording_file.exists(): - path = self.recording_file.read_text().strip() - - if path: - self.send_recording_notification(path) - - if self.recording_start_time_file.exists(): - self.recording_start_time_file.unlink() - - if self.recording_file.exists(): - self.recording_file.unlink() - - return True + def stop_recording(self) -> bool: + """Stop the active recording.""" + return self._recorder.stop() @Property(bool, "readable", default_value=False) - def is_recording(self): - return is_app_running("wf-recorder") or is_app_running("gpu-screen-recorder") + def is_recording(self) -> bool: + """Whether a recorder process is currently running.""" + return self._recorder.active + + def _on_captured(self, _service: Screenshot, path: str) -> None: + self.screenshot_taken(path) + + def _on_screenshot_failed(self, _service: Screenshot, _reason: str) -> None: + self.screenshot_taken(None) + + def _on_recording_started(self, _service: ScreenRecorder, path: str) -> None: + self.recording_started(path) + + def _on_recording_stopped(self, _service: ScreenRecorder, path: str) -> None: + self.recording_stopped(path) -def __getattr__(name): +def __getattr__(name: str): if name == "screen_capture_service": import sys diff --git a/src/services/screenrecorder.py b/src/services/screenrecorder.py new file mode 100644 index 0000000..f1922b5 --- /dev/null +++ b/src/services/screenrecorder.py @@ -0,0 +1,281 @@ +"""Screen recording service. + +Owns recording only; screenshots live in :mod:`services.screenshot`. +Recording state is held entirely in memory as Fabric properties -- no +temporary files are used. +""" + +import time +from datetime import datetime +from pathlib import Path + +from fabric.core.service import Property, Service, Signal +from fabric.utils import GLib, exec_shell_command_async + +from shared.capture import ( + CaptureBackend, + CaptureNotifier, + CaptureResult, + HyprlandCaptureBackend, + RecordingRequest, +) +from utils.functions import is_app_running + +_RECORDER_PROCESSES = ("wf-recorder", "gpu-screen-recorder") + + +class ScreenRecorder(Service): + """Record the screen through a pluggable :class:`CaptureBackend`.""" + + @Signal + def started(self, path: str) -> None: + """Emitted with the output path when a recording begins.""" + + @Signal + def stopped(self, path: str) -> None: + """Emitted with the output path when a recording ends.""" + + @Signal("paused") + def paused_signal(self) -> None: + """Emitted when an active recording is paused.""" + + @Signal + def resumed(self) -> None: + """Emitted when a paused recording resumes.""" + + @Signal + def failed(self, reason: str) -> None: + """Emitted when a recording is cancelled or fails.""" + + _instance: "ScreenRecorder | None" = None + + def __new__(cls) -> "ScreenRecorder": + if cls._instance is None: + cls._instance = super().__new__(cls) + return cls._instance + + def __init__(self, backend: CaptureBackend | None = None, **kwargs) -> None: + if getattr(self, "_initialized", False): + return + super().__init__(**kwargs) + self._initialized = True + + self._backend = backend or HyprlandCaptureBackend() + self._notifier = CaptureNotifier() + + self._recording = False + self._paused = False + self._output_file: str | None = None + self._start_time: float | None = None + self._tick_source_id: int | None = None + + self.recordings_dir = Path.home() / "Videos" / "Recordings" + self.recordings_dir.mkdir(parents=True, exist_ok=True) + + # -- properties -------------------------------------------------------- + + @Property(bool, "readable", default_value=False) + def recording(self) -> bool: + """In-memory recording state managed by this service.""" + return self._recording + + @Property(bool, "readable", default_value=False) + def paused(self) -> bool: + """Whether the current recording is paused.""" + return self._paused + + @Property(int, "readable", default_value=0) + def elapsed_seconds(self) -> int: + """Seconds elapsed since the current recording started.""" + if self._start_time is None: + return 0 + return int(time.monotonic() - self._start_time) + + @Property(str, "readable", default_value="") + def output_file(self) -> str: + """Path of the current or most recent recording.""" + return self._output_file or "" + + @Property(str, "readable", default_value="") + def backend(self) -> str: + """Name of the active capture backend.""" + return self._backend.name + + @Property(bool, "readable", default_value=False) + def busy(self) -> bool: + """Whether a recording is active (in-memory state).""" + return self._recording + + @Property(bool, "readable", default_value=False) + def available(self) -> bool: + """Whether the backend can record on this system.""" + return self._backend.capabilities.can_record + + @Property(bool, "readable", default_value=False) + def supports_pause(self) -> bool: + """Whether the backend supports pause/resume.""" + return self._backend.capabilities.supports_pause + + @Property(bool, "readable", default_value=False) + def active(self) -> bool: + """Whether a recorder process is actually running. + + Unlike :attr:`recording`, this reflects the real process state and + also detects recordings started outside this service. + """ + return any(is_app_running(proc) for proc in _RECORDER_PROCESSES) + + # -- public API -------------------------------------------------------- + + def start( + self, + target: str = "selection", + use_audio: bool = False, + show_cursor: bool = False, + ) -> bool: + """Start a recording. + + :param target: ``selection``, ``active`` or a specific monitor name. + :param use_audio: capture audio. + :param show_cursor: show the mouse cursor in the recording. + :returns: ``True`` if the recording attempt was launched. + """ + if self.active: + self.failed("A recording is already running") + return False + if not self._backend.capabilities.can_record: + self.failed("Recording backend is unavailable") + return False + + request = self._build_request(target, use_audio, show_cursor) + return self._backend.start_recording(request, self._on_start_result) + + def stop(self) -> bool: + """Stop the active recording.""" + if not (self._recording or self.active): + return False + path = self._output_file + self._backend.stop_recording() + self._teardown_state() + if path: + self._notify_saved(path) + self.stopped(path) + return True + + def pause(self) -> bool: + """Pause the active recording, if the backend supports it.""" + if not self._recording or self._paused: + return False + if not self._backend.pause_recording(): + return False + self._paused = True + self.notify("paused") + self.paused_signal() + return True + + def resume(self) -> bool: + """Resume a paused recording, if the backend supports it.""" + if not self._recording or not self._paused: + return False + if not self._backend.resume_recording(): + return False + self._paused = False + self.notify("paused") + self.resumed() + return True + + def toggle(self) -> bool: + """Stop when recording, otherwise start with defaults.""" + if self._recording or self.active: + return self.stop() + return self.start() + + # -- internals --------------------------------------------------------- + + def _build_request( + self, target: str, use_audio: bool, show_cursor: bool + ) -> RecordingRequest: + timestamp = datetime.now().strftime("%Y-%m-%d_%H-%M-%S") + return RecordingRequest( + target=target, + output_file=self.recordings_dir / f"{timestamp}.mkv", + use_audio=use_audio, + show_cursor=show_cursor, + monitor=self._resolve_monitor(target), + ) + + @staticmethod + def _resolve_monitor(target: str) -> str | None: + if target != "active": + return None + from services.modus import get_active_monitor_name + + return get_active_monitor_name() + + def _on_start_result(self, result: CaptureResult) -> None: + if result.success and result.path: + self._setup_state(result.path) + self.started(result.path) + elif result.cancelled: + self._notify_cancelled() + self.failed("cancelled") + else: + self.failed(result.error or "Recording failed") + + def _setup_state(self, path: str) -> None: + self._output_file = path + self._recording = True + self._paused = False + self._start_time = time.monotonic() + self._start_tick() + self.notify("output-file") + self.notify("recording") + self.notify("busy") + + def _teardown_state(self) -> None: + self._stop_tick() + self._recording = False + self._paused = False + self._start_time = None + self.notify("recording") + self.notify("busy") + self.notify("elapsed-seconds") + + def _start_tick(self) -> None: + self._stop_tick() + self._tick_source_id = GLib.timeout_add_seconds(1, self._on_tick) + + def _stop_tick(self) -> None: + if self._tick_source_id is not None: + GLib.source_remove(self._tick_source_id) + self._tick_source_id = None + + def _on_tick(self) -> bool: + if not self._recording: + self._tick_source_id = None + return False + self.notify("elapsed-seconds") + return True + + def _notify_saved(self, path: str) -> None: + self._notifier.notify( + "Screencast Saved", + f"Saved Screencast at {path}", + icon="camera-video-symbolic", + actions=[("files", "Show in Files"), ("view", "View")], + on_action=lambda action: self._handle_action(action, path), + ) + + def _handle_action(self, action: str, path: str) -> None: + match action: + case "files": + exec_shell_command_async(f"xdg-open {self.recordings_dir}") + case "view": + exec_shell_command_async(f"xdg-open {path}") + + def _notify_cancelled(self) -> None: + self._notifier.notify( + "Recording cancelled", + "Selection was cancelled", + icon="camera-video-symbolic", + ) diff --git a/src/services/screenshot.py b/src/services/screenshot.py new file mode 100644 index 0000000..5382c92 --- /dev/null +++ b/src/services/screenshot.py @@ -0,0 +1,187 @@ +"""Screenshot service. + +Owns screenshot capture only; recording lives in +:mod:`services.screenrecorder`. The service resolves the desired target +into a concrete request, hands it to a :class:`CaptureBackend`, then raises +notifications and signals based on the result. +""" + +from datetime import datetime +from pathlib import Path + +from fabric.core.service import Property, Service, Signal +from fabric.utils import exec_shell_command_async + +from shared.capture import ( + CaptureBackend, + CaptureNotifier, + CaptureResult, + HyprlandCaptureBackend, + ScreenshotRequest, +) +from utils.functions import find_binary + + +class Screenshot(Service): + """Capture screenshots through a pluggable :class:`CaptureBackend`.""" + + @Signal + def captured(self, path: str) -> None: + """Emitted with the saved path when a screenshot succeeds.""" + + @Signal + def failed(self, reason: str) -> None: + """Emitted when a screenshot is cancelled or fails.""" + + _instance: "Screenshot | None" = None + + def __new__(cls) -> "Screenshot": + if cls._instance is None: + cls._instance = super().__new__(cls) + return cls._instance + + def __init__(self, backend: CaptureBackend | None = None, **kwargs) -> None: + if getattr(self, "_initialized", False): + return + super().__init__(**kwargs) + self._initialized = True + + self._backend = backend or HyprlandCaptureBackend() + self._notifier = CaptureNotifier() + self._busy = False + self._last_file: str | None = None + self._has_satty = find_binary("satty") is not None + + self.screenshots_dir = Path.home() / "Pictures" / "Screenshots" + self.screenshots_dir.mkdir(parents=True, exist_ok=True) + + # -- properties -------------------------------------------------------- + + @Property(bool, "readable", default_value=False) + def busy(self) -> bool: + """Whether a capture is currently in progress.""" + return self._busy + + @Property(str, "readable", default_value="") + def last_file(self) -> str: + """Path of the most recently captured screenshot.""" + return self._last_file or "" + + @Property(str, "readable", default_value="") + def backend(self) -> str: + """Name of the active capture backend.""" + return self._backend.name + + @Property(bool, "readable", default_value=False) + def available(self) -> bool: + """Whether the backend can take screenshots on this system.""" + return self._backend.capabilities.can_screenshot + + @Property(bool, "readable", default_value=False) + def can_edit(self) -> bool: + """Whether an external editor (``satty``) is available.""" + return self._has_satty + + # -- public API -------------------------------------------------------- + + def screenshot( + self, + target: str = "region", + output_dir: str | Path | None = None, + show_cursor: bool = False, + ) -> bool: + """Take a screenshot. + + :param target: ``region``, ``window``, ``active``, ``output`` or a + specific monitor name. + :param output_dir: destination directory; defaults to the standard + screenshots directory. + :param show_cursor: include the mouse cursor in the capture. + :returns: ``True`` if the capture attempt was launched. + """ + if not self._backend.capabilities.can_screenshot: + self._fail("Screenshot backend is unavailable") + return False + + request = self._build_request(target, output_dir, show_cursor) + self._set_busy(True) + launched = self._backend.screenshot(request, self._on_result) + if not launched: + self._set_busy(False) + return launched + + # -- internals --------------------------------------------------------- + + def _build_request( + self, target: str, output_dir: str | Path | None, show_cursor: bool + ) -> ScreenshotRequest: + save_dir = Path(output_dir).expanduser() if output_dir else self.screenshots_dir + timestamp = datetime.now().strftime("%Y-%m-%d_%H-%M-%S") + monitor = self._resolve_monitor(target) + return ScreenshotRequest( + target=target, + save_path=save_dir / f"{timestamp}.png", + show_cursor=show_cursor, + monitor=monitor, + ) + + @staticmethod + def _resolve_monitor(target: str) -> str | None: + if target != "active": + return None + from services.modus import get_active_monitor_name + + return get_active_monitor_name() + + def _on_result(self, result: CaptureResult) -> None: + self._set_busy(False) + if result.success and result.path: + self._on_success(result.path) + elif result.cancelled: + self._notify_cancelled() + self.failed("cancelled") + else: + self._fail(result.error or "Screenshot failed") + + def _on_success(self, path: str) -> None: + self._last_file = path + self.notify("last-file") + self._notify_saved(path) + self.captured(path) + + def _notify_saved(self, path: str) -> None: + actions = [("files", "Show in Files"), ("view", "View")] + if self._has_satty: + actions.append(("edit", "Edit")) + self._notifier.notify( + "Screenshot Saved", + f"Saved Screenshot at {path}", + icon="camera-photo-symbolic", + actions=actions, + hints=[f"STRING:image-path:{path}"], + on_action=lambda action: self._handle_action(action, path), + ) + + def _handle_action(self, action: str, path: str) -> None: + match action: + case "files": + exec_shell_command_async(f"xdg-open {self.screenshots_dir}") + case "view": + exec_shell_command_async(f"xdg-open {path}") + case "edit": + exec_shell_command_async(f"satty -f {path}") + + def _notify_cancelled(self) -> None: + self._notifier.notify( + "Screenshot cancelled", + "Selection was cancelled", + icon="camera-photo-symbolic", + ) + + def _fail(self, reason: str) -> None: + self.failed(reason) + + def _set_busy(self, value: bool) -> None: + if value != self._busy: + self._busy = value + self.notify("busy") diff --git a/src/shared/capture.py b/src/shared/capture.py new file mode 100644 index 0000000..e2224d2 --- /dev/null +++ b/src/shared/capture.py @@ -0,0 +1,482 @@ +"""Shared capture infrastructure: backend abstraction + notifications. + +Services never invoke capture tools (``hyprshot``, ``grim``, ``slurp``, +``wf-recorder``) directly. They speak only to a :class:`CaptureBackend`, +which encapsulates the concrete tooling. This makes it possible to add +alternative backends (GPU Screen Recorder, PipeWire, XDG Desktop Portal) +later without touching the services or the UI. + +:class:`CaptureNotifier` provides the reusable ``notify-send`` plumbing used +by both the screenshot and recorder services so the notification code is not +duplicated in each. +""" + +from __future__ import annotations + +from abc import ABC, abstractmethod +from dataclasses import dataclass, field +from typing import TYPE_CHECKING, Callable, Iterable + +from fabric.utils import Gio, logger + +from utils.functions import find_binary, kill_process + +if TYPE_CHECKING: + from pathlib import Path + +ResultCallback = Callable[["CaptureResult"], None] +NotificationAction = tuple[str, str] +ActionCallback = Callable[[str], None] + + +@dataclass(slots=True) +class CaptureResult: + """Outcome of a capture operation reported back to a service.""" + + success: bool + path: str | None = None + cancelled: bool = False + error: str | None = None + + @classmethod + def ok(cls, path: str) -> CaptureResult: + return cls(success=True, path=path) + + @classmethod + def user_cancelled(cls) -> CaptureResult: + return cls(success=False, cancelled=True) + + @classmethod + def failure(cls, error: str) -> CaptureResult: + return cls(success=False, error=error) + + +@dataclass(slots=True) +class ScreenshotRequest: + """Description of a single screenshot to capture. + + :param target: ``region``, ``window``, ``active``, ``output`` or a + specific monitor name. + :param save_path: absolute path the resulting image is written to. + :param show_cursor: whether the mouse cursor is included. + :param monitor: resolved monitor name (used for ``active`` / ``output``). + """ + + target: str + save_path: Path + show_cursor: bool = False + monitor: str | None = None + + +@dataclass(slots=True) +class RecordingRequest: + """Description of a screen recording to start. + + :param target: ``selection``, ``active`` or a specific monitor name. + :param output_file: absolute path the recording is written to. + :param use_audio: whether audio is captured. + :param show_cursor: whether the mouse cursor is recorded. + :param monitor: resolved monitor name (used for ``active``). + """ + + target: str + output_file: Path + use_audio: bool = False + show_cursor: bool = False + monitor: str | None = None + + +@dataclass(slots=True) +class BackendCapabilities: + """Static description of what a backend can do on this system.""" + + tools: dict[str, bool] = field(default_factory=dict) + can_screenshot: bool = False + can_record: bool = False + supports_cursor: bool = False + supports_pause: bool = False + + +class CaptureBackend(ABC): + """Abstract capture/recording backend. + + Concrete backends translate high level requests into whatever tooling + they wrap and report the outcome through the supplied callback. + """ + + name: str = "abstract" + + @property + @abstractmethod + def capabilities(self) -> BackendCapabilities: + """Return the static capabilities of this backend.""" + + @abstractmethod + def screenshot(self, request: ScreenshotRequest, on_result: ResultCallback) -> bool: + """Capture a screenshot; return ``True`` if the attempt was launched.""" + + @abstractmethod + def start_recording( + self, request: RecordingRequest, on_result: ResultCallback + ) -> bool: + """Begin a recording; return ``True`` if the attempt was launched.""" + + @abstractmethod + def stop_recording(self) -> bool: + """Stop the active recording; return ``True`` if one was stopped.""" + + def pause_recording(self) -> bool: + """Pause the active recording. Unsupported by default.""" + return False + + def resume_recording(self) -> bool: + """Resume a paused recording. Unsupported by default.""" + return False + + +class HyprlandCaptureBackend(CaptureBackend): + """Capture backend built on ``hyprshot``, ``grim``, ``slurp`` and + ``wf-recorder`` for Hyprland/wlroots sessions.""" + + name = "hyprland" + + _RECORDER = "wf-recorder" + + def __init__(self) -> None: + self._tools = { + tool: find_binary(tool) is not None + for tool in ("hyprshot", "grim", "slurp", self._RECORDER) + } + + @property + def capabilities(self) -> BackendCapabilities: + tools = self._tools + return BackendCapabilities( + tools=dict(tools), + can_screenshot=tools.get("hyprshot", False) or tools.get("grim", False), + can_record=tools.get(self._RECORDER, False), + supports_cursor=tools.get("grim", False), + supports_pause=False, + ) + + def has_tool(self, tool: str) -> bool: + """Return whether a wrapped tool is available on this system.""" + return self._tools.get(tool, False) + + # screenshots + + def screenshot(self, request: ScreenshotRequest, on_result: ResultCallback) -> bool: + request.save_path.parent.mkdir(parents=True, exist_ok=True) + if request.show_cursor and request.target in ("region", "active", "output"): + return self._screenshot_with_cursor(request, on_result) + return self._screenshot_hyprshot(request, on_result) + + def _screenshot_hyprshot( + self, request: ScreenshotRequest, on_result: ResultCallback + ) -> bool: + if not self.has_tool("hyprshot"): + on_result(CaptureResult.failure("hyprshot is not installed")) + return False + + command = [ + "hyprshot", + "-s", + "-o", + str(request.save_path.parent), + "-f", + request.save_path.name, + *self._hyprshot_mode(request), + ] + return self._spawn_capture(command, request.save_path, on_result) + + @staticmethod + def _hyprshot_mode(request: ScreenshotRequest) -> list[str]: + target = request.target + if target == "region": + return ["-m", "region"] + if target == "window": + return ["-m", "window"] + if target == "active": + return ["-m", "output", "-m", request.monitor or ""] + if target == "output": + return ["-m", "output"] + return ["-m", "output", "-m", target] + + def _screenshot_with_cursor( + self, request: ScreenshotRequest, on_result: ResultCallback + ) -> bool: + if not self.has_tool("grim"): + on_result(CaptureResult.failure("grim is not installed")) + return False + + if request.target == "region": + return self._region_with_cursor(request, on_result) + + monitor = request.monitor if request.target == "active" else None + return self._grim_capture(request.save_path, on_result, monitor=monitor) + + def _region_with_cursor( + self, request: ScreenshotRequest, on_result: ResultCallback + ) -> bool: + def on_geometry(geometry: str | None) -> None: + if not geometry: + on_result(CaptureResult.user_cancelled()) + return + self._grim_capture(request.save_path, on_result, geometry=geometry) + + return self._select_geometry(on_geometry) + + def _grim_capture( + self, + save_path: Path, + on_result: ResultCallback, + geometry: str | None = None, + monitor: str | None = None, + ) -> bool: + command = ["grim", "-c"] + if geometry: + command += ["-g", geometry] + elif monitor: + command += ["-o", monitor] + command.append(str(save_path)) + return self._spawn_capture(command, save_path, on_result) + + def _spawn_capture( + self, command: list[str], expected: Path, on_result: ResultCallback + ) -> bool: + def on_done(proc: Gio.Subprocess, task: Gio.Task) -> None: + try: + proc.wait_finish(task) + except Exception as error: + on_result(CaptureResult.failure(str(error))) + return + if expected.exists(): + on_result(CaptureResult.ok(str(expected))) + else: + on_result(CaptureResult.user_cancelled()) + + proc = self._spawn(command, Gio.SubprocessFlags.STDERR_SILENCE) + if proc is None: + on_result(CaptureResult.failure(f"failed to launch {command[0]}")) + return False + proc.wait_async(None, on_done) + return True + + # recording + + def start_recording( + self, request: RecordingRequest, on_result: ResultCallback + ) -> bool: + if not self.has_tool(self._RECORDER): + on_result(CaptureResult.failure(f"{self._RECORDER} is not installed")) + return False + request.output_file.parent.mkdir(parents=True, exist_ok=True) + + if request.target == "selection": + return self._record_geometry(request, on_result, self._select_geometry) + if request.target == "window": + return self._record_geometry(request, on_result, self._select_window) + + monitor = request.monitor or request.target + return self._launch_recorder(request, on_result, monitor=monitor) + + def _record_geometry( + self, + request: RecordingRequest, + on_result: ResultCallback, + selector: Callable[[Callable[[str | None], None]], bool], + ) -> bool: + def on_geometry(geometry: str | None) -> None: + if not geometry: + on_result(CaptureResult.user_cancelled()) + return + self._launch_recorder(request, on_result, geometry=geometry) + + return selector(on_geometry) + + def _launch_recorder( + self, + request: RecordingRequest, + on_result: ResultCallback, + geometry: str | None = None, + monitor: str | None = None, + ) -> bool: + command = [ + self._RECORDER, + "--file", + str(request.output_file), + "--pixel-format", + "yuv420p", + ] + if not request.use_audio: + command.append("--no-audio") + if request.show_cursor: + command.append("--show-cursor") + if geometry: + command += ["-g", geometry] + elif monitor: + command += ["-o", monitor] + + proc = self._spawn(command, Gio.SubprocessFlags.NONE) + if proc is None: + on_result(CaptureResult.failure(f"failed to launch {self._RECORDER}")) + return False + on_result(CaptureResult.ok(str(request.output_file))) + return True + + def stop_recording(self) -> bool: + kill_process(self._RECORDER) + return True + + # helpers + + def _select_geometry(self, on_geometry: Callable[[str | None], None]) -> bool: + if not self.has_tool("slurp"): + on_geometry(None) + return False + + def on_done(proc: Gio.Subprocess, task: Gio.Task) -> None: + try: + _, stdout, _ = proc.communicate_utf8_finish(task) + except Exception: + on_geometry(None) + return + on_geometry(stdout.strip() if stdout else None) + + proc = self._spawn( + ["slurp"], + Gio.SubprocessFlags.STDOUT_PIPE | Gio.SubprocessFlags.STDERR_SILENCE, + ) + if proc is None: + on_geometry(None) + return False + proc.communicate_utf8_async(None, None, on_done) + return True + + def _select_window(self, on_geometry: Callable[[str | None], None]) -> bool: + """Let the user pick a visible window and return its geometry. + + Mirrors how ``hyprshot`` implements window selection: the geometry of + every mapped window on an active workspace is fed to ``slurp -r`` so a + single click on a window selects it. + """ + if not self.has_tool("slurp"): + on_geometry(None) + return False + + boxes = self._window_boxes() + if not boxes: + on_geometry(None) + return False + + def on_done(proc: Gio.Subprocess, task: Gio.Task) -> None: + try: + _, stdout, _ = proc.communicate_utf8_finish(task) + except Exception: + on_geometry(None) + return + on_geometry(stdout.strip() if stdout else None) + + proc = self._spawn( + ["slurp", "-r"], + Gio.SubprocessFlags.STDIN_PIPE + | Gio.SubprocessFlags.STDOUT_PIPE + | Gio.SubprocessFlags.STDERR_SILENCE, + ) + if proc is None: + on_geometry(None) + return False + proc.communicate_utf8_async(boxes, None, on_done) + return True + + @staticmethod + def _window_boxes() -> str: + """Build a ``slurp``-compatible box list of currently visible windows.""" + from services.modus import get_clients, get_monitors + + active_ids = { + m.get("activeWorkspace", {}).get("id") for m in get_monitors() + } + boxes: list[str] = [] + for client in get_clients(): + if not client.get("mapped", True): + continue + if client.get("workspace", {}).get("id") not in active_ids: + continue + at = client.get("at") + size = client.get("size") + if not at or not size: + continue + boxes.append(f"{at[0]},{at[1]} {size[0]}x{size[1]}") + return "\n".join(boxes) + + @staticmethod + def _spawn(command: list[str], flags: Gio.SubprocessFlags) -> Gio.Subprocess | None: + try: + return Gio.Subprocess.new(command, flags) + except Exception as error: + logger.error(f"[CaptureBackend] Failed to spawn {command}: {error}") + return None + + +class CaptureNotifier: + """Thin wrapper around ``notify-send`` with async action handling.""" + + def __init__(self, app_name: str = "Fabric Screenshot Utility") -> None: + self._app_name = app_name + + def notify( + self, + title: str, + message: str, + *, + icon: str | None = None, + actions: Iterable[NotificationAction] | None = None, + hints: Iterable[str] | None = None, + on_action: ActionCallback | None = None, + ) -> None: + """Send a notification. + + :param actions: iterable of ``(id, label)`` pairs shown as buttons. + :param hints: raw ``notify-send`` ``-h`` hint strings. + :param on_action: invoked with the chosen action id when a button is + pressed. When omitted the notification is fire-and-forget. + """ + command = ["notify-send", "-a", self._app_name] + if icon: + command += ["-i", icon] + for hint in hints or (): + command += ["-h", hint] + for action_id, label in actions or (): + command += ["-A", f"{action_id}={label}"] + command += [title, message] + + if on_action is None: + self._spawn(command, Gio.SubprocessFlags.STDOUT_SILENCE) + return + self._spawn_with_action(command, on_action) + + def _spawn_with_action(self, command: list[str], on_action: ActionCallback) -> None: + proc = self._spawn(command, Gio.SubprocessFlags.STDOUT_PIPE) + if proc is None: + return + + def on_done(process: Gio.Subprocess, task: Gio.Task) -> None: + try: + _, stdout, _ = process.communicate_utf8_finish(task) + except Exception as error: + logger.error(f"[CaptureNotifier] Failed to read action: {error}") + return + action = stdout.strip() if stdout else "" + if action: + on_action(action) + + proc.communicate_utf8_async(None, None, on_done) + + @staticmethod + def _spawn(command: list[str], flags: Gio.SubprocessFlags) -> Gio.Subprocess | None: + try: + return Gio.Subprocess.new(command, flags) + except Exception as error: + logger.error(f"[CaptureNotifier] Failed to notify: {error}") + return None diff --git a/src/utils/inhibit.py b/src/utils/inhibit.py deleted file mode 100644 index ed9eaf6..0000000 --- a/src/utils/inhibit.py +++ /dev/null @@ -1,193 +0,0 @@ -# From https://github.com/stwa/wayland-idle-inhibitor -# License: WTFPL Version 2 -import argparse -import os -import subprocess -import sys -from dataclasses import dataclass -from pathlib import Path - -from fabric.utils import logger - -# Add the project root to sys.path to allow absolute imports from 'utils' -sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) - -from signal import SIGINT, SIGTERM, signal -from threading import Event, Timer - -from pywayland.client.display import Display -from pywayland.protocol.idle_inhibit_unstable_v1.zwp_idle_inhibit_manager_v1 import ( - ZwpIdleInhibitManagerV1, -) -from pywayland.protocol.wayland.wl_compositor import WlCompositor -from pywayland.protocol.wayland.wl_registry import WlRegistryProxy -from pywayland.protocol.wayland.wl_surface import WlSurface - -from utils.functions import set_process_name - - -@dataclass -class GlobalRegistry: - surface: WlSurface | None = None - inhibit_manager: ZwpIdleInhibitManagerV1 | None = None - - -def parse_duration(duration_str: str) -> int: - """Parse duration string into seconds. - Examples: '1h', '30m', '45s', '1.5h', '1.5m', '30.5s', 'on', 'off' - """ - try: - if duration_str.lower() in ["on", "off"]: - return 0 - elif duration_str.endswith("h"): - return int(float(duration_str[:-1]) * 3600) - elif duration_str.endswith("m"): - return int(float(duration_str[:-1]) * 60) - elif duration_str.endswith("s"): - return int(float(duration_str[:-1])) - else: - return int(duration_str) - except ValueError as err: - raise ValueError( - "Invalid duration format. Use '1h', '30m', '45s', 'on', 'off', etc." - ) from err - - -def kill_existing_inhibit_processes(): - """Kill any existing modus-inhibit processes.""" - try: - # Find all modus-inhibit processes except the current one - output = subprocess.check_output(["pgrep", "-f", "modus-inhibit"], text=True) - pids = output.strip().split("\n") - current_pid = str(os.getpid()) - - killed_count = 0 - for pid in pids: - pid = pid.strip() - if pid and pid != current_pid: - try: - subprocess.run(["kill", pid], check=True) - killed_count += 1 - except subprocess.CalledProcessError: - pass # Process might have already exited - - if killed_count > 0: - print(f"Stopped {killed_count} existing inhibit process(es)") - else: - print("No existing inhibit processes were running") - return killed_count - - except subprocess.CalledProcessError: - # No processes found - logger.error("No existing inhibit processes were running") - return 0 - except Exception as e: - logger.error(f"Error stopping existing processes: {e}") - return 0 - - -def handle_registry_global( - wl_registry: WlRegistryProxy, id_num: int, iface_name: str, version: int -) -> None: - global_registry: GlobalRegistry = wl_registry.user_data or GlobalRegistry() - - if iface_name == "wl_compositor": - compositor = wl_registry.bind(id_num, WlCompositor, version) - global_registry.surface = compositor.create_surface() # type: ignore - elif iface_name == "zwp_idle_inhibit_manager_v1": - global_registry.inhibit_manager = wl_registry.bind( - id_num, ZwpIdleInhibitManagerV1, version - ) - - -def main() -> None: - parser = argparse.ArgumentParser( - description="Inhibit system idle for a specified duration" - ) - parser.add_argument( - "duration", - nargs="?", - default="0", - help='Duration to inhibit (e.g., "1h", "30m", "45s", "on", "off"). Use "on" for indefinite, "off" to stop.', - ) - args = parser.parse_args() - - # Handle "off" argument - kill existing processes and exit - if args.duration.lower() == "off": - kill_existing_inhibit_processes() - sys.exit(0) - - done = Event() - signal(SIGINT, lambda _, __: done.set()) - signal(SIGTERM, lambda _, __: done.set()) - - global_registry = GlobalRegistry() - - try: - display = Display() - display.connect() - - registry = display.get_registry() # type: ignore - registry.user_data = global_registry - registry.dispatcher["global"] = handle_registry_global - - def shutdown() -> None: - display.dispatch() - display.roundtrip() - display.disconnect() - - display.dispatch() - display.roundtrip() - - if global_registry.surface is None: - print("Error: Failed to create Wayland surface.") - shutdown() - sys.exit(1) - - if global_registry.inhibit_manager is None: - print("Error: Your Wayland compositor does not support idle inhibition.") - print("This usually means either:") - print( - "1. Your compositor (like Hyprland) doesn't support the idle-inhibit protocol" - ) - print("2. The protocol is not enabled in your compositor") - print("\nFor Hyprland, you can enable it by adding to your config:") - print("misc:disable_autoreload = true") - print("misc:enable_wayland_protocols = true") - shutdown() - sys.exit(1) - - inhibitor = global_registry.inhibit_manager.create_inhibitor( # type: ignore - global_registry.surface - ) - - display.dispatch() - display.roundtrip() - - duration = parse_duration(args.duration) - if duration > 0: - print(f"Inhibiting idle for {args.duration}...") - # Set up timer to release inhibition - timer = Timer(duration, lambda: done.set()) - timer.start() - else: - print("Inhibiting idle indefinitely...") - - done.wait() - print("Shutting down...") - - inhibitor.destroy() - if duration > 0: - timer.cancel() - - shutdown() - - except Exception as e: - logger.error(f"Error: {e!s}") - logger.error("Make sure you're running this under a Wayland session.") - sys.exit(1) - - -if __name__ == "__main__": - set_process_name("modus-inhibit") - main() diff --git a/src/utils/roam.py b/src/utils/roam.py index f35bf98..bfeef21 100644 --- a/src/utils/roam.py +++ b/src/utils/roam.py @@ -14,6 +14,8 @@ def __getattr__(name): "notification_service", "audio_service", "screen_capture_service", + "screenshot_service", + "screen_recorder_service", ): return getattr(_mod(), name) raise AttributeError(f"module {__name__!r} has no attribute {name!r}") diff --git a/src/window/screencapture/main.py b/src/window/screencapture/main.py index 1dc8517..fd1b551 100644 --- a/src/window/screencapture/main.py +++ b/src/window/screencapture/main.py @@ -219,7 +219,11 @@ class ScreenCaptureWindow(Window): screen_capture_service.record( "active", use_audio=use_audio, show_cursor=show_cursor ) - elif tool == "record-window" or tool == "record-region": + elif tool == "record-window": + screen_capture_service.record( + "window", use_audio=use_audio, show_cursor=show_cursor + ) + elif tool == "record-region": screen_capture_service.record( "selection", use_audio=use_audio, show_cursor=show_cursor ) @@ -323,7 +327,7 @@ class ScreenCaptureWindow(Window): target_map = { "region": "selection", "fullscreen": "active", - "selectwindow": "selection", + "selectwindow": "window", } use_audio = ( self._options_menu.get_mic_enabled() if self._options_menu else False