first commit
This commit is contained in:
440
app/gui.py
Normal file
440
app/gui.py
Normal file
@@ -0,0 +1,440 @@
|
||||
# gui.py — Tk-App mit Interface-Dropdown, Link Up/Down, Settings-View/Save + CAN-Trace
|
||||
from __future__ import annotations
|
||||
import json
|
||||
import threading
|
||||
import time
|
||||
import tkinter as tk
|
||||
from tkinter import ttk, messagebox
|
||||
from collections import deque, defaultdict
|
||||
|
||||
import can # nur für Trace-Reader
|
||||
|
||||
from .config import load_settings, setup_logging, SETTINGS_PATH, APP_ROOT
|
||||
from .simulator import EcuState, DrivelineModel
|
||||
from .can import (
|
||||
ObdResponder, make_speed_response, make_rpm_response,
|
||||
list_can_ifaces, link_up, link_down,
|
||||
have_cap_netadmin, link_state, link_kind
|
||||
)
|
||||
|
||||
|
||||
# ---------- kleine Trace-Helfer ----------
|
||||
class TraceCollector:
|
||||
"""
|
||||
Liest mit eigenem BufferedReader vom SocketCAN und sammelt Frames.
|
||||
- stream_buffer: deque mit (ts, id, dlc, data_bytes)
|
||||
- aggregate: dict[(id, dir)] -> {count, last_ts, last_data}
|
||||
"""
|
||||
def __init__(self, channel: str):
|
||||
self.channel = channel
|
||||
self.bus = None
|
||||
self._run = threading.Event(); self._run.set()
|
||||
self._thread = threading.Thread(target=self._rx_loop, name="CAN-TRACE", daemon=True)
|
||||
self.stream_buffer = deque(maxlen=2000)
|
||||
self.lock = threading.Lock()
|
||||
|
||||
def _open(self):
|
||||
self._close()
|
||||
self.bus = can.interface.Bus(channel=self.channel, interface="socketcan")
|
||||
|
||||
def _close(self):
|
||||
try:
|
||||
if self.bus: self.bus.shutdown()
|
||||
except Exception: pass
|
||||
self.bus = None
|
||||
|
||||
def start(self):
|
||||
self._thread.start()
|
||||
|
||||
def stop(self):
|
||||
self._run.clear()
|
||||
try: self._thread.join(timeout=1.0)
|
||||
except RuntimeError: pass
|
||||
self._close()
|
||||
|
||||
def _rx_loop(self):
|
||||
backoff = 0.5
|
||||
while self._run.is_set():
|
||||
if self.bus is None:
|
||||
if link_state(self.channel) == "UP":
|
||||
try:
|
||||
self._open()
|
||||
backoff = 0.5
|
||||
except Exception:
|
||||
time.sleep(backoff); backoff = min(5.0, backoff*1.7)
|
||||
continue
|
||||
else:
|
||||
time.sleep(0.5); continue
|
||||
try:
|
||||
msg = self.bus.recv(0.05)
|
||||
if msg and not msg.is_error_frame and not msg.is_remote_frame:
|
||||
ts = time.time()
|
||||
with self.lock:
|
||||
self.stream_buffer.append((ts, msg.arbitration_id, msg.dlc, bytes(msg.data)))
|
||||
except (can.CanOperationError, OSError):
|
||||
# IF down → ruhig schließen, kein Traceback
|
||||
self._close()
|
||||
time.sleep(0.5)
|
||||
except Exception:
|
||||
time.sleep(0.05)
|
||||
|
||||
def snapshot_stream(self):
|
||||
with self.lock:
|
||||
return list(self.stream_buffer)
|
||||
|
||||
|
||||
def launch_gui():
|
||||
cfg = load_settings()
|
||||
logger = setup_logging(cfg)
|
||||
|
||||
# read config values
|
||||
can_iface = (cfg.get("can", {}).get("interface")) or "can0"
|
||||
resp_id_raw = (cfg.get("can", {}).get("resp_id")) or "0x7E8"
|
||||
try:
|
||||
resp_id = int(resp_id_raw, 16) if isinstance(resp_id_raw, str) else int(resp_id_raw)
|
||||
except Exception:
|
||||
resp_id = 0x7E8
|
||||
timeout_ms = cfg.get("can", {}).get("timeout_ms", 200)
|
||||
bitrate = cfg.get("can", {}).get("baudrate", 500000)
|
||||
|
||||
ecu = EcuState(DrivelineModel())
|
||||
responder = ObdResponder(interface=can_iface, resp_id=resp_id, timeout_ms=timeout_ms, logger=logger)
|
||||
|
||||
# register providers
|
||||
responder.register_pid(0x0D, lambda: make_speed_response(int(round(ecu.snapshot()[3]))))
|
||||
responder.register_pid(0x0C, lambda: make_rpm_response(int(ecu.snapshot()[2])))
|
||||
|
||||
# physics thread
|
||||
running = True
|
||||
def physics_loop():
|
||||
while running:
|
||||
ecu.update()
|
||||
time.sleep(0.02)
|
||||
t = threading.Thread(target=physics_loop, daemon=True)
|
||||
t.start()
|
||||
|
||||
# Trace-Collector (eigener Bus, hört alles auf can_iface)
|
||||
tracer = TraceCollector(can_iface)
|
||||
tracer.start()
|
||||
|
||||
# --- Tk UI ---
|
||||
root = tk.Tk()
|
||||
root.title("OBD-II ECU Simulator – SocketCAN")
|
||||
|
||||
# window size from cfg
|
||||
try:
|
||||
w = int(cfg["ui"]["window"]["width"]); h = int(cfg["ui"]["window"]["height"])
|
||||
root.geometry(f"{w}x{h}")
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
# fonts/styles
|
||||
family = cfg.get("ui", {}).get("font_family", "TkDefaultFont")
|
||||
size = int(cfg.get("ui", {}).get("font_size", 10))
|
||||
style = ttk.Style()
|
||||
style.configure("TLabel", font=(family, size))
|
||||
style.configure("Header.TLabel", font=(family, size+2, "bold"))
|
||||
style.configure("TButton", font=(family, size))
|
||||
|
||||
# layout
|
||||
root.columnconfigure(0, weight=1); root.rowconfigure(0, weight=1)
|
||||
main = ttk.Frame(root, padding=10); main.grid(row=0, column=0, sticky="nsew")
|
||||
main.columnconfigure(1, weight=1)
|
||||
|
||||
# === Controls: Gear + Throttle ===
|
||||
ttk.Label(main, text="Gang").grid(row=0, column=0, sticky="w")
|
||||
gear_var = tk.IntVar(value=0)
|
||||
gear_box = ttk.Combobox(main, textvariable=gear_var, state="readonly", values=[0,1,2,3,4,5,6], width=5)
|
||||
gear_box.grid(row=0, column=1, sticky="w", padx=(6,12))
|
||||
gear_box.bind("<<ComboboxSelected>>", lambda _e: ecu.set_gear(gear_var.get()))
|
||||
|
||||
ttk.Label(main, text="Gas (%)").grid(row=1, column=0, sticky="w")
|
||||
thr = ttk.Scale(main, from_=0, to=100, orient="horizontal",
|
||||
command=lambda v: ecu.set_throttle(int(float(v))))
|
||||
thr.set(0)
|
||||
thr.grid(row=1, column=1, sticky="ew", padx=(6,12))
|
||||
|
||||
lbl_speed = ttk.Label(main, text="Speed: 0 km/h", style="Header.TLabel")
|
||||
lbl_rpm = ttk.Label(main, text="RPM: 0")
|
||||
lbl_speed.grid(row=2, column=0, columnspan=2, sticky="w", pady=(10,0))
|
||||
lbl_rpm.grid(row=3, column=0, columnspan=2, sticky="w")
|
||||
|
||||
# === CAN Panel ===
|
||||
sep = ttk.Separator(main); sep.grid(row=4, column=0, columnspan=2, sticky="ew", pady=(10,10))
|
||||
|
||||
can_frame = ttk.LabelFrame(main, text="CAN & Settings", padding=10)
|
||||
can_frame.grid(row=5, column=0, columnspan=2, sticky="nsew")
|
||||
can_frame.columnconfigure(1, weight=1)
|
||||
|
||||
ttk.Label(can_frame, text="Interface").grid(row=0, column=0, sticky="w")
|
||||
iface_var = tk.StringVar(value=can_iface)
|
||||
iface_list = list_can_ifaces() or [can_iface]
|
||||
iface_dd = ttk.Combobox(can_frame, textvariable=iface_var, values=iface_list, state="readonly", width=12)
|
||||
iface_dd.grid(row=0, column=1, sticky="w", padx=(6,12))
|
||||
|
||||
def refresh_ifaces():
|
||||
lst = list_can_ifaces()
|
||||
if not lst:
|
||||
messagebox.showwarning("Interfaces", "Keine can*/vcan* Interfaces gefunden.")
|
||||
return
|
||||
iface_dd.config(values=lst)
|
||||
ttk.Button(can_frame, text="Refresh", command=refresh_ifaces).grid(row=0, column=2, padx=4)
|
||||
|
||||
ttk.Label(can_frame, text="RESP-ID (hex)").grid(row=1, column=0, sticky="w")
|
||||
resp_var = tk.StringVar(value=f"0x{resp_id:03X}")
|
||||
resp_entry = ttk.Entry(can_frame, textvariable=resp_var, width=10)
|
||||
resp_entry.grid(row=1, column=1, sticky="w", padx=(6,12))
|
||||
|
||||
ttk.Label(can_frame, text="Timeout (ms)").grid(row=2, column=0, sticky="w")
|
||||
to_var = tk.IntVar(value=int(timeout_ms))
|
||||
to_spin = ttk.Spinbox(can_frame, from_=10, to=5000, increment=10, textvariable=to_var, width=8)
|
||||
to_spin.grid(row=2, column=1, sticky="w", padx=(6,12))
|
||||
|
||||
ttk.Label(can_frame, text="Bitrate").grid(row=3, column=0, sticky="w")
|
||||
br_var = tk.IntVar(value=int(bitrate))
|
||||
br_spin = ttk.Spinbox(can_frame, from_=20000, to=1000000, increment=10000, textvariable=br_var, width=10)
|
||||
br_spin.grid(row=3, column=1, sticky="w", padx=(6,12))
|
||||
|
||||
# unter Bitrate-Spinbox
|
||||
set_params = tk.BooleanVar(value=True)
|
||||
ttk.Checkbutton(can_frame, text="Bitrate beim UP setzen", variable=set_params).grid(row=3, column=2, sticky="w")
|
||||
|
||||
# add Kind-Anzeige
|
||||
kind_label = ttk.Label(can_frame, text=f"Kind: {link_kind(can_iface)}")
|
||||
kind_label.grid(row=0, column=3, sticky="w", padx=(12,0))
|
||||
|
||||
# Link control
|
||||
def do_link_up():
|
||||
try:
|
||||
# Kind-Anzeige aktualisieren (falls Interface gewechselt)
|
||||
kind_label.config(text=f"Kind: {link_kind(iface_var.get())}")
|
||||
|
||||
if link_state(iface_var.get()) == "UP":
|
||||
messagebox.showinfo("CAN", f"{iface_var.get()} ist bereits UP")
|
||||
return
|
||||
# NEU: set_params aus Checkbox
|
||||
link_up(iface_var.get(), bitrate=br_var.get(), fd=False, set_params=set_params.get())
|
||||
msg = f"{iface_var.get()} ist UP"
|
||||
if set_params.get():
|
||||
msg += f" @ {br_var.get()} bit/s (falls vom Treiber unterstützt)"
|
||||
else:
|
||||
msg += " (Bitrate unverändert)"
|
||||
messagebox.showinfo("CAN", msg)
|
||||
except PermissionError as e:
|
||||
messagebox.showerror("Berechtigung", str(e))
|
||||
except Exception as e:
|
||||
messagebox.showerror("CAN", f"Link UP fehlgeschlagen:\n{e}")
|
||||
|
||||
def do_link_down():
|
||||
try:
|
||||
if link_state(iface_var.get()) == "DOWN":
|
||||
messagebox.showinfo("CAN", f"{iface_var.get()} ist bereits DOWN")
|
||||
return
|
||||
link_down(iface_var.get())
|
||||
messagebox.showinfo("CAN", f"{iface_var.get()} ist DOWN")
|
||||
except PermissionError as e:
|
||||
messagebox.showerror("Berechtigung", str(e))
|
||||
except Exception as e:
|
||||
messagebox.showerror("CAN", f"Link DOWN fehlgeschlagen:\n{e}")
|
||||
|
||||
btn_up = ttk.Button(can_frame, text="Link UP", command=do_link_up)
|
||||
btn_down = ttk.Button(can_frame, text="Link DOWN", command=do_link_down)
|
||||
btn_up.grid(row=4, column=0, pady=(8,0), sticky="w")
|
||||
btn_down.grid(row=4, column=1, pady=(8,0), sticky="w")
|
||||
|
||||
# Rebind responder
|
||||
def do_rebind():
|
||||
nonlocal can_iface, resp_id, timeout_ms, bitrate, tracer
|
||||
can_iface = iface_var.get()
|
||||
try:
|
||||
new_resp = int(resp_var.get(), 16)
|
||||
except Exception:
|
||||
messagebox.showerror("RESP-ID", "Bitte gültige Hex-Zahl, z.B. 0x7E8")
|
||||
return
|
||||
resp_id = new_resp
|
||||
timeout_ms = to_var.get()
|
||||
bitrate = br_var.get()
|
||||
try:
|
||||
responder.rebind(interface=can_iface, resp_id=resp_id)
|
||||
# Trace-Collector auf neues IF neu binden
|
||||
try:
|
||||
tracer.stop()
|
||||
except Exception:
|
||||
pass
|
||||
tracer = TraceCollector(can_iface)
|
||||
tracer.start()
|
||||
messagebox.showinfo("CAN", f"Responder neu gebunden: {can_iface}, RESP 0x{resp_id:03X}")
|
||||
except Exception as e:
|
||||
messagebox.showerror("CAN", f"Rebind fehlgeschlagen:\n{e}")
|
||||
|
||||
ttk.Button(can_frame, text="Responder Rebind", command=do_rebind).grid(row=4, column=2, pady=(8,0), sticky="w")
|
||||
|
||||
# CAP-Status
|
||||
caps_ok = have_cap_netadmin()
|
||||
cap_label = ttk.Label(can_frame, text=f"CAP_NET_ADMIN: {'yes' if caps_ok else 'no'}")
|
||||
cap_label.grid(row=6, column=0, columnspan=2, sticky="w", pady=(6,0))
|
||||
if not caps_ok:
|
||||
btn_up.state(["disabled"]); btn_down.state(["disabled"])
|
||||
|
||||
# Statusbar
|
||||
status = ttk.Label(main, text=f"CAN: {can_iface} | RESP-ID: 0x{resp_id:03X}", relief="sunken", anchor="w")
|
||||
status.grid(row=6, column=0, columnspan=2, sticky="ew", pady=(10,0))
|
||||
|
||||
# === TRACE-FENSTER (unten) ===
|
||||
trace_frame = ttk.LabelFrame(root, text="CAN Trace", padding=6)
|
||||
trace_frame.grid(row=1, column=0, sticky="nsew")
|
||||
root.rowconfigure(1, weight=1)
|
||||
trace_frame.columnconfigure(0, weight=1)
|
||||
trace_frame.rowconfigure(1, weight=1)
|
||||
|
||||
# Controls: Mode, Pause, Clear, Autoscroll
|
||||
ctrl = ttk.Frame(trace_frame)
|
||||
ctrl.grid(row=0, column=0, sticky="ew", pady=(0,4))
|
||||
ctrl.columnconfigure(5, weight=1)
|
||||
|
||||
mode_var = tk.StringVar(value="stream") # "stream" | "aggregate"
|
||||
ttk.Label(ctrl, text="Modus:").grid(row=0, column=0, sticky="w")
|
||||
mode_dd = ttk.Combobox(ctrl, textvariable=mode_var, state="readonly", width=10,
|
||||
values=["stream", "aggregate"])
|
||||
mode_dd.grid(row=0, column=1, sticky="w", padx=(4,12))
|
||||
|
||||
paused = tk.BooleanVar(value=False)
|
||||
ttk.Checkbutton(ctrl, text="Pause", variable=paused).grid(row=0, column=2, sticky="w")
|
||||
|
||||
autoscroll = tk.BooleanVar(value=True)
|
||||
ttk.Checkbutton(ctrl, text="Auto-Scroll", variable=autoscroll).grid(row=0, column=3, sticky="w")
|
||||
|
||||
def do_clear():
|
||||
nonlocal aggregate_cache
|
||||
tree.delete(*tree.get_children())
|
||||
aggregate_cache.clear()
|
||||
ttk.Button(ctrl, text="Clear", command=do_clear).grid(row=0, column=4, padx=(8,0), sticky="w")
|
||||
|
||||
# Treeview
|
||||
cols_stream = ("time", "dir", "id", "dlc", "data")
|
||||
cols_agg = ("id", "dir", "count", "last_time", "last_dlc", "last_data")
|
||||
|
||||
tree = ttk.Treeview(trace_frame, columns=cols_stream, show="headings", height=10)
|
||||
tree.grid(row=1, column=0, sticky="nsew")
|
||||
sb_y = ttk.Scrollbar(trace_frame, orient="vertical", command=tree.yview)
|
||||
tree.configure(yscrollcommand=sb_y.set)
|
||||
sb_y.grid(row=1, column=1, sticky="ns")
|
||||
|
||||
def setup_columns(mode: str):
|
||||
tree.delete(*tree.get_children())
|
||||
if mode == "stream":
|
||||
tree.config(columns=cols_stream)
|
||||
headings = [("time","Time"),("dir","Dir"),("id","ID"),("dlc","DLC"),("data","Data")]
|
||||
widths = [140, 60, 90, 60, 520]
|
||||
else:
|
||||
tree.config(columns=cols_agg)
|
||||
headings = [("id","ID"),("dir","Dir"),("count","Count"),("last_time","Last Time"),("last_dlc","DLC"),("last_data","Last Data")]
|
||||
widths = [90, 60, 80, 140, 60, 520]
|
||||
for (col, text), w in zip(headings, widths):
|
||||
tree.heading(col, text=text)
|
||||
tree.column(col, width=w, anchor="w")
|
||||
setup_columns("stream")
|
||||
|
||||
aggregate_cache: dict[tuple[int,str], dict] = {}
|
||||
|
||||
def fmt_time(ts: float) -> str:
|
||||
# hh:mm:ss.mmm
|
||||
lt = time.localtime(ts)
|
||||
return time.strftime("%H:%M:%S", lt) + f".{int((ts%1)*1000):03d}"
|
||||
|
||||
def fmt_id(i: int) -> str:
|
||||
return f"0x{i:03X}"
|
||||
|
||||
def fmt_data(b: bytes) -> str:
|
||||
return " ".join(f"{x:02X}" for x in b)
|
||||
|
||||
# periodic UI update
|
||||
last_index = 0
|
||||
def tick():
|
||||
nonlocal can_iface, resp_id, last_index
|
||||
# Top-Status
|
||||
g, tval, rpm, spd = ecu.snapshot()
|
||||
caps = "CAP:yes" if have_cap_netadmin() else "CAP:no"
|
||||
st = link_state(can_iface)
|
||||
lbl_speed.config(text=f"Speed: {int(round(spd))} km/h")
|
||||
lbl_rpm.config(text=f"RPM: {rpm}")
|
||||
st = link_state(can_iface)
|
||||
kd = link_kind(can_iface)
|
||||
status.config(text=f"CAN: {can_iface}({st},{kd}) | RESP-ID: 0x{resp_id:03X} | Gear {g} | Throttle {tval}% | {caps}")
|
||||
|
||||
|
||||
# Trace
|
||||
if not paused.get():
|
||||
mode = mode_var.get()
|
||||
if mode == "stream":
|
||||
setup_columns("stream") if tree["columns"] != cols_stream else None
|
||||
# append new items
|
||||
buf = tracer.snapshot_stream()
|
||||
# nur neue ab letztem Index
|
||||
for ts, cid, dlc, data in buf[last_index:]:
|
||||
# Richtung heuristisch
|
||||
if cid == 0x7DF:
|
||||
d = "RX"
|
||||
elif cid == resp_id:
|
||||
d = "TX"
|
||||
else:
|
||||
d = "?"
|
||||
tree.insert("", "end",
|
||||
values=(fmt_time(ts), d, fmt_id(cid), dlc, fmt_data(data)))
|
||||
# autoscroll
|
||||
if autoscroll.get() and buf[last_index:]:
|
||||
tree.see(tree.get_children()[-1])
|
||||
last_index = len(buf)
|
||||
else:
|
||||
setup_columns("aggregate") if tree["columns"] != cols_agg else None
|
||||
# baue Aggregat neu (leicht, schnell)
|
||||
buf = tracer.snapshot_stream()
|
||||
agg: dict[tuple[int,str], dict] = {}
|
||||
for ts, cid, dlc, data in buf:
|
||||
if cid == 0x7DF:
|
||||
d = "RX"
|
||||
elif cid == resp_id:
|
||||
d = "TX"
|
||||
else:
|
||||
d = "?"
|
||||
key = (cid, d)
|
||||
entry = agg.get(key)
|
||||
if entry is None:
|
||||
agg[key] = {"count":1, "last_ts":ts, "last_dlc":dlc, "last_data":data}
|
||||
else:
|
||||
entry["count"] += 1
|
||||
if ts >= entry["last_ts"]:
|
||||
entry["last_ts"] = ts
|
||||
entry["last_dlc"] = dlc
|
||||
entry["last_data"] = data
|
||||
# nur neu zeichnen, wenn sich was ändert
|
||||
if agg != aggregate_cache:
|
||||
tree.delete(*tree.get_children())
|
||||
# sortiert nach ID, RX vor TX
|
||||
for (cid, d) in sorted(agg.keys(), key=lambda k:(k[0], 0 if k[1]=="RX" else 1)):
|
||||
e = agg[(cid,d)]
|
||||
tree.insert("", "end",
|
||||
values=(fmt_id(cid), d, e["count"],
|
||||
fmt_time(e["last_ts"]),
|
||||
e["last_dlc"], fmt_data(e["last_data"])))
|
||||
aggregate_cache.clear()
|
||||
aggregate_cache.update(agg)
|
||||
|
||||
root.after(50, tick)
|
||||
|
||||
tick()
|
||||
|
||||
def on_close():
|
||||
nonlocal running
|
||||
running = False
|
||||
try:
|
||||
tracer.stop()
|
||||
except Exception:
|
||||
pass
|
||||
try:
|
||||
responder.stop()
|
||||
finally:
|
||||
root.destroy()
|
||||
|
||||
root.protocol("WM_DELETE_WINDOW", on_close)
|
||||
root.mainloop()
|
||||
Reference in New Issue
Block a user