fix(recording): track backend subprocess for active state

active shelled out to pidof on every read, blocking the main loop and
missing our own recorder handle. Track the spawned Gio.Subprocess for
an exact answer and cache the external pidof probe for 2s.

Resets the cache at ownership boundaries (start/stop).
This commit is contained in:
sankalp committed 2026-08-03 07:39:30 +05:45
1 parent 8318bdf0fd
commit 195143664b
2 files changed
+43 -3

No files matched your search

+28 -3
View File
@@ -24,6 +24,12 @@ from utils.functions import is_app_running, resolve_monitor
_RECORDER_PROCESSES = ("wf-recorder", "gpu-screen-recorder")
# How long a cached external-recorder probe stays valid. Detecting a recorder
# spawned outside this service requires a ``pidof`` call; caching the result
# means reading the ``active`` property never spawns a subprocess on the main
# loop more than once per this window.
_ACTIVE_PROBE_TTL = 2.0
class ScreenRecorder(Service):
"""Record the screen through a pluggable :class:`CaptureBackend`."""
@@ -69,6 +75,8 @@ class ScreenRecorder(Service):
self._output_file: str | None = None
self._start_time: float | None = None
self._tick_source_id: int | None = None
self._external_recorder_active = False
self._external_probe_time = 0.0
self.recordings_dir = Path.home() / "Videos" / "Recordings"
self.recordings_dir.mkdir(parents=True, exist_ok=True)
@@ -121,10 +129,21 @@ class ScreenRecorder(Service):
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.
A recording started through this service is tracked exactly via its
subprocess handle. Recordings started outside modus are detected with
a short-lived cached ``pidof`` probe, so reading this property never
blocks the main loop more than once per ``_ACTIVE_PROBE_TTL`` seconds.
"""
return any(is_app_running(proc) for proc in _RECORDER_PROCESSES)
ours = self._backend.is_recorder_running()
if ours is not None:
return ours
now = time.monotonic()
if now - self._external_probe_time >= _ACTIVE_PROBE_TTL:
self._external_recorder_active = any(
is_app_running(proc) for proc in _RECORDER_PROCESSES
)
self._external_probe_time = now
return self._external_recorder_active
# -- public API --------------------------------------------------------
@@ -220,6 +239,10 @@ class ScreenRecorder(Service):
self._recording = True
self._paused = False
self._start_time = time.monotonic()
# This service now owns the recorder: drop the external cache so the
# next ``active`` read resolves exactly from the tracked subprocess.
self._external_recorder_active = False
self._external_probe_time = 0.0
self._start_tick()
self.notify("output-file")
self.notify("recording")
@@ -230,6 +253,8 @@ class ScreenRecorder(Service):
self._recording = False
self._paused = False
self._start_time = None
# Force a fresh external probe on the next ``active`` read.
self._external_probe_time = 0.0
self.notify("recording")
self.notify("busy")
self.notify("elapsed-seconds")
+15
View File
@@ -147,6 +147,7 @@ class HyprlandCaptureBackend(CaptureBackend):
tool: find_binary(tool) is not None
for tool in ("hyprshot", "grim", "slurp", self._RECORDER)
}
self._recorder_proc: Gio.Subprocess | None = None
@property
def capabilities(self) -> BackendCapabilities:
@@ -163,6 +164,18 @@ class HyprlandCaptureBackend(CaptureBackend):
"""Return whether a wrapped tool is available on this system."""
return self._tools.get(tool, False)
def is_recorder_running(self) -> bool | None:
"""Return whether the recorder spawned by this backend is still alive.
``None`` means no recorder was started through this backend, so the
caller must fall back to external detection. Cheap: reads the already
tracked subprocess instead of shelling out.
"""
proc = self._recorder_proc
if proc is None:
return None
return not proc.get_if_exited()
# screenshots
def screenshot(self, request: ScreenshotRequest, on_result: ResultCallback) -> bool:
@@ -322,10 +335,12 @@ class HyprlandCaptureBackend(CaptureBackend):
if proc is None:
on_result(CaptureResult.failure(f"failed to launch {self._RECORDER}"))
return False
self._recorder_proc = proc
on_result(CaptureResult.ok(str(request.output_file)))
return True
def stop_recording(self) -> bool:
self._recorder_proc = None
kill_process(self._RECORDER)
return True