mawei/StarPilot
1
0
Fork 0
mirror of https://github.com/firestar5683/StarPilot.git synced 2026-07-19 08:12:07 +08:00
Code Issues Packages Projects Releases Wiki Activity

openpilot v0.9.0 release

Browse Source 
date: 2022-11-16T06:35:40
master commit: 58b84fb401a804967aa0dd5ee66fafa90194fd30
This commit is contained in:
Vehicle Researcher
2022-11-16 06:35:41 +00:00
parent 108478216b
commit b47ad523a3
406 changed files with 20722 additions and 9442 deletions
Download Patch File Download Diff File Expand all files Collapse all files
panda/python/__init__.py
+130 -81
View File
@@ -1,15 +1,18 @@
# python library to interface with panda
import datetime
import os
import sys
import time
import usb1
import struct
import hashlib
import usb1
import os
import time
import datetime
import traceback
import sys
import warnings
import logging
from functools import wraps
from typing import Optional
from itertools import accumulate
from .dfu import PandaDFU, MCU_TYPE_F2, MCU_TYPE_F4, MCU_TYPE_H7 # pylint: disable=import-error
from .flash_release import flash_release # noqa pylint: disable=import-error
from .update import ensure_st_up_to_date # noqa pylint: disable=import-error
@@ -19,6 +22,11 @@ from .config import DEFAULT_FW_FN, DEFAULT_H7_FW_FN, SECTOR_SIZES_FX, SECTOR_SIZ
__version__ = '0.0.10'
# setup logging
LOGLEVEL = os.environ.get('LOGLEVEL', 'INFO').upper()
logging.basicConfig(level=LOGLEVEL, format='%(message)s')
BASEDIR = os.path.join(os.path.dirname(os.path.realpath(__file__)), "../")
DEBUG = os.getenv("PANDADEBUG") is not None
@@ -48,7 +56,7 @@ def pack_can_buffer(arr):
snds.append(b'')
idx += 1
#Apply counter to each 64 byte packet
# Apply counter to each 64 byte packet
for idx in range(len(snds)):
tx = b''
counter = 0
@@ -116,6 +124,16 @@ def ensure_can_packet_version(fn):
return fn(self, *args, **kwargs)
return wrapper
def ensure_can_health_packet_version(fn):
@wraps(fn)
def wrapper(self, *args, **kwargs):
if self.can_health_version < self.CAN_HEALTH_PACKET_VERSION:
raise RuntimeError("Panda firmware has outdated CAN health packet definition. Reflash panda firmware.")
elif self.can_health_version > self.CAN_HEALTH_PACKET_VERSION:
raise RuntimeError("Panda python library has outdated CAN health packet definition. Update panda python library.")
return fn(self, *args, **kwargs)
return wrapper
class ALTERNATIVE_EXPERIENCE:
DEFAULT = 0
DISABLE_DISENGAGE_ON_GAS = 1
@@ -171,14 +189,21 @@ class Panda:
HW_TYPE_UNO = b'\x05'
HW_TYPE_DOS = b'\x06'
HW_TYPE_RED_PANDA = b'\x07'
HW_TYPE_RED_PANDA_V2 = b'\x08'
HW_TYPE_TRES = b'\x09'
CAN_PACKET_VERSION = 2
HEALTH_PACKET_VERSION = 8
HEALTH_STRUCT = struct.Struct("<IIIIIIIIBBBBBBBHBBBHIfB")
HEALTH_PACKET_VERSION = 11
CAN_HEALTH_PACKET_VERSION = 3
HEALTH_STRUCT = struct.Struct("<IIIIIIIIIBBBBBBHBBBHfBB")
CAN_HEALTH_STRUCT = struct.Struct("<BIBBBBBBBBIIIIIIHHBBB")
F2_DEVICES = (HW_TYPE_PEDAL, )
F4_DEVICES = (HW_TYPE_WHITE_PANDA, HW_TYPE_GREY_PANDA, HW_TYPE_BLACK_PANDA, HW_TYPE_UNO, HW_TYPE_DOS)
H7_DEVICES = (HW_TYPE_RED_PANDA, )
H7_DEVICES = (HW_TYPE_RED_PANDA, HW_TYPE_RED_PANDA_V2, HW_TYPE_TRES)
INTERNAL_DEVICES = (HW_TYPE_UNO, HW_TYPE_DOS)
HAS_OBD = (HW_TYPE_BLACK_PANDA, HW_TYPE_UNO, HW_TYPE_DOS, HW_TYPE_RED_PANDA, HW_TYPE_RED_PANDA_V2, HW_TYPE_TRES)
CLOCK_SOURCE_MODE_DISABLED = 0
CLOCK_SOURCE_MODE_FREE_RUNNING = 1
@@ -196,9 +221,8 @@ class Panda:
FLAG_HYUNDAI_HYBRID_GAS = 2
FLAG_HYUNDAI_LONG = 4
FLAG_HYUNDAI_CAMERA_SCC = 8
FLAG_HYUNDAI_CANFD_HDA2 = 1
FLAG_HYUNDAI_CANFD_ALT_BUTTONS = 2
FLAG_HYUNDAI_CANFD_HDA2 = 16
FLAG_HYUNDAI_CANFD_ALT_BUTTONS = 32
FLAG_TESLA_POWERTRAIN = 1
FLAG_TESLA_LONG_CONTROL = 2
@@ -211,8 +235,9 @@ class Panda:
FLAG_SUBARU_GEN2 = 1
FLAG_GM_HW_CAM = 1
FLAG_GM_HW_CAM_LONG = 2
def __init__(self, serial: Optional[str] = None, claim: bool = True, disable_checks: bool = True):
def __init__(self, serial: Optional[str] = None, claim: bool = True, spi: bool = False, disable_checks: bool = True):
self._serial = serial
self._disable_checks = disable_checks
@@ -220,9 +245,9 @@ class Panda:
self._bcd_device = None
# connect and set mcu type
self._spi = spi
self.connect(claim)
def close(self):
self._handle.close()
self._handle = None
@@ -230,10 +255,32 @@ class Panda:
def connect(self, claim=True, wait=False):
if self._handle is not None:
self.close()
context = usb1.USBContext()
self._handle = None
if self._spi:
# TODO: move this back. need to wait until next AGNOS build
from .spi import SpiHandle # noqa pylint: disable=import-error
self._handle = SpiHandle()
# TODO implement
self._serial = "SPIDEV"
self.bootstub = False
else:
self.usb_connect(claim=claim, wait=wait)
assert self._handle is not None
self._mcu_type = self.get_mcu_type()
self.health_version, self.can_version, self.can_health_version = self.get_packets_versions()
print("connected")
# disable openpilot's heartbeat checks
if self._disable_checks:
self.set_heartbeat_disabled()
self.set_power_save(0)
def usb_connect(self, claim=True, wait=False):
context = usb1.USBContext()
while 1:
try:
for device in context.getDeviceList(skip_on_error=True):
@@ -266,16 +313,6 @@ class Panda:
break
context = usb1.USBContext() # New context needed so new devices show up
assert self._handle is not None
self._mcu_type = self.get_mcu_type()
self.health_version, self.can_version = self.get_packets_versions()
print("connected")
# disable openpilot's heartbeat checks
if self._disable_checks:
self.set_heartbeat_disabled()
self.set_power_save(0)
def reset(self, enter_bootstub=False, enter_bootloader=False, reconnect=True):
try:
if enter_bootloader:
@@ -410,7 +447,11 @@ class Panda:
for device in context.getDeviceList(skip_on_error=True):
if device.getVendorID() == 0xbbaa and device.getProductID() in (0xddcc, 0xddee):
try:
ret.append(device.getSerialNumber())
serial = device.getSerialNumber()
if len(serial) == 24:
ret.append(serial)
else:
warnings.warn(f"found device with panda descriptors but invalid serial: {serial}", RuntimeWarning)
except Exception:
continue
except Exception:
@@ -430,26 +471,64 @@ class Panda:
"uptime": a[0],
"voltage": a[1],
"current": a[2],
"can_rx_errs": a[3],
"can_send_errs": a[4],
"can_fwd_errs": a[5],
"gmlan_send_errs": a[6],
"faults": a[7],
"ignition_line": a[8],
"ignition_can": a[9],
"controls_allowed": a[10],
"gas_interceptor_detected": a[11],
"car_harness_status": a[12],
"usb_power_mode": a[13],
"safety_tx_blocked": a[3],
"safety_rx_invalid": a[4],
"tx_buffer_overflow": a[5],
"rx_buffer_overflow": a[6],
"gmlan_send_errs": a[7],
"faults": a[8],
"ignition_line": a[9],
"ignition_can": a[10],
"controls_allowed": a[11],
"gas_interceptor_detected": a[12],
"car_harness_status": a[13],
"safety_mode": a[14],
"safety_param": a[15],
"fault_status": a[16],
"power_save_enabled": a[17],
"heartbeat_lost": a[18],
"alternative_experience": a[19],
"blocked_msg_cnt": a[20],
"interrupt_load": a[21],
"fan_power": a[22],
"interrupt_load": a[20],
"fan_power": a[21],
"safety_rx_checks_invalid": a[22],
}
@ensure_can_health_packet_version
def can_health(self, can_number):
LEC_ERROR_CODE = {
0: "No error",
1: "Stuff error",
2: "Form error",
3: "AckError",
4: "Bit1Error",
5: "Bit0Error",
6: "CRCError",
7: "NoChange",
}
dat = self._handle.controlRead(Panda.REQUEST_IN, 0xc2, int(can_number), 0, self.CAN_HEALTH_STRUCT.size)
a = self.CAN_HEALTH_STRUCT.unpack(dat)
return {
"bus_off": a[0],
"bus_off_cnt": a[1],
"error_warning": a[2],
"error_passive": a[3],
"last_error": LEC_ERROR_CODE[a[4]],
"last_stored_error": LEC_ERROR_CODE[a[5]],
"last_data_error": LEC_ERROR_CODE[a[6]],
"last_data_stored_error": LEC_ERROR_CODE[a[7]],
"receive_error_cnt": a[8],
"transmit_error_cnt": a[9],
"total_error_cnt": a[10],
"total_tx_lost_cnt": a[11],
"total_rx_lost_cnt": a[12],
"total_tx_cnt": a[13],
"total_rx_cnt": a[14],
"total_fwd_cnt": a[15],
"can_speed": a[16],
"can_data_speed": a[17],
"canfd_enabled": a[18],
"brs_enabled": a[19],
"canfd_non_iso": a[20],
}
# ******************* control *******************
@@ -486,33 +565,12 @@ class Panda:
# Returns tuple with health packet version and CAN packet/USB packet version
def get_packets_versions(self):
dat = self._handle.controlRead(Panda.REQUEST_IN, 0xdd, 0, 0, 2)
if dat:
a = struct.unpack("BB", dat)
return (a[0], a[1])
dat = self._handle.controlRead(Panda.REQUEST_IN, 0xdd, 0, 0, 3)
if dat and len(dat) == 3:
a = struct.unpack("BBB", dat)
return (a[0], a[1], a[2])
else:
return (0, 0)
def is_white(self):
return self.get_type() == Panda.HW_TYPE_WHITE_PANDA
def is_grey(self):
return self.get_type() == Panda.HW_TYPE_GREY_PANDA
def is_black(self):
return self.get_type() == Panda.HW_TYPE_BLACK_PANDA
def is_pedal(self):
return self.get_type() == Panda.HW_TYPE_PEDAL
def is_uno(self):
return self.get_type() == Panda.HW_TYPE_UNO
def is_dos(self):
return self.get_type() == Panda.HW_TYPE_DOS
def is_red(self):
return self.get_type() == Panda.HW_TYPE_RED_PANDA
return (0, 0, 0)
def get_mcu_type(self):
hw_type = self.get_type()
@@ -525,13 +583,10 @@ class Panda:
return None
def has_obd(self):
return (self.is_uno() or self.is_dos() or self.is_black() or self.is_red())
return self.get_type() in Panda.HAS_OBD
def has_canfd(self) -> bool:
return self.get_type() in Panda.H7_DEVICES
def is_internal(self) -> bool:
return self.get_type() in (Panda.HW_TYPE_UNO, Panda.HW_TYPE_DOS)
def is_internal(self):
return self.get_type() in Panda.INTERNAL_DEVICES
def get_serial(self):
dat = self._handle.controlRead(Panda.REQUEST_IN, 0xd0, 0, 0, 0x20)
@@ -591,14 +646,8 @@ class Panda:
def set_can_data_speed_kbps(self, bus, speed):
self._handle.controlWrite(Panda.REQUEST_OUT, 0xf9, bus, int(speed * 10), b'')
# CAN FD and BRS status
def get_canfd_status(self, bus):
dat = self._handle.controlRead(Panda.REQUEST_IN, 0xfa, bus, 0, 2)
if dat:
a = struct.unpack("BB", dat)
return (a[0], a[1])
else:
return (None, None)
def set_canfd_non_iso(self, bus, non_iso):
self._handle.controlWrite(Panda.REQUEST_OUT, 0xfc, bus, int(non_iso), b'')
def set_uart_baud(self, uart, rate):
self._handle.controlWrite(Panda.REQUEST_OUT, 0xe4, uart, int(rate / 300), b'')