Merge pull request gpiozero#430 from waveform80/spi-tests

Fix gpiozero#421

Author: waveform80

docs/images/spi_device_hierarchy.dot

@@ -11,6 +11,7 @@ digraph classes {
     MCP3xxx;
     MCP30xx;
     MCP32xx;
+    MCP3xx2;
     MCP33xx;
 
     /* Concrete classes */
@@ -21,6 +22,8 @@ digraph classes {
     MCP30xx->MCP3xxx;
     MCP32xx->MCP3xxx;
     MCP33xx->MCP3xxx;
+    MCP3xx2->MCP3xxx;
+
     MCP3001->MCP30xx;
     MCP3002->MCP30xx;
     MCP3004->MCP30xx;
@@ -29,6 +32,8 @@ digraph classes {
     MCP3202->MCP32xx;
     MCP3204->MCP32xx;
     MCP3208->MCP32xx;
+    MCP3002->MCP3xx2;
+    MCP3202->MCP3xx2;
     MCP3301->MCP33xx;
     MCP3302->MCP33xx;
     MCP3304->MCP33xx;

docs/images/spi_device_hierarchy.pdf

@@ -49,6 +49,9 @@ class SPIError(GPIOZeroError):
 class SPIBadArgs(SPIError, ValueError):
     "Error raised when invalid arguments are given while constructing :class:`SPIDevice`"
 
+class SPIBadChannel(SPIError, ValueError):
+    "Error raised when an invalid channel is given to an :class:`AnalogInputDevice`"
+
 class GPIODeviceError(GPIOZeroError):
     "Base class for errors specific to the GPIODevice hierarchy"
 

docs/images/spi_device_hierarchy.png

@@ -54,7 +54,8 @@ def __new__(cls, number):
             self._when_changed = None
             self.clear_states()
             return self
-        if old_pin.__class__ != cls:
+        # Ensure the pin class expected supports PWM (or not)
+        if issubclass(cls, MockPWMPin) != isinstance(old_pin, MockPWMPin):
             raise ValueError('pin %d is already in use as a %s' % (number, old_pin.__class__.__name__))
         return old_pin
 
@@ -249,7 +250,6 @@ class MockPWMPin(MockPin):
     """
     This derivative of :class:`MockPin` adds PWM support.
     """
-
     def __init__(self, number):
         super(MockPWMPin, self).__init__()
         self._frequency = None
@@ -275,3 +275,141 @@ def _set_frequency(self, value):
         if value is None:
             self._change_state(0.0)
 
+
+class MockSPIClockPin(MockPin):
+    """
+    This derivative of :class:`MockPin` is intended to be used as the clock pin
+    of a mock SPI device. It is not intended for direct construction in tests;
+    rather, construct a :class:`MockSPIDevice` with various pin numbers, and
+    this class will be used for the clock pin.
+    """
+    def __init__(self, number):
+        super(MockSPIClockPin, self).__init__()
+        if not hasattr(self, 'spi_devices'):
+            self.spi_devices = []
+
+    def _set_state(self, value):
+        super(MockSPIClockPin, self)._set_state(value)
+        for dev in self.spi_devices:
+            dev.on_clock()
+
+
+class MockSPISelectPin(MockPin):
+    """
+    This derivative of :class:`MockPin` is intended to be used as the select
+    pin of a mock SPI device. It is not intended for direct construction in
+    tests; rather, construct a :class:`MockSPIDevice` with various pin numbers,
+    and this class will be used for the select pin.
+    """
+    def __init__(self, number):
+        super(MockSPISelectPin, self).__init__()
+        if not hasattr(self, 'spi_device'):
+            self.spi_device = None
+
+    def _set_state(self, value):
+        super(MockSPISelectPin, self)._set_state(value)
+        if self.spi_device:
+            self.spi_device.on_select()
+
+
+class MockSPIDevice(object):
+    def __init__(
+            self, clock_pin, mosi_pin, miso_pin, select_pin=None,
+            clock_polarity=False, clock_phase=False, lsb_first=False,
+            bits_per_word=8, select_high=False):
+        self.clock_pin = MockSPIClockPin(clock_pin)
+        self.mosi_pin = None if mosi_pin is None else MockPin(mosi_pin)
+        self.miso_pin = None if miso_pin is None else MockPin(miso_pin)
+        self.select_pin = None if select_pin is None else MockSPISelectPin(select_pin)
+        self.clock_polarity = clock_polarity
+        self.clock_phase = clock_phase
+        self.lsb_first = lsb_first
+        self.bits_per_word = bits_per_word
+        self.select_high = select_high
+        self.rx_bit = 0
+        self.rx_buf = []
+        self.tx_buf = []
+        self.clock_pin.spi_devices.append(self)
+        self.select_pin.spi_device = self
+
+    def __enter__(self):
+        return self
+
+    def __exit__(self, exc_type, exc_value, exc_tb):
+        self.close()
+
+    def close(self):
+        if self in self.clock_pin.spi_devices:
+            self.clock_pin.spi_devices.remove(self)
+        if self.select_pin is not None:
+            self.select_pin.spi_device = None
+
+    def on_select(self):
+        if self.select_pin.state == self.select_high:
+            self.on_start()
+
+    def on_clock(self):
+        # Don't do anything if this SPI device isn't currently selected
+        if self.select_pin is None or self.select_pin.state == self.select_high:
+            # The XOR of the clock pin's values, polarity and phase indicates
+            # whether we're meant to be acting on this edge
+            if self.clock_pin.state ^ self.clock_polarity ^ self.clock_phase:
+                self.rx_bit += 1
+                if self.mosi_pin is not None:
+                    self.rx_buf.append(self.mosi_pin.state)
+                if self.miso_pin is not None:
+                    try:
+                        tx_value = self.tx_buf.pop(0)
+                    except IndexError:
+                        tx_value = 0
+                    if tx_value:
+                        self.miso_pin.drive_high()
+                    else:
+                        self.miso_pin.drive_low()
+                self.on_bit()
+
+    def on_start(self):
+        """
+        Override this in descendents to detect when the mock SPI device's
+        select line is activated.
+        """
+        self.rx_bit = 0
+        self.rx_buf = []
+        self.tx_buf = []
+
+    def on_bit(self):
+        """
+        Override this in descendents to react to receiving a bit.
+
+        The :attr:`rx_bit` attribute gives the index of the bit received (this
+        is reset to 0 by default by :meth:`on_select`). The :attr:`rx_buf`
+        sequence gives the sequence of 1s and 0s that have been recevied so
+        far. The :attr:`tx_buf` sequence gives the sequence of 1s and 0s to
+        transmit on the next clock pulses. All these attributes can be modified
+        within this method.
+
+        The :meth:`rx_word` and :meth:`tx_word` methods can also be used to
+        read and append to the buffers using integers instead of bool bits.
+        """
+        pass
+
+    def rx_word(self):
+        result = 0
+        bits = reversed(self.rx_buf) if self.lsb_first else self.rx_buf
+        for bit in bits:
+            result <<= 1
+            result |= bit
+        return result
+
+    def tx_word(self, value, bits_per_word=None):
+        if bits_per_word is None:
+            bits_per_word = self.bits_per_word
+        bits = [0] * bits_per_word
+        for bit in range(bits_per_word):
+            bits[bit] = value & 1
+            value >>= 1
+        assert not value
+        if not self.lsb_first:
+            bits = reversed(bits)
+        self.tx_buf.extend(bits)
+

docs/images/spi_device_hierarchy.svg

@@ -85,16 +85,16 @@ def _set_clock_mode(self, value):
         self._device.mode = value
 
     def _get_clock_polarity(self):
-        return bool(self.mode & 2)
+        return bool(self.clock_mode & 2)
 
     def _set_clock_polarity(self, value):
-        self.mode = self.mode & (~2) | (bool(value) << 1)
+        self.clock_mode = self.clock_mode & (~2) | (bool(value) << 1)
 
     def _get_clock_phase(self):
-        return bool(self.mode & 1)
+        return bool(self.clock_mode & 1)
 
     def _set_clock_phase(self, value):
-        self.mode = self.mode & (~1) | bool(value)
+        self.clock_mode = self.clock_mode & (~1) | bool(value)
 
     def _get_lsb_first(self):
         return self._device.lsbfirst
@@ -130,9 +130,6 @@ def __init__(self, clock_pin, mosi_pin, miso_pin):
         self.miso = None
         super(SPISoftwareBus, self).__init__()
         self.lock = RLock()
-        self.clock_phase = False
-        self.lsb_first = False
-        self.bits_per_word = 8
         try:
             self.clock = OutputDevice(clock_pin, active_high=True)
             if mosi_pin is not None:
@@ -166,33 +163,27 @@ def closed(self):
     def _shared_key(cls, clock_pin, mosi_pin, miso_pin):
         return (clock_pin, mosi_pin, miso_pin)
 
-    def read(self, n):
-        return self.transfer((0,) * n)
-
-    def write(self, data):
-        return len(self.transfer(data))
-
-    def transfer(self, data):
+    def transfer(self, data, clock_phase=False, lsb_first=False, bits_per_word=8):
         """
         Writes data (a list of integer words where each word is assumed to have
         :attr:`bits_per_word` bits or less) to the SPI interface, and reads an
         equivalent number of words, returning them as a list of integers.
         """
         result = []
         with self.lock:
-            shift = operator.lshift if self.lsb_first else operator.rshift
+            shift = operator.lshift if lsb_first else operator.rshift
             for write_word in data:
-                mask = 1 if self.lsb_first else 1 << (self.bits_per_word - 1)
+                mask = 1 if lsb_first else 1 << (bits_per_word - 1)
                 read_word = 0
-                for _ in range(self.bits_per_word):
+                for _ in range(bits_per_word):
                     if self.mosi is not None:
                         self.mosi.value = bool(write_word & mask)
                     self.clock.on()
-                    if self.miso is not None and not self.clock_phase:
+                    if self.miso is not None and not clock_phase:
                         if self.miso.value:
                             read_word |= mask
                     self.clock.off()
-                    if self.miso is not None and self.clock_phase:
+                    if self.miso is not None and clock_phase:
                         if self.miso.value:
                             read_word |= mask
                     mask = shift(mask, 1)
@@ -205,6 +196,9 @@ def __init__(self, clock_pin, mosi_pin, miso_pin, select_pin):
         self._bus = None
         super(SPISoftwareInterface, self).__init__(select_pin, active_high=False)
         try:
+            self._clock_phase = False
+            self._lsb_first = False
+            self._bits_per_word = 8
             self._bus = SPISoftwareBus(clock_pin, mosi_pin, miso_pin)
         except:
             self.close()
@@ -230,16 +224,17 @@ def __repr__(self):
             return "software SPI closed"
 
     def read(self, n):
-        return self._bus.read(n)
+        return self.transfer((0,) * n)
 
     def write(self, data):
-        return self._bus.write(data)
+        return len(self.transfer(data))
 
     def transfer(self, data):
         with self._bus.lock:
             self.on()
             try:
-                return self._bus.transfer(data)
+                return self._bus.transfer(
+                    data, self._clock_phase, self._lsb_first, self._bits_per_word)
             finally:
                 self.off()
 
@@ -250,40 +245,37 @@ def _set_clock_mode(self, value):
         value = int(value)
         if not 0 <= value <= 3:
             raise ValueError('clock_mode must be a value between 0 and 3 inclusive')
-        with self._bus.lock:
-            self._bus.clock.active_high = not (value & 2)
-            self._bus.clock.off()
-            self._bus.clock_phase = bool(value & 1)
+        self.clock_polarity = bool(value & 2)
+        self.clock_phase = bool(value & 1)
 
     def _get_clock_polarity(self):
-        return not self._bus.clock.active_high
+        with self._bus.lock:
+            return not self._bus.clock.active_high
 
     def _set_clock_polarity(self, value):
         with self._bus.lock:
             self._bus.clock.active_high = not value
+            self._bus.clock.off()
 
     def _get_clock_phase(self):
-        return self._bus.clock_phase
+        return self._clock_phase
 
     def _set_clock_phase(self, value):
-        with self._bus.lock:
-            self._bus.clock_phase = bool(value)
+        self._clock_phase = bool(value)
 
     def _get_lsb_first(self):
-        return self._bus.lsb_first
+        return self._lsb_first
 
     def _set_lsb_first(self, value):
-        with self._bus.lock:
-            self._bus.lsb_first = bool(value)
+        self._lsb_first = bool(value)
 
     def _get_bits_per_word(self):
-        return self._bus.bits_per_word
+        return self._bits_per_word
 
     def _set_bits_per_word(self, value):
         if value < 1:
             raise ValueError('bits_per_word must be positive')
-        with self._bus.lock:
-            self._bus.bits_per_word = int(value)
+        self._bits_per_word = int(value)
 
     def _get_select_high(self):
         return self.active_high