mirror of
https://github.com/openglow-org/forgefirm.git
synced 2026-09-27 16:51:12 -07:00
bench: platform drills, cutting-power lid-IR and expected-stop live drills, SIGTERM lifecycle case
- platform_drills.py (on the board, forgectrl stopped): dead-man trip readback, rmmod/modprobe with concurrent attr reads, decay/microstep readback, LED sequence. - live_fire_drills.py: ircut (S/F selectable characterization job), expstop (armed job + POST /controller/stop, controller left stopped), ctrlstart (separate, operator-approved resume); the token header is sent in exact case. - laser_lifecycle_test.py: sigterm-mid-job - SIGTERM during an armed job must stop it, relock the latch and exit promptly.
This commit is contained in:
ScottW514
@@ -274,6 +274,53 @@ def test_verdict_blocks_arm():
|
||||
s.close()
|
||||
|
||||
|
||||
def test_sigterm_mid_job():
|
||||
"""Rule 5: a termination signal during an armed job is a STOP, not a
|
||||
"finish the job first". The supervisor's expected-stop path (mode
|
||||
switch, diagnostics, the emergency lever) delivers SIGTERM; the
|
||||
controller must bring motion to a controlled stop, close the armed
|
||||
window (latch relocked) and exit promptly - long before the job
|
||||
would have ended and inside the supervisor's SIGKILL grace."""
|
||||
s = Session("sigterm-mid-job", disarm_s=60)
|
||||
try:
|
||||
send_line(s.sock, "M4 S100", s.log)
|
||||
send_line(s.sock, "G1 X300 F60", s.log) # a 5-minute move
|
||||
if not wait_for(s.log, ARMED, 5, s.sock):
|
||||
fail("[sigterm-mid-job] the job did not arm")
|
||||
# Let the run get under way.
|
||||
deadline = time.time() + 5
|
||||
running = False
|
||||
while time.time() < deadline and not running:
|
||||
s.sock.sendall(b"?")
|
||||
read_avail(s.sock, s.log, 0.3)
|
||||
running = "<Run" in "".join(s.log[-5:])
|
||||
if not running:
|
||||
fail("[sigterm-mid-job] the job never reported Run")
|
||||
t0 = time.time()
|
||||
s.proc.send_signal(signal.SIGTERM)
|
||||
# Keep draining the sender socket so the disarm report is captured.
|
||||
exited = False
|
||||
while time.time() - t0 < 10:
|
||||
read_avail(s.sock, s.log, 0.2)
|
||||
if s.proc.poll() is not None:
|
||||
exited = True
|
||||
break
|
||||
dt = time.time() - t0
|
||||
if not exited:
|
||||
fail("[sigterm-mid-job] the controller did not exit within 10 s of "
|
||||
"SIGTERM during a job (it must stop, not finish the job)")
|
||||
if dt > 3.0:
|
||||
fail("[sigterm-mid-job] exit took %.1f s after SIGTERM (want < 3 s)" % dt)
|
||||
if s.disarmed_count() < 1:
|
||||
fail("[sigterm-mid-job] no disarm (latch relock) reported before exit")
|
||||
if s.proc.returncode != 0:
|
||||
fail("[sigterm-mid-job] exit status %s (want a clean 0)" % s.proc.returncode)
|
||||
print("PASS [sigterm-mid-job]: SIGTERM during a job stopped it, relocked "
|
||||
"the latch and exited in %.2f s" % dt)
|
||||
finally:
|
||||
s.close()
|
||||
|
||||
|
||||
def main():
|
||||
if not os.path.isfile(BIN):
|
||||
fail("controller binary not found at %s" % BIN)
|
||||
@@ -281,6 +328,7 @@ def main():
|
||||
test_sender_change()
|
||||
test_hold_grace()
|
||||
test_verdict_blocks_arm()
|
||||
test_sigterm_mid_job()
|
||||
print("PASS: the armed-window lifecycle holds")
|
||||
|
||||
|
||||
|
||||
@@ -22,6 +22,23 @@ Drills (pass a name):
|
||||
underrun (position no longer trusted), a subsequent armed
|
||||
job must refuse to cut at the stale origin - the sender
|
||||
alarms and re-home is required. Reads homed via /status.
|
||||
ircut Lid-IR fire characterization at cutting power: a 30 mm
|
||||
square at S<power> (default 1000 = full) and F<feed>
|
||||
(default 300) on scrap, sampled like `witness`. Prints the
|
||||
per-channel peak delta over the ambient baseline and the
|
||||
engine's own "run telemetry" line is the record. Run it
|
||||
>= 3 times on representative material; the highest peak
|
||||
delta sizes cool_fire_ir_delta.
|
||||
ircut [host] [S] [F] e.g. ircut 192.0.2.1 1000 300
|
||||
expstop Armed kill on the EXPECTED-stop path: start a mark job,
|
||||
then mid-burn POST /controller/stop (the supervisor stops
|
||||
the controller: SIGTERM, reap, exit safing). PASS: emission
|
||||
drops to 0 within a few samples of the stop and stays 0,
|
||||
the kernel is not running, and POST /controller/start
|
||||
is a SEPARATE step (`ctrlstart`, run after the operator has
|
||||
judged the stop). Needs the panel token in GF_TOKEN
|
||||
(cat /data/forgefirm/panel.token on the board).
|
||||
ctrlstart POST /controller/start after an expstop; no motion, no laser.
|
||||
|
||||
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.
|
||||
@@ -339,13 +356,144 @@ def drill_faultpos(g):
|
||||
g.cmd('M5', timeout=1)
|
||||
|
||||
|
||||
def drill_ircut(g):
|
||||
power = int(sys.argv[3]) if len(sys.argv) > 3 else 1000
|
||||
feed = int(sys.argv[4]) if len(sys.argv) > 4 else 300
|
||||
print('=== lid-IR characterization: S%d F%d 30 mm square ===' % (power, feed))
|
||||
print('connect: %s' % prepare(g))
|
||||
base = sample_forgectrl()
|
||||
print('pre-fire: %s' % base)
|
||||
arm_cue()
|
||||
job = [
|
||||
'G91', 'G21', 'M4', 'S%d' % power,
|
||||
'G1 X30 F%d' % feed, 'G1 Y30 F%d' % feed,
|
||||
'G1 X-30 F%d' % feed, 'G1 Y-30 F%d' % feed,
|
||||
'M5', 'G90', 'M2',
|
||||
]
|
||||
samples = run_and_sample(g, job, overall_timeout=400)
|
||||
ir_peak = [0, 0, 0, 0]
|
||||
ir_min = [10 ** 6] * 4
|
||||
hv_vals = []
|
||||
emis = []
|
||||
fw = set()
|
||||
for s in samples:
|
||||
if s['ir'] and len(s['ir']) == 4:
|
||||
for i in range(4):
|
||||
ir_peak[i] = max(ir_peak[i], s['ir'][i])
|
||||
ir_min[i] = min(ir_min[i], s['ir'][i])
|
||||
if s['hv'] is not None:
|
||||
hv_vals.append(s['hv'])
|
||||
if s['emission'] is not None:
|
||||
emis.append(s['emission'])
|
||||
if s['fire_watch']:
|
||||
fw.add(s['fire_watch'])
|
||||
print('\n--- results ---')
|
||||
print('samples: %d emission peak=%s fire_watch states=%s'
|
||||
% (len(samples), max(emis) if emis else '-', sorted(fw)))
|
||||
delta = [round(ir_peak[i] - IR_BASELINE[i], 1) for i in range(4)]
|
||||
print('lid_ir min=%s peak=%s baseline=%s peak delta=%s'
|
||||
% (ir_min, ir_peak, IR_BASELINE, delta))
|
||||
print('hv_current range: %s..%s' % (min(hv_vals) if hv_vals else '-',
|
||||
max(hv_vals) if hv_vals else '-'))
|
||||
worst = max(delta)
|
||||
print('worst peak delta this job: %s counts -> cool_fire_ir_delta must sit '
|
||||
'above the worst across ALL jobs (>= 2x it, never < 15)' % worst)
|
||||
print('the engine logged its own "run telemetry: lid IR ..." line for this job')
|
||||
return samples
|
||||
|
||||
|
||||
def post_ctrl(action):
|
||||
# http.client preserves the header-name case exactly as given.
|
||||
import http.client
|
||||
tok = os.environ.get('GF_TOKEN', '')
|
||||
c = http.client.HTTPConnection(HOST, 8080, timeout=8)
|
||||
c.putrequest('POST', '/controller/' + action)
|
||||
c.putheader('X-ForgeFIRM-Token', tok)
|
||||
c.putheader('Content-Length', '0')
|
||||
c.endheaders()
|
||||
r = c.getresponse()
|
||||
body = r.read().decode()
|
||||
c.close()
|
||||
return r.status, body
|
||||
|
||||
|
||||
def drill_expstop(g):
|
||||
print('=== armed kill on the expected-stop path (POST /controller/stop) ===')
|
||||
if not os.environ.get('GF_TOKEN'):
|
||||
raise SystemExit('set GF_TOKEN to the panel token first')
|
||||
print('connect: %s' % prepare(g))
|
||||
arm_cue()
|
||||
job = ['G91', 'G21', 'M4', 'S400',
|
||||
'G1 X40 F200', 'G1 Y40 F200', 'G1 X-40 F200', 'G1 Y-40 F200',
|
||||
'M5', 'G90', 'M2']
|
||||
for ln in job:
|
||||
g.s.sendall(ln.encode() + b'\n')
|
||||
# Wait for the burn to be under way (emission > 0), then stop.
|
||||
t0 = time.time()
|
||||
seen = False
|
||||
while time.time() - t0 < 240:
|
||||
smp = sample_forgectrl()
|
||||
if smp and smp['emission'] and smp['emission'] > 0:
|
||||
seen = True
|
||||
break
|
||||
time.sleep(0.15)
|
||||
if not seen:
|
||||
print('no emission seen within the wait - operator did not arm? ABORT')
|
||||
return []
|
||||
print('emission live (%s) - stopping the controller NOW' % smp['emission'])
|
||||
t_stop = time.time()
|
||||
code, body = post_ctrl('stop')
|
||||
print('POST /controller/stop -> %s %s (%.2f s)' % (code, body.strip(), time.time() - t_stop))
|
||||
trail = []
|
||||
for _ in range(40): # ~5 s at 8 Hz
|
||||
smp = sample_forgectrl()
|
||||
if smp:
|
||||
trail.append((round(time.time() - t_stop, 2), smp['emission'], smp['kstate'], smp['armed']))
|
||||
time.sleep(0.12)
|
||||
print('post-stop trail (t, emission_samples, kstate, armed):')
|
||||
for t in trail:
|
||||
print(' %s' % (t,))
|
||||
zero_at = next((t for t, e, _, _ in trail if e == 0), None)
|
||||
tail_zero = all(e == 0 for _, e, _, _ in trail[-16:])
|
||||
not_running = all(k != 'running' for _, _, k, _ in trail[-16:])
|
||||
print('emission first 0 at +%s s; last 2 s all zero: %s; kernel not running: %s'
|
||||
% (zero_at, tail_zero, not_running))
|
||||
try:
|
||||
mode = get_json('/mode')
|
||||
except Exception as e:
|
||||
mode = str(e)
|
||||
print('/mode after stop: %s' % mode)
|
||||
ok = zero_at is not None and zero_at < 2.5 and tail_zero and not_running
|
||||
print('EXPSTOP %s' % ('PASS' if ok else 'REVIEW'))
|
||||
print('the controller is left STOPPED (supervision held); resume it with '
|
||||
'the ctrlstart step once the operator has judged the stop')
|
||||
return trail
|
||||
|
||||
|
||||
def drill_ctrlstart(g):
|
||||
"""Resume supervision after expstop: POST /controller/start, then
|
||||
report /mode. No motion, no laser."""
|
||||
code, body = post_ctrl('start')
|
||||
print('POST /controller/start -> %s %s' % (code, body.strip()))
|
||||
time.sleep(6)
|
||||
try:
|
||||
print('/mode after start: %s' % get_json('/mode'))
|
||||
except Exception as e:
|
||||
print('/mode after start: %s' % e)
|
||||
return []
|
||||
|
||||
|
||||
def main():
|
||||
drill = sys.argv[1] if len(sys.argv) > 1 else ''
|
||||
drills = {'witness': drill_witness, 'hold': drill_hold,
|
||||
'faultpos': drill_faultpos}
|
||||
'faultpos': drill_faultpos, 'ircut': drill_ircut,
|
||||
'expstop': drill_expstop, 'ctrlstart': drill_ctrlstart}
|
||||
if drill not in drills:
|
||||
print(__doc__)
|
||||
return 2
|
||||
if drill == 'ctrlstart':
|
||||
drills[drill](None)
|
||||
return 0
|
||||
g = Grbl(HOST, PORT)
|
||||
try:
|
||||
drills[drill](g)
|
||||
|
||||
@@ -0,0 +1,309 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Kernel platform drills - runs ON the board, with forgectrl stopped
|
||||
(/etc/init.d/forgectrl stop) so the pulse device is free. Restart forgectrl
|
||||
afterward. Usage: platform_drills.py deadman|rmmod|decay|led|all
|
||||
|
||||
deadman Trip the kernel dead-man mid-run and read back what it touched.
|
||||
Before the run: heater and TEC on, pump/exhaust/intake at a known
|
||||
duty, measure laser + UV LED lit, Z driver enabled. Open the pulse
|
||||
device with flock, motor_lock=15 (nothing moves), stream pads at
|
||||
10 kHz, run, then CLOSE the fd mid-program - the final close is the
|
||||
dead-man trip. PASS: the run stops; heater and TEC are off; the
|
||||
measure laser, UV LED and Z driver are off; pump, exhaust and intake
|
||||
are UNCHANGED (airflow and circulation stay with the engine).
|
||||
rmmod Three rmmod/modprobe cycles while another thread reads cnc/state
|
||||
and cnc/position in a tight loop the whole time. PASS: every cycle
|
||||
completes, the attrs come back, no oops/BUG/WARNING in dmesg.
|
||||
decay Set every decay mode (0/1/2) and microstep mode on each axis and
|
||||
read each back. PASS: readback equals the write, every time.
|
||||
led Drive the button and lid LEDs through target/pulse settings and
|
||||
back to their resting state. Operator confirms visually.
|
||||
"""
|
||||
import errno, fcntl, os, subprocess, sys, threading, time
|
||||
|
||||
TICK_HZ = 10000
|
||||
PAD = b'\x00'
|
||||
POWER0 = bytes([0x80])
|
||||
G = '/sys/glowforge/'
|
||||
LEDS = '/sys/class/leds/'
|
||||
|
||||
|
||||
def wr(attr, val):
|
||||
with open(G + attr, 'w') as f:
|
||||
f.write(str(val))
|
||||
|
||||
|
||||
def rd(attr):
|
||||
with open(G + attr) as f:
|
||||
return f.read().strip()
|
||||
|
||||
|
||||
def sh(cmd):
|
||||
return subprocess.run(cmd, shell=True, capture_output=True, text=True).stdout
|
||||
|
||||
|
||||
def open_pulsedev():
|
||||
try:
|
||||
fd = os.open('/dev/glowforge', os.O_WRONLY)
|
||||
except OSError as e:
|
||||
if e.errno == errno.EBUSY:
|
||||
print('ABORT: /dev/glowforge is busy - stop forgectrl first '
|
||||
'(/etc/init.d/forgectrl stop), then re-run')
|
||||
sys.exit(1)
|
||||
raise
|
||||
fcntl.flock(fd, fcntl.LOCK_EX)
|
||||
return fd
|
||||
|
||||
|
||||
def wait_state(want, timeout, poll=0.05):
|
||||
t0 = time.time()
|
||||
while time.time() - t0 < timeout:
|
||||
s = rd('cnc/state')
|
||||
if s == want:
|
||||
return s
|
||||
time.sleep(poll)
|
||||
return rd('cnc/state')
|
||||
|
||||
|
||||
def dmesg_since(marker_ts):
|
||||
out = sh('dmesg')
|
||||
keep = []
|
||||
for ln in out.splitlines():
|
||||
try:
|
||||
ts = float(ln.split(']')[0].strip('[ '))
|
||||
except (ValueError, IndexError):
|
||||
continue
|
||||
if ts >= marker_ts:
|
||||
keep.append(ln)
|
||||
return keep
|
||||
|
||||
|
||||
def uptime():
|
||||
with open('/proc/uptime') as f:
|
||||
return float(f.read().split()[0])
|
||||
|
||||
|
||||
# ------------------------------------------------------------- deadman
|
||||
|
||||
def drill_deadman():
|
||||
print('=== dead-man trip: what the kernel touches ===')
|
||||
watch = ['thermal/heater_pwm', 'thermal/tec_on', 'thermal/water_pump_on',
|
||||
'thermal/exhaust_pwm', 'thermal/intake_pwm',
|
||||
'head/measure_laser', 'head/uv_led', 'head/z_enable',
|
||||
'head/air_assist_pwm', 'head/white_led']
|
||||
before = {a: rd(a) for a in watch}
|
||||
print('resting: %s' % before)
|
||||
# Known pre-trip posture (all harmless for a few seconds).
|
||||
wr('thermal/heater_pwm', 30)
|
||||
wr('thermal/tec_on', 1)
|
||||
wr('thermal/water_pump_on', 1)
|
||||
wr('thermal/exhaust_pwm', 40)
|
||||
wr('thermal/intake_pwm', 40)
|
||||
wr('head/air_assist_pwm', 40)
|
||||
wr('head/measure_laser', 20)
|
||||
wr('head/uv_led', 20)
|
||||
wr('head/z_enable', 1)
|
||||
time.sleep(0.3)
|
||||
armed = {a: rd(a) for a in watch}
|
||||
print('pre-trip: %s' % armed)
|
||||
|
||||
fd = open_pulsedev()
|
||||
t_mark = uptime()
|
||||
wr('cnc/motor_lock', 15)
|
||||
wr('cnc/laser_latch', 1)
|
||||
wr('cnc/step_freq', TICK_HZ)
|
||||
os.lseek(fd, 1, os.SEEK_SET) # clear ring + counters
|
||||
os.write(fd, POWER0 + PAD * (TICK_HZ * 6)) # 6 s of pads
|
||||
wr('cnc/run', 1)
|
||||
st = wait_state('running', 2)
|
||||
print('run: state=%s' % st)
|
||||
time.sleep(1.0)
|
||||
print('closing the flock\'d fd mid-program (dead-man trip)')
|
||||
os.close(fd)
|
||||
time.sleep(0.5)
|
||||
after = {a: rd(a) for a in watch}
|
||||
print('post-trip: state=%s %s' % (rd('cnc/state'), after))
|
||||
for ln in dmesg_since(t_mark):
|
||||
if 'glowforge' in ln:
|
||||
print(' dmesg: %s' % ln)
|
||||
|
||||
ok = True
|
||||
def expect(attr, val):
|
||||
nonlocal ok
|
||||
got = after[attr]
|
||||
good = got == str(val)
|
||||
ok &= good
|
||||
print(' %-24s %s (want %s) %s' % (attr, got, val, 'ok' if good else 'FAIL'))
|
||||
def unchanged(attr):
|
||||
nonlocal ok
|
||||
good = after[attr] == armed[attr]
|
||||
ok &= good
|
||||
print(' %-24s %s (want unchanged %s) %s'
|
||||
% (attr, after[attr], armed[attr], 'ok' if good else 'FAIL'))
|
||||
print('--- heat sources off:')
|
||||
expect('thermal/heater_pwm', 0)
|
||||
expect('thermal/tec_on', 0)
|
||||
print('--- head safe:')
|
||||
expect('head/measure_laser', 0)
|
||||
expect('head/uv_led', 0)
|
||||
expect('head/z_enable', 0)
|
||||
print('--- airflow and circulation left to the engine:')
|
||||
unchanged('thermal/water_pump_on')
|
||||
unchanged('thermal/exhaust_pwm')
|
||||
unchanged('thermal/intake_pwm')
|
||||
unchanged('head/air_assist_pwm')
|
||||
stopped = rd('cnc/state') != 'running'
|
||||
ok &= stopped
|
||||
print(' run stopped: %s' % stopped)
|
||||
# Restore the resting posture we found.
|
||||
for a, v in before.items():
|
||||
try:
|
||||
wr(a, v)
|
||||
except OSError:
|
||||
pass
|
||||
wr('cnc/motor_lock', 0)
|
||||
print('DEADMAN %s' % ('PASS' if ok else 'FAIL'))
|
||||
return 0 if ok else 1
|
||||
|
||||
|
||||
# --------------------------------------------------------------- rmmod
|
||||
|
||||
def drill_rmmod():
|
||||
print('=== rmmod/modprobe x3 with concurrent attr reads ===')
|
||||
stop = threading.Event()
|
||||
reads = {'n': 0, 'err': 0}
|
||||
|
||||
def reader():
|
||||
while not stop.is_set():
|
||||
for a in ('cnc/state', 'cnc/position', 'cnc/faults', 'head/hall_sensor'):
|
||||
try:
|
||||
with open(G + a, 'rb') as f:
|
||||
f.read(64)
|
||||
reads['n'] += 1
|
||||
except OSError:
|
||||
reads['err'] += 1 # expected while the module is out
|
||||
th = threading.Thread(target=reader, daemon=True)
|
||||
th.start()
|
||||
t_mark = uptime()
|
||||
ok = True
|
||||
for i in range(3):
|
||||
r1 = subprocess.run('rmmod glowforge', shell=True, capture_output=True, text=True)
|
||||
gone = not os.path.exists(G + 'cnc/state')
|
||||
r2 = subprocess.run('modprobe glowforge', shell=True, capture_output=True, text=True)
|
||||
time.sleep(1.0)
|
||||
back = os.path.exists(G + 'cnc/state') and rd('cnc/state') in ('idle', 'disabled')
|
||||
print('cycle %d: rmmod rc=%d %s| unloaded=%s | modprobe rc=%d %s| back=%s state=%s'
|
||||
% (i + 1, r1.returncode, r1.stderr.strip(), gone, r2.returncode,
|
||||
r2.stderr.strip(), back, rd('cnc/state') if back else '-'))
|
||||
ok &= r1.returncode == 0 and gone and r2.returncode == 0 and back
|
||||
stop.set()
|
||||
th.join(2)
|
||||
print('concurrent reads: %d ok, %d refused while unloaded' % (reads['n'], reads['err']))
|
||||
bad = [ln for ln in dmesg_since(t_mark)
|
||||
if any(k in ln for k in ('Oops', 'BUG', 'WARNING', 'Call trace', 'Unable to handle'))]
|
||||
for ln in dmesg_since(t_mark):
|
||||
if 'glowforge' in ln and ('probe' in ln or 'init' in ln or 'remove' in ln):
|
||||
print(' dmesg: %s' % ln)
|
||||
if bad:
|
||||
ok = False
|
||||
print('KERNEL COMPLAINTS:')
|
||||
for ln in bad:
|
||||
print(' ' + ln)
|
||||
print('note: a module reload resets the analog config and the lid LED; the')
|
||||
print(' controller re-applies its analog config at start, relight the lid')
|
||||
print(' LED via /sys/class/leds/lid_led_*/target if wanted')
|
||||
print('RMMOD %s' % ('PASS' if ok else 'FAIL'))
|
||||
return 0 if ok else 1
|
||||
|
||||
|
||||
# --------------------------------------------------------------- decay
|
||||
|
||||
def drill_decay():
|
||||
print('=== decay + microstep mode readback per axis ===')
|
||||
ok = True
|
||||
saved = {a: rd(a) for a in ('cnc/x_decay', 'cnc/y_decay', 'cnc/x_mode', 'cnc/y_mode')}
|
||||
print('resting: %s' % saved)
|
||||
for axis in ('x', 'y'):
|
||||
for v in (0, 1, 2, 1):
|
||||
wr('cnc/%s_decay' % axis, v)
|
||||
got = rd('cnc/%s_decay' % axis)
|
||||
good = got == str(v)
|
||||
ok &= good
|
||||
print(' %s_decay <- %d reads %s %s' % (axis, v, got, 'ok' if good else 'FAIL'))
|
||||
for v in (1, 2, 4, 8, 16, 32):
|
||||
wr('cnc/%s_mode' % axis, v)
|
||||
got = rd('cnc/%s_mode' % axis)
|
||||
good = got == str(v)
|
||||
ok &= good
|
||||
print(' %s_mode <- %-2d reads %s %s' % (axis, v, got, 'ok' if good else 'FAIL'))
|
||||
# Out-of-range writes must be refused, not wrapped.
|
||||
for a, bad in (('%s_decay' % axis, 3), ('%s_mode' % axis, 3)):
|
||||
try:
|
||||
wr('cnc/' + a, bad)
|
||||
print(' %s <- %d ACCEPTED (FAIL: should be refused)' % (a, bad))
|
||||
ok = False
|
||||
except OSError as e:
|
||||
print(' %s <- %d refused (%s) ok' % (a, bad, errno.errorcode.get(e.errno, e.errno)))
|
||||
for a, v in saved.items():
|
||||
wr(a, v)
|
||||
print('restored: %s' % {a: rd(a) for a in saved})
|
||||
print('DECAY %s' % ('PASS' if ok else 'FAIL'))
|
||||
return 0 if ok else 1
|
||||
|
||||
|
||||
# ----------------------------------------------------------------- led
|
||||
|
||||
def led_wr(name, attr, val):
|
||||
with open(LEDS + name + '/' + attr, 'w') as f:
|
||||
f.write(str(val))
|
||||
|
||||
|
||||
def led_rd(name, attr):
|
||||
with open(LEDS + name + '/' + attr) as f:
|
||||
return f.read().strip()
|
||||
|
||||
|
||||
def drill_led():
|
||||
print('=== LED behavior (operator: watch the button and the lid) ===')
|
||||
names = [n for n in os.listdir(LEDS) if n.startswith('button_led_') or n.startswith('lid_led')]
|
||||
names.sort()
|
||||
saved = {n: (led_rd(n, 'target'), led_rd(n, 'pulse_on'), led_rd(n, 'pulse_off')) for n in names}
|
||||
print('leds: %s resting=%s' % (names, saved))
|
||||
print(' all to 255 (bright) for 2 s')
|
||||
for n in names:
|
||||
led_wr(n, 'pulse_on', 0); led_wr(n, 'pulse_off', 0); led_wr(n, 'target', 255)
|
||||
time.sleep(2)
|
||||
print(' all to 0 (dark) for 2 s')
|
||||
for n in names:
|
||||
led_wr(n, 'target', 0)
|
||||
time.sleep(2)
|
||||
print(' button LEDs pulsing 300/300 ms for 4 s')
|
||||
for n in names:
|
||||
if n.startswith('button_led_'):
|
||||
led_wr(n, 'pulse_on', 300); led_wr(n, 'pulse_off', 300)
|
||||
time.sleep(4)
|
||||
print(' restoring')
|
||||
for n, (t, on, off) in saved.items():
|
||||
led_wr(n, 'pulse_on', on); led_wr(n, 'pulse_off', off); led_wr(n, 'target', t)
|
||||
print('LED sequence done - operator verdict: bright / dark / pulse / restored?')
|
||||
return 0
|
||||
|
||||
|
||||
def main():
|
||||
mode = sys.argv[1] if len(sys.argv) > 1 else ''
|
||||
drills = {'deadman': drill_deadman, 'rmmod': drill_rmmod,
|
||||
'decay': drill_decay, 'led': drill_led}
|
||||
if mode == 'all':
|
||||
rc = 0
|
||||
for name in ('decay', 'led', 'deadman', 'rmmod'):
|
||||
rc |= drills[name]()
|
||||
print()
|
||||
return rc
|
||||
if mode not in drills:
|
||||
print(__doc__)
|
||||
return 2
|
||||
return drills[mode]()
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
sys.exit(main())
|
||||
Reference in New Issue
Block a user