scgraph.spanning

  1from heapq import heappush, heappop
  2
  3
  4class SpanningTree:
  5    @staticmethod
  6    def makowskis_spanning_tree(graph: list[dict], node_id: int) -> dict:
  7        """
  8        Function:
  9
 10        - Calculate the shortest path tree of a graph using Makowski's modified Dijkstra algorithm
 11            - Modifications allow for a sparse distance matrix to be used instead of a dense distance matrix
 12            - Improvements include only computing future potential nodes based on the open leaves for each branch
 13                - Open leaves are nodes that have not been visited yet but are adjacent to other visited nodes
 14            - This can dramatically reduce the memory and compute requirements of the algorithm
 15            - For particularly sparse graphs, this algorithm runs close to O(n log n) time
 16                - Where n is the number of nodes in the graph
 17            - For dense graphs, this algorithm runs closer to O(n^2) time (similar to the standard Dijkstra algorithm)
 18                - Where n is the number of nodes in the graph
 19
 20        Required Arguments:
 21
 22        - `graph`:
 23            - Type: list of dictionaries
 24            - See: https://connor-makowski.github.io/scgraph/scgraph/graph.html#Graph.validate_graph
 25        - `node_id`
 26            - Type: int
 27            - What: The id of the node from which to calculate the shortest path tree
 28
 29        Returns:
 30
 31        - A dictionary with the following keys:
 32            - `node_id`: The id of the node from which the shortest path tree was calculated
 33            - `predecessors`: A list of node ids referring to the predecessor of each node given the shortest path tree
 34                - Note: For disconnected graphs, nodes that are not connected to the origin node will have a predecessor of None
 35            - `distance_matrix`: A list of distances from the origin node to each node in the graph
 36                - Note: For disconnected graphs, nodes that are not connected to the origin node will have a distance of float("inf")
 37
 38        """
 39        # Input Validation
 40        assert isinstance(node_id, int), "node_id must be an integer"
 41        assert 0 <= node_id < len(graph), "node_id must be a valid node id"
 42        # Variable Initialization
 43        distance_matrix = [float("inf")] * len(graph)
 44        distance_matrix[node_id] = 0
 45        open_leaves = []
 46        heappush(open_leaves, (0, node_id))
 47        predecessor = [-1] * len(graph)
 48
 49        while open_leaves:
 50            current_distance, current_id = heappop(open_leaves)
 51            for connected_id, connected_distance in graph[current_id].items():
 52                possible_distance = current_distance + connected_distance
 53                if possible_distance < distance_matrix[connected_id]:
 54                    distance_matrix[connected_id] = possible_distance
 55                    predecessor[connected_id] = current_id
 56                    heappush(open_leaves, (possible_distance, connected_id))
 57
 58        return {
 59            "node_id": node_id,
 60            "predecessors": predecessor,
 61            "distance_matrix": distance_matrix,
 62        }
 63
 64    @staticmethod
 65    def get_path(
 66        origin_id: int,
 67        destination_id: int,
 68        spanning_tree: dict,
 69        length_only: bool = False,
 70    ) -> dict:
 71        """
 72        Function:
 73
 74        - Get the path from the origin node to the destination node using the provided spanning tree
 75        - Return a list of node ids in the order they are visited as well as the length of the path
 76        - If the destination node is not reachable from the origin node, return None
 77
 78        Required Arguments:
 79
 80        - `origin_id`
 81            - Type: int
 82            - What: The id of the origin node from the graph dictionary to start the shortest path from
 83        - `destination_id`
 84            - Type: int
 85            - What: The id of the destination node from the graph dictionary to end the shortest path at
 86        - `spanning_tree`
 87            - Type: dict
 88            - What: The output from the makowskis_spanning_tree function
 89
 90        Optional Arguments:
 91
 92        - `length_only`
 93            - Type: bool
 94            - What: If True, only returns the length of the path
 95            - Default: False
 96
 97        Returns:
 98
 99        - A dictionary with the following keys:
100            - `path`: A list of node ids in the order they are visited from the origin node to the destination node
101            - `length`: The length of the path from the origin node to the destination node
102        """
103        if spanning_tree["node_id"] != origin_id:
104            raise Exception(
105                "The origin node must be the same as the spanning node for this function to work."
106            )
107        destination_distance = spanning_tree["distance_matrix"][destination_id]
108        if destination_distance == float("inf"):
109            raise Exception(
110                "Something went wrong, the origin and destination nodes are not connected."
111            )
112        if length_only:
113            return {"length": destination_distance}
114        current_id = destination_id
115        current_path = [destination_id]
116        while current_id != origin_id:
117            current_id = spanning_tree["predecessors"][current_id]
118            current_path.append(current_id)
119        current_path.reverse()
120        return {
121            "path": current_path,
122            "length": destination_distance,
123        }