Bench: plugcut.py — a real power cut on the HP with three ways back on; the night's round table written before round 1

scripts/CHANGELOG.md

@@ -1,3 +1,10 @@
+## 2026-10-14 (night) — plugcut: a real power cut on the HP, with three ways back on
+
+- New `scripts/bench/plugcut.py` + `test_plugcut.py` (12 tests): refuses any plug but `hp` (N100 only with
+  `--allow-n100`), without `--i-mean-a-power-cut`, over 60 s without `--long`, over 3600 s. The relay off carries the
+  plug countdown (DP 9) in the same command; a detached watchdog is armed first; `finally:` re-arms. Red-proofed: each
+  of nine guards removed in turn, a test failed each. `plugswitch.py` is unchanged.
+
 ## 2026-10-13 (night) — site_gates: the machine estimator moves from UNLISTED to TWINS (R-997)
 
 - `site_gates.py`: the `gepigeny`/`en/sizing` pair is a normal twin; `UNLISTED` is empty and gate 25 stays for the next

scripts/bench/README.md

@@ -53,3 +53,16 @@ Same tools, `--plug hp` (config `~/.config/felhom-bench/plug-hp.json`, the plug
 the HP at `192.168.0.104`). `plugswitch.py off --plug hp` refuses until the HP's switched-off figure (`standby_w`) is
 measured. The wizard reads both plugs from one Tuya account; the credentials are `TUYA_DEVICE_ID_N100` and
 `TUYA_DEVICE_ID_HP` (renamed from `TUYA_DEVICE_ID` by the operator).
+
+## plugcut.py — a real power cut, on purpose (2026-10-13 night)
+
+`plugswitch.py off` stays „not a power-cut button". `plugcut.py` is the separate tool that cuts a running machine:
+
+    ~/.venvs/plugmeter/bin/python scripts/bench/plugcut.py --plug hp --off-seconds 10 --i-mean-a-power-cut --log DIR
+
+- Only the plug `hp` (the N100's only with `--allow-n100`); only with `--i-mean-a-power-cut`; 1–60 s, up to 3600 s
+  only with `--long`.
+- Three ways back on: the plug's own countdown (data point 9, sent in the same command as the off, then read back —
+  not shown running → on at once); a detached watchdog process started before the off (on after off-seconds + 15 s,
+  survives the tool being killed); `finally:` in the tool.
+- One line per action to stdout and `DIR/plugcut.log`. Pinned by `test_plugcut.py` (12 tests, each guard red-proofed).

scripts/bench/plugcut.py

@@ -0,0 +1,153 @@
+#!/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:]))

scripts/bench/test_plugcut.py

@@ -0,0 +1,117 @@
+#!/usr/bin/env python3
+"""Pins plugcut.py: its refusals, the countdown sent WITH the off, the watchdog armed BEFORE the off,
+the fail-safe when the countdown does not show, and the `finally:` re-arm. A fake plug, no network."""
+import argparse, os, sys, unittest
+sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
+import plugcut  # noqa: E402
+
+
+def args(**kw):
+    d = dict(plug='hp', off_seconds=10, i_mean_a_power_cut=True, long=False, allow_n100=False)
+    d.update(kw)
+    return argparse.Namespace(**d)
+
+
+class FakePlug:
+    """Records every command. `countdown_works`: the plug turns itself on when the countdown ends."""
+    def __init__(self, countdown_works=True, show_countdown=True, fail_after_off=False):
+        self.on, self.cd, self.log = True, 0, []
+        self.countdown_works, self.show_countdown, self.fail_after_off = countdown_works, show_countdown, fail_after_off
+        self.clock = 0.0
+
+    def set_multiple_values(self, v):
+        self.log.append(('set', dict(v)))
+        if '1' in v:
+            self.on = v['1']
+        if '9' in v and self.show_countdown:
+            self.cd = v['9']
+
+    def turn_on(self):
+        self.log.append(('on',))
+        self.on, self.cd = True, 0
+
+    def status(self):
+        if self.fail_after_off and not self.on and any(e[0] == 'status' for e in self.log):
+            self.log.append(('status',))
+            raise RuntimeError('plug gone')
+        self.log.append(('status',))
+        return {'dps': {'1': self.on, '9': self.cd}}
+
+    def tick(self, s):
+        self.clock += s
+        if self.countdown_works and not self.on and self.cd and self.clock >= self.cd:
+            self.on, self.cd = True, 0
+
+
+class Refusals(unittest.TestCase):
+    def test_allowed(self):
+        self.assertIsNone(plugcut.refuse_reason(args()))
+
+    def test_without_the_long_flag(self):
+        self.assertIn('--i-mean-a-power-cut', plugcut.refuse_reason(args(i_mean_a_power_cut=False)))
+
+    def test_n100_without_operator_word(self):
+        self.assertIn('N100', plugcut.refuse_reason(args(plug='n100')))
+        self.assertIsNone(plugcut.refuse_reason(args(plug='n100', allow_n100=True)))
+
+    def test_unknown_plug(self):
+        self.assertIn('unknown', plugcut.refuse_reason(args(plug='dooplex')))
+
+    def test_over_60_needs_long(self):
+        self.assertIn('--long', plugcut.refuse_reason(args(off_seconds=61)))
+        self.assertIsNone(plugcut.refuse_reason(args(off_seconds=3000, long=True)))
+
+    def test_long_capped_at_3600(self):
+        self.assertIsNotNone(plugcut.refuse_reason(args(off_seconds=3601, long=True)))
+
+    def test_zero_seconds(self):
+        self.assertIsNotNone(plugcut.refuse_reason(args(off_seconds=0)))
+
+
+class Cut(unittest.TestCase):
+    def run_cut(self, plug, off_s=10):
+        order = []
+        def watchdog(p, after):
+            order.append(('watchdog', after, len(plug.log)))
+            return 4242
+        rc = plugcut.cut(plug, 'hp', off_s, watchdog=watchdog, sleep=plug.tick, now=lambda: plug.clock)
+        return rc, order
+
+    def test_countdown_sent_with_the_off(self):
+        p = FakePlug()
+        rc, _ = self.run_cut(p)
+        first_off = next(e for e in p.log if e[0] == 'set' and e[1].get('1') is False)
+        self.assertEqual(first_off[1].get('9'), 10, 'the off must carry the countdown in the same command')
+        self.assertEqual(rc, 0)
+        self.assertTrue(p.on)
+
+    def test_watchdog_armed_before_the_off(self):
+        p = FakePlug()
+        _, order = self.run_cut(p)
+        self.assertEqual(order[0][2], 0, 'the watchdog must start before any command reaches the plug')
+        self.assertEqual(order[0][1], 10 + plugcut.WATCHDOG_EXTRA_S)
+
+    def test_failsafe_when_countdown_not_shown(self):
+        p = FakePlug(show_countdown=False)
+        rc, _ = self.run_cut(p)
+        self.assertTrue(p.on)
+        self.assertIn(('on',), p.log)
+        self.assertEqual(rc, 0)
+
+    def test_finally_rearms_when_countdown_never_fires(self):
+        p = FakePlug(countdown_works=False)
+        rc, _ = self.run_cut(p)
+        self.assertEqual(rc, 1, 'a countdown that did not fire is a failed cut')
+        self.assertTrue(p.on, 'finally: must switch the relay back on')
+
+    def test_finally_rearms_after_an_exception(self):
+        p = FakePlug(fail_after_off=True)
+        try:
+            self.run_cut(p)
+        except Exception:
+            pass
+        self.assertTrue(p.on, 'finally: must switch the relay back on after an error')
+
+
+if __name__ == '__main__':
+    unittest.main()