Skip to main content
`Ctrl`+`K`

# Source code for networkx.algorithms.shortest_paths.astar

    """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:]))