105 lines
3.3 KiB
Python
105 lines
3.3 KiB
Python
|
from itertools import chain
|
||
|
|
|
||
|
|
import networkx as nx
|
||
|
|
from networkx.utils import not_implemented_for, pairwise
|
||
|
|
|
||
|
|
__all__ = ["metric_closure", "steiner_tree"]
|
||
|
|
|
||
|
|
|
||
|
|
@not_implemented_for("directed")
|
||
|
|
def metric_closure(G, weight="weight"):
|
||
|
|
"""Return the metric closure of a graph.
|
||
|
|
|
||
|
|
The metric closure of a graph *G* is the complete graph in which each edge
|
||
|
|
is weighted by the shortest path distance between the nodes in *G* .
|
||
|
|
|
||
|
|
Parameters
|
||
|
|
----------
|
||
|
|
G : NetworkX graph
|
||
|
|
|
||
|
|
Returns
|
||
|
|
-------
|
||
|
|
NetworkX graph
|
||
|
|
Metric closure of the graph `G`.
|
||
|
|
|
||
|
|
"""
|
||
|
|
M = nx.Graph()
|
||
|
|
|
||
|
|
Gnodes = set(G)
|
||
|
|
|
||
|
|
# check for connected graph while processing first node
|
||
|
|
all_paths_iter = nx.all_pairs_dijkstra(G, weight=weight)
|
||
|
|
u, (distance, path) = next(all_paths_iter)
|
||
|
|
if Gnodes - set(distance):
|
||
|
|
msg = "G is not a connected graph. metric_closure is not defined."
|
||
|
|
raise nx.NetworkXError(msg)
|
||
|
|
Gnodes.remove(u)
|
||
|
|
for v in Gnodes:
|
||
|
|
M.add_edge(u, v, distance=distance[v], path=path[v])
|
||
|
|
|
||
|
|
# first node done -- now process the rest
|
||
|
|
for u, (distance, path) in all_paths_iter:
|
||
|
|
Gnodes.remove(u)
|
||
|
|
for v in Gnodes:
|
||
|
|
M.add_edge(u, v, distance=distance[v], path=path[v])
|
||
|
|
|
||
|
|
return M
|
||
|
|
|
||
|
|
|
||
|
|
@not_implemented_for("directed")
|
||
|
|
def steiner_tree(G, terminal_nodes, weight="weight"):
|
||
|
|
"""Return an approximation to the minimum Steiner tree of a graph.
|
||
|
|
|
||
|
|
The minimum Steiner tree of `G` w.r.t a set of `terminal_nodes`
|
||
|
|
is a tree within `G` that spans those nodes and has minimum size
|
||
|
|
(sum of edge weights) among all such trees.
|
||
|
|
|
||
|
|
The minimum Steiner tree can be approximated by computing the minimum
|
||
|
|
spanning tree of the subgraph of the metric closure of *G* induced by the
|
||
|
|
terminal nodes, where the metric closure of *G* is the complete graph in
|
||
|
|
which each edge is weighted by the shortest path distance between the
|
||
|
|
nodes in *G* .
|
||
|
|
This algorithm produces a tree whose weight is within a (2 - (2 / t))
|
||
|
|
factor of the weight of the optimal Steiner tree where *t* is number of
|
||
|
|
terminal nodes.
|
||
|
|
|
||
|
|
Parameters
|
||
|
|
----------
|
||
|
|
G : NetworkX graph
|
||
|
|
|
||
|
|
terminal_nodes : list
|
||
|
|
A list of terminal nodes for which minimum steiner tree is
|
||
|
|
to be found.
|
||
|
|
|
||
|
|
Returns
|
||
|
|
-------
|
||
|
|
NetworkX graph
|
||
|
|
Approximation to the minimum steiner tree of `G` induced by
|
||
|
|
`terminal_nodes` .
|
||
|
|
|
||
|
|
Notes
|
||
|
|
-----
|
||
|
|
For multigraphs, the edge between two nodes with minimum weight is the
|
||
|
|
edge put into the Steiner tree.
|
||
|
|
|
||
|
|
|
||
|
|
References
|
||
|
|
----------
|
||
|
|
.. [1] Steiner_tree_problem on Wikipedia.
|
||
|
|
https://en.wikipedia.org/wiki/Steiner_tree_problem
|
||
|
|
"""
|
||
|
|
# H is the subgraph induced by terminal_nodes in the metric closure M of G.
|
||
|
|
M = metric_closure(G, weight=weight)
|
||
|
|
H = M.subgraph(terminal_nodes)
|
||
|
|
# Use the 'distance' attribute of each edge provided by M.
|
||
|
|
mst_edges = nx.minimum_spanning_edges(H, weight="distance", data=True)
|
||
|
|
# Create an iterator over each edge in each shortest path; repeats are okay
|
||
|
|
edges = chain.from_iterable(pairwise(d["path"]) for u, v, d in mst_edges)
|
||
|
|
# For multigraph we should add the minimal weight edge keys
|
||
|
|
if G.is_multigraph():
|
||
|
|
edges = (
|
||
|
|
(u, v, min(G[u][v], key=lambda k: G[u][v][k][weight])) for u, v in edges
|
||
|
|
)
|
||
|
|
T = G.edge_subgraph(edges)
|
||
|
|
return T
|