fuxino/xidle-inhibit
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Keep inhibits across daemon restarts

Browse Source 
Save held inhibits to $XDG_RUNTIME_DIR/xautolock-inhibit.json and restore them on start if the previous instance stopped within --restore-window (default 30s) on the same bus and the senders are still connected. The file is touched every 10s while inhibits are held, so crashes are timed too.
This commit is contained in:
Daniele Fucini
2026-10-17 16:49:40 +02:00
parent a36e700df6
commit 265966ffe9
3 changed files with 144 additions and 17 deletions
Download Patch File Download Diff File Expand all files Collapse all files
xautolock-inhibit
+139 -14
View File
@@ -6,13 +6,15 @@ While an inhibit is active, X screen blanking is also paused (like mpv does).
import argparse
import ctypes
import itertools
import json
import os
import signal
import subprocess
import sys
import time
from dataclasses import dataclass
from datetime import timedelta
from typing import Any
from gi.repository import Gio, GLib
@@ -26,6 +28,8 @@ BUS_NAME = "org.freedesktop.ScreenSaver"
OBJECT_PATHS = ("/org/freedesktop/ScreenSaver", "/ScreenSaver") # Vivaldi uses the first, Zen the second
EXIT_NAME_TAKEN = 2
ACTIVITY_QUIET = 300 # seconds without SimulateUserActivity that end a logged burst
STATE_FILE = "xautolock-inhibit.json" # in $XDG_RUNTIME_DIR
STATE_HEARTBEAT = 10 # seconds between state file touches while inhibits are held
Cookie = tuple[str, str, str, bool] # sender, application_name, reason, counts
@@ -120,12 +124,19 @@ class ActivityBurst:
class Inhibitor:
def __init__(self, ignored_reasons: set[str], release_delay: int, blanking: XBlanking) -> None:
def __init__(
self, ignored_reasons: set[str], release_delay: int, blanking: XBlanking,
state_path: str | None, restore_window: int,
) -> None:
self.ignored_reasons = ignored_reasons
self.blanking = blanking
self.release_delay = release_delay
self.state_path = state_path
self.restore_window = restore_window
self.state_ready = False # set once this instance owns the name and may write the state file
self.heartbeat_timer = 0
self.cookies: dict[int, Cookie] = {}
self.next_cookie = itertools.count(1)
self.last_cookie = 0
self.proc_names: dict[str, str] = {} # sender -> "comm[pid]", for logging only
self.activity: dict[str, ActivityBurst] = {} # sender -> SimulateUserActivity burst
self.disabled = False
@@ -171,20 +182,121 @@ class Inhibitor:
if self.release_timer:
GLib.source_remove(self.release_timer)
self.release_timer = 0
self.save(stopped=True)
if self.disabled:
self.unblock()
# --- state file -------------------------------------------------------
def start(self) -> None:
"""Take over inhibits saved by a previous instance, then set xautolock to match."""
self.restore()
self.state_ready = True
self.save()
if self.active_count():
self.block()
else:
# Start from a known state, in case a previous instance died while disabled.
xautolock("-enable")
def restore(self) -> None:
if not self.state_path:
log("XDG_RUNTIME_DIR is not set; inhibits will not survive a restart")
return
assert self.conn is not None
try:
with open(self.state_path) as f:
state = json.load(f)
age = time.time() - (state["stopped_at"] or os.stat(self.state_path).st_mtime)
bus_id = state["bus_id"]
last_cookie = int(state["last_cookie"])
saved = [(int(c), s, a, r) for c, s, a, r in state["cookies"]]
except FileNotFoundError:
return
except (OSError, ValueError, KeyError, TypeError) as e:
log(f"Ignoring unreadable state file {self.state_path}: {e}")
return
if bus_id != self.conn.get_guid():
# Unique names like :1.42 are reused by a new bus, so they would mean other clients.
if saved:
log(f"Not restoring {len(saved)} inhibit(s): saved on another session bus")
return
# Keep numbering after old cookies, so a late UnInhibit can't hit a new one.
self.last_cookie = last_cookie
if not saved:
return
if age > self.restore_window:
# An UnInhibit sent while no instance ran is lost; don't trust an old file.
log(f"Not restoring {len(saved)} inhibit(s): saved {age:.0f}s ago, "
f"over the {self.restore_window}s window")
return
for cookie, sender, app, reason in saved:
if not self.has_owner(sender):
log(f"Not restoring {cookie}: '{app}' '{reason}' ({sender} left the bus)")
continue
counts = reason not in self.ignored_reasons
self.cookies[cookie] = (sender, app, reason, counts)
note = "" if counts else " (reason ignored, does not block locking)"
log(f"Restored {cookie}: '{app}' '{reason}' from {self.proc_name(sender)}{note}")
def save(self, stopped: bool = False) -> None:
"""Write the held inhibits atomically; while any are held, touch the file regularly.
The touches keep the mtime recent, so after a crash the next instance can tell how
long ago it happened. A clean stop records the time in `stopped_at` instead.
"""
if not self.state_ready or not self.state_path:
return
assert self.conn is not None
state = {
"bus_id": self.conn.get_guid(),
"last_cookie": self.last_cookie,
"stopped_at": time.time() if stopped else None,
"cookies": [[c, s, a, r] for c, (s, a, r, _) in self.cookies.items()],
}
tmp = f"{self.state_path}.tmp"
try:
with open(tmp, "w") as f:
json.dump(state, f)
os.replace(tmp, self.state_path)
except OSError as e:
log(f"Cannot write state file {self.state_path}: {e}")
if self.cookies and not stopped:
if not self.heartbeat_timer:
self.heartbeat_timer = GLib.timeout_add_seconds(STATE_HEARTBEAT, self.on_heartbeat)
elif self.heartbeat_timer:
GLib.source_remove(self.heartbeat_timer)
self.heartbeat_timer = 0
def on_heartbeat(self) -> bool:
assert self.state_path is not None
try:
os.utime(self.state_path)
except OSError as e:
log(f"Cannot touch state file {self.state_path}: {e}")
return GLib.SOURCE_CONTINUE
# --- D-Bus ------------------------------------------------------------
def bus_call(self, method: str, name: str, reply_type: str) -> Any:
"""Call an org.freedesktop.DBus method that takes a bus name; return its single result."""
assert self.conn is not None # set in on_bus_acquired, before any method call
return self.conn.call_sync(
"org.freedesktop.DBus", "/org/freedesktop/DBus", "org.freedesktop.DBus",
method, GLib.Variant("(s)", (name,)),
GLib.VariantType(reply_type), Gio.DBusCallFlags.NONE, 1000, None,
).unpack()[0]
def has_owner(self, name: str) -> bool:
try:
return bool(self.bus_call("NameHasOwner", name, "(b)"))
except GLib.Error:
return False
def proc_name(self, sender: str) -> str:
if sender not in self.proc_names:
assert self.conn is not None # set in on_bus_acquired, before any method call
try:
pid = self.conn.call_sync(
"org.freedesktop.DBus", "/org/freedesktop/DBus", "org.freedesktop.DBus",
"GetConnectionUnixProcessID", GLib.Variant("(s)", (sender,)),
GLib.VariantType("(u)"), Gio.DBusCallFlags.NONE, 1000, None,
).unpack()[0]
pid = self.bus_call("GetConnectionUnixProcessID", sender, "(u)")
with open(f"/proc/{pid}/comm") as f:
self.proc_names[sender] = f"{f.read().strip()}[{pid}]"
except (GLib.Error, OSError):
@@ -192,11 +304,13 @@ class Inhibitor:
return f"{sender} {self.proc_names[sender]}"
def inhibit(self, sender: str, app: str, reason: str) -> int:
cookie = next(self.next_cookie)
self.last_cookie += 1
cookie = self.last_cookie
counts = reason not in self.ignored_reasons
self.cookies[cookie] = (sender, app, reason, counts)
note = "" if counts else " (reason ignored, does not block locking)"
log(f"Inhibit {cookie}: '{app}' '{reason}' from {self.proc_name(sender)}{note}")
self.save()
self.update()
return cookie
@@ -208,6 +322,7 @@ class Inhibitor:
del self.cookies[cookie]
_, app, reason, _ = entry
log(f"UnInhibit {cookie}: '{app}' '{reason}'")
self.save()
self.update()
def simulate_activity(self, sender: str) -> None:
@@ -247,6 +362,7 @@ class Inhibitor:
self.end_activity(sender)
self.proc_names.pop(sender, None)
if gone:
self.save()
self.update()
def on_method_call(
@@ -304,6 +420,11 @@ def main() -> None:
"-d", "--release-delay", type=int, default=5, metavar="SECONDS",
help="wait this long after the last release before re-enabling xautolock (default: 5)",
)
parser.add_argument(
"--restore-window", type=int, default=30, metavar="SECONDS",
help="take over inhibits from a previous instance that stopped at most this long ago "
"(default: 30; 0 disables)",
)
args = parser.parse_args()
ignored: set[str]
@@ -311,7 +432,11 @@ def main() -> None:
ignored = set()
else:
ignored = set(args.ignore_reason or ["Playing audio"])
inhibitor = Inhibitor(ignored, max(args.release_delay, 0), XBlanking())
runtime_dir = os.environ.get("XDG_RUNTIME_DIR")
state_path = os.path.join(runtime_dir, STATE_FILE) if runtime_dir else None
inhibitor = Inhibitor(
ignored, max(args.release_delay, 0), XBlanking(), state_path, max(args.restore_window, 0),
)
loop = GLib.MainLoop()
exit_code = 0
acquired = False
@@ -320,9 +445,9 @@ def main() -> None:
nonlocal acquired
acquired = True
log(f"Owning {name}; ignored reasons: {sorted(ignored) or 'none'}")
# Start from a known state, in case a previous instance died while disabled.
# Only after owning the name, so a second instance can't undo an active inhibit.
xautolock("-enable")
# Only after owning the name, so a second instance can't undo an active inhibit
# or overwrite the state file.
inhibitor.start()
def on_name_lost(conn: Gio.DBusConnection | None, name: str) -> None:
nonlocal exit_code