sankalp/Modus
1
0 Fork 0
Code Issues 2 Pull Requests Actions Packages Projects Releases 1 Wiki Activity

refactor: extract reusable utils, remove screencapture facade

Browse Source 
- Add ManagedTimer to replace repeated source_remove/timeout_add boilerplate
- Add EvdevLockService base class for numlock/capslock (cut 154→28 lines)
- Add BaseNotchIndicator for 5 near-identical indicator classes (cut 404→171 lines)
- Add load_pixbuf_thumbnail and resolve_monitor to utils
- Delete screencapture.py facade; callers use Screenshot/ScreenRecorder directly
This commit is contained in:
sankalp committed 2026-07-15 10:01:40 +05:45
1 parent ebe1238b3d
commit e63a2c04a5
16 files changed
+401 -612
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

src/services/capslock.py
+12 -72
View File
@@ -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
src/services/modus.py
-6
View File
@@ -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
src/services/numlock.py
+12 -72
View File
@@ -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
src/services/screencapture.py
-116
View File
@@ -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}")
src/services/screenrecorder.py
+2 -10
View File
@@ -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)
src/services/screenshot.py
+2 -10
View File
@@ -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:
src/utils/evdev_lock.py
+90
View File
@@ -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()
src/utils/functions.py
+13
View File
@@ -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.
src/utils/gtk_utils.py
+26 -1
View File
@@ -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
src/utils/roam.py
-1
View File
@@ -13,7 +13,6 @@ def __getattr__(name):
"modus_service",
"notification_service",
"audio_service",
"screen_capture_service",
"screenshot_service",
"screen_recorder_service",
):
src/utils/timer.py
+56
View File
@@ -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
src/window/panel/notch/indicator_base.py
+80
View File
@@ -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."""
src/window/panel/notch/indicators.py
+46 -266
View File
@@ -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()
src/window/panel/notch/notch.py
+5 -5
View File
@@ -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()
src/window/panel/notch/recording.py
+3 -5
View File
@@ -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()
src/window/screencapture/main.py
+54 -48
View File
@@ -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,