"""Shortest paths and path lengths using the A* ("A star") algorithm.""" from heapq import heappop, heappush from itertools import count import networkx as nx from networkx.algorithms.shortest_paths.weighted import _weight_function __all__ = ["astar_path", "astar_path_length"] @nx._dispatchable(edge_attrs="weight", preserve_node_attrs="heuristic") def astar_path(G, source, target, heuristic=None, weight="weight", *, cutoff=None): """Returns a list of nodes in a shortest path between source and target using the A* ("A-star") algorithm. There may be more than one shortest path. This returns only one. Parameters ---------- G : NetworkX graph source : node Starting node for path target : node Ending node for path heuristic : function A function to evaluate the estimate of the distance from the a node to the target. The function takes two nodes arguments and must return a number. If the heuristic is inadmissible (if it might overestimate the cost of reaching the goal from a node), the result may not be a shortest path. The algorithm does not support updating heuristic values for the same node due to caching the first heuristic calculation per node. weight : string or function If this is a string, then edge weights will be accessed via the edge attribute with this key (that is, the weight of the edge joining `u` to `v` will be ``G.edges[u, v][weight]``). If no such edge attribute exists, the weight of the edge is assumed to be one. If this is a function, the weight of an edge is the value returned by the function. The function must accept exactly three positional arguments: the two endpoints of an edge and the dictionary of edge attributes for that edge. The function must return a number or None to indicate a hidden edge. cutoff : float, optional If this is provided, the search will be bounded to this value. I.e. if the evaluation function surpasses this value for a node n, the node will not be expanded further and will be ignored. More formally, let h'(n) be the heuristic function, and g(n) be the cost of reaching n from the source node. Then, if g(n) + h'(n) > cutoff, the node will not be explored further. Note that if the heuristic is inadmissible, it is possible that paths are ignored even though they satisfy the cutoff. Raises ------ NetworkXNoPath If no path exists between source and target. Examples -------- >>> G = nx.path_graph(5) >>> print(nx.astar_path(G, 0, 4)) [0, 1, 2, 3, 4] >>> G = nx.grid_graph(dim=[3, 3]) # nodes are two-tuples (x,y) >>> nx.set_edge_attributes(G, {e: e[1][0] * 2 for e in G.edges()}, "cost") >>> def dist(a, b): ... (x1, y1) = a ... (x2, y2) = b ... return ((x1 - x2) ** 2 + (y1 - y2) ** 2) ** 0.5 >>> print(nx.astar_path(G, (0, 0), (2, 2), heuristic=dist, weight="cost")) [(0, 0), (0, 1), (0, 2), (1, 2), (2, 2)] Notes ----- Edge weight attributes must be numerical. Distances are calculated as sums of weighted edges traversed. The weight function can be used to hide edges by returning None. So ``weight = lambda u, v, d: 1 if d['color']=="red" else None`` will find the shortest red path. See Also -------- shortest_path, dijkstra_path """ if source not in G: raise nx.NodeNotFound(f"Source {source} is not in G") if target not in G: raise nx.NodeNotFound(f"Target {target} is not in G") if heuristic is None: # The default heuristic is h=0 - same as Dijkstra's algorithm def heuristic(u, v): return 0 push = heappush pop = heappop weight = _weight_function(G, weight) G_succ = G._adj # For speed-up (and works for both directed and undirected graphs) # The queue stores priority, node, cost to reach, and parent. # Uses Python heapq to keep in priority order. # Add a counter to the queue to prevent the underlying heap from # attempting to compare the nodes themselves. The hash breaks ties in the # priority and is guaranteed unique for all nodes in the graph. c = count() queue = [(0, next(c), source, 0, None)] # Maps enqueued nodes to distance of discovered paths and the # computed heuristics to target. We avoid computing the heuristics # more than once and inserting the node into the queue too many times. enqueued = {} # Maps explored nodes to parent closest to the source. explored = {} while queue: # Pop the smallest item from queue. _, __, curnode, dist, parent = pop(queue) if curnode == target: path = [curnode] node = parent while node is not None: path.append(node) node = explored[node] path.reverse() return path if curnode in explored: # Do not override the parent of starting node if explored[curnode] is None: continue # Skip bad paths that were enqueued before finding a better one qcost, h = enqueued[curnode] if qcost < dist: continue explored[curnode] = parent for neighbor, w in G_succ[curnode].items(): cost = weight(curnode, neighbor, w) if cost is None: continue ncost = dist + cost if neighbor in enqueued: qcost, h = enqueued[neighbor] # if qcost <= ncost, a less costly path from the # neighbor to the source was already determined. # Therefore, we won't attempt to push this neighbor # to the queue if qcost <= ncost: continue else: h = heuristic(neighbor, target) if cutoff and ncost + h > cutoff: continue enqueued[neighbor] = ncost, h push(queue, (ncost + h, next(c), neighbor, ncost, curnode)) raise nx.NetworkXNoPath(f"Node {target} not reachable from {source}") @nx._dispatchable(edge_attrs="weight", preserve_node_attrs="heuristic") def astar_path_length( G, source, target, heuristic=None, weight="weight", *, cutoff=None ): """Returns the length of the shortest path between source and target using the A* ("A-star") algorithm. Parameters ---------- G : NetworkX graph source : node Starting node for path target : node Ending node for path heuristic : function A function to evaluate the estimate of the distance from the a node to the target. The function takes two nodes arguments and must return a number. If the heuristic is inadmissible (if it might overestimate the cost of reaching the goal from a node), the result may not be a shortest path. The algorithm does not support updating heuristic values for the same node due to caching the first heuristic calculation per node. weight : string or function If this is a string, then edge weights will be accessed via the edge attribute with this key (that is, the weight of the edge joining `u` to `v` will be ``G.edges[u, v][weight]``). If no such edge attribute exists, the weight of the edge is assumed to be one. If this is a function, the weight of an edge is the value returned by the function. The function must accept exactly three positional arguments: the two endpoints of an edge and the dictionary of edge attributes for that edge. The function must return a number or None to indicate a hidden edge. cutoff : float, optional If this is provided, the search will be bounded to this value. I.e. if the evaluation function surpasses this value for a node n, the node will not be expanded further and will be ignored. More formally, let h'(n) be the heuristic function, and g(n) be the cost of reaching n from the source node. Then, if g(n) + h'(n) > cutoff, the node will not be explored further. Note that if the heuristic is inadmissible, it is possible that paths are ignored even though they satisfy the cutoff. Raises ------ NetworkXNoPath If no path exists between source and target. See Also -------- astar_path """ if source not in G or target not in G: msg = f"Either source {source} or target {target} is not in G" raise nx.NodeNotFound(msg) weight = _weight_function(G, weight) path = astar_path(G, source, target, heuristic, weight, cutoff=cutoff) return sum(weight(u, v, G[u][v]) for u, v in zip(path[:-1], path[1:]))