BMW_E8x_EPS/tools/extract_dio_timeline.py
Luca c32c4645b5 Reverse-engineer BMW E8x EPS for standalone operation
Tooling and findings for running an E8x/E9x electric power steering unit
outside its donor car, e.g. in an EV conversion.

Headline result: the EPS needs only two CAN messages plus a 12V enable
wire, not the 69-message set the car puts on the bus:
  0x130 CAS terminal status (100ms) brings the unit up
  0x1A0 DSC road speed (20ms) sets the assist level
Total required rate is 60 frames/s. Protocol write-up, including what is
proven vs. inferred and the open questions, is in eps-comms/.

Contents:
  adapters/   CANdapter (its SLCAN dialect differs) and generic SLCAN
  gateway/    car<->EPS relay, replay, message bench, EPS controller,
              4-tab Streamlit UI
  decoder/    PT-CAN frame decoding and live/replay sources
  can-io/     XIAO ESP32-S3 firmware: CAN IO board + USB-CAN bridge with
              a CAN-independent digital IO channel
  tools/      capture, bitrate scan, startup-order and session analysis,
              checksum solver
  captures/   reference working session + the replay set eps_control reads
2026-08-29 19:34:43 +02:00

59 lines
2 KiB
Python

"""Extract the CAN-IO board's own digital IO timeline (IN1-4/OUT1-2) from a
gateway session log, by pulling out its status frames (arbitration ID
0x100, see can-io/PROTOCOL.md) - the board sends one immediately on any
input/output change plus a 1s heartbeat, so this recovers when each
channel was high/low over the session (at that resolution - a change that
gets undone within a couple of debounce/scheduling ticks might not get its
own frame, only the surrounding samples).
Usage:
python tools/extract_dio_timeline.py captures/gateway_20260829_164540.csv \\
--out captures/dio_20260829.csv
"""
import argparse
import csv
from pathlib import Path
CAN_ID_STATUS = 0x100
def bits(n: int, count: int) -> str:
return "".join(str((n >> i) & 1) for i in range(count))
def main() -> int:
parser = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
parser.add_argument("gateway_log")
parser.add_argument("--out", required=True, help="Output CSV path (t, inputs, outputs)")
args = parser.parse_args()
samples = []
with open(args.gateway_log, newline="") as f:
for row in csv.DictReader(f):
if int(row["arbitration_id"], 16) != CAN_ID_STATUS:
continue
data = bytes.fromhex(row["data_hex"])
if len(data) < 2:
continue
samples.append((float(row["t"]), data[0], data[1]))
with open(args.out, "w", newline="") as f:
w = csv.writer(f)
w.writerow(["t", "inputs", "outputs"])
for t, ins, outs in samples:
w.writerow([t, ins, outs])
print(f"Wrote {len(samples)} status samples to {args.out}\n")
print("Transitions (IN1-4 / OUT1-2, bit0 first):")
prev = None
for t, ins, outs in samples:
cur = (ins, outs)
if cur != prev:
print(f" t={t:8.3f}s IN={bits(ins, 4)} OUT={bits(outs, 2)}")
prev = cur
return 0
if __name__ == "__main__":
raise SystemExit(main())