Fix: OBD-II reads DTCs the standard way and asks which PIDs exist

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,21 +172,162 @@
     },
 }
 
+
+
+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 <count> <hi> <lo> ...``: 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]
 
 
 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))
     except Exception:
         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.
 

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):
         """

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

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