Spaces:
Paused
Paused
| # utils.py | |
| import networkx as nx | |
| import config | |
| def calculate_metric(G, metric_name): | |
| """ | |
| Calculates metrics. Includes: | |
| Originals: Density, Clustering, Cycles, Interdependence, etc | |
| """ | |
| try: | |
| n = G.number_of_nodes() | |
| if metric_name == "Nodes": return str(n) | |
| if metric_name == "Edges": return str(G.number_of_edges()) | |
| if n == 0: return "0" | |
| if metric_name == "Density": | |
| return f"{nx.density(G):.3f}" | |
| if metric_name == "Avg Degree": | |
| avg_deg = sum([d for _, d in G.degree()]) / n | |
| return f"{avg_deg:.2f}" | |
| if metric_name == "Avg Cycle Length": | |
| try: | |
| cycles = list(nx.simple_cycles(G)) | |
| if not cycles: return "0.0 (None)" | |
| lengths = [len(c) for c in cycles] | |
| avg = sum(lengths) / len(lengths) | |
| return f"{avg:.2f} {lengths}" | |
| except: return "Err" | |
| if metric_name == "Interdependence": | |
| try: | |
| m = G.number_of_edges() | |
| if m == 0: return "0.000" | |
| cross_boundary_edges = 0 | |
| for u, v in G.edges(): | |
| agent_u = G.nodes[u].get('agent', 'Unassigned') | |
| agent_v = G.nodes[v].get('agent', 'Unassigned') | |
| if agent_u != agent_v: cross_boundary_edges += 1 | |
| return f"{(cross_boundary_edges / m):.3f}" | |
| except: return "Err" | |
| if metric_name == "Cyclomatic Number": | |
| try: | |
| e = G.number_of_edges() | |
| p = nx.number_weakly_connected_components(G) | |
| return str(e - n + p) | |
| except: return "Err" | |
| if metric_name == "Critical Loop Nodes": | |
| try: | |
| fvs = nx.approximation.min_weighted_feedback_vertex_set(G) | |
| return str(len(fvs)) | |
| except: return "0" | |
| if metric_name == "Total Cycles": | |
| try: | |
| count = 0 | |
| for _ in nx.simple_cycles(G): | |
| count += 1 | |
| if count > 100: return "100+" | |
| return str(count) | |
| except: return "Err" | |
| if metric_name == "Global Efficiency": | |
| # Measures how integrated the system is (0.0 to 1.0) | |
| try: | |
| # Treated as undirected to measure potential for information flow | |
| eff = nx.global_efficiency(G.to_undirected()) | |
| return f"{eff:.3f}" | |
| except: return "Err" | |
| if metric_name == "Modularity": | |
| # Detects if system splits into distinct groups (Q-Score) | |
| try: | |
| communities = nx.community.greedy_modularity_communities(G.to_undirected()) | |
| num_communities = len(communities) | |
| q_score = nx.community.modularity(G.to_undirected(), communities) | |
| return f"Q={q_score:.2f} ({num_communities} Grps)" | |
| except: return "Err" | |
| if metric_name == "Brittleness Ratio": # This isn't a great name, will rename the button | |
| # A. Ratio of Soft to Hard edges | |
| soft = sum(1 for u, v, d in G.edges(data=True) if d.get('type') == config.EDGE_TYPE_SOFT) | |
| hard = sum(1 for u, v, d in G.edges(data=True) if d.get('type') == config.EDGE_TYPE_HARD) | |
| if hard == 0: return "Infinite (No Hard Edges)" | |
| ratio = soft / hard | |
| return f"{ratio:.2f} (S:{soft}/H:{hard})" | |
| if metric_name == "Supportive Gain": | |
| # B. Supportive Gain: Efficiency(Total) - Efficiency(HardOnly) | |
| # Measures how much integration is lost if soft edges fail. | |
| # 1. Total Efficiency | |
| eff_total = nx.global_efficiency(G.to_undirected()) | |
| # 2. Hard-edges Only Efficiency | |
| hard_edges = [(u, v) for u, v, d in G.edges(data=True) if d.get('type') == config.EDGE_TYPE_HARD] | |
| G_hard = nx.Graph() | |
| G_hard.add_nodes_from(G.nodes()) | |
| G_hard.add_edges_from(hard_edges) | |
| eff_hard = nx.global_efficiency(G_hard) | |
| gain = eff_total - eff_hard | |
| return f"{gain:.3f} (Tot: {eff_total:.2f})" | |
| if metric_name == "Critical Vulnerability": | |
| # C. Critical Path Vulnerability: | |
| # Does the graph fragment if soft edges are removed? | |
| # Measures Connected Components of the Hard-Only graph. | |
| hard_edges = [(u, v) for u, v, d in G.edges(data=True) if d.get('type') == config.EDGE_TYPE_HARD] | |
| G_hard = nx.DiGraph() # di-graph to catch strict flow breaks | |
| G_hard.add_nodes_from(G.nodes()) | |
| G_hard.add_edges_from(hard_edges) | |
| # Weakly connected components (islands of connectivity) | |
| num_components = nx.number_weakly_connected_components(G_hard) | |
| if num_components == 1: | |
| return "Robust (1 Comp)" | |
| else: | |
| return f"Fractured ({num_components} Comps)" | |
| ### Metric Ideas for analyzing SHared Authority | |
| if metric_name == "Functional Redundancy": | |
| # "Average number of agents" capable of performing a function. | |
| # Idea: 1.0 = Brittle (No backup), >1.2 = Resilient (more backups) | |
| total_agents = 0 | |
| func_count = 0 | |
| for n, d in G.nodes(data=True): | |
| if d.get('type') == 'Function': | |
| # Get list of agents, handling single strings if necessary | |
| ag = d.get('agent', []) | |
| if not isinstance(ag, list): ag = [ag] | |
| # Count real agents (exclude "Unassigned") | |
| real_agents = [x for x in ag if x != "Unassigned"] | |
| total_agents += len(real_agents) | |
| func_count += 1 | |
| if func_count == 0: return "0.0" | |
| avg = total_agents / func_count | |
| return f"{avg:.2f} (Avg Agents)" | |
| if metric_name == "Agent Criticality": | |
| # try to find the agent with the most "authority" in the system | |
| # for the most functions? (If they disconnect, these functions fail) | |
| agent_sole_counts = {} | |
| for n, d in G.nodes(data=True): | |
| if d.get('type') == 'Function': | |
| ag = d.get('agent', []) | |
| if not isinstance(ag, list): ag = [ag] | |
| real_agents = [x for x in ag if x != "Unassigned"] | |
| # If exactly one agent owns this, they are critical to it | |
| if len(real_agents) == 1: | |
| sole_agent = real_agents[0] | |
| agent_sole_counts[sole_agent] = agent_sole_counts.get(sole_agent, 0) + 1 | |
| if not agent_sole_counts: | |
| return "None (Robust)" | |
| # Find the agent with the highest count | |
| worst_agent = max(agent_sole_counts, key=agent_sole_counts.get) | |
| count = agent_sole_counts[worst_agent] | |
| return f"{worst_agent} ({count} Sole Tasks)" | |
| if metric_name == "Collaboration Ratio": | |
| # Percent of functions that involve multiple agents | |
| shared_funcs = 0 | |
| total_funcs = 0 | |
| for n, d in G.nodes(data=True): | |
| if d.get('type') == 'Function': | |
| ag = d.get('agent', []) | |
| if not isinstance(ag, list): ag = [ag] | |
| # Filter out 'Unassigned' (only count other agents) | |
| real_agents = [x for x in ag if x != "Unassigned"] | |
| if len(real_agents) > 0: | |
| total_funcs += 1 | |
| # if more than 1 agent is assigned then it's a collaborative task | |
| if len(real_agents) > 1: | |
| shared_funcs += 1 | |
| if total_funcs == 0: return "0.0%" | |
| ratio = (shared_funcs / total_funcs) * 100 | |
| return f"{ratio:.1f}%" | |
| except Exception as e: | |
| print(f"Error calculating {metric_name}: {e}") | |
| return "Err" | |
| return "" |