From 8c1b740471beb79c7cc26b190c582c8751626365 Mon Sep 17 00:00:00 2001 From: Sherin Joseph Roy Date: Thu, 24 Sep 2026 00:25:01 +0530 Subject: [PATCH] fix: OBD-II reads DTCs the standard way and asks which PIDs exist Three defects in the diagnostics path, found auditing it against SAE J1979: - Read DTC sent only UDS service 0x19. Every OBD-II vehicle answers mode 03 (stored codes) and 07 (pending); only UDS-capable ECUs answer 0x19. It now asks with 03 and 07 and falls back to 0x19. - When nothing answered, the tab said "No DTCs found." That is false reassurance about a car that was never heard from. The scanner now records whether any ECU answered, and the tab says so. - The PID scan requested every PID in the table, one timeout each, without asking the car. It now reads the supported-PID masks (0x00, 0x20, ...) and requests only those, and stops after one question if nothing speaks mode 01. The table grows from 26 PIDs to 78: bank 2 fuel trims, fuel and rail pressures, eight O2 sensor voltages and equivalence ratios, catalyst temperatures, evaporative system pressures, secondary O2 trims, injection timing, engine torque, odometer. Formulas are J1979's. PID 0x49 was named "Throttle Pos D"; it is Accelerator Pedal Position D. A payload shorter than a PID's answer is refused instead of misread. Tests script an ECU that supports four PIDs and holds one stored and one pending code: the scan sends exactly six requests, the DTC read never reaches UDS, and a silent bus is reported as silent. --- canlab/core/obd2_pids.py | 154 ++++++++++++++++++++++++++++++++- canlab/core/uds.py | 62 +++++++++++-- canlab/tabs/diagnostics_tab.py | 13 ++- tests/test_obd2_pids.py | 52 +++++++++++ tests/test_uds.py | 96 ++++++++++++++++++++ 5 files changed, 366 insertions(+), 11 deletions(-) diff --git a/canlab/core/obd2_pids.py b/canlab/core/obd2_pids.py index 73f6eca..98faf6c 100644 --- a/canlab/core/obd2_pids.py +++ b/canlab/core/obd2_pids.py @@ -2,6 +2,15 @@ Each entry: {name, unit, min, max, decode} decode(data: bytes) -> float (data = response payload bytes A, B, C, D...) + +The formulas are the published SAE J1979 ones, written in the A, B, C, D +byte notation the standard uses. Only PIDs whose value is a single number +are here; bit-encoded status PIDs (monitor status, fuel system status, O2 +sensor presence) are not. + +Two helpers for the rest of the standard live beside the table: reading the +"which PIDs do you support" masks so a scan asks only for those, and +decoding the DTC lists that modes 03, 07 and 0A return. """ PID_TABLE: dict[int, dict] = { @@ -138,7 +147,7 @@ "decode": lambda b: b[0] * 100 / 255, }, 0x49: { - "name": "Throttle Pos D", + "name": "Accelerator Pedal Position D", "unit": "%", "min": 0, "max": 100, "decode": lambda b: b[0] * 100 / 255, @@ -163,6 +172,124 @@ }, } + + +def _u16(b) -> int: + return (b[0] << 8) | b[1] + + +def _s16(b) -> int: + v = _u16(b) + return v - 0x10000 if v & 0x8000 else v + + +def _add(pid, name, unit, lo, hi, decode): + PID_TABLE.setdefault(pid, {"name": name, "unit": unit, "min": lo, "max": hi, + "decode": decode}) + + +# Fuel trims share one formula: 100/128 * A - 100. +def _trim(b): + return (b[0] - 128) * 100 / 128 + + +def _pct(b): + return b[0] * 100 / 255 + + +_add(0x08, "Short Fuel Trim B2", "%", -100, 99.2, _trim) +_add(0x09, "Long Fuel Trim B2", "%", -100, 99.2, _trim) +_add(0x0A, "Fuel Pressure", "kPa", 0, 765, lambda b: 3 * b[0]) +for _n in range(8): + # 0x14..0x1B: voltage in A; B is that sensor's short term trim + _add(0x14 + _n, f"O2 Sensor {_n + 1} Voltage", "V", 0, 1.275, + lambda b: b[0] / 200) +_add(0x22, "Fuel Rail Pressure (vacuum ref.)", "kPa", 0, 5177.265, + lambda b: 0.079 * _u16(b)) +_add(0x23, "Fuel Rail Gauge Pressure", "kPa", 0, 655350, lambda b: 10 * _u16(b)) +for _n in range(8): + # 0x24..0x2B: wide-range sensors report an equivalence ratio in A, B + _add(0x24 + _n, f"O2 Sensor {_n + 1} Equivalence Ratio", "", 0, 2, + lambda b: 2 / 65536 * _u16(b)) +_add(0x2D, "EGR Error", "%", -100, 99.2, _trim) +_add(0x2E, "Commanded Evaporative Purge", "%", 0, 100, _pct) +_add(0x30, "Warm-ups Since Codes Cleared", "", 0, 255, lambda b: b[0]) +_add(0x32, "Evap System Vapor Pressure", "Pa", -8192, 8191.75, lambda b: _s16(b) / 4) +for _n, _where in enumerate(("Bank 1, Sensor 1", "Bank 2, Sensor 1", + "Bank 1, Sensor 2", "Bank 2, Sensor 2")): + _add(0x3C + _n, f"Catalyst Temperature {_where}", "°C", -40, 6513.5, + lambda b: _u16(b) / 10 - 40) +_add(0x44, "Commanded Equivalence Ratio", "", 0, 2, lambda b: 2 / 65536 * _u16(b)) +_add(0x48, "Absolute Throttle Position C", "%", 0, 100, _pct) +_add(0x4A, "Accelerator Pedal Position E", "%", 0, 100, _pct) +_add(0x4B, "Accelerator Pedal Position F", "%", 0, 100, _pct) +_add(0x4D, "Time Run With MIL On", "min", 0, 65535, lambda b: _u16(b)) +_add(0x4E, "Time Since Trouble Codes Cleared", "min", 0, 65535, lambda b: _u16(b)) +_add(0x50, "Maximum Air Flow Rate", "g/s", 0, 2550, lambda b: b[0] * 10) +_add(0x52, "Ethanol Fuel", "%", 0, 100, _pct) +_add(0x53, "Absolute Evap System Vapor Pressure", "kPa", 0, 327.675, + lambda b: _u16(b) / 200) +_add(0x54, "Evap System Vapor Pressure (wide)", "Pa", -32768, 32767, lambda b: _s16(b)) +_add(0x55, "Short Secondary O2 Trim B1", "%", -100, 99.2, _trim) +_add(0x56, "Long Secondary O2 Trim B1", "%", -100, 99.2, _trim) +_add(0x57, "Short Secondary O2 Trim B2", "%", -100, 99.2, _trim) +_add(0x58, "Long Secondary O2 Trim B2", "%", -100, 99.2, _trim) +_add(0x59, "Fuel Rail Absolute Pressure", "kPa", 0, 655350, lambda b: 10 * _u16(b)) +_add(0x5A, "Relative Accelerator Pedal Position", "%", 0, 100, _pct) +_add(0x5B, "Hybrid Battery Remaining Life", "%", 0, 100, _pct) +_add(0x5D, "Fuel Injection Timing", "°", -210, 301.992, lambda b: _u16(b) / 128 - 210) +_add(0x61, "Driver's Demand Engine Torque", "%", -125, 130, lambda b: b[0] - 125) +_add(0x62, "Actual Engine Torque", "%", -125, 130, lambda b: b[0] - 125) +_add(0x63, "Engine Reference Torque", "Nm", 0, 65535, lambda b: _u16(b)) +_add(0x8E, "Engine Friction Torque", "%", -125, 130, lambda b: b[0] - 125) +_add(0xA6, "Odometer", "km", 0, 429496729.5, + lambda b: ((b[0] << 24) | (b[1] << 16) | (b[2] << 8) | b[3]) / 10) + +#: How many data bytes each PID's answer carries, for the ones that are not one. +_TWO_BYTE = {0x0C, 0x10, 0x1F, 0x21, 0x22, 0x23, 0x31, 0x32, 0x3C, 0x3D, 0x3E, 0x3F, + 0x42, 0x43, 0x44, 0x4D, 0x4E, 0x53, 0x54, 0x59, 0x5D, 0x5E, 0x63, + *range(0x24, 0x2C)} +_FOUR_BYTE = {0xA6} + + +def response_length(pid: int) -> int: + """Data bytes in a Mode 01 answer for ``pid`` (after 41 and the PID).""" + if pid in _FOUR_BYTE: + return 4 + return 2 if pid in _TWO_BYTE else 1 + + +# ── diagnostic trouble codes (modes 03, 07, 0A) ────────────────────────────── + +#: Positive response byte for each DTC-reading mode, and what it lists. +DTC_MODES = {0x03: (0x43, "stored"), 0x07: (0x47, "pending"), 0x0A: (0x4A, "permanent")} + + +def decode_obd_dtcs(payload: bytes, mode: int = 0x03) -> list[str] | None: + """The codes in a mode 03, 07 or 0A answer, or None if it is not one. + + Over CAN (ISO 15765-4) the answer is ``43 ...``: a count + byte, then two bytes per code. An empty list means the ECU answered and + has no codes, which is not the same thing as no answer at all. + """ + from canlab.core.uds import decode_dtc + expect = DTC_MODES.get(mode, (0x43, ""))[0] + if not payload or payload[0] != expect: + return None + body = payload[1:] + if body and len(body) % 2 == 1: + count, body = body[0], body[1:] + else: + count = len(body) // 2 + codes = [] + for i in range(0, min(len(body), 2 * count), 2): + hi, lo = body[i], body[i + 1] + if hi == 0 and lo == 0: + continue # padding, not P0000 + codes.append(decode_dtc(hi, lo)) + return codes + + # Subset shown by default on the gauge tab DEFAULT_PIDS = [0x0C, 0x0D, 0x05, 0x11, 0x10, 0x2F] @@ -170,7 +297,7 @@ def decode_pid(pid: int, data: bytes) -> float | None: """Decode a Mode 01 PID response payload (bytes after SID/PID stripped).""" entry = PID_TABLE.get(pid) - if entry is None or not data: + if entry is None or len(data) < response_length(pid): return None try: return float(entry["decode"](data)) @@ -178,6 +305,29 @@ def decode_pid(pid: int, data: bytes) -> float | None: return None +def discover_supported(request, bases=(0x00, 0x20, 0x40, 0x60, 0x80, 0xA0, 0xC0)): + """Ask which PIDs the vehicle supports; ``request(bytes)`` returns the + response payload or None. Returns (supported pids, answered at all). + + PID 0x00 lists 0x01-0x20, and if 0x20 is among them, PID 0x20 lists the + next window, and so on. A vehicle that does not answer 0x00 does not + speak Mode 01, and asking it for sixty PIDs one timeout at a time tells + you nothing more. + """ + found: list[int] = [] + answered = False + for base in bases: + payload = request(bytes([0x01, base])) + if not (payload and len(payload) >= 6 and payload[0] == 0x41 and payload[1] == base): + break + answered = True + window = supported_pids_from_mask(payload[2:6], base=base) + found.extend(window) + if (base + 0x20) not in window: + break + return found, answered + + def supported_pids_from_mask(mask_data: bytes, base: int = 0x00) -> list[int]: """Parse a 4-byte 'supported PIDs' bit-mask into a list of PID numbers. diff --git a/canlab/core/uds.py b/canlab/core/uds.py index cd2c99b..8e749bd 100644 --- a/canlab/core/uds.py +++ b/canlab/core/uds.py @@ -151,6 +151,9 @@ def __init__(self, bus, mode: str = "PID", ecu_addr: int = 0x7DF, # "which services are supported" probe cannot reset ECUs or clear DTCs # on a live bus. self._allow_unsafe = allow_unsafe + #: After a DTC read: did any ECU answer? "No codes" and "no answer" are + #: different findings, and only the first means the car is clean. + self.dtc_answered = False def stop(self): self._running = False @@ -209,8 +212,33 @@ def _send_and_recv(self, data: bytes, timeout: float = 0.5): return None def _scan_pids(self): - self.status.emit("Scanning OBD-II PIDs…") - for pid, entry in PID_TABLE.items(): + """Ask which PIDs the vehicle supports, then read those. + + This used to request every PID in the table, one timeout each, whether + or not the vehicle had it. The vehicle says which it supports in the + 0x00, 0x20, ... masks; asking only for those is what J1979 intends and + is several times faster on a real car. + """ + from canlab.core.obd2_pids import discover_supported + self.status.emit("Asking which OBD-II PIDs are supported…") + + def ask(data): + resp = self._send_and_recv(data) + return resp.data if resp is not None else None + + supported, answered = discover_supported(ask) + if not answered: + self.status.emit("No ECU answered PID 0x00, so nothing speaks OBD-II " + "Mode 01 on this bus (or the bitrate is wrong).") + return + # 0x20, 0x40, ... only say "ask me about the next window"; not readings + readings = [pid for pid in supported if pid % 0x20] + wanted = [pid for pid in readings if pid in PID_TABLE] + unknown = [pid for pid in readings if pid not in PID_TABLE] + self.status.emit(f"{len(readings)} PIDs supported, reading {len(wanted)}" + + (f"; {len(unknown)} have no decoder here" if unknown else "")) + for pid in wanted: + entry = PID_TABLE[pid] if not self._running: break resp = self._send_and_recv(bytes([0x01, pid])) @@ -226,9 +254,33 @@ def _scan_pids(self): entry.get("unit", "") or "") def _read_dtc(self): - self.status.emit("Reading DTCs (service 0x19)…") - resp = self._send_and_recv(bytes([0x19, 0x02, 0xFF]), timeout=0.5) - self.dtc_result.emit(decode_dtc_records(resp.data) if resp else []) + """Stored and pending codes the OBD-II way, then UDS if that fails. + + Every OBD-II vehicle answers modes 03 (stored) and 07 (pending); only + UDS-capable ECUs answer service 0x19, which is what this used to send + alone, so an older car reported "no DTCs" when it had simply not been + asked in a language it speaks. + """ + from canlab.core.obd2_pids import decode_obd_dtcs + codes: list[str] = [] + self.dtc_answered = False + for mode, label in ((0x03, "stored"), (0x07, "pending")): + self.status.emit(f"Reading {label} DTCs (OBD-II mode {mode:02X})…") + resp = self._send_and_recv(bytes([mode]), timeout=0.5) + found = decode_obd_dtcs(resp.data, mode) if resp is not None else None + if found is None: + continue + self.dtc_answered = True + codes += found if mode == 0x03 else [f"{c} (pending)" for c in found] + if not self.dtc_answered: + self.status.emit("No answer to OBD-II modes 03/07; trying UDS 0x19…") + resp = self._send_and_recv(bytes([0x19, 0x02, 0xFF]), timeout=0.5) + if resp is not None and resp.data[:1] == b"\x59": + self.dtc_answered = True + codes = decode_dtc_records(resp.data) + if not self.dtc_answered: + self.status.emit("No ECU answered a DTC request.") + self.dtc_result.emit(codes) def _deep_scan(self): """ diff --git a/canlab/tabs/diagnostics_tab.py b/canlab/tabs/diagnostics_tab.py index eed253c..7bb6001 100644 --- a/canlab/tabs/diagnostics_tab.py +++ b/canlab/tabs/diagnostics_tab.py @@ -477,11 +477,16 @@ def _read_dtc(self): self._dtc_worker.start() def _on_dtc_result(self, dtcs: list): - if not dtcs: - self.dtc_text.setPlainText("No DTCs found.") - else: + answered = getattr(self._dtc_worker, "dtc_answered", True) + if dtcs: self.dtc_text.setPlainText(" ".join(dtcs)) - self.uds_log.append(f"DTCs: {dtcs}") + elif answered: + self.dtc_text.setPlainText("No DTCs stored or pending.") + else: + # Not "no DTCs": nothing replied, so nothing is known about them. + self.dtc_text.setPlainText("No ECU answered. Check the bitrate and that " + "the ignition is on; this is not a clean result.") + self.uds_log.append(f"DTCs: {dtcs}" if answered else "DTCs: no answer") def _clear_dtc(self): import can diff --git a/tests/test_obd2_pids.py b/tests/test_obd2_pids.py index 5f42754..249f794 100644 --- a/tests/test_obd2_pids.py +++ b/tests/test_obd2_pids.py @@ -1,3 +1,4 @@ +import pytest """Tests for OBD-II supported-PID mask decoding, including continuation windows above 0x20 (#17).""" from canlab.core.obd2_pids import supported_pids_from_mask @@ -23,3 +24,54 @@ def test_continuation_window_offsets_by_base(): def test_short_mask_returns_empty(): assert supported_pids_from_mask(b"\x00\x00") == [] + + +# ── the formulas, against the worked values SAE J1979 publishes ────────────── + +from canlab.core.obd2_pids import ( # noqa: E402 + PID_TABLE, decode_obd_dtcs, decode_pid, response_length, +) + + +@pytest.mark.parametrize("pid,data,expected", [ + (0x0C, [0x1A, 0xF8], 1726.0), # (256A + B) / 4 + (0x05, [0x7B], 83.0), # A - 40 + (0x06, [0x80], 0.0), # 100/128 A - 100 + (0x0A, [0x64], 300.0), # 3A + (0x14, [0xC8, 0x80], 1.0), # A / 200 + (0x23, [0x01, 0x00], 2560.0), # 10 (256A + B) + (0x24, [0x80, 0x00], 1.0), # 2/65536 (256A + B): stoichiometric + (0x32, [0xFF, 0xFC], -1.0), # signed (256A + B) / 4 + (0x3C, [0x11, 0x94], 410.0), # (256A + B) / 10 - 40 + (0x49, [0xFF], 100.0), + (0x54, [0x80, 0x00], -32768.0), # signed + (0x5D, [0x69, 0x00], 0.0), # (256A + B) / 128 - 210 + (0x61, [0x7D], 0.0), # A - 125 + (0xA6, [0x00, 0x01, 0xE2, 0x40], 12345.6), # 4 bytes / 10 +]) +def test_published_formulas(pid, data, expected): + assert decode_pid(pid, bytes(data)) == pytest.approx(expected, abs=1e-3) + + +def test_pid_0x49_is_the_accelerator_pedal_not_a_throttle(): + assert PID_TABLE[0x49]["name"] == "Accelerator Pedal Position D" + + +def test_a_short_answer_is_refused_not_misread(): + assert response_length(0x0C) == 2 and decode_pid(0x0C, bytes([0x1A])) is None + assert response_length(0xA6) == 4 and decode_pid(0xA6, bytes([0, 1, 2])) is None + + +def test_every_pid_decodes_zeros_inside_its_range(): + for pid, entry in PID_TABLE.items(): + value = decode_pid(pid, bytes(response_length(pid) + 1)) + assert value is not None, hex(pid) + assert entry["min"] <= value <= entry["max"], (hex(pid), value) + + +def test_dtc_lists_from_modes_03_07_and_0a(): + assert decode_obd_dtcs(bytes([0x43, 0x02, 0x01, 0x33, 0x03, 0x01])) == ["P0133", "P0301"] + assert decode_obd_dtcs(bytes([0x47, 0x01, 0xC1, 0x00]), 0x07) == ["U0100"] + assert decode_obd_dtcs(bytes([0x43, 0x00])) == [] # answered, no codes + assert decode_obd_dtcs(bytes([0x7F, 0x03, 0x11])) is None # a refusal is not a list + assert decode_obd_dtcs(b"") is None diff --git a/tests/test_uds.py b/tests/test_uds.py index 804f6cc..d554353 100644 --- a/tests/test_uds.py +++ b/tests/test_uds.py @@ -151,3 +151,99 @@ def test_nothing_is_transmitted_while_disarmed(): assert bus.sent == [] finally: safety.set_armed(True) + + +# ── OBD-II the way J1979 intends ───────────────────────────────────────────── + +def _mask(*pids, base=0x00): + """A supported-PIDs mask for the window starting at ``base``.""" + m = 0 + for pid in pids: + m |= 1 << (32 - (pid - base)) + return list(m.to_bytes(4, "big")) + + +class _ScriptedCar: + """Answers like a car that supports four PIDs and has one stored and one + pending code. Anything else gets no reply, as on a real bus.""" + + ANSWERS = { + (0x01, 0x00): [0x41, 0x00, *_mask(0x0C, 0x0D, 0x20)], + (0x01, 0x20): [0x41, 0x20, *_mask(0x2F, 0x33, base=0x20)], + (0x01, 0x0C): [0x41, 0x0C, 0x1A, 0xF8], + (0x01, 0x0D): [0x41, 0x0D, 0x3C], + (0x01, 0x2F): [0x41, 0x2F, 0x80], + (0x01, 0x33): [0x41, 0x33, 0x65], + (0x03,): [0x43, 0x01, 0x01, 0x33], + (0x07,): [0x47, 0x01, 0x03, 0x01], + } + + def __init__(self): + self.requests = [] + + def __call__(self, msg): + n = msg.data[0] & 0x0F + req = tuple(msg.data[1:1 + n]) + self.requests.append(req) + key = req[:2] if req[0] == 0x01 else req[:1] + payload = self.ANSWERS.get(key) + return sf(0x7E8, payload) if payload else None + + +def test_pid_scan_asks_what_is_supported_and_reads_only_that(): + car = _ScriptedCar() + scanner = UDSScanner(RecordingBus(on_send=car), mode="PID") + got, notes = [], [] + scanner.pid_result.connect(lambda pid, name, value, unit: got.append((pid, value, unit))) + scanner.status.connect(notes.append) + scanner._scan_pids() + assert car.requests == [(0x01, 0x00), (0x01, 0x20), (0x01, 0x0C), (0x01, 0x0D), + (0x01, 0x2F), (0x01, 0x33)] + assert got == [(0x0C, 1726.0, "rpm"), (0x0D, 60.0, "km/h"), + (0x2F, 50.2, "%"), (0x33, 101.0, "kPa")] + assert any("4 PIDs supported, reading 4" in n for n in notes) + + +def test_pid_scan_stops_when_nothing_speaks_mode_01(): + bus = RecordingBus() + scanner = UDSScanner(bus, mode="PID") + notes = [] + scanner.status.connect(notes.append) + scanner._scan_pids() + assert len(bus.sent) == 1 # one question, not seventy-eight + assert any("nothing speaks OBD-II" in n for n in notes) + + +def test_dtcs_are_read_with_modes_03_and_07_before_uds(): + car = _ScriptedCar() + scanner = UDSScanner(RecordingBus(on_send=car), mode="DTC") + got = [] + scanner.dtc_result.connect(got.append) + scanner._read_dtc() + assert got == [["P0133", "P0301 (pending)"]] + assert scanner.dtc_answered + assert (0x19, 0x02, 0xFF) not in car.requests # OBD-II answered, no UDS needed + + +def test_no_answer_is_not_reported_as_no_codes(): + bus = RecordingBus() + scanner = UDSScanner(bus, mode="DTC") + got = [] + scanner.dtc_result.connect(got.append) + scanner._read_dtc() + assert got == [[]] and not scanner.dtc_answered + tried = [bytes(m.data[1:1 + m.data[0]]) for m in bus.sent] + assert tried == [b"\x03", b"\x07", b"\x19\x02\xff"] + + from types import SimpleNamespace + from canlab.tabs.diagnostics_tab import DiagnosticsTab + shown = {} + fake = SimpleNamespace( + _dtc_worker=scanner, + dtc_text=SimpleNamespace(setPlainText=lambda t: shown.update(text=t)), + uds_log=SimpleNamespace(append=lambda t: None)) + DiagnosticsTab._on_dtc_result(fake, []) + assert "not a clean result" in shown["text"] + scanner.dtc_answered = True + DiagnosticsTab._on_dtc_result(fake, []) + assert shown["text"] == "No DTCs stored or pending."