¿Cómo usar el argumento `pos` en` networkx` para crear un gráfico de estilo de diagrama de flujo?(Python 3)

matplotlib plot

Question

Estoy tratando de crear un gráfico de red lineal usando Python (preferiblemente con matplotlib y networkx aunque estaría interesado en bokeh ) similar en concepto al que se muestra a continuación.

¿Cómo se puede construir esta gráfica de forma eficiente ( pos ?) En Python usando networkx ? Quiero usar esto para ejemplos más complicados, así que creo que no será útil codificar las posiciones para este ejemplo simple :(. ¿Tiene la networkx una solución para esto?

pos (diccionario, opcional) - Un diccionario con nodos como claves y posiciones como valores. Si no se especifica, se calculará un posicionamiento del resorte. Consulte networkx.layout para conocer las funciones que computan las posiciones de los nodos.

No he visto ningún tutorial sobre cómo se puede lograr esto en networkx por lo que creo que esta pregunta será un recurso confiable para la comunidad. He networkx tutoriales de networkx y nada de esto está ahí. Los diseños para networkx harían que este tipo de red sea imposible de interpretar sin un uso cuidadoso del argumento pos ... que creo que es mi única opción. Ninguno de los diseños precomputados en la documentación https://networkx.github.io/documentation/networkx-1.9/reference/drawing.html parece manejar bien este tipo de estructura de red.

Ejemplo simple:

(A) cada clave externa es la iteración en el gráfico que se mueve de izquierda a derecha (por ejemplo, la iteración 0 representa muestras, la iteración 1 tiene grupos 1 - 3, la misma iteración 2, la iteración 3 tiene Grupos 1 - 2, etc.). (B) El diccionario interno contiene la agrupación actual en esa iteración en particular, y las ponderaciones de los grupos anteriores que se fusionan representan el grupo actual (por ejemplo, la iteration 3 tiene el Group 1 y el Group 2 y para la iteration 4 todo iteration 3''s Group 2 de la iteration 3''s ha ido en iteration 4''s Group 2 iteration 4''s pero iteration 3''s Group 1 iteration 3''s se ha dividido. Los pesos siempre suman 1.

Mi código para las conexiones w / pesas para la trama de arriba:

D_iter_current_previous = {
1: {
"Group 1":{"sample_0":0.5, "sample_1":0.5, "sample_2":0, "sample_3":0, "sample_4":0},
"Group 2":{"sample_0":0, "sample_1":0, "sample_2":1, "sample_3":0, "sample_4":0},
"Group 3":{"sample_0":0, "sample_1":0, "sample_2":0, "sample_3":0.5, "sample_4":0.5}
},
2: {
"Group 1":{"Group 1":1, "Group 2":0, "Group 3":0},
"Group 2":{"Group 1":0, "Group 2":1, "Group 3":0},
"Group 3":{"Group 1":0, "Group 2":0, "Group 3":1}
},
3: {
"Group 1":{"Group 1":0.25, "Group 2":0, "Group 3":0.75},
"Group 2":{"Group 1":0.25, "Group 2":0.75, "Group 3":0}
},
4: {
"Group 1":{"Group 1":1, "Group 2":0},
"Group 2":{"Group 1":0.25, "Group 2":0.75}
}
}

Esto es lo que sucedió cuando hice el gráfico en networkx :

import networkx
import matplotlib.pyplot as plt
# Create Directed Graph
G = nx.DiGraph()
# Iterate through all connections
for iter_n, D_current_previous in D_iter_current_previous.items():
for current_group, D_previous_weights in D_current_previous.items():
for previous_group, weight in D_previous_weights.items():
if weight > 0:
# Define connections using `|__|` as a delimiter for the names
previous_node = "%d|__|%s"%(iter_n - 1, previous_group)
current_node = "%d|__|%s"%(iter_n, current_group)
connection = (previous_node, current_node)
G.add_edge(*connection, weight=weight)
# Draw Graph with labels and width thickness
nx.draw(G, with_labels=True, width=[G[u][v][''weight''] for u,v in G.edges()])

Nota: la única otra manera, se me ocurre hacer esto sería en matplotlib creando un diagrama de dispersión con cada tic que representa una iteración (5 incluyendo las muestras iniciales) y luego conectando los puntos entre sí con diferentes pesos. Esto sería un código bastante desordenado, especialmente tratando de alinear los bordes de los marcadores con las conexiones ... Sin embargo, no estoy seguro si esto y networkx es la mejor manera de hacerlo o si hay una herramienta (por ejemplo, bokeh o plotly ) que está diseñado para este tipo de trazado.

Answer 1

Networkx tiene instalaciones de trazado decentes para el análisis de datos exploratorios, no es la herramienta para hacer cifras de calidad de publicación, por varias razones por las que no quiero entrar aquí. Por lo tanto, reescribí esa parte del código base desde cero, e hice un módulo de dibujo independiente llamado netgraph que se puede encontrar here (como el original puramente basado en matplotlib). El API es muy, muy similar y está bien documentado, por lo que no debería ser demasiado difícil adaptarlo a sus propósitos.

A partir de eso obtengo el siguiente resultado:

Elegí el color para denotar la fuerza del borde como puedas
1) indicar valores negativos, y
2) Distinguir mejor los valores pequeños.
Sin embargo, también puede pasar un ancho de borde a netgraph (consulte netgraph.draw_edges() ).

El orden diferente de las ramas es el resultado de su estructura de datos (un dict), que indica que no hay un orden inherente. Tendría que modificar su estructura de datos y la función _parse_input() continuación para solucionar ese problema.

Código:

import itertools
import numpy as np
import matplotlib.pyplot as plt
import netgraph; reload(netgraph)
def plot_layered_network(weight_matrices,
distance_between_layers=2,
distance_between_nodes=1,
layer_labels=None,
**kwargs):
"""
Convenience function to plot layered network.
Arguments:
----------
weight_matrices: [w1, w2, ..., wn]
list of weight matrices defining the connectivity between layers;
each weight matrix is a 2-D ndarray with rows indexing source and columns indexing targets;
the number of sources has to match the number of targets in the last layer
distance_between_layers: int
distance_between_nodes: int
layer_labels: [str1, str2, ..., strn+1]
labels of layers
**kwargs: passed to netgraph.draw()
Returns:
--------
ax: matplotlib axis instance
"""
nodes_per_layer = _get_nodes_per_layer(weight_matrices)
node_positions = _get_node_positions(nodes_per_layer,
distance_between_layers,
distance_between_nodes)
w = _combine_weight_matrices(weight_matrices, nodes_per_layer)
ax = netgraph.draw(w, node_positions, **kwargs)
if not layer_labels is None:
ax.set_xticks(distance_between_layers*np.arange(len(weight_matrices)+1))
ax.set_xticklabels(layer_labels)
ax.xaxis.set_ticks_position(''bottom'')
return ax
def _get_nodes_per_layer(weight_matrices):
nodes_per_layer = []
for w in weight_matrices:
sources, targets = w.shape
nodes_per_layer.append(sources)
nodes_per_layer.append(targets)
return nodes_per_layer
def _get_node_positions(nodes_per_layer,
distance_between_layers,
distance_between_nodes):
x = []
y = []
for ii, n in enumerate(nodes_per_layer):
x.append(distance_between_nodes * np.arange(0., n))
y.append(ii * distance_between_layers * np.ones((n)))
x = np.concatenate(x)
y = np.concatenate(y)
return np.c_[y,x]
def _combine_weight_matrices(weight_matrices, nodes_per_layer):
total_nodes = np.sum(nodes_per_layer)
w = np.full((total_nodes, total_nodes), np.nan, np.float)
a = 0
b = nodes_per_layer[0]
for ii, ww in enumerate(weight_matrices):
w[a:a+ww.shape[0], b:b+ww.shape[1]] = ww
a += nodes_per_layer[ii]
b += nodes_per_layer[ii+1]
return w
def test():
w1 = np.random.rand(4,5) #< 0.50
w2 = np.random.rand(5,6) #< 0.25
w3 = np.random.rand(6,3) #< 0.75
import string
node_labels = dict(zip(range(18), list(string.ascii_lowercase)))
fig, ax = plt.subplots(1,1)
plot_layered_network([w1,w2,w3],
layer_labels=[''start'', ''step 1'', ''step 2'', ''finish''],
ax=ax,
node_size=20,
node_edge_width=2,
node_labels=node_labels,
edge_width=5,
)
plt.show()
return
def test_example(input_dict):
weight_matrices, node_labels = _parse_input(input_dict)
fig, ax = plt.subplots(1,1)
plot_layered_network(weight_matrices,
layer_labels=['''', ''1'', ''2'', ''3'', ''4''],
distance_between_layers=10,
distance_between_nodes=8,
ax=ax,
node_size=300,
node_edge_width=10,
node_labels=node_labels,
edge_width=50,
)
plt.show()
return
def _parse_input(input_dict):
weight_matrices = []
node_labels = []
# initialise sources
sources = set()
for v in input_dict[1].values():
for s in v.keys():
sources.add(s)
sources = list(sources)
for ii in range(len(input_dict)):
inner_dict = input_dict[ii+1]
targets = inner_dict.keys()
w = np.full((len(sources), len(targets)), np.nan, np.float)
for ii, s in enumerate(sources):
for jj, t in enumerate(targets):
try:
w[ii,jj] = inner_dict[t][s]
except KeyError:
pass
weight_matrices.append(w)
node_labels.append(sources)
sources = targets
node_labels.append(targets)
node_labels = list(itertools.chain.from_iterable(node_labels))
node_labels = dict(enumerate(node_labels))
return weight_matrices, node_labels
# --------------------------------------------------------------------------------
# script
# --------------------------------------------------------------------------------
if __name__ == "__main__":
# test()
input_dict = {
1: {
"Group 1":{"sample_0":0.5, "sample_1":0.5, "sample_2":0, "sample_3":0, "sample_4":0},
"Group 2":{"sample_0":0, "sample_1":0, "sample_2":1, "sample_3":0, "sample_4":0},
"Group 3":{"sample_0":0, "sample_1":0, "sample_2":0, "sample_3":0.5, "sample_4":0.5}
},
2: {
"Group 1":{"Group 1":1, "Group 2":0, "Group 3":0},
"Group 2":{"Group 1":0, "Group 2":1, "Group 3":0},
"Group 3":{"Group 1":0, "Group 2":0, "Group 3":1}
},
3: {
"Group 1":{"Group 1":0.25, "Group 2":0, "Group 3":0.75},
"Group 2":{"Group 1":0.25, "Group 2":0.75, "Group 3":0}
},
4: {
"Group 1":{"Group 1":1, "Group 2":0},
"Group 2":{"Group 1":0.25, "Group 2":0.75}
}
}
test_example(input_dict)
pass