Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

usb_power_delivery: support Revision 3.2 Extended Data #118

Open
wants to merge 1 commit into
base: master
Choose a base branch
from
Open

usb_power_delivery: support Revision 3.2 Extended Data #118

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
107 changes: 90 additions & 17 deletions decoders/usb_power_delivery/pd.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -40,19 +40,22 @@
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',
16: 'Not Supported',
9: 'DR_Swap',
10: 'PR_Swap',
11: 'VCONN_Swap',
12: 'Wait',
13: 'Soft_Reset',
14: 'Data_Reset',
15: 'Data_Reset_Complete',
16: 'Not_Supported',
17: 'Get_Source_Cap_Extended',
18: 'Get_Status',
19: 'FR_Swap',
20: 'Get_PPS_Status',
21: 'Get_Country_Codes',
22: 'Get_Sink_Cap_Extended',
23: 'Get_Source_Info',
24: 'Get_Revision'
}

# Data message type
Expand All @@ -64,9 +67,37 @@
5: 'Battery_Status',
6: 'Alert',
7: 'Get_Country_Info',
8: 'Enter_USB',
9: 'EPR_Request',
10: 'EPR_Mode',
11: 'Source_Info',
12: 'Revision',
15: 'VDM'
}

# Extended message type
EXT_TYPES = {
1: 'Source_Capabilities_Extended',
2: 'Status',
3: 'Get_Battery_Cap',
4: 'Get_Battery_Status',
5: 'Battery_Capabilities',
6: 'Get_Manufacturer_Info',
7: 'Manufacturer_Info',
8: 'Security_Request',
9: 'Security_Response',
10: 'Firmware_Update_Request',
11: 'Firmware_Update_Response',
12: 'PPS_Status',
13: 'Country_Info',
14: 'Country_Codes',
15: 'Sink_Capabilities_Extended',
16: 'Extended_Control',
17: 'EPR_Source_Capabilities',
18: 'EPR_Sink_Capabilities',
30: 'Vendor_Defined_Extended'
}

# 4b5b encoding of the symbols
DEC4B5B = [
0x10, # Error 00000
Expand Down Expand Up @@ -377,6 +408,10 @@ def get_bist(self, idx, data):
return 'mode %s' % (mode_name) if idx == 0 else 'invalid BRO'

def putpayload(self, s0, s1, idx):
if self.head_ext() == 1:
# TODO: to decode Extended Message Types
return

t = self.head_type()
txt = '['+str(idx+1)+'] '
if t == 2:
Expand All @@ -396,14 +431,21 @@ def puthead(self):
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]
if self.head_ext() == 1:
shortm = EXT_TYPES[t] if t in EXT_TYPES else 'EXT???'
elif self.head_count() == 0:
shortm = CTRL_TYPES[t] if t in CTRL_TYPES else 'CTR???'
else:
shortm = DATA_TYPES[t] if t in DATA_TYPES else 'DAT???'

longm = '(r{:d}) {:s}[{:d}]: {:s}'.format(self.head_rev(), role, self.head_id(), shortm)
self.putx(0, -1, [ann_type, [longm, shortm]])
self.text += longm

def head_ext(self):
if self.head_rev() == 3:
return (self.head >> 15) & 1
return 0

def head_id(self):
return (self.head >> 9) & 7
Expand All @@ -418,20 +460,29 @@ def head_rev(self):
return ((self.head >> 6) & 3) + 1

def head_type(self):
if self.head_rev() == 3:
return self.head & 0x1F
return self.head & 0xF

def head_count(self):
return (self.head >> 12) & 7

def exthead_size(self):
return (self.exthead) & 511

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]))
if self.head_ext() == 1:
bdata = struct.pack('<HH'+'B'*len(self.extdata), self.head & 0xffff, self.exthead & 0xffff,
*tuple([d & 0xff for d in self.extdata]))
else:
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):
Expand All @@ -445,6 +496,18 @@ def get_sym(self, i, rec=True):
self.rec_sym(i, sym)
return sym

def get_byte(self):
i = self.idx
# Check it's not a truncated packet.
if len(self.bits) - i <= 10:
self.putwarn('Truncated', '!')
return 0x0BAD
k = [self.get_sym(i), self.get_sym(i+5)]
# TODO: Check bad symbols.
val = k[0] | (k[1] << 4)
self.idx += 10
return val

def get_short(self):
i = self.idx
# Check it's not a truncated packet.
Expand Down Expand Up @@ -539,6 +602,7 @@ def us2samples(self, us):

def decode_packet(self):
self.data = []
self.extdata = []
self.idx = 0
self.text = ''

Expand All @@ -561,11 +625,20 @@ def decode_packet(self):
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)
if self.head_ext() == 1:
self.exthead = self.get_short()
self.putx(self.idx-20, self.idx, [3, ['EH:%04x' % (self.exthead), 'EHD']])
for i in range(self.exthead_size()):
self.extdata.append(self.get_byte())
self.putx(self.idx-10, self.idx,
[4, ['[%d]%02x' % (i, self.extdata[i]), 'D%d' % (i)]])
#self.putpayload(self.idx-40, self.idx, i)
else:
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()
Expand Down