Restructure to use Future

Author: chrisruk

buildhat/devices.py

@@ -3,6 +3,7 @@
 import os
 import sys
 import weakref
+from concurrent.futures import Future
 
 from .exc import DeviceError
 from .serinterface import BuildHAT
@@ -202,11 +203,10 @@ def get(self):
             idx = self._combimode
         else:
             raise DeviceError("Not in simple or combimode")
+        ftr = Future()
+        self._hat.portcond[self.port].append(ftr)
         self._write(f"port {self.port} ; selonce {idx}\r")
-        # wait for data
-        with Device._instance.portcond[self.port]:
-            Device._instance.portcond[self.port].wait()
-        return self._conn.data
+        return ftr.result()
 
     def mode(self, modev):
         """Set combimode or simple mode

buildhat/motors.py

@@ -3,6 +3,7 @@
 import threading
 import time
 from collections import deque
+from concurrent.futures import Future
 from enum import Enum
 from threading import Condition
 
@@ -205,9 +206,10 @@ def _run_positional_ramp(self, pos, newpos, speed):
         cmd = (f"port {self.port}; combi 0 {self._combi} ; select 0 ; selrate {self._interval}; "
                f"pid {self.port} 0 1 s4 0.0027777778 0 5 0 .1 3; "
                f"set ramp {pos} {newpos} {dur} 0\r")
+        ftr = Future()
+        self._hat.rampcond[self.port].append(ftr)
         self._write(cmd)
-        with self._hat.rampcond[self.port]:
-            self._hat.rampcond[self.port].wait()
+        ftr.result()
         if self._release:
             time.sleep(0.2)
             self.coast()
@@ -228,9 +230,7 @@ def run_for_degrees(self, degrees, speed=None, blocking=True):
         if not (speed >= -100 and speed <= 100):
             raise MotorError("Invalid Speed")
         if not blocking:
-            th = threading.Thread(target=self._run_for_degrees, args=(degrees, speed))
-            th.daemon = True
-            th.start()
+            Device._instance.mqueue.put((self._run_for_degrees, (degrees, speed)))
         else:
             self._run_for_degrees(degrees, speed)
 
@@ -251,9 +251,7 @@ def run_to_position(self, degrees, speed=None, blocking=True, direction="shortes
         if degrees < -180 or degrees > 180:
             raise MotorError("Invalid angle")
         if not blocking:
-            th = threading.Thread(target=self._run_to_position, args=(degrees, speed, direction))
-            th.daemon = True
-            th.start()
+            Device._instance.mqueue.put((self._run_to_position, (degrees, speed, direction)))
         else:
             self._run_to_position(degrees, speed, direction)
 
@@ -262,9 +260,10 @@ def _run_for_seconds(self, seconds, speed):
         cmd = (f"port {self.port} ; combi 0 {self._combi} ; select 0 ; selrate {self._interval}; "
                f"pid {self.port} 0 0 s1 1 0 0.003 0.01 0 100; "
                f"set pulse {speed} 0.0 {seconds} 0\r")
+        ftr = Future()
+        self._hat.pulsecond[self.port].append(ftr)
         self._write(cmd)
-        with self._hat.pulsecond[self.port]:
-            self._hat.pulsecond[self.port].wait()
+        ftr.result()
         if self._release:
             self.coast()
         self._runmode = MotorRunmode.NONE
@@ -283,9 +282,7 @@ def run_for_seconds(self, seconds, speed=None, blocking=True):
         if not (speed >= -100 and speed <= 100):
             raise MotorError("Invalid Speed")
         if not blocking:
-            th = threading.Thread(target=self._run_for_seconds, args=(seconds, speed))
-            th.daemon = True
-            th.start()
+            Device._instance.mqueue.put((self._run_for_seconds, (seconds, speed)))
         else:
             self._run_for_seconds(seconds, speed)
 

buildhat/serinterface.py

@@ -98,9 +98,9 @@ def __init__(self, firmware, signature, version, device="/dev/serial0", debug=Fa
 
         for _ in range(4):
             self.connections.append(Connection())
-            self.portcond.append(Condition())
-            self.pulsecond.append(Condition())
-            self.rampcond.append(Condition())
+            self.portcond.append([])
+            self.pulsecond.append([])
+            self.rampcond.append([])
 
         self.ser = serial.Serial(device, 115200, timeout=5)
         # Check if we're in the bootloader or the firmware
@@ -150,6 +150,11 @@ def __init__(self, firmware, signature, version, device="/dev/serial0", debug=Fa
         self.cb.daemon = True
         self.cb.start()
 
+        self.mqueue = queue.Queue()
+        self.ml = threading.Thread(target=self.motorloop, args=(self.mqueue,))
+        self.ml.daemon = True
+        self.ml.start()
+
         # Drop timeout value to 1s
         listevt = threading.Event()
         self.ser.timeout = 1
@@ -266,6 +271,7 @@ def shutdown(self):
             self.running = False
             self.th.join()
             self.cbqueue.put(())
+            self.mqueue.put((None, None))
             self.cb.join()
             turnoff = ""
             for p in range(4):
@@ -278,6 +284,18 @@ def shutdown(self):
             self.write(f"{turnoff}\r".encode())
             self.write(b"port 0 ; select ; port 1 ; select ; port 2 ; select ; port 3 ; select ; echo 0\r")
 
+    def motorloop(self, q):
+        """Event handling for non-blocking motor commands
+
+        :param q: Queue of motor functions
+        """
+        while self.running:
+            func, data = q.get()
+            if func is None:
+                break
+            else:
+                func(*data)
+
     def callbackloop(self, q):
         """Event handling for callbacks
 
@@ -330,11 +348,11 @@ def loop(self, cond, uselist, q, listevt):
                     if uselist and listevt.is_set():
                         count += 1
                 elif cmp(msg, BuildHAT.RAMPDONE):
-                    with self.rampcond[portid]:
-                        self.rampcond[portid].notify()
+                    ftr = self.rampcond[portid].pop()
+                    ftr.set_result(True)
                 elif cmp(msg, BuildHAT.PULSEDONE):
-                    with self.pulsecond[portid]:
-                        self.pulsecond[portid].notify()
+                    ftr = self.pulsecond[portid].pop()
+                    ftr.set_result(True)
 
             if uselist and count == 4:
                 with cond:
@@ -362,8 +380,11 @@ def runit():
                 if callit is not None:
                     q.put((callit, newdata))
                 self.connections[portid].data = newdata
-                with self.portcond[portid]:
-                    self.portcond[portid].notify()
+                try:
+                    ftr = self.portcond[portid].pop()
+                    ftr.set_result(newdata)
+                except IndexError:
+                    pass
 
             if len(line) >= 5 and line[1] == "." and line.endswith(" V"):
                 vin = float(line.split(" ")[0])