Need help? Have a feature request? Please check out the
tethne-users group
.
Source code for tethne.classes.graphcollection
"""
A :class:`.GraphCollection` is a set of graphs generated from a
:class:`.Corpus` or model.
"""
import networkx as nx
from collections import defaultdict
import warnings
from tethne import networks
from tethne.utilities import _iterable
import sys
PYTHON_3 = sys.version_info[0] == 3
if PYTHON_3:
unicode = str
xrange = range
str = bytes
[docs]class GraphCollection(dict):
"""
A :class:`.GraphCollection` is an indexed set of
:ref:`networkx.Graph <networkx:graph>`\s.
When you add a :ref:`networkx.Graph <networkx:graph>`, the nodes are indexed
and relabeled.
.. code-block:: python
>>> from tethne import GraphCollection
>>> import networkx as nx
>>> G = GraphCollection()
>>> g = nx.Graph()
>>> g.add_node('A', yes='no')
>>> g.add_edge('A', 'B', c='d')
>>> G['graph1'] = g # You can also use G.add('graph1', g)
>>> G.graph1.nodes(data=True)
[(0, {}), (1, {'yes': 'no'})]
>>> G.node_index, G.node_lookup
({0: 'B', 1: 'A', -1: None}, {'A': 1, None: -1, 'B': 0})
To build a :class:`.GraphCollection` from a :class:`.Corpus`, pass it and
a method to the constructor, or use :meth:`.GraphCollection.build`\.
.. code-block:: python
>>> corpus = read(datapath)
>>> G = GraphCollection(corpus, coauthors)
>>> G.build(corpus, authors)
"""
def __init__(self, corpus=None, method=None, slice_kwargs={},
method_kwargs={}, directed=False):
"""
Parameters
----------
corpus : :class:`.Corpus`
method : str or func
If str, looks for ``method`` in the ``tethne`` namespace.
slice_kwargs : dict
Keyword arguments to pass to ``corpus``' ``slice`` method.
method_kwargs : dict
Keyword arguments to pass to ``method`` along with ``corpus``.
directed : bool
If True, graphs will be treated as directed during indexing.
"""
self.directed = directed
if directed:
self.master_graph = nx.MultiDiGraph()
else:
self.master_graph = nx.MultiGraph()
self.node_index = {-1: None}
self.node_lookup = {None: -1}
self.graphs_containing = defaultdict(list)
if corpus and method:
self.build(corpus, method, slice_kwargs, method_kwargs)
def __setitem__(self, name, graph):
self.add(name, graph)
def __getattr__(self, name):
if name in self:
return self[name]
try: # hasttr() causes endless recursion in Python 3.x
return object.__getattr__(self, name)
except AttributeError:
pass
raise AttributeError('GraphCollection has no such attribute or graph.')
[docs] def build(self, corpus, method, slice_kwargs={}, method_kwargs={}):
"""
Generate a set of :ref:`networkx.Graph <networkx:graph>`\s using
``method`` on the slices in ``corpus``\.
Parameters
----------
corpus : :class:`.Corpus`
method : str or func
If str, looks for ``method`` in the ``tethne`` namespace.
slice_kwargs : dict
Keyword arguments to pass to ``corpus``' ``slice`` method.
method_kwargs : dict
Keyword arguments to pass to ``method`` along with ``corpus``.
"""
if not hasattr(method, '__call__'):
if not hasattr(networks, method):
raise NameError('No such method')
method = getattr(networks, method)
for key, subcorpus in corpus.slice(**slice_kwargs):
graph = method(subcorpus, **method_kwargs)
self.add(key, graph)
[docs] def add(self, name, graph):
"""
Index and add a :ref:`networkx.Graph <networkx:graph>` to the
:class:`.GraphCollection`.
Parameters
----------
name : hashable
Unique name used to identify the `graph`.
graph : :ref:`networkx.Graph <networkx:graph>`
Raises
------
ValueError
If `name` has already been used in this :class:`.GraphCollection`\.
"""
if name in self:
raise ValueError("{0} exists in this GraphCollection".format(name))
elif hasattr(self, unicode(name)):
raise ValueError("Name conflicts with an existing attribute")
indexed_graph = self.index(name, graph)
# Add all edges to the `master_graph`.
for s, t, attrs in indexed_graph.edges(data=True):
attrs.update({'graph': name})
self.master_graph.add_edge(s, t, **attrs)
# Add all node attributes to the `master_graph`.
for n, attrs in indexed_graph.nodes(data=True):
for k,v in attrs.iteritems():
if k not in self.master_graph.node[n]:
self.master_graph.node[n][k] = {}
self.master_graph.node[n][k][name] = v
dict.__setitem__(self, name, indexed_graph)
[docs] def index(self, name, graph):
"""
Index any new nodes in `graph`, and relabel the nodes in `graph` using
the index.
Parameters
----------
name : hashable
Unique name used to identify the `graph`.
graph : networkx.Graph
Returns
-------
indexed_graph : networkx.Graph
"""
nodes = graph.nodes()
# Index new nodes.
new_nodes = list(set(nodes) - set(self.node_index.values()))
start = max(len(self.node_index) - 1, max(self.node_index.keys()))
for i in xrange(start, start + len(new_nodes)):
n = new_nodes.pop()
self.node_index[i], self.node_lookup[n] = n, i
self.graphs_containing[n].append(name)
# Relabel nodes in `graph`.
new_labels = {n: self.node_lookup[n] for n in nodes}
indexed_graph = nx.relabel.relabel_nodes(graph, new_labels, copy=True)
return indexed_graph
[docs] def nodes(self, data=False, native=True):
"""
Returns a list of all nodes in the :class:`.GraphCollection`\.
Parameters
----------
data : bool
(default: False) If True, returns a list of 2-tuples containing
node labels and attributes.
Returns
-------
nodes : list
"""
nodes = self.master_graph.nodes(data=data)
if native:
if data:
nodes = [(self.node_index[n], attrs) for n, attrs in nodes]
else:
nodes = [self.node_index[n] for n in nodes]
return nodes
[docs] def edges(self, data=False, native=True):
"""
Returns a list of all edges in the :class:`.GraphCollection`\.
Parameters
----------
data : bool
(default: False) If True, returns a list of 3-tuples containing
source and target node labels, and attributes.
Returns
-------
edges : list
"""
edges = self.master_graph.edges(data=data)
if native:
if data:
edges = [(self.node_index[s], self.node_index[t], attrs)
for s, t, attrs in edges]
else:
edges = [(self.node_index[s], self.node_index[t])
for s, t in edges]
return edges
[docs] def order(self, piecewise=False):
"""
Returns the total number of nodes in the :class:`.GraphCollection`\.
"""
if piecewise:
return {k: v.order() for k, v in self.items()}
return self.master_graph.order()
[docs] def node_distribution(self):
warnings.warn("node_distribution will be removed in v0.8. Use" +
" order(piecewise=True) instead.", DeprecationWarning)
return self.order(piecewise=True)
[docs] def size(self, piecewise=False):
"""
Returns the total number of edges in the :class:`.GraphCollection`\.
"""
if piecewise:
return {k: v.size() for k, v in self.items()}
return self.master_graph.size()
[docs] def edge_distribution(self):
warnings.warn("edge_distribution will be removed in v0.8. Use" +
" size(piecewise=True) instead.", DeprecationWarning)
return self.order(piecewise=True)
[docs] def collapse(self, weight_attr='weight'):
"""
Returns a :ref:`networkx.Graph <networkx:graph>` or :class:`networkx.DiGraph` in which
the edges between each pair of nodes are collapsed into a single
weighted edge.
"""
if self.directed:
graph = nx.DiGraph()
else:
graph = nx.Graph()
# Transfer all nodes and attributes.
for n, attrs in self.master_graph.nodes(data=True):
graph.add_node(n, attrs)
for s, t, attrs in self.master_graph.edges(data=True):
if not graph.has_edge(s, t):
graph.add_edge(s, t)
if 'weight' not in graph[s][t]:
graph[s][t]['weight'] = 0.
if weight_attr in attrs:
graph[s][t]['weight'] += attrs[weight_attr]
else:
graph[s][t]['weight'] += 1.
gname = attrs['graph']
for k, v in attrs.iteritems():
if k in [weight_attr, 'graph']:
continue
if k not in graph[s][t]:
graph[s][t][k] = {}
graph[s][t][k][gname] = v
return graph
[docs] def analyze(self, method_name, mapper=map, invert=False, **kwargs):
"""
Apply a method from NetworkX to each of the graphs in the
:class:`.GraphCollection`\.
Examples
--------
.. code-block:: python
>>> G.analyze('betweenness_centrality')
{'test': {0: 1.0, 1: 0.0, 2: 0.0},
'test2': {0: 0.0, 1: 0.0, 2: 0.0, 3: 0.0}}
>>> G.analyze('betweenness_centrality', results_by='node')
{0: {'test': 1.0, 'test2': 0.0},
1: {'test': 0.0, 'test2': 0.0},
2: {'test': 0.0, 'test2': 0.0},
3: {'test2': 0.0}}
Parameters
----------
method : str or iterable
Must be the name of a method accessible directly from the
`networkx` namespace. If an iterable, should be the complete
dot-path to the method, e.g. ``nx.connected.is_connected`` would be
written as ``['connected', 'is_connected']``.
mapper : func
A mapping function. Be default uses Python's builtin ``map``
function. MUST return results in order.
results_by : str
(default: 'graph'). By default, the top-level key in the results
are graph names. If results_by='node', node labels are used as
top-level keys.
kwargs : kwargs
Any additional kwargs are passed to the NetworkX method.
Returns
-------
dict
"""
# Find the analysis method, if possible.
if hasattr(method_name, '__iter__'):
mpath = method_name
if type(mpath) in [str, unicode]:
mpath = [mpath]
root = nx
while len(mpath) > 0:
head = getattr(root, mpath.pop(0))
if hasattr(head, '__call__'):
method = head
else:
raise NameError("No matching method in NetworkX")
elif hasattr(nx, method_name):
method = getattr(nx, method_name)
else:
raise AttributeError('No such method in NetworkX')
# Farm out the analysis using ``mapper``. This allows us to use
# multiprocessing in the future, or to add pre- or post-processing
# routines.
keys, graphs = zip(*self.items())
results = mapper(method, graphs, **kwargs)
# Group the results by graph.
by_graph = dict(zip(keys, results))
# Invert results.
inverse = defaultdict(dict)
for gname, result in by_graph.iteritems():
if hasattr(result, '__iter__'):
for n, val in result.iteritems():
inverse[n].update({gname: val})
if type(list(by_graph.values())[0]) is dict:
# Look for a result set that we can inspect.
i = 0
while True:
if len(by_graph.values()[i]) > 0:
inspect = by_graph.values()[i]
break
i += 1
if type(list(inspect.keys())[0]) is tuple:
# Results correspond to edges.
by_edge = dict(inverse)
# Set edge attributes in each graph.
for graph, attrs in by_graph.iteritems():
nx.set_edge_attributes(self[graph], method_name, attrs)
# Set edge attributes in the master graph.
for (s, t), v in by_edge.iteritems():
for i, attrs in self.master_graph.edge[s][t].iteritems():
val = v[attrs['graph']]
self.master_graph.edge[s][t][i][method_name] = val
if invert:
return by_edge
else:
# Results correspond to nodes.
by_node = dict(inverse)
# Set node attributes for each graph.
for graph, attrs in by_graph.iteritems():
nx.set_node_attributes(self[graph], method_name, attrs)
# Store node attributes in the master graph.
nx.set_node_attributes(self.master_graph, method_name, by_node)
if invert:
return by_node
return by_graph
[docs] def node_history(self, node, attribute):
"""
Returns a dictionary of attribute values for each :ref:`networkx.Graph <networkx:graph>` in
the :class:`.GraphCollection` for a single node.
Parameters
----------
node : str
The node of interest.
attribute : str
The attribute of interest; e.g. 'betweenness_centrality'
Returns
-------
history : dict
"""
return self.master_graph.node[node][attribute]
[docs] def edge_history(self, source, target, attribute):
"""
Returns a dictionary of attribute vales for each Graph in the
:class:`.GraphCollection` for a single edge.
Parameters
----------
source : str
Identifier for source node.
target : str
Identifier for target node.
attribute : str
The attribute of interest; e.g. 'betweenness_centrality'
Returns
-------
history : dict
"""
return {attr['graph']: attr[attribute] for i, attr
in self.master_graph.edge[source][target].items()}
[docs] def union(self, weight_attr='_weight'):
"""
Returns the union of all graphs in this :class:`.GraphCollection`\.
The number of graphs in which an edge exists between each node pair `u` and `v`
is stored in the edge attribute given be `weight_attr` (default: `_weight`).
Parameters
----------
weight_attr : str
(default: '_weight') Name of the edge attribute used to store the number of
graphs in which an edge exists between node pairs.
Returns
-------
graph : :ref:`networkx.Graph <networkx:graph>`
"""
if type(self.master_graph) is nx.MultiDiGraph:
graph = nx.DiGraph()
else:
graph = nx.Graph()
edge_attrs = defaultdict(list)
for u, v, a in self.master_graph.edges(data=True):
if not graph.has_edge(u, v):
graph.add_edge(u, v)
graph[u][v]['graphs'] = []
graph[u][v][weight_attr] = 0.
for key, value in a.iteritems():
if key not in graph[u][v]:
graph[u][v][key] = []
graph[u][v][key].append(value)
graph[u][v]['graphs'].append(a['graph'])
graph[u][v][weight_attr] += 1.
for u, a in self.master_graph.nodes(data=True):
for key, value in a.iteritems():
graph.node[u][key] = value
return graph
