#!/usr/bin/env python3
"""app-timers: close apps after a set time with no activity, so reopening them needs the
(password-protected) wallet again. (This replaced Signal guard's idle timer; it works for any app.)

  app-timers --list                 app-id, minutes and "headset" if set, one per line
  app-timers --set APP-ID MINUTES   set a timer (0 = none); also what the installer window uses
  app-timers --headset APP-ID on|off   also close it ~10 s after the headset is taken off
  app-timers --status               what's running and how long you've been idle
  app-timers --run                  the watcher (the app-timers user service runs this)
  app-timers --import-signal-guard  take over Signal guard's settings (Signal's installer does)

No activity means the headset isn't worn (its proximity sensor) and there's no keyboard or mouse
input (read from /dev/input, so it counts in both the desktop and VR sessions). The Frame's VR
controllers aren't input devices there: while you use them, wearing the headset is what counts.
Time the Frame spends asleep counts too. Apps are found by the systemd scope they run in
(app-flatpak-<id>-….scope, app-<id>-….scope), however they were started, else by their
program, and asked to quit (SIGTERM, which Electron apps handle as a normal quit).
Timers live in ~/.config/frame-tweaks/app-timers.conf: "app-id = minutes [headset]" (0: never).
"""
import glob
import os
import re
import select
import shlex
import shutil
import signal
import struct
import subprocess
import sys
import time
from pathlib import Path

sys.path.insert(0, os.environ.get("FRAME_TWEAKS_PYLIB") or str(Path.home() / ".local/share/frame-tweaks/i18n"))   # (lib/common.sh install_i18n)
try:
    from i18n import _, N_, ngettext
except ImportError:                       # (not installed by Frame Tweaks: English)
    def _(text):
        return text

    def ngettext(singular, plural, n):
        return singular if n == 1 else plural
    N_ = _

CONF = Path.home() / ".config/frame-tweaks/app-timers.conf"
SIGNAL_CONF = Path.home() / ".config/signal-guard.conf"     # Signal guard's, read once to migrate
SIGNAL_ID = "org.signal.Signal"
STATE = Path(f"/run/user/{os.getuid()}/app-timers.idle")
CGROUP_ROOT = Path("/sys/fs/cgroup")
IGNORE_DEVICES = re.compile(r"^(als\d+)")      # sensors that report constantly by themselves
EV_KEY, EV_REL, EV_ABS = 1, 2, 3
EVENT = "llHHi"
EVENT_SIZE = struct.calcsize(EVENT)
# Headset proximity sensor (vcnl4000), as documented by curiousjtuber/steam-frame-utils.
PROX = glob.glob("/sys/bus/iio/devices/iio:device*/in_proximity_raw")
HEADSET_OFF_GRACE = 10          # seconds off-head before closing


# ---- settings ------------------------------------------------------------------------------
# "app-id = minutes [headset]": an app id is everything before " = " (a menu entry's name can have
# spaces, "My App.desktop").
CONF_LINE = re.compile(r"\s*(.+?)\s*=\s*(\d+(?:\.\d+)?)\s*(headset)?\s*$")


def read_conf(never_too=False):
    """app-id -> {"minutes": float, "headset": bool}; never_too: also the apps set to Never
    (minutes 0, kept so that Signal's installer doesn't give it its 15 minutes again)."""
    timers = {}
    if CONF.exists():
        for line in CONF.read_text().splitlines():
            m = None if line.lstrip().startswith("#") else CONF_LINE.match(line)
            if m and (never_too or float(m[2]) > 0):
                timers[m[1]] = {"minutes": float(m[2]), "headset": bool(m[3])}
    return timers


def write_conf(timers):
    CONF.parent.mkdir(parents=True, exist_ok=True)
    CONF.write_text("# app-id = minutes without activity before the app is closed (0: never) [headset: also\n"
                    "# when the headset is taken off]. Set in the installer window (Wallet apps) or app-timers.\n"
                    + "".join(f"{a} = {t['minutes']:g}{' headset' if t['headset'] else ''}\n"
                              for a, t in sorted(timers.items())))


def set_timer(app_id, minutes):
    timers = read_conf(never_too=True)
    timers[app_id] = {"minutes": minutes, "headset": minutes > 0 and timers.get(app_id, {}).get("headset", False)}
    write_conf(timers)


def set_headset(app_id, on):
    timers = read_conf(never_too=True)
    if not timers.get(app_id, {}).get("minutes"):
        sys.exit(_("{app_id} has no timer: set one first").format(app_id=app_id))
    timers[app_id]["headset"] = on
    write_conf(timers)


def import_signal_guard():
    """Signal guard's IDLE_MINUTES / LOCK_ON_HEADSET_OFF become Signal's timer, once: not again
    once Signal has a time, Never included (it ran at every Signal reinstall and update, and
    turned Never back into 15 minutes)."""
    timers = read_conf(never_too=True)
    if SIGNAL_ID in timers:
        return
    text = SIGNAL_CONF.read_text() if SIGNAL_CONF.exists() else ""
    m = re.search(r"^\s*IDLE_MINUTES\s*=\s*(\d+(?:\.\d+)?)", text, re.M)
    timers[SIGNAL_ID] = {"minutes": float(m[1]) if m else 15.0,
                         "headset": bool(re.search(r"^\s*LOCK_ON_HEADSET_OFF\s*=\s*1", text, re.M))}
    write_conf(timers)


# ---- finding an app's processes --------------------------------------------------------------
def scope_pattern(app_id):
    # Flatpak: app-flatpak-<id>-<pid>.scope. Launched by Steam or the desktop: app-<id>-<n>.scope,
    # or app-<launcher>-<id>@<random>.service (KDE).
    return re.compile(rf"/app-(?:flatpak-|[\w.]+-)?{re.escape(app_id)}(?:-\d+\.scope|@\w*\.service)$")


def app_cgroups(app_id):
    """cgroup directories the app is running in (empty when it isn't running)."""
    pat, found = scope_pattern(app_id), set()
    for d in Path("/proc").iterdir():
        if not d.name.isdigit():
            continue
        try:
            cg = (d / "cgroup").read_text().strip().split("::", 1)[-1]
        except OSError:
            continue
        if pat.search(cg):
            found.add(CGROUP_ROOT / cg.lstrip("/"))
    return found


# Host programs whose menu command is a launcher script, not the program itself.
KNOWN_PROGRAMS = {"com.anthropic.Claude": str(Path.home() / ".local/opt/claude-desktop/claude-desktop")}
WRAPPERS = ("wallet-gate", "frame-outer-bus", "signal-guard", "--", "env")   # (signal-guard: older entries)
# Shells and interpreters also run your session's scripts, terminals and tools (sh -c …, python3 …):
# an app started through one is found by its scope only, never by that program.
INTERPRETERS = re.compile(r"(?:ba|da|z|k|mk|fi|tc|c)?sh|busybox|python[\d.]*|perl[\d.]*|ruby[\d.]*|"
                          r"node(?:js)?|java|lua[\d.]*|php[\d.]*|tclsh[\d.]*|wish[\d.]*")


def host_program(app_id):
    """The program a non-Flatpak app runs (from its menu entry), or None."""
    if app_id in KNOWN_PROGRAMS:
        return KNOWN_PROGRAMS[app_id]
    for d in (Path.home() / ".local/share/applications", Path("/usr/share/applications")):
        f = d / f"{app_id}.desktop"
        if f.exists():
            m = re.search(r"^Exec=(.+)$", f.read_text(errors="replace"), re.M)
            try:
                args = shlex.split(m[1]) if m else []
            except ValueError:
                return None
            while args and (os.path.basename(args[0]) in WRAPPERS or "=" in args[0]):
                args = args[1:]
            if not args or os.path.basename(args[0]) == "flatpak":
                return None                 # Flatpaks are found by their scope
            prog = shutil.which(os.path.expanduser(args[0]))
            if not prog or INTERPRETERS.fullmatch(os.path.basename(os.path.realpath(prog))):
                return None
            return os.path.realpath(prog)
    return None


def host_main_pids(app_id):
    """Apps opened from desktop mode share the desktop's cgroup: find the app's main process by
    its program instead (one whose parent isn't the same program: Electron's helpers are)."""
    prog = host_program(app_id)
    if not prog:
        return []
    exe = {}
    for d in Path("/proc").iterdir():
        if d.name.isdigit():
            try:
                exe[int(d.name)] = (os.readlink(d / "exe"), int((d / "stat").read_text().rsplit(")", 1)[1].split()[1]))
            except (OSError, ValueError, IndexError):
                pass
    return [pid for pid, (path, ppid) in exe.items()
            if path == prog and exe.get(ppid, ("", 0))[0] != prog]


def app_pids(app_id):
    pids = []
    for cg in app_cgroups(app_id):
        try:
            pids += [int(p) for p in (cg / "cgroup.procs").read_text().split()]
        except OSError:
            pass
    return pids or host_main_pids(app_id)


def quit_app(app_id, why):
    pids = app_pids(app_id)
    for pid in pids:
        try:
            os.kill(pid, signal.SIGTERM)
        except OSError:
            pass
    if pids:
        subprocess.run(["notify-send", "-a", _("App timers"), _("{app} closed").format(app=app_name(app_id)),
                        why], stderr=subprocess.DEVNULL)
    return bool(pids)


def app_name(app_id):
    for d in (Path.home() / ".local/share/applications", Path("/var/lib/flatpak/exports/share/applications"),
              Path.home() / ".local/share/flatpak/exports/share/applications", Path("/usr/share/applications")):
        f = d / f"{app_id}.desktop"
        if f.exists():
            m = re.search(r"^Name=(.+)$", f.read_text(errors="replace"), re.M)
            if m:
                return m[1].strip()
    return app_id


# ---- activity ------------------------------------------------------------------------------
def input_devices():
    devs = []
    for blk in Path("/proc/bus/input/devices").read_text().split("\n\n"):
        name = re.search(r'N: Name="([^"]*)"', blk)
        ev = re.search(r"H: Handlers=.*\b(event\d+)", blk)
        if name and ev and not IGNORE_DEVICES.match(name[1]):
            devs.append(f"/dev/input/{ev[1]}")
    return devs


def has_user_input(data):
    for i in range(0, len(data) - EVENT_SIZE + 1, EVENT_SIZE):
        if struct.unpack_from(EVENT, data, i)[2] in (EV_KEY, EV_REL, EV_ABS):
            return True
    return False


def headset_state():
    """True: the headset is worn; False: it's off; None: no proximity sensor, or no reading."""
    if not PROX:
        return None
    try:
        raw = float(Path(PROX[0]).read_text())
    except (OSError, ValueError):
        return None
    return raw > 14             # SteamVR's rule: worn while raw > prox_far * 0.8; prox_far is ~17-18


def boot_clock():
    """Seconds since boot, the time asleep included (time.monotonic() stands still while the Frame
    sleeps: a night asleep didn't count as idle, and the app was still open when it woke)."""
    return time.clock_gettime(time.CLOCK_BOOTTIME)


class Watch:
    """What the watcher decides, apart from reading the devices; times are boot_clock()'s."""

    def __init__(self, now):
        self.last_active = now
        self.off_since = None          # headset taken off at, for "headset" timers
        self.closed = {}               # app-id -> the activity it was closed after: once per idle stretch

    def seen(self, now, pressed, worn):
        """Activity: keyboard or mouse input, or the headset worn. (The Frame's VR controllers
        aren't input devices: using only them counted as idle, and an app was closed mid-use.)
        worn: headset_state(); without a reading, the input alone counts, and Headset off waits."""
        if pressed or worn:
            self.last_active = now
        self.off_since = (self.off_since or now) if worn is False else None

    def due(self, now, timers, idle_too=True):
        """[(app-id, why)]: the apps to close now (each only once per idle stretch, or per time the
        headset is taken off)."""
        due = {}
        if self.off_since is not None and now - self.off_since >= HEADSET_OFF_GRACE:
            due.update((a, _("The headset was taken off.")) for a, t in timers.items() if t["headset"])
            self.off_since = float("inf")             # once per time it's taken off
        idle = now - self.last_active
        for app_id, t in timers.items():
            if idle_too and idle >= t["minutes"] * 60 and self.closed.get(app_id) != self.last_active:
                due.setdefault(app_id, _("No activity for {minutes} minutes.").format(minutes=f"{t['minutes']:g}"))
        return list(due.items())

    def close(self, due):
        for app_id, why in due:
            if quit_app(app_id, why):
                self.closed[app_id] = self.last_active


def run():
    fds, last_scan, last_check = {}, 0.0, 0.0
    conf_mtime, timers = None, {}
    watch = Watch(boot_clock())
    asleep = boot_clock() - time.monotonic()        # time asleep since boot: the clocks' difference
    while True:
        try:
            mtime = CONF.stat().st_mtime
        except OSError:
            mtime = 0
        if mtime != conf_mtime:
            conf_mtime, timers = mtime, read_conf()
        if not timers:                 # nothing to time: don't even watch input
            for f in fds.values():
                os.close(f)
            fds.clear()
            time.sleep(10)
            watch, asleep = Watch(boot_clock()), boot_clock() - time.monotonic()
            continue
        now = boot_clock()
        if now - last_scan > 30:       # hot-plugged devices
            for path in input_devices():
                if path not in fds:
                    try:
                        fds[path] = os.open(path, os.O_RDONLY | os.O_NONBLOCK)
                    except OSError:
                        pass
            last_scan = now
        ready, _w, _x = select.select(list(fds.values()), [], [], 2)
        now = boot_clock()
        # Back from sleep? Close what was due by then before anything counts as activity: the key
        # press, or the headset, that woke the Frame would otherwise reset the idle time first.
        if now - time.monotonic() - asleep > 30:
            watch.close(watch.due(now, timers))
            last_check = now
        asleep = now - time.monotonic()
        pressed = False
        for fd in ready:
            try:
                pressed = has_user_input(os.read(fd, EVENT_SIZE * 64)) or pressed
            except OSError:            # unplugged
                for p, f in list(fds.items()):
                    if f == fd:
                        os.close(f)
                        del fds[p]
        watch.seen(now, pressed, headset_state())
        try:
            STATE.write_text(str(int(now - watch.last_active)))
        except OSError:
            pass
        idle_check = now - last_check >= 15     # finding processes costs a /proc scan: 4x a minute
        if idle_check:
            last_check = now
        watch.close(watch.due(now, timers, idle_check))


def main():
    a = sys.argv[1:]
    if a[:1] == ["--list"]:
        for app_id, t in sorted(read_conf().items()):
            print(f"{app_id} {t['minutes']:g}" + (" headset" if t["headset"] else ""))
    elif a[:1] == ["--set"] and len(a) == 3:
        try:
            minutes = float(a[2])
        except ValueError:
            sys.exit(_("minutes must be a number"))
        set_timer(a[1], max(0.0, minutes))
    elif a[:1] == ["--headset"] and len(a) == 3 and a[2] in ("on", "off"):
        set_headset(a[1], a[2] == "on")
    elif a[:1] == ["--import-signal-guard"]:
        import_signal_guard()
    elif a[:1] == ["--status"]:
        idle = STATE.read_text() if STATE.exists() else "?"
        print(_("idle {seconds}s").format(seconds=idle))
        for app_id, t in sorted(read_conf().items()):
            line = (_("{app_id}: {minutes} min, and headset off, {state}") if t["headset"] else
                    _("{app_id}: {minutes} min, {state}"))
            print("  " + line.format(app_id=app_id, minutes=f"{t['minutes']:g}",
                                     state=_("running") if app_pids(app_id) else _("not running")))
    elif a[:1] == ["--run"]:
        run()
    else:
        sys.exit(__doc__)


if __name__ == "__main__":
    main()
