mawei/dragonpilot
1
0
Fork 0
mirror of https://github.com/dragonpilot/dragonpilot.git synced 2026-02-26 23:03:53 +08:00
Code Issues Packages Projects Releases Wiki Activity

Merge panda subtree

Browse Source
This commit is contained in:
Vehicle Researcher
2019-10-09 18:31:56 +00:00
parent d17745fb5e 38faf7f8a4
commit f2de9784f8
98 changed files with 1314 additions and 796 deletions
Download Patch File Download Diff File Expand all files Collapse all files

54
panda/python/isotp.py
View File

@@ -1,13 +1,15 @@
import binascii
DEBUG = False
def msg(x):
if DEBUG:
print "S:",x.encode("hex")
print("S:", binascii.hexlify(x))
if len(x) <= 7:
ret = chr(len(x)) + x
else:
assert False
return ret.ljust(8, "\x00")
return ret.ljust(8, b"\x00")
kmsgs = []
def recv(panda, cnt, addr, nbus):
@@ -24,35 +26,35 @@ def recv(panda, cnt, addr, nbus):
# leave around
nmsgs.append((ids, ts, dat, bus))
kmsgs = nmsgs[-256:]
return map(str, ret)
return ret
def isotp_recv_subaddr(panda, addr, bus, sendaddr, subaddr):
msg = recv(panda, 1, addr, bus)[0]
# TODO: handle other subaddr also communicating
assert ord(msg[0]) == subaddr
assert msg[0] == subaddr
if ord(msg[1])&0xf0 == 0x10:
if msg[1]&0xf0 == 0x10:
# first
tlen = ((ord(msg[1]) & 0xf) << 8) | ord(msg[2])
tlen = ((msg[1] & 0xf) << 8) | msg[2]
dat = msg[3:]
# 0 block size?
CONTINUE = chr(subaddr) + "\x30" + "\x00"*6
CONTINUE = chr(subaddr).encode("utf8") + b"\x30" + b"\x00"*6
panda.can_send(sendaddr, CONTINUE, bus)
idx = 1
for mm in recv(panda, (tlen-len(dat) + 5)/6, addr, bus):
assert ord(mm[0]) == subaddr
assert ord(mm[1]) == (0x20 | (idx&0xF))
for mm in recv(panda, (tlen-len(dat) + 5)//6, addr, bus):
assert mm[0] == subaddr
assert mm[1] == (0x20 | (idx&0xF))
dat += mm[2:]
idx += 1
elif ord(msg[1])&0xf0 == 0x00:
elif msg[1]&0xf0 == 0x00:
# single
tlen = ord(msg[1]) & 0xf
tlen = msg[1] & 0xf
dat = msg[2:]
else:
print msg.encode("hex")
print(binascii.hexlify(msg))
assert False
return dat[0:tlen]
@@ -69,26 +71,26 @@ def isotp_send(panda, x, addr, bus=0, recvaddr=None, subaddr=None):
panda.can_send(addr, chr(subaddr)+msg(x)[0:7], bus)
else:
if subaddr:
ss = chr(subaddr) + chr(0x10 + (len(x)>>8)) + chr(len(x)&0xFF) + x[0:5]
ss = (chr(subaddr) + chr(0x10 + (len(x)>>8)) + chr(len(x)&0xFF)).encode("utf8") + x[0:5]
x = x[5:]
else:
ss = chr(0x10 + (len(x)>>8)) + chr(len(x)&0xFF) + x[0:6]
ss = (chr(0x10 + (len(x)>>8)) + chr(len(x)&0xFF)).encode("utf8") + x[0:6]
x = x[6:]
idx = 1
sends = []
while len(x) > 0:
if subaddr:
sends.append(((chr(subaddr) + chr(0x20 + (idx&0xF)) + x[0:6]).ljust(8, "\x00")))
sends.append((((chr(subaddr) + chr(0x20 + (idx&0xF))).encode('utf8') + x[0:6]).ljust(8, b"\x00")))
x = x[6:]
else:
sends.append(((chr(0x20 + (idx&0xF)) + x[0:7]).ljust(8, "\x00")))
sends.append(((chr(0x20 + (idx&0xF)).encode("utf8") + x[0:7]).ljust(8, b"\x00")))
x = x[7:]
idx += 1
# actually send
panda.can_send(addr, ss, bus)
rr = recv(panda, 1, recvaddr, bus)[0]
if rr.find("\x30\x01") != -1:
if rr.find(b"\x30\x01") != -1:
for s in sends[:-1]:
panda.can_send(addr, s, 0)
rr = recv(panda, 1, recvaddr, bus)[0]
@@ -105,31 +107,31 @@ def isotp_recv(panda, addr, bus=0, sendaddr=None, subaddr=None):
else:
msg = recv(panda, 1, addr, bus)[0]
if ord(msg[0])&0xf0 == 0x10:
if msg[0]&0xf0 == 0x10:
# first
tlen = ((ord(msg[0]) & 0xf) << 8) | ord(msg[1])
tlen = ((msg[0] & 0xf) << 8) | msg[1]
dat = msg[2:]
# 0 block size?
CONTINUE = "\x30" + "\x00"*7
CONTINUE = b"\x30" + b"\x00"*7
panda.can_send(sendaddr, CONTINUE, bus)
idx = 1
for mm in recv(panda, (tlen-len(dat) + 6)/7, addr, bus):
assert ord(mm[0]) == (0x20 | (idx&0xF))
for mm in recv(panda, (tlen-len(dat) + 6)//7, addr, bus):
assert mm[0] == (0x20 | (idx&0xF))
dat += mm[1:]
idx += 1
elif ord(msg[0])&0xf0 == 0x00:
elif msg[0]&0xf0 == 0x00:
# single
tlen = ord(msg[0]) & 0xf
tlen = msg[0] & 0xf
dat = msg[1:]
else:
assert False
dat = dat[0:tlen]
if DEBUG:
print "R:",dat.encode("hex")
print("R:", binascii.hexlify(dat))
return dat