ResiNet-LLM-topology / f1_edges_main.py
KholoudLIA's picture
Update f1_edges_main.py
2699e6b verified
Raw
History Blame Contribute Delete
24.4 kB
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()