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
95 lines
4 KiB
Python
95 lines
4 KiB
Python
"""Work out the message order that precedes the EPS coming online.
|
|
|
|
Takes a gateway session log where the EPS *did* wake up, finds the moment
|
|
it first transmits (its own IDs, not relayed car traffic), and reports what
|
|
the car sent before that - first-seen times, and how the CAS terminal
|
|
message 0x130 progressed. That progression is the thing a synthetic
|
|
transmit set has to reproduce: a static payload can't walk the EPS through
|
|
the same states the car does.
|
|
|
|
Usage:
|
|
python tools/analyze_startup_order.py captures/gateway_20260829_164540.csv
|
|
"""
|
|
import argparse
|
|
import csv
|
|
from collections import defaultdict
|
|
from pathlib import Path
|
|
|
|
# The EPS's own transmissions, per eps-comms/findings.md. 0x100 is our
|
|
# CAN-IO board's status frame, so it is deliberately not in this set.
|
|
EPS_IDS = {0x1FB, 0x4B0, 0x5B0}
|
|
|
|
TERMINAL = {0x00: "off", 0x40: "terminal_R", 0x41: "terminal_15",
|
|
0x45: "engine_running", 0x55: "cranking", 0x80: "wake_up"}
|
|
|
|
|
|
def main() -> int:
|
|
parser = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
|
|
parser.add_argument("gateway_log")
|
|
parser.add_argument("--window", type=float, default=None,
|
|
help="Only consider car frames within N seconds before the EPS wakes")
|
|
args = parser.parse_args()
|
|
|
|
rows = []
|
|
with open(args.gateway_log, newline="") as f:
|
|
for row in csv.DictReader(f):
|
|
rows.append((float(row["t"]), row["direction"], int(row["arbitration_id"], 16),
|
|
bytes.fromhex(row["data_hex"])))
|
|
rows.sort(key=lambda r: r[0])
|
|
|
|
wake = next((t for t, d, aid, _ in rows if d == "eps->car" and aid in EPS_IDS), None)
|
|
if wake is None:
|
|
print("The EPS never transmitted in this log - pick one where it came online.")
|
|
return 1
|
|
print(f"EPS first transmits at t={wake:.3f}s\n")
|
|
|
|
first_seen = {}
|
|
counts = defaultdict(int)
|
|
for t, direction, aid, _ in rows:
|
|
if direction != "car->eps":
|
|
continue
|
|
counts[aid] += 1
|
|
first_seen.setdefault(aid, t)
|
|
|
|
before = {aid: t for aid, t in first_seen.items() if t < wake}
|
|
after = {aid: t for aid, t in first_seen.items() if t >= wake}
|
|
if args.window is not None:
|
|
before = {aid: t for aid, t in before.items() if t >= wake - args.window}
|
|
|
|
print(f"Car IDs present BEFORE the EPS woke ({len(before)}):")
|
|
for aid, t in sorted(before.items(), key=lambda kv: kv[1]):
|
|
print(f" t={t:7.3f}s 0x{aid:03X} ({counts[aid]} frames total)")
|
|
|
|
if after:
|
|
print(f"\nCar IDs that only appeared AFTER ({len(after)}) - not needed to wake it:")
|
|
for aid, t in sorted(after.items(), key=lambda kv: kv[1]):
|
|
print(f" t={t:7.3f}s 0x{aid:03X}")
|
|
|
|
print("\n0x130 CAS terminal progression (the state walk to reproduce):")
|
|
prev = None
|
|
for t, direction, aid, data in rows:
|
|
if direction != "car->eps" or aid != 0x130 or not data:
|
|
continue
|
|
if data[0] != prev:
|
|
marker = " <-- EPS wakes around here" if prev is not None and t >= wake > 0 and abs(t - wake) < 1.5 else ""
|
|
print(f" t={t:7.3f}s b0=0x{data[0]:02X} {TERMINAL.get(data[0], '?'):15s}{marker}")
|
|
prev = data[0]
|
|
|
|
print("\nPayload variety in the pre-wake window (how much each ID actually moves):")
|
|
seen_payloads = defaultdict(set)
|
|
for t, direction, aid, data in rows:
|
|
if direction == "car->eps" and t < wake:
|
|
seen_payloads[aid].add(data)
|
|
static = [aid for aid, s in seen_payloads.items() if len(s) == 1]
|
|
varying = sorted(((len(s), aid) for aid, s in seen_payloads.items() if len(s) > 1), reverse=True)
|
|
print(f" {len(static)} IDs sent one fixed payload: " +
|
|
", ".join(f"0x{a:03X}" for a in sorted(static)))
|
|
print(" IDs whose payload changed (these need live counters/values, not a frozen capture):")
|
|
for n, aid in varying[:15]:
|
|
print(f" 0x{aid:03X}: {n} distinct payloads")
|
|
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|