forked from teddy1122/UDS_Python_Prj
-
Notifications
You must be signed in to change notification settings - Fork 0
/
ISO15765_Tp.py
216 lines (208 loc) · 10.4 KB
/
ISO15765_Tp.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
#!/usr/bin/python
#coding=utf-8
#junfeng.luan 2019.07.24
import CAN_If
# import UDS
import time
from enum import IntEnum, Enum
import threading
import File_data
import copy
class CanTpMessageType(IntEnum):
SINGLE_FRAME = 0
FIRST_FRAME = 1
CONSECUTIVE_FRAME = 2
FLOW_CONTROL = 3
class CanTpState(Enum):
IDLE = 0
SEND_SINGLE_FRAME = 1
SEND_FIRST_FRAME = 2
SEND_CONSECUTIVE_FRAME = 3
SEND_FLOW_CONTROL = 4
WAIT_FLOW_CONTROL = 5
WAIT_STMIN_TIMEOUT = 6
WAIT_WAIT_TIMEOUT = 7
RECEIVING_CONSECUTIVE_FRAME = 8
RECEIVED_SINGLE_FARME = 9
RECEIVED_FIRST_FRAME = 10
BUSY = 255
class CAN_Tp(threading.Thread):
def __init__(self):
super().__init__()
self.CAN_TP_Rx_Status = CanTpState.IDLE
self.CAN_TP_Tx_Status = CanTpState.IDLE
self.First_frame_len = 0
self.CAN_TP_Num_of_Bock_Size = 0
self.CAN_TP_Tx_Data = []
self.CAN_TP_Tx_Data_Id = []
self.CAN_TP_Tx_Data_Temp = []
self.CAN_TP_Tx_Data_Id_Temp = []
self.CAN_TP_Tx_Data_len_Temp = 0
self.CAN_TP_Multi_Fram_Inedex = 0
self.CanIf_Receive_Flag = 0
self.CAN_TP_Rx_Data = []
self.CAN_To_DCM_Data = []
self.CAN_TP_Wait_Flow_Control_Flag = 0
self.CAN_TP_Wait_Flow_Control_count = 0
self.Tp_running = True
self.tptxdata = []
print("CAN_Tp Init\n")
def run(self):
print("Tp_MainFunction Running\n")
while self.Tp_running:
time.sleep(0.0005)
self.Rx_MainFunction()
self.Tx_MainFunction()
def Rx_MainFunction(self):
#CAN_TP_Rx_Data = []
receive_data = CAN_If.CAN_Device.can_read()
if self.CAN_TP_Rx_Status == CanTpState.IDLE:
if receive_data!=None and receive_data[0] == File_data.RES_ID:#RES ID
#print("Rx data:{0}".format(receive_data[2]))
receive_frame_type = (receive_data[2][0] & 0xF0) >> 4
if receive_frame_type == CanTpMessageType.SINGLE_FRAME:#singe frame
self.CAN_TP_Rx_Data.append(receive_data[2][1:])
#print("Rx Append:{0}".format(self.CAN_TP_Rx_Data))
#UDS.Uds.Dcm_RxIndication(UDS,CAN_TP_Rx_Data)
self.CAN_To_DCM_Data = copy.deepcopy(self.CAN_TP_Rx_Data)
self.CAN_TP_Rx_Data.clear()
self.CanIf_Receive_Flag = 1
# self.CAN_TP_Rx_Data.clear()
self.CAN_TP_Rx_Status = CanTpState.IDLE
elif receive_frame_type == CanTpMessageType.FIRST_FRAME:#首帧
self.First_frame_len = ((receive_data[2][0]&0x0F)*256 + receive_data[2][1])
self.CAN_TP_Rx_Data.append(receive_data[2][2:])
self.First_frame_len -= 6
self.CAN_TP_Tx_Status = CanTpState.SEND_FLOW_CONTROL
self.CAN_TP_Rx_Status = CanTpState.BUSY#wait for send flow control
elif receive_frame_type == CanTpMessageType.CONSECUTIVE_FRAME:#连续�?
#self.CAN_TP_Rx_Status = CanTpState.RECEIVING_CONSECUTIVE_FRAME
if self.First_frame_len <= 7:
self.CAN_TP_Rx_Data.append(receive_data[2][1:(self.First_frame_len+1)])
self.CAN_TP_Rx_Status = CanTpState.IDLE
#UDS.Uds.Dcm_RxIndication(UDS,CAN_TP_Rx_Data)
self.CAN_To_DCM_Data = copy.deepcopy(self.CAN_TP_Rx_Data)
self.CAN_TP_Rx_Data.clear()
self.CanIf_Receive_Flag = 1
else:
self.CAN_TP_Rx_Data.append(receive_data[2][1:8])
self.First_frame_len -= 7
elif receive_frame_type == CanTpMessageType.FLOW_CONTROL:#流控
if self.CAN_TP_Tx_Status == CanTpState.WAIT_FLOW_CONTROL or self.CAN_TP_Tx_Status == CanTpState.SEND_CONSECUTIVE_FRAME:
self.CAN_TP_Tx_Status = CanTpState.SEND_CONSECUTIVE_FRAME
self.CAN_TP_Num_of_Bock_Size = receive_data[2][1]
#self.CAN_TP_Multi_Fram_Inedex = 0
self.CAN_TP_Wait_Flow_Control_Flag = 0
#print("ISOTP receive flow control frame:block:{0} state:{1}\n".format(self.CAN_TP_Num_of_Bock_Size,self.CAN_TP_Tx_Status))
else:
pass
else:
pass
elif self.CAN_TP_Rx_Status == CanTpState.BUSY:
pass
else:
pass
def Tx_MainFunction(self):
pdu_data = [0xAA,0xAA,0xAA,0xAA,0xAA,0xAA,0xAA,0xAA]
# print("TX Main:{0}".format(self.CAN_TP_Tx_Data))
if self.CAN_TP_Tx_Status == CanTpState.IDLE:
if len(self.CAN_TP_Tx_Data) != 0:
#print("Tx Main data1:{0}\r\n".format(self.CAN_TP_Tx_Data))
self.CAN_TP_Tx_Data_Temp = self.CAN_TP_Tx_Data.pop(0)
self.CAN_TP_Tx_Data_len_Temp = len(self.CAN_TP_Tx_Data_Temp)
self.CAN_TP_Tx_Data_Id_Temp = self.CAN_TP_Tx_Data_Id.pop(0)
#print("Tx Main data2:{0}\r\n".format(self.CAN_TP_Tx_Data))
if self.CAN_TP_Tx_Data_len_Temp <= 7:#单帧发�?
self.CAN_TP_Tx_Status = CanTpState.SEND_SINGLE_FRAME
else:#多帧发送�?�帧
self.CAN_TP_Tx_Status = CanTpState.SEND_FIRST_FRAME
else:
pass
if self.CAN_TP_Tx_Status == CanTpState.SEND_FLOW_CONTROL:
data = (0x30,0x00,0x00,0x00,0x00,0x00,0x00,0x00)
CAN_If.CAN_Device.can_write(data, self.CAN_TP_Tx_Data_Id_Temp, 8)
self.CAN_TP_Rx_Status = CanTpState.IDLE
self.CAN_TP_Tx_Status = CanTpState.IDLE
elif self.CAN_TP_Tx_Status == CanTpState.SEND_SINGLE_FRAME:
pdu_data[0] = self.CAN_TP_Tx_Data_len_Temp
pdu_data[1:(self.CAN_TP_Tx_Data_len_Temp+1)] = self.CAN_TP_Tx_Data_Temp[0:self.CAN_TP_Tx_Data_len_Temp]
CAN_If.CAN_Device.can_write(pdu_data, self.CAN_TP_Tx_Data_Id_Temp, self.CAN_TP_Tx_Data_len_Temp+1)
self.CAN_TP_Tx_Status = CanTpState.IDLE
self.Canif_Tp_Tx_over = 1
print('Single frame send\n')
elif self.CAN_TP_Tx_Status == CanTpState.SEND_FIRST_FRAME:
pdu_data[0] = 0x10 | (self.CAN_TP_Tx_Data_len_Temp//256)
pdu_data[1] = self.CAN_TP_Tx_Data_len_Temp%256
pdu_data[2:8] = self.CAN_TP_Tx_Data_Temp[0:6]
self.CAN_TP_Tx_Data_Temp = self.CAN_TP_Tx_Data_Temp[6:]
self.CAN_TP_Multi_Fram_Inedex = 1
self.CAN_TP_Tx_Data_len_Temp -= 6
CAN_If.CAN_Device.can_write(pdu_data, self.CAN_TP_Tx_Data_Id_Temp, 8)
self.CAN_TP_Tx_Status = CanTpState.WAIT_FLOW_CONTROL
#record time
print('First frame send\n')
elif self.CAN_TP_Tx_Status == CanTpState.SEND_CONSECUTIVE_FRAME:
if ((self.CAN_TP_Num_of_Bock_Size != 0) and (self.CAN_TP_Wait_Flow_Control_Flag == 1)):
#self.CAN_TP_Tx_Status = CanTpState.WAIT_FLOW_CONTROL
#print("Tx MultiFram wait for flow control:{0}\n".format(self.CAN_TP_Multi_Fram_Inedex))
pass
else:
if CAN_If.CAN_Device.can_Tx_confirmation() != 1:
if self.CAN_TP_Tx_Data_len_Temp <= 7:
pdu_data[0] = 0x20|(self.CAN_TP_Multi_Fram_Inedex%16)
pdu_data[1:self.CAN_TP_Tx_Data_len_Temp+1] = self.CAN_TP_Tx_Data_Temp[0:self.CAN_TP_Tx_Data_len_Temp]
self.CAN_TP_Tx_Status = CanTpState.IDLE
self.CAN_TP_Multi_Fram_Inedex = 0
self.CAN_TP_Wait_Flow_Control_count = 0
CAN_If.CAN_Device.can_write(pdu_data, self.CAN_TP_Tx_Data_Id_Temp, 8)
print("Multi frame send end\n")
self.Canif_Tp_Tx_over = 1
else:
pdu_data[0] = 0x20|(self.CAN_TP_Multi_Fram_Inedex%16)
pdu_data[1:8] = self.CAN_TP_Tx_Data_Temp[0:7]
self.CAN_TP_Tx_Data_Temp = self.CAN_TP_Tx_Data_Temp[7:]
self.CAN_TP_Tx_Data_len_Temp -= 7
self.CAN_TP_Multi_Fram_Inedex += 1
self.CAN_TP_Tx_Status = CanTpState.SEND_CONSECUTIVE_FRAME
CAN_If.CAN_Device.can_write(pdu_data, self.CAN_TP_Tx_Data_Id_Temp, 8)
#print("continuous frame:{0}\n".format(self.CAN_TP_Multi_Fram_Inedex))
if (self.CAN_TP_Num_of_Bock_Size != 0) and ((self.CAN_TP_Wait_Flow_Control_count)%(self.CAN_TP_Num_of_Bock_Size) == 0) and (self.CAN_TP_Wait_Flow_Control_count != 0):
self.CAN_TP_Wait_Flow_Control_count = 0
self.CAN_TP_Wait_Flow_Control_Flag = 1
else:
pass
self.CAN_TP_Wait_Flow_Control_count+=1
else:
pass
elif self.CAN_TP_Tx_Status == CanTpState.WAIT_FLOW_CONTROL:
#time out ?
print("Tx MultiFram wait for flow control:{0}\r\n".format(self.CAN_TP_Tx_Status))
pass
else:
pass
def Tp_Transmit(self,data,Fun_Phy):
self.tptxdata = copy.deepcopy(data)
self.CAN_TP_Tx_Data.append(self.tptxdata)
self.CAN_TP_Tx_Data_Id.append(Fun_Phy)
#print("Tp Transmit:{0} len:{1} ID:{2} Len:{3}".format(self.CAN_TP_Tx_Data,len(self.CAN_TP_Tx_Data[0]),self.CAN_TP_Tx_Data_Id,len(self.CAN_TP_Tx_Data_Id)))
def CanIf_Indication_To_Dcm(self):
if self.CanIf_Receive_Flag == 1:
#print("flag:{0} data;{1}".format(self.CanIf_Receive_Flag,self.CAN_To_DCM_Data))
ret = copy.deepcopy(self.CAN_To_DCM_Data.pop(0))
#print("ret:{0}".format(ret))
return (self.CanIf_Receive_Flag,ret)
else:
return (0,0)
def CanIf_Clear_Indica_To_Dcm_Flag(self):
#print("flagaa:{0} dataaa:{1}".format(self.CanIf_Receive_Flag,self.CAN_To_DCM_Data))
if len(self.CAN_To_DCM_Data) == 0:
self.CanIf_Receive_Flag = 0
else:
self.CanIf_Receive_Flag = 1
#self.CAN_TP_Rx_Data.clear()
def CanIf_Tx_Over_To_DCM(self):
return self.Canif_Tp_Tx_over
def CanIf_Tx_Flag_Clear(self):
self.Canif_Tp_Tx_over = 0
def terminate(self):
self.Tp_running = False