import numpy as np
import networkx as nx
import math
import matplotlib.pyplot as plt
from sklearn.metrics import confusion_matrix
from scipy.optimize import linear_sum_assignment as linear_assignment
import scipy.special as ss
import pandas as pd
import matplotlib.colors as mcolors
import random
import collections
def structural_pattern_int(statistics):
"""return the structural pattern of a graph for integer nodal statistics:
# input is the dictionary of values (node, measure value)
# return a dictionary having as key the values in statistics codomain and as values the orbit/class (as set of nodes)
"""
measure=dict(statistics)
# input is the dictionary of values (node, measure value)
# compute the orbits, return a dictionary with key all taken measure value and object the set of nodes
orbit_set={}
for m in set(measure.values()):
orbit_set[m]=set([k for k,v in measure.items() if v==m])
return orbit_set
def structural_pattern(statistics,epsilon):
"""return the structural pattern of a graph for dense nodal statistics:
# input is the dictionary of values (node, measure value) and epsilon which corresponds to the number of significativ digit to be used for the values comparison
# return a dictionary having as key the values in statistics codomain and as values the orbit/class (as set of nodes)
"""
epsilon = "{:."+str(epsilon)+"}"
measure=dict(statistics)
#input is the dictionary of values (node, measure value)
#compute the orbits, return a dictionary with key all taken measure value and object the set of nodes
orbit_set={}
for m in set(measure.values()):
orbit_set[epsilon.format(float(m))]=set([k for k,v in measure.items() if epsilon.format(float(v))==epsilon.format(float(m))])
return orbit_set
def nodes_in_trivial_classes(structural_pattern):
""""return a list of nodes in trivial class of the given structural pattern of a graph
#input is the structural pattern of a graph (as set of classes)
"""
#taken a dictionary of orbits
#return a list of fixed point
fixed_point=[]
for (k,o) in structural_pattern.items():
if len(o)==1:
fixed_point.append(list(o)[0])
return fixed_point
def intersection_structural_patterns(orbits_set_1,orbits_set_2):
"""given two structural pattern associated with different statistics, return their intersection
#input are two structural patterns on the same graph (as orbits set)
#output it the intersection of structural patterns
"""
#for each orbit set in orbits1 compute the intersection with each orbitset in orbits2
#keep only the non-empty set
#repeat for every orbitset in orbits1
intersection_orbits={}
for (k,o1) in orbits_set_1.items():
for (c,o2) in orbits_set_2.items():
f=o1.intersection(o2)
if len(f)!=0:
intersection_orbits[(k,c)]=f
return intersection_orbits
def count_nodes_in_non_trivial_class(structural_pattern,num_nodes):
"""return the number of nodes in non-trivial class for a given orbits set and the total number of nodes"""
return num_nodes-len(nodes_in_trivial_classes(structural_pattern))
def graph_structural_pattern(G, statistics=["d","b","cc","cs","s"],precision=6):
""" return the structural pattern of the graph associated to each single statistics in statistics list
implemented only for the following nodal statistics:
"d" degree,"b" betweenness centrality,"cc" clustering coefficient,cs" closeness centrality,"s" second order centrality
#output is in the format of dictionary with keys the statistics identifier
"""
orbits_G={}
error_case = {"0":set(G.nodes())}
if "d" in statistics:
try:
degrees=G.degree()
orbits_G["d"] = structural_pattern_int(degrees)
except:
orbits_G["d"] = error_case
if "b" in statistics:
try:
betw=nx.betweenness_centrality(G)
orbits_G["b"] = structural_pattern(betw,precision)
except:
orbits_G["b"] = error_case
if "cc" in statistics:
try:
cc=nx.clustering(G)
orbits_G["cc"] = structural_pattern(cc,precision)
except:
orbits_G["cc"] = error_case
if "cs" in statistics:
try:
cs = nx.closeness_centrality(G)
orbits_G["cs"] = structural_pattern(cs,precision)
except:
orbits_G["cs"] = error_case
if "s" in statistics:
try:
s=nx.second_order_centrality(G)
orbits_G["s"] = structural_pattern(s,precision)
except:
orbits_G["s"] = error_case
return orbits_G
def orthogonality_in_G(G,statistics=["d","b","cc","cs","s"],precision=6):
"""
return a measure of orthogonality of a collection of nodal statistics,
implemented only for the following nodal statistics:
"d" degree,"b" betweenness centrality,"cc" clustering coefficient,cs" closeness centrality,"s" second order centrality
0 means perfectely orthogonal (all nodes of the intersection are in trivial orbits)
1 means all nodes in the intersection have at least one equivalent node"""
N = len(G.nodes())
orbit_set_of_G=graph_structural_pattern(G,statistics=statistics,precision=precision)
inter=orbit_set_of_G.get(measure[0])
for m in measure:
o = orbit_set_of_G.get(m)
inter = intersection_structural_patterns(inter,o)
return count_nodes_in_non_trivial_class(inter,N)/N
def orthogonal_score(orbits_set_1,orbits_set_2,N):
"""return a measure of orthogonality of two orbits sets,
0 means perfectely orthogonal (all nodes of the intersection are in trivial class)
1 means all nodes have at least one equivalent node"""
inter = intersection_structural_patterns(orbit_set_1,orbit_set_2)
return count_nodes_in_non_trivial_orbits(inter,N)/N
import scipy.special as ss
def open_data(path, dtype=float, skiprows=0, usecols=range(90)):
"""open correlation matrices file given its path, check usecols to be range(number of nodes)"""
#open a single csv file and convert into a numpy matrix
#in our case it'd be a 90x90 matrix, for gin-chuga files need to skip a row and usecols from 1-91
corr_matrix=np.loadtxt(fname=path,dtype=dtype,skiprows=skiprows,usecols=usecols)
return corr_matrix
import math
from networkx import tree
def adj_matrix_connected(corr_matrix,sparsity_value):
"""
given the correlation matrix and the expected sparsity coefficient it can
happen that the corresponding thresholded matrix results in a disconnected graph
here we force the graph to be fully connected by the computation of the minimum
spanning tree and adding the required edges in order to have a unique connected component
"""
if sparsity_value == 1.0:
adj_matrix=np.ones(corr_matrix.shape)
np.fill_diagonal(adj_matrix,0)
return adj_matrix
corr_matrix =abs(corr_matrix)
max_num_edges = ss.comb(corr_matrix.shape[0],2)
num_edges = int(max_num_edges*sparsity_value)
num_regions=corr_matrix.shape[0]
#total number of regions in the graph
totalgraph=nx.from_numpy_matrix(1-abs(corr_matrix))
#extraction of a complete graph having has weight 1-abs(correlation)
#we need to take 1-abs since the mst is taking the minimum weight graph and we want the most correlated edges to be there
MST=nx.adjacency_matrix(tree.minimum_spanning_tree(totalgraph).to_undirected()).todense()
MST_adj_mat=MST
MST_adj_mat[MST>0]==1
MST_adj_mat=np.triu(MST_adj_mat) #put zeros in the inferior triangular matrix
#put zeros in the diagonal of the corr matrix
for i in range(num_regions):
corr_matrix[i,i]=0
values_corr=abs(np.triu(corr_matrix))
cor_wo_MST=values_corr[np.triu(MST_adj_mat)==0]
#we do not consider the correlation values which do not involve edges that are already in the MST
values=list(cor_wo_MST.flatten())
values.sort(reverse=True)
#we select the maximum value of correlation to have the expected num of edges - num of edges in the mst (num regions-1)
value_thresh=values[num_edges-(num_regions-1)-1] #-1 index start at 0
adj_matrix=np.zeros(corr_matrix.shape)
#we put an edge if the value of correlation is higher than the found threshold or if the edges is required by the mst
adj_matrix[values_corr>=value_thresh]=1
adj_matrix[MST_adj_mat!=0]=1
adj_matrix=np.triu(adj_matrix)+np.transpose(np.triu(adj_matrix)) #simmetry of the adj matrix
return adj_matrix
def adj_matrix_connected_num_edges(corr_matrix,num_edges):
"""given the correlation matrix and the expected # edges it can
happen that the corresponding thresholded matrix results in a disconnected graph
here we force the graph to be fully connected by the computation of the minimum
spanning tree and adding the required edges in order to have a single connected component
"""
max_num_edges = ss.comb(corr_matrix.shape[0],2)
if num_edges == max_num_edges:
adj_matrix=np.ones(corr_matrix.shape)
np.fill_diagonal(adj_matrix,0)
return adj_matrix
num_regions=corr_matrix.shape[0]
#total number of regions in the graph
totalgraph=nx.from_numpy_matrix(1-abs(corr_matrix))
#extraction of a complete graph having has weight 1-abs(correlation)
#we need to take 1-abs since the mst is taking the minimum weight graph and we want the most correlated edges to be there
MST=nx.adjacency_matrix(tree.minimum_spanning_tree(totalgraph).to_undirected()).todense()
MST_adj_mat=MST
MST_adj_mat[MST>0]==1
MST_adj_mat=np.triu(MST_adj_mat) #put zeros in the inferior triangular matrix
#put zeros in the diagonal of the corr matrix
for i in range(num_regions):
corr_matrix[i,i]=0
values_corr=abs(np.triu(corr_matrix))
cor_wo_MST=values_corr[np.triu(MST_adj_mat)==0]
#we do not consider the correlation values which do not involve edges that are already in the MST
values=list(cor_wo_MST.flatten())
values.sort(reverse=True)
#we select the maximum value of correlation to have the expected num of edges - num of edges in the mst (num regions-1)
value_thresh=values[num_edges-(num_regions-1)-1] #-1 index start at 0
adj_matrix=np.zeros(corr_matrix.shape)
#we put an edge if the value of correlation is higher than the found threshold or if the edges is required by the mst
adj_matrix[values_corr>=value_thresh]=1
adj_matrix[MST_adj_mat!=0]=1
adj_matrix=np.triu(adj_matrix)+np.transpose(np.triu(adj_matrix)) #simmetry of the adj matrix
return adj_matrix
def adj_converstion_to_structural_patterns(dataset_adj,statistics,precision=6):
""" convert a dataset of adjacency matrices to one of structural pattern associated to the collection of statistics in statistics,
available statistics are "d" degree,"b" betweenness centrality,"cc" clustering coefficient,cs" closeness centrality,"s" second order centrality
"""
dataset_orbits = {}
for key,adj in dataset_adj.items():
G = nx.from_numpy_array(adj)
orbits_of_G = graph_structural_pattern(G, statistics=statistics, precision=precision)
intersection = orbits_of_G.get(statistics[0])
if len(statistics) > 1:
for s in statistics[1:]:
intersection = intersection_structural_patterns(intersection,orbits_of_G.get(s))
dataset_orbits[key] = intersection
return dataset_orbits
def graphs_converstion_to_structural_patterns(dataset_graph,statistics,precision=6):
""" convert a dataset of graphs to one of structural pattern associated to the given list of statistics
available statistics are "d" degree,"b" betweenness centrality,"cc" clustering coefficient,cs" closeness centrality,"s" second order centrality
"""
dataset_orbits = {}
for key,G in dataset_graph.items():
orbits_of_G = graph_structural_pattern(G, statistics=statistics, precision=precision)
intersection = orbits_of_G.get(statistics[0])
if len(statistics) > 1:
for s in statistics[1:]:
intersection = intersection_structural_patterns(intersection,orbits_of_G.get(s))
dataset_orbits[key] = intersection
return dataset_orbits
def orthogonality_of_dataset(dataset, statistics =["d","b","cc","cs","s"],precision = 6):
""" evaluate the orthogonality of the collection of statistics on a given dataset, dataset can be either adjacency dataset or graph dataset
available statistics are "d" degree,"b" betweenness centrality,"cc" clustering coefficient,cs" closeness centrality,"s" second order centrality
"""
#check the dataset type
sample = list(dataset.values())[0]
if isinstance(sample,np.ndarray):
#the dataset contains adjacency matrix
dataset_converted = adj_converstion_to_structural_patterns(dataset,statistics,precision=precision)
N = sample.shape[0]
elif isinstance(sample,nx.Graph):
#the dataset contains graphs
dataset_converted = graphs_converstion_to_structural_patterns(dataset,statistics,precision=precision)
N = len(sample.nodes())
else:
print("the dataset is not in a good format,check it is composed either of nx.Graph or np.ndarray ")
return None
ortho = []
for key,str_pat in dataset_converted.items() :
ortho.append((N-len(nodes_in_trivial_orbits(str_pat)))/N)
return ortho
def orthogonality_of_dataset_over_sparsity(dataset, sparsity =np.linspace(0.1,1,10),statistics =["d","b","cc","cs","s"],precision = 6)
""" evaluate the orthogonality of the given collection of statistics at different sparsity level, dataset is a correlation matrices dataset, return the mean orthogonality score and the std orthogonality
available statistics are "d" degree,"b" betweenness centrality,"cc" clustering coefficient,cs" closeness centrality,"s" second order centrality
return a tuple of mean value and std
"""
ortho_mean={}
ortho_std={}
for s in sparsity:
adj_dataset = {}
for k,corr_matrix in dataset.items():
adj_dataset[k] = adj_matrix_connected(corr_matrix,s)
dataset_converted = adj_converstion_to_structural_patterns(dataset,statistics,precision=precision)
orth_s = orthogonality_of_dataset(adj_dataset, statistics = statistics, precision = precision)
ortho_mean[s] = np.mean(orth_s)
ortho_std[s] = np.std(orth_s)
return ortho_mean, ortho_std
def nodal_percentage_participation_no_class_distinction(structural_patterns_of_a_group,num_region):
"""input is a dictionary with keys (class,subject) and values the structural pattern,
return a dictionary whose keys are the nodes and values are the nodal percentage participation in non trivial class
with no class_distiction!
"""
counting_classe = {}
for i in range(num_region):
counting_classe[i]=0
for o in structural_patterns_of_a_group.values():
n = nodes_in_trivial_class(o)
for i in range(num_region):
if i not in n:
counting_classe[i]=counting_classe[i]+1
for i in range(num_region):
counting_classe[i] = counting_classe[i]/len(list(orbits_dataset.keys()))
return counting_classe
def nodal_percentage_participation_dataset(structural_pattern_dataset,num_region):
"""dataset is dataset of structural_pattern_dataset at an already fixed sparsity and associated with previously given collection of statistics,
input is a dictionary with keys (class,subject) and values the structural pattern,
return a dictionary whose keys are the class in the dataset and values is a dictionary keyd by nodes and values are nodal percentage participation in non trivial class
"""
counting_dataset = {}
for classe in set([k[0] for k in orbits_dataset.keys()]):
counting_classe={}
for i in range(num_region):
counting_classe[i]=0
all_classe=[orb for k,orb in list(structural_pattern_dataset.items()) if k[0]==classe]
for o in all_classe:
n = nodes_in_trivial_class(o)
for i in range(num_region):
if i not in n:
counting_classe[i]=counting_classe[i]+1
for i in range(num_region):
counting_classe[i] = counting_classe[i]/len(all_classe)
counting_dataset[classe] = counting_classe
return counting_dataset
def help_to_structural_patterns(dataset_adj,statistics,precision=6):
""" convert a dataset of adjacency matrices to one of structural pattern associated to all statistics in the given list of statistics
implemented only for the following nodal statistics:
"d" degree,"b" betweenness centrality,"cc" clustering coefficient,cs" closeness centrality,"s" second order centrality
"""
dataset_orbits = {}
for key,adj in dataset_adj.items():
G = nx.from_numpy_array(adj)
orbits_of_G = graph_structural_pattern(G, statistics=statistics, precision=precision)
dataset_orbits[key] = orbits_of_G
return dataset_orbits
from itertools import combinations
def orthogonality_of_dataset_over_pairs(dataset, statistics = ["d","b","cc","cs","s"],precision = 6):
""" evaluate the orthogonality of all possible pairs of nodal statistics in the input list on a given dataset,
dataset can be either adjacencies dataset or graph dataset (class,subjectID):adj_mat/(class,subjectID):graph
implemented only for the following nodal statistics:
"d" degree,"b" betweenness centrality,"cc" clustering coefficient,cs" closeness centrality,"s" second order centrality
return a dictionary keys by statistics pairs with values a list of orthogonality score of the elements in dataset
"""
dataset_converted = help_to_structural_patterns(dataset,statistics,precision=precision)
sample = list(dataset.values())[0]
N = sample.shape[0]
ortho = {}
couple_statistics = list(combinations(statistics,2))
for k in range(len(couple_statistics)):
couple = couple_statistics[k]
ortho[couple] = []
for k in range(len(couple_statistics)):
couple = couple_statistics[k]
for key,str_pat in dataset_converted.items() :
intersection = str_pat.get(couple[0])
intersection = intersection_structural_patterns(intersection,str_pat.get(couple[1]))
o = (N-len(nodes_in_trivial_class(intersection)))/N
ortho[couple].append(o)
return ortho