vh/forgefirm
1
0
Fork 0
mirror of https://github.com/openglow-org/forgefirm.git synced 2026-09-27 16:51:12 -07:00
Code Issues Packages Projects Releases Wiki Activity

cooling.fan-gate-trips: a fan under its floor is a fault; pin the airflow gates

Browse Source 
The acceptance catalog gains the test behind forgectrl 5a31d66: an
exhaust floor at the legal maximum must trip AIRFLOW after the grace and
three ticks (hold, fire blocked, no resume, the exhaust gate TRIPPED and
named in the reason); a purge current floor at the ADC rail must trip
the same way; an exhaust floor of zero must read off in gates_off and
trip nothing; restored, the next session must run OK with every fan at
or above its floor. Three host cases against a scripted engine that
models the grace and the readings. The covers map names src/airflow.*.

COOLING.md gains 3a (the airflow gates) and the five settings with their
ranges and bands; SERVICES.md the gate rule, the rows, the AIRFLOW
verdict and fan_gates; SAFETY.md names the floors among the verdict's
inputs; BRINGUP item 19 records the gates as in with provisional floors
and the catalog at 44. Pins: forgectrl 5a31d66, forgefirm-app 81027ff
(0.1.15+git); fetch-verified.
This commit is contained in:
ScottW514
2026-08-21 19:00:06 -04:00
parent a8a02596cd
commit 0bddc9eeab
7 changed files with 329 additions and 20 deletions
Download Patch File Download Diff File Expand all files Collapse all files
forgetest/forgetest/suite/cooling.py
+117 -7
View File
@@ -1,16 +1,18 @@
"""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, and 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)."""
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)."""
import time
from ..catalog import test
from .. import hw
_COOL_COVERS = [("forgectrl", "src/cool.*"), ("forgectrl", "src/diag.*"),
("forgectrl", "src/gates.*"), ("forgectrl", "src/settings.*"),
("forgectrl", "src/gates.*"), ("forgectrl", "src/airflow.*"),
("forgectrl", "src/settings.*"),
("forgectrl", "src/status.*"), ("forgectrl", "src/ui/**"),
("grblhal-glowforge", "src/glowforge_cooling.*"),
("kernel-module-glowforge", "src/thermal*"),
@@ -215,7 +217,7 @@ def _session_ended(ctx, fc, what):
return False
def _run_session(ctx, g, fc, until, what):
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."""
@@ -224,7 +226,7 @@ def _run_session(ctx, g, fc, until, what):
try:
t0 = time.time()
c = {}
while time.time() - t0 < VERDICT_WAIT_S:
while time.time() - t0 < (wait or VERDICT_WAIT_S):
ctx.sleep(1)
c = _cool(fc)
if until(c):
@@ -337,3 +339,111 @@ def gate_off(ctx):
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")
FAN_KEYS = ("cool_tach_exhaust_min_rpm", "cool_purge_min_current", "cool_fan_grace_s")
FAN_GRACE_S = "2" # the shortest grace the test can wait out with margin
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, 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 run while the cut fan profile is applied. 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); 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"))
# 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 []),
"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")
forgetest/tests/test_cooling_suite.py
+145
View File
@@ -374,3 +374,148 @@ class GateOffTests(unittest.TestCase):
self.assertEqual(run.evidence["orig"], {"cool_temp_max": "30", "cool_temp_resume": "28"})
self.assertEqual(self.settings_posts()[-1], {"cool_temp_max": "30", "cool_temp_resume": "28"})
self.assertEqual(self.fc.state["settings"]["cool_temp_max"], "30")
class FanGateTests(unittest.TestCase):
"""cooling.fan-gate-trips against a scripted engine: the fake reads the
floors at every M8 (the engine reloads at run start), holds every fan
in grace for the configured seconds, then judges the bench readings
(exhaust 6753, intakes 3212/3328, air assist 10997 rpm, purge 628)
against the floors: a floor a reading cannot meet trips AIRFLOW three
ticks after the grace, a floor of zero reads off."""
READINGS = {"exhaust": 6753, "intake_1": 3212, "intake_2": 3328, "air_assist": 10997, "purge": 628}
FLOORS = {"exhaust": ("cool_tach_exhaust_min_rpm", 3700.0, 20000.0),
"intake_1": ("cool_tach_intake_min_rpm", 1800.0, 20000.0),
"intake_2": ("cool_tach_intake_min_rpm", 1800.0, 20000.0),
"air_assist": ("cool_tach_air_assist_min_rpm", 6000.0, 30000.0),
"purge": ("cool_purge_min_current", 300.0, 1023.0)}
GATE_OF = {"exhaust": "exhaust", "intake_1": "intake", "intake_2": "intake",
"air_assist": "air_assist", "purge": "purge"}
def setUp(self):
self.fc = helpers.FakeForgectrl().start()
self.grbl = FakeGrbl()
self.saved = (cooling.VERDICT_WAIT_S, cooling.SESSION_END_WAIT_S, cooling.FAN_TRIP_WAIT_S)
cooling.VERDICT_WAIT_S = 4
cooling.SESSION_END_WAIT_S = 3
cooling.FAN_TRIP_WAIT_S = 4
self.fc.state["status"] = dict(self.fc.state["status"], gates_off=[])
self.fc.state["cool"] = {"phase": "idle", "verdict": "OK", "fire_ok": False, "hold": False,
"gates_off": [], "fan_gates": {}}
for key, _d, _h in self.FLOORS.values():
self.fc.state["settings"].setdefault(key, "")
self.fc.state["settings"].setdefault("cool_fan_grace_s", "")
self.trips = True # the engine trips an unmeetable floor
self.reports_off = True # the engine reports a zero floor as off
self.grace_scale = 0.1 # seconds of fake grace per configured second
self._describe()
self.grbl.on_command = self._engine
self.fc.on_post = self._on_post
def tearDown(self):
cooling.VERDICT_WAIT_S, cooling.SESSION_END_WAIT_S, cooling.FAN_TRIP_WAIT_S = self.saved
self.grbl.close()
self.fc.stop()
def setting(self, key, default):
v = self.fc.state["settings"].get(key) or ""
return float(v) if v else default
def _describe(self):
gates = {}
for fan, (key, default, hi) in self.FLOORS.items():
v = self.setting(key, default)
gates[key] = {"gate": self.GATE_OF[fan], "def": default, "lo": 0.0, "hi": hi,
"band": [default * 0.7, default * 1.4], "off": "low", "value": v,
"state": "off" if v <= 0 else "ok"}
g = self.setting("cool_fan_grace_s", 15.0)
gates["cool_fan_grace_s"] = {"gate": None, "def": 15.0, "lo": 0.0, "hi": 120.0, "band": [5, 30],
"off": "none", "value": g, "state": "ok"}
self.fc.state["settings"]["gates"] = gates
def _on_post(self, path, form):
if path != "/settings":
return None
self.fc.state["settings"].update(form)
self._describe()
return (200, self.fc.state["settings"])
def _engine(self, line):
self._describe()
cool = self.fc.state["cool"]
if line == "M9":
def end():
time.sleep(0.3)
cool["phase"] = "idle"
threading.Thread(target=end, daemon=True).start()
return
if line != "M8":
return
cool["phase"] = "run"
floors = {fan: self.setting(key, default) for fan, (key, default, _h) in self.FLOORS.items()}
off = sorted({self.GATE_OF[f] for f, v in floors.items() if v <= 0}) if self.reports_off else []
cool.update(verdict="OK", fire_ok=True, hold=False, resume_ok=True, reason="", gates_off=off,
fan_gates={f: {"reading": self.READINGS[f], "floor": floors[f],
"state": "off" if floors[f] <= 0 else "grace"} for f in floors})
self.fc.state["status"]["gates_off"] = off
grace = self.setting("cool_fan_grace_s", 15.0) * self.grace_scale
def judge():
time.sleep(grace)
if cool["phase"] != "run":
return
for f in floors:
if floors[f] > 0:
cool["fan_gates"][f]["state"] = "ok" if self.READINGS[f] >= floors[f] else "under"
time.sleep(0.3)
if cool["phase"] != "run" or not self.trips:
return
for f in floors:
if 0 < floors[f] > self.READINGS[f]:
cool["fan_gates"][f]["state"] = "TRIPPED"
cool.update(verdict="AIRFLOW", fire_ok=False, hold=True, resume_ok=False,
reason="AIRFLOW: %s %d under the %d floor for 3 s - hold, no resume this job"
% (f, self.READINGS[f], floors[f]))
break
threading.Thread(target=judge, daemon=True).start()
def run_test(self):
run = Run("test", "cooling.fan-gate-trips", "t")
ctx = Context(run, None, helpers.make_test("cooling.fan-gate-trips", []))
cooling.fan_gate_trips(ctx)
return run
def settings_posts(self):
return [f for p, f in self.fc.posts if p == "/settings"]
def test_trip_purge_off_and_restore_pass(self):
run = self.run_test()
ev = run.evidence
self.assertEqual(ev["exhaust_trip"]["cool"]["verdict"], "AIRFLOW")
self.assertEqual(ev["exhaust_trip"]["gate"]["state"], "TRIPPED")
self.assertEqual(ev["purge_trip"]["gate"]["state"], "TRIPPED")
self.assertEqual(ev["exhaust_off"]["cool"]["gates_off"], ["exhaust"])
self.assertTrue(all(g["state"] == "ok" for g in ev["restored"].values()))
posts = self.settings_posts()
self.assertEqual(posts[0], {"cool_tach_exhaust_min_rpm": "20000.0", "cool_fan_grace_s": "2"})
self.assertEqual(posts[-1], {"cool_tach_exhaust_min_rpm": "", "cool_purge_min_current": "",
"cool_fan_grace_s": ""})
self.assertEqual(self.grbl.commands.count("M8"), 4)
self.assertEqual(self.grbl.commands.count("M9"), 4)
def test_an_engine_that_does_not_trip_fails_and_restores(self):
self.trips = False
with self.assertRaises(Failed) as cm:
self.run_test()
self.assertIn("did not trip", str(cm.exception))
self.assertEqual(self.settings_posts()[-1], {"cool_tach_exhaust_min_rpm": "", "cool_purge_min_current": "",
"cool_fan_grace_s": ""})
self.assertEqual(self.fc.state["cool"]["verdict"], "OK")
def test_an_engine_that_hides_an_off_floor_fails(self):
self.reports_off = False
with self.assertRaises(Failed) as cm:
self.run_test()
self.assertIn("lacks exhaust", str(cm.exception))
self.assertEqual(self.fc.state["settings"]["cool_tach_exhaust_min_rpm"], "")