Fix #421

Added SPI tests, simplified the shared SPI software bus implementation,
and fixed several protocol errors in our MCP3xxx classes (the x2 and x1
protocols were wrong)

tests/test_spi.py

@@ -0,0 +1,170 @@
+from __future__ import (
+    unicode_literals,
+    absolute_import,
+    print_function,
+    division,
+    )
+str = type('')
+
+
+import sys
+import mock
+import pytest
+from collections import namedtuple
+
+from gpiozero import *
+from gpiozero.pins.mock import MockPin, MockSPIDevice
+from gpiozero.spi import *
+
+
+def setup_function(function):
+    import gpiozero.devices
+    gpiozero.devices.pin_factory = MockPin
+
+def teardown_function(function):
+    MockPin.clear_pins()
+
+
+def test_spi_hardware_params():
+    with mock.patch('gpiozero.spi.SpiDev') as spidev:
+        with SPI() as device:
+            assert isinstance(device, SPIHardwareInterface)
+        with SPI(port=0, device=0) as device:
+            assert isinstance(device, SPIHardwareInterface)
+        with SPI(port=0, device=1) as device:
+            assert isinstance(device, SPIHardwareInterface)
+        with SPI(clock_pin=11) as device:
+            assert isinstance(device, SPIHardwareInterface)
+        with SPI(clock_pin=11, mosi_pin=10, select_pin=8) as device:
+            assert isinstance(device, SPIHardwareInterface)
+        with SPI(clock_pin=11, mosi_pin=10, select_pin=7) as device:
+            assert isinstance(device, SPIHardwareInterface)
+        with SPI(shared=True) as device:
+            assert isinstance(device, SharedSPIHardwareInterface)
+        with pytest.raises(ValueError):
+            SPI(port=1)
+        with pytest.raises(ValueError):
+            SPI(device=2)
+        with pytest.raises(ValueError):
+            SPI(port=0, clock_pin=12)
+        with pytest.raises(ValueError):
+            SPI(foo='bar')
+
+def test_spi_software_params():
+    with mock.patch('gpiozero.spi.SpiDev') as spidev:
+        with SPI(select_pin=6) as device:
+            assert isinstance(device, SPISoftwareInterface)
+        with SPI(clock_pin=11, mosi_pin=9, miso_pin=10) as device:
+            assert isinstance(device, SPISoftwareInterface)
+        with SPI(select_pin=6, shared=True) as device:
+            assert isinstance(device, SharedSPISoftwareInterface)
+    # Ensure software fallback works when SpiDev isn't present
+    with SPI() as device:
+        assert isinstance(device, SPISoftwareInterface)
+
+def test_spi_hardware_conflict():
+    with mock.patch('gpiozero.spi.SpiDev') as spidev:
+        with LED(11) as led:
+            with pytest.raises(GPIOPinInUse):
+                SPI(port=0, device=0)
+
+def test_spi_hardware_read():
+    with mock.patch('gpiozero.spi.SpiDev') as spidev:
+        spidev.return_value.xfer2.side_effect = lambda data: list(range(10))[:len(data)]
+        with SPI() as device:
+            assert device.read(3) == [0, 1, 2]
+            assert device.read(6) == list(range(6))
+
+def test_spi_hardware_write():
+    with mock.patch('gpiozero.spi.SpiDev') as spidev:
+        spidev.return_value.xfer2.side_effect = lambda data: list(range(10))[:len(data)]
+        with SPI() as device:
+            assert device.write([0, 1, 2]) == 3
+            assert spidev.return_value.xfer2.called_with([0, 1, 2])
+            assert device.write(list(range(6))) == 6
+            assert spidev.return_value.xfer2.called_with(list(range(6)))
+
+def test_spi_hardware_modes():
+    with mock.patch('gpiozero.spi.SpiDev') as spidev:
+        spidev.return_value.mode = 0
+        spidev.return_value.lsbfirst = False
+        spidev.return_value.cshigh = True
+        spidev.return_value.bits_per_word = 8
+        with SPI() as device:
+            assert device.clock_mode == 0
+            assert not device.clock_polarity
+            assert not device.clock_phase
+            device.clock_polarity = False
+            assert device.clock_mode == 0
+            device.clock_polarity = True
+            assert device.clock_mode == 2
+            device.clock_phase = True
+            assert device.clock_mode == 3
+            assert not device.lsb_first
+            assert device.select_high
+            assert device.bits_per_word == 8
+            device.select_high = False
+            device.lsb_first = True
+            device.bits_per_word = 12
+            assert not spidev.return_value.cshigh
+            assert spidev.return_value.lsbfirst
+            assert spidev.return_value.bits_per_word == 12
+
+def test_spi_software_read():
+    class SPISlave(MockSPIDevice):
+        def on_start(self):
+            super(SPISlave, self).on_start()
+            for i in range(10):
+                self.tx_word(i)
+    with SPISlave(11, 10, 9, 8) as slave, SPI() as master:
+        assert master.read(3) == [0, 1, 2]
+        assert master.read(6) == [0, 1, 2, 3, 4, 5]
+        slave.clock_phase = True
+        master.clock_phase = True
+        assert master.read(3) == [0, 1, 2]
+        assert master.read(6) == [0, 1, 2, 3, 4, 5]
+
+def test_spi_software_write():
+    with MockSPIDevice(11, 10, 9, 8) as test_device, SPI() as master:
+        master.write([0])
+        assert test_device.rx_word() == 0
+        master.write([2, 0])
+        assert test_device.rx_word() == 512
+        master.write([0, 1, 1])
+        assert test_device.rx_word() == 257
+
+def test_spi_software_clock_mode():
+    with SPI() as master:
+        assert master.clock_mode == 0
+        assert not master.clock_polarity
+        assert not master.clock_phase
+        master.clock_polarity = False
+        assert master.clock_mode == 0
+        master.clock_polarity = True
+        assert master.clock_mode == 2
+        master.clock_phase = True
+        assert master.clock_mode == 3
+        master.clock_mode = 0
+        assert not master.clock_polarity
+        assert not master.clock_phase
+        with pytest.raises(ValueError):
+            master.clock_mode = 5
+
+def test_spi_software_attr():
+    with SPI() as master:
+        assert not master.lsb_first
+        assert not master.select_high
+        assert master.bits_per_word == 8
+        master.bits_per_word = 12
+        assert master.bits_per_word == 12
+        master.lsb_first = True
+        assert master.lsb_first
+        master.select_high = True
+        assert master.select_high
+        with pytest.raises(ValueError):
+            master.bits_per_word = 0
+
+
+# XXX Test two simultaneous SPI devices sharing clock, MOSI, and MISO, with
+# separate select pins (including threaded tests which attempt simultaneous
+# reading/writing)

tests/test_spi_devices.py

@@ -0,0 +1,339 @@
+from __future__ import (
+    unicode_literals,
+    absolute_import,
+    print_function,
+    division,
+    )
+str = type('')
+
+
+import sys
+import pytest
+from collections import namedtuple
+try:
+    from math import isclose
+except ImportError:
+    from gpiozero.compat import isclose
+
+from gpiozero import *
+from gpiozero.pins.mock import MockSPIDevice, MockPin
+
+
+def setup_function(function):
+    import gpiozero.devices
+    gpiozero.devices.pin_factory = MockPin
+
+def teardown_function(function):
+    MockPin.clear_pins()
+
+def clamp(v, min_value, max_value):
+    return min(max_value, max(min_value, v))
+
+def scale(v, ref, bits):
+    v /= ref
+    vmin = -(2 ** bits)
+    vmax = -vmin - 1
+    vrange = vmax - vmin
+    return int(((v + 1) / 2.0) * vrange + vmin)
+
+
+class MockMCP3xxx(MockSPIDevice):
+    def __init__(
+            self, clock_pin, mosi_pin, miso_pin, select_pin=None,
+            channels=8, bits=10):
+        super(MockMCP3xxx, self).__init__(
+            clock_pin, mosi_pin, miso_pin, select_pin)
+        self.vref = 3.3
+        self.channels = [0.0] * channels
+        self.channel_bits = 3
+        self.bits = bits
+        self.state = 'idle'
+
+    def on_start(self):
+        super(MockMCP3xxx, self).on_start()
+        self.state = 'idle'
+
+    def on_bit(self):
+        if self.state == 'idle':
+            if self.rx_buf[-1]:
+                self.state = 'mode'
+                self.rx_buf = []
+        elif self.state == 'mode':
+            if self.rx_buf[-1]:
+                self.state = 'single'
+            else:
+                self.state = 'diff'
+            self.rx_buf = []
+        elif self.state in ('single', 'diff'):
+            if len(self.rx_buf) == self.channel_bits:
+                self.on_result(self.state == 'diff', self.rx_word())
+                self.state = 'result'
+        elif self.state == 'result':
+            if not self.tx_buf:
+                self.state = 'idle'
+                self.rx_buf = []
+        else:
+            assert False
+
+    def on_result(self, differential, channel):
+        if differential:
+            pos_channel = channel
+            neg_channel = pos_channel ^ 1
+            result = self.channels[pos_channel] - self.channels[neg_channel]
+            result = clamp(result, 0, self.vref)
+        else:
+            result = clamp(self.channels[channel], 0, self.vref)
+        result = scale(result, self.vref, self.bits)
+        self.tx_word(result, self.bits + 2)
+
+
+class MockMCP3xx1(MockMCP3xxx):
+    def __init__(self, clock_pin, mosi_pin, miso_pin, select_pin=None, bits=10):
+        super(MockMCP3xx1, self).__init__(
+            clock_pin, mosi_pin, miso_pin, select_pin, channels=2, bits=bits)
+
+    def on_start(self):
+        super(MockMCP3xx1, self).on_start()
+        result = self.channels[0] - self.channels[1]
+        result = clamp(result, 0, self.vref)
+        result = scale(result, self.vref, self.bits)
+        self.tx_word(result, self.bits + 3)
+
+    def on_bit(self):
+        pass
+
+
+class MockMCP3xx2(MockMCP3xxx):
+    def __init__(
+            self, clock_pin, mosi_pin, miso_pin, select_pin=None,
+            bits=10):
+        super(MockMCP3xx2, self).__init__(
+            clock_pin, mosi_pin, miso_pin, select_pin, channels=2, bits=bits)
+        self.channel_bits = 1
+
+
+class MockMCP33xx(MockMCP3xxx):
+    def __init__(
+            self, clock_pin, mosi_pin, miso_pin, select_pin=None,
+            channels=8):
+        super(MockMCP33xx, self).__init__(
+            clock_pin, mosi_pin, miso_pin, select_pin, channels, 12)
+
+    def on_result(self, differential, channel):
+        if differential:
+            pos_channel = channel
+            neg_channel = pos_channel ^ 1
+            result = self.channels[pos_channel] - self.channels[neg_channel]
+            result = clamp(result, -self.vref, self.vref)
+        else:
+            result = clamp(self.channels[channel], 0, self.vref)
+        result = scale(result, self.vref, self.bits)
+        if result < 0:
+            result += 8192
+        self.tx_word(result, self.bits + 3)
+
+
+class MockMCP3001(MockMCP3xx1):
+    def __init__(self, clock_pin, mosi_pin, miso_pin, select_pin=None):
+        super(MockMCP3001, self).__init__(
+            clock_pin, mosi_pin, miso_pin, select_pin, bits=10)
+
+
+class MockMCP3002(MockMCP3xx2):
+    def __init__(self, clock_pin, mosi_pin, miso_pin, select_pin=None):
+        super(MockMCP3002, self).__init__(
+            clock_pin, mosi_pin, miso_pin, select_pin, bits=10)
+
+
+class MockMCP3004(MockMCP3xxx):
+    def __init__(self, clock_pin, mosi_pin, miso_pin, select_pin=None):
+        super(MockMCP3004, self).__init__(
+            clock_pin, mosi_pin, miso_pin, select_pin, channels=4, bits=10)
+
+
+class MockMCP3008(MockMCP3xxx):
+    def __init__(self, clock_pin, mosi_pin, miso_pin, select_pin=None):
+        super(MockMCP3008, self).__init__(
+            clock_pin, mosi_pin, miso_pin, select_pin, channels=8, bits=10)
+
+
+class MockMCP3201(MockMCP3xx1):
+    def __init__(self, clock_pin, mosi_pin, miso_pin, select_pin=None):
+        super(MockMCP3201, self).__init__(
+            clock_pin, mosi_pin, miso_pin, select_pin, bits=12)
+
+
+class MockMCP3202(MockMCP3xx2):
+    def __init__(self, clock_pin, mosi_pin, miso_pin, select_pin=None):
+        super(MockMCP3202, self).__init__(
+            clock_pin, mosi_pin, miso_pin, select_pin, bits=12)
+
+
+class MockMCP3204(MockMCP3xxx):
+    def __init__(self, clock_pin, mosi_pin, miso_pin, select_pin=None):
+        super(MockMCP3204, self).__init__(
+            clock_pin, mosi_pin, miso_pin, select_pin, channels=4, bits=12)
+
+
+class MockMCP3208(MockMCP3xxx):
+    def __init__(self, clock_pin, mosi_pin, miso_pin, select_pin=None):
+        super(MockMCP3208, self).__init__(
+            clock_pin, mosi_pin, miso_pin, select_pin, channels=8, bits=12)
+
+
+class MockMCP3301(MockMCP3xxx):
+    def __init__(self, clock_pin, mosi_pin, miso_pin, select_pin=None):
+        super(MockMCP3301, self).__init__(
+            clock_pin, mosi_pin, miso_pin, select_pin, channels=2, bits=12)
+
+    def on_start(self):
+        super(MockMCP3301, self).on_start()
+        result = self.channels[0] - self.channels[1]
+        result = clamp(result, -self.vref, self.vref)
+        result = scale(result, self.vref, self.bits)
+        if result < 0:
+            result += 8192
+        self.tx_word(result, self.bits + 4)
+
+
+class MockMCP3302(MockMCP33xx):
+    def __init__(self, clock_pin, mosi_pin, miso_pin, select_pin=None):
+        super(MockMCP3302, self).__init__(
+            clock_pin, mosi_pin, miso_pin, select_pin, channels=4)
+
+
+class MockMCP3304(MockMCP33xx):
+    def __init__(self, clock_pin, mosi_pin, miso_pin, select_pin=None):
+        super(MockMCP3304, self).__init__(
+            clock_pin, mosi_pin, miso_pin, select_pin, channels=8)
+
+
+def single_mcp_test(mock, pot, channel, bits):
+    scale = 2**bits
+    tolerance = 1 / scale
+    mock.channels[channel] = 0.0
+    assert pot.raw_value == 0
+    assert isclose(pot.value, 0.0, abs_tol=tolerance)
+    mock.channels[channel] = mock.vref / 2
+    assert pot.raw_value == (scale / 2) - 1
+    assert isclose(pot.value, 0.5, abs_tol=tolerance)
+    mock.channels[channel] = mock.vref
+    assert pot.raw_value == scale - 1
+    assert isclose(pot.value, 1.0, abs_tol=tolerance)
+
+def differential_mcp_test(mock, pot, pos_channel, neg_channel, bits, full=False):
+    scale = 2**bits
+    tolerance = 1 / scale
+    mock.channels[pos_channel] = 0.0
+    mock.channels[neg_channel] = 0.0
+    assert pot.raw_value == 0
+    assert isclose(pot.value, 0.0, abs_tol=tolerance)
+    mock.channels[pos_channel] = mock.vref / 2
+    assert pot.raw_value == (scale / 2) - 1
+    assert isclose(pot.value, 0.5, abs_tol=tolerance)
+    mock.channels[pos_channel] = mock.vref
+    assert pot.raw_value == scale - 1
+    assert isclose(pot.value, 1.0, abs_tol=tolerance)
+    mock.channels[neg_channel] = mock.vref / 2
+    assert pot.raw_value == (scale / 2) - 1
+    assert isclose(pot.value, 0.5, abs_tol=tolerance)
+    mock.channels[pos_channel] = mock.vref / 2
+    assert pot.raw_value == 0
+    assert isclose(pot.value, 0.0, abs_tol=tolerance)
+    mock.channels[pos_channel] = 0.0
+    mock.channels[neg_channel] = mock.vref
+    if full:
+        assert pot.raw_value == -scale
+        assert isclose(pot.value, -1.0, abs_tol=tolerance)
+    else:
+        assert pot.raw_value == 0
+        assert isclose(pot.value, 0.0, abs_tol=tolerance)
+
+
+def test_MCP3001():
+    mock = MockMCP3001(11, 10, 9, 8)
+    with MCP3001() as pot:
+        differential_mcp_test(mock, pot, 0, 1, 10)
+
+def test_MCP3002():
+    mock = MockMCP3002(11, 10, 9, 8)
+    with pytest.raises(ValueError):
+        MCP3002(channel=5)
+    with MCP3002(channel=1) as pot:
+        single_mcp_test(mock, pot, 1, 10)
+    with MCP3002(channel=1, differential=True) as pot:
+        differential_mcp_test(mock, pot, 1, 0, 10)
+
+def test_MCP3004():
+    mock = MockMCP3004(11, 10, 9, 8)
+    with pytest.raises(ValueError):
+        MCP3004(channel=5)
+    with MCP3004(channel=3) as pot:
+        single_mcp_test(mock, pot, 3, 10)
+    with MCP3004(channel=3, differential=True) as pot:
+        differential_mcp_test(mock, pot, 3, 2, 10)
+
+def test_MCP3008():
+    mock = MockMCP3008(11, 10, 9, 8)
+    with pytest.raises(ValueError):
+        MCP3008(channel=9)
+    with MCP3008(channel=0) as pot:
+        single_mcp_test(mock, pot, 0, 10)
+    with MCP3008(channel=0, differential=True) as pot:
+        differential_mcp_test(mock, pot, 0, 1, 10)
+
+def test_MCP3201():
+    mock = MockMCP3201(11, 10, 9, 8)
+    with MCP3201() as pot:
+        differential_mcp_test(mock, pot, 0, 1, 12)
+
+def test_MCP3202():
+    mock = MockMCP3202(11, 10, 9, 8)
+    with pytest.raises(ValueError):
+        MCP3202(channel=5)
+    with MCP3202(channel=1) as pot:
+        single_mcp_test(mock, pot, 1, 12)
+    with MCP3202(channel=1, differential=True) as pot:
+        differential_mcp_test(mock, pot, 1, 0, 12)
+
+def test_MCP3204():
+    mock = MockMCP3204(11, 10, 9, 8)
+    with pytest.raises(ValueError):
+        MCP3204(channel=5)
+    with MCP3204(channel=1) as pot:
+        single_mcp_test(mock, pot, 1, 12)
+    with MCP3204(channel=1, differential=True) as pot:
+        differential_mcp_test(mock, pot, 1, 0, 12)
+
+def test_MCP3208():
+    mock = MockMCP3208(11, 10, 9, 8)
+    with pytest.raises(ValueError):
+        MCP3208(channel=9)
+    with MCP3208(channel=7) as pot:
+        single_mcp_test(mock, pot, 7, 12)
+    with MCP3208(channel=7, differential=True) as pot:
+        differential_mcp_test(mock, pot, 7, 6, 12)
+
+def test_MCP3301():
+    mock = MockMCP3301(11, 10, 9, 8)
+    with MCP3301() as pot:
+        differential_mcp_test(mock, pot, 0, 1, 12, full=True)
+
+def test_MCP3302():
+    mock = MockMCP3302(11, 10, 9, 8)
+    with pytest.raises(ValueError):
+        MCP3302(channel=4)
+    with MCP3302(channel=0) as pot:
+        single_mcp_test(mock, pot, 0, 12)
+    with MCP3302(channel=0, differential=True) as pot:
+        differential_mcp_test(mock, pot, 0, 1, 12, full=True)
+
+def test_MCP3304():
+    mock = MockMCP3304(11, 10, 9, 8)
+    with pytest.raises(ValueError):
+        MCP3304(channel=9)
+    with MCP3304(channel=5) as pot:
+        single_mcp_test(mock, pot, 5, 12)
+    with MCP3304(channel=5, differential=True) as pot:
+        differential_mcp_test(mock, pot, 5, 4, 12, full=True)
+