Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
441b86448e
|
||
|
|
6548bbb232
|
||
|
|
0fc1ab2f47
|
||
|
|
ef9c8b858b
|
||
|
|
ddd1f6f40a
|
||
|
|
265966ffe9
|
||
|
|
a36e700df6
|
||
|
|
13c1afd7b6
|
||
|
|
0234456af8
|
||
|
|
51e18766a4
|
||
|
|
fa882bb0d7
|
||
|
|
af82b10d3c
|
||
|
|
1b0f2a2a66
|
@@ -1 +1,4 @@
|
||||
CLAUDE.md
|
||||
__pycache__/
|
||||
.mypy_cache/
|
||||
.ruff_cache/
|
||||
|
||||
@@ -2,26 +2,30 @@
|
||||
|
||||
# Maintainer: Daniele Fucini <dfucini [at] gmail [dot] com>
|
||||
|
||||
pkgname=xautolock-inhibit
|
||||
pkgver=0.1
|
||||
pkgname=xidle-inhibit
|
||||
pkgver=0.3
|
||||
pkgrel=1
|
||||
pkgdesc='Honor org.freedesktop.ScreenSaver inhibit requests by disabling xautolock'
|
||||
pkgdesc='Honor org.freedesktop.ScreenSaver inhibit requests by blocking xautolock or xss-lock'
|
||||
arch=('any')
|
||||
url="https://git.shouldnt.work/fuxino/${pkgname}"
|
||||
url="https://git.shouldnt.work/fuxino/$pkgname"
|
||||
license=('GPL-3.0-or-later')
|
||||
depends=('libx11'
|
||||
'libxss'
|
||||
'python'
|
||||
'python-gobject'
|
||||
'xautolock')
|
||||
'python-gobject')
|
||||
optdepends=('xautolock: xautolock backend'
|
||||
'xss-lock: xss-lock backend')
|
||||
makedepends=('git')
|
||||
source=(git+${url}?signed#tag=${pkgver})
|
||||
conflicts=('xautolock-inhibit')
|
||||
replaces=('xautolock-inhibit')
|
||||
source=("git+$url?signed#tag=$pkgver")
|
||||
validpgpkeys=('7E12BC1FF3B6EDB2CD8053EB981A2B2A3BBF5514')
|
||||
sha256sums=('SKIP')
|
||||
sha256sums=('13390705ad8e0abea5136dbe57f21b7d9509681ee8628109dd0a302497188fbf')
|
||||
|
||||
package()
|
||||
{
|
||||
cd "${srcdir}/${pkgname}"
|
||||
install -Dm755 "${pkgname}" "${pkgdir}/usr/bin/${pkgname}"
|
||||
install -Dm644 "${pkgname}.service" "${pkgdir}/usr/lib/systemd/user/${pkgname}.service"
|
||||
cd "$srcdir"/"$pkgname"
|
||||
install -Dm755 "$pkgname" "$pkgdir"/usr/bin/"$pkgname"
|
||||
install -Dm644 "$pkgname".service "$pkgdir"/usr/lib/systemd/user/"$pkgname".service
|
||||
install -Dm644 README.md "$pkgdir"/usr/share/doc/"$pkgname"/README.md
|
||||
}
|
||||
|
||||
@@ -0,0 +1,147 @@
|
||||
# xidle-inhibit
|
||||
|
||||
Makes screen locking with `xautolock` or `xss-lock` respect the "don't lock the screen" requests that apps send, on X11 sessions without a desktop environment (for example a plain window manager like xmonad).
|
||||
|
||||
Browsers, Steam, gamemode and other apps ask the desktop not to lock or blank the screen while a video or game is running, through the D-Bus service `org.freedesktop.ScreenSaver`. Without a desktop environment nothing provides that service, so the requests are dropped and the screen locks in the middle of a video. xidle-inhibit provides it:
|
||||
|
||||
- While at least one request is active, it pauses X screen blanking and DPMS, and with xautolock also runs `xautolock -disable`. xss-lock locks when the X screensaver starts, so pausing it is enough.
|
||||
- When the last request is released, it waits a few seconds, then resumes blanking (and runs `xautolock -enable`).
|
||||
|
||||
Your locker setup stays as it is; xidle-inhibit only holds it off.
|
||||
|
||||
Formerly `xautolock-inhibit`, which supported only xautolock.
|
||||
|
||||
## How it works
|
||||
|
||||
- Owns `org.freedesktop.ScreenSaver` on the session bus at both `/org/freedesktop/ScreenSaver` and `/ScreenSaver` (apps differ on the path), and implements `Inhibit`, `UnInhibit` and `Lock`.
|
||||
- Tracks requests per D-Bus client. If a client disconnects without releasing them (for example a browser crash), its requests are released.
|
||||
- Blocks and unblocks only when the number of active requests goes from 0 to 1 or back to 0. Unblocking waits a few seconds (`--release-delay`), because browsers send short bursts of requests when a video starts.
|
||||
- Pauses X blanking with `XScreenSaverSuspend`, the same mechanism mpv uses, instead of changing `xset` settings. The X server ends the pause by itself if the daemon exits or crashes.
|
||||
- Requests whose reason is on an ignore list (default: `Playing audio`) are accepted and logged, but don't block locking. Firefox-based browsers send one for audio playback, which would otherwise keep the screen unlocked while music plays with nobody at the computer.
|
||||
- Keeps the requests it holds in `$XDG_RUNTIME_DIR/xidle-inhibit.json`, so a restarted or crashed daemon takes them over. They are taken over only if the previous daemon stopped at most 30 seconds ago (`--restore-window`) on the same session bus, and only for clients still connected. A request released while no daemon was running can't be seen, so an old file isn't trusted.
|
||||
- With xautolock, it re-enables xautolock on exit (including SIGTERM). The systemd unit also runs `xidle-inhibit --cleanup` when it stops, which does the same, as a fallback.
|
||||
- `Lock` runs `xautolock -locknow`, or `loginctl lock-session` with xss-lock.
|
||||
- If another program (a desktop environment) already owns `org.freedesktop.ScreenSaver`, it exits with status 2 instead of taking over.
|
||||
|
||||
## Requirements
|
||||
|
||||
- `xautolock` or `xss-lock`, already running in the session
|
||||
- `python`, `python-gobject`
|
||||
- `libx11`, `libxss`
|
||||
- For `xidle-dim`: a compositor, such as picom
|
||||
|
||||
## Installation
|
||||
|
||||
Build and install the package with the included PKGBUILD, which builds the signed release tag set in `pkgver`:
|
||||
|
||||
```
|
||||
makepkg -si
|
||||
```
|
||||
|
||||
Or, with aurutils, build it into a local repository with `aur build` and install it with pacman.
|
||||
|
||||
Then start it from your window manager's startup, after the locker. For xmonad with xautolock:
|
||||
|
||||
```haskell
|
||||
spawnOnce "xautolock -time 10 -locker slock -detectsleep"
|
||||
spawnOnce "systemctl --user start xidle-inhibit"
|
||||
```
|
||||
|
||||
With xss-lock, the lock timeout is the X screensaver timeout (`xset s TIMEOUT CYCLE`). This dims the screen with `xidle-dim` at 9:50 and locks at 10:00:
|
||||
|
||||
```haskell
|
||||
spawnOnce "xset s 590 10"
|
||||
spawnOnce "xss-lock -n xidle-dim -- slock"
|
||||
spawnOnce "systemctl --user start xidle-inhibit"
|
||||
```
|
||||
|
||||
Without dimming, use `xset s 600 0` and `xss-lock -- slock`. Start xss-lock from the window manager, not from a terminal: terminals like kitty run commands outside the login session, and xss-lock then ignores `loginctl lock-session`.
|
||||
|
||||
The unit has no `[Install]` section on purpose. Enabling it at login would also start it in a desktop environment, or before X is up; starting it from the window manager limits it to that session. The user service manager needs `DISPLAY` (and `XAUTHORITY`) in its environment, which `/etc/X11/xinit/xinitrc.d/50-systemd-user.sh` sets up for `startx` sessions.
|
||||
|
||||
### Upgrading from xautolock-inhibit
|
||||
|
||||
The package replaces `xautolock-inhibit`. After upgrading, change the `systemctl --user start` line in your window manager's startup, then switch the running daemon:
|
||||
|
||||
```
|
||||
systemctl --user stop xautolock-inhibit
|
||||
systemctl --user start xidle-inhibit
|
||||
```
|
||||
|
||||
Requests held by the old daemon are taken over if the new one starts within `--restore-window`.
|
||||
|
||||
## Configuration
|
||||
|
||||
The backend is set in `~/.config/xidle-inhibit/config` (`$XDG_CONFIG_HOME`):
|
||||
|
||||
```
|
||||
backend = xss-lock
|
||||
```
|
||||
|
||||
Values are `xautolock` and `xss-lock`. Without the file, xidle-inhibit uses whichever of the two is installed, and xautolock if both are. Restart the service after changing it.
|
||||
|
||||
## Usage
|
||||
|
||||
```
|
||||
xidle-inhibit [-i REASON] [--count-all] [-d SECONDS] [--restore-window SECONDS] [--cleanup]
|
||||
```
|
||||
|
||||
- `-i`, `--ignore-reason REASON`: request reason that doesn't block locking. Can be repeated; replaces the default (`Playing audio`).
|
||||
- `--count-all`: every request blocks locking; no reasons are ignored.
|
||||
- `-d`, `--release-delay SECONDS`: wait this long after the last release before unblocking locking (default 5).
|
||||
- `--restore-window SECONDS`: take over requests from a previous daemon that stopped at most this long ago (default 30; 0 disables).
|
||||
- `--cleanup`: run `xautolock -enable` if xautolock is the backend, then exit. Used by the unit's `ExecStopPost`.
|
||||
|
||||
To change the options used by the service:
|
||||
|
||||
```
|
||||
systemctl --user edit xidle-inhibit.service
|
||||
```
|
||||
|
||||
```ini
|
||||
[Service]
|
||||
ExecStart=
|
||||
ExecStart=/usr/bin/xidle-inhibit --count-all
|
||||
```
|
||||
|
||||
Every request, release and locker command is logged to the journal:
|
||||
|
||||
```
|
||||
journalctl --user -u xidle-inhibit
|
||||
```
|
||||
|
||||
### Dimming before the lock
|
||||
|
||||
`xidle-dim` is an xss-lock notifier (`-n`). It fades a black window over the screen until xss-lock stops it, on user activity or when the locker starts. It needs a compositor such as picom and does nothing without one. Because it doesn't change gamma, it works alongside redshift.
|
||||
|
||||
```
|
||||
xidle-dim [--opacity 0-1] [--fade SECONDS]
|
||||
```
|
||||
|
||||
- `--opacity`: final darkness (default 0.6).
|
||||
- `--fade`: fade-in time (default 1.5).
|
||||
|
||||
Options go inside the notifier command: `xss-lock -n "xidle-dim --opacity 0.8" -- slock`.
|
||||
|
||||
## App notes
|
||||
|
||||
- **Sandboxed apps** may be blocked from reaching the service. For a firejail profile that filters D-Bus (`dbus-user filter`, as the Firefox-based profiles do), allow it in the app's `.local` file, e.g. `zen-browser.local`:
|
||||
|
||||
```
|
||||
dbus-user.talk org.freedesktop.ScreenSaver
|
||||
```
|
||||
|
||||
- **Games**: Steam's overlay and gamemode (`inhibit_screensaver=1`) send requests on their own.
|
||||
- **scrcpy** only sends a request when started with `--disable-screensaver`.
|
||||
- **mpv** talks to the X server directly and doesn't need this daemon.
|
||||
|
||||
## Limitations
|
||||
|
||||
- A release sent while the daemon is down is lost: if the daemon is back within `--restore-window`, that request stays active until its app exits. After a longer stop, all held requests are dropped and apps don't send them again.
|
||||
- With xss-lock, requests block only the idle timeout: suspending, or `loginctl lock-session` from elsewhere, still locks.
|
||||
- `SimulateUserActivity` is accepted but does nothing. Calls are logged once when a client starts sending them and once, with a count, after 5 minutes without any.
|
||||
- `GetActive` always returns `false`.
|
||||
|
||||
## License
|
||||
|
||||
GPL-3.0-or-later. See [LICENSE](LICENSE).
|
||||
@@ -0,0 +1,11 @@
|
||||
[tool.ruff]
|
||||
line-length = 120
|
||||
extend-include = ["xidle-inhibit", "xidle-dim"]
|
||||
|
||||
[tool.ruff.lint]
|
||||
select = ["E", "F", "W", "I", "B", "UP"]
|
||||
|
||||
[tool.mypy]
|
||||
# Scripts both map to __main__, so check xidle-dim separately: mypy xidle-dim
|
||||
files = ["xidle-inhibit"]
|
||||
strict = true
|
||||
@@ -1,306 +0,0 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Own org.freedesktop.ScreenSaver and turn inhibit requests into xautolock -disable/-enable.
|
||||
|
||||
While an inhibit is active, X screen blanking is also paused (like mpv does).
|
||||
"""
|
||||
|
||||
import argparse
|
||||
import ctypes
|
||||
import itertools
|
||||
import signal
|
||||
import subprocess
|
||||
import sys
|
||||
|
||||
from gi.repository import Gio, GLib
|
||||
|
||||
try:
|
||||
from gi.repository import GLibUnix # GLib >= 2.80
|
||||
signal_add = GLibUnix.signal_add
|
||||
except ImportError:
|
||||
signal_add = GLib.unix_signal_add
|
||||
|
||||
BUS_NAME = "org.freedesktop.ScreenSaver"
|
||||
OBJECT_PATHS = ("/org/freedesktop/ScreenSaver", "/ScreenSaver") # Vivaldi uses the first, Zen the second
|
||||
EXIT_NAME_TAKEN = 2
|
||||
|
||||
INTROSPECTION = """
|
||||
<node>
|
||||
<interface name="org.freedesktop.ScreenSaver">
|
||||
<method name="Inhibit">
|
||||
<arg name="application_name" type="s" direction="in"/>
|
||||
<arg name="reason_for_inhibit" type="s" direction="in"/>
|
||||
<arg name="cookie" type="u" direction="out"/>
|
||||
</method>
|
||||
<method name="UnInhibit">
|
||||
<arg name="cookie" type="u" direction="in"/>
|
||||
</method>
|
||||
<method name="SimulateUserActivity"/>
|
||||
<method name="Lock"/>
|
||||
<method name="GetActive">
|
||||
<arg type="b" direction="out"/>
|
||||
</method>
|
||||
<method name="SetActive">
|
||||
<arg type="b" direction="in"/>
|
||||
<arg type="b" direction="out"/>
|
||||
</method>
|
||||
<method name="GetActiveTime">
|
||||
<arg type="u" direction="out"/>
|
||||
</method>
|
||||
<method name="GetSessionIdleTime">
|
||||
<arg type="u" direction="out"/>
|
||||
</method>
|
||||
</interface>
|
||||
</node>
|
||||
"""
|
||||
|
||||
|
||||
def log(msg):
|
||||
print(msg, flush=True)
|
||||
|
||||
|
||||
def xautolock(arg):
|
||||
try:
|
||||
subprocess.run(["xautolock", arg], check=True, timeout=10)
|
||||
log(f"xautolock {arg}")
|
||||
except (OSError, subprocess.SubprocessError) as e:
|
||||
log(f"xautolock {arg} failed: {e}")
|
||||
|
||||
|
||||
class XBlanking:
|
||||
"""Pause X screen blanking with XScreenSaverSuspend (MIT-SCREEN-SAVER extension).
|
||||
|
||||
Unlike `xset s off -dpms` this changes no global setting: the X server ends the
|
||||
pause by itself when this client releases it or its connection closes (exit, crash).
|
||||
"""
|
||||
|
||||
def __init__(self):
|
||||
self.dpy = None
|
||||
try:
|
||||
self.x11 = ctypes.CDLL("libX11.so.6")
|
||||
self.xss = ctypes.CDLL("libXss.so.1")
|
||||
except OSError as e:
|
||||
log(f"Cannot load X11 libraries ({e}); X blanking will not be paused")
|
||||
return
|
||||
self.x11.XOpenDisplay.argtypes = [ctypes.c_char_p]
|
||||
self.x11.XOpenDisplay.restype = ctypes.c_void_p
|
||||
self.x11.XFlush.argtypes = [ctypes.c_void_p]
|
||||
self.xss.XScreenSaverQueryExtension.argtypes = [
|
||||
ctypes.c_void_p, ctypes.POINTER(ctypes.c_int), ctypes.POINTER(ctypes.c_int)]
|
||||
self.xss.XScreenSaverSuspend.argtypes = [ctypes.c_void_p, ctypes.c_int]
|
||||
|
||||
dpy = self.x11.XOpenDisplay(None)
|
||||
if not dpy:
|
||||
log("Cannot open the X display; X blanking will not be paused")
|
||||
return
|
||||
event_base, error_base = ctypes.c_int(), ctypes.c_int()
|
||||
if not self.xss.XScreenSaverQueryExtension(dpy, ctypes.byref(event_base), ctypes.byref(error_base)):
|
||||
log("X server lacks the MIT-SCREEN-SAVER extension; X blanking will not be paused")
|
||||
return
|
||||
self.dpy = dpy
|
||||
|
||||
def suspend(self, on):
|
||||
if self.dpy:
|
||||
self.xss.XScreenSaverSuspend(self.dpy, int(on))
|
||||
self.x11.XFlush(self.dpy)
|
||||
log(f"X blanking {'paused' if on else 'resumed'}")
|
||||
|
||||
|
||||
class Inhibitor:
|
||||
def __init__(self, ignored_reasons, release_delay, blanking):
|
||||
self.ignored_reasons = ignored_reasons
|
||||
self.blanking = blanking
|
||||
self.release_delay = release_delay
|
||||
self.cookies = {} # cookie -> (sender, application_name, reason, counts)
|
||||
self.next_cookie = itertools.count(1)
|
||||
self.proc_names = {} # sender -> "comm[pid]", for logging only
|
||||
self.disabled = False
|
||||
self.release_timer = 0
|
||||
self.conn = None
|
||||
|
||||
# --- lock state -------------------------------------------------------
|
||||
|
||||
def active_count(self):
|
||||
return sum(1 for *_, counts in self.cookies.values() if counts)
|
||||
|
||||
def update(self):
|
||||
"""Run xautolock only on 0->1 and 1->0 transitions; delay re-enabling."""
|
||||
active = self.active_count()
|
||||
if active and self.release_timer:
|
||||
GLib.source_remove(self.release_timer)
|
||||
self.release_timer = 0
|
||||
if active and not self.disabled:
|
||||
self.block()
|
||||
elif not active and self.disabled and not self.release_timer:
|
||||
# Browsers send quick Inhibit/UnInhibit bursts when a video starts.
|
||||
self.release_timer = GLib.timeout_add_seconds(self.release_delay, self.on_release_timeout)
|
||||
|
||||
def on_release_timeout(self):
|
||||
self.release_timer = 0
|
||||
if not self.active_count() and self.disabled:
|
||||
self.unblock()
|
||||
return GLib.SOURCE_REMOVE
|
||||
|
||||
def block(self):
|
||||
xautolock("-disable")
|
||||
self.blanking.suspend(True)
|
||||
self.disabled = True
|
||||
|
||||
def unblock(self):
|
||||
xautolock("-enable")
|
||||
self.blanking.suspend(False)
|
||||
self.disabled = False
|
||||
|
||||
def shutdown(self):
|
||||
if self.release_timer:
|
||||
GLib.source_remove(self.release_timer)
|
||||
self.release_timer = 0
|
||||
if self.disabled:
|
||||
self.unblock()
|
||||
|
||||
# --- D-Bus ------------------------------------------------------------
|
||||
|
||||
def proc_name(self, sender):
|
||||
if sender not in self.proc_names:
|
||||
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]
|
||||
with open(f"/proc/{pid}/comm") as f:
|
||||
self.proc_names[sender] = f"{f.read().strip()}[{pid}]"
|
||||
except (GLib.Error, OSError):
|
||||
self.proc_names[sender] = "?"
|
||||
return f"{sender} {self.proc_names[sender]}"
|
||||
|
||||
def inhibit(self, sender, app, reason):
|
||||
cookie = next(self.next_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.update()
|
||||
return cookie
|
||||
|
||||
def uninhibit(self, sender, cookie):
|
||||
entry = self.cookies.get(cookie)
|
||||
if entry is None or entry[0] != sender:
|
||||
log(f"UnInhibit {cookie} from {self.proc_name(sender)}: unknown cookie, ignored")
|
||||
return
|
||||
del self.cookies[cookie]
|
||||
_, app, reason, _ = entry
|
||||
log(f"UnInhibit {cookie}: '{app}' '{reason}'")
|
||||
self.update()
|
||||
|
||||
def release_sender(self, sender):
|
||||
gone = [c for c, (s, *_) in self.cookies.items() if s == sender]
|
||||
for c in gone:
|
||||
_, app, reason, _ = self.cookies.pop(c)
|
||||
log(f"Released {c}: '{app}' '{reason}' ({sender} left the bus)")
|
||||
self.proc_names.pop(sender, None)
|
||||
if gone:
|
||||
self.update()
|
||||
|
||||
def on_method_call(self, conn, sender, path, iface, method, params, invocation):
|
||||
if method == "Inhibit":
|
||||
app, reason = params.unpack()
|
||||
invocation.return_value(GLib.Variant("(u)", (self.inhibit(sender, app, reason),)))
|
||||
elif method == "UnInhibit":
|
||||
self.uninhibit(sender, params.unpack()[0])
|
||||
invocation.return_value(None)
|
||||
elif method == "SimulateUserActivity":
|
||||
log(f"SimulateUserActivity from {self.proc_name(sender)} (not implemented)")
|
||||
invocation.return_value(None)
|
||||
elif method == "Lock":
|
||||
log(f"Lock from {self.proc_name(sender)}")
|
||||
xautolock("-locknow")
|
||||
invocation.return_value(None)
|
||||
elif method in ("GetActive", "SetActive"):
|
||||
invocation.return_value(GLib.Variant("(b)", (False,)))
|
||||
elif method in ("GetActiveTime", "GetSessionIdleTime"):
|
||||
invocation.return_value(GLib.Variant("(u)", (0,)))
|
||||
|
||||
def on_name_owner_changed(self, conn, sender, path, iface, signal_name, params):
|
||||
name, _old, new = params.unpack()
|
||||
if not new and name.startswith(":"):
|
||||
self.release_sender(name)
|
||||
|
||||
def on_bus_acquired(self, conn, name):
|
||||
self.conn = conn
|
||||
iface = Gio.DBusNodeInfo.new_for_xml(INTROSPECTION).interfaces[0]
|
||||
for path in OBJECT_PATHS:
|
||||
conn.register_object(path, iface, self.on_method_call, None, None)
|
||||
conn.signal_subscribe(
|
||||
"org.freedesktop.DBus", "org.freedesktop.DBus", "NameOwnerChanged",
|
||||
"/org/freedesktop/DBus", None, Gio.DBusSignalFlags.NONE, self.on_name_owner_changed,
|
||||
)
|
||||
|
||||
|
||||
def main():
|
||||
parser = argparse.ArgumentParser(description=__doc__)
|
||||
parser.add_argument(
|
||||
"-i", "--ignore-reason", action="append", metavar="REASON",
|
||||
help="inhibit reason that does not block locking (repeatable; default: 'Playing audio')",
|
||||
)
|
||||
parser.add_argument(
|
||||
"--count-all", action="store_true",
|
||||
help="let every inhibit block locking, ignoring no reasons",
|
||||
)
|
||||
parser.add_argument(
|
||||
"-d", "--release-delay", type=int, default=5, metavar="SECONDS",
|
||||
help="wait this long after the last release before re-enabling xautolock (default: 5)",
|
||||
)
|
||||
args = parser.parse_args()
|
||||
|
||||
if args.count_all:
|
||||
ignored = set()
|
||||
else:
|
||||
ignored = set(args.ignore_reason or ["Playing audio"])
|
||||
inhibitor = Inhibitor(ignored, max(args.release_delay, 0), XBlanking())
|
||||
loop = GLib.MainLoop()
|
||||
exit_code = 0
|
||||
acquired = False
|
||||
|
||||
def on_name_acquired(conn, name):
|
||||
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")
|
||||
|
||||
def on_name_lost(conn, name):
|
||||
nonlocal exit_code
|
||||
if conn is None:
|
||||
log("Could not connect to the session bus")
|
||||
exit_code = 1
|
||||
elif acquired:
|
||||
log(f"Lost {name} (bus connection closed?)")
|
||||
exit_code = 1
|
||||
else:
|
||||
log(f"{name} is owned by another process (a desktop environment?); exiting")
|
||||
exit_code = EXIT_NAME_TAKEN
|
||||
loop.quit()
|
||||
|
||||
def on_signal():
|
||||
log("Signal received, exiting")
|
||||
loop.quit()
|
||||
return GLib.SOURCE_REMOVE
|
||||
|
||||
for sig in (signal.SIGTERM, signal.SIGINT, signal.SIGHUP):
|
||||
signal_add(GLib.PRIORITY_HIGH, sig, on_signal)
|
||||
|
||||
owner_id = Gio.bus_own_name(
|
||||
Gio.BusType.SESSION, BUS_NAME, Gio.BusNameOwnerFlags.DO_NOT_QUEUE,
|
||||
inhibitor.on_bus_acquired, on_name_acquired, on_name_lost,
|
||||
)
|
||||
try:
|
||||
loop.run()
|
||||
finally:
|
||||
inhibitor.shutdown()
|
||||
Gio.bus_unown_name(owner_id)
|
||||
sys.exit(exit_code)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
@@ -1,10 +0,0 @@
|
||||
[Unit]
|
||||
Description=Turn org.freedesktop.ScreenSaver inhibit requests into xautolock control
|
||||
|
||||
[Service]
|
||||
Type=dbus
|
||||
BusName=org.freedesktop.ScreenSaver
|
||||
ExecStart=/usr/bin/xautolock-inhibit
|
||||
ExecStopPost=-/usr/bin/xautolock -enable
|
||||
Restart=on-failure
|
||||
RestartPreventExitStatus=2
|
||||
@@ -0,0 +1,119 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Dim the screen until killed, as an xss-lock notifier: xss-lock -n xidle-dim -- slock.
|
||||
|
||||
Maps a black window over the whole screen and fades in its opacity through the compositor; gamma is left alone,
|
||||
so it doesn't clash with redshift. Without a compositor it does nothing, as the window would be solid black.
|
||||
"""
|
||||
|
||||
import argparse
|
||||
import ctypes
|
||||
import signal
|
||||
import sys
|
||||
import time
|
||||
from types import FrameType
|
||||
|
||||
CW_BACK_PIXEL = 1 << 1
|
||||
CW_OVERRIDE_REDIRECT = 1 << 9
|
||||
INPUT_OUTPUT = 1
|
||||
XA_CARDINAL = 6
|
||||
PROP_MODE_REPLACE = 0
|
||||
OPACITY_MAX = 0xFFFFFFFF
|
||||
|
||||
|
||||
class XSetWindowAttributes(ctypes.Structure):
|
||||
_fields_ = [
|
||||
("background_pixmap", ctypes.c_ulong),
|
||||
("background_pixel", ctypes.c_ulong),
|
||||
("border_pixmap", ctypes.c_ulong),
|
||||
("border_pixel", ctypes.c_ulong),
|
||||
("bit_gravity", ctypes.c_int),
|
||||
("win_gravity", ctypes.c_int),
|
||||
("backing_store", ctypes.c_int),
|
||||
("backing_planes", ctypes.c_ulong),
|
||||
("backing_pixel", ctypes.c_ulong),
|
||||
("save_under", ctypes.c_int),
|
||||
("event_mask", ctypes.c_long),
|
||||
("do_not_propagate_mask", ctypes.c_long),
|
||||
("override_redirect", ctypes.c_int),
|
||||
("colormap", ctypes.c_ulong),
|
||||
("cursor", ctypes.c_ulong),
|
||||
]
|
||||
|
||||
|
||||
def load_x11() -> ctypes.CDLL:
|
||||
x11 = ctypes.CDLL("libX11.so.6")
|
||||
p, ul, i = ctypes.c_void_p, ctypes.c_ulong, ctypes.c_int
|
||||
x11.XOpenDisplay.argtypes = [ctypes.c_char_p]
|
||||
x11.XOpenDisplay.restype = p
|
||||
for name in ("XDefaultScreen",):
|
||||
getattr(x11, name).argtypes = [p]
|
||||
for name in ("XRootWindow", "XBlackPixel"):
|
||||
getattr(x11, name).argtypes = [p, i]
|
||||
getattr(x11, name).restype = ul
|
||||
for name in ("XDisplayWidth", "XDisplayHeight"):
|
||||
getattr(x11, name).argtypes = [p, i]
|
||||
x11.XInternAtom.argtypes = [p, ctypes.c_char_p, i]
|
||||
x11.XInternAtom.restype = ul
|
||||
x11.XGetSelectionOwner.argtypes = [p, ul]
|
||||
x11.XGetSelectionOwner.restype = ul
|
||||
x11.XCreateWindow.argtypes = [p, ul, i, i, ctypes.c_uint, ctypes.c_uint, ctypes.c_uint, i, ctypes.c_uint, p, ul,
|
||||
ctypes.POINTER(XSetWindowAttributes)]
|
||||
x11.XCreateWindow.restype = ul
|
||||
x11.XChangeProperty.argtypes = [p, ul, ul, ul, i, i, p, i]
|
||||
x11.XMapRaised.argtypes = [p, ul]
|
||||
x11.XFlush.argtypes = [p]
|
||||
return x11
|
||||
|
||||
|
||||
def parse_args() -> argparse.Namespace:
|
||||
parser = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
|
||||
parser.add_argument("--opacity", type=float, default=0.6, help="final darkness, 0 to 1 (default: %(default)s)")
|
||||
parser.add_argument("--fade", type=float, default=1.5, help="fade-in time in seconds (default: %(default)s)")
|
||||
args = parser.parse_args()
|
||||
if not 0 <= args.opacity <= 1 or args.fade < 0:
|
||||
parser.error("--opacity must be between 0 and 1 and --fade not negative")
|
||||
return args
|
||||
|
||||
|
||||
def exit_on_signal(_signum: int, _frame: FrameType | None) -> None:
|
||||
sys.exit(0) # the X server removes the window when the connection closes
|
||||
|
||||
|
||||
def main() -> None:
|
||||
args = parse_args()
|
||||
# xss-lock sends SIGHUP on user activity and SIGTERM when the locker starts.
|
||||
for sig in (signal.SIGHUP, signal.SIGTERM, signal.SIGINT):
|
||||
signal.signal(sig, exit_on_signal)
|
||||
|
||||
x11 = load_x11()
|
||||
dpy = x11.XOpenDisplay(None)
|
||||
if not dpy:
|
||||
sys.exit("xidle-dim: cannot open display")
|
||||
screen = x11.XDefaultScreen(dpy)
|
||||
if not x11.XGetSelectionOwner(dpy, x11.XInternAtom(dpy, f"_NET_WM_CM_S{screen}".encode(), False)):
|
||||
print("xidle-dim: no compositor running, not dimming", file=sys.stderr)
|
||||
signal.pause() # stay alive like a notifier should, until xss-lock kills us
|
||||
|
||||
attrs = XSetWindowAttributes(background_pixel=x11.XBlackPixel(dpy, screen), override_redirect=True)
|
||||
win = x11.XCreateWindow(dpy, x11.XRootWindow(dpy, screen), 0, 0, x11.XDisplayWidth(dpy, screen),
|
||||
x11.XDisplayHeight(dpy, screen), 0, 0, INPUT_OUTPUT, None,
|
||||
CW_BACK_PIXEL | CW_OVERRIDE_REDIRECT, ctypes.byref(attrs))
|
||||
opacity_atom = x11.XInternAtom(dpy, b"_NET_WM_WINDOW_OPACITY", False)
|
||||
|
||||
def set_opacity(level: float) -> None:
|
||||
value = ctypes.c_ulong(int(level * OPACITY_MAX)) # format 32 properties are passed as longs
|
||||
x11.XChangeProperty(dpy, win, opacity_atom, XA_CARDINAL, 32, PROP_MODE_REPLACE, ctypes.byref(value), 1)
|
||||
x11.XFlush(dpy)
|
||||
|
||||
set_opacity(0)
|
||||
x11.XMapRaised(dpy, win)
|
||||
steps = max(1, int(args.fade * 30))
|
||||
for n in range(1, steps + 1):
|
||||
set_opacity(args.opacity * n / steps)
|
||||
time.sleep(args.fade / steps)
|
||||
while True:
|
||||
signal.pause()
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Executable
+551
@@ -0,0 +1,551 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Own org.freedesktop.ScreenSaver and block screen locking by xautolock or xss-lock while inhibits are held.
|
||||
|
||||
While an inhibit is active, X screen blanking is paused (like mpv does), which also keeps xss-lock from
|
||||
locking; with xautolock, it is also switched off with xautolock -disable.
|
||||
"""
|
||||
|
||||
import argparse
|
||||
import ctypes
|
||||
import json
|
||||
import os
|
||||
import shutil
|
||||
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
|
||||
|
||||
try:
|
||||
from gi.repository import GLibUnix # GLib >= 2.80
|
||||
signal_add = GLibUnix.signal_add
|
||||
except ImportError:
|
||||
signal_add = GLib.unix_signal_add
|
||||
|
||||
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 = "xidle-inhibit.json" # in $XDG_RUNTIME_DIR
|
||||
OLD_STATE_FILE = "xautolock-inhibit.json" # before the rename (0.2); read once when upgrading
|
||||
STATE_HEARTBEAT = 10 # seconds between state file touches while inhibits are held
|
||||
CONFIG_FILE = "xidle-inhibit/config" # in $XDG_CONFIG_HOME
|
||||
BACKENDS = ("xautolock", "xss-lock")
|
||||
|
||||
Cookie = tuple[str, str, str, bool] # sender, application_name, reason, counts
|
||||
|
||||
INTROSPECTION = """
|
||||
<node>
|
||||
<interface name="org.freedesktop.ScreenSaver">
|
||||
<method name="Inhibit">
|
||||
<arg name="application_name" type="s" direction="in"/>
|
||||
<arg name="reason_for_inhibit" type="s" direction="in"/>
|
||||
<arg name="cookie" type="u" direction="out"/>
|
||||
</method>
|
||||
<method name="UnInhibit">
|
||||
<arg name="cookie" type="u" direction="in"/>
|
||||
</method>
|
||||
<method name="SimulateUserActivity"/>
|
||||
<method name="Lock"/>
|
||||
<method name="GetActive">
|
||||
<arg type="b" direction="out"/>
|
||||
</method>
|
||||
<method name="SetActive">
|
||||
<arg type="b" direction="in"/>
|
||||
<arg type="b" direction="out"/>
|
||||
</method>
|
||||
<method name="GetActiveTime">
|
||||
<arg type="u" direction="out"/>
|
||||
</method>
|
||||
<method name="GetSessionIdleTime">
|
||||
<arg type="u" direction="out"/>
|
||||
</method>
|
||||
</interface>
|
||||
</node>
|
||||
"""
|
||||
|
||||
|
||||
def log(msg: str) -> None:
|
||||
print(msg, flush=True)
|
||||
|
||||
|
||||
def run(*cmd: str) -> None:
|
||||
try:
|
||||
subprocess.run(cmd, check=True, timeout=10)
|
||||
log(" ".join(cmd))
|
||||
except (OSError, subprocess.SubprocessError) as e:
|
||||
log(f"{' '.join(cmd)} failed: {e}")
|
||||
|
||||
|
||||
def xautolock(arg: str) -> None:
|
||||
run("xautolock", arg)
|
||||
|
||||
|
||||
def read_backend() -> str:
|
||||
"""Return the backend set in the config file, else the installed locker (xautolock if both are)."""
|
||||
config_home = os.environ.get("XDG_CONFIG_HOME") or os.path.expanduser("~/.config")
|
||||
path = os.path.join(config_home, CONFIG_FILE)
|
||||
try:
|
||||
with open(path) as f:
|
||||
for line in f:
|
||||
key, sep, value = line.partition("=")
|
||||
if not sep or key.strip() != "backend":
|
||||
continue
|
||||
value = value.strip().strip("\"'")
|
||||
if value in BACKENDS:
|
||||
return value
|
||||
log(f"{path}: unknown backend '{value}' (expected {' or '.join(BACKENDS)}); detecting")
|
||||
except FileNotFoundError:
|
||||
pass
|
||||
except OSError as e:
|
||||
log(f"Cannot read {path}: {e}; detecting the backend")
|
||||
if not shutil.which("xautolock") and shutil.which("xss-lock"):
|
||||
return "xss-lock"
|
||||
return "xautolock"
|
||||
|
||||
|
||||
class XBlanking:
|
||||
"""Pause X screen blanking with XScreenSaverSuspend (MIT-SCREEN-SAVER extension).
|
||||
|
||||
Unlike `xset s off -dpms` this changes no global setting: the X server ends the
|
||||
pause by itself when this client releases it or its connection closes (exit, crash).
|
||||
"""
|
||||
|
||||
def __init__(self) -> None:
|
||||
self.dpy: int | None = None
|
||||
try:
|
||||
self.x11 = ctypes.CDLL("libX11.so.6")
|
||||
self.xss = ctypes.CDLL("libXss.so.1")
|
||||
except OSError as e:
|
||||
log(f"Cannot load X11 libraries ({e}); X blanking will not be paused")
|
||||
return
|
||||
self.x11.XOpenDisplay.argtypes = [ctypes.c_char_p]
|
||||
self.x11.XOpenDisplay.restype = ctypes.c_void_p
|
||||
self.x11.XFlush.argtypes = [ctypes.c_void_p]
|
||||
self.xss.XScreenSaverQueryExtension.argtypes = [
|
||||
ctypes.c_void_p, ctypes.POINTER(ctypes.c_int), ctypes.POINTER(ctypes.c_int)]
|
||||
self.xss.XScreenSaverSuspend.argtypes = [ctypes.c_void_p, ctypes.c_int]
|
||||
|
||||
dpy = self.x11.XOpenDisplay(None)
|
||||
if not dpy:
|
||||
log("Cannot open the X display; X blanking will not be paused")
|
||||
return
|
||||
event_base, error_base = ctypes.c_int(), ctypes.c_int()
|
||||
if not self.xss.XScreenSaverQueryExtension(dpy, ctypes.byref(event_base), ctypes.byref(error_base)):
|
||||
log("X server lacks the MIT-SCREEN-SAVER extension; X blanking will not be paused")
|
||||
return
|
||||
self.dpy = dpy
|
||||
|
||||
def suspend(self, on: bool) -> None:
|
||||
if self.dpy:
|
||||
self.xss.XScreenSaverSuspend(self.dpy, int(on))
|
||||
self.x11.XFlush(self.dpy)
|
||||
log(f"X blanking {'paused' if on else 'resumed'}")
|
||||
|
||||
|
||||
@dataclass
|
||||
class ActivityBurst:
|
||||
first: float
|
||||
last: float
|
||||
calls: int = 0
|
||||
timer: int = 0
|
||||
|
||||
|
||||
class Inhibitor:
|
||||
def __init__(
|
||||
self, backend: str, ignored_reasons: set[str], release_delay: int, blanking: XBlanking,
|
||||
state_path: str | None, restore_window: int,
|
||||
) -> None:
|
||||
self.xautolock = backend == "xautolock" # else xss-lock: pausing X blanking is enough
|
||||
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.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
|
||||
self.release_timer = 0
|
||||
self.conn: Gio.DBusConnection | None = None
|
||||
|
||||
# --- lock state -------------------------------------------------------
|
||||
|
||||
def active_count(self) -> int:
|
||||
return sum(1 for *_, counts in self.cookies.values() if counts)
|
||||
|
||||
def update(self) -> None:
|
||||
"""Block or unblock only on 0->1 and 1->0 transitions; delay unblocking."""
|
||||
active = self.active_count()
|
||||
if active and self.release_timer:
|
||||
GLib.source_remove(self.release_timer)
|
||||
self.release_timer = 0
|
||||
if active and not self.disabled:
|
||||
self.block()
|
||||
elif not active and self.disabled and not self.release_timer:
|
||||
# Browsers send quick Inhibit/UnInhibit bursts when a video starts.
|
||||
self.release_timer = GLib.timeout_add_seconds(self.release_delay, self.on_release_timeout)
|
||||
|
||||
def on_release_timeout(self) -> bool:
|
||||
self.release_timer = 0
|
||||
if not self.active_count() and self.disabled:
|
||||
self.unblock()
|
||||
return GLib.SOURCE_REMOVE
|
||||
|
||||
def block(self) -> None:
|
||||
if self.xautolock:
|
||||
xautolock("-disable")
|
||||
self.blanking.suspend(True)
|
||||
self.disabled = True
|
||||
|
||||
def unblock(self) -> None:
|
||||
if self.xautolock:
|
||||
xautolock("-enable")
|
||||
self.blanking.suspend(False)
|
||||
self.disabled = False
|
||||
|
||||
def shutdown(self) -> None:
|
||||
for sender in list(self.activity):
|
||||
self.end_activity(sender)
|
||||
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 the locker to match."""
|
||||
self.restore()
|
||||
self.state_ready = True
|
||||
self.save()
|
||||
if self.state_path:
|
||||
old = os.path.join(os.path.dirname(self.state_path), OLD_STATE_FILE)
|
||||
try:
|
||||
os.remove(old)
|
||||
except FileNotFoundError:
|
||||
pass
|
||||
except OSError as e:
|
||||
log(f"Cannot remove old state file {old}: {e}")
|
||||
if self.active_count():
|
||||
self.block()
|
||||
elif self.xautolock:
|
||||
# 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
|
||||
path = self.state_path
|
||||
old = os.path.join(os.path.dirname(path), OLD_STATE_FILE)
|
||||
if not os.path.exists(path) and os.path.exists(old):
|
||||
path = old
|
||||
try:
|
||||
with open(path) as f:
|
||||
state = json.load(f)
|
||||
age = time.time() - (state["stopped_at"] or os.stat(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 {path}: {e}")
|
||||
return
|
||||
if path == old:
|
||||
log(f"Reading inhibits from {old} (xautolock-inhibit)")
|
||||
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:
|
||||
try:
|
||||
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):
|
||||
self.proc_names[sender] = "?"
|
||||
return f"{sender} {self.proc_names[sender]}"
|
||||
|
||||
def inhibit(self, sender: str, app: str, reason: str) -> int:
|
||||
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
|
||||
|
||||
def uninhibit(self, sender: str, cookie: int) -> None:
|
||||
entry = self.cookies.get(cookie)
|
||||
if entry is None or entry[0] != sender:
|
||||
log(f"UnInhibit {cookie} from {self.proc_name(sender)}: unknown cookie, ignored")
|
||||
return
|
||||
del self.cookies[cookie]
|
||||
_, app, reason, _ = entry
|
||||
log(f"UnInhibit {cookie}: '{app}' '{reason}'")
|
||||
self.save()
|
||||
self.update()
|
||||
|
||||
def simulate_activity(self, sender: str) -> None:
|
||||
"""Log a sender's calls once per burst: Steam sends thousands per hour."""
|
||||
now = time.monotonic()
|
||||
burst = self.activity.get(sender)
|
||||
if burst is None:
|
||||
log(f"SimulateUserActivity from {self.proc_name(sender)} (not implemented)")
|
||||
burst = self.activity[sender] = ActivityBurst(first=now, last=now)
|
||||
else:
|
||||
GLib.source_remove(burst.timer)
|
||||
burst.calls += 1
|
||||
burst.last = now
|
||||
burst.timer = GLib.timeout_add_seconds(ACTIVITY_QUIET, self.on_activity_quiet, sender)
|
||||
|
||||
def on_activity_quiet(self, sender: str) -> bool:
|
||||
self.activity[sender].timer = 0
|
||||
self.end_activity(sender)
|
||||
return GLib.SOURCE_REMOVE
|
||||
|
||||
def end_activity(self, sender: str) -> None:
|
||||
burst = self.activity.pop(sender, None)
|
||||
if burst is None:
|
||||
return
|
||||
if burst.timer:
|
||||
GLib.source_remove(burst.timer)
|
||||
if burst.calls == 1:
|
||||
return
|
||||
span = timedelta(seconds=round(burst.last - burst.first))
|
||||
log(f"SimulateUserActivity from {self.proc_name(sender)}: {burst.calls} calls over {span}")
|
||||
|
||||
def release_sender(self, sender: str) -> None:
|
||||
gone = [c for c, (s, *_) in self.cookies.items() if s == sender]
|
||||
for c in gone:
|
||||
_, app, reason, _ = self.cookies.pop(c)
|
||||
log(f"Released {c}: '{app}' '{reason}' ({sender} left the bus)")
|
||||
self.end_activity(sender)
|
||||
self.proc_names.pop(sender, None)
|
||||
if gone:
|
||||
self.save()
|
||||
self.update()
|
||||
|
||||
def on_method_call(
|
||||
self, conn: Gio.DBusConnection, sender: str, path: str, iface: str, method: str,
|
||||
params: GLib.Variant, invocation: Gio.DBusMethodInvocation,
|
||||
) -> None:
|
||||
if method == "Inhibit":
|
||||
app, reason = params.unpack()
|
||||
invocation.return_value(GLib.Variant("(u)", (self.inhibit(sender, app, reason),)))
|
||||
elif method == "UnInhibit":
|
||||
self.uninhibit(sender, params.unpack()[0])
|
||||
invocation.return_value(None)
|
||||
elif method == "SimulateUserActivity":
|
||||
self.simulate_activity(sender)
|
||||
invocation.return_value(None)
|
||||
elif method == "Lock":
|
||||
log(f"Lock from {self.proc_name(sender)}")
|
||||
if self.xautolock:
|
||||
xautolock("-locknow")
|
||||
else:
|
||||
run("loginctl", "lock-session")
|
||||
invocation.return_value(None)
|
||||
elif method in ("GetActive", "SetActive"):
|
||||
invocation.return_value(GLib.Variant("(b)", (False,)))
|
||||
elif method in ("GetActiveTime", "GetSessionIdleTime"):
|
||||
invocation.return_value(GLib.Variant("(u)", (0,)))
|
||||
|
||||
def on_name_owner_changed(
|
||||
self, conn: Gio.DBusConnection, sender: str | None, path: str, iface: str, signal_name: str,
|
||||
params: GLib.Variant,
|
||||
) -> None:
|
||||
name, _old, new = params.unpack()
|
||||
if not new and name.startswith(":"):
|
||||
self.release_sender(name)
|
||||
|
||||
def on_bus_acquired(self, conn: Gio.DBusConnection, name: str) -> None:
|
||||
self.conn = conn
|
||||
iface = Gio.DBusNodeInfo.new_for_xml(INTROSPECTION).interfaces[0]
|
||||
for path in OBJECT_PATHS:
|
||||
conn.register_object(path, iface, self.on_method_call, None, None)
|
||||
conn.signal_subscribe(
|
||||
"org.freedesktop.DBus", "org.freedesktop.DBus", "NameOwnerChanged",
|
||||
"/org/freedesktop/DBus", None, Gio.DBusSignalFlags.NONE, self.on_name_owner_changed,
|
||||
)
|
||||
|
||||
|
||||
def main() -> None:
|
||||
parser = argparse.ArgumentParser(description=__doc__)
|
||||
parser.add_argument(
|
||||
"-i", "--ignore-reason", action="append", metavar="REASON",
|
||||
help="inhibit reason that does not block locking (repeatable; default: 'Playing audio')",
|
||||
)
|
||||
parser.add_argument(
|
||||
"--count-all", action="store_true",
|
||||
help="let every inhibit block locking, ignoring no reasons",
|
||||
)
|
||||
parser.add_argument(
|
||||
"-d", "--release-delay", type=int, default=5, metavar="SECONDS",
|
||||
help="wait this long after the last release before unblocking locking (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)",
|
||||
)
|
||||
parser.add_argument(
|
||||
"--cleanup", action="store_true",
|
||||
help="re-enable xautolock if it is the backend, then exit (for the unit's ExecStopPost)",
|
||||
)
|
||||
args = parser.parse_args()
|
||||
|
||||
backend = read_backend()
|
||||
if args.cleanup:
|
||||
if backend == "xautolock":
|
||||
xautolock("-enable")
|
||||
return
|
||||
|
||||
ignored: set[str]
|
||||
if args.count_all:
|
||||
ignored = set()
|
||||
else:
|
||||
ignored = set(args.ignore_reason or ["Playing audio"])
|
||||
runtime_dir = os.environ.get("XDG_RUNTIME_DIR")
|
||||
state_path = os.path.join(runtime_dir, STATE_FILE) if runtime_dir else None
|
||||
blanking = XBlanking()
|
||||
if backend == "xss-lock" and not blanking.dpy:
|
||||
log("Warning: with xss-lock, inhibits block locking only through X blanking, which is unavailable")
|
||||
inhibitor = Inhibitor(
|
||||
backend, ignored, max(args.release_delay, 0), blanking, state_path, max(args.restore_window, 0),
|
||||
)
|
||||
loop = GLib.MainLoop()
|
||||
exit_code = 0
|
||||
acquired = False
|
||||
|
||||
def on_name_acquired(conn: Gio.DBusConnection, name: str) -> None:
|
||||
nonlocal acquired
|
||||
acquired = True
|
||||
log(f"Owning {name}; backend: {backend}; ignored reasons: {sorted(ignored) or 'none'}")
|
||||
# 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
|
||||
if conn is None:
|
||||
log("Could not connect to the session bus")
|
||||
exit_code = 1
|
||||
elif acquired:
|
||||
log(f"Lost {name} (bus connection closed?)")
|
||||
exit_code = 1
|
||||
else:
|
||||
log(f"{name} is owned by another process (a desktop environment?); exiting")
|
||||
exit_code = EXIT_NAME_TAKEN
|
||||
loop.quit()
|
||||
|
||||
def on_signal() -> bool:
|
||||
log("Signal received, exiting")
|
||||
loop.quit()
|
||||
return GLib.SOURCE_REMOVE
|
||||
|
||||
for sig in (signal.SIGTERM, signal.SIGINT, signal.SIGHUP):
|
||||
signal_add(GLib.PRIORITY_HIGH, sig, on_signal)
|
||||
|
||||
owner_id = Gio.bus_own_name(
|
||||
Gio.BusType.SESSION, BUS_NAME, Gio.BusNameOwnerFlags.DO_NOT_QUEUE,
|
||||
inhibitor.on_bus_acquired, on_name_acquired, on_name_lost,
|
||||
)
|
||||
try:
|
||||
loop.run()
|
||||
finally:
|
||||
inhibitor.shutdown()
|
||||
Gio.bus_unown_name(owner_id)
|
||||
sys.exit(exit_code)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
@@ -0,0 +1,10 @@
|
||||
[Unit]
|
||||
Description=Block screen locking (xautolock or xss-lock) on org.freedesktop.ScreenSaver inhibit requests
|
||||
|
||||
[Service]
|
||||
Type=dbus
|
||||
BusName=org.freedesktop.ScreenSaver
|
||||
ExecStart=/usr/bin/xidle-inhibit
|
||||
ExecStopPost=-/usr/bin/xidle-inhibit --cleanup
|
||||
Restart=on-failure
|
||||
RestartPreventExitStatus=2
|
||||
Reference in New Issue
Block a user