laser.emission-witness: the read rule ends where the sampling stopped, and the rapid is placed inside the window

The dark stretch after the M5 runs on past the close, and three of its
judgments did not hold with it.

The read rule took the last sample served as the trail's last edge, and
sample() returns None for a request the daemon does not answer. So an
outage after the served close (one request that hangs to the client's
10 s timeout is enough) left a trail that ended at the close with no
gap, and the window that latches after the close was never read: the
test passed whole with a 30 ms fire at the end of the return rapid.
The sampling could also stop as early as about 1.8 s past the close,
and with a 0.9 s unread tail allowed, the last reading could come
before the latch of the window in flight: the review's tail model
(the kernel's ~1 s window latch, the sampler's GET timing, a fire just
before the close) finds 4 runs in 20000, 21 with a wider GET spread,
where that window goes unread and the judge passes.

The placement and the disarm timing searched the whole stretch. A
window that closed before the head was back (in the dwell at +20 mm
or on the return leg, up to about 2 s early) found the head back after
the close and passed "the rapid ran inside the dark stretch", with a
negative disarm time.

The rapid's sample count in the record ran past the close while its
span stopped at the close, so the record read about 8 Hz for a 5 Hz
loop.

sample_to_disarm moves from emission_witness to module level, ends two
windows and EMISSION_READ_GAP_S past the close, and returns the time
the sampling stopped beside the samples. read_gaps takes that time as
the trail's last edge (t_end), so a tail the daemon served none of is a
gap. rapid_place judges the placement over the armed part alone, up to
the first disarmed sample, and the disarm time must be 0 or more. The
rapid block counts its samples to the close and records how long the
sampling ran past it (sampled_past_close_s), and the log line says
both. The relit, HV and read judgments still cover the whole stretch.
The description says what is judged. tests/test_laser_dwell.py drops
relit(), a copy of the judge that no change to laser.py could fail.
Every change makes the judge stricter: no energy or motion path
changes.

Proof: tests/test_laser_dwell.py, the Unit tests step of
forgetest-ci.yml (python -B -m unittest discover -s tests), 13 tests OK
on Windows with python 3.14 and on Ubuntu 24.04 with python 3.12:
test_a_tail_the_daemon_never_served_fails (every sample served read,
then nothing for 2.9 s before the sampling stopped: gap 2.9 s, the rule
fails; served through, it holds),
test_the_sampling_runs_on_past_the_close_and_ends_where_it_stopped
(sample_to_disarm on a fake clock, the daemon serving the close and
nothing after it: the sampling ends at least 2.9 s past the close and
the rule fails), and
test_a_window_closed_before_the_head_is_back_does_not_place_the_rapid
(a close after the return places out, back and the close; a close in
the dwell at +20 mm leaves back None). test_cooling_suite.py (3) and
test_setup_burns.py (22) pass. Negative controls on scratch copies:
read_gaps without the end edge fails the first two cases; the end at
two windows fails the second; sample_to_disarm that returns the samples
alone errors in the second; rapid_place over the whole stretch fails
the third. emission_witness run whole on a fake clock with only its I/O
stubbed: a window closing 0.2, 1.0, 1.9 or 2.5 s before the head is
back fails the placement (the code before this change passed the first
three), and a normal close passes with 22 samples over 4.07 s to the
close in the record (32 before). Against a simulated daemon and kernel
latch with that 30 ms fire: an outage after the served close and one
request hung to the 10 s timeout fail the read rule (gaps 3.05 s and
10.32 s; both passed before); a tail served through with no fire
passes. In the review's tail model, the end edge alone and the longer
sampling alone each leave 0 of 20000 runs with the window in flight
unread and the judge passing, in both GET spreads. pyflakes reports no
undefined name.

Acceptance: laser.emission-witness's fingerprint moves (its own text
and laser.py's shared text), laser.armed-kill and
laser.pause-resume-lid-cancel move through the shared text, and
cooling.flow-under-load and cooling.fail-tier-stop move because they
import from .laser in their own functions; the other 38 do not move
(source_sha compared before and after). Against the campaign this
series already needs, that adds no run: 850498f moved all 43 once.
laser.emission-witness is always run, and the next campaign judges the
rapid with the end edge on the bench.
This commit is contained in:
ScottW514
2026-10-16 14:40:53 -04:00
parent 41ca7fab82
commit bed46e961e
2 changed files with 178 additions and 81 deletions
+88 -61
View File
@@ -291,11 +291,13 @@ def dwell_gap(samples):
"hv_enable_dipped": dip is not None, "hv_enable_back_lit": back_lit} "hv_enable_dipped": dip is not None, "hv_enable_back_lit": back_lit}
def read_gaps(samples, key): def read_gaps(samples, key, t_end=None):
"""Over a sample trail: how many samples carry a reading for `key`, """Over a sample trail: how many samples carry a reading for `key`,
how many do not, and the longest stretch of seconds between two how many do not, and the longest stretch of seconds between two
readings, the trail's ends counted as edges (an unread lead or tail readings, the trail's ends counted as edges (an unread lead or tail
is a gap too; a trail with no reading is one gap, its whole span). is a gap too; a trail with no reading is one gap, its whole span).
The trail ends when its sampling stopped (`t_end`), not at the last
sample served: a tail the daemon did not serve at all is a gap too.
A reading the daemon could not serve is a gap, never a dark sample: A reading the daemon could not serve is a gap, never a dark sample:
/status leaves the laser block out when the kernel's count did not /status leaves the laser block out when the kernel's count did not
read, and the count latches once per ~1 s window, so a window with read, and the count latches once per ~1 s window, so a window with
@@ -303,11 +305,60 @@ def read_gaps(samples, key):
if not samples: if not samples:
return {"read": 0, "unread": 0, "gap_max_s": None} return {"read": 0, "unread": 0, "gap_max_s": None}
ts = [s["t"] for s in samples if s.get(key) is not None] ts = [s["t"] for s in samples if s.get(key) is not None]
edges = [samples[0]["t"]] + ts + [samples[-1]["t"]] last = samples[-1]["t"] if t_end is None else max(t_end, samples[-1]["t"])
edges = [samples[0]["t"]] + ts + [last]
gap = max(b - a for a, b in zip(edges, edges[1:])) gap = max(b - a for a, b in zip(edges, edges[1:]))
return {"read": len(ts), "unread": len(samples) - len(ts), "gap_max_s": round(gap, 3)} return {"read": len(ts), "unread": len(samples) - len(ts), "gap_max_s": round(gap, 3)}
def rapid_place(stretch, x0):
"""Where the head's reports place the rapid after the M5, judged over
the armed part of the dark stretch alone: its samples up to the first
disarmed one, that one included. Returns (i_out, i_back, i_closed):
the first sample with the head at least 19.5 mm out, the first after
it with the head back at x0 (within 0.05 mm), and the first disarmed
sample; each None when there is none. A window that closes before
the head is back leaves i_back None: the samples after the close
never place the rapid inside the window."""
i_closed = next((i for i, s in enumerate(stretch) if not s["armed"]), None)
part = stretch if i_closed is None else stretch[:i_closed + 1]
i_out = next((i for i, s in enumerate(part) if s.get("x") is not None and s["x"] - x0 >= 19.5), None)
i_back = (next((i for i in range(i_out + 1, len(part))
if part[i].get("x") is not None and abs(part[i]["x"] - x0) < 0.05), None)
if i_out is not None else None)
return i_out, i_back, i_closed
def sample_to_disarm(ctx, g, timeout=20.0, sample_hz=8):
"""Sample until the armed window closes, or the timeout, with the
controller's state and machine X beside each sample ("x"). A G4 dwell
reports Idle, so Idle cannot end this; the window closing at the
program end does, after two more kernel windows and the read gap
allowed: the LASER_ON count for the window in flight at the close
latches up to one window later and is held through the next, so the
sampling runs on until that count has been read. Returns the samples
served and the time the sampling stopped, the trail's end for the
read rule."""
samples = []
period = 1.0 / sample_hz
t0 = time.time()
end = t0 + timeout
t_closed = None
while time.time() < end:
ctx.checkpoint()
smp = sample(ctx)
rep = g.status_report()
if smp:
smp["gstate"] = rep.get("state")
smp["x"] = (rep.get("MPos") or (None,))[0]
samples.append(smp)
if t_closed is None and not smp["armed"]:
t_closed = time.time()
end = t_closed + 2 * EMISSION_WINDOW_S + EMISSION_READ_GAP_S
time.sleep(period)
return samples, time.time()
def run_and_sample(ctx, g, job, sample_hz=8, overall_timeout=200): def run_and_sample(ctx, g, job, sample_hz=8, overall_timeout=200):
"""Stream the job; sample forgectrl through arm -> fire -> disarm. """Stream the job; sample forgectrl through arm -> fire -> disarm.
Completes on: emission seen then Idle > 3 s; or armed then disarmed Completes on: emission seen then Idle > 3 s; or armed then disarmed
@@ -411,13 +462,14 @@ def wait_disarm(ctx, timeout):
"second press. The square ends at M5 with the window still open; once the " "second press. The square ends at M5 with the window still open; once the "
"LASER_ON count reads 0 (the dark point), a rapid runs 20 mm out over the " "LASER_ON count reads 0 (the dark point), a rapid runs 20 mm out over the "
"first side and back, then the program ends. From the dark point until the " "first side and back, then the program ends. From the dark point until the "
"window closes, every sample reads no emission and the HV current at its idle " "window closes, and on until the count of the window in flight at the close is "
"reading, the kernel's LASER_ON count read in every ~1 s window of that stretch " "read, every sample reads no emission and the HV current at its idle reading, "
"and in the window that latches after the close (a reading the daemon could not " "the kernel's LASER_ON count read in every ~1 s window of that stretch (a "
"serve is a gap, never a dark sample), and the head's reports place the rapid " "reading the daemon could not serve is a gap, never a dark sample, and so is a "
"inside that stretch: LASER_ON stays off with FIRE clear while the chain can " "tail it served none of), and the head's reports place the rapid inside the "
"fire. The M2 program end " "window, between the dark point and the close: LASER_ON stays off with FIRE "
"disarms promptly (job-based, not the 60 s idle grace) and the button goes " "clear while the chain can fire. The M2 program end disarms promptly after the "
"head is back (job-based, not the 60 s idle grace) and the button goes "
"dark. The operator confirms all four sides.") "dark. The operator confirms all four sides.")
def emission_witness(ctx): def emission_witness(ctx):
ev = ctx.evidence ev = ctx.evidence
@@ -425,33 +477,6 @@ def emission_witness(ctx):
# first side and back, then the program end. # first side and back, then the program end.
rapid_tail = ["G0 X20", "G4 P2.5", "G0 X-20", "G4 P0.5", "G90", "M2"] rapid_tail = ["G0 X20", "G4 P2.5", "G0 X-20", "G4 P0.5", "G90", "M2"]
def sample_to_disarm(g, timeout=20.0, sample_hz=8):
"""Sample until the armed window closes, or the timeout, with the
controller's state and machine X beside each sample ("x"). A G4
dwell reports Idle, so Idle cannot end this; the window closing at
the program end does, after two more kernel windows: the LASER_ON
count for the window in flight at the close latches up to one
window later and is held through the next, so the sampling runs
on until that count has been read."""
samples = []
period = 1.0 / sample_hz
t0 = time.time()
end = t0 + timeout
t_closed = None
while time.time() < end:
ctx.checkpoint()
smp = sample(ctx)
rep = g.status_report()
if smp:
smp["gstate"] = rep.get("state")
smp["x"] = (rep.get("MPos") or (None,))[0]
samples.append(smp)
if t_closed is None and not smp["armed"]:
t_closed = time.time()
end = t_closed + 2 * EMISSION_WINDOW_S
time.sleep(period)
return samples
with ctx.grbl() as g, LiveJob(ctx, g): with ctx.grbl() as g, LiveJob(ctx, g):
prepare(ctx, g) prepare(ctx, g)
base = sample(ctx) base = sample(ctx)
@@ -490,7 +515,7 @@ def emission_witness(ctx):
dark["x"] = (rep.get("MPos") or (None,))[0] dark["x"] = (rep.get("MPos") or (None,))[0]
ctx.check(dark["x"] is not None, "the controller reports no machine position: %s", rep) ctx.check(dark["x"] is not None, "the controller reports no machine position: %s", rep)
stream(g, rapid_tail) stream(g, rapid_tail)
tail = sample_to_disarm(g) tail, t_end = sample_to_disarm(ctx, g)
finally: finally:
ctx.clear_notice() ctx.clear_notice()
samples = cut + [dark] + tail samples = cut + [dark] + tail
@@ -545,39 +570,40 @@ def emission_witness(ctx):
ctx.log("exhaust at the first emission sample: %s rpm (floor %.0f)", ctx.log("exhaust at the first emission sample: %s rpm (floor %.0f)",
ev["exhaust_at_first_fire"], floor) ev["exhaust_at_first_fire"], floor)
# The rapid after the M5, with the window open and FIRE clear: every # The rapid after the M5, with the window open and FIRE clear: every
# sample from the dark point until the window closes reads no emission # sample from the dark point until the sampling stops (two windows and
# and the HV current at its idle reading, and the kernel's LASER_ON # the read gap allowed past the close) reads no emission and the HV
# count was read in every ~1 s window of that stretch and in the one # current at its idle reading, and the kernel's LASER_ON count was read
# that latches after the close (the stretch runs two windows past it). # in every ~1 s window of that stretch, the one that latches after the
# A reading the daemon could not serve is a gap, never a dark sample. # close included: the end of the sampling is the trail's last edge. A
# The head's reports place the rapid inside the stretch: out to # reading the daemon could not serve is a gap, never a dark sample.
# +20 mm, then back at the start. # The head's reports place the rapid inside the window, from the dark
# point to the close: out to +20 mm, then back at the start.
stretch = [dark] + tail stretch = [dark] + tail
x0 = dark["x"] x0 = dark["x"]
i_out = next((i for i, s in enumerate(stretch) if s.get("x") is not None and s["x"] - x0 >= 19.5), None) i_out, i_back, i_closed = rapid_place(stretch, x0)
i_back = (next((i for i in range(i_out + 1, len(stretch)) closed = stretch[i_closed] if i_closed is not None else None
if stretch[i].get("x") is not None and abs(stretch[i]["x"] - x0) < 0.05), None) armed_part = stretch if i_closed is None else stretch[:i_closed + 1]
if i_out is not None else None)
closed = next((s for s in stretch if not s["armed"]), None)
relit = [s for s in stretch if s["emission"]] relit = [s for s in stretch if s["emission"]]
hv_dark = max((s["hv"] for s in stretch if s["hv"] is not None), default=0) hv_dark = max((s["hv"] for s in stretch if s["hv"] is not None), default=0)
reads = read_gaps(stretch, "emission") reads = read_gaps(stretch, "emission", t_end)
hv_reads = read_gaps(stretch, "hv") hv_reads = read_gaps(stretch, "hv", t_end)
dt = (closed["t"] - stretch[i_back]["t"]) if (closed and i_back is not None) else None dt = (closed["t"] - stretch[i_back]["t"]) if (closed and i_back is not None) else None
ev["rapid"] = {"samples": len(stretch), "span_s": round((closed or stretch[-1])["t"] - dark["t"], 2), ev["rapid"] = {"samples": len(armed_part), "span_s": round((closed or stretch[-1])["t"] - dark["t"], 2),
"x_start": x0, "x_max": max((s["x"] for s in stretch if s.get("x") is not None), default=None), "x_start": x0, "x_max": max((s["x"] for s in armed_part if s.get("x") is not None), default=None),
"out_at_s": round(stretch[i_out]["t"] - dark["t"], 2) if i_out is not None else None, "out_at_s": round(stretch[i_out]["t"] - dark["t"], 2) if i_out is not None else None,
"back_at_s": round(stretch[i_back]["t"] - dark["t"], 2) if i_back is not None else None, "back_at_s": round(stretch[i_back]["t"] - dark["t"], 2) if i_back is not None else None,
"relit": [(round(s["t"] - dark["t"], 2), s["emission"]) for s in relit][:8], "relit": [(round(s["t"] - dark["t"], 2), s["emission"]) for s in relit][:8],
"hv_max": hv_dark, "locked_at_dark": dark["locked"], "hv_max": hv_dark, "locked_at_dark": dark["locked"],
"locked_at_out": stretch[i_out]["locked"] if i_out is not None else None, "locked_at_out": stretch[i_out]["locked"] if i_out is not None else None,
"sampled_past_close_s": round(t_end - closed["t"], 2) if closed else None,
"emission_reads": reads, "hv_reads": hv_reads} "emission_reads": reads, "hv_reads": hv_reads}
ev["disarm_after_idle_s"] = round(dt, 1) if dt is not None else None ev["disarm_after_idle_s"] = round(dt, 1) if dt is not None else None
ctx.log("the rapid after the M5: %d samples over %.1f s from the dark point, out at +%s s, back at +%s s, " ctx.log("the rapid after the M5: %d samples over %.1f s from the dark point to the close, out at +%s s, "
"%d with emission, HV max %s; %d of %d samples carried the LASER_ON count, longest unread " "back at +%s s; sampled %s s past the close, %d of %d samples with emission, HV max %s, %d of %d "
"stretch %s s; the window closed %s s after the head was back", len(stretch), "carried the LASER_ON count, longest unread stretch %s s; the window closed %s s after the head "
ev["rapid"]["span_s"], ev["rapid"]["out_at_s"], ev["rapid"]["back_at_s"], len(relit), hv_dark, "was back", len(armed_part), ev["rapid"]["span_s"], ev["rapid"]["out_at_s"], ev["rapid"]["back_at_s"],
reads["read"], len(stretch), reads["gap_max_s"], ev["disarm_after_idle_s"]) ev["rapid"]["sampled_past_close_s"], len(relit), len(stretch), hv_dark, reads["read"],
len(stretch), reads["gap_max_s"], ev["disarm_after_idle_s"])
ctx.check(armed_seen, "the armed window was never observed (arm refused, or no button press)") ctx.check(armed_seen, "the armed window was never observed (arm refused, or no button press)")
ctx.check(peak > 0, "no emission witnessed on the square (emission_samples stayed 0)") ctx.check(peak > 0, "no emission witnessed on the square (emission_samples stayed 0)")
first_idle_fire = ev["idle_airflow_fire"][0] if idle_fire else (None, None, None) first_idle_fire = ev["idle_airflow_fire"][0] if idle_fire else (None, None, None)
@@ -592,8 +618,9 @@ def emission_witness(ctx):
ctx.check(hv and max(hv) > min(hv), "HV current did not rise during the burn (%s..%s)", ctx.check(hv and max(hv) > min(hv), "HV current did not rise during the burn (%s..%s)",
ev["hv_min"], ev["hv_max"]) ev["hv_min"], ev["hv_max"])
ctx.check(i_out is not None and i_back is not None, ctx.check(i_out is not None and i_back is not None,
"the rapid did not run inside the dark stretch: out at %s, back at %s (X from %s, max %s)", "the rapid did not run inside the dark stretch with the window open: out at %s, back at %s "
ev["rapid"]["out_at_s"], ev["rapid"]["back_at_s"], x0, ev["rapid"]["x_max"]) "(X from %s, max %s), the window closed at %s", ev["rapid"]["out_at_s"], ev["rapid"]["back_at_s"],
x0, ev["rapid"]["x_max"], round(closed["t"] - dark["t"], 2) if closed else None)
ctx.check(stretch[i_out]["locked"] is False, "the laser latch was locked with the head at +20 mm " ctx.check(stretch[i_out]["locked"] is False, "the laser latch was locked with the head at +20 mm "
"(laser_locked %s): the rapid ran unarmed", stretch[i_out]["locked"]) "(laser_locked %s): the rapid ran unarmed", stretch[i_out]["locked"])
ctx.check(reads["read"] >= RAPID_MIN_READS and reads["gap_max_s"] is not None ctx.check(reads["read"] >= RAPID_MIN_READS and reads["gap_max_s"] is not None
@@ -609,7 +636,7 @@ def emission_witness(ctx):
hv_reads["read"], len(stretch), RAPID_MIN_READS) hv_reads["read"], len(stretch), RAPID_MIN_READS)
ctx.check(hv_dark <= HV_DARK_MAX, "HV current %s after the M5 (idle reads ~0): the discharge ran " ctx.check(hv_dark <= HV_DARK_MAX, "HV current %s after the M5 (idle reads ~0): the discharge ran "
"through the rapid", hv_dark) "through the rapid", hv_dark)
ctx.check(dt is not None and dt < 10.0, ctx.check(dt is not None and 0 <= dt < 10.0,
"the M2 job did not disarm promptly (%s s after the head was back; the idle grace is ~60 s)", "the M2 job did not disarm promptly (%s s after the head was back; the idle grace is ~60 s)",
ev["disarm_after_idle_s"]) ev["disarm_after_idle_s"])
ctx.check(gap["button_latch_unlocked_samples"] and gap["button_latch_unlocked_max"] == 0, ctx.check(gap["button_latch_unlocked_samples"] and gap["button_latch_unlocked_max"] == 0,
+90 -20
View File
@@ -9,8 +9,11 @@ whose readback word shows the laser latch unlocked, both bits from that
word, so the relocked tail (where the relock itself sets the latch and word, so the relocked tail (where the relock itself sets the latch and
the emission counter still reads nonzero) is never judged. And its the emission counter still reads nonzero) is never judged. And its
read-gap verdict over the dark stretch after the M5: a sample whose read-gap verdict over the dark stretch after the M5: a sample whose
/status carried no laser block is a gap, never a dark sample, and a /status carried no laser block is a gap, never a dark sample, a tail the
stretch with a kernel window unread fails the rule.""" daemon did not serve before the sampling stopped is a gap too, and a
stretch with a kernel window unread fails the rule; the sampling runs on
past the close; and the head's reports place the rapid inside the armed
window only."""
import unittest import unittest
from forgetest.suite import laser from forgetest.suite import laser
@@ -71,20 +74,12 @@ def covered(r):
and r["gap_max_s"] <= laser.EMISSION_READ_GAP_S) and r["gap_max_s"] <= laser.EMISSION_READ_GAP_S)
def relit(trail):
"""The samples that read emission: the relit judge alone, which an
unread sample never reaches."""
return [s for s in trail if s["emission"]]
class RapidReadTests(unittest.TestCase): class RapidReadTests(unittest.TestCase):
def test_every_sample_read_is_inside_the_rule(self): def test_every_sample_read_is_inside_the_rule(self):
trail = rapid_trail(25) r = laser.read_gaps(rapid_trail(25), "emission")
r = laser.read_gaps(trail, "emission")
self.assertEqual((r["read"], r["unread"]), (25, 0)) self.assertEqual((r["read"], r["unread"]), (25, 0))
self.assertAlmostEqual(r["gap_max_s"], 0.2) self.assertAlmostEqual(r["gap_max_s"], 0.2)
self.assertTrue(covered(r)) self.assertTrue(covered(r))
self.assertEqual(relit(trail), [])
def test_one_transient_miss_is_inside_the_rule(self): def test_one_transient_miss_is_inside_the_rule(self):
r = laser.read_gaps(rapid_trail(25, holes=(12,)), "emission") r = laser.read_gaps(rapid_trail(25, holes=(12,)), "emission")
@@ -98,28 +93,34 @@ class RapidReadTests(unittest.TestCase):
self.assertTrue(covered(r)) self.assertTrue(covered(r))
def test_a_hole_longer_than_a_window_fails(self): def test_a_hole_longer_than_a_window_fails(self):
trail = rapid_trail(25, holes=(10, 11, 12, 13, 14)) r = laser.read_gaps(rapid_trail(25, holes=(10, 11, 12, 13, 14)), "emission")
r = laser.read_gaps(trail, "emission")
self.assertAlmostEqual(r["gap_max_s"], 1.2) self.assertAlmostEqual(r["gap_max_s"], 1.2)
self.assertFalse(covered(r)) self.assertFalse(covered(r))
self.assertEqual(relit(trail), [])
def test_the_block_gone_after_the_dark_point_fails(self): def test_the_block_gone_after_the_dark_point_fails(self):
trail = rapid_trail(25, holes=tuple(range(1, 25))) r = laser.read_gaps(rapid_trail(25, holes=tuple(range(1, 25))), "emission")
r = laser.read_gaps(trail, "emission")
self.assertEqual((r["read"], r["unread"]), (1, 24)) self.assertEqual((r["read"], r["unread"]), (1, 24))
self.assertAlmostEqual(r["gap_max_s"], 4.8) self.assertAlmostEqual(r["gap_max_s"], 4.8)
self.assertFalse(covered(r)) self.assertFalse(covered(r))
self.assertEqual(relit(trail), [])
def test_the_block_gone_for_the_last_window_and_a_half_fails(self): def test_the_block_gone_for_the_last_window_and_a_half_fails(self):
# the tail is an edge: the window that latches after the close # the tail is an edge: the window that latches after the close
# goes unread # goes unread
trail = rapid_trail(25, holes=tuple(range(17, 25))) r = laser.read_gaps(rapid_trail(25, holes=tuple(range(17, 25))), "emission")
r = laser.read_gaps(trail, "emission")
self.assertAlmostEqual(r["gap_max_s"], 1.6) self.assertAlmostEqual(r["gap_max_s"], 1.6)
self.assertFalse(covered(r)) self.assertFalse(covered(r))
self.assertEqual(relit(trail), [])
def test_a_tail_the_daemon_never_served_fails(self):
# every sample served was read, and the served samples end at the
# close: the sampling ran on past it with nothing served, so the
# stretch from the last reading to where the sampling stopped is a gap
trail = rapid_trail(25)
tail = 2 * laser.EMISSION_WINDOW_S + laser.EMISSION_READ_GAP_S
r = laser.read_gaps(trail, "emission", trail[-1]["t"] + tail)
self.assertAlmostEqual(r["gap_max_s"], tail)
self.assertFalse(covered(r))
# served through: the sampling stops one loop after the last sample
self.assertTrue(covered(laser.read_gaps(trail, "emission", trail[-1]["t"] + 0.125)))
def test_a_stretch_of_one_sample_fails(self): def test_a_stretch_of_one_sample_fails(self):
r = laser.read_gaps(rapid_trail(1), "emission") r = laser.read_gaps(rapid_trail(1), "emission")
@@ -127,5 +128,74 @@ class RapidReadTests(unittest.TestCase):
self.assertFalse(covered(r)) self.assertFalse(covered(r))
class Clock:
"""time.time and time.sleep on a clock the test moves."""
def __init__(self, now=1000.0):
self.now = now
def time(self):
return self.now
def sleep(self, s):
self.now += s
class RapidTailTests(unittest.TestCase):
def test_the_sampling_runs_on_past_the_close_and_ends_where_it_stopped(self):
# The window closes 3 s in, and the daemon serves the closed
# sample and nothing after it: the sampling still runs two kernel
# windows and the read gap allowed past the close, and the time it
# stopped is the trail's end, so the tail it never served fails the
# read rule.
clock = Clock()
close = clock.now + 3.0
served = []
def sample(_ctx):
clock.now += 0.06 # the two GETs
if served and not served[-1]["armed"]:
return None # nothing served after the close
s = {"t": clock.now, "armed": clock.now < close, "emission": 0, "hv": 0}
served.append(s)
return s
class Ctx:
def checkpoint(self):
pass
class Grbl:
def status_report(self):
return {"state": "Idle", "MPos": (0.0, 0.0, 0.0)}
saved = laser.time, laser.sample
self.addCleanup(setattr, laser, "time", saved[0])
self.addCleanup(setattr, laser, "sample", saved[1])
laser.time, laser.sample = clock, sample
tail, t_end = laser.sample_to_disarm(Ctx(), Grbl())
closed = tail[-1]
self.assertFalse(closed["armed"])
self.assertTrue(all(s["armed"] for s in tail[:-1]))
self.assertGreaterEqual(t_end - closed["t"], 2 * laser.EMISSION_WINDOW_S + laser.EMISSION_READ_GAP_S)
self.assertFalse(covered(laser.read_gaps(tail, "emission", t_end)))
def place_trail(xs, closes_at):
"""A dark stretch with the head at machine X `xs` (one sample each,
0.2 s apart), the window closing at sample `closes_at`."""
return [{"t": 100.0 + i * 0.2, "x": float(x), "armed": i < closes_at} for i, x in enumerate(xs)]
class RapidPlaceTests(unittest.TestCase):
# out to +20 mm, the dwell there, back at the start, the dwell before the M2
XS = [0, 5, 20, 20, 20, 10, 0, 0, 0, 0]
def test_a_window_closed_before_the_head_is_back_does_not_place_the_rapid(self):
# the window closes after the head is back: out at 2, back at 6, closed at 8
self.assertEqual(laser.rapid_place(place_trail(self.XS, 8), 0.0), (2, 6, 8))
# the window closes in the dwell at +20 mm and the head comes back
# after the close: the samples past the close do not place it
self.assertEqual(laser.rapid_place(place_trail(self.XS, 4), 0.0), (2, None, 4))
if __name__ == "__main__": if __name__ == "__main__":
unittest.main() unittest.main()