mirror of https://github.com/commaai/panda.git
Created python package and implemented industry best practices.
Supports python 2 and 3 (to the best of my testing ability at the time)
This commit is contained in:
Jessy Diamond Exum
parent
0e46ca4a5d
commit
6a25791fea
|
|
@ -2,4 +2,5 @@
|
|||
.*.swp
|
||||
*.o
|
||||
a.out
|
||||
|
||||
*~
|
||||
.#*
|
||||
|
|
@ -1 +0,0 @@
|
|||
obj/*
|
||||
|
|
@ -1,14 +1,18 @@
|
|||
# python library to interface with panda
|
||||
from __future__ import print_function
|
||||
import binascii
|
||||
import struct
|
||||
import hashlib
|
||||
import socket
|
||||
import usb1
|
||||
from usb1 import USBErrorIO, USBErrorOverflow
|
||||
|
||||
try:
|
||||
from hexdump import hexdump
|
||||
except:
|
||||
pass
|
||||
__version__ = '0.0.1'
|
||||
|
||||
class PandaHashMismatchException(Exception):
|
||||
def __init__(self, hash_, expected_hash):
|
||||
super(PandaHashMismatchException, self).__init__(
|
||||
"Hash '%s' did not match the expected hash '%s'"%\
|
||||
(binascii.hexlify(hash_), binascii.hexlify(expected_hash)))
|
||||
|
||||
def parse_can_buffer(dat):
|
||||
ret = []
|
||||
|
|
@ -33,7 +37,7 @@ class PandaWifiStreaming(object):
|
|||
|
||||
def can_recv(self):
|
||||
ret = []
|
||||
while 1:
|
||||
while True:
|
||||
try:
|
||||
dat, addr = self.sock.recvfrom(0x200*0x10)
|
||||
if addr == (self.ip, self.port):
|
||||
|
|
@ -61,7 +65,8 @@ class WifiHandle(object):
|
|||
return self.__recv()
|
||||
|
||||
def bulkWrite(self, endpoint, data, timeout=0):
|
||||
assert len(data) <= 0x10
|
||||
if len(data) > 0x10:
|
||||
raise ValueError("Data must not be longer than 0x10")
|
||||
self.sock.send(struct.pack("HH", endpoint, len(data))+data)
|
||||
self.__recv() # to /dev/null
|
||||
|
||||
|
|
@ -73,10 +78,12 @@ class WifiHandle(object):
|
|||
self.sock.close()
|
||||
|
||||
class Panda(object):
|
||||
REQUEST_TYPE = usb1.TYPE_VENDOR | usb1.RECIPIENT_DEVICE
|
||||
|
||||
def __init__(self, serial=None, claim=True):
|
||||
if serial == "WIFI":
|
||||
self.handle = WifiHandle()
|
||||
print "opening WIFI device"
|
||||
print("opening WIFI device")
|
||||
else:
|
||||
context = usb1.USBContext()
|
||||
|
||||
|
|
@ -84,7 +91,7 @@ class Panda(object):
|
|||
for device in context.getDeviceList(skip_on_error=True):
|
||||
if device.getVendorID() == 0xbbaa and device.getProductID() == 0xddcc:
|
||||
if serial is None or device.getSerialNumber() == serial:
|
||||
print "opening device", device.getSerialNumber()
|
||||
print("opening device", device.getSerialNumber())
|
||||
self.handle = device.open()
|
||||
if claim:
|
||||
self.handle.claimInterface(0)
|
||||
|
|
@ -109,7 +116,7 @@ class Panda(object):
|
|||
# ******************* health *******************
|
||||
|
||||
def health(self):
|
||||
dat = self.handle.controlRead(usb1.TYPE_VENDOR | usb1.RECIPIENT_DEVICE, 0xd2, 0, 0, 13)
|
||||
dat = self.handle.controlRead(Panda.REQUEST_TYPE, 0xd2, 0, 0, 13)
|
||||
a = struct.unpack("IIBBBBB", dat)
|
||||
return {"voltage": a[0], "current": a[1],
|
||||
"started": a[2], "controls_allowed": a[3],
|
||||
|
|
@ -121,82 +128,79 @@ class Panda(object):
|
|||
|
||||
def enter_bootloader(self):
|
||||
try:
|
||||
self.handle.controlWrite(usb1.TYPE_VENDOR | usb1.RECIPIENT_DEVICE, 0xd1, 0, 0, '')
|
||||
except Exception:
|
||||
self.handle.controlWrite(Panda.REQUEST_TYPE, 0xd1, 0, 0, b'')
|
||||
except Exception as e:
|
||||
print(e)
|
||||
pass
|
||||
|
||||
def get_serial(self):
|
||||
dat = str(self.handle.controlRead(usb1.TYPE_VENDOR | usb1.RECIPIENT_DEVICE, 0xd0, 0, 0, 0x20))
|
||||
assert dat[0x1c:] == hashlib.sha1(dat[0:0x1c]).digest()[0:4]
|
||||
dat = self.handle.controlRead(Panda.REQUEST_TYPE, 0xd0, 0, 0, 0x20)
|
||||
hashsig, calc_hash = dat[0x1c:], hashlib.sha1(dat[0:0x1c]).digest()[0:4]
|
||||
if hashsig != calc_hash:
|
||||
raise PandaHashMismatchException(calc_hash, hashsig)
|
||||
return [dat[0:0x10], dat[0x10:0x10+10]]
|
||||
|
||||
def get_secret(self):
|
||||
dat = str(self.handle.controlRead(usb1.TYPE_VENDOR | usb1.RECIPIENT_DEVICE, 0xd0, 1, 0, 0x10))
|
||||
return dat.encode("hex")
|
||||
return self.handle.controlRead(Panda.REQUEST_TYPE, 0xd0, 1, 0, 0x10)
|
||||
|
||||
# ******************* configuration *******************
|
||||
|
||||
def set_controls_allowed(self, on):
|
||||
if on:
|
||||
self.handle.controlWrite(usb1.TYPE_VENDOR | usb1.RECIPIENT_DEVICE, 0xdc, 0x1337, 0, '')
|
||||
else:
|
||||
self.handle.controlWrite(usb1.TYPE_VENDOR | usb1.RECIPIENT_DEVICE, 0xdc, 0, 0, '')
|
||||
self.handle.controlWrite(Panda.REQUEST_TYPE, 0xdc, (0x1337 if on else 0), 0, b'')
|
||||
|
||||
def set_gmlan(self, on, bus=2):
|
||||
if on:
|
||||
self.handle.controlWrite(usb1.TYPE_VENDOR | usb1.RECIPIENT_DEVICE, 0xdb, 1, bus, '')
|
||||
else:
|
||||
self.handle.controlWrite(usb1.TYPE_VENDOR | usb1.RECIPIENT_DEVICE, 0xdb, 0, bus, '')
|
||||
self.handle.controlWrite(Panda.REQUEST_TYPE, 0xdb, 1, bus, b'')
|
||||
|
||||
def set_uart_baud(self, uart, rate):
|
||||
self.handle.controlWrite(usb1.TYPE_VENDOR | usb1.RECIPIENT_DEVICE, 0xe1, uart, rate, '')
|
||||
self.handle.controlWrite(Panda.REQUEST_TYPE, 0xe1, uart, rate, b'')
|
||||
|
||||
def set_uart_parity(self, uart, parity):
|
||||
# parity, 0=off, 1=even, 2=odd
|
||||
self.handle.controlWrite(usb1.TYPE_VENDOR | usb1.RECIPIENT_DEVICE, 0xe2, uart, parity, '')
|
||||
self.handle.controlWrite(Panda.REQUEST_TYPE, 0xe2, uart, parity, b'')
|
||||
|
||||
def set_uart_callback(self, uart, install):
|
||||
self.handle.controlWrite(usb1.TYPE_VENDOR | usb1.RECIPIENT_DEVICE, 0xe3, uart, int(install), '')
|
||||
self.handle.controlWrite(Panda.REQUEST_TYPE, 0xe3, uart, int(install), b'')
|
||||
|
||||
# ******************* can *******************
|
||||
|
||||
def can_send_many(self, arr):
|
||||
snds = []
|
||||
for addr, _, dat, bus in arr:
|
||||
transmit = 1
|
||||
extended = 4
|
||||
for addr, _, dat, bus in arr:
|
||||
if addr >= 0x800:
|
||||
rir = (addr << 3) | transmit | extended
|
||||
else:
|
||||
rir = (addr << 21) | transmit
|
||||
snd = struct.pack("II", rir, len(dat) | (bus << 4)) + dat
|
||||
snd = snd.ljust(0x10, '\x00')
|
||||
snd = snd.ljust(0x10, b'\x00')
|
||||
snds.append(snd)
|
||||
|
||||
while 1:
|
||||
while True:
|
||||
try:
|
||||
self.handle.bulkWrite(3, ''.join(snds))
|
||||
print("DAT: %s"%b''.join(snds).__repr__())
|
||||
self.handle.bulkWrite(3, b''.join(snds))
|
||||
break
|
||||
except (USBErrorIO, USBErrorOverflow):
|
||||
print "CAN: BAD SEND MANY, RETRYING"
|
||||
except (usb1.USBErrorIO, usb1.USBErrorOverflow):
|
||||
print("CAN: BAD SEND MANY, RETRYING")
|
||||
|
||||
def can_send(self, addr, dat, bus):
|
||||
self.can_send_many([[addr, None, dat, bus]])
|
||||
|
||||
def can_recv(self):
|
||||
dat = ""
|
||||
while 1:
|
||||
dat = bytearray()
|
||||
while True:
|
||||
try:
|
||||
dat = self.handle.bulkRead(1, 0x10*256)
|
||||
break
|
||||
except (USBErrorIO, USBErrorOverflow):
|
||||
print "CAN: BAD RECV, RETRYING"
|
||||
except (usb1.USBErrorIO, usb1.USBErrorOverflow):
|
||||
print("CAN: BAD RECV, RETRYING")
|
||||
return parse_can_buffer(dat)
|
||||
|
||||
# ******************* serial *******************
|
||||
|
||||
def serial_read(self, port_number):
|
||||
return self.handle.controlRead(usb1.TYPE_VENDOR | usb1.RECIPIENT_DEVICE, 0xe0, port_number, 0, 0x40)
|
||||
return self.handle.controlRead(Panda.REQUEST_TYPE, 0xe0, port_number, 0, 0x40)
|
||||
|
||||
def serial_write(self, port_number, ln):
|
||||
return self.handle.bulkWrite(2, chr(port_number) + ln)
|
||||
|
|
@ -205,22 +209,22 @@ class Panda(object):
|
|||
|
||||
# pulse low for wakeup
|
||||
def kline_wakeup(self):
|
||||
ret = self.handle.controlWrite(usb1.TYPE_VENDOR | usb1.RECIPIENT_DEVICE, 0xf0, 0, 0, "")
|
||||
self.handle.controlWrite(Panda.REQUEST_TYPE, 0xf0, 0, 0, b'')
|
||||
|
||||
def kline_drain(self, bus=2):
|
||||
# drain buffer
|
||||
bret = ""
|
||||
while 1:
|
||||
ret = self.handle.controlRead(usb1.TYPE_VENDOR | usb1.RECIPIENT_DEVICE, 0xe0, bus, 0, 0x40)
|
||||
bret = bytearray()
|
||||
while True:
|
||||
ret = self.handle.controlRead(Panda.REQUEST_TYPE, 0xe0, bus, 0, 0x40)
|
||||
if len(ret) == 0:
|
||||
break
|
||||
bret += str(ret)
|
||||
bret += ret
|
||||
return bret
|
||||
|
||||
def kline_ll_recv(self, cnt, bus=2):
|
||||
echo = ""
|
||||
echo = bytearray()
|
||||
while len(echo) != cnt:
|
||||
echo += str(self.handle.controlRead(usb1.TYPE_VENDOR | usb1.RECIPIENT_DEVICE, 0xe0, bus, 0, cnt-len(echo)))
|
||||
echo += self.handle.controlRead(Panda.REQUEST_TYPE, 0xe0, bus, 0, cnt-len(echo))
|
||||
return echo
|
||||
|
||||
def kline_send(self, x, bus=2, checksum=True):
|
||||
|
|
@ -238,13 +242,12 @@ class Panda(object):
|
|||
self.handle.bulkWrite(2, chr(bus)+ts)
|
||||
echo = self.kline_ll_recv(len(ts), bus=bus)
|
||||
if echo != ts:
|
||||
print "**** ECHO ERROR %d ****" % i
|
||||
print echo.encode("hex")
|
||||
print ts.encode("hex")
|
||||
print("**** ECHO ERROR %d ****" % i)
|
||||
print(binascii.hexlify(echo))
|
||||
print(binascii.hexlify(ts))
|
||||
assert echo == ts
|
||||
|
||||
def kline_recv(self, bus=2):
|
||||
msg = self.kline_ll_recv(2, bus=bus)
|
||||
msg += self.kline_ll_recv(ord(msg[1])-2, bus=bus)
|
||||
return msg
|
||||
|
||||
|
|
@ -0,0 +1,64 @@
|
|||
#-*- coding: utf-8 -*-
|
||||
|
||||
"""
|
||||
Panda CAN Controller Dongle
|
||||
~~~~~
|
||||
|
||||
Setup
|
||||
`````
|
||||
|
||||
$ pip install . # or python setup.py install
|
||||
"""
|
||||
|
||||
import codecs
|
||||
import os
|
||||
import re
|
||||
from setuptools import setup, Extension
|
||||
|
||||
here = os.path.abspath(os.path.dirname(__file__))
|
||||
|
||||
def read(*parts):
|
||||
"""Taken from pypa pip setup.py:
|
||||
intentionally *not* adding an encoding option to open, See:
|
||||
https://github.com/pypa/virtualenv/issues/201#issuecomment-3145690
|
||||
"""
|
||||
return codecs.open(os.path.join(here, *parts), 'r').read()
|
||||
|
||||
|
||||
def find_version(*file_paths):
|
||||
version_file = read(*file_paths)
|
||||
version_match = re.search(r"^__version__ = ['\"]([^'\"]*)['\"]",
|
||||
version_file, re.M)
|
||||
if version_match:
|
||||
return version_match.group(1)
|
||||
raise RuntimeError("Unable to find version string.")
|
||||
|
||||
setup(
|
||||
name='panda',
|
||||
version=find_version("panda", "__init__.py"),
|
||||
url='https://github.com/commaai/panda',
|
||||
author='Comma.ai',
|
||||
author_email='',
|
||||
packages=[
|
||||
'panda',
|
||||
],
|
||||
platforms='any',
|
||||
license='MIT',
|
||||
install_requires=[
|
||||
'libusb1 >= 1.6.4',
|
||||
'hexdump >= 3.3',
|
||||
'pycrypto >= 2.6.1',
|
||||
'tqdm >= 4.14.0',
|
||||
],
|
||||
ext_modules = [],
|
||||
description="Code powering the comma.ai panda",
|
||||
long_description=open(os.path.join(os.path.dirname(__file__),
|
||||
'README.md')).read(),
|
||||
classifiers=[
|
||||
'Development Status :: 2 - Pre-Alpha',
|
||||
"Natural Language :: English",
|
||||
"Programming Language :: Python :: 2",
|
||||
"Programming Language :: Python :: 3",
|
||||
"Topic :: System :: Hardware",
|
||||
],
|
||||
)
|
||||
|
|
@ -1,9 +1,12 @@
|
|||
#!/usr/bin/env python
|
||||
from __future__ import print_function
|
||||
import os
|
||||
import struct
|
||||
import sys
|
||||
import time
|
||||
from collections import defaultdict
|
||||
from panda.lib.panda import Panda
|
||||
|
||||
sys.path.append(os.path.join(os.path.dirname(os.path.realpath(__file__)), ".."))
|
||||
from panda import Panda
|
||||
|
||||
# fake
|
||||
def sec_since_boot():
|
||||
|
|
@ -16,7 +19,7 @@ def can_printer():
|
|||
lp = sec_since_boot()
|
||||
msgs = defaultdict(list)
|
||||
canbus = int(os.getenv("CAN", 0))
|
||||
while 1:
|
||||
while True:
|
||||
can_recv = p.can_recv()
|
||||
for address, _, dat, src in can_recv:
|
||||
if src == canbus:
|
||||
|
|
@ -27,9 +30,8 @@ def can_printer():
|
|||
dd += "%5.2f\n" % (sec_since_boot() - start)
|
||||
for k,v in sorted(zip(msgs.keys(), map(lambda x: x[-1].encode("hex"), msgs.values()))):
|
||||
dd += "%s(%6d) %s\n" % ("%04X(%4d)" % (k,k),len(msgs[k]), v)
|
||||
print dd
|
||||
print(dd)
|
||||
lp = sec_since_boot()
|
||||
|
||||
if __name__ == "__main__":
|
||||
can_printer()
|
||||
|
||||
|
|
|
|||
|
|
@ -1,10 +1,12 @@
|
|||
#!/usr/bin/env python
|
||||
from __future__ import print_function
|
||||
import os
|
||||
import sys
|
||||
import usb1
|
||||
import time
|
||||
import select
|
||||
from panda.lib.panda import Panda
|
||||
|
||||
sys.path.append(os.path.join(os.path.dirname(os.path.realpath(__file__)), ".."))
|
||||
from panda import Panda
|
||||
|
||||
setcolor = ["\033[1;32;40m", "\033[1;31;40m"]
|
||||
unsetcolor = "\033[00m"
|
||||
|
|
@ -17,17 +19,16 @@ if __name__ == "__main__":
|
|||
serials = filter(lambda x: x==os.getenv("SERIAL"), serials)
|
||||
|
||||
pandas = map(lambda x: Panda(x, False), serials)
|
||||
while 1:
|
||||
while True:
|
||||
for i, panda in enumerate(pandas):
|
||||
while 1:
|
||||
while True:
|
||||
ret = panda.serial_read(port_number)
|
||||
if len(ret) > 0:
|
||||
sys.stdout.write(setcolor[i] + ret + unsetcolor)
|
||||
sys.stdout.write(setcolor[i] + ret.decode('utf8') + unsetcolor)
|
||||
sys.stdout.flush()
|
||||
else:
|
||||
break
|
||||
if select.select([sys.stdin], [], [], 0) == ([sys.stdin], [], []):
|
||||
if select.select([sys.stdin], [], [], 0)[0][0] == sys.stdin:
|
||||
ln = sys.stdin.readline()
|
||||
panda.serial_write(port_number, ln)
|
||||
time.sleep(0.01)
|
||||
|
||||
|
|
|
|||
|
|
@ -1,23 +1,25 @@
|
|||
#!/usr/bin/env python
|
||||
from __future__ import print_function
|
||||
import os
|
||||
import sys
|
||||
import time
|
||||
import usb1
|
||||
import random
|
||||
import struct
|
||||
from panda.lib.panda import Panda
|
||||
|
||||
from hexdump import hexdump
|
||||
from itertools import permutations
|
||||
|
||||
sys.path.append(os.path.join(os.path.dirname(os.path.realpath(__file__)), ".."))
|
||||
from panda import Panda
|
||||
|
||||
def get_test_string():
|
||||
return "test"+os.urandom(10)
|
||||
|
||||
def run_test():
|
||||
pandas = Panda.list()
|
||||
print pandas
|
||||
print(pandas)
|
||||
|
||||
if len(pandas) == 0:
|
||||
print "NO PANDAS"
|
||||
print("NO PANDAS")
|
||||
assert False
|
||||
|
||||
if len(pandas) == 1:
|
||||
|
|
@ -27,17 +29,17 @@ def run_test():
|
|||
|
||||
def run_test_w_pandas(pandas):
|
||||
h = map(lambda x: Panda(x), pandas)
|
||||
print h
|
||||
print(h)
|
||||
|
||||
for hh in h:
|
||||
hh.set_controls_allowed(True)
|
||||
|
||||
# test both directions
|
||||
for ho in permutations(range(len(h)), r=2):
|
||||
print "***************** TESTING", ho
|
||||
print("***************** TESTING", ho)
|
||||
|
||||
# **** test health packet ****
|
||||
print "health", ho[0], h[ho[0]].health()
|
||||
print("health", ho[0], h[ho[0]].health())
|
||||
|
||||
# **** test K/L line loopback ****
|
||||
for bus in [2,3]:
|
||||
|
|
@ -55,11 +57,11 @@ def run_test_w_pandas(pandas):
|
|||
hexdump(st)
|
||||
hexdump(ret)
|
||||
assert st == ret
|
||||
print "K/L pass", bus, ho
|
||||
print("K/L pass", bus, ho)
|
||||
|
||||
# **** test can line loopback ****
|
||||
for bus in [0,1,4,5,6]:
|
||||
print "test can", bus
|
||||
print("test can", bus)
|
||||
# flush
|
||||
cans_echo = h[ho[0]].can_recv()
|
||||
cans_loop = h[ho[1]].can_recv()
|
||||
|
|
@ -89,7 +91,7 @@ def run_test_w_pandas(pandas):
|
|||
cans_echo = h[ho[0]].can_recv()
|
||||
cans_loop = h[ho[1]].can_recv()
|
||||
|
||||
print bus, cans_echo, cans_loop
|
||||
print(bus, cans_echo, cans_loop)
|
||||
|
||||
assert len(cans_echo) == 1
|
||||
assert len(cans_loop) == 1
|
||||
|
|
@ -102,10 +104,10 @@ def run_test_w_pandas(pandas):
|
|||
|
||||
assert cans_echo[0][3] == bus+2
|
||||
if cans_loop[0][3] != bus:
|
||||
print "EXPECTED %d GOT %d" % (bus, cans_loop[0][3])
|
||||
print("EXPECTED %d GOT %d" % (bus, cans_loop[0][3]))
|
||||
assert cans_loop[0][3] == bus
|
||||
|
||||
print "CAN pass", bus, ho
|
||||
print("CAN pass", bus, ho)
|
||||
|
||||
if __name__ == "__main__":
|
||||
if len(sys.argv) > 1:
|
||||
|
|
@ -113,8 +115,7 @@ if __name__ == "__main__":
|
|||
run_test()
|
||||
else :
|
||||
i = 0
|
||||
while 1:
|
||||
print "************* testing %d" % i
|
||||
while True:
|
||||
print("************* testing %d" % i)
|
||||
run_test()
|
||||
i += 1
|
||||
|
||||
|
|
|
|||
|
|
@ -1,29 +1,32 @@
|
|||
#!/usr/bin/env python
|
||||
import os
|
||||
import sys
|
||||
import struct
|
||||
import time
|
||||
from panda.lib.panda import Panda
|
||||
|
||||
sys.path.append(os.path.join(os.path.dirname(os.path.realpath(__file__)), ".."))
|
||||
from panda import Panda
|
||||
|
||||
if __name__ == "__main__":
|
||||
if os.getenv("WIFI") is not None:
|
||||
p = Panda("WIFI")
|
||||
else:
|
||||
p = Panda()
|
||||
print p.get_serial()
|
||||
print p.health()
|
||||
print(p.get_serial())
|
||||
print(p.health())
|
||||
|
||||
t1 = time.time()
|
||||
for i in range(100):
|
||||
p.get_serial()
|
||||
t2 = time.time()
|
||||
print "100 requests took %.2f ms" % ((t2-t1)*1000)
|
||||
print("100 requests took %.2f ms" % ((t2-t1)*1000))
|
||||
|
||||
p.set_controls_allowed(True)
|
||||
|
||||
a = 0
|
||||
while 1:
|
||||
while True:
|
||||
# flood
|
||||
msg = "\xaa"*4 + struct.pack("I", a)
|
||||
msg = b"\xaa"*4 + struct.pack("I", a)
|
||||
p.can_send(0xaa, msg, 0)
|
||||
p.can_send(0xaa, msg, 1)
|
||||
p.can_send(0xaa, msg, 4)
|
||||
|
|
@ -31,6 +34,5 @@ if __name__ == "__main__":
|
|||
|
||||
dat = p.can_recv()
|
||||
if len(dat) > 0:
|
||||
print dat
|
||||
print(dat)
|
||||
a += 1
|
||||
|
||||
|
|
|
|||
|
|
@ -1,25 +1,29 @@
|
|||
#!/usr/bin/env python
|
||||
from __future__ import print_function
|
||||
import os
|
||||
import sys
|
||||
import struct
|
||||
import time
|
||||
from panda.lib.panda import Panda, PandaWifiStreaming
|
||||
from tqdm import tqdm
|
||||
|
||||
sys.path.append(os.path.join(os.path.dirname(os.path.realpath(__file__)), ".."))
|
||||
from panda import Panda, PandaWifiStreaming
|
||||
|
||||
# test throughput between USB and wifi
|
||||
|
||||
if __name__ == "__main__":
|
||||
print Panda.list()
|
||||
print(Panda.list())
|
||||
p_out = Panda("108018800f51363038363036")
|
||||
print p_out.get_serial()
|
||||
print(p_out.get_serial())
|
||||
#p_in = Panda("02001b000f51363038363036")
|
||||
p_in = Panda("WIFI")
|
||||
print p_in.get_serial()
|
||||
print(p_in.get_serial())
|
||||
|
||||
p_in = PandaWifiStreaming()
|
||||
|
||||
#while 1:
|
||||
#while True:
|
||||
# p_in.can_recv()
|
||||
#exit(0)
|
||||
#sys.exit(0)
|
||||
|
||||
p_out.set_controls_allowed(True)
|
||||
|
||||
|
|
@ -32,13 +36,13 @@ if __name__ == "__main__":
|
|||
BATCH_SIZE = 16
|
||||
for a in tqdm(range(0, 10000, BATCH_SIZE)):
|
||||
for b in range(0, BATCH_SIZE):
|
||||
msg = "\xaa"*4 + struct.pack("I", a+b)
|
||||
msg = b"\xaa"*4 + struct.pack("I", a+b)
|
||||
if a%1 == 0:
|
||||
p_out.can_send(0xaa, msg, 0)
|
||||
|
||||
dat_out, dat_in = p_out.can_recv(), p_in.can_recv()
|
||||
if len(dat_in) != 0:
|
||||
print len(dat_in)
|
||||
print(len(dat_in))
|
||||
|
||||
num_out = [struct.unpack("I", i[4:])[0] for _, _, i, _ in dat_out]
|
||||
num_in = [struct.unpack("I", i[4:])[0] for _, _, i, _ in dat_in]
|
||||
|
|
@ -47,14 +51,14 @@ if __name__ == "__main__":
|
|||
set_out.update(num_out)
|
||||
|
||||
# swag
|
||||
print "waiting for packets"
|
||||
print("waiting for packets")
|
||||
time.sleep(2.0)
|
||||
dat_in = p_in.can_recv()
|
||||
print len(dat_in)
|
||||
print(len(dat_in))
|
||||
num_in = [struct.unpack("I", i[4:])[0] for _, _, i, _ in dat_in]
|
||||
set_in.update(num_in)
|
||||
|
||||
if len(set_out - set_in):
|
||||
print "MISSING %d" % len(set_out - set_in)
|
||||
print("MISSING %d" % len(set_out - set_in))
|
||||
if len(set_out - set_in) < 256:
|
||||
print map(hex, sorted(list(set_out - set_in)))
|
||||
print(map(hex, sorted(list(set_out - set_in))))
|
||||
|
|
|
|||
Loading…
Reference in New Issue