Merge pull request #31 from kbsriram/fix-phase

Move SPI bit writes to the right clock phase.

adafruit_bitbangio.py

@@ -323,9 +323,6 @@ def __init__(
         self._mosi = None
         self._miso = None
 
-        self.configure()
-        self.unlock()
-
         # Set pins as outputs/inputs.
         self._sclk = DigitalInOut(clock)
         self._sclk.switch_to_output()
@@ -338,6 +335,9 @@ def __init__(
             self._miso = DigitalInOut(MISO)
             self._miso.switch_to_input()
 
+        self.configure()
+        self.unlock()
+
     def deinit(self) -> None:
         """Free any hardware used by the object."""
         self._sclk.deinit()
@@ -372,12 +372,30 @@ def configure(
             self._bits = bits
             self._half_period = (1 / self._baudrate) / 2  # 50% Duty Cyle delay
 
+            # Initialize the clock to the idle state. This is important to
+            # guarantee that the clock is at a known (idle) state before
+            # any read/write operations.
+            self._sclk.value = self._polarity
+
     def _wait(self, start: Optional[int] = None) -> float:
         """Wait for up to one half cycle"""
         while (start + self._half_period) > monotonic():
             pass
         return monotonic()  # Return current time
 
+    def _should_write(self, to_active: Literal[0, 1]) -> bool:
+        """Return true if a bit should be written on the given clock transition."""
+        # phase 0: write when active is 0
+        # phase 1: write when active is 1
+        return self._phase == to_active
+
+    def _should_read(self, to_active: Literal[0, 1]) -> bool:
+        """Return true if a bit should be read on the given clock transition."""
+        # phase 0: read when active is 1
+        # phase 1: read when active is 0
+        # Data is read on the idle->active transition only when the phase is 1
+        return self._phase == 1 - to_active
+
     def write(
         self, buffer: ReadableBuffer, start: int = 0, end: Optional[int] = None
     ) -> None:
@@ -392,24 +410,26 @@ def write(
 
         if self._check_lock():
             start_time = monotonic()
+            # Note: when we come here, our clock must always be its idle state.
             for byte in buffer[start:end]:
                 for bit_position in range(self._bits):
                     bit_value = byte & 0x80 >> bit_position
-                    # Set clock to base
-                    if not self._phase:  # Mode 0, 2
+                    # clock: idle, or has made an active->idle transition.
+                    if self._should_write(to_active=0):
                         self._mosi.value = bit_value
-                    self._sclk.value = not self._polarity
+                    # clock: wait in idle for half a period
                     start_time = self._wait(start_time)
-
-                    # Flip clock off base
-                    if self._phase:  # Mode 1, 3
+                    # clock: idle->active
+                    self._sclk.value = not self._polarity
+                    if self._should_write(to_active=1):
                         self._mosi.value = bit_value
-                    self._sclk.value = self._polarity
+                    # clock: wait in active for half a period
                     start_time = self._wait(start_time)
-
-            # Return pins to base positions
-            self._mosi.value = 0
-            self._sclk.value = self._polarity
+                    # clock: active->idle
+                    self._sclk.value = self._polarity
+            # clock: stay in idle for the last active->idle transition
+            # to settle.
+            start_time = self._wait(start_time)
 
     # pylint: disable=too-many-branches
     def readinto(
@@ -433,36 +453,38 @@ def readinto(
                 for bit_position in range(self._bits):
                     bit_mask = 0x80 >> bit_position
                     bit_value = write_value & 0x80 >> bit_position
-                    # Return clock to base
-                    self._sclk.value = self._polarity
-                    start_time = self._wait(start_time)
-                    # Handle read on leading edge of clock.
-                    if not self._phase:  # Mode 0, 2
+                    # clock: idle, or has made an active->idle transition.
+                    if self._should_write(to_active=0):
                         if self._mosi is not None:
                             self._mosi.value = bit_value
+                    # clock: wait half a period.
+                    start_time = self._wait(start_time)
+                    # clock: idle->active
+                    self._sclk.value = not self._polarity
+                    if self._should_read(to_active=1):
                         if self._miso.value:
                             # Set bit to 1 at appropriate location.
                             buffer[byte_position] |= bit_mask
                         else:
                             # Set bit to 0 at appropriate location.
                             buffer[byte_position] &= ~bit_mask
-                    # Flip clock off base
-                    self._sclk.value = not self._polarity
-                    start_time = self._wait(start_time)
-                    # Handle read on trailing edge of clock.
-                    if self._phase:  # Mode 1, 3
+                    if self._should_write(to_active=1):
                         if self._mosi is not None:
                             self._mosi.value = bit_value
+                    # clock: wait half a period
+                    start_time = self._wait(start_time)
+                    # Clock: active->idle
+                    self._sclk.value = self._polarity
+                    if self._should_read(to_active=0):
                         if self._miso.value:
                             # Set bit to 1 at appropriate location.
                             buffer[byte_position] |= bit_mask
                         else:
                             # Set bit to 0 at appropriate location.
                             buffer[byte_position] &= ~bit_mask
 
-            # Return pins to base positions
-            self._mosi.value = 0
-            self._sclk.value = self._polarity
+            # clock: wait another half period for the last transition.
+            start_time = self._wait(start_time)
 
     def write_readinto(
         self,
@@ -499,34 +521,34 @@ def write_readinto(
                         buffer_out[byte_position + out_start] & 0x80 >> bit_position
                     )
                     in_byte_position = byte_position + in_start
-                    # Return clock to 0
-                    self._sclk.value = self._polarity
-                    start_time = self._wait(start_time)
-                    # Handle read on leading edge of clock.
-                    if not self._phase:  # Mode 0, 2
+                    # clock: idle, or has made an active->idle transition.
+                    if self._should_write(to_active=0):
                         self._mosi.value = bit_value
+                    # clock: wait half a period.
+                    start_time = self._wait(start_time)
+                    # clock: idle->active
+                    self._sclk.value = not self._polarity
+                    if self._should_read(to_active=1):
                         if self._miso.value:
                             # Set bit to 1 at appropriate location.
                             buffer_in[in_byte_position] |= bit_mask
                         else:
-                            # Set bit to 0 at appropriate location.
                             buffer_in[in_byte_position] &= ~bit_mask
-                    # Flip clock off base
-                    self._sclk.value = not self._polarity
-                    start_time = self._wait(start_time)
-                    # Handle read on trailing edge of clock.
-                    if self._phase:  # Mode 1, 3
+                    if self._should_write(to_active=1):
                         self._mosi.value = bit_value
+                    # clock: wait half a period
+                    start_time = self._wait(start_time)
+                    # Clock: active->idle
+                    self._sclk.value = self._polarity
+                    if self._should_read(to_active=0):
                         if self._miso.value:
                             # Set bit to 1 at appropriate location.
                             buffer_in[in_byte_position] |= bit_mask
                         else:
-                            # Set bit to 0 at appropriate location.
                             buffer_in[in_byte_position] &= ~bit_mask
 
-            # Return pins to base positions
-            self._mosi.value = 0
-            self._sclk.value = self._polarity
+            # clock: wait another half period for the last transition.
+            start_time = self._wait(start_time)
 
     # pylint: enable=too-many-branches
 

tests/README.rst

@@ -0,0 +1,53 @@
+..
+  SPDX-FileCopyrightText: KB Sriram
+  SPDX-License-Identifier: MIT
+..
+
+Bitbangio Tests
+===============
+
+These tests run under CPython, and are intended to verify that the
+library passes some sanity checks, using a lightweight simulator as
+the target device.
+
+These tests run automatically from the standard `circuitpython github
+workflow <wf_>`_. To run them manually, first install these packages
+if necessary::
+
+  $ pip3 install pytest
+
+Then ensure you're in the *root* directory of the repository and run
+the following command::
+
+  $ python -m pytest
+
+Notes on the simulator
+======================
+
+`simulator.py` implements a small logic level simulator and a few test
+doubles so the library can run under CPython.
+
+The `Engine` class is used as a singleton in the module to co-ordinate
+the simulation.
+
+A `Net` holds a list of `FakePins` that are connected together. It
+also resolves the overall logic level of the net when a `FakePin` is
+updated. It can optionally hold a history of logic level changes,
+which may be useful for testing some timing expectations, or export
+them as a VCD file for `Pulseview <pv_>`_. Test code can also register
+listeners on a `Net` when the net's level changes, so it can simulate
+device behavior.
+
+A `FakePin` is a test double for the CircuitPython `Pin` class, and
+implements all the functionality so it behaves appropriately in
+CPython.
+
+A simulated device can create a `FakePin` for each of its terminals,
+and connect them to one or more `Net` instances. It can listen for
+level changes on the `Net`, and bitbang the `FakePin` to simulate
+behavior. `simulated_spi_device.py` implements a peripheral device
+that writes a constant value onto an SPI bus.
+
+
+.. _wf: https://github.com/adafruit/workflows-circuitpython-libs/blob/6e1562eaabced4db1bd91173b698b1cc1dfd35ab/build/action.yml#L78-L84
+.. _pv: https://sigrok.org/wiki/PulseView

tests/simulated_spi.py

@@ -0,0 +1,67 @@
+# SPDX-FileCopyrightText: KB Sriram
+# SPDX-License-Identifier: MIT
+"""Implementation of testable SPI devices."""
+
+import dataclasses
+import simulator as sim
+
+
+@dataclasses.dataclass(frozen=True)
+class SpiBus:
+    enable: sim.Net
+    clock: sim.Net
+    copi: sim.Net
+    cipo: sim.Net
+
+
+class Constant:
+    """Device that always writes a constant."""
+
+    def __init__(self, data: bytearray, bus: SpiBus, polarity: int, phase: int) -> None:
+        # convert to binary string array of bits for convenience
+        datalen = 8 * len(data)
+        self._data = f"{int.from_bytes(data, 'big'):0{datalen}b}"
+        self._bit_position = 0
+        self._clock = sim.FakePin("const_clock_pin", bus.clock)
+        self._last_clock_level = bus.clock.level
+        self._cipo = sim.FakePin("const_cipo_pin", bus.cipo)
+        self._enable = sim.FakePin("const_enable_pin", bus.enable)
+        self._cipo.init(sim.Mode.OUT)
+        self._phase = phase
+        self._polarity = sim.Level.HIGH if polarity else sim.Level.LOW
+        self._enabled = False
+        bus.clock.on_level_change(self._on_level_change)
+        bus.enable.on_level_change(self._on_level_change)
+
+    def write_bit(self) -> None:
+        """Writes the next bit to the cipo net."""
+        if self._bit_position >= len(self._data):
+            # Just write a zero
+            self._cipo.value(0)  # pylint: disable=not-callable
+            return
+        self._cipo.value(
+            int(self._data[self._bit_position])  # pylint: disable=not-callable
+        )
+        self._bit_position += 1
+
+    def _on_level_change(self, net: sim.Net) -> None:
+        if net == self._enable.net:
+            # Assumes enable is active high.
+            self._enabled = net.level == sim.Level.HIGH
+            if self._enabled:
+                self._bit_position = 0
+                if self._phase == 0:
+                    # Write on enable or idle->active
+                    self.write_bit()
+            return
+        if not self._enabled:
+            return
+        if net != self._clock.net:
+            return
+        cur_clock_level = net.level
+        if cur_clock_level == self._last_clock_level:
+            return
+        active = 0 if cur_clock_level == self._polarity else 1
+        if self._phase == active:
+            self.write_bit()
+        self._last_clock_level = cur_clock_level