Classroom_Desktop/scripts/classroom-reboot-when-idle

#!/usr/bin/env python3
"""Reboot after an update, but only once nobody is using the computer.

The nightly update check (ekklesia-classroom-update.timer, 03:00, and
Persistent so that a computer switched off at 03:00 checks when it is next
switched on) used to reboot the moment an update was installed. At 03:00
that is harmless. Switched on in the morning after a night off, it meant a
classroom computer restarting itself half an hour into the day, in front
of a teacher.

So the updater installs, then starts this, and this reboots only when:

    nothing has touched any keyboard, mouse, touchpad or touchscreen for
    IDLE_SECONDS — seen here as kernel input events, which is what every
    one of them produces, whatever the desktop is doing; and

    no sound has played for as long — a video to the room has no input for
    an hour and is not an idle computer, and the end of it is not the
    moment to restart either. Seen as the kernel's own playback state
    (/proc/asound/.../status says RUNNING while a stream plays).

At 03:00 that is about 03:30. In the middle of a morning it is half an
hour after the teacher last touched it. If the computer is switched off
first, nothing is lost: the update is installed, and the next start is on
it. Until then the running session carries on as it was.

The time starts when this starts: input from before cannot be seen, so a
computer is only called idle after this has watched it be idle.

    classroom-reboot-when-idle [--idle-seconds N] [--reboot-command CMD...]

The options are for the lab; the service runs it with none.
"""

import argparse
import glob
import os
import select
import subprocess
import sys
import time

IDLE_SECONDS = 30 * 60

# How often it looks: input is noticed at once, the rest once a minute is
# plenty for a half-hour decision.
CHECK_SECONDS = 60

INPUT_GLOB = "/dev/input/event*"
ASOUND_ROOT = "/proc/asound"


def log(message):
    print(f"[classroom-reboot-when-idle] {message}", flush=True)


def audio_playing(root=ASOUND_ROOT):
    """Is any sound card playing right now?"""
    for status in glob.glob(os.path.join(root, "card*", "pcm*p", "sub*", "status")):
        try:
            with open(status, encoding="utf-8", errors="replace") as handle:
                if handle.readline().strip() == "state: RUNNING":
                    return True
        except OSError:
            continue

    return False


def should_reboot(now, last_input, idle_seconds, playing):
    """The decision: long enough without input, and nothing playing."""
    return not playing and now - last_input >= idle_seconds


class Inputs:
    """Every input device the kernel has, reopened as they come and go."""

    def __init__(self, pattern=INPUT_GLOB):
        self.pattern = pattern
        self.open = {}

    def rescan(self):
        present = set(glob.glob(self.pattern))

        for fd, path in list(self.open.items()):
            if path not in present:
                os.close(fd)
                del self.open[fd]

        known = set(self.open.values())

        for path in sorted(present - known):
            try:
                self.open[os.open(path, os.O_RDONLY | os.O_NONBLOCK)] = path
            except OSError:
                continue

    def wait(self, seconds):
        """True if anything happened on any device within SECONDS."""
        if not self.open:
            time.sleep(seconds)
            return False

        ready, _, _ = select.select(list(self.open), [], [], seconds)
        seen = False

        for fd in ready:
            try:
                while os.read(fd, 4096):
                    seen = True
            except BlockingIOError:
                pass
            except OSError:
                # A device that went away mid-read; the next rescan drops it.
                pass

        return seen


def main(argv=None):
    parser = argparse.ArgumentParser(description=__doc__.splitlines()[0])
    parser.add_argument("--idle-seconds", type=float, default=IDLE_SECONDS)
    parser.add_argument("--check-seconds", type=float, default=CHECK_SECONDS)
    parser.add_argument("--input-glob", default=INPUT_GLOB)
    parser.add_argument("--asound-root", default=ASOUND_ROOT)
    parser.add_argument("--reboot-command", nargs="+", default=["systemctl", "reboot"])
    args = parser.parse_args(argv)

    inputs = Inputs(args.input_glob)
    inputs.rescan()
    last_input = time.monotonic()
    log(f"an update is installed; rebooting once nobody has used this computer for "
        f"{round(args.idle_seconds / 60)} minutes and nothing is playing "
        f"(watching {len(inputs.open)} input devices)")

    while True:
        deadline = time.monotonic() + args.check_seconds

        while time.monotonic() < deadline:
            if inputs.wait(max(0.0, min(5.0, deadline - time.monotonic()))):
                last_input = time.monotonic()

        inputs.rescan()
        playing = audio_playing(args.asound_root)

        if playing:
            # Sound counts as someone there: a 40-minute video ending is
            # not the moment to restart, so the half hour starts after it.
            last_input = time.monotonic()

        if should_reboot(time.monotonic(), last_input, args.idle_seconds, playing):
            log("nobody has used it, and nothing is playing: rebooting")
            return subprocess.run(args.reboot_command).returncode


if __name__ == "__main__":
    sys.exit(main())