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
69 lines
2.5 KiB
Python
69 lines
2.5 KiB
Python
"""CLI entry point for the car<->EPS gateway (headless, no GUI).
|
|
|
|
Usage:
|
|
python gateway/run_gateway.py \\
|
|
--car-port /dev/cu.usbserial-DNBJV4F5 --car-bitrate 500000 \\
|
|
--eps-port /dev/cu.usbmodemXXXX --eps-bitrate 500000 \\
|
|
[--block 0x1D6 0x380] [--log captures/gateway_log.csv]
|
|
|
|
Starts fully transparent (every frame relayed both ways) unless --block is
|
|
given. For interactive control (toggle IDs while it's running, live
|
|
monitor), use gateway/gateway_app.py (Streamlit) instead.
|
|
"""
|
|
import argparse
|
|
import sys
|
|
import time
|
|
from pathlib import Path
|
|
|
|
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
|
|
from adapters.candapter import Candapter # noqa: E402
|
|
from adapters.slcan_adapter import SlcanAdapter # noqa: E402
|
|
from gateway.gateway import Gateway, GatewayLogger # noqa: E402
|
|
from gateway.rules import FilterRules # noqa: E402
|
|
|
|
DEFAULT_CAR_PORT = "/dev/cu.usbserial-DNBJV4F5"
|
|
|
|
|
|
def main() -> int:
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument("--car-port", default=DEFAULT_CAR_PORT)
|
|
parser.add_argument("--car-bitrate", type=int, default=500_000)
|
|
parser.add_argument("--eps-port", required=True, help="Serial port of the CAN-IO board's USB bridge")
|
|
parser.add_argument("--eps-bitrate", type=int, default=500_000)
|
|
parser.add_argument("--block", nargs="*", default=[], help="Arbitration IDs (hex) to block in BOTH directions")
|
|
parser.add_argument("--log", default=None, help="CSV path to log every frame seen (relayed or not)")
|
|
args = parser.parse_args()
|
|
|
|
car = Candapter(args.car_port, args.car_bitrate, timestamps=True)
|
|
eps = SlcanAdapter(args.eps_port, args.eps_bitrate)
|
|
|
|
car_to_eps = FilterRules()
|
|
eps_to_car = FilterRules()
|
|
for tok in args.block:
|
|
aid = int(tok, 16)
|
|
car_to_eps.set_allow(aid, False)
|
|
eps_to_car.set_allow(aid, False)
|
|
|
|
logger = GatewayLogger(Path(args.log)) if args.log else None
|
|
gw = Gateway(car, eps, car_to_eps, eps_to_car, logger=logger)
|
|
gw.start()
|
|
|
|
print(f"Gateway running: car({args.car_port}) <-> eps({args.eps_port}). Ctrl+C to stop.")
|
|
try:
|
|
while True:
|
|
time.sleep(2)
|
|
passed, blocked = gw.stats.snapshot()
|
|
print(f"passed IDs: {len(passed)} blocked IDs: {len(blocked)} "
|
|
f"total passed: {sum(passed.values())} total blocked: {sum(blocked.values())}")
|
|
except KeyboardInterrupt:
|
|
pass
|
|
finally:
|
|
gw.stop()
|
|
car.close()
|
|
eps.close()
|
|
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|