CANveyor/vehicle/base.py

# File: vehicle/base.py
# 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 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
    factor: float = 1.0
    offset: float = 0.0
    is_signed: bool = False
    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_indices = _byte_indices(sig)
    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)

    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)
        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[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()

    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)
    sub = df.loc[df_ids == norm].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,
    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"),
                )
            ],
            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>"
                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),
        ),
        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