Implement plotly resamper and precompute data #5
@@ -9,13 +9,13 @@ import polars as pl
|
||||
import dash
|
||||
from dash import dcc, html, Input, Output
|
||||
import dash_bootstrap_components as dbc
|
||||
import numpy as np
|
||||
|
||||
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 stats.id_viewer import prepare_data, plot_bits
|
||||
from stats.correlation import calculate_correlation, plot_correlation_heatmap
|
||||
from stats.entropy import calculate_byte_entropy, plot_entropy_heatmap
|
||||
|
||||
@@ -55,7 +55,35 @@ DATA = {
|
||||
"Bus 2": load_data(BUS2_DECODED)
|
||||
}
|
||||
|
||||
PRECOMPUTED_FIGURES = {}
|
||||
DATA_BY_ID = {}
|
||||
CORR_CACHE = {}
|
||||
|
||||
for bus, df in DATA.items():
|
||||
PRECOMPUTED_FIGURES[f"{bus}_freq"] = plot_frequency(calculate_frequency(df), title=f"{bus} Frequency")
|
||||
PRECOMPUTED_FIGURES[f"{bus}_entropy"] = plot_entropy_heatmap(calculate_byte_entropy(df), title=f"{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('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[can_id] = (group, byte_cols)
|
||||
|
||||
DATA_BY_ID[bus] = grouped
|
||||
|
||||
app = dash.Dash(__name__, external_stylesheets=[dbc.themes.BOOTSTRAP])
|
||||
app.config.suppress_callback_exceptions = True
|
||||
|
||||
app.layout = dbc.Container([
|
||||
html.H1("CAN Bus Analyzer", className="my-4"),
|
||||
@@ -86,12 +114,10 @@ def render_content(tab, bus):
|
||||
df = DATA[bus]
|
||||
|
||||
if tab == 'freq':
|
||||
stats = calculate_frequency(df)
|
||||
fig = plot_frequency(stats, title=f"{bus} Frequency")
|
||||
return dcc.Graph(figure=fig, style={'height': '80vh'})
|
||||
return dcc.Graph(figure=PRECOMPUTED_FIGURES[f"{bus}_freq"], style={'height': '80vh'})
|
||||
|
||||
elif tab == 'id_viewer':
|
||||
ids = sorted(df['ID'].unique().tolist())
|
||||
ids = sorted(DATA_BY_ID[bus].keys())
|
||||
return html.Div([
|
||||
html.Label("Select CAN ID:"),
|
||||
dcc.Dropdown(
|
||||
@@ -105,7 +131,7 @@ def render_content(tab, bus):
|
||||
])
|
||||
|
||||
elif tab == 'corr':
|
||||
ids = sorted(df['ID'].unique().tolist())
|
||||
ids = sorted(DATA_BY_ID[bus].keys())
|
||||
return html.Div([
|
||||
dbc.Row([
|
||||
dbc.Col(html.Label("Method:"), width=1, className="mt-2"),
|
||||
@@ -127,9 +153,7 @@ def render_content(tab, bus):
|
||||
])
|
||||
|
||||
elif tab == 'entropy':
|
||||
entropy_df = calculate_byte_entropy(df)
|
||||
fig = plot_entropy_heatmap(entropy_df, title=f"{bus} Byte-Level Entropy")
|
||||
return dcc.Graph(figure=fig, style={'height': '80vh'})
|
||||
return dcc.Graph(figure=PRECOMPUTED_FIGURES[f"{bus}_entropy"], style={'height': '80vh'})
|
||||
|
||||
return html.Div("Tab not found")
|
||||
|
||||
@@ -142,8 +166,12 @@ def render_content(tab, bus):
|
||||
def update_id_viewer(selected_id, bus, tab):
|
||||
if tab != 'id_viewer' or not selected_id:
|
||||
return dash.no_update
|
||||
df = DATA[bus]
|
||||
filtered_df, byte_cols = prepare_data(df, selected_id)
|
||||
|
||||
grouped_data = DATA_BY_ID.get(bus, {})
|
||||
if selected_id not in grouped_data:
|
||||
return dash.no_update
|
||||
|
||||
filtered_df, byte_cols = grouped_data[selected_id]
|
||||
return plot_bits(filtered_df, byte_cols, selected_id, title=f"{bus} Byte Visualization")
|
||||
|
||||
@app.callback(
|
||||
@@ -156,12 +184,21 @@ def update_id_viewer(selected_id, bus, tab):
|
||||
def update_corr(method, target, bus, tab):
|
||||
if tab != 'corr':
|
||||
return dash.no_update
|
||||
df = DATA[bus]
|
||||
|
||||
target_id = None if target == 'all' or not target else target
|
||||
corr_df = calculate_correlation(df, method=method, target_id=target_id)
|
||||
cache_key = (bus, method, target_id)
|
||||
|
||||
if cache_key not in CORR_CACHE:
|
||||
df = DATA[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]
|
||||
|
||||
title = f"{bus} Correlation"
|
||||
if target_id:
|
||||
title += f" ({target_id})"
|
||||
|
||||
return plot_correlation_heatmap(corr_df, target_id=target_id, title=title)
|
||||
|
||||
if __name__ == '__main__':
|
||||
|
||||
+8
-5
@@ -7,6 +7,7 @@ from pathlib import Path
|
||||
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:
|
||||
@@ -40,22 +41,24 @@ def prepare_data(df, target_id):
|
||||
return filtered, byte_cols
|
||||
|
||||
def plot_bits(df, byte_cols, can_id, title):
|
||||
fig = go.Figure()
|
||||
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([])
|
||||
for i, col in enumerate(byte_cols):
|
||||
y = df[col].to_numpy(dtype=np.float32, copy=False) if not df.empty else np.array([])
|
||||
fig.add_trace(go.Scattergl(
|
||||
x=x,
|
||||
y=y,
|
||||
|
||||
fig.add_trace(go.Scatter(
|
||||
mode='lines',
|
||||
line=dict(shape='hv', width=2, color=colors[i % len(colors)]),
|
||||
name=col.upper(),
|
||||
legendgroup=col.upper(),
|
||||
hovertemplate=f"<b>{col.upper()}</b><br>Time: %{{x}}<br>Value: %{{y}}<extra></extra>",
|
||||
))
|
||||
), hf_x=x, hf_y=y)
|
||||
|
||||
all_button = dict(label='ALL', method='restyle', args=[{'visible': [True] * n}])
|
||||
none_button = dict(label='NONE', method='restyle', args=[{'visible': ['legendonly'] * n}])
|
||||
|
||||
Reference in New Issue
Block a user