From 20cc603922983f154d10b7d56c6b49e5e02997bd Mon Sep 17 00:00:00 2001
From: tao_z <tzj0429@163.com>
Date: Mon, 06 Jun 2022 00:13:05 +0800
Subject: [PATCH] 初步bootloader实现。dbc解析存在解析值不正确问题。
---
widgets/ShifterTool.py | 254 ++++++++++++++++++++++++++++----------------------
1 files changed, 144 insertions(+), 110 deletions(-)
diff --git a/widgets/ShifterTool.py b/widgets/ShifterTool.py
index 020d602..3ea4ac8 100644
--- a/widgets/ShifterTool.py
+++ b/widgets/ShifterTool.py
@@ -1,6 +1,6 @@
-
from ast import Not, Pass
from concurrent.futures import thread
+from curses import flash
from logging import exception
import threading
from typing import List
@@ -14,17 +14,19 @@
from isotp import CanMessage
import queue
from USBCAN import *
-from Shifter import AsciiCodec, ShifterClass
+from Shifter import *
+from styleSheet import *
from mainwindows import Ui_MainWindow
from PySide2 import QtWidgets, QtCore, QtGui
import struct
import time
import datetime
from ShifterDefine import *
-from multiprocessing import Array, Pool, Process, Queue, Value, Pipe
+from multiprocessing import Process, Queue, Value, Pipe
MAX_RCV_NUM = 20
-
+APP_ADDR_LOCATION_OFFSET = 8
+APP_LENGTH_LOCATION_OFFSET = 12
logger = logging.getLogger()
logger.setLevel(logging.DEBUG)
formatter = logging.Formatter(
@@ -33,8 +35,9 @@
consoleHandler = logging.StreamHandler()
consoleHandler.setLevel(logging.DEBUG)
-fileHandler = logging.FileHandler(
- './log/ShiftTool.log', mode='a', encoding='UTF-8')
+fileHandler = logging.FileHandler('./log/ShiftTool.log',
+ mode='a',
+ encoding='UTF-8')
fileHandler.setLevel(logging.NOTSET)
consoleHandler.setFormatter(formatter)
@@ -55,6 +58,7 @@
class HardwreDevice():
+
def __init__(self):
self.device_type = 4
self.channel = 0 # can_index
@@ -63,6 +67,7 @@
class PeriodSendThread(object):
+
def __init__(self, period_func, args=[], kwargs={}):
self._thread = threading.Thread(target=self._run)
self._function = period_func
@@ -117,12 +122,13 @@
self.isotp_layer = isotp_layer
assert isinstance(
- self.isotp_layer, isotp.TransportLayer), 'isotp_layer must be a valid isotp.TransportLayer '
+ self.isotp_layer, isotp.TransportLayer
+ ), 'isotp_layer must be a valid isotp.TransportLayer '
def open(self):
self.exit_requested = False
- self._read_thread = threading.Thread(
- None, target=self.rxthread_task)
+ self._read_thread = threading.Thread(None,
+ target=self.rxthread_task)
self._read_thread.start()
self.opened = True
logger.info('Connection opened')
@@ -150,7 +156,8 @@
if self.mtu is not None:
if len(payload) > self.mtu:
logger.warning(
- "Truncating payload to be set to a length of %d" % (self.mtu))
+ "Truncating payload to be set to a length of %d" %
+ (self.mtu))
payload = payload[0:self.mtu]
# isotp.protocol.TransportLayer uses byte array. udsoncan is strict on bytes format
@@ -172,12 +179,14 @@
if timedout:
raise TimeoutException(
- "Did not receive frame IsoTP Transport layer in time (timeout=%s sec)" % timeout)
+ "Did not receive frame IsoTP Transport layer in time (timeout=%s sec)"
+ % timeout)
if self.mtu is not None:
if frame is not None and len(frame) > self.mtu:
logger.warning(
- "Truncating received payload to a length of %d" % (self.mtu))
+ "Truncating received payload to a length of %d" %
+ (self.mtu))
frame = frame[0:self.mtu]
# isotp.protocol.TransportLayer uses bytearray. udsoncan is strict on bytes format
@@ -204,8 +213,9 @@
while self.isotp_layer.available():
# isotp_conn.send(self.isotp_layer.recv())
self.IsoTPtoUDSQueue.put(self.isotp_layer.recv())
- self.logger.debug("IsoTPtoUDSQueue queue size is now %d" % (
- self.IsoTPtoUDSQueue.qsize()))
+ self.logger.debug(
+ "IsoTPtoUDSQueue queue size is now %d" %
+ (self.IsoTPtoUDSQueue.qsize()))
# time.sleep(self.isotp_layer.sleep_time())
time.sleep(0.001)
@@ -237,8 +247,11 @@
self.DeviceInit()
self.WidgetsInit()
self.ChnInfoUpdate(self._isOpen)
- self.drv_data: bytes = [0x1c, 0x01, 0x06, 0x80,
- 0x1f, 0x01, 0x06, 0x80, 0xfb, 0x0a]
+ self.in_programming = False
+ self.startfromboot = False
+ self.drv_data = [
+ 0x1c, 0x01, 0x06, 0x80, 0x1f, 0x01, 0x06, 0x80, 0xfb, 0x0a
+ ]
self.app_data: bytes = None
def DeviceInit(self):
@@ -275,26 +288,33 @@
'squash_stmin_requirement': False
}
self._isotpaddr_PHYS = isotp.Address(
- isotp.AddressingMode.Normal_11bits, txid=self.shifter.canid.phy_rxId, rxid=self.shifter.canid.phy_txId)
+ isotp.AddressingMode.Normal_11bits,
+ txid=self.shifter.canid.phy_rxId,
+ rxid=self.shifter.canid.phy_txId)
self._isotpaddr_FUNC = isotp.Address(
- isotp.AddressingMode.Normal_11bits, txid=self.shifter.canid.fun_rxId, rxid=self.shifter.canid.phy_txId)
+ isotp.AddressingMode.Normal_11bits,
+ txid=self.shifter.canid.fun_rxId,
+ rxid=self.shifter.canid.phy_txId)
# self._isotpaddr_EPS = isotp.Address(
# isotp.AddressingMode.Normal_11bits, txid=EPS_RX_ID_PHYS, rxid=EPS_TX_ID)
# self._isotpaddr_EPS4wd = isotp.Address(
# isotp.AddressingMode.Normal_11bits, txid=EPS4wd_RX_ID_PHYS, rxid=EPS4wd_TX_ID)
- self.isotp_layer = isotp.TransportLayer(
- rxfn=self.isotp_rcv, txfn=self.isotp_send, address=self._isotpaddr_PHYS, params=self.isotp_params)
+ self.isotp_layer = isotp.TransportLayer(rxfn=self.isotp_rcv,
+ txfn=self.isotp_send,
+ address=self._isotpaddr_PHYS,
+ params=self.isotp_params)
self.conn = ShifterTool.IsoTpConnection(isotp_layer=self.isotp_layer)
self.udsclient = Client(self.conn, request_timeout=2)
self.udsclient.config['security_algo'] = self.SecAlgo
self.udsclient.config['security_algo_params'] = [
- 0x11223344, 0x20AA097B, 0x11223344, 0x70237577]
+ 0x11223344, 0x20AA097B, 0x11223344, 0x70237577
+ ]
self.udsclient.config['data_identifiers'] = self.shifter.did_config
self.udsclient.config['server_address_format'] = 32
self.udsclient.config['server_memorysize_format'] = 32
- def SecAlgo(self, level, seed, params):
+ def SecAlgo(self, level, seed, params=None):
"""
Builds the security key to unlock a security level.
@@ -311,8 +331,8 @@
self.output_key[2] = ((temp_key[1] & 0xFC) >> 2) | (temp_key[0] & 0xC0)
self.output_key[3] = ((temp_key[0] & 0x0F) << 4) | (temp_key[3] & 0x0F)
"""
- temp_key = (seed[0] << 24) | (
- seed[1] << 16) | (seed[2] << 8) | (seed[3])
+ temp_key = (seed[0] << 24) | (seed[1] << 16) | (seed[2] << 8) | (
+ seed[3])
if level == 0x01:
output_key_temp = 0x20AA097B
# output_key_temp = ((((temp_key >> 4) ^ temp_key)
@@ -337,8 +357,10 @@
else:
output_key_temp = temp_key
- output_key = struct.pack('BBBB', (output_key_temp >> 24) & 0xFF, (
- output_key_temp >> 16) & 0xFF, (output_key_temp >> 8) & 0xFF, output_key_temp & 0xFF)
+ output_key = struct.pack('BBBB', (output_key_temp >> 24) & 0xFF,
+ (output_key_temp >> 16) & 0xFF,
+ (output_key_temp >> 8) & 0xFF,
+ output_key_temp & 0xFF)
return output_key
@@ -381,7 +403,7 @@
self.UI.comboBox_2.addItem('USBCAN-II')
# 波特率
- self.UI.comboBox_3.addItems([('%dK' % (i/1000))
+ self.UI.comboBox_3.addItems([('%dK' % (i / 1000))
for i in TIMING_DICT.keys()])
# CHANNEL
self.UI.comboBox_5.addItems(['CH1/3', 'CH2/4'])
@@ -430,7 +452,8 @@
self.UI.radioButton.toggled.connect(self.TestPresentEvent)
self.UI.pushButton_48.clicked.connect(self.start_programming)
- seff.UI.pushButton_46.clicked.connect(self.loadAppfile)
+ self.UI.pushButton_46.clicked.connect(self.loadAppfile)
+ self.UI.radioButton_2.toggled.connect(self.SetFromBootFlag)
def _formatMsgData(self, index, item, received):
'''msg data to list
@@ -463,8 +486,8 @@
else:
data.append('标准帧')
- data.append(' '.join(['0x' +
- '{:0<2x}'.format(a).upper() for a in list(item.data)]))
+ data.append(' '.join(
+ ['0x' + '{:0<2x}'.format(a).upper() for a in list(item.data)]))
return data
def can_thread_stop(self):
@@ -491,7 +514,6 @@
# conn.send(msg[i]) # pipe connection send
self.CANtoIsoTPQueue.put(
msg[i]) # send msg to isotp
- print('收到77A')
self.msgQueue.put(msg[i])
# logger.info('time:'.format(msg[i].timestamp)+'Rx: ID=0x{:0<3x} '.format(msg[i].arbitration_id) + ' '.join(['0x' +
@@ -522,13 +544,15 @@
if self._isOpen.value == 0:
if self._usbcan is None:
self.update_HardwareDevice()
- can_filters = [
- {'can_id': 0x420, 'can_mask': 0xFFFFFFFF}]
+ can_filters = [{'can_id': 0x420, 'can_mask': 0xFFFFFFFF}]
- bitrate = list(TIMING_DICT.keys())[
- self.devicedescription.baudrate]
- self._usbcan = USBCAN(device_type=4, device_index=0,
- can_index=0, bitrate=bitrate, can_filters=can_filters)
+ bitrate = list(
+ TIMING_DICT.keys())[self.devicedescription.baudrate]
+ self._usbcan = USBCAN(device_type=4,
+ device_index=0,
+ can_index=0,
+ bitrate=bitrate,
+ can_filters=can_filters)
if not self._usbcan.InitAndStart():
logger.info("Open usbcan device fail.")
@@ -550,7 +574,8 @@
def update_HardwareDevice(self):
- self.devicedescription.device_type = self.UI.comboBox_2.currentIndex()+3
+ self.devicedescription.device_type = self.UI.comboBox_2.currentIndex(
+ ) + 3
self.devicedescription.channel = self.UI.comboBox_5.currentIndex()
self.devicedescription.baudrate = self.UI.comboBox_3.currentIndex()
@@ -559,13 +584,14 @@
self.UI.pushButton.setText("Close CAN")
elif self.needdisconnect.value == 1:
self.UI.pushButton.setText("Open CAN")
+
# set ui info
def _StartlistenMsgProcess(self):
- self.msgProcess = Process(
- name='pyUSBCANListener',
- target=self._usbcan.ListeningMsg,
- args=(self._isOpen, self.needdisconnect, self.msgQueue, self.sendQueue))
+ self.msgProcess = Process(name='pyUSBCANListener',
+ target=self._usbcan.ListeningMsg,
+ args=(self._isOpen, self.needdisconnect,
+ self.msgQueue, self.sendQueue))
self.msgProcess.daemon = True
self.msgProcess.start()
# 1.5s后检测连接状态，该值可能需要标定
@@ -575,15 +601,19 @@
time.sleep(0.1)
def disp_string(self, out_s):
- dt = datetime.now()
+ dt = datetime.datetime.now()
nowtime_str = dt.strftime('%I:%M:%S') # time
- self.textEdit_2.insertPlainText(nowtime_str + ': ')
- cursor = self.textEdit_2.textCursor()
+ self.UI.textEdit_2.insertPlainText(nowtime_str + ': ')
+ cursor = self.UI.textEdit_2.textCursor()
cursor.movePosition(QtGui.QTextCursor.End)
- self.textEdit_2.setTextCursor(cursor)
- self.textEdit_2.insertPlainText(str(out_s+"\t\n"))
+ self.UI.textEdit_2.setTextCursor(cursor)
+ self.UI.textEdit_2.insertPlainText(str(out_s + "\t\n"))
# self.statusbar.showMessage(str(out))
+ def dispShiftstatus(self):
+ self.UI.pushButton_41.setStyleSheet(
+ Style_dic[self.shifter.UnlockButton])
+ self.UI.pushButton_40.setStyleSheet(Style_dic[self.shifter.Pbutton])
def _updateRootList(self):
_dataSize = self.msgQueue.qsize()
@@ -594,9 +624,11 @@
for i in range(_dataSize):
receiveNum += 1
msg = self.msgQueue.get()
+ self.shifter.FramUnpack(msg.arbitration_id, msg.data)
formateddata.append(self._formatMsgData(
receiveNum, msg, True)) # return a data list
self._insertDataSmooth(data=formateddata, datasize=_dataSize)
+ # self.dispShiftstatus()
def _insertDataSmooth(self, data, datasize):
# row = 6-datasize
@@ -628,8 +660,8 @@
def sessioncontrol_req(self):
sesson = self.UI.comboBox.currentIndex() + 1
- t = threading.Thread(
- target=self.udsclient.change_session, args=(sesson,))
+ t = threading.Thread(target=self.udsclient.change_session,
+ args=(sesson, ))
t.start()
t.join()
# self.udsclient.change_session(sesson)
@@ -682,15 +714,15 @@
def DID_display(self):
# print(self.UI.comboBox_6.currentText())
- self.UI.lineEdit.setText("0x{:0<4x}".
- format((DID_dic[self.UI.comboBox_6.currentText()])))
+ self.UI.lineEdit.setText("0x{:0<4x}".format(
+ (DID_dic[self.UI.comboBox_6.currentText()])))
def ReadDataByID(self):
tempdid = DID_dic[self.UI.comboBox_6.currentText()]
self.UI.lineEdit_2.clear()
data = self.ReadByDID(tempdid)
if data is not None and len(str(data[tempdid])):
- self.UI.lineEdit_2.setText(str(data[tempdid])+' ')
+ self.UI.lineEdit_2.setText(str(data[tempdid]) + ' ')
def communicationControl_req(self):
req = 0x00
@@ -701,8 +733,8 @@
req = 1
else:
req = 3
- self.udsclient.communication_control(
- control_type=0, communication_type=req)
+ self.udsclient.communication_control(control_type=0,
+ communication_type=req)
def StartCalibraiton(self):
startcmd = bytearray([0x0, 0x0, 0x0, 0x0])
@@ -768,35 +800,45 @@
resp_1 = self.udsclient.start_routine(routine_id=routine_id, data=data)
return resp_1.positive
+ def SetFromBootFlag(self):
+ self.startfromboot = self.UI.radioButton_2.isChecked()
+
def pre_programming(self):
+ if not self.startfromboot:
+ self.disp_string('# 预编程步骤')
- self.disp_string('# 预编程步骤')
+ # 进入extended session
+ self.disp_string('>>> 进入扩展模式')
+ response = self.udsclient.change_session(3)
+ if response.positive:
+ self.disp_string('>>> 解锁安全访问')
+ response = self.udsclient.unlock_security_access(9)
- # 进入extended session
- self.disp_string('>>> 进入扩展模式')
- response = self.udsclient.change_session(3)
- if response.positive:
- # 检查编程条件
- self.disp_string('>>> 检查编程条件')
- response = self.udsclient.start_routine(0xff02)
- if response.positive:
- # 关闭DTC的存储
- self.disp_string('>>> 关闭DTC的存储')
- response = self.udsclient.control_dtc_setting(2)
- # print(response)
- if response.positive:
- # 关闭与诊断无关的报文
- self.disp_string('>>> 关闭与诊断无关的报文')
- response = self.udsclient.communication_control(0x03, 0x01)
- else:
- self.disp_string('>>> 预编程失败')
+ if response.positive:
+ # 检查编程条件
+ self.disp_string('>>> 检查编程条件')
+ response = self.udsclient.start_routine(0xff02)
+ if response.positive:
+ # 关闭DTC的存储
+ self.disp_string('>>> 关闭DTC的存储')
+ response = self.udsclient.control_dtc_setting(2)
+ # print(response)
+ if response.positive:
+ # 关闭与诊断无关的报文
+ self.disp_string('>>> 关闭与诊断无关的报文')
+ response = self.udsclient.communication_control(0x03, 0x01)
+ else:
+ self.disp_string('>>> 预编程失败')
- return response.positive
+ return response.positive
# print(response)
def main_programming(self):
-
self.disp_string('# 主编程步骤')
+
+ if self.startfromboot:
+ self.disp_string('>>> 进入扩展模式')
+ response = self.udsclient.change_session(3)
# 进入programming session
self.disp_string('>>> 进入编程模式')
@@ -807,49 +849,40 @@
self.disp_string('>>> 安全访问')
response = self.udsclient.unlock_security_access(9)
# print(response)
-
+ if response.positive:
+ self.disp_string('>>> 请求下载驱动文件')
+ address = 0x20004600
+ memorysize = 10
+ memory_location = MemoryLocation(address, memorysize, 32, 32)
+ response = self.udsclient.request_download(memory_location)
# print(response)
if response.positive:
# 发送driver文件
self.disp_string('>>> 发送driver文件')
drv_file = self.drv_data
- address = drv_file[16] << 24 | drv_file[17] << 16 | drv_file[
- 18] << 8 | drv_file[19] << 0
- memorysize = drv_file[20] << 24 | drv_file[21] << 16 | drv_file[
- 22] << 8 | drv_file[23] << 0
- memory_location = MemoryLocation(address, memorysize, 32, 32)
- response = self.udsclient.request_download(memory_location)
- # print(response)
-
- max_length = response.service_data.max_length
-
- # 有效数据长度, 去除sid和sequence两个字节
- payload_length = max_length - 2
-
- count = (len(drv_file) + payload_length - 1) // payload_length
-
- base = self.get_progressbar_pos()
- for i in range(count):
- start = i * payload_length
- end = start + payload_length
- response = self.udsclient.transfer_data((i + 1) % 256,
- drv_file[start:end])
- self.set_progressbar_pos(base + end)
+ response = self.udsclient.transfer_data(1, bytes(drv_file))
+ self.set_progressbar_pos(len(drv_file))
# print(response)
if response.positive:
+ self.disp_string('>>> 发送驱动结束，请求推出')
response = self.udsclient.request_transfer_exit()
# print(response)
if response.positive:
# driver文件完整性检验
self.disp_string('>>> driver文件完整性检验')
- response = self.udsclient.start_routine(0xf001, drv_file[0:4])
+ drivecrc = bytes([0x9B, 0x39, 0xf2, 0xec])
+ response = self.udsclient.start_routine(0xf001, drivecrc)
# print(response)
if response.positive:
app_file = self.app_data
- address = app_file[16] << 24 | app_file[17] << 16 | app_file[
- 18] << 8 | app_file[19] << 0
- memorysize = app_file[20] << 24 | app_file[21] << 16 | app_file[
- 22] << 8 | app_file[23] << 0
+ address = app_file[APP_ADDR_LOCATION_OFFSET] << 24 | app_file[
+ APP_ADDR_LOCATION_OFFSET + 1] << 16 | app_file[
+ APP_ADDR_LOCATION_OFFSET +
+ 2] << 8 | app_file[APP_ADDR_LOCATION_OFFSET + 3] << 0
+ memorysize = app_file[APP_LENGTH_LOCATION_OFFSET] << 24 | app_file[
+ APP_LENGTH_LOCATION_OFFSET + 1] << 16 | app_file[
+ APP_LENGTH_LOCATION_OFFSET +
+ 2] << 8 | app_file[APP_LENGTH_LOCATION_OFFSET + 3] << 0
memory_location = MemoryLocation(address, memorysize, 32, 32)
# 删除app存储空间
@@ -864,11 +897,11 @@
# 发送app文件
self.disp_string('>>> 发送app文件')
response = self.udsclient.request_download(memory_location)
- # print(response)
+ # print(response)
max_length = response.service_data.max_length
- # 有效数据长度, 去除sid和sequence两个字节
+ # 有效数据长度, 去除sid和sequence两个字节
payload_length = max_length - 2
count = (len(app_file) + payload_length - 1) // payload_length
@@ -898,12 +931,11 @@
# 回到default session
self.disp_string('>>> 回到默认模式')
- self.udsclient.suppress_positive_response.enabled = True
response = self.udsclient.change_session(1)
return response.positive
def start_programming(self):
-
+ self.UI.pushButton_48.setDisabled(True)
self.set_progressbar_pos(0)
t1 = time.time()
@@ -915,13 +947,14 @@
self.post_programming()
except Exception as e:
- self.disp_string('%s' % e, '#ff0000')
+ self.disp_string('%s' % e)
+ # self.UI.pushButton_48.setDisabled(False)
t2 = time.time()
self.disp_string('finished in %.2f sec' % (t2 - t1))
self.in_programming = False
- self.UI.pushButton_48.setDisabled(True)
+ self.UI.pushButton_48.setDisabled(False)
def set_progressbar_pos(self, pos):
@@ -944,7 +977,8 @@
return
fileName, _ = QtWidgets.QFileDialog.getOpenFileName(
- self, 'comboBox_4', '.', 'All Files (*);;Bin Files (*.bin);;Hex Files (*.hex)')
+ None, 'comboBox_4', '.',
+ 'All Files (*);;Bin Files (*.bin);;Hex Files (*.hex)')
if fileName != '':
self.UI.comboBox_4.addItem(fileName)
--
Gitblit v1.8.0