Normalize CAN IDs to strictly match same number of bit

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: