forgefirm/forgetest/forgetest/suite/kernel.py

"""kernel.* - glowforge.ko safety readbacks and the pulse-engine drills.

The drills are the bench scripts `scripts/bench/gate_a_kernel_drills.py`
(K1/K2/K3) and `scripts/bench/fire_test.py` (A/B/U) with their proven
sequences kept intact; they run under a hardware takeover (forgectrl and
the controller stopped, the pulse device free) and judge the software
witnesses: cnc/state, laser_enable (the FIRE line), laser_on and
laser_on_sampled (the gated LASER_ON output), interlock_circuit bit 3
(the commanded latch), faults and underruns. Every drill forces duty to
zero before any FIRE bit, keeps motor_lock=15 (no axis moves), and
re-locks the latch on every exit path. K3 and fire B/U unlock the latch
for their run and therefore refuse to proceed while laser_pgood reports
the HV supply good (the operator opens the lid: the safety chain holds
HV off).
"""
import errno
import os
import struct
import time

try:
    import fcntl
except ImportError:          # host unit tests import the suite off-target
    fcntl = None

from ..catalog import test
from .. import hw
from ..runner import Failed

# interlock_circuit bits (UAPI.md): bit 3 = the driven latch line, set = locked.
LATCH_BIT = 1 << 3

TICK_HZ = 10000
FIRE = b"\x10"
PAD = b"\x00"
XSTEP = b"\x01"                 # +X (DIR clear)
XSTEP_BACK = b"\x03"            # -X (DIR set)
POWER0 = bytes([0x80])          # power byte, duty 0

_KERNEL_COVERS = [("kernel-module-glowforge", "**"), ("linux-fslc", "**")]


# ---------------------------------------------------------------- helpers

def wr(attr, val):
    hw.sysfs_write(attr, val)


def rd(attr):
    v = hw.sysfs_read(attr)
    if v is None:
        raise Failed("cannot read %s" % attr)
    return v


def rd_pos():
    with open(hw.sysfs_root() + "cnc/position", "rb") as f:
        raw = f.read(32)
    return struct.unpack("<5i", raw[:20])


def snap(ctx, tag):
    line = ("%s: state=%s laser_enable=%s laser_on=%s laser_on_sampled=%s interlock=%s"
            % (tag, rd("cnc/state"), rd("cnc/laser_enable"), rd("cnc/laser_on"),
               rd("cnc/laser_on_sampled"), rd("cnc/interlock_circuit")))
    ctx.log(line)
    return line


def wait_state(ctx, want, timeout, poll=0.05):
    t0 = time.time()
    while time.time() - t0 < timeout:
        ctx.checkpoint()
        s = rd("cnc/state")
        if s == want:
            return s
        time.sleep(poll)
    return rd("cnc/state")


def watch_laser_until_idle(ctx, timeout):
    """Tight-loop laser_enable/laser_on watch; returns (hits, end_state)."""
    hits = []
    t0 = time.time()
    state = "running"
    n = 0
    while time.time() - t0 < timeout:
        en = rd("cnc/laser_enable")
        on = rd("cnc/laser_on")
        if en != "0" or on != "0":
            hits.append((round(time.time() - t0, 4), en, on))
        state = rd("cnc/state")
        if state != "running":
            break
        n += 1
        if n % 200 == 0:
            ctx.checkpoint()
    return hits, state


class PulseDevice:
    """Exclusive hold of /dev/glowforge for one drill."""

    def __init__(self, ctx):
        self.ctx = ctx
        self.fd = None

    def __enter__(self):
        try:
            self.fd = os.open("/dev/glowforge", os.O_WRONLY)
        except OSError as e:
            if e.errno == errno.EBUSY:
                raise Failed("/dev/glowforge is busy - the takeover did not free the pulse device")
            raise
        fcntl.flock(self.fd, fcntl.LOCK_EX)
        return self

    def write(self, data):
        os.write(self.fd, data)

    def rewind(self):
        os.lseek(self.fd, 1, os.SEEK_SET)

    def __exit__(self, *exc):
        try:
            wr("cnc/laser_latch", 1)         # re-lock unconditionally
        finally:
            fcntl.flock(self.fd, fcntl.LOCK_UN)
            os.close(self.fd)
        return False


def require_hv_not_good(ctx):
    """K3 and fire B/U unlock the latch: refuse while HV reports good.
    Called before the takeover; gives the operator one chance to drop it
    (open the lid)."""
    pgood = rd("cnc/laser_pgood")
    if pgood != "0":
        ctx.log("laser_pgood=%s: the HV supply reports good", pgood)
        ctx.instruct("This drill unlocks the laser latch with a zero-duty stream and must run "
                     "with the HV supply NOT good. Open the lid (the safety chain holds HV off), "
                     "then Done.")
        pgood = rd("cnc/laser_pgood")
    ctx.evidence["laser_pgood"] = pgood
    ctx.check(pgood == "0", "laser_pgood=%s (HV supply reports good) - refusing the latch unlock", pgood)


def check_hv_not_good(ctx):
    """The hard check right before an unlock (no prompt: forgectrl is down)."""
    pgood = rd("cnc/laser_pgood")
    ctx.check(pgood == "0", "laser_pgood=%s (HV supply reports good) - refusing the latch unlock", pgood)


# ---------------------------------------------------------------- readbacks

@test("kernel.latch-locked-idle", title="Laser latch locked at idle", subsystem="kernel",
      kind="auto", always=True, est_min=1,
      covers=_KERNEL_COVERS + [("forgectrl", "src/super.c"),
                               ("grblhal-glowforge", "src/glowforge_laser.c"),
                               ("grblhal-glowforge", "src/driver.c")],
      description="With the machine idle the kernel latch reads locked, the FIRE line is not "
                  "driven, no LASER_ON sample is seen, and no stepper fault is pending; forgectrl "
                  "agrees.")
def latch_locked_idle(ctx):
    ev = ctx.evidence
    state = hw.sysfs_read("cnc/state")
    ev["cnc_state"] = state
    ctx.log("cnc/state: %s", state)
    ctx.check(state is not None, "cnc/state unreadable")
    ctx.check(state in ("idle", "disabled"), "machine is %r, run this test at idle", state)

    ilk = hw.sysfs_int("cnc/interlock_circuit")
    ev["interlock_circuit"] = ilk
    ctx.check(ilk is not None, "cnc/interlock_circuit unreadable")
    ctx.log("interlock_circuit: %d (0x%x)", ilk, ilk)
    ctx.check(ilk & LATCH_BIT, "latch line reads unlocked at idle (bit 3 clear)")

    fire = hw.sysfs_int("cnc/laser_enable")
    ev["laser_enable"] = fire
    ctx.log("laser_enable (FIRE line): %s", fire)
    ctx.check(fire == 0, "FIRE line driven at idle (laser_enable=%s)", fire)

    on = hw.sysfs_int("cnc/laser_on")
    on_s = hw.sysfs_int("cnc/laser_on_sampled")
    ev["laser_on"] = on
    ev["laser_on_sampled"] = on_s
    ctx.log("laser_on: %s, laser_on_sampled: %s", on, on_s)
    ctx.check(on == 0, "LASER_ON active at idle")
    ctx.check(on_s == 0, "LASER_ON samples seen at idle (%s)", on_s)

    faults = hw.sysfs_int("cnc/faults")
    ev["faults"] = faults
    ctx.log("faults: %s", faults)
    ctx.check(faults == 0, "stepper faults pending: %s", faults)

    st = ctx.forgectrl.status()
    ev["forgectrl_laser_locked"] = st.get("laser_locked")
    ctx.log("forgectrl /status laser_locked=%s state=%s", st.get("laser_locked"), st.get("state"))
    ctx.check(st.get("laser_locked") is True, "forgectrl reports the latch unlocked")


# ---------------------------------------------------------------- K1 + K2

@test("kernel.k1-k2", title="Controlled-stop floor and resume honors the locked latch",
      subsystem="kernel", kind="auto", hardware="takeover", always=True, est_min=2,
      covers=_KERNEL_COVERS,
      requires=["kernel.latch-locked-idle"],
      description="K1: a controlled stop mid-run ramps the step frequency down (tens of ms), "
                  "never consumes the tail as a burst or hangs. K2: with the latch locked, a "
                  "stop inside the leading pads and a resume with a positive waypoint replays "
                  "a 2 s FIRE window with laser_enable/laser_on at 0 throughout. Motors locked; "
                  "duty zero.")
def k1_k2(ctx):
    ev = ctx.evidence
    with ctx.takeover():
        # ---- K1
        stream = POWER0 + PAD * (6 * TICK_HZ)
        ctx.log("K1: %d bytes = %.1f s of pads at %d Hz, ramp 125000 Hz/s",
                len(stream), len(stream) / TICK_HZ, TICK_HZ)
        snap(ctx, "K1 pre")
        wr("cnc/motor_lock", 15)
        wr("cnc/laser_latch", 1)
        wr("cnc/ramp_rate", 125000)
        wr("cnc/step_freq", TICK_HZ)
        with PulseDevice(ctx) as dev:
            dev.rewind()
            wr("cnc/enable", 1)
            ctx.sleep(0.5)
            dev.write(stream)
            wr("cnc/run", 1)
            ctx.sleep(1.5)                      # well past the accel ramp
            st = rd("cnc/state")
            ctx.check(st == "running", "K1: expected running before the stop, got %s", st)
            t0 = time.time()
            wr("cnc/stop", 1)
            while time.time() - t0 < 5:
                if rd("cnc/state") != "running":
                    break
            dt = time.time() - t0
            state = rd("cnc/state")
            faults = rd("cnc/faults")
            ev["k1"] = {"stop_to_idle_s": round(dt, 4), "state": state, "faults": faults}
            ctx.log("K1 controlled stop: state=%s after %.4f s, faults=%s", state, dt, faults)
            # drain the paused remainder laser-less so the device ends clean
            wr("cnc/resume", 0)
            wait_state(ctx, "running", 2, poll=0.005)
            wait_state(ctx, "idle", 10)
        ctx.check(state == "idle", "K1: state %s after the stop", state)
        ctx.check(dt >= 0.02, "K1: stop consumed the tail as a burst (%.4f s) - decel floor broken", dt)
        ctx.check(dt <= 3.0, "K1: stop took %.4f s", dt)
        ctx.check(faults == "0", "K1: faults=%s", faults)
        ctx.log("K1 PASS: decelerating tail %.4f s, no burst, no fault", dt)

        # ---- K2
        # 1000 X steps out and 1000 back at 2 kHz (masked by motor_lock):
        # the position counters end where they started
        step_sec = (XSTEP + PAD * 4) * 1000 + (XSTEP_BACK + PAD * 4) * 1000
        stream = (POWER0 + PAD * TICK_HZ + step_sec + PAD * (TICK_HZ // 2)
                  + FIRE * (2 * TICK_HZ) + PAD * TICK_HZ)
        ctx.log("K2: %d bytes = %.1f s; latch stays LOCKED; waypoint +200; motor_lock=15",
                len(stream), len(stream) / TICK_HZ)
        snap(ctx, "K2 pre")
        wr("cnc/motor_lock", 15)
        wr("cnc/laser_latch", 1)
        wr("cnc/ramp_rate", 125000)
        wr("cnc/step_freq", TICK_HZ)
        with PulseDevice(ctx) as dev:
            dev.rewind()
            wr("cnc/enable", 1)
            ctx.sleep(0.5)
            dev.write(stream)
            pos_before = rd_pos()
            wr("cnc/run", 1)
            ctx.sleep(0.4)                      # inside the initial pads
            wr("cnc/stop", 1)
            state = wait_state(ctx, "idle", 5, poll=0.01)
            ctx.check(state == "idle", "K2: controlled stop did not reach idle (state=%s)", state)
            ctx.log("K2: paused inside the pads; resuming with waypoint +200 (latch LOCKED)")
            wr("cnc/resume", 200)
            wait_state(ctx, "running", 2, poll=0.005)
            hits, state = watch_laser_until_idle(ctx, 20)
            pos_after = rd_pos()
            snap(ctx, "K2 post")
            ev["k2"] = {"hits": hits[:10], "end_state": state, "pos_before": pos_before,
                        "pos_after": pos_after, "laser_on_sampled": rd("cnc/laser_on_sampled"),
                        "underruns": rd("cnc/underruns"), "faults": rd("cnc/faults")}
            ctx.log("K2 done: state=%s pos before=%s after=%s", state, pos_before, pos_after)
        ctx.check(not hits, "K2: laser asserted with the latch locked: %s", hits[:10])
        ctx.check(pos_before[:3] == pos_after[:3],
                  "K2: position counters did not return to start (%s -> %s)", pos_before[:3], pos_after[:3])
        ctx.log("K2 PASS: FIRE window replayed after the resume waypoint with "
                "laser_enable/laser_on at 0 throughout")


# ---------------------------------------------------------------- K3

@test("kernel.k3-unlock", title="Mid-run latch unlock never re-arms FIRE", subsystem="kernel",
      kind="operator", hardware="takeover", always=True, est_min=2,
      covers=_KERNEL_COVERS,
      requires=["kernel.k1-k2"],
      steps=["If the HV supply reports good the drill asks you to open the lid first (the "
             "safety chain holds HV off); zero duty throughout."],
      description="Stream of FIRE bits run with the latch locked (laser-less by the run-start "
                  "guard); the latch is unlocked during the accel ramp. The unlock drives the "
                  "latch pin (interlock bit 3 clears) but must not restore the FIRE drive while "
                  "the run is in flight: laser_enable stays 0 for the entire run.")
def k3_unlock(ctx):
    ev = ctx.evidence
    require_hv_not_good(ctx)
    with ctx.takeover():
        check_hv_not_good(ctx)
        stream = POWER0 + FIRE * (3 * TICK_HZ) + PAD * (TICK_HZ // 2)
        ctx.log("K3: %d bytes = %.1f s of FIRE bits; ramp_rate 10000 Hz/s (~0.9 s accel "
                "window); unlock at t=+0.15 s", len(stream), len(stream) / TICK_HZ)
        snap(ctx, "K3 pre")
        wr("cnc/motor_lock", 15)
        wr("cnc/laser_latch", 1)
        wr("cnc/step_freq", TICK_HZ)
        wr("cnc/ramp_rate", 10000)
        try:
            with PulseDevice(ctx) as dev:
                dev.rewind()
                wr("cnc/enable", 1)
                ctx.sleep(0.5)
                dev.write(stream)
                wr("cnc/run", 1)
                time.sleep(0.15)                # inside the accel ramp
                wr("cnc/laser_latch", 0)
                ilk = rd("cnc/interlock_circuit")
                ctx.log("K3: latch UNLOCKED mid-ramp; interlock=%s (bit 3 should read 0)", ilk)
                hits, state = watch_laser_until_idle(ctx, 20)
                snap(ctx, "K3 post")
                ev["k3"] = {"interlock_after_unlock": ilk, "hits": hits[:10], "end_state": state,
                            "laser_on_sampled": rd("cnc/laser_on_sampled"),
                            "underruns": rd("cnc/underruns"), "faults": rd("cnc/faults")}
        finally:
            wr("cnc/laser_latch", 1)
            try:
                wr("cnc/ramp_rate", 125000)
            except OSError:
                ctx.log("WARNING: could not restore ramp_rate=125000")
        ctx.check((int(ilk) & LATCH_BIT) == 0, "K3: the unlock did not drive the latch pin (interlock=%s)", ilk)
        ctx.check(not hits, "K3: FIRE drive re-armed by a mid-run unlock: %s", hits[:10])
        ctx.log("K3 PASS: laser_enable stayed 0 for the entire run after the mid-ramp unlock")


# ---------------------------------------------------------------- FIRE A/B/U

def _fire_stream():
    return (POWER0 +                    # duty zero before any FIRE bit
            PAD * TICK_HZ +             # 1 s baseline
            FIRE * (2 * TICK_HZ) +      # 2.000 s FIRE window (bounded by pads)
            PAD * TICK_HZ +             # 1 s gap
            FIRE * (2 * TICK_HZ))       # 2.000 s FIRE window ending AT end-of-data


def _fire_phase(ctx, mode):
    """One phase of fire_test.py; returns the evidence dict."""
    unlock = mode in ("B", "U")
    underrun_mode = mode == "U"
    stream = _fire_stream()
    ctx.log("fire %s: stream %d bytes = %.3f s", mode, len(stream), len(stream) / TICK_HZ)
    if unlock:
        check_hv_not_good(ctx)
    snap(ctx, "fire %s pre" % mode)
    wr("cnc/motor_lock", 15)
    wr("cnc/step_freq", TICK_HZ)
    wr("cnc/laser_latch", 1)
    underruns_before = int(rd("cnc/underruns"))
    mid = None
    tail = None
    with PulseDevice(ctx) as dev:
        dev.rewind()
        wr("cnc/enable", 1)
        ctx.sleep(0.5)
        dev.write(stream)
        pos_before = rd_pos()
        if underrun_mode:
            wr("cnc/streaming", 1)      # end-of-data mid-run = true underrun
            ctx.log("fire U: streaming=1, the terminal end-of-data will be a TRUE UNDERRUN")
        try:
            if unlock:
                wr("cnc/laser_latch", 0)
                ctx.log("fire %s: latch UNLOCKED for this run", mode)
            wr("cnc/run", 1)
            t0 = time.time()
            state = ""
            samples = []
            while time.time() - t0 < 20:
                ctx.checkpoint()
                state = rd("cnc/state")
                dt = time.time() - t0
                en, on, ons = rd("cnc/laser_enable"), rd("cnc/laser_on"), rd("cnc/laser_on_sampled")
                samples.append((round(dt, 2), state, en, on, ons))
                if mid is None and 1.5 < dt < 3.0:
                    mid = {"t": round(dt, 2), "laser_enable": en, "laser_on": on,
                           "laser_on_sampled": ons, "interlock": rd("cnc/interlock_circuit")}
                    ctx.log("fire %s mid (inside FIRE window): laser_enable=%s laser_on=%s "
                            "laser_on_sampled=%s interlock=%s", mode, en, on, ons, mid["interlock"])
                if state != "running":
                    break
                time.sleep(0.05)
            end_dt = time.time() - t0
            tail = {"state": state, "after_s": round(end_dt, 2), "laser_enable": rd("cnc/laser_enable"),
                    "laser_on": rd("cnc/laser_on")}
            ctx.log("fire %s done: state=%s after %.1f s (laser_enable=%s)", mode, state, end_dt,
                    tail["laser_enable"])
            if underrun_mode:
                if state == "underrun":
                    ctx.log("fire U: underrun state reached as EXPECTED; acking via stop")
                else:
                    ctx.log("fire U: WARNING expected underrun state, got %s", state)
                wr("cnc/stop", 1)
                wr("cnc/streaming", 0)
                tail["acked_state"] = rd("cnc/state")
                ctx.log("fire U: acked: state=%s", tail["acked_state"])
        finally:
            wr("cnc/laser_latch", 1)
            if underrun_mode:
                try:
                    wr("cnc/streaming", 0)
                except OSError:
                    pass
    pos_after = rd_pos()
    snap(ctx, "fire %s post" % mode)
    ev = {"mid": mid, "tail": tail, "moved": pos_before[:3] != pos_after[:3],
          "underruns_before": underruns_before, "underruns_after": int(rd("cnc/underruns")),
          "faults": rd("cnc/faults"),
          "any_laser_on": any(s[3] != "0" or s[4] != "0" for s in samples),
          "any_fire_driven": any(s[2] != "0" for s in samples)}
    ctx.log("fire %s: moved=%s underruns %d->%d faults=%s", mode, ev["moved"],
            ev["underruns_before"], ev["underruns_after"], ev["faults"])
    return ev


@test("kernel.fire-abu", title="FIRE line: latch locked, unlocked-unarmed, true underrun",
      subsystem="kernel", kind="operator", hardware="takeover", always=True, est_min=3,
      covers=_KERNEL_COVERS,
      requires=["kernel.k1-k2"],
      steps=["Phases B and U unlock the latch with a zero-duty stream: if the HV supply reports "
             "good the drill asks you to open the lid first (the safety chain holds HV off)."],
      description="A: latch locked, 40 000 streamed FIRE bits, nothing on the FIRE/LASER_ON "
                  "nets. B: latch unlocked with the chain unarmed, the FIRE line is driven "
                  "mid-window and LASER_ON stays off (the safety AND-gate holds), FIRE clear at "
                  "end-of-data. U: streaming declared, the terminal end-of-data is a true "
                  "underrun, the backstop drops FIRE and stop acks it.")
def fire_abu(ctx):
    ev = ctx.evidence
    require_hv_not_good(ctx)
    with ctx.takeover():
        try:
            a = _fire_phase(ctx, "A")
            ev["A"] = a
            ctx.check(a["mid"] is not None, "A: no mid-window sample")
            ctx.check(not a["any_fire_driven"], "A: FIRE line driven with the latch locked")
            ctx.check(not a["any_laser_on"], "A: LASER_ON seen with the latch locked")
            ctx.check(a["tail"]["state"] == "idle", "A: ended in %s", a["tail"]["state"])
            ctx.check(not a["moved"], "A: position moved with motors locked")
            ctx.log("fire A PASS: latch locked, no FIRE drive, no LASER_ON")

            b = _fire_phase(ctx, "B")
            ev["B"] = b
            ctx.check(b["mid"] is not None, "B: no mid-window sample")
            ctx.check(b["mid"]["laser_enable"] != "0",
                      "B: FIRE line not driven mid-window with the latch unlocked (%s)", b["mid"])
            ctx.check(not b["any_laser_on"], "B: LASER_ON active with the chain unarmed - the AND-gate did not hold")
            ctx.check(b["tail"]["state"] == "idle", "B: ended in %s", b["tail"]["state"])
            ctx.check(b["tail"]["laser_enable"] == "0", "B: FIRE still driven after end-of-data")
            ctx.check(b["underruns_after"] == b["underruns_before"], "B: underrun counted on a normal completion")
            ctx.log("fire B PASS: FIRE driven mid-window, LASER_ON off, FIRE clear at end-of-data")

            u = _fire_phase(ctx, "U")
            ev["U"] = u
            ctx.check(u["tail"]["state"] == "underrun", "U: expected the underrun state, got %s", u["tail"]["state"])
            ctx.check(u["tail"]["laser_enable"] == "0", "U: FIRE still driven after the underrun")
            ctx.check(not u["any_laser_on"], "U: LASER_ON active with the chain unarmed")
            ctx.check(u["tail"].get("acked_state") == "idle", "U: stop did not ack the underrun (state %s)",
                      u["tail"].get("acked_state"))
            ctx.check(u["underruns_after"] == u["underruns_before"] + 1,
                      "U: underrun counter %d -> %d", u["underruns_before"], u["underruns_after"])
            ctx.log("fire U PASS: true underrun, backstop dropped FIRE, stop acked")
        finally:
            wr("cnc/laser_latch", 1)
            try:
                wr("cnc/disable", 1)            # the script's safe state
            except OSError:
                pass
            ctx.log("safe state restored: state=%s latch=LOCKED", rd("cnc/state"))