add some tests for the virtual grbl controller

spf/grbl/grbl_interactive.py

@@ -1,6 +1,7 @@
 import argparse
 import logging
 import queue
+import select
 import sys
 import threading
 import time
@@ -480,6 +481,9 @@ def write(self, x):
     def readline(self):
         return self.rq.get()
 
+    def close(self):
+        pass
+
 
 class GRBLController:
     def __init__(self, serial_fn, dynamics, channel_to_motor_map):
@@ -839,6 +843,9 @@ def v1_calibrate(self):
             ),
         ]
 
+    def close(self):
+        self.controller.close()
+
 
 def get_default_dynamics(unsafe=False):
     return GRBLDynamics(
@@ -862,6 +869,13 @@ def get_default_gm(serial_fn, routine, unsafe=False):
     return GRBLManager(controller, routine=routine)
 
 
+def exit_fine():
+    global run_grbl
+    run_grbl = False
+    print("EXIT!!")
+    sys.exit(0)
+
+
 if __name__ == "__main__":
     parser = argparse.ArgumentParser()
     parser.add_argument("serial", help="serial device to use")
@@ -891,43 +905,55 @@ def get_default_gm(serial_fn, routine, unsafe=False):
     )
     for k in gm.routines:
         print("        ", k)
-    for line in sys.stdin:
-        line = line.strip()
-        if line in gm.routines:
-            gm.routines[line]()
-            gm.get_ready()
-            gm.run()
-        elif len(line) == 0:
-            continue
-        elif line == "q":
-            sys.exit(1)
-        elif line == "s":
-            p = gm.controller.update_status()
-            print(p)
-        elif len(line.split()) == 3:
-            target_channel = int(line.split()[0])
-            points_iter = {
-                target_channel: iter(
-                    a_to_b_in_stepsize(
-                        gm.controller.update_status()["xy"][target_channel],
-                        np.array([float(x) for x in line.split()[1:]]),
-                        5,
+    while run_grbl:
+        if select.select(
+            [
+                sys.stdin,
+            ],
+            [],
+            [],
+            0.1,
+        )[0]:
+            line = sys.stdin.readline()
+            line = line.strip()
+            if line in gm.routines:
+                gm.routines[line]()
+                gm.get_ready()
+                gm.run()
+            elif line.split()[0] == "timer":
+                t = threading.Timer(float(line.split()[1]), exit_fine)
+                t.start()
+            elif len(line) == 0:
+                continue
+            elif line == "q":
+                sys.exit(0)
+            elif line == "s":
+                p = gm.controller.update_status()
+                print(p)
+            elif len(line.split()) == 3:
+                target_channel = int(line.split()[0])
+                points_iter = {
+                    target_channel: iter(
+                        a_to_b_in_stepsize(
+                            gm.controller.update_status()["xy"][target_channel],
+                            np.array([float(x) for x in line.split()[1:]]),
+                            5,
+                        )
                     )
-                )
-            }
-            gm.controller.move_to_iter(points_iter)
-        else:
-            points_iter = {
-                c: iter(
-                    a_to_b_in_stepsize(
-                        gm.controller.update_status()["xy"][c],
-                        np.array([float(x) for x in line.split()]),
-                        5,
+                }
+                gm.controller.move_to_iter(points_iter)
+            else:
+                points_iter = {
+                    c: iter(
+                        a_to_b_in_stepsize(
+                            gm.controller.update_status()["xy"][c],
+                            np.array([float(x) for x in line.split()]),
+                            5,
+                        )
                     )
-                )
-                for c in [0, 1]
-            }
-            gm.controller.move_to_iter(points_iter)
-        time.sleep(0.01)
+                    for c in [0, 1]
+                }
+                gm.controller.move_to_iter(points_iter)
+            time.sleep(0.01)
 
     gm.close()

tests/grbl_test_bounce

@@ -0,0 +1,4 @@
+600 600
+timer 10
+bounce
+

tests/grbl_test_simple_move

@@ -0,0 +1,4 @@
+600 600
+550 580
+q
+

tests/test_grbl.py

@@ -1,9 +1,13 @@
+import os
+import subprocess
+import sys
 import time
 
 import matplotlib.path as pltpath
 import numpy as np
 from shapely import geometry
 
+import spf
 from spf.grbl.grbl_interactive import (
     BouncePlanner,
     GRBLDynamics,
@@ -13,6 +17,14 @@
     home_pB,
 )
 
+root_dir = os.path.dirname(os.path.dirname(spf.__file__))
+
+
+def get_env():
+    env = os.environ.copy()
+    env["PYTHONPATH"] = ":".join(sys.path)
+    return env
+
 
 def test_steps_and_steps_inverse():
     dynamics = GRBLDynamics(
@@ -113,3 +125,23 @@ def test_binary_search_edge():
     lp, nd = planner.get_bounce_pos_and_new_direction(lp, nd)
     planner.single_bounce(direction, p)
     [x for x in planner.bounce(p, 10)]
+
+
+def test_grbl_simple_move(script_runner):
+    subprocess.check_output(
+        f"cat {root_dir}/tests/grbl_test_simple_move | python3 {root_dir}/spf/grbl/grbl_interactive.py none",
+        timeout=180,
+        shell=True,
+        env=get_env(),
+        stderr=subprocess.STDOUT,
+    ).decode()
+
+
+def test_grbl_bounce(script_runner):
+    subprocess.check_output(
+        f"cat {root_dir}/tests/grbl_test_bounce | python3 {root_dir}/spf/grbl/grbl_interactive.py none",
+        timeout=180,
+        shell=True,
+        env=get_env(),
+        stderr=subprocess.STDOUT,
+    ).decode()

tests/test_mavlink_radio_collect.py

@@ -1,4 +1,4 @@
-def test_foo_bar(script_runner):
+def test_mavlink_radio_collect(script_runner):
     result = script_runner.run(
         [
             "./spf/mavlink_radio_collection.py",