Compare commits
3 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| d0e70dad2a | |||
| f39d23fc64 | |||
| 7daf4c8e08 |
+180
@@ -0,0 +1,180 @@
|
||||
# File: entropy.py
|
||||
# Copyright (C) 2026 Erick Ahmed
|
||||
# SPDX-License-Identifier: AGPL-3.0-or-later
|
||||
|
||||
import argparse
|
||||
from pathlib import Path
|
||||
|
||||
import numpy as np
|
||||
import pandas as pd
|
||||
import plotly.graph_objects as go
|
||||
|
||||
from utils.extractor import load_data
|
||||
|
||||
|
||||
def _to_int(x):
|
||||
"""Convert a hex string or integer to int, returning NaN on failure."""
|
||||
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 calculate_byte_entropy(df: pd.DataFrame) -> pd.DataFrame:
|
||||
"""Calculates Shannon entropy per byte position for each identifier."""
|
||||
byte_cols = [f"b{i}" for i in range(8)]
|
||||
available_cols = [col for col in byte_cols if col in df.columns]
|
||||
if not available_cols:
|
||||
raise ValueError("No byte columns (b0-b7) found in the DataFrame")
|
||||
|
||||
df_bytes = df[available_cols].copy()
|
||||
for col in available_cols:
|
||||
df_bytes[col] = df_bytes[col].apply(_to_int)
|
||||
|
||||
def entropy(s: pd.Series) -> float:
|
||||
s = s.dropna()
|
||||
if s.empty:
|
||||
return 0.0
|
||||
p = s.value_counts(normalize=True)
|
||||
return -np.sum(p * np.log2(p))
|
||||
|
||||
return df.groupby("Identifier")[available_cols].agg(entropy)
|
||||
|
||||
|
||||
def plot_entropy_heatmap(entropy_df: pd.DataFrame, title: str) -> go.Figure:
|
||||
"""Generates an interactive heatmap of byte-level Shannon entropy."""
|
||||
x = entropy_df.columns.tolist()
|
||||
y = entropy_df.index.tolist()
|
||||
z = entropy_df.values
|
||||
|
||||
fig = go.Figure(
|
||||
data=go.Heatmap(
|
||||
z=z,
|
||||
x=x,
|
||||
y=y,
|
||||
colorscale=[
|
||||
[0.0, "#ffffff"],
|
||||
[0.15, "#fff7ec"],
|
||||
[0.35, "#fee8c8"],
|
||||
[0.55, "#fdd49e"],
|
||||
[0.75, "#fdbb84"],
|
||||
[1.0, "#ef6548"],
|
||||
],
|
||||
xgap=3,
|
||||
ygap=3,
|
||||
text=np.round(z, 2),
|
||||
texttemplate="%{text}",
|
||||
textfont={
|
||||
"size": 11,
|
||||
"color": "#2a2a2a",
|
||||
"family": "Segoe UI, Arial, sans-serif",
|
||||
},
|
||||
hoverongaps=False,
|
||||
hovertemplate=(
|
||||
"<b>%{y}</b><br>"
|
||||
"Byte %{x}: %{z:.2f} bits<extra></extra>"
|
||||
),
|
||||
colorbar=dict(
|
||||
title=dict(
|
||||
text="Entropy (bits)",
|
||||
side="top",
|
||||
font=dict(size=13, color="#1a1a1a"),
|
||||
),
|
||||
orientation="h",
|
||||
thickness=15,
|
||||
len=0.35,
|
||||
x=1.0,
|
||||
xanchor="right",
|
||||
y=1.02,
|
||||
yanchor="bottom",
|
||||
tickfont=dict(size=11, color="#2a2a2a"),
|
||||
tickformat=".1f",
|
||||
outlinewidth=0.5,
|
||||
outlinecolor="#cccccc",
|
||||
),
|
||||
)
|
||||
)
|
||||
|
||||
fig.update_layout(
|
||||
title=dict(
|
||||
text=title,
|
||||
font=dict(size=20, color="#1a1a1a"),
|
||||
x=0.5,
|
||||
xanchor="center",
|
||||
pad=dict(b=20),
|
||||
),
|
||||
height=max(600, len(y) * 28 + 150),
|
||||
autosize=True,
|
||||
template="plotly_white",
|
||||
xaxis=dict(
|
||||
title=dict(text="Byte Position", font=dict(size=13, color="#1a1a1a")),
|
||||
side="top",
|
||||
dtick=1,
|
||||
showgrid=False,
|
||||
linecolor="#bdbdbd",
|
||||
tickfont=dict(size=12, color="#2a2a2a"),
|
||||
ticks="outside",
|
||||
ticklen=4,
|
||||
tickcolor="#cccccc",
|
||||
),
|
||||
yaxis=dict(
|
||||
title=dict(text="PGN or CAN ID", font=dict(size=13, color="#1a1a1a")),
|
||||
autorange="reversed",
|
||||
showgrid=False,
|
||||
linecolor="#bdbdbd",
|
||||
tickfont=dict(size=12, color="#2a2a2a"),
|
||||
ticks="outside",
|
||||
ticklen=4,
|
||||
tickcolor="#cccccc",
|
||||
automargin=True,
|
||||
),
|
||||
font=dict(family="Segoe UI, Arial, sans-serif", size=12, color="#2a2a2a"),
|
||||
hoverlabel=dict(
|
||||
bgcolor="white",
|
||||
font_size=13,
|
||||
font_family="Segoe UI",
|
||||
bordercolor="#cccccc",
|
||||
),
|
||||
margin=dict(l=200, r=40, t=120, b=60),
|
||||
)
|
||||
return fig
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
parser = argparse.ArgumentParser(
|
||||
description="Analyze CAN bus byte-level entropy"
|
||||
)
|
||||
parser.add_argument(
|
||||
"input", type=Path, help="Path to the input CAN log file"
|
||||
)
|
||||
parser.add_argument(
|
||||
"output",
|
||||
type=Path,
|
||||
nargs="?",
|
||||
default=Path("entropy_report.html"),
|
||||
help="Path to the output HTML report",
|
||||
)
|
||||
parser.add_argument(
|
||||
"title",
|
||||
nargs="?",
|
||||
default="CAN Bus Byte-Level Entropy",
|
||||
help="Title for the HTML report",
|
||||
)
|
||||
args = parser.parse_args()
|
||||
|
||||
df = load_data(args.input)
|
||||
entropy_df = calculate_byte_entropy(df)
|
||||
fig = plot_entropy_heatmap(entropy_df, title=args.title)
|
||||
|
||||
config = {
|
||||
"responsive": True,
|
||||
"displaylogo": False,
|
||||
"scrollZoom": True,
|
||||
"modeBarButtonsToAdd": ["toggleSpikelines"],
|
||||
"toImageButtonOptions": {"format": "png", "scale": 2},
|
||||
}
|
||||
fig.write_html(str(args.output), include_plotlyjs="cdn", config=config)
|
||||
@@ -1,27 +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 pandas as pd
|
||||
|
||||
def extract_id(row: pd.Series) -> str:
|
||||
"""Extracts PGN from metadata or falls back to CAN ID."""
|
||||
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:
|
||||
"""Loads Parquet file and adds an Identifier column."""
|
||||
df = pd.read_parquet(file_path)
|
||||
df['Identifier'] = df.apply(extract_id, axis=1)
|
||||
return df
|
||||
Reference in New Issue
Block a user