mirror of
https://github.com/openglow-org/forgefirm.git
synced 2026-10-18 02:21:41 -07:00
ScottW514
5f19ae01cd
The lens now takes its hall-edge reference before any controller starts, so Z is referenced on every start and M103 is gone. The laser-stream harness opened its Z session by referencing the lens the way a commissioning card did; it no longer has to, because Z is already open by the time the session runs. forgectrl.panel-serves gains the assertions for the per-axis reference: homed_axes is an axis mask, homed agrees with it, and with a controller running Z is referenced and reads inside the lens reach the same document reports. That last check is the one that catches a panel showing nothing for a Z the controller holds. commission.check-motion already exercised the new path, because the motion wizard's probe runs the same sequence the supervisor does, so its covers map gains lenshome.c and its description names the lens reference and the hard fault behind it.
331 lines
19 KiB
Python
331 lines
19 KiB
Python
"""commission.check-* - the setup's checks (the dark wizards), driven the
|
|
way the page drives them: POST /wiz/<id>/start, GET /wiz/dark polled,
|
|
the prompts answered from here (the bench fixture works the lid and the
|
|
button; a confirmation is answered yes once the snapshot exists), the
|
|
result judged, and every setting a check wrote put back as found. Each
|
|
check re-runs on the real record: a completed check completes again at
|
|
the same version, so the record reads as before.
|
|
"""
|
|
import json
|
|
import os
|
|
import time
|
|
from ..catalog import test
|
|
from .. import hw
|
|
from ..baseline import read_position
|
|
from .commission import wiz, Restore # noqa: F401 - Restore is re-exported for the sheet
|
|
|
|
POLL_S = 1.0
|
|
DARK_COVERS = [("forgectrl", "src/wizdark.*"), ("forgectrl", "src/wizcalc.*"),
|
|
("forgectrl", "src/wiz.*"), ("forgectrl", "src/commission.*"),
|
|
("forgectrl", "src/main.c"), ("forgectrl", "src/ui/wizard.*")]
|
|
|
|
|
|
def dark(fc):
|
|
st, body = fc.get("/wiz/dark")
|
|
if st != 200 or not isinstance(body, dict):
|
|
raise hw.HwError("GET /wiz/dark -> %s" % st)
|
|
return body
|
|
|
|
|
|
def start(ctx, fc, wid):
|
|
ev = ctx.evidence
|
|
st, body = fc.post("/wiz/%s/start" % wid)
|
|
if st == 409 and dark(fc).get("id") == wid and dark(fc).get("running"):
|
|
# The same check left running by an earlier run of this test (an
|
|
# abort mid-wait): this run owns the machine, so it ends that one
|
|
# and starts its own.
|
|
ctx.log("the %s check is still running from an earlier run: aborting it", wid)
|
|
fc.post("/wiz/%s/abort" % wid)
|
|
ctx.wait_for(lambda: not dark(fc).get("running"), 30)
|
|
st, body = fc.post("/wiz/%s/start" % wid)
|
|
ctx.log("POST /wiz/%s/start -> %s %s", wid, st, body if isinstance(body, dict) else "")
|
|
ctx.check(st == 200, "the %s check did not start (%s %s)", wid, st, body)
|
|
ev["started"] = st == 200
|
|
|
|
|
|
def answer(fc, wid, prompt, value):
|
|
return fc.post("/wiz/%s/answer" % wid, data={"seq": str(prompt["seq"]), "value": value})
|
|
|
|
|
|
def run_check(ctx, wid, on_prompt, timeout_s):
|
|
"""Start `wid` and follow it to its end. on_prompt(prompt) is called
|
|
once per prompt (by sequence number) and returns the value to answer
|
|
with, or None when it handled the prompt another way (a fixture
|
|
action). Returns the final status document."""
|
|
fc = ctx.forgectrl
|
|
ev = ctx.evidence
|
|
start(ctx, fc, wid)
|
|
t0 = time.time()
|
|
seen = set()
|
|
phases = []
|
|
last = None
|
|
try:
|
|
while time.time() - t0 < timeout_s:
|
|
ctx.checkpoint()
|
|
d = dark(fc)
|
|
if d.get("id") != wid:
|
|
raise hw.HwError("another check took the slot: %s" % d.get("id"))
|
|
if d.get("phase") and (not phases or phases[-1] != d["phase"]):
|
|
phases.append(d["phase"])
|
|
ctx.log("%s: %s", wid, d["phase"])
|
|
p = d.get("prompt")
|
|
if p and p.get("seq") not in seen:
|
|
seen.add(p["seq"])
|
|
ctx.log("prompt %s (%s): %s", p.get("id"), p.get("kind"), p.get("text"))
|
|
value = on_prompt(p)
|
|
if value is not None:
|
|
st, body = answer(fc, wid, p, value)
|
|
ctx.check(st == 200, "the answer to %s -> %s %s", p.get("id"), st, body)
|
|
if not d.get("running"):
|
|
last = d
|
|
break
|
|
time.sleep(POLL_S)
|
|
except BaseException:
|
|
# An abort or a failure on the way out leaves no check running
|
|
# behind this test: the next run must be able to start its own.
|
|
fc.post("/wiz/%s/abort" % wid)
|
|
raise
|
|
ev["elapsed_s"] = round(time.time() - t0, 1)
|
|
ev["phases"] = phases[-12:]
|
|
ctx.check(last is not None, "the %s check did not end within %d s", wid, timeout_s)
|
|
if last is None:
|
|
fc.post("/wiz/%s/abort" % wid)
|
|
return {}
|
|
ev["log"] = last.get("log", [])[-12:]
|
|
ev["error"] = last.get("error")
|
|
ev["result"] = last.get("result")
|
|
ctx.check(not last.get("error"), "the %s check failed: %s", wid, last.get("error"))
|
|
ctx.check(isinstance(last.get("result"), dict), "the %s check ended without a result", wid)
|
|
w = wiz(fc)
|
|
ctx.check((w.get("versions") or {}).get(wid) == 1, "the record does not carry %s at version 1", wid)
|
|
return last
|
|
|
|
|
|
@test("commission.check-switches", title="The switches check follows the lid and the button",
|
|
subsystem="commission", kind="operator", est_min=3,
|
|
covers=DARK_COVERS + [("forgectrl", "src/status.c")],
|
|
requires=["forgectrl.auth"], actions=["lid", "button"],
|
|
description="POST /wiz/switches/start; the check asks for the lid to open and close and for "
|
|
"a press, each as a 'wait' prompt that the switch edge itself answers; the bench "
|
|
"fixture works the lid and the button. On a Basic or Plus the interlock loop reads "
|
|
"satisfied without a prompt. The result says lid, button, head present, and "
|
|
"whether the HV enable readback followed the lid; the record carries switches at "
|
|
"version 1.")
|
|
def check_switches(ctx):
|
|
def on_prompt(p):
|
|
pid = p.get("id", "")
|
|
if pid in ("lid-open", "lid-close-first", "lid-close"):
|
|
ctx.act("lid", "open" if pid == "lid-open" else "close",
|
|
until=lambda: (dark(ctx.forgectrl).get("prompt") or {}).get("seq") != p["seq"],
|
|
text=p.get("text", ""))
|
|
elif pid == "button-press":
|
|
ctx.act("button", "press",
|
|
until=lambda: (dark(ctx.forgectrl).get("prompt") or {}).get("id") != "button-press",
|
|
text=p.get("text", ""))
|
|
elif pid == "button-release":
|
|
pass # the fixture's press releases on its own
|
|
elif pid.startswith("interlock-"):
|
|
ctx.act("interlock", "open" if pid == "interlock-open" else "close",
|
|
until=lambda: (dark(ctx.forgectrl).get("prompt") or {}).get("seq") != p["seq"],
|
|
text=p.get("text", ""))
|
|
return None
|
|
last = run_check(ctx, "switches", on_prompt, 600)
|
|
r = last.get("result") or {}
|
|
ctx.check(r.get("lid") is True and r.get("button") is True, "lid or button not proven: %s", r)
|
|
ctx.check(r.get("head_present") is True, "the head did not read present: %s", r)
|
|
ctx.log("switches: %s", json.dumps(r))
|
|
|
|
|
|
@test("commission.check-sensors", title="The sensors check reads a plausible machine at rest",
|
|
subsystem="commission", kind="auto", est_min=2,
|
|
covers=DARK_COVERS + [("forgectrl", "src/status.c"), ("forgectrl", "src/accel.c"),
|
|
("forgectrl", "src/cool.c")],
|
|
requires=["forgectrl.auth"],
|
|
description="POST /wiz/sensors/start; ten seconds of readings, then the room-temperature "
|
|
"prompt, answered Skip so no offset is written. The result carries both coolant "
|
|
"temperatures, the chassis and SoC, the lid IR maxima, the accelerometer event "
|
|
"count, the supply power-good, the HV current, and the idle fan speeds; the "
|
|
"record carries sensors at version 1 and cool_temp_offset_c reads as before.")
|
|
def check_sensors(ctx):
|
|
with Restore(ctx, ["cool_temp_offset_c"]):
|
|
last = run_check(ctx, "sensors", lambda p: "Skip" if p.get("id") == "room-temp" else None, 120)
|
|
r = last.get("result") or {}
|
|
for k in ("coolant_down_c", "coolant_up_c", "lid_ir_max", "laser_pgood", "hv_current_max",
|
|
"exhaust_rpm_idle", "intake_rpm_idle"):
|
|
ctx.check(k in r, "the result lacks %s: %s", k, r)
|
|
ctx.check(r.get("laser_pgood") == 1, "power-good read %s", r.get("laser_pgood"))
|
|
ctx.log("sensors: coolant %s/%s C, IR %s, HV %s", r.get("coolant_down_c"), r.get("coolant_up_c"),
|
|
r.get("lid_ir_max"), r.get("hv_current_max"))
|
|
|
|
|
|
@test("commission.check-airflow", title="The airflow check measures the fans and sets the floors",
|
|
subsystem="commission", kind="auto", hardware="takeover", est_min=3,
|
|
covers=DARK_COVERS + [("forgectrl", "src/cool.c"), ("forgectrl", "src/airflow.*"),
|
|
("forgectrl", "src/gates.c"), ("forgectrl", "src/super.c")],
|
|
requires=["forgectrl.auth", "cooling.fan-gate-trips"],
|
|
description="POST /wiz/airflow/start: the controller stops, the fans run at the cut profile "
|
|
"for 35 s, the purge is measured on and off, and five settings are written: the "
|
|
"four floors at 55 percent of steady and the grace from the slowest spin-up. Each "
|
|
"fan must read at least 1000 rpm. The five settings are put back as found at the "
|
|
"end and the controller must be running again.")
|
|
def check_airflow(ctx):
|
|
keys = ["cool_tach_exhaust_min_rpm", "cool_tach_intake_min_rpm", "cool_tach_air_assist_min_rpm",
|
|
"cool_purge_min_current", "cool_fan_grace_s"]
|
|
with Restore(ctx, keys):
|
|
last = run_check(ctx, "airflow", lambda p: None, 240)
|
|
r = last.get("result") or {}
|
|
fans = r.get("fans") or {}
|
|
for name in ("exhaust", "intake 1", "intake 2", "air assist"):
|
|
f = fans.get(name) or {}
|
|
ctx.check(f.get("ok") is True, "%s: %s", name, f)
|
|
ctx.log("%s: %s rpm steady, 90 percent at %s s", name, f.get("steady_rpm"), f.get("spinup_s"))
|
|
floors = r.get("floors") or {}
|
|
for k in keys:
|
|
ctx.check(k in floors, "no floor for %s: %s", k, floors)
|
|
s = ctx.forgectrl.settings() or {}
|
|
for k in keys:
|
|
ctx.check(s.get(k) == floors.get(k), "%s reads %r, the check wrote %r", k, s.get(k), floors.get(k))
|
|
ctx.log("floors: %s", json.dumps(floors))
|
|
ok = ctx.wait_for(lambda: (ctx.forgectrl.get("/mode")[1] or {}).get("controller") == "running", 60)
|
|
ctx.check(ok is not None, "the controller did not come back after the check")
|
|
|
|
|
|
@test("commission.check-cameras", title="The cameras check captures both cameras",
|
|
subsystem="commission", kind="operator", est_min=2,
|
|
covers=DARK_COVERS + [("forgectrl", "src/cam.c")],
|
|
requires=["forgectrl.auth", "camera.snapshot"], actions=["lid"],
|
|
description="POST /wiz/cameras/start with the lid closed: a lid snapshot, the question, a "
|
|
"head snapshot, the question. Each question is answered yes once GET /wiz/shot "
|
|
"serves the snapshot as image/jpeg. The result names the sensor and says both "
|
|
"views were accepted.")
|
|
def check_cameras(ctx):
|
|
fc = ctx.forgectrl
|
|
ev = ctx.evidence
|
|
ctx.act("lid", "close")
|
|
shots = {}
|
|
|
|
def on_prompt(p):
|
|
pid = p.get("id", "")
|
|
if pid in ("lid-view", "head-view"):
|
|
cam = "lid" if pid == "lid-view" else "head"
|
|
st, body = fc.get("/wiz/shot", params={"cam": cam}, raw=True)
|
|
shots[cam] = {"status": st, "bytes": len(body) if body else 0,
|
|
"jpeg": bool(body) and body[:2] == b"\xff\xd8"}
|
|
ctx.check(st == 200 and shots[cam]["jpeg"], "no %s snapshot to judge: %s", cam, shots[cam])
|
|
return "Yes"
|
|
if pid == "lid-close":
|
|
ctx.act("lid", "close", until=lambda: (dark(fc).get("prompt") or {}).get("seq") != p["seq"])
|
|
return None
|
|
last = run_check(ctx, "cameras", on_prompt, 180)
|
|
ev["shots"] = shots
|
|
r = last.get("result") or {}
|
|
ctx.check(r.get("lid_ok") is True and r.get("head_ok") is True, "a view was not accepted: %s", r)
|
|
ctx.check(bool(r.get("sensor")) and r.get("sensor") != "unknown", "no sensor named: %s", r)
|
|
ctx.log("cameras: %s, lid %d bytes, head %d bytes", r.get("sensor"),
|
|
shots.get("lid", {}).get("bytes", 0), shots.get("head", {}).get("bytes", 0))
|
|
|
|
|
|
@test("commission.check-motion", title="The motion check proves the rail, the lens reference, and the jogs",
|
|
subsystem="commission", kind="auto", hardware="takeover", est_min=5,
|
|
covers=DARK_COVERS + [("forgectrl", "src/super.c"), ("forgectrl", "src/liveness.c"),
|
|
("forgectrl", "src/lenshome.c"),
|
|
("forgectrl", "src/accel.c"), ("forgectrl", "src/cool.c")],
|
|
requires=["forgectrl.auth", "motion.pacing"],
|
|
description="POST /wiz/motion/start: the controller stops, the liveness probe runs and the "
|
|
"lens takes its hall-edge reference behind it (a lens that cannot reach its "
|
|
"edge is a motion fault, and no controller starts), the lens "
|
|
"finds the hall reference in five agreeing passes, the controller comes back in "
|
|
"loopback posture, and the head jogs 50 mm each way on X and Y with the "
|
|
"accelerometer as the witness. The one prompt (the jogs are about to move) is "
|
|
"answered Continue. The result carries the probe line, the passes, and a "
|
|
"witnessed reading for each jog; the controller must be back in its normal "
|
|
"posture, and the kernel position counters must read what they read before: "
|
|
"every move played to its end (the probe's return leg under the cooling "
|
|
"engine's dead-man, the last jog under the controller stop).")
|
|
def check_motion(ctx):
|
|
fc = ctx.forgectrl
|
|
before = read_position()
|
|
ctx.notice("The head moves 50 mm each way on X and on Y. Keep the bed clear.")
|
|
last = run_check(ctx, "motion", lambda p: "Continue" if p.get("id") == "jogs" else None, 420)
|
|
ctx.clear_notice()
|
|
r = last.get("result") or {}
|
|
ctx.check(r.get("z_referenced") is True, "the lens was not referenced: %s", r)
|
|
ctx.check(len(r.get("z_passes") or []) == 5, "not five reference passes: %s", r.get("z_passes"))
|
|
moves = r.get("moves") or {}
|
|
for name in ("+X", "-X", "+Y", "-Y"):
|
|
m = moves.get(name) or {}
|
|
ctx.check(m.get("witnessed") is True, "the %s jog was not witnessed: %s", name, m)
|
|
ok = ctx.wait_for(lambda: (fc.get("/mode")[1] or {}).get("local") is False
|
|
and (fc.get("/mode")[1] or {}).get("controller") == "running", 90)
|
|
ctx.check(ok is not None, "the controller is not back in its normal posture")
|
|
after = read_position()
|
|
if before is not None and after is not None:
|
|
ctx.check(all(abs(a - b) <= 2 for a, b in zip(after[:2], before[:2])),
|
|
"the check lost motion: position %s before, %s after (steps)", before, after)
|
|
ctx.log("motion: probe '%s', passes %s, rest %s, moves %s", r.get("probe"), r.get("z_passes"),
|
|
r.get("rest"), {k: (v.get("p2p_lp_x"), v.get("p2p_lp_y")) for k, v in moves.items()})
|
|
|
|
|
|
@test("commission.check-flow-verify", title="The flow check runs as a setup check",
|
|
subsystem="commission", kind="auto", hardware="takeover", est_min=5,
|
|
covers=DARK_COVERS + [("forgectrl", "src/diag.c")],
|
|
requires=["forgectrl.auth", "cooling.aa-offset-calibrate", "cooling.flow-verify"],
|
|
description="POST /wiz/cooling.flow-verify/start drives the flow-verify diagnostic through the "
|
|
"check runner: the diagnostic's phases show as the check's, its result becomes the "
|
|
"check's result with pass true, and the record carries cooling.flow-verify at "
|
|
"version 1. A thin margin recommends the calibration in the record's flags. The "
|
|
"air-assist offset calibration is named first among the prerequisites so a queue "
|
|
"keeps it ahead of every heater tool this check pulls forward.")
|
|
def check_flow_verify(ctx):
|
|
last = run_check(ctx, "cooling.flow-verify", lambda p: None, 600)
|
|
r = last.get("result") or {}
|
|
ctx.check(r.get("pass") is True, "the flow check did not pass: %s", r)
|
|
ctx.log("flow check: threshold %s, flow %s, no-flow %s, thin %s", r.get("threshold"),
|
|
r.get("flow_rise"), r.get("noflow_rise"), r.get("thin_margin"))
|
|
|
|
|
|
@test("commission.cloud-header-capture", title="The cloud header check takes one print's envelope",
|
|
subsystem="commission", kind="operator", hardware="takeover", mode="grbl", est_min=8,
|
|
covers=DARK_COVERS + [("forgectrl", "src/super.c"),
|
|
("python3-gfhardware", "gfhardware/machine.py"),
|
|
("python3-gfhardware", "forgefirm-app/gfcloud.py"),
|
|
("python3-gfhardware", "forgefirm-app/ffmachine.py")],
|
|
requires=["forgectrl.auth", "cloud.mode-switch"], hands=["workstation"],
|
|
steps=["The Glowforge app open on the workstation, signed in to this machine. The test "
|
|
"turns cloud mode on itself and puts it back.",
|
|
"Once the app shows the machine online, place any small design on the bed image "
|
|
"and press Print. That is the only action: the check takes the print's header and "
|
|
"cancels the print before it arms. Do not press the machine's button.",
|
|
"The machine returns to GRBL mode on its own."],
|
|
description="POST /wiz/cloud.header/start with cloud mode on: the wizard writes the "
|
|
"client's one-start capture marker, starts cloud mode in the wizard's posture, "
|
|
"asks for a print, and takes the header the client writes to the run directory "
|
|
"before it cancels the print. The result carries the tag count, the "
|
|
"calibration-bearing tags, the limits, and the machine's own numbers; the log "
|
|
"has the tagged block; the machine is back in GRBL mode with the marker gone.")
|
|
def cloud_header_capture(ctx):
|
|
fc = ctx.forgectrl
|
|
ev = ctx.evidence
|
|
with Restore(ctx, ["cloud_enabled", "homing_mode", "controller_mode"]):
|
|
s = fc.settings() or {}
|
|
if s.get("cloud_enabled") != "1":
|
|
st, body = fc.post("/settings", data={"cloud_enabled": "1"})
|
|
ctx.check(st == 200, "cloud_enabled=1 -> %s %s", st, body)
|
|
ctx.log("cloud mode turned on for the test")
|
|
ctx.notice("Print any small design from the Glowforge app once it shows the machine "
|
|
"online. Do not press the machine's button.")
|
|
last = run_check(ctx, "cloud.header",
|
|
lambda p: "Continue" if p.get("id") == "print" else None, 900)
|
|
ctx.clear_notice()
|
|
r = last.get("result") or {}
|
|
ctx.check(isinstance(r.get("tag_count"), int) and r["tag_count"] > 100,
|
|
"the header carried %s tags", r.get("tag_count"))
|
|
kept = r.get("calibration_tags") or {}
|
|
ctx.check(len(kept) > 10, "only %d calibration-bearing tags kept", len(kept))
|
|
ev["kept"] = sorted(kept)[:40]
|
|
ctx.check(not os.path.exists("/run/gfcloud-capture"), "the capture marker is still down")
|
|
ok = ctx.wait_for(lambda: (fc.get("/mode")[1] or {}).get("mode") == "grbl"
|
|
and (fc.get("/mode")[1] or {}).get("controller") == "running", 120)
|
|
ctx.check(ok is not None, "the machine did not return to GRBL mode")
|
|
ctx.log("header: %d tags, %d kept; job %s", r.get("tag_count", 0), len(kept), r.get("job_id"))
|