import json import sys import argparse import re import os from pathlib import Path from typing import List, Dict, Set, Tuple import numpy as np from collections import Counter from f1_edges_functions import* def _normalize_name(name: str) -> str: return re.sub(r'[^a-z0-9]', '', str(name).lower()) def _normalize_if_type(if_name: str) -> str: n = if_name.lower() if 'se' in n: return "Serial" if 'eth' in n: return "Ethernet" return "Ethernet" def custom_print(message: str = "", end: str = "\n"): global LOG_FILE full_message = message + end if LOG_FILE: LOG_FILE.write(full_message) LOG_FILE.flush() sys.stdout.write(full_message) sys.stdout.flush() def main(): parser = argparse.ArgumentParser(description="Calculate edge F1 metrics (TP/FP/FN/P/R/F1) for topology JSON.") parser.add_argument("--gen", type=Path, nargs="+", required=True, help="One or more generated JSON files or folders containing generated JSON files to evaluate.") parser.add_argument("--ref", type=Path, help="Reference JSON file to use (optional). If omitted, the script looks for topo_ref_case_*.json.") parser.add_argument("--min-sim", type=float, default=9.0, help="Minimum similarity threshold (%) for Hungarian matching.") parser.add_argument("--output", type=Path, default=Path("output_f1_scores.txt"), help="Output file for detailed results.") args = parser.parse_args() input_base = args.gen[0] if args.gen[0].is_dir() else args.gen[0].parent results_dir = input_base / "results" results_dir.mkdir(parents=True, exist_ok=True) if not args.output.is_absolute() and args.output.parent == Path('.'): args.output = results_dir / args.output global LOG_FILE try: LOG_FILE = open(args.output, "w", encoding="utf-8") custom_print(f"{C.OKGREEN}✅ Detailed output saved to: {args.output}{C.ENDC}") except Exception as e: custom_print(f"{C.FAIL}❌ ERROR: Unable to open output file {args.output}. Logging will only print to console. Error: {e}{C.ENDC}") LOG_FILE = None summary_table = [] f1_scores = [] current_ref_path = None ref_nodes = [] gen_files = [] for p in args.gen: path = Path(p) if path.is_dir(): for json_file in sorted(path.glob("*.json")): gen_files.append(json_file) elif path.is_file() and path.suffix.lower() == ".json": gen_files.append(path) else: custom_print(f"{C.WARNING} Skipped unsupported path or format: {path}{C.ENDC}") if not gen_files: custom_print(f"{C.FAIL}No JSON files found in the provided paths!{C.ENDC}") sys.exit(1) for gen_file in gen_files: gen_path = gen_file if args.ref: ref_path = args.ref.resolve() if not ref_path.exists(): raise FileNotFoundError(f"Specified reference file not found: {ref_path}") else: match = re.search(r'case_(\d+)', gen_path.name) if not match: custom_print(f"No case_X found in filename {gen_path.name}") continue case_id = match.group(1) ref_path = gen_path.parent / f"topo_ref_case_{case_id}.json" if not ref_path.exists(): custom_print(f"Missing reference for {gen_path.name}") continue if ref_path != current_ref_path: if args.ref: custom_print(f"\n{C.HEADER}*** SPECIFIED REFERENCE: {ref_path.name} ({ref_path.parent}) ***{C.ENDC}") else: custom_print(f"\n{C.HEADER}*** NEW REFERENCE: {ref_path.name} ({ref_path.parent}) ***{C.ENDC}") with open(ref_path) as f: ref_nodes = load_nodes(build_edges(json.load(f))) current_ref_path = ref_path custom_print(f"\n{C.HEADER}{'='*70}") custom_print(f"ANALYSIS: {gen_path.name}") custom_print(f"{'='*70}{C.ENDC}") # Load the generated file with open(gen_path) as f: gen_data = json.load(f) # Check if devices list is empty gen_devices = gen_data.get("result", {}).get("network_topology", {}).get("devices", []) if len(gen_devices) == 0: custom_print(f"{C.WARNING}Empty devices list detected - format with devices: []{C.ENDC}") custom_print(f"{C.FAIL}Generated topology is empty, no mapping processing will be performed{C.ENDC}") # Calculate FN = total connections in the reference fn_final = count_total_connections(ref_nodes) tp_final = 0 fp_final = 0 custom_print(f"{C.OKCYAN}Immediate results:{C.ENDC}") custom_print(f" Final TP: {tp_final}") custom_print(f" Final FP: {fp_final}") custom_print(f" Final FN: {fn_final}") # Compute F1 p = tp_final / (tp_final + fp_final) if (tp_final + fp_final) > 0 else 0 r = tp_final / (tp_final + fn_final) if (tp_final + fn_final) > 0 else 0 f1 = 2 * p * r / (p + r) if (p + r) > 0 else 0 custom_print(f"{C.OKGREEN}FINAL RESULTS (EMPTY DEVICES):{C.ENDC}") custom_print(f" Precision (P): {p:.4f}") custom_print(f" Recall (R): {r:.4f}") custom_print(f" F1 score: {f1:.4f}") f1_scores.append(f1) summary_table.append({ 'test': gen_path.name, 'ref': ref_path.name, 'TP': tp_final, 'FP': fp_final, 'FN': fn_final, 'P': p, 'R': r, 'F1': f1, 'method': "Devices Vide (No Mapping)", 'mapping_count': 0, 'structural_mapping_count': 0, 'ref_count': len(ref_nodes), 'gen_count': 0, 'tp_structural': 0, 'fp_structural': 0, 'fn_structural': fn_final }) continue # If devices is not empty, continue normal processing gen_nodes = load_nodes(build_edges(gen_data)) # A. Normalize GEN topology to REF with Hungarian algorithm gen_nodes, site_mapping, type_mapping = normalize_gen_to_ref_topology(ref_nodes, gen_nodes, args.min_sim) # B. Initial mapping & FN/FP for nodes the model missed or created custom_print(f"{C.OKCYAN}FULL DEBUG - NODE ANALYSIS{C.ENDC}") custom_print(f"Original total REF nodes: {len(ref_nodes)}") custom_print(f"Original total GEN nodes: {len(gen_nodes)}") mapping = get_mapping(ref_nodes, gen_nodes, args.min_sim) mapped_fp_names = set(mapping.values()) mapped_fn_names = set(mapping.keys()) custom_print(f"Mapping found: {len(mapping)} correspondences") custom_print(f"mapped_fn_names (REF): {len(mapped_fn_names)} nodes") custom_print(f"mapped_fp_names (GEN): {len(mapped_fp_names)} nodes") # Debug: list all nodes all_ref_names = {n['device_name'] for n in ref_nodes} all_gen_names = {n['device_name'] for n in gen_nodes} custom_print(f"All REF names: {all_ref_names}") custom_print(f"All GEN names: {all_gen_names}") # Initial FN: connections from REF nodes without a GEN correspondence fn_connections = set() unmapped_ref_nodes = [] for n in ref_nodes: if n['device_name'] not in mapped_fn_names: # REF nodes not mapped = FN unmapped_ref_nodes.append(n['device_name']) node_layer = n.get('network_layer') node_priority = get_layer_priority(node_layer) for c in n.get('connections', []): peer_layer = get_peer_layer(c['peer_device'], ref_nodes) is_endpoint_intra_link = (node_layer.lower() == 'endpoint' and peer_layer.lower() == 'endpoint') # Apply filtering rules if is_endpoint_intra_link or should_count_connection(node_layer, peer_layer, node_priority): link_id = tuple(sorted([n['device_name'], c['peer_device']])) fn_connections.add(link_id) fn_initial_nodes = len(fn_connections) # Initial FP: connections from GEN nodes without a REF correspondence fp_connections = set() unmapped_gen_nodes = [] for n in gen_nodes: if n['device_name'] not in mapped_fp_names: # GEN nodes not mapped = FP unmapped_gen_nodes.append(n['device_name']) node_layer = n.get('network_layer') node_priority = get_layer_priority(node_layer) for c in n.get('connections', []): peer_layer = get_peer_layer(c['peer_device'], gen_nodes) # Apply filtering rules if should_count_connection(node_layer, peer_layer, node_priority): link_id = tuple(sorted([n['device_name'], c['peer_device']])) fp_connections.add(link_id) fp_initial_nodes = len(fp_connections) # Detailed debug custom_print(f"{C.WARNING}Unmapped REF nodes (FN): {unmapped_ref_nodes}{C.ENDC}") custom_print(f"{C.WARNING}Unmapped GEN nodes (FP): {unmapped_gen_nodes}{C.ENDC}") custom_print(f"Number of unmapped REF nodes: {len(unmapped_ref_nodes)}") custom_print(f"Number of unmapped GEN nodes: {len(unmapped_gen_nodes)}") custom_print(f"{C.OKBLUE}Initial FN (missing connections): {fn_initial_nodes}{C.ENDC}") custom_print(f"{C.OKBLUE}Initial FP (extra connections): {fp_initial_nodes}{C.ENDC}") # Display mapped names custom_print(f"{C.OKGREEN}Mapped REF names ({len(mapped_fn_names)}): {', '.join(list(mapped_fn_names)[:5])}{'...' if len(mapped_fn_names) > 5 else ''}{C.ENDC}") custom_print(f"{C.OKGREEN}Mapped GEN names ({len(mapped_fp_names)}): {', '.join(list(mapped_fp_names)[:5])}{'...' if len(mapped_fp_names) > 5 else ''}{C.ENDC}") # B. Full normalization (overlay REF structure onto GEN) ref_info = {n['device_name']: n for n in ref_nodes} # Reverse mapping to get GEN -> REF gen_to_ref = {v: k for k, v in mapping.items()} # Normaliser les types d'interface pour les REF for r in ref_nodes: for c in r.get('connections', []): c['interface_type'] = _normalize_if_type(c['interface_name']) for g in gen_nodes: if g['device_name'] in gen_to_ref: rn = gen_to_ref[g['device_name']] g['device_name'] = rn for c in g.get('connections', []): if c['peer_device'] in gen_to_ref: c['peer_device'] = gen_to_ref[c['peer_device']] c['interface_type'] = _normalize_if_type(c['interface_name']) # C. Summary of correct edges e_ref, _ = get_correct_edges(ref_nodes) e_gen, efp = get_correct_edges(gen_nodes) custom_print(f"{C.HEADER}{'-'*10}{C.ENDC}\n\n{C.OKGREEN}e_ref: ({e_ref}){C.ENDC}\n\n{C.OKBLUE}e_gen: ({e_gen}){C.ENDC}\n\n{C.WARNING}efp: ({efp}){C.ENDC}\n{C.HEADER}{'-'*10}{C.ENDC}") # D. Structural remapping (based on neighborhood vectors) custom_print(f"{C.HEADER}{'='*60}{C.ENDC}") custom_print(f"{C.HEADER}STRUCTURAL REMAPPING AFTER NORMALIZATION{C.ENDC}") custom_print(f"{C.HEADER}{'='*60}{C.ENDC}") # Extract all normalized types from the REF topology all_types = get_all_normalized_types(ref_nodes) custom_print(f"{C.OKBLUE}Types found in REF: {all_types}{C.ENDC}") # Create neighborhood vectors ref_vectors = {} gen_vectors = {} for node in ref_nodes: vector = create_neighbor_vector(node, ref_nodes, all_types) ref_vectors[node['device_name']] = vector for node in gen_nodes: vector = create_neighbor_vector(node, gen_nodes, all_types) gen_vectors[node['device_name']] = vector # Group nodes by (site, layer, type) ref_groups = group_nodes_by_criteria(ref_nodes) gen_groups = group_nodes_by_criteria(gen_nodes) # Structural remapping structural_mapping = {} for group_key in ref_groups: if group_key in gen_groups: ref_group = ref_groups[group_key] gen_group = gen_groups[group_key] site, layer, device_type = group_key custom_print(f"{C.OKCYAN}Group: {site} - {layer} - {device_type}{C.ENDC}") custom_print(f" REF: {[n['device_name'] for n in ref_group]}") custom_print(f" GEN: {[n['device_name'] for n in gen_group]}") # Calculate distances and find the best matches group_mapping = find_best_structural_matches(ref_group, gen_group, ref_vectors, gen_vectors, all_types) for ref_name, gen_name in group_mapping.items(): structural_mapping[ref_name] = gen_name distance = manhattan_distance(ref_vectors[ref_name], gen_vectors[gen_name]) custom_print(f" {C.OKGREEN}→ {ref_name} ↔ {gen_name} (distance: {distance}){C.ENDC}") custom_print(f"{C.OKGREEN}Structural remapping found: {len(structural_mapping)} correspondences{C.ENDC}") # Display new structural mappings custom_print(f"{C.HEADER}{'='*60}{C.ENDC}") custom_print(f"{C.HEADER}NEW STRUCTURAL MAPPINGS{C.ENDC}") custom_print(f"{C.HEADER}{'='*60}{C.ENDC}") # Compare with the initial mapping custom_print(f"{C.OKCYAN}Mapping comparison:{C.ENDC}") custom_print(f"{C.OKBLUE}Initial mapping (based on names):{C.ENDC}") for ref_name, gen_name in mapping.items(): custom_print(f" {ref_name} ↔ {gen_name}") custom_print(f"{C.OKGREEN}New structural mapping (based on connections):{C.ENDC}") for ref_name, gen_name in structural_mapping.items(): # Check if this changed from the initial mapping initial_gen = mapping.get(ref_name, "UNMAPPED") if initial_gen != gen_name: custom_print(f" {C.WARNING}→ {ref_name} ↔ {gen_name} (changed from: {initial_gen}){C.ENDC}") else: custom_print(f" {ref_name} ↔ {gen_name} (unchanged)") # Show changes changes = [] for ref_name, new_gen in structural_mapping.items(): old_gen = mapping.get(ref_name) if old_gen and old_gen != new_gen: changes.append((ref_name, old_gen, new_gen)) if changes: custom_print(f"{C.WARNING}Mapping changes ({len(changes)}):{C.ENDC}") for ref_name, old_gen, new_gen in changes: custom_print(f" {ref_name}: {old_gen} → {new_gen}") else: custom_print(f"{C.OKBLUE}No mapping changes detected{C.ENDC}") # E. STRUCTURAL REMAPPING F1 CALCULATION custom_print(f"{C.HEADER}{'='*60}{C.ENDC}") custom_print(f"{C.HEADER}STRUCTURAL REMAPPING F1 CALCULATION{C.ENDC}") custom_print(f"{C.HEADER}{'='*60}{C.ENDC}") # Apply structural remapping to GEN nodes gen_nodes_remapped = apply_structural_mapping_to_nodes(gen_nodes, structural_mapping) custom_print(f"{C.OKBLUE}GEN nodes after structural remapping: {len(gen_nodes_remapped)}{C.ENDC}") # Display REF and GEN topologies after remapping custom_print(f"{C.HEADER}{'='*60}{C.ENDC}") custom_print(f"{C.HEADER}TOPOLOGIES AFTER STRUCTURAL REMAPPING{C.ENDC}") custom_print(f"{C.HEADER}{'='*60}{C.ENDC}") # Display REF topology (intact) custom_print(f"{C.OKCYAN}REF TOPOLOGY (intact):{C.ENDC}") custom_print(f" Number of nodes: {len(ref_nodes)}") for ref_node in ref_nodes: connections = [conn['peer_device'] for conn in ref_node.get('connections', [])] custom_print(f" {C.OKGREEN}{ref_node['device_name']}{C.ENDC} ({ref_node['device_type']}, {ref_node['network_layer']}) → {connections}") custom_print(f"") # Display GEN topology after first mapping (normalization) custom_print(f"{C.OKCYAN}GEN TOPOLOGY (after first mapping):{C.ENDC}") custom_print(f" Number of nodes: {len(gen_nodes)}") for gen_node in gen_nodes: connections = [conn['peer_device'] for conn in gen_node.get('connections', [])] custom_print(f" {C.OKGREEN}{gen_node['device_name']}{C.ENDC} ({gen_node['device_type']}, {gen_node['network_layer']}) → {connections}") custom_print(f"") # Display GEN topology after remapping custom_print(f"{C.OKCYAN}GEN TOPOLOGY (after structural remapping):{C.ENDC}") custom_print(f" Number of nodes: {len(gen_nodes_remapped)}") # Retrieve the REF target nodes from structural remapping mapped_ref_names = set(structural_mapping.values()) for gen_node in gen_nodes_remapped: connections = [conn['peer_device'] for conn in gen_node.get('connections', [])] # Check if this node was structurally remapped was_remapped = gen_node['device_name'] in mapped_ref_names color = C.WARNING if was_remapped else C.OKGREEN marker = " [REMAP]" if was_remapped else "" custom_print(f" {color}{gen_node['device_name']}{marker}{C.ENDC} ({gen_node['device_type']}, {gen_node['network_layer']}) → {connections}") custom_print(f"") # Calculate structural TP/FP/FN tp_structural, fp_structural, fn_structural = calculate_structural_f1_scores( ref_nodes, gen_nodes_remapped, structural_mapping ) # Show TP/FP/FN details # display_detailed_f1_components(ref_nodes, gen_nodes_remapped, structural_mapping) custom_print(f"{C.OKGREEN}Structural TP: {tp_structural}{C.ENDC}") custom_print(f"{C.WARNING}Structural FP: {fp_structural}{C.ENDC}") custom_print(f"{C.FAIL}Structural FN: {fn_structural}{C.ENDC}") gen_filename = os.path.splitext(os.path.basename(gen_path))[0] output_dir = os.path.join(os.path.dirname(gen_path), "second_mapped") os.makedirs(output_dir, exist_ok=True) output_path = os.path.join(output_dir, f"{gen_filename}.json") # Save mapped_count = save_cleaned_topology( gen_nodes_remapped, mapping, output_path ) custom_print(f"{C.OKCYAN}Remapped GEN topology saved:{C.ENDC}") custom_print(f" File: {output_path}") custom_print(f" Mapped nodes: {mapped_count}") custom_print(f"") # Final F1 calculation with all components tp_final = tp_structural fp_final = fp_structural + efp + fp_initial_nodes fn_final = fn_structural + fn_initial_nodes custom_print(f"{C.OKCYAN}Additional components:{C.ENDC}") custom_print(f" efp (edges FP): {efp}") custom_print(f" fp_initial_nodes: {fp_initial_nodes}") custom_print(f" fn_initial_nodes: {fn_initial_nodes}") custom_print(f"{C.OKCYAN}Final totals:{C.ENDC}") custom_print(f" TP final: {tp_final}") custom_print(f" FP final: {fp_final}") custom_print(f" FN final: {fn_final}") # Compute F1 p = tp_final / (tp_final + fp_final) if (tp_final + fp_final) > 0 else 0 r = tp_final / (tp_final + fn_final) if (tp_final + fn_final) > 0 else 0 f1 = 2 * p * r / (p + r) if (p + r) > 0 else 0 custom_print(f"{C.OKGREEN}FINAL RESULTS WITH STRUCTURAL REMAPPING:{C.ENDC}") custom_print(f" Precision (P): {p:.4f}") custom_print(f" Recall (R): {r:.4f}") custom_print(f" F1 Score: {f1:.4f}") # Add to summary tables method_used = "Structural Remapping" custom_print(f"{C.HEADER}METHOD USED: {method_used}{C.ENDC}") custom_print(f"{C.OKGREEN}Final score: F1={f1:.4f} | P={p:.4f} | R={r:.4f}{C.ENDC}") f1_scores.append(f1) summary_table.append({ 'test': gen_path.name, 'ref': ref_path.name, 'TP': tp_final, 'FP': fp_final, 'FN': fn_final, 'P': p, 'R': r, 'F1': f1, 'method': method_used, 'mapping_count': len(mapping), 'structural_mapping_count': len(structural_mapping), 'ref_count': len(ref_nodes), 'gen_count': len(gen_nodes), 'tp_structural': tp_structural, 'fp_structural': fp_structural, 'fn_structural': fn_structural }) # # --- FINAL TABLE AND STATISTICS --- if summary_table: custom_print(f"\n{C.HEADER}{C.BOLD}===== GLOBAL SUMMARY TABLE ====={C.ENDC}") # console display custom_print(f"{'Test':<25} {'RefDir':<10} {'TP':<5} {'FP':<5} {'FN':<5} {'P':<6} {'R':<6} {'F1':<6} {'Map':<5} {'Str':<5}") custom_print("-" * 95) for s in summary_table: custom_print(f"{s['test']:<25} {s['ref']:<10} {s['TP']:<5} {s['FP']:<5} {s['FN']:<5} " f"{s['P']:.4f} {s['R']:.4f} {s['F1']:.4f} {s['mapping_count']:<5} {s['structural_mapping_count']:<5}") # Compute global statistics m_f1, s_f1 = np.mean(f1_scores), np.std(f1_scores) custom_print("-" * 95) custom_print(f"{'MEAN':<25} {'':<10} {'':<5} {'':<5} {'':<5} {'':<6} {'':<6} {C.OKGREEN}{m_f1:.4f}{C.ENDC}{'':<6} {'':<5} {'':<5}") custom_print(f"{'STD DEV':<25} {'':<10} {'':<5} {'':<5} {'':<5} {'':<6} {'':<6} {C.OKGREEN}{s_f1:.4f}{C.ENDC}{'':<6} {'':<5} {'':<5}") custom_print(f"{'NUMBER OF TESTS':<25} {'':<10} {'':<5} {'':<5} {'':<5} {'':<6} {'':<6} {len(f1_scores):.0f}{'':<6} {'':<5} {'':<5}") custom_print("=" * 95) if gen_files: first_gen_path = gen_files[0] summary_path = results_dir / f"{first_gen_path.parent.name}_F1_structural.txt" if LOG_FILE and args.output.exists(): with open(args.output, "r", encoding="utf-8") as log_file: log_content = log_file.read() with open(summary_path, "w", encoding="utf-8") as f: f.write(log_content) else: with open(summary_path, "w", encoding="utf-8") as f: f.write("Edge analysis - global summary with structural remapping\n") f.write(f"{'Test':<25} {'RefDir':<10} {'TP':<5} {'FP':<5} {'FN':<5} {'P':<6} {'R':<6} {'F1':<6} {'Map':<5} {'Str':<5}\n") f.write("-" * 95 + "\n") for s in summary_table: f.write(f"{s['test']:<25} {s['ref']:<10} {s['TP']:<5} {s['FP']:<5} {s['FN']:<5} " f"{s['P']:.4f} {s['R']:.4f} {s['F1']:.4f} {s['mapping_count']:<5} {s['structural_mapping_count']:<5}\n") if f1_scores: f.write("-" * 95 + "\n") f.write(f"{'MEAN':<25} {'':<10} {'':<5} {'':<5} {'':<5} {'':<6} {'':<6} {m_f1:.4f}{'':<6} {'':<5} {'':<5}\n") f.write(f"{'STD DEV':<25} {'':<10} {'':<5} {'':<5} {'':<5} {'':<6} {'':<6} {s_f1:.4f}{'':<6} {'':<5} {'':<5}\n") f.write(f"{'NUMBER OF TESTS':<25} {'':<10} {'':<5} {'':<5} {'':<5} {'':<6} {'':<6} {len(f1_scores):.0f}{'':<6} {'':<5} {'':<5}\n") custom_print(f"\n{C.OKGREEN} Global summary saved to:{C.ENDC} {summary_path}") if LOG_FILE: LOG_FILE.close() if __name__ == "__main__": main()