felhom.eu/scripts/bench/plugswitch.py

#!/usr/bin/env python3
"""plugswitch.py — switch the N100 bench's Wi-Fi plug: on, off, status.

  ~/.venvs/plugmeter/bin/python scripts/bench/plugswitch.py status
  ~/.venvs/plugmeter/bin/python scripts/bench/plugswitch.py off
  ~/.venvs/plugmeter/bin/python scripts/bench/plugswitch.py on

Reads ~/.config/felhom-bench/plug.json (the same file plugmeter.py reads). `off` also needs, in that file:
  "standby_w":  the plug's measured reading with the machine powered off (the A2 figure);
  "host_ping":  the machine's addresses, e.g. ["192.168.0.162"].

THE RULE (it keeps this tool from being a power-cut button): `off` refuses unless, for the last 20 s, every
plug reading was below standby_w + 1 W AND no address in host_ping answered a ping. It prints why it refused
and exits 2. No standby_w or no host_ping → refuse. A failed plug read counts as „not known to be off" →
refuse. There is no flag to skip the rule.

The key and the device id are never printed. Pinned by scripts/bench/test_plugswitch.py.
"""
import json, os, subprocess, sys, time

CONF = os.path.expanduser('~/.config/felhom-bench/plug.json')
WINDOW_S = 20
STEP_S = 2
MARGIN_W = 1.0


def load():
    return json.load(open(CONF))


def device(c):
    import tinytuya
    d = tinytuya.OutletDevice(c['id'], c['ip'], c['key'], version=float(c['version']))
    d.set_socketTimeout(5)
    d.set_socketRetryLimit(1)
    return d


def read_watts(dev, c):
    dp, div = c['dps']['watts']
    dev.updatedps([int(dp)])
    s = dev.status()
    if not isinstance(s, dict) or 'dps' not in s or dp not in s['dps']:
        raise RuntimeError(f"plug read failed ({s.get('Error') if isinstance(s, dict) else s})")
    return s['dps'][dp] / div, s['dps'].get('1')


def pings(addr):
    return subprocess.run(['ping', '-c', '1', '-W', '1', addr], stdout=subprocess.DEVNULL,
                          stderr=subprocess.DEVNULL).returncode == 0


def off_allowed(samples, standby_w, window_s=WINDOW_S):
    """samples: [(t, watts or None, host_answered bool)], oldest first. Returns (ok, reason)."""
    if standby_w is None:
        return False, 'no standby_w in plug.json — measure the switched-off reading first'
    if not samples:
        return False, 'no readings'
    span = samples[-1][0] - samples[0][0]
    if span < window_s - STEP_S:
        return False, f'only {span:.0f} s of readings, need {window_s} s'
    limit = standby_w + MARGIN_W
    for t, w, up in samples:
        if up:
            return False, 'the machine answered a ping — it is running'
        if w is None:
            return False, 'a plug reading failed — cannot show the machine is off'
        if w >= limit:
            return False, f'the plug read {w:g} W, not below {limit:g} W (standby {standby_w:g} + {MARGIN_W:g}) — the machine draws power'
    return True, f'{len(samples)} readings over {span:.0f} s, all below {limit:g} W, no ping answer'


def main(argv):
    if len(argv) != 2 or argv[1] not in ('on', 'off', 'status'):
        print('usage: plugswitch.py on|off|status', file=sys.stderr)
        return 64
    cmd = argv[1]
    c = load()
    dev = device(c)
    if cmd == 'status':
        w, on = read_watts(dev, c)
        print(f'switch={"on" if on else "off"} watts={w:g}')
        return 0
    if cmd == 'on':
        r = dev.turn_on()
        w, on = read_watts(dev, c)
        print(f'switch={"on" if on else "off"} watts={w:g}' + ('' if on else f' (turn_on answered {r})'))
        return 0 if on else 1
    hosts = c.get('host_ping') or []
    if not hosts:
        print('REFUSED: no host_ping in plug.json — cannot check the machine is off', file=sys.stderr)
        return 2
    samples, start = [], time.monotonic()
    # Read until the readings themselves span the window: a slow plug read (up to 8 s) must not shorten it.
    while not samples or samples[-1][0] - samples[0][0] < WINDOW_S:
        if time.monotonic() - start > 3 * WINDOW_S:
            print('REFUSED: the plug answered too slowly to fill the window', file=sys.stderr)
            return 2
        t0 = time.monotonic()
        try:
            w, _ = read_watts(dev, c)
        except Exception as e:
            print(f'plug read failed: {type(e).__name__}', file=sys.stderr)
            w = None
        up = any(pings(h) for h in hosts)
        samples.append((t0 - start, w, up))
        print(f'  check {t0 - start:4.0f}s  watts={w}  ping={"answered" if up else "none"}', file=sys.stderr)
        ok, why = off_allowed(samples, c.get('standby_w'), window_s=0)  # early exit on the first bad sample
        if not ok and 'readings' not in why:
            print(f'REFUSED: {why}', file=sys.stderr)
            return 2
        time.sleep(max(0.0, STEP_S - (time.monotonic() - t0)))
    ok, why = off_allowed(samples, c.get('standby_w'))
    if not ok:
        print(f'REFUSED: {why}', file=sys.stderr)
        return 2
    print(f'allowed: {why}', file=sys.stderr)
    dev.turn_off()
    w, on = read_watts(dev, c)
    print(f'switch={"on" if on else "off"} watts={w:g}')
    return 0 if not on else 1


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