"""Jog panel: a small web page (http://127.0.0.1:8765) to home, jog and calibrate the table. Keeps the USB connection open the whole time, so the Uno doesn't restart between moves. Start it with `python3 -m sandtable ui`. """ import json import threading import time import webbrowser from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from .grbl import RT_HOLD, RT_JOG_CANCEL, GrblError PAGE = r""" Sand Table Jog

Sand table: jog & calibrate

State-

Rho (mm)-

Theta (deg)-

Home switch-

not homed since connecting

Rho

+ moves the magnet outward. An inward jog stops by itself when it presses the home switch; nothing stops it at the outer end, so watch the carriage there.

Theta

+ turns the arm counter-clockwise, seen from above.

Calibrate

Move a known amount with the buttons above, measure what really happened, and enter both. The new steps value is written to the Uno.

Rho commanded mm, measured

Theta commanded deg, measured

Now: rho - steps/mm, theta - steps/deg

""" def _log(msg): try: from .web import log log(msg) except Exception: print(msg, flush=True) def open_usb(cfg): """Find the Uno and open it, for reconnecting. Raises OSError with a plain message.""" import serial from serial.tools import list_ports from .cli import USB_IDS dev = cfg['serial']['port'] if not dev or dev == 'auto': cands = [p.device for p in list_ports.comports() if p.vid in USB_IDS] if not cands: raise OSError('the Uno is not connected: plug in its USB cable (it reconnects by itself)') dev = cands[0] try: return serial.Serial(dev, cfg['serial']['baud'], timeout=0.02, write_timeout=2) except Exception as e: if getattr(e, 'errno', None) == 13 or 'Permission denied' in str(e): raise OSError(f'no permission for {dev}: run sudo chmod a+rw {dev} (it reconnects by itself)') raise OSError(f'cannot open {dev}: {e}') class Panel: def __init__(self, ctrl, cfg, connected=True, opener=open_usb): self.c = ctrl self.cfg = cfg self.lock = threading.Lock() # one serial conversation at a time self.homed = False self.error = '' self.steps = {} self.busy = '' self.jog_in = False # the last jog moved rho inward, toward the switch self.connected = connected self.opener = opener self._next_try = 0.0 if connected: with self.lock: self._read_steps() else: self.error = 'the Uno is not connected: plug in its USB cable (it reconnects by itself)' threading.Thread(target=self._poll, daemon=True).start() def lost(self, e): """The USB link dropped: forget the position and start trying to reconnect.""" _log(f'USB connection lost: {e!r}') self.connected = False self.homed = False self.error = f'USB connection lost ({e}). Plug the Uno back in; the page reconnects by itself.' try: self.c.port.close() except Exception: pass def _reconnect(self): if time.monotonic() < self._next_try: return self._next_try = time.monotonic() + 2.0 try: self.c.reopen(self.opener(self.cfg)) self.c.query() self._read_steps() except Exception as e: self.error = str(e) return self.connected = True self.error = '' _log('USB connection back') def _read_steps(self): s = self.c.settings() self.steps = {k: s.get(k, '') for k in ('$100', '$101')} def _poll(self): while True: fast = False if self.lock.acquire(timeout=0.05): try: if not self.connected: self._reconnect() else: st = self.c.query() fast = st.state == 'Jog' if fast and self.jog_in and 'Y' in st.pins: # inward jog reached the home switch: stop before the carriage hits the end stop self.c.realtime(RT_JOG_CANCEL) self.jog_in = False self.error = 'inward jog stopped: the home switch is pressed' except OSError as e: # pyserial's errors are OSErrors: the cable came out self.lost(e) except Exception as e: # keep the page alive; show the problem if str(e) != self.error: _log(f'panel: {e}') self.error = str(e) finally: self.lock.release() time.sleep(0.05 if fast else 0.25) def status(self): st = self.c.status state = 'Motors off' if st.state == 'Sleep' else st.state + (':' + st.sub if st.sub else '') return {'state': 'Disconnected' if not self.connected else self.busy or state, 'connected': self.connected, 'theta': st.wpos[0], 'rho': st.wpos[1], 'switch': 'Y' in st.pins, 'pins': st.pins, 'homed': self.homed, 'steps': self.steps, 'error': self.error} # ---------------------------------------------------------- actions def jog(self, axis, d): axis = 'X' if axis == 'X' else 'Y' d = max(-360.0, min(360.0, float(d))) feed = 1200 if axis == 'X' else 600 # 20 deg/s, 10 mm/s with self.lock: st = self.c.query() if st.state == 'Sleep': # motors switched off after a design: wake them self.c.unlock() st = self.c.query() if st.state == 'Alarm': return False, 'alarm: press Unlock first' inward = axis == 'Y' and d < 0 if inward and 'Y' in st.pins: return False, 'the home switch is pressed: rho can only move outward (+) from here' self.error = '' self.jog_in = inward self.c.command(f'$J=G91 G21 {axis}{d:.3f} F{feed}') return True, f'jog {"theta" if axis == "X" else "rho"} {d:+g}' def home(self): if not self.lock.acquire(timeout=1): return False, 'busy' try: self.busy = 'Homing' self.jog_in = False self.c.abort = False self.c.home_rho(self.cfg, log=lambda m: None) self.homed = True self.error = '' return True, 'homed' except GrblError as e: self.homed = False return False, str(e) finally: self.busy = '' self.lock.release() def stop(self): # real-time bytes go straight out, even while another request holds the lock self.c.abort = True self.c.realtime(RT_JOG_CANCEL) self.c.realtime(RT_HOLD) with self.lock: self.c.stop() # waits for the stop, clears the queue, keeps the position return True, 'stopped' def unlock(self): with self.lock: self.c.unlock() # also wakes the motors after a design self.c.command('$X') return True, 'unlocked' def calibrate(self, key, commanded, measured): if key not in ('$100', '$101'): return False, 'bad setting' with self.lock: old = float(self.c.settings().get(key, '0')) new = old * float(commanded) / float(measured) if not 0.2 * old < new < 5 * old: return False, f'that would change {key} from {old:g} to {new:.3f}; check the numbers' self.c.command(f'{key}={new:.3f}') self._read_steps() what = 'rho steps/mm' if key == '$101' else 'theta steps/deg' return True, (f'{what}: {old:g} -> {new:.3f}. Saved on the Uno. Home again, then copy it into ' f'sandtable.toml so `setup` keeps it.') def serve(ctrl, cfg, port=8765, open_browser=True): panel = Panel(ctrl, cfg) class H(BaseHTTPRequestHandler): def log_message(self, *a): pass def _send(self, code, body, ctype='application/json'): data = body.encode() if isinstance(body, str) else body self.send_response(code) self.send_header('Content-Type', ctype) self.send_header('Content-Length', str(len(data))) self.end_headers() self.wfile.write(data) def do_GET(self): if self.path in ('/', '/index.html'): self._send(200, PAGE, 'text/html; charset=utf-8') elif self.path == '/api/status': self._send(200, json.dumps(panel.status())) else: self._send(404, '{}') def do_POST(self): n = int(self.headers.get('Content-Length') or 0) body = json.loads(self.rfile.read(n) or b'{}') act = self.path.rsplit('/', 1)[-1] try: if act == 'jog': ok, msg = panel.jog(body.get('axis'), body.get('d', 0)) elif act == 'home': ok, msg = panel.home() elif act == 'stop': ok, msg = panel.stop() elif act == 'unlock': ok, msg = panel.unlock() elif act == 'calibrate': ok, msg = panel.calibrate(body.get('key'), body.get('commanded'), body.get('measured')) else: ok, msg = False, 'unknown action' except Exception as e: ok, msg = False, str(e) self._send(200, json.dumps({'ok': ok, 'msg': msg})) srv = ThreadingHTTPServer(('127.0.0.1', port), H) url = f'http://127.0.0.1:{port}' print(f'Jog panel at {url} (Ctrl-C here to quit)', flush=True) if open_browser: try: webbrowser.open(url) except Exception: pass try: srv.serve_forever() except KeyboardInterrupt: pass finally: srv.server_close()