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

cooling.gate-off: let each run session end before opening the next

Browse Source 
On the bench the test tripped the gate and then failed its off leg: its
M9 and the next M8 were 300 ms apart, the GRBL client reports at 1 Hz
and the engine samples at 1 Hz, so the engine never saw the session
end, the next M8 was not a new run start, and the ceiling was never
re-read. Every M9 now waits for the engine's phase to leave run before
the next M8, and an M8 refuses to open on a session still running.

The failure path restores the settings and then cycles a run session
too: the engine reads settings at run start only, so restoring the file
alone left the bench holding OVERTEMP against the test's 6 C ceiling
until the operator's next job. The scripted engine in the host cases
models the report period, and two cases pin both rules.
This commit is contained in:
ScottW514
2026-08-21 17:26:14 -04:00
parent bf3483e994
commit f274eb1137
2 changed files with 78 additions and 12 deletions
Download Patch File Download Diff File Expand all files Collapse all files
forgetest/forgetest/suite/cooling.py
+37 -9
View File
@@ -178,6 +178,7 @@ def fans_quiet(ctx):
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
@@ -200,9 +201,25 @@ 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):
"""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."""
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()
@@ -217,6 +234,7 @@ def _run_session(ctx, g, fc, until, what):
return c
finally:
g.command("M9")
_session_ended(ctx, fc, what)
def _tail_has(fc, needle):
@@ -256,10 +274,10 @@ def gate_off(ctx):
ctx.check(isinstance(top, (int, float)) and isinstance(bottom, (int, float)), "no range in the reply: %s", g_default)
restored = False
try:
with ctx.grbl() as grbl:
st = grbl.status_report()
ctx.check(st["state"].startswith("Idle"), "controller is %s", st["state"])
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.
@@ -301,10 +319,20 @@ def gate_off(ctx):
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)
finally:
if not restored:
st, body = fc.post("/settings", params=orig)
ctx.log("restore on failure: POST /settings %s -> %s", orig, st)
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})
forgetest/tests/test_cooling_suite.py
+41 -3
View File
@@ -224,8 +224,9 @@ class GateOffTests(unittest.TestCase):
def setUp(self):
self.fc = helpers.FakeForgectrl().start()
self.grbl = FakeGrbl()
self.saved = cooling.VERDICT_WAIT_S
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": 22.3, "pump": True, "tec": False},
gates_off=[])
@@ -235,12 +236,13 @@ class GateOffTests(unittest.TestCase):
self.log_line = True # the engine writes its run-start line
self.report_off = True # the engine reports the off gate
self.trips = True # the engine trips a low ceiling
self.sessions = 0 # run sessions the engine saw (M8 with the phase not run)
self._describe()
self.grbl.on_command = self._engine
self.fc.on_post = self._on_post
def tearDown(self):
cooling.VERDICT_WAIT_S = self.saved
cooling.VERDICT_WAIT_S, cooling.SESSION_END_WAIT_S = self.saved
self.grbl.close()
self.fc.stop()
@@ -267,11 +269,25 @@ class GateOffTests(unittest.TestCase):
return (200, self.fc.state["settings"])
def _engine(self, line):
"""M8 opens a run session: the engine re-reads the ceiling and
ticks the gate. M9 ends it a report period later (the phase
leaves run); a hold taken against the old ceiling stands until
the next session re-reads."""
self._describe()
cool = self.fc.state["cool"]
if line == "M9":
def end():
time.sleep(0.3)
cool["phase"] = "smoke"
time.sleep(0.2)
cool["phase"] = "idle"
threading.Thread(target=end, daemon=True).start()
return
if line != "M8":
return
self.sessions += 1
cool["phase"] = "run"
v = self.ceiling()
cool = self.fc.state["cool"]
off = v >= self.TOP
if off and self.log_line:
self.fc.state["logs_tail"]["text"] += (
@@ -314,6 +330,28 @@ class GateOffTests(unittest.TestCase):
self.assertIn("did not trip", str(cm.exception))
self.assertEqual(self.settings_posts()[-1], {"cool_temp_max": "", "cool_temp_resume": ""})
self.assertEqual(self.fc.state["settings"]["cool_temp_max"], "")
# The restore cycles a run session so the engine re-reads the
# restored values; the bench is not left holding on the test's.
self.assertEqual(self.grbl.commands.count("M8"), 2)
self.assertEqual(self.grbl.commands.count("M9"), 2)
self.assertEqual(self.fc.state["cool"]["verdict"], "OK")
def test_every_session_waits_for_the_previous_one_to_end(self):
"""Each M8 must find the engine out of phase run, or the engine
never re-reads: the bench failure behind this case sent M9 and
the next M8 300 ms apart and the 1 Hz report pipeline swallowed
the session end."""
seen = []
inner = self._engine
def engine(line):
if line == "M8":
seen.append(self.fc.state["cool"]["phase"])
inner(line)
self.grbl.on_command = engine
self.run_test()
self.assertEqual(seen, ["idle", "idle", "idle"])
self.assertEqual(self.sessions, 3)
def test_an_engine_that_hides_the_off_gate_fails(self):
self.report_off = False