From 4f280da033293b90513921ca5b63ab1994836895 Mon Sep 17 00:00:00 2001 From: Erick Ahmed Date: Thu, 23 Jul 2026 16:01:34 +0200 Subject: [PATCH] Refactor CANveyor dashboard architecture - 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. --- main.py | 682 +++++++++++++++++++++++---------------- stats/correlation.py | 19 +- stats/entropy.py | 17 +- stats/frequency.py | 16 +- stats/id_viewer.py | 36 ++- stats/utils/extractor.py | 63 ---- vehicle/base.py | 107 ++++-- vehicle/komatsu.py | 75 +++-- 8 files changed, 575 insertions(+), 440 deletions(-) delete mode 100644 stats/utils/extractor.py diff --git a/main.py b/main.py index d65b600..de72cab 100644 --- a/main.py +++ b/main.py @@ -2,205 +2,306 @@ # Copyright (C) 2026 Erick Ahmed # SPDX-License-Identifier: AGPL-3.0-or-later -import os -from pathlib import Path -from concurrent.futures import ThreadPoolExecutor +"""CANveyor dashboard entry point. + +Handles raw log ingestion, J1939 decoding, precomputation of +statistical figures, and exposes a Dash application for browsing the +processed data. +""" + +import os +from concurrent.futures import ThreadPoolExecutor +from pathlib import Path +from typing import Dict, List, Tuple -import polars as pl import dash -from dash import dcc, html, Input, Output, State import dash_bootstrap_components as dbc import numpy as np import pandas as pd +import polars as pl +from dash import dcc, html, Input, Output, State -from parser import parse_log, parse_csv from decoder import decode_j1939_frames -from stats.utils.extractor import load_data -from stats.id_viewer import _format_can_id_vec, plot_bits -from stats.frequency import calculate_frequency, plot_frequency +from logs.view import get_logs_table_component, prepare_logs_data +from parser import parse_csv, parse_log 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 stats.frequency import calculate_frequency, plot_frequency +from stats.id_viewer import _format_can_id_vec, plot_bits +from stats.utils.loader import load_data from vehicle import get_vehicle_module RAW_LOG_DIR = "data/logs" -PAGE_SIZE = 25000 +CSV_DIR = "data/csv" +PARQUET_DIR = "data/parquet" +PAGE_SIZE = 25_000 +BYTE_COLS = [f"b{i}" for i in range(8)] +BUS_OPTIONS = [ + {"label": "Bus 1", "value": "Bus 1"}, + {"label": "Bus 2", "value": "Bus 2"}, +] -def parse_vehicle_from_filename(filename: str): +DATA: Dict[str, Dict[str, pd.DataFrame]] = {} +VEHICLE_META: Dict[str, Dict[str, str]] = {} +PRECOMPUTED_FIGURES: Dict[str, object] = {} +DATA_BY_ID: Dict[Tuple[str, str], Dict[str, Tuple[pd.DataFrame, List[str]]]] = {} +CORR_CACHE: Dict[Tuple, object] = {} +PREPARED_LOGS_CACHE: Dict[Tuple[str, str], pd.DataFrame] = {} + + +def parse_vehicle_from_filename(filename: str) -> Tuple[str, str, str]: + """Derive (vehicle, brand, model) from a log file name.""" stem = Path(filename).stem - if '-' in stem: - brand, model_part = stem.split('-', 1) + if "-" in stem: + brand, model_part = stem.split("-", 1) else: brand, model_part = stem, "Unknown" - model = model_part.replace('_', ' ') + model = model_part.replace("_", " ") vehicle = f"{brand} {model}".strip() return vehicle, brand, model -def run_pipeline(): - os.makedirs(RAW_LOG_DIR, exist_ok=True) - os.makedirs("data/csv", exist_ok=True) - os.makedirs("data/parquet", exist_ok=True) + +def _vehicle_paths(vehicle: str) -> Dict[str, str]: + """Return all intermediate file paths for a given vehicle.""" + return { + "bus1_csv": f"{CSV_DIR}/{vehicle}_bus1.csv", + "bus2_csv": f"{CSV_DIR}/{vehicle}_bus2.csv", + "bus1_parquet": f"{PARQUET_DIR}/{vehicle}_bus1.parquet", + "bus2_parquet": f"{PARQUET_DIR}/{vehicle}_bus2.parquet", + "bus1_decoded": f"{PARQUET_DIR}/{vehicle}_bus1_decoded.parquet", + "bus2_decoded": f"{PARQUET_DIR}/{vehicle}_bus2_decoded.parquet", + } + + +def run_pipeline() -> None: + """Parse raw log files, convert to parquet, and decode J1939 frames.""" + for directory in (RAW_LOG_DIR, CSV_DIR, PARQUET_DIR): + os.makedirs(directory, exist_ok=True) + for log_file in Path(RAW_LOG_DIR).glob("*.txt"): + vehicle, _, _ = parse_vehicle_from_filename(log_file.name) + paths = _vehicle_paths(vehicle) + + if Path(paths["bus1_decoded"]).exists() and Path(paths["bus2_decoded"]).exists(): + continue + + print(f"Parsing raw log: {log_file.name}...") + parse_log(str(log_file), paths["bus1_csv"], paths["bus2_csv"]) + + print("Converting to parquet...") + parse_csv(paths["bus1_csv"]).sink_parquet(paths["bus1_parquet"]) + parse_csv(paths["bus2_csv"]).sink_parquet(paths["bus2_parquet"]) + + print("Decoding J1939...") + df1 = pl.read_parquet(paths["bus1_parquet"]) + df2 = pl.read_parquet(paths["bus2_parquet"]) + decode_j1939_frames(df1).write_parquet(paths["bus1_decoded"]) + decode_j1939_frames(df2).write_parquet(paths["bus2_decoded"]) + + +def load_vehicle_data() -> None: + """Load all decoded parquet files into the in-memory DATA store.""" for log_file in Path(RAW_LOG_DIR).glob("*.txt"): vehicle, brand, model = parse_vehicle_from_filename(log_file.name) + VEHICLE_META[vehicle] = {"brand": brand, "model": model} - bus1_csv = f"data/csv/{vehicle}_bus1.csv" - bus2_csv = f"data/csv/{vehicle}_bus2.csv" - bus1_parquet = f"data/parquet/{vehicle}_bus1.parquet" - bus2_parquet = f"data/parquet/{vehicle}_bus2.parquet" - bus1_decoded = f"data/parquet/{vehicle}_bus1_decoded.parquet" - bus2_decoded = f"data/parquet/{vehicle}_bus2_decoded.parquet" + paths = _vehicle_paths(vehicle) + if Path(paths["bus1_decoded"]).exists() and Path(paths["bus2_decoded"]).exists(): + DATA[vehicle] = { + "Bus 1": load_data(paths["bus1_decoded"]), + "Bus 2": load_data(paths["bus2_decoded"]), + } - if not Path(bus1_decoded).exists() or not Path(bus2_decoded).exists(): - print(f"Parsing raw log: {log_file.name}...") - parse_log(str(log_file), bus1_csv, bus2_csv) - print("Converting to parquet...") - parse_csv(bus1_csv).sink_parquet(bus1_parquet) - parse_csv(bus2_csv).sink_parquet(bus2_parquet) +def _can_id_column(df: pd.DataFrame) -> str: + return "ID" if "ID" in df.columns else "Identifier" - print("Decoding J1939...") - df1 = pl.read_parquet(bus1_parquet) - df2 = pl.read_parquet(bus2_parquet) - dec1 = decode_j1939_frames(df1) - dec2 = decode_j1939_frames(df2) - dec1.write_parquet(bus1_decoded) - dec2.write_parquet(bus2_decoded) -run_pipeline() +def _downsample_unchanged(group: pd.DataFrame, byte_cols: List[str]) -> pd.DataFrame: + """Keep only rows where at least one byte changed vs. the previous row.""" + if group.empty or not byte_cols: + return group + arr = group[byte_cols].to_numpy(dtype=np.float32, copy=False) + if len(arr) <= 1: + return group + changed = np.any(arr[1:] != arr[:-1], axis=1) + keep = np.concatenate(([True], changed)) + return group.iloc[keep] -print("Loading data into memory...") -DATA = {} -VEHICLE_META = {} -for log_file in Path(RAW_LOG_DIR).glob("*.txt"): - vehicle, brand, model = parse_vehicle_from_filename(log_file.name) - VEHICLE_META[vehicle] = {"brand": brand, "model": model} +def process_bus_data( + vehicle: str, bus: str, df: pd.DataFrame +) -> Tuple[str, str, Dict[str, object], Dict[str, Tuple[pd.DataFrame, List[str]]]]: + """Compute per-bus figures and ID-grouped, downsampled frames.""" + precomp = { + f"{vehicle}_{bus}_freq": plot_frequency( + calculate_frequency(df), title=f"{vehicle} {bus} Frequency" + ), + f"{vehicle}_{bus}_entropy": plot_entropy_heatmap( + calculate_byte_entropy(df), title=f"{vehicle} {bus} Byte-Level Entropy" + ), + } - bus1_decoded = f"data/parquet/{vehicle}_bus1_decoded.parquet" - bus2_decoded = f"data/parquet/{vehicle}_bus2_decoded.parquet" + df = df.assign(Formatted_ID=_format_can_id_vec(df[_can_id_column(df)])) + df = df.sort_values(["Formatted_ID", "Timestamp"], kind="stable") - if Path(bus1_decoded).exists() and Path(bus2_decoded).exists(): - DATA[vehicle] = { - "Bus 1": load_data(bus1_decoded), - "Bus 2": load_data(bus2_decoded) - } - -PRECOMPUTED_FIGURES = {} -DATA_BY_ID = {} -CORR_CACHE = {} -PREPARED_LOGS_CACHE = {} - -def process_bus_data(vehicle, bus, df): - precomp = {} - precomp[f"{vehicle}_{bus}_freq"] = plot_frequency(calculate_frequency(df), title=f"{vehicle} {bus} Frequency") - precomp[f"{vehicle}_{bus}_entropy"] = plot_entropy_heatmap(calculate_byte_entropy(df), title=f"{vehicle} {bus} Byte-Level Entropy") - - can_id_col = 'ID' if 'ID' in df.columns else 'Identifier' - formatted = _format_can_id_vec(df[can_id_col]) - df = df.assign(Formatted_ID=formatted) - df = df.sort_values(['Formatted_ID', 'Timestamp'], kind='stable') - - grouped = {} - for can_id, group in df.groupby(by='Formatted_ID'): - byte_cols = [f"b{i}" for i in range(8) if f"b{i}" in group.columns] - if not group.empty and len(byte_cols) > 0: - arr = group[byte_cols].to_numpy(dtype=np.float32, copy=False) - if len(arr) > 1: - changed = np.any(arr[1:] != arr[:-1], axis=1) - keep = np.concatenate(([True], changed)) - group = group.iloc[keep] + grouped: Dict[str, Tuple[pd.DataFrame, List[str]]] = {} + for can_id, group in df.groupby(by="Formatted_ID"): + byte_cols = [c for c in BYTE_COLS if c in group.columns] + group = _downsample_unchanged(group, byte_cols) grouped[can_id] = (group, byte_cols) return vehicle, bus, precomp, grouped -with ThreadPoolExecutor() as executor: - futures = [] - for vehicle, buses in DATA.items(): - for bus, df in buses.items(): - futures.append(executor.submit(process_bus_data, vehicle, bus, df)) - for future in futures: - v, b, precomp, grouped = future.result() - PRECOMPUTED_FIGURES.update(precomp) - DATA_BY_ID[(v, b)] = grouped + +def precompute_all() -> None: + """Run :func:`process_bus_data` across every vehicle/bus pair in parallel.""" + with ThreadPoolExecutor() as executor: + futures = [ + executor.submit(process_bus_data, vehicle, bus, df) + for vehicle, buses in DATA.items() + for bus, df in buses.items() + ] + for future in futures: + v, b, precomp, grouped = future.result() + PRECOMPUTED_FIGURES.update(precomp) + DATA_BY_ID[(v, b)] = grouped + +run_pipeline() + +print("Loading data into memory...") +load_vehicle_data() +precompute_all() app = dash.Dash(__name__, external_stylesheets=[dbc.themes.BOOTSTRAP]) app.config.suppress_callback_exceptions = True -app.layout = dbc.Container([ - html.H1("CANveyor", className="my-4"), - dbc.Tabs([ - 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("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("Vehicle:", className="mt-2"), width="auto"), - dbc.Col(dcc.Dropdown( - id='logs-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"), - dbc.Col(html.Label("Bus:", className="mt-2"), width="auto"), - dbc.Col(dcc.Dropdown( - id='logs-bus-selector', - options=[{'label': 'Bus 1', 'value': 'Bus 1'}, {'label': 'Bus 2', 'value': 'Bus 2'}], - value='Bus 1', - clearable=False - ), width=2), - ], className="mb-3 mt-3", align="end"), - get_logs_table_component() - ]), - dbc.Tab(label="Statistics", tab_id="statistics", children=[ - dbc.Row([ - dbc.Col(html.Label("Vehicle:", className="mt-2"), width="auto"), - dbc.Col(dcc.Dropdown( - id='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"), - dbc.Col(html.Label("Bus:", className="mt-2"), width="auto"), - dbc.Col(dcc.Dropdown( - id='bus-selector', - options=[{'label': 'Bus 1', 'value': 'Bus 1'}, {'label': 'Bus 2', 'value': 'Bus 2'}], - value='Bus 1', - clearable=False - ), width=2), - ], className="mb-3 mt-3", align="end"), - dbc.Tabs([ - dbc.Tab(label="Frequency", tab_id="freq"), - dbc.Tab(label="ID Viewer", tab_id="id_viewer"), - dbc.Tab(label="Correlation", tab_id="corr"), - dbc.Tab(label="Entropy", tab_id="entropy"), - ], id="tabs", active_tab="freq"), - html.Div(id="tab-content", className="mt-3") - ]) - ], id="main-tabs", active_tab="statistics") -], fluid=True) -def get_prepared_logs(vehicle, bus): +def _vehicle_dropdown(dropdown_id: str) -> dcc.Dropdown: + return dcc.Dropdown( + id=dropdown_id, + options=[{"label": v, "value": v} for v in DATA.keys()], + value=list(DATA.keys())[0] if DATA else None, + clearable=False, + ) + + +def _bus_dropdown(dropdown_id: str) -> dcc.Dropdown: + return dcc.Dropdown( + id=dropdown_id, + options=BUS_OPTIONS, + value="Bus 1", + clearable=False, + ) + + +def _label(text: str) -> html.Label: + return html.Label(text, className="mt-2") + + +app.layout = dbc.Container( + [ + html.H1("CANveyor", className="my-4"), + dbc.Tabs( + [ + 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(_label("Vehicle:"), width="auto"), + dbc.Col( + _vehicle_dropdown("vehicles-vehicle-selector"), + 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(_label("Vehicle:"), width="auto"), + dbc.Col( + _vehicle_dropdown("logs-vehicle-selector"), + width=3, className="me-4", + ), + dbc.Col(_label("Bus:"), width="auto"), + dbc.Col( + _bus_dropdown("logs-bus-selector"), + width=2, + ), + ], + className="mb-3 mt-3", align="end", + ), + get_logs_table_component(), + ], + ), + dbc.Tab( + label="Statistics", + tab_id="statistics", + children=[ + dbc.Row( + [ + dbc.Col(_label("Vehicle:"), width="auto"), + dbc.Col( + _vehicle_dropdown("vehicle-selector"), + width=3, className="me-4", + ), + dbc.Col(_label("Bus:"), width="auto"), + dbc.Col( + _bus_dropdown("bus-selector"), + width=2, + ), + ], + className="mb-3 mt-3", align="end", + ), + dbc.Tabs( + [ + dbc.Tab(label="Frequency", tab_id="freq"), + dbc.Tab(label="ID Viewer", tab_id="id_viewer"), + dbc.Tab(label="Correlation", tab_id="corr"), + dbc.Tab(label="Entropy", tab_id="entropy"), + ], + id="tabs", + active_tab="freq", + ), + html.Div(id="tab-content", className="mt-3"), + ], + ), + ], + id="main-tabs", + active_tab="statistics", + ), + ], + fluid=True, +) + + +def get_prepared_logs(vehicle: str, bus: str) -> pd.DataFrame: + """Lazily prepare and cache log table data for a vehicle/bus pair.""" cache_key = (vehicle, bus) if cache_key not in PREPARED_LOGS_CACHE: - df = DATA[vehicle][bus] - PREPARED_LOGS_CACHE[cache_key] = prepare_logs_data(df) + PREPARED_LOGS_CACHE[cache_key] = prepare_logs_data(DATA[vehicle][bus]) return PREPARED_LOGS_CACHE[cache_key] -def build_page_buttons(current_page: int, total_pages: int, max_buttons: int = 15) -> list: - buttons: list = [] + +def build_page_buttons( + current_page: int, total_pages: int, max_buttons: int = 15 +) -> List: + """Build the pagination button list with ellipses where appropriate.""" + buttons: List = [] if total_pages <= 1: return buttons @@ -212,17 +313,21 @@ def build_page_buttons(current_page: int, total_pages: int, max_buttons: int = 1 if start > 0: buttons.append( - dbc.Button("1", id={'type': 'page-btn', 'index': 0}, color="secondary", outline=True, size="sm", className="me-1") + dbc.Button( + "1", + id={"type": "page-btn", "index": 0}, + color="secondary", outline=True, size="sm", className="me-1", + ) ) if start > 1: buttons.append(html.Span("…", className="mx-1 align-middle")) for i in range(start, end): - is_current = (i == current_page) + is_current = i == current_page buttons.append( dbc.Button( str(i + 1), - id={'type': 'page-btn', 'index': i}, + id={"type": "page-btn", "index": i}, size="sm", color="primary" if is_current else "secondary", outline=not is_current, @@ -237,33 +342,33 @@ def build_page_buttons(current_page: int, total_pages: int, max_buttons: int = 1 buttons.append( dbc.Button( str(total_pages), - id={'type': 'page-btn', 'index': total_pages - 1}, + id={"type": "page-btn", "index": total_pages - 1}, color="secondary", outline=True, size="sm", className="me-1", ) ) return buttons + @app.callback( - Output('logs-table', 'data'), - Output('logs-table', 'columns'), - Output('logs-info-text', 'children'), - Output('logs-page-nav', 'children'), - Output('logs-current-page', 'data'), - Input('logs-vehicle-selector', 'value'), - Input('logs-bus-selector', 'value'), - Input('logs-prev-btn', 'n_clicks'), - Input('logs-next-btn', 'n_clicks'), - Input({'type': 'page-btn', 'index': dash.ALL}, 'n_clicks'), - State('logs-current-page', 'data'), + Output("logs-table", "data"), + Output("logs-table", "columns"), + Output("logs-info-text", "children"), + Output("logs-page-nav", "children"), + Output("logs-current-page", "data"), + Input("logs-vehicle-selector", "value"), + Input("logs-bus-selector", "value"), + Input("logs-prev-btn", "n_clicks"), + Input("logs-next-btn", "n_clicks"), + Input({"type": "page-btn", "index": dash.ALL}, "n_clicks"), + State("logs-current-page", "data"), ) def update_logs_table(vehicle, bus, prev_clicks, next_clicks, page_btn_clicks, current_page): - if (not vehicle or not bus or vehicle not in DATA or bus not in DATA[vehicle]): + if not vehicle or not bus or vehicle not in DATA or bus not in DATA[vehicle]: return [], [], "No data available", [], 0 prepared_df = get_prepared_logs(vehicle, bus) total_rows = len(prepared_df) - if total_rows == 0: return [], [], "No data available", [], 0 @@ -271,43 +376,37 @@ def update_logs_table(vehicle, bus, prev_clicks, next_clicks, page_btn_clicks, c ctx = dash.callback_context triggered_id = ctx.triggered_id - current_page = current_page if current_page is not None else 0 - if triggered_id in ('logs-vehicle-selector', 'logs-bus-selector'): + if triggered_id in ("logs-vehicle-selector", "logs-bus-selector"): current_page = 0 - - elif triggered_id == 'logs-prev-btn': + elif triggered_id == "logs-prev-btn": current_page = max(0, current_page - 1) - - elif triggered_id == 'logs-next-btn': + elif triggered_id == "logs-next-btn": current_page = current_page + 1 - - elif isinstance(triggered_id, dict) and triggered_id.get('type') == 'page-btn': - if ctx.triggered and ctx.triggered[0]['value']: - current_page = triggered_id['index'] + elif isinstance(triggered_id, dict) and triggered_id.get("type") == "page-btn": + if ctx.triggered and ctx.triggered[0]["value"]: + current_page = triggered_id["index"] current_page = max(0, min(current_page, total_pages - 1)) start_idx = current_page * PAGE_SIZE end_idx = min(start_idx + PAGE_SIZE, total_rows) - page_data = prepared_df.iloc[start_idx:end_idx].to_dict('records') - - columns = [{"name": i, "id": i} for i in prepared_df.columns] - - info_text = (f"Page {current_page + 1} of {total_pages} | " - f"Showing rows {start_idx + 1:,}–{end_idx:,} " - f"of {total_rows:,} total frames") - + page_data = prepared_df.iloc[start_idx:end_idx].to_dict("records") + columns = [{"name": c, "id": c} for c in prepared_df.columns] + info_text = ( + f"Page {current_page + 1} of {total_pages} | " + f"Showing rows {start_idx + 1:,}–{end_idx:,} " + f"of {total_rows:,} total frames" + ) page_buttons = build_page_buttons(current_page, total_pages) - return page_data, columns, info_text, page_buttons, current_page @app.callback( - Output('tab-content', 'children'), - Input('tabs', 'active_tab'), - Input('vehicle-selector', 'value'), - Input('bus-selector', 'value') + Output("tab-content", "children"), + Input("tabs", "active_tab"), + Input("vehicle-selector", "value"), + Input("bus-selector", "value"), ) def render_content(tab, vehicle, bus): if not vehicle or not bus or vehicle not in DATA or bus not in DATA[vehicle]: @@ -315,59 +414,82 @@ def render_content(tab, vehicle, bus): df = DATA[vehicle][bus] - if tab == 'freq': - return dcc.Graph(figure=PRECOMPUTED_FIGURES[f"{vehicle}_{bus}_freq"], style={'height': '80vh'}) + if tab == "freq": + return dcc.Graph( + figure=PRECOMPUTED_FIGURES[f"{vehicle}_{bus}_freq"], + style={"height": "80vh"}, + ) - elif tab == 'id_viewer': + if tab == "id_viewer": ids = sorted(DATA_BY_ID.get((vehicle, bus), {}).keys()) - return html.Div([ - html.Label("CAN ID:"), - dcc.Dropdown( - id='id-selector', - options=[{'label': i, 'value': i} for i in ids], - value=ids[0] if ids else None, - clearable=False, - style={'width': '50%', 'marginBottom': '10px'} - ), - dcc.Graph(id='id-viewer-graph', style={'height': '70vh'}) - ]) + return html.Div( + [ + html.Label("CAN ID:"), + dcc.Dropdown( + id="id-selector", + options=[{"label": i, "value": i} for i in ids], + value=ids[0] if ids else None, + clearable=False, + style={"width": "50%", "marginBottom": "10px"}, + ), + dcc.Graph(id="id-viewer-graph", style={"height": "70vh"}), + ] + ) - elif tab == 'corr': + if tab == "corr": ids = sorted(DATA_BY_ID.get((vehicle, bus), {}).keys()) - return html.Div([ - dbc.Row([ - dbc.Col(html.Label("Method:"), width=1, className="mt-2"), - dbc.Col(dcc.Dropdown( - id='corr-method', - options=[{'label': 'Pearson', 'value': 'pearson'}, {'label': 'Spearman', 'value': 'spearman'}], - value='pearson', - clearable=False - ), width=2), - dbc.Col(html.Label("Target ID:"), width=1, className="mt-2"), - dbc.Col(dcc.Dropdown( - id='corr-target', - options=[{'label': 'All IDs (Max Corr)', 'value': 'all'}] + [{'label': i, 'value': i} for i in ids], - value='all', - clearable=True - ), width=4), - ], className="mb-3"), - dcc.Graph(id='corr-graph', style={'height': '80vh'}) - ]) + return html.Div( + [ + dbc.Row( + [ + dbc.Col(html.Label("Method:"), width=1, className="mt-2"), + dbc.Col( + dcc.Dropdown( + id="corr-method", + options=[ + {"label": "Pearson", "value": "pearson"}, + {"label": "Spearman", "value": "spearman"}, + ], + value="pearson", + clearable=False, + ), + width=2, + ), + dbc.Col(html.Label("Target ID:"), width=1, className="mt-2"), + dbc.Col( + dcc.Dropdown( + id="corr-target", + options=[{"label": "All IDs (Max Corr)", "value": "all"}] + + [{"label": i, "value": i} for i in ids], + value="all", + clearable=True, + ), + width=4, + ), + ], + className="mb-3", + ), + dcc.Graph(id="corr-graph", style={"height": "80vh"}), + ] + ) - elif tab == 'entropy': - return dcc.Graph(figure=PRECOMPUTED_FIGURES[f"{vehicle}_{bus}_entropy"], style={'height': '80vh'}) + if tab == "entropy": + return dcc.Graph( + figure=PRECOMPUTED_FIGURES[f"{vehicle}_{bus}_entropy"], + style={"height": "80vh"}, + ) return html.Div("Tab not found") @app.callback( - Output('id-viewer-graph', 'figure'), - Input('id-selector', 'value'), - Input('vehicle-selector', 'value'), - Input('bus-selector', 'value'), - Input('tabs', 'active_tab'), + Output("id-viewer-graph", "figure"), + Input("id-selector", "value"), + Input("vehicle-selector", "value"), + Input("bus-selector", "value"), + Input("tabs", "active_tab"), ) def update_id_viewer(selected_id, vehicle, bus, tab): - if tab != 'id_viewer' or not selected_id or not vehicle or not bus: + if tab != "id_viewer" or not selected_id or not vehicle or not bus: return dash.no_update grouped_data = DATA_BY_ID.get((vehicle, bus), {}) @@ -375,39 +497,41 @@ def update_id_viewer(selected_id, vehicle, bus, tab): return dash.no_update filtered_df, byte_cols = grouped_data[selected_id] - return plot_bits(filtered_df, byte_cols, selected_id, title=f"{vehicle} {bus} Byte Visualization") + return plot_bits( + filtered_df, byte_cols, selected_id, + title=f"{vehicle} {bus} Byte Visualization", + ) @app.callback( - Output('corr-graph', 'figure'), - Input('corr-method', 'value'), - Input('corr-target', 'value'), - Input('vehicle-selector', 'value'), - Input('bus-selector', 'value'), - Input('tabs', 'active_tab'), + Output("corr-graph", "figure"), + Input("corr-method", "value"), + Input("corr-target", "value"), + Input("vehicle-selector", "value"), + Input("bus-selector", "value"), + Input("tabs", "active_tab"), ) def update_corr(method, target, vehicle, bus, tab): - if tab != 'corr' or not vehicle or not bus: + if tab != "corr" or not vehicle or not bus: return dash.no_update - target_id = None if target == 'all' or not target else target + target_id = None if target == "all" or not target else target cache_key = (vehicle, bus, method, target_id) if cache_key not in CORR_CACHE: df = DATA[vehicle][bus] - corr_df = calculate_correlation(df, method=method, target_id=target_id) - CORR_CACHE[cache_key] = corr_df - else: - corr_df = CORR_CACHE[cache_key] + CORR_CACHE[cache_key] = calculate_correlation( + df, method=method, target_id=target_id + ) + corr_df = CORR_CACHE[cache_key] title = f"{vehicle} {bus} Correlation" if target_id: title += f" ({target_id})" - return plot_correlation_heatmap(corr_df, target_id=target_id, title=title) @app.callback( - Output('vehicles-content', 'children'), - Input('vehicles-vehicle-selector', 'value') + Output("vehicles-content", "children"), + Input("vehicles-vehicle-selector", "value"), ) def render_vehicles(vehicle): if not vehicle or vehicle not in DATA: @@ -416,54 +540,58 @@ def render_vehicles(vehicle): 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) - + dfs = list(DATA[vehicle].values()) 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) + 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 item in frame_def.signals: - if hasattr(item, 'plot_func') and callable(item.plot_func): + if hasattr(item, "plot_func") and callable(item.plot_func): title = f"{frame_def.can_id} - {item.name}" fig = item.plot_func(decoded, frame_def.color) else: sig = item - if getattr(sig, 'skip_plot', False): + if getattr(sig, "skip_plot", False): continue - 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, color=frame_def.color) - - 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") + fig = vehicle_module.plot_signal( + decoded, sig.name, title=title, color=frame_def.color + ) cards.append( - dbc.Col(card, xs=12, sm=6, md=4, lg=3, className="mb-3") + dbc.Col( + dbc.Card( + [ + dbc.CardBody( + [ + dcc.Graph( + figure=fig, + config={"displayModeBar": False}, + style={"height": "280px"}, + ) + ], + className="p-2", + ), + ], + className="shadow-sm border-0 h-100", + ), + 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" + className="text-muted", ) - return dbc.Row(cards) -if __name__ == '__main__': +if __name__ == "__main__": app.run(debug=False) diff --git a/stats/correlation.py b/stats/correlation.py index 6cbca29..4ae5812 100644 --- a/stats/correlation.py +++ b/stats/correlation.py @@ -1,25 +1,23 @@ -# File: correlation.py +# File: stats/correlation.py # Copyright (C) 2026 Erick Ahmed # SPDX-License-Identifier: AGPL-3.0-or-later +"""CAN bus inter-byte correlation analyzer and plotter.""" + import argparse from pathlib import Path from concurrent.futures import ThreadPoolExecutor +from typing import List import numpy as np import pandas as pd import plotly.graph_objects as go -from stats.utils.extractor import load_data -from stats.utils.extractor import to_int +from stats.utils.converter import format_can_id_vec as _format_can_id_vec, to_int +from stats.utils.loader import load_data -def _format_can_id_vec(s: pd.Series) -> pd.Series: - s = s.astype('string').str.strip() - s = s.str.replace(r'^0x', '', case=False, regex=True) - s = s.str.upper() - return s.fillna('UNKNOWN').replace('', 'UNKNOWN') -def _ensure_int_bytes(df: pd.DataFrame, cols: list) -> pd.DataFrame: +def _ensure_int_bytes(df: pd.DataFrame, cols: List[str]) -> pd.DataFrame: needs = [c for c in cols if not pd.api.types.is_numeric_dtype(df[c])] if needs: df = df.copy() @@ -27,6 +25,7 @@ def _ensure_int_bytes(df: pd.DataFrame, cols: list) -> pd.DataFrame: df[c] = df[c].apply(to_int) return df + def calculate_correlation(df: pd.DataFrame, method: str, target_id: str | None = None) -> pd.DataFrame: available_cols = [f"b{i}" for i in range(8) if f"b{i}" in df.columns] @@ -70,7 +69,7 @@ def calculate_correlation(df: pd.DataFrame, method: str, target_id: str | None = else: groups = [] - def _process_group(sub): + def _process_group(sub: np.ndarray) -> np.ndarray: mask = ~np.isnan(sub).any(axis=1) sub = sub[mask] if len(sub) > 1: diff --git a/stats/entropy.py b/stats/entropy.py index e39e5d2..47eb5c8 100644 --- a/stats/entropy.py +++ b/stats/entropy.py @@ -1,23 +1,21 @@ -# File: entropy.py +# File: stats/entropy.py # Copyright (C) 2026 Erick Ahmed # SPDX-License-Identifier: AGPL-3.0-or-later +"""CAN bus byte-level entropy analyzer and plotter.""" + import argparse from pathlib import Path from concurrent.futures import ThreadPoolExecutor +from typing import List import numpy as np import pandas as pd import plotly.graph_objects as go -from stats.utils.extractor import load_data -from stats.utils.extractor import to_int +from stats.utils.converter import format_can_id_vec as _format_can_id_vec, to_int +from stats.utils.loader import load_data -def _format_can_id_vec(s: pd.Series) -> pd.Series: - s = s.astype('string').str.strip() - s = s.str.replace(r'^0x', '', case=False, regex=True) - s = s.str.upper() - return s.fillna('UNKNOWN').replace('', 'UNKNOWN') def _entropy_col(a: np.ndarray) -> float: a = a[~np.isnan(a)] @@ -36,6 +34,7 @@ def _entropy_col(a: np.ndarray) -> float: p = counts / counts.sum() return float(-np.sum(p * np.log2(p))) + def calculate_byte_entropy(df: pd.DataFrame) -> pd.DataFrame: available_cols = [f"b{i}" for i in range(8) if f"b{i}" in df.columns] if not available_cols: @@ -64,7 +63,7 @@ def calculate_byte_entropy(df: pd.DataFrame) -> pd.DataFrame: else: groups = [] - def _process_group(sub): + def _process_group(sub: np.ndarray) -> np.ndarray: res = np.zeros(n_cols, dtype=np.float64) for ci in range(n_cols): res[ci] = _entropy_col(sub[:, ci]) diff --git a/stats/frequency.py b/stats/frequency.py index 57b1f5f..ca9bc50 100644 --- a/stats/frequency.py +++ b/stats/frequency.py @@ -1,19 +1,19 @@ -# File: frequency.py +# File: stats/frequency.py # Copyright (C) 2026 Erick Ahmed # SPDX-License-Identifier: AGPL-3.0-or-later +"""CAN bus message frequency analyzer and plotter.""" + import argparse from pathlib import Path + import numpy as np import pandas as pd import plotly.graph_objects as go -from stats.utils.extractor import load_data -def _format_can_id_vec(s: pd.Series) -> pd.Series: - s = s.astype('string').str.strip() - s = s.str.replace(r'^0x', '', case=False, regex=True) - s = s.str.upper() - return s.fillna('UNKNOWN').replace('', 'UNKNOWN') +from stats.utils.converter import format_can_id_vec as _format_can_id_vec +from stats.utils.loader import load_data + def calculate_frequency(df: pd.DataFrame) -> pd.DataFrame: can_id_col = 'ID' if 'ID' in df.columns else 'Identifier' @@ -28,6 +28,7 @@ def calculate_frequency(df: pd.DataFrame) -> pd.DataFrame: freq_df['Percentage'] = np.round(freq_df['Count'] / total * 100, 2) if total else 0.0 return freq_df.sort_values('Count', ascending=True).reset_index(drop=True) + def plot_frequency(stats_df: pd.DataFrame, title: str) -> go.Figure: n = len(stats_df) fig = go.Figure(go.Bar( @@ -106,6 +107,7 @@ def plot_frequency(stats_df: pd.DataFrame, title: str) -> go.Figure: ) return fig + if __name__ == "__main__": parser = argparse.ArgumentParser(description="Analyze CAN bus message frequency") parser.add_argument("input", type=Path, help="Path to the input CAN log file") diff --git a/stats/id_viewer.py b/stats/id_viewer.py index f3e30c8..768e477 100644 --- a/stats/id_viewer.py +++ b/stats/id_viewer.py @@ -1,25 +1,30 @@ -# File: id_viewer.py +# File: stats/id_viewer.py # Copyright (C) 2026 Erick Ahmed # SPDX-License-Identifier: AGPL-3.0-or-later +"""Interactive CAN bus byte-change visualizer.""" + import argparse from pathlib import Path +from typing import List, Tuple + import numpy as np import pandas as pd import plotly.graph_objects as go from plotly_resampler import FigureResampler -from stats.utils.extractor import load_data -def _format_can_id_vec(s: pd.Series) -> pd.Series: - s = s.astype('string').str.strip() - s = s.str.replace(r'^0x', '', case=False, regex=True) - s = s.str.upper() - return s.fillna('UNKNOWN').replace('', 'UNKNOWN') +from stats.utils.converter import format_can_id_vec as _format_can_id_vec +from stats.utils.loader import load_data -def prepare_data(df, target_id): +_BYTE_COLORS = [ + '#e41a1c', '#377eb8', '#4daf4a', '#984ea3', + '#ff7f00', '#ffff33', '#a65628', '#f781bf', +] + + +def prepare_data(df: pd.DataFrame, target_id: str) -> Tuple[pd.DataFrame, List[str]]: can_id_col = 'ID' if 'ID' in df.columns else 'Identifier' - formatted = _format_can_id_vec(df[can_id_col]) - df = df.assign(Formatted_ID=formatted) + df = df.assign(Formatted_ID=_format_can_id_vec(df[can_id_col])) target_id_clean = _format_can_id_vec(pd.Series([target_id])).iloc[0] filtered = df[df['Formatted_ID'] == target_id_clean] @@ -29,7 +34,9 @@ def prepare_data(df, target_id): for col in byte_cols: if not pd.api.types.is_numeric_dtype(filtered[col]): - filtered = filtered.assign(**{col: pd.to_numeric(filtered[col], errors='coerce').astype('float32')}) + filtered = filtered.assign( + **{col: pd.to_numeric(filtered[col], errors='coerce').astype('float32')} + ) filtered = filtered.sort_values('Timestamp', kind='stable') arr = filtered[byte_cols].to_numpy(dtype=np.float32, copy=False) @@ -40,12 +47,12 @@ def prepare_data(df, target_id): return filtered, byte_cols -def plot_bits(df, byte_cols, can_id, title): + +def plot_bits(df: pd.DataFrame, byte_cols: List[str], can_id: str, title: str) -> FigureResampler: fig = FigureResampler( resampled_trace_prefix_suffix=("", ""), show_mean_aggregation_size=False ) - colors = ['#e41a1c', '#377eb8', '#4daf4a', '#984ea3', '#ff7f00', '#ffff33', '#a65628', '#f781bf'] n = len(byte_cols) x = df['Timestamp'].to_numpy() if not df.empty else np.array([]) @@ -54,7 +61,7 @@ def plot_bits(df, byte_cols, can_id, title): fig.add_trace(go.Scatter( mode='lines', - line=dict(shape='hv', width=2, color=colors[i % len(colors)]), + line=dict(shape='hv', width=2, color=_BYTE_COLORS[i % len(_BYTE_COLORS)]), name=col.upper(), legendgroup=col.upper(), hovertemplate=f"{col.upper()}
Time: %{{x}}
Value: %{{y}}", @@ -121,6 +128,7 @@ def plot_bits(df, byte_cols, can_id, title): ) return fig + if __name__ == "__main__": parser = argparse.ArgumentParser(description="Visualize CAN bus byte changes over time") parser.add_argument("input", type=Path, help="Path to the input CAN log file") diff --git a/stats/utils/extractor.py b/stats/utils/extractor.py deleted file mode 100644 index 3472f07..0000000 --- a/stats/utils/extractor.py +++ /dev/null @@ -1,63 +0,0 @@ -# File: extractor.py -# Copyright (C) 2026 Erick Ahmed -# SPDX-License-Identifier: AGPL-3.0-or-later - -import json -from pathlib import Path - -import numpy as np -import pandas as pd -import polars as pl - -def to_int(x): - if isinstance(x, (int, np.integer)): - return int(x) - if isinstance(x, str): - try: - return int(x, 16) - except ValueError: - return np.nan - return np.nan - -def extract_id(row: pd.Series) -> str: - meta = row.get('j1939_metadata') - if pd.isna(meta): - return f"ID: {row['ID']}" - if isinstance(meta, str): - try: - meta = json.loads(meta) - except json.JSONDecodeError: - return f"ID: {row['ID']}" - if isinstance(meta, dict) and 'PGN' in meta: - return f"PGN: {meta['PGN']}" - return f"ID: {row['ID']}" - -def load_data(file_path: Path) -> pd.DataFrame: - lf = pl.scan_parquet(file_path) - schema = lf.collect_schema() - names = schema.names() - - byte_cols = [f"b{i}" for i in range(8) if f"b{i}" in names] - if byte_cols: - lf = lf.with_columns([ - pl.col(c).str.to_integer(base=16, strict=False).cast(pl.Int16).alias(c) - for c in byte_cols - ]) - - id_col = 'ID' if 'ID' in names else 'Identifier' - id_expr = pl.col(id_col).cast(pl.Utf8) - - if 'j1939_metadata' in names: - try: - lf = lf.with_columns( - pl.when(pl.col('j1939_metadata').is_not_null()) - .then(pl.lit('PGN: ') + pl.col('j1939_metadata').struct.field('PGN').cast(pl.Utf8)) - .otherwise(pl.lit('ID: ') + id_expr) - .alias('Identifier') - ) - except Exception: - lf = lf.with_columns((pl.lit('ID: ') + id_expr).alias('Identifier')) - else: - lf = lf.with_columns((pl.lit('ID: ') + id_expr).alias('Identifier')) - - return lf.collect().to_pandas() diff --git a/vehicle/base.py b/vehicle/base.py index 56e20db..4e23a57 100644 --- a/vehicle/base.py +++ b/vehicle/base.py @@ -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"{signal_name}
Time: %{{x}}
Value: %{{y:.2f}}", - )) + 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"{signal_name}
Time: %{{x}}
" + f"Value: %{{y:.2f}}" + ), + ) + ) 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 diff --git a/vehicle/komatsu.py b/vehicle/komatsu.py index f1ab977..0e933ca 100644 --- a/vehicle/komatsu.py +++ b/vehicle/komatsu.py @@ -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) -- 2.52.0