Skip to content

Commit c122e06

Browse files
kbsriramkbsriram
committed
Add I2C clock stretching.
- Also fixed a corner case with I2C checking whether the device acknowledged sent data. - Added tests to cover most of the major operations. Fixes #16
1 parent fb37622 commit c122e06

File tree

3 files changed

+406
-4
lines changed

3 files changed

+406
-4
lines changed

‎adafruit_bitbangio.py

+9-4
Original file line numberDiff line numberDiff line change
@@ -190,13 +190,17 @@ def _sda_low(self) -> None:
190190
self._sda.switch_to_output(value=False)
191191

192192
def _scl_release(self) -> None:
193-
"""Release and let the pullups lift"""
194-
# Use self._timeout to add clock stretching
193+
"""Release and wait for the pullups to lift."""
195194
self._scl.switch_to_input()
195+
# Wait at most self._timeout seconds for any clock stretching.
196+
end = monotonic() + self._timeout
197+
while not self._scl.value and end > monotonic():
198+
pass
199+
if not self._scl.value:
200+
raise RuntimeError("Bus timed out.")
196201

197202
def _sda_release(self) -> None:
198203
"""Release and let the pullups lift"""
199-
# Use self._timeout to add clock stretching
200204
self._sda.switch_to_input()
201205

202206
def _start(self) -> None:
@@ -288,7 +292,8 @@ def _write(self, address: int, buffer: ReadableBuffer, transmit_stop: bool) -> N
288292
# raise RuntimeError("Device not responding at 0x{:02X}".format(address))
289293
raise RuntimeError(f"Device not responding at 0x{address:02X}")
290294
for byte in buffer:
291-
self._write_byte(byte)
295+
if not self._write_byte(byte):
296+
raise RuntimeError(f"Device not responding at 0x{address:02X}")
292297
if transmit_stop:
293298
self._stop()
294299

‎tests/simulated_i2c.py

+209
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,209 @@
1+
# SPDX-FileCopyrightText: KB Sriram
2+
# SPDX-License-Identifier: MIT
3+
"""Implementation of testable I2C devices."""
4+
5+
from typing import Any, Callable, Optional, Union
6+
import dataclasses
7+
import enum
8+
import signal
9+
import types
10+
from typing_extensions import TypeAlias
11+
import simulator as sim
12+
13+
_SignalHandler: TypeAlias = Union[
14+
Callable[[int, Optional[types.FrameType]], Any], int, None
15+
]
16+
17+
18+
@enum.unique
19+
class State(enum.Enum):
20+
IDLE = "idle"
21+
ADDRESS = "address"
22+
ACK = "ack"
23+
ACK_DONE = "ack_done"
24+
WAIT_ACK = "wait_ack"
25+
READ = "read"
26+
WRITE = "write"
27+
28+
29+
@dataclasses.dataclass(frozen=True)
30+
class I2CBus:
31+
scl: sim.Net
32+
sda: sim.Net
33+
34+
35+
def _switch_to_output(pin: sim.FakePin, value: bool) -> None:
36+
pin.mode = sim.Mode.OUT
37+
pin.value(1 if value else 0)
38+
39+
40+
def _switch_to_input(pin: sim.FakePin) -> None:
41+
pin.init(mode=sim.Mode.IN)
42+
pin.level = sim.Level.HIGH
43+
44+
45+
class Constant:
46+
"""I2C device that sinks all data and can send a constant."""
47+
48+
# pylint:disable=too-many-instance-attributes
49+
# pylint:disable=too-many-arguments
50+
def __init__(
51+
self,
52+
name: str,
53+
address: int,
54+
bus: I2CBus,
55+
ack_data: bool = True,
56+
clock_stretch_sec: int = 0,
57+
data_to_send: int = 0,
58+
) -> None:
59+
self._address = address
60+
self._scl = sim.FakePin(f"{name}_scl_pin", bus.scl)
61+
self._sda = sim.FakePin(f"{name}_sda_pin", bus.sda)
62+
self._last_scl_level = bus.scl.level
63+
self._ack_data = ack_data
64+
self._clock_stretch_sec = clock_stretch_sec
65+
self._prev_signal: _SignalHandler = None
66+
self._state = State.IDLE
67+
self._bit_count = 0
68+
self._received = 0
69+
self._all_received = bytearray()
70+
self._send_data = data_to_send
71+
self._sent_bit_count = 0
72+
self._in_write = 0
73+
74+
bus.scl.on_level_change(self._on_level_change)
75+
bus.sda.on_level_change(self._on_level_change)
76+
77+
def _move_state(self, nstate: State) -> None:
78+
self._state = nstate
79+
80+
def _on_start(self) -> None:
81+
# This resets our state machine unconditionally and
82+
# starts waiting for an address.
83+
self._bit_count = 0
84+
self._received = 0
85+
self._move_state(State.ADDRESS)
86+
87+
def _on_stop(self) -> None:
88+
# Reset and start idling.
89+
self._reset()
90+
91+
def _reset(self) -> None:
92+
self._bit_count = 0
93+
self._received = 0
94+
self._move_state(State.IDLE)
95+
96+
def _clock_release(
97+
self, ignored_signum: int, ignored_frame: Optional[types.FrameType] = None
98+
) -> None:
99+
# First release the scl line
100+
_switch_to_input(self._scl)
101+
# Remove alarms
102+
signal.alarm(0)
103+
# Restore any existing signal.
104+
if self._prev_signal:
105+
signal.signal(signal.SIGALRM, self._prev_signal)
106+
self._prev_signal = None
107+
108+
def _maybe_clock_stretch(self) -> None:
109+
if not self._clock_stretch_sec:
110+
return
111+
if self._state == State.IDLE:
112+
return
113+
# pull the clock line low
114+
_switch_to_output(self._scl, value=False)
115+
# Set an alarm to release the line after some time.
116+
self._prev_signal = signal.signal(signal.SIGALRM, self._clock_release)
117+
signal.alarm(self._clock_stretch_sec)
118+
119+
def _on_byte_read(self) -> None:
120+
self._all_received.append(self._received)
121+
122+
def _on_clock_fall(self) -> None:
123+
self._maybe_clock_stretch()
124+
125+
# Return early unless we need to send data.
126+
if self._state not in (State.ACK, State.ACK_DONE, State.WRITE):
127+
return
128+
129+
if self._state == State.ACK:
130+
# pull down the data line to start the ack. We want to hold
131+
# it down until the next clock falling edge.
132+
if self._ack_data or not self._all_received:
133+
_switch_to_output(self._sda, value=False)
134+
self._move_state(State.ACK_DONE)
135+
return
136+
if self._state == State.ACK_DONE:
137+
# The data line has been held between one pair of falling edges - we can
138+
# let go now if we need to start reading.
139+
if self._in_write:
140+
# Note: this will also write out the first bit later in this method.
141+
self._move_state(State.WRITE)
142+
else:
143+
_switch_to_input(self._sda)
144+
self._move_state(State.READ)
145+
146+
if self._state == State.WRITE:
147+
if self._sent_bit_count == 8:
148+
_switch_to_input(self._sda)
149+
self._sent_bit_count = 0
150+
self._move_state(State.WAIT_ACK)
151+
else:
152+
bit_value = (self._send_data >> (7 - self._sent_bit_count)) & 0x1
153+
_switch_to_output(self._sda, value=bit_value == 1)
154+
self._sent_bit_count += 1
155+
156+
def _on_clock_rise(self) -> None:
157+
if self._state not in (State.ADDRESS, State.READ, State.WAIT_ACK):
158+
return
159+
bit_value = 1 if self._sda.net.level == sim.Level.HIGH else 0
160+
if self._state == State.WAIT_ACK:
161+
if bit_value:
162+
# NACK, just reset.
163+
self._move_state(State.IDLE)
164+
else:
165+
# ACK, continue writing.
166+
self._move_state(State.ACK_DONE)
167+
return
168+
self._received = (self._received << 1) | bit_value
169+
self._bit_count += 1
170+
if self._bit_count < 8:
171+
return
172+
173+
# We've read 8 bits of either address or data sent to us.
174+
if self._state == State.ADDRESS and self._address != (self._received >> 1):
175+
# This message isn't for us, reset and start idling.
176+
self._reset()
177+
return
178+
# This message is for us, ack it.
179+
if self._state == State.ADDRESS:
180+
self._in_write = self._received & 0x1
181+
elif self._state == State.READ:
182+
self._on_byte_read()
183+
self._bit_count = 0
184+
self._received = 0
185+
self._move_state(State.ACK)
186+
187+
def _on_level_change(self, net: sim.Net) -> None:
188+
# Handle start/stop events directly.
189+
if net == self._sda.net and self._scl.net.level == sim.Level.HIGH:
190+
if net.level == sim.Level.LOW:
191+
# sda hi->low with scl high
192+
self._on_start()
193+
else:
194+
# sda low->hi with scl high
195+
self._on_stop()
196+
return
197+
198+
# Everything else can be handled as state changes that occur
199+
# either on the clock rising or falling edge.
200+
if net == self._scl.net:
201+
if net.level == sim.Level.HIGH:
202+
# scl low->high
203+
self._on_clock_rise()
204+
else:
205+
# scl high->low
206+
self._on_clock_fall()
207+
208+
def all_received_data(self) -> bytearray:
209+
return self._all_received

0 commit comments

Comments
 (0)