eeeck/CANveyor
1
0
Fork 0
Code Issues 1 Pull Requests Projects Releases 6 Wiki Activity

Refactor CANveyor dashboard architecture

Browse Source 
- Replace `stats.utils.extractor` with dedicated `loader` and
  `converter`
  modules to improve code organization.
- Implement explicit pipeline stages for ingestion, decoding, and
  precomputation with caching.
- Standardize data loading and J1939 parsing logic across sub-modules.
- Enhance dashboard responsiveness by pre-calculating figures and
  downsampling ID-grouped data.
- Enforce strict typing and add docstrings to public components.
This commit is contained in:
Erick
2026-07-23 16:01:34 +02:00
parent 89a9124a83
commit 4f280da033
8 changed files with 575 additions and 440 deletions
Download Patch File Download Diff File Expand all files Collapse all files
vehicle/base.py
+77 -30
View File
@@ -2,14 +2,20 @@
# Copyright (C) 2026 Erick Ahmed
# SPDX-License-Identifier: AGPL-3.0-or-later
"""Core data structures and helpers for J1939/CAN signal decoding."""
from dataclasses import dataclass, field
from typing import List
from typing import Dict, List
import numpy as np
import pandas as pd
import plotly.graph_objects as go
@dataclass
class SignalDef:
"""Definition of a single signal within a CAN frame."""
name: str
bit_start: int
bit_length: int
@@ -19,27 +25,38 @@ class SignalDef:
byte_order: str = "little"
unit: str = ""
@dataclass
class FrameDef:
"""Definition of a CAN frame and its contained signals."""
can_id: str
description: str = ""
color: str = "#377eb8"
signals: List[SignalDef] = field(default_factory=list)
def normalize_id(can_id: str) -> str:
"""Normalize a CAN ID string to uppercase hex without leading zeros/0x."""
s = str(can_id).strip().upper()
if s.startswith("0X"):
s = s[2:]
return s.lstrip("0") or "0"
def _byte_indices(sig: SignalDef) -> List[int]:
"""Return the in-range byte positions spanned by *sig*."""
byte_lo = sig.bit_start // 8
byte_hi = (sig.bit_start + sig.bit_length - 1) // 8
return [i for i in range(byte_lo, byte_hi + 1) if 0 <= i < 8]
def _extract_signal(bytes_arr: np.ndarray, sig: SignalDef) -> np.ndarray:
"""Extract raw signal values from an (N, 8) byte array and apply scaling."""
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]
byte_indices = _byte_indices(sig)
if not byte_indices:
return np.full(bytes_arr.shape[0], np.nan, dtype=np.float64)
@@ -51,11 +68,8 @@ def _extract_signal(bytes_arr: np.ndarray, sig: SignalDef) -> np.ndarray:
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
raw = raw >> (sig.bit_start % 8)
raw = raw & ((1 << sig.bit_length) - 1)
if sig.is_signed and sig.bit_length < 64:
sign_bit = 1 << (sig.bit_length - 1)
@@ -63,7 +77,11 @@ def _extract_signal(bytes_arr: np.ndarray, sig: SignalDef) -> np.ndarray:
return raw.astype(np.float64) * sig.factor + sig.offset
def decode_dataframe(df: pd.DataFrame, can_id: str, decoder_rules: dict) -> pd.DataFrame:
def decode_dataframe(
df: pd.DataFrame, can_id: str, decoder_rules: Dict[str, FrameDef]
) -> pd.DataFrame:
"""Decode all signals for *can_id* from *df* into a new DataFrame."""
norm = normalize_id(can_id)
if norm not in decoder_rules:
return pd.DataFrame()
@@ -72,8 +90,7 @@ def decode_dataframe(df: pd.DataFrame, can_id: str, decoder_rules: dict) -> pd.D
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()
sub = df.loc[df_ids == norm].copy()
if sub.empty:
return pd.DataFrame()
@@ -83,41 +100,69 @@ def decode_dataframe(df: pd.DataFrame, can_id: str, decoder_rules: dict) -> pd.D
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()
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))
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, color: str = "#377eb8", height: int = 280) -> go.Figure:
def plot_signal(
df: pd.DataFrame,
signal_name: str,
title: str,
color: str = "#377eb8",
height: int = 280,
) -> go.Figure:
"""Plot a single signal over time as a line chart."""
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"))],
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=color),
name=signal_name,
hovertemplate=f"<b>{signal_name}</b><br>Time: %{{x}}<br>Value: %{{y:.2f}}<extra></extra>",
))
fig.add_trace(
go.Scatter(
x=df["Timestamp"],
y=df[signal_name],
mode="lines",
line=dict(width=2, color=color),
name=signal_name,
hovertemplate=(
f"<b>{signal_name}</b><br>Time: %{{x}}<br>"
f"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)),
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",
@@ -133,7 +178,9 @@ def plot_signal(df: pd.DataFrame, signal_name: str, title: str, color: str = "#3
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"),
hoverlabel=dict(
bgcolor="white", font_size=12,
font_family="Segoe UI", bordercolor="#cccccc",
),
)
return fig
vehicle/komatsu.py
+45 -30
View File
@@ -2,17 +2,28 @@
# Copyright (C) 2026 Erick Ahmed
# SPDX-License-Identifier: AGPL-3.0-or-later
import plotly.express as px
import pandas as pd
from dataclasses import dataclass
from typing import Callable
"""Komatsu-specific CAN frame decoder rules and custom plot definitions."""
from copy import deepcopy
from dataclasses import dataclass
from typing import Callable, Dict
import pandas as pd
import plotly.express as px
from vehicle.base import (
SignalDef, FrameDef, normalize_id, decode_dataframe as _decode_dataframe, plot_signal
FrameDef,
SignalDef,
decode_dataframe as _decode_dataframe,
normalize_id,
plot_signal,
)
@dataclass
class CustomPlotDef:
"""A non-signal entry in a FrameDef that carries its own plotting function."""
name: str
plot_func: Callable
@@ -28,7 +39,7 @@ load_state_sig = SignalDef(
)
load_state_sig.skip_plot = True
DECODER_RULES = {
DECODER_RULES: Dict[str, FrameDef] = {
normalize_id("0x011F"): FrameDef(
can_id="0x011F",
description="Engine ECM Main Broadcast",
@@ -84,7 +95,7 @@ DECODER_RULES = {
load_state_sig,
CustomPlotDef(
name="Engine Load State",
plot_func=lambda decoded, color: plot_load_state_pie(decoded, color)
plot_func=lambda decoded, color: plot_load_state_pie(decoded, color),
),
],
),
@@ -93,54 +104,58 @@ DECODER_RULES = {
LOAD_STATE_MAP = {
0: "Boot up",
16: "Normal load",
32: "High load"
32: "High load",
}
_LOAD_STATE_COLORS = {
"Boot up": "#ff9900",
"Normal load": "#00cc00",
"High load": "#cc0000",
"Unknown": "#808080",
}
def plot_load_state_pie(decoded_df, color):
"""Render a pie chart showing the distribution of engine load states."""
if decoded_df is None or decoded_df.empty or "Engine Load State" not in decoded_df.columns:
return px.pie(title="No data for Engine Load State")
states = pd.to_numeric(decoded_df["Engine Load State"], errors='coerce').dropna().astype(int)
states = pd.to_numeric(decoded_df["Engine Load State"], errors="coerce").dropna().astype(int)
labels = states.map(LOAD_STATE_MAP).fillna("Unknown")
counts = labels.value_counts().reset_index()
counts.columns = ['State', 'Count']
counts.columns = ["State", "Count"]
total = counts['Count'].sum()
counts['Percentage'] = (counts['Count'] / total * 100).round(1)
counts['Legend'] = counts['State'] + " (" + counts['Percentage'].astype(str) + "%)"
total = counts["Count"].sum()
counts["Percentage"] = (counts["Count"] / total * 100).round(1)
counts["Legend"] = counts["State"] + " (" + counts["Percentage"].astype(str) + "%)"
fig = px.pie(
counts,
values='Count',
names='Legend',
color='State',
title='Engine Load State Distribution',
color_discrete_map={
"Boot up": "#ff9900",
"Normal load": "#00cc00",
"High load": "#cc0000",
"Unknown": "#808080"
}
values="Count",
names="Legend",
color="State",
title="Engine Load State Distribution",
color_discrete_map=_LOAD_STATE_COLORS,
)
fig.update_traces(
textinfo='none',
hoverinfo='label+percent+value',
domain={'x': [0.05, 0.55], 'y': [0.05, 0.95]}
textinfo="none",
hoverinfo="label+percent+value",
domain={"x": [0.05, 0.55], "y": [0.05, 0.95]},
)
fig.update_layout(
margin=dict(l=0, r=10, t=40, b=0),
legend=dict(x=0.6, y=0.5)
legend=dict(x=0.6, y=0.5),
)
return fig
def decode_dataframe(df, can_id):
filtered_rules = {}
"""Decode *can_id* from *df*, filtering out non-SignalDef entries first."""
filtered_rules: Dict[str, FrameDef] = {}
for nid, frame in DECODER_RULES.items():
filtered_signals = [sig for sig in frame.signals if isinstance(sig, SignalDef)]
new_frame = deepcopy(frame)
new_frame.signals = filtered_signals
new_frame.signals = [s for s in frame.signals if isinstance(s, SignalDef)]
filtered_rules[nid] = new_frame
return _decode_dataframe(df, can_id, filtered_rules)