From 8294019152ac91351595c917c4c4ffdb0f7e78d5 Mon Sep 17 00:00:00 2001 From: Fabio Pugliese Ornellas Date: Wed, 1 Sep 2021 21:50:20 +0100 Subject: [PATCH 1/3] Add upstream uart decoder --- libsigrokdecode4DSL/decoders/uart/__init__.py | 40 ++ libsigrokdecode4DSL/decoders/uart/pd.py | 575 ++++++++++++++++++ 2 files changed, 615 insertions(+) create mode 100644 libsigrokdecode4DSL/decoders/uart/__init__.py create mode 100644 libsigrokdecode4DSL/decoders/uart/pd.py diff --git a/libsigrokdecode4DSL/decoders/uart/__init__.py b/libsigrokdecode4DSL/decoders/uart/__init__.py new file mode 100644 index 00000000..ce6136f1 --- /dev/null +++ b/libsigrokdecode4DSL/decoders/uart/__init__.py @@ -0,0 +1,40 @@ +## +## This file is part of the libsigrokdecode project. +## +## Copyright (C) 2012 Uwe Hermann +## +## This program is free software; you can redistribute it and/or modify +## it under the terms of the GNU General Public License as published by +## the Free Software Foundation; either version 2 of the License, or +## (at your option) any later version. +## +## This program is distributed in the hope that it will be useful, +## but WITHOUT ANY WARRANTY; without even the implied warranty of +## MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +## GNU General Public License for more details. +## +## You should have received a copy of the GNU General Public License +## along with this program; if not, see . +## + +''' +UART (Universal Asynchronous Receiver Transmitter) is a simple serial +communication protocol which allows two devices to talk to each other. + +This decoder should work on all "UART-like" async protocols with one +start bit (0), 5-9 databits, an (optional) parity bit, and one or more +stop bits (1), in this order. + +It can be run on one signal line (RX or TX) only, or on two lines (RX + TX). + +There are various standards for the physical layer specification of the +signals, including RS232, (TTL) UART, RS485, and others. However, the logic +level of the respective pins is only relevant when acquiring the data via +a logic analyzer (you have to select the correct logic analyzer and/or +the correct place where to probe). Once the data is in digital form and +matches the "UART" description above, this protocol decoder can work with +it though, no matter whether the source was on TTL UART levels, or RS232, +or others. +''' + +from .pd import Decoder diff --git a/libsigrokdecode4DSL/decoders/uart/pd.py b/libsigrokdecode4DSL/decoders/uart/pd.py new file mode 100644 index 00000000..0c501b0d --- /dev/null +++ b/libsigrokdecode4DSL/decoders/uart/pd.py @@ -0,0 +1,575 @@ +## +## This file is part of the libsigrokdecode project. +## +## Copyright (C) 2011-2014 Uwe Hermann +## +## This program is free software; you can redistribute it and/or modify +## it under the terms of the GNU General Public License as published by +## the Free Software Foundation; either version 2 of the License, or +## (at your option) any later version. +## +## This program is distributed in the hope that it will be useful, +## but WITHOUT ANY WARRANTY; without even the implied warranty of +## MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +## GNU General Public License for more details. +## +## You should have received a copy of the GNU General Public License +## along with this program; if not, see . +## + +import sigrokdecode as srd +from common.srdhelper import bitpack +from math import floor, ceil + +''' +OUTPUT_PYTHON format: + +Packet: +[, , ] + +This is the list of s and their respective values: + - 'STARTBIT': The data is the (integer) value of the start bit (0/1). + - 'DATA': This is always a tuple containing two items: + - 1st item: the (integer) value of the UART data. Valid values + range from 0 to 511 (as the data can be up to 9 bits in size). + - 2nd item: the list of individual data bits and their ss/es numbers. + - 'PARITYBIT': The data is the (integer) value of the parity bit (0/1). + - 'STOPBIT': The data is the (integer) value of the stop bit (0 or 1). + - 'INVALID STARTBIT': The data is the (integer) value of the start bit (0/1). + - 'INVALID STOPBIT': The data is the (integer) value of the stop bit (0/1). + - '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. + - 'BREAK': The data is always 0. + - 'FRAME': The data is always a tuple containing two items: The (integer) + value of the UART data, and a boolean which reflects the validity of the + UART frame. + - 'IDLE': The data is always 0. + +The field is 0 for RX packets, 1 for TX packets. +''' + +# Used for differentiating between the two data directions. +RX = 0 +TX = 1 + +# 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, +# usually 8 bits) return True if the parity is correct, False otherwise. +# 'none' is _not_ allowed as value for 'parity_type'. +def parity_ok(parity_type, parity_bit, data, data_bits): + + if parity_type == 'ignore': + return True + + # Handle easy cases first (parity bit is always 1 or 0). + if parity_type == 'zero': + return parity_bit == 0 + elif parity_type == 'one': + return parity_bit == 1 + + # Count number of 1 (high) bits in the data (and the parity bit itself!). + ones = bin(data).count('1') + parity_bit + + # Check for odd/even parity. + if parity_type == 'odd': + return (ones % 2) == 1 + elif parity_type == 'even': + return (ones % 2) == 0 + +class SamplerateError(Exception): + pass + +class ChannelError(Exception): + pass + +class Ann: + RX_DATA, TX_DATA, RX_START, TX_START, RX_PARITY_OK, TX_PARITY_OK, \ + RX_PARITY_ERR, TX_PARITY_ERR, RX_STOP, TX_STOP, RX_WARN, TX_WARN, \ + RX_DATA_BIT, TX_DATA_BIT, RX_BREAK, TX_BREAK, RX_PACKET, TX_PACKET = \ + range(18) + +class Bin: + RX, TX, RXTX = range(3) + +class Decoder(srd.Decoder): + api_version = 3 + id = 'uart' + name = 'UART' + longname = 'Universal Asynchronous Receiver/Transmitter' + desc = 'Asynchronous, serial bus.' + license = 'gplv2+' + inputs = ['logic'] + outputs = ['uart'] + tags = ['Embedded/industrial'] + optional_channels = ( + # Allow specifying only one of the signals, e.g. if only one data + # direction exists (or is relevant). + {'id': 'rx', 'name': 'RX', 'desc': 'UART receive line'}, + {'id': 'tx', 'name': 'TX', 'desc': 'UART transmit line'}, + ) + options = ( + {'id': 'baudrate', 'desc': 'Baud rate', 'default': 115200}, + {'id': 'data_bits', 'desc': 'Data bits', 'default': 8, + 'values': (5, 6, 7, 8, 9)}, + {'id': 'parity', 'desc': 'Parity', 'default': 'none', + 'values': ('none', 'odd', 'even', 'zero', 'one', 'ignore')}, + {'id': 'stop_bits', 'desc': 'Stop bits', 'default': 1.0, + 'values': (0.0, 0.5, 1.0, 1.5)}, + {'id': 'bit_order', 'desc': 'Bit order', 'default': 'lsb-first', + 'values': ('lsb-first', 'msb-first')}, + {'id': 'format', 'desc': 'Data format', 'default': 'hex', + 'values': ('ascii', 'dec', 'hex', 'oct', 'bin')}, + {'id': 'invert_rx', 'desc': 'Invert RX', 'default': 'no', + 'values': ('yes', 'no')}, + {'id': 'invert_tx', 'desc': 'Invert TX', 'default': 'no', + 'values': ('yes', 'no')}, + {'id': 'sample_point', 'desc': 'Sample point (%)', 'default': 50}, + {'id': 'rx_packet_delim', 'desc': 'RX packet delimiter (decimal)', + 'default': -1}, + {'id': 'tx_packet_delim', 'desc': 'TX packet delimiter (decimal)', + 'default': -1}, + {'id': 'rx_packet_len', 'desc': 'RX packet length', 'default': -1}, + {'id': 'tx_packet_len', 'desc': 'TX packet length', 'default': -1}, + ) + annotations = ( + ('rx-data', 'RX data'), + ('tx-data', 'TX data'), + ('rx-start', 'RX start bit'), + ('tx-start', 'TX start bit'), + ('rx-parity-ok', 'RX parity OK bit'), + ('tx-parity-ok', 'TX parity OK bit'), + ('rx-parity-err', 'RX parity error bit'), + ('tx-parity-err', 'TX parity error bit'), + ('rx-stop', 'RX stop bit'), + ('tx-stop', 'TX stop bit'), + ('rx-warning', 'RX warning'), + ('tx-warning', 'TX warning'), + ('rx-data-bit', 'RX data bit'), + ('tx-data-bit', 'TX data bit'), + ('rx-break', 'RX break'), + ('tx-break', 'TX break'), + ('rx-packet', 'RX packet'), + ('tx-packet', 'TX packet'), + ) + annotation_rows = ( + ('rx-data-bits', 'RX bits', (Ann.RX_DATA_BIT,)), + ('rx-data-vals', 'RX data', (Ann.RX_DATA, Ann.RX_START, Ann.RX_PARITY_OK, Ann.RX_PARITY_ERR, Ann.RX_STOP)), + ('rx-warnings', 'RX warnings', (Ann.RX_WARN,)), + ('rx-breaks', 'RX breaks', (Ann.RX_BREAK,)), + ('rx-packets', 'RX packets', (Ann.RX_PACKET,)), + ('tx-data-bits', 'TX bits', (Ann.TX_DATA_BIT,)), + ('tx-data-vals', 'TX data', (Ann.TX_DATA, Ann.TX_START, Ann.TX_PARITY_OK, Ann.TX_PARITY_ERR, Ann.TX_STOP)), + ('tx-warnings', 'TX warnings', (Ann.TX_WARN,)), + ('tx-breaks', 'TX breaks', (Ann.TX_BREAK,)), + ('tx-packets', 'TX packets', (Ann.TX_PACKET,)), + ) + binary = ( + ('rx', 'RX dump'), + ('tx', 'TX dump'), + ('rxtx', 'RX/TX dump'), + ) + idle_state = ['WAIT FOR START BIT', 'WAIT FOR START BIT'] + + def putx(self, rxtx, data): + s, halfbit = self.startsample[rxtx], self.bit_width / 2.0 + self.put(s - floor(halfbit), self.samplenum + ceil(halfbit), self.out_ann, data) + + def putx_packet(self, rxtx, data): + s, halfbit = self.ss_packet[rxtx], self.bit_width / 2.0 + self.put(s - floor(halfbit), self.samplenum + ceil(halfbit), self.out_ann, data) + + def putpx(self, rxtx, data): + s, halfbit = self.startsample[rxtx], self.bit_width / 2.0 + self.put(s - floor(halfbit), self.samplenum + ceil(halfbit), self.out_python, data) + + def putg(self, data): + s, halfbit = self.samplenum, self.bit_width / 2.0 + self.put(s - floor(halfbit), s + ceil(halfbit), self.out_ann, data) + + def putp(self, data): + s, halfbit = self.samplenum, self.bit_width / 2.0 + self.put(s - floor(halfbit), s + ceil(halfbit), self.out_python, data) + + def putgse(self, ss, es, data): + self.put(ss, es, self.out_ann, data) + + def putpse(self, ss, es, data): + self.put(ss, es, self.out_python, data) + + def putbin(self, rxtx, data): + s, halfbit = self.startsample[rxtx], self.bit_width / 2.0 + self.put(s - floor(halfbit), self.samplenum + ceil(halfbit), self.out_binary, data) + + def __init__(self): + self.reset() + + def reset(self): + self.samplerate = None + self.frame_start = [-1, -1] + self.frame_valid = [None, None] + self.startbit = [-1, -1] + self.cur_data_bit = [0, 0] + self.datavalue = [0, 0] + self.paritybit = [-1, -1] + self.stopbit1 = [-1, -1] + self.startsample = [-1, -1] + self.state = ['WAIT FOR START BIT', 'WAIT FOR START BIT'] + self.databits = [[], []] + self.break_start = [None, None] + self.packet_cache = [[], []] + self.ss_packet, self.es_packet = [None, None], [None, None] + self.idle_start = [None, None] + + def start(self): + self.out_python = self.register(srd.OUTPUT_PYTHON) + self.out_binary = self.register(srd.OUTPUT_BINARY) + self.out_ann = self.register(srd.OUTPUT_ANN) + self.bw = (self.options['data_bits'] + 7) // 8 + + def metadata(self, key, value): + if key == srd.SRD_CONF_SAMPLERATE: + self.samplerate = value + # The width of one UART bit in number of samples. + self.bit_width = float(self.samplerate) / float(self.options['baudrate']) + + def get_sample_point(self, rxtx, bitnum): + # Determine absolute sample number of a bit slot's sample point. + # Counts for UART bits start from 0 (0 = start bit, 1..x = data, + # x+1 = parity bit (if used) or the first stop bit, and so on). + # Accept a position in the range of 1-99% of the full bit width. + # Assume 50% for invalid input specs for backwards compatibility. + perc = self.options['sample_point'] or 50 + if not perc or perc not in range(1, 100): + perc = 50 + perc /= 100.0 + bitpos = (self.bit_width - 1) * perc + bitpos += self.frame_start[rxtx] + bitpos += bitnum * self.bit_width + return bitpos + + def wait_for_start_bit(self, rxtx, signal): + # Save the sample number where the start bit begins. + self.frame_start[rxtx] = self.samplenum + self.frame_valid[rxtx] = True + + self.state[rxtx] = 'GET START BIT' + + def get_start_bit(self, rxtx, signal): + self.startbit[rxtx] = signal + + # The startbit must be 0. If not, we report an error and wait + # for the next start bit (assuming this one was spurious). + if self.startbit[rxtx] != 0: + self.putp(['INVALID STARTBIT', rxtx, self.startbit[rxtx]]) + self.putg([Ann.RX_WARN + rxtx, ['Frame error', 'Frame err', 'FE']]) + self.frame_valid[rxtx] = False + es = self.samplenum + ceil(self.bit_width / 2.0) + self.putpse(self.frame_start[rxtx], es, ['FRAME', rxtx, + (self.datavalue[rxtx], self.frame_valid[rxtx])]) + self.state[rxtx] = 'WAIT FOR START BIT' + return + + self.cur_data_bit[rxtx] = 0 + self.datavalue[rxtx] = 0 + self.startsample[rxtx] = -1 + + self.putp(['STARTBIT', rxtx, self.startbit[rxtx]]) + self.putg([Ann.RX_START + rxtx, ['Start bit', 'Start', 'S']]) + + self.state[rxtx] = 'GET DATA BITS' + + def handle_packet(self, rxtx): + d = 'rx' if (rxtx == RX) else 'tx' + delim = self.options[d + '_packet_delim'] + plen = self.options[d + '_packet_len'] + if delim == -1 and plen == -1: + return + + # Cache data values until we see the delimiter and/or the specified + # packet length has been reached (whichever happens first). + if len(self.packet_cache[rxtx]) == 0: + self.ss_packet[rxtx] = self.startsample[rxtx] + self.packet_cache[rxtx].append(self.datavalue[rxtx]) + if self.datavalue[rxtx] == delim or len(self.packet_cache[rxtx]) == plen: + self.es_packet[rxtx] = self.samplenum + s = '' + for b in self.packet_cache[rxtx]: + s += self.format_value(b) + if self.options['format'] != 'ascii': + s += ' ' + if self.options['format'] != 'ascii' and s[-1] == ' ': + s = s[:-1] # Drop trailing space. + self.putx_packet(rxtx, [Ann.RX_PACKET + rxtx, [s]]) + self.packet_cache[rxtx] = [] + + def get_data_bits(self, rxtx, signal): + # Save the sample number of the middle of the first data bit. + if self.startsample[rxtx] == -1: + self.startsample[rxtx] = self.samplenum + + self.putg([Ann.RX_DATA_BIT + rxtx, ['%d' % signal]]) + + # Store individual data bits and their start/end samplenumbers. + s, halfbit = self.samplenum, int(self.bit_width / 2) + self.databits[rxtx].append([signal, s - halfbit, s + halfbit]) + + # Return here, unless we already received all data bits. + self.cur_data_bit[rxtx] += 1 + if self.cur_data_bit[rxtx] < self.options['data_bits']: + return + + # Convert accumulated data bits to a data value. + bits = [b[0] for b in self.databits[rxtx]] + if self.options['bit_order'] == 'msb-first': + bits.reverse() + self.datavalue[rxtx] = bitpack(bits) + self.putpx(rxtx, ['DATA', rxtx, + (self.datavalue[rxtx], self.databits[rxtx])]) + + b = self.datavalue[rxtx] + formatted = self.format_value(b) + if formatted is not None: + self.putx(rxtx, [rxtx, [formatted]]) + + bdata = b.to_bytes(self.bw, byteorder='big') + self.putbin(rxtx, [Bin.RX + rxtx, bdata]) + self.putbin(rxtx, [Bin.RXTX, bdata]) + + self.handle_packet(rxtx) + + self.databits[rxtx] = [] + + # Advance to either reception of the parity bit, or reception of + # the STOP bits if parity is not applicable. + self.state[rxtx] = 'GET PARITY BIT' + if self.options['parity'] == 'none': + self.state[rxtx] = 'GET STOP BITS' + + def format_value(self, v): + # Format value 'v' according to configured options. + # Reflects the user selected kind of representation, as well as + # the number of data bits in the UART frames. + + fmt, bits = self.options['format'], self.options['data_bits'] + + # Assume "is printable" for values from 32 to including 126, + # below 32 is "control" and thus not printable, above 127 is + # "not ASCII" in its strict sense, 127 (DEL) is not printable, + # fall back to hex representation for non-printables. + if fmt == 'ascii': + if v in range(32, 126 + 1): + return chr(v) + hexfmt = "[{:02X}]" if bits <= 8 else "[{:03X}]" + return hexfmt.format(v) + + # Mere number to text conversion without prefix and padding + # for the "decimal" output format. + if fmt == 'dec': + return "{:d}".format(v) + + # Padding with leading zeroes for hex/oct/bin formats, but + # without a prefix for density -- since the format is user + # specified, there is no ambiguity. + if fmt == 'hex': + digits = (bits + 4 - 1) // 4 + fmtchar = "X" + elif fmt == 'oct': + digits = (bits + 3 - 1) // 3 + fmtchar = "o" + elif fmt == 'bin': + digits = bits + fmtchar = "b" + else: + fmtchar = None + if fmtchar is not None: + fmt = "{{:0{:d}{:s}}}".format(digits, fmtchar) + return fmt.format(v) + + return None + + def get_parity_bit(self, rxtx, signal): + self.paritybit[rxtx] = signal + + if parity_ok(self.options['parity'], self.paritybit[rxtx], + self.datavalue[rxtx], self.options['data_bits']): + self.putp(['PARITYBIT', rxtx, self.paritybit[rxtx]]) + self.putg([Ann.RX_PARITY_OK + rxtx, ['Parity bit', 'Parity', 'P']]) + else: + # TODO: Return expected/actual parity values. + self.putp(['PARITY ERROR', rxtx, (0, 1)]) # FIXME: Dummy tuple... + self.putg([Ann.RX_PARITY_ERR + rxtx, ['Parity error', 'Parity err', 'PE']]) + self.frame_valid[rxtx] = False + + self.state[rxtx] = 'GET STOP BITS' + + # TODO: Currently only supports 1 stop bit. + def get_stop_bits(self, rxtx, signal): + self.stopbit1[rxtx] = signal + + # Stop bits must be 1. If not, we report an error. + if self.stopbit1[rxtx] != 1: + self.putp(['INVALID STOPBIT', rxtx, self.stopbit1[rxtx]]) + self.putg([Ann.RX_WARN + rxtx, ['Frame error', 'Frame err', 'FE']]) + self.frame_valid[rxtx] = False + + self.putp(['STOPBIT', rxtx, self.stopbit1[rxtx]]) + self.putg([Ann.RX_STOP + rxtx, ['Stop bit', 'Stop', 'T']]) + + # Pass the complete UART frame to upper layers. + es = self.samplenum + ceil(self.bit_width / 2.0) + self.putpse(self.frame_start[rxtx], es, ['FRAME', rxtx, + (self.datavalue[rxtx], self.frame_valid[rxtx])]) + + self.state[rxtx] = 'WAIT FOR START BIT' + self.idle_start[rxtx] = self.frame_start[rxtx] + self.frame_len_sample_count + + def handle_break(self, rxtx): + self.putpse(self.frame_start[rxtx], self.samplenum, + ['BREAK', rxtx, 0]) + self.putgse(self.frame_start[rxtx], self.samplenum, + [Ann.RX_BREAK + rxtx, ['Break condition', 'Break', 'Brk', 'B']]) + self.state[rxtx] = 'WAIT FOR START BIT' + + def get_wait_cond(self, rxtx, inv): + # Return condititions that are suitable for Decoder.wait(). Those + # conditions either match the falling edge of the START bit, or + # the sample point of the next bit time. + state = self.state[rxtx] + if state == 'WAIT FOR START BIT': + return {rxtx: 'r' if inv else 'f'} + if state == 'GET START BIT': + bitnum = 0 + elif state == 'GET DATA BITS': + bitnum = 1 + self.cur_data_bit[rxtx] + elif state == 'GET PARITY BIT': + bitnum = 1 + self.options['data_bits'] + elif state == 'GET STOP BITS': + bitnum = 1 + self.options['data_bits'] + bitnum += 0 if self.options['parity'] == 'none' else 1 + want_num = ceil(self.get_sample_point(rxtx, bitnum)) + return {'skip': want_num - self.samplenum} + + def get_idle_cond(self, rxtx, inv): + # Return a condition that corresponds to the (expected) end of + # the next frame, assuming that it will be an "idle frame" + # (constant high input level for the frame's length). + if self.idle_start[rxtx] is None: + return None + end_of_frame = self.idle_start[rxtx] + self.frame_len_sample_count + if end_of_frame < self.samplenum: + return None + return {'skip': end_of_frame - self.samplenum} + + def inspect_sample(self, rxtx, signal, inv): + # Inspect a sample returned by .wait() for the specified UART line. + if inv: + signal = not signal + + state = self.state[rxtx] + if state == 'WAIT FOR START BIT': + self.wait_for_start_bit(rxtx, signal) + elif state == 'GET START BIT': + self.get_start_bit(rxtx, signal) + elif state == 'GET DATA BITS': + self.get_data_bits(rxtx, signal) + elif state == 'GET PARITY BIT': + self.get_parity_bit(rxtx, signal) + elif state == 'GET STOP BITS': + self.get_stop_bits(rxtx, signal) + + def inspect_edge(self, rxtx, signal, inv): + # Inspect edges, independently from traffic, to detect break conditions. + if inv: + signal = not signal + if not signal: + # Signal went low. Start another interval. + self.break_start[rxtx] = self.samplenum + return + # Signal went high. Was there an extended period with low signal? + if self.break_start[rxtx] is None: + return + diff = self.samplenum - self.break_start[rxtx] + if diff >= self.break_min_sample_count: + self.handle_break(rxtx) + self.break_start[rxtx] = None + + def inspect_idle(self, rxtx, signal, inv): + # Check each edge and each period of stable input (either level). + # Can derive the "idle frame period has passed" condition. + if inv: + signal = not signal + if not signal: + # Low input, cease inspection. + self.idle_start[rxtx] = None + return + # High input, either just reached, or still stable. + if self.idle_start[rxtx] is None: + self.idle_start[rxtx] = self.samplenum + diff = self.samplenum - self.idle_start[rxtx] + if diff < self.frame_len_sample_count: + return + ss, es = self.idle_start[rxtx], self.samplenum + self.putpse(ss, es, ['IDLE', rxtx, 0]) + self.idle_start[rxtx] = self.samplenum + + def decode(self): + if not self.samplerate: + raise SamplerateError('Cannot decode without samplerate.') + + has_pin = [self.has_channel(ch) for ch in (RX, TX)] + if not True in has_pin: + raise ChannelError('Need at least one of TX or RX pins.') + + opt = self.options + inv = [opt['invert_rx'] == 'yes', opt['invert_tx'] == 'yes'] + cond_data_idx = [None] * len(has_pin) + + # Determine the number of samples for a complete frame's time span. + # A period of low signal (at least) that long is a break condition. + frame_samples = 1 # START + frame_samples += self.options['data_bits'] + frame_samples += 0 if self.options['parity'] == 'none' else 1 + frame_samples += self.options['stop_bits'] + frame_samples *= self.bit_width + self.frame_len_sample_count = ceil(frame_samples) + self.break_min_sample_count = self.frame_len_sample_count + cond_edge_idx = [None] * len(has_pin) + cond_idle_idx = [None] * len(has_pin) + + while True: + conds = [] + if has_pin[RX]: + cond_data_idx[RX] = len(conds) + conds.append(self.get_wait_cond(RX, inv[RX])) + cond_edge_idx[RX] = len(conds) + conds.append({RX: 'e'}) + cond_idle_idx[RX] = None + idle_cond = self.get_idle_cond(RX, inv[RX]) + if idle_cond: + cond_idle_idx[RX] = len(conds) + conds.append(idle_cond) + if has_pin[TX]: + cond_data_idx[TX] = len(conds) + conds.append(self.get_wait_cond(TX, inv[TX])) + cond_edge_idx[TX] = len(conds) + conds.append({TX: 'e'}) + cond_idle_idx[TX] = None + idle_cond = self.get_idle_cond(TX, inv[TX]) + if idle_cond: + cond_idle_idx[TX] = len(conds) + conds.append(idle_cond) + (rx, tx) = self.wait(conds) + if cond_data_idx[RX] is not None and self.matched[cond_data_idx[RX]]: + self.inspect_sample(RX, rx, inv[RX]) + if cond_edge_idx[RX] is not None and self.matched[cond_edge_idx[RX]]: + self.inspect_edge(RX, rx, inv[RX]) + self.inspect_idle(RX, rx, inv[RX]) + if cond_idle_idx[RX] is not None and self.matched[cond_idle_idx[RX]]: + self.inspect_idle(RX, rx, inv[RX]) + if cond_data_idx[TX] is not None and self.matched[cond_data_idx[TX]]: + self.inspect_sample(TX, tx, inv[TX]) + if cond_edge_idx[TX] is not None and self.matched[cond_edge_idx[TX]]: + self.inspect_edge(TX, tx, inv[TX]) + self.inspect_idle(TX, tx, inv[TX]) + if cond_idle_idx[TX] is not None and self.matched[cond_idle_idx[TX]]: + self.inspect_idle(TX, tx, inv[TX]) From 718db8e128c16a939e0047726e249cd677de62ff Mon Sep 17 00:00:00 2001 From: Fabio Pugliese Ornellas Date: Wed, 1 Sep 2021 21:51:09 +0100 Subject: [PATCH 2/3] Patch to fix issue with #430 --- libsigrokdecode4DSL/decoders/uart/pd.py | 13 +++++++------ 1 file changed, 7 insertions(+), 6 deletions(-) diff --git a/libsigrokdecode4DSL/decoders/uart/pd.py b/libsigrokdecode4DSL/decoders/uart/pd.py index 0c501b0d..457e14b3 100644 --- a/libsigrokdecode4DSL/decoders/uart/pd.py +++ b/libsigrokdecode4DSL/decoders/uart/pd.py @@ -559,17 +559,18 @@ def decode(self): cond_idle_idx[TX] = len(conds) conds.append(idle_cond) (rx, tx) = self.wait(conds) - if cond_data_idx[RX] is not None and self.matched[cond_data_idx[RX]]: + + if cond_data_idx[RX] is not None and (self.matched & (0b1 << cond_data_idx[RX])): self.inspect_sample(RX, rx, inv[RX]) - if cond_edge_idx[RX] is not None and self.matched[cond_edge_idx[RX]]: + if cond_edge_idx[RX] is not None and (self.matched & (0b1 << cond_edge_idx[RX])): self.inspect_edge(RX, rx, inv[RX]) self.inspect_idle(RX, rx, inv[RX]) - if cond_idle_idx[RX] is not None and self.matched[cond_idle_idx[RX]]: + if cond_idle_idx[RX] is not None and (self.matched & (0b1 << cond_idle_idx[RX])): self.inspect_idle(RX, rx, inv[RX]) - if cond_data_idx[TX] is not None and self.matched[cond_data_idx[TX]]: + if cond_data_idx[TX] is not None and (self.matched & (0b1 << cond_data_idx[TX])): self.inspect_sample(TX, tx, inv[TX]) - if cond_edge_idx[TX] is not None and self.matched[cond_edge_idx[TX]]: + if cond_edge_idx[TX] is not None and (self.matched & (0b1 << cond_edge_idx[TX])): self.inspect_edge(TX, tx, inv[TX]) self.inspect_idle(TX, tx, inv[TX]) - if cond_idle_idx[TX] is not None and self.matched[cond_idle_idx[TX]]: + if cond_idle_idx[TX] is not None and (self.matched & (0b1 << cond_idle_idx[TX])): self.inspect_idle(TX, tx, inv[TX]) From 8de9dde5097a51256ad2a4a7be093654425a4117 Mon Sep 17 00:00:00 2001 From: Fabio Pugliese Ornellas Date: Mon, 5 Sep 2022 22:26:25 +0100 Subject: [PATCH 3/3] use sigrok's uart decoder in favor of 1:uart --- libsigrokdecode4DSL/decoders/1-uart/pd.py | 517 +++++++++++----- libsigrokdecode4DSL/decoders/uart/__init__.py | 40 -- libsigrokdecode4DSL/decoders/uart/pd.py | 576 ------------------ 3 files changed, 348 insertions(+), 785 deletions(-) delete mode 100644 libsigrokdecode4DSL/decoders/uart/__init__.py delete mode 100644 libsigrokdecode4DSL/decoders/uart/pd.py diff --git a/libsigrokdecode4DSL/decoders/1-uart/pd.py b/libsigrokdecode4DSL/decoders/1-uart/pd.py index b8b26201..457e14b3 100644 --- a/libsigrokdecode4DSL/decoders/1-uart/pd.py +++ b/libsigrokdecode4DSL/decoders/1-uart/pd.py @@ -2,7 +2,6 @@ ## This file is part of the libsigrokdecode project. ## ## Copyright (C) 2011-2014 Uwe Hermann -## Copyright (C) 2019 DreamSourceLab ## ## This program is free software; you can redistribute it and/or modify ## it under the terms of the GNU General Public License as published by @@ -44,14 +43,23 @@ - 'FRAME': The data is always a tuple containing two items: The (integer) value of the UART data, and a boolean which reflects the validity of the UART frame. + - 'IDLE': The data is always 0. +The field is 0 for RX packets, 1 for TX packets. ''' +# Used for differentiating between the two data directions. +RX = 0 +TX = 1 + # 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, # usually 8 bits) return True if the parity is correct, False otherwise. # 'none' is _not_ allowed as value for 'parity_type'. -def parity_ok(parity_type, parity_bit, data, num_data_bits): +def parity_ok(parity_type, parity_bit, data, data_bits): + + if parity_type == 'ignore': + return True # Handle easy cases first (parity bit is always 1 or 0). if parity_type == 'zero': @@ -74,68 +82,104 @@ class SamplerateError(Exception): class ChannelError(Exception): pass +class Ann: + RX_DATA, TX_DATA, RX_START, TX_START, RX_PARITY_OK, TX_PARITY_OK, \ + RX_PARITY_ERR, TX_PARITY_ERR, RX_STOP, TX_STOP, RX_WARN, TX_WARN, \ + RX_DATA_BIT, TX_DATA_BIT, RX_BREAK, TX_BREAK, RX_PACKET, TX_PACKET = \ + range(18) + +class Bin: + RX, TX, RXTX = range(3) + class Decoder(srd.Decoder): api_version = 3 - id = '1:uart' - name = '1:UART' + id = 'uart' + name = 'UART' longname = 'Universal Asynchronous Receiver/Transmitter' desc = 'Asynchronous, serial bus.' license = 'gplv2+' inputs = ['logic'] outputs = ['uart'] tags = ['Embedded/industrial'] - channels = ( - {'id': 'rxtx', 'type': 209, 'name': 'RX/TX', 'desc': 'UART transceive line'}, + optional_channels = ( + # Allow specifying only one of the signals, e.g. if only one data + # direction exists (or is relevant). + {'id': 'rx', 'name': 'RX', 'desc': 'UART receive line'}, + {'id': 'tx', 'name': 'TX', 'desc': 'UART transmit line'}, ) options = ( {'id': 'baudrate', 'desc': 'Baud rate', 'default': 115200}, - {'id': 'num_data_bits', 'desc': 'Data bits', 'default': 8, - 'values': tuple(range(4,129,1))}, - {'id': 'parity_type', 'desc': 'Parity type', 'default': 'none', - 'values': ('none', 'odd', 'even', 'zero', 'one')}, - {'id': 'parity_check', 'desc': 'Check parity?', 'default': 'yes', - 'values': ('yes', 'no')}, - {'id': 'num_stop_bits', 'desc': 'Stop bits', 'default': 1.0, - 'values': (0.0, 0.5, 1.0, 1.5, 2.0, 2.5)}, + {'id': 'data_bits', 'desc': 'Data bits', 'default': 8, + 'values': (5, 6, 7, 8, 9)}, + {'id': 'parity', 'desc': 'Parity', 'default': 'none', + 'values': ('none', 'odd', 'even', 'zero', 'one', 'ignore')}, + {'id': 'stop_bits', 'desc': 'Stop bits', 'default': 1.0, + 'values': (0.0, 0.5, 1.0, 1.5)}, {'id': 'bit_order', 'desc': 'Bit order', 'default': 'lsb-first', 'values': ('lsb-first', 'msb-first')}, {'id': 'format', 'desc': 'Data format', 'default': 'hex', 'values': ('ascii', 'dec', 'hex', 'oct', 'bin')}, - {'id': 'invert', 'desc': 'Invert Signal?', 'default': 'no', + {'id': 'invert_rx', 'desc': 'Invert RX', 'default': 'no', 'values': ('yes', 'no')}, - {'id': 'anno_startstop', 'desc': 'Display Start/Stop?', 'default': 'yes', + {'id': 'invert_tx', 'desc': 'Invert TX', 'default': 'no', 'values': ('yes', 'no')}, + {'id': 'sample_point', 'desc': 'Sample point (%)', 'default': 50}, + {'id': 'rx_packet_delim', 'desc': 'RX packet delimiter (decimal)', + 'default': -1}, + {'id': 'tx_packet_delim', 'desc': 'TX packet delimiter (decimal)', + 'default': -1}, + {'id': 'rx_packet_len', 'desc': 'RX packet length', 'default': -1}, + {'id': 'tx_packet_len', 'desc': 'TX packet length', 'default': -1}, ) annotations = ( - ('108', 'data', 'data'), - ('7', 'start', 'start bits'), - ('6', 'parity-ok', 'parity OK bits'), - ('0', 'parity-err', 'parity error bits'), - ('1', 'stop', 'stop bits'), - ('1000', 'warnings', 'warnings'), - ('209', 'data-bits', 'data bits'), - ('10', 'break', 'break'), + ('rx-data', 'RX data'), + ('tx-data', 'TX data'), + ('rx-start', 'RX start bit'), + ('tx-start', 'TX start bit'), + ('rx-parity-ok', 'RX parity OK bit'), + ('tx-parity-ok', 'TX parity OK bit'), + ('rx-parity-err', 'RX parity error bit'), + ('tx-parity-err', 'TX parity error bit'), + ('rx-stop', 'RX stop bit'), + ('tx-stop', 'TX stop bit'), + ('rx-warning', 'RX warning'), + ('tx-warning', 'TX warning'), + ('rx-data-bit', 'RX data bit'), + ('tx-data-bit', 'TX data bit'), + ('rx-break', 'RX break'), + ('tx-break', 'TX break'), + ('rx-packet', 'RX packet'), + ('tx-packet', 'TX packet'), ) annotation_rows = ( - ('data', 'RX/TX', (0, 1, 2, 3, 4)), - ('data-bits', 'Bits', (6,)), - ('warnings', 'Warnings', (5,)), - ('break', 'break', (7,)), + ('rx-data-bits', 'RX bits', (Ann.RX_DATA_BIT,)), + ('rx-data-vals', 'RX data', (Ann.RX_DATA, Ann.RX_START, Ann.RX_PARITY_OK, Ann.RX_PARITY_ERR, Ann.RX_STOP)), + ('rx-warnings', 'RX warnings', (Ann.RX_WARN,)), + ('rx-breaks', 'RX breaks', (Ann.RX_BREAK,)), + ('rx-packets', 'RX packets', (Ann.RX_PACKET,)), + ('tx-data-bits', 'TX bits', (Ann.TX_DATA_BIT,)), + ('tx-data-vals', 'TX data', (Ann.TX_DATA, Ann.TX_START, Ann.TX_PARITY_OK, Ann.TX_PARITY_ERR, Ann.TX_STOP)), + ('tx-warnings', 'TX warnings', (Ann.TX_WARN,)), + ('tx-breaks', 'TX breaks', (Ann.TX_BREAK,)), + ('tx-packets', 'TX packets', (Ann.TX_PACKET,)), ) binary = ( + ('rx', 'RX dump'), + ('tx', 'TX dump'), ('rxtx', 'RX/TX dump'), ) - idle_state = 'WAIT FOR START BIT' + idle_state = ['WAIT FOR START BIT', 'WAIT FOR START BIT'] + + def putx(self, rxtx, data): + s, halfbit = self.startsample[rxtx], self.bit_width / 2.0 + self.put(s - floor(halfbit), self.samplenum + ceil(halfbit), self.out_ann, data) - def putx(self, data): - s, halfbit = self.startsample, self.bit_width / 2.0 - if self.options['anno_startstop'] == 'yes' : - self.put(s - floor(halfbit), self.samplenum + ceil(halfbit), self.out_ann, data) - else : - self.put(self.frame_start, self.samplenum + ceil(halfbit * (1+self.options['num_stop_bits'])), self.out_ann, data) + def putx_packet(self, rxtx, data): + s, halfbit = self.ss_packet[rxtx], self.bit_width / 2.0 + self.put(s - floor(halfbit), self.samplenum + ceil(halfbit), self.out_ann, data) - def putpx(self, data): - s, halfbit = self.startsample, self.bit_width / 2.0 + def putpx(self, rxtx, data): + s, halfbit = self.startsample[rxtx], self.bit_width / 2.0 self.put(s - floor(halfbit), self.samplenum + ceil(halfbit), self.out_python, data) def putg(self, data): @@ -152,34 +196,35 @@ def putgse(self, ss, es, data): def putpse(self, ss, es, data): self.put(ss, es, self.out_python, data) - def putbin(self, data): - s, halfbit = self.startsample, self.bit_width / 2.0 + def putbin(self, rxtx, data): + s, halfbit = self.startsample[rxtx], self.bit_width / 2.0 self.put(s - floor(halfbit), self.samplenum + ceil(halfbit), self.out_binary, data) - def __init__(self): self.reset() def reset(self): self.samplerate = None - self.samplenum = 0 - self.frame_start = -1 - self.frame_valid = None - self.startbit = -1 - self.cur_data_bit = 0 - self.datavalue = 0 - self.paritybit = -1 - self.stopbit1 = -1 - self.startsample = -1 - self.state = 'WAIT FOR START BIT' - self.databits = [] - self.break_start = None + self.frame_start = [-1, -1] + self.frame_valid = [None, None] + self.startbit = [-1, -1] + self.cur_data_bit = [0, 0] + self.datavalue = [0, 0] + self.paritybit = [-1, -1] + self.stopbit1 = [-1, -1] + self.startsample = [-1, -1] + self.state = ['WAIT FOR START BIT', 'WAIT FOR START BIT'] + self.databits = [[], []] + self.break_start = [None, None] + self.packet_cache = [[], []] + self.ss_packet, self.es_packet = [None, None], [None, None] + self.idle_start = [None, None] def start(self): self.out_python = self.register(srd.OUTPUT_PYTHON) self.out_binary = self.register(srd.OUTPUT_BINARY) self.out_ann = self.register(srd.OUTPUT_ANN) - self.bw = (self.options['num_data_bits'] + 7) // 8 + self.bw = (self.options['data_bits'] + 7) // 8 def metadata(self, key, value): if key == srd.SRD_CONF_SAMPLERATE: @@ -187,211 +232,345 @@ def metadata(self, key, value): # The width of one UART bit in number of samples. self.bit_width = float(self.samplerate) / float(self.options['baudrate']) - def get_sample_point(self, bitnum): + def get_sample_point(self, rxtx, bitnum): # Determine absolute sample number of a bit slot's sample point. - # 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). - # The samples within bit are 0, 1, ..., (bit_width - 1), therefore - # index of the middle sample within bit window is (bit_width - 1) / 2. - bitpos = self.frame_start + (self.bit_width - 1) / 2.0 + # Counts for UART bits start from 0 (0 = start bit, 1..x = data, + # x+1 = parity bit (if used) or the first stop bit, and so on). + # Accept a position in the range of 1-99% of the full bit width. + # Assume 50% for invalid input specs for backwards compatibility. + perc = self.options['sample_point'] or 50 + if not perc or perc not in range(1, 100): + perc = 50 + perc /= 100.0 + bitpos = (self.bit_width - 1) * perc + bitpos += self.frame_start[rxtx] bitpos += bitnum * self.bit_width return bitpos - def wait_for_start_bit(self, signal): + def wait_for_start_bit(self, rxtx, signal): # Save the sample number where the start bit begins. - self.frame_start = self.samplenum - self.frame_valid = True + self.frame_start[rxtx] = self.samplenum + self.frame_valid[rxtx] = True - self.state = 'GET START BIT' + self.state[rxtx] = 'GET START BIT' - def get_start_bit(self, signal): - self.startbit = signal + def get_start_bit(self, rxtx, signal): + self.startbit[rxtx] = signal # The startbit must be 0. If not, we report an error and wait # for the next start bit (assuming this one was spurious). - if self.startbit != 0: - self.putp(['INVALID STARTBIT', 0, self.startbit]) - self.putg([5, ['Frame error', 'Frame err', 'FE']]) - self.frame_valid = False + if self.startbit[rxtx] != 0: + self.putp(['INVALID STARTBIT', rxtx, self.startbit[rxtx]]) + self.putg([Ann.RX_WARN + rxtx, ['Frame error', 'Frame err', 'FE']]) + self.frame_valid[rxtx] = False es = self.samplenum + ceil(self.bit_width / 2.0) - self.putpse(self.frame_start, es, ['FRAME', 0, - (self.datavalue, self.frame_valid)]) - self.state = 'WAIT FOR START BIT' + self.putpse(self.frame_start[rxtx], es, ['FRAME', rxtx, + (self.datavalue[rxtx], self.frame_valid[rxtx])]) + self.state[rxtx] = 'WAIT FOR START BIT' return - self.cur_data_bit = 0 - self.datavalue = 0 - self.startsample = -1 + self.cur_data_bit[rxtx] = 0 + self.datavalue[rxtx] = 0 + self.startsample[rxtx] = -1 + + self.putp(['STARTBIT', rxtx, self.startbit[rxtx]]) + self.putg([Ann.RX_START + rxtx, ['Start bit', 'Start', 'S']]) - self.putp(['STARTBIT', 0, self.startbit]) - if self.options['anno_startstop'] == 'yes': - self.putg([1, ['Start bit', 'Start', 'S']]) + self.state[rxtx] = 'GET DATA BITS' - self.state = 'GET DATA BITS' + def handle_packet(self, rxtx): + d = 'rx' if (rxtx == RX) else 'tx' + delim = self.options[d + '_packet_delim'] + plen = self.options[d + '_packet_len'] + if delim == -1 and plen == -1: + return - def get_data_bits(self, signal): + # Cache data values until we see the delimiter and/or the specified + # packet length has been reached (whichever happens first). + if len(self.packet_cache[rxtx]) == 0: + self.ss_packet[rxtx] = self.startsample[rxtx] + self.packet_cache[rxtx].append(self.datavalue[rxtx]) + if self.datavalue[rxtx] == delim or len(self.packet_cache[rxtx]) == plen: + self.es_packet[rxtx] = self.samplenum + s = '' + for b in self.packet_cache[rxtx]: + s += self.format_value(b) + if self.options['format'] != 'ascii': + s += ' ' + if self.options['format'] != 'ascii' and s[-1] == ' ': + s = s[:-1] # Drop trailing space. + self.putx_packet(rxtx, [Ann.RX_PACKET + rxtx, [s]]) + self.packet_cache[rxtx] = [] + + def get_data_bits(self, rxtx, signal): # Save the sample number of the middle of the first data bit. - if self.startsample == -1: - self.startsample = self.samplenum + if self.startsample[rxtx] == -1: + self.startsample[rxtx] = self.samplenum - self.putg([6, ['%d' % signal]]) + self.putg([Ann.RX_DATA_BIT + rxtx, ['%d' % signal]]) # Store individual data bits and their start/end samplenumbers. s, halfbit = self.samplenum, int(self.bit_width / 2) - self.databits.append([signal, s - halfbit, s + halfbit]) + self.databits[rxtx].append([signal, s - halfbit, s + halfbit]) # Return here, unless we already received all data bits. - self.cur_data_bit += 1 - if self.cur_data_bit < self.options['num_data_bits']: + self.cur_data_bit[rxtx] += 1 + if self.cur_data_bit[rxtx] < self.options['data_bits']: return # Convert accumulated data bits to a data value. - bits = [b[0] for b in self.databits] + bits = [b[0] for b in self.databits[rxtx]] if self.options['bit_order'] == 'msb-first': bits.reverse() - self.datavalue = bitpack(bits) - self.putpx(['DATA', 0, (self.datavalue, self.databits)]) + self.datavalue[rxtx] = bitpack(bits) + self.putpx(rxtx, ['DATA', rxtx, + (self.datavalue[rxtx], self.databits[rxtx])]) - self.putx([0, ['@%02X' % self.datavalue]]) + b = self.datavalue[rxtx] + formatted = self.format_value(b) + if formatted is not None: + self.putx(rxtx, [rxtx, [formatted]]) - b = self.datavalue - bdata = b.to_bytes(self.bw, byteorder='big') - self.putbin([0, bdata]) - self.putbin([1, bdata]) + self.putbin(rxtx, [Bin.RX + rxtx, bdata]) + self.putbin(rxtx, [Bin.RXTX, bdata]) - self.databits = [] + self.handle_packet(rxtx) + + self.databits[rxtx] = [] # Advance to either reception of the parity bit, or reception of # the STOP bits if parity is not applicable. - self.state = 'GET PARITY BIT' - if self.options['parity_type'] == 'none': - self.state = 'GET STOP BITS' - - - def get_parity_bit(self, signal): - self.paritybit = signal - - if parity_ok(self.options['parity_type'], self.paritybit, - self.datavalue, self.options['num_data_bits']): - self.putp(['PARITYBIT', 0, self.paritybit]) - self.putg([2, ['Parity bit', 'Parity', 'P']]) + self.state[rxtx] = 'GET PARITY BIT' + if self.options['parity'] == 'none': + self.state[rxtx] = 'GET STOP BITS' + + def format_value(self, v): + # Format value 'v' according to configured options. + # Reflects the user selected kind of representation, as well as + # the number of data bits in the UART frames. + + fmt, bits = self.options['format'], self.options['data_bits'] + + # Assume "is printable" for values from 32 to including 126, + # below 32 is "control" and thus not printable, above 127 is + # "not ASCII" in its strict sense, 127 (DEL) is not printable, + # fall back to hex representation for non-printables. + if fmt == 'ascii': + if v in range(32, 126 + 1): + return chr(v) + hexfmt = "[{:02X}]" if bits <= 8 else "[{:03X}]" + return hexfmt.format(v) + + # Mere number to text conversion without prefix and padding + # for the "decimal" output format. + if fmt == 'dec': + return "{:d}".format(v) + + # Padding with leading zeroes for hex/oct/bin formats, but + # without a prefix for density -- since the format is user + # specified, there is no ambiguity. + if fmt == 'hex': + digits = (bits + 4 - 1) // 4 + fmtchar = "X" + elif fmt == 'oct': + digits = (bits + 3 - 1) // 3 + fmtchar = "o" + elif fmt == 'bin': + digits = bits + fmtchar = "b" + else: + fmtchar = None + if fmtchar is not None: + fmt = "{{:0{:d}{:s}}}".format(digits, fmtchar) + return fmt.format(v) + + return None + + def get_parity_bit(self, rxtx, signal): + self.paritybit[rxtx] = signal + + if parity_ok(self.options['parity'], self.paritybit[rxtx], + self.datavalue[rxtx], self.options['data_bits']): + self.putp(['PARITYBIT', rxtx, self.paritybit[rxtx]]) + self.putg([Ann.RX_PARITY_OK + rxtx, ['Parity bit', 'Parity', 'P']]) else: # TODO: Return expected/actual parity values. - self.putp(['PARITY ERROR', 0, (0, 1)]) # FIXME: Dummy tuple... - self.putg([3, ['Parity error', 'Parity err', 'PE']]) - self.frame_valid = False + self.putp(['PARITY ERROR', rxtx, (0, 1)]) # FIXME: Dummy tuple... + self.putg([Ann.RX_PARITY_ERR + rxtx, ['Parity error', 'Parity err', 'PE']]) + self.frame_valid[rxtx] = False - self.state = 'GET STOP BITS' + self.state[rxtx] = 'GET STOP BITS' # TODO: Currently only supports 1 stop bit. - def get_stop_bits(self, signal): - self.stopbit1 = signal + def get_stop_bits(self, rxtx, signal): + self.stopbit1[rxtx] = signal # Stop bits must be 1. If not, we report an error. - if self.stopbit1 != 1: - self.putp(['INVALID STOPBIT', 0, self.stopbit1]) - self.putg([5, ['Frame error', 'Frame err', 'FE']]) - self.frame_valid = False + if self.stopbit1[rxtx] != 1: + self.putp(['INVALID STOPBIT', rxtx, self.stopbit1[rxtx]]) + self.putg([Ann.RX_WARN + rxtx, ['Frame error', 'Frame err', 'FE']]) + self.frame_valid[rxtx] = False - self.putp(['STOPBIT', 0, self.stopbit1]) - if self.options['anno_startstop'] == 'yes': - self.putg([2, ['Stop bit', 'Stop', 'T']]) + self.putp(['STOPBIT', rxtx, self.stopbit1[rxtx]]) + self.putg([Ann.RX_STOP + rxtx, ['Stop bit', 'Stop', 'T']]) # Pass the complete UART frame to upper layers. es = self.samplenum + ceil(self.bit_width / 2.0) - self.putpse(self.frame_start, es, ['FRAME', 0, - (self.datavalue, self.frame_valid)]) + self.putpse(self.frame_start[rxtx], es, ['FRAME', rxtx, + (self.datavalue[rxtx], self.frame_valid[rxtx])]) - self.state = 'WAIT FOR START BIT' + self.state[rxtx] = 'WAIT FOR START BIT' + self.idle_start[rxtx] = self.frame_start[rxtx] + self.frame_len_sample_count - def handle_break(self): - self.putpse(self.frame_start, self.samplenum, - ['BREAK', 0, 0]) - self.putgse(self.frame_start, self.samplenum, - [7, ['Break condition', 'Break', 'Brk', 'B']]) - self.state = 'WAIT FOR START BIT' + def handle_break(self, rxtx): + self.putpse(self.frame_start[rxtx], self.samplenum, + ['BREAK', rxtx, 0]) + self.putgse(self.frame_start[rxtx], self.samplenum, + [Ann.RX_BREAK + rxtx, ['Break condition', 'Break', 'Brk', 'B']]) + self.state[rxtx] = 'WAIT FOR START BIT' - def get_wait_cond(self, inv): + def get_wait_cond(self, rxtx, inv): # Return condititions that are suitable for Decoder.wait(). Those # conditions either match the falling edge of the START bit, or # the sample point of the next bit time. - state = self.state + state = self.state[rxtx] if state == 'WAIT FOR START BIT': - return {0: 'r' if inv else 'f'} + return {rxtx: 'r' if inv else 'f'} if state == 'GET START BIT': bitnum = 0 elif state == 'GET DATA BITS': - bitnum = 1 + self.cur_data_bit + bitnum = 1 + self.cur_data_bit[rxtx] elif state == 'GET PARITY BIT': - bitnum = 1 + self.options['num_data_bits'] + bitnum = 1 + self.options['data_bits'] elif state == 'GET STOP BITS': - bitnum = 1 + self.options['num_data_bits'] - bitnum += 0 if self.options['parity_type'] == 'none' else 1 - want_num = ceil(self.get_sample_point(bitnum)) + bitnum = 1 + self.options['data_bits'] + bitnum += 0 if self.options['parity'] == 'none' else 1 + want_num = ceil(self.get_sample_point(rxtx, bitnum)) return {'skip': want_num - self.samplenum} - def inspect_sample(self, signal, inv): + def get_idle_cond(self, rxtx, inv): + # Return a condition that corresponds to the (expected) end of + # the next frame, assuming that it will be an "idle frame" + # (constant high input level for the frame's length). + if self.idle_start[rxtx] is None: + return None + end_of_frame = self.idle_start[rxtx] + self.frame_len_sample_count + if end_of_frame < self.samplenum: + return None + return {'skip': end_of_frame - self.samplenum} + + def inspect_sample(self, rxtx, signal, inv): # Inspect a sample returned by .wait() for the specified UART line. if inv: signal = not signal - state = self.state + state = self.state[rxtx] if state == 'WAIT FOR START BIT': - self.wait_for_start_bit(signal) + self.wait_for_start_bit(rxtx, signal) elif state == 'GET START BIT': - self.get_start_bit(signal) + self.get_start_bit(rxtx, signal) elif state == 'GET DATA BITS': - self.get_data_bits(signal) + self.get_data_bits(rxtx, signal) elif state == 'GET PARITY BIT': - self.get_parity_bit(signal) + self.get_parity_bit(rxtx, signal) elif state == 'GET STOP BITS': - self.get_stop_bits(signal) + self.get_stop_bits(rxtx, signal) - def inspect_edge(self, signal, inv): + def inspect_edge(self, rxtx, signal, inv): # Inspect edges, independently from traffic, to detect break conditions. if inv: signal = not signal if not signal: # Signal went low. Start another interval. - self.break_start = self.samplenum + self.break_start[rxtx] = self.samplenum return # Signal went high. Was there an extended period with low signal? - if self.break_start is None: + if self.break_start[rxtx] is None: return - diff = self.samplenum - self.break_start + diff = self.samplenum - self.break_start[rxtx] if diff >= self.break_min_sample_count: - self.handle_break() - self.break_start = None + self.handle_break(rxtx) + self.break_start[rxtx] = None + + def inspect_idle(self, rxtx, signal, inv): + # Check each edge and each period of stable input (either level). + # Can derive the "idle frame period has passed" condition. + if inv: + signal = not signal + if not signal: + # Low input, cease inspection. + self.idle_start[rxtx] = None + return + # High input, either just reached, or still stable. + if self.idle_start[rxtx] is None: + self.idle_start[rxtx] = self.samplenum + diff = self.samplenum - self.idle_start[rxtx] + if diff < self.frame_len_sample_count: + return + ss, es = self.idle_start[rxtx], self.samplenum + self.putpse(ss, es, ['IDLE', rxtx, 0]) + self.idle_start[rxtx] = self.samplenum def decode(self): if not self.samplerate: raise SamplerateError('Cannot decode without samplerate.') - inv = self.options['invert'] == 'yes' - cond_data_idx = None + has_pin = [self.has_channel(ch) for ch in (RX, TX)] + if not True in has_pin: + raise ChannelError('Need at least one of TX or RX pins.') + + opt = self.options + inv = [opt['invert_rx'] == 'yes', opt['invert_tx'] == 'yes'] + cond_data_idx = [None] * len(has_pin) # Determine the number of samples for a complete frame's time span. # A period of low signal (at least) that long is a break condition. frame_samples = 1 # START - frame_samples += self.options['num_data_bits'] - frame_samples += 0 if self.options['parity_type'] == 'none' else 1 - frame_samples += self.options['num_stop_bits'] + frame_samples += self.options['data_bits'] + frame_samples += 0 if self.options['parity'] == 'none' else 1 + frame_samples += self.options['stop_bits'] frame_samples *= self.bit_width - self.break_min_sample_count = ceil(frame_samples) - cond_edge_idx = None + self.frame_len_sample_count = ceil(frame_samples) + self.break_min_sample_count = self.frame_len_sample_count + cond_edge_idx = [None] * len(has_pin) + cond_idle_idx = [None] * len(has_pin) while True: conds = [] - - cond_data_idx = len(conds) - conds.append(self.get_wait_cond(inv)) - cond_edge_idx = len(conds) - conds.append({0: 'e'}) - - (rxtx, ) = self.wait(conds) - if cond_data_idx is not None and (self.matched & (0b1 << cond_data_idx)): - self.inspect_sample(rxtx, inv) - if cond_edge_idx is not None and (self.matched & (0b1 << cond_edge_idx)): - self.inspect_edge(rxtx, inv) + if has_pin[RX]: + cond_data_idx[RX] = len(conds) + conds.append(self.get_wait_cond(RX, inv[RX])) + cond_edge_idx[RX] = len(conds) + conds.append({RX: 'e'}) + cond_idle_idx[RX] = None + idle_cond = self.get_idle_cond(RX, inv[RX]) + if idle_cond: + cond_idle_idx[RX] = len(conds) + conds.append(idle_cond) + if has_pin[TX]: + cond_data_idx[TX] = len(conds) + conds.append(self.get_wait_cond(TX, inv[TX])) + cond_edge_idx[TX] = len(conds) + conds.append({TX: 'e'}) + cond_idle_idx[TX] = None + idle_cond = self.get_idle_cond(TX, inv[TX]) + if idle_cond: + cond_idle_idx[TX] = len(conds) + conds.append(idle_cond) + (rx, tx) = self.wait(conds) + + if cond_data_idx[RX] is not None and (self.matched & (0b1 << cond_data_idx[RX])): + self.inspect_sample(RX, rx, inv[RX]) + if cond_edge_idx[RX] is not None and (self.matched & (0b1 << cond_edge_idx[RX])): + self.inspect_edge(RX, rx, inv[RX]) + self.inspect_idle(RX, rx, inv[RX]) + if cond_idle_idx[RX] is not None and (self.matched & (0b1 << cond_idle_idx[RX])): + self.inspect_idle(RX, rx, inv[RX]) + if cond_data_idx[TX] is not None and (self.matched & (0b1 << cond_data_idx[TX])): + self.inspect_sample(TX, tx, inv[TX]) + if cond_edge_idx[TX] is not None and (self.matched & (0b1 << cond_edge_idx[TX])): + self.inspect_edge(TX, tx, inv[TX]) + self.inspect_idle(TX, tx, inv[TX]) + if cond_idle_idx[TX] is not None and (self.matched & (0b1 << cond_idle_idx[TX])): + self.inspect_idle(TX, tx, inv[TX]) diff --git a/libsigrokdecode4DSL/decoders/uart/__init__.py b/libsigrokdecode4DSL/decoders/uart/__init__.py deleted file mode 100644 index ce6136f1..00000000 --- a/libsigrokdecode4DSL/decoders/uart/__init__.py +++ /dev/null @@ -1,40 +0,0 @@ -## -## This file is part of the libsigrokdecode project. -## -## Copyright (C) 2012 Uwe Hermann -## -## This program is free software; you can redistribute it and/or modify -## it under the terms of the GNU General Public License as published by -## the Free Software Foundation; either version 2 of the License, or -## (at your option) any later version. -## -## This program is distributed in the hope that it will be useful, -## but WITHOUT ANY WARRANTY; without even the implied warranty of -## MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the -## GNU General Public License for more details. -## -## You should have received a copy of the GNU General Public License -## along with this program; if not, see . -## - -''' -UART (Universal Asynchronous Receiver Transmitter) is a simple serial -communication protocol which allows two devices to talk to each other. - -This decoder should work on all "UART-like" async protocols with one -start bit (0), 5-9 databits, an (optional) parity bit, and one or more -stop bits (1), in this order. - -It can be run on one signal line (RX or TX) only, or on two lines (RX + TX). - -There are various standards for the physical layer specification of the -signals, including RS232, (TTL) UART, RS485, and others. However, the logic -level of the respective pins is only relevant when acquiring the data via -a logic analyzer (you have to select the correct logic analyzer and/or -the correct place where to probe). Once the data is in digital form and -matches the "UART" description above, this protocol decoder can work with -it though, no matter whether the source was on TTL UART levels, or RS232, -or others. -''' - -from .pd import Decoder diff --git a/libsigrokdecode4DSL/decoders/uart/pd.py b/libsigrokdecode4DSL/decoders/uart/pd.py deleted file mode 100644 index 457e14b3..00000000 --- a/libsigrokdecode4DSL/decoders/uart/pd.py +++ /dev/null @@ -1,576 +0,0 @@ -## -## This file is part of the libsigrokdecode project. -## -## Copyright (C) 2011-2014 Uwe Hermann -## -## This program is free software; you can redistribute it and/or modify -## it under the terms of the GNU General Public License as published by -## the Free Software Foundation; either version 2 of the License, or -## (at your option) any later version. -## -## This program is distributed in the hope that it will be useful, -## but WITHOUT ANY WARRANTY; without even the implied warranty of -## MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the -## GNU General Public License for more details. -## -## You should have received a copy of the GNU General Public License -## along with this program; if not, see . -## - -import sigrokdecode as srd -from common.srdhelper import bitpack -from math import floor, ceil - -''' -OUTPUT_PYTHON format: - -Packet: -[, , ] - -This is the list of s and their respective values: - - 'STARTBIT': The data is the (integer) value of the start bit (0/1). - - 'DATA': This is always a tuple containing two items: - - 1st item: the (integer) value of the UART data. Valid values - range from 0 to 511 (as the data can be up to 9 bits in size). - - 2nd item: the list of individual data bits and their ss/es numbers. - - 'PARITYBIT': The data is the (integer) value of the parity bit (0/1). - - 'STOPBIT': The data is the (integer) value of the stop bit (0 or 1). - - 'INVALID STARTBIT': The data is the (integer) value of the start bit (0/1). - - 'INVALID STOPBIT': The data is the (integer) value of the stop bit (0/1). - - '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. - - 'BREAK': The data is always 0. - - 'FRAME': The data is always a tuple containing two items: The (integer) - value of the UART data, and a boolean which reflects the validity of the - UART frame. - - 'IDLE': The data is always 0. - -The field is 0 for RX packets, 1 for TX packets. -''' - -# Used for differentiating between the two data directions. -RX = 0 -TX = 1 - -# 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, -# usually 8 bits) return True if the parity is correct, False otherwise. -# 'none' is _not_ allowed as value for 'parity_type'. -def parity_ok(parity_type, parity_bit, data, data_bits): - - if parity_type == 'ignore': - return True - - # Handle easy cases first (parity bit is always 1 or 0). - if parity_type == 'zero': - return parity_bit == 0 - elif parity_type == 'one': - return parity_bit == 1 - - # Count number of 1 (high) bits in the data (and the parity bit itself!). - ones = bin(data).count('1') + parity_bit - - # Check for odd/even parity. - if parity_type == 'odd': - return (ones % 2) == 1 - elif parity_type == 'even': - return (ones % 2) == 0 - -class SamplerateError(Exception): - pass - -class ChannelError(Exception): - pass - -class Ann: - RX_DATA, TX_DATA, RX_START, TX_START, RX_PARITY_OK, TX_PARITY_OK, \ - RX_PARITY_ERR, TX_PARITY_ERR, RX_STOP, TX_STOP, RX_WARN, TX_WARN, \ - RX_DATA_BIT, TX_DATA_BIT, RX_BREAK, TX_BREAK, RX_PACKET, TX_PACKET = \ - range(18) - -class Bin: - RX, TX, RXTX = range(3) - -class Decoder(srd.Decoder): - api_version = 3 - id = 'uart' - name = 'UART' - longname = 'Universal Asynchronous Receiver/Transmitter' - desc = 'Asynchronous, serial bus.' - license = 'gplv2+' - inputs = ['logic'] - outputs = ['uart'] - tags = ['Embedded/industrial'] - optional_channels = ( - # Allow specifying only one of the signals, e.g. if only one data - # direction exists (or is relevant). - {'id': 'rx', 'name': 'RX', 'desc': 'UART receive line'}, - {'id': 'tx', 'name': 'TX', 'desc': 'UART transmit line'}, - ) - options = ( - {'id': 'baudrate', 'desc': 'Baud rate', 'default': 115200}, - {'id': 'data_bits', 'desc': 'Data bits', 'default': 8, - 'values': (5, 6, 7, 8, 9)}, - {'id': 'parity', 'desc': 'Parity', 'default': 'none', - 'values': ('none', 'odd', 'even', 'zero', 'one', 'ignore')}, - {'id': 'stop_bits', 'desc': 'Stop bits', 'default': 1.0, - 'values': (0.0, 0.5, 1.0, 1.5)}, - {'id': 'bit_order', 'desc': 'Bit order', 'default': 'lsb-first', - 'values': ('lsb-first', 'msb-first')}, - {'id': 'format', 'desc': 'Data format', 'default': 'hex', - 'values': ('ascii', 'dec', 'hex', 'oct', 'bin')}, - {'id': 'invert_rx', 'desc': 'Invert RX', 'default': 'no', - 'values': ('yes', 'no')}, - {'id': 'invert_tx', 'desc': 'Invert TX', 'default': 'no', - 'values': ('yes', 'no')}, - {'id': 'sample_point', 'desc': 'Sample point (%)', 'default': 50}, - {'id': 'rx_packet_delim', 'desc': 'RX packet delimiter (decimal)', - 'default': -1}, - {'id': 'tx_packet_delim', 'desc': 'TX packet delimiter (decimal)', - 'default': -1}, - {'id': 'rx_packet_len', 'desc': 'RX packet length', 'default': -1}, - {'id': 'tx_packet_len', 'desc': 'TX packet length', 'default': -1}, - ) - annotations = ( - ('rx-data', 'RX data'), - ('tx-data', 'TX data'), - ('rx-start', 'RX start bit'), - ('tx-start', 'TX start bit'), - ('rx-parity-ok', 'RX parity OK bit'), - ('tx-parity-ok', 'TX parity OK bit'), - ('rx-parity-err', 'RX parity error bit'), - ('tx-parity-err', 'TX parity error bit'), - ('rx-stop', 'RX stop bit'), - ('tx-stop', 'TX stop bit'), - ('rx-warning', 'RX warning'), - ('tx-warning', 'TX warning'), - ('rx-data-bit', 'RX data bit'), - ('tx-data-bit', 'TX data bit'), - ('rx-break', 'RX break'), - ('tx-break', 'TX break'), - ('rx-packet', 'RX packet'), - ('tx-packet', 'TX packet'), - ) - annotation_rows = ( - ('rx-data-bits', 'RX bits', (Ann.RX_DATA_BIT,)), - ('rx-data-vals', 'RX data', (Ann.RX_DATA, Ann.RX_START, Ann.RX_PARITY_OK, Ann.RX_PARITY_ERR, Ann.RX_STOP)), - ('rx-warnings', 'RX warnings', (Ann.RX_WARN,)), - ('rx-breaks', 'RX breaks', (Ann.RX_BREAK,)), - ('rx-packets', 'RX packets', (Ann.RX_PACKET,)), - ('tx-data-bits', 'TX bits', (Ann.TX_DATA_BIT,)), - ('tx-data-vals', 'TX data', (Ann.TX_DATA, Ann.TX_START, Ann.TX_PARITY_OK, Ann.TX_PARITY_ERR, Ann.TX_STOP)), - ('tx-warnings', 'TX warnings', (Ann.TX_WARN,)), - ('tx-breaks', 'TX breaks', (Ann.TX_BREAK,)), - ('tx-packets', 'TX packets', (Ann.TX_PACKET,)), - ) - binary = ( - ('rx', 'RX dump'), - ('tx', 'TX dump'), - ('rxtx', 'RX/TX dump'), - ) - idle_state = ['WAIT FOR START BIT', 'WAIT FOR START BIT'] - - def putx(self, rxtx, data): - s, halfbit = self.startsample[rxtx], self.bit_width / 2.0 - self.put(s - floor(halfbit), self.samplenum + ceil(halfbit), self.out_ann, data) - - def putx_packet(self, rxtx, data): - s, halfbit = self.ss_packet[rxtx], self.bit_width / 2.0 - self.put(s - floor(halfbit), self.samplenum + ceil(halfbit), self.out_ann, data) - - def putpx(self, rxtx, data): - s, halfbit = self.startsample[rxtx], self.bit_width / 2.0 - self.put(s - floor(halfbit), self.samplenum + ceil(halfbit), self.out_python, data) - - def putg(self, data): - s, halfbit = self.samplenum, self.bit_width / 2.0 - self.put(s - floor(halfbit), s + ceil(halfbit), self.out_ann, data) - - def putp(self, data): - s, halfbit = self.samplenum, self.bit_width / 2.0 - self.put(s - floor(halfbit), s + ceil(halfbit), self.out_python, data) - - def putgse(self, ss, es, data): - self.put(ss, es, self.out_ann, data) - - def putpse(self, ss, es, data): - self.put(ss, es, self.out_python, data) - - def putbin(self, rxtx, data): - s, halfbit = self.startsample[rxtx], self.bit_width / 2.0 - self.put(s - floor(halfbit), self.samplenum + ceil(halfbit), self.out_binary, data) - - def __init__(self): - self.reset() - - def reset(self): - self.samplerate = None - self.frame_start = [-1, -1] - self.frame_valid = [None, None] - self.startbit = [-1, -1] - self.cur_data_bit = [0, 0] - self.datavalue = [0, 0] - self.paritybit = [-1, -1] - self.stopbit1 = [-1, -1] - self.startsample = [-1, -1] - self.state = ['WAIT FOR START BIT', 'WAIT FOR START BIT'] - self.databits = [[], []] - self.break_start = [None, None] - self.packet_cache = [[], []] - self.ss_packet, self.es_packet = [None, None], [None, None] - self.idle_start = [None, None] - - def start(self): - self.out_python = self.register(srd.OUTPUT_PYTHON) - self.out_binary = self.register(srd.OUTPUT_BINARY) - self.out_ann = self.register(srd.OUTPUT_ANN) - self.bw = (self.options['data_bits'] + 7) // 8 - - def metadata(self, key, value): - if key == srd.SRD_CONF_SAMPLERATE: - self.samplerate = value - # The width of one UART bit in number of samples. - self.bit_width = float(self.samplerate) / float(self.options['baudrate']) - - def get_sample_point(self, rxtx, bitnum): - # Determine absolute sample number of a bit slot's sample point. - # Counts for UART bits start from 0 (0 = start bit, 1..x = data, - # x+1 = parity bit (if used) or the first stop bit, and so on). - # Accept a position in the range of 1-99% of the full bit width. - # Assume 50% for invalid input specs for backwards compatibility. - perc = self.options['sample_point'] or 50 - if not perc or perc not in range(1, 100): - perc = 50 - perc /= 100.0 - bitpos = (self.bit_width - 1) * perc - bitpos += self.frame_start[rxtx] - bitpos += bitnum * self.bit_width - return bitpos - - def wait_for_start_bit(self, rxtx, signal): - # Save the sample number where the start bit begins. - self.frame_start[rxtx] = self.samplenum - self.frame_valid[rxtx] = True - - self.state[rxtx] = 'GET START BIT' - - def get_start_bit(self, rxtx, signal): - self.startbit[rxtx] = signal - - # The startbit must be 0. If not, we report an error and wait - # for the next start bit (assuming this one was spurious). - if self.startbit[rxtx] != 0: - self.putp(['INVALID STARTBIT', rxtx, self.startbit[rxtx]]) - self.putg([Ann.RX_WARN + rxtx, ['Frame error', 'Frame err', 'FE']]) - self.frame_valid[rxtx] = False - es = self.samplenum + ceil(self.bit_width / 2.0) - self.putpse(self.frame_start[rxtx], es, ['FRAME', rxtx, - (self.datavalue[rxtx], self.frame_valid[rxtx])]) - self.state[rxtx] = 'WAIT FOR START BIT' - return - - self.cur_data_bit[rxtx] = 0 - self.datavalue[rxtx] = 0 - self.startsample[rxtx] = -1 - - self.putp(['STARTBIT', rxtx, self.startbit[rxtx]]) - self.putg([Ann.RX_START + rxtx, ['Start bit', 'Start', 'S']]) - - self.state[rxtx] = 'GET DATA BITS' - - def handle_packet(self, rxtx): - d = 'rx' if (rxtx == RX) else 'tx' - delim = self.options[d + '_packet_delim'] - plen = self.options[d + '_packet_len'] - if delim == -1 and plen == -1: - return - - # Cache data values until we see the delimiter and/or the specified - # packet length has been reached (whichever happens first). - if len(self.packet_cache[rxtx]) == 0: - self.ss_packet[rxtx] = self.startsample[rxtx] - self.packet_cache[rxtx].append(self.datavalue[rxtx]) - if self.datavalue[rxtx] == delim or len(self.packet_cache[rxtx]) == plen: - self.es_packet[rxtx] = self.samplenum - s = '' - for b in self.packet_cache[rxtx]: - s += self.format_value(b) - if self.options['format'] != 'ascii': - s += ' ' - if self.options['format'] != 'ascii' and s[-1] == ' ': - s = s[:-1] # Drop trailing space. - self.putx_packet(rxtx, [Ann.RX_PACKET + rxtx, [s]]) - self.packet_cache[rxtx] = [] - - def get_data_bits(self, rxtx, signal): - # Save the sample number of the middle of the first data bit. - if self.startsample[rxtx] == -1: - self.startsample[rxtx] = self.samplenum - - self.putg([Ann.RX_DATA_BIT + rxtx, ['%d' % signal]]) - - # Store individual data bits and their start/end samplenumbers. - s, halfbit = self.samplenum, int(self.bit_width / 2) - self.databits[rxtx].append([signal, s - halfbit, s + halfbit]) - - # Return here, unless we already received all data bits. - self.cur_data_bit[rxtx] += 1 - if self.cur_data_bit[rxtx] < self.options['data_bits']: - return - - # Convert accumulated data bits to a data value. - bits = [b[0] for b in self.databits[rxtx]] - if self.options['bit_order'] == 'msb-first': - bits.reverse() - self.datavalue[rxtx] = bitpack(bits) - self.putpx(rxtx, ['DATA', rxtx, - (self.datavalue[rxtx], self.databits[rxtx])]) - - b = self.datavalue[rxtx] - formatted = self.format_value(b) - if formatted is not None: - self.putx(rxtx, [rxtx, [formatted]]) - - bdata = b.to_bytes(self.bw, byteorder='big') - self.putbin(rxtx, [Bin.RX + rxtx, bdata]) - self.putbin(rxtx, [Bin.RXTX, bdata]) - - self.handle_packet(rxtx) - - self.databits[rxtx] = [] - - # Advance to either reception of the parity bit, or reception of - # the STOP bits if parity is not applicable. - self.state[rxtx] = 'GET PARITY BIT' - if self.options['parity'] == 'none': - self.state[rxtx] = 'GET STOP BITS' - - def format_value(self, v): - # Format value 'v' according to configured options. - # Reflects the user selected kind of representation, as well as - # the number of data bits in the UART frames. - - fmt, bits = self.options['format'], self.options['data_bits'] - - # Assume "is printable" for values from 32 to including 126, - # below 32 is "control" and thus not printable, above 127 is - # "not ASCII" in its strict sense, 127 (DEL) is not printable, - # fall back to hex representation for non-printables. - if fmt == 'ascii': - if v in range(32, 126 + 1): - return chr(v) - hexfmt = "[{:02X}]" if bits <= 8 else "[{:03X}]" - return hexfmt.format(v) - - # Mere number to text conversion without prefix and padding - # for the "decimal" output format. - if fmt == 'dec': - return "{:d}".format(v) - - # Padding with leading zeroes for hex/oct/bin formats, but - # without a prefix for density -- since the format is user - # specified, there is no ambiguity. - if fmt == 'hex': - digits = (bits + 4 - 1) // 4 - fmtchar = "X" - elif fmt == 'oct': - digits = (bits + 3 - 1) // 3 - fmtchar = "o" - elif fmt == 'bin': - digits = bits - fmtchar = "b" - else: - fmtchar = None - if fmtchar is not None: - fmt = "{{:0{:d}{:s}}}".format(digits, fmtchar) - return fmt.format(v) - - return None - - def get_parity_bit(self, rxtx, signal): - self.paritybit[rxtx] = signal - - if parity_ok(self.options['parity'], self.paritybit[rxtx], - self.datavalue[rxtx], self.options['data_bits']): - self.putp(['PARITYBIT', rxtx, self.paritybit[rxtx]]) - self.putg([Ann.RX_PARITY_OK + rxtx, ['Parity bit', 'Parity', 'P']]) - else: - # TODO: Return expected/actual parity values. - self.putp(['PARITY ERROR', rxtx, (0, 1)]) # FIXME: Dummy tuple... - self.putg([Ann.RX_PARITY_ERR + rxtx, ['Parity error', 'Parity err', 'PE']]) - self.frame_valid[rxtx] = False - - self.state[rxtx] = 'GET STOP BITS' - - # TODO: Currently only supports 1 stop bit. - def get_stop_bits(self, rxtx, signal): - self.stopbit1[rxtx] = signal - - # Stop bits must be 1. If not, we report an error. - if self.stopbit1[rxtx] != 1: - self.putp(['INVALID STOPBIT', rxtx, self.stopbit1[rxtx]]) - self.putg([Ann.RX_WARN + rxtx, ['Frame error', 'Frame err', 'FE']]) - self.frame_valid[rxtx] = False - - self.putp(['STOPBIT', rxtx, self.stopbit1[rxtx]]) - self.putg([Ann.RX_STOP + rxtx, ['Stop bit', 'Stop', 'T']]) - - # Pass the complete UART frame to upper layers. - es = self.samplenum + ceil(self.bit_width / 2.0) - self.putpse(self.frame_start[rxtx], es, ['FRAME', rxtx, - (self.datavalue[rxtx], self.frame_valid[rxtx])]) - - self.state[rxtx] = 'WAIT FOR START BIT' - self.idle_start[rxtx] = self.frame_start[rxtx] + self.frame_len_sample_count - - def handle_break(self, rxtx): - self.putpse(self.frame_start[rxtx], self.samplenum, - ['BREAK', rxtx, 0]) - self.putgse(self.frame_start[rxtx], self.samplenum, - [Ann.RX_BREAK + rxtx, ['Break condition', 'Break', 'Brk', 'B']]) - self.state[rxtx] = 'WAIT FOR START BIT' - - def get_wait_cond(self, rxtx, inv): - # Return condititions that are suitable for Decoder.wait(). Those - # conditions either match the falling edge of the START bit, or - # the sample point of the next bit time. - state = self.state[rxtx] - if state == 'WAIT FOR START BIT': - return {rxtx: 'r' if inv else 'f'} - if state == 'GET START BIT': - bitnum = 0 - elif state == 'GET DATA BITS': - bitnum = 1 + self.cur_data_bit[rxtx] - elif state == 'GET PARITY BIT': - bitnum = 1 + self.options['data_bits'] - elif state == 'GET STOP BITS': - bitnum = 1 + self.options['data_bits'] - bitnum += 0 if self.options['parity'] == 'none' else 1 - want_num = ceil(self.get_sample_point(rxtx, bitnum)) - return {'skip': want_num - self.samplenum} - - def get_idle_cond(self, rxtx, inv): - # Return a condition that corresponds to the (expected) end of - # the next frame, assuming that it will be an "idle frame" - # (constant high input level for the frame's length). - if self.idle_start[rxtx] is None: - return None - end_of_frame = self.idle_start[rxtx] + self.frame_len_sample_count - if end_of_frame < self.samplenum: - return None - return {'skip': end_of_frame - self.samplenum} - - def inspect_sample(self, rxtx, signal, inv): - # Inspect a sample returned by .wait() for the specified UART line. - if inv: - signal = not signal - - state = self.state[rxtx] - if state == 'WAIT FOR START BIT': - self.wait_for_start_bit(rxtx, signal) - elif state == 'GET START BIT': - self.get_start_bit(rxtx, signal) - elif state == 'GET DATA BITS': - self.get_data_bits(rxtx, signal) - elif state == 'GET PARITY BIT': - self.get_parity_bit(rxtx, signal) - elif state == 'GET STOP BITS': - self.get_stop_bits(rxtx, signal) - - def inspect_edge(self, rxtx, signal, inv): - # Inspect edges, independently from traffic, to detect break conditions. - if inv: - signal = not signal - if not signal: - # Signal went low. Start another interval. - self.break_start[rxtx] = self.samplenum - return - # Signal went high. Was there an extended period with low signal? - if self.break_start[rxtx] is None: - return - diff = self.samplenum - self.break_start[rxtx] - if diff >= self.break_min_sample_count: - self.handle_break(rxtx) - self.break_start[rxtx] = None - - def inspect_idle(self, rxtx, signal, inv): - # Check each edge and each period of stable input (either level). - # Can derive the "idle frame period has passed" condition. - if inv: - signal = not signal - if not signal: - # Low input, cease inspection. - self.idle_start[rxtx] = None - return - # High input, either just reached, or still stable. - if self.idle_start[rxtx] is None: - self.idle_start[rxtx] = self.samplenum - diff = self.samplenum - self.idle_start[rxtx] - if diff < self.frame_len_sample_count: - return - ss, es = self.idle_start[rxtx], self.samplenum - self.putpse(ss, es, ['IDLE', rxtx, 0]) - self.idle_start[rxtx] = self.samplenum - - def decode(self): - if not self.samplerate: - raise SamplerateError('Cannot decode without samplerate.') - - has_pin = [self.has_channel(ch) for ch in (RX, TX)] - if not True in has_pin: - raise ChannelError('Need at least one of TX or RX pins.') - - opt = self.options - inv = [opt['invert_rx'] == 'yes', opt['invert_tx'] == 'yes'] - cond_data_idx = [None] * len(has_pin) - - # Determine the number of samples for a complete frame's time span. - # A period of low signal (at least) that long is a break condition. - frame_samples = 1 # START - frame_samples += self.options['data_bits'] - frame_samples += 0 if self.options['parity'] == 'none' else 1 - frame_samples += self.options['stop_bits'] - frame_samples *= self.bit_width - self.frame_len_sample_count = ceil(frame_samples) - self.break_min_sample_count = self.frame_len_sample_count - cond_edge_idx = [None] * len(has_pin) - cond_idle_idx = [None] * len(has_pin) - - while True: - conds = [] - if has_pin[RX]: - cond_data_idx[RX] = len(conds) - conds.append(self.get_wait_cond(RX, inv[RX])) - cond_edge_idx[RX] = len(conds) - conds.append({RX: 'e'}) - cond_idle_idx[RX] = None - idle_cond = self.get_idle_cond(RX, inv[RX]) - if idle_cond: - cond_idle_idx[RX] = len(conds) - conds.append(idle_cond) - if has_pin[TX]: - cond_data_idx[TX] = len(conds) - conds.append(self.get_wait_cond(TX, inv[TX])) - cond_edge_idx[TX] = len(conds) - conds.append({TX: 'e'}) - cond_idle_idx[TX] = None - idle_cond = self.get_idle_cond(TX, inv[TX]) - if idle_cond: - cond_idle_idx[TX] = len(conds) - conds.append(idle_cond) - (rx, tx) = self.wait(conds) - - if cond_data_idx[RX] is not None and (self.matched & (0b1 << cond_data_idx[RX])): - self.inspect_sample(RX, rx, inv[RX]) - if cond_edge_idx[RX] is not None and (self.matched & (0b1 << cond_edge_idx[RX])): - self.inspect_edge(RX, rx, inv[RX]) - self.inspect_idle(RX, rx, inv[RX]) - if cond_idle_idx[RX] is not None and (self.matched & (0b1 << cond_idle_idx[RX])): - self.inspect_idle(RX, rx, inv[RX]) - if cond_data_idx[TX] is not None and (self.matched & (0b1 << cond_data_idx[TX])): - self.inspect_sample(TX, tx, inv[TX]) - if cond_edge_idx[TX] is not None and (self.matched & (0b1 << cond_edge_idx[TX])): - self.inspect_edge(TX, tx, inv[TX]) - self.inspect_idle(TX, tx, inv[TX]) - if cond_idle_idx[TX] is not None and (self.matched & (0b1 << cond_idle_idx[TX])): - self.inspect_idle(TX, tx, inv[TX])