"""Live/replay dashboard for the BMW E8x PT-CAN capture. "/Users/luca/Projects/Stuurhuis CAN/.venv/bin/python" -m streamlit run dashboard/app.py Two data sources, selectable in the sidebar: - Live capture: reads the CANdapter in the background and feeds the dashboard in real time. Optionally records the session to a new CSV. - Replay file: steps through an existing capture_*.csv at an adjustable speed, so you can scrub through what happened without the car running. Decoding uses decoder/ptcan_decoder.py, which wraps the signal definitions from files/decode_ptcan.py / files/PTCAN_protocol.md / files/bmw_e8x_ptcan.dbc. """ from __future__ import annotations import sys import time from pathlib import Path import pandas as pd import plotly.graph_objects as go import streamlit as st from plotly.subplots import make_subplots ROOT = Path(__file__).resolve().parents[1] sys.path.insert(0, str(ROOT)) from decoder.live_source import LiveCapture # noqa: E402 from decoder.ptcan_decoder import FrameStore, NAMES # noqa: E402 from decoder.replay_source import ReplayPlayer, load_capture # noqa: E402 CAPTURES_DIR = ROOT / "captures" DEFAULT_PORT = "/dev/cu.usbserial-DNBJV4F5" REFRESH_INTERVAL = 0.25 # seconds between dashboard reruns while a source is active st.set_page_config(page_title="PT-CAN — BMW E8x", layout="wide") @st.cache_resource def get_store() -> FrameStore: return FrameStore() @st.cache_resource def get_live() -> LiveCapture: return LiveCapture() @st.cache_resource def get_player(csv_path: str) -> ReplayPlayer: return ReplayPlayer(load_capture(Path(csv_path))) store = get_store() live = get_live() # ---------------------------------------------------------------- sidebar -- st.sidebar.title("PT-CAN — BMW E8x") mode = st.sidebar.radio("Source", ["Live capture", "Replay file"]) active = False # whether the dashboard should keep auto-rerunning if mode == "Live capture": port = st.sidebar.text_input("Port", DEFAULT_PORT) bitrate = st.sidebar.selectbox("Bitrate", [500_000, 250_000, 125_000, 1_000_000], index=0) record = st.sidebar.checkbox("Record session to CSV", value=True) c1, c2 = st.sidebar.columns(2) if c1.button("Start", disabled=live.is_running(), width="stretch"): store.reset() record_path = CAPTURES_DIR / time.strftime("capture_%Y%m%d_%H%M%S.csv") if record else None live.start(port, bitrate, store, record_path=record_path) if record_path: st.sidebar.success(f"Recording to {record_path.name}") if c2.button("Stop", disabled=not live.is_running(), width="stretch"): live.stop() if live.error: st.sidebar.error(f"Adapter error: {live.error}") elif live.is_running(): st.sidebar.success(f"Connected — {live.frame_count} frames") else: st.sidebar.caption("Stopped") active = live.is_running() else: files = sorted(p.name for p in CAPTURES_DIR.glob("capture*.csv")) if not files: st.sidebar.warning("No capture_*.csv files found in captures/.") st.stop() chosen = st.sidebar.selectbox("Capture file", files) player = get_player(str(CAPTURES_DIR / chosen)) if "replay_playing" not in st.session_state: st.session_state.replay_playing = False st.session_state.replay_wall = None speed = st.sidebar.slider("Playback speed", 0.25, 8.0, 1.0, step=0.25) c1, c2, c3 = st.sidebar.columns(3) label = "Pause" if st.session_state.replay_playing else "Play" if c1.button(label, width="stretch"): st.session_state.replay_playing = not st.session_state.replay_playing st.session_state.replay_wall = time.monotonic() if c2.button("Restart", width="stretch"): player.reset() store.reset() st.session_state.replay_playing = False if c3.button("Seek", width="stretch"): pass # handled via the slider below, this just forces a rerun seek_to = st.sidebar.slider("Position (s)", 0.0, max(player.duration, 0.1), min(player.clock, player.duration), step=0.5) if abs(seek_to - player.clock) > 0.5: player.seek(seek_to, store) if st.session_state.replay_playing: now = time.monotonic() dt = min(now - (st.session_state.replay_wall or now), 1.0) st.session_state.replay_wall = now player.advance(dt, speed, store) if player.at_end(): st.session_state.replay_playing = False st.sidebar.caption(f"{player.clock:.1f}s / {player.duration:.1f}s") active = st.session_state.replay_playing st.sidebar.divider() window_s = st.sidebar.slider("Chart window (s)", 5, 120, 30) if st.sidebar.button("Clear dashboard data"): store.reset() # --------------------------------------------------------------- helpers -- COLORS = {"rpm": "#FFA733", "steer": "#FF5C4D", "speed": "#56AEF2", "term": "#5FD6A0", "volt": "#B48CF2", "fuel": "#E8C87A"} def signal_value(snapshot: dict, arbitration_id: int, name: str): entry = snapshot["latest"].get(arbitration_id) if not entry: return None, "" return entry["decoded"].get(name, (None, "")) snap = store.snapshot() # ------------------------------------------------------------------ KPIs -- st.subheader("Key signals") k1, k2, k3, k4, k5, k6 = st.columns(6) angle, _ = signal_value(snap, 0x0C4, "steering_angle") rate, _ = signal_value(snap, 0x0C4, "steering_rate") rpm, _ = signal_value(snap, 0x0AA, "engine_speed") speed_kmh, _ = signal_value(snap, 0x1A0, "road_speed") volt, _ = signal_value(snap, 0x0A9, "system_voltage") fuel, _ = signal_value(snap, 0x1D0, "fuel_accum") terminal, _ = signal_value(snap, 0x130, "terminal") k1.metric("Steering angle (0x0C4)", f"{angle:.1f}°" if angle is not None else "—") k2.metric("Steering rate (0x0C4)", f"{rate:.0f}°/s" if rate is not None else "—") k3.metric("Engine speed (0x0AA)", f"{rpm:.0f} rpm" if rpm is not None else "—") k4.metric("Road speed (0x1A0)", f"{speed_kmh:.1f} km/h" if speed_kmh is not None else "—") k5.metric("System voltage (0x0A9)", f"{volt:.2f} V" if volt is not None else "—") k6.metric("Terminal (0x130)", str(terminal) if terminal is not None else "—") # --------------------------------------------------------------- charts -- st.subheader("Decoded channels over time") fig = make_subplots( rows=4, cols=1, shared_xaxes=True, vertical_spacing=0.04, specs=[[{}], [{"secondary_y": True}], [{}], [{}]], subplot_titles=("Engine speed (rpm)", "Steering angle / rate", "System voltage (V)", "Road speed (km/h)"), ) xs, ys = store.series(0x0AA, "engine_speed", window_s) fig.add_trace(go.Scatter(x=xs, y=ys, mode="lines", line=dict(color=COLORS["rpm"]), name="rpm"), row=1, col=1) xs, ys = store.series(0x0C4, "steering_angle", window_s) fig.add_trace(go.Scatter(x=xs, y=ys, mode="lines", line=dict(color=COLORS["steer"]), name="angle (deg)"), row=2, col=1, secondary_y=False) xs, ys = store.series(0x0C4, "steering_rate", window_s) fig.add_trace(go.Scatter(x=xs, y=ys, mode="lines", line=dict(color="#7a4238"), name="rate (deg/s)"), row=2, col=1, secondary_y=True) xs, ys = store.series(0x0A9, "system_voltage", window_s) fig.add_trace(go.Scatter(x=xs, y=ys, mode="lines", line=dict(color=COLORS["volt"]), name="voltage"), row=3, col=1) xs, ys = store.series(0x1A0, "road_speed", window_s) fig.add_trace(go.Scatter(x=xs, y=ys, mode="lines", line=dict(color=COLORS["speed"]), name="speed"), row=4, col=1) fig.update_layout(height=650, showlegend=False, margin=dict(l=40, r=20, t=30, b=20), template="plotly_dark") st.plotly_chart(fig, width="stretch") # --------------------------------------------------------------- monitor -- st.subheader("Bus monitor") rows = [] for aid in sorted(snap["latest"]): entry = snap["latest"][aid] decoded = ", ".join(f"{k}={v[0]:.2f}{v[1]}" if isinstance(v[0], float) else f"{k}={v[0]}{v[1]}" for k, v in entry["decoded"].items()) rows.append( { "ID": f"0x{aid:03X}", "Message": NAMES.get(aid, ""), "Hz": round(store.hz(aid), 1), "Count": snap["counts"].get(aid, 0), "Data": entry["data"].hex(" ").upper(), "Decoded": decoded, } ) if rows: df = pd.DataFrame(rows).sort_values("ID") st.dataframe(df, width="stretch", height=380, hide_index=True) else: st.info("No frames yet — start a live capture or play a replay file.") # ------------------------------------------------------------- bit view -- st.subheader("Bit inspector") st.caption("Pick an ID to see its raw bytes bit-by-bit. Highlighted bits changed since the previous frame — the fastest way to spot an unidentified signal (candidates from the protocol notes: 0x1B6 and 0x335).") ids_present = sorted(snap["latest"]) if ids_present: default_idx = ids_present.index(0x1B6) if 0x1B6 in ids_present else 0 picked = st.selectbox("Arbitration ID", ids_present, index=default_idx, format_func=lambda i: f"0x{i:03X} {NAMES.get(i, '')}") entry = snap["latest"][picked] data, changed = entry["data"], entry["changed"] html = ["
"] html.append("" + "".join(f"" for b in range(7, -1, -1)) + "") for i, byte in enumerate(data): cells = [] chg_byte = changed[i] if changed else 0 for bit in range(7, -1, -1): on = (byte >> bit) & 1 toggled = (chg_byte >> bit) & 1 bg = "#FF5C4D" if toggled else ("#3a4a58" if on else "#151D25") fg = "#0E1419" if on or toggled else "#485A67" cells.append(f"") html.append(f"{''.join(cells)}") html.append("
byte{b}hex
{on}
{i}0x{byte:02X}
") st.markdown("".join(html), unsafe_allow_html=True) if entry["decoded"]: st.write({name: f"{val:.3f}{unit}" if isinstance(val, float) else f"{val}{unit}" for name, (val, unit) in entry["decoded"].items()}) else: st.info("No frames yet.") # ----------------------------------------------------------- keep alive -- if active: time.sleep(REFRESH_INTERVAL) st.rerun()