X-Git-Url: https://sigrok.org/gitweb/?p=libsigrokdecode.git;a=blobdiff_plain;f=decoders%2Fusb_power_delivery%2Fpd.py;h=702c22a525bf7e04aa7be4576a92cb797bb4d173;hp=ad28d97a3c34c45aaafa5e41572043bfd72fa64e;hb=a886ba3b903d2e900020affe743bb486e8880464;hpb=669f30f4b1035a8eb0a5350246a3bc50cf8b376d diff --git a/decoders/usb_power_delivery/pd.py b/decoders/usb_power_delivery/pd.py index ad28d97..702c22a 100644 --- a/decoders/usb_power_delivery/pd.py +++ b/decoders/usb_power_delivery/pd.py @@ -2,6 +2,7 @@ ## This file is part of the libsigrokdecode project. ## ## Copyright (C) 2015 Google, Inc +## Copyright (C) 2018 Peter Hazenberg ## ## 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 @@ -14,8 +15,7 @@ ## 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, write to the Free Software -## Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA +## along with this program; if not, see . ## import sigrokdecode as srd @@ -102,14 +102,23 @@ EOP = 0x16 SYNC_CODES = [SYNC1, SYNC2, SYNC3] HRST_CODES = [RST1, RST1, RST1, RST2] +SOP_SEQUENCES = [ + (SYNC1, SYNC1, SYNC1, SYNC2), + (SYNC1, SYNC1, SYNC3, SYNC3), + (SYNC1, SYNC3, SYNC1, SYNC3), + (SYNC1, RST2, RST2, SYNC3), + (SYNC1, RST2, SYNC3, SYNC2), + (RST1, SYNC1, RST1, SYNC3), + (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', + SOP_SEQUENCES[0]: 'SOP', + SOP_SEQUENCES[1]: "SOP'", + SOP_SEQUENCES[2]: 'SOP"', + SOP_SEQUENCES[3]: "SOP' Debug", + SOP_SEQUENCES[4]: 'SOP" Debug', + SOP_SEQUENCES[5]: 'Cable Reset', + SOP_SEQUENCES[6]: 'Hard Reset', } SYM_NAME = [ @@ -144,7 +153,7 @@ RDO_FLAGS = { (1 << 26): 'cap_mismatch', (1 << 27): 'give_back' } -PDO_TYPE = ['', 'BATT:', 'VAR:', ''] + PDO_FLAGS = { (1 << 29): 'dual_role_power', (1 << 28): 'suspend', @@ -178,11 +187,12 @@ VDM_CMDS = { } VDM_ACK = ['REQ', 'ACK', 'NAK', 'BSY'] + class SamplerateError(Exception): pass class Decoder(srd.Decoder): - api_version = 2 + api_version = 3 id = 'usb_power_delivery' name = 'USB PD' longname = 'USB Power Delivery' @@ -191,7 +201,10 @@ class Decoder(srd.Decoder): inputs = ['logic'] outputs = ['usb_pd'] channels = ( - {'id': 'cc', 'name': 'CC', 'desc': 'Control channel'}, + {'id': 'cc1', 'name': 'CC1', 'desc': 'Control channel 1'}, + ) + optional_channels = ( + {'id': 'cc2', 'name': 'CC2', 'desc': 'Control channel 2'}, ) options = ( {'id': 'fulltext', 'desc': 'full text decoding of the packet', @@ -224,63 +237,57 @@ class Decoder(srd.Decoder): ('raw-data', 'RAW binary data'), ) + stored_pdos = {} + + def get_request(self, rdo): pos = (rdo >> 28) & 7 - op_ma = ((rdo >> 10) & 0x3ff) * 10 - max_ma = (rdo & 0x3ff) * 10 + op_ma = ((rdo >> 10) & 0x3ff) * 0.01 + max_ma = (rdo & 0x3ff) * 0.01 flags = '' - for f in RDO_FLAGS.keys(): + for f in sorted(RDO_FLAGS.keys(), reverse = True): 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) + flags += ' [' + RDO_FLAGS[f] + ']' + if pos in self.stored_pdos.keys(): + return '(PDO #%d: %s) %gA (operating) / %gA (max)%s' % (pos, self.stored_pdos[pos], op_ma, max_ma, flags) else: - p = '' + return '(PDO #%d) %gA (operating) / %gA (max)%s' % (pos, op_ma, max_ma, flags) + + def get_source_sink_cap(self, pdo, idx): + t1 = (pdo >> 30) & 3 + if t1 == 0: + t_name = 'Fixed' + mv = ((pdo >> 10) & 0x3ff) * 0.05 + ma = ((pdo >> 0) & 0x3ff) * 0.01 + p = '%gV %gA (%gW)' % (mv, ma, mv*ma) + self.stored_pdos[idx] = '%s %gV' % (t_name, mv) + elif t1 == 1: + t_name = 'Battery' + minv = ((pdo >> 10) & 0x3ff) * 0.05 + maxv = ((pdo >> 20) & 0x3ff) * 0.05 + mw = ((pdo >> 0) & 0x3ff) * 0.25 + p = '%g/%gV %gW' % (minv, maxv, mw) + self.stored_pdos[idx] = '%s %g/%gV' % (t_name, minv, maxv) + elif t1 == 2: + t_name = 'Variable' + minv = ((pdo >> 10) & 0x3ff) * 0.05 + maxv = ((pdo >> 20) & 0x3ff) * 0.05 + ma = ((pdo >> 0) & 0x3ff) * 0.01 + p = '%g/%gV %gA' % (minv, maxv, ma) + self.stored_pdos[idx] = '%s %g/%gV' % (t_name, minv, maxv) + elif t1 == 3: + t2 = (pdo >> 28) & 3 + if t2 == 0: + t_name = 'Programmable' + p = 'TODO: PPS support' + else: + t_name = 'Reserved APDO: '+bin(t2) + p = '' flags = '' - for f in PDO_FLAGS.keys(): + for f in sorted(PDO_FLAGS.keys(), reverse = True): 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) + flags += ' [' + PDO_FLAGS[f] + ']' + return '[%s] %s%s' % (t_name, p, flags) def get_vdm(self, idx, data): if idx == 0: # VDM header @@ -314,17 +321,15 @@ class Decoder(srd.Decoder): def putpayload(self, s0, s1, idx): t = self.head_type() - txt = '???' + txt = '['+str(idx+1)+'] ' 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]) + txt += self.get_request(self.data[idx]) + elif t == 1 or t == 4: + txt += self.get_source_sink_cap(self.data[idx], idx+1) elif t == 15: - txt = self.get_vdm(idx, self.data[idx]) + txt += self.get_vdm(idx, self.data[idx]) elif t == 3: - txt = self.get_bist(idx, self.data[idx]) + txt += self.get_bist(idx, self.data[idx]) self.putx(s0, s1, [11, [txt, txt]]) self.text += ' - ' + txt @@ -404,7 +409,7 @@ class Decoder(srd.Decoder): 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(): + for seq in SOP_SEQUENCES: if [k[i] == seq[i] for i in range(len(k))].count(True) >= 3: return START_OF_PACKETS[seq] return None @@ -413,7 +418,7 @@ class Decoder(srd.Decoder): 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 + sym = START_OF_PACKETS.get(k, None) if not sym: sym = self.find_corrupted_sop(k) # We have an interesting symbol sequence @@ -439,19 +444,21 @@ class Decoder(srd.Decoder): self.putwarn('No start of packet found', 'XXX') return -1 # No Start Of Packet - def __init__(self, **kwargs): + def __init__(self): + self.reset() + + def reset(self): 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 + self.stored_pdos = {} def metadata(self, key, value): if key == srd.SRD_CONF_SAMPLERATE: @@ -471,8 +478,6 @@ class Decoder(srd.Decoder): ) def us2samples(self, us): - if not self.samplerate: - raise SamplerateError('Need the samplerate.') return int(us * self.samplerate / 1000000) def decode_packet(self): @@ -529,15 +534,12 @@ class Decoder(srd.Decoder): # Raw binary data (BMC decoded) self.put(es, ss, self.out_binary, [0, bytes(self.bits)]) - def decode(self, ss, es, data): + def decode(self): if not self.samplerate: raise SamplerateError('Cannot decode without samplerate.') - for (self.samplenum, pins) in data: - # find edges ... - if self.oldpins == pins: - continue + while True: + pins = self.wait([{0: 'e'}, {1: 'e'}, {'skip': 100*1000}]) - self.oldpins, (cc, ) = pins, pins # First sample of the packet, just record the start date if not self.startsample: