Log in
dotter
main 0 open branches 33 files Download ZIP Browse files
scripts/golden/gen_golden.py 143 lines · 4.5 KB · python
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()