##
# TODO: Look into arbitration, collision detection, clock synchronisation, etc.
-# TODO: Implement support for 10bit slave addresses.
# TODO: Implement support for inverting SDA/SCL levels (0->1 and 1->0).
# TODO: Implement support for detecting various bus errors.
'DATA WRITE': [9, 'Data write', 'DW'],
}
-class SamplerateError(Exception):
- pass
-
class Decoder(srd.Decoder):
api_version = 3
id = 'i2c'
license = 'gplv2+'
inputs = ['logic']
outputs = ['i2c']
+ tags = ['Embedded/industrial']
channels = (
{'id': 'scl', 'name': 'SCL', 'desc': 'Serial clock line'},
{'id': 'sda', 'name': 'SDA', 'desc': 'Serial data line'},
('address-write', 'Address write'),
('data-read', 'Data read'),
('data-write', 'Data write'),
- ('warnings', 'Human-readable warnings'),
+ ('warning', 'Warning'),
)
annotation_rows = (
('bits', 'Bits', (5,)),
- ('addr-data', 'Address/Data', (0, 1, 2, 3, 4, 6, 7, 8, 9)),
+ ('addr-data', 'Address/data', (0, 1, 2, 3, 4, 6, 7, 8, 9)),
('warnings', 'Warnings', (10,)),
)
binary = (
)
def __init__(self):
+ self.reset()
+
+ def reset(self):
self.samplerate = None
self.ss = self.es = self.ss_byte = -1
self.bitcount = 0
self.databyte = 0
- self.wr = -1
- self.is_repeat_start = 0
+ self.is_write = None
+ self.rem_addr_bytes = None
+ self.is_repeat_start = False
self.state = 'FIND START'
self.pdu_start = None
self.pdu_bits = 0
- self.bits = []
+ self.data_bits = []
def metadata(self, key, value):
if key == srd.SRD_CONF_SAMPLERATE:
self.out_bitrate = self.register(srd.OUTPUT_META,
meta=(int, 'Bitrate', 'Bitrate from Start bit to Stop bit'))
- # Assume that the initial SCL/SDA pin state is high (logic 1).
- # This is a good default, since both pins have pullups as per spec.
- self.initial_pins = [1, 1]
-
def putx(self, data):
self.put(self.ss, self.es, self.out_ann, data)
self.ss, self.es = self.samplenum, self.samplenum
self.pdu_start = self.samplenum
self.pdu_bits = 0
- cmd = 'START REPEAT' if (self.is_repeat_start == 1) else 'START'
+ cmd = 'START REPEAT' if self.is_repeat_start else 'START'
self.putp([cmd, None])
self.putx([proto[cmd][0], proto[cmd][1:]])
self.state = 'FIND ADDRESS'
self.bitcount = self.databyte = 0
- self.is_repeat_start = 1
- self.wr = -1
- self.bits = []
+ self.is_repeat_start = True
+ self.is_write = None
+ self.rem_addr_bytes = None
+ self.data_bits = []
# Gather 8 bits of data plus the ACK/NACK bit.
def handle_address_or_data(self, pins):
# Store individual bits and their start/end samplenumbers.
# In the list, index 0 represents the LSB (I²C transmits MSB-first).
- self.bits.insert(0, [sda, self.samplenum, self.samplenum])
+ self.data_bits.insert(0, [sda, self.samplenum, self.samplenum])
if self.bitcount > 0:
- self.bits[1][2] = self.samplenum
+ self.data_bits[1][2] = self.samplenum
if self.bitcount == 7:
- self.bitwidth = self.bits[1][2] - self.bits[2][2]
- self.bits[0][2] += self.bitwidth
+ self.bitwidth = self.data_bits[1][2] - self.data_bits[2][2]
+ self.data_bits[0][2] += self.bitwidth
# Return if we haven't collected all 8 + 1 bits, yet.
if self.bitcount < 7:
d = self.databyte
if self.state == 'FIND ADDRESS':
- # The READ/WRITE bit is only in address bytes, not data bytes.
- self.wr = 0 if (self.databyte & 1) else 1
- if self.options['address_format'] == 'shifted':
- d = d >> 1
+ # The READ/WRITE bit is only in the first address byte, not
+ # in data bytes. Address bit pattern 0b1111_0xxx means that
+ # this is a 10bit slave address, another byte follows. Get
+ # the R/W direction and the address bytes count from the
+ # first byte in the I2C transfer.
+ addr_byte = d
+ if self.rem_addr_bytes is None:
+ if (addr_byte & 0xf8) == 0xf0:
+ self.rem_addr_bytes = 2
+ self.slave_addr_7 = None
+ self.slave_addr_10 = addr_byte & 0x06
+ self.slave_addr_10 <<= 7
+ else:
+ self.rem_addr_bytes = 1
+ self.slave_addr_7 = addr_byte >> 1
+ self.slave_addr_10 = None
+ is_seven = self.slave_addr_7 is not None
+ if self.is_write is None:
+ read_bit = bool(addr_byte & 1)
+ shift_seven = self.options['address_format'] == 'shifted'
+ if is_seven and shift_seven:
+ d = d >> 1
+ self.is_write = False if read_bit else True
+ else:
+ self.slave_addr_10 |= addr_byte
bin_class = -1
- if self.state == 'FIND ADDRESS' and self.wr == 1:
+ if self.state == 'FIND ADDRESS' and self.is_write:
cmd = 'ADDRESS WRITE'
bin_class = 1
- elif self.state == 'FIND ADDRESS' and self.wr == 0:
+ elif self.state == 'FIND ADDRESS' and not self.is_write:
cmd = 'ADDRESS READ'
bin_class = 0
- elif self.state == 'FIND DATA' and self.wr == 1:
+ elif self.state == 'FIND DATA' and self.is_write:
cmd = 'DATA WRITE'
bin_class = 3
- elif self.state == 'FIND DATA' and self.wr == 0:
+ elif self.state == 'FIND DATA' and not self.is_write:
cmd = 'DATA READ'
bin_class = 2
self.ss, self.es = self.ss_byte, self.samplenum + self.bitwidth
- self.putp(['BITS', self.bits])
+ self.putp(['BITS', self.data_bits])
self.putp([cmd, d])
self.putb([bin_class, bytes([d])])
- for bit in self.bits:
+ for bit in self.data_bits:
self.put(bit[1], bit[2], self.out_ann, [5, ['%d' % bit[0]]])
- if cmd.startswith('ADDRESS'):
+ if cmd.startswith('ADDRESS') and is_seven:
self.ss, self.es = self.samplenum, self.samplenum + self.bitwidth
- w = ['Write', 'Wr', 'W'] if self.wr else ['Read', 'Rd', 'R']
+ w = ['Write', 'Wr', 'W'] if self.is_write else ['Read', 'Rd', 'R']
self.putx([proto[cmd][0], w])
self.ss, self.es = self.ss_byte, self.samplenum
# Done with this packet.
self.bitcount = self.databyte = 0
- self.bits = []
+ self.data_bits = []
self.state = 'FIND ACK'
def get_ack(self, pins):
cmd = 'NACK' if (sda == 1) else 'ACK'
self.putp([cmd, None])
self.putx([proto[cmd][0], proto[cmd][1:]])
- # There could be multiple data bytes in a row, so either find
- # another data byte or a STOP condition next.
- self.state = 'FIND DATA'
+ # Slave addresses can span one or two bytes, before data bytes
+ # follow. There can be an arbitrary number of data bytes. Stick
+ # with getting more address bytes if applicable, or enter or
+ # remain in the data phase of the transfer otherwise.
+ if self.rem_addr_bytes:
+ self.rem_addr_bytes -= 1
+ if self.rem_addr_bytes:
+ self.state = 'FIND ADDRESS'
+ else:
+ self.state = 'FIND DATA'
def handle_stop(self, pins):
# Meta bitrate
- elapsed = 1 / float(self.samplerate) * (self.samplenum - self.pdu_start + 1)
- bitrate = int(1 / elapsed * self.pdu_bits)
- self.put(self.ss_byte, self.samplenum, self.out_bitrate, bitrate)
+ if self.samplerate:
+ elapsed = 1 / float(self.samplerate) * (self.samplenum - self.pdu_start + 1)
+ bitrate = int(1 / elapsed * self.pdu_bits)
+ self.put(self.ss_byte, self.samplenum, self.out_bitrate, bitrate)
cmd = 'STOP'
self.ss, self.es = self.samplenum, self.samplenum
self.putp([cmd, None])
self.putx([proto[cmd][0], proto[cmd][1:]])
self.state = 'FIND START'
- self.is_repeat_start = 0
- self.wr = -1
- self.bits = []
+ self.is_repeat_start = False
+ self.is_write = None
+ self.data_bits = []
def decode(self):
- if not self.samplerate:
- raise SamplerateError('Cannot decode without samplerate.')
-
- self.wait({})
-
while True:
# State machine.
if self.state == 'FIND START':
# a) Data sampling of receiver: SCL = rising, and/or
# b) START condition (S): SCL = high, SDA = falling, and/or
# c) STOP condition (P): SCL = high, SDA = rising
- conds = [{0: 'r'}, {0: 'h', 1: 'f'}, {0: 'h', 1: 'r'}]
- pins = self.wait(conds[:]) # TODO
+ pins = self.wait([{0: 'r'}, {0: 'h', 1: 'f'}, {0: 'h', 1: 'r'}])
# Check which of the condition(s) matched and handle them.
if self.matched[0]:
projects / libsigrokdecode.git / blobdiff