forked from peterh/pyobd
-
Notifications
You must be signed in to change notification settings - Fork 30
/
obd_io.py
326 lines (267 loc) · 11.3 KB
/
obd_io.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
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
#!/usr/bin/env python
###########################################################################
# odb_io.py
#
# Copyright 2004 Donour Sizemore ([email protected])
# Copyright 2009 Secons Ltd. (www.obdtester.com)
#
# This file is part of pyOBD.
#
# pyOBD is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation; either version 2 of the License, or
# (at your option) any later version.
#
# pyOBD is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with pyOBD; if not, write to the Free Software
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
###########################################################################
import serial
import string
import time
from math import ceil
from datetime import datetime
import obd_sensors
from obd_sensors import hex_to_int
GET_DTC_COMMAND = "03"
CLEAR_DTC_COMMAND = "04"
GET_FREEZE_DTC_COMMAND = "07"
from debugEvent import debug_display
#__________________________________________________________________________
def decrypt_dtc_code(code):
"""Returns the 5-digit DTC code from hex encoding"""
dtc = []
current = code
for i in range(0,3):
if len(current)<4:
raise "Tried to decode bad DTC: %s" % code
tc = obd_sensors.hex_to_int(current[0]) #typecode
tc = tc >> 2
if tc == 0:
type = "P"
elif tc == 1:
type = "C"
elif tc == 2:
type = "B"
elif tc == 3:
type = "U"
else:
raise tc
dig1 = str(obd_sensors.hex_to_int(current[0]) & 3)
dig2 = str(obd_sensors.hex_to_int(current[1]))
dig3 = str(obd_sensors.hex_to_int(current[2]))
dig4 = str(obd_sensors.hex_to_int(current[3]))
dtc.append(type+dig1+dig2+dig3+dig4)
current = current[4:]
return dtc
#__________________________________________________________________________
class OBDPort:
""" OBDPort abstracts all communication with OBD-II device."""
def __init__(self,portnum,_notify_window,SERTIMEOUT,RECONNATTEMPTS):
"""Initializes port by resetting device and gettings supported PIDs. """
# These should really be set by the user.
baud = 38400
databits = 8
par = serial.PARITY_NONE # parity
sb = 1 # stop bits
to = SERTIMEOUT
self.ELMver = "Unknown"
self.State = 1 #state SERIAL is 1 connected, 0 disconnected (connection failed)
self.port = None
self._notify_window=_notify_window
debug_display(self._notify_window, 1, "Opening interface (serial port)")
try:
self.port = serial.Serial(portnum,baud, \
parity = par, stopbits = sb, bytesize = databits,timeout = to)
except serial.SerialException as e:
print e
self.State = 0
return None
debug_display(self._notify_window, 1, "Interface successfully " + self.port.portstr + " opened")
debug_display(self._notify_window, 1, "Connecting to ECU...")
try:
self.send_command("atz") # initialize
time.sleep(1)
except serial.SerialException:
self.State = 0
return None
self.ELMver = self.get_result()
if(self.ELMver is None):
self.State = 0
return None
debug_display(self._notify_window, 2, "atz response:" + self.ELMver)
self.send_command("ate0") # echo off
debug_display(self._notify_window, 2, "ate0 response:" + self.get_result())
self.send_command("0100")
ready = self.get_result()
if(ready is None):
self.State = 0
return None
debug_display(self._notify_window, 2, "0100 response:" + ready)
return None
def close(self):
""" Resets device and closes all associated filehandles"""
if (self.port!= None) and self.State==1:
self.send_command("atz")
self.port.close()
self.port = None
self.ELMver = "Unknown"
def send_command(self, cmd):
"""Internal use only: not a public interface"""
if self.port:
self.port.flushOutput()
self.port.flushInput()
for c in cmd:
self.port.write(c)
self.port.write("\r\n")
#debug_display(self._notify_window, 3, "Send command:" + cmd)
def interpret_result(self,code):
"""Internal use only: not a public interface"""
# Code will be the string returned from the device.
# It should look something like this:
# '41 11 0 0\r\r'
# 9 seems to be the length of the shortest valid response
if len(code) < 7:
#raise Exception("BogusCode")
print "boguscode?"+code
# get the first thing returned, echo should be off
code = string.split(code, "\r")
code = code[0]
#remove whitespace
code = string.split(code)
code = string.join(code, "")
#cables can behave differently
if code[:6] == "NODATA": # there is no such sensor
return "NODATA"
# first 4 characters are code from ELM
code = code[4:]
return code
def get_result(self):
"""Internal use only: not a public interface"""
#time.sleep(0.01)
repeat_count = 0
if self.port is not None:
buffer = ""
while 1:
c = self.port.read(1)
if len(c) == 0:
if(repeat_count == 5):
break
print "Got nothing\n"
repeat_count = repeat_count + 1
continue
if c == '\r':
continue
if c == ">":
break;
if buffer != "" or c != ">": #if something is in buffer, add everything
buffer = buffer + c
#debug_display(self._notify_window, 3, "Get result:" + buffer)
if(buffer == ""):
return None
return buffer
else:
debug_display(self._notify_window, 3, "NO self.port!")
return None
# get sensor value from command
def get_sensor_value(self,sensor):
"""Internal use only: not a public interface"""
cmd = sensor.cmd
self.send_command(cmd)
data = self.get_result()
if data:
data = self.interpret_result(data)
if data != "NODATA":
data = sensor.value(data)
else:
return "NORESPONSE"
return data
# return string of sensor name and value from sensor index
def sensor(self , sensor_index):
"""Returns 3-tuple of given sensors. 3-tuple consists of
(Sensor Name (string), Sensor Value (string), Sensor Unit (string) ) """
sensor = obd_sensors.SENSORS[sensor_index]
r = self.get_sensor_value(sensor)
return (sensor.name,r, sensor.unit)
def sensor_names(self):
"""Internal use only: not a public interface"""
names = []
for s in obd_sensors.SENSORS:
names.append(s.name)
return names
def get_tests_MIL(self):
statusText=["Unsupported","Supported - Completed","Unsupported","Supported - Incompleted"]
statusRes = self.sensor(1)[1] #GET values
statusTrans = [] #translate values to text
statusTrans.append(str(statusRes[0])) #DTCs
if statusRes[1]==0: #MIL
statusTrans.append("Off")
else:
statusTrans.append("On")
for i in range(2,len(statusRes)): #Tests
statusTrans.append(statusText[statusRes[i]])
return statusTrans
#
# fixme: j1979 specifies that the program should poll until the number
# of returned DTCs matches the number indicated by a call to PID 01
#
def get_dtc(self):
"""Returns a list of all pending DTC codes. Each element consists of
a 2-tuple: (DTC code (string), Code description (string) )"""
dtcLetters = ["P", "C", "B", "U"]
r = self.sensor(1)[1] #data
dtcNumber = r[0]
mil = r[1]
DTCCodes = []
print "Number of stored DTC:" + str(dtcNumber) + " MIL: " + str(mil)
# get all DTC, 3 per mesg response
for i in range(0, ((dtcNumber+2)/3)):
self.send_command(GET_DTC_COMMAND)
res = self.get_result()
print "DTC result:" + res
for i in range(0, 3):
val1 = hex_to_int(res[3+i*6:5+i*6])
val2 = hex_to_int(res[6+i*6:8+i*6]) #get DTC codes from response (3 DTC each 2 bytes)
val = (val1<<8)+val2 #DTC val as int
if val==0: #skip fill of last packet
break
DTCStr=dtcLetters[(val&0xC000)>14]+str((val&0x3000)>>12)+str((val&0x0f00)>>8)+str((val&0x00f0)>>4)+str(val&0x000f)
DTCCodes.append(["Active",DTCStr])
#read mode 7
self.send_command(GET_FREEZE_DTC_COMMAND)
res = self.get_result()
if res[:7] == "NO DATA": #no freeze frame
return DTCCodes
print "DTC freeze result:" + res
for i in range(0, 3):
val1 = hex_to_int(res[3+i*6:5+i*6])
val2 = hex_to_int(res[6+i*6:8+i*6]) #get DTC codes from response (3 DTC each 2 bytes)
val = (val1<<8)+val2 #DTC val as int
if val==0: #skip fill of last packet
break
DTCStr=dtcLetters[(val&0xC000)>14]+str((val&0x3000)>>12)+str((val&0x0f00)>>8)+str((val&0x00f0)>>4)+str(val&0x000f)
DTCCodes.append(["Passive",DTCStr])
return DTCCodes
def clear_dtc(self):
"""Clears all DTCs and freeze frame data"""
self.send_command(CLEAR_DTC_COMMAND)
r = self.get_result()
return r
def log(self, sensor_index, filename):
file = open(filename, "w")
start_time = time.time()
if file:
data = self.sensor(sensor_index)
file.write("%s \t%s(%s)\n" % \
("Time", string.strip(data[0]), data[2]))
while 1:
now = time.time()
data = self.sensor(sensor_index)
line = "%.6f,\t%s\n" % (now - start_time, data[1])
file.write(line)
file.flush()