Software: pattern player, web page with a design library, GRBL tools
- python3 -m sandtable web: local web page to play designs from a library of 27 built-in patterns, upload .thr or .svg, pause/skip/stop, jog and calibrate. Reconnects by itself when the USB cable comes back. - Plays .thr files and built-in generators with an erase spiral between them, constant ball speed, host-side homing on stock GRBL 1.1 (Uno + CNC shield), FluidNC for the ESP32 board. - Home switch guard: patterns never press the switch, so a press means lost steps and stops the machine at once with an alert. - lineart.py turns SVG line drawings into one continuous path. - Tests against a GRBL simulator. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
1 parent
1400cc8afd
commit
95ad4d25de
52 files changed
+388117
No files matched your search
@@ -0,0 +1,18 @@
|
||||
"""Try the web page without the table: the real design library, driven against the GRBL simulator.
|
||||
Run from the software folder: python3 tests/demo_web.py then open http://127.0.0.1:8766"""
|
||||
import os
|
||||
import sys
|
||||
|
||||
sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..'))
|
||||
sys.path.insert(0, os.path.dirname(__file__))
|
||||
|
||||
from sandtable import config
|
||||
from sandtable.grbl import Controller
|
||||
from sandtable.web import serve
|
||||
from fakegrbl import FakeGrbl
|
||||
|
||||
if __name__ == '__main__':
|
||||
fake = FakeGrbl(speedup=float(os.environ.get('SPEEDUP', '30')), carriage_mm=40.0)
|
||||
c = Controller(fake, 'grbl', log=lambda *a: None)
|
||||
c.wake(wait=0.3)
|
||||
serve(c, config.load(), port=int(os.environ.get('PORT', '8766')), open_browser='--no-browser' not in sys.argv)
|
||||
@@ -0,0 +1,210 @@
|
||||
"""A small GRBL 1.1 simulator that behaves like a serial port, for testing without the table.
|
||||
|
||||
- Lines go into a 128-byte receive buffer; overflowing it raises (that would corrupt a real job).
|
||||
- Lines move into a 15-block planner and are acknowledged with 'ok' as they do on GRBL.
|
||||
- G1 / $J moves run in simulated time (speed-up factor), so status reports show motion.
|
||||
- A home switch closes when the carriage's real position is at or below `switch_at`.
|
||||
- A reset (Ctrl-X) while moving leaves it in ALARM, as on GRBL; $X clears it.
|
||||
"""
|
||||
import math
|
||||
import re
|
||||
import threading
|
||||
import time
|
||||
|
||||
WORD = re.compile(r'([A-Z])(-?\d*\.?\d+)')
|
||||
|
||||
|
||||
class FakeGrbl:
|
||||
RX = 128
|
||||
PLANNER = 15
|
||||
|
||||
def __init__(self, speedup=50.0, carriage_mm=60.0, switch_at=2.0, jog_speedup=None):
|
||||
self.speedup = speedup
|
||||
# jogs (homing) run at most 10x real time, so the switch is found as precisely as on the
|
||||
# table (at 100x the simulated carriage overshoots it by millimetres and fakes a trip)
|
||||
self.jog_speedup = min(speedup, 10.0) if jog_speedup is None else jog_speedup
|
||||
self.out = bytearray()
|
||||
self.rx = bytearray()
|
||||
self.planner = [] # [kind, target(x,y), feed mm/min]
|
||||
self.mpos = [0.0, 0.0]
|
||||
self.wco = [0.0, 0.0]
|
||||
self.offset_true = carriage_mm # real rho = mpos y + offset_true
|
||||
self.switch_at = switch_at
|
||||
self.state = 'Idle'
|
||||
self.hold = False
|
||||
self.alarm = False
|
||||
self.unplugged = False # set to pretend the USB cable came out
|
||||
self.absolute = True
|
||||
self.settings = {'$3': '0', '$100': '250.000', '$101': '250.000', '$110': '500', '$111': '500'}
|
||||
self.max_rx = 0
|
||||
self.lines = [] # every line received, in order
|
||||
self.lock = threading.Lock()
|
||||
self.alive = True
|
||||
self.t_last = time.monotonic()
|
||||
self.th = threading.Thread(target=self._run, daemon=True)
|
||||
self.th.start()
|
||||
self._emit('Grbl 1.1h [\'$\' for help]')
|
||||
|
||||
# ---------------------------------------------------------- serial-port interface
|
||||
@property
|
||||
def in_waiting(self):
|
||||
if self.unplugged:
|
||||
raise OSError(5, 'Input/output error')
|
||||
with self.lock:
|
||||
return len(self.out)
|
||||
|
||||
def read(self, n=1):
|
||||
if self.unplugged:
|
||||
raise OSError(5, 'Input/output error')
|
||||
t_end = time.monotonic() + 0.02
|
||||
while True:
|
||||
with self.lock:
|
||||
if self.out:
|
||||
data = bytes(self.out[:n])
|
||||
del self.out[:n]
|
||||
return data
|
||||
if time.monotonic() > t_end:
|
||||
return b''
|
||||
time.sleep(0.001)
|
||||
|
||||
def write(self, data):
|
||||
if self.unplugged:
|
||||
raise OSError(5, 'write failed: Input/output error')
|
||||
with self.lock:
|
||||
for b in data:
|
||||
if b == ord('?'):
|
||||
self._status()
|
||||
elif b == ord('!'):
|
||||
self.hold = True
|
||||
elif b == ord('~'):
|
||||
self.hold = False
|
||||
elif b == 0x18:
|
||||
moving = bool(self.planner) and not self.hold
|
||||
if moving:
|
||||
self.alarm = True
|
||||
self._emit('ALARM:3') # reset while in motion: position lost
|
||||
self.planner.clear()
|
||||
self.rx.clear()
|
||||
self.hold = False
|
||||
self.wco = [0.0, 0.0] # GRBL clears the G92 offset on reset
|
||||
self.state = 'Alarm' if self.alarm else 'Idle'
|
||||
self._emit('Grbl 1.1h [\'$\' for help]')
|
||||
if self.alarm:
|
||||
self._emit("[MSG:'$H'|'$X' to unlock]")
|
||||
elif b == 0x85:
|
||||
self.planner = [p for p in self.planner if p[0] != 'jog']
|
||||
else:
|
||||
self.rx.append(b)
|
||||
self.max_rx = max(self.max_rx, len(self.rx))
|
||||
if len(self.rx) > self.RX:
|
||||
raise AssertionError('GRBL receive buffer overflow')
|
||||
return len(data)
|
||||
|
||||
def reset_input_buffer(self):
|
||||
with self.lock:
|
||||
self.out.clear()
|
||||
|
||||
def close(self):
|
||||
self.alive = False
|
||||
|
||||
# ---------------------------------------------------------- simulation
|
||||
def _emit(self, s):
|
||||
self.out += (s + '\r\n').encode()
|
||||
|
||||
@property
|
||||
def pins(self):
|
||||
return 'Y' if self.mpos[1] + self.offset_true <= self.switch_at else ''
|
||||
|
||||
def _status(self):
|
||||
st = self.state
|
||||
if self.hold:
|
||||
st = 'Hold:0' # the simulator stops at once
|
||||
pn = f'|Pn:{self.pins}' if self.pins else ''
|
||||
self._emit(f'<{st}|MPos:{self.mpos[0]:.3f},{self.mpos[1]:.3f},0.000|FS:0,0|WCO:'
|
||||
f'{self.wco[0]:.3f},{self.wco[1]:.3f},0.000{pn}>')
|
||||
|
||||
def _take_line(self):
|
||||
if b'\n' not in self.rx or len(self.planner) >= self.PLANNER:
|
||||
return
|
||||
raw, _, rest = bytes(self.rx).partition(b'\n')
|
||||
self.rx = bytearray(rest)
|
||||
line = raw.decode().strip().upper()
|
||||
self.lines.append(line)
|
||||
line = re.sub(r'\(.*?\)', '', line).replace(' ', '')
|
||||
if not line:
|
||||
self._emit('ok')
|
||||
return
|
||||
if line == '$$':
|
||||
for k, v in self.settings.items():
|
||||
self._emit(f'{k}={v}')
|
||||
elif line.startswith('$J='):
|
||||
body = line[3:]
|
||||
w = dict(WORD.findall(body.replace('G91', 'g').replace('G21', 'g').replace('G90', 'g')))
|
||||
rel = 'G91' in body
|
||||
tgt = list(self.mpos)
|
||||
for i, ax in enumerate('XY'):
|
||||
if ax in w:
|
||||
tgt[i] = (self.mpos[i] + float(w[ax])) if rel else float(w[ax]) + self.wco[i]
|
||||
self.planner.append(['jog', tgt, float(w.get('F', 600))])
|
||||
elif line.startswith('$X'):
|
||||
self.alarm = False
|
||||
self.state = 'Idle'
|
||||
elif self.alarm and not line.startswith('$'):
|
||||
self._emit('error:9') # G-code is locked out during an alarm
|
||||
return
|
||||
elif line.startswith('$') and '=' in line:
|
||||
k, v = line.split('=', 1)
|
||||
self.settings[k] = v
|
||||
elif line.startswith('G92'):
|
||||
w = dict(WORD.findall(line[3:]))
|
||||
for i, ax in enumerate('XY'):
|
||||
if ax in w:
|
||||
self.wco[i] = self.mpos[i] - float(w[ax])
|
||||
elif line.startswith('G10L20P1'):
|
||||
w = dict(WORD.findall(line[8:]))
|
||||
for i, ax in enumerate('XY'):
|
||||
if ax in w:
|
||||
self.wco[i] = self.mpos[i] - float(w[ax])
|
||||
else:
|
||||
if 'G91' in line:
|
||||
self.absolute = False
|
||||
if 'G90' in line:
|
||||
self.absolute = True
|
||||
w = dict(WORD.findall(re.sub(r'G\d+', '', line)))
|
||||
if 'X' in w or 'Y' in w:
|
||||
base = self.planner[-1][1] if self.planner else self.mpos
|
||||
tgt = list(base)
|
||||
for i, ax in enumerate('XY'):
|
||||
if ax in w:
|
||||
tgt[i] = (base[i] + float(w[ax])) if not self.absolute else float(w[ax]) + self.wco[i]
|
||||
self.planner.append(['move', tgt, float(w.get('F', 600))])
|
||||
self._emit('ok')
|
||||
|
||||
def _run(self):
|
||||
while self.alive:
|
||||
time.sleep(0.002)
|
||||
now = time.monotonic()
|
||||
with self.lock:
|
||||
real_dt = now - self.t_last
|
||||
dt = real_dt * self.speedup
|
||||
self.t_last = now
|
||||
self._take_line()
|
||||
if self.alarm:
|
||||
self.state = 'Alarm'
|
||||
continue
|
||||
if self.hold or not self.planner:
|
||||
self.state = 'Idle' if not self.planner else self.state
|
||||
continue
|
||||
kind, tgt, feed = self.planner[0]
|
||||
self.state = 'Jog' if kind == 'jog' else 'Run'
|
||||
if kind == 'jog':
|
||||
dt = real_dt * self.jog_speedup
|
||||
dx, dy = tgt[0] - self.mpos[0], tgt[1] - self.mpos[1]
|
||||
d = math.hypot(dx, dy)
|
||||
step = feed / 60.0 * dt
|
||||
if d <= step or d == 0:
|
||||
self.mpos = list(tgt)
|
||||
self.planner.pop(0)
|
||||
else:
|
||||
self.mpos[0] += dx / d * step
|
||||
self.mpos[1] += dy / d * step
|
||||
@@ -0,0 +1,103 @@
|
||||
"""Line drawings (SVG and the fan-art designs) joined into one sand-table path.
|
||||
Run from the software folder: python3 -m unittest discover -s tests -v"""
|
||||
import math
|
||||
import os
|
||||
import sys
|
||||
import tempfile
|
||||
import unittest
|
||||
|
||||
os.environ['SANDTABLE_QUIET'] = '1'
|
||||
sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..'))
|
||||
sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..', 'tools'))
|
||||
|
||||
from sandtable import config, thr
|
||||
from sandtable.lineart import drawing_to_polar, join_strokes, path_polylines, svg_strokes
|
||||
|
||||
SVG = """<svg xmlns="http://www.w3.org/2000/svg" viewBox="0 0 100 100">
|
||||
<defs><circle cx="0" cy="0" r="99"/></defs>
|
||||
<rect x="10" y="10" width="20" height="20"/>
|
||||
<g transform="translate(50 50) scale(2)"><circle cx="0" cy="0" r="10"/></g>
|
||||
<path d="M10 80 C20 70 30 90 40 80 Q50 70 60 80 A10 10 0 0 1 80 80 L90 90 h-5 v5 z"/>
|
||||
<text x="0" y="0">ignored</text>
|
||||
</svg>"""
|
||||
|
||||
|
||||
def unit_xy(pts):
|
||||
return [(r * math.cos(t), r * math.sin(t)) for t, r in pts]
|
||||
|
||||
|
||||
class TestSvg(unittest.TestCase):
|
||||
def test_shapes_and_transforms(self):
|
||||
s = svg_strokes(SVG)
|
||||
self.assertEqual(len(s), 3) # defs and text are skipped
|
||||
circle = s[1]
|
||||
self.assertAlmostEqual(max(p[0] for p in circle), 70, delta=0.1) # r 10 scaled 2x, moved to 50
|
||||
self.assertAlmostEqual(min(p[0] for p in circle), 30, delta=0.1)
|
||||
path = s[2]
|
||||
self.assertAlmostEqual(math.dist(path[0], path[-1]), 0, places=6) # z closes it
|
||||
self.assertAlmostEqual(min(p[1] for p in path), 70, delta=0.6) # the arc's top (centre 70,80, r 10)
|
||||
|
||||
def test_arc_and_relative(self):
|
||||
(semi,) = path_polylines('M0 0 a10 10 0 0 1 20 0', tol=0.1)
|
||||
self.assertAlmostEqual(min(p[1] for p in semi), -10, delta=0.05) # sweep 1 bulges up the page (y down)
|
||||
lines = path_polylines('m0 0 l10 0 10 0 M0 5 H20 V9')
|
||||
self.assertEqual(lines[0][-1], (20.0, 0.0)) # implicit repeated l
|
||||
self.assertEqual(lines[1][-1], (20.0, 9.0))
|
||||
|
||||
|
||||
class TestJoin(unittest.TestCase):
|
||||
def check_path(self, xy, strokes):
|
||||
for a, b in zip(xy, xy[1:]):
|
||||
self.assertLess(math.dist(a, b), 0.008) # one continuous line
|
||||
for end in (xy[0], xy[-1]):
|
||||
r = math.hypot(*end)
|
||||
self.assertTrue(r < 1e-9 or abs(r - 1) < 1e-9, end) # starts and ends at the centre or the edge
|
||||
for st in strokes: # every stroke is drawn
|
||||
for p in st[::7]:
|
||||
self.assertLess(min(math.dist(p, q) for q in xy), 0.01)
|
||||
|
||||
def test_triangle_from_three_lines_needs_no_jumps(self):
|
||||
a, b, c = (0.0, 0.85), (-0.74, -0.42), (0.74, -0.42)
|
||||
strokes = [[a, b], [c, b], [c, a]]
|
||||
stats = {}
|
||||
xy = join_strokes(strokes, stats=stats)
|
||||
self.check_path(xy, strokes)
|
||||
self.assertLess(stats['jump'], 0.35) # only the short lines in from and out to the edge
|
||||
|
||||
def test_separate_shapes(self):
|
||||
sq = lambda x, y, d: [(x, y), (x + d, y), (x + d, y + d), (x, y + d), (x, y)]
|
||||
strokes = [sq(-0.6, -0.6, 0.3), sq(0.3, 0.3, 0.3), sq(0.3, -0.6, 0.3), [(-0.5, 0.5), (-0.2, 0.2)]]
|
||||
xy = join_strokes(strokes)
|
||||
self.check_path(xy, strokes)
|
||||
|
||||
def test_fan_art_designs(self):
|
||||
try:
|
||||
import make_fan_art # not in the public repository (trademarks)
|
||||
except ImportError:
|
||||
self.skipTest('tools/make_fan_art.py is not here')
|
||||
for name, (fn, opts) in make_fan_art.DESIGNS.items():
|
||||
pts = drawing_to_polar(fn(), **opts)
|
||||
self.assertGreater(len(pts), 100, name)
|
||||
for end in (pts[0][1], pts[-1][1]):
|
||||
self.assertTrue(end < 0.02 or end > 0.98, f'{name} ends at rho {end}')
|
||||
xy = unit_xy(pts)
|
||||
for a, b in zip(xy, xy[1:]):
|
||||
self.assertLess(math.dist(a, b), 0.06, name)
|
||||
|
||||
|
||||
class TestUpload(unittest.TestCase):
|
||||
def test_svg_upload_becomes_a_design(self):
|
||||
from sandtable.web import Library
|
||||
with tempfile.TemporaryDirectory() as d:
|
||||
lib = Library(config.load('/nonexistent.toml'), root=os.path.join(d, 'p'), cache=os.path.join(d, 'c'),
|
||||
removed=os.path.join(d, 'r'))
|
||||
ident = lib.add('My Logo.svg', SVG)
|
||||
self.assertEqual(ident, 'uploads/My-Logo.thr')
|
||||
pts = thr.load(lib.path(ident))
|
||||
self.assertGreater(len(pts), 100)
|
||||
with self.assertRaises(ValueError):
|
||||
lib.add('empty.svg', '<svg xmlns="http://www.w3.org/2000/svg"><text>hi</text></svg>')
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
unittest.main()
|
||||
@@ -0,0 +1,247 @@
|
||||
"""Run from the software folder: python3 -m unittest discover -s tests -v"""
|
||||
import math
|
||||
import os
|
||||
import sys
|
||||
import tempfile
|
||||
import time
|
||||
import unittest
|
||||
|
||||
os.environ['SANDTABLE_QUIET'] = '1' # never pop up desktop notifications from the tests
|
||||
|
||||
sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..'))
|
||||
sys.path.insert(0, os.path.dirname(__file__))
|
||||
|
||||
from sandtable import config, patterns, thr
|
||||
from sandtable.grbl import Controller, SwitchTripped
|
||||
from sandtable.motion import Planner, phys_len
|
||||
from fakegrbl import FakeGrbl
|
||||
|
||||
CFG = config.load('/nonexistent.toml') # defaults only
|
||||
|
||||
|
||||
class TestThr(unittest.TestCase):
|
||||
def test_parse(self):
|
||||
pts = thr.loads('# comment\n// also\n0 0\n1.5, 0.5\nbad line\n6.3 1.2\n')
|
||||
self.assertEqual(pts, [(0.0, 0.0), (1.5, 0.5), (6.3, 1.0)]) # rho clamped to 1
|
||||
|
||||
def test_roundtrip(self):
|
||||
pts = patterns.spiral(loops=2)
|
||||
with tempfile.TemporaryDirectory() as d:
|
||||
p = os.path.join(d, 'a.thr')
|
||||
thr.save(p, pts, header='test')
|
||||
back = thr.load(p)
|
||||
self.assertEqual(len(back), len(pts))
|
||||
self.assertAlmostEqual(back[-1][0], pts[-1][0], places=4)
|
||||
|
||||
|
||||
class TestPatterns(unittest.TestCase):
|
||||
def test_all_generators(self):
|
||||
for name in patterns.GENERATORS:
|
||||
pts = patterns.generate(name)
|
||||
self.assertGreater(len(pts), 10, name)
|
||||
rhos = [r for _, r in pts]
|
||||
self.assertTrue(all(0 <= r <= 1 for r in rhos), name)
|
||||
for end in (rhos[0], rhos[-1]): # start and finish at the centre or the edge
|
||||
self.assertTrue(end < 0.02 or end > 0.98, f'{name} ends at rho {end}')
|
||||
for (t1, r1), (t2, r2) in zip(pts, pts[1:]): # no jumps across the sand
|
||||
gap = math.hypot(r1 * math.cos(t1) - r2 * math.cos(t2), r1 * math.sin(t1) - r2 * math.sin(t2))
|
||||
self.assertLess(gap, 0.06, name)
|
||||
|
||||
def test_lissajous_copies_join(self):
|
||||
pts = patterns.generate('lissajous', copies='3', twist_deg='20')
|
||||
for (t1, r1), (t2, r2) in zip(pts, pts[1:]):
|
||||
gap = math.hypot(r1 * math.cos(t1) - r2 * math.cos(t2), r1 * math.sin(t1) - r2 * math.sin(t2))
|
||||
self.assertLess(gap, 0.06)
|
||||
self.assertGreater(abs(pts[-1][1]), 0.98)
|
||||
|
||||
def test_params(self):
|
||||
self.assertGreater(len(patterns.generate('rose', petals='7')), 100)
|
||||
with self.assertRaises(KeyError):
|
||||
patterns.generate('rose', nope=1)
|
||||
|
||||
|
||||
class TestMotion(unittest.TestCase):
|
||||
def test_feed_keeps_ball_speed(self):
|
||||
pl = Planner(CFG)
|
||||
# 1 degree at rho 100 mm is 1.745 mm on the sand: at 25 mm/s that is 1 deg per 0.07 s
|
||||
f = pl.feed(0, 100, 1, 100, 25)
|
||||
self.assertAlmostEqual(f, 60 * 25 / (100 * math.pi / 180), delta=0.5)
|
||||
# pure rho move: machine units are mm, so F = speed * 60
|
||||
self.assertAlmostEqual(pl.feed(0, 10, 0, 20, 25), 1500, delta=0.1)
|
||||
|
||||
def test_densify_limits(self):
|
||||
pl = Planner(CFG)
|
||||
pts = pl.densify([(0, 0), (720, 133)])
|
||||
for (t1, r1), (t2, r2) in zip(pts, pts[1:]):
|
||||
self.assertLessEqual(phys_len(t1, r1, t2, r2), CFG['motion']['max_segment_mm'] + 1e-6)
|
||||
self.assertLessEqual(abs(t2 - t1), CFG['motion']['max_segment_deg'] + 1e-6)
|
||||
|
||||
def test_erase_single_leg(self):
|
||||
pl = Planner(CFG, 0.0, 2.0)
|
||||
path, shift = pl.erase_path(37.0, 133.0, 7.0)
|
||||
self.assertEqual(len(path), 2)
|
||||
end_t, end_r = path[-1]
|
||||
self.assertAlmostEqual(end_r, 133.0)
|
||||
self.assertAlmostEqual((end_t - 37.0) % 360, 0, places=6)
|
||||
self.assertAlmostEqual(abs(end_t) / 360, (133 - 2) / 7, delta=0.6) # about pitch-spaced
|
||||
|
||||
def test_erase_unwinds(self):
|
||||
pl = Planner(CFG, 50000.0, 133.0) # arm has turned a lot one way
|
||||
path, _ = pl.erase_path(10.0, 133.0, 7.0)
|
||||
rmin = CFG['machine']['rho_min_mm']
|
||||
self.assertEqual([round(r, 3) for _, r in path], [133, rmin, 133]) # in to the floor and back out
|
||||
self.assertLess(path[-1][0], 50000.0)
|
||||
|
||||
def test_never_inside_floor(self):
|
||||
# patterns through the centre stop at rho_min, so they never press the home switch
|
||||
rmin = CFG['machine']['rho_min_mm']
|
||||
pl = Planner(CFG, 0.0, rmin) # (just after homing it starts at 2 mm and only moves out)
|
||||
for gen in ('rose', 'test_f', 'polygons'):
|
||||
lines, _, _ = pl.program(patterns.generate(gen))
|
||||
ys = [float(l.split('Y')[1].split('F')[0]) for l in lines if l.startswith('G1')]
|
||||
self.assertGreaterEqual(min(ys), rmin - 1e-9, gen)
|
||||
self.assertLess(min(ys), rmin + 0.01, gen) # and still reach it
|
||||
|
||||
def test_program_continuity(self):
|
||||
pl = Planner(CFG, 0.0, 2.0)
|
||||
lines, segs, est = pl.program(patterns.rose(petals=3))
|
||||
self.assertTrue(lines[0].startswith('(---'))
|
||||
self.assertGreater(est, 60)
|
||||
# the pattern part starts where the erase ended
|
||||
self.assertEqual(segs[0][1][-1], segs[1][1][0])
|
||||
# and the planner now knows where the ball is
|
||||
self.assertEqual((pl.theta, pl.rho), segs[-1][1][-1])
|
||||
|
||||
|
||||
class TestController(unittest.TestCase):
|
||||
def make(self, **kw):
|
||||
fake = FakeGrbl(**kw)
|
||||
c = Controller(fake, 'grbl', log=lambda *a: None)
|
||||
c.wake(wait=0.3)
|
||||
return fake, c
|
||||
|
||||
def test_stream_never_overflows(self):
|
||||
fake, c = self.make(speedup=400)
|
||||
pl = Planner(CFG, 0.0, 0.0)
|
||||
lines, _, _ = pl.program(patterns.spiral(loops=3), erase=False)
|
||||
lines = [l for l in lines if not l.startswith('(')]
|
||||
c.stream(lines)
|
||||
self.assertLessEqual(fake.max_rx, 128)
|
||||
got = [l for l in fake.lines if l.startswith('G1')]
|
||||
self.assertEqual(got, [l.upper() for l in lines])
|
||||
st = c.query()
|
||||
self.assertAlmostEqual(st.wpos[0], pl.theta, places=2)
|
||||
self.assertAlmostEqual(st.wpos[1], pl.rho, places=2)
|
||||
fake.close()
|
||||
|
||||
def test_homing(self):
|
||||
fake, c = self.make(speedup=8, carriage_mm=40.0, switch_at=2.0)
|
||||
st = c.home_rho(CFG, log=lambda *a: None)
|
||||
self.assertAlmostEqual(st.wpos[1], CFG['homing']['switch_rho_mm'], places=2)
|
||||
real = fake.mpos[1] + fake.offset_true # where the carriage really is
|
||||
self.assertLess(abs(real - fake.switch_at), 0.3)
|
||||
fake.close()
|
||||
|
||||
def test_homing_from_pressed_switch(self):
|
||||
fake, c = self.make(speedup=8, carriage_mm=1.0, switch_at=2.0)
|
||||
st = c.home_rho(CFG, log=lambda *a: None)
|
||||
self.assertAlmostEqual(st.wpos[1], CFG['homing']['switch_rho_mm'], places=2)
|
||||
fake.close()
|
||||
|
||||
def test_guard_quiet_on_a_normal_run(self):
|
||||
# homing leaves the switch pressed; moving off it and drawing must not trip the guard
|
||||
fake, c = self.make(speedup=60, carriage_mm=30.0, switch_at=2.0)
|
||||
c.home_rho(CFG, log=lambda *a: None)
|
||||
c.arm_guard(CFG)
|
||||
st = c.query()
|
||||
pl = Planner(CFG, st.wpos[0], st.wpos[1])
|
||||
for d in ('out', 'in'):
|
||||
lines, _, _ = pl.program(patterns.spiral(loops=2, direction=d), erase=False)
|
||||
c.stream([l for l in lines if not l.startswith('(')])
|
||||
self.assertEqual(c.guard, 'armed')
|
||||
self.assertFalse(fake.alarm)
|
||||
fake.close()
|
||||
|
||||
def test_guard_stops_a_lost_carriage(self):
|
||||
fake, c = self.make(speedup=3, carriage_mm=30.0, switch_at=2.0)
|
||||
c.home_rho(CFG, log=lambda *a: None)
|
||||
c.arm_guard(CFG)
|
||||
c.stream(['G1X0Y60F6000'])
|
||||
with fake.lock:
|
||||
fake.offset_true -= 50.0 # belt skipped: the carriage is 50 mm further in than GRBL thinks
|
||||
with self.assertRaises(SwitchTripped) as cm:
|
||||
c.stream(['G1X0Y5F600'])
|
||||
self.assertIn('lost its place', str(cm.exception))
|
||||
y = float(str(cm.exception).split('rho ')[1].split(' mm')[0])
|
||||
self.assertTrue(40 < y < 53, y) # caught soon after the switch closed (rho 52)
|
||||
time.sleep(0.1)
|
||||
self.assertTrue(fake.alarm) # stopped by a reset while moving
|
||||
self.assertEqual(fake.planner, [])
|
||||
fake.close()
|
||||
|
||||
def test_guard_spots_a_stuck_switch(self):
|
||||
fake, c = self.make(speedup=10, carriage_mm=30.0, switch_at=2.0)
|
||||
c.home_rho(CFG, log=lambda *a: None)
|
||||
with fake.lock:
|
||||
fake.switch_at = 1000.0 # switch now reads pressed everywhere
|
||||
c.arm_guard(CFG)
|
||||
with self.assertRaises(SwitchTripped) as cm:
|
||||
c.stream(['G1X0Y60F1200'])
|
||||
self.assertIn('stuck', str(cm.exception))
|
||||
fake.close()
|
||||
|
||||
def test_stop_keeps_position(self):
|
||||
fake, c = self.make(speedup=5)
|
||||
c.command('G92 X0 Y10') # as homing leaves it; a reset clears this
|
||||
c.command('G1X0Y100F6000')
|
||||
time.sleep(0.2)
|
||||
st = c.stop()
|
||||
y = st.wpos[1]
|
||||
self.assertGreater(y, 0)
|
||||
self.assertEqual(c.query().state, 'Idle')
|
||||
self.assertAlmostEqual(c.query().wpos[1], y, places=3)
|
||||
fake.close()
|
||||
|
||||
|
||||
class TestCommands(unittest.TestCase):
|
||||
"""The real commands, end to end, against the simulator instead of a serial port."""
|
||||
|
||||
def setUp(self):
|
||||
from sandtable import grbl
|
||||
self.fakes = []
|
||||
orig = grbl.Controller.open
|
||||
|
||||
def fake_open(device, baud=115200, kind='grbl', log=print):
|
||||
f = FakeGrbl(speedup=150, carriage_mm=30.0) # faster and the simulated homing overshoots
|
||||
self.fakes.append(f)
|
||||
c = grbl.Controller(f, kind, log=lambda *a: None)
|
||||
c.wake(wait=0.2)
|
||||
return c
|
||||
grbl.Controller.open = staticmethod(fake_open)
|
||||
self.addCleanup(setattr, grbl.Controller, 'open', orig)
|
||||
|
||||
def run_cli(self, *argv):
|
||||
from sandtable import cli
|
||||
import contextlib
|
||||
import io
|
||||
buf = io.StringIO()
|
||||
with contextlib.redirect_stdout(buf):
|
||||
cli.main(['--port', 'sim', *argv])
|
||||
return buf.getvalue()
|
||||
|
||||
def test_setup_writes_settings(self):
|
||||
out = self.run_cli('setup', '--yes')
|
||||
self.assertIn('written', out)
|
||||
self.assertEqual(self.fakes[-1].settings['$100'], '71.111')
|
||||
self.assertEqual(self.fakes[-1].settings['$22'], '0')
|
||||
|
||||
def test_play_generated_pattern(self):
|
||||
self.run_cli('play', 'gen:spiral:loops=2', '--speed', '40')
|
||||
f = self.fakes[-1]
|
||||
self.assertTrue(any(l.startswith('$J=') for l in f.lines)) # homed first
|
||||
self.assertGreater(sum(1 for l in f.lines if l.startswith('G1X')), 100)
|
||||
self.assertLessEqual(f.max_rx, 128)
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
unittest.main()
|
||||
@@ -0,0 +1,246 @@
|
||||
"""The web app's library, player and HTTP API, against the GRBL simulator.
|
||||
Run from the software folder: python3 -m unittest discover -s tests -v"""
|
||||
import json
|
||||
import os
|
||||
import socket
|
||||
import sys
|
||||
import tempfile
|
||||
import threading
|
||||
import time
|
||||
import unittest
|
||||
|
||||
os.environ['SANDTABLE_QUIET'] = '1' # never pop up desktop notifications from the tests
|
||||
import urllib.request
|
||||
|
||||
sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..'))
|
||||
sys.path.insert(0, os.path.dirname(__file__))
|
||||
|
||||
from sandtable import config, patterns, thr
|
||||
from sandtable.grbl import Controller
|
||||
from sandtable.ui import Panel
|
||||
from sandtable.web import Library, Player, serve
|
||||
from fakegrbl import FakeGrbl
|
||||
|
||||
CFG = config.load('/nonexistent.toml')
|
||||
|
||||
|
||||
def make_library(tmp):
|
||||
root = os.path.join(tmp, 'patterns')
|
||||
os.makedirs(os.path.join(root, 'spirals'))
|
||||
os.makedirs(os.path.join(root, 'flowers'))
|
||||
thr.save(os.path.join(root, 'spirals', 'tiny.thr'), patterns.spiral(loops=2))
|
||||
thr.save(os.path.join(root, 'flowers', 'rose.thr'), patterns.rose(petals=3))
|
||||
return Library(CFG, root=root, cache=os.path.join(tmp, 'cache'), removed=os.path.join(tmp, 'removed'))
|
||||
|
||||
|
||||
def make_controller(speedup):
|
||||
fake = FakeGrbl(speedup=speedup, carriage_mm=30.0, switch_at=2.0)
|
||||
c = Controller(fake, 'grbl', log=lambda *a: None)
|
||||
c.wake(wait=0.3)
|
||||
return fake, c
|
||||
|
||||
|
||||
def wait_for(cond, timeout=30):
|
||||
t_end = time.monotonic() + timeout
|
||||
while time.monotonic() < t_end:
|
||||
if cond():
|
||||
return True
|
||||
time.sleep(0.05)
|
||||
return False
|
||||
|
||||
|
||||
class TestLibrary(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.tmp = tempfile.TemporaryDirectory()
|
||||
self.lib = make_library(self.tmp.name)
|
||||
|
||||
def tearDown(self):
|
||||
self.tmp.cleanup()
|
||||
|
||||
def test_items_and_thumbs(self):
|
||||
items = self.lib.items()
|
||||
self.assertEqual([d['id'] for d in items], ['flowers/rose.thr', 'spirals/tiny.thr'])
|
||||
png = self.lib.thumb('spirals/tiny.thr')
|
||||
self.assertTrue(os.path.getsize(png) > 1000)
|
||||
self.assertGreater(self.lib.items()[1]['est'], 0) # running time known once the thumbnail is made
|
||||
|
||||
def test_refuses_paths_outside(self):
|
||||
for bad in ('../cache/index.json', '../../tiny.thr', '/etc/passwd', 'spirals/../../x.thr', 'spirals/nope.thr'):
|
||||
with self.assertRaises(KeyError):
|
||||
self.lib.path(bad)
|
||||
|
||||
def test_upload_and_remove(self):
|
||||
ident = self.lib.add('My Design (1).thr', '# sandify\n0 0\n3.14 1\n')
|
||||
self.assertEqual(ident, 'uploads/My-Design-1.thr')
|
||||
self.assertEqual(self.lib.add('My Design (1).thr', '0 0\n1 1\n'), 'uploads/My-Design-1-2.thr')
|
||||
with self.assertRaises(ValueError):
|
||||
self.lib.add('junk.thr', 'hello\nworld\n')
|
||||
self.lib.remove(ident)
|
||||
self.assertNotIn(ident, [d['id'] for d in self.lib.items()])
|
||||
self.assertTrue(os.path.exists(os.path.join(self.tmp.name, 'removed', ident)))
|
||||
|
||||
|
||||
class TestPlayer(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.tmp = tempfile.TemporaryDirectory()
|
||||
self.lib = make_library(self.tmp.name)
|
||||
|
||||
def tearDown(self):
|
||||
self.tmp.cleanup()
|
||||
|
||||
def make(self, speedup):
|
||||
self.fake, self.c = make_controller(speedup)
|
||||
self.panel = Panel(self.c, CFG)
|
||||
self.alerts = []
|
||||
self.player = Player(self.panel, CFG, self.lib, alert=lambda t, m: self.alerts.append(m))
|
||||
self.player.speed = 40
|
||||
|
||||
def test_plays_to_the_end(self):
|
||||
self.make(100)
|
||||
self.player.play(['spirals/tiny.thr'])
|
||||
self.assertTrue(wait_for(lambda: not self.player.running(), 60))
|
||||
self.assertEqual(self.player.state, 'idle')
|
||||
self.assertEqual(self.player.message, '')
|
||||
self.assertTrue(self.panel.homed)
|
||||
self.assertLessEqual(self.fake.max_rx, 128)
|
||||
self.assertTrue(any(l.startswith('$J=') for l in self.fake.lines)) # homed first
|
||||
self.fake.close()
|
||||
|
||||
def test_stop_keeps_the_position(self):
|
||||
self.make(20)
|
||||
self.player.play(['flowers/rose.thr'])
|
||||
self.assertTrue(wait_for(lambda: self.player.now and self.player.now['frac'] > 0.02, 60))
|
||||
self.player.stop()
|
||||
self.assertFalse(self.player.running())
|
||||
self.assertEqual(self.player.state, 'idle')
|
||||
st = self.c.query()
|
||||
self.assertEqual(st.state, 'Idle')
|
||||
self.assertFalse(self.fake.alarm)
|
||||
real = self.fake.mpos[1] + self.fake.offset_true # where the carriage really is
|
||||
# still homed: rho matches the carriage to within the simulator's homing overshoot. Without
|
||||
# putting the G92 offset back after the reset it would be ~30 mm out.
|
||||
self.assertAlmostEqual(st.wpos[1], real, delta=0.2)
|
||||
self.assertTrue(self.panel.homed)
|
||||
self.fake.close()
|
||||
|
||||
def test_pause_resume_and_skip(self):
|
||||
self.make(20)
|
||||
self.player.play(['flowers/rose.thr', 'spirals/tiny.thr'])
|
||||
self.assertTrue(wait_for(lambda: self.player.now and self.player.now['name'] == 'rose' and self.player.now['frac'] > 0.01, 60))
|
||||
self.player.pause()
|
||||
self.assertEqual(self.player.state, 'paused')
|
||||
time.sleep(0.3)
|
||||
y = self.c.status.mpos[1]
|
||||
time.sleep(0.3)
|
||||
self.assertAlmostEqual(self.c.status.mpos[1], y, places=3) # held still
|
||||
self.player.resume()
|
||||
self.player.skip()
|
||||
self.assertTrue(wait_for(lambda: self.player.now and self.player.now['name'] == 'tiny', 30))
|
||||
self.player.stop()
|
||||
self.assertEqual(self.player.state, 'idle')
|
||||
self.fake.close()
|
||||
|
||||
def test_switch_trip_is_reported(self):
|
||||
self.make(20)
|
||||
self.player.play(['flowers/rose.thr'])
|
||||
self.assertTrue(wait_for(lambda: self.player.now and self.player.now['frac'] > 0.05, 60))
|
||||
with self.fake.lock:
|
||||
self.fake.switch_at = 1000.0 # the switch closes everywhere
|
||||
self.assertTrue(wait_for(lambda: not self.player.running(), 30))
|
||||
self.assertIn('home switch', self.player.message)
|
||||
self.assertEqual(len(self.alerts), 1)
|
||||
self.assertFalse(self.panel.homed)
|
||||
self.fake.close()
|
||||
|
||||
|
||||
class TestReconnect(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.tmp = tempfile.TemporaryDirectory()
|
||||
self.lib = make_library(self.tmp.name)
|
||||
|
||||
def tearDown(self):
|
||||
self.tmp.cleanup()
|
||||
|
||||
def test_cable_out_and_back_in(self):
|
||||
fake, c = make_controller(20)
|
||||
fresh = []
|
||||
|
||||
def opener(cfg): # "plugging back in" gives a new simulated Uno
|
||||
if not fresh:
|
||||
raise OSError('the Uno is not connected: plug in its USB cable (it reconnects by itself)')
|
||||
return fresh[-1]
|
||||
panel = Panel(c, CFG, opener=opener)
|
||||
player = Player(panel, CFG, self.lib)
|
||||
player.play(['flowers/rose.thr'])
|
||||
self.assertTrue(wait_for(lambda: player.now and player.now['frac'] > 0.02, 60))
|
||||
fake.unplugged = True
|
||||
self.assertTrue(wait_for(lambda: not player.running(), 10))
|
||||
self.assertIn('USB connection', player.message)
|
||||
self.assertTrue(wait_for(lambda: not panel.connected, 5))
|
||||
self.assertFalse(panel.homed)
|
||||
self.assertTrue(wait_for(lambda: 'not connected' in panel.error, 5)) # trying to reconnect
|
||||
new = FakeGrbl(speedup=20, carriage_mm=30.0)
|
||||
fresh.append(new)
|
||||
self.assertTrue(wait_for(lambda: panel.connected, 10))
|
||||
self.assertEqual(panel.error, '')
|
||||
self.assertEqual(c.query().state, 'Idle')
|
||||
player.play(['spirals/tiny.thr']) # homes again, then draws
|
||||
self.assertTrue(wait_for(lambda: player.now and player.now['name'] == 'tiny', 60))
|
||||
player.stop()
|
||||
self.assertTrue(panel.homed)
|
||||
new.close()
|
||||
fake.close()
|
||||
|
||||
|
||||
class TestHttp(unittest.TestCase):
|
||||
@classmethod
|
||||
def setUpClass(cls):
|
||||
cls.tmp = tempfile.TemporaryDirectory()
|
||||
lib = make_library(cls.tmp.name)
|
||||
cls.fake, c = make_controller(100)
|
||||
s = socket.socket()
|
||||
s.bind(('127.0.0.1', 0))
|
||||
cls.port = s.getsockname()[1]
|
||||
s.close()
|
||||
threading.Thread(target=serve, args=(c, CFG), kwargs={'port': cls.port, 'open_browser': False, 'library': lib},
|
||||
daemon=True).start()
|
||||
assert wait_for(lambda: cls._up(), 10)
|
||||
|
||||
@classmethod
|
||||
def _up(cls):
|
||||
try:
|
||||
urllib.request.urlopen(f'http://127.0.0.1:{cls.port}/api/status', timeout=1)
|
||||
return True
|
||||
except OSError:
|
||||
return False
|
||||
|
||||
def get(self, path):
|
||||
with urllib.request.urlopen(f'http://127.0.0.1:{self.port}{path}', timeout=10) as r:
|
||||
return r.read()
|
||||
|
||||
def post(self, path, body=None, raw=None):
|
||||
data = raw if raw is not None else json.dumps(body or {}).encode()
|
||||
req = urllib.request.Request(f'http://127.0.0.1:{self.port}/api/{path}', data=data, method='POST')
|
||||
with urllib.request.urlopen(req, timeout=40) as r:
|
||||
return json.loads(r.read())
|
||||
|
||||
def test_pages_and_play(self):
|
||||
self.assertIn(b'Sand Table', self.get('/'))
|
||||
self.assertIn(b'jog & calibrate', self.get('/jog'))
|
||||
lib = json.loads(self.get('/api/library'))
|
||||
self.assertEqual(len(lib), 2)
|
||||
self.assertTrue(self.get('/thumb?id=spirals/tiny.thr').startswith(b'\x89PNG'))
|
||||
r = self.post('play', {'ids': ['spirals/tiny.thr']})
|
||||
self.assertTrue(r['ok'], r)
|
||||
self.assertFalse(self.post('jog', {'axis': 'Y', 'd': 1})['ok']) # refused while playing
|
||||
self.assertTrue(wait_for(lambda: json.loads(self.get('/api/status'))['player']['state'] == 'idle', 60))
|
||||
st = json.loads(self.get('/api/status'))
|
||||
self.assertTrue(st['machine']['homed'])
|
||||
self.assertEqual(st['player']['message'], '')
|
||||
self.assertFalse(self.post('play', {'ids': ['../../etc/passwd']})['ok'])
|
||||
up = self.post('upload?name=new.thr', raw=b'0 0\n6.28 1\n')
|
||||
self.assertEqual(up['msg'], 'uploads/new.thr')
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
unittest.main()
|
||||
Reference in new issue
Block a user