Allow custom device interval for callbacks

Author: chrisruk

buildhat/devices.py

@@ -58,6 +58,7 @@ def __init__(self, port):
         self._simplemode = -1
         self._combimode = -1
         self._typeid = self._conn.typeid
+        self._interval = 100
         if (
             self._typeid in Device._device_names
             and Device._device_names[self._typeid][0] != type(self).__name__  # noqa: W503
@@ -239,7 +240,7 @@ def select(self):
             idx = self._combimode
         else:
             raise DeviceError("Not in simple or combimode")
-        self._write(f"port {self.port} ; select {idx}\r")
+        self._write(f"port {self.port} ; select {idx} ; selrate {self._interval}\r")
 
     def on(self):
         """Turn on sensor"""
@@ -274,3 +275,28 @@ def callback(self, func):
             self._conn.callit = None
         else:
             self._conn.callit = weakref.WeakMethod(func)
+
+    @property
+    def interval(self):
+        """Interval between data points in milliseconds
+
+        :getter: Gets interval
+        :setter: Sets interval
+        :return: Device interval
+        :rtype: int
+        """
+        return self._interval
+
+    @interval.setter
+    def interval(self, value):
+        """Interval between data points in milliseconds
+
+        :param value: Interval
+        :type value: int
+        :raises DeviceError: Occurs if invalid interval passed
+        """
+        if isinstance(value, int) and value >= 0 and value <= 1000000000:
+            self._interval = value
+            self._write(f"port {self.port} ; selrate {self._interval}\r")
+        else:
+            raise DeviceError("Invalid interval")

buildhat/motors.py

@@ -202,7 +202,8 @@ def _run_positional_ramp(self, pos, newpos, speed):
         # Collapse speed range to -5 to 5
         speed *= 0.05
         dur = abs((newpos - pos) / speed)
-        cmd = (f"port {self.port}; combi 0 {self._combi} ; select 0 ; pid {self.port} 0 1 s4 0.0027777778 0 5 0 .1 3 ; "
+        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")
         self._write(cmd)
         with self._hat.rampcond[self.port]:
@@ -258,7 +259,8 @@ def run_to_position(self, degrees, speed=None, blocking=True, direction="shortes
 
     def _run_for_seconds(self, seconds, speed):
         self._runmode = MotorRunmode.SECONDS
-        cmd = (f"port {self.port} ; combi 0 {self._combi} ; select 0 ; pid {self.port} 0 0 s1 1 0 0.003 0.01 0 100; "
+        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")
         self._write(cmd)
         with self._hat.pulsecond[self.port]:
@@ -308,7 +310,8 @@ def start(self, speed=None):
                 raise MotorError("Invalid Speed")
         cmd = f"port {self.port} ; set {speed}\r"
         if self._runmode == MotorRunmode.NONE:
-            cmd = (f"port {self.port} ; combi 0 {self._combi} ; select 0 ; pid {self.port} 0 0 s1 1 0 0.003 0.01 0 100; "
+            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 {speed}\r")
         self._runmode = MotorRunmode.FREE
         self._currentspeed = speed

test/motors.py

@@ -81,6 +81,23 @@ def handle_motor(speed, pos, apos):
         m.run_for_seconds(1)
         self.assertGreater(handle_motor.evt, 0)
 
+    def test_callback_interval(self):
+        """Test setting callback and interval"""
+        m = Motor('A')
+        m.interval = 10
+
+        def handle_motor(speed, pos, apos):
+            handle_motor.evt += 1
+        handle_motor.evt = 0
+        m.when_rotated = handle_motor
+        m.run_for_seconds(5)
+        self.assertGreater(handle_motor.evt, 0.8 * ((1 / ((m.interval) * 1e-3)) * 5))
+
+        handle_motor.evt = 0
+        m.interval = 5
+        m.run_for_seconds(5)
+        self.assertGreater(handle_motor.evt, 0.8 * ((1 / ((m.interval) * 1e-3)) * 5))
+
     def test_none_callback(self):
         """Test setting empty callback"""
         m = Motor('A')