#!/usr/bin/env python3 import os import torch import numpy as np import glob import matplotlib.pyplot as plt from pathlib import Path import sys import json from tqdm import tqdm import inspect import argparse from collections import OrderedDict, defaultdict import time # <<< IMPORT TIME >>> # Add the project root to path sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), '..'))) # Try to force deterministic algorithms torch.backends.cudnn.benchmark = False torch.backends.cudnn.deterministic = True print("PYTHON SCRIPT: Set cuDNN benchmark=False, deterministic=True") # Import model wrappers from pytracking.features.net_wrappers import DiMPTorchScriptWrapper # For loading AtomIoUNet from source from ltr.models.bbreg.atom_iou_net import AtomIoUNet # Add import for new modular comparison from model_comparison.bbreg_comparison import compare_debug_tensors SCRIPT_DIR_FOR_INIT = os.path.dirname(os.path.abspath(__file__)) ROOT_DIR_FOR_INIT = os.path.dirname(SCRIPT_DIR_FOR_INIT) # --- Model Configurations --- def get_model_configs(root_dir_param): # ... (rest of get_model_configs, ensuring it uses root_dir_param if needed) # For now, assume it doesn't strictly need root_dir_param for paths if they are relative to script # or if model_dir in DiMPTorchScriptWrapper handles it. return { # ... (existing model_configs definitions) 'ResNet': { 'python_model_loader': lambda: DiMPTorchScriptWrapper(os.path.join(root_dir_param, 'pytracking_models/dimp50_ausdruck_ep0050.pth.tar')), 'cpp_output_subdir': 'resnet', 'python_output_subdir': 'resnet_py', 'outputs_to_compare': { 'Conv1': ('conv1_output.pt', 'conv1'), 'Debug ResNet Conv1->BN1 Input': ('debug_resnet_conv1_output_for_bn1_input.pt', 'conv1_pre_bn'), # BN1 final output (manual C++ vs manual Python pre-ReLU) 'BN1': ('bn1_output.pt', 'bn1_post_relu_pre'), # BN1 Intermediate comparisons 'BN1 Centered X': ('bn1_centered_x.pt', 'bn1_centered_x_py'), 'BN1 Var+Eps': ('bn1_variance_plus_eps.pt', 'bn1_variance_plus_eps_py'), 'BN1 InvStd': ('bn1_inv_std.pt', 'bn1_inv_std_py'), 'BN1 Normalized X': ('bn1_normalized_x.pt', 'bn1_normalized_x_py'), 'ReLU1': ('relu1_output.pt', 'conv1'), 'MaxPool': 'maxpool_output.pt', 'Features': 'features.pt', 'Layer1': 'layer1.pt', 'Layer2': 'layer2.pt', 'Layer3': 'layer3.pt', 'Layer4': 'layer4.pt', 'Layer1.0 Shortcut': 'layer1_0_shortcut_output.pt' } }, 'Classifier': { 'python_model_loader': lambda: DiMPTorchScriptWrapper(os.path.join(root_dir_param, 'pytracking_models/dimp50_ Ausdruck_ep0050.pth.tar')), 'cpp_output_subdir': 'classifier', 'python_output_subdir': 'classifier_py', 'outputs_to_compare': { 'Features': 'features.pt', } }, 'BBRegressor': { 'python_model_loader': lambda: DiMPTorchScriptWrapper(os.path.join(root_dir_param, 'pytracking_models/dimp50_ Ausdruck_ep0050.pth.tar')), 'cpp_output_subdir': 'bb_regressor', 'python_output_subdir': 'bb_regressor_py', 'outputs_to_compare': { 'IoUPred': 'iou_scores.pt', 'PyIoUFeat0': ('iou_feat0.pt', True), # True indicates Python-specific output name 'CppIoUFeat0': 'iou_feat0.pt', 'PyIoUFeat1': ('iou_feat1.pt', True), 'CppIoUFeat1': 'iou_feat1.pt', 'PyMod0': ('mod_vec0.pt', True), 'CppMod0': 'mod_vec0.pt', 'PyMod1': ('mod_vec1.pt', True), 'CppMod1': 'mod_vec1.pt', } }, } class ComparisonRunner: def __init__(self, root_dir, model_configs, cpp_output_dir, python_output_dir, num_samples=-1, plot_histograms=True, plot_scatter=True): self.root_dir = root_dir self.model_configs = model_configs self.cpp_output_dir = cpp_output_dir self.python_output_dir = python_output_dir self.num_samples = num_samples self.plot_histograms = plot_histograms self.plot_scatter = plot_scatter self.all_comparison_stats = defaultdict(lambda: defaultdict(list)) self.python_wrapper = None # ADDED: To store the DiMPTorchScriptWrapper instance self.models = {} # To store loaded Python sub-models like ResNet, Classifier self.device = torch.device("cuda" if torch.cuda.is_available() else "cpu") # Ensure comparison directory exists self.comparison_dir = os.path.join(self.root_dir, "test/comparison") if not os.path.exists(self.comparison_dir): os.makedirs(self.comparison_dir) print("PYTHON: Attempting to load 'traced_resnet50.pth'...") try: self.models['ResNet'] = torch.jit.load('traced_resnet50.pth', map_location=self.device) print("PYTHON: Successfully loaded 'traced_resnet50.pth'.") self.models['ResNet'].eval() print("PYTHON: ResNet JIT model set to eval().") except Exception as e: print(f"PYTHON: CRITICAL ERROR loading 'traced_resnet50.pth': {e}") self.models['ResNet'] = None # Ensure it's None if loading failed # Print sums of ResNet.bn1 running_mean and running_var from state_dict print("PYTHON: Attempting to access ResNet state_dict (if model loaded)...") if self.models.get('ResNet'): try: resnet_state_dict = self.models['ResNet'].state_dict() print("PYTHON ResNet state_dict keys:", list(resnet_state_dict.keys())) # PRINT ALL KEYS py_bn1_running_mean = resnet_state_dict.get('bn1.running_mean') py_bn1_running_var = resnet_state_dict.get('bn1.running_var') if py_bn1_running_mean is not None and py_bn1_running_var is not None: print(f"PYTHON ResNet.bn1 running_mean sum (from state_dict): {py_bn1_running_mean.sum().item():.10f}") print(f"PYTHON ResNet.bn1 running_var sum (from state_dict): {py_bn1_running_var.sum().item():.10f}") else: print("PYTHON: ResNet.bn1 running_mean or running_var is None in state_dict.") except Exception as e: print(f"PYTHON: Error accessing ResNet.bn1 state_dict: {e}") # Load other models if necessary (e.g., BBRegressor, Classifier) def load_python_models(self): print("DEBUG: ComparisonRunner.load_python_models() ENTERED") # DEBUG PRINT """Initialize Python models""" print("Loading Python models...") self.python_wrapper = DiMPTorchScriptWrapper( model_dir=str(Path(self.root_dir) / 'exported_weights'), device=self.device, backbone_sd='backbone_regenerated', # CORRECTED: Ensure this uses regenerated weights classifier_sd='classifier', bbregressor_sd='bb_regressor' ) # Populate self.models AFTER python_wrapper is initialized if self.python_wrapper: # Check if wrapper was successfully initialized print("DEBUG: self.python_wrapper initialized. Populating self.models.") # DEBUG PRINT if hasattr(self.python_wrapper, 'backbone') and self.python_wrapper.backbone is not None: # Check for 'backbone' self.models['ResNet'] = self.python_wrapper.backbone # Assign from .backbone print(f"DEBUG: self.models['ResNet'] populated with type: {type(self.models['ResNet'])}") # DEBUG PRINT else: print("ERROR: python_wrapper does not have a 'backbone' attribute or it is None.") self.models['ResNet'] = None if hasattr(self.python_wrapper, 'classifier') and self.python_wrapper.classifier is not None: self.models['Classifier'] = self.python_wrapper.classifier print(f"DEBUG: self.models['Classifier'] populated with type: {type(self.models['Classifier'])}") # DEBUG PRINT else: print("ERROR: python_wrapper does not have a 'classifier' attribute or it is None.") self.models['Classifier'] = None if hasattr(self.python_wrapper, 'bb_regressor') and self.python_wrapper.bb_regressor is not None: self.models['BBRegressor'] = self.python_wrapper.bb_regressor print(f"DEBUG: self.models['BBRegressor'] populated with type: {type(self.models['BBRegressor'])}") # DEBUG PRINT else: print("ERROR: python_wrapper does not have a 'bb_regressor' attribute or it is None.") self.models['BBRegressor'] = None else: print("CRITICAL ERROR: self.python_wrapper is None after DiMPTorchScriptWrapper instantiation.") # Ensure self.models has keys to prevent crashes later, though values will be None self.models['ResNet'] = None self.models['Classifier'] = None self.models['BBRegressor'] = None # Initialize BBRegressor from source for get_modulation fallback self.bb_regressor_from_source = AtomIoUNet( input_dim=(512, 1024), pred_input_dim=(256, 256), pred_inter_dim=(256, 256) ) ComparisonRunner.load_weights_for_custom_model( self.bb_regressor_from_source, 'bb_regressor', # model_name for path and doc file self.root_dir, self.device ) self.bb_regressor_from_source.eval().to(self.device) print("Python models loaded.") # New check: Compare the conv1.weight actually used by Python ResNet vs C++ ResNet from backbone_regenerated print("\n--- COMPARING CURRENTLY USED conv1.weight (Python vs C++) ---") python_resnet_conv1_weight = None if self.models.get('ResNet') and hasattr(self.models['ResNet'], 'conv1'): python_resnet_conv1_weight = self.models['ResNet'].conv1.weight.detach().cpu() print(f" Python ResNet model's conv1.weight shape: {python_resnet_conv1_weight.shape}") cpp_conv1_path = os.path.join(self.root_dir, "exported_weights/backbone_regenerated/conv1_weight.pt") cpp_resnet_conv1_weight = None if os.path.exists(cpp_conv1_path): try: cpp_resnet_conv1_weight = torch.load(cpp_conv1_path, map_location='cpu', weights_only=False) print(f" C++ (loaded from {cpp_conv1_path}) conv1.weight shape: {cpp_resnet_conv1_weight.shape}") except Exception as e: print(f" Error loading C++ conv1.weight from {cpp_conv1_path}: {e}") if python_resnet_conv1_weight is not None and cpp_resnet_conv1_weight is not None: if isinstance(python_resnet_conv1_weight, torch.Tensor) and isinstance(cpp_resnet_conv1_weight, torch.Tensor): print(f" torch.allclose(python_model_conv1, cpp_loaded_conv1): {torch.allclose(python_resnet_conv1_weight, cpp_resnet_conv1_weight)}") abs_diff = torch.abs(python_resnet_conv1_weight - cpp_resnet_conv1_weight) print(f" Max abs diff for conv1.weight: {torch.max(abs_diff).item()}") print(f" Mean abs diff for conv1.weight: {torch.mean(abs_diff).item()}") else: print(" Skipping conv1.weight comparison due to type mismatch after loading.") else: print(" Skipping conv1.weight comparison because one or both tensors could not be obtained.") print("--- END CURRENTLY USED conv1.weight COMPARISON ---\n") # New check: Compare ResNet bn1 parameters print("\n--- COMPARING CURRENTLY USED bn1 PARAMS (Python vs C++) ---") bn1_param_names = ['weight', 'bias', 'running_mean', 'running_var'] python_resnet_bn1_params = {} if self.models.get('ResNet') and hasattr(self.models['ResNet'], 'bn1'): bn1_module = self.models['ResNet'].bn1 for p_name in bn1_param_names: if hasattr(bn1_module, p_name): param_tensor = getattr(bn1_module, p_name) if param_tensor is not None: python_resnet_bn1_params[p_name] = param_tensor.detach().cpu() print(f" Python ResNet model's bn1.{p_name} shape: {python_resnet_bn1_params[p_name].shape}") else: print(f" Python ResNet model's bn1.{p_name} is None.") else: print(f" Python ResNet model's bn1 does not have attribute {p_name}.") cpp_resnet_bn1_params = {} for p_name in bn1_param_names: # Adjust filename for C++ saved tensors (e.g., bn1_running_mean.pt) cpp_param_filename = f"bn1_{p_name.replace('.', '_')}.pt" cpp_param_path = os.path.join(self.root_dir, "exported_weights/backbone_regenerated", cpp_param_filename) if os.path.exists(cpp_param_path): try: cpp_resnet_bn1_params[p_name] = torch.load(cpp_param_path, map_location='cpu', weights_only=False) print(f" C++ (loaded from {cpp_param_path}) bn1.{p_name} shape: {cpp_resnet_bn1_params[p_name].shape}") except Exception as e: print(f" Error loading C++ bn1.{p_name} from {cpp_param_path}: {e}") else: print(f" C++ bn1 parameter file not found: {cpp_param_path}") for p_name in bn1_param_names: py_tensor = python_resnet_bn1_params.get(p_name) cpp_tensor = cpp_resnet_bn1_params.get(p_name) print(f" Comparison for bn1.{p_name}:") if py_tensor is not None and cpp_tensor is not None: if isinstance(py_tensor, torch.Tensor) and isinstance(cpp_tensor, torch.Tensor): print(f" torch.allclose(python_bn1_{p_name}, cpp_bn1_{p_name}): {torch.allclose(py_tensor, cpp_tensor)}") abs_diff = torch.abs(py_tensor - cpp_tensor) print(f" Max abs diff for bn1.{p_name}: {torch.max(abs_diff).item()}") print(f" Mean abs diff for bn1.{p_name}: {torch.mean(abs_diff).item()}") else: print(f" Skipping bn1.{p_name} comparison due to type mismatch after loading.") else: print(f" Skipping bn1.{p_name} comparison because one or both tensors could not be obtained.") print("--- END CURRENTLY USED bn1 PARAMS COMPARISON ---\n") # New check: Compare ResNet layer1.0 parameters print("\n--- COMPARING CURRENTLY USED layer1.0 PARAMS (Python vs C++) ---") layer1_0_block_prefix = "layer1.0." layer1_0_components = { "conv1": ["weight"], "bn1": ["weight", "bias", "running_mean", "running_var"], "conv2": ["weight"], "bn2": ["weight", "bias", "running_mean", "running_var"], "conv3": ["weight"], "bn3": ["weight", "bias", "running_mean", "running_var"], "downsample.0": ["weight"], # Downsample Conv "downsample.1": ["weight", "bias", "running_mean", "running_var"] # Downsample BN } if self.models.get('ResNet') and hasattr(self.models['ResNet'], 'layer1') and len(self.models['ResNet'].layer1) > 0: py_layer1_0_module = self.models['ResNet'].layer1[0] for comp_name, param_list in layer1_0_components.items(): py_comp_module = py_layer1_0_module try: # Handle nested modules like downsample.0 for part_name in comp_name.split('.'): py_comp_module = getattr(py_comp_module, part_name) except AttributeError: print(f" Python ResNet model's layer1.0 does not have component {comp_name}. Skipping.") continue for p_name in param_list: py_param_tensor_name = f"{layer1_0_block_prefix}{comp_name}.{p_name}" cpp_param_filename = f"{layer1_0_block_prefix.replace('.', '_')}{comp_name.replace('.', '_')}_{p_name}.pt" py_param_tensor = None if hasattr(py_comp_module, p_name): param_tensor_val = getattr(py_comp_module, p_name) if param_tensor_val is not None: py_param_tensor = param_tensor_val.detach().cpu() print(f" Python ResNet {py_param_tensor_name} shape: {py_param_tensor.shape}") else: print(f" Python ResNet {py_param_tensor_name} is None.") else: print(f" Python ResNet module {comp_name} does not have param {p_name}.") cpp_param_path = os.path.join(self.root_dir, "exported_weights/backbone_regenerated", cpp_param_filename) cpp_param_tensor = None if os.path.exists(cpp_param_path): try: cpp_param_tensor = torch.load(cpp_param_path, map_location='cpu', weights_only=False) # print(f" C++ (loaded from {cpp_param_path}) {cpp_param_filename} shape: {cpp_param_tensor.shape}") # Optional: less verbose except Exception as e: print(f" Error loading C++ {cpp_param_filename} from {cpp_param_path}: {e}") # Adjusted to cpp_param_filename else: print(f" Warning: C++ {cpp_param_filename} file not found: {cpp_param_path}") # Adjusted print(f" Comparison for {py_param_tensor_name} vs {cpp_param_filename}:") # More specific if py_param_tensor is not None and cpp_param_tensor is not None: if isinstance(py_param_tensor, torch.Tensor) and isinstance(cpp_param_tensor, torch.Tensor): all_close = torch.allclose(py_param_tensor, cpp_param_tensor) print(f" torch.allclose: {all_close}") if not all_close: abs_diff = torch.abs(py_param_tensor - cpp_param_tensor) print(f" Max abs diff: {torch.max(abs_diff).item()}") print(f" Mean abs diff: {torch.mean(abs_diff).item()}") else: print(f" Skipping comparison due to type mismatch after loading.") else: print(f" Skipping comparison because one or both tensors could not be obtained.") else: print(" Skipping layer1.0 parameter comparison: ResNet model or its layer1 not found/empty.") print("--- END CURRENTLY USED layer1.0 PARAMS COMPARISON ---\n") # Corrected to \n # --- START WEIGHT COMPARISON FOR layer1.1 and layer1.2 --- for block_idx_in_layer1 in [1, 2]: # For layer1.1 and layer1.2 print(f"\n--- COMPARING CURRENTLY USED layer1.{block_idx_in_layer1} PARAMS (Python vs C++) ---") layer1_block_prefix = f"layer1.{block_idx_in_layer1}." # Components within a standard bottleneck block (no downsample for these) block_components = { "conv1": ["weight"], "bn1": ["weight", "bias", "running_mean", "running_var", "num_batches_tracked"], "conv2": ["weight"], "bn2": ["weight", "bias", "running_mean", "running_var", "num_batches_tracked"], "conv3": ["weight"], "bn3": ["weight", "bias", "running_mean", "running_var", "num_batches_tracked"], } if self.models.get('ResNet') and hasattr(self.models['ResNet'], 'layer1') and len(self.models['ResNet'].layer1) > block_idx_in_layer1: py_layer1_block_module = self.models['ResNet'].layer1[block_idx_in_layer1] for comp_name, param_list in block_components.items(): py_comp_module = py_layer1_block_module try: # No nested modules like 'downsample' for these blocks py_comp_module = getattr(py_comp_module, comp_name) except AttributeError: print(f" Python ResNet model's layer1.{block_idx_in_layer1} does not have component {comp_name}. Skipping.") continue for p_name in param_list: py_param_tensor_name = f"{layer1_block_prefix}{comp_name}.{p_name}" # C++ saves files like layer1_0_bn1_weight.pt or layer1_1_bn1_weight.pt cpp_param_filename = f"{layer1_block_prefix.replace('.', '_')}{comp_name.replace('.', '_')}_{p_name}.pt" py_param_tensor = None if hasattr(py_comp_module, p_name): param_tensor_val = getattr(py_comp_module, p_name) if param_tensor_val is not None: py_param_tensor = param_tensor_val.detach().cpu() print(f" Python ResNet {py_param_tensor_name} shape: {py_param_tensor.shape}") else: print(f" Python ResNet {py_param_tensor_name} is None.") elif p_name == "num_batches_tracked" and isinstance(py_comp_module, torch.nn.BatchNorm2d): # PyTorch stores num_batches_tracked in _buffers, not as a direct attribute usually if py_comp_module.num_batches_tracked is not None: py_param_tensor = py_comp_module.num_batches_tracked.detach().cpu() print(f" Python ResNet {py_param_tensor_name} (from buffer) shape: {py_param_tensor.shape}") else: print(f" Python ResNet {py_param_tensor_name} (from buffer) is None.") else: print(f" Python ResNet module {comp_name} does not have param/buffer {p_name}.") cpp_param_path = os.path.join(self.root_dir, "exported_weights/backbone_regenerated", cpp_param_filename) cpp_param_tensor = None if os.path.exists(cpp_param_path): try: cpp_param_tensor = torch.load(cpp_param_path, map_location='cpu', weights_only=False) except Exception as e: print(f" Error loading C++ {cpp_param_filename} from {cpp_param_path}: {e}") else: print(f" Warning: C++ {cpp_param_filename} file not found: {cpp_param_path}") print(f" Comparison for {py_param_tensor_name} vs {cpp_param_filename}:") if py_param_tensor is not None and cpp_param_tensor is not None: if isinstance(py_param_tensor, torch.Tensor) and isinstance(cpp_param_tensor, torch.Tensor): # Ensure tensors are float for allclose if one is int (e.g. num_batches_tracked) py_param_tensor_float = py_param_tensor.float() cpp_param_tensor_float = cpp_param_tensor.float() all_close = torch.allclose(py_param_tensor_float, cpp_param_tensor_float) print(f" torch.allclose: {all_close}") if not all_close: abs_diff = torch.abs(py_param_tensor_float - cpp_param_tensor_float) mae = torch.mean(abs_diff).item() max_abs_err = torch.max(abs_diff).item() print(f" MAE (Weight/Buffer): {mae:.4e}") print(f" Max Abs Err (Weight/Buffer): {max_abs_err:.4e}") # Also print L2 norms for context l2_py = torch.linalg.norm(py_param_tensor_float.flatten()).item() l2_cpp = torch.linalg.norm(cpp_param_tensor_float.flatten()).item() print(f" L2 Norm Python: {l2_py:.4e}") print(f" L2 Norm C++: {l2_cpp:.4e}") else: print(f" Skipping comparison due to type mismatch after loading for {py_param_tensor_name}.") else: print(f" Skipping comparison because one or both tensors could not be obtained for {py_param_tensor_name}.") else: print(f" Skipping layer1.{block_idx_in_layer1} parameter comparison: ResNet model or its layer1 not found/long enough.") print(f"--- END CURRENTLY USED layer1.{block_idx_in_layer1} PARAMS COMPARISON ---\n") # --- END WEIGHT COMPARISON FOR layer1.1 and layer1.2 --- # --- END TEMPORARY WEIGHT COMPARISON --- # This marker is now after layer1.0 checks print("\n--- Types at END of load_python_models: ---") if 'ResNet' in self.models: print(f" self.models['ResNet'] type: {type(self.models['ResNet'])}") if 'Classifier' in self.models: print(f" self.models['Classifier'] type: {type(self.models['Classifier'])}") if 'BBRegressor' in self.models: print(f" self.models['BBRegressor'] type: {type(self.models['BBRegressor'])}") def compare_classifier(self): """Compare classifier model outputs between Python and C++""" print("\nComparing classifier outputs...") # Python model needs C++ ResNet output as its input cpp_input_dir_path = Path(os.path.join(self.cpp_output_dir, 'resnet')) cpp_output_classifier_dir = Path(os.path.join(self.cpp_output_dir, 'classifier')) if not cpp_input_dir_path.exists() or not cpp_output_classifier_dir.exists(): print(f"Classifier input (C++ ResNet features from {cpp_input_dir_path}) or C++ Classifier output dir ({cpp_output_classifier_dir}) not found. Skipping Classifier comparison.") # Populate NaN for all expected Classifier comparisons if dirs are missing for i in range(self.num_samples): sample_key_base = f"Clf_Sample_{i}" current_errors = {} self._compare_tensor_data(None, None, "Classifier Features", i, current_errors) self.all_comparison_stats[sample_key_base] = current_errors return print("\nClassifier - Comparing Samples...") for i in tqdm(range(self.num_samples), desc="Classifier samples"): current_errors = {} # For this sample py_clf_feat = None cpp_clf_feat = None # Input for Python classifier is the layer3 output of C++ ResNet cpp_resnet_layer3_for_py_path = cpp_input_dir_path / f'sample_{i}_layer3.pt' # C++ classifier output cpp_classifier_feat_path = cpp_output_classifier_dir / f'sample_{i}_features.pt' if not cpp_resnet_layer3_for_py_path.exists() or not cpp_classifier_feat_path.exists(): print(f"Warning: Skipping classifier sample {i}, files not found: C++ ResNet output {cpp_resnet_layer3_for_py_path} or C++ Clf output {cpp_classifier_feat_path}.") else: feat_from_cpp_resnet = self.load_cpp_tensor(cpp_resnet_layer3_for_py_path, self.device) if feat_from_cpp_resnet is None: print(f"Critical: Failed to load C++ ResNet output tensor {cpp_resnet_layer3_for_py_path} for classifier sample {i}.") else: try: with torch.no_grad(): if self.models.get('Classifier'): py_clf_feat = self.models['Classifier'].extract_classification_feat(feat_from_cpp_resnet) else: print("ERROR: Python Classifier model not found in self.models") except Exception as e: print(f"ERROR: Python model extract_classification_feat failed for sample {i}: {e}") cpp_clf_feat = self.load_cpp_tensor(cpp_classifier_feat_path, self.device) if cpp_clf_feat is None: print(f"Warning: Failed to load C++ output tensor {cpp_classifier_feat_path} for classifier sample {i}.") self._compare_tensor_data(py_clf_feat, cpp_clf_feat, "Classifier Features", i, current_errors) if current_errors: self.all_comparison_stats[f"Clf_Sample_{i}"] = current_errors # Removed the separate "Test Samples" loop for classifier for simplification # The C++ test_models only produces one set of classifier outputs per sample. def compare_bb_regressor(self): """Compare bb_regressor model outputs between Python and C++""" print("\nComparing bb_regressor outputs...") # Python model inputs come from 'common' C++ generated/loaded files # C++ model outputs are in 'output/bb_regressor' py_input_common_dir = os.path.join(self.root_dir, 'test', 'input_samples', 'common') cpp_output_bb_reg_dir = os.path.join(self.cpp_output_dir, 'bb_regressor') cpp_resnet_output_dir = os.path.join(self.cpp_output_dir, 'resnet') # Convert to Path for exists check py_input_common_dir_path = Path(py_input_common_dir) cpp_output_bb_reg_dir_path = Path(cpp_output_bb_reg_dir) cpp_resnet_output_dir_path = Path(cpp_resnet_output_dir) if not py_input_common_dir_path.exists() or not cpp_output_bb_reg_dir_path.exists() or not cpp_resnet_output_dir_path.exists(): print(f"BB Regressor input ({py_input_common_dir_path}), C++ ResNet output ({cpp_resnet_output_dir_path}), or C++ BB Reg output dir ({cpp_output_bb_reg_dir_path}) not found. Skipping BB Regressor comparison.") # Populate NaN for all expected BB Regressor comparisons if dirs are missing for i in range(self.num_samples): sample_key_base = f"BBReg_Sample_{i}" current_errors = {} self._compare_tensor_data(None, None, "BBReg PyIoUFeat0 vs CppIoUFeat0", i, current_errors) self._compare_tensor_data(None, None, "BBReg PyIoUFeat1 vs CppIoUFeat1", i, current_errors) self._compare_tensor_data(None, None, "BBReg PyMod0 vs CppMod0", i, current_errors) self._compare_tensor_data(None, None, "BBReg PyMod1 vs CppMod1", i, current_errors) self._compare_tensor_data(None, None, "BBReg IoUPred", i, current_errors) self.all_comparison_stats[sample_key_base] = current_errors return for i in tqdm(range(self.num_samples), desc="BB Regressor samples"): current_errors = {} # For this sample # --- Python Model Path --- # For BBRegressor, the Python model needs to run its own ResNet pass # using the common input image. py_image_input_path = py_input_common_dir_path / f'sample_{i}_image.pt' py_init_bbox_path = py_input_common_dir_path / f'sample_{i}_bb.pt' py_proposals_path = py_input_common_dir_path / f'sample_{i}_proposals.pt' # --- C++ Model Outputs --- cpp_iou_feat0_path = cpp_output_bb_reg_dir_path / f'sample_{i}_iou_feat0.pt' cpp_iou_feat1_path = cpp_output_bb_reg_dir_path / f'sample_{i}_iou_feat1.pt' cpp_mod_vec0_path = cpp_output_bb_reg_dir_path / f'sample_{i}_mod_vec0.pt' cpp_mod_vec1_path = cpp_output_bb_reg_dir_path / f'sample_{i}_mod_vec1.pt' cpp_iou_scores_path = cpp_output_bb_reg_dir_path / f'sample_{i}_iou_scores.pt' # Paths for debug C++ outputs cpp_debug_conv3_1t_path = cpp_output_bb_reg_dir_path / f'sample_{i}_debug_conv3_1t_output.pt' cpp_debug_conv4_1t_path = cpp_output_bb_reg_dir_path / f'sample_{i}_debug_conv4_1t_output.pt' # Load initial inputs for Python model py_image_tensor = self.load_cpp_tensor(py_image_input_path, self.device) py_init_bbox = self.load_cpp_tensor(py_init_bbox_path, self.device) py_proposals = self.load_cpp_tensor(py_proposals_path, self.device) py_feat_layer2, py_feat_layer3 = None, None if py_image_tensor is not None: try: with torch.no_grad(): # Run Python ResNet backbone via the wrapper's method to include preprocessing if self.python_wrapper: py_backbone_outputs = self.python_wrapper.extract_backbone(py_image_tensor) else: print("ERROR: self.python_wrapper is None, cannot extract backbone features.") py_backbone_outputs = {} # Ensure it's a dict # Assign ResNet outputs to be used by BB Regressor py_feat_layer2 = py_backbone_outputs.get('layer2') py_feat_layer3 = py_backbone_outputs.get('layer3') except Exception as e: print(f"ERROR: Python ResNet backbone failed for sample {i}: {e}") else: print(f"Warning: Skipping Python BB Regressor for sample {i}, image input not found at {py_image_input_path}") # ---- Intermediate debug outputs for conv3_1t and conv4_1t ---- py_debug_conv3_1t_out = None py_debug_conv4_1t_out = None if py_feat_layer2 is not None: try: _feat2_for_debug_conv3_1t = py_feat_layer2 if _feat2_for_debug_conv3_1t.dim() == 5: _feat2_for_debug_conv3_1t = _feat2_for_debug_conv3_1t.reshape(-1, *_feat2_for_debug_conv3_1t.shape[-3:]) with torch.no_grad(): # Ensure no_grad context py_debug_conv3_1t_out = self.bb_regressor_from_source.conv3_1t(_feat2_for_debug_conv3_1t) except Exception as e: print(f"ERROR calculating Python Debug_Conv3_1t for sample {i}: {e}") if py_feat_layer3 is not None: try: _feat3_for_debug_conv4_1t = py_feat_layer3 if _feat3_for_debug_conv4_1t.dim() == 5: _feat3_for_debug_conv4_1t = _feat3_for_debug_conv4_1t.reshape(-1, *_feat3_for_debug_conv4_1t.shape[-3:]) with torch.no_grad(): # Ensure no_grad context py_debug_conv4_1t_out = self.bb_regressor_from_source.conv4_1t(_feat3_for_debug_conv4_1t) except Exception as e: print(f"ERROR calculating Python Debug_Conv4_1t for sample {i}: {e}") # ---- End intermediate debug outputs ---- # Get Python IoU features py_iou_feat_list = [None, None] # Initialize as a list of two Nones if py_feat_layer2 is not None and py_feat_layer3 is not None: try: # Use from-source get_iou_feat for consistent 256-channel features # DiMPTorchScriptWrapper.bb_regressor.get_iou_feat returns features with different channel counts temp_iou_feat = self.bb_regressor_from_source.get_iou_feat([py_feat_layer2, py_feat_layer3], i) if isinstance(temp_iou_feat, tuple): temp_iou_feat = list(temp_iou_feat) if len(temp_iou_feat) >= 2: py_iou_feat_list = [temp_iou_feat[0], temp_iou_feat[1]] elif len(temp_iou_feat) == 1: py_iou_feat_list[0] = temp_iou_feat[0] # print(f"Sample {i}: Py from-source get_iou_feat. Shapes: {[f.shape for f in py_iou_feat_list if f is not None]}") except Exception as e_iou_source: print(f"Sample {i}: Py from-source get_iou_feat failed: {e_iou_source}") # Get Python modulation vectors py_modulation_list = [None, None] # Initialize as a list of two Nones if py_feat_layer2 is not None and py_feat_layer3 is not None and py_init_bbox is not None: py_features_list = [py_feat_layer2, py_feat_layer3] squeezed_init_bbox = py_init_bbox if py_init_bbox.ndim == 3 and py_init_bbox.shape[0] > 0 and py_init_bbox.shape[1] == 1: squeezed_init_bbox = py_init_bbox.squeeze(1) try: # Using Torchscript model for modulation if self.python_wrapper and self.python_wrapper.bb_regressor: temp_mod = self.python_wrapper.bb_regressor.get_modulation(py_features_list, squeezed_init_bbox) else: print("ERROR: self.python_wrapper.bb_regressor is not available for get_modulation.") temp_mod = [None, None] if isinstance(temp_mod, tuple): temp_mod = list(temp_mod) if len(temp_mod) >= 2: py_modulation_list = [temp_mod[0], temp_mod[1]] elif len(temp_mod) == 1: py_modulation_list[0] = temp_mod[0] # print(f"Sample {i}: Py TorchScript get_modulation. Shapes: {[f.shape for f in py_modulation_list if f is not None]}") except Exception as e_ts: print(f"Sample {i}: Py TorchScript get_modulation failed: {e_ts}. Trying from-source.") try: temp_mod_source = self.bb_regressor_from_source.get_modulation(py_features_list, squeezed_init_bbox) if isinstance(temp_mod_source, tuple): temp_mod_source = list(temp_mod_source) if len(temp_mod_source) >=2: py_modulation_list = [temp_mod_source[0], temp_mod_source[1]] elif len(temp_mod_source) == 1: py_modulation_list[0] = temp_mod_source[0] # print(f"Sample {i}: Py from-source get_modulation. Shapes: {[f.shape for f in py_modulation_list if f is not None]}") except Exception as e_source: print(f"Sample {i}: Py from-source get_modulation also failed: {e_source}") # Run Python bb_regressor's predict_iou (from TorchScript model) py_iou_pred = None if all(f is not None for f in py_iou_feat_list) and \ all(m is not None for m in py_modulation_list) and \ py_proposals is not None: try: with torch.no_grad(): if self.python_wrapper and self.python_wrapper.bb_regressor: py_iou_pred = self.python_wrapper.bb_regressor.predict_iou(py_modulation_list, py_iou_feat_list, py_proposals) else: print("ERROR: self.python_wrapper.bb_regressor is not available for predict_iou.") py_iou_pred = None # print(f"Sample {i}: Py predict_iou output shape: {py_iou_pred.shape if py_iou_pred is not None else 'N/A'}") except Exception as e: print(f"ERROR: Python model predict_iou failed for sample {i}: {e}") # Load C++ outputs cpp_iou_feat0 = self.load_cpp_tensor(cpp_iou_feat0_path, self.device) cpp_iou_feat1 = self.load_cpp_tensor(cpp_iou_feat1_path, self.device) cpp_mod_vec0 = self.load_cpp_tensor(cpp_mod_vec0_path, self.device) cpp_mod_vec1 = self.load_cpp_tensor(cpp_mod_vec1_path, self.device) cpp_iou_scores = self.load_cpp_tensor(cpp_iou_scores_path, self.device) # Load debug C++ tensors cpp_debug_conv3_1t_tensor = self.load_cpp_tensor(cpp_debug_conv3_1t_path, self.device) cpp_debug_conv4_1t_tensor = self.load_cpp_tensor(cpp_debug_conv4_1t_path, self.device) # Comparisons self._compare_tensor_data(py_debug_conv3_1t_out, cpp_debug_conv3_1t_tensor, "BBReg Debug_Conv3_1t", i, current_errors) self._compare_tensor_data(py_debug_conv4_1t_out, cpp_debug_conv4_1t_tensor, "BBReg Debug_Conv4_1t", i, current_errors) self._compare_tensor_data(py_iou_feat_list[0], cpp_iou_feat0, "BBReg PyIoUFeat0 vs CppIoUFeat0", i, current_errors) self._compare_tensor_data(py_iou_feat_list[1], cpp_iou_feat1, "BBReg PyIoUFeat1 vs CppIoUFeat1", i, current_errors) self._compare_tensor_data(py_modulation_list[0], cpp_mod_vec0, "BBReg PyMod0 vs CppMod0", i, current_errors) self._compare_tensor_data(py_modulation_list[1], cpp_mod_vec1, "BBReg PyMod1 vs CppMod1", i, current_errors) self._compare_tensor_data(py_iou_pred, cpp_iou_scores, "BBReg IoUPred", i, current_errors) if current_errors: self.all_comparison_stats[f"BBReg_Sample_{i}"] = current_errors # For sample 0, save these tensors for comparison if i == 0: out_dir = Path('test/input_samples/common') out_dir.mkdir(parents=True, exist_ok=True) if py_feat_layer2 is not None: save_tensor_jit(py_feat_layer2.cpu(), out_dir / f'sample_0_layer2.pt') if py_feat_layer3 is not None: save_tensor_jit(py_feat_layer3.cpu(), out_dir / f'sample_0_layer3.pt') if py_proposals is not None: save_tensor_jit(py_proposals.cpu(), out_dir / f'sample_0_proposals.pt') if py_init_bbox is not None: save_tensor_jit(py_init_bbox.cpu(), out_dir / f'sample_0_bb.pt') def compare_resnet_outputs(self): print("\\n--- Comparing ResNet Outputs ---") if not self.models.get('ResNet'): print("PYTHON: ResNet model not loaded, skipping ResNet comparison.") return resnet_model = self.models['ResNet'] config = self.model_configs['ResNet'] cpp_resnet_dir = os.path.join(self.cpp_output_dir, config['cpp_output_subdir']) python_resnet_save_dir = os.path.join(self.python_output_dir, config.get('python_output_subdir', config['cpp_output_subdir'])) if not os.path.exists(python_resnet_save_dir): os.makedirs(python_resnet_save_dir, exist_ok=True) num_samples_to_process = self.num_samples if num_samples_to_process == -1: # If -1, determine from available C++ output files # This logic can be complex if C++ output is sparse. For now, let's assume if -1 it means process all *common* inputs. # A safer way for -1 would be to count common input samples first. common_input_glob = os.path.join(self.root_dir, "test", "input_samples", "common", "sample_*_image.pt") num_samples_to_process = len(glob.glob(common_input_glob)) print(f"INFO: num_samples set to -1, determined {num_samples_to_process} common input samples.") processed_samples_count = 0 # Renamed from processed_samples to avoid conflict sample_input_base_dir = os.path.join(self.root_dir, "test", "input_samples", "common") # Loop exactly self.num_samples times (or detected count if -1) for sample_idx in tqdm(range(num_samples_to_process), desc="Comparing ResNet samples"): current_errors = {} # Initialize for each sample python_intermediate_outputs_cache = {} # Reset for each sample # Construct the input file path based on sample_idx sample_input_file_path = os.path.join(sample_input_base_dir, f"sample_{sample_idx}_image.pt") if not os.path.exists(sample_input_file_path): print(f"Warning: Input sample file {sample_input_file_path} not found for sample index {sample_idx}. Skipping ResNet sample.") empty_errors_for_skipped_sample = {} for output_key_config in config['outputs_to_compare'].keys(): self._compare_tensor_data(None, None, output_key_config, sample_idx, empty_errors_for_skipped_sample) if empty_errors_for_skipped_sample: self.all_comparison_stats[f"ResNet_Sample_{sample_idx}"] = empty_errors_for_skipped_sample continue # --- START REINSTATED INPUT LOADING AND PREPROCESSING --- input_tensor = self.load_cpp_tensor(sample_input_file_path, self.device, is_image=True) if input_tensor is None: print(f"Warning: Failed to load a valid tensor for ResNet input sample {sample_input_file_path} (sample {sample_idx}) using self.load_cpp_tensor. Skipping.") # Populate NaNs for all expected outputs for this sample empty_errors_for_skipped_sample = {} for output_key_config in config['outputs_to_compare'].keys(): self._compare_tensor_data(None, None, output_key_config, sample_idx, empty_errors_for_skipped_sample) if empty_errors_for_skipped_sample: self.all_comparison_stats[f"ResNet_Sample_{sample_idx}"] = empty_errors_for_skipped_sample continue if not isinstance(input_tensor, torch.Tensor): print(f"Warning: self.load_cpp_tensor for {sample_input_file_path} did not return a Tensor (got {type(input_tensor)}). Skipping sample {sample_idx}.") # Populate NaNs for all expected outputs for this sample empty_errors_for_skipped_sample = {} for output_key_config in config['outputs_to_compare'].keys(): self._compare_tensor_data(None, None, output_key_config, sample_idx, empty_errors_for_skipped_sample) if empty_errors_for_skipped_sample: self.all_comparison_stats[f"ResNet_Sample_{sample_idx}"] = empty_errors_for_skipped_sample continue # Preprocess the input tensor for Python's ResNet if hasattr(self.python_wrapper, 'preprocess_image'): processed_input_tensor = self.python_wrapper.preprocess_image(input_tensor.clone()) # Use clone else: print("Warning: python_wrapper.preprocess_image not found. Using input_tensor as is.") processed_input_tensor = input_tensor.to(self.device) # Ensure device # --- END REINSTATED INPUT LOADING AND PREPROCESSING --- # --- Save preprocessed input for sample 0 --- if sample_idx == 0: preprocessed_dir = Path(self.cpp_output_dir) / 'resnet' preprocessed_dir.mkdir(parents=True, exist_ok=True) py_preprocessed_path = preprocessed_dir / f'sample_{sample_idx}_image_preprocessed_python.pt' torch.save(processed_input_tensor.cpu(), py_preprocessed_path) print(f"Saved Python preprocessed image for sample {sample_idx} to {py_preprocessed_path}") # --- END save preprocessed input --- # Initialize dictionaries to store Python-side outputs for the current sample python_outputs = {} # To store outputs from the Python model for this sample try: # Python ResNet forward pass (assuming it's a JIT model or similar) # The output of a JIT ResNet model might be a dictionary or a list/tuple of tensors # We need to ensure we can map these to the 'outputs_to_compare' keys print(f"PYTHON ResNet forward pass for sample {sample_idx}...") # For ResNet, the output is a dictionary from its forward method. # output_layers = list(config['outputs_to_compare'].keys()) # This might be too broad initially # Define the layers we actually need from the Python ResNet forward pass. # These should match the keys used in the Python ResNet's forward method. # e.g., ['layer1', 'layer2', 'layer3', 'layer4', 'conv1_output', 'bn1_output', etc.] # For now, let's define specific layers needed for the comparison. # The JIT ResNet model we have should output a dictionary. py_output_layers_needed = ['conv1', 'layer1', 'layer2', 'layer3', 'layer4'] # Only request layers that the Python ResNet model actually supports # The Python ResNet model only supports standard layers, not intermediate debug layers # Add 'fc' if configured, though not typically used in these comparisons if 'fc' in config['outputs_to_compare']: py_output_layers_needed.append('fc') # Deduplicate, just in case (though construction above should be fine) py_output_layers_needed = list(OrderedDict.fromkeys(py_output_layers_needed)) print(f"DEBUG: Requesting these layers from Python ResNet: {py_output_layers_needed}") # Call the Python ResNet forward # The `self.models['ResNet']` should be the loaded JIT model # It expects the output_layers argument. # The DiMPTorchScriptWrapper's backbone should also support this. if hasattr(resnet_model, 'forward') and callable(getattr(resnet_model, 'forward')) and 'output_layers' in inspect.signature(resnet_model.forward).parameters: python_model_outputs_dict = resnet_model.forward(processed_input_tensor, output_layers=py_output_layers_needed) elif hasattr(self.python_wrapper, 'extract_backbone') and callable(getattr(self.python_wrapper, 'extract_backbone')): # This is the case if ResNet is accessed via the DiMPTorchScriptWrapper's extract_backbone, # which internally calls the backbone's forward with output_layers. python_model_outputs_dict = self.python_wrapper.extract_backbone(input_tensor.clone()) # extract_backbone handles preprocessing else: print(f"ERROR: Cannot call forward on Python ResNet model. Type: {type(resnet_model)}") continue # DEBUG: Print keys from Python model output if isinstance(python_model_outputs_dict, dict): print(f"DEBUG RN_CMP: Keys from python_model_outputs_dict (sample {sample_idx}): {list(python_model_outputs_dict.keys())}") else: print(f"DEBUG RN_CMP: python_model_outputs_dict is not a dict (sample {sample_idx}), type: {type(python_model_outputs_dict)}") # Populate python_outputs based on the python_model_outputs_dict # This maps the Python output names to the keys used in 'outputs_to_compare' if isinstance(python_model_outputs_dict, dict): python_outputs = python_model_outputs_dict # If 'features' is an alias for 'layer4' in Python output if 'layer4' in python_outputs and 'features' not in python_outputs: python_outputs['features'] = python_outputs['layer4'] if 'conv1_output' in python_outputs: python_intermediate_outputs_cache['conv1_output'] = python_outputs['conv1_output'] else: print(f"ERROR: Python ResNet output is not a dict. Got {type(python_model_outputs_dict)}") # Handle tuple/list output if necessary, mapping by order or specific logic. # For now, we assume dict output from our ResNet. continue except Exception as e: print(f"Error during Python ResNet forward pass for sample {sample_idx}: {e}") import traceback traceback.print_exc() continue # Skip to next sample for output_key, cpp_output_filename_or_tuple in config['outputs_to_compare'].items(): is_python_specific_name = isinstance(cpp_output_filename_or_tuple, tuple) cpp_output_filename = cpp_output_filename_or_tuple[0] if is_python_specific_name else cpp_output_filename_or_tuple # Corrected path construction for C++ ResNet tensors: # The sample index is already part of the cpp_output_filename for ResNet outputs from C++. # (e.g., sample_0_conv1_output.pt) # So, we join cpp_resnet_dir directly with this filename. # However, the C++ code actually saves ResNet outputs as sample_X_LAYERNAME.pt directly in cpp_resnet_dir, # not in a per-sample subdirectory for ResNet outputs. # Let's check how test_models.cpp saves them. # test_models.cpp -> save_resnet_outputs -> file_path = resnet_output_dir + "/sample_" + std::to_string(sample_idx) + "_" + output_name; # This means filenames are like "sample_0_conv1_output.pt" directly in "../test/output/resnet/" correct_cpp_tensor_filename = f"sample_{sample_idx}_{cpp_output_filename}" cpp_tensor_path = os.path.join(cpp_resnet_dir, correct_cpp_tensor_filename) # <<< START ADDED DEBUG PRINTS >>> print(f"DEBUG RN_CMP: Attempting to load C++ tensor for '{output_key}' (sample {sample_idx}) from: {cpp_tensor_path}") # <<< END ADDED DEBUG PRINTS >>> try: cpp_tensor = self.load_cpp_tensor(cpp_tensor_path, self.device) # <<< START ADDED DEBUG PRINTS >>> loaded_status = "None" if cpp_tensor is not None: loaded_status = f"Tensor with shape {cpp_tensor.shape}, dtype {cpp_tensor.dtype}, device {cpp_tensor.device}" print(f"DEBUG RN_CMP: Loaded C++ tensor for '{output_key}' (sample {sample_idx}): {loaded_status}") # <<< END ADDED DEBUG PRINTS >>> if cpp_tensor is None: print(f"Warning: C++ tensor {cpp_output_filename} for sample {sample_idx} ('{output_key}') is None or loading failed. Skipping comparison for this output.") # _compare_tensor_data will be called with cpp_tensor=None, which handles NaN population # Fall through to _compare_tensor_data to record NaNs # continue # This would skip the _compare_tensor_data call entirely # Get the corresponding Python tensor python_tensor = None python_output_save_path = os.path.join(python_resnet_save_dir, f"sample_{sample_idx}", cpp_output_filename) # Save with same name as C++ for consistency # Map the 'output_key' from config to the key used in 'python_outputs' dictionary # This requires knowing how 'outputs_to_compare' keys map to Python model output dict keys. # Example: 'Conv1' maps to 'conv1_output', 'Features' to 'features' (which might be 'layer4'), etc. py_dict_key = None if output_key == 'Conv1': py_dict_key = 'conv1' # Python ResNet outputs conv1 directly elif output_key == 'Debug ResNet Conv1->BN1 Input': print(f"Warning: Python ResNet does not support intermediate debug layers. Skipping {output_key}.") continue elif output_key == 'BN1': print(f"Warning: Python ResNet does not support intermediate BN1 output. Skipping {output_key}.") continue elif output_key == 'BN1 Centered X': print(f"Warning: Python ResNet does not support intermediate BN1 layers. Skipping {output_key}.") continue elif output_key == 'BN1 Var+Eps': print(f"Warning: Python ResNet does not support intermediate BN1 layers. Skipping {output_key}.") continue elif output_key == 'BN1 InvStd': print(f"Warning: Python ResNet does not support intermediate BN1 layers. Skipping {output_key}.") continue elif output_key == 'BN1 Normalized X': print(f"Warning: Python ResNet does not support intermediate BN1 layers. Skipping {output_key}.") continue elif output_key == 'ReLU1': print(f"Warning: Python ResNet does not support intermediate ReLU1 output. Skipping {output_key}.") continue elif output_key == 'MaxPool': # MaxPool is applied *after* 'conv1' (conv1+bn1+relu) block in Python ResNet. # However, the Python ResNet forward doesn't have a separate 'maxpool' output key. # The output of layer1 is *after* maxpool. # C++ saves maxpool_output.pt *before* layer1. # This means we need to save python_outputs['conv1'] (after conv1,bn1,relu) then apply maxpool to it manually for comparison. # OR, recognize that C++ output for maxpool is input to layer1. # For now, this is tricky. Let's see if layer1 input in C++ matches python maxpool output. # The Python output named 'layer1' is after the nn.Sequential that IS layer1. # The input to C++ layer1 is the output of C++ maxpool. # The input to Python model.layer1 is the output of model.maxpool(model.relu(model.bn1(model.conv1(x)))). # So, Python's 'conv1' output, when passed through an nn.MaxPool2d, should match C++ 'maxpool_output.pt'. print(f"Warning: Direct Python equivalent for C++ 'MaxPool' output is complex. Requires manual maxpool application to Python's 'conv1' output. Skipping {output_key} for now.") continue # Skip this key for now elif output_key == 'Layer1': py_dict_key = 'layer1' elif output_key == 'Layer2': py_dict_key = 'layer2' elif output_key == 'Layer3': py_dict_key = 'layer3' elif output_key == 'Layer4': py_dict_key = 'layer4' elif output_key == 'Features': py_dict_key = 'layer4' # 'Features' is an alias for 'layer4' elif output_key == 'Layer1.0 Shortcut': # Shortcut outputs are not available from the Python ResNet forward method. print(f"Warning: Shortcut output '{output_key}' cannot be directly fetched from Python ResNet. Skipping.") continue else: print(f"Warning: Unknown output_key '{output_key}' in ResNet config for Python tensor mapping. Skipping.") continue if py_dict_key and py_dict_key in python_outputs: python_tensor = python_outputs[py_dict_key] else: # DEBUG: Print info if key is not found print(f"DEBUG RN_CMP: py_dict_key '{py_dict_key}' not found in python_outputs (keys: {list(python_outputs.keys())}) for output_key '{output_key}', sample {sample_idx}") if python_tensor is None: print(f"Warning: Python tensor for {output_key} is None for sample {sample_idx}. Skipping.") continue # Save the Python tensor (always, for record-keeping) os.makedirs(os.path.dirname(python_output_save_path), exist_ok=True) torch.save(python_tensor.cpu(), python_output_save_path) # print(f"Saved Python tensor for {output_key} (sample {sample_idx}) to {python_output_save_path}") # Perform comparison self._compare_tensor_data(python_tensor.to(self.device) if python_tensor is not None else None, cpp_tensor, output_key, sample_idx, current_errors) # current_errors is populated in place # The line above was changed to handle python_tensor being None before .to(self.device) # current_errors is populated by _compare_tensor_data directly. # self.all_comparison_stats is updated after this inner loop completes for the sample. except FileNotFoundError: print(f"Warning: C++ output file not found: {cpp_tensor_path}. Skipping for sample {sample_idx}, output {output_key}.") # Populate NaNs for this missing C++ file self._compare_tensor_data(None, None, output_key, sample_idx, current_errors) except Exception as e: print(f"Error comparing {output_key} for sample {sample_idx}: {e}") import traceback traceback.print_exc() # Populate NaNs on error self._compare_tensor_data(None, None, output_key, sample_idx, current_errors) # After processing all output_keys for this sample, store the collected current_errors if current_errors: # If any comparisons were attempted (even if they resulted in NaNs) self.all_comparison_stats[f"ResNet_Sample_{sample_idx}"] = current_errors # processed_samples += 1 # This variable is no longer used as loop is range-based print("--- ResNet Output Comparison Complete ---") def generate_html_report(self): print("\nGenerating HTML report...") report_path = os.path.join(self.comparison_dir, "report.html") # Prepare data for the report: group by model and comparison type report_data = { } for sample_key, comparisons in self.all_comparison_stats.items(): # sample_key examples: "Clf_Train_Sample_0", "Clf_Test_Sample_0", "BBReg_Sample_0" parts = sample_key.split("_") model_prefix = parts[0] # Clf, BBReg, ResNet sample_type_str = "" sample_idx = -1 if model_prefix == "Clf": sample_type_str = parts[1] # Train or Test sample_idx = int(parts[-1]) model_name_key = f"Classifier {sample_type_str}" elif model_prefix == "BBReg": sample_idx = int(parts[-1]) model_name_key = "BB Regressor" elif model_prefix == "ResNet": # Added this case sample_idx = int(parts[-1]) model_name_key = "ResNet" else: print(f"WARNING: Unknown sample key format in all_comparison_stats: {sample_key}") continue for comparison_name, stats in comparisons.items(): # comparison_name examples: "Classifier Features Train", "BBReg PyIoUFeat0 vs CppIoUFeat0" # Unpack all 11 metrics now mae, max_err, diff_arr, mean_py_val, std_abs_err, \ l2_py, l2_cpp, l2_diff, cos_sim, pearson, mre = stats full_comparison_key = f"{model_name_key} - {comparison_name}" if full_comparison_key not in report_data: report_data[full_comparison_key] = { "samples": {}, "all_maes": [], "all_max_errs": [], "all_mean_py_vals": [], "all_std_abs_errs": [], # Renamed from all_std_errs "all_l2_py_vals": [], "all_l2_cpp_vals": [], "all_l2_diff_vals": [], "all_cos_sim_vals": [], "all_pearson_vals": [], "all_mre_vals": [] } relative_plot_path = None # Initialize relative_plot_path plot_filename = f"{model_name_key.replace(' ', '_')}_{comparison_name.replace(' ', '_')}_{sample_idx}.png" plot_abs_path = os.path.join(self.comparison_dir, plot_filename) if os.path.exists(plot_abs_path): # relative_plot_path = Path(plot_filename) # Old relative_plot_path = plot_filename # New, already just filename self._generate_single_plot(diff_arr, comparison_name, plot_abs_path, mean_py_val, std_abs_err, mae, max_err) report_data[full_comparison_key]["samples"][sample_idx] = { "mae": mae, "max_err": max_err, "mean_py_val": mean_py_val, "std_abs_err": std_abs_err, # Renamed from std_err "l2_py": l2_py, "l2_cpp": l2_cpp, "l2_diff": l2_diff, "cos_sim": cos_sim, "pearson": pearson, "mre": mre, "plot_path": relative_plot_path # Store relative path for HTML } if not np.isnan(mae): report_data[full_comparison_key]["all_maes"].append(mae) if not np.isnan(max_err): report_data[full_comparison_key]["all_max_errs"].append(max_err) if not np.isnan(mean_py_val): report_data[full_comparison_key]["all_mean_py_vals"].append(mean_py_val) if not np.isnan(std_abs_err): report_data[full_comparison_key]["all_std_abs_errs"].append(std_abs_err) if not np.isnan(l2_py): report_data[full_comparison_key]["all_l2_py_vals"].append(l2_py) if not np.isnan(l2_cpp): report_data[full_comparison_key]["all_l2_cpp_vals"].append(l2_cpp) if not np.isnan(l2_diff): report_data[full_comparison_key]["all_l2_diff_vals"].append(l2_diff) if not np.isnan(cos_sim): report_data[full_comparison_key]["all_cos_sim_vals"].append(cos_sim) if not np.isnan(pearson): report_data[full_comparison_key]["all_pearson_vals"].append(pearson) if not np.isnan(mre): report_data[full_comparison_key]["all_mre_vals"].append(mre) # Calculate overall stats for comp_key, data in report_data.items(): data["overall_mae_mean"] = np.mean(data["all_maes"]) if data["all_maes"] else float('nan') data["overall_mae_std"] = np.std(data["all_maes"]) if data["all_maes"] else float('nan') data["overall_max_err_mean"] = np.mean(data["all_max_errs"]) if data["all_max_errs"] else float('nan') data["overall_mean_py_val_mean"] = np.mean(data["all_mean_py_vals"]) if data["all_mean_py_vals"] else float('nan') data["overall_std_abs_err_mean"] = np.mean(data["all_std_abs_errs"]) if data["all_std_abs_errs"] else float('nan') # Renamed data["overall_l2_py_mean"] = np.mean(data["all_l2_py_vals"]) if data["all_l2_py_vals"] else float('nan') data["overall_l2_cpp_mean"] = np.mean(data["all_l2_cpp_vals"]) if data["all_l2_cpp_vals"] else float('nan') data["overall_l2_diff_mean"] = np.mean(data["all_l2_diff_vals"]) if data["all_l2_diff_vals"] else float('nan') data["overall_cos_sim_mean"] = np.mean(data["all_cos_sim_vals"]) if data["all_cos_sim_vals"] else float('nan') data["overall_pearson_mean"] = np.mean(data["all_pearson_vals"]) if data["all_pearson_vals"] else float('nan') data["overall_mre_mean"] = np.mean(data["all_mre_vals"]) if data["all_mre_vals"] else float('nan') # HTML Generation html_content = """
Number of samples per model component: {self.num_samples}
mean(abs(py - cpp))
). The "Mean MAE" in the summary table is the average of these MAEs over all samples for a given comparison.mean(max(abs(py - cpp)))
over samples).mean(mean(py_tensor_sample_N))
).abs(py - cpp)
) for each sample. The "Err Std" in plot titles is this value for that specific sample.py - cpp
) over all samples.dot(py, cpp) / (norm(py) * norm(cpp))
.mean(abs(py - cpp) / (abs(py) + epsilon))
. Epsilon is a small value to prevent division by zero.Comparison Key | Mean MAE | Std MAE | Mean Max Error | Mean Py Val | Mean Std Abs Err | Mean L2 Py | Mean L2 Cpp | Mean L2 Diff | Mean Cosine Sim | Mean Pearson Corr | Mean MRE |
---|---|---|---|---|---|---|---|---|---|---|---|
{comp_key} | {f"{data['overall_mae_mean']:.4e}" if not np.isnan(data['overall_mae_mean']) else 'N/A'} | {f"{data['overall_mae_std']:.4e}" if not np.isnan(data['overall_mae_std']) else 'N/A'} | {f"{data['overall_max_err_mean']:.4e}" if not np.isnan(data['overall_max_err_mean']) else 'N/A'} | {f"{data['overall_mean_py_val_mean']:.4e}" if not np.isnan(data['overall_mean_py_val_mean']) else 'N/A'} | {f"{data['overall_std_abs_err_mean']:.4e}" if not np.isnan(data['overall_std_abs_err_mean']) else 'N/A'} | {f"{data['overall_l2_py_mean']:.4e}" if not np.isnan(data['overall_l2_py_mean']) else 'N/A'} | {f"{data['overall_l2_cpp_mean']:.4e}" if not np.isnan(data['overall_l2_cpp_mean']) else 'N/A'} | {f"{data['overall_l2_diff_mean']:.4e}" if not np.isnan(data['overall_l2_diff_mean']) else 'N/A'} | {f"{data['overall_cos_sim_mean']:.4f}" if not np.isnan(data['overall_cos_sim_mean']) else 'N/A'} | {f"{data['overall_pearson_mean']:.4f}" if not np.isnan(data['overall_pearson_mean']) else 'N/A'} | {f"{data['overall_mre_mean']:.4e}" if not np.isnan(data['overall_mre_mean']) else 'N/A'} |
Overall Mean MAE: {f'{data["overall_mae_mean"]:.4e}' if not np.isnan(data['overall_mae_mean']) else 'N/A'}
""" html_content += "Sample Index | MAE | Max Error | Mean Py Val | Std Abs Err | L2 Py | L2 Cpp | L2 Diff | Cosine Sim | Pearson Corr | MRE | Error Distribution Plot |
---|---|---|---|---|---|---|---|---|---|---|---|
{sample_idx} | {f"{sample_data['mae']:.4e}" if not np.isnan(sample_data['mae']) else 'N/A'} | {f"{sample_data['max_err']:.4e}" if not np.isnan(sample_data['max_err']) else 'N/A'} | {f"{sample_data['mean_py_val']:.4e}" if not np.isnan(sample_data['mean_py_val']) else 'N/A'} | {f"{sample_data['std_abs_err']:.4e}" if not np.isnan(sample_data['std_abs_err']) else 'N/A'} | {f"{sample_data['l2_py']:.4e}" if not np.isnan(sample_data['l2_py']) else 'N/A'} | {f"{sample_data['l2_cpp']:.4e}" if not np.isnan(sample_data['l2_cpp']) else 'N/A'} | {f"{sample_data['l2_diff']:.4e}" if not np.isnan(sample_data['l2_diff']) else 'N/A'} | {f"{sample_data['cos_sim']:.4f}" if not np.isnan(sample_data['cos_sim']) else 'N/A'} | {f"{sample_data['pearson']:.4f}" if not np.isnan(sample_data['pearson']) else 'N/A'} | {f"{sample_data['mre']:.4e}" if not np.isnan(sample_data['mre']) else 'N/A'} | {img_tag} |