Code: Select all
#!/usr/bin/env python3
"""
Minimal J2534 CAN poller — read-only
- Mode 01 PID 00 (supported PIDs)
- Mode 09 PID 00 (supported Mode9)
- Mode 09 PID 02 VIN
- Mode 09 PID 04 CALID
- Mode 09 PID 06 CVN
"""
import ctypes, time, sys
from datetime import datetime
# ==== J2534 constants (from SAE J2534 FEB2002) ====
PROTOCOL_CAN = 0x00000005 # raw CAN, we do ISO-TP ourselves
BAUD_500K = 500000
PASS_FILTER = 0x00001
STATUS_NOERROR = 0
TX_ID = 0x7E0 # physical request
RX_ID = 0x7E8 # typical ECU response (0x7E8-0x7EF also valid)
class PASSTHRU_MSG(ctypes.Structure):
_fields_ = [
("ProtocolID", ctypes.c_ulong),
("RxStatus", ctypes.c_ulong),
("TxFlags", ctypes.c_ulong),
("Timestamp", ctypes.c_ulong),
("DataSize", ctypes.c_ulong),
("ExtraDataIndex", ctypes.c_ulong),
("Data", ctypes.c_ubyte * 4128)
]
def hexd(b): return ' '.join(f'{x:02X}' for x in b)
class J2534:
def __init__(self, dll_path):
self.dll = ctypes.WinDLL(dll_path)
self.dev = ctypes.c_ulong()
self.chan = ctypes.c_ulong()
def open(self):
assert self.dll.PassThruOpen(None, ctypes.byref(self.dev)) == STATUS_NOERROR
def connect(self):
assert self.dll.PassThruConnect(self.dev, PROTOCOL_CAN, 0, BAUD_500K, ctypes.byref(self.chan)) == STATUS_NOERROR
def close(self):
self.dll.PassThruDisconnect(self.chan)
self.dll.PassThruClose(self.dev)
def start_filter(self):
# pass only 0x7E8-0x7EF
mask = PASSTHRU_MSG(); pat = PASSTHRU_MSG()
for m in (mask, pat):
m.ProtocolID = PROTOCOL_CAN
m.DataSize = 4
# mask = 0x7F8, pattern = 0x7E8
for i, b in enumerate([(0x7F8>>24)&0xFF, (0x7F8>>16)&0xFF, (0x7F8>>8)&0xFF, 0x7F8&0xFF]):
mask.Data[i] = b
for i, b in enumerate([(0x7E8>>24)&0xFF, (0x7E8>>16)&0xFF, (0x7E8>>8)&0xFF, 0x7E8&0xFF]):
pat.Data[i] = b
fid = ctypes.c_ulong()
self.dll.PassThruStartMsgFilter(self.chan, PASS_FILTER, ctypes.byref(mask), ctypes.byref(pat), None, ctypes.byref(fid))
def send_can(self, can_id, data8):
msg = PASSTHRU_MSG()
msg.ProtocolID = PROTOCOL_CAN
msg.DataSize = 4 + len(data8)
msg.Data[0] = (can_id>>24)&0xFF; msg.Data[1] = (can_id>>16)&0xFF
msg.Data[2] = (can_id>>8)&0xFF; msg.Data[3] = can_id&0xFF
for i,b in enumerate(data8): msg.Data[4+i]=b
n = ctypes.c_ulong(1)
self.dll.PassThruWriteMsgs(self.chan, ctypes.byref(msg), ctypes.byref(n), 100)
def read_can(self, timeout_ms=50):
msg = PASSTHRU_MSG(); n = ctypes.c_ulong(1)
if self.dll.PassThruReadMsgs(self.chan, ctypes.byref(msg), ctypes.byref(n), timeout_ms)!= STATUS_NOERROR or n.value==0:
return None
can_id = (msg.Data[0]<<24)|(msg.Data[1]<<16)|(msg.Data[2]<<8)|msg.Data[3]
data = bytes(msg.Data[4:4+msg.DataSize-4])
return can_id, data
def flush(self):
while self.read_can(10): pass
# ==== ISO-TP ====
def iso_tp_request(j, service, pid, tx=TX_ID, rx=RX_ID, timeout=1.0):
j.flush()
# build single-frame request: PCI=0x02, [svc,pid]
req = bytes([0x02, service, pid, 0,0,0,0,0])
j.send_can(tx, req)
start = time.time()
buf = bytearray()
expected_len = None
seq = 1
while time.time()-start < timeout:
r = j.read_can(50)
if not r: continue
cid, data = r
if cid!= rx or not data: continue
pci = data[0]
# Single Frame
if (pci & 0xF0) == 0x00:
length = pci & 0x0F
return data[1:1+length] # includes SID,PID,...
# First Frame
if (pci & 0xF0) == 0x10:
if len(data) < 2: continue
expected_len = ((pci & 0x0F)<<8) | data[1]
buf.extend(data[2:8]) # take up to 6 bytes
# send FC to TX_ID
fc = bytes([0x30, 0x00, 0x0A, 0,0,0]) # BS=0, ST=10ms
j.send_can(tx, fc)
continue
# Consecutive Frame
if (pci & 0xF0) == 0x20:
if expected_len is None: continue
if (pci & 0x0F)!= seq:
seq = pci & 0x0F # resync
buf.extend(data[1:8])
seq = (seq+1) & 0x0F
if len(buf) >= expected_len:
return bytes(buf[:expected_len])
# Negative Response (inside SF)
if len(data) >= 3 and data[1] == 0x7F:
return data[1:4] # 7F, sid, nrc
return None
# ==== decoders (correct offsets) ====
def decode_0100(p):
if not p or p[0]!=0x41 or p[1]!=0x00: return "bad"
return f"01-20 supported: {hexd(p[2:6])}"
def decode_0900(p):
if not p or p[0]!=0x49 or p[1]!=0x00: return "bad"
return f"09-00 supported: {hexd(p[2:6])}"
def decode_vin(p):
# p = 49 02 01 [17 ascii]
if not p or p[0]!=0x49 or p[1]!=0x02: return "bad"
vin = p[3:3+17].decode('ascii', errors='ignore').strip('\x00')
return f"VIN: {vin}"
def decode_calid(p):
if not p or p[0]!=0x49 or p[1]!=0x04: return "bad"
cal = p[2:].decode('ascii', errors='ignore').strip('\x00')
return f"CALID: {cal}"
def decode_cvn(p):
if not p or p[0]!=0x49 or p[1]!=0x06: return "bad"
cvns = [hexd(p[i:i+4]) for i in range(2, len(p), 4)]
return f"CVN: {' '.join(cvns)}"
def main():
dll = r"C:\Program Files (x86)\OBDX Pro\J2534\OBDX Pro FT\OBDXFT_J2534_32bit.dll"
j = J2534(dll)
j.open(); j.connect(); j.start_filter()
print("Connected on CAN 500k")
tests = [
(0x01,0x00, decode_0100, "Mode1 PIDs"),
(0x09,0x00, decode_0900, "Mode9 PIDs"),
(0x09,0x02, decode_vin, "VIN"),
(0x09,0x04, decode_calid,"CALID"),
(0x09,0x06, decode_cvn, "CVN"),
]
for svc,pid,dec,name in tests:
resp = iso_tp_request(j, svc, pid)
if resp and len(resp)>=2 and resp[0]==0x7F:
print(f"{name}: NRC {resp[2]:02X}")
else:
print(f"{name}: {dec(resp) if resp else 'no response'}")
time.sleep(0.1)
j.close()
if __name__ == "__main__":
main()
Notice the pcm changed it's VIN to match the truck.
If HPT actually over wrote where I read the vin, I wonder if it sends a custom bootloader to do a write, or a proprietary ford command that hasn't been openly documented yet. I plan to pull the bin and check it out
