ResiNet-LLM-topology / evaluate_connectivity.py
KholoudLIA's picture
Upload folder using huggingface_hub
9696f7a verified
Raw
History Blame Contribute Delete
24.8 kB
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()