diff --git a/forgetest/forgetest/suite/cooling.py b/forgetest/forgetest/suite/cooling.py index 4b948bf..cc6349c 100644 --- a/forgetest/forgetest/suite/cooling.py +++ b/forgetest/forgetest/suite/cooling.py @@ -3,13 +3,14 @@ # https://community.openglow.org # SPDX-License-Identifier: MIT -"""cooling.* - the cooling engine: flow verification through forgectrl's -diagnostics runner (the same check the fire gate runs), the fan profile -returning to idle after motion, the gate settings (a value inside the -legal range trips the gate, the far end of the range turns it off by -value, and both are said out loud: the settings reply, /status, the -engine's run-start log line), and the airflow gates (a fan under its -floor past the spin-up grace is a fault for the rest of the run).""" +"""cooling.* - the cooling engine on the machine: the air-assist ground +shift on the coolant readings, flow verification through forgectrl's +diagnostics runner (the same check the fire gate runs) and under a lit +tube, and the fail tier ending the controller on the head's own move +inside an armed window. The engine's decisions on a reading a host can +emulate (the gate settings, the phases' fan duties, the TEC, the fire and +crash watches' tiers and gates) are forgectrl's cool_flow_test and +cool_gate_test.""" import os import time @@ -83,15 +84,6 @@ def aa_offset_calibrate(ctx): ctx.check(fc.wait_idle(120, abort=ctx.aborted), "machine did not return to idle after the diagnostic") -IDLE_DUTY = {"thermal/exhaust_pwm": 0, "thermal/intake_pwm": 0} # forgectrl's idle posture -TACH_KEYS = ("exhaust", "intake_1", "intake_2") -SAMPLE_S = 5 # tach sampling period (the big exhaust fan coasts for tens of seconds) -STABLE_SAMPLES = 3 # consecutive agreeing samples that make an idle reference -IDLE_REF_TIMEOUT_S = 180 # a previous test's cooldown + spin-down -COOLDOWN_TIMEOUT_S = 240 # the engine's smoke phase + idle drop + spin-down -RESTART_IDLE_TIMEOUT_S = 90 # a daemon restart: the controller's start, its liveness probe, the engine's tick - - @test("cooling.flow-verify", title="Coolant flow check separates flow from no-flow", subsystem="cooling", kind="auto", est_min=4, covers=_COOL_COVERS, requires=["kernel.latch-locked-idle"], @@ -147,144 +139,6 @@ def flow_verify(ctx): AA_CAL_TIMEOUT_S = 540 # the tool's 420 s settle limit plus its six edges -def _duties(): - return {k: hw.sysfs_int(k) for k in IDLE_DUTY} - - -def _tach_close(now, ref): - """Every tach at or below its idle reference, within the tolerance a - tach reading wanders by (150 rpm or 15 %). Lower is quieter, never a - fault: the reference is an idle level, not a target.""" - return all(now.get(k, 0) - ref.get(k, 0) <= max(150, 0.15 * max(ref.get(k, 0), 1)) - for k in TACH_KEYS) - - -def _tach_stable(a, b): - return all(abs(a.get(k, 0) - b.get(k, 0)) <= max(100, 0.10 * max(a.get(k, 0), 1)) for k in TACH_KEYS) - - -@test("cooling.fans-quiet-after-motion", title="Fan profile returns to idle after motion, after M8/M9, " - "and after a daemon restart on a busy machine", - subsystem="cooling", kind="auto", mode="grbl", est_min=5, - covers=_COOL_COVERS + [("forgectrl", "src/super.c")], requires=["motion.pacing"], - steps=["Bed clear; the head needs 20 mm of free +X travel."], - description="A dry jog and an M8/M9 cycle must not leave the run fan profile on: within " - "the cooldown the engine is back at its idle duty and the exhaust/intake tachs " - "are back at (or below) the idle level they held before the test. The idle " - "reference is taken only once the engine is idle and the tachs have stopped " - "changing, so a previous test's spin-down cannot be mistaken for idle. Then a " - "forgectrl restart with the kernel not idle (a takeover's safe state): the " - "engine starts in its cooldown posture and must take the idle duties back once " - "the machine is idle, not keep the exhaust at cooldown duty for good.") -def fans_quiet(ctx): - fc = ctx.forgectrl - ev = ctx.evidence - - def fans(): - s = fc.status() - return dict(s.get("fans") or {}) - - def phase(): - st, c = fc.get("/cool/status") - return (c or {}).get("phase") if st == 200 and isinstance(c, dict) else None - - # The idle reference: engine idle, idle duty applied, and tachs that - # have stopped changing (STABLE_SAMPLES consecutive samples that - # agree). A test that ran just before leaves the fans spinning down - # for tens of seconds after the engine goes idle; a reference taken - # then is not idle, and two samples can agree by chance mid-coast. - t0 = time.time() - before = None - recent = [] - while time.time() - t0 < IDLE_REF_TIMEOUT_S: - ctx.sleep(SAMPLE_S) - now, ph, d = fans(), phase(), _duties() - ctx.log(" idle ref: phase %s duty %s fans %s", ph, d, now) - recent = (recent + [now])[-STABLE_SAMPLES:] if ph == "idle" and d == IDLE_DUTY else [] - if len(recent) == STABLE_SAMPLES and all(_tach_stable(a, b) for a, b in zip(recent, recent[1:])): - before = now - break - ev["before"] = before - ev["idle_ref_s"] = round(time.time() - t0, 1) - ctx.check(before is not None, "the fans never settled to an idle reference within %d s (last %s, " - "phase %s, duty %s)", IDLE_REF_TIMEOUT_S, recent[-1:] or None, phase(), _duties()) - ctx.log("idle reference after %s s: %s", ev["idle_ref_s"], before) - - with ctx.grbl() as g: - st = g.status_report() - ctx.check(st["state"].startswith("Idle"), "controller is %s", st["state"]) - g.command("G91") - g.command("$J=G91X20F2400") - t0 = time.time() - while time.time() - t0 < 30 and not g.status_report()["state"].startswith("Idle"): - ctx.sleep(0.2) - g.command("$J=G91X-20F2400") - while time.time() - t0 < 60 and not g.status_report()["state"].startswith("Idle"): - ctx.sleep(0.2) - g.command("M8") - ctx.sleep(3) - during = fans() - ev["during_m8"] = during - ev["duty_m8"] = _duties() - ctx.log("fans during M8: %s (duty %s)", during, ev["duty_m8"]) - ctx.check(ev["duty_m8"] != IDLE_DUTY, "M8 did not raise the fan duty off idle: %s", ev["duty_m8"]) - g.command("M9") - g.command("G90") - # Cooldown: the engine's smoke phase, then the drop to idle duty, then - # the tachs coast down. Every sample is logged: a quiet pane here - # looks like a hang. - settle = None - t0 = time.time() - while time.time() - t0 < COOLDOWN_TIMEOUT_S: - ctx.sleep(SAMPLE_S) - now, ph, d = fans(), phase(), _duties() - ctx.log(" cooldown +%3.0f s: phase %s duty %s fans %s", time.time() - t0, ph, d, now) - if ph == "idle" and d == IDLE_DUTY and _tach_close(now, before): - settle = time.time() - t0 - break - ev["after"] = fans() - ev["duty_after"] = _duties() - ev["settle_s"] = round(settle, 1) if settle is not None else None - ctx.log("fans after: %s duty %s (settled in %s s)", ev["after"], ev["duty_after"], ev["settle_s"]) - ctx.check(settle is not None, "fans did not return to the idle profile within %d s: %s, duty %s, " - "phase %s (idle reference %s)", COOLDOWN_TIMEOUT_S, ev["after"], ev["duty_after"], phase(), before) - - # A daemon restart while the kernel is not idle: forgectrl stopped, the - # kernel put in the takeover's safe state (disabled), forgectrl started. - # The engine judges the machine busy and takes the cooldown airflow; the - # controller's start enables the kernel and the machine is idle again, - # and the engine must follow it to the idle duties. - rc, out = hw.initd("forgectrl", "stop") - ctx.log("forgectrl stop -> rc %s", rc) - ctx.check(rc == 0, "forgectrl stop failed: %s", (out or "").strip()[:200]) - hw.sysfs_write("cnc/disable", "1") - ev["kstate_at_restart"] = hw.sysfs_read("cnc/state") - rc, out = hw.initd("forgectrl", "start") - ctx.log("forgectrl start with cnc/state=%s -> rc %s", ev["kstate_at_restart"], rc) - ctx.check(rc == 0, "forgectrl start failed: %s", (out or "").strip()[:200]) - back = None - t0 = time.time() - while time.time() - t0 < RESTART_IDLE_TIMEOUT_S: - ctx.sleep(SAMPLE_S) - st, m = fc.get("/mode") - d = _duties() - ctrl = m.get("controller") if isinstance(m, dict) else st - ctx.log(" after the restart +%3.0f s: controller %s duty %s", time.time() - t0, ctrl, d) - if ctrl == "running" and d == IDLE_DUTY: - back = time.time() - t0 - break - ev["restart_idle_s"] = round(back, 1) if back is not None else None - ev["duty_after_restart"] = _duties() - ctx.check(back is not None, "the engine kept the busy-start airflow after the restart (duty %s " - "after %d s)", ev["duty_after_restart"], RESTART_IDLE_TIMEOUT_S) - - -GATE_KEYS = ("cool_temp_max", "cool_temp_resume") -VERDICT_WAIT_S = 20 # the engine reloads settings at run start and ticks at 1 Hz -SESSION_END_WAIT_S = 15 # M9 -> the engine's phase leaves "run" (1 Hz reports, 1 Hz ticks) -GATE_LOG_LINES = "400" # how far back the run-start gate lines can sit in the forgectrl log - - def _cool(fc): st, c = fc.get("/cool/status") return c if st == 200 and isinstance(c, dict) else {} @@ -304,961 +158,7 @@ def _set_gates(ctx, fc, values): return body -def _session_ended(ctx, fc, what): - """After M9 the GRBL client's next report ends the engine's run - session; the client reports at 1 Hz and the engine samples at 1 Hz, - so an M8 sent inside that window is not a new session and nothing is - re-read. Wait until the phase has left "run".""" - t0 = time.time() - while time.time() - t0 < SESSION_END_WAIT_S: - ctx.sleep(1) - if _cool(fc).get("phase") != "run": - return True - ctx.log("%s: the engine is still in phase run %d s after M9", what, SESSION_END_WAIT_S) - return False - - -def _run_session(ctx, g, fc, until, what, wait=None): - """M8 opens a run session (the engine re-reads its settings there and - ticks the gates at 1 Hz); wait for `until(cool)` to hold, then M9 and - wait for the session to end, so the next M8 is a new one.""" - ctx.check(_cool(fc).get("phase") != "run", "%s: a run session is already open", what) - g.command("M8") - try: - t0 = time.time() - c = {} - while time.time() - t0 < (wait or VERDICT_WAIT_S): - ctx.sleep(1) - c = _cool(fc) - if until(c): - break - ctx.log("%s: verdict %s fire_ok %s hold %s gates_off %s (after %.0f s)", what, - c.get("verdict"), c.get("fire_ok"), c.get("hold"), c.get("gates_off"), time.time() - t0) - return c - finally: - g.command("M9") - _session_ended(ctx, fc, what) - - -def _after_session(ctx, fc, wait=5, until=None): - """The engine's state a few ticks after a session ended: the first - sample out of phase run that satisfies `until` (default: verdict OK), - or the last sample taken.""" - until = until or (lambda c: c.get("verdict") == "OK") - c = {} - t0 = time.time() - while time.time() - t0 < wait: - ctx.sleep(1) - c = _cool(fc) - if c.get("phase") != "run" and until(c): - break - return c - - -def _tail_has(fc, needle): - st, body = fc.get("/logs/tail", params={"name": "forgectrl", "lines": GATE_LOG_LINES}) - text = body.get("text", "") if st == 200 and isinstance(body, dict) else "" - return needle in text - - -@test("cooling.gate-off", title="A gate setting trips inside its range and is off at its far end", - subsystem="cooling", kind="auto", mode="grbl", est_min=3, - covers=_COOL_COVERS, requires=["kernel.latch-locked-idle"], - steps=["Machine idle, coolant at room temperature (above 8 C). The test writes the coolant " - "ceiling and resume gate and restores them; three short M8/M9 cycles spin the fans."], - description="The coolant ceiling is a plain setting with a wide legal range whose top " - "turns the gate off by value. Set just above its legal minimum it must trip " - "(OVERTEMP, hold, fire blocked) at the next run start; set to its top the " - "engine must skip the gate (verdict OK), report it in gates_off on /status " - "and /cool/status, say so in the settings reply, and log the run-start line; " - "restored, everything reads as before. Alongside: /status carries the watched " - "board temperatures (chassis and SoC die in degrees, supply as a raw count, the " - "CPU unthrottled) and every run session ends with them ranged into one log line.") -def gate_off(ctx): - from .motion import FORGECTRL_LOG, _log_lines, _log_offset - fc = ctx.forgectrl - ev = ctx.evidence - before = fc.settings() - orig = {k: before.get(k, "") for k in GATE_KEYS} - ev["orig"] = orig - ctx.log("original: %s", orig) - st0 = fc.status() - up = (st0.get("coolant") or {}).get("up_c") - ctx.check(up is not None and up > 8.0, "coolant too cold for the trip leg (up_c %s)", up) - # The board temperatures ride /status, watched and not gated: the - # chassis in degrees, the supply as its raw count. - temps = st0.get("temps") or {} - ev["temps"] = temps - ctx.check(isinstance(temps.get("chassis_c"), (int, float)) and 5.0 <= temps["chassis_c"] <= 80.0, - "/status temps.chassis_c is not a plausible chassis temperature: %s", temps) - ctx.check(isinstance(temps.get("supply_raw"), int) and 0 <= temps["supply_raw"] <= 1023, - "/status temps.supply_raw is not a 10-bit count: %s", temps) - ctx.check(isinstance(temps.get("soc_c"), (int, float)) and 10.0 <= temps["soc_c"] <= 100.0, - "/status temps.soc_c is not a plausible die temperature: %s", temps) - ctx.check(temps.get("soc_throttle") == 0, - "the CPU is throttled (or its cooling device is missing) at idle: %s", temps) - c0 = _cool(fc) - ctx.check(c0.get("verdict") == "OK", "engine is not at OK before the test: %s", c0) - ctx.check(c0.get("gates_off") == [], "a gate is already off: %s", c0.get("gates_off")) - g_default = (before.get("gates") or {}).get("cool_temp_max") or {} - ctx.check(g_default.get("gate") == "coolant_max" and g_default.get("off") == "high", - "settings reply does not describe the ceiling as the coolant_max gate, off at its top: %s", g_default) - top = g_default.get("hi") - bottom = g_default.get("lo") - ctx.check(isinstance(top, (int, float)) and isinstance(bottom, (int, float)), "no range in the reply: %s", g_default) - - restored = False - with ctx.grbl() as grbl: - st = grbl.status_report() - ctx.check(st["state"].startswith("Idle"), "controller is %s", st["state"]) - try: - - # Leg 1: a ceiling the coolant is already over. Legal, outside - # the band (warned), and it must trip at the next run start. - _set_gates(ctx, fc, {"cool_temp_max": str(bottom + 1), "cool_temp_resume": str(bottom)}) - state, val = _gate_state(fc, "cool_temp_max") - ev["trip_state"] = state - ctx.check(state == "warn", "a ceiling of %s reports state %r, expected warn", val, state) - c = _run_session(ctx, grbl, fc, lambda c: c.get("verdict") == "OVERTEMP", "trip leg") - ev["trip"] = c - ctx.check(c.get("verdict") == "OVERTEMP", "ceiling %s C with coolant at %.1f C did not trip: %s", - bottom + 1, up, c) - ctx.check(c.get("fire_ok") is False and c.get("hold") is True, - "OVERTEMP without fire blocked and a hold: %s", c) - ctx.check(c.get("gates_off") == [], "a tripped gate is not an off gate: %s", c.get("gates_off")) - - # Leg 2: the ceiling at its top. Off by value: no gate, verdict - # back to OK at the next run start, and said out loud. - at_off = _log_offset(FORGECTRL_LOG) # the off gate's line is one written after this - _set_gates(ctx, fc, {"cool_temp_max": str(top), "cool_temp_resume": orig["cool_temp_resume"]}) - state, val = _gate_state(fc, "cool_temp_max") - ev["off_state"] = state - ctx.check(state == "off", "a ceiling of %s reports state %r, expected off", val, state) - c = _run_session(ctx, grbl, fc, lambda c: c.get("verdict") == "OK" and c.get("gates_off"), "off leg") - ev["off"] = c - ctx.check(c.get("verdict") == "OK", "ceiling at %s did not clear the gate: %s", top, c) - ctx.check(c.get("gates_off") == ["coolant_max"], "/cool/status gates_off %s, expected [coolant_max]", - c.get("gates_off")) - s_off = fc.status().get("gates_off") - ctx.check(s_off == ["coolant_max"], "/status gates_off %s, expected [coolant_max]", s_off) - # A log line reaches the file a moment after the engine acts. - line = "gate coolant_max OFF: cool_temp_max = %g" % top - ctx.wait_for(lambda: bool(_log_lines(FORGECTRL_LOG, at_off, line)), 5, poll=0.25) - ctx.check(_log_lines(FORGECTRL_LOG, at_off, line), - "the run-start log line for the off gate is missing from the forgectrl log") - - # Restore, and prove the restore: the next run start reloads. - _set_gates(ctx, fc, orig) - restored = True - state, val = _gate_state(fc, "cool_temp_max") - at_restore = _log_offset(FORGECTRL_LOG) # the session's run-end line is one written after this - c = _run_session(ctx, grbl, fc, lambda c: c.get("verdict") == "OK" and not c.get("gates_off"), - "restored") - ev["restored"] = c - ctx.check(c.get("verdict") == "OK" and c.get("gates_off") == [], - "engine did not return to OK with no gate off after the restore: %s", c) - ctx.log("restored ceiling %s reports state %s", val, state) - # Every run session ends with the board temperatures ranged - # into one line. - ctx.wait_for(lambda: bool(_log_lines(FORGECTRL_LOG, at_restore, "temps this job: chassis ")), 5, poll=0.25) - temps = _log_lines(FORGECTRL_LOG, at_restore, "temps this job: chassis ") - ev["temps_line"] = temps[-1] if temps else None - ctx.check(temps and ", soc " in temps[-1], - "the run-end board-temperature line (chassis, soc, supply) is missing from the " - "forgectrl log") - finally: - if not restored: - # The engine reads settings at run start only: restoring - # the file is not enough, a hold taken against the test's - # ceiling would stand until the operator's next job. - st, body = fc.post("/settings", params=orig) - ctx.log("restore on failure: POST /settings %s -> %s", orig, st) - try: - c = _run_session(ctx, grbl, fc, - lambda c: c.get("verdict") == "OK" and not c.get("gates_off"), - "restore on failure") - ctx.log("restore on failure: engine %s gates_off %s", c.get("verdict"), c.get("gates_off")) - except Exception as e: # the original failure is the one to report - ctx.log("restore on failure: run session did not complete (%s)", e) - after = fc.settings() - ctx.check(all(after.get(k, "") == orig[k] for k in GATE_KEYS), - "settings not restored: %s", {k: after.get(k) for k in GATE_KEYS}) - ctx.check(fc.wait_idle(60, abort=ctx.aborted), "machine did not return to idle") - - -CRIT_KEYS = ("cool_temp_max", "cool_temp_resume", "cool_temp_critical_c") - - -LOW_KEYS = ("cool_temp_min", "cool_temp_start") -WARMUP_ABOVE_C = 1.0 # the start gate this far above the loop: a few heater minutes -WARMUP_RELEASE_S = 720 # the flow heater warms the bulk 0.4 to 0.8 C a minute -HEATER_PWM = "thermal/heater_pwm" - - -def _hold_session(ctx, g, fc, until, what, wait): - """M8 opens a run session and leaves it open: wait for `until(cool)` - and return the sample. The caller ends the session.""" - ctx.check(_cool(fc).get("phase") != "run", "%s: a run session is already open", what) - g.command("M8") - t0 = time.time() - c = {} - while time.time() - t0 < wait: - ctx.sleep(1) - c = _cool(fc) - if until(c): - break - ctx.log("%s: verdict %s phase %s fire_ok %s hold %s (after %.0f s)", what, c.get("verdict"), - c.get("phase"), c.get("fire_ok"), c.get("hold"), time.time() - t0) - return c - - -@test("cooling.floor-and-warm-up", title="The coolant floor blocks fire and the warm-up gate holds a cold start", - subsystem="cooling", kind="auto", mode="grbl", est_min=9, - covers=_COOL_COVERS, requires=["kernel.latch-locked-idle", "cooling.gate-off"], - steps=["Machine idle, coolant at room temperature (10 to 30 C). The test writes the coolant " - "floor and the warm-up gate and restores them; four M8/M9 sessions, the first with " - "the loop heater on for a few minutes."], - description="Both low-side gates, proven at room temperature by moving them above the " - "loop. Warm-up: the start gate set 1 C above the coolant makes the next run " - "session hold (WARMUP, hold, fire blocked, phase warm-up) with the loop heater " - "on and the fans idle; the coolant climbs to the gate and the session releases " - "into run with the verdict OK, the heater off and the run fans up. Floor: with " - "the start gate off, a floor set 2 C above the coolant makes the next session " - "hold COLD with fire blocked. Both at 0 are off, reported in gates_off with the " - "run-start log line, and the session reads OK. Restored, everything reads as " - "before.") -def floor_and_warm_up(ctx): - from .motion import FORGECTRL_LOG, _log_lines, _log_offset - fc = ctx.forgectrl - ev = ctx.evidence - before = fc.settings() - orig = {k: before.get(k, "") for k in LOW_KEYS} - ev["orig"] = orig - ctx.log("original: %s", orig) - up = (fc.status().get("coolant") or {}).get("up_c") - ctx.check(up is not None and 10.0 <= up <= 30.0, "coolant outside the test's range (up_c %s)", up) - c0 = _cool(fc) - ctx.check(c0.get("verdict") == "OK" and c0.get("gates_off") == [], - "engine is not at OK with every gate on before the test: %s", c0) - gates = before.get("gates") or {} - g_min, g_start = gates.get("cool_temp_min") or {}, gates.get("cool_temp_start") or {} - ctx.check(g_min.get("gate") == "coolant_min" and g_min.get("off") == "low", - "settings reply does not describe cool_temp_min as the coolant_min gate, off at zero: %s", g_min) - ctx.check(g_start.get("gate") == "warm_up" and g_start.get("off") == "low", - "settings reply does not describe cool_temp_start as the warm_up gate, off at zero: %s", g_start) - - restored = False - with ctx.grbl() as grbl: - ctx.check(grbl.status_report()["state"].startswith("Idle"), "controller is not idle") - try: - # Leg 1: the warm-up hold and its release. The gate sits just - # above the loop, so the heater reaches it in a few minutes. - gate = round(up + WARMUP_ABOVE_C, 1) - at_warmup = _log_offset(FORGECTRL_LOG) # the release line is one written after this - _set_gates(ctx, fc, {"cool_temp_start": str(gate), "cool_temp_min": orig["cool_temp_min"]}) - c = _hold_session(ctx, grbl, fc, lambda c: c.get("verdict") == "WARMUP", "warm-up hold", VERDICT_WAIT_S) - ev["warmup_hold"] = c - heater = hw.sysfs_int(HEATER_PWM) - duty = _duties() - ev["warmup_heater_pwm"] = heater - ev["warmup_duty"] = duty - ctx.check(c.get("verdict") == "WARMUP", "start gate %s C over coolant %.1f C did not hold the session: %s", - gate, up, c) - ctx.check(c.get("fire_ok") is False and c.get("hold") is True, "WARMUP without fire blocked and a hold: %s", c) - ctx.check(c.get("phase") == "warm-up", "phase %r during the warm-up hold, expected warm-up", c.get("phase")) - ctx.check(heater is not None and heater > 0, "the loop heater is not on during the warm-up (%s=%s)", HEATER_PWM, heater) - ctx.check(duty == IDLE_DUTY, "the fans are not idle during the warm-up: %s", duty) - t0 = time.time() - c = {} - while time.time() - t0 < WARMUP_RELEASE_S: - ctx.checkpoint() - ctx.sleep(5) - c = _cool(fc) - if c.get("verdict") != "WARMUP" or c.get("phase") != "warm-up": - break - ev["warmup_release_s"] = round(time.time() - t0) - ev["warmup_release"] = c - ctx.log("warm-up release after %s s: verdict %s phase %s up %s (gate %s)", ev["warmup_release_s"], - c.get("verdict"), c.get("phase"), c.get("up_c"), gate) - ctx.check(c.get("verdict") == "OK" and c.get("phase") == "run", - "the warm-up did not release into a run session within %d s: %s", WARMUP_RELEASE_S, c) - ctx.check(ev["warmup_release_s"] >= 60, - "the warm-up released in %s s: an instant reading, not the bulk (the release " - "judges a one-minute rolling minimum)", ev["warmup_release_s"]) - ctx.check(c.get("fire_ok") is True and c.get("hold") is False, "released warm-up without fire_ok and no hold: %s", c) - ctx.sleep(2) - heater = hw.sysfs_int(HEATER_PWM) - duty = _duties() - ev["run_heater_pwm"] = heater - ev["run_duty"] = duty - ctx.check(heater == 0, "the loop heater stayed on after the release (%s=%s)", HEATER_PWM, heater) - ctx.check(duty != IDLE_DUTY, "the run fans did not come up at the release: %s", duty) - # A log line reaches the file a moment after the engine acts. - line = "warm-up complete: coolant " - ctx.wait_for(lambda: bool(_log_lines(FORGECTRL_LOG, at_warmup, line)), 5, poll=0.25) - ctx.check(_log_lines(FORGECTRL_LOG, at_warmup, line), - "the warm-up release line is missing from the forgectrl log") - grbl.command("M9") - _session_ended(ctx, fc, "warm-up") - - # Leg 2: the floor. The start gate off, the floor above the - # (now warmer) loop. - up2 = (fc.status().get("coolant") or {}).get("up_c") or up - floor = round(min(up2 + 2.0, 32.0), 1) - ctx.check(up2 < floor, "coolant %.1f C leaves no room under the ceiling for a floor above it", up2) - _set_gates(ctx, fc, {"cool_temp_start": "0", "cool_temp_min": str(floor)}) - c = _run_session(ctx, grbl, fc, lambda c: c.get("verdict") == "COLD", "floor") - ev["floor"] = c - ctx.check(c.get("verdict") == "COLD", "floor %s C over coolant %.1f C did not trip: %s", floor, up2, c) - ctx.check(c.get("fire_ok") is False and c.get("hold") is True, "COLD without fire blocked and a hold: %s", c) - ctx.check(c.get("gates_off") == ["warm_up"], "gates_off %s with the start gate at 0, expected [warm_up]", - c.get("gates_off")) - - # Leg 3: both at zero: off, said so, and the session reads OK. - at_min_off = _log_offset(FORGECTRL_LOG) # the floor's off line is one written after this - _set_gates(ctx, fc, {"cool_temp_start": "0", "cool_temp_min": "0"}) - c = _run_session(ctx, grbl, fc, lambda c: c.get("verdict") == "OK" and len(c.get("gates_off") or []) == 2, - "both off") - ev["both_off"] = c - ctx.check(c.get("verdict") == "OK", "gates at 0 did not clear the hold: %s", c) - ctx.check(sorted(c.get("gates_off") or []) == ["coolant_min", "warm_up"], - "gates_off %s, expected [coolant_min, warm_up]", c.get("gates_off")) - line = "gate coolant_min OFF: cool_temp_min = 0" - ctx.wait_for(lambda: bool(_log_lines(FORGECTRL_LOG, at_min_off, line)), 5, poll=0.25) - ctx.check(_log_lines(FORGECTRL_LOG, at_min_off, line), - "the run-start log line for the floor off is missing from the forgectrl log") - - # Restore, and prove the restore at the next run start. - _set_gates(ctx, fc, orig) - restored = True - c = _run_session(ctx, grbl, fc, lambda c: c.get("verdict") == "OK" and not c.get("gates_off"), "restored") - ev["restored"] = c - ctx.check(c.get("verdict") == "OK" and c.get("gates_off") == [], - "engine did not return to OK with no gate off after the restore: %s", c) - finally: - if not restored: - st, body = fc.post("/settings", params=orig) - ctx.log("restore on failure: POST /settings %s -> %s", orig, st) - try: - grbl.command("M9") - _session_ended(ctx, fc, "restore on failure") - c = _run_session(ctx, grbl, fc, - lambda c: c.get("verdict") == "OK" and not c.get("gates_off"), - "restore on failure") - ctx.log("restore on failure: engine %s gates_off %s", c.get("verdict"), c.get("gates_off")) - except Exception as e: # the original failure is the one to report - ctx.log("restore on failure: run session did not complete (%s)", e) - after = fc.settings() - ctx.check(all(after.get(k, "") == orig[k] for k in LOW_KEYS), - "settings not restored: %s", {k: after.get(k) for k in LOW_KEYS}) - ctx.check(hw.sysfs_int(HEATER_PWM) == 0, "the loop heater is on after the test") - ctx.check(fc.wait_idle(60, abort=ctx.aborted), "machine did not return to idle") - - -TEC_KEYS = ("cool_tec_present", "cool_tec_on_c", "cool_tec_off_c") -TEC_ATTR = "thermal/tec_on" -TEC_WAIT_S = 20 # settings re-read at run start, 1 Hz ticks - - -@test("cooling.tec-drive", title="The TEC runs on its hysteresis only with airflow and only when declared fitted", - subsystem="cooling", kind="auto", mode="grbl", est_min=5, - covers=_COOL_COVERS, requires=["kernel.latch-locked-idle"], - steps=["Machine idle, coolant at room temperature (10 to 30 C). The test declares the TEC " - "fitted, moves its thresholds under the loop, runs two short M8/M9 sessions and " - "restores everything. On a machine without the part the line toggles into nothing."], - description="TEC policy at room temperature, thresholds moved under the loop so the " - "hysteresis wants the cooler on. Declared fitted, a run session drives " - "thermal/tec_on to 1 (airflow up) and the session's end returns it to 0 (no " - "airflow at idle). Declared not fitted, the same session leaves the line at 0. " - "" - "The cross-check keeps off under on. The line has no readback, so the last written value is " - "what the test reads; the drive rule is the engine's, never the cloud's.") -def tec_drive(ctx): - from .motion import FORGECTRL_LOG, _log_lines, _log_offset - fc = ctx.forgectrl - ev = ctx.evidence - before = fc.settings() - orig = {k: before.get(k, "") for k in TEC_KEYS} - ev["orig"] = orig - ctx.log("original: %s", orig) - up = (fc.status().get("coolant") or {}).get("up_c") - ctx.check(up is not None and 10.0 <= up <= 30.0, "coolant outside the test's range (up_c %s)", up) - ctx.check(hw.sysfs_int(TEC_ATTR) == 0, "the TEC line is not 0 at idle before the test") - on_c, off_c = round(up - 2.0, 1), round(up - 4.0, 1) - - # The cross-check, before anything runs: off must sit under on. The - # floor relation is the engine's runtime clamp, not a settings rule. - st, body = fc.post("/settings", params={"cool_tec_on_c": "18", "cool_tec_off_c": "19"}) - ctx.check(st == 400, "an off threshold over the on threshold was accepted (%s %s)", st, body) - - restored = False - with ctx.grbl() as grbl: - ctx.check(grbl.status_report()["state"].startswith("Idle"), "controller is not idle") - try: - # Leg 1: fitted, thresholds under the loop: on with the session's - # airflow, off again at idle. - _set_gates(ctx, fc, {"cool_tec_present": "1", "cool_tec_on_c": str(on_c), - "cool_tec_off_c": str(off_c)}) - ctx.check(_cool(fc).get("phase") != "run", "a run session is already open") - off = _log_offset(FORGECTRL_LOG) # the TEC-on line is one written after this - grbl.command("M8") - t0 = time.time() - tec = 0 - while time.time() - t0 < TEC_WAIT_S: - ctx.sleep(1) - tec = hw.sysfs_int(TEC_ATTR) - if tec: - break - ev["tec_on_after_s"] = round(time.time() - t0) - ctx.log("fitted, in session: %s=%s after %s s", TEC_ATTR, tec, ev["tec_on_after_s"]) - ctx.check(tec == 1, "the TEC did not come on in a run session with its threshold under the loop") - # The engine writes the line first and logs it after, and the log - # line reaches the file a moment later still. - ctx.wait_for(lambda: bool(_log_lines(FORGECTRL_LOG, off, "TEC on: coolant ")), 5, poll=0.25) - ev["tec_on_line"] = (_log_lines(FORGECTRL_LOG, off, "TEC on: coolant ") or [None])[0] - ctx.check(ev["tec_on_line"], "the TEC-on line is missing from the forgectrl log") - grbl.command("M9") - _session_ended(ctx, fc, "tec leg 1") - t0 = time.time() - while time.time() - t0 < 30: - ctx.sleep(1) - if _cool(fc).get("phase") == "idle" and hw.sysfs_int(TEC_ATTR) == 0: - break - tec = hw.sysfs_int(TEC_ATTR) - ev["tec_after_session"] = tec - ctx.log("after the session: phase %s %s=%s", _cool(fc).get("phase"), TEC_ATTR, tec) - ctx.check(tec == 0, "the TEC stayed on with no airflow (idle after the session)") - - # Leg 2: not fitted: the same session leaves the line alone. - _set_gates(ctx, fc, {"cool_tec_present": "0", "cool_tec_on_c": str(on_c), - "cool_tec_off_c": str(off_c)}) - c = _run_session(ctx, grbl, fc, lambda c: c.get("phase") == "run", "tec leg 2") - tec = hw.sysfs_int(TEC_ATTR) - ev["tec_not_fitted"] = tec - ctx.check(tec == 0, "the TEC was driven while declared not fitted (%s=%s)", TEC_ATTR, tec) - - # Restore, and prove the next session reads it back. - _set_gates(ctx, fc, orig) - restored = True - c = _run_session(ctx, grbl, fc, lambda c: c.get("verdict") == "OK", "restored") - ev["restored"] = c - finally: - if not restored: - st, body = fc.post("/settings", params=orig) - ctx.log("restore on failure: POST /settings %s -> %s", orig, st) - try: - grbl.command("M9") - _session_ended(ctx, fc, "restore on failure") - except Exception as e: - ctx.log("restore on failure: session end did not complete (%s)", e) - after = fc.settings() - ctx.check(all(after.get(k, "") == orig[k] for k in TEC_KEYS), - "settings not restored: %s", {k: after.get(k) for k in TEC_KEYS}) - ctx.check(hw.sysfs_int(TEC_ATTR) == 0, "the TEC line is not 0 after the test") - ctx.check(fc.wait_idle(60, abort=ctx.aborted), "machine did not return to idle") - - -FIRE_KEYS = ("cool_fire_q1_alert", "cool_fire_q1_critical", "cool_fire_q2_alert", "cool_fire_q2_critical") - - -def _tail_wait(ctx, fc, needle, wait=8): - """A log line lands through rsyslog a beat after the engine acts; - poll the tail instead of reading it once.""" - t0 = time.time() - while time.time() - t0 < wait: - if _tail_has(fc, needle): - return True - ctx.sleep(1) - return False - -FIRE_WAIT_S = 25 # settings re-read at run start, two-tick breach, 1 Hz - - -@test("cooling.fire-watch-tiers", title="The lid-IR fire watch pauses at its alert and fails at its critical", - subsystem="cooling", kind="auto", mode="grbl", est_min=6, - covers=_COOL_COVERS + [("forgectrl", "src/super.*")], requires=["kernel.latch-locked-idle"], - steps=["Machine idle, lid closed, lid lamp at its resting level. The test moves the flame " - "thresholds under the lamp's own reading and restores them; three M8/M9 sessions, " - "no fire."], - description="The factory-shaped fire watch, proven with the lid lamp as the flame " - "stand-in: the four lid-IR channels read the resting lamp at a known level, " - "so a q1 alert moved under that level must hold the session (FLAME, hold, " - "fire blocked, fire_watch alert) and release once the threshold is restored " - "mid-session; a q1 critical moved under it must stop motion, lock the laser " - "latch and latch FIRE for the rest of the session; all four thresholds at 0 " - "read as the four flame gates off and the watch reads watch. Restored, the " - "watch reads armed at OK.") -def fire_watch_tiers(ctx): - from .motion import FORGECTRL_LOG, _log_lines, _log_offset - fc = ctx.forgectrl - ev = ctx.evidence - before = fc.settings() - orig = {k: before.get(k, "") for k in FIRE_KEYS} - ev["orig"] = orig - ctx.log("original: %s", orig) - st0 = fc.status() - irs = st0.get("lid_ir") - ctx.check(isinstance(irs, list) and len(irs) == 4 and all(isinstance(v, int) for v in irs), - "/status lid_ir is not four readings: %s", irs) - q1 = sorted(irs)[0] - ev["lid_ir_idle"] = irs - ctx.check(20 <= q1 <= 250, "the resting lamp reads %s on the lowest channel; the test needs 20 to 250", q1) - c0 = _cool(fc) - ctx.check(c0.get("verdict") == "OK" and c0.get("fire_watch") == "armed", - "engine not at OK with the watch armed before the test: %s", c0) - - restored = False - with ctx.grbl() as grbl: - ctx.check(grbl.status_report()["state"].startswith("Idle"), "controller is not idle") - try: - # Leg 1: the pause tier. The q1 alert under the lamp reading - # holds the session; restoring it mid-session releases the hold. - _set_gates(ctx, fc, {"cool_fire_q1_alert": str(max(1, q1 - 10))}) - grbl.command("M8") - t0 = time.time() - c = {} - while time.time() - t0 < FIRE_WAIT_S: - ctx.sleep(1) - c = _cool(fc) - if c.get("verdict") == "FLAME": - break - ev["alert"] = c - ctx.log("alert leg: verdict %s fire_watch %s hold %s fire_ok %s", c.get("verdict"), - c.get("fire_watch"), c.get("hold"), c.get("fire_ok")) - ctx.check(c.get("verdict") == "FLAME", "q1 alert under the lamp did not hold the session: %s", c) - ctx.check(c.get("hold") is True and c.get("fire_ok") is False, "FLAME without a hold and fire blocked: %s", c) - ctx.check(c.get("fire_watch") == "alert", "fire_watch %r during the alert, expected alert", c.get("fire_watch")) - ctx.check(_tail_wait(ctx, fc, "lid IR flame alert (quartiles "), - "the flame-alert line is missing from the forgectrl log") - # The release: the settings re-read is at run start, so the - # release lever mid-session is the env-free path - restore the - # threshold and end the session; the alert must not survive - # into the next one. - _set_gates(ctx, fc, {"cool_fire_q1_alert": orig["cool_fire_q1_alert"]}) - grbl.command("M9") - _session_ended(ctx, fc, "alert leg") - c = _run_session(ctx, grbl, fc, lambda c: c.get("verdict") == "OK", "post-alert") - ctx.check(c.get("verdict") == "OK", "the alert survived into a fresh session: %s", c) - - # Leg 2: the fail tier. The q1 critical under the lamp reading - # latches FIRE and locks the laser; the fail tier then stops - # the controller through the supervisor and starts it again, - # so no run start can relight what was locked and the sender - # sees the job end. The Grbl socket dies with the restart - - # FIRE is read from the engine, the latch from sysfs, the stop - # from the supervisor's log, and a fresh session is opened for - # the legs that follow. - st, m0 = fc.get("/mode") - pid0 = m0.get("pid") if isinstance(m0, dict) else None - off = _log_offset(FORGECTRL_LOG) - _set_gates(ctx, fc, {"cool_fire_q1_critical": str(max(2, q1 - 5)), - "cool_fire_q1_alert": str(max(1, q1 - 10))}) - grbl.command("M8") - t0 = time.time() - c = {} - while time.time() - t0 < FIRE_WAIT_S: - ctx.sleep(1) - c = _cool(fc) - if c.get("verdict") == "FIRE": - break - ilk = hw.sysfs_int("cnc/interlock_circuit") - ev["critical"] = c - ev["interlock_circuit"] = ilk - ctx.log("critical leg: verdict %s fire_watch %s interlock_circuit %s", - c.get("verdict"), c.get("fire_watch"), ilk) - ctx.check(c.get("verdict") == "FIRE", "q1 critical under the lamp did not latch FIRE: %s", c) - ctx.check(c.get("fire_watch") == "ALARM", "fire_watch %r during FIRE, expected ALARM", c.get("fire_watch")) - ctx.check(ilk is not None and (ilk & 8), "the laser latch is not locked under FIRE (interlock_circuit=%s)", ilk) - ctx.check(_tail_wait(ctx, fc, "LID IR FIRE SIGNAL (quartiles "), - "the FIRE line is missing from the forgectrl log") - # the fail tier stops the controller and starts it again; the - # Grbl socket dies with it. Wait out the restart, prove the - # supervisor logged the stop, restore the thresholds, and open - # a fresh session on the new controller. - grbl.close() - m1 = None - t0 = time.time() - while time.time() - t0 < 60: - st, m1 = fc.get("/mode") - if isinstance(m1, dict) and m1.get("controller") == "running" and m1.get("pid") != pid0: - break - ctx.sleep(0.5) - ev["restart"] = {"pid_before": pid0, "mode_after": m1} - ctx.log("after FIRE: %s", ev["restart"]) - ctx.check(m1 and m1.get("controller") == "running" and m1.get("pid") != pid0, - "the controller was not stopped and started again after FIRE: %s", m1) - ctx.check(_log_lines(FORGECTRL_LOG, off, "lid IR fire signal - the controller is stopped"), - "the supervisor did not log the fail-tier stop after FIRE") - _set_gates(ctx, fc, {"cool_fire_q1_critical": orig["cool_fire_q1_critical"], - "cool_fire_q1_alert": orig["cool_fire_q1_alert"]}) - ctx.check(fc.wait_idle(30, abort=ctx.aborted), "machine not idle after the FIRE restart") - grbl.connect() # a fresh session on the restarted controller - - # Leg 3: all four at zero: the watch off, said so, session OK. - _set_gates(ctx, fc, {k: "0" for k in FIRE_KEYS}) - c = _run_session(ctx, grbl, fc, lambda c: c.get("verdict") == "OK" and len(c.get("gates_off") or []) == 4, - "watch off") - ev["off"] = c - ctx.check(c.get("verdict") == "OK", "the watch at zero did not read OK: %s", c) - ctx.check(sorted(c.get("gates_off") or []) == ["flame_q1_alert", "flame_q1_critical", - "flame_q2_alert", "flame_q2_critical"], - "gates_off %s, expected the four flame gates", c.get("gates_off")) - ctx.check(c.get("fire_watch") == "watch", "fire_watch %r with every tier at 0, expected watch", - c.get("fire_watch")) - - # Restore, and prove the watch is armed again. - _set_gates(ctx, fc, orig) - restored = True - c = _run_session(ctx, grbl, fc, - lambda c: c.get("verdict") == "OK" and not c.get("gates_off"), "restored") - ev["restored"] = c - ctx.check(c.get("verdict") == "OK" and c.get("fire_watch") == "armed", - "engine did not return to OK with the watch armed: %s", c) - finally: - if not restored: - st, body = fc.post("/settings", params=orig) - ctx.log("restore on failure: POST /settings %s -> %s", orig, st) - try: - grbl.command("M9") - _session_ended(ctx, fc, "restore on failure") - c = _run_session(ctx, grbl, fc, - lambda c: c.get("verdict") == "OK" and not c.get("gates_off"), - "restore on failure") - ctx.log("restore on failure: engine %s gates_off %s", c.get("verdict"), c.get("gates_off")) - except Exception as e: - ctx.log("restore on failure: run session did not complete (%s)", e) - after = fc.settings() - ctx.check(all(after.get(k, "") == orig[k] for k in FIRE_KEYS), - "settings not restored: %s", {k: after.get(k) for k in FIRE_KEYS}) - ctx.check(fc.wait_idle(60, abort=ctx.aborted), "machine did not return to idle") - - CRASH_KEYS = ("cool_accel_x_alert", "cool_accel_y_alert", "cool_accel_abort") -CRASH_GATES = ["crash_abort", "crash_x_alert", "crash_y_alert"] - - -@test("cooling.crash-watch-plumbing", title="The crash watch stays off the accelerometer unarmed and its gates say off", - subsystem="cooling", kind="auto", mode="grbl", est_min=3, - covers=_COOL_COVERS, requires=["kernel.latch-locked-idle"], - steps=["Machine idle, lid closed. The test writes the three crash thresholds and restores " - "them; two M8/M9 sessions, no fire, no motion."], - description="The head-accelerometer crash watch's plumbing, provable without an armed " - "window: /cool/status carries accel_watch at watch with the factory-seeded " - "thresholds standing; an unarmed run session never arms the watch (the " - "coexistence contract: liveness and homing own the accel outside the armed " - "window), so accel_watch reads watch through it; the three thresholds at 0 " - "read as the three crash gates off by value (gates_off names them) with the " - "session OK; a threshold past the register range is refused; restored, the " - "gates read ok. The tiers themselves (BUMP, CRASH) trip only on a physical " - "knock inside an armed window and are proven at the bench, not here.") -def crash_watch_plumbing(ctx): - fc = ctx.forgectrl - ev = ctx.evidence - before = fc.settings() - orig = {k: before.get(k, "") for k in CRASH_KEYS} - ev["orig"] = orig - ctx.log("original: %s", orig) - gates = (before.get("gates") or {}) - for k in CRASH_KEYS: - g = gates.get(k) or {} - ctx.check(g.get("gate", "").startswith("crash_"), "settings gates has no %s row: %s", k, g) - c0 = _cool(fc) - ctx.check(c0.get("accel_watch") == "watch", - "accel_watch %r at idle with thresholds set, expected watch", c0.get("accel_watch")) - - st, body = fc.post("/settings", params={"cool_accel_x_alert": "300"}) - ctx.check(st != 200 or body.get("cool_accel_x_alert") != "300", - "a threshold past the register range (300) was accepted") - - restored = False - with ctx.grbl() as grbl: - ctx.check(grbl.status_report()["state"].startswith("Idle"), "controller is not idle") - try: - # Leg 1: an unarmed session never arms the watch. - c = _run_session(ctx, grbl, fc, lambda c: c.get("phase") == "run", "unarmed session") - ev["unarmed"] = c - ctx.check(c.get("accel_watch") == "watch", - "accel_watch %r in an unarmed run session, expected watch", c.get("accel_watch")) - ctx.check(c.get("verdict") == "OK", "unarmed session not OK: %s", c) - - # Leg 2: the three thresholds at zero are the gates off. - _set_gates(ctx, fc, {k: "0" for k in CRASH_KEYS}) - c = _run_session(ctx, grbl, fc, - lambda c: c.get("verdict") == "OK" and - set(CRASH_GATES) <= set(c.get("gates_off") or []), - "watch off") - ev["off"] = c - ctx.check(set(CRASH_GATES) <= set(c.get("gates_off") or []), - "gates_off %s does not name the three crash gates", c.get("gates_off")) - ctx.check(c.get("verdict") == "OK", "the watch at zero did not read OK: %s", c) - - # Restore, and prove the gates read ok again. - _set_gates(ctx, fc, orig) - restored = True - c = _run_session(ctx, grbl, fc, - lambda c: not (set(CRASH_GATES) & set(c.get("gates_off") or [])), - "restored") - ev["restored"] = c - ctx.check(not (set(CRASH_GATES) & set(c.get("gates_off") or [])), - "the crash gates stayed off after the restore: %s", c.get("gates_off")) - finally: - if not restored: - st, body = fc.post("/settings", params=orig) - ctx.log("restore on failure: POST /settings %s -> %s", orig, st) - after = fc.settings() - ctx.check(all(after.get(k, "") == orig[k] for k in CRASH_KEYS), - "settings not restored: %s", {k: after.get(k) for k in CRASH_KEYS}) - ctx.check(fc.wait_idle(60, abort=ctx.aborted), "machine did not return to idle") - - -@test("cooling.critical-tier", title="The coolant critical line is a fault above the ceiling's pause", - subsystem="cooling", kind="auto", mode="grbl", est_min=3, - covers=_COOL_COVERS + [("forgectrl", "src/main.c")], requires=["cooling.gate-off"], - steps=["Machine idle, coolant at room temperature (above 8 C). The test writes the coolant " - "ceiling, the resume gate and the critical line and restores them; three short M8/M9 " - "cycles spin the fans."], - description="Two tiers on the upstream coolant sensor: the ceiling pauses (OVERTEMP, resume " - "below the resume gate), the critical line above it is a fault. With the ceiling " - "and the critical line both under the coolant's temperature a run session must " - "read CRITICAL rather than OVERTEMP (fire blocked, hold, no resume), the fault " - "must end with the session, and the settings API must refuse a critical line at " - "or below the ceiling; with the critical line at its top the gate is off by " - "value (gates_off names it) and the ceiling alone pauses; restored, the next " - "session runs OK with nothing off.") -def critical_tier(ctx): - from .motion import FORGECTRL_LOG, _log_lines, _log_offset - fc = ctx.forgectrl - ev = ctx.evidence - before = fc.settings() - orig = {k: before.get(k, "") for k in CRIT_KEYS} - ev["orig"] = orig - ctx.log("original: %s", orig) - up = (fc.status().get("coolant") or {}).get("up_c") - ctx.check(up is not None and up > 8.0, "coolant too cold for the trip leg (up_c %s)", up) - c0 = _cool(fc) - ctx.check(c0.get("verdict") == "OK" and c0.get("gates_off") == [], - "engine is not at OK with every gate on before the test: %s", c0) - gates = before.get("gates") or {} - ceil = gates.get("cool_temp_max") or {} - crit = gates.get("cool_temp_critical_c") or {} - ctx.check(crit.get("gate") == "coolant_critical" and crit.get("off") == "high", - "settings reply does not describe the critical line as the coolant_critical gate, off at " - "its top: %s", crit) - bottom, top = ceil.get("lo"), crit.get("hi") - ctx.check(isinstance(bottom, (int, float)) and isinstance(top, (int, float)), - "no ranges in the reply: %s %s", ceil, crit) - ctx.check(isinstance(crit.get("lo"), (int, float)) and crit.get("lo") > bottom, - "the critical line's floor %s is not above the ceiling's %s", crit.get("lo"), bottom) - - # The cross-check: a critical line at or below the ceiling is refused - # before anything is written. - st, body = fc.post("/settings", params={"cool_temp_max": "33", "cool_temp_critical_c": "33"}) - ev["cross_check"] = {"status": st, "body": body} - if st == 200: - fc.post("/settings", params=orig) # undo before failing - ctx.check(st == 400, "a critical line equal to the ceiling was accepted: %s %s", st, body) - after = fc.settings() - ctx.check(all(after.get(k, "") == orig[k] for k in CRIT_KEYS), - "the refused POST changed a setting: %s", {k: after.get(k) for k in CRIT_KEYS}) - # A ceiling at its off end is no ceiling: the default line stands - # alone, and the POST that turns the ceiling off is accepted. - st, body = fc.post("/settings", params={"cool_temp_max": str(ceil.get("hi"))}) - ev["ceiling_off_accepted"] = st - if st == 200: - fc.post("/settings", params=orig) - ctx.check(st == 200, "the ceiling at its off end %s was refused against the critical line: %s %s", - ceil.get("hi"), st, body) - - restored = False - with ctx.grbl() as grbl: - st = grbl.status_report() - ctx.check(st["state"].startswith("Idle"), "controller is %s", st["state"]) - try: - # Leg 1: ceiling and critical line both under the coolant's - # temperature. The fail tier wins: CRITICAL, not OVERTEMP. - low = {"cool_temp_max": str(bottom + 1), "cool_temp_resume": str(bottom), - "cool_temp_critical_c": str(bottom + 2)} - _set_gates(ctx, fc, low) - c = _run_session(ctx, grbl, fc, lambda c: c.get("verdict") == "CRITICAL", "critical leg") - ev["critical"] = c - ctx.check(c.get("verdict") == "CRITICAL", - "critical line %s C with coolant at %.1f C did not fault (verdict %s): %s", - bottom + 2, up, c.get("verdict"), c) - ctx.check(c.get("fire_ok") is False and c.get("hold") is True, - "CRITICAL without fire blocked and a hold: %s", c) - ctx.check(c.get("resume_ok") is not True, "a coolant fault offered a resume: %s", c) - ctx.check("CRITICAL" in (c.get("reason") or "") and "coolant" in (c.get("reason") or ""), - "the reason does not name the tier and the coolant: %r", c.get("reason")) - ctx.check(c.get("gates_off") == [], "a tripped gate is not an off gate: %s", c.get("gates_off")) - # The fault ends with the session; the ceiling, still under the - # coolant, keeps its pause (OVERTEMP), never CRITICAL. - c = _after_session(ctx, fc, until=lambda c: c.get("verdict") != "CRITICAL") - ev["critical_after"] = c - ctx.check(c.get("verdict") == "OVERTEMP", - "after the faulted session the engine reads %s, expected the ceiling's OVERTEMP: %s", - c.get("verdict"), c) - ctx.check("CRITICAL" not in (c.get("reason") or ""), - "the reason still names the ended critical fault under the ceiling's hold: %r", - c.get("reason")) - - # Leg 2: the critical line at its top is the gate off; the - # ceiling alone pauses, and gates_off says so. - at_off = _log_offset(FORGECTRL_LOG) # the off gate's line is one written after this - _set_gates(ctx, fc, {"cool_temp_critical_c": str(top)}) - c = _run_session(ctx, grbl, fc, - lambda c: c.get("verdict") == "OVERTEMP" and "coolant_critical" in (c.get("gates_off") or []), - "critical off leg") - ev["critical_off"] = c - ctx.check(c.get("verdict") == "OVERTEMP", - "with the critical line off the ceiling did not pause (verdict %s): %s", c.get("verdict"), c) - ctx.check("coolant_critical" in (c.get("gates_off") or []), - "gates_off %s lacks coolant_critical", c.get("gates_off")) - # A log line reaches the file a moment after the engine acts. - line = "gate coolant_critical OFF: cool_temp_critical_c = %g" % top - ctx.wait_for(lambda: bool(_log_lines(FORGECTRL_LOG, at_off, line)), 5, poll=0.25) - ctx.check(_log_lines(FORGECTRL_LOG, at_off, line), - "the run-start log line for the off gate is missing from the forgectrl log") - - # Restore, and prove it: OK, nothing off. - _set_gates(ctx, fc, orig) - restored = True - c = _run_session(ctx, grbl, fc, lambda c: c.get("verdict") == "OK" and not c.get("gates_off"), - "restored") - ev["restored"] = c - ctx.check(c.get("verdict") == "OK" and c.get("gates_off") == [], - "engine did not return to OK with every gate on after the restore: %s", c) - finally: - if not restored: - st, body = fc.post("/settings", params=orig) - ctx.log("restore on failure: POST /settings %s -> %s", orig, st) - try: - c = _run_session(ctx, grbl, fc, - lambda c: c.get("verdict") == "OK" and not c.get("gates_off"), - "restore on failure") - ctx.log("restore on failure: engine %s gates_off %s", c.get("verdict"), c.get("gates_off")) - except Exception as e: # the original failure is the one to report - ctx.log("restore on failure: run session did not complete (%s)", e) - after = fc.settings() - ctx.check(all(after.get(k, "") == orig[k] for k in CRIT_KEYS), - "settings not restored: %s", {k: after.get(k) for k in CRIT_KEYS}) - - -FAN_KEYS = ("cool_tach_exhaust_min_rpm", "cool_purge_min_current", "cool_fan_grace_s") -FAN_GRACE_S = "8" # past the intakes' 7 s to 90 percent, so only the leg's floor trips -FAN_TRIP_WAIT_S = 20 # grace + three ticks, with slack for the 1 Hz pipeline - - -def _fan_gate(c, name): - return ((c.get("fan_gates") or {}).get(name) or {}) - - -@test("cooling.fan-gate-trips", title="A fan under its floor past the grace is a fault; a floor of zero is off", - subsystem="cooling", kind="auto", mode="grbl", est_min=4, - covers=_COOL_COVERS + [("forgectrl", "src/main.c")], requires=["cooling.gate-off"], - steps=["Machine idle, fans quiet. The test writes the exhaust floor, the purge current floor and " - "the spin-up grace and restores them; four short M8/M9 cycles spin the fans."], - description="The airflow gates judge a fan commanded at the cut fan profile, which a bare M8 " - "applies, armed or not. An exhaust floor no fan " - "can meet must trip AIRFLOW after the grace plus three ticks (hold, fire blocked, no " - "resume while the session lasts) and the fault must end with the session (verdict OK, " - "no hold, once the session is over); a purge current floor at the ADC rail must trip " - "the same way; a floor of zero must read off in gates_off and trip nothing; restored, " - "the next session runs OK with every fan reading at or above its floor.") -def fan_gate_trips(ctx): - fc = ctx.forgectrl - ev = ctx.evidence - before = fc.settings() - orig = {k: before.get(k, "") for k in FAN_KEYS} - ev["orig"] = orig - ctx.log("original: %s", orig) - gates = before.get("gates") or {} - exh = gates.get("cool_tach_exhaust_min_rpm") or {} - prg = gates.get("cool_purge_min_current") or {} - ctx.check(exh.get("gate") == "exhaust" and exh.get("off") == "low", - "settings reply does not describe the exhaust floor as the exhaust gate, off at zero: %s", exh) - ctx.check(prg.get("gate") == "purge" and prg.get("off") == "low", - "settings reply does not describe the purge floor as the purge gate, off at zero: %s", prg) - c0 = _cool(fc) - ctx.check(c0.get("verdict") == "OK" and c0.get("gates_off") == [], - "engine is not at OK with every gate on before the test: %s", c0) - - restored = False - with ctx.grbl() as grbl: - st = grbl.status_report() - ctx.check(st["state"].startswith("Idle"), "controller is %s", st["state"]) - try: - # Leg 1: an exhaust floor at the legal maximum. No fan reaches it, - # so after the grace and three ticks the gate trips a fault. - _set_gates(ctx, fc, {"cool_tach_exhaust_min_rpm": str(exh.get("hi")), "cool_fan_grace_s": FAN_GRACE_S}) - c = _run_session(ctx, grbl, fc, lambda c: c.get("verdict") == "AIRFLOW", "exhaust trip leg", - wait=FAN_TRIP_WAIT_S) - ev["exhaust_trip"] = {"cool": c, "gate": _fan_gate(c, "exhaust")} - ctx.check(c.get("verdict") == "AIRFLOW", "an exhaust floor of %s rpm did not trip: %s", exh.get("hi"), c) - ctx.check(c.get("fire_ok") is False and c.get("hold") is True, - "AIRFLOW without fire blocked and a hold: %s", c) - ctx.check(c.get("resume_ok") is not True, "a fan fault offered a resume: %s", c) - ctx.check(_fan_gate(c, "exhaust").get("state") == "TRIPPED", - "the exhaust gate does not read TRIPPED: %s", c.get("fan_gates")) - ctx.check("exhaust" in (c.get("reason") or ""), "the reason does not name the fan: %r", c.get("reason")) - # The fault is the session's: with the session over, the - # verdict is back to OK and nothing holds (jogs and the next - # job's pre-check must not see a fan fault at idle). - c = _after_session(ctx, fc) - ev["exhaust_trip_after"] = c - ctx.check(c.get("verdict") == "OK" and c.get("hold") is False, - "the fan fault outlived its run session: %s", c) - - # Leg 2: the purge fan by current, floor at the ADC rail. - _set_gates(ctx, fc, {"cool_tach_exhaust_min_rpm": orig["cool_tach_exhaust_min_rpm"], - "cool_purge_min_current": str(prg.get("hi"))}) - c = _run_session(ctx, grbl, fc, lambda c: c.get("verdict") == "AIRFLOW", "purge trip leg", - wait=FAN_TRIP_WAIT_S) - ev["purge_trip"] = {"cool": c, "gate": _fan_gate(c, "purge")} - ctx.check(c.get("verdict") == "AIRFLOW" and _fan_gate(c, "purge").get("state") == "TRIPPED", - "a purge current floor of %s did not trip: %s", prg.get("hi"), c) - - # Leg 3: the exhaust floor at zero is the gate off: nothing trips, - # gates_off says so, and the other fans are judged on their own. - _set_gates(ctx, fc, {"cool_tach_exhaust_min_rpm": "0", - "cool_purge_min_current": orig["cool_purge_min_current"]}) - c = _run_session(ctx, grbl, fc, lambda c: c.get("verdict") == "OK" and "exhaust" in (c.get("gates_off") or []) - and _fan_gate(c, "exhaust").get("state") == "off", - "exhaust off leg") - ev["exhaust_off"] = {"cool": c, "gate": _fan_gate(c, "exhaust")} - ctx.check(c.get("verdict") == "OK", "the exhaust floor at zero did not clear the gate: %s", c) - ctx.check("exhaust" in (c.get("gates_off") or []), "gates_off %s lacks exhaust", c.get("gates_off")) - ctx.check(_fan_gate(c, "exhaust").get("state") == "off", - "the exhaust gate does not read off: %s", c.get("fan_gates")) - - # Restore, and prove it: every fan at or above its floor, nothing off. - _set_gates(ctx, fc, orig) - restored = True - # The restored grace is the shipped one, so this leg waits it out. - c = _run_session(ctx, grbl, fc, - lambda c: c.get("verdict") == "OK" and not c.get("gates_off") - and all(g.get("state") == "ok" for g in (c.get("fan_gates") or {}).values()), - "restored", wait=FAN_TRIP_WAIT_S + 20) - ev["restored"] = c.get("fan_gates") - ctx.check(c.get("verdict") == "OK" and c.get("gates_off") == [], - "engine did not return to OK with every gate on after the restore: %s", c) - bad = {k: g for k, g in (c.get("fan_gates") or {}).items() if g.get("state") != "ok"} - ctx.check(not bad, "fans not at or above their floors after the grace: %s", bad) - ctx.log("fans at run duty: %s", {k: "%s/%s" % (g.get("reading"), g.get("floor")) - for k, g in (c.get("fan_gates") or {}).items()}) - finally: - if not restored: - st, body = fc.post("/settings", params=orig) - ctx.log("restore on failure: POST /settings %s -> %s", orig, st) - try: - c = _run_session(ctx, grbl, fc, lambda c: c.get("verdict") == "OK" and not c.get("gates_off"), - "restore on failure") - ctx.log("restore on failure: engine %s gates_off %s", c.get("verdict"), c.get("gates_off")) - except Exception as e: # the original failure is the one to report - ctx.log("restore on failure: run session did not complete (%s)", e) - after = fc.settings() - ctx.check(all(after.get(k, "") == orig[k] for k in FAN_KEYS), - "settings not restored: %s", {k: after.get(k) for k in FAN_KEYS}) - ctx.check(fc.wait_idle(60, abort=ctx.aborted), "machine did not return to idle") # The flow check under a lit tube. With a prompt press the tube is lit for @@ -1408,7 +308,7 @@ def flow_under_load(ctx): "locks the laser, and the supervisor restarts the controller", subsystem="cooling", kind="operator", mode="grbl", est_min=4, covers=_COOL_COVERS + [("forgectrl", "src/super.*")], - requires=["cooling.crash-watch-plumbing", "motion.deadman"], actions=["button"], + requires=["motion.deadman"], actions=["button"], steps=["Bed clear, lid closed, 40 mm of free +X travel. Nothing fires: the job is dark (S0).", "Press the physical button when it lights white, once. The head moves and stops on " "its own; the controller restarts."], diff --git a/forgetest/forgetest/suite/exthost.py b/forgetest/forgetest/suite/exthost.py index 55cc775..2503d9b 100644 --- a/forgetest/forgetest/suite/exthost.py +++ b/forgetest/forgetest/suite/exthost.py @@ -1171,7 +1171,7 @@ def _frozen(id_): covers=[("forgeext", "src/super.*"), ("forgeext", "src/run.*"), ("forgeext", "src/machine.*"), ("forgeext", "src/cgroup.*"), ("forgectrl", "src/cool.*"), ("forgectrl", "src/cam.*"), ("forgectrl", "src/main.c")], - requires=["exthost.service", "cooling.floor-and-warm-up"], actions=["button"], + requires=["exthost.service"], actions=["button"], steps=[OFFLINE_STEP, "Bed clear (the job is dark: a 30 s square at S0, nothing fires). Press the button " "when it lights."], diff --git a/forgetest/forgetest/suite/image.py b/forgetest/forgetest/suite/image.py index 12c02a1..3457e5f 100644 --- a/forgetest/forgetest/suite/image.py +++ b/forgetest/forgetest/suite/image.py @@ -112,6 +112,7 @@ def fds_of(pid): @test("image.health", title="Post-flash image health", subsystem="image", kind="auto", always=True, est_min=1, covers=[("forgectrl", "init/**"), ("forgeext", "init/**"), ("forgectrl", "src/main.c"), ("forgectrl", "src/auth.c"), + ("forgectrl", "src/status.*"), ("forgectrl", "CMakeLists.txt"), ("grblhal-glowforge", "src/boards/**"), ("grblhal-glowforge", "CMakeLists.txt"), ("kernel-module-glowforge", "**"), ("linux-fslc", "**")], @@ -119,7 +120,9 @@ def fds_of(pid): "version stamped in /etc/forgefirm-version and under the machine mark in the " "console banner and the motd, " "the kernel options, the module, the pulse ring it maps and the SDMA clocks it holds, " - "the daemon ownership, " + "the daemon ownership, the board temperatures /status reports (the chassis and " + "the SoC die in a plausible range, the supply as a 10-bit count, the CPU " + "unthrottled), " "the init ordering, the extension host as one process that starts after forgectrl and " "stops before it on a machine with no package installed and nothing running under a " "pool account, the data directory open for search on a machine with an account, the file modes the release depends on, and the mounts: the " @@ -260,6 +263,21 @@ def image_health(ctx): "controller is %r (expected running or standby)", mode.get("controller")) ctx.check(mode.get("motion") != "fault", "supervisor reports a motion fault") + # 4b. the board temperatures /status carries, watched and not gated: + # the chassis and the SoC die in degrees, the supply as its raw count, + # and the CPU unthrottled at idle. + temps = (ctx.forgectrl.status() or {}).get("temps") or {} + ev["temps"] = temps + ctx.log("temps: %s", temps) + ctx.check(isinstance(temps.get("chassis_c"), (int, float)) and 5.0 <= temps["chassis_c"] <= 80.0, + "/status temps.chassis_c is not a plausible chassis temperature: %s", temps) + ctx.check(isinstance(temps.get("soc_c"), (int, float)) and 10.0 <= temps["soc_c"] <= 100.0, + "/status temps.soc_c is not a plausible die temperature: %s", temps) + ctx.check(isinstance(temps.get("supply_raw"), int) and 0 <= temps["supply_raw"] <= 1023, + "/status temps.supply_raw is not a 10-bit count: %s", temps) + ctx.check(temps.get("soc_throttle") == 0, + "the CPU is throttled (or its cooling device is missing) at idle: %s", temps) + # 5. init ordering: controllers stop before the daemon k = sorted(os.path.basename(p) for p in glob.glob("/etc/rc6.d/K*")) ev["rc6_kill"] = k diff --git a/forgetest/forgetest/suite/laser.py b/forgetest/forgetest/suite/laser.py index 50a4c57..1e9cf0e 100644 --- a/forgetest/forgetest/suite/laser.py +++ b/forgetest/forgetest/suite/laser.py @@ -28,8 +28,7 @@ import time from ..catalog import test from .. import hw from ..runner import Failed -from .motion import (kernel_xy_mm, kernel_start, check_kernel_returned, Watch, machine_idle, - _job, _job_post, _job_wait, _lease_holder, _port_state, _words) +from .motion import kernel_start, check_kernel_returned, Watch _LASER_COVERS = [("grblhal-glowforge", "src/**"), ("kernel-module-glowforge", "**"), ("forgectrl", "src/super.c"), ("forgectrl", "src/cool.c"), @@ -706,7 +705,7 @@ def armed_kill(ctx): subsystem="laser", kind="live", mode="grbl", est_min=7, covers=_LASER_COVERS + [("grblhal-glowforge", "src/glowforge_switches.c"), ("grblhal-glowforge", "src/glowforge_switch_map.h")], - requires=["laser.emission-witness", "motion.lid-cancel-home"], + requires=["laser.emission-witness"], actions=["button", "lid"], steps=["Scrap under the head with 40 mm of free +X and +Y travel; lid closed; exhaust on.", "Press the physical button when it lights white (arm). Once the cut is under way the " @@ -864,119 +863,3 @@ def pause_resume_lid_cancel(ctx): ctx.log("PASS: button paused the burn (emission 0, armed kept, latch unlocked) and resumed it; " "the lid then canceled it - emission 0 at +%s s, reset without alarm, returned (drift " "%.3f mm), button latch SET", zero_at, drift) - - -# ---------------------------------------------------------------- the recorder, as a job of the runner's - -def _laser_keys(fc): - s = fc.settings() - return {k: s.get(k) for k in ("laser_floor_density", "laser_dose_curve")} - - -def _arming(fc): - rep = ((fc.status().get("grbl") or {}).get("report") or {}) - return bool((rep.get("laser") or {}).get("arming")) - - -@test("laser.recorder-dark", title="The dose-curve recorder takes the machine as a job, arms, and is stopped dark", - subsystem="laser", kind="auto", mode="grbl", est_min=2, - covers=_LASER_COVERS + [("forgectrl", "src/curverec.*"), ("forgectrl", "src/jobrun.*"), - ("forgectrl", "src/jobstream.*"), ("forgectrl", "src/lease.*"), - ("forgectrl", "src/settings.*")], - requires=["motion.job"], - steps=["Lid closed, bed clear. Nobody presses the button: the recorder is stopped before its press.", - "The ladder begins at the controller's X0 Y0: the head goes there, and is brought back."], - description="POST /curve/record starts the dose ladder as a job of the job runner's. While it " - "waits for a press that never comes: /status names the recorder as the lease " - "holder of kind sender, GET /job reports the run (owner recorder, not a posted " - "program), the floor and the curve are overridden (0 and off), the controller " - "reaches its arm wait, POST /job/abort refuses to stop a run it did not start, a " - "posted job is refused in the recorder's name, and the kernel's LASER_ON sample " - "count and the HV current stay dark. POST /curve/stop then ends it: the recorder " - "says it was stopped before the ladder fired, the lease is free, and both laser " - "keys read as they did before. The head is returned by a posted job.") -def recorder_dark(ctx): - ev = ctx.evidence - fc = hw.Forgectrl() - machine_idle(ctx) - x0, y0 = kernel_start(ctx) - keys0 = _laser_keys(fc) - base = sample(ctx) - ctx.check(base and not base["emission"], "emission_samples nonzero before the recording (%s)", base) - ctx.check(not _lease_holder(fc), "the machine is not free at the start: %s", _lease_holder(fc)) - started = False - samples = [] - try: - st, body = fc.post("/curve/record") - ctx.check(st == 200 and isinstance(body, dict) and body.get("state") == "waiting", - "POST /curve/record -> %s %s", st, _words(body)[:200]) - started = True - ok = ctx.wait_for(lambda: _lease_holder(fc).get("owner") == "recorder", 5, poll=0.1) - held = _lease_holder(fc) - ev["holder"] = held - ctx.check(ok is not None and held.get("kind") == "sender", "the lease does not name the recorder: %s", held) - rec = _job(fc) - ev["job_record"] = rec - ctx.check(rec["state"] == "running" and rec["owner"] == "recorder" and rec["program"] is False, - "GET /job while the recorder runs: %s", rec) - over = _laser_keys(fc) - ev["override"] = over - ctx.check(str(over["laser_dose_curve"]) == "off" and float(over["laser_floor_density"] or 0) == 0.0, - "the floor and the curve are not overridden for the ladder: %s", over) - - ok = ctx.wait_for(lambda: _arming(fc), 40, poll=0.25) - ctx.check(ok is not None, "the controller never reached its arm wait: %s", - (fc.status().get("grbl") or {}).get("report")) - ctx.log("the recorder holds the machine as %s, and the controller waits for the press", held) - st, body = fc.post("/job/abort") - ev["abort_route"] = [st, _words(body)[:160]] - ctx.check(st == 409 and "stop it where it was started" in _words(body), - "POST /job/abort on the recorder's run -> %s %s", st, _words(body)[:160]) - st, body = _job_post(fc, "G21\n") - ev["job_beside"] = [st, _words(body)[:160]] - ctx.check(st == 409 and "the dose-curve recorder holds the machine" in _words(body), - "a posted job beside the recorder -> %s %s", st, _words(body)[:160]) - for _ in range(16): # two seconds in the arm wait, by the witnesses - samples.append(sample(ctx)) - ctx.sleep(0.125) - ctx.check(_lease_holder(fc).get("owner") == "recorder", "a refused request took the lease away") - finally: - if started: - st, body = fc.post("/curve/stop") - ev["stopped"] = body if isinstance(body, dict) else _words(body)[:160] - ctx.check(st == 200 and isinstance(body, dict) and body.get("state") == "failed" - and "before the ladder fired" in body.get("reason", ""), "POST /curve/stop -> %s %s", - st, _words(body)[:200]) - ok = ctx.wait_for(lambda: not _lease_holder(fc), 10, poll=0.2) - ctx.check(ok is not None, "the lease was not given back: %s", _lease_holder(fc)) - keys1 = _laser_keys(fc) - ev["keys"] = {"before": keys0, "after": keys1} - ctx.check(keys1 == keys0, "the laser keys were not restored: %s, were %s", keys1, keys0) - - lit = [s for s in samples if s and s["emission"]] - hv_max = max((s["hv"] for s in samples if s and s["hv"] is not None), default=0) - ev["witnesses"] = {"samples": len(samples), "with_emission": len(lit), "hv_max": hv_max, - "job": _job(fc)["emission"]} - ctx.check(not lit and hv_max <= HV_DARK_MAX and _job(fc)["emission"]["laser_on_samples"] == 0, - "the recorder's arm wait was not dark: %s", ev["witnesses"]) - - # The ladder's first move took the head to the controller's X0 Y0. - ctx.sleep(1.0) - dx, dy = (a - b for a, b in zip(kernel_xy_mm(ctx), (x0, y0))) - ev["moved_mm"] = [round(dx, 3), round(dy, 3)] - ev["controller_after_stop"] = _port_state(fc).get("state") - ctx.log("after the stop the controller is in %s; the head is %.3f, %.3f mm from where it was found", - ev["controller_after_stop"], dx, dy) - if abs(dx) > 0.05 or abs(dy) > 0.05: - st, body = _job_post(fc, "G21\nG91\nG1 X%.3f Y%.3f F2400\nG90\n" % (-dx, -dy), unlock="1") - ctx.check(st == 200, "the returning job -> %s %s", st, _words(body)[:200]) - rec, _far = _job_wait(ctx, fc, 60) - ctx.check(rec["state"] == "done", "the returning job: %s", rec) - elif str(ev["controller_after_stop"]).startswith("Alarm"): - st, body = _job_post(fc, "G21\n", unlock="1") - rec, _far = _job_wait(ctx, fc, 20) - ctx.check(st == 200 and rec["state"] == "done", "the unlocking job: %s %s", st, rec) - check_kernel_returned(ctx, ev, (x0, y0), tag="recorder") - machine_idle(ctx) - ctx.log("PASS: the recorder ran as a job of the runner's, held the machine as a sender through its arm " - "wait with every witness dark, and its stop freed the machine and restored the laser keys") diff --git a/forgetest/forgetest/suite/motion.py b/forgetest/forgetest/suite/motion.py index 448f486..554de0e 100644 --- a/forgetest/forgetest/suite/motion.py +++ b/forgetest/forgetest/suite/motion.py @@ -131,47 +131,6 @@ def clear_error(ctx, g): "the controller answered %s to a blank line at the start - not a clean parser", r) -MOVE_START_S = 2.0 # Run is there within 100 ms; a loaded board gets this long - - -def start_move(ctx, g, line): - """Start a travel job with `line` and see the controller in Run. The - wait is bounded, not a fixed sleep. When Run never comes, the failure - names what happened instead: the reply (a refused block answers - error), the state the controller sits in (Hold, Door, Alarm, or an - Idle that never took the block), every line it said meanwhile, and - the machine as forgectrl sees it - the evidence a bare "did not - start" leaves out. Recorded in the evidence either way.""" - reply = g.command(line, timeout=1.0) - end = time.time() + MOVE_START_S - t0 = time.time() - said = "" - st = None - while time.time() < end: - ctx.checkpoint() - st = g.status_report() - if st["state"].startswith("Run"): - break - said += g.drain() - time.sleep(0.1) - rec = {"line": line, "reply": reply, "state": st["state"] if st else None, - "after_s": round(time.time() - t0, 2)} - ctx.evidence.setdefault("move_start", []).append(rec) - if st is not None and st["state"].startswith("Run"): - return st - msgs = [ln.strip() for ln in said.splitlines() if ln.strip()] - rec["said"] = msgs - s = ctx.forgectrl.status() or {} - gr = ((s.get("grbl") or {}).get("report") or {}) - rec["forgectrl"] = {"state": s.get("state"), "switches": s.get("switches"), - "laser_locked": s.get("laser_locked"), "grbl": gr.get("state"), - "alarm": gr.get("alarm"), "sender": (gr.get("sender") or {}).get("connected")} - ctx.check(False, "the move did not start: %s answered %s; controller %s %.1f s later, said %s; " - "forgectrl sees kernel %s, grbl %s (alarm %s), switches %s, laser locked %s", - line, reply, st["state"] if st else "no report", rec["after_s"], msgs or "nothing", - s.get("state"), gr.get("state"), gr.get("alarm"), s.get("switches"), s.get("laser_locked")) - - @test("motion.pacing", title="Protocol-loop pacing (idle, parked, moving) and hold/resume position", subsystem="motion", kind="auto", mode="grbl", est_min=1, covers=_MOTION_COVERS, requires=["kernel.latch-locked-idle"], @@ -600,99 +559,6 @@ def _wait_controller(ctx, fc, states, timeout, motion=None): return m if isinstance(m, dict) else {} -def _button_led(ctx, window_s=1.2): - """The commanded level of the button's red, green, and blue channels - (ledtrig_smooth's target), sampled over a blink period or two and - reduced to the maximum per channel: with a pulse set, target reads 0 - through the off half of every blink. None when the LEDs are not - readable.""" - best = None - t0 = time.time() - while time.time() - t0 < window_s: - out = [] - for ch in (1, 2, 3): - try: - with open("/sys/class/leds/button_led_%d/target" % ch, "r", encoding="utf-8") as f: - out.append(int(f.read().strip())) - except (OSError, ValueError): - return None - best = out if best is None else [max(a, b) for a, b in zip(best, out)] - ctx.sleep(0.1) - return best - - -@test("motion.gate-waits-for-lid", title="The motion gate waits for the lid", subsystem="motion", - kind="operator", est_min=2, - covers=[("forgectrl", "src/super.*"), ("forgectrl", "src/liveness.*"), ("forgectrl", "src/led.*"), - ("forgectrl", "src/ui/panel.js")], - requires=["motion.liveness-probe"], actions=["lid"], - steps=["The lid opens, forgectrl is restarted (the gate runs at the first spawn of a " - "session: power-on, a daemon restart), the lid closes: the head does not move " - "while the lid is open, then makes the 15 mm probe move (+X first)."], - description="With the lid open when forgectrl starts (power-on with the lid up, or a " - "daemon restart) the supervisor starts nothing: GET /mode reports controller " - "waiting with why naming the lid, motion unverified, no pid, the button blinks " - "amber, and no probe line reaches the log. When the lid closes the probe runs " - "(MOTION OK), the lens takes its reference, and the controller comes up " - "verified with the button handed back.") -def gate_waits_for_lid(ctx): - fc = ctx.forgectrl - ev = ctx.evidence - ctx.check(fc.wait_idle(15, abort=ctx.aborted), "machine not idle at the start") - try: - ctx.act("lid", "open") - off = _log_offset(FORGECTRL_LOG) - rc, out = hw.initd("forgectrl", "restart") - ctx.check(rc == 0, "forgectrl restart -> rc %s", rc) - m = _wait_controller(ctx, fc, ("waiting", "running", "motion-fault"), 60) - ev["waiting"] = m - ctx.log("mode with the lid open: %s", m) - ctx.check(m.get("controller") == "waiting", - "controller is %r with the lid open, expected waiting", m.get("controller")) - ctx.check("lid" in (m.get("why") or ""), "why does not name the lid: %r", m.get("why")) - ctx.check(m.get("motion") == "unverified", "motion reads %r while waiting", m.get("motion")) - ctx.check(not m.get("pid"), "a controller (pid %s) runs while the gate waits", m.get("pid")) - led = _button_led(ctx) - ev["led_waiting"] = led - ctx.log("button LED targets while waiting (r, g, b): %s", led) - ctx.check(led is not None and led[0] > 0 and led[1] > 0 and led[2] == 0, - "the button does not blink amber while the gate waits: %s", led) - ctx.sleep(3) - st, m2 = fc.get("/mode") - ctx.check(isinstance(m2, dict) and m2.get("controller") == "waiting", "the wait did not hold: %s", m2) - lines = _probe_lines(FORGECTRL_LOG, off) - ctx.check(not lines, "the probe ran with the lid open: %s", lines[:1]) - - t_close = time.time() - ctx.act("lid", "close") - m = _wait_controller(ctx, fc, ("running", "motion-fault"), 150, motion="verified") - ev["after_close"] = m - ev["close_to_running_s"] = round(time.time() - t_close, 1) - ctx.log("mode %.1f s after the lid closed: %s", time.time() - t_close, m) - ctx.check(m.get("controller") == "running" and m.get("motion") == "verified", - "the controller did not come up verified after the lid closed: %s", m) - t1 = time.time() - lines = _probe_lines(FORGECTRL_LOG, off) - while not lines and time.time() - t1 < 10: - ctx.sleep(0.5) - lines = _probe_lines(FORGECTRL_LOG, off) - ev["probe_lines"] = lines[-3:] - ctx.check(lines and "MOTION OK" in lines[0], - "the probe after the lid closed was not MOTION OK: %s", lines[:1]) - led = _button_led(ctx) - ev["led_running"] = led - ctx.check(led is None or led[0] == 0 or led[2] > 0, - "the amber wait pattern is still on the button after the start: %s", led) - finally: - if ctx.switch("lid") is False: - ctx.act("lid", "close", fail=False) - m = _wait_controller(ctx, fc, ("running", "motion-fault"), 150) - if m.get("controller") != "running": - fc.post("/controller/start") - _wait_controller(ctx, fc, ("running", "motion-fault"), 150) - ctx.check(fc.wait_idle(15, abort=ctx.aborted), "machine not idle after the drill") - - # ---------------------------------------------------------------- cancel / abort @test("motion.cancel-abort", title="Jog cancel and controlled abort recover cleanly", subsystem="motion", @@ -1115,11 +981,6 @@ def homing_kill(ctx, wait_running): # ------------------------------------------------- lid / button (the factory's) -_LID_COVERS = _MOTION_COVERS + [("grblhal-glowforge", "src/glowforge_switches.c"), - ("grblhal-glowforge", "src/glowforge_switch_map.h"), - ("grblhal-glowforge", "src/glowforge_laser.c")] - - def kernel_xy_mm(ctx): """The kernel's own position counters, in mm (forgectrl /status pos): what the machine physically did, independent of what grbl believes.""" @@ -1187,134 +1048,6 @@ def drain_text(g, seconds): return text -def expect_cancel_and_return(ctx, g, ev, start, k0, why, tag): - """The cancel policy's whole tail, shared by every trigger that ends a - job this way (lid or interlock, from Run or from a hold): the reason is - reported, the controller resets without an alarm and with the position - kept, and the head goes back to where the job started - which the KERNEL - counters have to confirm, not grbl's belief about them.""" - text = drain_text(g, 3.0) - msgs = [ln for ln in text.splitlines() - if ln.startswith("[MSG:") or "help]" in ln or ln.startswith("ALARM")] - ev[tag + "_messages"] = msgs - ctx.log("[%s] controller: %s", tag, msgs) - cancel_msg = "%s - job canceled" % why - ctx.check(cancel_msg in text, "[%s] the job was not canceled with %r as the reason", tag, why) - ctx.check("help]" in text, "[%s] no reset banner after the cancel (the sender must see the job end)", tag) - ctx.check("ALARM" not in text, "[%s] an alarm was raised on the cancel (position should be kept)", tag) - t0 = time.time() - returned = "returned to the job start" in text - while not returned and time.time() - t0 < 30: - ctx.checkpoint() - text += g.drain() - returned = "returned to the job start" in text - time.sleep(0.2) - ev[tag + "_returned_message"] = returned - ctx.check(returned, "[%s] the head did not report returning to the job start within 30 s", tag) - st = wait_state(ctx, g, "Idle", 5) - ctx.check(st is not None, "[%s] not Idle after the return (state %s)", tag, - g.status_report()["state"]) - drift = max(abs(st["MPos"][i] - start[i]) for i in range(2)) - ev[tag + "_drift_mm"] = round(drift, 3) - ctx.log("[%s] back at the job start: drift %.3f mm", tag, drift) - ctx.check(drift <= 0.05, "[%s] head not back at the job start (drift %.3f mm)", tag, drift) - machine_idle(ctx, 10) - check_kernel_returned(ctx, ev, k0, tag=tag) - return drift - - -@test("motion.lid-cancel-home", title="Lid open during a job - running or paused - cancels it and returns " - "to the job start", - subsystem="motion", kind="operator", mode="grbl", est_min=5, - covers=_LID_COVERS, requires=["motion.pacing", "motion.cancel-abort"], - actions=["lid", "button"], - steps=["Bed clear; the head needs 40 mm of free +X travel. No laser is involved.", - "Twice, on Ready the head starts an 8 s move. First: open the lid while it moves and " - "leave it open until the head has come back, then close it. Second: press the button " - "once while it moves (pause), then open the lid, leave it open until the head is back, " - "and close it."], - description="A travel job is running when the lid opens: the job parks (planned deceleration), " - "the reason is reported, the controller resets (position kept, no alarm - the " - "sender's job is over), and the head returns on its own to where the job started " - "with the lid still open; the controller ends Idle at the start position. The same " - "job paused on the button takes the same path - a lid open from the hold cancels " - "and returns, it never resumes. With lid_policy=cancel (the default).") -def lid_cancel_home(ctx): - ev = ctx.evidence - policy = (ctx.forgectrl.settings() or {}).get("lid_policy") or "cancel" - ev["lid_policy"] = policy - ctx.check(policy == "cancel", "lid_policy is %r; this test needs cancel", policy) - with ctx.grbl() as g: - clean_slate(ctx, g) - k0 = kernel_start(ctx) - start = g.status_report()["MPos"] - ev["kernel_start"] = k0 - ev["start"] = start - ctx.ready("On Ready the head starts an 8 s move along +X. Open the lid while it moves and " - "leave it open until the head has come back on its own.") - g.command("M5") - g.command("G91") - start_move(ctx, g, "G1X40F300") # an 8 s move - g.drain() # the message window opens here - ctx.act("lid", "open", text="Leave it open until the head has come back.", timeout=12) - drift = expect_cancel_and_return(ctx, g, ev, start, k0, "lid opened", "running") - sw = (ctx.forgectrl.status().get("switches") or {}) - ev["lid_at_return"] = sw.get("lid") - ctx.act("lid", "close") - ctx.sleep(1) - # a jog afterward proves the controller is usable without $X - r = g.command("$J=G91X5F1200") - ctx.check(not any(x.startswith("error") for x in r), "jog refused after the cancel: %s", r) - wait_idle(ctx, g, 15) - g.command("$J=G91X-5F1200") - wait_idle(ctx, g, 15) - - # -- the same cancel, entered from a hold --------------------------- - # A job paused on the button must not be resumable past a lid open: - # the armed window and the hardware button latch have to agree, so - # the lid ends the job here exactly as it does from Run. - # The reference for this phase is taken only once the MACHINE is at - # rest: the jogs above waited for grblHAL's Idle, which arrives while - # the kernel is still playing their tail. - k1 = kernel_start(ctx) - start2 = g.status_report()["MPos"] - ev["hold_start"] = start2 - ctx.ready("On Ready the head starts the 8 s move again. Press the button ONCE while it " - "moves (the job pauses); then open the lid and leave it open until the head has " - "come back.") - g.command("G91") # the reset restored G90 - g.command("G1X40F300", timeout=0.5) - ctx.sleep(0.5) - ctx.check(g.status_report()["state"].startswith("Run"), "the second move did not start") - g.drain() - w = Watch(g) - ctx.act("button", "press", text="The job pauses.", until=w.in_state("Hold"), timeout=12, - fail=False) - st, held = w.last, w.text - ctx.check(st is not None and st["state"].startswith("Hold"), - "the press did not hold the job (state %s)", st["state"] if st else "?") - ev["hold_state"] = st["state"] - ev["hold_pause_message"] = "job paused" in held - ctx.log("paused: %s; message seen: %s", st["state"], ev["hold_pause_message"]) - g.drain() - ctx.act("lid", "open", text="The job is paused: leave the lid open until the head has come " - "back.", timeout=60) - hold_drift = expect_cancel_and_return(ctx, g, ev, start2, k1, "lid opened", "hold") - ctx.check(not g.status_report()["state"].startswith("Hold"), - "the controller is still holding after the lid canceled the paused job") - ctx.act("lid", "close") - ctx.sleep(1) - r = g.command("$J=G91X5F1200") - ctx.check(not any(x.startswith("error") for x in r), "jog refused after the paused cancel: %s", r) - wait_idle(ctx, g, 15) - g.command("$J=G91X-5F1200") - wait_idle(ctx, g, 15) - g.command("G90") - machine_idle(ctx) - ctx.log("PASS: lid open canceled the job from Run (drift %.3f mm) and from the hold (drift %.3f mm), " - "reset without alarm, head returned to the start both times", drift, hold_drift) - - GRBLHAL_LOG = "/data/log/forgefirm/grblhal/grblhal.log" SCHED_FIFO = 1 @@ -1422,10 +1155,9 @@ def step_timing_under_load(ctx): ev["legs"]) -# ------------------------------------------------- manual home, motor release +# ------------------------------------------------- motor release HOLD_CURRENTS = ("33", "5") # pic/x_step_current, pic/y_step_current at idle -MANUAL_NOTICE = "Manual home: position set where the head was placed" def _step_currents(): @@ -1600,139 +1332,6 @@ def motor_release(ctx): "energized with no fault, and the liveness probe and the witnessed jogs say the drivers are alive") -@test("homing.manual", title="Manual home: $H declares the gantry-stop position and jogs to an offset origin", - subsystem="motion", kind="auto", mode="grbl", est_min=5, - covers=_MOTION_COVERS + [("forgectrl", "src/status.*"), ("forgectrl", "src/main.c")], - requires=["motion.jog-roundtrip"], - steps=["Bed clear; the head needs 35 mm of free +X travel and 3 mm of free +Y travel."], - description="With homing_mode = manual and no manual home offset, $H ships nothing (the kernel's " - "position record shows no byte played across it), sets X and Y to X0 Y0 where the head " - "stands, marks X and Y homed with their soft limits on (a jog past the envelope is " - "refused with error 15), leaves Z where it was, tells the sender the home was set by " - "hand, and the anchor's source reads back as manual through forgectrl. With " - "manual_home_x = 5 and manual_home_y = 3, $H declares the head at X-5 Y-3, says so, and " - "jogs it to X0 Y0 before it answers: the kernel plays the move, Grbl and forgectrl both " - "read X0 Y0, and the envelope starts at the origin. Whatever a check decides, the " - "settings are restored, the false reference is dropped and the head returned before the " - "machine is handed back.") -def manual_home(ctx): - ev = ctx.evidence - fc = hw.Forgectrl() - settings = fc.settings() or {} - keys = ("homing_mode", "manual_home_x", "manual_home_y") - ev["found"] = {k: settings.get(k) for k in keys} - offset = (5.0, 3.0) - moved = None # how far the head is from where the run found it - for k, v in (("homing_mode", "manual"), ("manual_home_x", ""), ("manual_home_y", "")): - st, body = (fc.post("/settings", data={k: v}) if v else fc.post("/settings", params={k: ""})) - ctx.check(st == 200, "%s=%r -> %s %s", k, v, st, body) - try: - with ctx.grbl() as g: - clean_slate(ctx, g) - r = g.command("$J=G91X30F2400") # away from wherever the origin was - ctx.check(not any(x.startswith("error") for x in r), "the outbound jog was refused: %s", r) - moved = [30.0, 0.0] - wait_idle(ctx, g) - z0 = g.status_report()["MPos"][2] - ctx.sleep(1.5) # the kernel has played the jog's tail - k0 = _kernel_position() - r = g.command("$H", timeout=15) - ctx.check(not any(x.startswith("error") for x in r), "$H under manual refused: %s", r) - text = "\n".join(r) + drain_text(g, 1.5) - k1 = _kernel_position() - rep = g.status_report() - st = fc.status() - ev.update(kernel_position=[list(k0), list(k1)], mpos=rep["MPos"], state=rep["state"], - homed_axes=st.get("homed_axes"), home_source=st.get("home_source"), - pos=st.get("pos")) - ctx.log("$H: %s MPos %s; kernel record %s -> %s; forgectrl homed_axes %s source %s pos %s", - rep["state"], rep["MPos"], k0, k1, ev["homed_axes"], ev["home_source"], ev["pos"]) - ctx.check(k1[3:] == k0[3:], "$H played pulse bytes: %s -> %s", k0[3:], k1[3:]) - ctx.check(k1[0] == 0 and k1[1] == 0, "the kernel counters were not cleared: %s", k1) - ctx.check(rep["MPos"][0] == 0 and rep["MPos"][1] == 0, "X and Y are %s, not X0 Y0", rep["MPos"][:2]) - ctx.check(rep["MPos"][2] == z0, "Z changed across the home: %s -> %s", z0, rep["MPos"][2]) - ctx.check(rep["state"].startswith("Idle"), "not idle after the home: %s", rep["state"]) - ctx.check(MANUAL_NOTICE in text, "the sender was not told the home was set by hand") - ctx.check((ev["homed_axes"] or 0) & 3 == 3, "forgectrl does not show X and Y homed") - ctx.check(ev["home_source"] == "manual", "the anchor's source reads %r", ev["home_source"]) - ctx.check(ev["pos"] and abs(ev["pos"]["x"]) < 0.02 and abs(ev["pos"]["y"]) < 0.02, - "forgectrl's position is not X0 Y0: %s", ev["pos"]) - # The envelope starts at the origin. - ev["past_envelope"] = _refused(ctx, g, "$J=G91X-5F600") - ctx.check(ev["past_envelope"] == "error:15", - "a jog past the envelope after the home got %s, not error:15", ev["past_envelope"]) - - # With an offset: the head at the gantry stops is minus it, and $H jogs - # it to the origin before it answers. - st, body = fc.post("/settings", data={"manual_home_x": "%g" % offset[0], - "manual_home_y": "%g" % offset[1]}) - ctx.check(st == 200, "manual_home_x/_y = %s -> %s %s", offset, st, body) - k2 = _kernel_position() - r = g.command("$H", timeout=15) - refused = any(x.startswith("error") for x in r) - text = "\n".join(r) + drain_text(g, 1.5) - rep = g.status_report() - if not refused: # declared at minus the offset, then moved - moved = [30.0 + rep["MPos"][0] + offset[0], rep["MPos"][1] + offset[1]] - ctx.check(not refused, "$H with an offset refused: %s", r) - k3 = _kernel_position() - st = fc.status() - ev.update(offset_kernel_position=[list(k2), list(k3)], offset_mpos=rep["MPos"], - offset_state=rep["state"], offset_pos=st.get("pos")) - ctx.log("$H with offset %s: %s MPos %s; kernel record %s -> %s; forgectrl pos %s", - offset, rep["state"], rep["MPos"], k2, k3, ev["offset_pos"]) - said = [ln for ln in text.splitlines() if "Manual home: the head is at X" in ln] - ev["offset_said"] = said - try: - words = said[-1].split("the head is at ")[1].split(";")[0].split() - declared = [float(words[0][1:]), float(words[1][1:])] - except (IndexError, ValueError): - declared = None - ctx.check(declared is not None and abs(declared[0] + offset[0]) < 0.01 - and abs(declared[1] + offset[1]) < 0.01, - "the sender was told the head stood at %s, not minus %s", declared, offset) - ctx.check(k3[3:] != k2[3:], "the move to the origin played no pulse byte: %s -> %s", k2[3:], k3[3:]) - ctx.check(k3[0] != 0 and k3[1] != 0, "the kernel counters did not move on X and Y: %s", k3) - ctx.check(rep["MPos"][0] == 0 and rep["MPos"][1] == 0, - "the head is at %s when $H answered, not X0 Y0", rep["MPos"][:2]) - ctx.check(rep["state"].startswith("Idle"), "not idle when $H answered: %s", rep["state"]) - ctx.check(ev["offset_pos"] and abs(ev["offset_pos"]["x"]) < 0.02 and abs(ev["offset_pos"]["y"]) < 0.02, - "forgectrl's position is not X0 Y0: %s", ev["offset_pos"]) - ev["offset_past_envelope"] = _refused(ctx, g, "$J=G91Y-1F600") - ctx.check(ev["offset_past_envelope"] == "error:15", - "a jog behind the origin got %s, not error:15", ev["offset_past_envelope"]) - finally: - # Whatever a check above decided, the machine is handed back as it was - # found: the settings restored, the false home dropped, the head returned. - for k in keys: - v = ev["found"][k] - st, body = (fc.post("/settings", data={k: v}) if v else fc.post("/settings", params={k: ""})) - ctx.log("restore %s=%r -> %s", k, v, st) - if moved: - _drop_reference(ctx, fc) - with ctx.grbl() as g: - clean_slate(ctx, g) - r = g.command("$J=G91X%.3fY%.3fF2400" % (-moved[0], -moved[1])) - ctx.check(not any(x.startswith("error") for x in r), "the return jog was refused: %s", r) - wait_idle(ctx, g) - machine_idle(ctx) - # The home cleared the kernel's counters away from where the run - # began, so the return leaves them off zero with the head where - # the run found it. The hand-back compares counters, and a - # controller start zeroes them: left like this, the next test - # that restarts the controller reads as out of place and the - # hand-back "returns" a head that never moved (on the bench - # reference it drove the head into the gantry stops). One more - # start, with the head back, leaves the counters at zero where the - # run began, which is what the rest of the catalog counts on. - _drop_reference(ctx, fc) - ctx.counters_rezeroed() - machine_idle(ctx) - ctx.log("PASS: a manual home with no offset shipped nothing and declared X0 Y0; with %s it declared " - "minus that and jogged to X0 Y0 before answering; the soft limits start at the origin, Z was " - "kept, and the home reads back as manual", offset) - - def _port_state(fc): st, body = fc.get("/motion/state") if st != 200 or not isinstance(body, dict): diff --git a/forgetest/forgetest/suite/setup_dark.py b/forgetest/forgetest/suite/setup_dark.py index c6bb5c2..007c29c 100644 --- a/forgetest/forgetest/suite/setup_dark.py +++ b/forgetest/forgetest/suite/setup_dark.py @@ -210,7 +210,7 @@ def check_sensors(ctx): subsystem="setup", 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"], precheck=intakes_counted, + requires=["forgectrl.auth"], precheck=intakes_counted, 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 the five floors are written: the " "four at 55 percent of steady and the grace from the slowest spin-up. The intake " @@ -296,7 +296,7 @@ def check_airflow(ctx): @test("setup.check-intakes", title="The intake fans step counts the machine's intake fans", subsystem="setup", kind="auto", hardware="takeover", est_min=2, covers=DARK_COVERS + [("forgectrl", "src/cool.c"), ("forgectrl", "src/airflow.*")], - requires=["forgectrl.auth", "cooling.fan-gate-trips"], precheck=intakes_counted, + requires=["forgectrl.auth"], precheck=intakes_counted, description="POST /wiz/airflow.intakes/start with the intake fan count set (the precheck " "refuses the start without it: the gate would hold every test after this one). " "The controller stops, the fans run at the run profile for 20 s, and the tachs "