140 lines
4.4 KiB
Python
140 lines
4.4 KiB
Python
# 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 = ""
|
|
color: str = "#377eb8"
|
|
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, color: str = "#377eb8", 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=color),
|
|
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
|