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
57 lines
2 KiB
Python
57 lines
2 KiB
Python
"""Load a capture CSV and step through it as a virtual live feed."""
|
|
from __future__ import annotations
|
|
|
|
import csv
|
|
from pathlib import Path
|
|
|
|
|
|
def load_capture(path: Path) -> list[tuple[float, int, bytes]]:
|
|
"""Returns (t_seconds, arbitration_id, data) sorted by time, t relative to first frame.
|
|
|
|
Mirrors files/decode_ptcan.py: sort by timestamp_ms first since these logs can contain
|
|
the odd corrupt row, and the adapter's ms counter wraps every 60000ms anyway.
|
|
"""
|
|
rows = []
|
|
with open(path, newline="") as f:
|
|
for row in csv.DictReader(f):
|
|
hx = row["data_hex"].strip()
|
|
rows.append((int(row["timestamp_ms"]), int(row["arbitration_id"], 16), bytes.fromhex(hx)))
|
|
rows.sort(key=lambda r: r[0])
|
|
if not rows:
|
|
return []
|
|
t0 = rows[0][0]
|
|
return [((ms - t0) / 1000, aid, data) for ms, aid, data in rows]
|
|
|
|
|
|
class ReplayPlayer:
|
|
"""Feeds a FrameStore at a scaled real-time rate, tracking position in the log."""
|
|
|
|
def __init__(self, frames: list[tuple[float, int, bytes]]):
|
|
self.frames = frames
|
|
self.duration = frames[-1][0] if frames else 0.0
|
|
self.index = 0
|
|
self.clock = 0.0
|
|
|
|
def reset(self) -> None:
|
|
self.index = 0
|
|
self.clock = 0.0
|
|
|
|
def at_end(self) -> bool:
|
|
return self.index >= len(self.frames)
|
|
|
|
def advance(self, dt_wall: float, speed: float, store) -> None:
|
|
self.clock += dt_wall * speed
|
|
while self.index < len(self.frames) and self.frames[self.index][0] <= self.clock:
|
|
t, aid, data = self.frames[self.index]
|
|
store.feed(aid, data, t)
|
|
self.index += 1
|
|
|
|
def seek(self, t: float, store) -> None:
|
|
"""Jump to time t, replaying everything up to it into a freshly reset store."""
|
|
store.reset()
|
|
self.index = 0
|
|
self.clock = t
|
|
while self.index < len(self.frames) and self.frames[self.index][0] <= self.clock:
|
|
frame_t, aid, data = self.frames[self.index]
|
|
store.feed(aid, data, frame_t)
|
|
self.index += 1
|