import networkx as nx import json import sys import argparse from pathlib import Path from collections import defaultdict from typing import List, Dict, Tuple import numpy as np try: from tabulate import tabulate HAS_TABULATE = True except ImportError: HAS_TABULATE = False def tabulate(data, headers=None, tablefmt="grid", floatfmt=None): """Fallback simple pour tabulate""" if not data: return "" if headers: all_rows = [headers] + data else: all_rows = data col_widths = [] for i in range(len(all_rows[0])): max_width = max(len(str(row[i])) for row in all_rows) col_widths.append(max_width) result = [] separator = "+" + "+".join("-" * (w + 2) for w in col_widths) + "+" result.append(separator) if headers: header_row = "|" + "|".join(f" {str(headers[i]).ljust(col_widths[i])} " for i in range(len(headers))) + "|" result.append(header_row) result.append(separator) for row in data: data_row = "|" + "|".join(f" {str(row[i]).ljust(col_widths[i])} " for i in range(len(row))) + "|" result.append(data_row) result.append(separator) return "\n".join(result) try: from scipy.optimize import linear_sum_assignment HAS_SCIPY = True except ImportError: HAS_SCIPY = False print("SciPy is not installed; falling back to simple mapping.") try: from rapidfuzz import fuzz except ImportError: import difflib class fuzz: @staticmethod def ratio(a, b): return difflib.SequenceMatcher(None, a, b).ratio() * 100 @staticmethod def token_set_ratio(a, b): tokens_a = set(a.lower().split()) tokens_b = set(b.lower().split()) if not tokens_a and not tokens_b: return 100.0 if not tokens_a or not tokens_b: return 0.0 common_tokens = tokens_a.intersection(tokens_b) if not common_tokens: return 0.0 similarity = len(common_tokens) / max(len(tokens_a), len(tokens_b)) return similarity * 100 def load_json_file(file_path): """Charge un fichier JSON et retourne les données""" try: with open(file_path, 'r', encoding='utf-8') as f: return json.load(f) except FileNotFoundError: print(f"Error: File '{file_path}' not found.") sys.exit(1) except json.JSONDecodeError as e: print(f" Error: The file '{file_path}' is not a valid JSON: {e}") sys.exit(1) def create_networkx_graph(data): G = nx.Graph() if "result" in data: devices_data = data["result"]["network_topology"]["devices"] else: devices_data = data["network_topology"]["devices"] for device in devices_data: device_name = device.get("device_name") node_attrs = { "type": device.get("device_type"), "layer": device.get("network_layer"), "site": device.get("location_site"), "zone": device.get("security_zone") } G.add_node(device_name, **node_attrs) connections = [] if "connections" in device: connections = device["connections"] elif "device_interfaces" in device and "used_interfaces" in device["device_interfaces"]: for interface in device["device_interfaces"]["used_interfaces"]: if "peer_connection" in interface: peer_conn = interface["peer_connection"] connections.append({ "interface_name": interface.get("interface_name"), "interface_type": interface.get("interface_type"), "peer_device": peer_conn.get("peer_device"), "peer_interface": peer_conn.get("peer_interface"), "peer_interface_type": peer_conn.get("peer_interface_type") }) for connection in connections: if connection.get("peer_device"): target_device = connection["peer_device"] link_attrs = { "source_interface": connection.get("interface_name"), "target_interface": connection.get("peer_interface"), "type": connection.get("interface_type"), "source_type": connection.get("interface_type"), "target_type": connection.get("interface_type") } G.add_edge(device_name, target_device, **link_attrs) return G def analyze_resilience(G, client, server_primary, server_secondary, nodes_to_exclude, edges_to_exclude): """Analyze the resilience of the graph according to the specified methodology Returns: Dict: Dictionary containing the analysis results """ V_CS = 0 V_CC = 0 V_CS_Link = 0 V_CC_Link = 0 try: shortest_path = nx.shortest_path(G, source=client, target=server_primary) print(f" Shortest path ({client} -> {server_primary}):\n {shortest_path}") except nx.NetworkXNoPath: shortest_path = [] print(f" No path found between {client} et {server_primary}.") print("All resilience metrics have been reset to 0.") results = { 'V_CS': 0, 'V_CC': 0, 'dim_N_Eval': 0, 'V_CS_Link': 0, 'V_CC_Link': 0, 'dim_E_Eval': 0, 'rapport_CS_nodes': 0.0, 'rapport_CC_nodes': 0.0, 'rapport_CS_links': 0.0, 'rapport_CC_links': 0.0 } return results N_Eval = [node for node in shortest_path if node not in nodes_to_exclude] E_Eval = [] for i in range(len(shortest_path) - 1): u = shortest_path[i] v = shortest_path[i+1] current_edge = tuple(sorted((u, v))) if current_edge not in edges_to_exclude: E_Eval.append(current_edge) servers_content = {server_primary, server_secondary} dim_N_Eval = len(N_Eval) dim_E_Eval = len(E_Eval) dim_Servers = len(servers_content) print("\n--- Set Contents and Dimensions ---") print(f"**Excluded reference nodes :** {nodes_to_exclude}") print(f"**Excluded links from E_Eval :** {edges_to_exclude}") print(f"\n**ENSEMBLE N_Eval** (Nodes to remove from the path):") print(f" Values: {N_Eval}") print(f" Dimension: |N_Eval| = {dim_N_Eval}") print(f"\n**ENSEMBLE E_Eval** (Links to remove from the path):") print(f" Values: {E_Eval}") print(f" Dimension: |E_Eval| = {dim_E_Eval}") print(f"\n**ENSEMBLE Servers_Content** (Servers):") print(f" Values: {servers_content}") print(f" Dimension: |Servers_Content| = {dim_Servers}") print("\n--- Resilience Evaluation by Node Removal (N_Eval) ---") for node_to_remove in N_Eval: G_temp_node = G.copy() if G_temp_node.has_node(node_to_remove): G_temp_node.remove_node(node_to_remove) path_to_primary_exists = False if node_to_remove != server_primary: if nx.has_path(G_temp_node, client, server_primary): path_to_primary_exists = True if path_to_primary_exists: V_CS += 1 V_CC += 1 else: path_to_secondary_exists = False if node_to_remove != server_secondary: if nx.has_path(G_temp_node, client, server_secondary): path_to_secondary_exists = True if path_to_secondary_exists: V_CC += 1 print(f"\n**Node Evaluation Results (based on |N_Eval| = {dim_N_Eval}):") print(f"CS success cases (V_CS): {V_CS}") print(f"CC success cases (V_CC): {V_CC}") if dim_N_Eval > 0: print(f"Ratio V_CS / |N_Eval| = {V_CS}/{dim_N_Eval} (Primary failure resilience)") print(f"Ratio V_CC / |N_Eval| = {V_CC}/{dim_N_Eval} (Secondary failure resilience)") else: print("Rapports Nœuds: N_Eval est vide.") print("\n--- Edge Evaluation Results (E_Eval) ---") for edge_to_remove in E_Eval: G_temp_edge = G.copy() if G_temp_edge.has_edge(*edge_to_remove): G_temp_edge.remove_edge(*edge_to_remove) if nx.has_path(G_temp_edge, client, server_primary): V_CS_Link += 1 V_CC_Link += 1 else: if nx.has_path(G_temp_edge, client, server_secondary): V_CC_Link += 1 print(f"\n**Edge Evaluation Results (based on |E_Eval| = {dim_E_Eval}):") print(f"CS success cases (V_CS_Link): {V_CS_Link}") print(f"CC success cases (V_CC_Link): {V_CC_Link}") results = { 'V_CS': V_CS, 'V_CC': V_CC, 'dim_N_Eval': dim_N_Eval, 'V_CS_Link': V_CS_Link, 'V_CC_Link': V_CC_Link, 'dim_E_Eval': dim_E_Eval, 'rapport_CS_nodes': V_CS / dim_N_Eval if dim_N_Eval > 0 else 0.0, 'rapport_CC_nodes': V_CC / dim_N_Eval if dim_N_Eval > 0 else 0.0, 'rapport_CS_links': V_CS_Link / dim_E_Eval if dim_E_Eval > 0 else 0.0, 'rapport_CC_links': V_CC_Link / dim_E_Eval if dim_E_Eval > 0 else 0.0 } if dim_E_Eval > 0: print(f"Ratio V_CS / |E_Eval| = {V_CS_Link}/{dim_E_Eval} (Primary failure resilience)") print(f"Ratio V_CC / |E_Eval| = {V_CC_Link}/{dim_E_Eval} (Secondary failure resilience)") else: print("Ratio Links: E_Eval is empty.") return results def rf_score_nodes(a: str, b: str) -> float: """Similarity score for node names.""" if not a or not b: return 0.0 a_norm, b_norm = a.lower(), b.lower() return fuzz.ratio(a_norm, b_norm) def build_similarity_matrix_nodes(ref_names: List[str], gen_names: List[str]) -> np.ndarray: """Build similarity matrix.""" 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): """Apply Hungarian algorithm.""" 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 get_mapping(ref_nodes, gen_nodes, min_sim: float = 9.0): """Mapping by layer using Hungarian algorithm.""" if not gen_nodes: return {} mapping: Dict[str, str] = {} ref_groups: Dict[str, List[str]] = {} gen_groups: Dict[str, List[str]] = {} for n in ref_nodes: if n["device_name"] and isinstance(n["device_name"], str): key = n["network_layer"] ref_groups.setdefault(key, []).append(n["device_name"]) for n in gen_nodes: if n["device_name"] and isinstance(n["device_name"], str): key = n["network_layer"] gen_groups.setdefault(key, []).append(n["device_name"]) all_keys = set(ref_groups.keys()) | set(gen_groups.keys()) print(f"\n===== Mapping by Layer (Hungarian Algorithm) =====") for layer in sorted(all_keys, key=lambda x: (x != 'null', x)): ref_names = ref_groups.get(layer, []) gen_names = gen_groups.get(layer, []) display_layer = layer.upper() if layer != 'null' else layer print(f"\n--- Group: {display_layer} ({len(ref_names)} REF vs {len(gen_names)} GEN) ---") if not ref_names or not gen_names: continue sim = build_similarity_matrix_nodes(ref_names, gen_names) pairs = hungarian_match_nodes(sim) layer_matches: Dict[str, str] = {} matches = [] for i, j, s in pairs: r_name = ref_names[i] g_name = gen_names[j] if s >= min_sim: layer_matches[r_name] = g_name matches.append(f"{r_name} -> {g_name} (sim: {s:.1f}%)") for match in matches: print(f" {match}") mapping.update(layer_matches) return mapping def main(): """Main function to analyze resilience of multiple network topologies""" parser = argparse.ArgumentParser(description="Analyze network topology resilience") parser.add_argument("json_path", type=Path, help="Folder containing JSON topology files to analyze") args = parser.parse_args() input_path = args.json_path if not input_path.exists(): print(f"[-] Error: The path '{input_path}' does not exist.") sys.exit(1) if input_path.is_file(): if input_path.suffix.lower() == '.json': json_files = [input_path] else: print(f"[-] Error: The file '{input_path}' is not a JSON file.") sys.exit(1) elif input_path.is_dir(): json_files = sorted(list(input_path.glob("*.json"))) if not json_files: print(f"[-] Error: No JSON files found in folder '{input_path}'.") sys.exit(1) else: print(f"[-] Error: Unsupported path type for '{input_path}'.") sys.exit(1) if input_path.is_file(): print(f"[*] Analyzing single JSON file: {input_path.name}") else: print(f"[*] Analyzing {len(json_files)} JSON files in folder: {input_path}") all_results = [] for json_file in json_files: print(f"\n{'='*60}") print(f" Analyzing file: {json_file.name}") print(f"{'='*60}") try: json_data = load_json_file(json_file) G = create_networkx_graph(json_data) print("× NetworkX graph created from topology.") if "result" in json_data: devices_data = json_data["result"]["network_topology"]["devices"] else: devices_data = json_data["network_topology"]["devices"] gen_nodes = [] for device in devices_data: connections = [] for conn in device.get("connections", []): connections.append({ "peer_device": conn.get("peer_device"), "interface_name": conn.get("interface_name"), "peer_interface": conn.get("peer_interface"), "interface_type": conn.get("interface_type", "Ethernet") }) gen_nodes.append({ "device_name": device.get("device_name"), "network_layer": str(device.get("network_layer", "access")).lower(), "connections": connections }) ref_nodes = [ {"device_name": "Avignon_PC1", "network_layer": "endpoint"}, {"device_name": "Avignon_Switch1", "network_layer": "access"}, {"device_name": "Marseille_Primary_Video_Delivery_Server", "network_layer": "endpoint"}, {"device_name": "Marseille_Backup_Video_Delivery_Server", "network_layer": "endpoint"} ] if HAS_SCIPY: mapping = get_mapping(ref_nodes, gen_nodes, min_sim=9.0) client = mapping.get("Avignon_PC1") client_switch = mapping.get("Avignon_Switch1") server_primary = mapping.get("Marseille_Primary_Video_Delivery_Server") server_secondary = mapping.get("Marseille_Backup_Video_Delivery_Server") print(f"\n@ Mapping with Hungarian Algorithm:") print(f" Client: {client}") print(f" Switch client: {client_switch}") print(f" Primary Serveur : {server_primary}") print(f" Secondary Serveur : {server_secondary}") nodes_to_exclude = [node for node in [client, client_switch] if node is not None] edges_to_exclude = [] if client is not None and client_switch is not None: edges_to_exclude = [tuple(sorted((client, client_switch)))] print(f"\n@ Final Configuration:") print(f" Nodes to exclude: {nodes_to_exclude}") print(f" Edges to exclude: {edges_to_exclude}") if client is None or server_primary is None or server_secondary is None: print(f" Mapping incomplete - all metrics set to 0") results = { 'V_CS': 0, 'V_CC': 0, 'dim_N_Eval': 0, 'V_CS_Link': 0, 'V_CC_Link': 0, 'dim_E_Eval': 0, 'rapport_CS_nodes': 0.0, 'rapport_CC_nodes': 0.0, 'rapport_CS_links': 0.0, 'rapport_CC_links': 0.0 } else: results = analyze_resilience(G, client, server_primary, server_secondary, nodes_to_exclude, edges_to_exclude) results['fichier'] = json_file.name all_results.append(results) except Exception as e: print(f" Error parsing file {json_file.name}: {e}") continue if all_results: display_results_table(all_results) calculate_summary_statistics(all_results) else: print("\n No results to display.") def display_results_table(results: List[Dict]): print(f"\n{'='*80}") print(" RESULTS SUMMARY TABLE") print(f"{'='*80}") table_data = [] headers = ["File", "CS Nodes", "CC Nodes", "CS Links", "CC Links"] for result in results: cs_nodes_display = "0" if result['dim_N_Eval'] == 0 else f"{result['V_CS']}/{result['dim_N_Eval']}" cc_nodes_display = "0" if result['dim_N_Eval'] == 0 else f"{result['V_CC']}/{result['dim_N_Eval']}" cs_links_display = "0" if result['dim_E_Eval'] == 0 else f"{result['V_CS_Link']}/{result['dim_E_Eval']}" cc_links_display = "0" if result['dim_E_Eval'] == 0 else f"{result['V_CC_Link']}/{result['dim_E_Eval']}" row = [ result['fichier'], cs_nodes_display, cc_nodes_display, cs_links_display, cc_links_display ] table_data.append(row) print(tabulate(table_data, headers=headers, tablefmt="grid")) def calculate_summary_statistics(results: List[Dict]): """Calculate and display summary statistics""" print(f"\n{'='*80}") print(" SUMMARY STATISTICS") print(f"{'='*80}") if not results: print("No results to analyze.") return import math total_cs_nodes = sum(r['V_CS'] for r in results) total_cc_nodes = sum(r['V_CC'] for r in results) total_cs_links = sum(r['V_CS_Link'] for r in results) total_cc_links = sum(r['V_CC_Link'] for r in results) if results: dim_N_Eval = results[0]['dim_N_Eval'] dim_E_Eval = results[0]['dim_E_Eval'] sum_cs_nodes_ratio = sum(r['V_CS'] / r['dim_N_Eval'] if r['dim_N_Eval'] > 0 else 0.0 for r in results) sum_cc_nodes_ratio = sum(r['V_CC'] / r['dim_N_Eval'] if r['dim_N_Eval'] > 0 else 0.0 for r in results) sum_cs_links_ratio = sum(r['V_CS_Link'] / r['dim_E_Eval'] if r['dim_E_Eval'] > 0 else 0.0 for r in results) sum_cc_links_ratio = sum(r['V_CC_Link'] / r['dim_E_Eval'] if r['dim_E_Eval'] > 0 else 0.0 for r in results) avg_cs_nodes_ratio = sum_cs_nodes_ratio / len(results) avg_cc_nodes_ratio = sum_cc_nodes_ratio / len(results) avg_cs_links_ratio = sum_cs_links_ratio / len(results) avg_cc_links_ratio = sum_cc_links_ratio / len(results) global_avg_ratio = (avg_cs_nodes_ratio + avg_cc_nodes_ratio + avg_cs_links_ratio + avg_cc_links_ratio) / 4 avg_cs_nodes = sum(r['V_CS'] for r in results) / len(results) avg_cc_nodes = sum(r['V_CC'] for r in results) / len(results) avg_cs_links = sum(r['V_CS_Link'] for r in results) / len(results) avg_cc_links = sum(r['V_CC_Link'] for r in results) / len(results) global_avg = (avg_cs_nodes + avg_cc_nodes + avg_cs_links + avg_cc_links) / 4 else: avg_cs_nodes = avg_cc_nodes = avg_cs_links = avg_cc_links = global_avg = 0.0 avg_cs_nodes_ratio = avg_cc_nodes_ratio = avg_cs_links_ratio = avg_cc_links_ratio = global_avg_ratio = 0.0 dim_N_Eval = dim_E_Eval = 0 def std_dev(values): if len(values) <= 1: return 0.0 mean = sum(values) / len(values) variance = sum((x - mean) ** 2 for x in values) / len(values) return math.sqrt(variance) cs_nodes_values = [r['V_CS'] / r['dim_N_Eval'] if r['dim_N_Eval'] > 0 else 0.0 for r in results] cc_nodes_values = [r['V_CC'] / r['dim_N_Eval'] if r['dim_N_Eval'] > 0 else 0.0 for r in results] cs_links_values = [r['V_CS_Link'] / r['dim_E_Eval'] if r['dim_E_Eval'] > 0 else 0.0 for r in results] cc_links_values = [r['V_CC_Link'] / r['dim_E_Eval'] if r['dim_E_Eval'] > 0 else 0.0 for r in results] std_cs_nodes = std_dev(cs_nodes_values) std_cc_nodes = std_dev(cc_nodes_values) std_cs_links = std_dev(cs_links_values) std_cc_links = std_dev(cc_links_values) all_values = cs_nodes_values + cc_nodes_values + cs_links_values + cc_links_values std_global = std_dev(all_values) stats_table = [ ["Avg Node CS", f"{avg_cs_nodes:.4f}/{dim_N_Eval} ({avg_cs_nodes/dim_N_Eval:.4f})" if dim_N_Eval > 0 else f"{avg_cs_nodes:.4f}/0 (0.0000)"], ["Avg Node CC", f"{avg_cc_nodes:.4f}/{dim_N_Eval} ({avg_cc_nodes/dim_N_Eval:.4f})" if dim_N_Eval > 0 else f"{avg_cc_nodes:.4f}/0 (0.0000)"], ["Avg Link CS", f"{avg_cs_links:.4f}/{dim_E_Eval} ({avg_cs_links/dim_E_Eval:.4f})" if dim_E_Eval > 0 else f"{avg_cs_links:.4f}/0 (0.0000)"], ["Avg Link CC", f"{avg_cc_links:.4f}/{dim_E_Eval} ({avg_cc_links/dim_E_Eval:.4f})" if dim_E_Eval > 0 else f"{avg_cc_links:.4f}/0 (0.0000)"], ["Avg Global", f"{global_avg:.4f}"] ] print(tabulate(stats_table, headers=["Metric", "Ratio (Value)"], tablefmt="grid")) std_table = [ ["Std Dev CS Nodes", f"{std_cs_nodes:.4f}"], ["Std Dev CC Nodes", f"{std_cc_nodes:.4f}"], ["Std Dev CS Links", f"{std_cs_links:.4f}"], ["Std Dev CC Links", f"{std_cc_links:.4f}"], ["Std Dev Global", f"{std_global:.4f}"] ] print(tabulate(std_table, headers=["Metric", "Value"], tablefmt="grid", floatfmt=".4f")) print(f"\n Number of files analyzed: {len(results)}") print(f" Total number of measurements: {len(all_values)}") print(f" Global performance: {global_avg:.2%}") print(f"\n{'='*80}") print(" SUMMARY OF MEANS AND STANDARD DEVIATIONS (PERCENTAGES)") print(f"{'='*80}") pct_cs_nodes = avg_cs_nodes_ratio * 100 pct_cc_nodes = avg_cc_nodes_ratio * 100 pct_cs_links = avg_cs_links_ratio * 100 pct_cc_links = avg_cc_links_ratio * 100 pct_global = global_avg_ratio * 100 pct_std_cs_nodes = std_cs_nodes * 100 pct_std_cc_nodes = std_cc_nodes * 100 pct_std_cs_links = std_cs_links * 100 pct_std_cc_links = std_cc_links * 100 pct_std_global = std_global * 100 pct_table = [ ["CS Nodes (%)", f"{pct_cs_nodes:.2f}%"], ["CC Nodes (%)", f"{pct_cc_nodes:.2f}%"], ["CS Links (%)", f"{pct_cs_links:.2f}%"], ["CC Links (%)", f"{pct_cc_links:.2f}%"], ["Avg Global (%)", f"{pct_global:.2f}%"], ["Std Dev CS Nodes (%)", f"{pct_std_cs_nodes:.2f}%"], ["Std Dev CC Nodes (%)", f"{pct_std_cc_nodes:.2f}%"], ["Std Dev CS Links (%)", f"{pct_std_cs_links:.2f}%"], ["Std Dev CC Links (%)", f"{pct_std_cc_links:.2f}%"], ["Std Dev Global (%)", f"{pct_std_global:.2f}%"] ] print(tabulate(pct_table, headers=["Metric", "Percentage"], tablefmt="grid")) if __name__ == "__main__": main()