mawei/sunnypilot
1
0
Fork 0
mirror of https://github.com/sunnypilot/sunnypilot.git synced 2026-02-19 14:43:57 +08:00
Code Issues Packages Projects Releases Wiki Activity

Remove old panda subtree

Browse Source 
old-commit-hash: 27803e787b
This commit is contained in:
Vehicle Researcher
2019-06-28 21:09:11 +00:00
parent e338e9b840
commit 6c3cb4dfe9
311 changed files with 0 additions and 30420 deletions
Download Patch File Download Diff File Expand all files Collapse all files

561
panda/python/__init__.py
View File

@@ -1,561 +0,0 @@
# python library to interface with panda
from __future__ import print_function
import binascii
import struct
import hashlib
import socket
import usb1
import os
import time
import traceback
import subprocess
from dfu import PandaDFU
from esptool import ESPROM, CesantaFlasher
from flash_release import flash_release
from update import ensure_st_up_to_date
from serial import PandaSerial
from isotp import isotp_send, isotp_recv
__version__ = '0.0.8'
BASEDIR = os.path.join(os.path.dirname(os.path.realpath(__file__)), "../")
DEBUG = os.getenv("PANDADEBUG") is not None
# *** wifi mode ***
def build_st(target, mkfile="Makefile"):
from panda import BASEDIR
cmd = 'cd %s && make -f %s clean && make -f %s %s >/dev/null' % (os.path.join(BASEDIR, "board"), mkfile, mkfile, target)
try:
output = subprocess.check_output(cmd, stderr=subprocess.STDOUT, shell=True)
except subprocess.CalledProcessError as exception:
output = exception.output
returncode = exception.returncode
raise
def parse_can_buffer(dat):
ret = []
for j in range(0, len(dat), 0x10):
ddat = dat[j:j+0x10]
f1, f2 = struct.unpack("II", ddat[0:8])
extended = 4
if f1 & extended:
address = f1 >> 3
else:
address = f1 >> 21
dddat = ddat[8:8+(f2&0xF)]
if DEBUG:
print(" R %x: %s" % (address, str(dddat).encode("hex")))
ret.append((address, f2>>16, dddat, (f2>>4)&0xFF))
return ret
class PandaWifiStreaming(object):
def __init__(self, ip="192.168.0.10", port=1338):
self.sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
self.sock.setblocking(0)
self.ip = ip
self.port = port
self.kick()
def kick(self):
# must be called at least every 5 seconds
self.sock.sendto("hello", (self.ip, self.port))
def can_recv(self):
ret = []
while True:
try:
dat, addr = self.sock.recvfrom(0x200*0x10)
if addr == (self.ip, self.port):
ret += parse_can_buffer(dat)
except socket.error as e:
if e.errno != 35 and e.errno != 11:
traceback.print_exc()
break
return ret
# stupid tunneling of USB over wifi and SPI
class WifiHandle(object):
def __init__(self, ip="192.168.0.10", port=1337):
self.sock = socket.create_connection((ip, port))
def __recv(self):
ret = self.sock.recv(0x44)
length = struct.unpack("I", ret[0:4])[0]
return ret[4:4+length]
def controlWrite(self, request_type, request, value, index, data, timeout=0):
# ignore data in reply, panda doesn't use it
return self.controlRead(request_type, request, value, index, 0, timeout)
def controlRead(self, request_type, request, value, index, length, timeout=0):
self.sock.send(struct.pack("HHBBHHH", 0, 0, request_type, request, value, index, length))
return self.__recv()
def bulkWrite(self, endpoint, data, timeout=0):
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
def bulkRead(self, endpoint, length, timeout=0):
self.sock.send(struct.pack("HH", endpoint, 0))
return self.__recv()
def close(self):
self.sock.close()
# *** normal mode ***
class Panda(object):
SAFETY_NOOUTPUT = 0
SAFETY_HONDA = 1
SAFETY_TOYOTA = 2
SAFETY_GM = 3
SAFETY_HONDA_BOSCH = 4
SAFETY_FORD = 5
SAFETY_CADILLAC = 6
SAFETY_HYUNDAI = 7
SAFETY_TESLA = 8
SAFETY_CHRYSLER = 9
SAFETY_TOYOTA_IPAS = 0x1335
SAFETY_TOYOTA_NOLIMITS = 0x1336
SAFETY_ALLOUTPUT = 0x1337
SAFETY_ELM327 = 0xE327
SERIAL_DEBUG = 0
SERIAL_ESP = 1
SERIAL_LIN1 = 2
SERIAL_LIN2 = 3
GMLAN_CAN2 = 1
GMLAN_CAN3 = 2
REQUEST_IN = usb1.ENDPOINT_IN | usb1.TYPE_VENDOR | usb1.RECIPIENT_DEVICE
REQUEST_OUT = usb1.ENDPOINT_OUT | usb1.TYPE_VENDOR | usb1.RECIPIENT_DEVICE
def __init__(self, serial=None, claim=True):
self._serial = serial
self._handle = None
self.connect(claim)
def close(self):
self._handle.close()
self._handle = None
def connect(self, claim=True, wait=False):
if self._handle != None:
self.close()
if self._serial == "WIFI":
self._handle = WifiHandle()
print("opening WIFI device")
self.wifi = True
else:
context = usb1.USBContext()
self._handle = None
self.wifi = False
while 1:
try:
for device in context.getDeviceList(skip_on_error=True):
#print(device)
if device.getVendorID() == 0xbbaa and device.getProductID() in [0xddcc, 0xddee]:
try:
this_serial = device.getSerialNumber()
except Exception:
continue
if self._serial is None or this_serial == self._serial:
self._serial = this_serial
print("opening device", self._serial, hex(device.getProductID()))
time.sleep(1)
self.bootstub = device.getProductID() == 0xddee
self.legacy = (device.getbcdDevice() != 0x2300)
self._handle = device.open()
if claim:
self._handle.claimInterface(0)
#self._handle.setInterfaceAltSetting(0, 0) #Issue in USB stack
break
except Exception as e:
print("exception", e)
traceback.print_exc()
if wait == False or self._handle != None:
break
context = usb1.USBContext() #New context needed so new devices show up
assert(self._handle != None)
print("connected")
def reset(self, enter_bootstub=False, enter_bootloader=False):
# reset
try:
if enter_bootloader:
self._handle.controlWrite(Panda.REQUEST_IN, 0xd1, 0, 0, b'')
else:
if enter_bootstub:
self._handle.controlWrite(Panda.REQUEST_IN, 0xd1, 1, 0, b'')
else:
self._handle.controlWrite(Panda.REQUEST_IN, 0xd8, 0, 0, b'')
except Exception:
pass
if not enter_bootloader:
self.reconnect()
def reconnect(self):
self.close()
time.sleep(1.0)
success = False
# wait up to 15 seconds
for i in range(0, 15):
try:
self.connect()
success = True
break
except Exception:
print("reconnecting is taking %d seconds..." % (i+1))
try:
dfu = PandaDFU(PandaDFU.st_serial_to_dfu_serial(self._serial))
dfu.recover()
except Exception:
pass
time.sleep(1.0)
if not success:
raise Exception("reconnect failed")
@staticmethod
def flash_static(handle, code):
# confirm flasher is present
fr = handle.controlRead(Panda.REQUEST_IN, 0xb0, 0, 0, 0xc)
assert fr[4:8] == "\xde\xad\xd0\x0d"
# unlock flash
print("flash: unlocking")
handle.controlWrite(Panda.REQUEST_IN, 0xb1, 0, 0, b'')
# erase sectors 1 and 2
print("flash: erasing")
handle.controlWrite(Panda.REQUEST_IN, 0xb2, 1, 0, b'')
handle.controlWrite(Panda.REQUEST_IN, 0xb2, 2, 0, b'')
# flash over EP2
STEP = 0x10
print("flash: flashing")
for i in range(0, len(code), STEP):
handle.bulkWrite(2, code[i:i+STEP])
# reset
print("flash: resetting")
try:
handle.controlWrite(Panda.REQUEST_IN, 0xd8, 0, 0, b'')
except Exception:
pass
def flash(self, fn=None, code=None, reconnect=True):
print("flash: main version is " + self.get_version())
if not self.bootstub:
self.reset(enter_bootstub=True)
assert(self.bootstub)
if fn is None and code is None:
if self.legacy:
fn = "obj/comma.bin"
print("building legacy st code")
build_st(fn, "Makefile.legacy")
else:
fn = "obj/panda.bin"
print("building panda st code")
build_st(fn)
fn = os.path.join(BASEDIR, "board", fn)
if code is None:
with open(fn) as f:
code = f.read()
# get version
print("flash: bootstub version is " + self.get_version())
# do flash
Panda.flash_static(self._handle, code)
# reconnect
if reconnect:
self.reconnect()
def recover(self, timeout=None):
self.reset(enter_bootloader=True)
t_start = time.time()
while len(PandaDFU.list()) == 0:
print("waiting for DFU...")
time.sleep(0.1)
if timeout is not None and (time.time() - t_start) > timeout:
return False
dfu = PandaDFU(PandaDFU.st_serial_to_dfu_serial(self._serial))
dfu.recover()
# reflash after recover
self.connect(True, True)
self.flash()
return True
@staticmethod
def flash_ota_st():
ret = os.system("cd %s && make clean && make ota" % (os.path.join(BASEDIR, "board")))
time.sleep(1)
return ret==0
@staticmethod
def flash_ota_wifi(release=False):
release_str = "RELEASE=1" if release else ""
ret = os.system("cd {} && make clean && {} make ota".format(os.path.join(BASEDIR, "boardesp"),release_str))
time.sleep(1)
return ret==0
@staticmethod
def list():
context = usb1.USBContext()
ret = []
try:
for device in context.getDeviceList(skip_on_error=True):
if device.getVendorID() == 0xbbaa and device.getProductID() in [0xddcc, 0xddee]:
try:
ret.append(device.getSerialNumber())
except Exception:
continue
except Exception:
pass
# TODO: detect if this is real
#ret += ["WIFI"]
return ret
def call_control_api(self, msg):
self._handle.controlWrite(Panda.REQUEST_OUT, msg, 0, 0, b'')
# ******************* health *******************
def health(self):
dat = self._handle.controlRead(Panda.REQUEST_IN, 0xd2, 0, 0, 13)
a = struct.unpack("IIBBBBB", dat)
return {"voltage": a[0], "current": a[1],
"started": a[2], "controls_allowed": a[3],
"gas_interceptor_detected": a[4],
"started_signal_detected": a[5],
"started_alt": a[6]}
# ******************* control *******************
def enter_bootloader(self):
try:
self._handle.controlWrite(Panda.REQUEST_OUT, 0xd1, 0, 0, b'')
except Exception as e:
print(e)
pass
def get_version(self):
return self._handle.controlRead(Panda.REQUEST_IN, 0xd6, 0, 0, 0x40)
def is_grey(self):
ret = self._handle.controlRead(Panda.REQUEST_IN, 0xc1, 0, 0, 0x40)
return ret == "\x01"
def get_serial(self):
dat = self._handle.controlRead(Panda.REQUEST_IN, 0xd0, 0, 0, 0x20)
hashsig, calc_hash = dat[0x1c:], hashlib.sha1(dat[0:0x1c]).digest()[0:4]
assert(hashsig == calc_hash)
return [dat[0:0x10], dat[0x10:0x10+10]]
def get_secret(self):
return self._handle.controlRead(Panda.REQUEST_IN, 0xd0, 1, 0, 0x10)
# ******************* configuration *******************
def set_usb_power(self, on):
self._handle.controlWrite(Panda.REQUEST_OUT, 0xe6, int(on), 0, b'')
def set_esp_power(self, on):
self._handle.controlWrite(Panda.REQUEST_OUT, 0xd9, int(on), 0, b'')
def esp_reset(self, bootmode=0):
self._handle.controlWrite(Panda.REQUEST_OUT, 0xda, int(bootmode), 0, b'')
time.sleep(0.2)
def set_safety_mode(self, mode=SAFETY_NOOUTPUT):
self._handle.controlWrite(Panda.REQUEST_OUT, 0xdc, mode, 0, b'')
def set_can_forwarding(self, from_bus, to_bus):
# TODO: This feature may not work correctly with saturated buses
self._handle.controlWrite(Panda.REQUEST_OUT, 0xdd, from_bus, to_bus, b'')
def set_gmlan(self, bus=2):
if bus is None:
self._handle.controlWrite(Panda.REQUEST_OUT, 0xdb, 0, 0, b'')
elif bus in [Panda.GMLAN_CAN2, Panda.GMLAN_CAN3]:
self._handle.controlWrite(Panda.REQUEST_OUT, 0xdb, 1, bus, b'')
def set_can_loopback(self, enable):
# set can loopback mode for all buses
self._handle.controlWrite(Panda.REQUEST_OUT, 0xe5, int(enable), 0, b'')
def set_can_enable(self, bus_num, enable):
# sets the can transciever enable pin
self._handle.controlWrite(Panda.REQUEST_OUT, 0xf4, int(bus_num), int(enable), b'')
def set_can_speed_kbps(self, bus, speed):
self._handle.controlWrite(Panda.REQUEST_OUT, 0xde, bus, int(speed*10), b'')
def set_uart_baud(self, uart, rate):
self._handle.controlWrite(Panda.REQUEST_OUT, 0xe4, uart, rate/300, b'')
def set_uart_parity(self, uart, parity):
# parity, 0=off, 1=even, 2=odd
self._handle.controlWrite(Panda.REQUEST_OUT, 0xe2, uart, parity, b'')
def set_uart_callback(self, uart, install):
self._handle.controlWrite(Panda.REQUEST_OUT, 0xe3, uart, int(install), b'')
# ******************* can *******************
def can_send_many(self, arr):
snds = []
transmit = 1
extended = 4
for addr, _, dat, bus in arr:
assert len(dat) <= 8
if DEBUG:
print(" W %x: %s" % (addr, dat.encode("hex")))
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, b'\x00')
snds.append(snd)
while True:
try:
#print("DAT: %s"%b''.join(snds).__repr__())
if self.wifi:
for s in snds:
self._handle.bulkWrite(3, s)
else:
self._handle.bulkWrite(3, b''.join(snds))
break
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 = bytearray()
while True:
try:
dat = self._handle.bulkRead(1, 0x10*256)
break
except (usb1.USBErrorIO, usb1.USBErrorOverflow):
print("CAN: BAD RECV, RETRYING")
return parse_can_buffer(dat)
def can_clear(self, bus):
"""Clears all messages from the specified internal CAN ringbuffer as
though it were drained.
Args:
bus (int): can bus number to clear a tx queue, or 0xFFFF to clear the
global can rx queue.
"""
self._handle.controlWrite(Panda.REQUEST_OUT, 0xf1, bus, 0, b'')
# ******************* isotp *******************
def isotp_send(self, addr, dat, bus, recvaddr=None, subaddr=None):
return isotp_send(self, dat, addr, bus, recvaddr, subaddr)
def isotp_recv(self, addr, bus=0, sendaddr=None, subaddr=None):
return isotp_recv(self, addr, bus, sendaddr, subaddr)
# ******************* serial *******************
def serial_read(self, port_number):
ret = []
while 1:
lret = bytes(self._handle.controlRead(Panda.REQUEST_IN, 0xe0, port_number, 0, 0x40))
if len(lret) == 0:
break
ret.append(lret)
return b''.join(ret)
def serial_write(self, port_number, ln):
ret = 0
for i in range(0, len(ln), 0x20):
ret += self._handle.bulkWrite(2, struct.pack("B", port_number) + ln[i:i+0x20])
return ret
def serial_clear(self, port_number):
"""Clears all messages (tx and rx) from the specified internal uart
ringbuffer as though it were drained.
Args:
port_number (int): port number of the uart to clear.
"""
self._handle.controlWrite(Panda.REQUEST_OUT, 0xf2, port_number, 0, b'')
# ******************* kline *******************
# pulse low for wakeup
def kline_wakeup(self):
if DEBUG:
print("kline wakeup...")
self._handle.controlWrite(Panda.REQUEST_OUT, 0xf0, 0, 0, b'')
if DEBUG:
print("kline wakeup done")
def kline_drain(self, bus=2):
# drain buffer
bret = bytearray()
while True:
ret = self._handle.controlRead(Panda.REQUEST_IN, 0xe0, bus, 0, 0x40)
if len(ret) == 0:
break
elif DEBUG:
print("kline drain: "+str(ret).encode("hex"))
bret += ret
return bytes(bret)
def kline_ll_recv(self, cnt, bus=2):
echo = bytearray()
while len(echo) != cnt:
ret = str(self._handle.controlRead(Panda.REQUEST_OUT, 0xe0, bus, 0, cnt-len(echo)))
if DEBUG and len(ret) > 0:
print("kline recv: "+ret.encode("hex"))
echo += ret
return str(echo)
def kline_send(self, x, bus=2, checksum=True):
def get_checksum(dat):
result = 0
result += sum(map(ord, dat)) if isinstance(b'dat', str) else sum(dat)
result = -result
return struct.pack("B", result % 0x100)
self.kline_drain(bus=bus)
if checksum:
x += get_checksum(x)
for i in range(0, len(x), 0xf):
ts = x[i:i+0xf]
if DEBUG:
print("kline send: "+ts.encode("hex"))
self._handle.bulkWrite(2, chr(bus).encode()+ts)
echo = self.kline_ll_recv(len(ts), bus=bus)
if echo != ts:
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

122
panda/python/dfu.py
View File

@@ -1,122 +0,0 @@
from __future__ import print_function
import os
import usb1
import struct
import time
# *** DFU mode ***
DFU_DNLOAD = 1
DFU_UPLOAD = 2
DFU_GETSTATUS = 3
DFU_CLRSTATUS = 4
DFU_ABORT = 6
class PandaDFU(object):
def __init__(self, dfu_serial):
context = usb1.USBContext()
for device in context.getDeviceList(skip_on_error=True):
if device.getVendorID() == 0x0483 and device.getProductID() == 0xdf11:
try:
this_dfu_serial = device._getASCIIStringDescriptor(3)
except Exception:
continue
if this_dfu_serial == dfu_serial or dfu_serial is None:
self._handle = device.open()
self.legacy = "07*128Kg" in self._handle.getASCIIStringDescriptor(4)
return
raise Exception("failed to open "+dfu_serial)
@staticmethod
def list():
context = usb1.USBContext()
dfu_serials = []
try:
for device in context.getDeviceList(skip_on_error=True):
if device.getVendorID() == 0x0483 and device.getProductID() == 0xdf11:
try:
dfu_serials.append(device._getASCIIStringDescriptor(3))
except Exception:
pass
except Exception:
pass
return dfu_serials
@staticmethod
def st_serial_to_dfu_serial(st):
if st == None or st == "none":
return None
uid_base = struct.unpack("H"*6, st.decode("hex"))
return struct.pack("!HHH", uid_base[1] + uid_base[5], uid_base[0] + uid_base[4] + 0xA, uid_base[3]).encode("hex").upper()
def status(self):
while 1:
dat = str(self._handle.controlRead(0x21, DFU_GETSTATUS, 0, 0, 6))
if dat[1] == "\x00":
break
def clear_status(self):
# Clear status
stat = str(self._handle.controlRead(0x21, DFU_GETSTATUS, 0, 0, 6))
if stat[4] == "\x0a":
self._handle.controlRead(0x21, DFU_CLRSTATUS, 0, 0, 0)
elif stat[4] == "\x09":
self._handle.controlWrite(0x21, DFU_ABORT, 0, 0, "")
self.status()
stat = str(self._handle.controlRead(0x21, DFU_GETSTATUS, 0, 0, 6))
def erase(self, address):
self._handle.controlWrite(0x21, DFU_DNLOAD, 0, 0, "\x41" + struct.pack("I", address))
self.status()
def program(self, address, dat, block_size=None):
if block_size == None:
block_size = len(dat)
# Set Address Pointer
self._handle.controlWrite(0x21, DFU_DNLOAD, 0, 0, "\x21" + struct.pack("I", address))
self.status()
# Program
dat += "\xFF"*((block_size-len(dat)) % block_size)
for i in range(0, len(dat)/block_size):
ldat = dat[i*block_size:(i+1)*block_size]
print("programming %d with length %d" % (i, len(ldat)))
self._handle.controlWrite(0x21, DFU_DNLOAD, 2+i, 0, ldat)
self.status()
def program_bootstub(self, code_bootstub):
self.clear_status()
self.erase(0x8004000)
self.erase(0x8000000)
self.program(0x8000000, code_bootstub, 0x800)
self.reset()
def recover(self):
from panda import BASEDIR, build_st
if self.legacy:
fn = "obj/bootstub.comma.bin"
print("building legacy bootstub")
build_st(fn, "Makefile.legacy")
else:
fn = "obj/bootstub.panda.bin"
print("building panda bootstub")
build_st(fn)
fn = os.path.join(BASEDIR, "board", fn)
with open(fn) as f:
code = f.read()
self.program_bootstub(code)
def reset(self):
# **** Reset ****
self._handle.controlWrite(0x21, DFU_DNLOAD, 0, 0, "\x21" + struct.pack("I", 0x8000000))
self.status()
try:
self._handle.controlWrite(0x21, DFU_DNLOAD, 2, 0, "")
stat = str(self._handle.controlRead(0x21, DFU_GETSTATUS, 0, 0, 6))
except Exception:
pass

1314
panda/python/esptool.py
View File

File diff suppressed because it is too large Load Diff

95
panda/python/flash_release.py
View File

@@ -1,95 +0,0 @@
#!/usr/bin/env python
from __future__ import print_function
import sys
import time
import requests
import json
import StringIO
def flash_release(path=None, st_serial=None):
from panda import Panda, PandaDFU, ESPROM, CesantaFlasher
from zipfile import ZipFile
def status(x):
print("\033[1;32;40m"+x+"\033[00m")
if st_serial == None:
# look for Panda
panda_list = Panda.list()
if len(panda_list) == 0:
raise Exception("panda not found, make sure it's connected and your user can access it")
elif len(panda_list) > 1:
raise Exception("Please only connect one panda")
st_serial = panda_list[0]
print("Using panda with serial %s" % st_serial)
if path == None:
print("Fetching latest firmware from github.com/commaai/panda-artifacts")
r = requests.get("https://raw.githubusercontent.com/commaai/panda-artifacts/master/latest.json")
url = json.loads(r.text)['url']
r = requests.get(url)
print("Fetching firmware from %s" % url)
path = StringIO.StringIO(r.content)
zf = ZipFile(path)
zf.printdir()
version = zf.read("version")
status("0. Preparing to flash "+version)
code_bootstub = zf.read("bootstub.panda.bin")
code_panda = zf.read("panda.bin")
code_boot_15 = zf.read("boot_v1.5.bin")
code_boot_15 = code_boot_15[0:2] + "\x00\x30" + code_boot_15[4:]
code_user1 = zf.read("user1.bin")
code_user2 = zf.read("user2.bin")
# enter DFU mode
status("1. Entering DFU mode")
panda = Panda(st_serial)
panda.enter_bootloader()
time.sleep(1)
# program bootstub
status("2. Programming bootstub")
dfu = PandaDFU(PandaDFU.st_serial_to_dfu_serial(st_serial))
dfu.program_bootstub(code_bootstub)
time.sleep(1)
# flash main code
status("3. Flashing main code")
panda = Panda(st_serial)
panda.flash(code=code_panda)
panda.close()
# flashing ESP
status("4. Flashing ESP (slow!)")
align = lambda x, sz=0x1000: x+"\xFF"*((sz-len(x)) % sz)
esp = ESPROM(st_serial)
esp.connect()
flasher = CesantaFlasher(esp, 230400)
flasher.flash_write(0x0, align(code_boot_15), True)
flasher.flash_write(0x1000, align(code_user1), True)
flasher.flash_write(0x81000, align(code_user2), True)
flasher.flash_write(0x3FE000, "\xFF"*0x1000)
flasher.boot_fw()
del flasher
del esp
time.sleep(1)
# check for connection
status("5. Verifying version")
panda = Panda(st_serial)
my_version = panda.get_version()
print("dongle id: %s" % panda.get_serial()[0])
print(my_version, "should be", version)
assert(str(version) == str(my_version))
# done!
status("6. Success!")
if __name__ == "__main__":
flash_release(*sys.argv[1:])

135
panda/python/isotp.py
View File

@@ -1,135 +0,0 @@
DEBUG = False
def msg(x):
if DEBUG:
print "S:",x.encode("hex")
if len(x) <= 7:
ret = chr(len(x)) + x
else:
assert False
return ret.ljust(8, "\x00")
kmsgs = []
def recv(panda, cnt, addr, nbus):
global kmsgs
ret = []
while len(ret) < cnt:
kmsgs += panda.can_recv()
nmsgs = []
for ids, ts, dat, bus in kmsgs:
if ids == addr and bus == nbus and len(ret) < cnt:
ret.append(dat)
else:
# leave around
nmsgs.append((ids, ts, dat, bus))
kmsgs = nmsgs[-256:]
return map(str, 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
if ord(msg[1])&0xf0 == 0x10:
# first
tlen = ((ord(msg[1]) & 0xf) << 8) | ord(msg[2])
dat = msg[3:]
# 0 block size?
CONTINUE = chr(subaddr) + "\x30" + "\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))
dat += mm[2:]
idx += 1
elif ord(msg[1])&0xf0 == 0x00:
# single
tlen = ord(msg[1]) & 0xf
dat = msg[2:]
else:
print msg.encode("hex")
assert False
return dat[0:tlen]
# **** import below this line ****
def isotp_send(panda, x, addr, bus=0, recvaddr=None, subaddr=None):
if recvaddr is None:
recvaddr = addr+8
if len(x) <= 7 and subaddr is None:
panda.can_send(addr, msg(x), bus)
elif len(x) <= 6 and subaddr is not 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]
x = x[5:]
else:
ss = chr(0x10 + (len(x)>>8)) + chr(len(x)&0xFF) + 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")))
x = x[6:]
else:
sends.append(((chr(0x20 + (idx&0xF)) + x[0:7]).ljust(8, "\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:
for s in sends[:-1]:
panda.can_send(addr, s, 0)
rr = recv(panda, 1, recvaddr, bus)[0]
panda.can_send(addr, sends[-1], 0)
else:
panda.can_send_many([(addr, None, s, 0) for s in sends])
def isotp_recv(panda, addr, bus=0, sendaddr=None, subaddr=None):
if sendaddr is None:
sendaddr = addr-8
if subaddr is not None:
dat = isotp_recv_subaddr(panda, addr, bus, sendaddr, subaddr)
else:
msg = recv(panda, 1, addr, bus)[0]
if ord(msg[0])&0xf0 == 0x10:
# first
tlen = ((ord(msg[0]) & 0xf) << 8) | ord(msg[1])
dat = msg[2:]
# 0 block size?
CONTINUE = "\x30" + "\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))
dat += mm[1:]
idx += 1
elif ord(msg[0])&0xf0 == 0x00:
# single
tlen = ord(msg[0]) & 0xf
dat = msg[1:]
else:
assert False
dat = dat[0:tlen]
if DEBUG:
print "R:",dat.encode("hex")
return dat

27
panda/python/serial.py
View File

@@ -1,27 +0,0 @@
# mimic a python serial port
class PandaSerial(object):
def __init__(self, panda, port, baud):
self.panda = panda
self.port = port
self.panda.set_uart_parity(self.port, 0)
self.panda.set_uart_baud(self.port, baud)
self.buf = ""
def read(self, l=1):
tt = self.panda.serial_read(self.port)
if len(tt) > 0:
#print "R: ", tt.encode("hex")
self.buf += tt
ret = self.buf[0:l]
self.buf = self.buf[l:]
return ret
def write(self, dat):
#print "W: ", dat.encode("hex")
#print ' pigeon_send("' + ''.join(map(lambda x: "\\x%02X" % ord(x), dat)) + '");'
return self.panda.serial_write(self.port, dat)
def close(self):
pass

46
panda/python/update.py
View File

@@ -1,46 +0,0 @@
#!/usr/bin/env python
import os
import time
def ensure_st_up_to_date():
from panda import Panda, PandaDFU, BASEDIR
with open(os.path.join(BASEDIR, "VERSION")) as f:
repo_version = f.read()
repo_version += "-EON" if os.path.isfile('/EON') else "-DEV"
panda = None
panda_dfu = None
should_flash_recover = False
while 1:
# break on normal mode Panda
panda_list = Panda.list()
if len(panda_list) > 0:
panda = Panda(panda_list[0])
break
# flash on DFU mode Panda
panda_dfu = PandaDFU.list()
if len(panda_dfu) > 0:
panda_dfu = PandaDFU(panda_dfu[0])
panda_dfu.recover()
print "waiting for board..."
time.sleep(1)
if panda.bootstub or not panda.get_version().startswith(repo_version):
panda.flash()
if panda.bootstub:
panda.recover()
assert(not panda.bootstub)
version = str(panda.get_version())
print("%s should be %s" % (version, repo_version))
assert(version.startswith(repo_version))
if __name__ == "__main__":
ensure_st_up_to_date()