+##
+## This file is part of the libsigrokdecode project.
+##
+## Copyright 2015 Google, Inc
+##
+## 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.
+##
+
+import sigrokdecode as srd
+import struct
+import zlib # for crc32
+
+# BMC encoding with a 600Khz datarate
+UI_US = 1000000/600000.0
+
+# Threshold to discriminate half-1 from 0 in Binary Mark Conding
+THRESHOLD_US = (UI_US + 2 * UI_US) / 2
+
+# Control Message type
+CTRL_TYPES = {
+ 0: "reserved",
+ 1: "GOOD CRC",
+ 2: "GOTO MIN",
+ 3: "ACCEPT",
+ 4: "REJECT",
+ 5: "PING",
+ 6: "PS RDY",
+ 7: "GET SOURCE CAP",
+ 8: "GET SINK CAP",
+ 9: "DR SWAP",
+ 10: "PR SWAP",
+ 11: "VCONN SWAP",
+ 12: "WAIT",
+ 13: "SOFT RESET",
+ 14: "reserved",
+ 15: "reserved"
+}
+
+# Data message type
+DATA_TYPES = {
+ 1: "SOURCE CAP",
+ 2: "REQUEST",
+ 3: "BIST",
+ 4: "SINK CAP",
+ 15: "VDM"
+}
+
+# 4b5b encoding of the symbols
+DEC4B5B = [
+ 0x10, # Error 00000
+ 0x10, # Error 00001
+ 0x10, # Error 00010
+ 0x10, # Error 00011
+ 0x10, # Error 00100
+ 0x10, # Error 00101
+ 0x13, # Sync-3 00110
+ 0x14, # RST-1 00111
+ 0x10, # Error 01000
+ 0x01, # 1 = 0001 01001
+ 0x04, # 4 = 0100 01010
+ 0x05, # 5 = 0101 01011
+ 0x10, # Error 01100
+ 0x16, # EOP 01101
+ 0x06, # 6 = 0110 01110
+ 0x07, # 7 = 0111 01111
+ 0x10, # Error 10000
+ 0x12, # Sync-2 10001
+ 0x08, # 8 = 1000 10010
+ 0x09, # 9 = 1001 10011
+ 0x02, # 2 = 0010 10100
+ 0x03, # 3 = 0011 10101
+ 0x0A, # A = 1010 10110
+ 0x0B, # B = 1011 10111
+ 0x11, # Sync-1 11000
+ 0x15, # RST-2 11001
+ 0x0C, # C = 1100 11010
+ 0x0D, # D = 1101 11011
+ 0x0E, # E = 1110 11100
+ 0x0F, # F = 1111 11101
+ 0x00, # 0 = 0000 11110
+ 0x10, # Error 11111
+]
+SYM_ERR = 0x10
+SYNC1 = 0x11
+SYNC2 = 0x12
+SYNC3 = 0x13
+RST1 = 0x14
+RST2 = 0x15
+EOP = 0x16
+SYNC_CODES = [SYNC1, SYNC2, SYNC3]
+HRST_CODES = [RST1, RST1, RST1, RST2]
+
+START_OF_PACKETS = {
+ (SYNC1, SYNC1, SYNC1, SYNC2): "SOP",
+ (SYNC1, SYNC1, SYNC3, SYNC3): "SOP'",
+ (SYNC1, SYNC3, SYNC1, SYNC3): 'SOP"',
+ (SYNC1, RST2, RST2, SYNC3): "SOP' Debug",
+ (SYNC1, RST2, SYNC3, SYNC2): 'SOP" Debug',
+ (RST1, SYNC1, RST1, SYNC3): "Cable Reset",
+ (RST1, RST1, RST1, RST2): "Hard Reset",
+}
+
+SYM_NAME = [
+ ['0x0', '0'],
+ ['0x1', '1'],
+ ['0x2', '2'],
+ ['0x3', '3'],
+ ['0x4', '4'],
+ ['0x5', '5'],
+ ['0x6', '6'],
+ ['0x7', '7'],
+ ['0x8', '8'],
+ ['0x9', '9'],
+ ['0xA', 'A'],
+ ['0xB', 'B'],
+ ['0xC', 'C'],
+ ['0xD', 'D'],
+ ['0xE', 'E'],
+ ['0xF', 'F'],
+ ['ERROR', 'X'],
+ ['SYNC-1', 'S1'],
+ ['SYNC-2', 'S2'],
+ ['SYNC-3', 'S3'],
+ ['RST-1', 'R1'],
+ ['RST-2', 'R2'],
+ ['EOP', '#'],
+]
+
+RDO_FLAGS = {
+ (1 << 24): "no_suspend",
+ (1 << 25): "comm_cap",
+ (1 << 26): "cap_mismatch",
+ (1 << 27): "give_back"
+}
+PDO_TYPE = ["", "BATT:", "VAR:", "<bad>"]
+PDO_FLAGS = {
+ (1 << 29): "dual_role_power",
+ (1 << 28): "suspend",
+ (1 << 27): "ext",
+ (1 << 26): "comm_cap",
+ (1 << 25): "dual_role_data"
+}
+
+BIST_MODES = {
+ 0: "Receiver",
+ 1: "Transmit",
+ 2: "Counters",
+ 3: "Carrier 0",
+ 4: "Carrier 1",
+ 5: "Carrier 2",
+ 6: "Carrier 3",
+ 7: "Eye",
+}
+
+VDM_CMDS = {
+ 1: "Disc Ident",
+ 2: "Disc SVID",
+ 3: "Disc Mode",
+ 4: "Enter Mode",
+ 5: "Exit Mode",
+ 6: "Attention",
+ # 16..31: SVID Specific Commands
+ # DisplayPort Commands
+ 16: "DP Status",
+ 17: "DP Configure",
+}
+VDM_ACK = ["REQ", "ACK", "NAK", "BSY"]
+
+
+class Decoder(srd.Decoder):
+ api_version = 2
+ id = 'usb_power_delivery'
+ name = 'USB PD'
+ longname = 'USB Power Delivery'
+ desc = 'USB Power Delivery protocol.'
+ license = 'gplv2+'
+ inputs = ['logic']
+ outputs = ['usb_pd']
+ channels = (
+ {'id': 'cc', 'name': 'CC', 'desc': 'Control channel'},
+ )
+ options = (
+ {'id': 'fulltext', 'desc': 'full text decoding of the packet',
+ 'default': 'no', 'values': ('yes', 'no')},
+ )
+ annotations = (
+ ('type', 'Packet Type'),
+ ('Preamble', 'Preamble'),
+ ('SOP', 'Start of Packet'),
+ ('Head', 'Header'),
+ ('Data', 'Data'),
+ ('CRC', 'Checksum'),
+ ('EOP', 'End Of Packet'),
+ ('Sym', '4b5b symbols'),
+ ('warnings', 'Warnings'),
+ ('src', 'Source Message'),
+ ('snk', 'Sink Message'),
+ ('payload', 'Payload'),
+ ('text', 'Plain text'),
+ )
+ annotation_rows = (
+ ('4B5B', 'symbols', (7, )),
+ ('Phase', 'parts', (1, 2, 3, 4, 5, 6, )),
+ ('payload', 'Payload', (11, )),
+ ('type', 'Type', (0, 9, 10, )),
+ ('warnings', 'Warnings', (8, )),
+ ('text', 'Full text', (12, )),
+ )
+ binary = (
+ ('raw-data', 'RAW binary data'),
+ )
+
+ def get_request(self, rdo):
+ pos = (rdo >> 28) & 7
+ op_ma = ((rdo >> 10) & 0x3ff) * 10
+ max_ma = (rdo & 0x3ff) * 10
+ flags = ""
+ for f in RDO_FLAGS.keys():
+ if rdo & f:
+ flags += " " + RDO_FLAGS[f]
+ return "[%d]%d/%d mA%s" % (pos, op_ma, max_ma, flags)
+
+ def get_source_cap(self, pdo):
+ t = (pdo >> 30) & 3
+ if t == 0:
+ mv = ((pdo >> 10) & 0x3ff) * 50
+ ma = ((pdo >> 0) & 0x3ff) * 10
+ p = "%.1fV %.1fA" % (mv/1000.0, ma/1000.0)
+ elif t == 1:
+ minv = ((pdo >> 10) & 0x3ff) * 50
+ maxv = ((pdo >> 20) & 0x3ff) * 50
+ mw = ((pdo >> 0) & 0x3ff) * 250
+ p = "%.1f/%.1fV %.1fW" % (minv/1000.0, maxv/1000.0, mw/1000.0)
+ elif t == 2:
+ minv = ((pdo >> 10) & 0x3ff) * 50
+ maxv = ((pdo >> 20) & 0x3ff) * 50
+ ma = ((pdo >> 0) & 0x3ff) * 10
+ p = "%.1f/%.1fV %.1fA" % (minv/1000.0, maxv/1000.0, ma/1000.0)
+ else:
+ p = ""
+ flags = ""
+ for f in PDO_FLAGS.keys():
+ if pdo & f:
+ flags += " " + PDO_FLAGS[f]
+ return "%s%s%s" % (PDO_TYPE[t], p, flags)
+
+ def get_sink_cap(self, pdo):
+ t = (pdo >> 30) & 3
+ if t == 0:
+ mv = ((pdo >> 10) & 0x3ff) * 50
+ ma = ((pdo >> 0) & 0x3ff) * 10
+ p = "%.1fV %.1fA" % (mv/1000.0, ma/1000.0)
+ elif t == 1:
+ minv = ((pdo >> 10) & 0x3ff) * 50
+ maxv = ((pdo >> 20) & 0x3ff) * 50
+ mw = ((pdo >> 0) & 0x3ff) * 250
+ p = "%.1f/%.1fV %.1fW" % (minv/1000.0, maxv/1000.0, mw/1000.0)
+ elif t == 2:
+ minv = ((pdo >> 10) & 0x3ff) * 50
+ maxv = ((pdo >> 20) & 0x3ff) * 50
+ ma = ((pdo >> 0) & 0x3ff) * 10
+ p = "%.1f/%.1fV %.1fA" % (minv/1000.0, maxv/1000.0, ma/1000.0)
+ else:
+ p = ""
+ flags = ""
+ for f in PDO_FLAGS.keys():
+ if pdo & f:
+ flags += " " + PDO_FLAGS[f]
+ return "%s%s%s" % (PDO_TYPE[t], p, flags)
+
+ def get_vdm(self, idx, data):
+ if idx == 0: # VDM header
+ vid = data >> 16
+ struct = data & (1 << 15)
+ txt = "VDM"
+ if struct: # Structured VDM
+ cmd = data & 0x1f
+ src = data & (1 << 5)
+ ack = (data >> 6) & 3
+ pos = (data >> 8) & 7
+ ver = (data >> 13) & 3
+ txt = VDM_ACK[ack] + " "
+ txt += VDM_CMDS[cmd] if cmd in VDM_CMDS else "cmd?"
+ txt += " pos %d" % (pos) if pos else " "
+ else: # Unstructured VDM
+ txt = "unstruct [%04x]" % (data & 0x7fff)
+ txt += " SVID:%04x" % (vid)
+ else: # VDM payload
+ txt = "VDO:%08x" % (data)
+ return txt
+
+ def get_bist(self, idx, data):
+ mode = data >> 28
+ counter = data & 0xffff
+ mode_name = BIST_MODES[mode] if mode in BIST_MODES else "INVALID"
+ if mode == 2:
+ mode_name = "Counter[= %d]" % (counter)
+ # TODO check all 0 bits are 0 / emit warnings
+ return "mode %s" % (mode_name) if idx == 0 else "invalid BRO"
+
+ def putpayload(self, s0, s1, idx):
+ t = self.head_type()
+ txt = "???"
+ if t == 2:
+ txt = self.get_request(self.data[idx])
+ elif t == 1:
+ txt = self.get_source_cap(self.data[idx])
+ elif t == 4:
+ txt = self.get_sink_cap(self.data[idx])
+ elif t == 15:
+ txt = self.get_vdm(idx, self.data[idx])
+ elif t == 3:
+ txt = self.get_bist(idx, self.data[idx])
+ self.putx(s0, s1, [11, [txt, txt]])
+ self.text += " - " + txt
+
+ def puthead(self):
+ ann_type = 9 if self.head_power_role() else 10
+ role = "SRC" if self.head_power_role() else "SNK"
+ if self.head_data_role() != self.head_power_role():
+ role += "/DFP" if self.head_data_role() else "/UFP"
+ t = self.head_type()
+ if self.head_count() == 0:
+ shortm = CTRL_TYPES[t]
+ else:
+ shortm = DATA_TYPES[t] if t in DATA_TYPES else "DAT???"
+
+ longm = "{:s}[{:d}]:{:s}".format(role, self.head_id(), shortm)
+ self.putx(0, -1, [ann_type, [longm, shortm]])
+ self.text += longm
+
+ def head_id(self):
+ return (self.head >> 9) & 7
+
+ def head_power_role(self):
+ return (self.head >> 8) & 1
+
+ def head_data_role(self):
+ return (self.head >> 5) & 1
+
+ def head_rev(self):
+ return ((self.head >> 6) & 3) + 1
+
+ def head_type(self):
+ return self.head & 0xF
+
+ def head_count(self):
+ return (self.head >> 12) & 7
+
+ def putx(self, s0, s1, data):
+ self.put(self.edges[s0], self.edges[s1], self.out_ann, data)
+
+ def putwarn(self, longm, shortm):
+ self.putx(0, -1, [8, [longm, shortm]])
+
+ def compute_crc32(self):
+ bdata = struct.pack("<H"+"I"*len(self.data), self.head & 0xffff,
+ *tuple([d & 0xffffffff for d in self.data]))
+ return zlib.crc32(bdata)
+
+ def rec_sym(self, i, sym):
+ self.putx(i, i+5, [7, SYM_NAME[sym]])
+
+ def get_sym(self, i, rec=True):
+ v = (self.bits[i] | (self.bits[i+1] << 1) | (self.bits[i+2] << 2) |
+ (self.bits[i+3] << 3) | (self.bits[i+4] << 4))
+ sym = DEC4B5B[v]
+ if rec:
+ self.rec_sym(i, sym)
+ return sym
+
+ def get_short(self):
+ i = self.idx
+ # Check it's not a truncated packet
+ if len(self.bits) - i <= 20:
+ self.putwarn("Truncated", "!")
+ return 0x0BAD
+ k = [self.get_sym(i), self.get_sym(i+5),
+ self.get_sym(i+10), self.get_sym(i+15)]
+ # TODO check bad symbols
+ val = k[0] | (k[1] << 4) | (k[2] << 8) | (k[3] << 12)
+ self.idx += 20
+ return val
+
+ def get_word(self):
+ lo = self.get_short()
+ hi = self.get_short()
+ return lo | (hi << 16)
+
+ def find_corrupted_sop(self, k):
+ # Start of packet are valid even if they have only 3 correct symbols
+ # out of 4.
+ for seq in START_OF_PACKETS.keys():
+ if [k[i] == seq[i] for i in range(len(k))].count(True) >= 3:
+ return START_OF_PACKETS[seq]
+ return None
+
+ def scan_eop(self):
+ for i in range(len(self.bits) - 19):
+ k = (self.get_sym(i, rec=False), self.get_sym(i+5, rec=False),
+ self.get_sym(i+10, rec=False), self.get_sym(i+15, rec=False))
+ sym = START_OF_PACKETS[k] if k in START_OF_PACKETS else None
+ if not sym:
+ sym = self.find_corrupted_sop(k)
+ # We have an interesting symbol sequence
+ if sym:
+ # annotate the preamble
+ self.putx(0, i, [1, ['Preamble', '...']])
+ # annotate each symbol
+ self.rec_sym(i, k[0])
+ self.rec_sym(i+5, k[1])
+ self.rec_sym(i+10, k[2])
+ self.rec_sym(i+15, k[3])
+ if sym == 'Hard Reset':
+ self.text += "HRST"
+ return -1 # Hard reset
+ elif sym == 'Cable Reset':
+ self.text += "CRST"
+ return -1 # Cable reset
+ else:
+ self.putx(i, i+20, [2, [sym, 'S']])
+ return i+20
+ self.putx(0, len(self.bits), [1, ['Junk???', 'XXX']])
+ self.text += "Junk???"
+ self.putwarn("No start of packet found", "XXX")
+ return -1 # No Start Of Packet
+
+ def __init__(self, **kwargs):
+ self.samplerate = None
+ self.idx = 0
+ self.packet_seq = 0
+
+ self.samplenum = 0
+ self.previous = 0
+ self.oldpins = [0]
+ self.startsample = None
+ self.bits = []
+ self.edges = []
+ self.bad = []
+ self.half_one = False
+ self.start_one = 0
+
+ def metadata(self, key, value):
+ if key == srd.SRD_CONF_SAMPLERATE:
+ self.samplerate = value
+ # 0 is 2 UI, space larger than 1.5x 0 is definitely wrong
+ self.maxbit = self.us2samples(3 * UI_US)
+ # duration threshold between half 1 and 0
+ self.threshold = self.us2samples(THRESHOLD_US)
+
+ def start(self):
+ self.out_python = self.register(srd.OUTPUT_PYTHON)
+ self.out_ann = self.register(srd.OUTPUT_ANN)
+ self.out_binary = self.register(srd.OUTPUT_BINARY)
+ self.out_bitrate = self.register(
+ srd.OUTPUT_META,
+ meta=(int, 'Bitrate', 'Bitrate during the packet')
+ )
+
+ def us2samples(self, us):
+ if self.samplerate is None:
+ raise Exception("Need the samplerate.")
+ return int(us * self.samplerate / 1000000)
+
+ def decode_packet(self):
+ self.data = []
+ self.idx = 0
+ self.text = ""
+
+ if len(self.edges) < 50:
+ return # Not a real PD packet
+
+ self.packet_seq += 1
+ tstamp = float(self.startsample) / self.samplerate
+ self.text += "#%-4d (%8.6fms): " % (self.packet_seq, tstamp*1000)
+
+ self.idx = self.scan_eop()
+ if self.idx < 0:
+ # Full text trace of the issue
+ self.putx(0, self.idx, [12, [self.text, '...']])
+ return # No real packet: ABORT
+
+ # Packet header
+ self.head = self.get_short()
+ self.putx(self.idx-20, self.idx, [3, ['H:%04x' % (self.head), 'HD']])
+ self.puthead()
+
+ # Decode data payload
+ for i in range(self.head_count()):
+ self.data.append(self.get_word())
+ self.putx(self.idx-40, self.idx,
+ [4, ['[%d]%08x' % (i, self.data[i]), 'D%d' % (i)]])
+ self.putpayload(self.idx-40, self.idx, i)
+
+ # CRC check
+ self.crc = self.get_word()
+ ccrc = self.compute_crc32()
+ if self.crc != ccrc:
+ self.putwarn("Bad CRC %08x != %08x" % (self.crc, ccrc), "CRC!")
+ self.putx(self.idx-40, self.idx, [5, ['CRC:%08x' % (self.crc), 'CRC']])
+
+ # End of Packet
+ if len(self.bits) >= self.idx + 5 and self.get_sym(self.idx) == EOP:
+ self.putx(self.idx, self.idx + 5, [6, ['EOP', 'E']])
+ self.idx += 5
+ else:
+ self.putwarn("No EOP", "EOP!")
+ # Full text trace
+ if self.options['fulltext'] == 'yes':
+ self.putx(0, self.idx, [12, [self.text, '...']])
+
+ # Meta data for bitrate
+ ss, es = self.edges[0], self.edges[-1]
+ bitrate = self.samplerate*len(self.bits) / float(es - ss)
+ self.put(es, ss, self.out_bitrate, int(bitrate))
+ # Raw binary data (BMC decoded)
+ self.put(es, ss, self.out_binary, (0, bytes(self.bits)))
+
+ def decode(self, ss, es, data):
+ if self.samplerate is None:
+ raise Exception("Cannot decode without samplerate.")
+ for (self.samplenum, pins) in data:
+ # find edges ...
+ if self.oldpins == pins:
+ continue
+
+ self.oldpins, (cc, ) = pins, pins
+
+ # First sample of the packet, just record the start date
+ if not self.startsample:
+ self.startsample = self.samplenum
+ self.previous = self.samplenum
+ continue
+
+ diff = self.samplenum - self.previous
+
+ # Large idle: use it as the end of packet
+ if diff > self.maxbit:
+ # the last edge of the packet
+ self.edges.append(self.previous)
+ # Export the packet
+ self.decode_packet()
+ # Reset for next packet
+ self.startsample = self.samplenum
+ self.bits = []
+ self.edges = []
+ self.bad = []
+ self.half_one = False
+ self.start_one = 0
+ else: # add the bit to the packet
+ is_zero = diff > self.threshold
+ if is_zero and not self.half_one:
+ self.bits.append(0)
+ self.edges.append(self.previous)
+ elif not is_zero and self.half_one:
+ self.bits.append(1)
+ self.edges.append(self.start_one)
+ self.half_one = False
+ elif not is_zero and not self.half_one:
+ self.half_one = True
+ self.start_one = self.previous
+ else: # Invalid BMC sequence
+ self.bad.append((self.start_one, self.previous))
+ # TODO try to recover
+ self.bits.append(0)
+ self.edges.append(self.previous)
+ self.half_one = False
+ self.previous = self.samplenum
projects / libsigrokdecode.git / commitdiff