tgrosinger
a9f97b078a
swayidle obeys idle inhibitors, so Waybar's "stay awake" toggle stopped idle events entirely and work time kept accumulating. Add break-idle, a stdlib-only Wayland client that uses ext-idle-notify-v1's get_input_idle_notification (v2), which reports only real user input. swayidle is still used for before-sleep and after-resume.
99 lines
3.8 KiB
Python
Executable File
99 lines
3.8 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""
|
|
break-idle - Report idle/resume to break-event, ignoring idle inhibitors
|
|
|
|
swayidle obeys idle inhibitors (e.g. Waybar's idle_inhibitor), which stops
|
|
activity tracking entirely while "stay awake" is on. This speaks the Wayland
|
|
wire protocol directly and uses ext-idle-notify-v1's version 2
|
|
get_input_idle_notification, which only considers real user input.
|
|
|
|
Usage: break-idle [timeout_seconds] (default 5)
|
|
"""
|
|
|
|
import os
|
|
import socket
|
|
import struct
|
|
import subprocess
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
DISPLAY_ID, REGISTRY_ID, CALLBACK_ID, SEAT_ID, NOTIFIER_ID, NOTIFICATION_ID = range(1, 7)
|
|
BREAK_EVENT = Path(__file__).resolve().parent / "break-event"
|
|
|
|
|
|
def encode_string(s: str) -> bytes:
|
|
data = s.encode() + b"\0"
|
|
return struct.pack("=I", len(data)) + data + b"\0" * (-len(data) % 4)
|
|
|
|
|
|
def decode_string(payload: bytes, offset: int) -> tuple[str, int]:
|
|
(length,) = struct.unpack_from("=I", payload, offset)
|
|
start = offset + 4
|
|
return payload[start:start + length - 1].decode(), start + length + (-length % 4)
|
|
|
|
|
|
def send(sock: socket.socket, obj: int, opcode: int, args: bytes) -> None:
|
|
sock.sendall(struct.pack("=II", obj, (8 + len(args)) << 16 | opcode) + args)
|
|
|
|
|
|
def messages(sock: socket.socket):
|
|
buf = b""
|
|
while True:
|
|
chunk = sock.recv(4096)
|
|
if not chunk:
|
|
sys.exit("break-idle: compositor closed the connection")
|
|
buf += chunk
|
|
while len(buf) >= 8:
|
|
obj, size_op = struct.unpack_from("=II", buf)
|
|
size = size_op >> 16
|
|
if len(buf) < size:
|
|
break
|
|
yield obj, size_op & 0xFFFF, buf[8:size]
|
|
buf = buf[size:]
|
|
|
|
|
|
def main() -> None:
|
|
timeout = int(sys.argv[1]) if len(sys.argv) > 1 else 5
|
|
|
|
display = os.environ.get("WAYLAND_DISPLAY", "wayland-0")
|
|
path = display if display.startswith("/") else os.path.join(os.environ["XDG_RUNTIME_DIR"], display)
|
|
sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
|
|
sock.connect(path)
|
|
|
|
send(sock, DISPLAY_ID, 1, struct.pack("=I", REGISTRY_ID)) # wl_display.get_registry
|
|
send(sock, DISPLAY_ID, 0, struct.pack("=I", CALLBACK_ID)) # wl_display.sync
|
|
|
|
globals_: dict[str, tuple[int, int]] = {}
|
|
for obj, opcode, payload in messages(sock):
|
|
if obj == DISPLAY_ID and opcode == 0: # wl_display.error
|
|
sys.exit(f"break-idle: protocol error: {decode_string(payload, 8)[0]}")
|
|
if obj == REGISTRY_ID and opcode == 0: # wl_registry.global
|
|
(name,) = struct.unpack_from("=I", payload)
|
|
interface, offset = decode_string(payload, 4)
|
|
(version,) = struct.unpack_from("=I", payload, offset)
|
|
globals_.setdefault(interface, (name, version))
|
|
elif obj == CALLBACK_ID: # wl_callback.done: all globals received
|
|
break
|
|
|
|
notifier = globals_.get("ext_idle_notifier_v1")
|
|
if "wl_seat" not in globals_ or not notifier or notifier[1] < 2:
|
|
sys.exit("break-idle: compositor lacks ext_idle_notifier_v1 version 2 "
|
|
"(needed to ignore idle inhibitors)")
|
|
|
|
for interface, obj, version in (("wl_seat", SEAT_ID, 1), ("ext_idle_notifier_v1", NOTIFIER_ID, 2)):
|
|
args = struct.pack("=I", globals_[interface][0]) + encode_string(interface) + struct.pack("=II", version, obj)
|
|
send(sock, REGISTRY_ID, 0, args) # wl_registry.bind
|
|
|
|
# ext_idle_notifier_v1.get_input_idle_notification(id, timeout_ms, seat)
|
|
send(sock, NOTIFIER_ID, 2, struct.pack("=III", NOTIFICATION_ID, timeout * 1000, SEAT_ID))
|
|
|
|
for obj, opcode, payload in messages(sock):
|
|
if obj == DISPLAY_ID and opcode == 0:
|
|
sys.exit(f"break-idle: protocol error: {decode_string(payload, 8)[0]}")
|
|
if obj == NOTIFICATION_ID: # 0 = idled, 1 = resumed
|
|
subprocess.run([BREAK_EVENT, "idle_start" if opcode == 0 else "idle_end"])
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|