#!/usr/bin/env python3 """plugmeter.py — read the N100 bench's metered Wi-Fi plug and print CSV. ~/.venvs/plugmeter/bin/python scripts/bench/plugmeter.py --once ~/.venvs/plugmeter/bin/python scripts/bench/plugmeter.py --duration 600 > run.csv ~/.venvs/plugmeter/bin/python scripts/bench/plugmeter.py --source ha --once CSV columns: utc_time, watts, milliamps, volts, kwh_total. Sources: tuya (default) the plug directly on the LAN (tinytuya, protocol 3.4). Reads ~/.config/felhom-bench/plug.json: id, key, ip, version, dps. ha Home Assistant on demo-hp's guest (Tuya cloud integration). Reads ~/.config/felhom-bench/ha.json (url, connect_ip, entities) and ~/.config/felhom-bench/ha.token (a non-admin long-lived token). connect_ip sends the request to the guest on the LAN while TLS still checks the certificate of the url's host name: the felhom-bench user is "local access only", so the public (Cloudflare) path is refused. Data points (read from the device 2026-10-13, documentation/audits/plugmeter-2026-10-13/): 19 cur_power W x10, 18 cur_current mA, 20 cur_voltage V x10, 17 add_ele kWh x1000. kwh_total is the plug's add_ele counter: energy since the plug last reported to the Tuya cloud, NOT a lifetime total. It can drop back to 0. Use watts x time for energy. A failed read logs one line to stderr and the loop continues. The key and the token are never printed. Secrets live only in ~/.config/felhom-bench/ (700/600), never in a repo. """ import argparse, datetime, http.client, json, os, socket, ssl, sys, time CONF = os.path.expanduser('~/.config/felhom-bench') def log(msg): print(f'{now()} plugmeter: {msg}', file=sys.stderr, flush=True) def now(): return datetime.datetime.now(datetime.timezone.utc).strftime('%Y-%m-%dT%H:%M:%S.%fZ')[:-4] + 'Z' class TuyaSource: def __init__(self): import tinytuya c = json.load(open(os.path.join(CONF, 'plug.json'))) self.dps = c['dps'] self.dev = tinytuya.OutletDevice(c['id'], c['ip'], c['key'], version=float(c['version'])) self.dev.set_socketTimeout(5) self.dev.set_socketRetryLimit(1) # updatedps asks the plug to re-measure; without it status() repeats a cached reading. self.refresh = [int(self.dps[k][0]) for k in ('milliamps', 'watts', 'volts')] # The plug answers a refresh with the reading it already holds; the new one arrives later. So the # first read of a run can be minutes old (measured 2026-10-13: 18.7 W first, 6.8 W two seconds # later). Ask once now, so the first printed row is already fresh. self.dev.updatedps(self.refresh) time.sleep(1) def read(self): self.dev.updatedps(self.refresh) s = self.dev.status() if not isinstance(s, dict) or 'dps' not in s: # tinytuya returns an error dict, not an exception; it never holds the key. raise RuntimeError(f"tuya error {s.get('Err') if isinstance(s, dict) else ''} {s.get('Error') if isinstance(s, dict) else s}") d = s['dps'] def val(name): dp, div = self.dps[name] return d[dp] / div if dp in d else None return val('watts'), val('milliamps'), val('volts'), val('kwh_added') class HASource: def __init__(self): c = json.load(open(os.path.join(CONF, 'ha.json'))) self.host = c['url'].split('://', 1)[1].rstrip('/') self.ip = c.get('connect_ip') or self.host self.entities = c['entities'] # {"watts": "sensor.…", "milliamps": …, "volts": …, "kwh_total": …} self.token = open(os.path.join(CONF, 'ha.token')).read().strip() self.ctx = ssl.create_default_context() def state(self, entity): sock = socket.create_connection((self.ip, 443), timeout=5) conn = http.client.HTTPSConnection(self.host, timeout=5, context=self.ctx) conn.sock = self.ctx.wrap_socket(sock, server_hostname=self.host) # SNI + cert check on the name try: conn.request('GET', f'/api/states/{entity}', headers={'Authorization': 'Bearer ' + self.token}) r = conn.getresponse() body = r.read() if r.status != 200: raise RuntimeError(f'HA {entity}: HTTP {r.status}') return json.loads(body) finally: conn.close() def read(self): out = [] for name in ('watts', 'milliamps', 'volts', 'kwh_total'): ent = self.entities.get(name) if not ent: out.append(None) continue st = self.state(ent) try: v = float(st['state']) except (TypeError, ValueError): v = None if v is not None and name == 'milliamps' and st.get('attributes', {}).get('unit_of_measurement') == 'A': v *= 1000 out.append(v) return tuple(out) def fmt(v): return '' if v is None else f'{v:g}' def main(): ap = argparse.ArgumentParser(description=__doc__.split('\n')[0]) ap.add_argument('--source', choices=('tuya', 'ha'), default='tuya') ap.add_argument('--interval', type=float, default=2.0, help='seconds between reads (default 2)') ap.add_argument('--duration', type=float, default=0, help='seconds to run; 0 = until stopped') ap.add_argument('--once', action='store_true', help='one read, then exit (rc 1 if it failed)') a = ap.parse_args() src = TuyaSource() if a.source == 'tuya' else HASource() print('utc_time,watts,milliamps,volts,kwh_total', flush=True) start = time.monotonic() fails = 0 while True: t0 = time.monotonic() try: w, ma, v, kwh = src.read() print(f'{now()},{fmt(w)},{fmt(ma)},{fmt(v)},{fmt(kwh)}', flush=True) except Exception as e: # one bad read must not end a measurement run fails += 1 log(f'read failed ({type(e).__name__}: {e})') if a.once: return 1 if a.once: return 0 if a.duration and time.monotonic() - start >= a.duration: log(f'done, {fails} failed reads') return 0 time.sleep(max(0.0, a.interval - (time.monotonic() - t0))) if __name__ == '__main__': try: sys.exit(main()) except KeyboardInterrupt: sys.exit(0)