-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathmain.py
More file actions
2286 lines (1848 loc) · 89.8 KB
/
main.py
File metadata and controls
2286 lines (1848 loc) · 89.8 KB
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
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
934
935
936
937
938
939
940
941
942
943
944
945
946
947
948
949
950
951
952
953
954
955
956
957
958
959
960
961
962
963
964
965
966
967
968
969
970
971
972
973
974
975
976
977
978
979
980
981
982
983
984
985
986
987
988
989
990
991
992
993
994
995
996
997
998
999
1000
"""
ELM327 WiFi OBD-II Professional Scanner Tool
A comprehensive command-line interface for ELM327 OBD-II scanners
Author: Advanced OBD Tools
Version: 1.0.0
https://www.dashlogic.com/docs/technical/obdii_pids
"""
import os
import socket
import time
import threading
import json
import csv
import logging
import re
import sys
from datetime import datetime
from typing import Dict, List, Optional, Tuple, Any
from dataclasses import dataclass
from enum import Enum
import time
from collections import deque
import pids
import re
from typing import List, Optional
# Optional matplotlib import for graphing
try:
import matplotlib.pyplot as plt
import matplotlib.dates as mdates
from matplotlib.animation import FuncAnimation
MATPLOTLIB_AVAILABLE = True
except ImportError:
MATPLOTLIB_AVAILABLE = False
print("Warning: matplotlib not available. Graphing features disabled.")
with open("dtc.json", "r") as f:
dtc_dict = json.load(f)
_SYSTEM_STRINGS = ("SEARCHING", "BUSINIT", "BUS INIT", "STOPPED", "NODATA", "NO DATA")
# Configure logging
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(levelname)s - %(message)s',
handlers=[
logging.FileHandler('elm327_scanner.log'),
logging.StreamHandler()
]
)
class OBDProtocol(Enum):
"""OBD Protocol definitions"""
AUTO = "0"
SAE_J1850_PWM = "1"
SAE_J1850_VPW = "2"
ISO_9141_2 = "3"
ISO_14230_4_KWP = "4"
ISO_14230_4_KWP_FAST = "5"
ISO_15765_4_CAN_11_500 = "6"
ISO_15765_4_CAN_29_500 = "7"
ISO_15765_4_CAN_11_250 = "8"
ISO_15765_4_CAN_29_250 = "9"
SAE_J1939_CAN = "A"
@dataclass
class SensorReading:
"""Data structure for sensor readings"""
pid: str
name: str
signals: [{
'value': str,
'path': str,
'name': str,
'unit': str,
}] # type: ignore
timestamp: datetime
raw_data: str
@dataclass
class DTCInfo:
"""Data structure for Diagnostic Trouble Codes"""
code: str
description: str
status: str
class PIDDatabase:
"""Complete PID database with formulas and descriptions"""
# (PIDs 0114-011B)
PIDS = pids.PID_DICT
# {
# # Mode 01 PIDs
# "0100": {"name": "PIDs supported [01-20]", "unit": "bit", "formula": lambda x: x},
# "0101": {"name": "Monitor status since DTCs cleared", "unit": "bit", "formula": lambda x: x},
# "0102": {"name": "Freeze DTC", "unit": "code", "formula": lambda x: x},
# "0103": {"name": "Fuel system status", "unit": "status", "formula": lambda x: x},
# "0104": {"name": "Calculated engine load", "unit": "%", "formula": lambda x: x[0] * 100 / 255},
# "0105": {"name": "Engine coolant temperature", "unit": "°C", "formula": lambda x: x[0] - 40},
# "0106": {"name": "Short term fuel trim—Bank 1", "unit": "%", "formula": lambda x: (x[0] - 128) * 100 / 128},
# "0107": {"name": "Long term fuel trim—Bank 1", "unit": "%", "formula": lambda x: (x[0] - 128) * 100 / 128},
# "0108": {"name": "Short term fuel trim—Bank 2", "unit": "%", "formula": lambda x: (x[0] - 128) * 100 / 128},
# "0109": {"name": "Long term fuel trim—Bank 2", "unit": "%", "formula": lambda x: (x[0] - 128) * 100 / 128},
# "010A": {"name": "Fuel rail pressure", "unit": "kPa", "formula": lambda x: x[0] * 3},
# "010B": {"name": "Intake manifold absolute pressure", "unit": "kPa", "formula": lambda x: x[0]},
# "010C": {"name": "Engine RPM", "unit": "rpm", "formula": lambda x: (x[0] * 256 + x[1]) / 4},
# "010D": {"name": "Vehicle speed", "unit": "km/h", "formula": lambda x: x[0]},
# "010E": {"name": "Timing advance", "unit": "°", "formula": lambda x: (x[0] - 128) / 2},
# "010F": {"name": "Intake air temperature", "unit": "°C", "formula": lambda x: x[0] - 40},
# "0110": {"name": "MAF air flow rate", "unit": "g/s", "formula": lambda x: (x[0] * 256 + x[1]) / 100},
# "0111": {"name": "Throttle position", "unit": "%", "formula": lambda x: x[0] * 100 / 255},
# "0112": {"name": "Commanded secondary air status", "unit": "status", "formula": lambda x: x},
# "0113": {"name": "Oxygen sensors present", "unit": "bit", "formula": lambda x: x},
# "0114": {"name": "Oxygen sensor (narrowband) bank 1 sensor 1", "unit": "V", "formula": lambda x: x[0] / 200},
# "0115": {"name": "Oxygen sensor (narrowband) bank 1 sensor 2", "unit": "V", "formula": lambda x: x[0] / 200},
# "0116": {"name": "Oxygen sensor (narrowband) bank 1 sensor 3", "unit": "V", "formula": lambda x: x[0] / 200},
# "0117": {"name": "Oxygen sensor (narrowband) bank 1 sensor 4", "unit": "V", "formula": lambda x: x[0] / 200},
# "0118": {"name": "Oxygen sensor (narrowband) bank 2 sensor 1", "unit": "V", "formula": lambda x: x[0] / 200},
# "0119": {"name": "Oxygen sensor (narrowband) bank 2 sensor 2", "unit": "V", "formula": lambda x: x[0] / 200},
# "011A": {"name": "Oxygen sensor (narrowband) bank 2 sensor 3", "unit": "V", "formula": lambda x: x[0] / 200},
# "011B": {"name": "Oxygen sensor (narrowband) bank 2 sensor 4", "unit": "V", "formula": lambda x: x[0] / 200},
# "011C": {"name": "OBD standards", "unit": "code", "formula": lambda x: x},
# "011D": {"name": "O2 sensors present (4 banks)", "unit": "bit", "formula": lambda x: x},
# "011E": {"name": "Auxiliary input status", "unit": "bit", "formula": lambda x: x},
# "011F": {"name": "Run time since engine start", "unit": "s", "formula": lambda x: x[0] * 256 + x[1]},
# "0120": {"name": "PIDs supported [21-40]", "unit": "bit", "formula": lambda x: x},
# "0121": {"name": "Distance traveled with malfunction indicator lamp (MIL) on", "unit": "km", "formula": lambda x: x[0] * 256 + x[1]},
# "0122": {"name": "Fuel Rail Pressure", "unit": "kPa", "formula": lambda x: ((x[0] * 256 + x[1]) * 0.079)},
# "0123": {"name": "Fuel Rail Gauge Pressure", "unit": "kPa", "formula": lambda x: ((x[0] * 256 + x[1]) * 10)},
# "0124": {"name": "Oxygen sensor (wideband) bank 1 sensor 1", "unit": "ratio", "formula": lambda x: ((x[0] * 256 + x[1]) / 32768)},
# "0125": {"name": "Oxygen sensor (wideband) bank 1 sensor 2", "unit": "ratio", "formula": lambda x: ((x[0] * 256 + x[1]) / 32768)},
# "0126": {"name": "Oxygen sensor (wideband) bank 1 sensor 3", "unit": "ratio", "formula": lambda x: ((x[0] * 256 + x[1]) / 32768)},
# "0127": {"name": "Oxygen sensor (wideband) bank 1 sensor 4", "unit": "ratio", "formula": lambda x: ((x[0] * 256 + x[1]) / 32768)},
# "0128": {"name": "Oxygen sensor (wideband) bank 2 sensor 1", "unit": "ratio", "formula": lambda x: ((x[0] * 256 + x[1]) / 32768)},
# "0129": {"name": "Oxygen sensor (wideband) bank 2 sensor 2", "unit": "ratio", "formula": lambda x: ((x[0] * 256 + x[1]) / 32768)},
# "012A": {"name": "Oxygen sensor (wideband) bank 2 sensor 3", "unit": "ratio", "formula": lambda x: ((x[0] * 256 + x[1]) / 32768)},
# "012B": {"name": "Oxygen sensor (wideband) bank 2 sensor 4", "unit": "ratio", "formula": lambda x: ((x[0] * 256 + x[1]) / 32768)},
# "012C": {"name": "Commanded EGR", "unit": "%", "formula": lambda x: x[0] * 100 / 255},
# "012D": {"name": "EGR Error", "unit": "%", "formula": lambda x: (x[0] - 128) * 100 / 128},
# "012E": {"name": "Commanded evaporative purge", "unit": "%", "formula": lambda x: x[0] * 100 / 255},
# "012F": {"name": "Fuel Tank Level Input", "unit": "%", "formula": lambda x: x[0] * 100 / 255},
# "0130": {"name": "Warm-ups since codes cleared", "unit": "count", "formula": lambda x: x[0]},
# "0131": {"name": "Distance traveled since codes cleared", "unit": "km", "formula": lambda x: x[0] * 256 + x[1]},
# "0132": {"name": "Evap. System Vapor Pressure", "unit": "Pa", "formula": lambda x: ((x[0] * 256 + x[1]) / 4)},
# "0133": {"name": "Absolut load value", "unit": "%", "formula": lambda x: (x[0] * 256 + x[1]) * 100 / 255},
# "0134": {"name": "Command equivalence ratio", "unit": "ratio", "formula": lambda x: (x[0] * 256 + x[1]) / 32768},
# "0135": {"name": "Relative throttle position", "unit": "%", "formula": lambda x: x[0] * 100 / 255},
# "0136": {"name": "Ambient air temperature", "unit": "°C", "formula": lambda x: x[0] - 40},
# "0137": {"name": "Absolute throttle position B", "unit": "%", "formula": lambda x: x[0] * 100 / 255},
# "0138": {"name": "Absolute throttle position C", "unit": "%", "formula": lambda x: x[0] * 100 / 255},
# "0139": {"name": "Accelerator pedal position D", "unit": "%", "formula": lambda x: x[0] * 100 / 255},
# "013A": {"name": "Accelerator pedal position E", "unit": "%", "formula": lambda x: x[0] * 100 / 255},
# "013B": {"name": "Accelerator pedal position F", "unit": "%", "formula": lambda x: x[0] * 100 / 255},
# "013C": {"name": "Commanded throttle actuator", "unit": "%", "formula": lambda x: x[0] * 100 / 255},
# "013D": {"name": "Time run with MIL on", "unit": "min", "formula": lambda x: x[0] * 256 + x[1]},
# "013E": {"name": "Time since trouble codes cleared", "unit": "min", "formula": lambda x: x[0] * 256 + x[1]},
# "013F": {"name": "Maximum value for equivalence ratio, oxygen sensor voltage, oxygen sensor current, and intake manifold absolute pressure", "unit": "mixed", "formula": lambda x: x},
# "0140": {"name": "PIDs supported [41-60]", "unit": "bit", "formula": lambda x: x},
# "0141": {"name": "Maximum value for air flow rate from mass air flow sensor", "unit": "g/s", "formula": lambda x: x[0] * 10},
# "0142": {"name": "Fuel Type", "unit": "code", "formula": lambda x: x[0]},
# "0143": {"name": "Ethanol fuel %", "unit": "%", "formula": lambda x: x[0] * 100 / 255},
# "0144": {"name": "Absolute Evap system Vapor Pressure", "unit": "kPa", "formula": lambda x: (x[0] * 256 + x[1]) / 200},
# "0145": {"name": "Evap system vapor pressure", "unit": "Pa", "formula": lambda x: x[0] * 256 + x[1] - 32767},
# "0146": {"name": "Short term secondary oxygen sensor trim, A: bank 1, B: bank 3", "unit": "%", "formula": lambda x: (x[0] - 128) * 100 / 128},
# "0147": {"name": "Long term secondary oxygen sensor trim, A: bank 1, B: bank 3", "unit": "%", "formula": lambda x: (x[0] - 128) * 100 / 128},
# "0148": {"name": "Short term secondary oxygen sensor trim, A: bank 2, B: bank 4", "unit": "%", "formula": lambda x: (x[0] - 128) * 100 / 128},
# "0149": {"name": "Long term secondary oxygen sensor trim, A: bank 2, B: bank 4", "unit": "%", "formula": lambda x: (x[0] - 128) * 100 / 128},
# "014A": {"name": "Fuel rail absolute pressure", "unit": "kPa", "formula": lambda x: (x[0] * 256 + x[1]) * 10},
# "014B": {"name": "Relative accelerator pedal position", "unit": "%", "formula": lambda x: x[0] * 100 / 255},
# "014C": {"name": "Hybrid battery pack remaining life", "unit": "%", "formula": lambda x: x[0] * 100 / 255},
# "014D": {"name": "Engine oil temperature", "unit": "°C", "formula": lambda x: x[0] - 40},
# "014E": {"name": "Fuel injection timing", "unit": "°", "formula": lambda x: (x[0] * 256 + x[1] - 26880) / 128},
# "014F": {"name": "Engine fuel rate", "unit": "L/h", "formula": lambda x: (x[0] * 256 + x[1]) / 20},
# "0150": {"name": "Emission requirements", "unit": "bit", "formula": lambda x: x},
# "0151": {"name": "PIDs supported [51-70]", "unit": "bit", "formula": lambda x: x},
# "0152": {"name": "Driver's demand engine - percent torque", "unit": "%", "formula": lambda x: x[0] - 125},
# "0153": {"name": "Actual engine - percent torque", "unit": "%", "formula": lambda x: x[0] - 125},
# "0154": {"name": "Engine reference torque", "unit": "Nm", "formula": lambda x: x[0] * 256 + x[1]},
# }
@classmethod
def get_pid_info(cls, pid: str) -> Dict[str, Any]:
"""Get PID information"""
return cls.PIDS.get(pid.upper(), {
"name": f"Unknown PID {pid}",
"unit": "unknown",
"formula": lambda x: x
})
class DTC_Database:
"""Diagnostic Trouble Code database"""
DTC_PREFIXES = {
'P0': 'Powertrain - Generic',
'P1': 'Powertrain - Manufacturer',
'P2': 'Powertrain - Generic',
'P3': 'Powertrain - Reserved',
'B0': 'Body - Generic',
'B1': 'Body - Manufacturer',
'B2': 'Body - Generic',
'B3': 'Body - Reserved',
'C0': 'Chassis - Generic',
'C1': 'Chassis - Manufacturer',
'C2': 'Chassis - Generic',
'C3': 'Chassis - Reserved',
'U0': 'Network - Generic',
'U1': 'Network - Manufacturer',
'U2': 'Network - Generic',
'U3': 'Network - Reserved',
}
COMMON_DTCS = dtc_dict
# {
# 'P0100': 'Mass or Volume Air Flow Circuit Malfunction',
# 'P0101': 'Mass or Volume Air Flow Circuit Range/Performance Problem',
# 'P0102': 'Mass or Volume Air Flow Circuit Low Input',
# 'P0103': 'Mass or Volume Air Flow Circuit High Input',
# 'P0104': 'Mass or Volume Air Flow Circuit Intermittent',
# 'P0105': 'Manifold Absolute Pressure/Barometric Pressure Circuit Malfunction',
# 'P0106': 'Manifold Absolute Pressure/Barometric Pressure Circuit Range/Performance Problem',
# 'P0107': 'Manifold Absolute Pressure/Barometric Pressure Circuit Low Input',
# 'P0108': 'Manifold Absolute Pressure/Barometric Pressure Circuit High Input',
# 'P0109': 'Manifold Absolute Pressure/Barometric Pressure Circuit Intermittent',
# 'P0110': 'Intake Air Temperature Circuit Malfunction',
# 'P0111': 'Intake Air Temperature Circuit Range/Performance Problem',
# 'P0112': 'Intake Air Temperature Circuit Low Input',
# 'P0113': 'Intake Air Temperature Circuit High Input',
# 'P0114': 'Intake Air Temperature Circuit Intermittent',
# 'P0115': 'Engine Coolant Temperature Circuit Malfunction',
# 'P0116': 'Engine Coolant Temperature Circuit Range/Performance Problem',
# 'P0117': 'Engine Coolant Temperature Circuit Low Input',
# 'P0118': 'Engine Coolant Temperature Circuit High Input',
# 'P0119': 'Engine Coolant Temperature Circuit Intermittent',
# 'P0120': 'Throttle/Pedal Position Sensor/Switch "A" Circuit Malfunction',
# 'P0121': 'Throttle/Pedal Position Sensor/Switch "A" Circuit Range/Performance Problem',
# 'P0122': 'Throttle/Pedal Position Sensor/Switch "A" Circuit Low Input',
# 'P0123': 'Throttle/Pedal Position Sensor/Switch "A" Circuit High Input',
# 'P0124': 'Throttle/Pedal Position Sensor/Switch "A" Circuit Intermittent',
# 'P0125': 'Insufficient Coolant Temperature for Closed Loop Fuel Control',
# 'P0126': 'Insufficient Coolant Temperature for Stable Operation',
# 'P0127': 'Intake Air Temperature Too High',
# 'P0128': 'Coolant Thermostat (Coolant Temperature Below Thermostat Regulating Temperature)',
# 'P0129': 'Barometric Pressure Too Low',
# 'P0130': 'O2 Sensor Circuit Malfunction (Bank 1 Sensor 1)',
# 'P0131': 'O2 Sensor Circuit Low Voltage (Bank 1 Sensor 1)',
# 'P0132': 'O2 Sensor Circuit High Voltage (Bank 1 Sensor 1)',
# 'P0133': 'O2 Sensor Circuit Slow Response (Bank 1 Sensor 1)',
# 'P0134': 'O2 Sensor Circuit No Activity Detected (Bank 1 Sensor 1)',
# 'P0135': 'O2 Sensor Heater Circuit Malfunction (Bank 1 Sensor 1)',
# 'P0136': 'O2 Sensor Circuit Malfunction (Bank 1 Sensor 2)',
# 'P0137': 'O2 Sensor Circuit Low Voltage (Bank 1 Sensor 2)',
# 'P0138': 'O2 Sensor Circuit High Voltage (Bank 1 Sensor 2)',
# 'P0139': 'O2 Sensor Circuit Slow Response (Bank 1 Sensor 2)',
# 'P0140': 'O2 Sensor Circuit No Activity Detected (Bank 1 Sensor 2)',
# 'P0141': 'O2 Sensor Heater Circuit Malfunction (Bank 1 Sensor 2)',
# 'P0171': 'System Too Lean (Bank 1)',
# 'P0172': 'System Too Rich (Bank 1)',
# 'P0173': 'Fuel Trim Malfunction (Bank 2)',
# 'P0174': 'System Too Lean (Bank 2)',
# 'P0175': 'System Too Rich (Bank 2)',
# 'P0300': 'Random/Multiple Cylinder Misfire Detected',
# 'P0301': 'Cylinder 1 Misfire Detected',
# 'P0302': 'Cylinder 2 Misfire Detected',
# 'P0303': 'Cylinder 3 Misfire Detected',
# 'P0304': 'Cylinder 4 Misfire Detected',
# 'P0305': 'Cylinder 5 Misfire Detected',
# 'P0306': 'Cylinder 6 Misfire Detected',
# 'P0307': 'Cylinder 7 Misfire Detected',
# 'P0308': 'Cylinder 8 Misfire Detected',
# 'P0420': 'Catalyst System Efficiency Below Threshold (Bank 1)',
# 'P0430': 'Catalyst System Efficiency Below Threshold (Bank 2)',
# 'P0442': 'Evaporative Emission Control System Leak Detected (Small Leak)',
# 'P0446': 'Evaporative Emission Control System Vent Control Circuit Malfunction',
# 'P0455': 'Evaporative Emission Control System Leak Detected (Gross Leak)',
# 'P0456': 'Evaporative Emission Control System Leak Detected (Very Small Leak)',
# 'P0506': 'Idle Control System RPM Lower Than Expected',
# 'P0507': 'Idle Control System RPM Higher Than Expected',
# }
@classmethod
def get_dtc_description(cls, code: str) -> str:
"""Get DTC description"""
if code in cls.COMMON_DTCS:
return cls.COMMON_DTCS[code]
prefix = code[:2].upper()
if prefix in cls.DTC_PREFIXES:
return f"{cls.DTC_PREFIXES[prefix]} - {code}"
return f"Unknown DTC: {code}"
class StoreRadings:
def appendToJson(self, filename, new_reading):
filename = filename + ".json"
# Step 1: If file exists, read it. Otherwise, start with an empty list
if os.path.exists(filename):
with open(filename, "r") as f:
try:
data = json.load(f)
except json.JSONDecodeError:
data = []
else:
data = []
# Step 2: Append the new reading
data.append(new_reading)
# Step 3: Write it back to the file
with open(filename, "w") as f:
json.dump(data, f, indent=4)
print("Current Readings saved! in ", filename)
class ELM327Scanner:
"""Main ELM327 Scanner class with comprehensive functionality"""
def __init__(self, host: str = "192.168.0.10", port: int = 35000):
self.host = host
self.port = port
self.socket = None
self.connected = False
self.protocol = None
self.vin = None
self.ecu_info = {}
self.streaming = False
self.log_file = None
self.performance_data = {}
# Initialize logging
self.logger = logging.getLogger(__name__)
def connect(self) -> bool:
"""Connect to ELM327 device"""
try:
self.socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
self.socket.settimeout(10)
self.socket.connect((self.host, self.port))
self.connected = True
# Initialize ELM327
if self._initialize_elm327():
self.logger.info(f"Connected to ELM327 at {self.host}:{self.port}")
return True
else:
self.disconnect()
return False
except Exception as e:
self.logger.error(f"Connection failed: {e}")
self.connected = False
return False
def disconnect(self):
"""Disconnect from ELM327 device"""
if self.socket:
try:
self.socket.close()
except:
pass
self.socket = None
self.connected = False
self.streaming = False
self.logger.info("Disconnected from ELM327")
def _initialize_elm327(self) -> bool:
"""Initialize ELM327 with proper settings"""
try:
# Reset device
if not self._send_command("ATZ"):
return False
time.sleep(2)
# Turn off echo
if not self._send_command("ATE0"):
return False
# Turn off line feeds
if not self._send_command("ATL0"):
return False
# Turn off spaces
if not self._send_command("ATS0"):
return False
# Set headers off
if not self._send_command("ATH0"):
return False
# Adaptive timing auto
if not self._send_command("ATAT1"):
return False
# Try automatic protocol detection
response = self._send_command("ATSP0")
if response:
self.logger.info("ELM327 initialized successfully")
return True
return False
except Exception as e:
self.logger.error(f"ELM327 initialization failed: {e}")
return False
def parse_pid_response(self, resp: str):
"""
Assumes resp is already cleaned (e.g., '410C0FA0').
"""
mode = resp[:2] # '41'
pid = resp[2:4] # '0C'
data = resp[4:] # '0FA0' (remaining hex string)
return mode, pid, data
def _send_command(self, command: str, timeout: float = 5.0) -> str: ## !! checked :)
"""Send command to ELM327 and return response"""
if not self.connected or not self.socket:
return ""
try:
# Send command
cmd = f"{command}\r"
self.socket.send(cmd.encode('ascii'))
# Receive response
response = ""
start_time = time.time()
while time.time() - start_time < timeout:
try:
data = self.socket.recv(1024).decode('ascii', errors='ignore')
if not data:
break
response += data
# Check for prompt
if '>' in response:
break
except socket.timeout:
break
# Clean up response
response = response.replace('\r', '').replace('\n', '').replace('>', '').replace(" ", '').strip()
print(response, "R1", response.find(command) )
# Filter out echo and common responses
# If the command echo is in the response, remove it
if response.startswith(command):
response = response[len(command):].strip()
## ?? may it not sutable here
# mode, pid, data = self.parse_pid_response(response)
# print("ECU respond --> ",mode, pid, data, " ")
return response
except Exception as e:
self.logger.error(f"Command failed: {command} - {e}")
return ""
def _parse_hex_response(self, response: str) -> List[int]:
"""Parse hex response into byte array"""
try:
# Remove spaces and common error responses
clean_response = response.replace(' ', '').replace('\r', '').replace('\n', '')
# Check for error responses
error_responses = ['NODATA', 'ERROR', 'TIMEOUT', '?', 'UNABLETOCONNECT', 'BUSBUSY']
if any(err in clean_response.upper() for err in error_responses):
return []
# Convert hex pairs to integers
if len(clean_response) % 2 == 0:
return [int(clean_response[i:i+2], 16) for i in range(0, len(clean_response), 2)]
else:
return []
except Exception as e:
self.logger.error(f"Failed to parse hex response: {response} - {e}")
return []
def get_supported_pids(self, mode: str = "01") -> List[str]:
"""Get list of supported PIDs for given mode"""
supported_pids = []
try:
# Check PID support ranges
pid_ranges = ["00", "20", "40", "60", "80", "A0", "C0", "E0"]
for pid_range in pid_ranges:
command = f"{mode}{pid_range}"
response = self._send_command(command)
if response and "NODATA" not in response.upper():
modep, pid, r = self.parse_pid_response(response) ## for skipping pid and mode
data = self._parse_hex_response(r)
if len(data) >= 6: # Mode + PID + 4 bytes of data
# Skip mode and PID bytes we have skiped it in __send_command
# pid_data = data[2:6]
pid_data = data
# Parse supported PIDs from bitmask
base_pid = int(pid_range, 16)
for byte_idx, byte_val in enumerate(pid_data):
for bit_idx in range(8):
if byte_val & (0x80 >> bit_idx):
pid_num = base_pid + byte_idx * 8 + bit_idx + 1
if pid_num <= 255:
supported_pids.append(f"{mode}{pid_num:02X}")
except Exception as e:
self.logger.error(f"Failed to get supported PIDs: {e}")
return supported_pids
def signals_values(self, bits_list:list, data_bits: str) -> List: ## !! checked :)
"""give each signal it is value """
signalsv = []
dbits = data_bits
for b in bits_list:
value = dbits[b[1]-1:b[1]+b[0]-1]
print(value)
signalsv.append(value)
return signalsv
def read_pid(self, pid: str, skippedByet=0) -> Optional[SensorReading]: ## !! checked :)
"""Read a specific PID and return sensor reading"""
try:
response = self._send_command(pid)
if not response or "NODATA" in response.upper():
return None
rmode, rpid, r = self.parse_pid_response(response) ## for skipping pid and mode
data_bytes = self._parse_hex_response(r)
data_bytes = data_bytes[skippedByet:]
bits = bin(int(r, 16))[2:].zfill(len(r) * 4)
print(bits, " done1")
if len(data_bytes) < 1: # At least mode + PID + 1 data byte
return None
# Extract data bytes (skip mode and PID)
# data_bytes = data[2:]
dbyte = data_bytes
# Get PID info and calculate value
pid_info = PIDDatabase.get_pid_info(pid)
signals = pid_info["signals"].keys()
signals_list = []
try:
bits_list = []
for s in signals:
l = pid_info["signals"][s]['bit_length']
i = pid_info["signals"][s].setdefault("bit_index", 1)
bits_list.append([l, i])
print(bits_list, bits)
signalsv = self.signals_values(bits_list, bits)
print(signalsv)
# {
# 'value': str,
# 'path': str,
# 'name': str,
# 'unit': str,
# }
# "bit_length": 1,
# "bit_length": 16,
# "bit_length": 8,
if (len(signalsv) <= len(signals)):
c = 0
for s in signals:
try:
ss = pid_info["signals"][s]
l = pid_info["signals"][s]['bit_length']
formula = ss.setdefault('formula', lambda x: x)
if (ss['bit_length']/8 >= 1):
num = int(signalsv[c], 2) # convert to decimal
sss = {
'value': float(formula(num)),
'path': ss['path'],
'name': ss['name'],
'unit': ss['unit'],
}
signals_list.append(sss)
except Exception as err:
print(err)
c += 1
except Exception as e:
print("Read PID Error : ", e)
sss = {
'value': bits,
'path': 'everything',
'name': 'everything',
'unit': '',
}
signals_list.append(sss)
return SensorReading(
pid=pid,
name=pid_info["name"],
signals=signals_list,
timestamp=datetime.now(),
raw_data=response,
)
except Exception as e:
self.logger.error(f"Failed to read PID {pid}: {e}")
return None
def get_dtcs(self, mode: str = "03") -> list[DTCInfo]:
"""Read Diagnostic Trouble Codes (Mode 03)"""
dtcs = []
try:
response = self._send_command(mode)
if not response or "NODATA" in response.upper():
return dtcs
print(response)
# Remove spaces and split into bytes
hex_bytes = self._parse_hex_response(response) # should return list of ints
# ECU reply for Mode 03 (the first byte is 40 + 03).
hex_bytes = hex_bytes[1:]
print(hex_bytes)
# Each DTC is 2 bytes
for i in range(0, len(hex_bytes), 2):
if i + 1 >= len(hex_bytes):
break # incomplete DTC
b1, b2 = hex_bytes[i], hex_bytes[i + 1]
# Skip empty DTCs
if b1 == 0 and b2 == 0:
continue
dtc_code = self._decode_dtc([b1, b2])
if dtc_code:
dtcs.append(DTCInfo(
code=dtc_code,
description=DTC_Database.get_dtc_description(dtc_code),
status="Active"
))
except Exception as e:
self.logger.error(f"Failed to read DTCs: {e}")
return dtcs
def get_pending_dtcs(self) -> List[DTCInfo]:
"""Read Pending Diagnostic Trouble Codes"""
dtcs = []
try:
response = self._send_command("07")
if not response or "NODATA" in response.upper():
return dtcs
data = self._parse_hex_response(response)
if len(data) < 2:
return dtcs
# First byte is number of DTCs
num_dtcs = data[0]
# Each DTC is 2 bytes
for i in range(1, min(len(data), 1 + num_dtcs * 2), 2):
if i + 1 < len(data):
dtc_bytes = [data[i], data[i + 1]]
dtc_code = self._decode_dtc(dtc_bytes)
if dtc_code:
dtcs.append(DTCInfo(
code=dtc_code,
description=DTC_Database.get_dtc_description(dtc_code),
status="Pending"
))
except Exception as e:
self.logger.error(f"Failed to read pending DTCs: {e}")
return dtcs
def clear_dtcs(self) -> bool:
"""Clear Diagnostic Trouble Codes"""
try:
response = self._send_command("04")
return "OK" in response.upper() or response == ""
except Exception as e:
self.logger.error(f"Failed to clear DTCs: {e}")
return False
def _decode_dtc(self, dtc_bytes: List[int]) -> str:
"""Decode DTC bytes to standard format"""
if len(dtc_bytes) != 2:
return ""
try:
# First two bits determine the system
system_bits = (dtc_bytes[0] & 0xC0) >> 6
system_map = {0: 'P', 1: 'C', 2: 'B', 3: 'U'}
system = system_map.get(system_bits, 'P')
# Remaining 14 bits are the code
code_value = ((dtc_bytes[0] & 0x3F) << 8) | dtc_bytes[1]
return f"{system}{code_value:04X}"
except Exception as e:
self.logger.error(f"Failed to decode DTC: {dtc_bytes} - {e}")
return ""
def get_freeze_frame_data(self, dtc_index: int = 0) -> Dict[str, Any]:
"""Read freeze frame data
https://x-engineer.org/obd-diagnostic-service-mode-02-request-powertrain-freeze-frame-data/
"""
"""
Request Example (from the scantool to ECM): 02 00 01
02 – Service Identifier (SID) for freeze frame data request.
00 – PID to report supported PIDs.
01 – frame number (the ECU can store multiple freeze frames).
Response Example (from the ECM to scantool): 42 00 01 5F B8 00 00
42 – Response Service Identifier.
00 – PID to report supported PIDs.
01 – frame number (the ECU can store multiple freeze frames).
5F B8 00 00 – Bitmap indicating supported PIDs; each bit in the data bytes corresponds to a PID. If a bit is 1, the PID is supported.
"""
freeze_data = {}
try:
# Mode 02 - Freeze Frame Data
response = self._send_command(f"0200{dtc_index:02X}")
if response and "NODATA" not in response.upper():
data = self._parse_hex_response(response)
# Parse freeze frame PIDs (implementation depends on vehicle)
if len(data) >= 3:
freeze_data["dtc_index"] = dtc_index
freeze_data["timestamp"] = datetime.now()
freeze_data["raw_data"] = response
# Common freeze frame PIDs
freeze_pids = ['01', '02', '03', '04', '05', '06', '07', '08', '09', '0A', '0B', '0C', '0D', '0E', '0F', '10'] # Common freeze frame PIDs
for pid in freeze_pids:
reading = self.read_pid(f"02{pid}{dtc_index:02X}", skippedByet=1)
if reading:
freeze_data[pid] = reading
except Exception as e:
self.logger.error(f"Failed to read freeze frame data: {e}")
return freeze_data
def get_vehicle_info(self) -> Dict[str, str]: # !! checked :)
"""Get comprehensive vehicle information"""
info = {}
try:
# VIN (Mode 09, PID 02)
vin = self.get_vin()
if vin:
info["VIN"] = vin
# Calibration ID (Mode 09, PID 04)
cal_id = self._send_command("0904")
if cal_id and "NODATA" not in cal_id.upper():
info["Calibration_ID"] = cal_id
# CVN (Mode 09, PID 06)
cvn = self._send_command("0906")
if cvn and "NODATA" not in cvn.upper():
info["CVN"] = cvn
# ECU Name (Mode 09, PID 0A)
ecu_name = self._send_command("090A")
if ecu_name and "NODATA" not in ecu_name.upper():
info["ECU_Name"] = ecu_name
# Get protocol info
protocol_info = self._send_command("ATDP")
if protocol_info:
info["Protocol"] = protocol_info
except Exception as e:
self.logger.error(f"Failed to get vehicle info: {e}")
return info
def get_vin(self) -> str: # !! checked :)
"""Get Vehicle Identification Number"""
try:
if self.vin:
return self.vin
# Mode 09, PID 02 - VIN
response = self._send_command("0902")
if response and "NODATA" not in response.upper():
data = self._parse_hex_response(response)
# VIN is typically 17 ASCII characters
# if len(data) >= 19: # Mode + PID + 17 VIN bytes
# vin_bytes = data[2:19] # Skip mode and PID
vin = ''.join(chr(b) for b in data if 32 <= b <= 126)
if len(vin) == 17:
self.vin = vin
return vin
return ""
except Exception as e:
self.logger.error(f"Failed to get VIN: {e}")
return ""
def get_readiness_monitors(self) -> Dict[str, str]:
"""Get OBD monitor readiness status"""
monitors = {}
try:
# Mode 01, PID 01 - Monitor status
response = self._send_command("0101")
if response and "NODATA" not in response.upper():
data = self._parse_hex_response(response)
if len(data) >= 6: # Mode + PID + 4 data bytes
status_bytes = data[2:6]
# Parse monitor status bits
monitors["MIL_Status"] = "ON" if status_bytes[0] & 0x80 else "OFF"
monitors["DTC_Count"] = status_bytes[0] & 0x7F
# Continuous monitors (byte 1)
continuous = status_bytes[1]
monitors["Misfire"] = "Ready" if not (continuous & 0x01) else "Not Ready"
monitors["Fuel_System"] = "Ready" if not (continuous & 0x02) else "Not Ready"
monitors["Components"] = "Ready" if not (continuous & 0x04) else "Not Ready"
# Non-continuous monitors (bytes 2-3)
non_continuous_supported = (status_bytes[2] << 8) | status_bytes[3]
non_continuous_ready = (status_bytes[2] << 8) | status_bytes[3]
monitor_names = [
"Catalyst", "Heated_Catalyst", "Evaporative_System", "Secondary_Air",
"A/C_Refrigerant", "Oxygen_Sensor", "Oxygen_Sensor_Heater", "EGR_System"
]
for i, name in enumerate(monitor_names):
if non_continuous_supported & (1 << i):
monitors[name] = "Ready" if not (non_continuous_ready & (1 << (i + 8))) else "Not Ready"
except Exception as e:
self.logger.error(f"Failed to get readiness monitors: {e}")
return monitors
def start_live_data_stream(self, pids: List[str], interval: float = 1.0, enable_graph: bool = False):
"""Start streaming live sensor data"""
if self.streaming:
self.stop_live_data_stream()
self.streaming = True
# Initialize shared data structures for graphing
if enable_graph and MATPLOTLIB_AVAILABLE:
self.graph_data = {
'pids': pids,
'data_history': {},
'time_history': {},
'max_points': 50
}
for pid in pids:
self.graph_data['data_history'][pid] = deque(maxlen=50)
self.graph_data['time_history'][pid] = deque(maxlen=50)
self.stream_thread = threading.Thread(
target=self._stream_worker,
args=(pids, interval, enable_graph),
daemon=True
)
self.stream_thread.start()
# Start graph in main thread if enabled
if enable_graph and MATPLOTLIB_AVAILABLE:
self._start_graph_display()
def _start_graph_display(self):
"""Start graph display in main thread"""
try:
plt.ion()
self.fig, self.ax = plt.subplots(figsize=(12, 8))
self.lines = {}
# Set up the plot
for pid in self.graph_data['pids']:
pid_info = PIDDatabase.get_pid_info(pid)
label = f"{pid}: {pid_info['signals'][0]['name']} ({pid_info['signals'][0]['unit']})"
line, = self.ax.plot([], [], marker='o', markersize=3, label=label, linewidth=2)
self.lines[pid] = line
self.ax.set_xlabel("Time", fontsize=12)
self.ax.set_ylabel("Sensor Values", fontsize=12)
self.ax.set_title("Live OBD-II Sensor Data", fontsize=14, fontweight='bold')
self.ax.legend(bbox_to_anchor=(1.05, 1), loc='upper left')
self.ax.grid(True, alpha=0.3)
# Set up time formatting
self.ax.xaxis.set_major_formatter(mdates.DateFormatter('%H:%M:%S'))
self.ax.xaxis.set_major_locator(mdates.SecondLocator(interval=10))
# Limit number of ticks to prevent warnings
self.ax.yaxis.set_major_locator(ticker.MaxNLocator(nbins=8))
self.ax.xaxis.set_major_locator(ticker.MaxNLocator(nbins=6))
plt.tight_layout()
plt.show(block=False)
# Start update timer
self.graph_timer = threading.Timer(0.5, self._update_graph)
self.graph_timer.start()
except Exception as e:
self.logger.error(f"Failed to start graph display: {e}")
print(f"Graph display failed: {e}")
def _update_graph(self):
"""Update graph periodically"""
if not self.streaming:
return
try:
# Update lines with current data
for pid in self.graph_data['pids']:
if len(self.graph_data['time_history'][pid]) > 0:
times = list(self.graph_data['time_history'][pid])
values = list(self.graph_data['data_history'][pid])
self.lines[pid].set_xdata(times)
self.lines[pid].set_ydata(values)
# Rescale axes
self.ax.relim()
self.ax.autoscale_view()
# Format x-axis dates
self.fig.autofmt_xdate()
# Refresh display
self.fig.canvas.draw()
self.fig.canvas.flush_events()
# Schedule next update
if self.streaming:
self.graph_timer = threading.Timer(0.5, self._update_graph)
self.graph_timer.start()
except Exception as e:
self.logger.error(f"Graph update error: {e}")
# Continue streaming without graph
def stop_live_data_stream(self):
"""Stop streaming live sensor data"""
self.streaming = False
# Stop graph timer