mirror of
https://github.com/openglow-org/forgefirm.git
synced 2026-10-16 01:29:55 -07:00
forgetest: cooling.intake-one-fan, and setup.check-airflow reads the intake fan count
forgectrl c274f19 adds cool_intake_fans: at 1, a machine built with one
intake fan, the second intake is published not fitted and never judged,
and the setup's airflow check writes the count it finds.
cooling.intake-one-fan, in a suite module of its own, proves both halves
on a machine with two intake fans with the setting alone. With
cool_intake_fans 1 and an 8 s grace, a session at the cut fan profile
must publish intake_2 as not fitted with no floor while it reads a real
speed, run OK, and judge intake_1 ok at the intake floor; with the intake
floor at its legal maximum it must trip AIRFLOW (hold, fire blocked) with
intake_1 TRIPPED and intake_2 still not fitted, and the fault must end
with the session; restored, every fan reads ok at its floor. It starts
only from a machine with both intakes judged, requires
cooling.fan-gate-trips, and puts the count, the intake floor and the
grace back as found on every exit path.
setup.check-airflow now reads the count from the result: 2 needs all
four fans at least 1000 rpm, 1 needs intake 2 reported {"fitted": false}
and the other three ok; either way the setting must read what the check
found, and the count is put back with the five floors.
Fingerprints: the new test has a module of its own, so no cooling test
moves; the setup.check-airflow edit is inside its function. Diffed across
the change: 127 -> 128 tests, one added, setup.check-airflow the only
one moved.
Proof: the touched host tests pass (test_cooling_intake_suite 4,
test_setup_dark 10, test_cooling_suite 25) and pyflakes is clean.
test_cooling_intake_suite replays the test against a scripted engine,
and two broken engines fail it with the settings put back: one that
ignores the setting, and one that takes intake 1 out with intake 2.
test_setup_dark's new AirflowCheckTests take a two-fan and a one-fan
result, and fail a count the setting does not match and a result with
no count. On the bench reference, image 20260928194539 (dev) with these
files and forgectrl c274f19 bind-mounted, cooling.fan-gate-trips,
cooling.intake-one-fan and setup.check-airflow pass with a clean
baseline before and after each (forgectrl's commit has the readings).
This commit is contained in:
ScottW514
@@ -0,0 +1,176 @@
|
||||
# Copyright 2026 514 LLC d/b/a OpenGlow
|
||||
# Written by Scott Wiederhold
|
||||
# https://community.openglow.org
|
||||
# SPDX-License-Identifier: MIT
|
||||
|
||||
"""cooling.intake-one-fan replayed host-side under the real runner
|
||||
Context against a scripted engine: the fake reads cool_intake_fans and
|
||||
the floors at every M8 (the engine reloads at run start), publishes
|
||||
intake_2 not fitted with no floor when one intake fan is set, holds every
|
||||
fan in grace for the configured seconds, then judges the bench readings
|
||||
against the floors. The negative engines are the two ways the change
|
||||
could go wrong: the setting ignored (intake 2 still judged) and the
|
||||
setting taking intake 1 out with it."""
|
||||
import threading
|
||||
import time
|
||||
import unittest
|
||||
|
||||
import helpers
|
||||
from forgetest.runner import Context, Failed, Run
|
||||
from forgetest.suite import cooling, cooling_intake
|
||||
from test_cooling_suite import FakeGrbl
|
||||
|
||||
|
||||
class IntakeOneFanTests(unittest.TestCase):
|
||||
READINGS = {"exhaust": 11640, "intake_1": 4157, "intake_2": 4158, "air_assist": 11050, "purge": 625}
|
||||
FLOORS = {"exhaust": ("cool_tach_exhaust_min_rpm", 6400.0, 20000.0),
|
||||
"intake_1": ("cool_tach_intake_min_rpm", 2290.0, 20000.0),
|
||||
"intake_2": ("cool_tach_intake_min_rpm", 2290.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_intake.INTAKE_TRIP_WAIT_S)
|
||||
cooling.VERDICT_WAIT_S = 4
|
||||
cooling.SESSION_END_WAIT_S = 3
|
||||
cooling_intake.INTAKE_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.fc.state["settings"].setdefault("cool_intake_fans", "")
|
||||
self.honors_setting = True # the engine takes intake 2 out with one intake fan
|
||||
self.drops_both = False # the engine takes intake 1 out with it
|
||||
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_intake.INTAKE_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.9, default * 1.1], "off": "low", "value": v,
|
||||
"state": "off" if v <= 0 else "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"
|
||||
cool.update(verdict="OK", fire_ok=True, hold=False, reason="")
|
||||
for g in cool.get("fan_gates", {}).values():
|
||||
if g["state"] != "not fitted":
|
||||
g["state"] = "idle"
|
||||
threading.Thread(target=end, daemon=True).start()
|
||||
return
|
||||
if line != "M8":
|
||||
return
|
||||
cool["phase"] = "run"
|
||||
one = self.fc.state["settings"].get("cool_intake_fans") == "1" and self.honors_setting
|
||||
out = {"intake_2"} | ({"intake_1"} if self.drops_both else set()) if one else set()
|
||||
floors = {fan: 0.0 if fan in out else self.setting(key, default)
|
||||
for fan, (key, default, _h) in self.FLOORS.items()}
|
||||
cool.update(verdict="OK", fire_ok=True, hold=False, resume_ok=True, reason="", gates_off=[],
|
||||
fan_gates={f: {"reading": self.READINGS[f], "floor": floors[f],
|
||||
"state": "not fitted" if f in out else "grace"} for f in floors})
|
||||
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 f not in out:
|
||||
cool["fan_gates"][f]["state"] = "ok" if self.READINGS[f] >= floors[f] else "under"
|
||||
time.sleep(0.3)
|
||||
if cool["phase"] != "run":
|
||||
return
|
||||
for f in floors:
|
||||
if f not in out and floors[f] > self.READINGS[f]:
|
||||
cool["fan_gates"][f]["state"] = "TRIPPED"
|
||||
if cool["verdict"] != "AIRFLOW":
|
||||
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]))
|
||||
threading.Thread(target=judge, daemon=True).start()
|
||||
|
||||
def run_test(self):
|
||||
run = Run("test", "cooling.intake-one-fan", "t")
|
||||
ctx = Context(run, None, helpers.make_test("cooling.intake-one-fan", []))
|
||||
cooling_intake.intake_one_fan(ctx)
|
||||
return run
|
||||
|
||||
def settings_posts(self):
|
||||
return [f for p, f in self.fc.posts if p == "/settings"]
|
||||
|
||||
def restored(self):
|
||||
return {"cool_intake_fans": "", "cool_tach_intake_min_rpm": "", "cool_fan_grace_s": ""}
|
||||
|
||||
def test_one_fan_then_intake_1_trip_then_restore_pass(self):
|
||||
run = self.run_test()
|
||||
ev = run.evidence
|
||||
self.assertEqual(ev["one_fan"]["intake_2"]["state"], "not fitted")
|
||||
self.assertEqual(ev["one_fan"]["intake_1"]["state"], "ok")
|
||||
self.assertEqual(ev["intake_1_trip"]["intake_1"]["state"], "TRIPPED")
|
||||
self.assertEqual(ev["intake_1_trip"]["intake_2"]["state"], "not fitted")
|
||||
self.assertTrue(all(g["state"] == "ok" for g in ev["restored"].values()))
|
||||
posts = self.settings_posts()
|
||||
self.assertEqual(posts[0], {"cool_intake_fans": "1", "cool_fan_grace_s": "8"})
|
||||
self.assertEqual(posts[1], {"cool_tach_intake_min_rpm": "20000.0"})
|
||||
self.assertEqual(posts[-1], self.restored())
|
||||
self.assertEqual(self.grbl.commands.count("M8"), 3)
|
||||
self.assertEqual(self.grbl.commands.count("M9"), 3)
|
||||
|
||||
def test_an_engine_that_ignores_the_setting_fails_and_restores(self):
|
||||
self.honors_setting = False
|
||||
with self.assertRaises(Failed) as cm:
|
||||
self.run_test()
|
||||
self.assertIn("intake_2 is not published not fitted", str(cm.exception))
|
||||
self.assertEqual(self.settings_posts()[-1], self.restored())
|
||||
self.assertEqual(self.fc.state["cool"]["verdict"], "OK")
|
||||
|
||||
def test_an_engine_that_drops_intake_1_too_fails_and_restores(self):
|
||||
self.drops_both = True
|
||||
with self.assertRaises(Failed) as cm:
|
||||
self.run_test()
|
||||
self.assertIn("intake_1 is not judged ok at the intake floor", str(cm.exception))
|
||||
self.assertEqual(self.settings_posts()[-1], self.restored())
|
||||
|
||||
def test_a_machine_already_at_one_intake_fan_is_refused_untouched(self):
|
||||
self.fc.state["settings"]["cool_intake_fans"] = "1"
|
||||
with self.assertRaises(Failed) as cm:
|
||||
self.run_test()
|
||||
self.assertIn("already 1", str(cm.exception))
|
||||
self.assertEqual(self.settings_posts(), [])
|
||||
self.assertEqual(self.grbl.commands.count("M8"), 0)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -137,5 +137,124 @@ class SensorsCheckTests(unittest.TestCase):
|
||||
self.assertIn(("/wiz/sensors/abort", {}), [(p, f) for p, f in self.fake.posts])
|
||||
|
||||
|
||||
class AirflowCheckTests(unittest.TestCase):
|
||||
"""setup.check-airflow's judge against a scripted daemon: the check
|
||||
writes its floors and the intake fans fitted, and the test reads the
|
||||
fans the count says are there. A machine built with one intake fan
|
||||
reports intake 2 not fitted; the bench reference has two."""
|
||||
|
||||
FLOORS = {"cool_tach_exhaust_min_rpm": "6400", "cool_tach_intake_min_rpm": "2290",
|
||||
"cool_tach_air_assist_min_rpm": "6050", "cool_purge_min_current": "345",
|
||||
"cool_fan_grace_s": "12"}
|
||||
FAN = {"steady_rpm": 4150, "spinup_s": 6.8, "ok": True}
|
||||
|
||||
def result(self, intakes):
|
||||
fans = {"exhaust": dict(self.FAN), "intake 1": dict(self.FAN), "intake 2": dict(self.FAN),
|
||||
"air assist": dict(self.FAN)}
|
||||
if intakes == 1:
|
||||
fans["intake 2"] = {"fitted": False}
|
||||
return {"fans": fans, "intake_fans": intakes, "purge_current_on": 628, "purge_current_off": 74,
|
||||
"floors": dict(self.FLOORS)}
|
||||
|
||||
def setUp(self):
|
||||
import tempfile
|
||||
self.sysfs = tempfile.mkdtemp(prefix="forgetest-airflow-")
|
||||
os.makedirs(os.path.join(self.sysfs, "head"))
|
||||
with open(os.path.join(self.sysfs, "head", "purge_air"), "w") as f:
|
||||
f.write("1\n")
|
||||
self.saved_root = os.environ.get("GF_SYSFS_ROOT")
|
||||
os.environ["GF_SYSFS_ROOT"] = self.sysfs + "/"
|
||||
self.fake = helpers.FakeForgectrl().start()
|
||||
self.fake.state["settings"].update({"cool_tach_intake_min_rpm": "2300"})
|
||||
self.fake.state["cool"] = {"phase": "idle", "fan_gates": {"purge": {"reading": 628, "floor": 345}}}
|
||||
self.dark = {"id": "", "running": False, "phase": "", "progress": 0, "elapsed_s": 0,
|
||||
"log": [], "prompt": None, "result": None, "error": "",
|
||||
"shots": {"lid": False, "head": False}}
|
||||
self.versions = {}
|
||||
self.polls = 0
|
||||
self.wizard_result = self.result(2)
|
||||
self.wizard_writes = None # what the check writes; None = what its result says
|
||||
|
||||
def on_get(path, q):
|
||||
if path == "/wiz/dark":
|
||||
if self.dark["running"]:
|
||||
self.polls += 1
|
||||
if self.polls >= 2:
|
||||
r = self.wizard_result
|
||||
s = self.fake.state["settings"]
|
||||
s.update(r["floors"])
|
||||
s["cool_intake_fans"] = self.wizard_writes or str(r["intake_fans"])
|
||||
self.dark.update({"running": False, "progress": 100, "result": r})
|
||||
self.versions["airflow"] = 1
|
||||
return 200, dict(self.dark)
|
||||
if path == "/wiz":
|
||||
return 200, {"versions": dict(self.versions), "required": [], "completed": True}
|
||||
return None
|
||||
|
||||
def on_post(path, form):
|
||||
s = self.fake.state["settings"]
|
||||
if path == "/wiz/airflow/start":
|
||||
self.dark.update({"id": "airflow", "running": True, "phase": "fans to the run profile"})
|
||||
return 200, {"started": True, "id": "airflow"}
|
||||
if path == "/settings":
|
||||
for k, v in form.items():
|
||||
if v == "":
|
||||
s.pop(k, None)
|
||||
else:
|
||||
s[k] = v
|
||||
return 200, dict(s)
|
||||
return None
|
||||
self.fake.on_get = on_get
|
||||
self.fake.on_post = on_post
|
||||
|
||||
def tearDown(self):
|
||||
import shutil
|
||||
self.fake.stop()
|
||||
if self.saved_root is None:
|
||||
os.environ.pop("GF_SYSFS_ROOT", None)
|
||||
else:
|
||||
os.environ["GF_SYSFS_ROOT"] = self.saved_root
|
||||
shutil.rmtree(self.sysfs, ignore_errors=True)
|
||||
|
||||
def run_check(self):
|
||||
t = catalog.load_suite()["setup.check-airflow"]
|
||||
run = Run("test", t.id, t.title)
|
||||
t.fn(Context(run, None, t))
|
||||
return run
|
||||
|
||||
def assert_put_back(self):
|
||||
s = self.fake.state["settings"]
|
||||
self.assertEqual(s.get("cool_tach_intake_min_rpm"), "2300")
|
||||
self.assertNotIn("cool_intake_fans", s)
|
||||
self.assertNotIn("cool_tach_exhaust_min_rpm", s)
|
||||
|
||||
def test_two_intake_fans_pass_and_are_put_back(self):
|
||||
self.run_check()
|
||||
self.assert_put_back()
|
||||
|
||||
def test_one_intake_fan_passes_with_intake_2_not_fitted(self):
|
||||
self.wizard_result = self.result(1)
|
||||
self.run_check()
|
||||
self.assert_put_back()
|
||||
|
||||
def test_one_intake_fan_with_the_setting_left_at_two_fails(self):
|
||||
self.wizard_result = self.result(1)
|
||||
self.wizard_writes = "2"
|
||||
with self.assertRaises(Exception) as cm:
|
||||
self.run_check()
|
||||
self.assertIn("cool_intake_fans reads '2', the check found 1", str(cm.exception))
|
||||
self.assert_put_back()
|
||||
|
||||
def test_a_result_without_the_count_fails(self):
|
||||
r = self.result(2)
|
||||
del r["intake_fans"]
|
||||
self.wizard_result = r
|
||||
self.wizard_writes = "2"
|
||||
with self.assertRaises(Exception) as cm:
|
||||
self.run_check()
|
||||
self.assertIn("no intake fan count", str(cm.exception))
|
||||
self.assert_put_back()
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
|
||||
Reference in New Issue
Block a user