diff --git a/src/services/capslock.py b/src/services/capslock.py index 1d72345..f72b052 100644 --- a/src/services/capslock.py +++ b/src/services/capslock.py @@ -1,77 +1,17 @@ -from pathlib import Path - -from fabric.core.service import Property, Service, Signal -from fabric.utils import logger - -from utils.gtk_utils import EvdevLEDMonitor - -# Discover CapsLock LED device -capslock_leds = list(Path("/sys/class/leds").glob("input*::capslock")) -capslock_device = capslock_leds[0] if capslock_leds else None +from utils.evdev_lock import EvdevLockService -class CapsLock(Service): - instance = None +class CapsLock(EvdevLockService): + """Service for monitoring CapsLock LED state via evdev.""" + + _LED_GLOB = "input*::capslock" + _LED_INDEX = 1 # LED_CAPSL + _SERVICE_NAME = "CapsLock" + + _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._last_state = None - self._monitoring_started = False - - if capslock_device is None: - logger.warning("CapsLock device not found, CapsLock service disabled") - return - - # LED_CAPSL = 1 - self._monitor = EvdevLEDMonitor( - self.capslock_led_path, 1, self._on_state_changed - ) - - def _on_state_changed(self, is_on: bool): - if is_on != self._last_state: - self._last_state = is_on - self.emit("state_changed", is_on) - - def stop(self): - if hasattr(self, "_monitor"): - self._monitor.stop() - - 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._monitor._last_state = self._last_state - self._monitor.start() - - 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) as e: - logger.warning( - f"[capslock] return bool(int(brightness_path.read_text().strip())) failed: {e}" - ) - 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() + if CapsLock._instance is None: + CapsLock._instance = CapsLock() + return CapsLock._instance diff --git a/src/services/modus.py b/src/services/modus.py index e0fe7bd..1d3b195 100644 --- a/src/services/modus.py +++ b/src/services/modus.py @@ -37,12 +37,6 @@ def __getattr__(name): _inst = None setattr(mod, name, _inst) return _inst - if name == "screen_capture_service": - from services.screencapture import ScreenCapture - - _inst = ScreenCapture() - setattr(mod, name, _inst) - return _inst if name == "screenshot_service": from services.screenshot import Screenshot diff --git a/src/services/numlock.py b/src/services/numlock.py index 0f92017..4a809eb 100644 --- a/src/services/numlock.py +++ b/src/services/numlock.py @@ -1,77 +1,17 @@ -from pathlib import Path - -from fabric.core.service import Property, Service, Signal -from fabric.utils import logger - -from utils.gtk_utils import EvdevLEDMonitor - -# Discover NumLock LED device -numlock_leds = list(Path("/sys/class/leds").glob("input*::numlock")) -numlock_device = numlock_leds[0] if numlock_leds else None +from utils.evdev_lock import EvdevLockService -class NumLock(Service): - instance = None +class NumLock(EvdevLockService): + """Service for monitoring NumLock LED state via evdev.""" + + _LED_GLOB = "input*::numlock" + _LED_INDEX = 0 # LED_NUML + _SERVICE_NAME = "NumLock" + + _instance = None @staticmethod def get_initial(): - if NumLock.instance is None: - NumLock.instance = NumLock() - - return NumLock.instance - - @Signal - def state_changed(self, is_on: bool) -> None: - """Signal emitted when NumLock state changes.""" - - def __init__(self, **kwargs): - super().__init__(**kwargs) - - self.numlock_led_path = numlock_device - self._last_state = None - self._monitoring_started = False - - if numlock_device is None: - logger.warning("NumLock device not found, NumLock service disabled") - return - - # LED_NUML = 0 - self._monitor = EvdevLEDMonitor( - self.numlock_led_path, 0, self._on_state_changed - ) - - def _on_state_changed(self, is_on: bool): - if is_on != self._last_state: - self._last_state = is_on - self.emit("state_changed", is_on) - - def stop(self): - if hasattr(self, "_monitor"): - self._monitor.stop() - - def _ensure_monitoring_started(self): - if not self._monitoring_started and self.numlock_led_path: - self._monitoring_started = True - # Set initial state before starting monitoring - self._last_state = self._get_current_state() - self._monitor._last_state = self._last_state - self._monitor.start() - - def _get_current_state(self) -> bool: - if not self.numlock_led_path: - return False - brightness_path = self.numlock_led_path / "brightness" - if brightness_path.exists(): - try: - return bool(int(brightness_path.read_text().strip())) - except (ValueError, OSError) as e: - logger.warning( - f"[numlock] return bool(int(brightness_path.read_text().strip())) failed: {e}" - ) - 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() + if NumLock._instance is None: + NumLock._instance = NumLock() + return NumLock._instance diff --git a/src/services/screencapture.py b/src/services/screencapture.py deleted file mode 100644 index a4b1a14..0000000 --- a/src/services/screencapture.py +++ /dev/null @@ -1,116 +0,0 @@ -"""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.core.service import Property, Service, Signal - -from services.screenrecorder import ScreenRecorder -from services.screenshot import Screenshot - - -class ScreenCapture(Service): - """Facade delegating to :class:`Screenshot` and :class:`ScreenRecorder`.""" - - @Signal - def screenshot_taken(self, path: str) -> None: - """Emitted after a screenshot attempt (``None`` path on cancel/fail).""" - - @Signal - def recording_started(self, path: str) -> None: - """Emitted when a recording starts.""" - - @Signal - def recording_stopped(self, path: str) -> None: - """Emitted when a recording stops.""" - - _instance: "ScreenCapture | None" = None - - def __new__(cls) -> "ScreenCapture": - if cls._instance is None: - cls._instance = super().__new__(cls) - return cls._instance - - def __init__(self, **kwargs) -> None: - if getattr(self, "_initialized", False): - return - super().__init__(**kwargs) - self._initialized = True - - self._screenshot = Screenshot() - self._recorder = ScreenRecorder() - - self.screenshots_dir = self._screenshot.screenshots_dir - self.recordings_dir = self._recorder.recordings_dir - - 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) - - @property - def screenshot_service(self) -> Screenshot: - """The underlying screenshot service.""" - return self._screenshot - - @property - def recorder_service(self) -> ScreenRecorder: - """The underlying recorder service.""" - return self._recorder - - def screenshot( - self, - target: str = "region", - output_dir: str | Path | None = None, - show_cursor: bool = False, - ) -> bool: - """Take a screenshot. See :meth:`Screenshot.screenshot`.""" - return self._screenshot.screenshot( - target, output_dir=output_dir, show_cursor=show_cursor - ) - - 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 stop_recording(self) -> bool: - """Stop the active recording.""" - return self._recorder.stop() - - @Property(bool, "readable", default_value=False) - 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: str): - if name == "screen_capture_service": - import sys - - return sys.modules["services.modus"].screen_capture_service - raise AttributeError(f"module {__name__!r} has no attribute {name!r}") diff --git a/src/services/screenrecorder.py b/src/services/screenrecorder.py index f1922b5..4f8ad21 100644 --- a/src/services/screenrecorder.py +++ b/src/services/screenrecorder.py @@ -19,7 +19,7 @@ from shared.capture import ( HyprlandCaptureBackend, RecordingRequest, ) -from utils.functions import is_app_running +from utils.functions import is_app_running, resolve_monitor _RECORDER_PROCESSES = ("wf-recorder", "gpu-screen-recorder") @@ -201,17 +201,9 @@ class ScreenRecorder(Service): output_file=self.recordings_dir / f"{timestamp}.mkv", use_audio=use_audio, show_cursor=show_cursor, - monitor=self._resolve_monitor(target), + monitor=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) diff --git a/src/services/screenshot.py b/src/services/screenshot.py index 5382c92..a480e31 100644 --- a/src/services/screenshot.py +++ b/src/services/screenshot.py @@ -19,7 +19,7 @@ from shared.capture import ( HyprlandCaptureBackend, ScreenshotRequest, ) -from utils.functions import find_binary +from utils.functions import find_binary, resolve_monitor class Screenshot(Service): @@ -117,7 +117,7 @@ class Screenshot(Service): ) -> 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) + monitor = resolve_monitor(target) return ScreenshotRequest( target=target, save_path=save_dir / f"{timestamp}.png", @@ -125,14 +125,6 @@ class Screenshot(Service): 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: diff --git a/src/utils/evdev_lock.py b/src/utils/evdev_lock.py new file mode 100644 index 0000000..74ed2ec --- /dev/null +++ b/src/utils/evdev_lock.py @@ -0,0 +1,90 @@ +"""Parameterised evdev LED-lock service base. + +NumLock and CapsLock services are structurally identical; only the LED +glob pattern, LED index, and service name differ. This module provides a +single base class that both services can inherit from. +""" + +from pathlib import Path + +from fabric.core.service import Property, Service, Signal +from fabric.utils import logger + +from utils.gtk_utils import EvdevLEDMonitor + + +class EvdevLockService(Service): + """Base service for keyboard lock-key monitoring via evdev LEDs. + + Subclasses must set **class-level** attributes: + + * ``_LED_GLOB`` – sysfs glob for the LED device + (e.g. ``"input*::numlock"``). + * ``_LED_INDEX`` – integer LED code passed to :class:`EvdevLEDMonitor` + (0 = NUML, 1 = CAPSL). + * ``_SERVICE_NAME`` – human-readable name for log messages. + """ + + _LED_GLOB: str = "" + _LED_INDEX: int = 0 + _SERVICE_NAME: str = "EvdevLock" + + _instance = None + + @Signal + def state_changed(self, is_on: bool) -> None: + """Emitted when the lock-key state changes.""" + + def __new__(cls, *args, **kwargs): + if cls._instance is None: + cls._instance = super().__new__(cls) + return cls._instance + + def __init__(self, **kwargs): + super().__init__(**kwargs) + + leds = list(Path("/sys/class/leds").glob(self._LED_GLOB)) + self._led_path = leds[0] if leds else None + self._last_state = None + self._monitoring_started = False + + if self._led_path is None: + logger.warning(f"[{self._SERVICE_NAME}] Device not found, service disabled") + return + + self._monitor = EvdevLEDMonitor( + self._led_path, self._LED_INDEX, self._on_state_changed + ) + + def _on_state_changed(self, is_on: bool): + if is_on != self._last_state: + self._last_state = is_on + self.emit("state_changed", is_on) + + def stop(self): + if hasattr(self, "_monitor"): + self._monitor.stop() + + def _ensure_monitoring_started(self): + if not self._monitoring_started and self._led_path: + self._monitoring_started = True + self._last_state = self._get_current_state() + self._monitor._last_state = self._last_state + self._monitor.start() + + def _get_current_state(self) -> bool: + if not self._led_path: + return False + brightness_path = self._led_path / "brightness" + if brightness_path.exists(): + try: + return bool(int(brightness_path.read_text().strip())) + except (ValueError, OSError) as e: + logger.warning(f"[{self._SERVICE_NAME}] Failed to read brightness: {e}") + return False + return False + + @Property(bool, "read-write", default_value=False) + def is_on(self) -> bool: + self._ensure_monitoring_started() + return self._get_current_state() diff --git a/src/utils/functions.py b/src/utils/functions.py index 20f7467..3e38b89 100644 --- a/src/utils/functions.py +++ b/src/utils/functions.py @@ -418,6 +418,19 @@ def run_command( return CommandResult(1, "", str(e)) +def resolve_monitor(target: str) -> str | None: + """Resolve the ``"active"`` capture target to a concrete monitor name. + + Returns ``None`` for non-active targets (the capture backend handles + region/window/output directly). + """ + if target != "active": + return None + from services.modus import get_active_monitor_name + + return get_active_monitor_name() + + def fuzzy_score(query: str, text: str) -> int: """ Fuzzy match scoring. Returns score > 0 if all query chars found in order. diff --git a/src/utils/gtk_utils.py b/src/utils/gtk_utils.py index 8838439..04311c0 100644 --- a/src/utils/gtk_utils.py +++ b/src/utils/gtk_utils.py @@ -3,7 +3,14 @@ import struct from pathlib import Path from typing import Literal -from fabric.utils import Gdk, GLib, bulk_connect, exec_shell_command_async, logger +from fabric.utils import ( + Gdk, + GdkPixbuf, + GLib, + bulk_connect, + exec_shell_command_async, + logger, +) from fabric.widgets.svg import Svg @@ -168,3 +175,21 @@ class EvdevLEDMonitor: except OSError as e: logger.warning(f"[utils] os.close(self._fd) failed: {e}") self._fd = None + + +def load_pixbuf_thumbnail( + path: str, width: int, height: int, *, fallback: str | None = None +) -> GdkPixbuf.Pixbuf | None: + """Load a pixbuf scaled to fit within *width* x *height*. + + Returns ``None`` on failure (file missing, corrupt, etc.). Pass a + *fallback* path to try that on failure too. + """ + for p in (path, fallback): + if p is None: + continue + try: + return GdkPixbuf.Pixbuf.new_from_file_at_scale(p, width, height, True) + except Exception as e: + logger.warning(f"[gtk_utils] Failed to load pixbuf from {p}: {e}") + return None diff --git a/src/utils/roam.py b/src/utils/roam.py index bfeef21..297036b 100644 --- a/src/utils/roam.py +++ b/src/utils/roam.py @@ -13,7 +13,6 @@ def __getattr__(name): "modus_service", "notification_service", "audio_service", - "screen_capture_service", "screenshot_service", "screen_recorder_service", ): diff --git a/src/utils/timer.py b/src/utils/timer.py new file mode 100644 index 0000000..347352c --- /dev/null +++ b/src/utils/timer.py @@ -0,0 +1,56 @@ +"""Managed GLib timer helper. + +Replaces the repeated schedule/cancel/cleanup pattern:: + + if self._timer_id: + GLib.source_remove(self._timer_id) + self._timer_id = GLib.timeout_add(ms, self._on_timeout) + + # in destroy(): + if self._timer_id: + GLib.source_remove(self._timer_id) + +Usage:: + + self._hide_timer = ManagedTimer() + # schedule + self._hide_timer.schedule(DISPLAY_MS, self._do_hide) + # cancel + self._hide_timer.cancel() + # cleanup in destroy() + self._hide_timer.cancel() +""" + +from fabric.utils import GLib + + +class ManagedTimer: + """A single-shot GLib timer that auto-cancels any pending source.""" + + __slots__ = ("_source_id",) + + def __init__(self): + self._source_id: int | None = None + + def schedule(self, milliseconds: int, callback) -> None: + """Schedule *callback* after *milliseconds*. + + If a timer is already pending it is cancelled first. + """ + self.cancel() + self._source_id = GLib.timeout_add(milliseconds, self._on_fire, callback) + + def _on_fire(self, callback): + self._source_id = None + callback() + return False # one-shot + + def cancel(self) -> None: + """Cancel the pending timer, if any.""" + if self._source_id is not None: + GLib.source_remove(self._source_id) + self._source_id = None + + @property + def is_pending(self) -> bool: + return self._source_id is not None diff --git a/src/window/panel/notch/indicator_base.py b/src/window/panel/notch/indicator_base.py new file mode 100644 index 0000000..eb37b3f --- /dev/null +++ b/src/window/panel/notch/indicator_base.py @@ -0,0 +1,80 @@ +from fabric.utils import logger +from fabric.widgets.box import Box +from fabric.widgets.centerbox import CenterBox +from fabric.widgets.label import Label + +from utils.gtk_utils import svg_file +from utils.timer import ManagedTimer + +DISPLAY_MS = 2000 # how long each indicator stays visible (ms) + + +class BaseNotchIndicator(Box): + """Shared timer + layout scaffolding for notch indicators. + + Subclasses override :meth:`_on_activate` (called when the indicator + should become visible) and :meth:`_on_destroy` (optional extra cleanup). + """ + + def __init__( + self, + name: str, + icon_name: str, + label_text: str, + show_cb, + hide_cb, + *, + icon_size: int = 16, + icon_first: bool = True, + **kwargs, + ): + super().__init__( + name=name, + orientation="h", + h_expand=True, + v_align="center", + **kwargs, + ) + self._show_cb = show_cb + self._hide_cb = hide_cb + self._hide_timer = ManagedTimer() + + self._icon = svg_file( + icon_name, + size=icon_size, + name="notch-indicator-icon", + v_align="center", + ) + self._label = Label( + name="notch-indicator-label", + label=label_text, + v_align="center", + ) + + if icon_first: + start, end = self._icon, self._label + else: + start, end = self._label, self._icon + + self.add( + CenterBox( + orientation="h", + h_expand=True, + start_children=start, + end_children=end, + ) + ) + + def _schedule_hide(self): + self._hide_timer.schedule(DISPLAY_MS, self._hide_cb) + + def destroy(self): + self._hide_timer.cancel() + try: + self._on_destroy() + except Exception as e: + logger.warning(f"[NotchIndicators] {self.get_name()} destroy: {e}") + super().destroy() + + def _on_destroy(self): + """Override for extra signal disconnection.""" diff --git a/src/window/panel/notch/indicators.py b/src/window/panel/notch/indicators.py index 42e878d..01df828 100644 --- a/src/window/panel/notch/indicators.py +++ b/src/window/panel/notch/indicators.py @@ -1,54 +1,25 @@ from fabric.audio import Audio from fabric.utils import GLib, logger -from fabric.widgets.box import Box -from fabric.widgets.centerbox import CenterBox -from fabric.widgets.label import Label from services.battery import Battery from services.capslock import CapsLock from services.keyboard_layout import KeyboardLayout from services.numlock import NumLock -from utils.gtk_utils import svg_file -DISPLAY_MS = 2000 # how long each indicator stays visible (ms) +from .indicator_base import BaseNotchIndicator -class KeyboardLayoutIndicator(Box): +class KeyboardLayoutIndicator(BaseNotchIndicator): def __init__(self, show_cb, hide_cb, **kwargs): + self._kbd = KeyboardLayout.get_initial() super().__init__( name="notch-kbd-layout", - orientation="h", - h_expand=True, - v_align="center", + icon_name="notch/keyboard-layout.svg", + label_text=self._kbd.current_layout.upper(), + show_cb=show_cb, + hide_cb=hide_cb, **kwargs, ) - self._show_cb = show_cb - self._hide_cb = hide_cb - self._hide_timer_id = 0 - - self._kbd = KeyboardLayout.get_initial() - - self._icon = svg_file( - "notch/keyboard-layout.svg", - size=16, - name="notch-indicator-icon", - v_align="center", - ) - self._label = Label( - name="notch-indicator-label", - label=self._kbd.current_layout.upper(), - v_align="center", - ) - - self.add( - CenterBox( - orientation="h", - h_expand=True, - start_children=self._icon, - end_children=self._label, - ) - ) - self._kbd.connect("layout_changed", self._on_layout_changed) def _on_layout_changed(self, _service, layout: str): @@ -56,199 +27,68 @@ class KeyboardLayoutIndicator(Box): self._show_cb(self) self._schedule_hide() - def _schedule_hide(self): - if self._hide_timer_id: - GLib.source_remove(self._hide_timer_id) - self._hide_timer_id = GLib.timeout_add(DISPLAY_MS, self._do_hide) - - def _do_hide(self): - self._hide_timer_id = 0 - self._hide_cb() - return False - - def destroy(self): - if self._hide_timer_id: - GLib.source_remove(self._hide_timer_id) - self._hide_timer_id = 0 - try: - self._kbd.disconnect_by_func(self._on_layout_changed) - except Exception as e: - logger.warning(f"[NotchIndicators] kbd disconnect: {e}") - super().destroy() + def _on_destroy(self): + self._kbd.disconnect_by_func(self._on_layout_changed) -# ── CapsLock ─────────────────────────────────────────────────────────────────── - - -class CapsLockIndicator(Box): +class CapsLockIndicator(BaseNotchIndicator): def __init__(self, show_cb, hide_cb, **kwargs): + self._caps = CapsLock.get_initial() super().__init__( name="notch-capslock", - orientation="h", - h_expand=True, - v_align="center", + icon_name="notch/caps-lock.svg", + label_text="On" if self._caps.is_on else "Off", + show_cb=show_cb, + hide_cb=hide_cb, **kwargs, ) - self._show_cb = show_cb - self._hide_cb = hide_cb - self._hide_timer_id = 0 - - self._caps = CapsLock.get_initial() - - self._icon = svg_file( - "notch/caps-lock.svg", - size=16, - name="notch-indicator-icon", - v_align="center", - ) - self._label = Label( - name="notch-indicator-label", - label="On" if self._caps.is_on else "Off", - v_align="center", - ) - - self.add( - CenterBox( - orientation="h", - h_expand=True, - start_children=self._icon, - end_children=self._label, - ) - ) - self._caps.connect("state_changed", self._on_state_changed) def _on_state_changed(self, _service, is_on: bool): - # Show for BOTH On and Off self._label.set_label("On" if is_on else "Off") self._show_cb(self) self._schedule_hide() - def _schedule_hide(self): - if self._hide_timer_id: - GLib.source_remove(self._hide_timer_id) - self._hide_timer_id = GLib.timeout_add(DISPLAY_MS, self._do_hide) - - def _do_hide(self): - self._hide_timer_id = 0 - self._hide_cb() - return False - - def destroy(self): - if self._hide_timer_id: - GLib.source_remove(self._hide_timer_id) - self._hide_timer_id = 0 - try: - self._caps.disconnect_by_func(self._on_state_changed) - except Exception as e: - logger.warning(f"[NotchIndicators] caps disconnect: {e}") - super().destroy() + def _on_destroy(self): + self._caps.disconnect_by_func(self._on_state_changed) -class NumLockIndicator(Box): +class NumLockIndicator(BaseNotchIndicator): def __init__(self, show_cb, hide_cb, **kwargs): + self._num = NumLock.get_initial() super().__init__( name="notch-numlock", - orientation="h", - h_expand=True, - v_align="center", + icon_name="notch/num-lock.svg", + label_text="On" if self._num.is_on else "Off", + show_cb=show_cb, + hide_cb=hide_cb, **kwargs, ) - self._show_cb = show_cb - self._hide_cb = hide_cb - self._hide_timer_id = 0 - - self._num = NumLock.get_initial() - - self._icon = svg_file( - "notch/num-lock.svg", - size=16, - name="notch-indicator-icon", - v_align="center", - ) - self._label = Label( - name="notch-indicator-label", - label="On" if self._num.is_on else "Off", - v_align="center", - ) - - self.add( - CenterBox( - orientation="h", - h_expand=True, - start_children=self._icon, - end_children=self._label, - ) - ) - - # is_on access also starts the 500ms LED poll inside the service self._num.connect("state_changed", self._on_state_changed) def _on_state_changed(self, _service, is_on: bool): - # Show for BOTH On and Off self._label.set_label("On" if is_on else "Off") self._show_cb(self) self._schedule_hide() - def _schedule_hide(self): - if self._hide_timer_id: - GLib.source_remove(self._hide_timer_id) - self._hide_timer_id = GLib.timeout_add(DISPLAY_MS, self._do_hide) - - def _do_hide(self): - self._hide_timer_id = 0 - self._hide_cb() - return False - - def destroy(self): - if self._hide_timer_id: - GLib.source_remove(self._hide_timer_id) - self._hide_timer_id = 0 - try: - self._num.disconnect_by_func(self._on_state_changed) - except Exception as e: - logger.warning(f"[NotchIndicators] numlock disconnect: {e}") - super().destroy() + def _on_destroy(self): + self._num.disconnect_by_func(self._on_state_changed) -class ChargingIndicator(Box): +class ChargingIndicator(BaseNotchIndicator): def __init__(self, show_cb, hide_cb, **kwargs): + self._battery = Battery.get_initial() super().__init__( name="notch-charging", - orientation="h", - h_expand=True, - v_align="center", + icon_name="notch/charging.svg", + label_text="Charging", + show_cb=show_cb, + hide_cb=hide_cb, + icon_size=24, + icon_first=False, **kwargs, ) - self._show_cb = show_cb - self._hide_cb = hide_cb - self._hide_timer_id = 0 - - self._battery = Battery.get_initial() - - self._label = Label( - name="notch-indicator-label", - label="Charging", - v_align="center", - ) - self._icon = svg_file( - "notch/charging.svg", - size=24, - name="notch-indicator-icon", - v_align="center", - ) - - self.add( - CenterBox( - orientation="h", - h_expand=True, - start_children=self._label, - end_children=self._icon, - ) - ) - self._battery.connect("changed", self._on_battery_changed) - GLib.timeout_add(500, self._initial_check) def _is_charger_connected(self) -> bool: @@ -268,67 +108,23 @@ class ChargingIndicator(Box): self._schedule_hide() self._last_charger_state = connected - def _schedule_hide(self): - if self._hide_timer_id: - GLib.source_remove(self._hide_timer_id) - self._hide_timer_id = GLib.timeout_add(DISPLAY_MS, self._do_hide) - - def _do_hide(self): - self._hide_timer_id = 0 - self._hide_cb() - return False - - def destroy(self): - if self._hide_timer_id: - GLib.source_remove(self._hide_timer_id) - self._hide_timer_id = 0 - try: - self._battery.disconnect_by_func(self._on_battery_changed) - except Exception as e: - logger.warning(f"[NotchIndicators] charging disconnect: {e}") - super().destroy() + def _on_destroy(self): + self._battery.disconnect_by_func(self._on_battery_changed) -# ── Microphone ───────────────────────────────────────────────────────────────── - - -class MicrophoneIndicator(Box): +class MicrophoneIndicator(BaseNotchIndicator): def __init__(self, show_cb, hide_cb, **kwargs): - super().__init__( - name="notch-mic", - orientation="h", - h_expand=True, - v_align="center", - **kwargs, - ) - self._show_cb = show_cb - self._hide_cb = hide_cb - self._hide_timer_id = 0 - self._audio = Audio() self._mic_muted = True - - self._label = Label( - name="notch-indicator-label", - label="Microphone", - v_align="center", + super().__init__( + name="notch-mic", + icon_name="notch/mic-off.svg", + label_text="Microphone", + show_cb=show_cb, + hide_cb=hide_cb, + icon_first=False, + **kwargs, ) - self._state_icon = svg_file( - "notch/mic-off.svg", - size=16, - name="notch-indicator-icon", - v_align="center", - ) - - self.add( - CenterBox( - orientation="h", - h_expand=True, - start_children=self._label, - end_children=self._state_icon, - ) - ) - self._audio.connect("notify::microphone", self._on_mic_device_changed) if self._audio.microphone: self._connect_mic_signals() @@ -337,8 +133,6 @@ class MicrophoneIndicator(Box): mic = self._audio.microphone if mic is None: return - # Disconnect from a previously-connected device if it changed, and - # avoid stacking duplicate handlers on the same device. old = getattr(self, "_mic_device", None) if old is not None and old is not mic: try: @@ -370,24 +164,11 @@ class MicrophoneIndicator(Box): def _sync_state(self): if self._audio.microphone: self._mic_muted = self._audio.microphone.muted - self._state_icon.dynamic_file( + self._icon.dynamic_file( "notch/mic-off.svg" if self._mic_muted else "notch/mic-on.svg" ) - def _schedule_hide(self): - if self._hide_timer_id: - GLib.source_remove(self._hide_timer_id) - self._hide_timer_id = GLib.timeout_add(DISPLAY_MS, self._do_hide) - - def _do_hide(self): - self._hide_timer_id = 0 - self._hide_cb() - return False - - def destroy(self): - if self._hide_timer_id: - GLib.source_remove(self._hide_timer_id) - self._hide_timer_id = 0 + def _on_destroy(self): try: self._audio.disconnect_by_func(self._on_mic_device_changed) except Exception as e: @@ -401,4 +182,3 @@ class MicrophoneIndicator(Box): target.disconnect_by_func(self._on_mic_stream_changed) except Exception as e: logger.warning(f"[NotchIndicators] mic disconnect: {e}") - super().destroy() diff --git a/src/window/panel/notch/notch.py b/src/window/panel/notch/notch.py index ab2eb5a..34cd616 100644 --- a/src/window/panel/notch/notch.py +++ b/src/window/panel/notch/notch.py @@ -4,7 +4,7 @@ from fabric.widgets.centerbox import CenterBox from fabric.widgets.shapes import Corner from fabric.widgets.stack import Stack -from services.screencapture import screen_capture_service +from services.modus import screen_recorder_service from window.controlcenter.player import get_shared_mpris_manager from window.panel.notch.indicators import ( CapsLockIndicator, @@ -77,8 +77,8 @@ class Notch(Box): self._mpris.connect("new-player", self._on_new_player) self._mpris.connect("player-vanish", self._on_player_vanish) - screen_capture_service.connect("recording-started", self._on_recording_started) - screen_capture_service.connect("recording-stopped", self._on_recording_stopped) + screen_recorder_service.connect("started", self._on_recording_started) + screen_recorder_service.connect("stopped", self._on_recording_stopped) self.notch_stack = Stack( name="panel-notch-stack", @@ -156,7 +156,7 @@ class Notch(Box): # ── Init / MPRIS / recording ────────────────────────────────────────────── def _init_state(self): - self._is_active_recording = screen_capture_service.is_recording + self._is_active_recording = screen_recorder_service.recording if self._is_active_recording: self.recording_indicator.start_timer() @@ -241,7 +241,7 @@ class Notch(Box): now = GLib.get_monotonic_time() if now - self._last_recording_check > 3_000_000: self._last_recording_check = now - if self._is_active_recording and not screen_capture_service.is_recording: + if self._is_active_recording and not screen_recorder_service.recording: self._is_active_recording = False is_recording = False self.recording_indicator.stop_timer() diff --git a/src/window/panel/notch/recording.py b/src/window/panel/notch/recording.py index a9e93cf..53716af 100644 --- a/src/window/panel/notch/recording.py +++ b/src/window/panel/notch/recording.py @@ -4,7 +4,7 @@ from fabric.widgets.button import Button from fabric.widgets.centerbox import CenterBox from fabric.widgets.label import Label -from services.screencapture import screen_capture_service +from services.modus import screen_recorder_service from utils.gtk_utils import setup_cursor_hover, svg_file @@ -71,12 +71,10 @@ class RecordingIndicator(Box): def on_stop_recording(self, *args): try: - screen_capture_service.stop_recording() + screen_recorder_service.stop() self.stop_timer() except Exception as e: - logger.warning( - f"[recording] screen_capture_service.stop_recording() failed: {e}" - ) + logger.warning(f"[recording] screen_recorder_service.stop() failed: {e}") def destroy(self): self.stop_timer() diff --git a/src/window/screencapture/main.py b/src/window/screencapture/main.py index 39c5a56..aebabea 100644 --- a/src/window/screencapture/main.py +++ b/src/window/screencapture/main.py @@ -4,7 +4,7 @@ from fabric.widgets.button import Button from fabric.widgets.label import Label from fabric.widgets.wayland import WaylandWindow as Window -from services.screencapture import screen_capture_service +from services.modus import screen_recorder_service, screenshot_service from window.screencapture.options_menu import OptionsMenu from window.screencapture.tool_button import ToolButton, ToolGroup @@ -25,7 +25,7 @@ class ScreenCaptureWindow(Window): title="modus-screencapture", anchor="bottom", layer="top", - keyboard_mode="on-demand", + keyboard_mode="exclusive", margin="0px 0px 120px 0px", visible=False, **kwargs, @@ -101,8 +101,8 @@ class ScreenCaptureWindow(Window): self._tool_group.select_id("screenshot-screen") self._tool_group.on_change(self._on_tool_changed) - screen_capture_service.connect("recording-started", self._on_recording_started) - screen_capture_service.connect("recording-stopped", self._on_recording_stopped) + screen_recorder_service.connect("started", self._on_recording_started) + screen_recorder_service.connect("stopped", self._on_recording_stopped) self.connect("key-press-event", self._on_key_press) @@ -167,7 +167,7 @@ class ScreenCaptureWindow(Window): def _on_tool_changed(self, tool_id: str): is_recording_tool = tool_id.startswith("record-") - is_active = screen_capture_service.is_recording + is_active = screen_recorder_service.recording ctx = self._action_btn.get_style_context() if is_active: @@ -182,8 +182,8 @@ class ScreenCaptureWindow(Window): def _on_action(self, *_): # Stop an active recording - if screen_capture_service.is_recording: - screen_capture_service.stop_recording() + if screen_recorder_service.recording: + screen_recorder_service.stop() return delay = 0 @@ -216,15 +216,15 @@ class ScreenCaptureWindow(Window): elif tool == "screenshot-region": self._do_screenshot("region") elif tool == "record-screen": - screen_capture_service.record( + screen_recorder_service.start( "active", use_audio=use_audio, show_cursor=show_cursor ) elif tool == "record-window": - screen_capture_service.record( + screen_recorder_service.start( "window", use_audio=use_audio, show_cursor=show_cursor ) elif tool == "record-region": - screen_capture_service.record( + screen_recorder_service.start( "selection", use_audio=use_audio, show_cursor=show_cursor ) @@ -233,10 +233,10 @@ class ScreenCaptureWindow(Window): def _do_screenshot(self, target: str): """ Take a screenshot. - - On success: service emits screenshot_taken(path) → we stay hidden. - - On cancel: service emits screenshot_taken(None) → we re-show. - - If screenshot() itself fails immediately → we re-show. - The handler is always disconnected after one fire to avoid leaks. + - On success: service emits captured(path) → we stay hidden. + - On cancel: service emits failed(reason) → we re-show. + - If screenshot() itself fails immediately → we re-show. + The handlers are always disconnected after one fire to avoid leaks. """ # Gather options output_dir = None @@ -245,52 +245,58 @@ class ScreenCaptureWindow(Window): output_dir = self._options_menu.get_save_dir() show_cursor = self._options_menu.get_show_cursor() - _handler_id = [None] + _handler_ids = [None, None] - def _on_taken(svc, path, *_): - # Disconnect immediately — one-shot - if _handler_id[0] is not None: - try: - svc.disconnect(_handler_id[0]) - except Exception as e: - logger.warning(f"[main] svc.disconnect(_handler_id[0]) failed: {e}") - _handler_id[0] = None - # Re-show only on cancel (path is None) - if not path: - GLib.timeout_add(120, self.show) + def _on_done(svc, *_args): + for hid in _handler_ids: + if hid is not None: + try: + svc.disconnect(hid) + except Exception as e: + logger.warning(f"[main] disconnect failed: {e}") + _handler_ids[0] = _handler_ids[1] = None - _handler_id[0] = screen_capture_service.connect("screenshot-taken", _on_taken) + def _on_captured(svc, path, *_): + _on_done(svc) + # success — stay hidden + + def _on_failed(svc, _reason, *_): + _on_done(svc) + GLib.timeout_add(120, self.show) + + _handler_ids[0] = screenshot_service.connect("captured", _on_captured) + _handler_ids[1] = screenshot_service.connect("failed", _on_failed) def _safe_disconnect(): - if _handler_id[0] is not None: - try: - screen_capture_service.disconnect(_handler_id[0]) - except Exception as e: - logger.warning( - f"[main] screen_capture_service.disconnect(_handler_id[0]) failed: {e}" - ) - _handler_id[0] = None + if any(hid is not None for hid in _handler_ids): + for hid in _handler_ids: + if hid is not None: + try: + screenshot_service.disconnect(hid) + except Exception as e: + logger.warning(f"[main] safe disconnect failed: {e}") + _handler_ids[0] = _handler_ids[1] = None return False # Safety net: disconnect if the service never emits (e.g. capture failed # silently), otherwise the one-shot handler would leak on every shot. GLib.timeout_add_seconds(10, _safe_disconnect) - ok = screen_capture_service.screenshot( + ok = screenshot_service.screenshot( target, output_dir=output_dir, show_cursor=show_cursor, ) if not ok: # screenshot() returned False — clean up and re-show immediately - if _handler_id[0] is not None: - try: - screen_capture_service.disconnect(_handler_id[0]) - except Exception as e: - logger.warning( - f"[main] screen_capture_service.disconnect(_handler_id[0]) failed: {e}" - ) - _handler_id[0] = None + if any(hid is not None for hid in _handler_ids): + for hid in _handler_ids: + if hid is not None: + try: + screenshot_service.disconnect(hid) + except Exception as e: + logger.warning(f"[main] disconnect failed: {e}") + _handler_ids[0] = _handler_ids[1] = None GLib.timeout_add(120, self.show) def _on_recording_started(self, *_): @@ -320,12 +326,12 @@ class ScreenCaptureWindow(Window): } self.hide() target = target_map.get(ss, ss) - screen_capture_service.screenshot(target) + screenshot_service.screenshot(target) return if sr: - if screen_capture_service.is_recording: - screen_capture_service.stop_recording() + if screen_recorder_service.recording: + screen_recorder_service.stop() return target_map = { @@ -339,7 +345,7 @@ class ScreenCaptureWindow(Window): show_cursor = ( self._options_menu.get_show_cursor() if self._options_menu else False ) - screen_capture_service.record( + screen_recorder_service.start( target_map.get(sr, sr), use_audio=use_audio, show_cursor=show_cursor,