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