#!/usr/bin/env python3 """ Decode a BMW E8x PT-CAN capture (timestamp_ms, arbitration_id, is_extended, dlc, data_hex). python3 decode_ptcan.py capture_Start-Stop.csv # timeline + summary python3 decode_ptcan.py capture_Start-Stop.csv -o signals.csv # tidy signal table python3 decode_ptcan.py capture_Start-Stop.csv --check # verify checksums No dependencies beyond the standard library. The signal definitions match bmw_e8x_ptcan.dbc and PTCAN_protocol.md; edit SIGNALS to add your own. """ import argparse, csv, sys from collections import defaultdict u16le = lambda b, i: b[i] | (b[i + 1] << 8) def i16le(b, i): v = u16le(b, i) return v - 65536 if v > 32767 else v TERMINAL = {0x00: "off", 0x40: "terminal_R", 0x41: "terminal_15", 0x45: "engine_running", 0x55: "cranking", 0x80: "wake_up"} # id -> {signal name: (callable, unit)}. Confidence noted in PTCAN_protocol.md. SIGNALS = { 0x0AA: {"engine_speed": (lambda b: u16le(b, 4) / 6.4, "rpm"), "engine_flags": (lambda b: b[6], ""), "load_estimate": (lambda b: b[7], "")}, 0x0C4: {"steering_angle": (lambda b: i16le(b, 0) * 0.04395, "deg"), "steering_rate": (lambda b: i16le(b, 3) * 0.04395, "deg/s")}, 0x0C8: {"steering_angle_slow": (lambda b: i16le(b, 0) * 0.04395, "deg")}, 0x0A9: {"system_voltage": (lambda b: ((b[3] >> 4) | (b[4] << 4)) * 0.0555, "V")}, 0x130: {"terminal": (lambda b: TERMINAL.get(b[0], hex(b[0])), "")}, 0x1D0: {"temp_1": (lambda b: b[0] - 48, "degC"), "temp_2": (lambda b: b[1] - 48, "degC"), "fuel_accum": (lambda b: u16le(b, 4), "count")}, 0x1A0: {"road_speed": (lambda b: (b[0] | ((b[1] & 0x0F) << 8)) * 0.1, "km/h")}, 0x1B4: {"warning_lamps": (lambda b: (b[5] << 8) | b[4], ""), "cluster_speed": (lambda b: (b[0] | ((b[1] & 0x0F) << 8)) * 0.1, "km/h")}, 0x0CE: {"wheel_fl": (lambda b: u16le(b, 0), ""), "wheel_fr": (lambda b: u16le(b, 2), ""), "wheel_rl": (lambda b: u16le(b, 4), ""), "wheel_rr": (lambda b: u16le(b, 6), "")}, 0x1D6: {"mfl_buttons": (lambda b: u16le(b, 0), "")}, 0x380: {"vin_tail": (lambda b: bytes(b).decode("ascii", "replace"), "")}, 0x1A6: {"accum_fast": (lambda b: (b[6] >> 4) | ((b[7] & 0x0F) << 4), "count"), "accum_slow": (lambda b: b[0], "count")}, } NAMES = {0x0A8: "DME torque", 0x0A9: "DME voltage", 0x0AA: "DME engine speed", 0x0B6: "DSC counter", 0x0C4: "steering angle", 0x0C8: "steering angle 5Hz", 0x0CE: "DSC wheel speeds", 0x130: "CAS terminal", 0x19E: "DSC status", 0x1A0: "DSC road speed", 0x1A6: "DSC accumulator", 0x1B4: "instrument cluster", 0x1D0: "DME temps + fuel", 0x1D6: "MFL buttons", 0x380: "VIN tail"} # checksum byte index -> constant, for the IDs where one's-complement sum verifies 100% CHECKSUMS = {0x1B4: (7, 0xB6), 0x0B6: (0, 0xB7), 0x194: (0, 0x00), 0x1E1: (0, 0xE3), 0x200: (7, 0xBF)} # 4-bit alive counter position: (byte, shift) COUNTERS = {0x0A8: (1, 0), 0x0A9: (1, 0), 0x0AA: (1, 0), 0x0B6: (1, 0), 0x130: (4, 0), 0x1A0: (6, 4), 0x1B4: (3, 0), 0x194: (1, 0), 0x1E1: (1, 0), 0x1FB: (0, 0), 0x2F3: (0, 0), 0x2F1: (1, 0), 0x308: (0, 4)} def ocsum(bs): """One's-complement sum: add, folding each carry back into the low byte.""" c = 0 for b in bs: c += b if c > 0xFF: c = (c & 0xFF) + 1 return c & 0xFF def read(path): with open(path, newline="") as f: for row in csv.DictReader(f): hx = row["data_hex"].strip() yield (int(row["timestamp_ms"]), int(row["arbitration_id"], 16), bytes.fromhex(hx)) def main(): ap = argparse.ArgumentParser() ap.add_argument("csv") ap.add_argument("-o", "--out", help="write a tidy time,id,signal,value,unit table") ap.add_argument("--check", action="store_true", help="verify checksums and counters") a = ap.parse_args() frames = list(read(a.csv)) if not frames: sys.exit("no frames found — check the CSV columns") # the source capture has one row with a corrupt timestamp; sorting keeps the # rate maths and the timeline honest frames.sort(key=lambda f: f[0]) t0 = frames[0][0] counts, last_t, gaps, dsum = defaultdict(int), {}, defaultdict(int), defaultdict(float) events, rows = [], [] prev_term = prev_run = None bad_cs = defaultdict(int) bad_cnt = defaultdict(int) prev_cnt = {} for ms, aid, b in frames: t = (ms - t0) / 1000 counts[aid] += 1 if aid in last_t: dt = t - last_t[aid] if 0 < dt < 1: dsum[aid] += dt gaps[aid] += 1 last_t[aid] = t if a.check: if aid in CHECKSUMS: idx, k = CHECKSUMS[aid] if len(b) > idx: rest = [b[i] for i in range(len(b)) if i != idx] if (ocsum(rest) + k) & 0xFF != b[idx]: bad_cs[aid] += 1 if aid in COUNTERS: by, sh = COUNTERS[aid] if len(b) > by: c = (b[by] >> sh) & 0x0F p = prev_cnt.get(aid) if p is not None and c != (p + 1) % 15: bad_cnt[aid] += 1 prev_cnt[aid] = c if aid == 0x130 and b[0] != prev_term: events.append((t, "terminal", TERMINAL.get(b[0], hex(b[0])))) prev_term = b[0] if aid == 0x0AA: run = u16le(b, 4) > 0 if run != prev_run: events.append((t, "engine", "running" if run else "stopped")) prev_run = run if aid in SIGNALS: for name, (fn, unit) in SIGNALS[aid].items(): try: rows.append((round(t, 3), f"0x{aid:03X}", name, fn(b), unit)) except Exception: pass dur = (frames[-1][0] - t0) / 1000 print(f"{len(frames)} frames · {dur:.2f} s · {len(counts)} arbitration IDs\n") # bus-silent windows silent, pt = [], (frames[0][0] - t0) / 1000 for ms, _, _ in frames: t = (ms - t0) / 1000 if t - pt > 2: silent.append((pt, t)) pt = t for s, e in silent: print(f" bus silent {s:6.2f} → {e:6.2f} s ({e - s:.1f} s)") if silent: print() print("timeline") for t, kind, val in events: print(f" {t:6.2f}s {kind:9s} {val}") print("\n%-7s %-22s %7s %6s %s" % ("ID", "message", "Hz", "frames", "last data")) seen_last = {} for ms, aid, b in frames: seen_last[aid] = b for aid in sorted(counts): hz = gaps[aid] / dsum[aid] if dsum[aid] else counts[aid] / dur print("0x%03X %-22s %7.1f %6d %s" % (aid, NAMES.get(aid, ""), hz, counts[aid], seen_last[aid].hex().upper())) if a.check: print("\nchecksum / counter check") for aid in sorted(set(CHECKSUMS) | set(COUNTERS)): if aid not in counts: continue cs = f"{counts[aid] - bad_cs[aid]}/{counts[aid]}" if aid in CHECKSUMS else " —" ct = f"{counts[aid] - bad_cnt[aid]}/{counts[aid]}" if aid in COUNTERS else " —" print(f" 0x{aid:03X} checksum {cs:>12} counter {ct:>12}") if a.out: with open(a.out, "w", newline="") as f: w = csv.writer(f) w.writerow(["t_s", "id", "signal", "value", "unit"]) w.writerows(rows) print(f"\nwrote {len(rows)} signal samples to {a.out}") if __name__ == "__main__": main()