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

The coolant critical tier, with its catalog test

Browse Source 
forgectrl pin a1875a8: cool_temp_critical_c (default 38 C) is the fail
tier above the coolant ceiling's pause: at or over it in a run session the
verdict is CRITICAL (fire blocked, hold, no resume this job), the fault
ends with the session, the settings API keeps it above the ceiling, and
its top turns the gate off.

Catalog: cooling.critical-tier (auto, 45 tests), driven through the
settings API like cooling.gate-off: a critical line at the ceiling is
refused (and undone should it ever be accepted); with the ceiling, the
resume gate and the critical line all under the coolant's temperature a
session reads CRITICAL rather than OVERTEMP with no resume and the reason
naming the tier and the coolant; after the session the ceiling alone
holds (OVERTEMP); with the critical line at its top the gate is off
(gates_off, the run-start log line) and the ceiling alone pauses;
restored, OK with nothing off. Four unit cases against a scripted engine.
_after_session takes the condition to wait for.

Docs: COOLING (verdict table, section 5, the settings table, the quick
reference), BRINGUP item 19 and the catalog count, the CAMPAIGN-LOG
paragraph for the fan-fault session rule on dev image 20260822145201.
This commit is contained in:
ScottW514
2026-08-22 11:41:18 -04:00
parent c1591e4f69
commit f51140e528
6 changed files with 304 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files
forgetest/forgetest/suite/cooling.py
+125 -3
View File
@@ -239,14 +239,17 @@ def _run_session(ctx, g, fc, until, what, wait=None):
_session_ended(ctx, fc, what)
def _after_session(ctx, fc, wait=5):
"""The engine's state a few ticks after a session ended."""
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 c.get("verdict") == "OK":
if c.get("phase") != "run" and until(c):
break
return c
@@ -353,6 +356,125 @@ def gate_off(ctx):
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")
@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):
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})
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)
# Leg 2: the critical line at its top is the gate off; the
# ceiling alone pauses, and gates_off says so.
_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"))
ctx.check(_tail_has(fc, "gate coolant_critical OFF: cool_temp_critical_c = %g" % top),
"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
forgetest/tests/test_cooling_suite.py
+150
View File
@@ -376,6 +376,156 @@ class GateOffTests(unittest.TestCase):
self.assertEqual(self.fc.state["settings"]["cool_temp_max"], "30")
class CriticalTierTests(unittest.TestCase):
"""cooling.critical-tier against a scripted engine: two tiers on the
coolant, the fail tier winning over the pause tier in a run session
and ending with it, the critical line off at its top, and the settings
cross-check refusing a critical line at or below the ceiling."""
UP = 22.3
ROWS = { # key: (gate, default, lo, hi, band_lo, band_hi, off)
"cool_temp_max": ("coolant_max", 33.0, 5.0, 60.0, 25.0, 38.0, "high"),
"cool_temp_resume": (None, 31.0, 5.0, 59.0, 20.0, 36.0, "none"),
"cool_temp_critical_c": ("coolant_critical", 38.0, 6.0, 70.0, 36.0, 45.0, "high"),
}
def setUp(self):
self.fc = helpers.FakeForgectrl().start()
self.grbl = FakeGrbl()
self.saved = (cooling.VERDICT_WAIT_S, cooling.SESSION_END_WAIT_S)
cooling.VERDICT_WAIT_S = 3
cooling.SESSION_END_WAIT_S = 3
self.fc.state["status"] = dict(self.fc.state["status"],
coolant={"down_c": 22.4, "up_c": self.UP, "pump": True, "tec": False},
gates_off=[])
self.fc.state["cool"] = {"phase": "idle", "verdict": "OK", "fire_ok": False, "hold": False,
"gates_off": []}
for k in self.ROWS:
self.fc.state["settings"].setdefault(k, "")
self.fc.state["logs_tail"] = {"text": ""}
self.faults = True # the engine has the critical tier
self.ends_with_session = True
self.cross_checks = True # the settings API refuses crit <= max
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 = self.saved
self.grbl.close()
self.fc.stop()
def setting(self, key):
v = self.fc.state["settings"].get(key) or ""
return float(v) if v else self.ROWS[key][1]
def _describe(self):
gates = {}
for k, (gate, d, lo, hi, blo, bhi, off) in self.ROWS.items():
v = self.setting(k)
state = "off" if (off == "high" and v >= hi) else ("ok" if blo <= v <= bhi else "warn")
gates[k] = {"gate": gate, "def": d, "lo": lo, "hi": hi, "band": [blo, bhi], "off": off,
"value": v, "state": state}
self.fc.state["settings"]["gates"] = gates
def _on_post(self, path, form):
if path != "/settings":
return None
if self.cross_checks:
merged = dict(self.fc.state["settings"])
merged.update(form)
tmax = float(merged.get("cool_temp_max") or 33.0)
tcrit = float(merged.get("cool_temp_critical_c") or 38.0)
if tcrit <= tmax:
return (400, {"error": "cool_temp_critical_c must be above cool_temp_max"})
self.fc.state["settings"].update(form)
self._describe()
return (200, self.fc.state["settings"])
def _verdict_idle(self):
"""Outside a run only the ceiling's pause tier stands."""
return "OVERTEMP" if self.UP > self.setting("cool_temp_max") else "OK"
def _engine(self, line):
self._describe()
cool = self.fc.state["cool"]
if line == "M9":
def end():
time.sleep(0.3)
cool["phase"] = "idle"
if self.ends_with_session or cool["verdict"] != "CRITICAL":
v = self._verdict_idle()
cool.update(verdict=v, fire_ok=v == "OK", hold=v != "OK", reason="")
threading.Thread(target=end, daemon=True).start()
return
if line != "M8":
return
cool["phase"] = "run"
tmax, tcrit = self.setting("cool_temp_max"), self.setting("cool_temp_critical_c")
crit_off = tcrit >= self.ROWS["cool_temp_critical_c"][3]
off = ["coolant_critical"] if crit_off else []
if crit_off:
self.fc.state["logs_tail"]["text"] += (
"Aug 22 12:00:00 forgectrl: cool: gate coolant_critical OFF: cool_temp_critical_c = 70 "
"(the high end of 6 to 70; recommended 36 to 45, default 38)\n")
if self.faults and not crit_off and self.UP >= tcrit:
cool.update(verdict="CRITICAL", fire_ok=False, hold=True, resume_ok=False,
reason="CRITICAL: coolant %.1f C at or over the %.0f C critical line - hold, "
"no resume this job" % (self.UP, tcrit), gates_off=off)
elif self.UP > tmax:
cool.update(verdict="OVERTEMP", fire_ok=False, hold=True, resume_ok=False,
reason="coolant over the ceiling", gates_off=off)
else:
cool.update(verdict="OK", fire_ok=True, hold=False, resume_ok=True, reason="", gates_off=off)
self.fc.state["status"]["gates_off"] = off
def run_test(self):
run = Run("test", "cooling.critical-tier", "t")
ctx = Context(run, None, helpers.make_test("cooling.critical-tier", []))
cooling.critical_tier(ctx)
return run
def settings_posts(self):
return [f for p, f in self.fc.posts if p == "/settings"]
# -- cases ------------------------------------------------------------------
def test_fault_then_off_then_restored_passes(self):
run = self.run_test()
self.assertEqual(run.evidence["cross_check"]["status"], 400)
self.assertEqual(run.evidence["critical"]["verdict"], "CRITICAL")
self.assertEqual(run.evidence["critical_after"]["verdict"], "OVERTEMP")
self.assertEqual(run.evidence["critical_off"]["gates_off"], ["coolant_critical"])
self.assertEqual(run.evidence["restored"]["verdict"], "OK")
posts = self.settings_posts()
self.assertEqual(posts[1], {"cool_temp_max": "6.0", "cool_temp_resume": "5.0",
"cool_temp_critical_c": "7.0"})
self.assertEqual(posts[-1], {"cool_temp_max": "", "cool_temp_resume": "", "cool_temp_critical_c": ""})
self.assertEqual(self.grbl.commands.count("M8"), 3)
self.assertEqual(self.grbl.commands.count("M9"), 3)
def test_an_engine_without_the_tier_fails_and_restores(self):
self.faults = False
with self.assertRaises(Failed) as cm:
self.run_test()
self.assertIn("did not fault", str(cm.exception))
self.assertEqual(self.settings_posts()[-1],
{"cool_temp_max": "", "cool_temp_resume": "", "cool_temp_critical_c": ""})
self.assertEqual(self.fc.state["cool"]["verdict"], "OK")
def test_a_fault_that_outlives_its_session_fails(self):
self.ends_with_session = False
with self.assertRaises(Failed) as cm:
self.run_test()
self.assertIn("after the faulted session", str(cm.exception))
def test_a_missing_cross_check_fails_before_any_session(self):
self.cross_checks = False
with self.assertRaises(Failed) as cm:
self.run_test()
self.assertIn("was accepted", str(cm.exception))
self.assertEqual(self.grbl.commands.count("M8"), 0)
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