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

forgetest: cloud tests prove the session from the client's log and hand the head back

Browse Source 
cloud.mode-switch took the optional connect-time firmware probe file as
the evidence of a live session and failed on a bench where that check is
off; the evidence is now gfcloud's own authenticate/ws-connect lines in
the unified log after the switch, the probe recorded when present. Cloud
mode's connect clears the kernel position counters at the head's start
and its hunt homes the head to the corner: the test tells the runner the
counters were re-zeroed (Context.counters_rezeroed) and jogs the head
back by the counter-measured displacement, and hands the lid lamp back
at the level it found. cloud.gfhome-homing documents that it leaves the
machine homed at the corner.

Baseline: a displaced head is jogged back along its own path by the
kernel-measured X/Y delta through the GRBL controller (bounded 100 mm,
waits out a controller respawn backoff); Z is never touched.
This commit is contained in:
ScottW514
2026-08-16 14:21:54 -04:00
parent d4e2f85538
commit a80d7aabfb
4 changed files with 170 additions and 22 deletions
Download Patch File Download Diff File Expand all files Collapse all files
forgetest/forgetest/suite/cloud.py
+102 -18
View File
@@ -12,6 +12,64 @@ _CLOUD_COVERS = [("forgefirm-app", "**"), ("python3-gfhardware", "**"), ("python
("forgectrl", "src/super.c"), ("forgectrl", "src/main.c")]
GF_LATEST = "/data/forgefirm/gf-latest.json"
GFCLOUD_LOG = "/data/log/forgefirm/gfcloud/gfcloud.log"
SESSION_MARKS = ("authenticate_machine SUCCESS", "ws_connect ESTABLISHED")
RETURN_MAX_MM = 600.0 # the head comes back from the home corner across the bed
def log_size(path):
try:
return os.path.getsize(path)
except OSError:
return 0
def session_lines(path, offset):
"""New gfcloud log lines since offset that carry a session mark."""
try:
with open(path, "rb") as f:
f.seek(offset)
data = f.read().decode("utf-8", "replace")
except OSError:
return []
return [ln.strip()[:160] for ln in data.splitlines() if any(m in ln for m in SESSION_MARKS)]
def session_established(lines):
return all(any(m in ln for ln in lines) for m in SESSION_MARKS)
def return_head(ctx, feed=2400):
"""Jog the head back to where the run found it: cloud mode re-zeroed the
kernel counters at the starting position, so the counters now read the
displacement (the home corner). Ends on the machine idle."""
fc = ctx.forgectrl
pos = fc.status().get("pos") or {}
x, y = float(pos.get("x", 0.0)), float(pos.get("y", 0.0))
ctx.log("head displacement since the switch: X %.3f Y %.3f mm", x, y)
ctx.check(abs(x) <= RETURN_MAX_MM and abs(y) <= RETURN_MAX_MM,
"displacement %.1f/%.1f mm exceeds %.0f mm - not jogging back", x, y, RETURN_MAX_MM)
if abs(x) < 0.05 and abs(y) < 0.05:
return
with ctx.grbl() as g:
st = g.status_report()["state"]
if st.startswith("Alarm"):
g.command("$X")
r = g.command("$J=G91X%.3fY%.3fF%d" % (-x, -y, feed))
ctx.check(not any(k.startswith("error") for k in r), "return jog refused: %s", r)
t0 = time.time()
while time.time() - t0 < 120:
ctx.checkpoint()
st = g.status_report()["state"]
if st.startswith("Idle") and time.time() - t0 > 0.5:
break
time.sleep(0.2)
g.command("G90")
ctx.check(fc.wait_idle(15, abort=ctx.aborted), "machine not idle after the return jog")
pos = fc.status().get("pos") or {}
ctx.log("head returned: counters X %.3f Y %.3f mm", float(pos.get("x", 0)), float(pos.get("y", 0)))
ctx.check(abs(float(pos.get("x", 0))) < 0.1 and abs(float(pos.get("y", 0))) < 0.1,
"head not back at the start after the return jog: %s", pos)
def wait_mode(ctx, fc, want_mode, want_controller="running", timeout=90):
@@ -43,12 +101,14 @@ def grbl_port_open(timeout=5):
@test("cloud.mode-switch", title="Controller mode switch grbl -> cloud -> grbl", subsystem="cloud",
kind="auto", est_min=4,
covers=_CLOUD_COVERS, requires=["forgectrl.auth", "motion.pacing"],
steps=["Bed clear (the supervisor's liveness probe may jog the head a few mm on a "
"controller spawn). Cloud credentials configured; the machine on the network."],
steps=["Bed clear: the cloud client homes the head to the corner on connect (the factory "
"hunt) and the test jogs it back to where it started afterward. Cloud credentials "
"configured; the machine on the network."],
description="POST /mode switches to the cloud controller: gfcloud comes up under "
"supervision and records its connect-time service probe (/status gfsvc); the "
"camera service survives the switch; switching back brings grblHAL up with the "
"Grbl port open and Idle.")
"supervision, authenticates and establishes its service session (its own log "
"lines are the evidence; the connect-time firmware probe is recorded when "
"configured); the camera service survives the switch; switching back brings "
"grblHAL up with the Grbl port open and Idle, and the head returns to its start.")
def mode_switch(ctx):
fc = ctx.forgectrl
ev = ctx.evidence
@@ -64,6 +124,8 @@ def mode_switch(ctx):
probe_before = os.stat(GF_LATEST).st_mtime
except OSError:
pass
log_offset = log_size(GFCLOUD_LOG)
lamp0 = hw.sysfs_read("pic/lid_led")
st, body = fc.post("/mode", data={"controller": "cloud"})
ctx.log("POST /mode controller=cloud -> %s %s", st, body)
@@ -73,22 +135,28 @@ def mode_switch(ctx):
ctx.log("mode after switch: %s", m)
ctx.check(m and m.get("mode") == "cloud" and m.get("controller") == "running",
"cloud controller did not come up: %s", m)
# the connect-time service probe is the evidence of a live cloud session
# the client's own session lines are the evidence of a live cloud session
t0 = time.time()
session = []
probe = None
while time.time() - t0 < 120:
ctx.checkpoint()
try:
mt = os.stat(GF_LATEST).st_mtime
if probe_before is None or mt > probe_before:
with open(GF_LATEST) as f:
probe = json.load(f)
break
except (OSError, ValueError):
pass
session = session_lines(GFCLOUD_LOG, log_offset)
if session_established(session):
break
time.sleep(2)
ev["session"] = session
for ln in session:
ctx.log(" gfcloud: %s", ln.split(" ", 1)[-1] if " " in ln else ln)
try:
mt = os.stat(GF_LATEST).st_mtime
if probe_before is None or mt > probe_before:
with open(GF_LATEST) as f:
probe = json.load(f)
except (OSError, ValueError):
pass
ev["gf_probe"] = probe
ctx.log("cloud service probe: %s", probe)
ctx.log("cloud session established: %s; firmware probe: %s", session_established(session), probe)
st, cam1 = fc.get("/cam/status")
ev["cam_during_cloud"] = cam1
ctx.check(st == 200 and isinstance(cam1, dict), "camera status lost during cloud mode (%s)", st)
@@ -109,18 +177,31 @@ def mode_switch(ctx):
ev["grbl_state"] = st
ctx.log("grbl state after: %s", st)
ctx.check(st.startswith("Idle") or st.startswith("Alarm"), "grbl reports %s", st)
ctx.check(probe is not None, "the cloud controller never recorded a service probe "
"(no credentials, no network, or the service refused) - cloud mode not proven")
ctx.check(session_established(session),
"the cloud client never established its service session (no credentials, no "
"network, or the service refused) - cloud mode not proven")
st, cam2 = fc.get("/cam/status")
ev["cam_after"] = cam2
ctx.check(st == 200, "camera status lost after the switch back")
# cloud mode's connect cleared the kernel counters at the starting position
# and its hunt homed the head: bring it back
ctx.counters_rezeroed()
return_head(ctx)
# cloud mode sets its own lid-lamp level (LLvl) and leaves it: hand back the level found
lamp1 = hw.sysfs_read("pic/lid_led")
ev["lid_lamp"] = {"before": lamp0, "after_cloud": lamp1}
if lamp0 is not None and lamp1 != lamp0:
hw.sysfs_write("pic/lid_led", lamp0)
ctx.log("lid lamp: cloud mode left %s, restored %s", lamp1, lamp0)
@test("cloud.gfhome-homing", title="Glowforge web-service homing ($H with homing_mode=gfcloud)",
subsystem="cloud", kind="operator", est_min=5,
covers=_CLOUD_COVERS + [("grblhal-glowforge", "src/**")], requires=["cloud.mode-switch"],
steps=["homing_mode = gfcloud and cloud credentials configured; bed clear, lid closed.",
"Watch the gantry: the service drives it to the corner with camera corrections."],
"Watch the gantry: the service drives it to the corner with camera corrections.",
"The machine ends homed, the head parked at the home corner (the position "
"counters are re-anchored there)."],
description="In grbl mode, $H runs gfhome: the web-service homing session with the "
"head-accelerometer motion witness. The controller returns to Idle with "
"homed:true within the session timeout, and the operator confirms the head "
@@ -161,3 +242,6 @@ def gfhome_homing(ctx):
ctx.log("homing: homed=%s after %.1f s (kernel %s, grbl %s)", homed, ev["homing_s"], state, gs)
ctx.check(homed, "homing did not complete (grbl %s)", gs)
ctx.confirm("Did the head travel to the home corner under camera corrections and stop there?")
# homed: the counters are re-anchored at the corner, where the head stays
ctx.counters_rezeroed()
ctx.check(fc.wait_idle(15, abort=ctx.aborted), "machine not idle after homing")