Implement a modular vehicle decoding system
- Add Komatsu specific rules - Possibility to expand to any brand
This commit is contained in:
@@ -11,6 +11,7 @@ import dash
|
||||
from dash import dcc, html, Input, Output, State
|
||||
import dash_bootstrap_components as dbc
|
||||
import numpy as np
|
||||
import pandas as pd
|
||||
|
||||
from parser import parse_log, parse_csv
|
||||
from decoder import decode_j1939_frames
|
||||
@@ -20,6 +21,7 @@ from stats.frequency import calculate_frequency, plot_frequency
|
||||
from stats.correlation import calculate_correlation, plot_correlation_heatmap
|
||||
from stats.entropy import calculate_byte_entropy, plot_entropy_heatmap
|
||||
from logs.view import get_logs_table_component, prepare_logs_data
|
||||
from vehicle import get_vehicle_module
|
||||
|
||||
RAW_LOG_DIR = "data/logs"
|
||||
PAGE_SIZE = 25000
|
||||
@@ -131,6 +133,18 @@ app.layout = dbc.Container([
|
||||
dbc.Tab(label="Overview", tab_id="overview", children=[
|
||||
html.Div(id="overview-content")
|
||||
]),
|
||||
dbc.Tab(label="Vehicles", tab_id="vehicles", children=[
|
||||
dbc.Row([
|
||||
dbc.Col(html.Label("Select Vehicle:", className="mt-2"), width="auto"),
|
||||
dbc.Col(dcc.Dropdown(
|
||||
id='vehicles-vehicle-selector',
|
||||
options=[{'label': v, 'value': v} for v in DATA.keys()],
|
||||
value=list(DATA.keys())[0] if DATA else None,
|
||||
clearable=False
|
||||
), width=3, className="me-4"),
|
||||
], className="mb-3 mt-3", align="end"),
|
||||
html.Div(id='vehicles-content', className="mt-3")
|
||||
]),
|
||||
dbc.Tab(label="Logs", tab_id="logs", children=[
|
||||
dbc.Row([
|
||||
dbc.Col(html.Label("Select Vehicle:", className="mt-2"), width="auto"),
|
||||
@@ -391,5 +405,58 @@ def update_corr(method, target, vehicle, bus, tab):
|
||||
|
||||
return plot_correlation_heatmap(corr_df, target_id=target_id, title=title)
|
||||
|
||||
@app.callback(
|
||||
Output('vehicles-content', 'children'),
|
||||
Input('vehicles-vehicle-selector', 'value')
|
||||
)
|
||||
def render_vehicles(vehicle):
|
||||
"""Render small graph boxes for every decoded signal, combining both buses."""
|
||||
if not vehicle or vehicle not in DATA:
|
||||
return html.Div("No data available", className="text-muted")
|
||||
|
||||
brand = VEHICLE_META.get(vehicle, {}).get("brand", "")
|
||||
|
||||
vehicle_module = get_vehicle_module(brand)
|
||||
|
||||
dfs = []
|
||||
for bus_df in DATA[vehicle].values():
|
||||
dfs.append(bus_df)
|
||||
|
||||
if not dfs:
|
||||
return html.Div("No data available", className="text-muted")
|
||||
|
||||
df = pd.concat(dfs, ignore_index=True)
|
||||
|
||||
if 'Timestamp' in df.columns:
|
||||
df = df.sort_values('Timestamp', kind='stable').reset_index(drop=True)
|
||||
|
||||
cards = []
|
||||
for nid, frame_def in vehicle_module.DECODER_RULES.items():
|
||||
decoded = vehicle_module.decode_dataframe(df, frame_def.can_id)
|
||||
|
||||
for sig in frame_def.signals:
|
||||
unit_str = f" ({sig.unit})" if sig.unit else ""
|
||||
title = f"{frame_def.can_id} - {sig.name}{unit_str}"
|
||||
fig = vehicle_module.plot_signal(decoded, sig.name, title=title)
|
||||
|
||||
card = dbc.Card([
|
||||
dbc.CardBody([
|
||||
dcc.Graph(figure=fig, config={'displayModeBar': False},
|
||||
style={'height': '280px'})
|
||||
], className="p-2"),
|
||||
], className="shadow-sm border-0 h-100")
|
||||
|
||||
cards.append(
|
||||
dbc.Col(card, xs=12, sm=6, md=4, lg=3, className="mb-3")
|
||||
)
|
||||
|
||||
if not cards:
|
||||
return html.Div(
|
||||
"No decoded signals available. Add rules in the vehicle module.",
|
||||
className="text-muted"
|
||||
)
|
||||
|
||||
return dbc.Row(cards)
|
||||
|
||||
if __name__ == '__main__':
|
||||
app.run(debug=False)
|
||||
|
||||
@@ -0,0 +1,19 @@
|
||||
# File: vehicle/__init__.py
|
||||
# Copyright (C) 2026 Erick Ahmed
|
||||
# SPDX-License-Identifier: AGPL-3.0-or-later
|
||||
|
||||
import importlib
|
||||
|
||||
def get_vehicle_module(brand: str):
|
||||
"""
|
||||
Dynamically imports the correct decoder module based on the vehicle brand.
|
||||
Falls back to 'vehicle.generic' if a specific brand module is not found.
|
||||
"""
|
||||
if not brand:
|
||||
return importlib.import_module("vehicle.generic")
|
||||
|
||||
module_name = f"vehicle.{brand.lower().replace(' ', '_')}"
|
||||
try:
|
||||
return importlib.import_module(module_name)
|
||||
except ModuleNotFoundError:
|
||||
return importlib.import_module("vehicle.generic")
|
||||
+138
@@ -0,0 +1,138 @@
|
||||
# File: vehicle/base.py
|
||||
# Copyright (C) 2026 Erick Ahmed
|
||||
# SPDX-License-Identifier: AGPL-3.0-or-later
|
||||
|
||||
from dataclasses import dataclass, field
|
||||
from typing import List
|
||||
import numpy as np
|
||||
import pandas as pd
|
||||
import plotly.graph_objects as go
|
||||
|
||||
@dataclass
|
||||
class SignalDef:
|
||||
name: str
|
||||
bit_start: int
|
||||
bit_length: int
|
||||
factor: float = 1.0
|
||||
offset: float = 0.0
|
||||
is_signed: bool = False
|
||||
byte_order: str = "little"
|
||||
unit: str = ""
|
||||
|
||||
@dataclass
|
||||
class FrameDef:
|
||||
can_id: str
|
||||
description: str = ""
|
||||
signals: List[SignalDef] = field(default_factory=list)
|
||||
|
||||
def normalize_id(can_id: str) -> str:
|
||||
s = str(can_id).strip().upper()
|
||||
if s.startswith("0X"):
|
||||
s = s[2:]
|
||||
return s.lstrip("0") or "0"
|
||||
|
||||
def _extract_signal(bytes_arr: np.ndarray, sig: SignalDef) -> np.ndarray:
|
||||
if bytes_arr.size == 0:
|
||||
return np.zeros(0, dtype=np.float64)
|
||||
|
||||
byte_lo = sig.bit_start // 8
|
||||
byte_hi = (sig.bit_start + sig.bit_length - 1) // 8
|
||||
byte_indices = [i for i in range(byte_lo, byte_hi + 1) if 0 <= i < 8]
|
||||
|
||||
if not byte_indices:
|
||||
return np.full(bytes_arr.shape[0], np.nan, dtype=np.float64)
|
||||
|
||||
raw = np.zeros(bytes_arr.shape[0], dtype=np.int64)
|
||||
if sig.byte_order == "little":
|
||||
for shift, bi in enumerate(byte_indices):
|
||||
raw += bytes_arr[:, bi].astype(np.int64) << (shift * 8)
|
||||
else:
|
||||
for shift, bi in enumerate(reversed(byte_indices)):
|
||||
raw += bytes_arr[:, bi].astype(np.int64) << (shift * 8)
|
||||
|
||||
intra_byte_shift = sig.bit_start % 8
|
||||
raw = raw >> intra_byte_shift
|
||||
|
||||
mask = (1 << sig.bit_length) - 1
|
||||
raw = raw & mask
|
||||
|
||||
if sig.is_signed and sig.bit_length < 64:
|
||||
sign_bit = 1 << (sig.bit_length - 1)
|
||||
raw = (raw ^ sign_bit) - sign_bit
|
||||
|
||||
return raw.astype(np.float64) * sig.factor + sig.offset
|
||||
|
||||
def decode_dataframe(df: pd.DataFrame, can_id: str, decoder_rules: dict) -> pd.DataFrame:
|
||||
norm = normalize_id(can_id)
|
||||
if norm not in decoder_rules:
|
||||
return pd.DataFrame()
|
||||
|
||||
frame_def = decoder_rules[norm]
|
||||
|
||||
id_col = "ID" if "ID" in df.columns else "Identifier"
|
||||
df_ids = df[id_col].astype(str).map(normalize_id)
|
||||
mask = df_ids == norm
|
||||
sub = df.loc[mask].copy()
|
||||
if sub.empty:
|
||||
return pd.DataFrame()
|
||||
|
||||
byte_cols = [f"b{i}" for i in range(8) if f"b{i}" in sub.columns]
|
||||
if not byte_cols:
|
||||
return pd.DataFrame()
|
||||
|
||||
arr = np.zeros((len(sub), 8), dtype=np.int64)
|
||||
for i, c in enumerate(byte_cols):
|
||||
arr[:, i] = pd.to_numeric(sub[c], errors="coerce").fillna(0).astype(np.int64).to_numpy()
|
||||
|
||||
out = pd.DataFrame()
|
||||
out["Timestamp"] = sub["Timestamp"].to_numpy() if "Timestamp" in sub.columns else np.arange(len(sub))
|
||||
|
||||
for sig in frame_def.signals:
|
||||
out[sig.name] = _extract_signal(arr, sig)
|
||||
|
||||
return out
|
||||
|
||||
def plot_signal(df: pd.DataFrame, signal_name: str, title: str, height: int = 280) -> go.Figure:
|
||||
fig = go.Figure()
|
||||
|
||||
if df.empty or signal_name not in df.columns:
|
||||
fig.update_layout(
|
||||
title=dict(text=title, font=dict(size=14)),
|
||||
annotations=[dict(text="No data", showarrow=False, x=0.5, y=0.5,
|
||||
font=dict(size=13, color="#888"))],
|
||||
height=height,
|
||||
template="plotly_white",
|
||||
)
|
||||
return fig
|
||||
|
||||
fig.add_trace(go.Scatter(
|
||||
x=df["Timestamp"],
|
||||
y=df[signal_name],
|
||||
mode="lines",
|
||||
line=dict(width=2, color="#377eb8"),
|
||||
name=signal_name,
|
||||
hovertemplate=f"<b>{signal_name}</b><br>Time: %{{x}}<br>Value: %{{y:.2f}}<extra></extra>",
|
||||
))
|
||||
|
||||
fig.update_layout(
|
||||
title=dict(text=title, font=dict(size=14, color="#1a1a1a"),
|
||||
x=0.5, xanchor="center", pad=dict(b=10)),
|
||||
height=height,
|
||||
autosize=True,
|
||||
template="plotly_white",
|
||||
margin=dict(l=55, r=20, t=55, b=45),
|
||||
xaxis=dict(
|
||||
title=dict(text="Time", font=dict(size=11)),
|
||||
showgrid=True, gridwidth=0.5, gridcolor="#eee",
|
||||
zeroline=False, linecolor="#bdbdbd",
|
||||
),
|
||||
yaxis=dict(
|
||||
title=dict(text=signal_name, font=dict(size=11)),
|
||||
showgrid=True, gridwidth=0.5, gridcolor="#eee",
|
||||
zeroline=False, linecolor="#bdbdbd",
|
||||
),
|
||||
font=dict(family="Segoe UI, Arial, sans-serif", size=11, color="#2a2a2a"),
|
||||
hoverlabel=dict(bgcolor="white", font_size=12,
|
||||
font_family="Segoe UI", bordercolor="#cccccc"),
|
||||
)
|
||||
return fig
|
||||
@@ -0,0 +1,40 @@
|
||||
# File: vehicle/komatsu.py
|
||||
# Copyright (C) 2026 Erick Ahmed
|
||||
# SPDX-License-Identifier: AGPL-3.0-or-later
|
||||
|
||||
from vehicle.base import (
|
||||
SignalDef, FrameDef, normalize_id, decode_dataframe as _decode_dataframe, plot_signal
|
||||
)
|
||||
|
||||
DECODER_RULES = {
|
||||
normalize_id("0x011F"): FrameDef(
|
||||
can_id="0x011F",
|
||||
description="Engine ECM Main Broadcast",
|
||||
signals=[
|
||||
SignalDef(
|
||||
name="Engine RPM (b0:b1)",
|
||||
bit_start=0,
|
||||
bit_length=16,
|
||||
factor=0.125,
|
||||
offset=0.0,
|
||||
is_signed=False,
|
||||
byte_order="big",
|
||||
unit="RPM",
|
||||
),
|
||||
SignalDef(
|
||||
name="Pressure / Load (b3:b2)",
|
||||
bit_start=16,
|
||||
bit_length=16,
|
||||
factor=1.0,
|
||||
offset=0.0,
|
||||
is_signed=False,
|
||||
byte_order="little",
|
||||
unit="raw",
|
||||
),
|
||||
],
|
||||
),
|
||||
|
||||
}
|
||||
|
||||
def decode_dataframe(df, can_id):
|
||||
return _decode_dataframe(df, can_id, DECODER_RULES)
|
||||
Reference in New Issue
Block a user