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 def count_total_connections(nodes): """Count the total number of connections in a topology.""" seen_pairs = set() total = 0 for node in nodes: for conn in node.get('connections', []): peer = conn['peer_device'] pair = tuple(sorted([node['device_name'], peer])) if pair not in seen_pairs: seen_pairs.add(pair) total += 1 return total def get_layer_priority(layer): """Define layer priority for connection filtering.""" layer_priority = { 'Core': 0, 'Distribution': 1, 'Access': 2, 'Endpoint': 3 } return layer_priority.get(layer, float('inf')) def get_peer_layer(peer_device, all_nodes): """Obtenir la couche du peer device""" for node in all_nodes: if node['device_name'] == peer_device: return node.get('network_layer', 'Unknown') return 'Unknown' def should_count_connection(node_layer, peer_layer, node_priority): """Determine whether a connection should be counted according to the specified rules.""" peer_priority = get_layer_priority(peer_layer) # Rule 1: connections within the same layer if node_layer == peer_layer: return True # Rule 2: connections to the previous layer (N+1: core=0, distribution=1, etc) # So if the peer has lower priority (higher layer) if peer_priority < node_priority: return True return False try: from rapidfuzz import fuzz except ImportError: import difflib class fuzz: @staticmethod def ratio(a, b): return difflib.SequenceMatcher(None, a, b).ratio() * 100 try: from scipy.optimize import linear_sum_assignment HAS_SCIPY = True except ImportError: HAS_SCIPY = False class C: HEADER = '\033[95m' OKBLUE = '\033[94m' OKCYAN = '\033[96m' OKGREEN = '\033[92m' WARNING = '\033[93m' FAIL = '\033[91m' ENDC = '\033[0m' BOLD = '\033[1m' LOG_FILE = None def custom_print(message: str = "", end: str = "\n"): global LOG_FILE full_message = message + end if LOG_FILE: with open(LOG_FILE, 'a', encoding='utf-8') as f: f.write(full_message) print(full_message, end='') 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 build_edges(input_json: Dict) -> Dict: """ Function that completes the JSON by creating reverse connections for each existing connection. """ output_json = json.loads(json.dumps(input_json)) devices = output_json["result"]["network_topology"]["devices"] device_map = {device["device_name"]: device for device in devices} # Iterate over all devices for device in devices: current_device_name = device.get("device_name") used_interfaces = device.get("device_interfaces", {}).get("used_interfaces", []) # For each existing connection, create the reverse connection for conn in used_interfaces: peer_connection = conn.get("peer_connection") if not peer_connection: continue peer_device_name = peer_connection.get("peer_device") if not peer_device_name: continue peer_device = device_map.get(peer_device_name) if not peer_device: continue # Check whether the reverse connection already exists peer_used_interfaces = peer_device.get("device_interfaces", {}).get("used_interfaces", []) reverse_exists = False for peer_conn in peer_used_interfaces: if peer_conn.get("peer_connection", {}).get("peer_device") == current_device_name: reverse_exists = True break # If the reverse connection does not exist, create it if not reverse_exists: interface_name = conn.get("interface_name") interface_type = conn.get("interface_type") peer_interface = conn.get("peer_connection", {}).get("peer_interface") peer_interface_type = conn.get("peer_connection", {}).get("peer_interface_type") new_reverse_conn = { "interface_name": peer_interface, "interface_type": peer_interface_type, "peer_connection": { "peer_device": current_device_name, "peer_interface": interface_name, "peer_interface_type": interface_type } } peer_device_interfaces = peer_device.get("device_interfaces", {}) # if "used_interfaces" not in peer_device_interfaces: # peer_device_interfaces["used_interfaces"] = [] peer_device_interfaces["used_interfaces"].append(new_reverse_conn) custom_print(f"{C.OKGREEN}Reverse connection added: {peer_device_name}({peer_interface}) -> {current_device_name}({interface_name}){C.ENDC}") return output_json def load_nodes(data: Dict) -> List[Dict]: nodes = [] devices = data["result"]["network_topology"].get("devices", []) for device in devices: # Extraire les connexions depuis les interfaces utilisées connections = [] interfaces = device.get("device_interfaces", {}) used_interfaces = interfaces.get("used_interfaces", []) for interface in used_interfaces: peer = interface.get("peer_connection", {}) if peer is None: peer = {} if peer.get("peer_device"): connections.append({ "peer_device": peer["peer_device"], "interface_name": interface.get("interface_name"), "peer_interface": peer.get("peer_interface"), "interface_type": _normalize_if_type(interface.get("interface_name", "")) }) nodes.append({ "device_name": device.get("device_name"), "network_layer": str(device.get("network_layer", "access")).lower(), "device_type": device.get("device_type", "router"), "location_site": str(device.get("location_site", "default")).lower(), "connections": connections }) return nodes def _normalize_name_nodes(name: str) -> str: """Standardize node names for comparison (based on gen_F1_nodes.py logic).""" return str(name or "").replace('-', '').replace('_', '').upper() def rf_score_nodes(a: str, b: str) -> float: """Calculate the similarity score between two node names.""" a_norm = _normalize_name_nodes(a) b_norm = _normalize_name_nodes(b) if hasattr(fuzz, 'token_set_ratio'): return float(fuzz.token_set_ratio(a_norm, b_norm)) else: return fuzz.ratio(a_norm, b_norm) def build_similarity_matrix_nodes(ref_names: List[str], gen_names: List[str]) -> np.ndarray: """Construit la matrice de similarité (logique gen_F1_nodes.py).""" M = np.zeros((len(ref_names), len(gen_names)), dtype=float) for i, a in enumerate(ref_names): for j, b in enumerate(gen_names): if a and b: M[i, j] = rf_score_nodes(a, b) return M def hungarian_match_nodes(sim: np.ndarray): """Applique l'algorithme Hongrois (logique gen_F1_nodes.py).""" if not HAS_SCIPY: return [] m, n = sim.shape cost = 100.0 - sim row_ind, col_ind = linear_sum_assignment(cost) pairs = [(r, c, float(sim[r, c])) for r, c in zip(row_ind, col_ind) if r < m and c < n] return pairs def normalize_gen_to_ref_topology(ref_nodes, gen_nodes, min_sim: float = 1.0): """Normalize GEN sites and types to match REF using the Hungarian algorithm.""" if not gen_nodes or not ref_nodes: return gen_nodes # 1. Extraire les sites et types uniques de REF (ceux qu'on veut conserver) ref_sites = list(set(n.get('location_site') for n in ref_nodes)) ref_types = list(set(n.get('device_type') for n in ref_nodes)) # 2. Extraire les sites et types uniques de GEN gen_sites = list(set(n.get('location_site') for n in gen_nodes)) gen_types = list(set(n.get('device_type') for n in gen_nodes)) site_mapping = {} type_mapping = {} # 3. Normalisation des sites : Approche "Plus Proche Voisin" (Many-to-One) if ref_sites and gen_sites: # Recompute or reuse the similarity matrix site_sim = build_similarity_matrix_nodes(ref_sites, gen_sites) for j, gen_site in enumerate(gen_sites): # Trouver l'index du site REF qui a le score maximal avec ce gen_site best_ref_idx = max(range(len(ref_sites)), key=lambda i: site_sim[i][j]) best_score = site_sim[best_ref_idx][j] # On applique le mapping si le score passe le seuil minimum (ex: 0.5) if best_score >= min_sim: site_mapping[gen_site] = ref_sites[best_ref_idx] print(f" GEN site: {gen_site} → REF site (Max Sim): {ref_sites[best_ref_idx]} (score: {best_score:.2f})") else: print(f" Warning: GEN site '{gen_site}' has no reliable match (Best score: {best_score:.2f})") # 4. Apply site normalization to GEN nodes print(f" Application de la normalisation des sites:") for node in gen_nodes: original_site = node.get('location_site') if original_site in site_mapping: new_site = site_mapping[original_site] node['location_site'] = new_site print(f" Node {node['device_name']}: site {original_site} → {new_site}") # 5. Pre-normalization: if a type contains "server", normalize it directly to "Server" print(f" Pre-normalization of types containing 'server':") for node in gen_nodes: original_type = node.get('device_type', '') if original_type and any(keyword in original_type.lower() for keyword in ['server', 'videodelivery', 'haproxy']): node['device_type'] = 'Server' print(f" Node {node['device_name']}: type {original_type} → Server (contient 'server')") # 6. Normalisation des types par layer avec algorithme hongrois if ref_types and gen_types: print(f" Normalisation des types par layer:") # Group nodes by layer ref_nodes_by_layer = {} gen_nodes_by_layer = {} for node in ref_nodes: layer = node.get('network_layer') if layer not in ref_nodes_by_layer: ref_nodes_by_layer[layer] = [] ref_nodes_by_layer[layer].append(node) for node in gen_nodes: layer = node.get('network_layer') if layer not in gen_nodes_by_layer: gen_nodes_by_layer[layer] = [] gen_nodes_by_layer[layer].append(node) # Process each layer separately all_layers = set(ref_nodes_by_layer.keys()) | set(gen_nodes_by_layer.keys()) for layer in sorted(all_layers): ref_layer_nodes = ref_nodes_by_layer.get(layer, []) gen_layer_nodes = gen_nodes_by_layer.get(layer, []) if not ref_layer_nodes or not gen_layer_nodes: continue # Extraire les types uniques pour ce layer ref_layer_types = list(set(n.get('device_type') for n in ref_layer_nodes)) gen_layer_types = list(set(n.get('device_type') for n in gen_layer_nodes)) print(f" Layer {layer}: {len(ref_layer_types)} types REF vs {len(gen_layer_types)} types GEN") # Appliquer l'algorithme hongrois pour ce layer type_sim = build_similarity_matrix_nodes(ref_layer_types, gen_layer_types) type_pairs = hungarian_match_nodes(type_sim) layer_type_mapping = {} for i, j, s in type_pairs: if s >= min_sim: # Utiliser directement le nom REF layer_type_mapping[gen_layer_types[j]] = ref_layer_types[i] print(f" Type GEN: {gen_layer_types[j]} → Type REF: {ref_layer_types[i]} (score: {s:.2f})") # Fusionner avec le mapping global type_mapping.update(layer_type_mapping) # 6. Apply type normalization to GEN nodes print(f" Application de la normalisation des types:") for node in gen_nodes: original_type = node.get('device_type') if original_type in type_mapping: new_type = type_mapping[original_type] node['device_type'] = new_type print(f" Node {node['device_name']}: type {original_type} → {new_type}") print(f"Normalization complete: {len(site_mapping)} sites and {len(type_mapping)} types normalized") return gen_nodes, site_mapping, type_mapping def get_mapping(ref_nodes, gen_nodes, min_sim: float = 1.0): """Hierarchical mapping: layer → site → type with the Hungarian algorithm (adapted from gen_F1_nodes.py logic).""" if not gen_nodes: return {} mapping: Dict[str, str] = {} # Hierarchical grouping: layer → site → type ref_groups: Dict[str, Dict[str, List[str]]] = {} gen_groups: Dict[str, Dict[str, List[str]]] = {} # Group REF nodes by layer → site → type for n in ref_nodes: if n["device_name"] and isinstance(n["device_name"], str): layer = n["network_layer"] site = n.get("location_site") device_type = n.get("device_type") if layer not in ref_groups: ref_groups[layer] = {} if site not in ref_groups[layer]: ref_groups[layer][site] = {} if device_type not in ref_groups[layer][site]: ref_groups[layer][site][device_type] = [] ref_groups[layer][site][device_type].append(n["device_name"]) # Group GEN nodes by layer → site → type for n in gen_nodes: if n["device_name"] and isinstance(n["device_name"], str): layer = n["network_layer"] site = n.get("location_site") device_type = n.get("device_type") if layer not in gen_groups: gen_groups[layer] = {} if site not in gen_groups[layer]: gen_groups[layer][site] = {} if device_type not in gen_groups[layer][site]: gen_groups[layer][site][device_type] = [] gen_groups[layer][site][device_type].append(n["device_name"]) all_keys = set(ref_groups.keys()) | set(gen_groups.keys()) custom_print(f"\n{C.HEADER}===== Hierarchical Mapping: Layer → Site → Type (Hungarian Algorithm) ====={C.ENDC}") for layer in sorted(all_keys, key=lambda x: (x != 'null', x)): ref_layer_dict = ref_groups.get(layer, {}) gen_layer_dict = gen_groups.get(layer, {}) display_layer = layer.upper() if layer != 'null' else layer custom_print(f"\n{C.OKBLUE}--- Layer: {display_layer} ---{C.ENDC}") # Traiter chaque site dans ce layer all_sites = set(ref_layer_dict.keys()) | set(gen_layer_dict.keys()) for site in sorted(all_sites): ref_site_types = ref_layer_dict.get(site, {}) gen_site_types = gen_layer_dict.get(site, {}) if not ref_site_types and not gen_site_types: continue custom_print(f" {C.OKCYAN}Site: {site}{C.ENDC}") # Traiter chaque type dans ce site/layer all_types = set(ref_site_types.keys()) | set(gen_site_types.keys()) for device_type in sorted(all_types): ref_names = ref_site_types.get(device_type, []) gen_names = gen_site_types.get(device_type, []) if not ref_names or not gen_names: if ref_names: custom_print(f" {C.FAIL}Local FN {device_type}: {', '.join(ref_names)}{C.ENDC}") if gen_names: custom_print(f" {C.FAIL}Local FP {device_type}: {', '.join(gen_names)}{C.ENDC}") continue # Appliquer l'algorithme hongrois pour ce groupe spécifique sim = build_similarity_matrix_nodes(ref_names, gen_names) pairs = hungarian_match_nodes(sim) group_matches = {} matches = [] for i, j, s in pairs: r_name = ref_names[i] g_name = gen_names[j] if s >= min_sim and r_name and g_name: group_matches[r_name] = g_name mapping[r_name] = g_name matches.append((r_name, g_name, s)) if matches: custom_print(f" {C.OKGREEN}TP {device_type}: {len(matches)} match(s){C.ENDC}") for r_name, g_name, score in matches: custom_print(f" - REF: {r_name:<20} <--> GEN: {g_name:<20} (Score: {score:.2f})") else: custom_print(f" {C.WARNING}No matches for {device_type} (threshold: {min_sim:.1f}%){C.ENDC}") # Afficher FN/FP locaux pour ce groupe fn_nodes = [name for name in ref_names if name not in group_matches] fp_nodes = [name for name in gen_names if name not in group_matches.values()] if fn_nodes: custom_print(f" {C.FAIL}FN Locaux {device_type} ({len(fn_nodes)}): {', '.join(fn_nodes[:10])}{'...' if len(fn_nodes) > 10 else ''}{C.ENDC}") if fp_nodes: custom_print(f" {C.FAIL}FP Locaux {device_type} ({len(fp_nodes)}): {', '.join(fp_nodes[:10])}{'...' if len(fp_nodes) > 10 else ''}{C.ENDC}") custom_print(f"\n{C.OKGREEN}Mapping final : {len(mapping)} correspondances{C.ENDC}") return mapping def get_correct_edges(nodes_list: List[Dict]): """ Fonction pour calculer les FP en vérifiant les connexions réciproques. Retourne les edges valides (connexions bidirectionnelles) et le nombre de FP. """ edges = set() extra_fp = 0 node_map = {n['device_name']: n for n in nodes_list} # Dictionary to track connections and their reciprocal state connections_seen = {} # (u, v, u_if, v_if) -> {"has_reverse": bool, "type": str} # Debug: stocker les détails des FP fp_links = [] reciprocal_links = [] bidirectional_pairs = set() # Pour éviter de compter deux fois la même paire # First pass: identify all connections for node in nodes_list: u = node['device_name'] for c in node.get('connections', []): v, u_if, v_if = c['peer_device'], c['interface_name'], c['peer_interface'] u_type = c.get('interface_type') # Create the unique connection key conn_key = (u, v, u_if, v_if) # Check whether the peer exists peer = node_map.get(v) if not peer: continue # Skip les connexions vers des devices inexistants # Initialize the connection entry if conn_key not in connections_seen: connections_seen[conn_key] = {"has_reverse": False, "type": u_type} # Second pass: verify reciprocal connections for conn_key, conn_info in connections_seen.items(): u, v, u_if, v_if = conn_key u_type = conn_info["type"] # Create the reverse connection key reverse_key = (v, u, v_if, u_if) # Check whether the reverse connection exists if reverse_key in connections_seen: # Both directions exist reverse_type = connections_seen[reverse_key]["type"] # Mark both connections as having a reciprocal connections_seen[conn_key]["has_reverse"] = True connections_seen[reverse_key]["has_reverse"] = True # Check interface type compatibility if u_type != reverse_type: # Incompatible types → FP extra_fp += 1 fp_links.append(f"FP_TYPE: {u}({u_type}) <-> {v}({reverse_type}) via {u_if}/{v_if}") else: # Compatible types and reciprocal connection → valid edge peer = node_map.get(v) u_device_type = node_map.get(u, {}).get('device_type') v_device_type = peer.get('device_type') # Create a unique signature for the bidirectional pair pair_signature = tuple(sorted([u, v])) # (R1, SW1) ou (SW1, R1) -> (R1, SW1) # Add only if the pair has not already been counted if pair_signature not in bidirectional_pairs: bidirectional_pairs.add(pair_signature) # Signature unique de l'edge edge_id = tuple(sorted([(u, u_device_type, u_if, u_type), (v, v_device_type, v_if, u_type)])) edges.add(edge_id) reciprocal_links.append(f"VALID: {u} <-> {v} via {u_if}/{v_if} ({u_type})") # Identify and remove ALL non-bidirectional connections non_bidirectional_connections = [] connections_to_remove = [] # Store connections to remove counted_pairs = set() # Avoid counting the same device pair twice for conn_key, conn_info in connections_seen.items(): if not conn_info["has_reverse"]: # If no reciprocal connection = non-bidirectional u, v, u_if, v_if = conn_key u_type = conn_info["type"] # Create a unique signature for the device pair (order-independent) pair_signature = tuple(sorted([u, v])) # Count each pair as 1 FP (not each direction) if pair_signature not in counted_pairs: extra_fp += 1 counted_pairs.add(pair_signature) fp_links.append(f"FP_NON_BIDIRECTIONAL: {u} <-> {v} (non-reciprocal connection)") non_bidirectional_connections.append(f"{u} -> {v}") connections_to_remove.append((u, v, u_if, v_if)) # Physically remove all non-bidirectional connections if connections_to_remove: custom_print(f"{C.FAIL}PHYSICAL REMOVAL of non-bidirectional connections: {len(connections_to_remove)}{C.ENDC}") for u, v, u_if, v_if in connections_to_remove: u_node = node_map.get(u) if u_node: before_count = len(u_node.get('connections', [])) # Supprimer la connexion u -> v u_node['connections'] = [c for c in u_node.get('connections', []) if not (c['peer_device'] == v and c['interface_name'] == u_if and c['peer_interface'] == v_if)] after_count = len(u_node.get('connections', [])) removed = before_count - after_count custom_print(f" ❌ Removed {u} -> {v} via {u_if}/{v_if} ({removed} connection(s))") # Afficher les résultats if reciprocal_links: custom_print(f"{C.OKGREEN}Valid reciprocal connections: {len(reciprocal_links)}{C.ENDC}") for link in reciprocal_links[:3]: custom_print(f" ✓ {link}", C.OKGREEN) if len(reciprocal_links) > 3: custom_print(f" ... and {len(reciprocal_links) - 3} more", C.OKGREEN) if non_bidirectional_connections: custom_print(f"{C.WARNING}Non-bidirectional connections: {len(non_bidirectional_connections)}{C.ENDC}") for link in non_bidirectional_connections[:3]: custom_print(f" - {link}", C.WARNING) if len(non_bidirectional_connections) > 3: custom_print(f" ... and {len(non_bidirectional_connections) - 3} more", C.WARNING) if fp_links: custom_print(f"{C.FAIL}Total FP: {len(fp_links)}{C.ENDC}") for link in fp_links[:5]: custom_print(f" - {link}", C.FAIL) if len(fp_links) > 5: custom_print(f" ... and {len(fp_links) - 5} more", C.FAIL) return edges, extra_fp def get_neighbors(node_name: str, nodes_list: List[Dict]) -> Set[str]: """Extract the set of neighbors for a given node.""" for node in nodes_list: if node['device_name'] == node_name: return {conn['peer_device'] for conn in node.get('connections', [])} return set() def get_neighbors_with_type(node_name: str, nodes_list: List[Dict]) -> Dict[str, Set[str]]: """Extract neighbors of a given node grouped by device type.""" neighbors_by_type = {} for node in nodes_list: if node['device_name'] == node_name: for conn in node.get('connections', []): peer_name = conn['peer_device'] peer_node = next((n for n in nodes_list if n['device_name'] == peer_name), None) if peer_node: device_type = peer_node.get('device_type') if device_type not in neighbors_by_type: neighbors_by_type[device_type] = set() neighbors_by_type[device_type].add(peer_name) return neighbors_by_type # ==== Support functions for structural remapping ==== def get_all_normalized_types(nodes): """Extract ALL normalized node and neighbor types.""" types_set = set() # Add node types themselves for node in nodes: types_set.add(node['device_type']) # Ajouter les types des voisins for node in nodes: for conn in node.get('connections', []): peer_name = conn['peer_device'] # Lookup the neighbor node for peer_node in nodes: if peer_node['device_name'] == peer_name: types_set.add(peer_node['device_type']) break return sorted(list(types_set)) def create_neighbor_vector(node, all_nodes, type_order): """Create fixed-length neighborhood vector with zero padding""" vector = [0] * len(type_order) # Initialized to 0 everywhere for conn in node.get('connections', []): peer_name = conn['peer_device'] # Chercher le nœud voisin for peer_node in all_nodes: if peer_node['device_name'] == peer_name: peer_type = peer_node['device_type'] if peer_type in type_order: idx = type_order.index(peer_type) vector[idx] += 1 break return vector def manhattan_distance(vec1, vec2): """Calculer la distance Manhattan entre deux vecteurs""" return sum(abs(a - b) for a, b in zip(vec1, vec2)) def group_nodes_by_criteria(nodes): """Group nodes by (site, layer, type).""" groups = {} for node in nodes: key = (node['location_site'], node['network_layer'], node['device_type']) if key not in groups: groups[key] = [] groups[key].append(node) return groups def find_best_structural_matches(ref_group, gen_group, ref_vectors, gen_vectors, all_types): """Trouver les meilleurs matchs structurels en minimisant la distance Manhattan""" if not ref_group or not gen_group: return {} # Construire la matrice de distances n_ref = len(ref_group) n_gen = len(gen_group) # Si un groupe est beaucoup plus grand, utiliser une approche simple if n_ref > n_gen * 2 or n_gen > n_ref * 2: return _simple_greedy_matching(ref_group, gen_group, ref_vectors, gen_vectors) # Square matrix for the Hungarian algorithm size = max(n_ref, n_gen) cost_matrix = np.full((size, size), 999999.0) # Very large value to penalize dummy nodes # Fill the matrix with the real distances for i, ref_node in enumerate(ref_group): for j, gen_node in enumerate(gen_group): ref_name = ref_node['device_name'] gen_name = gen_node['device_name'] if ref_name in ref_vectors and gen_name in gen_vectors: distance = manhattan_distance(ref_vectors[ref_name], gen_vectors[gen_name]) cost_matrix[i, j] = distance # Appliquer l'algorithme hongrois try: row_indices, col_indices = linear_sum_assignment(cost_matrix) except ValueError: # If the Hungarian algorithm fails, use the greedy approach return _simple_greedy_matching(ref_group, gen_group, ref_vectors, gen_vectors) # Build the mapping mapping = {} used_gen = set() for i, j in zip(row_indices, col_indices): if i < n_ref and j < n_gen: ref_node = ref_group[i] gen_node = gen_group[j] # Check that the distance is not too large (invalid match) distance = cost_matrix[i, j] if distance < 999999.0 and gen_node['device_name'] not in used_gen: mapping[ref_node['device_name']] = gen_node['device_name'] used_gen.add(gen_node['device_name']) return mapping def _simple_greedy_matching(ref_group, gen_group, ref_vectors, gen_vectors): """Simple greedy matching approach for groups with very different sizes.""" mapping = {} used_gen = set() # For each REF node, find the best available GEN match for ref_node in ref_group: ref_name = ref_node['device_name'] best_gen = None best_distance = float('inf') for gen_node in gen_group: gen_name = gen_node['device_name'] if gen_name not in used_gen and ref_name in ref_vectors and gen_name in gen_vectors: distance = manhattan_distance(ref_vectors[ref_name], gen_vectors[gen_name]) if distance < best_distance: best_distance = distance best_gen = gen_name if best_gen is not None and best_distance < float('inf'): mapping[ref_name] = best_gen used_gen.add(best_gen) return mapping def find_node_by_name(name, nodes): """Find a node by its name.""" for node in nodes: if node['device_name'] == name: return node return None def apply_structural_mapping_to_nodes(gen_nodes, structural_mapping): """Apply the new structural mapping to GEN nodes and ALL their peer_device references.""" gen_nodes_remapped = [] for gen_node in gen_nodes: # Check if this GEN node is in the structural mapping new_name = None for ref_name, gen_name in structural_mapping.items(): if gen_node['device_name'] == gen_name: new_name = ref_name break # ALWAYS create a copy and update the connections remapped_node = gen_node.copy() # Update the node name if necessary if new_name: remapped_node['device_name'] = new_name # Update ALL peer_device entries in the connections updated_connections = [] for conn in remapped_node.get('connections', []): updated_conn = conn.copy() # Update the peer_device if it is a GEN node that was remapped peer_updated = False for ref_name, gen_name in structural_mapping.items(): if updated_conn['peer_device'] == gen_name: updated_conn['peer_device'] = ref_name peer_updated = True break updated_connections.append(updated_conn) remapped_node['connections'] = updated_connections gen_nodes_remapped.append(remapped_node) return gen_nodes_remapped def count_matching_connections(ref_node, gen_node): """Count matching connections between two nodes (including interface types).""" ref_connections = set() gen_connections = set() # Extract REF connections with interface type for conn in ref_node.get('connections', []): # Inclure le type d'interface dans l'identifiant de connexion interface_type = conn.get('interface_type') link_id = tuple(sorted([ref_node['device_name'], conn['peer_device'], interface_type])) ref_connections.add(link_id) # Extract GEN connections with interface type for conn in gen_node.get('connections', []): # Inclure le type d'interface dans l'identifiant de connexion interface_type = conn.get('interface_type') link_id = tuple(sorted([gen_node['device_name'], conn['peer_device'], interface_type])) gen_connections.add(link_id) # Count exact matches (peer_device + interface_type) matching = len(ref_connections & gen_connections) return matching def calculate_structural_f1_scores(ref_nodes, gen_nodes_remapped, structural_mapping): """ Calculate TP/FP/FN with structural remapping. Approach: REF (intact) vs GEN (modified to resemble REF). Uses the pair approach (u,v) and (v,u) with a seen boolean. """ # Helper to create unique pairs def get_pair(u, v): return tuple(sorted([u, v])) # Helper to extract peers from a node def get_peers(node): return {conn['peer_device'] for conn in node.get('connections', [])} # Helper to count unique pairs using seen set def count_pairs(pairs, seen_set): count = 0 for u, v in pairs: pair = get_pair(u, v) if pair not in seen_set: seen_set.add(pair) count += 1 return count mapped_ref_names = set(structural_mapping.values()) # 1. TP: common REF ∩ GEN connections tp_structural = 0 seen_tp = set() for gen_original, ref_target in structural_mapping.items(): ref_node = find_node_by_name(ref_target, ref_nodes) gen_node = find_node_by_name(ref_target, gen_nodes_remapped) if ref_node and gen_node: ref_peers = get_peers(ref_node) gen_peers = get_peers(gen_node) common_peers = ref_peers & gen_peers tp_pairs = [(ref_target, peer) for peer in common_peers] tp_structural += count_pairs(tp_pairs, seen_tp) # 2. FP: extra GEN connections fp_structural = 0 seen_fp = set() # FP from unmapped nodes for gen_node in gen_nodes_remapped: if gen_node['device_name'] not in mapped_ref_names: peers = get_peers(gen_node) fp_pairs = [(gen_node['device_name'], peer) for peer in peers] fp_structural += count_pairs(fp_pairs, seen_fp) # FP from mapped nodes with extra connections for gen_original, ref_target in structural_mapping.items(): ref_node = find_node_by_name(ref_target, ref_nodes) gen_node = find_node_by_name(ref_target, gen_nodes_remapped) if ref_node and gen_node: ref_peers = get_peers(ref_node) gen_peers = get_peers(gen_node) extra_peers = gen_peers - ref_peers fp_pairs = [(ref_target, peer) for peer in extra_peers] fp_structural += count_pairs(fp_pairs, seen_fp) # 3. FN: Total REF - TP fn_structural = 0 seen_fn = set() # Count all REF connections all_ref_pairs = [] for ref_node in ref_nodes: # peers = get_peers(ref_node) # all_ref_pairs.extend([(ref_node['device_name'], peer) for peer in peers]) if ref_node['device_name'] in mapped_ref_names: peers = get_peers(ref_node) all_ref_pairs.extend([(ref_node['device_name'], peer) for peer in peers]) total_ref = count_pairs(all_ref_pairs, seen_fn) fn_structural = total_ref - tp_structural return tp_structural, fp_structural, fn_structural def save_cleaned_topology(gen_nodes_remapped, first_mapping, output_path): """ Save the remapped GEN topology in JSON with the same format as the original. Contains only nodes that were originally in GEN and have been remapped with the first mapping (get_mapping). Connections are assumed to have already been cleaned by get_correct_edges(). Args: gen_nodes_remapped: List of GEN nodes after structural remapping first_mapping: REF → GEN mapping (first mapping from get_mapping) output_path: Output JSON file path """ # Retrieve the names of nodes mapped by the first mapping (REF names after normalization) mapped_ref_names = set(first_mapping.keys()) # Filter to keep only the nodes mapped by the first mapping mapped_nodes = [] for node in gen_nodes_remapped: if node['device_name'] in mapped_ref_names: mapped_nodes.append(node) # Count remaining connections (already cleaned) total_connections = 0 for node in mapped_nodes: total_connections += len(node.get('connections', [])) custom_print(f"{C.OKCYAN}Saving topology (connections already cleaned)...{C.ENDC}") custom_print(f"{C.OKGREEN}Remaining connections: {total_connections}{C.ENDC}") # Convertir les nœuds avec connexions nettoyées au format JSON complet cleaned_devices = [] for node in mapped_nodes: device_data = { "device_name": node["device_name"], "device_type": node.get("device_type", "router"), "network_layer": node.get("network_layer", "access"), "location_site": node.get("location_site", "default"), "device_interfaces": { "used_interfaces": [], "unused_interfaces": [] } } # Ajouter les connexions restantes (après nettoyage) for conn in node.get('connections', []): peer_device = conn['peer_device'] peer_interface = conn['peer_interface'] interface_name = conn['interface_name'] interface_type = conn.get('interface_type', 'Ethernet') device_data["device_interfaces"]["used_interfaces"].append({ "interface_name": interface_name, "interface_type": interface_type, "peer_connection": { "peer_device": peer_device, "peer_interface": peer_interface, "peer_interface_type": interface_type } }) cleaned_devices.append(device_data) # Create the JSON structure identical to the original topology_data = { "status": "SUCCESS", "result": { "network_topology": { "devices": cleaned_devices } } } # Sauvegarder en JSON with open(output_path, 'w', encoding='utf-8') as f: json.dump(topology_data, f, indent=2, ensure_ascii=False) custom_print(f"{C.OKGREEN}Topology saved: {output_path}{C.ENDC}") custom_print(f"{C.OKGREEN} - {len(cleaned_devices)} devices{C.ENDC}") custom_print(f"{C.OKGREEN} - {total_connections} connections{C.ENDC}") return len(cleaned_devices)