1#!/usr/bin/env python32"""Sample package power with powermetrics (requires sudo) at 1 s intervals into a JSONL file; a companion function integrates3energy over a benchmark's wall-clock window to produce joules per token. Usage:4 sudo powermetrics --samplers cpu_power,gpu_power,thermal -i 1000 -f plist | python3 power_sampler.py --out power.jsonl5"""6import argparse, json, plistlib, sys, time78def main():9 ap = argparse.ArgumentParser(); ap.add_argument("--out", required=True); a = ap.parse_args()10 buf = b""; out = open(a.out, "a")11 for chunk in iter(lambda: sys.stdin.buffer.read(65536), b""):12 buf += chunk13 while b"</plist>" in buf:14 i = buf.index(b"</plist>") + len(b"</plist>")15 doc, buf = buf[:i], buf[i:]16 s = doc.find(b"<?xml")17 if s < 0: continue18 try:19 d = plistlib.loads(doc[s:])20 except Exception:21 continue22 rec = {"t": time.time(), "elapsed_ns": d.get("elapsed_ns"),23 "cpu_mW": (d.get("processor") or {}).get("cpu_power"), "gpu_mW": (d.get("processor") or {}).get("gpu_power"),24 "ane_mW": (d.get("processor") or {}).get("ane_power"), "combined_mW": (d.get("processor") or {}).get("combined_power"),25 "thermal_pressure": (d.get("thermal_pressure"))}26 out.write(json.dumps(rec) + "\n"); out.flush()2728def energy_joules(power_jsonl, t0, t1, field="combined_mW"):29 """Integrate power (mW) over [t0, t1] (epoch seconds) with the trapezoid rule; returns joules."""30 pts = [json.loads(l) for l in open(power_jsonl)]31 pts = [(p["t"], p[field]) for p in pts if p.get(field) is not None and t0 - 1.5 <= p["t"] <= t1 + 1.5]32 if len(pts) < 2: return None33 e = 0.034 for (ta, pa), (tb, pb) in zip(pts, pts[1:]):35 lo, hi = max(ta, t0), min(tb, t1)36 if hi > lo: e += (pa + pb) / 2 * 1e-3 * (hi - lo)37 return e3839if __name__ == "__main__":40 main()41