Files
forgefirm/scripts/bench/flow_sustained.py
T
ScottW514 cd8a01a3d9 bench: every board-runnable tool ported to the bench page
The remaining bench diagnostics run from forgetest's #bench tab. The
tools that also run from a LAN host share scripts/bench/gfbench.py:
GF_HOST names a remote machine (host mode, sysfs through ssh, Grbl and
forgectrl over the LAN); unset, the tool runs on the board itself
(local mode, sysfs directly, everything on 127.0.0.1), which is how the
page runs them - with GF_HOST=127.0.0.1, the panel token in GF_TOKEN
and their data files under <data>/bench/ (FORGETEST_BENCH_DATA). The
helper also reads a machine setting from forgectrl, or from the settings
file on the board while forgectrl is stopped.

Ported: pwm_sweep / pwm_hold (scope = a takeover; the latch relocked,
the write refused if FIRE or LASER_ON reads active), pwm_stream_test
(PASS/FAIL exit), flow_characterize, flow_recheck_char,
flow_warm_validate and flow_matrix (takeovers: forgectrl owns the
thermal hardware, so the page's takeover replaces the tools' own
controller stop/restart, whose command line predated the supervisor;
results and logs in the bench data directory), flow_sustained,
fan_test, temp_calibrate (dry; watch bounded in seconds; the threshold
and the coolant conversion from the shared code), flow_escalate_drill
(cool_confirm_max_s shortened through forgectrl's settings for the
drill and restored; the setting's minimum is the default budget), and
live_fire_drills (<drill> [S] [F], all six drills, host from GF_HOST,
token from the board). flow_matrix joins the registry. What stays
unported cannot run against the machine at all: the two null-sink CI
harnesses and the .puls decoder.

Runner: a scope tool runs inside the takeover wrapper; the bench
environment above is passed to every tool. Tests: test_bench_registry
(registry <-> scripts/bench consistency, every ported tool builds its
command line, every script compiles, gfbench host/local modes) and the
server test (scope tool takeover, the environment reaching the tool).
Local mode smoke-run on the bench (temp_calibrate watch, setting, token)
from /tmp, removed after.

No catalog consequence: bench tools are not image components (dev-only
forgetest); the acceptance catalog is unchanged.
2026-08-16 16:28:39 -04:00

83 lines
2.6 KiB
Python

#!/usr/bin/env python3
"""Sustained-load test for periodic flow re-checks.
Question: what does repeated interrogation cost thermally over a long
job, and does the loop reach equilibrium or climb without bound? Runs
the driver's real check cadence (M8 held for the duration) with the
cut-profile fans, logging bulk coolant temperature and every verdict.
Needs the controller running with forgectrl's cooling engine (the
verdict lines arrive on the Grbl socket). Runs on the board or from a
host (gfbench: GF_HOST).
Usage: flow_sustained.py [minutes] (default 30)
"""
import re
import socket
import sys
import time
from gfbench import HOST, board, degc, setting
THRESHOLD = float(setting('cool_flow_rise', 14.4)) # forgectrl's configured threshold
def temps():
o = board('cat /sys/glowforge/pic/water_temp_1 /sys/glowforge/pic/water_temp_2').split()
return degc(o[0]), degc(o[1])
minutes = float(sys.argv[1]) if len(sys.argv) > 1 else 30
s = socket.create_connection((HOST, 23), timeout=5)
s.settimeout(0.3)
time.sleep(0.5)
try:
while s.recv(4096):
pass
except socket.timeout:
pass
d0, u0 = temps()
print('start: down=%.2f up=%.2f running M8 for %.0f min' % (d0, u0, minutes))
s.sendall(b'M8\n')
t0 = time.time()
verdicts = []
peak_up = u0
buf = ''
while time.time() - t0 < minutes * 60:
try:
buf += s.recv(4096).decode('ascii', 'replace')
except socket.timeout:
pass
for line in buf.split('\n'):
if 'flow' in line and ('verified' in line or 'FAULT' in line):
m = re.search(r'rise ([\d.]+) C', line)
if m and (not verdicts or verdicts[-1][1] != float(m.group(1))):
el = (time.time() - t0) / 60
verdicts.append((el, float(m.group(1)), 'FAULT' in line))
d, u = temps()
peak_up = max(peak_up, u)
print(' %5.1f min rise=%5.2f %-8s loop down=%.2f up=%.2f (%+.2f from start)'
% (el, float(m.group(1)), 'FAULT' if 'FAULT' in line else 'ok', d, u, u - u0),
flush=True)
buf = buf[-2000:]
time.sleep(5)
s.sendall(b'M9\n')
time.sleep(1)
s.close()
d, u = temps()
print()
print('checks run: %d over %.0f min' % (len(verdicts), minutes))
if verdicts:
rises = [v[1] for v in verdicts]
print('rise values: min=%.2f max=%.2f mean=%.2f (threshold %.1f)'
% (min(rises), max(rises), sum(rises) / len(rises), THRESHOLD))
print('false faults: %d' % sum(1 for v in verdicts if v[2]))
print('loop temperature: start %.2f -> end %.2f (%+.2f C), peak %.2f'
% (u0, u, u - u0, peak_up))
print('cadence: %.1f min between checks' % (minutes / max(1, len(verdicts))))