From c9d9f9bbec7d0c118e4ef2f286b07d3ce1b23fd9 Mon Sep 17 00:00:00 2001 From: Erick Ahmed Date: Mon, 13 Jul 2026 23:14:24 +0200 Subject: [PATCH] Normalize CAN IDs to strictly match same number of bit --- stat/correlation.py | 32 ++++++++++++++++++++++++++++---- stat/entropy.py | 25 ++++++++++++++++++++++--- stat/frequency.py | 37 +++++++++++++++++++++++++++++++------ 3 files changed, 81 insertions(+), 13 deletions(-) diff --git a/stat/correlation.py b/stat/correlation.py index 4d0f70c..eefed05 100644 --- a/stat/correlation.py +++ b/stat/correlation.py @@ -3,6 +3,7 @@ # SPDX-License-Identifier: AGPL-3.0-or-later import argparse +import numbers from pathlib import Path import numpy as np @@ -12,6 +13,21 @@ import plotly.graph_objects as go from utils.extractor import load_data from utils.extractor import to_int +def _format_can_id(x): + if pd.isna(x): + return "UNKNOWN" + if isinstance(x, numbers.Number): + return f"{int(x):08X}" + s = str(x).strip() + if s.lower().startswith('0x'): + s = s[2:] + if s.isdigit(): + return f"{int(s):08X}" + try: + return f"{int(s, 16):08X}" + except ValueError: + return s + def calculate_correlation(df: pd.DataFrame, method: str, target_id: str | None = None) -> pd.DataFrame: """Calculates inter-byte correlation grouped by identifier.""" byte_cols = [f"b{i}" for i in range(8)] @@ -20,12 +36,18 @@ def calculate_correlation(df: pd.DataFrame, method: str, target_id: str | None = if not available_cols: raise ValueError("No byte columns (b0-b7) found in the DataFrame") - df_bytes = df[["Identifier"] + available_cols].copy() + can_id_col = 'ID' if 'ID' in df.columns else 'Identifier' + identifiers = df[can_id_col].apply(_format_can_id) + + if target_id is not None: + target_id = _format_can_id(target_id) + + df_bytes = df[available_cols].copy() for col in available_cols: df_bytes[col] = df_bytes[col].apply(to_int) - if target_id: - group = df_bytes[df_bytes["Identifier"] == target_id] + if target_id is not None: + group = df_bytes[identifiers == target_id] if group.empty: raise ValueError(f"Identifier '{target_id}' not found in data") return group[available_cols].corr(method=method).fillna(0.0) @@ -35,7 +57,9 @@ def calculate_correlation(df: pd.DataFrame, method: str, target_id: str | None = np.fill_diagonal(corr_arr, 0.0) return pd.Series(corr_arr.max(axis=0), index=group.columns).fillna(0.0) - return df_bytes.groupby("Identifier")[available_cols].apply(max_abs_corr) + result = df_bytes.groupby(identifiers)[available_cols].apply(max_abs_corr) + result.index.name = 'Identifier' + return result def plot_correlation_heatmap(corr_df: pd.DataFrame, target_id: str | None, title: str) -> go.Figure: diff --git a/stat/entropy.py b/stat/entropy.py index ca314ea..6eb6299 100644 --- a/stat/entropy.py +++ b/stat/entropy.py @@ -3,6 +3,7 @@ # SPDX-License-Identifier: AGPL-3.0-or-later import argparse +import numbers from pathlib import Path import numpy as np @@ -12,6 +13,21 @@ import plotly.graph_objects as go from utils.extractor import load_data from utils.extractor import to_int +def _format_can_id(x): + if pd.isna(x): + return "UNKNOWN" + if isinstance(x, numbers.Number): + return f"{int(x):08X}" + s = str(x).strip() + if s.lower().startswith('0x'): + s = s[2:] + if s.isdigit(): + return f"{int(s):08X}" + try: + return f"{int(s, 16):08X}" + except ValueError: + return s + 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)] @@ -19,6 +35,9 @@ def calculate_byte_entropy(df: pd.DataFrame) -> pd.DataFrame: if not available_cols: raise ValueError("No byte columns (b0-b7) found in the DataFrame") + can_id_col = 'ID' if 'ID' in df.columns else 'Identifier' + identifiers = df[can_id_col].apply(_format_can_id) + df_bytes = df[available_cols].copy() for col in available_cols: df_bytes[col] = df_bytes[col].apply(to_int) @@ -30,8 +49,9 @@ def calculate_byte_entropy(df: pd.DataFrame) -> pd.DataFrame: p = s.value_counts(normalize=True) return -np.sum(p * np.log2(p)) - return df.groupby("Identifier")[available_cols].agg(entropy) - + result = df_bytes.groupby(identifiers).agg(entropy) + result.index.name = 'Identifier' + return result def plot_entropy_heatmap(entropy_df: pd.DataFrame, title: str) -> go.Figure: """Generates an interactive heatmap of byte-level Shannon entropy.""" @@ -131,7 +151,6 @@ def plot_entropy_heatmap(entropy_df: pd.DataFrame, title: str) -> go.Figure: ) return fig - if __name__ == "__main__": parser = argparse.ArgumentParser( description="Analyze CAN bus byte-level entropy" diff --git a/stat/frequency.py b/stat/frequency.py index b3e185a..3904d8b 100644 --- a/stat/frequency.py +++ b/stat/frequency.py @@ -3,21 +3,46 @@ # SPDX-License-Identifier: AGPL-3.0-or-later import argparse +import numbers from pathlib import Path import pandas as pd import plotly.express as px import plotly.graph_objects as go from utils.extractor import load_data -def calc_freq(df: pd.DataFrame) -> pd.DataFrame: +def calculate_frequency(df: pd.DataFrame) -> pd.DataFrame: """Calculates frequency counts and percentages for identifiers.""" - freq_df = df['Identifier'].value_counts().reset_index() + + can_id_col = 'ID' if 'ID' in df.columns else 'Identifier' + + def format_can_id(x): + if pd.isna(x): + return "UNKNOWN" + if isinstance(x, numbers.Number): + return f"{int(x):08X}" + s = str(x).strip() + if s.lower().startswith('0x'): + s = s[2:] + if s.isdigit(): + return f"{int(s):08X}" + try: + return f"{int(s, 16):08X}" + except ValueError: + return s + + raw_ids = df[can_id_col] + formatted_ids = raw_ids.apply(format_can_id) + + freq_df = formatted_ids.value_counts().reset_index() + freq_df.columns = ['Identifier', 'Count'] + total = freq_df['Count'].sum() freq_df['Percentage'] = (freq_df['Count'] / total * 100).round(2) + return freq_df.sort_values('Count', ascending=True) -def plot_freq(stats_df: pd.DataFrame, title: str) -> go.Figure: +def plot_frequency(stats_df: pd.DataFrame, title: str) -> go.Figure: """Generates interactive horizontal bar chart with log x-axis.""" fig = px.bar( stats_df, y='Identifier', x='Count', orientation='h', title=title, log_x=True, @@ -43,7 +68,6 @@ def plot_freq(stats_df: pd.DataFrame, title: str) -> go.Figure: ), yaxis=dict( title=dict(text="PGN or CAN ID", font=dict(size=13, color="#1a1a1a")), - #autorange="", showgrid=False, linecolor="#bdbdbd", tickfont=dict(size=12, color="#2a2a2a"), @@ -51,6 +75,7 @@ def plot_freq(stats_df: pd.DataFrame, title: str) -> go.Figure: ticklen=4, tickcolor="#cccccc", automargin=True, + type='category' # Force categorical axis to ensure every ID gets a tick ), font=dict(family="Segoe UI, Arial, sans-serif", size=12, color='#2a2a2a'), hoverlabel=dict(bgcolor="white", font_size=13, font_family="Segoe UI", @@ -104,8 +129,8 @@ if __name__ == "__main__": args = parser.parse_args() df = load_data(args.input) - stats = calc_freq(df) - fig = plot_freq(stats, title=args.title) + stats = calculate_frequency(df) + fig = plot_frequency(stats, title=args.title) config = { 'responsive': True, 'displaylogo': False,