felhom.eu/scripts/bench/plugcut.py

#!/usr/bin/env python3
"""plugcut.py — a REAL power cut on a bench machine, with three ways back on.

  ~/.venvs/plugmeter/bin/python scripts/bench/plugcut.py --plug hp --off-seconds 10 --i-mean-a-power-cut \
      [--long] [--log DIR]

This is NOT plugswitch.py. `plugswitch.py off` stays „not a power-cut button" (it refuses a running machine).
This tool cuts a running machine on purpose, so it is fenced:

  - only the plug `hp` (the N100's plug only with --allow-n100, the operator's „N100: B"; nothing else exists);
  - only with --i-mean-a-power-cut;
  - --off-seconds 1..60; up to 3600 only with --long.

Three ways the power comes back, so a dead DooPlex or a killed tool cannot leave a machine without power:
  1. THE PLUG'S OWN COUNTDOWN (data point 9, countdown_1): sent in the SAME command as the relay off, then read
     back. If the read-back does not show the countdown running, the relay is switched on at once.
  2. A WATCHDOG PROCESS, started (detached, own session) BEFORE the off: it sleeps off-seconds + 15 s and then
     switches the relay on. It survives this process being killed.
  3. `finally:` in this process: the relay is switched on (and read back) whatever happened after the off.

One line per action (UTC, plug, relay, countdown) to stdout and, with --log DIR, to DIR/plugcut.log.
The key and the device id are never printed. Pinned by scripts/bench/test_plugcut.py.
"""
import argparse, os, subprocess, sys, time
from datetime import datetime, timezone

sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
from plugswitch import load, device  # noqa: E402  (same plug files, same connection settings)

MAX_S = 60
MAX_LONG_S = 3600
WATCHDOG_EXTRA_S = 15
LOG_DIR = None


def say(plug, msg):
    line = f'{datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%S.%f")[:-3]}Z {plug} {msg}'
    print(line, flush=True)
    if LOG_DIR:
        with open(os.path.join(LOG_DIR, 'plugcut.log'), 'a') as f:
            f.write(line + '\n')


def refuse_reason(a):
    """None when allowed, else why not. Pure: tested without a plug."""
    if a.plug not in ('hp', 'n100'):
        return f'unknown plug {a.plug!r}'
    if a.plug == 'n100' and not a.allow_n100:
        return 'the N100 is not cut unless the operator said „N100: B" (--allow-n100)'
    if not a.i_mean_a_power_cut:
        return 'this cuts a running machine: add --i-mean-a-power-cut'
    if a.off_seconds < 1:
        return 'off-seconds must be at least 1'
    limit = MAX_LONG_S if a.long else MAX_S
    if a.off_seconds > limit:
        return f'off-seconds {a.off_seconds} is over {limit}' + ('' if a.long else ' (longer needs --long)')
    return None


def relay(dev):
    s = dev.status()
    d = s.get('dps', {}) if isinstance(s, dict) else {}
    return d.get('1'), d.get('9')


def start_watchdog(plug, after_s):
    """Detached process that switches the relay on after `after_s` s. Returns its pid."""
    args = [sys.executable, os.path.abspath(__file__), '--watchdog', str(after_s), '--plug', plug]
    if LOG_DIR:
        args += ['--log', LOG_DIR]
    p = subprocess.Popen(args, stdin=subprocess.DEVNULL, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL,
                         start_new_session=True)
    return p.pid


def turn_on_verified(dev, plug, who, tries=5):
    for i in range(tries):
        try:
            dev.turn_on()
            on, cd = relay(dev)
            say(plug, f'{who}: relay={"on" if on else "off?"} countdown={cd}')
            if on:
                return True
        except Exception as e:  # a slow plug read is late, not lost — try again
            say(plug, f'{who}: try {i + 1} failed ({type(e).__name__})')
        time.sleep(2)
    return False


def cut(dev, plug, off_s, watchdog=start_watchdog, sleep=time.sleep, now=time.monotonic):
    """The cut. Returns 0 when the relay is on again at the end."""
    pid = watchdog(plug, off_s + WATCHDOG_EXTRA_S)
    say(plug, f'watchdog armed (pid {pid}, relay on after {off_s + WATCHDOG_EXTRA_S} s at the latest)')
    try:
        dev.set_multiple_values({'1': False, '9': int(off_s)})
        say(plug, f'CUT: relay off + countdown {off_s} s sent in one command')
        on, cd = relay(dev)
        say(plug, f'read back: relay={"on" if on else "off"} countdown={cd}')
        if on is not False or not cd:
            say(plug, 'countdown NOT shown running — switching on now (fail-safe)')
            return 0 if turn_on_verified(dev, plug, 'fail-safe') else 1
        t0 = now()
        sleep(off_s)
        while now() - t0 < off_s + 10:
            try:
                on, cd = relay(dev)
            except Exception:
                on = None
            if on:
                say(plug, f'BACK: relay on by the plug countdown, +{now() - t0:.1f} s after the cut')
                return 0
            sleep(1)
        say(plug, 'the countdown did not switch the relay on within off-seconds + 10 s')
        return 1
    finally:
        try:
            on, cd = relay(dev)
        except Exception:
            on = None
        if not on:
            turn_on_verified(dev, plug, 'finally re-arm')


def main(argv):
    global LOG_DIR
    ap = argparse.ArgumentParser(description=__doc__.split('\n')[0])
    ap.add_argument('--plug', default='hp')
    ap.add_argument('--off-seconds', type=int, default=0)
    ap.add_argument('--i-mean-a-power-cut', action='store_true')
    ap.add_argument('--long', action='store_true')
    ap.add_argument('--allow-n100', action='store_true')
    ap.add_argument('--log')
    ap.add_argument('--watchdog', type=int, help=argparse.SUPPRESS)
    a = ap.parse_args(argv)
    LOG_DIR = a.log
    if a.watchdog is not None:  # the detached watchdog: sleep, then make sure the relay is on
        time.sleep(a.watchdog)
        dev = device(load(a.plug))
        on, _ = relay(dev)
        if on:
            say(a.plug, 'watchdog: relay already on — nothing to do')
            return 0
        return 0 if turn_on_verified(dev, a.plug, 'WATCHDOG') else 1
    why = refuse_reason(a)
    if why:
        print(f'REFUSED: {why}', file=sys.stderr)
        return 2
    dev = device(load(a.plug))
    return cut(dev, a.plug, a.off_seconds)


if __name__ == '__main__':
    sys.exit(main(sys.argv[1:]))