#!/usr/bin/env python3
# SPDX-License-Identifier: MIT
# Copyright (c) 2026 Olympian LED
"""current_probe_kit: a current probe on an oscilloscope channel, end to end, over documented SCPI only.

One file, Python 3.8 or newer, standard library only. It talks SCPI over the raw socket (TCP 5555 by default) and
uses only commands the Rigol DHO800/DHO900 and MHO900 programming guides document, so the same run works on Rigol's
own firmware, on a modified firmware and on NovaOS. It is written for the Micsig CP503B (the 5 A / 30 A version).

    python3 current_probe_kit.py amps    --host SCOPE --range 5A [...]   test 1: the channel reads amps, 0..2.9 A
    python3 current_probe_kit.py power   --host SCOPE --range 5A [...]   test 2: math CH1 x CH2 in W against the supply
    python3 current_probe_kit.py step    --host SCOPE --range 5A [...]   test 3: a 0.5 <-> 1.5 A step, a trigger at 1 A
    python3 current_probe_kit.py drift   --host SCOPE --minutes 30       test 4: the zero, nothing in the jaws
    python3 current_probe_kit.py persist-setup --host SCOPE              the unit and ratio through a saved setup
    python3 current_probe_kit.py persist-arm   --host SCOPE --state F    ... and through a power cycle (arm, then
    python3 current_probe_kit.py persist-check --host SCOPE --state F        cycle the scope, then check)
    python3 current_probe_kit.py report RESULT.json [...]               one table over runs (units, firmwares)
    python3 current_probe_kit.py <test> ... --dry-run                    the run against a simulated scope and
                                                                         bench: every command it would send

Every test reads each setting it will change first and puts it back at the end (on success, on an error and on
Ctrl-C), then reads them again and lists what still differs. Every command sent to the scope, the load and the supply
is logged with its time and its answer. The kit never sends *RST, a calibration, a screenshot (:DISPlay:DATA?), any
:MEASure command (:MEASure:ITEM? adds a live measurement item that stays) or a file command, except the one setup file
persist-setup saves and loads. It reads the channel from :WAVeform:DATA? and computes every figure itself.

Where a firmware does not take a unit, a ratio or a custom ratio, the run records it as a finding and goes on: that
is a result about the firmware, not a failure of the kit.

Exit codes: 0 ran (the JSON holds the verdicts and findings) and restored; 1 refused or failed, restored; 2 usage
error or an untested model; 3 the restore is incomplete (the JSON says what); 4 the scope stopped answering;
130 Ctrl-C, restored.
"""
import argparse
import csv
import datetime
import importlib
import json
import math
import random
import re
import socket
import statistics
import struct
import sys
import tempfile
import threading
import time
from pathlib import Path

VERSION = "1.0.0"
SCPI_PORT = 5555
HERE = Path(__file__).resolve().parent

# ============================================================================================ the probe and the bench
# Micsig CP503B, the 5 A / 30 A version, from Micsig's own documents (page numbers are the PDF's pages; METHOD.md
# has their URLs):
#   PI = "High Frequency AC/DC Current Probe CP Series, Product Introduction" (2026-09-21)
#   UM = "High Frequency AC/DC Current Probe CP Series User Manual" (2026-09-23)
# sens: output sensitivity, V/A (PI p.4 "1V / 1A (5A, 1X), 1V / 10A (30A, 10X)"; UM p.12 "30A: 0.1V/A, 5A: 1V/A").
# acc: warranted accuracy at DC, +-(rel x I + abs) (PI p.4; UM p.15 Table 1).
# band: this kit's pass band: the probe's accuracy with room for the reference ammeter and the scope.
# overload_apk: the overcurrent alarm (UM p.16 Table 2, typical).
PROBE = "Micsig CP503B (5 A / 30 A)"
RANGES = {
    "5A": dict(sens=1.0, acc_rel=0.01, acc_abs=0.001, band_rel=0.01, band_abs=0.005, overload_apk=7.0),
    "30A": dict(sens=0.1, acc_rel=0.01, acc_abs=0.010, band_rel=0.01, band_abs=0.020, overload_apk=50.0),
}
# The current steps of test 1, A. The top step stays under a 3 A supply's own limit (a Rigol DP2031's 32 V
# channels set at most 3 A, DP2000 data sheet p.8, with a programming accuracy of 0.15 % + 5 mA, p.9), so the supply
# stays in CV while the load regulates.
STEPS = (0.0, 0.1, 0.5, 1.0, 2.0, 2.9)
POWER_STEPS = (0.5, 1.0, 2.0, 2.9)
POWER_BAND = 0.02                       # test 2: math CH1 x CH2 against the supply's own V x I, within 2 %
STEP = dict(a=0.5, b=1.5, freq_khz=1.0, duty=50.0, level=1.0, singles=20)   # test 3
# The reference ammeter's accuracy, +-(rel x I + abs), for the datasheet budget: --ref-acc REL,ABS (a Rigol DP2031's
# current readback is 0.15 % + 5 mA on CH1 / CH2, DP2000 data sheet p.9). An electronic load's own readback is
# recorded beside it when a source gives one, never used as the reference (a DL3031's is +-(0.1 % + 0.1 % of 60 A),
# DL3000 data sheet p.5).
LOAD_CC_ACC = (0.001, 0.006)            # DL3031 CC programming accuracy, 0-6 A (DL3000 data sheet p.4)
# The scope's DC gain accuracy, +-1 % of full scale (8 divisions) above 5 mV/div (DHO900 data sheet p.10, MHO900
# data sheet p.11). A budget figure only: the kit's own readings subtract the zero taken at the same setting.
SCOPE_GAIN_FS = 0.01
# The probe ratios the guides list (MHO900 Programming Guide 3.6.9, DHO800/DHO900 Programming Guide 3.6.8). The MHO
# guide: "When the amplitude unit is set to AMPere, the range of <atten> is from 0.001 to 10, with the unit V/A."
RATIOS_MHO = (0.001, 0.002, 0.005, 0.01, 0.02, 0.05, 0.1, 0.2, 0.5, 1, 2, 5, 10, 15, 20, 50, 100, 150, 200, 500,
              1000, 1500, 2000, 5000, 10000, 15000, 20000, 50000)
RATIOS_DHO = (0.001, 0.002, 0.005, 0.01, 0.02, 0.05, 0.1, 0.2, 0.5, 1, 2, 5, 10, 20, 50, 100, 200, 500, 1000, 2000,
              5000, 10000, 20000, 50000)
SETUP_FILE = "C:/cpkit.stp"             # persist-setup: the one file the kit writes on the scope (internal disk C:/)


class KitError(Exception):
    """A failure the kit reports and exits on (after restoring)."""


class Refused(KitError):
    """A precondition or a reading refused the run (exit 1)."""


class ScpiTimeout(KitError):
    pass


class ScpiHung(KitError):
    """The scope stopped answering *IDN? on a fresh connection: nothing more can be sent, nor restored."""


def utc_iso(t):
    return datetime.datetime.fromtimestamp(t, datetime.timezone.utc).isoformat(timespec="milliseconds").replace("+00:00", "Z")


class Clock:
    def now(self):
        return time.monotonic()

    def wall(self):
        return time.time()

    def sleep(self, s):
        if s > 0:
            time.sleep(s)


class SimClock:
    """--dry-run and the tests: sleep() moves the time on and returns at once."""

    def __init__(self, start=1792252800.0):
        self.t, self.w = 0.0, start

    def now(self):
        return self.t

    def wall(self):
        return self.w + self.t

    def sleep(self, s):
        if s > 0:
            self.t += s


def parse_num(s):
    try:
        v = float(str(s).strip())
    except (TypeError, ValueError):
        return None
    return None if (math.isnan(v) or abs(v) >= 9.9e37) else v


def ladder_up(v):
    """The 1-2-5 value at or above v."""
    v = max(float(v), 1e-12)
    e = math.floor(math.log10(v))
    for m in (1, 2, 5, 10):
        x = m * 10 ** e
        if x >= v * (1 - 1e-9):
            return float("%.6g" % x)
    return float("%.6g" % (10 ** (e + 1)))


def same(a, b):
    """Two answers of one query are the same setting: numbers by value, switches 1/ON, words by their text."""
    if a is None or b is None:
        return a == b
    x, y = parse_num(a), parse_num(b)
    if x is not None and y is not None:
        return abs(x - y) <= 1e-6 * max(1e-9, abs(x), abs(y))
    norm = {"ON": "1", "OFF": "0"}
    a, b = a.strip().upper(), b.strip().upper()
    return norm.get(a, a) == norm.get(b, b)


# ============================================================================================ SCPI
# Never sent, whatever a caller asks: a reset, the setup block, calibration, a screenshot, any measurement command
# (:MEASure:ITEM? adds an item that stays; :MEASure:DELete clears the user's items), autoscale, files, a restart.
_REFUSED = [re.compile(p, re.I) for p in (
    r"^\*RST\b", r"^:?SYST(em)?:SET", r"^[:*]?CAL", r"^:?DISP(lay)?:DATA", r"^:?MEAS", r"^:?AUT",
    r"^:?SAVE", r"^:?LOAD", r"^:?SYST(em)?:(RES|REB|SHUT)", r"^:?QUIC")]
# ... but the save tree's overwrite switch, which persist-setup reads, sets and puts back (MHO900 guide 3.21.13)
_SAFE = re.compile(r"^:SAVE:OVERlap(\?| (ON|OFF|0|1))$", re.I)


class Scpi:
    """SCPI on the raw socket, LF-terminated. Every read has a deadline; after a timeout the connection is replaced
    and *IDN? checks the scope still answers (else ScpiHung). Every line is logged with its time, answer, latency."""

    def __init__(self, host, port=SCPI_PORT, timeout=5.0, clock=None, echo=False, pace=0.03, trace=None):
        self.host, self.port, self.timeout = host, port, timeout
        self.clock = clock or Clock()
        self.echo, self.pace = echo, pace
        self.trace = trace if trace is not None else []
        self.sock = None
        self.buf = b""
        self.t0 = self.clock.now()
        self.hide = []          # strings never written to the trace (the serial number)
        self.allow = set()      # exact file commands persist-setup may send

    def connect(self):
        self.close()
        try:
            self.sock = socket.create_connection((self.host, self.port), timeout=min(self.timeout, 6.0))
        except OSError as e:
            raise KitError("cannot connect to %s:%d (%s)" % (self.host, self.port, e))
        self.buf = b""

    def close(self):
        if self.sock is not None:
            try:
                self.sock.close()
            except OSError:
                pass
        self.sock = None

    def check(self, cmd):
        c = cmd.strip()
        if c in self.allow or _SAFE.match(c):
            return
        for rx in _REFUSED:
            if rx.search(c):
                raise KitError("refused to send %r: the kit never resets, calibrates, measures with the scope's own "
                               "items, takes a screenshot or writes files (but its one setup file)" % cmd)

    def _send(self, cmd):
        self.check(cmd)
        if self.sock is None:
            self.connect()
        try:
            self.sock.sendall((cmd + "\n").encode("ascii"))
        except OSError as e:
            raise KitError("send %r failed: %s" % (cmd, e))

    def _fill(self, deadline, what):
        left = deadline - time.monotonic()
        if left <= 0:
            raise ScpiTimeout(what)
        self.sock.settimeout(left)
        try:
            chunk = self.sock.recv(1 << 20)
        except socket.timeout:
            raise ScpiTimeout(what)
        except OSError as e:
            raise KitError("read failed: %s" % e)
        if not chunk:
            raise KitError("the scope closed the connection")
        self.buf += chunk

    def _readline(self, timeout):
        deadline = time.monotonic() + timeout
        while b"\n" not in self.buf:
            self._fill(deadline, "no answer")
        line, _, self.buf = self.buf.partition(b"\n")
        return line.decode("ascii", "replace").strip()

    def _clean(self, s):
        for h in self.hide:
            if h:
                s = s.replace(h, "<serial not recorded>")
        return s

    def _log(self, cmd, reply, t_send, t_reply, note=None):
        e = {"t": utc_iso(self.clock.wall()), "s": round(t_send - self.t0, 4), "dev": "scope", "send": cmd}
        if reply is not None:
            e["reply"] = self._clean(reply)
        if t_reply is not None:
            e["ms"] = round((t_reply - t_send) * 1000.0, 2)
        if note:
            e["note"] = note
        self.trace.append(e)
        if self.echo:
            sys.stderr.write("[scope %8.3f] %s%s\n" % (e["s"], cmd, "" if reply is None and not note else
                                                       "  -> " + (e.get("reply") if reply is not None else note)))

    def write(self, cmd, pace=None):
        t = self.clock.now()
        self._send(cmd)
        self._log(cmd, None, t, None)
        self.clock.sleep(self.pace if pace is None else pace)

    def query(self, cmd, timeout=None):
        tmo = self.timeout if timeout is None else timeout
        t = self.clock.now()
        self._send(cmd)
        try:
            r = self._readline(tmo)
        except ScpiTimeout:
            self._log(cmd, None, t, self.clock.now(), "TIMEOUT")
            self._recover()
            raise ScpiTimeout("%s: no answer within %.1f s" % (cmd, tmo))
        self._log(cmd, r, t, self.clock.now())
        return r

    def try_query(self, cmd, timeout=None):
        try:
            return self.query(cmd, timeout)
        except ScpiTimeout:
            return None

    def block(self, cmd, timeout=20.0):
        """An IEEE 488.2 definite-length block answer, as bytes."""
        t = self.clock.now()
        self._send(cmd)
        deadline = time.monotonic() + timeout

        def need(n):
            while len(self.buf) < n:
                self._fill(deadline, "%s: block incomplete after %.0f s" % (cmd, timeout))
        try:
            need(2)
            if self.buf[:1] != b"#":
                raise KitError("%s: not a block: %r" % (cmd, self.buf[:20]))
            nd = self.buf[1:2]
            if not nd.isdigit() or nd == b"0":
                raise KitError("%s: not a definite-length block: %r" % (cmd, self.buf[:20]))
            nd = int(nd)
            need(2 + nd)
            ln = int(self.buf[2:2 + nd])
            need(2 + nd + ln)
        except ScpiTimeout:
            self._log(cmd, None, t, self.clock.now(), "TIMEOUT (block)")
            self._recover()
            raise
        data = bytes(self.buf[2 + nd:2 + nd + ln])
        self.buf = self.buf[2 + nd + ln:]
        if self.buf[:1] == b"\n":
            self.buf = self.buf[1:]
        self._log(cmd, "<%d-byte block>" % ln, t, self.clock.now())
        return data

    def _recover(self):
        self.connect()
        t = self.clock.now()
        self._send("*IDN?")
        try:
            r = self._readline(6.0)
        except (ScpiTimeout, KitError):
            self._log("*IDN?", None, t, self.clock.now(), "TIMEOUT (after a timeout): the scope's SCPI is hung")
            raise ScpiHung("the scope stopped answering *IDN? on a fresh connection")
        self._log("*IDN?", r, t, self.clock.now(), "connection replaced after a timeout")

    def errors(self, limit=20):
        errs = []
        for _ in range(limit):
            r = self.try_query(":SYSTem:ERRor?", 3.0)
            if r is None or r.startswith("0,") or r.startswith("+0,"):
                break
            errs.append(r)
        return errs


class Identity:
    """*IDN?: maker, model, serial, firmware. The kit records the model and the firmware field only."""

    def __init__(self, idn):
        parts = [p.strip() for p in idn.split(",")]
        parts += [""] * (4 - len(parts))
        self.maker, self.model, self.serial = parts[:3]
        self.firmware = ",".join(parts[3:]).strip(",")
        m = re.match(r"^(DHO|MHO)(\d)(\d)(\d)", self.model.upper())
        self.series = None
        if m and (m.group(1) + m.group(2)) in ("DHO8", "DHO9", "MHO9"):
            self.series = m.group(1) + m.group(2) + "00"
        self.channels = int(m.group(4)) if m else 4
        self.has_50 = self.series == "MHO900"           # the DHO800 / DHO900 inputs are 1 MOhm only
        self.ratios = RATIOS_MHO if self.series == "MHO900" else RATIOS_DHO

    def as_dict(self):
        return {"model": self.model, "firmware": self.firmware, "series": self.series}


def ch(n):
    return "CHANnel%d" % n


# ============================================================================================ settings: read, restore
class Settings:
    """Every setting a test may change, read before it starts and put back after, in an order the scope accepts:
    a channel's unit before its ratio (the ratio's range follows the unit), the ratio before its scale and offset,
    the channel shown while those go back; the math and cursor sources before their switches; the channels' display
    and the run state last."""

    def __init__(self, s, ident, chans=(1,), math=False, cursor=False, save=False, clock=None):
        self.s, self.id = s, ident
        self.chans, self.math, self.cursor, self.save = tuple(chans), math, cursor, save
        self.clock = clock or Clock()
        self.snap, self.levels, self.unanswered = {}, {}, []
        self.state = None
        self.edge_source = None

    def chan_keys(self, n):
        k = [":CHANnel%d:UNITs" % n, ":CHANnel%d:PROBe" % n]
        if self.id.has_50:
            k.append(":CHANnel%d:IMPedance" % n)
        return k + [":CHANnel%d:%s" % (n, x) for x in ("COUPling", "BWLimit", "SCALe", "OFFSet")]

    def groups(self):
        g = [("display", [":CHANnel%d:DISPlay" % i for i in range(1, self.id.channels + 1)])]
        for n in self.chans:
            g.append(("ch%d" % n, self.chan_keys(n)))
        g.append(("acquire", [":ACQuire:TYPE", ":TIMebase:MAIN:SCALe", ":TIMebase:MAIN:OFFSet"]))
        g.append(("trigger", [":TRIGger:MODE", ":TRIGger:EDGE:SOURce", ":TRIGger:EDGE:SLOPe", ":TRIGger:SWEep"]))
        g.append(("waveform", [":WAVeform:SOURce", ":WAVeform:MODE", ":WAVeform:FORMat", ":WAVeform:STARt",
                               ":WAVeform:STOP"]))
        if self.math:
            g.append(("math", [":MATH1:OPERator", ":MATH1:SOURce1", ":MATH1:SOURce2", ":MATH1:SCALe",
                               ":MATH1:OFFSet", ":MATH1:DISPlay"]))
        if self.cursor:
            g.append(("cursor", [":CURSor:MANual:TYPE", ":CURSor:MANual:SOURce", ":CURSor:MANual:CAY",
                                 ":CURSor:MANual:CBY", ":CURSor:MODE"]))
        if self.save:
            g.append(("save", [":SAVE:OVERlap"]))
        return g

    def keys(self):
        return [k for _, ks in self.groups() for k in ks]

    def take(self):
        s = self.s
        s.errors(5)
        self.state = s.query(":TRIGger:STATus?")
        for k in self.keys():
            v = s.try_query(k + "?")
            if v is None:
                self.unanswered.append(k)
                s.errors(3)
            else:
                self.snap[k] = v
        # the edge level belongs to the selected source: read it on the found source and on each test channel
        self.edge_source = self.snap.get(":TRIGger:EDGE:SOURce")
        v = s.try_query(":TRIGger:EDGE:LEVel?")
        if v is not None and self.edge_source:
            self.levels[self.edge_source.upper()] = v
        for n in self.chans:
            mine = "CHAN%d" % n
            if self.edge_source and not same(self.edge_source, mine):
                s.write(":TRIGger:EDGE:SOURce " + ch(n))
                v = s.try_query(":TRIGger:EDGE:LEVel?")
                if v is not None:
                    self.levels[mine] = v
        if self.edge_source:
            s.write(":TRIGger:EDGE:SOURce " + self.edge_source)
        s.errors(3)
        return self.as_dict()

    def as_dict(self):
        return dict(state=self.state, values=dict(self.snap), edge_levels=dict(self.levels),
                    unanswered=list(self.unanswered), chans=list(self.chans), math=self.math, cursor=self.cursor,
                    save=self.save)

    @classmethod
    def from_dict(cls, s, ident, d, clock=None):
        st = cls(s, ident, d.get("chans", [1]), d.get("math", False), d.get("cursor", False), d.get("save", False),
                 clock)
        st.state, st.snap, st.levels = d.get("state"), dict(d.get("values", {})), dict(d.get("edge_levels", {}))
        st.unanswered = list(d.get("unanswered", []))
        st.edge_source = st.snap.get(":TRIGger:EDGE:SOURce")
        return st

    def restore(self):
        s, v = self.s, self.snap
        rep = {"written": [], "still_different": [], "not_readable": [], "errors": []}

        def cur(k):
            r = s.try_query(k + "?", 4.0)
            if r is None and k not in rep["not_readable"]:
                rep["not_readable"].append(k)
            return r

        def put(k):
            if k not in v:
                return
            now = cur(k)
            if now is None or same(now, v[k]):
                return
            s.write("%s %s" % (k, v[k]))
            rep["written"].append("%s %s" % (k, v[k]))

        def one_pass():
            for name, ks in self.groups():
                if name == "display":
                    continue
                if name.startswith("ch"):
                    n = int(name[2:])
                    dk = ":CHANnel%d:DISPlay" % n
                    now = cur(dk)
                    if now is not None and not same(now, "1"):
                        s.write(dk + " 1")       # a hidden channel's scale and offset are ignored (the stock)
                        rep["written"].append(dk + " 1")
                for k in ks:
                    put(k)
                if name == "trigger":
                    for src in sorted(self.levels):
                        s.write(":TRIGger:EDGE:SOURce " + src)
                        now = s.try_query(":TRIGger:EDGE:LEVel?")
                        if now is None:
                            rep["not_readable"].append(":TRIGger:EDGE:LEVel (%s)" % src)
                        elif not same(now, self.levels[src]):
                            s.write(":TRIGger:EDGE:LEVel " + self.levels[src])
                            rep["written"].append(":TRIGger:EDGE:LEVel %s (%s)" % (self.levels[src], src))
                    if self.edge_source:
                        s.write(":TRIGger:EDGE:SOURce " + self.edge_source)
            for k in dict(self.groups())["display"]:
                put(k)

        def compare():
            diff = []
            for k in self.keys():
                if k in v:
                    now = s.try_query(k + "?", 4.0)
                    if now is not None and not same(now, v[k]):
                        diff.append({"setting": k, "was": v[k], "now": now})
            for src in sorted(self.levels):
                s.write(":TRIGger:EDGE:SOURce " + src)
                now = s.try_query(":TRIGger:EDGE:LEVel?")
                if now is not None and not same(now, self.levels[src]):
                    diff.append({"setting": ":TRIGger:EDGE:LEVel (%s)" % src, "was": self.levels[src], "now": now})
            if self.edge_source:
                s.write(":TRIGger:EDGE:SOURce " + self.edge_source)
            return diff

        for _ in range(2):
            one_pass()
            self.clock.sleep(0.5)
            if not compare():
                break
        st = s.try_query(":TRIGger:STATus?")
        if self.state and st is not None and not self.state.upper().startswith("STOP") and st.upper().startswith("STOP"):
            s.write(":RUN")
            rep["written"].append(":RUN")
        rep["still_different"] = compare()
        rep["unanswered_in_snapshot"] = list(self.unanswered)
        rep["errors"] = s.errors(10)
        rep["complete"] = not rep["still_different"]
        if self.state and self.state.upper().startswith("STOP"):
            rep["note"] = "found stopped: the trace it was holding is replaced by the kit's last acquisition"
        return rep


# ============================================================================================ reading the channel
def preamble(raw):
    p = raw.split(",")
    if len(p) < 10:
        raise Refused("a preamble of %d fields: %r" % (len(p), raw))
    return dict(raw=raw, points=int(float(p[2])), xincrement=float(p[4]), xorigin=float(p[5]),
                xreference=float(p[6]), yincrement=float(p[7]), yorigin=float(p[8]), yreference=float(p[9]))


class Reader:
    """Single acquisitions and screen records (:WAVeform:MODE NORMal, WORD), in the channel's own unit."""

    def __init__(self, s, clock):
        self.s, self.clock = s, clock
        self.tb = 0.01

    def single(self, force=True, timeout=None):
        """One :SINGle; True once :TRIGger:STATus? reads STOP. AUTO sweep fires by itself; force adds one
        :TFORce once the record had time to fill. Without force, a single that does not stop within `timeout`
        returns False (test 3: the trigger did not fire)."""
        s = self.s
        s.write(":SINGle", pace=0.05)
        t0 = self.clock.now()
        forced = False
        limit = timeout if timeout is not None else 10 + 12 * self.tb
        while True:
            st = s.query(":TRIGger:STATus?")
            if st.upper().startswith("STOP"):
                return True
            if force and not forced and self.clock.now() - t0 > max(0.6, 12 * self.tb + 0.3):
                s.write(":TFORce")
                forced = True
            if self.clock.now() - t0 > limit:
                if force:
                    raise Refused("the single did not stop (%s)" % st)
                s.write(":STOP")
                return False
            self.clock.sleep(0.02)

    def record(self, source):
        s = self.s
        s.write(":WAVeform:SOURce %s" % source)
        s.write(":WAVeform:MODE NORMal")
        s.write(":WAVeform:FORMat WORD")
        pre = preamble(s.query(":WAVeform:PREamble?"))
        n = pre["points"]
        if n <= 0 or n > 200000:
            raise Refused("%s: a screen record of %d points" % (source, n))
        s.write(":WAVeform:STARt 1")
        s.write(":WAVeform:STOP %d" % n)
        b = s.block(":WAVeform:DATA?")
        if len(b) != 2 * n:
            raise Refused("%s: %d bytes, not %d" % (source, len(b), 2 * n))
        codes = struct.unpack("<%dH" % n, b)
        y = [(c - pre["yorigin"] - pre["yreference"]) * pre["yincrement"] for c in codes]
        return y, pre


def screen_clipped(y, scale, offset, margin=3.95):
    """True when a sample sits at or past the screen's edge (+-4 div about the offset's zero)."""
    if not scale:
        return False
    return any(abs((v + offset) / scale) >= margin for v in y)


def mean_std(y):
    m = statistics.fmean(y) if hasattr(statistics, "fmean") else sum(y) / len(y)
    return m, (statistics.pstdev(y) if len(y) > 1 else 0.0)


# ============================================================================================ the bench
class BenchLog:
    """Commands the load and the supply were sent, with their answers, into the run's trace."""

    def __init__(self, trace, clock, echo=False):
        self.trace, self.clock, self.echo = trace, clock, echo

    def add(self, dev, cmd, reply=None):
        e = {"t": utc_iso(self.clock.wall()), "dev": dev, "send": cmd}
        if reply is not None:
            e["reply"] = str(reply).strip()
        self.trace.append(e)
        if self.echo:
            sys.stderr.write("[%-5s] %s%s\n" % (dev, cmd, "" if reply is None else "  -> " + str(reply).strip()))


class NoBench:
    """No current source: test 4 (nothing in the jaws) and the persistence checks."""
    kind = "none"

    def preflight(self, top):
        return {"source": "none"}

    def set_current(self, amps):
        if amps:
            raise Refused("no current source in this run (--source none)")

    def transient(self, a, b, freq_khz, duty):
        raise Refused("test 3 needs a current source")

    def read(self):
        return {}

    def off(self):
        return {}


class ManualBench:
    """A person drives the source and reads a reference ammeter: the kit asks for each step and takes the reading
    typed in. For a bench without our instrument tools."""
    kind = "manual"

    def __init__(self, ask=None, say=None):
        self.ask = ask or input
        self.say = say or (lambda m: sys.stderr.write(m + "\n"))
        self.want = 0.0
        self.last = None

    def preflight(self, top):
        self.say("Manual source: for every step set the current through the jaws, then type the reference "
                 "ammeter's reading in A (a blank line takes the set value). Top step %.3g A." % top)
        return {"source": "manual"}

    def set_current(self, amps):
        self.want = amps
        self.last = None
        if amps == 0:
            self.say("Switch the current OFF (0 A), then press Enter.")
            self.ask("")
            self.last = 0.0
            return
        r = self.ask("Set %.4g A, wait for it to settle, type the reference reading in A: " % amps).strip()
        self.last = float(r) if r else amps

    def transient(self, a, b, freq_khz, duty):
        self.say("Set the source to step between %.3g A and %.3g A at %.4g kHz, %g %% duty, then press Enter." %
                 (a, b, freq_khz, duty))
        self.ask("")
        self.last = (a + b) / 2.0 * 1.0

    def read(self):
        return {} if self.last is None else {"ref_amps": self.last, "ref": "typed reference"}

    def off(self):
        self.say("Switch the current source off.")
        return {}


class SimBench:
    """--dry-run and the tests: a source that sets the simulated world's current and reads it back."""
    kind = "sim"

    def __init__(self, world, log, volts=12.0):
        self.w, self.log, self.volts = world, log, volts

    def preflight(self, top):
        self.log.add("psu", "*IDN?", "simulated supply")
        self.log.add("load", "*IDN?", "simulated load")
        return {"source": "sim", "supply": {"model": "simulated"}, "load": {"model": "simulated"}}

    def set_current(self, amps):
        self.w.transient = None
        self.w.current = amps
        if amps <= 0:
            self.log.add("load", ":SOURce:INPut:STATe OFF")
        else:
            self.log.add("load", ":SOURce:CURRent:LEVel:IMMediate %g" % amps)
            self.log.add("load", ":SOURce:INPut:STATe ON")

    def transient(self, a, b, freq_khz, duty):
        self.w.transient = (a, b, freq_khz * 1000.0, duty / 100.0)
        self.log.add("load", ":SOURce:CURRent:TRANsient:ALEVel %g" % a)
        self.log.add("load", ":SOURce:CURRent:TRANsient:BLEVel %g" % b)
        self.log.add("load", ":SOURce:TRANsient:STATe ON")

    def read(self):
        i = self.w.mean_current()
        self.log.add("psu", ":MEASure:SCALar:CURRent:DC?", "%.4f" % i)
        v = self.w.volts - 0.02 * i
        return {"ref_amps": i, "ref": "simulated supply readback", "supply_volts": v, "supply_watts": v * i,
                "supply_mode": "CV", "load_amps": i, "load_volts": v}

    def off(self):
        self.w.current, self.w.transient = 0.0, None
        self.log.add("load", ":SOURce:INPut:STATe OFF")
        return {"load": "off"}


# ============================================================================================ the tests
class Run:
    """One test on one scope: the connection, the snapshot, the bench, the findings, the points, the trace."""

    def __init__(self, a, s, ident, bench, clock):
        self.a, self.s, self.id, self.bench, self.clock = a, s, ident, bench, clock
        self.rd = Reader(s, clock)
        self.findings, self.points, self.notes = [], [], []
        self.extra = {}

    def finding(self, fid, what, **ev):
        self.findings.append(dict(id=fid, what=what, evidence=ev))

    def errors(self, what):
        e = self.s.errors()
        if e:
            self.finding("scpi-errors", "the scope queued errors after %s" % what, errors=e)
        return e

    # -- the probe channel
    def setup_probe_channel(self, n, sens, convention):
        """CH n to amperes with the probe's sensitivity: the guide's value (V/A, MHO900 guide 3.6.9) or, with
        --ratio-convention multiplier, its inverse (the X of a voltage probe). 1 MOhm (UM p.24), DC, 20 MHz band limit
        (UM p.26: for small currents). Returns what the scope took."""
        s = self.s
        s.write(":CHANnel%d:DISPlay 1" % n)
        s.write(":CHANnel%d:UNITs AMPere" % n)
        units = s.try_query(":CHANnel%d:UNITs?" % n)
        if units is None or not units.upper().startswith("AMP"):
            self.finding("units-not-taken", "CH%d did not take :CHANnel%d:UNITs AMPere" % (n, n), read_back=units,
                         errors=s.errors())
        value = sens if convention == "guide" else 1.0 / sens
        self.set_ratio(n, value)
        if self.id.has_50:
            s.write(":CHANnel%d:IMPedance OMEG" % n)
            imp = s.try_query(":CHANnel%d:IMPedance?" % n)
            if imp is None or not imp.upper().startswith("OMEG"):
                raise Refused("CH%d is not at 1 MOhm (%s): the probe's output needs a 1 MOhm input (UM p.24)" % (n, imp))
        s.write(":CHANnel%d:COUPling DC" % n)
        s.write(":CHANnel%d:BWLimit 20M" % n)
        self.errors("the probe channel's set-up")
        return {"units": units, "ratio_sent": value, "ratio_read": s.try_query(":CHANnel%d:PROBe?" % n),
                "convention": convention}

    def set_ratio(self, n, value):
        s = self.s
        word = "%.6g" % value
        s.write(":CHANnel%d:PROBe %s" % (n, word))
        got = s.try_query(":CHANnel%d:PROBe?" % n)
        listed = any(abs(value - r) <= r * 1e-6 for r in self.id.ratios)
        if got is None or not same(got, word):
            self.finding("ratio-not-taken" if listed else "custom-ratio-not-taken",
                         "CH%d did not take :CHANnel%d:PROBe %s%s" % (n, n, word, "" if listed else
                                                                      " (not in the guide's list: a custom ratio)"),
                         read_back=got, errors=s.errors())
        elif not listed:
            self.notes.append("CH%d took the custom ratio %s (not in the guide's list)" % (n, word))
        return got

    def timebase(self, tb):
        self.s.write(":TIMebase:MAIN:SCALe %g" % tb)
        self.s.write(":TIMebase:MAIN:OFFSet 0")
        self.rd.tb = tb

    def scale(self, n, scale, offset=0.0):
        s = self.s
        s.write(":CHANnel%d:SCALe %g" % (n, scale))
        s.write(":CHANnel%d:OFFSet %g" % (n, offset))
        got = parse_num(s.try_query(":CHANnel%d:SCALe?" % n))
        if got is None or abs(got / scale - 1) > 0.01:
            self.finding("scale-not-taken", "CH%d did not take %g per division in its unit" % (n, scale),
                         read_back=got)
        return got or scale

    def means(self, sources, k, scales):
        """k singles; per source the mean of each screen record, then their mean and spread."""
        per = {src: [] for src in sources}
        clipped = {src: False for src in sources}
        pre = {}
        for _ in range(k):
            self.rd.single()
            for src in sources:
                y, p = self.rd.record(src)
                m, sd = mean_std(y)
                per[src].append((m, sd))
                pre[src] = p
                sc, off = scales.get(src, (None, 0.0))
                if sc and screen_clipped(y, sc, off):
                    clipped[src] = True
        out = {}
        for src in sources:
            ms = [m for m, _ in per[src]]
            out[src] = dict(mean=sum(ms) / len(ms), spread=(max(ms) - min(ms)), noise=sum(sd for _, sd in per[src]) / len(ms),
                            records=k, clipped=clipped[src], yincrement=pre[src]["yincrement"])
        return out

    def settle(self, s=None):
        self.clock.sleep(self.a.settle if s is None else s)


def nearest_factor(f, sens):
    """What a reading/reference ratio means (the probe's range S V/A, the kit's ratio sent per the guide):
    1 the channel reads amps; S^2 the firmware multiplied by the value (the X of a voltage probe) instead of
    dividing by a V/A sensitivity; 1/10 or 10 the probe on its other range."""
    cands = {"reads amps": 1.0, "the firmware multiplies by the ratio (an X, not V/A)": sens * sens,
             "the probe on its other range (or the channel in V)": 0.1 if sens >= 1 else 10.0,
             "other range and multiplied": (0.1 if sens >= 1 else 10.0) * sens * sens}
    if f is None or f <= 0:
        return None, None
    best = min(cands.items(), key=lambda kv: abs(math.log(f / kv[1])))
    return best[0], best[1]


def band(rng, i):
    r = RANGES[rng]
    return r["band_rel"] * abs(i) + r["band_abs"]


def budget(rng, i, scale, ref_acc, turns=1):
    """The datasheets' worst case for one point: the probe (on turns x i), the reference ammeter, the scope's gain
    at full scale; in the lead's amperes."""
    r = RANGES[rng]
    probe = (r["acc_rel"] * abs(i) * turns + r["acc_abs"]) / turns
    ref = ref_acc[0] * abs(i) + ref_acc[1]
    scope = SCOPE_GAIN_FS * 8 * scale
    return dict(probe=probe, reference=ref, scope_gain=scope, total=probe + ref + scope)


def test_amps(run):
    """Test 1: the channel in amperes against the reference ammeter, 0..2.9 A; the trigger level and the cursors in
    amperes. Each step is a pair at the same vertical setting: the load off (the zero), then the load on."""
    a, s = run.a, run.s
    rng, n, turns = a.range, a.ch, a.turns
    sens = RANGES[rng]["sens"] * turns
    taken = run.setup_probe_channel(n, sens, a.ratio_convention)
    run.extra["channel"] = taken
    s.write(":ACQuire:TYPE NORMal")
    run.timebase(0.01)                              # 100 ms on screen: whole cycles of 50 Hz and 60 Hz
    s.write(":TRIGger:MODE EDGE")
    s.write(":TRIGger:EDGE:SOURce %s" % ch(n))
    s.write(":TRIGger:EDGE:SLOPe POSitive")
    s.write(":TRIGger:SWEep AUTO")
    s.write(":TRIGger:EDGE:LEVel 1")
    lv = s.try_query(":TRIGger:EDGE:LEVel?")
    run.extra["trigger_level"] = {"sent_amps": 1.0, "read": lv}
    if parse_num(lv) is None or abs(parse_num(lv) - 1.0) > 0.02:
        run.finding("trigger-level", "the trigger level set to 1 (A) read back %s" % lv, read_back=lv)
    run.errors("the acquisition and trigger set-up")
    ref_acc = a.ref_acc
    min_scale = ladder_up(0.005 / RANGES[rng]["sens"])
    steps = [x for x in STEPS if x <= a.top + 1e-9]
    for i_set in steps:
        sc = ladder_up(max(i_set, min_scale * 3) / 3.0)
        p = dict(set_amps=i_set)
        for attempt in range(3):
            # a reading at the screen's edge is no reading: ten times the scale, at most twice (a wrong ratio or the
            # probe's other range reads ten or a hundred times off)
            sc = run.scale(n, sc)
            run.bench.set_current(0.0)
            run.settle()
            z = run.means([ch(n)], a.records, {ch(n): (sc, 0.0)})[ch(n)]
            zref = run.bench.read()
            if i_set > 0:
                run.bench.set_current(i_set)
                run.settle()
                r0 = run.bench.read()
                m = run.means([ch(n)], a.records, {ch(n): (sc, 0.0)})[ch(n)]
                r1 = run.bench.read()
            if not (z["clipped"] or (i_set > 0 and m["clipped"])) or attempt == 2:
                break
            p.setdefault("rescaled_from", []).append(sc)
            sc = ladder_up(sc * 10)
        p.update(scale=sc, zero=z["mean"], zero_spread=z["spread"], zero_ref=zref.get("ref_amps"))
        if i_set > 0:
            refs = [x.get("ref_amps") for x in (r0, r1) if x.get("ref_amps") is not None]
            ref = sum(refs) / len(refs) if refs else None
            p.update(reading=m["mean"], spread=m["spread"], noise=m["noise"], clipped=m["clipped"],
                     ref_amps=ref, ref=r1.get("ref"), supply_mode=r1.get("supply_mode"),
                     load_amps=r1.get("load_amps"), supply_volts=r1.get("supply_volts"))
            corr = m["mean"] - z["mean"]
            p["corrected"] = corr
            if ref is not None:
                p["error"] = corr - ref
                p["raw_error"] = m["mean"] - ref
                p["band"] = band(rng, ref)
                p["budget"] = budget(rng, ref, sc, ref_acc, turns)
                p["factor"] = corr / ref if ref else None
                p["pass"] = abs(p["error"]) <= p["band"] and not m["clipped"]
                if str(r1.get("supply_mode") or "").upper().startswith("CC"):
                    p["pass"] = False
                    p["note"] = "the supply was in CC: not a valid point"
            if m["clipped"]:
                run.finding("clipped", "the %g A step reached the screen's edge at %g per division" % (i_set, sc))
            if abs(i_set - 1.0) < 1e-9:
                run.extra["cursor"] = cursor_check(run, n, m["mean"], z["mean"])
        run.points.append(p)
    run.bench.set_current(0.0)
    facs = [p["factor"] for p in run.points if p.get("factor") and (p.get("ref_amps") or 0) >= 0.4
            and not p.get("clipped")]
    fac = statistics.median(facs) if facs else None
    meaning, _ = nearest_factor(fac, sens)
    run.extra["factor"] = {"median": fac, "means": meaning}
    if fac is not None and abs(fac - 1) > 0.05:
        run.finding("not-amps", "the channel reads %.4g x the reference: %s" % (fac, meaning), factor=fac,
                    convention_sent=a.ratio_convention, ratio_sent=taken["ratio_sent"])
        if abs(sens - 1) > 1e-9 and steps:
            run.extra["other_convention"] = other_convention(run, n, sens, steps[-1])


def cursor_check(run, n, reading, zero):
    """The manual Y cursors on the channel: A at the step's reading, B at its zero; their values read back."""
    s = run.s
    s.write(":CURSor:MODE MANual")
    s.write(":CURSor:MANual:TYPE AMPLitude")
    s.write(":CURSor:MANual:SOURce %s" % ch(n))
    s.write(":CURSor:MANual:CAY %.6g" % reading)
    s.write(":CURSor:MANual:CBY %.6g" % zero)
    out = {"cay_sent": reading, "cby_sent": zero,
           "cay": s.try_query(":CURSor:MANual:CAY?"), "ay_value": s.try_query(":CURSor:MANual:AYValue?"),
           "by_value": s.try_query(":CURSor:MANual:BYValue?")}
    ay = parse_num(out["ay_value"])
    if ay is None or abs(ay - reading) > max(0.01 * abs(reading), 0.002):
        run.finding("cursor", "cursor A placed at %.4g (A) reads %s" % (reading, out["ay_value"]), **out)
    run.errors("the cursor check")
    return out


def other_convention(run, n, sens, i_set):
    """The guide's ratio did not give amps: the other reading of the ratio (1 / the sensitivity), one point."""
    other = 1.0 / sens if run.a.ratio_convention == "guide" else sens
    got = run.set_ratio(n, other)
    sc = ladder_up(i_set / 3.0)
    for attempt in range(3):
        sc = run.scale(n, sc)
        run.bench.set_current(0.0)
        run.settle()
        z = run.means([ch(n)], run.a.records, {ch(n): (sc, 0.0)})[ch(n)]
        run.bench.set_current(i_set)
        run.settle()
        m = run.means([ch(n)], run.a.records, {ch(n): (sc, 0.0)})[ch(n)]
        ref = run.bench.read().get("ref_amps")
        if not (z["clipped"] or m["clipped"]) or attempt == 2:
            break
        sc = ladder_up(sc * 10)
    run.bench.set_current(0.0)
    f = (m["mean"] - z["mean"]) / ref if ref else None
    out = {"ratio_sent": other, "ratio_read": got, "set_amps": i_set, "ref_amps": ref,
           "corrected": m["mean"] - z["mean"], "factor": f}
    if f is not None and abs(f - 1) <= 0.05:
        run.finding("ratio-convention", "the channel reads amps with :CHANnel%d:PROBe %.6g, the inverse of the "
                    "guide's V/A value: this firmware takes the ratio in amperes as an X" % (n, other), **out)
    return out


def test_power(run):
    """Test 2: CH2 on the supply's output (a 10x probe), math CH1 x CH2, its mean in W against the supply's own
    V x I readback, within 2 %."""
    a, s = run.a, run.s
    n, v = a.ch, a.vch
    sens = RANGES[a.range]["sens"] * a.turns
    run.extra["channel"] = run.setup_probe_channel(n, sens, a.ratio_convention)
    s.write(":CHANnel%d:DISPlay 1" % v)
    s.write(":CHANnel%d:UNITs VOLTage" % v)
    s.write(":CHANnel%d:PROBe 10" % v)
    if run.id.has_50:
        s.write(":CHANnel%d:IMPedance OMEG" % v)
    s.write(":CHANnel%d:COUPling DC" % v)
    s.write(":CHANnel%d:BWLimit 20M" % v)
    vsc = run.scale(v, ladder_up(a.volts / 3.0))
    s.write(":ACQuire:TYPE NORMal")
    run.timebase(0.01)
    s.write(":TRIGger:MODE EDGE")
    s.write(":TRIGger:EDGE:SOURce %s" % ch(v))
    s.write(":TRIGger:SWEep AUTO")
    s.write(":MATH1:OPERator MULTiply")
    s.write(":MATH1:SOURce1 %s" % ch(n))
    s.write(":MATH1:SOURce2 %s" % ch(v))
    s.write(":MATH1:DISPlay 1")
    run.errors("the power set-up")
    for i_set in [x for x in POWER_STEPS if x <= a.top + 1e-9]:
        sc = run.scale(n, ladder_up(i_set / 3.0))
        msc = ladder_up(i_set * a.volts / 3.0)
        s.write(":MATH1:SCALe %g" % msc)
        s.write(":MATH1:OFFSet 0")
        run.bench.set_current(i_set)
        run.settle()
        r0 = run.bench.read()
        m = run.means([ch(n), ch(v), "MATH1"], a.records, {ch(n): (sc, 0.0), ch(v): (vsc, 0.0), "MATH1": (msc, 0.0)})
        r1 = run.bench.read()
        ws = [x.get("supply_watts") for x in (r0, r1) if x.get("supply_watts") is not None]
        w = sum(ws) / len(ws) if ws else None
        p = dict(set_amps=i_set, amps=m[ch(n)]["mean"], volts=m[ch(v)]["mean"], math_watts=m["MATH1"]["mean"],
                 product_watts=m[ch(n)]["mean"] * m[ch(v)]["mean"], supply_watts=w, supply_volts=r1.get("supply_volts"),
                 ref_amps=r1.get("ref_amps"), math_scale=msc, clipped=any(m[x]["clipped"] for x in m))
        if w:
            p["math_error_pct"] = 100.0 * (p["math_watts"] - w) / w
            p["product_error_pct"] = 100.0 * (p["product_watts"] - w) / w
            p["pass"] = abs(p["math_watts"] - w) <= POWER_BAND * w and not p["clipped"]
        run.points.append(p)
    run.bench.set_current(0.0)


def test_step(run):
    """Test 3: the load stepping 0.5 <-> 1.5 A at 1 kHz; an edge trigger on the channel at 1 A, NORMal sweep: every
    single must fire; the step's levels, its rise and fall in amperes; the records' mean against the supply."""
    a, s = run.a, run.s
    n = a.ch
    sens = RANGES[a.range]["sens"] * a.turns
    run.extra["channel"] = run.setup_probe_channel(n, sens, a.ratio_convention)
    sc = run.scale(n, ladder_up(STEP["b"] / 3.0))
    s.write(":ACQuire:TYPE NORMal")
    run.timebase(200e-6)                            # 2 ms on screen: two periods
    s.write(":TRIGger:MODE EDGE")
    s.write(":TRIGger:EDGE:SOURce %s" % ch(n))
    s.write(":TRIGger:EDGE:SLOPe POSitive")
    s.write(":TRIGger:EDGE:LEVel %g" % STEP["level"])
    s.write(":TRIGger:SWEep NORMal")
    run.errors("the step set-up")
    run.bench.transient(STEP["a"], STEP["b"], STEP["freq_khz"], STEP["duty"])
    run.settle()
    fired, rows = 0, []
    for k in range(STEP["singles"]):
        ok = run.rd.single(force=False, timeout=2.0)
        if not ok:
            rows.append(dict(single=k + 1, fired=False))
            continue
        fired += 1
        y, pre = run.rd.record(ch(n))
        rows.append(dict(single=k + 1, fired=True, **step_shape(y, pre, STEP["level"])))
    ref = run.bench.read()
    run.bench.off()
    good = [r for r in rows if r.get("fired") and r.get("lo") is not None]

    def med(key):
        xs = [r[key] for r in good if r.get(key) is not None]
        return statistics.median(xs) if xs else None
    lo, hi, mean = med("lo"), med("hi"), med("mean")
    rng = a.range
    res = dict(singles=STEP["singles"], fired=fired, lo=lo, hi=hi, rise_s=med("rise_s"), fall_s=med("fall_s"),
               edges_per_record=med("rising"), at_trigger=med("at_trigger"), mean=mean, ref_amps=ref.get("ref_amps"),
               lo_band=band(rng, STEP["a"]) + LOAD_CC_ACC[0] * STEP["a"] + LOAD_CC_ACC[1],
               hi_band=band(rng, STEP["b"]) + LOAD_CC_ACC[0] * STEP["b"] + LOAD_CC_ACC[1], scale=sc)
    res["pass_fired"] = fired == STEP["singles"]
    res["pass_lo"] = lo is not None and abs(lo - STEP["a"]) <= res["lo_band"]
    res["pass_hi"] = hi is not None and abs(hi - STEP["b"]) <= res["hi_band"]
    if mean is not None and res["ref_amps"]:
        res["mean_error"] = mean - res["ref_amps"]
        res["pass_mean"] = abs(res["mean_error"]) <= band(rng, res["ref_amps"])
    res["pass"] = all(res.get(k, True) for k in ("pass_fired", "pass_lo", "pass_hi", "pass_mean"))
    run.extra["step"] = res
    run.points = rows


def step_shape(y, pre, level):
    """One record of the step: its two levels (the medians of the samples below and above the trigger level), the
    10-90 % rise and fall (median over the edges), the rising crossings, the value at the trigger point."""
    n = len(y)
    if n < 10:
        return {}
    lo_s = [v for v in y if v < level]
    hi_s = [v for v in y if v >= level]
    if not lo_s or not hi_s:
        return dict(lo=None, hi=None)
    lo, hi = statistics.median(lo_s), statistics.median(hi_s)
    dt = pre["xincrement"]
    t10, t90 = lo + 0.1 * (hi - lo), lo + 0.9 * (hi - lo)
    rises, falls, rising = [], [], 0
    state = None
    start = None
    for k, v in enumerate(y):
        if v <= t10:
            if state == "falling" and start is not None:
                falls.append((k - start) * dt)
            state, start = "low", None
        elif v >= t90:
            if state == "rising" and start is not None:
                rises.append((k - start) * dt)
                rising += 1
            state, start = "high", None
        else:
            if state == "low":
                state, start = "rising", k
            elif state == "high":
                state, start = "falling", k
    k0 = int(round((0.0 - pre["xorigin"]) / dt - pre["xreference"])) if dt else n // 2
    k0 = min(max(k0, 0), n - 1)
    return dict(lo=lo, hi=hi, mean=sum(y) / n, rise_s=statistics.median(rises) if rises else None,
                fall_s=statistics.median(falls) if falls else None, rising=rising, at_trigger=y[k0])


def test_drift(run):
    """Test 4: nothing in the jaws, after the probe's zero / degauss: the channel's mean every --interval seconds
    for --minutes; the drift from the first reading."""
    a, s = run.a, run.s
    n = a.ch
    run.extra["channel"] = run.setup_probe_channel(n, RANGES[a.range]["sens"], a.ratio_convention)
    sc = run.scale(n, ladder_up(0.005 / RANGES[a.range]["sens"]))
    s.write(":ACQuire:TYPE NORMal")
    run.timebase(0.01)
    s.write(":TRIGger:MODE EDGE")
    s.write(":TRIGger:EDGE:SOURce %s" % ch(n))
    s.write(":TRIGger:SWEep AUTO")
    run.errors("the drift set-up")
    t0 = run.clock.now()
    total = a.minutes * 60.0
    k = 0
    while True:
        t = run.clock.now() - t0
        m = run.means([ch(n)], a.records, {ch(n): (sc, 0.0)})[ch(n)]
        run.points.append(dict(t_min=round(t / 60.0, 3), utc=utc_iso(run.clock.wall()), mean=m["mean"],
                               spread=m["spread"], noise=m["noise"], clipped=m["clipped"]))
        k += 1
        nxt = k * a.interval
        if nxt > total + 1e-9:
            break
        run.clock.sleep(max(0.0, nxt - (run.clock.now() - t0)))
    ys = [p["mean"] for p in run.points]
    ts = [p["t_min"] for p in run.points]
    slope = None
    if len(ys) > 1:
        tm, ym = sum(ts) / len(ts), sum(ys) / len(ys)
        den = sum((x - tm) ** 2 for x in ts)
        slope = sum((x - tm) * (y - ym) for x, y in zip(ts, ys)) / den if den else None
    run.extra["drift"] = dict(first=ys[0], last=ys[-1], drift=ys[-1] - ys[0], span=max(ys) - min(ys),
                              slope_per_min=slope, samples=len(ys), scale=sc)


def readback(s, n):
    return {"units": s.try_query(":CHANnel%d:UNITs?" % n), "probe": s.try_query(":CHANnel%d:PROBe?" % n),
            "scale": s.try_query(":CHANnel%d:SCALe?" % n)}


def test_persist_setup(run):
    """The unit and the ratio through a saved setup: set, :SAVE:SETup the kit's file, change both, :LOAD:SETup it,
    read them back. The file stays on the scope (no documented command deletes it)."""
    a, s = run.a, run.s
    n = a.ch
    sens = RANGES[a.range]["sens"] * a.turns
    run.extra["channel"] = run.setup_probe_channel(n, sens, a.ratio_convention)
    want = readback(s, n)
    path = a.setup_path
    for c in (":SAVE:OVERlap ON", ":SAVE:SETup %s" % path, ":LOAD:SETup %s" % path):
        s.allow.add(c)
    s.write(":SAVE:OVERlap ON")
    s.write(":SAVE:SETup %s" % path, pace=2.0)
    run.errors(":SAVE:SETup")
    s.write(":CHANnel%d:UNITs VOLTage" % n)
    s.write(":CHANnel%d:PROBe 1" % n)
    changed = readback(s, n)
    s.write(":LOAD:SETup %s" % path, pace=3.0)
    run.errors(":LOAD:SETup")
    got = readback(s, n)
    s.allow.clear()
    ok_u = same(got["units"], want["units"])
    ok_p = same(got["probe"], want["probe"])
    run.extra["persist_setup"] = dict(path=path, set=want, changed_to=changed, after_load=got, units_kept=ok_u,
                                      ratio_kept=ok_p, pass_=ok_u and ok_p, file_left_on_scope=path)
    if not ok_u:
        run.finding("setup-units", "the unit did not come back from a saved setup", set=want, after_load=got)
    if not ok_p:
        run.finding("setup-ratio", "the ratio did not come back from a saved setup", set=want, after_load=got)


def test_persist_arm(run):
    """Before a power cycle: the channel to amperes and the ratio; the settings found are written to --state so
    persist-check can compare and put them back after the cycle. Nothing is restored now."""
    a, s = run.a, run.s
    n = a.ch
    sens = RANGES[a.range]["sens"] * a.turns
    run.extra["channel"] = run.setup_probe_channel(n, sens, a.ratio_convention)
    run.extra["armed"] = readback(s, n)
    run.no_restore = True


def test_persist_check(run):
    """After the power cycle: the unit and the ratio read back against what persist-arm set; then every setting
    persist-arm found is put back."""
    st = run.extra["state"]
    n = st["ch"]
    got = readback(run.s, n)
    want = st["armed"]
    ok_u, ok_p = same(got["units"], want["units"]), same(got["probe"], want["probe"])
    run.extra["persist_power"] = dict(armed=want, after_cycle=got, units_kept=ok_u, ratio_kept=ok_p,
                                      pass_=ok_u and ok_p, armed_at=st.get("armed_at"))
    if not ok_u:
        run.finding("power-cycle-units", "the unit did not survive the power cycle", armed=want, after=got)
    if not ok_p:
        run.finding("power-cycle-ratio", "the ratio did not survive the power cycle", armed=want, after=got)


TESTS = {"amps": (test_amps, dict(cursor=True)), "power": (test_power, dict(math=True, two=True)),
         "step": (test_step, {}), "drift": (test_drift, {}), "persist-setup": (test_persist_setup, dict(save=True)),
         "persist-arm": (test_persist_arm, {}), "persist-check": (test_persist_check, {})}
NEEDS_CURRENT = ("amps", "power", "step")


# ============================================================================================ a bench's own module
def bench_plugin():
    """An optional module beside the kit, current_probe_bench.py, that adds one bench's own instruments: a source
    that drives its load and supply, a unit name for --host, the unit's own date -u, extra commands. The published
    kit runs without it. Its hooks: add_args(parser, test), sources(), make_bench(args, log, clock),
    target(args) -> (host, key), unit_date(key), commands() -> {name: (add_parser_args, run)}, dry_model(args)."""
    if str(HERE) not in sys.path:
        sys.path.insert(0, str(HERE))
    try:
        mod = importlib.import_module("current_probe_bench")
    except ImportError:
        return None
    mod.bind(sys.modules[__name__])
    return mod


PLUGIN = bench_plugin()


def unit_stamp(s, clock, key=None):
    """The scope's own clock over SCPI (:SYSTem:DATE? / :SYSTem:TIME?), the unit's own `date -u` when a bench module
    can read it, and this computer's UTC on either side."""
    out = {"pc_utc_before": utc_iso(clock.wall())}
    out["scope_date"] = s.try_query(":SYSTem:DATE?", 3.0)
    out["scope_time"] = s.try_query(":SYSTem:TIME?", 3.0)
    if key is not None and PLUGIN is not None:
        try:
            out["unit_date_u"] = PLUGIN.unit_date(key)
        except (Exception, SystemExit) as e:  # noqa: BLE001
            out["unit_date_u"] = "unavailable (%s)" % type(e).__name__
    else:
        out["unit_date_u"] = "not asked"
    out["pc_utc_after"] = utc_iso(clock.wall())
    return out


# ============================================================================================ files
def write_json(path, doc):
    path = Path(path)
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text(json.dumps(doc, indent=1, default=str) + "\n", encoding="utf-8", newline="\n")


def write_csv(path, rows):
    if not rows:
        return
    keys = []
    for r in rows:
        for k, v in r.items():
            if k not in keys and not isinstance(v, (dict, list)):
                keys.append(k)
    with open(path, "w", newline="", encoding="utf-8") as f:
        w = csv.DictWriter(f, fieldnames=keys, extrasaction="ignore")
        w.writeheader()
        for r in rows:
            w.writerow({k: r.get(k) for k in keys})


def verdict(doc):
    t = doc["test"]
    ex = doc.get("results", {})
    if t == "amps":
        pts = [p for p in doc.get("points", []) if "pass" in p]
        return bool(pts) and all(p["pass"] for p in pts) and not any(f["id"] in ("units-not-taken", "not-amps",
                                                                               "ratio-not-taken")
                                                                   for f in doc["findings"])
    if t == "power":
        pts = [p for p in doc.get("points", []) if "pass" in p]
        return bool(pts) and all(p["pass"] for p in pts)
    if t == "step":
        return bool(ex.get("step", {}).get("pass"))
    if t == "persist-setup":
        return bool(ex.get("persist_setup", {}).get("pass_"))
    if t == "persist-check":
        return bool(ex.get("persist_power", {}).get("pass_"))
    return None


# ============================================================================================ one run
def run_test(a, clock=None, sim=None):
    """Connect, identify, snapshot, run the test, put everything back, write the files. Returns the exit code."""
    clock = clock or (SimClock() if sim else Clock())
    trace = []
    blog = BenchLog(trace, clock, echo=a.echo)
    host, port, key = a.host, a.port, None
    if sim is not None:
        host, port = "127.0.0.1", sim.port
    elif PLUGIN is not None:
        host, key = PLUGIN.target(a)
    if not host:
        raise SystemExit("give --host SCOPE")
    s = Scpi(host, port, timeout=a.timeout, clock=clock, echo=a.echo, trace=trace)
    t_start = clock.wall()
    doc = {"kit": "current_probe_kit", "version": VERSION, "test": a.test, "probe": PROBE, "range": a.range,
           "turns": a.turns, "label": a.label, "firmware_label": a.firmware,
           "unit_label": getattr(a, "unit", None) or a.label,
           "ratio_convention": a.ratio_convention, "started": utc_iso(t_start), "findings": [], "notes": [],
           "points": [], "results": {}}
    fn, opts = TESTS[a.test]
    code = 0
    run = None
    st = None
    bench = None
    state = None
    try:
        s.connect()
        idn = s.query("*IDN?")
        ident = Identity(idn)
        s.hide.append(ident.serial)
        for e in s.trace:
            if "reply" in e:
                e["reply"] = s._clean(e["reply"])
        doc["unit"] = ident.as_dict()
        if ident.series is None and not a.untested_model:
            raise SystemExit("%s is not a DHO800 / DHO900 / MHO900 (pass --untested-model to try anyway)" % ident.model)
        if s.query("*OPC?").strip() not in ("1", "+1"):
            raise KitError("*OPC? did not answer 1")
        doc["stamp_start"] = unit_stamp(s, clock, key)
        if a.test == "persist-check":
            state = json.loads(Path(a.state).read_text(encoding="utf-8"))
            if state.get("unit", {}).get("model") != ident.model:
                raise Refused("the state file is %s's, this is a %s" % (state.get("unit", {}).get("model"), ident.model))
            st = Settings.from_dict(s, ident, state["snapshot"], clock)
        else:
            chans = (a.ch, a.vch) if opts.get("two") else (a.ch,)
            st = Settings(s, ident, chans, math=opts.get("math", False), cursor=opts.get("cursor", False),
                          save=opts.get("save", False), clock=clock)
            doc["snapshot"] = st.take()
        if sim is not None:
            bench = SimBench(sim.world, blog, volts=a.volts)
        elif a.test in NEEDS_CURRENT and PLUGIN is not None and a.source in PLUGIN.sources():
            bench = PLUGIN.make_bench(a, blog, clock)
        elif a.test in NEEDS_CURRENT and a.source == "manual":
            bench = ManualBench()
        else:
            bench = NoBench()
        if a.test in NEEDS_CURRENT:
            doc["bench"] = bench.preflight(a.top if a.test != "step" else STEP["b"])
        run = Run(a, s, ident, bench, clock)
        if state is not None:
            run.extra["state"] = state
        fn(run)
    except SystemExit as e:
        doc["error"] = str(e)
        code = 2
    except ScpiHung as e:
        doc["error"] = str(e)
        code = 4
    except KeyboardInterrupt:
        doc["error"] = "interrupted"
        code = 130
    except (Refused, KitError) as e:
        doc["error"] = "%s: %s" % (type(e).__name__, e)
        code = 1
    except Exception as e:  # noqa: BLE001 -- recorded, restored, reported
        doc["error"] = "%s: %s" % (type(e).__name__, e)
        code = 1
    finally:
        if bench is not None:
            try:
                doc["bench_off"] = bench.off()
            except Exception as e:  # noqa: BLE001
                doc["bench_off"] = "FAILED: %s" % e
                code = code or 1
            if hasattr(bench, "close"):
                bench.close()
        if st is not None and code != 4:
            try:
                if run is not None and getattr(run, "no_restore", False):
                    doc["restore"] = {"skipped": "persist-arm leaves the channel armed; persist-check puts it back"}
                    state_doc = {"unit": doc.get("unit"), "snapshot": st.as_dict(), "armed": run.extra.get("armed"),
                                 "ch": a.ch, "armed_at": utc_iso(clock.wall())}
                    write_json(a.state, state_doc)
                    doc["state_file"] = str(a.state)
                else:
                    doc["restore"] = st.restore()
                    if not doc["restore"]["complete"] and code == 0:
                        code = 3
                doc["stamp_end"] = unit_stamp(s, clock, key)
            except ScpiHung as e:
                doc["error"] = (doc.get("error", "") + "; restore: " + str(e)).strip("; ")
                code = 4
        s.close()
    if run is not None:
        doc["findings"] = run.findings
        doc["notes"] = run.notes
        doc["points"] = run.points
        doc["results"] = {k: v for k, v in run.extra.items() if k != "state"}
    doc["verdict"] = verdict(doc) if code == 0 else None
    doc["trace"] = trace
    doc["finished"] = utc_iso(clock.wall())
    out = Path(a.out_dir) if a.out_dir else Path("current-probe-%s-%s-%s" % (
        re.sub(r"[^A-Za-z0-9.-]+", "-", doc.get("unit_label") or doc.get("unit", {}).get("model", "scope")),
        a.test,
        datetime.datetime.fromtimestamp(t_start, datetime.timezone.utc).strftime("%Y%m%dT%H%M%SZ")))
    out.mkdir(parents=True, exist_ok=True)
    stem = "%s-%s" % (a.test, a.range) if a.test in ("amps", "power", "step", "drift") else a.test
    write_json(out / (stem + ".json"), doc)
    write_csv(out / (stem + ".csv"), doc["points"])
    doc["_files"] = [str(out / (stem + ".json")), str(out / (stem + ".csv"))]
    if not a.quiet:
        for line in report_lines(doc):
            print(line)
        print("written: %s" % ", ".join(doc["_files"]))
    a._last = doc
    return code


def report_lines(doc):
    u = doc.get("unit", {})
    lines = ["%s %s on %s %s (%s), %s range, turns %s: %s" % (doc["kit"], doc["version"], u.get("model"),
                                                               u.get("firmware"), doc.get("firmware_label") or "-",
                                                               doc["range"], doc["turns"], doc["test"])]
    if doc.get("error"):
        lines.append("  ERROR: %s" % doc["error"])
    for p in doc.get("points", []):
        if doc["test"] == "amps" and "ref_amps" in p:
            lines.append("  %5.3f A set: ref %s, channel %s (zero %+.4f), error %s, band %s -> %s" % (
                p["set_amps"], fmt(p.get("ref_amps")), fmt(p.get("corrected")), p.get("zero", 0.0),
                fmt(p.get("error")), fmt(p.get("band")), "pass" if p.get("pass") else "FAIL"))
        elif doc["test"] == "power" and "supply_watts" in p:
            lines.append("  %5.3f A: math %s W, supply %s W (%s %%) -> %s" % (
                p["set_amps"], fmt(p.get("math_watts")), fmt(p.get("supply_watts")), fmt(p.get("math_error_pct"), 2),
                "pass" if p.get("pass") else "FAIL"))
    r = doc.get("results", {})
    if "factor" in r:
        lines.append("  the channel / the reference: %s (%s)" % (fmt(r["factor"].get("median")), r["factor"].get("means")))
    if "step" in r:
        x = r["step"]
        lines.append("  step: fired %s/%s, levels %s / %s A, rise %s s, fall %s s -> %s" % (
            x["fired"], x["singles"], fmt(x.get("lo")), fmt(x.get("hi")), fmt(x.get("rise_s")), fmt(x.get("fall_s")),
            "pass" if x.get("pass") else "FAIL"))
    if "drift" in r:
        x = r["drift"]
        lines.append("  drift over %d readings: %s A (span %s A, slope %s A/min)" % (
            x["samples"], fmt(x["drift"]), fmt(x["span"]), fmt(x.get("slope_per_min"))))
    for k in ("persist_setup", "persist_power"):
        if k in r:
            lines.append("  %s: units kept %s, ratio kept %s" % (k, r[k]["units_kept"], r[k]["ratio_kept"]))
    for f in doc.get("findings", []):
        lines.append("  finding [%s]: %s" % (f["id"], f["what"]))
    rs = doc.get("restore") or {}
    if rs.get("still_different"):
        lines.append("  RESTORE INCOMPLETE: %s" % rs["still_different"])
    lines.append("  verdict: %s" % {True: "pass", False: "FAIL", None: "-"}[doc.get("verdict")])
    return lines


def fmt(v, nd=4):
    if v is None:
        return "-"
    if isinstance(v, float):
        return ("%." + str(nd) + "g") % v
    return str(v)


# ============================================================================================ report over runs
def cmd_report(a):
    rows = []
    for f in a.files:
        d = json.loads(Path(f).read_text(encoding="utf-8"))
        u = d.get("unit", {})
        r = d.get("results", {})
        errs = [abs(p["error"]) for p in d.get("points", []) if p.get("error") is not None]
        rows.append([d.get("unit_label") or "-", "%s %s" % (u.get("model"), u.get("firmware")), d.get("firmware_label")
                     or "-", d["test"], d["range"], (r.get("channel") or {}).get("units") or "-",
                     fmt((r.get("factor") or {}).get("median")), fmt(max(errs)) if errs else "-",
                     ", ".join(x["id"] for x in d.get("findings", [])) or "-",
                     {True: "pass", False: "FAIL", None: "-"}[d.get("verdict")]])
    head = ["unit", "scope", "firmware", "test", "range", "units", "factor", "max |error| A", "findings", "verdict"]
    print("| " + " | ".join(head) + " |")
    print("|" + "---|" * len(head))
    for r in rows:
        print("| " + " | ".join(str(x) for x in r) + " |")
    return 0


# ============================================================================================ the simulated scope
class SimWorld:
    """What the probe is clamped on: a DC current or the load's step, the supply's voltage, the probe's zero."""

    def __init__(self, sens=1.0, turns=1, zero=0.0015, noise=0.0004, volts=12.0, seed=7):
        self.current, self.transient = 0.0, None
        self.sens, self.turns, self.zero, self.noise, self.volts = sens, turns, zero, noise, volts
        self.rng = random.Random(seed)

    def current_at(self, t):
        if self.transient is None:
            return self.current
        a, b, f, duty = self.transient
        T, tr = 1.0 / f, 20e-6
        p = (t + tr / 2.0) % T
        if p < tr:
            return a + (b - a) * p / tr
        if p < duty * T:
            return b
        if p < duty * T + tr:
            return b - (b - a) * (p - duty * T) / tr
        return a

    def mean_current(self):
        if self.transient is None:
            return self.current
        a, b, _, duty = self.transient
        return a + (b - a) * duty

    def probe_volts(self, t):
        return self.sens * self.turns * self.current_at(t) + self.zero + self.rng.gauss(0, self.noise)

    def supply_volts(self, t):
        return self.volts - 0.02 * self.current_at(t)


def short_form(word):
    if word == word.upper():
        return word
    m = re.match(r"([A-Z]+)[a-z]*(\d*)$", word)
    return (m.group(1) + m.group(2)) if m else word.upper()


class SimScope:
    """A Rigol-dialect scope on 127.0.0.1 for --dry-run and the tests. Its firmware's habits are parameters:
    `convention` how it reads :CHANnel:PROBe in amperes ("va", the MHO900 guide; "x", a multiplier), `units_ok`,
    `custom_ok` (a ratio off the guide's list), `keeps` (the unit and ratio across a power cycle), `file_ok`
    (:SAVE:SETup / :LOAD:SETup)."""

    def __init__(self, model="MHO934", world=None, convention="va", units_ok=True, custom_ok=True, keeps=True,
                 file_ok=True, firmware="00.01.00.00.26", serial="SIM0SERIAL0042"):
        self.model, self.mho = model, model.startswith("MHO")
        self.world = world or SimWorld()
        self.convention, self.units_ok, self.custom_ok, self.keeps, self.file_ok = (convention, units_ok, custom_ok,
                                                                                     keeps, file_ok)
        self.firmware, self.serial = firmware, serial
        self.ratios = RATIOS_MHO if self.mho else RATIOS_DHO
        self.errors, self.lines, self.files = [], [], {}
        self.v = {}
        for i in range(1, 5):
            c = ":CHAN%d:" % i
            self.v.update({c + "DISP": "1" if i in (1, 3) else "0", c + "UNIT": "VOLT", c + "PROB": 10.0,
                           c + "COUP": "AC" if i == 1 else "DC", c + "BWL": "OFF", c + "SCAL": 1.0,
                           c + "OFFS": 0.5 if i == 1 else 0.0})
            if self.mho:
                self.v[c + "IMP"] = "OMEG"
        self.v.update({":ACQ:TYPE": "AVER", ":TIM:MAIN:SCAL": 5e-7, ":TIM:MAIN:OFFS": 1e-6, ":TRIG:MODE": "EDGE",
                       ":TRIG:EDGE:SOUR": "CHAN3", ":TRIG:EDGE:SLOP": "NEG", ":TRIG:SWE": "AUTO",
                       ":WAV:SOUR": "CHAN3", ":WAV:MODE": "RAW", ":WAV:FORM": "BYTE", ":WAV:STAR": 1.0,
                       ":WAV:STOP": 1000.0, ":MATH1:DISP": "0", ":MATH1:OPER": "ADD", ":MATH1:SOUR1": "CHAN1",
                       ":MATH1:SOUR2": "CHAN2", ":MATH1:SCAL": 1.0, ":MATH1:OFFS": 0.0, ":CURS:MODE": "OFF",
                       ":CURS:MAN:TYPE": "TIME", ":CURS:MAN:SOUR": "CHAN1", ":CURS:MAN:CAY": 0.0, ":CURS:MAN:CBY": 0.0,
                       ":SAVE:OVER": "0"})
        self.levels = {"CHAN%d" % i: 0.1 * i for i in range(1, 5)}
        self.run = "RUN"
        self.initial = (json.dumps(self.v, sort_keys=True), json.dumps(self.levels, sort_keys=True), self.run)
        self.record = {}
        self.srv = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        self.srv.bind(("127.0.0.1", 0))
        self.srv.listen(4)
        self.port = self.srv.getsockname()[1]
        self.alive = True
        threading.Thread(target=self._accept, daemon=True).start()

    def close(self):
        self.alive = False
        try:
            self.srv.close()
        except OSError:
            pass

    def restored(self):
        return (json.dumps(self.v, sort_keys=True), json.dumps(self.levels, sort_keys=True), self.run) == self.initial

    def power_cycle(self):
        if not self.keeps:
            self.v[":CHAN1:UNIT"], self.v[":CHAN1:PROB"] = "VOLT", 1.0
        self.run = "RUN"

    def _accept(self):
        while self.alive:
            try:
                c, _ = self.srv.accept()
            except OSError:
                return
            threading.Thread(target=self._serve, args=(c,), daemon=True).start()

    def _serve(self, c):
        buf = b""
        with c:
            while True:
                try:
                    d = c.recv(65536)
                except OSError:
                    return
                if not d:
                    return
                buf += d
                while b"\n" in buf:
                    line, _, buf = buf.partition(b"\n")
                    out = self.handle(line.decode("ascii").strip())
                    if out is not None:
                        try:
                            c.sendall(out if isinstance(out, bytes) else (out + "\n").encode("ascii"))
                        except OSError:
                            return

    def err(self, code, text):
        self.errors.append('%d,"%s"' % (code, text))

    def fmt(self, k, v):
        if isinstance(v, float):
            if k.endswith(":PROB") and not self.mho:
                return "%g" % v
            return "%.6E" % v
        return v

    def handle(self, line):
        self.lines.append(line)
        head, _, arg = line.partition(" ")
        arg = arg.strip()
        q = head.endswith("?")
        k = re.sub(r"[a-z]", "", head.rstrip("?")).upper()
        if k == "*IDN":
            return "RIGOL TECHNOLOGIES,%s,%s,%s" % (self.model, self.serial, self.firmware)
        if k == "*OPC":
            return "1"
        if k == ":SYST:ERR":
            return self.errors.pop(0) if self.errors else '0,"No error"'
        if k == ":SYST:DATE":
            return "2026,10,17"
        if k == ":SYST:TIME":
            return "12,00,00"
        if k in (":RUN", ":STOP", ":SING", ":TFOR"):
            if k == ":SING":
                self.acquire()
            elif k == ":RUN":
                self.run = "RUN"
            elif k == ":STOP":
                self.run = "STOP"
            elif k == ":TFOR" and self.run == "WAIT":
                self.acquire(force=True)
            return None
        if k == ":TRIG:STAT":
            return {"RUN": "AUTO" if self.v[":TRIG:SWE"] == "AUTO" else "TD", "STOP": "STOP", "WAIT": "WAIT"}[self.run]
        if k == ":TRIG:EDGE:LEV":
            src = self.v[":TRIG:EDGE:SOUR"]
            if q:
                return "%.6E" % self.levels[src]
            self.levels[src] = float(arg)
            return None
        if k == ":WAV:PRE":
            return self.preamble()
        if k == ":WAV:DATA":
            return self.data()
        if k in (":CURS:MAN:AYV", ":CURS:MAN:BYV"):
            return "%.6E" % self.v[":CURS:MAN:CAY" if k.endswith("AYV") else ":CURS:MAN:CBY"]
        if k in (":SAVE:SET", ":LOAD:SET"):
            if not self.file_ok:
                self.err(-257, "File name error")
            elif k == ":SAVE:SET":
                self.files[arg] = dict(self.v)
            elif arg in self.files:
                self.v.update(self.files[arg])
            else:
                self.err(-256, "File name not found")
            return None
        if k not in self.v:
            self.err(-113, "Undefined header")
            return None
        if q:
            return self.fmt(k, self.v[k])
        return self.set(k, arg)

    def set(self, k, arg):
        cur = self.v[k]
        if isinstance(cur, float):
            try:
                x = float(arg)
            except ValueError:
                self.err(-224, "Illegal parameter value")
                return None
            if k.endswith(":PROB"):
                n = k[5]
                amp = self.v[":CHAN%s:UNIT" % n] == "AMP"
                listed = any(abs(x - r) <= r * 1e-6 for r in self.ratios)
                if amp and self.mho and x > 10 + 1e-9:
                    self.err(-224, "Illegal parameter value")
                    return None
                if not listed and not self.custom_ok:
                    self.err(-224, "Illegal parameter value")
                    return None
            self.v[k] = x
            return None
        w = short_form(arg)
        if k.endswith((":DISP", ":OVER")) and w.upper() in ("ON", "OFF"):
            w = "1" if w.upper() == "ON" else "0"
        if k.endswith(":UNIT") and w.startswith("AMP") and not self.units_ok:
            self.err(-221, "Settings conflict")
            return None
        self.v[k] = w
        return None

    # -- acquisitions
    def display(self, n, t):
        unit = self.v[":CHAN%d:UNIT" % n]
        p = self.v[":CHAN%d:PROB" % n]
        v = self.world.probe_volts(t) if n == 1 else self.world.supply_volts(t) / 10.0
        if unit == "AMP":
            return v / p if self.convention == "va" else v * p
        return v * p

    def acquire(self, force=False):
        tb = self.v[":TIM:MAIN:SCAL"]
        ts = [-5 * tb + i * (10 * tb / 1000.0) for i in range(1000)]
        rec = {"CHAN%d" % n: [self.display(n, t) for t in ts] for n in (1, 2)}
        if self.v[":MATH1:OPER"] == "MULT":
            a, b = rec[self.v[":MATH1:SOUR1"]], rec[self.v[":MATH1:SOUR2"]]
            rec["MATH1"] = [x * y for x, y in zip(a, b)]
        if self.v[":TRIG:SWE"] == "NORM" and not force:
            src = self.v[":TRIG:EDGE:SOUR"]
            lv = self.levels[src]
            y = rec.get(src, [])
            if not any(y[i] < lv <= y[i + 1] for i in range(len(y) - 1)):
                self.run = "WAIT"
                return
        self.record = {"t0": ts[0], "dt": ts[1] - ts[0], "y": rec}
        self.run = "STOP"

    def _scale(self, src):
        return self.v[":MATH1:SCAL"] if src == "MATH1" else self.v[":%s:SCAL" % src]

    def preamble(self):
        src = self.v[":WAV:SOUR"]
        yinc = self._scale(src) * 8 / 50000.0
        return "1,0,1000,1,%.6E,%.6E,0,%.6E,0,32768" % (self.record.get("dt", 1e-5), self.record.get("t0", 0.0), yinc)

    def data(self):
        src = self.v[":WAV:SOUR"]
        sc = self._scale(src)
        off = self.v[":MATH1:OFFS"] if src == "MATH1" else self.v[":%s:OFFS" % src]
        yinc = sc * 8 / 50000.0
        y = self.record.get("y", {}).get(src, [0.0] * 1000)
        a, b = int(self.v[":WAV:STAR"]), int(self.v[":WAV:STOP"])
        out = bytearray()
        for v in y[a - 1:b]:
            pos = max(-4.1, min(4.1, (v + off) / sc))
            code = int(round((pos * sc - off) / yinc)) + 32768
            out += struct.pack("<H", max(0, min(65535, code)))
        head = str(len(out)).encode()
        return b"#" + str(len(head)).encode() + head + bytes(out) + b"\n"


# ============================================================================================ the command line
def add_run_args(p, test):
    sources = ("manual", "none") + (tuple(PLUGIN.sources()) if PLUGIN is not None else ())
    default = (PLUGIN.sources()[0] if PLUGIN is not None else "manual") if test in NEEDS_CURRENT else "none"
    p.add_argument("--host", help="the scope, HOST or HOST:PORT")
    p.add_argument("--firmware", default=None, help="a label of what the scope runs (stock, a mod's name, ...)")
    p.add_argument("--label", default=None, help="a free label kept in the record and in the default folder name")
    p.add_argument("--range", choices=sorted(RANGES), default="5A", help="the probe's range button (default 5A)")
    p.add_argument("--ch", type=int, default=1, help="the channel the probe is on (default 1)")
    p.add_argument("--turns", type=int, default=1, help="turns of the lead through the jaws (a custom ratio when "
                                                         "turns x sensitivity is off the guide's list)")
    p.add_argument("--ratio-convention", choices=("guide", "multiplier"), default="guide",
                   help="how the ratio is sent in amperes: guide = the sensitivity in V/A (MHO900 guide 3.6.9), "
                        "multiplier = its inverse")
    p.add_argument("--records", type=int, default=5, help="single acquisitions averaged a reading (default 5)")
    p.add_argument("--settle", type=float, default=1.5, help="seconds after a current change (default 1.5)")
    p.add_argument("--top", type=float, default=STEPS[-1], help="the highest current step, A (default %g)" % STEPS[-1])
    p.add_argument("--source", choices=sources, default=default,
                   help="who sets the current and reads the reference: manual (you type each reading), none")
    p.add_argument("--ref-acc", type=ref_acc, default=(0.0015, 0.005), metavar="REL,ABS",
                   help="the reference ammeter's accuracy, for the datasheet budget (default 0.0015,0.005: "
                        "0.15 %% + 5 mA)")
    p.add_argument("--volts", type=float, default=12.0, help="test 2: the supply's voltage (sets CH2's scale)")
    p.add_argument("--vch", type=int, default=2, help="test 2: the channel with the 10x probe on the supply output")
    p.add_argument("--minutes", type=float, default=30.0, help="test 4: how long (default 30)")
    p.add_argument("--interval", type=float, default=60.0, help="test 4: seconds between readings (default 60)")
    p.add_argument("--setup-path", default=SETUP_FILE, help="persist-setup: the setup file on the scope")
    p.add_argument("--state", default="current-probe-persist-state.json", help="persist-arm / persist-check")
    p.add_argument("--out-dir", default=None)
    p.add_argument("--timeout", type=float, default=5.0)
    p.add_argument("--untested-model", action="store_true")
    p.add_argument("--echo", action="store_true", help="print every command as it is sent")
    p.add_argument("--quiet", action="store_true")
    p.add_argument("--dry-run", action="store_true", help="run against a simulated scope and bench and print every "
                                                          "command the run would send; touches no instrument")
    p.add_argument("--dry-model", default=None, help="--dry-run: the simulated model (MHO934 or DHO924S)")
    if PLUGIN is not None:
        PLUGIN.add_args(p, test)


def ref_acc(text):
    try:
        rel, ab = (float(x) for x in text.split(","))
    except ValueError:
        raise argparse.ArgumentTypeError("REL,ABS, for example 0.0015,0.005")
    return rel, ab


def test_plan_text(a):
    """What one test will do, in its own numbers."""
    r = RANGES[a.range]
    t = a.test
    sens = r["sens"] * a.turns
    lines = ["CH%d to amperes, ratio %.6g (%s), 1 MOhm, DC, 20 MHz band limit; the probe on its %s range "
             "(%g V/A x %d turn(s))" % (a.ch, sens if a.ratio_convention == "guide" else 1 / sens, a.ratio_convention,
                                      a.range, r["sens"], a.turns)]
    if t == "amps":
        lines.append("steps %s A, each a pair (load off = the zero, load on) of %d singles at 10 ms/div" % (
            " / ".join("%g" % x for x in STEPS if x <= a.top + 1e-9), a.records))
        lines.append("pass: |reading - zero - reference| <= %g %% + %g mA; the trigger level at 1 A and cursor A "
                     "read back in amperes" % (r["band_rel"] * 100, r["band_abs"] * 1000))
    elif t == "power":
        lines.append("CH%d 10x on the supply; math CH%d x CH%d at %s A; pass within %g %% of the supply's V x I" % (
            a.vch, a.ch, a.vch, " / ".join("%g" % x for x in POWER_STEPS if x <= a.top + 1e-9), POWER_BAND * 100))
    elif t == "step":
        lines.append("load %g <-> %g A at %g kHz, %g %% duty; edge trigger CH%d rising at %g A, NORMal, 200 us/div; "
                     "%d singles must fire" % (STEP["a"], STEP["b"], STEP["freq_khz"], STEP["duty"], a.ch,
                                               STEP["level"], STEP["singles"]))
    elif t == "drift":
        lines.append("nothing in the jaws; the mean every %g s for %g min" % (a.interval, a.minutes))
    elif t == "persist-setup":
        lines.append(":SAVE:SETup %s, the unit and ratio changed, :LOAD:SETup, read back" % a.setup_path)
    elif t in ("persist-arm", "persist-check"):
        lines.append("arm: the unit and ratio set and left; power-cycle the scope; check: read back, then restore")
    lines.append("reference: %s" % {"manual": "typed per step", "none": "none"}.get(
        a.source, "the %s source of the bench module" % a.source))
    return lines


def cmd_dry_run(a, test):
    """The run against SimScope + SimBench: prints the plan and each command, sends nothing to a real instrument."""
    model = a.dry_model or (PLUGIN.dry_model(a) if PLUGIN is not None else None) or "MHO934"
    world = SimWorld(sens=RANGES[a.range]["sens"], turns=a.turns)
    sim = SimScope(model=model, world=world)
    a.echo = True
    tmp = None
    if not a.out_dir:
        tmp = tempfile.mkdtemp(prefix="cpkit-dry-")
        a.out_dir = tmp
    if test == "persist-check":
        a.state = str(Path(a.out_dir) / "persist-state.json")
        b = argparse.Namespace(**vars(a))
        b.test, b.quiet = "persist-arm", True
        run_test(b, sim=sim)
        sim.power_cycle()
    if test == "drift":
        a.minutes, a.interval = min(a.minutes, 3.0), min(a.interval, 60.0)
    print("DRY RUN (a simulated %s and bench; nothing is sent to an instrument): %s" % (model, test))
    for line in test_plan_text(a):
        print("  plan: " + line)
    print("  every command follows on stderr, as it would be sent")
    try:
        return run_test(a, sim=sim)
    finally:
        sim.close()
        print("dry-run files in %s" % a.out_dir)


def main(argv=None):
    ap = argparse.ArgumentParser(description=__doc__.split("\n\n")[0])
    ap.add_argument("--version", action="version", version=VERSION)
    sub = ap.add_subparsers(dest="cmd", required=True)
    extra = PLUGIN.commands() if PLUGIN is not None else {}
    for name, (setup, _) in extra.items():
        setup(sub.add_parser(name, help=(setup.__doc__ or "").split("\n")[0]))
    for t in TESTS:
        add_run_args(sub.add_parser(t, help=(TESTS[t][0].__doc__ or "").split("\n")[0]), t)
    rp = sub.add_parser("report", help="one table over result JSONs")
    rp.add_argument("files", nargs="+")
    a = ap.parse_args(argv)
    if a.cmd in extra:
        return extra[a.cmd][1](a)
    if a.cmd == "report":
        return cmd_report(a)
    a.test = a.cmd
    if a.host and ":" in a.host:
        a.host, port = a.host.rsplit(":", 1)
        a.port = int(port)
    else:
        a.port = SCPI_PORT
    if a.test == "persist-check" and not a.dry_run and not Path(a.state).is_file():
        raise SystemExit("no state file %s: run persist-arm first" % a.state)
    if a.turns < 1:
        raise SystemExit("--turns is 1 or more")
    if a.dry_run:
        return cmd_dry_run(a, a.test)
    return run_test(a)


if __name__ == "__main__":
    sys.exit(main())