| 1 | """
|
| 2 | Record golden command streams from the official NextDraw Python API.
|
| 3 |
|
| 4 | Runs each scenario in test/golden/scenarios.json through the real `nextdraw`
|
| 5 | package against a fake EBB serial port (no hardware needed, nothing moves),
|
| 6 | and writes the exact commands sent to test/golden/expected/<name>.json.
|
| 7 | test/golden.test.js then checks that dotter sends the same commands.
|
| 8 |
|
| 9 | Usage (Python 3.10+):
|
| 10 | python -m venv .venv
|
| 11 | .venv/bin/pip install https://software-download.bantamtools.com/nd/api/nextdraw_api.zip
|
| 12 | .venv/bin/python scripts/golden/gen_golden.py
|
| 13 | """
|
| 14 |
|
| 15 | import json
|
| 16 | import math
|
| 17 | import pathlib
|
| 18 | import time
|
| 19 |
|
| 20 | from plotink import ebb3_serial
|
| 21 | from nextdraw import NextDraw
|
| 22 |
|
| 23 | ROOT = pathlib.Path(__file__).resolve().parents[2]
|
| 24 | SCENARIOS = ROOT / "test" / "golden" / "scenarios.json"
|
| 25 | OUT_DIR = ROOT / "test" / "golden" / "expected"
|
| 26 |
|
| 27 | MICROSTEPS = {1: 16, 2: 8, 3: 4, 4: 2, 5: 1, 0: 0}
|
| 28 |
|
| 29 |
|
| 30 | class FakeEBB:
|
| 31 | """Just enough of an EBB v3 (future syntax) to satisfy the driver."""
|
| 32 |
|
| 33 | def __init__(self, *args, **kwargs):
|
| 34 | self.out = []
|
| 35 | self.sent = []
|
| 36 | self.future = False
|
| 37 | self.vars = [0] * 32
|
| 38 | self.steps = [0, 0]
|
| 39 | self.enabled = [0, 0]
|
| 40 | self.pen_up = True
|
| 41 |
|
| 42 | # pyserial-like interface
|
| 43 | def reset_input_buffer(self):
|
| 44 | self.out = []
|
| 45 |
|
| 46 | def flush(self):
|
| 47 | pass
|
| 48 |
|
| 49 | def close(self):
|
| 50 | pass
|
| 51 |
|
| 52 | def write(self, data):
|
| 53 | for cmd in data.decode("ascii").split("\r"):
|
| 54 | if cmd:
|
| 55 | self.sent.append(cmd)
|
| 56 | self.handle(cmd)
|
| 57 |
|
| 58 | def readline(self):
|
| 59 | return self.out.pop(0).encode("ascii") if self.out else b""
|
| 60 |
|
| 61 | def reply(self, name, data=None):
|
| 62 | self.out.append(f"{name}\n" if data is None else f"{name},{data}\n")
|
| 63 |
|
| 64 | def handle(self, cmd):
|
| 65 | parts = cmd.split(",")
|
| 66 | name = parts[0].upper()
|
| 67 | args = parts[1:]
|
| 68 | if name == "V":
|
| 69 | self.out.append("EBBv13_and_above EB Firmware Version 3.0.3\r\n")
|
| 70 | elif name == "CU":
|
| 71 | if args[:2] == ["10", "1"]:
|
| 72 | self.future = True
|
| 73 | self.reply("CU")
|
| 74 | elif name == "QT":
|
| 75 | self.reply("QT")
|
| 76 | elif name == "QG":
|
| 77 | self.reply("QG", "%02X" % (16 if self.pen_up else 0))
|
| 78 | elif name == "QC":
|
| 79 | self.reply("QC", "400,305")
|
| 80 | elif name == "QE":
|
| 81 | self.reply("QE", f"{MICROSTEPS[self.enabled[0]]},{MICROSTEPS[self.enabled[1]]}")
|
| 82 | elif name == "EM":
|
| 83 | self.enabled = [int(args[0]), int(args[1]) if len(args) > 1 else self.enabled[1]]
|
| 84 | self.steps = [0, 0]
|
| 85 | self.reply("EM")
|
| 86 | elif name == "QS":
|
| 87 | self.reply("QS", f"{self.steps[0]},{self.steps[1]}")
|
| 88 | elif name == "CS":
|
| 89 | self.steps = [0, 0]
|
| 90 | self.reply("CS")
|
| 91 | elif name == "SL":
|
| 92 | self.vars[int(args[1])] = int(args[0]) & 255
|
| 93 | self.reply("SL")
|
| 94 | elif name == "QL":
|
| 95 | self.reply("QL", str(self.vars[int(args[0])]))
|
| 96 | elif name == "PI":
|
| 97 | self.reply("PI", "0")
|
| 98 | elif name == "SP":
|
| 99 | if args[0] == "0":
|
| 100 | self.pen_up = False
|
| 101 | elif args[0] in ("1", "2", "3"):
|
| 102 | self.pen_up = True
|
| 103 | self.reply("SP")
|
| 104 | else:
|
| 105 | self.reply(name[:2] if len(name) > 1 else name)
|
| 106 |
|
| 107 |
|
| 108 | def run_scenario(spec):
|
| 109 | fake = FakeEBB()
|
| 110 | ebb3_serial.serial.Serial = lambda *a, **k: fake
|
| 111 | ebb3_serial.EBB3.find_first = lambda self: setattr(self, "port_name", "FAKE")
|
| 112 |
|
| 113 | nd = NextDraw()
|
| 114 | nd.interactive()
|
| 115 | nd.options.homing = False
|
| 116 | for key, value in spec["options"].items():
|
| 117 | setattr(nd.options, key, value)
|
| 118 | if not nd.connect():
|
| 119 | raise RuntimeError("connect failed")
|
| 120 | for op, *args in spec["ops"]:
|
| 121 | if op == "draw_path_circle":
|
| 122 | cx, cy, r, n = args
|
| 123 | pts = [[cx + r * math.cos(2 * math.pi * i / n), cy + r * math.sin(2 * math.pi * i / n)]
|
| 124 | for i in range(n + 1)]
|
| 125 | nd.draw_path(pts)
|
| 126 | else:
|
| 127 | getattr(nd, op)(*args)
|
| 128 | nd.disconnect()
|
| 129 | return [c for c in fake.sent if c != "QG"]
|
| 130 |
|
| 131 |
|
| 132 | def main():
|
| 133 | time.sleep = lambda seconds: None # The driver paces long moves with sleeps; skip them.
|
| 134 | OUT_DIR.mkdir(parents=True, exist_ok=True)
|
| 135 | scenarios = json.loads(SCENARIOS.read_text())
|
| 136 | for name, spec in scenarios.items():
|
| 137 | commands = run_scenario(spec)
|
| 138 | (OUT_DIR / f"{name}.json").write_text(json.dumps(commands, indent=0) + "\n")
|
| 139 | print(f"{name}: {len(commands)} commands")
|
| 140 |
|
| 141 |
|
| 142 | if __name__ == "__main__":
|
| 143 | main()
|