Refactoring of encoder motors, new test case

Author: sorre97

coderbot.py

@@ -257,109 +257,109 @@ def _cb_button(self, gpio, level, tick):
                 cb()
 
     class MotorEncoder(object):
-        def __init__(self, parent, _pigpio, pin_enable, pin_forward, pin_backward, pin_encoder):
-            self._parent = parent
-            self._pigpio = _pigpio
-            self._pin_enable = pin_enable
-            self._pin_forward = pin_forward
-            self._pin_backward = pin_backward
-            self._pin_encoder = pin_encoder
-            self._direction = False
-            self._pin_duty = 0
-            self._pin_reverse = 0
-            self._power = 0.0
-            self._power_actual = 0.0
-            self._encoder_dist = 0
-            self._encoder_speed = 0.0
-            self._encoder_last_tick = 0
-            self._encoder_dist_target = 0
-            self._encoder_speed_target = 0.0
-            self._encoder_k_s_1 = 20
-            self._encoder_k_v_1 = 80
-            self._motor_stopping = False
-            self._motor_running = False
-            self._motor_stop_fast = True
-            self._pigpio.set_mode(self._pin_encoder, pigpio.INPUT)
-            self._cb = self._pigpio.callback(self._pin_encoder, pigpio.RISING_EDGE, self._cb_encoder)
-            self._motor_lock = threading.RLock()
-
-        def exit(self):
-            self._cb.cancel()
-
-        def _cb_encoder(self, gpio, level, tick):
-            self._motor_lock.acquire()
-            self._encoder_dist += 1
-            delta_ticks = tick - self._encoder_last_tick if tick > self._encoder_last_tick else tick - self._encoder_last_tick + 4294967295
-            self._encoder_last_tick = tick
-            self._encoder_speed = 1000000.0 / delta_ticks #convert speed in steps per second
-            #print "pin: " + str(self._pin_forward) + " dist: " + str(self._encoder_dist) + " target: " + str(self._encoder_dist_target)
-            if self._encoder_dist_target >= 0 and self._motor_stop_fast:
-                #delta_s is the delta (in steps)before the target  to reverse the motor in order to arrive at target
-                delta_s = max(min(self._encoder_speed / self._encoder_k_s_1, 100), 0)
-                #print "pin: " + str(self._pin_forward) + " dist: " + str(self._encoder_dist) + " target: " + str(self._encoder_dist_target) + " delta_s: " + str(delta_s)
-                if (self._encoder_dist >= self._encoder_dist_target - delta_s and
-                        not self._motor_stopping and self._motor_running):
-                    self._motor_stopping = True
-                    self._pigpio.write(self._pin_duty, 0)
-                    self._pigpio.set_PWM_dutycycle(self._pin_reverse, self._power)
-                elif (self._motor_running and
-                      ((self._motor_stopping and
-                        self._encoder_speed < self._encoder_k_v_1) or
-                       (self._motor_stopping and
-                        self._encoder_dist >= self._encoder_dist_target))):
-                    self.stop()
-                    logging.info("dist: " + str(self._encoder_dist) + " speed: " + str(self._encoder_speed))
-            if self._encoder_dist_target >= 0 and not self._motor_stop_fast:
-                if self._encoder_dist >= self._encoder_dist_target:
-                    self.stop()
-            self._parent._cb_encoder(self, gpio, level, tick)
-            self._motor_lock.release()
-            if not self._motor_running:
-                self._parent._check_complete()
-
-        def control(self, power=100.0, elapse=-1, speed=100.0, steps=-1):
-            self._motor_lock.acquire()
-            self._direction = speed > 0
-            self._encoder_dist_target = steps
-            self._motor_stopping = False
-            self._motor_running = True
-            self._encoder_dist = 0
-            self._encoder_speed_target = abs(speed)
-            self._power = abs(power) #TODO: initial power must be a function of desired speed
-            self._power_actual = abs(power) #TODO: initial power must be a function of desired speed
-            self._pin_duty = self._pin_forward if self._direction else self._pin_backward
-            self._pin_reverse = self._pin_backward if self._direction else self._pin_forward
-            self._pigpio.write(self._pin_reverse, 0)
-            self._pigpio.set_PWM_dutycycle(self._pin_duty, self._power)
-            self._pigpio.write(self._pin_enable, True)
-            self._motor_lock.release()
-            if elapse > 0:
-                time.sleep(elapse)
-                self.stop()
-
-        def stop(self):
-            self._motor_lock.acquire()
-            self._motor_stopping = False
-            self._motor_running = False
-            self._pigpio.write(self._pin_forward, 0)
-            self._pigpio.write(self._pin_backward, 0)
-            self._motor_lock.release()
-
-        def distance(self):
-            return self._encoder_dist
-
-        def speed(self):
-            return self._encoder_speed
-
-        def stopping(self):
-            return self._motor_stopping
-
-        def running(self):
-            return self._motor_running
-
-        def adjust_power(self, power_delta):
-            self._power_actual = min(max(self._power + power_delta, 0), 100)
-            self._pigpio.set_PWM_dutycycle(self._pin_duty, self._power_actual)
+         # def __init__(self, parent, _pigpio, pin_enable, pin_forward, pin_backward, pin_encoder):
+         #     self._parent = parent
+         #     #self._pigpio = _pigpio
+         #     #self._pin_enable = pin_enable
+         #     #self._pin_forward = pin_forward
+         #     #self._pin_backward = pin_backward
+         #     #self._pin_encoder = pin_encoder
+         #     #self._direction = False
+         #     self._pin_duty = 0
+         #     self._pin_reverse = 0
+         #     self._power = 0.0
+         #     #self._power_actual = 0.0
+         #     #self._encoder_dist = 0
+         #     #self._encoder_speed = 0.0
+         #     self._encoder_last_tick = 0
+         #     self._encoder_dist_target = 0
+         #     self._encoder_speed_target = 0.0
+         #     self._encoder_k_s_1 = 20
+         #     self._encoder_k_v_1 = 80
+         #     self._motor_stopping = False
+         #     self._motor_running = False
+         #     self._motor_stop_fast = True
+         #     #self._pigpio.set_mode(self._pin_encoder, pigpio.INPUT)
+         #     #self._cb = self._pigpio.callback(self._pin_encoder, pigpio.RISING_EDGE, self._cb_encoder)
+         #     #self._motor_lock = threading.RLock()
+
+         # def exit(self):
+         #     self._cb.cancel()
+
+         def _cb_encoder(self, gpio, level, tick):
+             self._motor_lock.acquire()
+             self._encoder_dist += 1
+             delta_ticks = tick - self._encoder_last_tick if tick > self._encoder_last_tick else tick - self._encoder_last_tick + 4294967295
+             self._encoder_last_tick = tick
+             self._encoder_speed = 1000000.0 / delta_ticks #convert speed in steps per second
+             #print "pin: " + str(self._pin_forward) + " dist: " + str(self._encoder_dist) + " target: " + str(self._encoder_dist_target)
+             if self._encoder_dist_target >= 0 and self._motor_stop_fast:
+                 #delta_s is the delta (in steps)before the target  to reverse the motor in order to arrive at target
+                 delta_s = max(min(self._encoder_speed / self._encoder_k_s_1, 100), 0)
+                 #print "pin: " + str(self._pin_forward) + " dist: " + str(self._encoder_dist) + " target: " + str(self._encoder_dist_target) + " delta_s: " + str(delta_s)
+                 if (self._encoder_dist >= self._encoder_dist_target - delta_s and
+                         not self._motor_stopping and self._motor_running):
+                     self._motor_stopping = True
+                     self._pigpio.write(self._pin_duty, 0)
+                     self._pigpio.set_PWM_dutycycle(self._pin_reverse, self._power)
+                 elif (self._motor_running and
+                       ((self._motor_stopping and
+                         self._encoder_speed < self._encoder_k_v_1) or
+                        (self._motor_stopping and
+                         self._encoder_dist >= self._encoder_dist_target))):
+                     self.stop()
+                     logging.info("dist: " + str(self._encoder_dist) + " speed: " + str(self._encoder_speed))
+             if self._encoder_dist_target >= 0 and not self._motor_stop_fast:
+                 if self._encoder_dist >= self._encoder_dist_target:
+                     self.stop()
+             self._parent._cb_encoder(self, gpio, level, tick)
+             self._motor_lock.release()
+             if not self._motor_running:
+                 self._parent._check_complete()
+
+         def control(self, power=100.0, elapse=-1, speed=100.0, steps=-1):
+             self._motor_lock.acquire()
+             self._direction = speed > 0
+             self._encoder_dist_target = steps
+             self._motor_stopping = False
+             self._motor_running = True
+             self._encoder_dist = 0
+             self._encoder_speed_target = abs(speed)
+             self._power = abs(power) #TODO: initial power must be a function of desired speed
+             self._power_actual = abs(power) #TODO: initial power must be a function of desired speed
+             self._pin_duty = self._pin_forward if self._direction else self._pin_backward
+             self._pin_reverse = self._pin_backward if self._direction else self._pin_forward
+             self._pigpio.write(self._pin_reverse, 0)
+             self._pigpio.set_PWM_dutycycle(self._pin_duty, self._power)
+             self._pigpio.write(self._pin_enable, True)
+             self._motor_lock.release()
+             if elapse > 0:
+                 time.sleep(elapse)
+                 self.stop()
+
+         # def stop(self):
+         #     self._motor_lock.acquire()
+         #     self._motor_stopping = False
+         #     self._motor_running = False
+         #     self._pigpio.write(self._pin_forward, 0)
+         #     self._pigpio.write(self._pin_backward, 0)
+         #     self._motor_lock.release()
+
+         # def distance(self):
+         #     return self._encoder_dist
+
+         # def speed(self):
+         #     return self._encoder_speed
+
+         def stopping(self):
+             return self._motor_stopping
+
+         def running(self):
+             return self._motor_running
+
+         def adjust_power(self, power_delta):
+             self._power_actual = min(max(self._power + power_delta, 0), 100)
+             self._pigpio.set_PWM_dutycycle(self._pin_duty, self._power_actual)
 
     class TwinMotorsEncoder(object):
         def __init__(self, apigpio, pin_enable, pin_forward_left, pin_backward_left, pin_encoder_left, pin_forward_right, pin_backward_right, pin_encoder_right):

rotary_encoder/motorencoder.py

@@ -0,0 +1,83 @@
+import pigpio
+import threading
+
+from rotarydecoder import RotaryDecoder
+
+class MotorEncoder:
+    """
+        Class that handles rotary decoder motors modelisation
+
+        The support class RotaryDecoder decodes mechanical rotary encoder
+        pulses. See the file for more.
+
+        Every movement method must acquire lock in order not to have
+        concurrency problems on GPIO READ/WRITE
+    """
+
+    # default constructor
+    def __init__(self, pi, enable_pin, forward_pin, backward_pin, encoder_feedback_pin):
+        # setting pin variables
+        self._pi = pi
+        self._enable_pin = enable_pin
+        self._forward_pin = forward_pin
+        self._backward_pin = backward_pin
+        self._encoder_feedback_pin = encoder_feedback_pin
+
+        # setting movement variables
+        self._direction = 0
+        self._distance = 0
+        self._ticks = 0
+        self._PWM_value = 0
+        self._speed = 0
+        self._is_moving = False
+
+        # other
+        self._callback = self._pi.callback()
+        self._motor_lock = threading.RLock()
+        self._rotary_decoder = RotaryDecoder(pi, encoder_feedback_pin, callback)
+
+
+    # GETTERS
+
+    # ticks
+    def ticks(self):
+        return self._ticks
+
+    # distance
+    def distance(self):
+        return self._distance
+
+    # direction
+    def direction(self):
+        return self._direction
+
+    # speed
+    def speed(self):
+        return self._speed
+
+    # is_moving
+    def is_moving(self):
+        return self._is_moving
+
+
+    # MOVEMENT
+
+    def control(self, ):
+
+
+    """ The stop function acquires the lock to operate on motor
+        then writes a 0 on movement pins to stop the motor
+        and releases the lock afterwards """
+    def stop(self):
+        self._motor_lock.acquire()
+        self._pi.write(self._backward_pin, 0)
+        self._pi.write(self._forward_pin, 0)
+        self._is_moving = False
+        self._motor_lock.release()
+
+    # OTHER
+
+    # callback cancelling
+    def cancel_callback(self):
+        self._rotary_decoder.cancel()
+

rotary_encoder/rotarydecoder.py

@@ -0,0 +1,40 @@
+#!/usr/bin/env python
+
+import pigpio
+
+class RotaryDecoder:
+
+   """ Class to decode mechanical rotary encoder pulses """
+
+   def __init__(self, pi, encoder_feedback_pin, callback):
+
+      self.pi = pi
+      self.encoder_feedback_pin = encoder_feedback_pin   # encoder feedback pin
+      self.callback = callback   # callback function on event
+      self.value = 0             # value of encoder feedback
+
+      # setting up GPIO
+      self.pi.set_mode(encoder_feedback_pin, pigpio.INPUT)
+      self.pi.set_pull_up_down(encoder_feedback_pin, pigpio.PUD_UP)
+
+      # callback function on EITHER_EDGE
+      self.callback_trigger = self.pi.callback(encoder_feedback_pin, pigpio.EITHER_EDGE, self._pulse)
+
+   # pulse is the callback function on EITHER_EDGE
+   # it returns a 1 if the feedback square wave is on a falling edge
+   def _pulse(self, level):
+      self.levB = level;
+
+      if self.levB == 1:
+          self.callback(1)
+
+
+   def cancel(self):
+
+      """
+      Cancel the rotary encoder decoder.
+      """
+
+      self.callback_trigger.cancel()
+
+