diff --git a/.gitignore b/.gitignore index ceb2b98..762fbdc 100644 --- a/.gitignore +++ b/.gitignore @@ -1 +1,4 @@ CLAUDE.md +__pycache__/ +.mypy_cache/ +.ruff_cache/ diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 0000000..659fa47 --- /dev/null +++ b/pyproject.toml @@ -0,0 +1,10 @@ +[tool.ruff] +line-length = 120 +extend-include = ["xautolock-inhibit"] + +[tool.ruff.lint] +select = ["E", "F", "W", "I", "B", "UP"] + +[tool.mypy] +files = ["xautolock-inhibit"] +strict = true diff --git a/xautolock-inhibit b/xautolock-inhibit index 3066fb2..754271a 100755 --- a/xautolock-inhibit +++ b/xautolock-inhibit @@ -17,7 +17,7 @@ from datetime import timedelta from gi.repository import Gio, GLib try: - from gi.repository import GLibUnix # GLib >= 2.80 + from gi.repository import GLibUnix # type: ignore[attr-defined] # GLib >= 2.80, not in pygobject-stubs signal_add = GLibUnix.signal_add except ImportError: signal_add = GLib.unix_signal_add @@ -27,6 +27,8 @@ OBJECT_PATHS = ("/org/freedesktop/ScreenSaver", "/ScreenSaver") # Vivaldi uses EXIT_NAME_TAKEN = 2 ACTIVITY_QUIET = 300 # seconds without SimulateUserActivity that end a logged burst +Cookie = tuple[str, str, str, bool] # sender, application_name, reason, counts + INTROSPECTION = """ @@ -58,11 +60,11 @@ INTROSPECTION = """ """ -def log(msg): +def log(msg: str) -> None: print(msg, flush=True) -def xautolock(arg): +def xautolock(arg: str) -> None: try: subprocess.run(["xautolock", arg], check=True, timeout=10) log(f"xautolock {arg}") @@ -77,8 +79,8 @@ class XBlanking: pause by itself when this client releases it or its connection closes (exit, crash). """ - def __init__(self): - self.dpy = None + def __init__(self) -> None: + self.dpy: int | None = None try: self.x11 = ctypes.CDLL("libX11.so.6") self.xss = ctypes.CDLL("libXss.so.1") @@ -102,7 +104,7 @@ class XBlanking: return self.dpy = dpy - def suspend(self, on): + def suspend(self, on: bool) -> None: if self.dpy: self.xss.XScreenSaverSuspend(self.dpy, int(on)) self.x11.XFlush(self.dpy) @@ -118,24 +120,24 @@ class ActivityBurst: class Inhibitor: - def __init__(self, ignored_reasons, release_delay, blanking): + def __init__(self, ignored_reasons: set[str], release_delay: int, blanking: XBlanking) -> None: self.ignored_reasons = ignored_reasons self.blanking = blanking self.release_delay = release_delay - self.cookies = {} # cookie -> (sender, application_name, reason, counts) + self.cookies: dict[int, Cookie] = {} self.next_cookie = itertools.count(1) - self.proc_names = {} # sender -> "comm[pid]", for logging only + 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 = None + self.conn: Gio.DBusConnection | None = None # --- lock state ------------------------------------------------------- - def active_count(self): + def active_count(self) -> int: return sum(1 for *_, counts in self.cookies.values() if counts) - def update(self): + def update(self) -> None: """Run xautolock only on 0->1 and 1->0 transitions; delay re-enabling.""" active = self.active_count() if active and self.release_timer: @@ -147,18 +149,18 @@ class Inhibitor: # 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): + 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): + def block(self) -> None: xautolock("-disable") self.blanking.suspend(True) self.disabled = True - def unblock(self): + def unblock(self) -> None: xautolock("-enable") self.blanking.suspend(False) self.disabled = False @@ -174,8 +176,9 @@ class Inhibitor: # --- D-Bus ------------------------------------------------------------ - def proc_name(self, sender): + 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", @@ -188,7 +191,7 @@ class Inhibitor: self.proc_names[sender] = "?" return f"{sender} {self.proc_names[sender]}" - def inhibit(self, sender, app, reason): + def inhibit(self, sender: str, app: str, reason: str) -> int: cookie = next(self.next_cookie) counts = reason not in self.ignored_reasons self.cookies[cookie] = (sender, app, reason, counts) @@ -197,7 +200,7 @@ class Inhibitor: self.update() return cookie - def uninhibit(self, sender, 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") @@ -246,7 +249,10 @@ class Inhibitor: if gone: self.update() - def on_method_call(self, conn, sender, path, iface, method, params, invocation): + 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),))) @@ -265,12 +271,15 @@ class Inhibitor: 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): + def on_name_owner_changed( + self, conn: Gio.DBusConnection, sender: str, 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, 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: @@ -281,7 +290,7 @@ class Inhibitor: ) -def main(): +def main() -> None: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument( "-i", "--ignore-reason", action="append", metavar="REASON", @@ -297,6 +306,7 @@ def main(): ) args = parser.parse_args() + ignored: set[str] if args.count_all: ignored = set() else: @@ -306,7 +316,7 @@ def main(): exit_code = 0 acquired = False - def on_name_acquired(conn, name): + def on_name_acquired(conn: Gio.DBusConnection, name: str) -> None: nonlocal acquired acquired = True log(f"Owning {name}; ignored reasons: {sorted(ignored) or 'none'}") @@ -314,7 +324,7 @@ def main(): # Only after owning the name, so a second instance can't undo an active inhibit. xautolock("-enable") - def on_name_lost(conn, name): + 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") @@ -327,7 +337,7 @@ def main(): exit_code = EXIT_NAME_TAKEN loop.quit() - def on_signal(): + def on_signal() -> bool: log("Signal received, exiting") loop.quit() return GLib.SOURCE_REMOVE