X-Git-Url: https://sigrok.org/gitweb/?a=blobdiff_plain;f=decoders%2Fuart.py;h=b8909ea461f02c4c4ca1cfb2c84032a72c2f82ee;hb=876f83fd3396ce581e700741e58414c2f4f6277e;hp=ee39697183618a5448be80dd92adbe6f02df217d;hpb=f44d2db2a027857bdfbfc18e40b9900cad1b0e1d;p=libsigrokdecode.git diff --git a/decoders/uart.py b/decoders/uart.py index ee39697..b8909ea 100644 --- a/decoders/uart.py +++ b/decoders/uart.py @@ -22,7 +22,99 @@ # UART protocol decoder # -import sigrok +# +# Universal Asynchronous Receiver Transmitter (UART) is a simple serial +# communication protocol which allows two devices to talk to each other. +# +# It uses just two data signals and a ground (GND) signal: +# - RX/RXD: Receive signal +# - TX/TXD: Transmit signal +# +# The protocol is asynchronous, i.e., there is no dedicated clock signal. +# Rather, both devices have to agree on a baudrate (number of bits to be +# transmitted per second) beforehand. Baudrates can be arbitrary in theory, +# but usually the choice is limited by the hardware UARTs that are used. +# Common values are 9600 or 115200. +# +# The protocol allows full-duplex transmission, i.e. both devices can send +# data at the same time. However, unlike SPI (which is always full-duplex, +# i.e., each send operation is automatically also a receive operation), UART +# allows one-way communication, too. In such a case only one signal (and GND) +# is required. +# +# The data is sent over the TX line in so-called 'frames', which consist of: +# - Exactly one start bit (always 0/low). +# - Between 5 and 9 data bits. +# - An (optional) parity bit. +# - One or more stop bit(s). +# +# The idle state of the RX/TX line is 1/high. As the start bit is 0/low, the +# receiver can continually monitor its RX line for a falling edge, in order +# to detect the start bit. +# +# Once detected, it can (due to the agreed-upon baudrate and thus the known +# width/duration of one UART bit) sample the state of the RX line "in the +# middle" of each (start/data/parity/stop) bit it wants to analyze. +# +# It is configurable whether there is a parity bit in a frame, and if yes, +# which type of parity is used: +# - None: No parity bit is included. +# - Odd: The number of 1 bits in the data (and parity bit itself) is odd. +# - Even: The number of 1 bits in the data (and parity bit itself) is even. +# - Mark/one: The parity bit is always 1/high (also called 'mark state'). +# - Space/zero: The parity bit is always 0/low (also called 'space state'). +# +# It is also configurable how many stop bits are to be used: +# - 1 stop bit (most common case) +# - 2 stop bits +# - 1.5 stop bits (i.e., one stop bit, but 1.5 times the UART bit width) +# - 0.5 stop bits (i.e., one stop bit, but 0.5 times the UART bit width) +# +# The bit order of the 5-9 data bits is LSB-first. +# +# Possible special cases: +# - One or both data lines could be inverted, which also means that the idle +# state of the signal line(s) is low instead of high. +# - Only the data bits on one or both data lines (and the parity bit) could +# be inverted (but the start/stop bits remain non-inverted). +# - The bit order could be MSB-first instead of LSB-first. +# - The baudrate could change in the middle of the communication. This only +# happens in very special cases, and can only work if both devices know +# to which baudrate they are to switch, and when. +# - Theoretically, the baudrate on RX and the one on TX could also be +# different, but that's a very obscure case and probably doesn't happen +# very often in practice. +# +# Error conditions: +# - If there is a parity bit, but it doesn't match the expected parity, +# this is called a 'parity error'. +# - If there are no stop bit(s), that's called a 'frame error'. +# +# More information: +# TODO: URLs +# + +# +# Protocol output format: +# +# UART packet: +# [, , ] +# +# This is the list of s and their respective : +# - T_START: The data is the (integer) value of the start bit (0 or 1). +# - T_DATA: The data is the (integer) value of the UART data. Valid values +# range from 0 to 512 (as the data can be up to 9 bits in size). +# - T_PARITY: The data is the (integer) value of the parity bit (0 or 1). +# - T_STOP: The data is the (integer) value of the stop bit (0 or 1). +# - T_INVALID_START: The data is the (integer) value of the start bit (0 or 1). +# - T_INVALID_STOP: The data is the (integer) value of the stop bit (0 or 1). +# - T_PARITY_ERROR: The data is a tuple with two entries. The first one is +# the expected parity value, the second is the actual parity value. +# +# The field is 0 for RX packets, 1 for TX packets. +# + +import sigrokdecode as srd # States WAIT_FOR_START_BIT = 0 @@ -31,6 +123,10 @@ GET_DATA_BITS = 2 GET_PARITY_BIT = 3 GET_STOP_BITS = 4 +# Used for differentiating between the two data directions. +RX = 0 +TX = 1 + # Parity options PARITY_NONE = 0 PARITY_ODD = 1 @@ -48,23 +144,21 @@ STOP_BITS_2 = 3 LSB_FIRST = 0 MSB_FIRST = 1 -# Output data formats -DATA_FORMAT_ASCII = 0 -DATA_FORMAT_HEX = 1 - -# TODO: Remove me later. -quick_hack = 1 - -class Sample(): - def __init__(self, data): - self.data = data - def probe(self, probe): - s = ord(self.data[probe / 8]) & (1 << (probe % 8)) - return True if s else False - -def sampleiter(data, unitsize): - for i in range(0, len(data), unitsize): - yield(Sample(data[i:i+unitsize])) +# Annotation feed formats +ANN_ASCII = 0 +ANN_DEC = 1 +ANN_HEX = 2 +ANN_OCT = 3 +ANN_BITS = 4 + +# Protocol output packet types +T_START = 0 +T_DATA = 1 +T_PARITY = 2 +T_STOP = 3 +T_INVALID_START = 4 +T_INVALID_STOP = 5 +T_PARITY_ERROR = 6 # Given a parity type to check (odd, even, zero, one), the value of the # parity bit, the value of the data, and the length of the data (5-9 bits, @@ -89,10 +183,10 @@ def parity_ok(parity_type, parity_bit, data, num_data_bits): else: raise Exception('Invalid parity type: %d' % parity_type) -class Decoder(sigrok.Decoder): +class Decoder(srd.Decoder): id = 'uart' name = 'UART' - longname = 'Universal Asynchronous Receiver/Transmitter (UART)' + longname = 'Universal Asynchronous Receiver/Transmitter' desc = 'Universal Asynchronous Receiver/Transmitter (UART)' longdesc = 'TODO.' author = 'Uwe Hermann' @@ -100,70 +194,60 @@ class Decoder(sigrok.Decoder): license = 'gplv2+' inputs = ['logic'] outputs = ['uart'] - probes = { + probes = [ # Allow specifying only one of the signals, e.g. if only one data # direction exists (or is relevant). - ## 'rx': {'ch': 0, 'name': 'RX', 'desc': 'UART receive line'}, - ## 'tx': {'ch': 1, 'name': 'TX', 'desc': 'UART transmit line'}, - 'rx': 0, - 'tx': 1, - } + {'id': 'rx', 'name': 'RX', 'desc': 'UART receive line'}, + {'id': 'tx', 'name': 'TX', 'desc': 'UART transmit line'}, + ] options = { - 'baudrate': ['UART baud rate', 115200], + 'baudrate': ['Baud rate', 115200], 'num_data_bits': ['Data bits', 8], # Valid: 5-9. 'parity': ['Parity', PARITY_NONE], 'parity_check': ['Check parity', True], 'num_stop_bits': ['Stop bit(s)', STOP_BITS_1], 'bit_order': ['Bit order', LSB_FIRST], - 'data_format': ['Output data format', DATA_FORMAT_ASCII], # TODO: Options to invert the signal(s). - # ... } + annotations = [ + ['ASCII', 'Data bytes as ASCII characters'], + ['Decimal', 'Databytes as decimal, integer values'], + ['Hex', 'Data bytes in hex format'], + ['Octal', 'Data bytes as octal numbers'], + ['Bits', 'Data bytes in bit notation (sequence of 0/1 digits)'], + ] - def __init__(self, **kwargs): - self.probes = Decoder.probes.copy() - - # Set defaults, can be overridden in 'start'. - self.baudrate = 115200 - self.num_data_bits = 8 - self.parity = PARITY_NONE - self.check_parity = True - self.num_stop_bits = 1 - self.bit_order = LSB_FIRST - self.data_format = DATA_FORMAT_ASCII + def putx(self, rxtx, data): + self.put(self.startsample[rxtx], self.samplenum - 1, self.out_ann, data) + def __init__(self, **kwargs): self.samplenum = 0 - self.frame_start = -1 - self.startbit = -1 - self.cur_data_bit = 0 - self.databyte = 0 - self.stopbit1 = -1 - self.startsample = -1 + self.frame_start = [-1, -1] + self.startbit = [-1, -1] + self.cur_data_bit = [0, 0] + self.databyte = [0, 0] + self.stopbit1 = [-1, -1] + self.startsample = [-1, -1] # Initial state. - self.staterx = WAIT_FOR_START_BIT + self.state = [WAIT_FOR_START_BIT, WAIT_FOR_START_BIT] - # Get the channel/probe number of the RX/TX signals. - ## self.rx_bit = self.probes['rx']['ch'] - ## self.tx_bit = self.probes['tx']['ch'] - self.rx_bit = self.probes['rx'] - self.tx_bit = self.probes['tx'] + self.oldbit = [None, None] - self.oldrx = None - self.oldtx = None + # Set protocol decoder option defaults. + self.baudrate = Decoder.options['baudrate'][1] + self.num_data_bits = Decoder.options['num_data_bits'][1] + self.parity = Decoder.options['parity'][1] + self.check_parity = Decoder.options['parity_check'][1] + self.num_stop_bits = Decoder.options['num_stop_bits'][1] + self.bit_order = Decoder.options['bit_order'][1] def start(self, metadata): - self.unitsize = metadata['unitsize'] self.samplerate = metadata['samplerate'] + self.out_proto = self.add(srd.OUTPUT_PROTO, 'uart') + self.out_ann = self.add(srd.OUTPUT_ANN, 'uart') - # TODO - ### self.baudrate = metadata['baudrate'] - ### self.num_data_bits = metadata['num_data_bits'] - ### self.parity = metadata['parity'] - ### self.parity_check = metadata['parity_check'] - ### self.num_stop_bits = metadata['num_stop_bits'] - ### self.bit_order = metadata['bit_order'] - ### self.data_format = metadata['data_format'] + # TODO: Override PD options, if user wants that. # The width of one UART bit in number of samples. self.bit_width = float(self.samplerate) / float(self.baudrate) @@ -172,196 +256,179 @@ class Decoder(sigrok.Decoder): pass # Return true if we reached the middle of the desired bit, false otherwise. - def reached_bit(self, bitnum): + def reached_bit(self, rxtx, bitnum): # bitpos is the samplenumber which is in the middle of the # specified UART bit (0 = start bit, 1..x = data, x+1 = parity bit # (if used) or the first stop bit, and so on). - bitpos = self.frame_start + (self.bit_width / 2.0) + bitpos = self.frame_start[rxtx] + (self.bit_width / 2.0) bitpos += bitnum * self.bit_width if self.samplenum >= bitpos: return True return False - def reached_bit_last(self, bitnum): - bitpos = self.frame_start + ((bitnum + 1) * self.bit_width) + def reached_bit_last(self, rxtx, bitnum): + bitpos = self.frame_start[rxtx] + ((bitnum + 1) * self.bit_width) if self.samplenum >= bitpos: return True return False - def wait_for_start_bit(self, old_signal, signal): + def wait_for_start_bit(self, rxtx, old_signal, signal): # The start bit is always 0 (low). As the idle UART (and the stop bit) # level is 1 (high), the beginning of a start bit is a falling edge. if not (old_signal == 1 and signal == 0): return # Save the sample number where the start bit begins. - self.frame_start = self.samplenum + self.frame_start[rxtx] = self.samplenum - self.staterx = GET_START_BIT + self.state[rxtx] = GET_START_BIT - def get_start_bit(self, signal): + def get_start_bit(self, rxtx, signal): # Skip samples until we're in the middle of the start bit. - if not self.reached_bit(0): - return [] - - self.startbit = signal + if not self.reached_bit(rxtx, 0): + return - if self.startbit != 0: - # TODO: Startbit must be 0. If not, we report an error. - pass + self.startbit[rxtx] = signal - self.cur_data_bit = 0 - self.databyte = 0 - self.startsample = -1 + # The startbit must be 0. If not, we report an error. + if self.startbit[rxtx] != 0: + self.put(self.frame_start[rxtx], self.samplenum, self.out_proto, + [T_INVALID_START, rxtx, self.startbit[rxtx]]) + # TODO: Abort? Ignore rest of the frame? - self.staterx = GET_DATA_BITS + self.cur_data_bit[rxtx] = 0 + self.databyte[rxtx] = 0 + self.startsample[rxtx] = -1 - if quick_hack: # TODO - return [] + self.state[rxtx] = GET_DATA_BITS - o = [{'type': 'S', 'range': (self.frame_start, self.samplenum), - 'data': None, 'ann': 'Start bit'}] - return o + self.put(self.frame_start[rxtx], self.samplenum, self.out_proto, + [T_START, rxtx, self.startbit[rxtx]]) + self.put(self.frame_start[rxtx], self.samplenum, self.out_ann, + [ANN_ASCII, ['Start bit', 'Start', 'S']]) - def get_data_bits(self, signal): + def get_data_bits(self, rxtx, signal): # Skip samples until we're in the middle of the desired data bit. - if not self.reached_bit(self.cur_data_bit + 1): - return [] + if not self.reached_bit(rxtx, self.cur_data_bit[rxtx] + 1): + return # Save the sample number where the data byte starts. - if self.startsample == -1: - self.startsample = self.samplenum + if self.startsample[rxtx] == -1: + self.startsample[rxtx] = self.samplenum # Get the next data bit in LSB-first or MSB-first fashion. if self.bit_order == LSB_FIRST: - self.databyte >>= 1 - self.databyte |= (signal << (self.num_data_bits - 1)) + self.databyte[rxtx] >>= 1 + self.databyte[rxtx] |= (signal << (self.num_data_bits - 1)) elif self.bit_order == MSB_FIRST: - self.databyte <<= 1 - self.databyte |= (signal << 0) + self.databyte[rxtx] <<= 1 + self.databyte[rxtx] |= (signal << 0) else: raise Exception('Invalid bit order value: %d', self.bit_order) # Return here, unless we already received all data bits. - if self.cur_data_bit < self.num_data_bits - 1: # TODO? Off-by-one? - self.cur_data_bit += 1 - return [] - - # Convert the data byte into the configured format. - if self.data_format == DATA_FORMAT_ASCII: - d = chr(self.databyte) - elif self.data_format == DATA_FORMAT_HEX: - d = '0x%02x' % self.databyte - else: - raise Exception('Invalid data format value: %d', self.data_format) - - self.staterx = GET_PARITY_BIT + if self.cur_data_bit[rxtx] < self.num_data_bits - 1: # TODO? Off-by-one? + self.cur_data_bit[rxtx] += 1 + return - if quick_hack: # TODO - return [d] + self.state[rxtx] = GET_PARITY_BIT - o = [{'type': 'D', 'range': (self.startsample, self.samplenum - 1), - 'data': d, 'ann': None}] + self.put(self.startsample[rxtx], self.samplenum - 1, self.out_proto, + [T_DATA, rxtx, self.databyte[rxtx]]) - return o + s = 'RX: ' if (rxtx == RX) else 'TX: ' + self.putx(rxtx, [ANN_ASCII, [s + chr(self.databyte[rxtx])]]) + self.putx(rxtx, [ANN_DEC, [s + str(self.databyte[rxtx])]]) + self.putx(rxtx, [ANN_HEX, [s + hex(self.databyte[rxtx]), + s + hex(self.databyte[rxtx])[2:]]]) + self.putx(rxtx, [ANN_OCT, [s + oct(self.databyte[rxtx]), + s + oct(self.databyte[rxtx])[2:]]]) + self.putx(rxtx, [ANN_BITS, [s + bin(self.databyte[rxtx]), + s + bin(self.databyte[rxtx])[2:]]]) - def get_parity_bit(self, signal): + def get_parity_bit(self, rxtx, signal): # If no parity is used/configured, skip to the next state immediately. if self.parity == PARITY_NONE: - self.staterx = GET_STOP_BITS - return [] + self.state[rxtx] = GET_STOP_BITS + return # Skip samples until we're in the middle of the parity bit. - if not self.reached_bit(self.num_data_bits + 1): - return [] + if not self.reached_bit(rxtx, self.num_data_bits + 1): + return - self.paritybit = signal + self.paritybit[rxtx] = signal - self.staterx = GET_STOP_BITS + self.state[rxtx] = GET_STOP_BITS - if parity_ok(self.parity, self.paritybit, self.databyte, - self.num_data_bits): - if quick_hack: # TODO - # return ['P'] - return [] + if parity_ok(self.parity[rxtx], self.paritybit[rxtx], + self.databyte[rxtx], self.num_data_bits): # TODO: Fix range. - o = [{'type': 'P', 'range': (self.samplenum, self.samplenum), - 'data': self.paritybit, 'ann': 'Parity bit'}] + self.put(self.samplenum, self.samplenum, self.out_proto, + [T_PARITY_BIT, rxtx, self.paritybit[rxtx]]) + self.put(self.samplenum, self.samplenum, self.out_ann, + [ANN_ASCII, ['Parity bit', 'Parity', 'P']]) else: - if quick_hack: # TODO - return ['PE'] - o = [{'type': 'PE', 'range': (self.samplenum, self.samplenum), - 'data': self.paritybit, 'ann': 'Parity error'}] - - return o + # TODO: Fix range. + # TODO: Return expected/actual parity values. + self.put(self.samplenum, self.samplenum, self.out_proto, + [T_PARITY_ERROR, rxtx, (0, 1)]) # FIXME: Dummy tuple... + self.put(self.samplenum, self.samplenum, self.out_ann, + [ANN_ASCII, ['Parity error', 'Parity err', 'PE']]) # TODO: Currently only supports 1 stop bit. - def get_stop_bits(self, signal): + def get_stop_bits(self, rxtx, signal): # Skip samples until we're in the middle of the stop bit(s). - skip_parity = 0 - if self.parity != PARITY_NONE: - skip_parity = 1 - if not self.reached_bit(self.num_data_bits + 1 + skip_parity): - return [] - - self.stopbit1 = signal + skip_parity = 0 if self.parity == PARITY_NONE else 1 + if not self.reached_bit(rxtx, self.num_data_bits + 1 + skip_parity): + return - if self.stopbit1 != 1: - # TODO: Stop bits must be 1. If not, we report an error. - pass + self.stopbit1[rxtx] = signal - self.staterx = WAIT_FOR_START_BIT + # Stop bits must be 1. If not, we report an error. + if self.stopbit1[rxtx] != 1: + self.put(self.frame_start[rxtx], self.samplenum, self.out_proto, + [T_INVALID_STOP, rxtx, self.stopbit1[rxtx]]) + # TODO: Abort? Ignore the frame? Other? - if quick_hack: # TODO - return [] + self.state[rxtx] = WAIT_FOR_START_BIT # TODO: Fix range. - o = [{'type': 'P', 'range': (self.samplenum, self.samplenum), - 'data': None, 'ann': 'Stop bit'}] - return o + self.put(self.samplenum, self.samplenum, self.out_proto, + [T_STOP, rxtx, self.stopbit1[rxtx]]) + self.put(self.samplenum, self.samplenum, self.out_ann, + [ANN_ASCII, ['Stop bit', 'Stop', 'P']]) - def decode(self, data): - """UART protocol decoder""" - - out = [] - - for sample in sampleiter(data["data"], self.unitsize): - - # TODO: Eliminate the need for ord(). - s = ord(sample.data) + def decode(self, ss, es, data): # TODO + for (samplenum, (rx, tx)) in data: # TODO: Start counting at 0 or 1? Increase before or after? self.samplenum += 1 # First sample: Save RX/TX value. - if self.oldrx == None: - # Get RX/TX bit values (0/1 for low/high) of the first sample. - self.oldrx = (s & (1 << self.rx_bit)) >> self.rx_bit - # self.oldtx = (s & (1 << self.tx_bit)) >> self.tx_bit + if self.oldbit[RX] == None: + self.oldbit[RX] = rx + continue + if self.oldbit[TX] == None: + self.oldbit[TX] = tx continue - - # Get RX/TX bit values (0/1 for low/high). - rx = (s & (1 << self.rx_bit)) >> self.rx_bit - # tx = (s & (1 << self.tx_bit)) >> self.tx_bit # State machine. - if self.staterx == WAIT_FOR_START_BIT: - self.wait_for_start_bit(self.oldrx, rx) - elif self.staterx == GET_START_BIT: - out += self.get_start_bit(rx) - elif self.staterx == GET_DATA_BITS: - out += self.get_data_bits(rx) - elif self.staterx == GET_PARITY_BIT: - out += self.get_parity_bit(rx) - elif self.staterx == GET_STOP_BITS: - out += self.get_stop_bits(rx) - else: - raise Exception('Invalid state: %s' % self.staterx) - - # Save current RX/TX values for the next round. - self.oldrx = rx - # self.oldtx = tx - - if out != []: - self.put(out) + for rxtx in (RX, TX): + signal = rx if (rxtx == RX) else tx + + if self.state[rxtx] == WAIT_FOR_START_BIT: + self.wait_for_start_bit(rxtx, self.oldbit[rxtx], signal) + elif self.state[rxtx] == GET_START_BIT: + self.get_start_bit(rxtx, signal) + elif self.state[rxtx] == GET_DATA_BITS: + self.get_data_bits(rxtx, signal) + elif self.state[rxtx] == GET_PARITY_BIT: + self.get_parity_bit(rxtx, signal) + elif self.state[rxtx] == GET_STOP_BITS: + self.get_stop_bits(rxtx, signal) + else: + raise Exception('Invalid state: %s' % self.state[rxtx]) + + # Save current RX/TX values for the next round. + self.oldbit[rxtx] = signal