Implement a modular vehicle decoding system

- Add Komatsu specific rules
- Possibility to expand to any brand

vehicle/__init__.py

@@ -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")

vehicle/base.py

@@ -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

vehicle/komatsu.py

@@ -0,0 +1,29 @@
+# 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 RPM",
+        signals=[
+            SignalDef(
+                name="RPM",
+                bit_start=0,
+                bit_length=8,
+                factor=20.0,
+                offset=0.0,
+                is_signed=False,
+                byte_order="little",
+                unit="rpm",
+            ),
+        ],
+    ),
+}
+
+def decode_dataframe(df, can_id):
+    return _decode_dataframe(df, can_id, DECODER_RULES)