mirror of
https://github.com/openglow-org/forgefirm.git
synced 2026-09-27 16:51:12 -07:00
XY microstep modes: the baseline, the catalog test and the bench tools
The baseline derives x/y_mode, step_freq, ramp_rate and the configured markers from the xy_microsteps setting; ramp_rate joins the GRBL controller's set (the driver writes it; the cloud client runs at the module's). motion.microstep-modes cycles 8, 16 and 32: the save restarts the idle controller, the kernel reads the mode with its tick and ramp, $100/$101 are the mode's and a typed $100 is overwritten, a 40 mm jog at top speed returns to Idle with the kernel counters over the mode agreeing with the commanded travel and the accelerometer seeing the head move; the setting is put back as found. Bench tools: xy_mode_test.py (the null-sink harness the grblHAL CI runs), raster_dry.py (a top-speed raster per mode), xy_pattern_accel.py (the operator's pattern from home with the machine silent and the head accelerometer listening), arc_tolerance_sweep.py (a $12 ladder on the 9 in circle: the chord rate the protocol loop feeds, about 300 a second, is the ceiling, not the core), and the xymode and xycircle live drills. BRINGUP carries the present state and the facts; CAMPAIGN-LOG the dated record, including the planner-blocks spin (a $398 of 255 or more loops forever at start, a core bug) and its recovery.
This commit is contained in:
ScottW514
@@ -35,6 +35,10 @@ page's takeover does that; from a host, stop them first.
|
||||
| `z_envelope_test.py` | Host-side Z envelope harness (null-sink controller): the Z soft limit belongs to the driver, not to `$20`, so the driver re-applies `sys.work_envelope`, `sys.homed` and `sys.soft_limits` for Z from the settings-changed chain. Checks that an unreferenced Z is collapsed to where the lens stands and blocks a move each way, that X and Y stay free, and that neither a `$20` write (the core clears the soft-limit mask in the setter) nor a `$132` write (which un-homes the axis as well) frees Z. Runs in the grblHAL repo's CI. |
|
||||
| `live_fire_drills.py` | **LIVE LASER** drills, on the board (the bench page) or from a LAN host (`GF_HOST`): `live_fire_drills.py <drill> [S] [F]` - `witness` (emission witness, lid-IR peaks vs the ambient baseline, HV current, job-based disarm on M2), `hold` (disarm grace in Hold), `faultpos` (armed job refuses a stale origin after an underrun), `ircut` (lid-IR characterization cut at S/F), `pthresh` (laser power-threshold ladder: 13 constant-power rungs from 2 % to 30 % of full on scrap; the lowest rung that marks is the tube's striking threshold and reads directly as the `$35` value - requires `$35` = 0 for the run), `dladder` (density ladder at a chosen base period), `pcurve` (laser performance-curve ladder: one 100 mm line per level at 10 mm/s under M3, the laser off between rungs and a mid-ladder rung repeated at the end; reads `pic/hv_current` and the head thermopile `head/beam_detect_analog` (a scatter detector in the beam path upstream of the final mirror, so it sees the beam, not the material) from sysfs at ~25 Hz on the board, brackets each rung on the controller's Run/Idle states, and reports per rung the current with a clipped-at-1023 flag, the thermopile delta over its laser-off baseline and in-line drift, then the normalized curve, monotonicity, a line fit with its threshold intercept and the repeat-rung drift; JSON record with the raw trace in the bench data directory; rungs follow `laser_power_model`, a comma list overrides; a curve measurement wants `$35` = 0), `dpatch [F] [pitch] [length]` (depth witness for the density dose curve: two rows of small serpentine-filled patches, row A CW at feeds giving relative doses 1.0 to 0.25 of the reference feed, row B at the reference feed at 100/80/60/45/30 % density; the operator matches each row-B patch to the row-A patch of equal depth, which reads the density's light fraction off the material beside the thermopile's prediction; JSON record), `m4feeds [S] [F1] [F2]` (the density time base across feeds: one out-and-back line pair per feed at the same S under M4 density, one armed run; the operator reads within-line evenness and reversal darkness at both feeds - M4's velocity scaling is what holds dose per mm through the accel), `m4corner [S] [F]` (M4 velocity-scaled power into corners: a corner-heavy vector pattern at 30 % under M4 density, one armed run; the operator confirms every commanded segment marks - the floor makes a dropout unreachable - and the drill asserts the arm report, one discharge window and dark after), `m5dark` (the rapids after an M5 ship dark: one 20 mm line at M3 S400, M5, dwell, rapid back, dwell, rapid forward; PASS when the 25 Hz current trace shows one discharge segment and reads dark after the M5 and `laser_on_sampled` never re-lights; the catalog's `laser.m5-rapid-dark` is its port), `flowload` (cooling under laser load, one armed run per invocation, the conf keys it writes put back at the end, the pump never commanded off: `t1` reproduces the flow-check trip with the check on at its defaults and two 30 x 4 mm CW fills at F1500 starting on the press with no dark dwell, and reports the engine's rise/dT verdict beside the 25 Hz trace of both coolant sensors, the current, the digital witness and the heater output in 5 s bins across the window, with the shape at fire start; `t2 <secs> [pct]` runs with the check off and one fill of about `secs` lit seconds at CW or at `pct` density, and reports the lag to each sensor, the rise per raw-second of `hv_current` and what a full 50 s window would add against the 1.6 C margin; `fit` fits rise against dose over every t2 record; JSON records), `expstop` (armed kill on the expected-stop path; needs the panel token - `GF_TOKEN`, or the board's token file) and `ctrlstart` (the separate controller restart after it). Every drill waits for the operator's physical arm press; eye protection, fire watch, extinguisher, and exhaust are mandatory. |
|
||||
| `pacing_test.py` | Protocol-loop pacing check (runs on the board, dry motion): idle and parked-in-Hold states are coarse-paced, active motion is tight-paced, and a feed-hold/resume mid-move preserves position with no feeder starve. |
|
||||
| `xy_mode_test.py` | Host-side XY microstep mode harness (null-sink controller): `xy_microsteps` in the shared config sets `$100`/`$101` and the machine tick, a typed `$100` is overwritten on the spot, a value that is not a mode falls back to x8 with a warning, and `$110`/`$111` are held under a tick lowered with `GFSINK_RATE`. Runs in the grblHAL repo's CI. |
|
||||
| `raster_dry.py` | A dry raster at top speed at each XY microstep mode (runs on the board; the controller in GRBL mode, idle, no other Grbl client): per mode given (default 8, 16, 32) it stores `xy_microsteps` through forgectrl, waits for the restarted controller, streams 60 passes of 150 mm at F12000 with the laser off, and reports the peak feed, the controller CPU, the kernel counters against the start, `cnc/underruns` and any clamped-event line. Needs 150 mm of free +X and 12 mm of free +Y travel. Ends at x8. |
|
||||
| `xy_pattern_accel.py` | The XY microstep modes by the head accelerometer with the machine silent (runs on the board; the machine homed and at home, the lid closed, no other Grbl client): per mode it takes the cooling engine's quiet hold with the pump (`POST /cool/quiet?on=1&pump=1`: every fan, the pump and the TEC off), waits the fixed 10 s, samples the LIS2HH12 over `/dev/i2c-3` in one SMBus block transaction per reading at about 600 Hz, and runs the pattern from home at F12000: to (18, 9) in, to (9, 9) in, a 9 in circle from its mid-bottom back to (9, 9), to (9, 0), home. Reports the cruise-window RMS and peak-to-peak per leg and overall, the leg times, the kernel counters against home and `cnc/underruns`; JSON with the trace in the bench data directory. Ends at x8. |
|
||||
| `arc_tolerance_sweep.py` | How fine an arc the controller can plan (runs on the board; the machine homed and at home, the lid closed, no other Grbl client, the machine silent through the quiet hold): a `$12` ladder on the 9 in circle at F12000 at the mode given (`--mode`, default 16). Per rung: the chords and the chord boundaries a second, the circle time against the ideal, the lowest feed mid-circle and the fewest free planner blocks, the controller CPU, clamped events, underruns and the accelerometer's cruise RMS. `$12` goes back to what it was on every exit path; the head returns home. |
|
||||
| `gfbench.py` | Not a tool: the helper the board/host tools share - `HOST`/`LOCAL` from `GF_HOST`, `board(cmd)` (local `sh -c` or ssh), the factory coolant conversion `degc()`, `data_path()` (`FORGETEST_BENCH_DATA` or next to the tool), forgectrl's HTTP API with the panel token, `setting(key)` (from forgectrl, or from `/data/forgefirm.conf` on the board while forgectrl is stopped). |
|
||||
| `lens_travel.py` | Lens travel drill (runs on the board; stops the controller through forgectrl and restarts it; lens motion only): steps the lens over sysfs the way the focus card's lens home does and counts, per condition and round, the hall's hysteresis band (down off the rising edge until the hall leaves home, back up until it reads home), the rising edge's height above the bottom stop after a drive onto it, and the top stop's height above the edge (a drive up, down until the hall leaves home, back up; the difference). Conditions: the card's half-step drive onto the stop, a short drive, a long one, the factory's full-step home mode; `--ladder` drives a list of half-step descents below the leave-home point and counts each back (exact until the stop, short by an even number after a stall), `--current` picks the run current, the hold current, or hold down / run up, `--settle` and `--cadence` vary the timing; `--park bottom|top|edge-N|edge+N` holds the lens there for a depth-gauge reading with the controller in standby, `--park edge` returns it to the rising edge and restarts the controller. Stall drills on the bench reference machine only. The posture found is put back. |
|
||||
| `lens_stop_accel.py` | Lens stop detection by the head accelerometer (runs on the board; stops the controller through forgectrl and restarts it; lens motion only; stall drills on the bench reference machine only): reads the LIS2HH12 straight over `/dev/i2c-3` in six-byte bursts at 800 Hz (the iio path waits a sample period per read), steps the lens one half-step at a time from the hall's rising edge toward each stop and past it, and prints per step the peak-to-peak on each axis. A free step rings strongly on every second half-step; at a stop the ring dies, and a rotor slip a few steps later is a burst three times any free ring. `--find N` runs the contact rule N times per stop (a strong-parity step ringing under `--thresh`, or a burst over four times it, calls contact; the lens backs off `--back` and the count home proves no slip). `--save` keeps the per-step sample traces as JSON. |
|
||||
|
||||
@@ -0,0 +1,237 @@
|
||||
#!/usr/bin/env python3
|
||||
"""How fine an arc the one core can plan: a $12 ladder on the 9 in circle (on the board).
|
||||
|
||||
Usage: arc_tolerance_sweep.py [--mode M] [tolerances...]
|
||||
(default: --mode 16, the ladder 0.002 0.001 0.0005 0.00025 0.0001)
|
||||
|
||||
grblHAL traces an arc as chords whose sagitta is $12, so the chord length
|
||||
is about 2 sqrt(2 r $12) and every chord is a planner block: on the 9 in
|
||||
circle at F12000 the default 0.002 mm gives 1.35 mm chords and 148 block
|
||||
boundaries a second, which is the tone the circle plays at every
|
||||
microstep mode. A finer $12 moves the tone up and asks the single core
|
||||
for more blocks a second. This drill finds where that stops holding:
|
||||
with the machine silent (the engine's quiet hold, the fixed 10 s wait)
|
||||
and the head accelerometer listening, it moves from home to (9, 9) in,
|
||||
then for each $12 in the ladder runs the 9 in circle from its mid-bottom
|
||||
at F12000 and reports the chords, the boundaries a second, the circle
|
||||
time against the ideal, the lowest feed in the middle of the circle and
|
||||
the fewest free planner blocks (a starved planner shows as both), the
|
||||
controller CPU over the circle, any clamped-event line the driver logged,
|
||||
cnc/underruns, and the accelerometer's cruise RMS. $12 goes back to what
|
||||
it was, on every exit path, and the head returns home.
|
||||
|
||||
Needs the controller in GRBL mode, homed and standing at home, the lid
|
||||
closed, no other Grbl client, and the bed clear across the circle.
|
||||
"""
|
||||
import json
|
||||
import math
|
||||
import os
|
||||
import re
|
||||
import subprocess
|
||||
import sys
|
||||
import time
|
||||
|
||||
from gfbench import data_path
|
||||
from xy_pattern_accel import (Accel, Sampler, Grbl, fail, preflight, quiet, set_mode, stats,
|
||||
sysfs, window, FEED, IN, QUIET_S, TRIM_S)
|
||||
|
||||
LOG = "/data/log/forgefirm/grblhal/grblhal.log"
|
||||
RADIUS = 4.5 * IN
|
||||
CIRCLE = "G2 X0 Y0 I0 J%.3f F%d" % (-RADIUS, FEED)
|
||||
DEFAULT_LADDER = [0.002, 0.001, 0.0005, 0.00025, 0.0001]
|
||||
|
||||
|
||||
def parse_args(argv):
|
||||
mode = 16
|
||||
tols = []
|
||||
it = iter(argv)
|
||||
for a in it:
|
||||
if a == "--mode":
|
||||
mode = int(next(it))
|
||||
else:
|
||||
tols += [float(x) for x in a.split()]
|
||||
return mode, tols or DEFAULT_LADDER
|
||||
|
||||
|
||||
def controller_pid():
|
||||
out = subprocess.run(["pidof", "grblHAL_glowforge"], capture_output=True, text=True).stdout.split()
|
||||
return int(out[0]) if out else None
|
||||
|
||||
|
||||
def cpu_ticks(pid):
|
||||
with open("/proc/%d/stat" % pid) as f:
|
||||
s = f.read().split()
|
||||
return int(s[13]) + int(s[14])
|
||||
|
||||
|
||||
def grbl_setting(g, key):
|
||||
g.s.sendall((key + "\n").encode())
|
||||
g.s.settimeout(1.0)
|
||||
buf = ""
|
||||
end = time.time() + 1.5
|
||||
while time.time() < end and "ok" not in buf:
|
||||
try:
|
||||
buf += g.s.recv(4096).decode(errors="replace")
|
||||
except OSError:
|
||||
break
|
||||
m = re.search(r"^%s=(\S+)" % re.escape(key), buf, re.M)
|
||||
return m.group(1) if m else None
|
||||
|
||||
|
||||
def chords(tol):
|
||||
seg = math.floor(math.pi * RADIUS / math.sqrt(tol * (2 * RADIUS - tol)))
|
||||
return seg, 2 * math.pi * RADIUS / seg
|
||||
|
||||
|
||||
def run_circle(g, pid):
|
||||
"""The circle, timed by the state transitions and watched at about
|
||||
40 Hz for the feed and the planner's free blocks."""
|
||||
k0 = int(sysfs("cnc/underruns"))
|
||||
log_off = os.path.getsize(LOG) if os.path.exists(LOG) else 0
|
||||
c0 = cpu_ticks(pid)
|
||||
g.s.sendall((CIRCLE + "\n").encode())
|
||||
t_run = t_idle = None
|
||||
feeds, bfs = [], []
|
||||
end = time.time() + 120
|
||||
while time.time() < end:
|
||||
g.s.sendall(b"?")
|
||||
g.s.settimeout(1.0)
|
||||
buf = ""
|
||||
t_end = time.time() + 1.0
|
||||
while time.time() < t_end and ">" not in buf:
|
||||
try:
|
||||
buf += g.s.recv(4096).decode(errors="replace")
|
||||
except OSError:
|
||||
break
|
||||
if "error:" in buf or "ALARM" in buf:
|
||||
fail("the controller answered %r" % buf.strip()[-200:])
|
||||
m = re.search(r"<(\w+)[^>]*", buf)
|
||||
st = m.group(1) if m else None
|
||||
now = time.monotonic()
|
||||
f = re.search(r"FS:([\d.]+),", buf)
|
||||
b = re.search(r"Bf:(\d+),", buf)
|
||||
if st and st.startswith("Run"):
|
||||
if t_run is None:
|
||||
t_run = now
|
||||
if f:
|
||||
feeds.append((now, float(f.group(1))))
|
||||
if b:
|
||||
bfs.append((now, int(b.group(1))))
|
||||
elif st and st.startswith("Idle") and t_run is not None:
|
||||
t_idle = now
|
||||
break
|
||||
elif st and st.startswith("Alarm"):
|
||||
fail("alarm during the circle: %s" % st)
|
||||
time.sleep(0.02)
|
||||
if t_idle is None:
|
||||
fail("the circle did not finish")
|
||||
cpu = 100.0 * (cpu_ticks(pid) - c0) / (os.sysconf("SC_CLK_TCK") * (t_idle - t_run))
|
||||
g.drain()
|
||||
clamped = []
|
||||
if os.path.exists(LOG):
|
||||
with open(LOG, errors="replace") as fh:
|
||||
fh.seek(log_off)
|
||||
clamped = [l.strip() for l in fh if "late events clamped" in l or "underrun" in l]
|
||||
# the middle of the circle: the ramps at each end left out
|
||||
mid = [(t, v) for t, v in feeds if t_run + TRIM_S <= t <= t_idle - TRIM_S]
|
||||
mid_bf = [v for t, v in bfs if t_run + TRIM_S <= t <= t_idle - TRIM_S]
|
||||
return {"t_run": t_run, "t_idle": t_idle, "seconds": round(t_idle - t_run, 2), "cpu_pct": round(cpu, 1),
|
||||
"peak_feed": max((v for _, v in feeds), default=0), "min_mid_feed": min((v for _, v in mid), default=0),
|
||||
"min_free_blocks": min(mid_bf, default=None), "max_free_blocks": max(mid_bf, default=None),
|
||||
"status_samples": len(feeds), "clamped": clamped, "underruns": [k0, int(sysfs("cnc/underruns"))]}
|
||||
|
||||
|
||||
def main():
|
||||
mode, ladder = parse_args(sys.argv[1:])
|
||||
home = preflight()
|
||||
set_mode(mode)
|
||||
pid = controller_pid()
|
||||
g = Grbl()
|
||||
accel = sampler = None
|
||||
orig = None
|
||||
rows = []
|
||||
stamp = time.strftime("%Y%m%d%H%M%S")
|
||||
try:
|
||||
st, pos, _ = g.status()
|
||||
if not st or not st.startswith("Idle"):
|
||||
fail("controller is %s, not Idle" % st)
|
||||
if abs(pos[0] - home[0]) > 0.05 or abs(pos[1] - home[1]) > 0.05:
|
||||
fail("the head is not at home (MPos %s): run $H first" % (pos,))
|
||||
orig = grbl_setting(g, "$12")
|
||||
if orig is None:
|
||||
fail("could not read $12")
|
||||
print("x%d: step_freq %s; $12 now %s; the ladder %s" % (mode, sysfs("cnc/step_freq"), orig, ladder))
|
||||
print("quiet hold: %s" % quiet(True))
|
||||
time.sleep(QUIET_S)
|
||||
accel = Accel()
|
||||
accel.start()
|
||||
sampler = Sampler(accel)
|
||||
sampler.start()
|
||||
g.cmd("G21")
|
||||
g.cmd("G91")
|
||||
g.run_leg("G1 X%.3f Y%.3f F%d" % (9 * IN, 9 * IN, FEED))
|
||||
ideal = 2 * math.pi * RADIUS * 60.0 / FEED
|
||||
for tol in ladder:
|
||||
# plain decimals only: the controller reads no exponent, so a
|
||||
# "5e-05" would land as 5 mm
|
||||
r = g.cmd("$12=%.6f" % tol)
|
||||
got = grbl_setting(g, "$12")
|
||||
if "ok" not in r or got is None:
|
||||
fail("$12=%g refused: %r" % (tol, r.strip()))
|
||||
# $$ shows three decimals; the stored float keeps what was typed
|
||||
# (the core parses the value with strtof and rounds nothing).
|
||||
if abs(float(got) - tol) > 0.0006:
|
||||
fail("$12=%g reads back as %s" % (tol, got))
|
||||
seg, chord = chords(tol)
|
||||
time.sleep(1.5)
|
||||
res = run_circle(g, pid)
|
||||
xs, ys = window(sampler.samples, res["t_run"] + TRIM_S, res["t_idle"] - TRIM_S)
|
||||
res.update({"tol": tol, "reported": got, "chords": seg, "chord_mm": round(chord, 3),
|
||||
"boundaries_per_s": round(seg / ideal, 0), "accel_x": stats(xs), "accel_y": stats(ys)})
|
||||
rows.append(res)
|
||||
ok = (res["min_mid_feed"] >= 0.97 * FEED and not res["clamped"] and res["underruns"][0] == res["underruns"][1]
|
||||
and res["seconds"] <= ideal + 0.6)
|
||||
res["result"] = "PASS" if ok else "FAIL"
|
||||
print("$12=%-8g %5d chords of %.3f mm, %4.0f/s | circle %.2f s (ideal %.2f) | mid feed min %5.0f | "
|
||||
"planner free min %s max %s | CPU %5.1f %% | clamped %d | underruns %d->%d | accel x rms %7s y rms %7s | %s"
|
||||
% (tol, seg, chord, res["boundaries_per_s"], res["seconds"], ideal, res["min_mid_feed"],
|
||||
res["min_free_blocks"], res["max_free_blocks"], res["cpu_pct"], len(res["clamped"]),
|
||||
res["underruns"][0], res["underruns"][1], res["accel_x"]["rms"], res["accel_y"]["rms"],
|
||||
res["result"]))
|
||||
for l in res["clamped"][:3]:
|
||||
print(" ", l)
|
||||
time.sleep(2.0)
|
||||
finally:
|
||||
try:
|
||||
if orig is not None:
|
||||
g.cmd("$12=%.6f" % float(orig))
|
||||
print("$12 restored to %s" % grbl_setting(g, "$12"))
|
||||
except Exception as e:
|
||||
print("WARNING: could not restore $12: %s" % e)
|
||||
try:
|
||||
quiet(False)
|
||||
except SystemExit:
|
||||
print("WARNING: the quiet hold did not release; the engine drops it itself within 600 s")
|
||||
try:
|
||||
if sampler is not None:
|
||||
sampler.stop()
|
||||
finally:
|
||||
if accel is not None:
|
||||
accel.restore()
|
||||
try:
|
||||
st, pos, _ = g.status()
|
||||
if pos and (abs(pos[0] - home[0]) > 0.05 or abs(pos[1] - home[1]) > 0.05):
|
||||
g.run_leg("G1 X%.3f Y%.3f F%d" % (home[0] - pos[0], home[1] - pos[1], FEED))
|
||||
print("head back at home: MPos %s" % (g.status()[1],))
|
||||
g.cmd("G90")
|
||||
finally:
|
||||
g.close()
|
||||
path = data_path("arc_tolerance_sweep_x%d_%s.json" % (mode, stamp))
|
||||
with open(path, "w") as f:
|
||||
json.dump({"mode": mode, "feed": FEED, "radius_mm": RADIUS, "rows": rows}, f)
|
||||
print("record: %s" % path)
|
||||
return 0 if rows and all(r["result"] == "PASS" for r in rows) else 1
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main())
|
||||
@@ -113,6 +113,26 @@ Drills (pass a name):
|
||||
the reversal darkness across the two feeds. Requires
|
||||
laser_power_model = density.
|
||||
m4feeds [S] [F1] [F2] e.g. m4feeds 600 1000 4000
|
||||
xymode The XY microstep mode under the laser: one out-and-back
|
||||
20 mm line pair per feed (default 1200 and 6000 mm/min) at
|
||||
the same S under M4 density, passes offset +Y, one armed
|
||||
run, the block shifted +X by an offset so the runs at 8, 16
|
||||
and 32 sit side by side for the operator's eye. Reports the
|
||||
mode, tick and ramp the kernel holds, the discharge window
|
||||
and the HV current per pass: with the laser ticks scaled to
|
||||
the tick in force, the current at cruise should read alike
|
||||
at every mode, and the operator compares the marks. Set the
|
||||
mode from the panel (xy_microsteps) before each run.
|
||||
xymode [S] [F1] [F2] [X offset mm] e.g. xymode 400 1200 6000 25
|
||||
xycircle The XY microstep mode under the laser, the operator's eye
|
||||
version: a dark rapid +X by an offset, then one full circle
|
||||
of the given diameter starting at its top (the center below
|
||||
the start in +Y) at S under M4 density, laser off, and a
|
||||
rapid back to the origin. One armed run; the runs at 8, 16
|
||||
and 32 use different offsets so the circles sit side by
|
||||
side. Reports the mode, tick and ramp, the discharge window
|
||||
and the HV current.
|
||||
xycircle [S] [F] [X offset mm] [diameter mm] e.g. xycircle 400 2400 76.2 152.4
|
||||
dpatch Depth witness for the dose curve of the configured
|
||||
laser_power_model: two rows of small engraved patches
|
||||
(serpentine G1 fills) on the stock. Row A is CW (S1000) at
|
||||
@@ -175,6 +195,7 @@ The G-4 arm-refuses-when-a-fire-gate-is-active drill is operator-manual
|
||||
(kill the pump during the button wait); this harness prints the cue.
|
||||
"""
|
||||
import json
|
||||
import math
|
||||
import os
|
||||
import re
|
||||
import socket
|
||||
@@ -3018,6 +3039,248 @@ def drill_m4feeds(g):
|
||||
return 0 if ok else 1
|
||||
|
||||
|
||||
XYM_MM = 20.0
|
||||
XYM_LEG_GAP = 0.6
|
||||
XYM_ROW_GAP = 5.0
|
||||
|
||||
|
||||
def drill_xymode(g):
|
||||
sval = int(sys.argv[2]) if len(sys.argv) > 2 else 400
|
||||
f1 = int(sys.argv[3]) if len(sys.argv) > 3 else 1200
|
||||
f2 = int(sys.argv[4]) if len(sys.argv) > 4 else 6000
|
||||
xoff = float(sys.argv[5]) if len(sys.argv) > 5 else 0.0
|
||||
if not 50 <= sval <= 1000 or not 300 <= f1 < f2 <= 12000 or not 0 <= xoff <= 300:
|
||||
print('usage: xymode [S 50..1000] [F1] [F2] [X offset 0..300] (300 <= F1 < F2 <= 12000)')
|
||||
return 2
|
||||
sampler = Sampler(PCURVE_SAMPLE_HZ)
|
||||
if not sampler.local:
|
||||
print('run this on the board: the witnesses are sysfs at 25 Hz')
|
||||
return 2
|
||||
model = conf_get('laser_power_model') or 'density'
|
||||
if model != 'density':
|
||||
print('PRECONDITION FAILED: laser_power_model is %s' % model)
|
||||
return 2
|
||||
|
||||
def sysfs(attr):
|
||||
with open(SYSFS + '/' + attr) as f:
|
||||
return f.read().strip()
|
||||
|
||||
fails = []
|
||||
|
||||
def check(cond, msg):
|
||||
print(' %s: %s' % ('ok' if cond else 'FAIL', msg))
|
||||
if not cond:
|
||||
fails.append(msg)
|
||||
|
||||
mode, tick, ramp = sysfs('cnc/x_mode'), sysfs('cnc/step_freq'), sysfs('cnc/ramp_rate')
|
||||
print('=== XY microstep mode under the laser: x%s, %s Hz tick, %s Hz/s ramp, S%d under M4 density at F%d and F%d ==='
|
||||
% (mode, tick, ramp, sval, f1, f2))
|
||||
check(sysfs('cnc/y_mode') == mode, 'both axes at x%s' % mode)
|
||||
print('connect: %s' % prepare(g))
|
||||
print('pre-fire: %s' % sample_forgectrl())
|
||||
arm_cue()
|
||||
print('>>> The block: %g mm along +X by %g mm along +Y, starting %g mm in +X from the head.'
|
||||
% (XYM_MM + 2, XYM_ROW_GAP + 4, xoff))
|
||||
print('>>> Pass 1 (F%d) first, pass 2 (F%d) %g mm past it in +Y; each pass an out leg'
|
||||
% (f1, f2, XYM_ROW_GAP))
|
||||
print('>>> and a return leg %g mm apart. The head comes back to where it started.\n' % XYM_LEG_GAP)
|
||||
sampler.start()
|
||||
aborted = False
|
||||
try:
|
||||
for ln in ('G91', 'G21'):
|
||||
g.cmd(ln)
|
||||
if xoff:
|
||||
g.cmd('G0 X%g' % xoff)
|
||||
g.wait_state('Idle', 30)
|
||||
for i, feed in enumerate((f1, f2)):
|
||||
g.s.sendall(b'M4 S%d\n' % sval)
|
||||
g.s.sendall(('G1 X%g F%d\n' % (XYM_MM, feed)).encode())
|
||||
g.s.sendall(('G1 Y%g F%d\n' % (XYM_LEG_GAP, feed)).encode())
|
||||
g.s.sendall(('G1 X%g F%d\n' % (-XYM_MM, feed)).encode())
|
||||
g.s.sendall(b'M5\n')
|
||||
g.s.sendall(('G0 Y%g\n' % -XYM_LEG_GAP).encode())
|
||||
st = g.wait_state('Run', 300 if i == 0 else 60)
|
||||
if not st.startswith('Run'):
|
||||
print('FAIL: pass %d never ran (state=%s)' % (i + 1, st))
|
||||
g.rt(b'\x18')
|
||||
aborted = True
|
||||
return 1
|
||||
g.wait_state('Idle', 120)
|
||||
time.sleep(0.5)
|
||||
g.drain()
|
||||
print(' pass at F%d ran' % feed)
|
||||
if i == 0:
|
||||
g.s.sendall(('G0 Y%g\n' % XYM_ROW_GAP).encode())
|
||||
g.wait_state('Idle', 30)
|
||||
g.s.sendall(('G0 Y%g\n' % -XYM_ROW_GAP).encode())
|
||||
g.wait_state('Idle', 30)
|
||||
if xoff:
|
||||
g.cmd('G0 X%g' % -xoff)
|
||||
g.wait_state('Idle', 30)
|
||||
g.cmd('G90')
|
||||
g.cmd('M2')
|
||||
t0 = time.time()
|
||||
while time.time() - t0 < 90:
|
||||
smp = sample_forgectrl()
|
||||
if smp and not smp['armed']:
|
||||
break
|
||||
time.sleep(0.2)
|
||||
time.sleep(1.5)
|
||||
except Exception as e:
|
||||
aborted = True
|
||||
print('ABORTED: %s' % e)
|
||||
g.rt(b'\x18')
|
||||
finally:
|
||||
sampler.stop()
|
||||
if aborted:
|
||||
return 1
|
||||
tr = sampler.samples
|
||||
segs, cur = [], None
|
||||
for smp in tr:
|
||||
on = smp['hv'] is not None and smp['hv'] > HV_DARK_MAX
|
||||
if on and cur is None:
|
||||
cur = [smp['t'], smp['t']]
|
||||
elif on:
|
||||
cur[1] = smp['t']
|
||||
elif cur is not None and smp['t'] - cur[1] > 1.5:
|
||||
segs.append(cur)
|
||||
cur = None
|
||||
if cur:
|
||||
segs.append(cur)
|
||||
print('\n--- results x%s (%d samples, %.1f Hz) ---' % (mode, len(tr), sampler.rate()))
|
||||
check(len(segs) == 2, '%d discharge window(s), expected 2 (one per feed)' % len(segs))
|
||||
for feed, (a, b) in zip((f1, f2), segs):
|
||||
hv = _stats(_window(tr, a + 0.2, b - 0.1, 'hv'))
|
||||
tp = _stats(_window(tr, a + 0.2, b - 0.1, 'tp'))
|
||||
print(' F%d pass: %.1f s lit, hv mean %.0f, tp mean %.0f' % (feed, b - a, hv['mean'], tp['mean'] or 0))
|
||||
if segs:
|
||||
t_end = segs[-1][1]
|
||||
hv_after = max((smp['hv'] for smp in tr if smp['t'] > t_end + 0.5 and smp['hv'] is not None), default=0)
|
||||
check(hv_after <= HV_DARK_MAX, 'dark after the last M5 (hv max %d)' % hv_after)
|
||||
check(sysfs('cnc/underruns') == '0', 'no underrun (cnc/underruns %s)' % sysfs('cnc/underruns'))
|
||||
print('\n--- the operator reads the material ---')
|
||||
print('Two line pairs, F%d nearest the start, F%d %g mm past it in +Y, this block %g mm in +X.'
|
||||
% (f1, f2, XYM_ROW_GAP, xoff))
|
||||
print(' Compare this block against the other modes: evenness along each line, the')
|
||||
print(' darkness at cruise, the reversal at the far end. The current per pass above')
|
||||
print(' should read alike across the modes: the laser ticks are scaled to the tick.')
|
||||
ok = not fails
|
||||
print('XYMODE x%s %s' % (mode, 'instrument checks PASS - the material verdict is yours'
|
||||
if ok else 'FAIL: %d check(s) failed' % len(fails)))
|
||||
return 0 if ok else 1
|
||||
|
||||
|
||||
def drill_xycircle(g):
|
||||
sval = int(sys.argv[2]) if len(sys.argv) > 2 else 400
|
||||
feed = int(sys.argv[3]) if len(sys.argv) > 3 else 2400
|
||||
xoff = float(sys.argv[4]) if len(sys.argv) > 4 else 76.2
|
||||
dia = float(sys.argv[5]) if len(sys.argv) > 5 else 152.4
|
||||
if not 50 <= sval <= 1000 or not 300 <= feed <= 12000 or not 0 <= xoff <= 300 or not 10 <= dia <= 250:
|
||||
print('usage: xycircle [S 50..1000] [F 300..12000] [X offset 0..300] [diameter 10..250]')
|
||||
return 2
|
||||
sampler = Sampler(PCURVE_SAMPLE_HZ)
|
||||
if not sampler.local:
|
||||
print('run this on the board: the witnesses are sysfs at 25 Hz')
|
||||
return 2
|
||||
model = conf_get('laser_power_model') or 'density'
|
||||
if model != 'density':
|
||||
print('PRECONDITION FAILED: laser_power_model is %s' % model)
|
||||
return 2
|
||||
|
||||
def sysfs(attr):
|
||||
with open(SYSFS + '/' + attr) as f:
|
||||
return f.read().strip()
|
||||
|
||||
fails = []
|
||||
|
||||
def check(cond, msg):
|
||||
print(' %s: %s' % ('ok' if cond else 'FAIL', msg))
|
||||
if not cond:
|
||||
fails.append(msg)
|
||||
|
||||
r = dia / 2.0
|
||||
mode, tick, ramp = sysfs('cnc/x_mode'), sysfs('cnc/step_freq'), sysfs('cnc/ramp_rate')
|
||||
print('=== XY microstep mode under the laser: x%s, %s Hz tick, %s Hz/s ramp, a %g mm circle at S%d under M4 density at F%d ==='
|
||||
% (mode, tick, ramp, dia, sval, feed))
|
||||
check(sysfs('cnc/y_mode') == mode, 'both axes at x%s' % mode)
|
||||
print('connect: %s' % prepare(g))
|
||||
print('pre-fire: %s' % sample_forgectrl())
|
||||
arm_cue()
|
||||
print('>>> A dark rapid %g mm in +X, then the circle from its top: %g mm across, its' % (xoff, dia))
|
||||
print('>>> center %g mm in +Y from the start, so it spans %g mm each side of the start' % (r, r))
|
||||
print('>>> and %g mm in +Y. Laser off, then a rapid back to the origin.\n' % dia)
|
||||
sampler.start()
|
||||
aborted = False
|
||||
try:
|
||||
for ln in ('G91', 'G21'):
|
||||
g.cmd(ln)
|
||||
if xoff:
|
||||
g.cmd('G0 X%g' % xoff)
|
||||
g.wait_state('Idle', 60)
|
||||
g.s.sendall(b'M4 S%d\n' % sval)
|
||||
g.s.sendall(('G2 X0 Y0 I0 J%g F%d\n' % (r, feed)).encode())
|
||||
g.s.sendall(b'M5\n')
|
||||
st = g.wait_state('Run', 300)
|
||||
if not st.startswith('Run'):
|
||||
print('FAIL: the circle never ran (state=%s)' % st)
|
||||
g.rt(b'\x18')
|
||||
aborted = True
|
||||
return 1
|
||||
g.wait_state('Idle', 300)
|
||||
time.sleep(0.5)
|
||||
g.drain()
|
||||
print(' the circle ran')
|
||||
if xoff:
|
||||
g.cmd('G0 X%g' % -xoff)
|
||||
g.wait_state('Idle', 60)
|
||||
g.cmd('G90')
|
||||
g.cmd('M2')
|
||||
t0 = time.time()
|
||||
while time.time() - t0 < 90:
|
||||
smp = sample_forgectrl()
|
||||
if smp and not smp['armed']:
|
||||
break
|
||||
time.sleep(0.2)
|
||||
time.sleep(1.5)
|
||||
except Exception as e:
|
||||
aborted = True
|
||||
print('ABORTED: %s' % e)
|
||||
g.rt(b'\x18')
|
||||
finally:
|
||||
sampler.stop()
|
||||
if aborted:
|
||||
return 1
|
||||
tr = sampler.samples
|
||||
segs, cur = [], None
|
||||
for smp in tr:
|
||||
on = smp['hv'] is not None and smp['hv'] > HV_DARK_MAX
|
||||
if on and cur is None:
|
||||
cur = [smp['t'], smp['t']]
|
||||
elif on:
|
||||
cur[1] = smp['t']
|
||||
elif cur is not None and smp['t'] - cur[1] > 1.5:
|
||||
segs.append(cur)
|
||||
cur = None
|
||||
if cur:
|
||||
segs.append(cur)
|
||||
print('\n--- results x%s (%d samples, %.1f Hz) ---' % (mode, len(tr), sampler.rate()))
|
||||
check(len(segs) == 1, '%d discharge window(s), expected 1' % len(segs))
|
||||
for (a, b) in segs:
|
||||
hv = _stats(_window(tr, a + 0.2, b - 0.1, 'hv'))
|
||||
tp = _stats(_window(tr, a + 0.2, b - 0.1, 'tp'))
|
||||
print(' circle: %.1f s lit (expected about %.1f s at F%d), hv mean %.0f, tp mean %.0f'
|
||||
% (b - a, math.pi * dia * 60.0 / feed, feed, hv['mean'], tp['mean'] or 0))
|
||||
if segs:
|
||||
t_end = segs[-1][1]
|
||||
hv_after = max((smp['hv'] for smp in tr if smp['t'] > t_end + 0.5 and smp['hv'] is not None), default=0)
|
||||
check(hv_after <= HV_DARK_MAX, 'dark after M5 (hv max %d)' % hv_after)
|
||||
check(sysfs('cnc/underruns') == '0', 'no underrun (cnc/underruns %s)' % sysfs('cnc/underruns'))
|
||||
ok = not fails
|
||||
print('XYCIRCLE x%s %s' % (mode, 'instrument checks PASS - the material verdict is yours'
|
||||
if ok else 'FAIL: %d check(s) failed' % len(fails)))
|
||||
return 0 if ok else 1
|
||||
|
||||
|
||||
def post_ctrl(action):
|
||||
# http.client preserves the header-name case exactly as given.
|
||||
import http.client
|
||||
@@ -3230,6 +3493,7 @@ def main():
|
||||
'pcurve': drill_pcurve, 'm5dark': drill_m5dark,
|
||||
'dpatch': drill_dpatch, 'flowload': drill_flowload,
|
||||
'm4corner': drill_m4corner, 'm4feeds': drill_m4feeds,
|
||||
'xymode': drill_xymode, 'xycircle': drill_xycircle,
|
||||
'senderchg': drill_senderchg, 'overrun': drill_overrun,
|
||||
'holdres': drill_holdres,
|
||||
'expstop': drill_expstop, 'ctrlstart': drill_ctrlstart}
|
||||
|
||||
@@ -0,0 +1,226 @@
|
||||
#!/usr/bin/env python3
|
||||
"""A dry raster at top speed at each XY microstep mode (on the board).
|
||||
|
||||
Usage: raster_dry.py [modes...] (default: 8 16 32)
|
||||
|
||||
Per mode: stores xy_microsteps through forgectrl (which restarts the
|
||||
idle GRBL controller), waits for the controller to come back at that
|
||||
mode, then streams a raster with the laser off (M5): LINES passes of
|
||||
LEN mm along X at F12000, Y stepped by PITCH between passes, and back to
|
||||
the start. Reports the peak feed, the controller CPU over the job, the
|
||||
kernel counters against the start (the head must come back within TOL mm
|
||||
on both axes), Grbl's own drift, cnc/underruns, and any "late events
|
||||
clamped" line the driver logged. Ends with the key cleared (x8).
|
||||
|
||||
Needs the controller in GRBL mode, idle, no other Grbl client, the bed
|
||||
clear, and LEN mm of free +X travel plus LINES x PITCH mm of free +Y
|
||||
travel from where the head stands.
|
||||
"""
|
||||
import json
|
||||
import os
|
||||
import re
|
||||
import socket
|
||||
import subprocess
|
||||
import sys
|
||||
import time
|
||||
import urllib.request
|
||||
|
||||
FC = "http://127.0.0.1"
|
||||
FC_TOKEN = "/data/forgefirm/panel.token"
|
||||
MODES = [int(a) for arg in sys.argv[1:] for a in arg.split()] or [8, 16, 32]
|
||||
LINES, LEN, PITCH, FEED, TOL = 60, 150.0, 0.2, 12000, 0.05
|
||||
LOG = "/data/log/forgefirm/grblhal/grblhal.log"
|
||||
|
||||
|
||||
def fc(path, data=None):
|
||||
token = open(FC_TOKEN).read().strip()
|
||||
body = None
|
||||
if data is not None:
|
||||
body = "&".join("%s=%s" % (k, v) for k, v in data.items()).encode()
|
||||
req = urllib.request.Request(FC + path, data=body,
|
||||
headers={"Authorization": "Bearer " + token, "X-ForgeFIRM-Token": token,
|
||||
"Content-Type": "application/x-www-form-urlencoded"})
|
||||
try:
|
||||
return json.load(urllib.request.urlopen(req, timeout=30))
|
||||
except urllib.error.HTTPError as e:
|
||||
return {"http": e.code, "body": e.read().decode()}
|
||||
|
||||
|
||||
def sysfs(attr):
|
||||
with open("/sys/glowforge/" + attr) as f:
|
||||
return f.read().strip()
|
||||
|
||||
|
||||
def kernel_mm():
|
||||
p = fc("/status").get("pos") or {}
|
||||
return float(p.get("x", 0)), float(p.get("y", 0))
|
||||
|
||||
|
||||
def controller_pid():
|
||||
out = subprocess.run(["pidof", "grblHAL_glowforge"], capture_output=True, text=True).stdout.split()
|
||||
return int(out[0]) if out else None
|
||||
|
||||
|
||||
def cpu_ticks(pid):
|
||||
with open("/proc/%d/stat" % pid) as f:
|
||||
s = f.read().split()
|
||||
return int(s[13]) + int(s[14])
|
||||
|
||||
|
||||
def set_mode(mode):
|
||||
r = fc("/settings?xy_microsteps=", {}) if mode == 8 else fc("/settings", {"xy_microsteps": str(mode)})
|
||||
if "http" in r:
|
||||
raise SystemExit("settings write refused: %s" % r)
|
||||
time.sleep(3)
|
||||
t0 = time.time()
|
||||
while time.time() - t0 < 180:
|
||||
m = fc("/mode")
|
||||
if m.get("controller") == "running" and m.get("motion") == "verified" and sysfs("cnc/x_mode") == str(mode):
|
||||
time.sleep(2)
|
||||
print("x%d: controller pid %s, step_freq %s, ramp_rate %s"
|
||||
% (mode, m.get("pid"), sysfs("cnc/step_freq"), sysfs("cnc/ramp_rate")))
|
||||
return
|
||||
time.sleep(2)
|
||||
raise SystemExit("x%d: the controller did not come back" % mode)
|
||||
|
||||
|
||||
class Grbl:
|
||||
def __init__(self):
|
||||
self.s = socket.create_connection(("127.0.0.1", 23), timeout=5)
|
||||
time.sleep(0.3)
|
||||
self.drain()
|
||||
|
||||
def drain(self):
|
||||
self.s.settimeout(0.2)
|
||||
out = b""
|
||||
try:
|
||||
while True:
|
||||
d = self.s.recv(4096)
|
||||
if not d:
|
||||
break
|
||||
out += d
|
||||
except (socket.timeout, OSError):
|
||||
pass
|
||||
return out.decode(errors="replace")
|
||||
|
||||
def cmd(self, line, timeout=30):
|
||||
self.s.sendall((line + "\n").encode())
|
||||
self.s.settimeout(timeout)
|
||||
buf = ""
|
||||
end = time.time() + timeout
|
||||
while time.time() < end:
|
||||
try:
|
||||
buf += self.s.recv(4096).decode(errors="replace")
|
||||
except socket.timeout:
|
||||
break
|
||||
if "ok\r\n" in buf or "error" in buf or "ALARM" in buf:
|
||||
break
|
||||
return buf
|
||||
|
||||
def status(self):
|
||||
self.s.sendall(b"?")
|
||||
self.s.settimeout(1.0)
|
||||
buf = ""
|
||||
end = time.time() + 1.0
|
||||
while time.time() < end and ">" not in buf:
|
||||
try:
|
||||
buf += self.s.recv(4096).decode(errors="replace")
|
||||
except socket.timeout:
|
||||
break
|
||||
m = re.search(r"<(\w+)[^>]*MPos:([-\d.]+),([-\d.]+),([-\d.]+)", buf)
|
||||
f = re.search(r"FS:([\d.]+),", buf)
|
||||
if not m:
|
||||
return None, None, 0.0
|
||||
return m.group(1), (float(m.group(2)), float(m.group(3))), float(f.group(1)) if f else 0.0
|
||||
|
||||
def wait_idle(self, timeout):
|
||||
peak = 0.0
|
||||
end = time.time() + timeout
|
||||
seen_run = False
|
||||
while time.time() < end:
|
||||
st, pos, f = self.status()
|
||||
peak = max(peak, f)
|
||||
if st and st.startswith(("Run", "Jog")):
|
||||
seen_run = True
|
||||
if st and st.startswith("Idle") and seen_run:
|
||||
return True, peak, pos
|
||||
if st and st.startswith("Alarm"):
|
||||
return False, peak, pos
|
||||
time.sleep(0.1)
|
||||
return False, peak, None
|
||||
|
||||
def close(self):
|
||||
self.s.close()
|
||||
|
||||
|
||||
def raster(mode):
|
||||
pid = controller_pid()
|
||||
k0 = kernel_mm()
|
||||
und0 = int(sysfs("cnc/underruns"))
|
||||
log_off = os.path.getsize(LOG) if os.path.exists(LOG) else 0
|
||||
g = Grbl()
|
||||
peak = 0.0
|
||||
try:
|
||||
st, pos0, _ = g.status()
|
||||
if not st or not st.startswith("Idle"):
|
||||
print("x%d: controller is %s, not Idle" % (mode, st))
|
||||
return False
|
||||
for line in ("M5", "G21", "G90", "G91"):
|
||||
g.cmd(line)
|
||||
c0 = cpu_ticks(pid)
|
||||
t0 = time.time()
|
||||
for i in range(LINES):
|
||||
dx = LEN if i % 2 == 0 else -LEN
|
||||
r = g.cmd("G1 X%.3f F%d" % (dx, FEED), timeout=60)
|
||||
if "error" in r or "ALARM" in r:
|
||||
print("x%d: line %d refused: %r" % (mode, i, r.strip()))
|
||||
break
|
||||
g.cmd("G1 Y%.3f F%d" % (PITCH, FEED), timeout=60)
|
||||
peak = max(peak, g.status()[2])
|
||||
g.cmd("G1 Y%.3f F%d" % (-LINES * PITCH, FEED), timeout=60)
|
||||
ok, peak_end, pos1 = g.wait_idle(120)
|
||||
peak = max(peak, peak_end)
|
||||
g.cmd("G90")
|
||||
elapsed = time.time() - t0
|
||||
cpu = 100.0 * (cpu_ticks(pid) - c0) / (os.sysconf("SC_CLK_TCK") * elapsed)
|
||||
finally:
|
||||
g.close()
|
||||
# the machine itself idle: the kernel plays out the depth behind Idle
|
||||
end = time.time() + 20
|
||||
while time.time() < end and fc("/status").get("state") != "idle":
|
||||
time.sleep(0.5)
|
||||
k1 = kernel_mm()
|
||||
und1 = int(sysfs("cnc/underruns"))
|
||||
clamped = []
|
||||
if os.path.exists(LOG):
|
||||
with open(LOG, errors="replace") as f:
|
||||
f.seek(log_off)
|
||||
clamped = [l.strip() for l in f if "late events clamped" in l or "underrun" in l]
|
||||
drift = (pos1[0] - pos0[0], pos1[1] - pos0[1]) if (pos1 and pos0) else None
|
||||
kd = (k1[0] - k0[0], k1[1] - k0[1])
|
||||
verdict = ok and abs(kd[0]) <= TOL and abs(kd[1]) <= TOL and und1 == und0 and not clamped
|
||||
print("x%d: %s | %d passes of %.0f mm at F%d in %.1f s, peak %.0f mm/min, controller CPU %.1f %% | "
|
||||
"kernel return dx=%.3f dy=%.3f mm | grbl drift %s | underruns %d->%d | clamped lines %d"
|
||||
% (mode, "PASS" if verdict else "FAIL", LINES, LEN, FEED, elapsed, peak, cpu, kd[0], kd[1],
|
||||
drift, und0, und1, len(clamped)))
|
||||
for l in clamped[:5]:
|
||||
print(" ", l)
|
||||
return verdict
|
||||
|
||||
|
||||
def main():
|
||||
verdicts = {}
|
||||
for mode in MODES:
|
||||
print("=== x%d ===" % mode)
|
||||
set_mode(mode)
|
||||
verdicts[mode] = raster(mode)
|
||||
if MODES[-1] != 8:
|
||||
print("=== restore x8 ===")
|
||||
set_mode(8)
|
||||
print("settings xy_microsteps:", repr(fc("/settings").get("xy_microsteps")))
|
||||
print("RESULT:", " ".join("x%d=%s" % (m, "PASS" if v else "FAIL") for m, v in verdicts.items()))
|
||||
return 0 if all(verdicts.values()) else 1
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main())
|
||||
@@ -0,0 +1,227 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Host-side verification of the XY microstep mode wiring.
|
||||
|
||||
The XY microstep mode is one number in the shared config, xy_microsteps
|
||||
(8, 16 or 32), and the driver derives three things from it at start and
|
||||
never takes them typed: $100/$101, the machine tick, and the kernel stop
|
||||
ramp. This harness drives the native grblHAL_glowforge binary in
|
||||
null-sink mode (no hardware, no root) over TCP, one process per config:
|
||||
|
||||
1. no key: x8, $100/$101 = 53.333, the 28160 Hz tick
|
||||
2. xy_microsteps = 16: 106.667, the 56320 Hz tick; a typed $100 is
|
||||
overwritten on the spot; $110 is left alone under a tick that
|
||||
carries it
|
||||
3. xy_microsteps = 32: 213.333, the 112640 Hz tick
|
||||
4. a value that is not a mode: x8 with a warning in the log
|
||||
5. GFSINK_RATE lowered under the mode's tick: $110/$111 are held at
|
||||
the feed the tick carries, one step per tick per axis
|
||||
|
||||
The binary keeps its settings in EEPROM.DAT in the working directory, so
|
||||
each run starts from defaults in a temporary directory and leaves
|
||||
nothing behind.
|
||||
|
||||
Usage: xy_mode_test.py [path-to-binary]
|
||||
(default ./build-native/grblHAL_glowforge)
|
||||
"""
|
||||
import os
|
||||
import re
|
||||
import shutil
|
||||
import socket
|
||||
import subprocess
|
||||
import sys
|
||||
import tempfile
|
||||
import time
|
||||
|
||||
BIN = os.path.abspath(sys.argv[1] if len(sys.argv) > 1 else "build-native/grblHAL_glowforge")
|
||||
PORT = 2401
|
||||
|
||||
|
||||
def fail(msg):
|
||||
print("FAIL: %s" % msg)
|
||||
sys.exit(1)
|
||||
|
||||
|
||||
class Session:
|
||||
"""One null-sink controller process with its own settings store and
|
||||
its own shared config file."""
|
||||
|
||||
def __init__(self, conf=None, env_extra=None):
|
||||
self.workdir = tempfile.mkdtemp(prefix="xy-mode-")
|
||||
env = dict(os.environ, GF_STATE_DIR=self.workdir, FFLOG_STDERR="1",
|
||||
FFLOG_LEVEL="info")
|
||||
for key in ("GFSINK", "GFSINK_RATE", "GF_SWITCH_FILE", "GF_VERDICT_FILE"):
|
||||
env.pop(key, None)
|
||||
conf_path = os.path.join(self.workdir, "forgefirm.conf")
|
||||
with open(conf_path, "w") as f:
|
||||
f.write(conf or "")
|
||||
env["GFHOME_CONF"] = conf_path
|
||||
env.update(env_extra or {})
|
||||
self.proc = subprocess.Popen([BIN, "-p", str(PORT)],
|
||||
cwd=self.workdir, env=env,
|
||||
stdout=subprocess.DEVNULL,
|
||||
stderr=subprocess.PIPE)
|
||||
self.sock = self.connect()
|
||||
self.read(1.0) # banner and any boot messages
|
||||
|
||||
def connect(self):
|
||||
for _ in range(50):
|
||||
try:
|
||||
return socket.create_connection(("127.0.0.1", PORT), timeout=1)
|
||||
except OSError:
|
||||
time.sleep(0.1)
|
||||
err = b""
|
||||
if self.proc.poll() is not None:
|
||||
err = self.proc.stderr.read() or b""
|
||||
fail("cannot connect to the controller (exit=%s)\n%s"
|
||||
% (self.proc.poll(), err.decode(errors="replace")))
|
||||
|
||||
def read(self, timeout=0.8):
|
||||
out = b""
|
||||
end = time.time() + timeout
|
||||
while time.time() < end:
|
||||
self.sock.settimeout(max(0.05, end - time.time()))
|
||||
try:
|
||||
data = self.sock.recv(4096)
|
||||
except (socket.timeout, OSError):
|
||||
break
|
||||
if not data:
|
||||
break
|
||||
out += data
|
||||
return out.decode(errors="replace")
|
||||
|
||||
def send(self, line, timeout=1.0):
|
||||
self.sock.sendall((line + "\n").encode())
|
||||
return self.read(timeout)
|
||||
|
||||
def setting(self, key):
|
||||
reply = self.send(key)
|
||||
m = re.search(r"^%s=(\S+)" % re.escape(key), reply, re.M)
|
||||
if not m:
|
||||
fail("no value reported for %s: %r" % (key, reply))
|
||||
return float(m.group(1))
|
||||
|
||||
def write_setting(self, assignment):
|
||||
reply = self.send(assignment)
|
||||
if "ok" not in reply:
|
||||
fail("settings write %r refused: %r" % (assignment, reply))
|
||||
|
||||
def close(self):
|
||||
"""Stop the process and return everything it logged."""
|
||||
try:
|
||||
self.sock.close()
|
||||
except OSError:
|
||||
pass
|
||||
self.proc.terminate()
|
||||
try:
|
||||
_out, err = self.proc.communicate(timeout=5)
|
||||
except subprocess.TimeoutExpired:
|
||||
self.proc.kill()
|
||||
_out, err = self.proc.communicate()
|
||||
shutil.rmtree(self.workdir, ignore_errors=True)
|
||||
return (err or b"").decode(errors="replace")
|
||||
|
||||
|
||||
def check(ok, ok_msg, fail_msg):
|
||||
if not ok:
|
||||
fail(fail_msg)
|
||||
print("ok %s" % ok_msg)
|
||||
|
||||
|
||||
def near(got, want, tol=0.0005):
|
||||
return abs(got - want) <= tol
|
||||
|
||||
|
||||
def expect_scale(s, mode, spm):
|
||||
for key in ("$100", "$101"):
|
||||
got = s.setting(key)
|
||||
check(near(got, spm), "%s=%.3f at x%d" % (key, got, mode),
|
||||
"%s is %.3f at x%d, want %.3f" % (key, got, mode, spm))
|
||||
|
||||
|
||||
def expect_log(log, needle, what):
|
||||
check(needle in log, what, "the log does not say %r:\n%s" % (needle, log))
|
||||
|
||||
|
||||
def case_default():
|
||||
s = Session()
|
||||
try:
|
||||
expect_scale(s, 8, 53.333)
|
||||
rate = s.setting("$110")
|
||||
check(near(rate, 12000.0), "$110=%.0f stands at x8" % rate,
|
||||
"$110 is %.3f at x8, want 12000" % rate)
|
||||
finally:
|
||||
log = s.close()
|
||||
expect_log(log, "x8 microsteps, 28160 Hz machine tick", "no key: x8, the 28160 Hz tick")
|
||||
|
||||
|
||||
def case_x16():
|
||||
s = Session("xy_microsteps = 16\n")
|
||||
try:
|
||||
expect_scale(s, 16, 106.667)
|
||||
# A typed $100 is accepted and overwritten on the spot: the scale
|
||||
# is the mode's, never the sender's.
|
||||
s.write_setting("$100=53.333")
|
||||
got = s.setting("$100")
|
||||
check(near(got, 106.667), "a typed $100 is back at %.3f" % got,
|
||||
"a typed $100 stuck at %.3f under x16" % got)
|
||||
rate = s.setting("$110")
|
||||
check(near(rate, 12000.0), "$110=%.0f stands under a tick that carries it" % rate,
|
||||
"$110 is %.3f under x16, want 12000" % rate)
|
||||
finally:
|
||||
log = s.close()
|
||||
expect_log(log, "x16 microsteps, 56320 Hz machine tick", "x16: the 56320 Hz tick")
|
||||
|
||||
|
||||
def case_x32():
|
||||
s = Session("xy_microsteps = 32\n")
|
||||
try:
|
||||
expect_scale(s, 32, 213.333)
|
||||
finally:
|
||||
log = s.close()
|
||||
expect_log(log, "x32 microsteps, 112640 Hz machine tick", "x32: the 112640 Hz tick")
|
||||
|
||||
|
||||
def case_invalid():
|
||||
s = Session("xy_microsteps = 24\n")
|
||||
try:
|
||||
expect_scale(s, 8, 53.333)
|
||||
finally:
|
||||
log = s.close()
|
||||
expect_log(log, "xy_microsteps '24' is not 8, 16 or 32", "a value that is not a mode is refused with a warning")
|
||||
expect_log(log, "x8 microsteps, 28160 Hz machine tick", "and the machine runs at x8")
|
||||
|
||||
|
||||
def case_low_tick():
|
||||
# 1000 Hz carries 1000 steps/s per axis: 562.5 mm/min at 106.667
|
||||
# steps/mm. $110/$111 are held there, and a typed value above it
|
||||
# comes back held.
|
||||
s = Session("xy_microsteps = 16\n", {"GFSINK_RATE": "1000"})
|
||||
try:
|
||||
expect_scale(s, 16, 106.667)
|
||||
for key in ("$110", "$111"):
|
||||
got = s.setting(key)
|
||||
check(near(got, 562.5, 0.05), "%s held at %.3f under a 1000 Hz tick" % (key, got),
|
||||
"%s is %.3f under a 1000 Hz tick, want 562.5" % (key, got))
|
||||
s.write_setting("$110=6000")
|
||||
got = s.setting("$110")
|
||||
check(near(got, 562.5, 0.05), "a typed $110=6000 comes back held at %.3f" % got,
|
||||
"a typed $110=6000 stuck at %.3f under a 1000 Hz tick" % got)
|
||||
finally:
|
||||
log = s.close()
|
||||
expect_log(log, "1000 Hz machine tick", "GFSINK_RATE took the tick to 1000 Hz")
|
||||
expect_log(log, "held at the ceiling", "the hold is logged")
|
||||
|
||||
|
||||
def main():
|
||||
if not os.path.isfile(BIN):
|
||||
fail("no controller binary at %s" % BIN)
|
||||
case_default()
|
||||
case_x16()
|
||||
case_x32()
|
||||
case_invalid()
|
||||
case_low_tick()
|
||||
print("PASS xy_mode_test")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
@@ -0,0 +1,414 @@
|
||||
#!/usr/bin/env python3
|
||||
"""The XY microstep modes by the head accelerometer, the machine silent (on the board).
|
||||
|
||||
Usage: xy_pattern_accel.py [modes...] (default: 8 16 32)
|
||||
|
||||
The pattern, from home at full speed (F12000), one leg at a time:
|
||||
|
||||
1. home to X 18 in, Y 9 in (457.2, 228.6 mm)
|
||||
2. to X 9 in, Y 9 in (228.6, 228.6 mm)
|
||||
3. a 9 in circle from its mid-bottom, the center 4.5 in toward the
|
||||
back, ending back at X 9 in, Y 9 in
|
||||
4. to X 9 in, Y 0 (228.6, 0 mm)
|
||||
5. to X 0, Y 0 home
|
||||
|
||||
Per mode: stores xy_microsteps through forgectrl (which restarts the idle
|
||||
GRBL controller), waits for the controller at that mode, takes the
|
||||
engine's quiet hold (POST /cool/quiet: air assist, exhaust, intake and
|
||||
purge fans, the coolant pump and the TEC all off, so the machine is silent
|
||||
and the modes can be heard as well as measured; a dry pattern, the laser
|
||||
latched), waits the fixed QUIET_S, then samples the head accelerometer
|
||||
(the crash watch's LIS2HH12 on i2c-3, straight over the bus at 800 Hz)
|
||||
through the pattern. Reports, per leg and overall, the cruise-window
|
||||
RMS and peak-to-peak of X and Y with the mean removed, the leg times,
|
||||
the kernel counters against home, and cnc/underruns; JSON with the raw
|
||||
trace to the bench data directory. Releases the hold and puts the
|
||||
accelerometer back on every exit path. Ends at x8.
|
||||
|
||||
Needs the controller in GRBL mode, homed and standing at home, the lid
|
||||
closed, no other Grbl client, and the bed clear across the pattern.
|
||||
"""
|
||||
import ctypes
|
||||
import fcntl
|
||||
import json
|
||||
import math
|
||||
import os
|
||||
import re
|
||||
import socket
|
||||
import struct
|
||||
import sys
|
||||
import threading
|
||||
import time
|
||||
|
||||
from gfbench import data_path, forgectrl_get, forgectrl_post
|
||||
|
||||
def modes_from_argv(argv):
|
||||
"""The modes named on the command line ("8 16 32" as one argument or
|
||||
several), the three by default."""
|
||||
return [int(a) for arg in argv for a in arg.split()] or [8, 16, 32]
|
||||
|
||||
|
||||
FEED = 12000
|
||||
IN = 25.4
|
||||
LEGS = [("home to (18, 9) in", "G1 X%.3f Y%.3f F%d" % (18 * IN, 9 * IN, FEED)),
|
||||
("to (9, 9) in", "G1 X%.3f F%d" % (-9 * IN, FEED)),
|
||||
("9 in circle", "G2 X0 Y0 I0 J%.3f F%d" % (-4.5 * IN, FEED)),
|
||||
("to (9, 0) in", "G1 Y%.3f F%d" % (-9 * IN, FEED)),
|
||||
("to home", "G1 X%.3f F%d" % (-9 * IN, FEED))]
|
||||
QUIET_S = 10.0 # the fixed wait after every fan is commanded off
|
||||
TRIM_S = 0.35 # the ramps at each end of a leg, outside the cruise window
|
||||
TOL_MM = 0.05
|
||||
SYS = "/sys/glowforge/"
|
||||
|
||||
I2C_DEV = "/dev/i2c-3"
|
||||
I2C_SLAVE_FORCE = 0x0706
|
||||
I2C_SMBUS = 0x0720
|
||||
I2C_SMBUS_READ = 1
|
||||
I2C_SMBUS_I2C_BLOCK_DATA = 8
|
||||
ADDR = 0x1E
|
||||
CTRL1 = 0x20
|
||||
CTRL1_RUN = 0x6F # 800 Hz, BDU, XYZ on (the crash watch's run value)
|
||||
CTRL4 = 0x23
|
||||
OUT_X_L = 0x28
|
||||
|
||||
|
||||
class SmbusData(ctypes.Union):
|
||||
_fields_ = [("byte", ctypes.c_uint8), ("word", ctypes.c_uint16), ("block", ctypes.c_uint8 * 34)]
|
||||
|
||||
|
||||
class SmbusIoctl(ctypes.Structure):
|
||||
_fields_ = [("read_write", ctypes.c_uint8), ("command", ctypes.c_uint8),
|
||||
("size", ctypes.c_uint32), ("data", ctypes.POINTER(SmbusData))]
|
||||
|
||||
|
||||
class Accel:
|
||||
"""The chip over the bus. The six output registers come in ONE bus
|
||||
transaction (an SMBus block read), so another reader on the bus (the
|
||||
crash watch) cannot slip a register pointer between the write and
|
||||
the read."""
|
||||
|
||||
def __init__(self):
|
||||
self.fd = os.open(I2C_DEV, os.O_RDWR)
|
||||
fcntl.ioctl(self.fd, I2C_SLAVE_FORCE, ADDR)
|
||||
self.ctrl1_found = self.reg(CTRL1)
|
||||
self.ctrl4 = self.reg(CTRL4)
|
||||
|
||||
def reg(self, r):
|
||||
os.write(self.fd, bytes([r]))
|
||||
return os.read(self.fd, 1)[0]
|
||||
|
||||
def set_reg(self, r, v):
|
||||
os.write(self.fd, bytes([r, v]))
|
||||
|
||||
def start(self):
|
||||
self.set_reg(CTRL1, CTRL1_RUN)
|
||||
time.sleep(0.02)
|
||||
|
||||
def restore(self):
|
||||
try:
|
||||
self.set_reg(CTRL1, self.ctrl1_found)
|
||||
finally:
|
||||
os.close(self.fd)
|
||||
|
||||
def read(self):
|
||||
data = SmbusData()
|
||||
data.block[0] = 6
|
||||
req = SmbusIoctl(I2C_SMBUS_READ, OUT_X_L, I2C_SMBUS_I2C_BLOCK_DATA, ctypes.pointer(data))
|
||||
fcntl.ioctl(self.fd, I2C_SMBUS, req)
|
||||
return struct.unpack("<hhh", bytes(data.block[1:7]))
|
||||
|
||||
|
||||
class Sampler(threading.Thread):
|
||||
def __init__(self, accel):
|
||||
super().__init__(daemon=True)
|
||||
self.accel = accel
|
||||
self.samples = [] # (t, x, y, z)
|
||||
self.errors = 0
|
||||
self._halt = threading.Event()
|
||||
|
||||
def run(self):
|
||||
while not self._halt.is_set():
|
||||
try:
|
||||
x, y, z = self.accel.read()
|
||||
self.samples.append((time.monotonic(), x, y, z))
|
||||
except OSError:
|
||||
self.errors += 1
|
||||
time.sleep(0.002)
|
||||
|
||||
def stop(self):
|
||||
self._halt.set()
|
||||
self.join(2.0)
|
||||
|
||||
|
||||
def sysfs(attr):
|
||||
with open(SYS + attr) as f:
|
||||
return f.read().strip()
|
||||
|
||||
|
||||
def fc_get(path):
|
||||
st, body = forgectrl_get(path)
|
||||
return body if st == 200 and isinstance(body, dict) else {}
|
||||
|
||||
|
||||
def fail(msg):
|
||||
print("FAIL: %s" % msg)
|
||||
raise SystemExit(1)
|
||||
|
||||
|
||||
def set_mode(mode):
|
||||
if mode == 8:
|
||||
st, body = forgectrl_post("/settings", params={"xy_microsteps": ""})
|
||||
else:
|
||||
st, body = forgectrl_post("/settings", data={"xy_microsteps": str(mode)})
|
||||
if st != 200:
|
||||
fail("settings write refused: %s %s" % (st, body))
|
||||
time.sleep(3)
|
||||
t0 = time.time()
|
||||
while time.time() - t0 < 180:
|
||||
m = fc_get("/mode")
|
||||
if m.get("controller") == "running" and m.get("motion") == "verified" and sysfs("cnc/x_mode") == str(mode):
|
||||
time.sleep(2)
|
||||
print("x%d: controller pid %s, step_freq %s, ramp_rate %s"
|
||||
% (mode, m.get("pid"), sysfs("cnc/step_freq"), sysfs("cnc/ramp_rate")))
|
||||
return
|
||||
time.sleep(2)
|
||||
fail("x%d: the controller did not come back" % mode)
|
||||
|
||||
|
||||
def quiet(on):
|
||||
"""The engine's quiet hold with the pump: every fan, the coolant pump
|
||||
and the TEC off (on), or the phase's posture back (off)."""
|
||||
st, body = forgectrl_post("/cool/quiet", params={"on": "1" if on else "0", "pump": "1"})
|
||||
if st != 200:
|
||||
fail("quiet hold %s refused: %s %s" % ("on" if on else "off", st, body))
|
||||
return body
|
||||
|
||||
|
||||
class Grbl:
|
||||
def __init__(self):
|
||||
self.s = socket.create_connection(("127.0.0.1", 23), timeout=5)
|
||||
time.sleep(0.3)
|
||||
self.drain()
|
||||
|
||||
def drain(self):
|
||||
self.s.settimeout(0.2)
|
||||
try:
|
||||
while self.s.recv(4096):
|
||||
pass
|
||||
except (socket.timeout, OSError):
|
||||
pass
|
||||
|
||||
def cmd(self, line, timeout=30):
|
||||
self.s.sendall((line + "\n").encode())
|
||||
self.s.settimeout(timeout)
|
||||
buf = ""
|
||||
end = time.time() + timeout
|
||||
while time.time() < end:
|
||||
try:
|
||||
buf += self.s.recv(4096).decode(errors="replace")
|
||||
except socket.timeout:
|
||||
break
|
||||
if "ok\r\n" in buf or "error" in buf or "ALARM" in buf:
|
||||
break
|
||||
return buf
|
||||
|
||||
def status(self):
|
||||
self.s.sendall(b"?")
|
||||
self.s.settimeout(1.0)
|
||||
buf = ""
|
||||
end = time.time() + 1.0
|
||||
while time.time() < end and ">" not in buf:
|
||||
try:
|
||||
buf += self.s.recv(4096).decode(errors="replace")
|
||||
except socket.timeout:
|
||||
break
|
||||
if "error:" in buf or "ALARM" in buf:
|
||||
fail("the controller answered %r" % buf.strip()[-200:])
|
||||
m = re.search(r"<(\w+)[^>]*MPos:([-\d.]+),([-\d.]+),([-\d.]+)", buf)
|
||||
f = re.search(r"FS:([\d.]+),", buf)
|
||||
if not m:
|
||||
return None, None, 0.0
|
||||
return m.group(1), (float(m.group(2)), float(m.group(3))), float(f.group(1)) if f else 0.0
|
||||
|
||||
def run_leg(self, gcode, timeout=120):
|
||||
"""Send one move and time it by the state transitions: t_run at
|
||||
the first Run report, t_idle at the Idle after it. The line's own
|
||||
"ok" is not waited for first: an arc is fed to the planner
|
||||
segment by segment and its ok arrives late in the motion, so it
|
||||
is drained along with the reports instead."""
|
||||
self.s.sendall((gcode + "\n").encode())
|
||||
t_run = t_idle = None
|
||||
peak = 0.0
|
||||
pos = None
|
||||
end = time.time() + timeout
|
||||
while time.time() < end:
|
||||
st, pos, fs = self.status()
|
||||
now = time.monotonic()
|
||||
peak = max(peak, fs)
|
||||
if st and st.startswith(("Run", "Jog")) and t_run is None:
|
||||
t_run = now
|
||||
if st and st.startswith("Idle") and t_run is not None:
|
||||
t_idle = now
|
||||
break
|
||||
if st and st.startswith("Alarm"):
|
||||
fail("alarm during %r: %s" % (gcode, st))
|
||||
time.sleep(0.02)
|
||||
if t_idle is None:
|
||||
fail("the move %r did not finish" % gcode)
|
||||
self.drain()
|
||||
return t_run, t_idle, peak, pos
|
||||
|
||||
def close(self):
|
||||
self.s.close()
|
||||
|
||||
|
||||
def window(samples, t0, t1):
|
||||
xs = [s[1] for s in samples if t0 <= s[0] <= t1]
|
||||
ys = [s[2] for s in samples if t0 <= s[0] <= t1]
|
||||
return xs, ys
|
||||
|
||||
|
||||
def stats(vals):
|
||||
if len(vals) < 2:
|
||||
return {"n": len(vals), "rms": None, "p2p": None}
|
||||
mean = sum(vals) / len(vals)
|
||||
rms = math.sqrt(sum((v - mean) ** 2 for v in vals) / len(vals))
|
||||
return {"n": len(vals), "rms": round(rms, 1), "p2p": max(vals) - min(vals)}
|
||||
|
||||
|
||||
def preflight():
|
||||
st = fc_get("/status")
|
||||
m = fc_get("/mode")
|
||||
if m.get("mode") != "grbl" or m.get("controller") != "running":
|
||||
fail("the GRBL controller is not running (%s)" % m)
|
||||
if not st.get("homed"):
|
||||
fail("the machine is not homed: run $H first, the pattern starts from home")
|
||||
if not (st.get("switches") or {}).get("lid"):
|
||||
fail("the lid is open")
|
||||
if st.get("state") != "idle":
|
||||
fail("the machine is not idle (%s)" % st.get("state"))
|
||||
home = (float((fc_get("/settings") or {}).get("gfcloud_home_x") or 0),
|
||||
float((fc_get("/settings") or {}).get("gfcloud_home_y") or 0))
|
||||
return home
|
||||
|
||||
|
||||
def run_pattern(mode, home):
|
||||
g = Grbl()
|
||||
accel = None
|
||||
sampler = None
|
||||
record = {"mode": mode, "step_freq": sysfs("cnc/step_freq"), "ramp_rate": sysfs("cnc/ramp_rate"),
|
||||
"feed": FEED, "legs": [], "quiet_s": QUIET_S}
|
||||
try:
|
||||
st, pos, _ = g.status()
|
||||
if not st or not st.startswith("Idle"):
|
||||
fail("controller is %s, not Idle" % st)
|
||||
if abs(pos[0] - home[0]) > TOL_MM or abs(pos[1] - home[1]) > TOL_MM:
|
||||
fail("the head is not at home (MPos %s, home %s): run $H first" % (pos, home))
|
||||
k0 = fc_get("/status").get("pos") or {}
|
||||
und0 = int(sysfs("cnc/underruns"))
|
||||
print("x%d: quiet hold: %s" % (mode, quiet(True)))
|
||||
pwm = {a: sysfs(a) for a in ("thermal/exhaust_pwm", "thermal/intake_pwm", "head/air_assist_pwm",
|
||||
"head/purge_air", "thermal/water_pump_on", "thermal/tec_on")}
|
||||
record["fans"] = pwm
|
||||
if any(v != "0" for v in pwm.values()):
|
||||
fail("a fan, the pump or the TEC is still commanded on: %s" % pwm)
|
||||
print("x%d: every fan, the pump and the TEC commanded off (%s); waiting %.0f s" % (mode, pwm, QUIET_S))
|
||||
time.sleep(QUIET_S)
|
||||
accel = Accel()
|
||||
record["ctrl4"] = accel.ctrl4
|
||||
accel.start()
|
||||
sampler = Sampler(accel)
|
||||
sampler.start()
|
||||
time.sleep(1.0)
|
||||
t_base0, t_base1 = time.monotonic() - 1.0, time.monotonic()
|
||||
g.cmd("G21")
|
||||
g.cmd("G91")
|
||||
for name, gcode in LEGS:
|
||||
t_run, t_idle, peak, pos = g.run_leg(gcode)
|
||||
record["legs"].append({"leg": name, "gcode": gcode, "t_run": t_run, "t_idle": t_idle,
|
||||
"peak_feed": peak, "end_mpos": pos})
|
||||
print(" %-22s %5.2f s, peak %5.0f mm/min, end MPos (%.3f, %.3f)"
|
||||
% (name, t_idle - t_run, peak, pos[0], pos[1]))
|
||||
g.cmd("G90")
|
||||
time.sleep(1.0)
|
||||
end = time.time() + 20
|
||||
while time.time() < end and fc_get("/status").get("state") != "idle":
|
||||
time.sleep(0.5)
|
||||
k1 = fc_get("/status").get("pos") or {}
|
||||
und1 = int(sysfs("cnc/underruns"))
|
||||
finally:
|
||||
# The fans first, whatever else fails: the hold must not outlive
|
||||
# the listening (the engine would drop it itself within 600 s).
|
||||
try:
|
||||
quiet(False)
|
||||
except SystemExit:
|
||||
print("WARNING: the quiet hold did not release; the engine drops it itself within 600 s")
|
||||
try:
|
||||
if sampler is not None:
|
||||
sampler.stop()
|
||||
finally:
|
||||
if accel is not None:
|
||||
accel.restore()
|
||||
g.close()
|
||||
samples = sampler.samples
|
||||
record["samples"] = len(samples)
|
||||
record["read_errors"] = sampler.errors
|
||||
record["rate_hz"] = round(len(samples) / (samples[-1][0] - samples[0][0]), 1) if len(samples) > 1 else 0
|
||||
bx, by = window(samples, t_base0, t_base1)
|
||||
record["baseline"] = {"x": stats(bx), "y": stats(by)}
|
||||
print(" baseline at rest (fans off): x rms %s p2p %s, y rms %s p2p %s, %d samples at %.0f Hz, %d read errors"
|
||||
% (record["baseline"]["x"]["rms"], record["baseline"]["x"]["p2p"],
|
||||
record["baseline"]["y"]["rms"], record["baseline"]["y"]["p2p"],
|
||||
len(samples), record["rate_hz"], sampler.errors))
|
||||
allx, ally = [], []
|
||||
for leg in record["legs"]:
|
||||
t0, t1 = leg["t_run"] + TRIM_S, leg["t_idle"] - TRIM_S
|
||||
xs, ys = window(samples, t0, t1)
|
||||
leg["cruise"] = {"x": stats(xs), "y": stats(ys), "seconds": round(max(0.0, t1 - t0), 2)}
|
||||
allx += xs
|
||||
ally += ys
|
||||
print(" %-22s cruise %4.2f s: x rms %6s p2p %6s | y rms %6s p2p %6s (%d samples)"
|
||||
% (leg["leg"], leg["cruise"]["seconds"], leg["cruise"]["x"]["rms"], leg["cruise"]["x"]["p2p"],
|
||||
leg["cruise"]["y"]["rms"], leg["cruise"]["y"]["p2p"], leg["cruise"]["x"]["n"]))
|
||||
record["overall"] = {"x": stats(allx), "y": stats(ally)}
|
||||
kd = (float(k1.get("x", 0)) - float(k0.get("x", 0)), float(k1.get("y", 0)) - float(k0.get("y", 0)))
|
||||
record["kernel_return_mm"] = kd
|
||||
record["underruns"] = [und0, und1]
|
||||
ok = abs(kd[0]) <= TOL_MM and abs(kd[1]) <= TOL_MM and und1 == und0
|
||||
record["result"] = "PASS" if ok else "FAIL"
|
||||
print(" overall cruise: x rms %s p2p %s | y rms %s p2p %s | kernel return dx=%.3f dy=%.3f | underruns %d->%d | %s"
|
||||
% (record["overall"]["x"]["rms"], record["overall"]["x"]["p2p"], record["overall"]["y"]["rms"],
|
||||
record["overall"]["y"]["p2p"], kd[0], kd[1], und0, und1, record["result"]))
|
||||
record["trace"] = [(round(s[0] - samples[0][0], 4), s[1], s[2], s[3]) for s in samples]
|
||||
return record
|
||||
|
||||
|
||||
def main():
|
||||
modes = modes_from_argv(sys.argv[1:])
|
||||
home = preflight()
|
||||
stamp = time.strftime("%Y%m%d%H%M%S")
|
||||
results = []
|
||||
for mode in modes:
|
||||
print("=== x%d ===" % mode)
|
||||
set_mode(mode)
|
||||
rec = run_pattern(mode, home)
|
||||
path = data_path("xy_pattern_accel_x%d_%s.json" % (mode, stamp))
|
||||
with open(path, "w") as f:
|
||||
json.dump(rec, f)
|
||||
print(" record: %s" % path)
|
||||
results.append(rec)
|
||||
if modes[-1] != 8:
|
||||
print("=== restore x8 ===")
|
||||
set_mode(8)
|
||||
print("\n=== summary (cruise RMS with the mean removed, raw counts; CTRL4 0x%02x) ==="
|
||||
% results[0].get("ctrl4", 0))
|
||||
print("%-6s %10s %10s %10s %10s %8s" % ("mode", "x rms", "x p2p", "y rms", "y p2p", "result"))
|
||||
for rec in results:
|
||||
o = rec["overall"]
|
||||
print("%-6s %10s %10s %10s %10s %8s" % ("x%d" % rec["mode"], o["x"]["rms"], o["x"]["p2p"],
|
||||
o["y"]["rms"], o["y"]["p2p"], rec["result"]))
|
||||
return 0 if all(r["result"] == "PASS" for r in results) else 1
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main())
|
||||
Reference in New Issue
Block a user