Luca
c32c4645b5
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
56 lines
2.1 KiB
Python
56 lines
2.1 KiB
Python
"""Convert a gateway session log (t, direction, arbitration_id, is_extended,
|
|
dlc, data_hex, relayed) into a plain capture CSV (timestamp_ms,
|
|
arbitration_id, is_extended, dlc, data_hex) that replay_source.py /
|
|
gateway/replay_to_bus.py can play back.
|
|
|
|
Usage:
|
|
python tools/gateway_log_to_capture.py captures/gateway_20260829_164540.csv \\
|
|
--direction car->eps --out captures/replay_car_to_eps.csv
|
|
|
|
Only rows that were actually relayed are kept by default (--include-blocked
|
|
to keep everything, e.g. to inspect what was withheld).
|
|
"""
|
|
import argparse
|
|
import csv
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
|
|
def main() -> int:
|
|
parser = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
|
|
parser.add_argument("gateway_log", help="Gateway session CSV to convert")
|
|
parser.add_argument("--direction", default="car->eps", choices=["car->eps", "eps->car"])
|
|
parser.add_argument("--out", required=True, help="Output capture CSV path")
|
|
parser.add_argument("--include-blocked", action="store_true", help="Also keep frames that were NOT relayed")
|
|
args = parser.parse_args()
|
|
|
|
src = Path(args.gateway_log)
|
|
out = Path(args.out)
|
|
|
|
written = 0
|
|
with open(src, newline="") as fin, open(out, "w", newline="") as fout:
|
|
reader = csv.DictReader(fin)
|
|
writer = csv.writer(fout)
|
|
writer.writerow(["timestamp_ms", "arbitration_id", "is_extended", "dlc", "data_hex"])
|
|
for row in reader:
|
|
if row["direction"] != args.direction:
|
|
continue
|
|
if not args.include_blocked and row["relayed"] != "1":
|
|
continue
|
|
writer.writerow(
|
|
[
|
|
round(float(row["t"]) * 1000),
|
|
row["arbitration_id"],
|
|
row["is_extended"],
|
|
row["dlc"],
|
|
row["data_hex"],
|
|
]
|
|
)
|
|
written += 1
|
|
|
|
print(f"Wrote {written} frames ({args.direction}) to {out}", file=sys.stderr)
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|