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."