Find the critical path of PERT using breadth-first search and depth-first search
background
Atcoder has stopped moving from brown I started studying algorithms in earnest. Now that I have a little understanding of breadth-first search (BFS) and depth-first search (DFS) As a review, the questions of the Basic Information Technology Engineer Examination are also given. The earliest connection point time, latest connection point time, and critical path in PERT I decided to ask.
Purpose
By entering the information of the graph, in PERT ・ The earliest connection point time ・ The latest connection point time ·Critical path Implement a program that calculates.
About breadth-first search (BFS) and depth-first search (DFS)
@ drken's article was very easy to understand.
・ DFS (Depth-first Search) Super Introduction! ~ Entrance to the world of graph algorithms ~ [Part 1] ・ DFS (Depth-first Search) Super Introduction! ~ Entrance to the world of graph algorithms ~ [Part 2] ・ BFS (Breadth-first Search) Super Introduction! ~ Use the queue vividly ~
About PERT
By showing the order of work and dependencies on an arrow diagram, It is a method to clarify the shortest time required to complete the project and the most important management work. Please see the article by @sakamoto_t. ・ What is WBS?
Example
The following arrow diagram is used as an example.
Implementation code
main.py
from collections import deque
import numpy as np
import sys
def main():
node_num, path_num, start, goal, path = input_data()
Graph, Graph_critical, Graph_reverse, need_day = made_graph(node_num, path_num, path)
earliest_node_time = bfs_normal(node_num, start, Graph, need_day)
latest_node_time = bfs_reverse(node_num, goal, earliest_node_time, Graph_reverse, need_day)
candidate = check_critical_candidate(node_num, latest_node_time, earliest_node_time)
temp = list(check_critical_path(start, goal, candidate, Graph_critical))
critical_path = [str(path) for path in temp]
#-------------Output part-----------------
print('------------------')
print('Earliest_node_time')
print(earliest_node_time)
print('------------------')
print('Latest_node_time')
print(latest_node_time)
print('------------------')
print('Critical path')
print(' -> '.join(critical_path))
return 0
#---------------Input part--------------------
def input_data():
node_num, path_num = map(int, input().split())
start, goal = 1, node_num
path = [input().split() for i in range(path_num)]
return node_num, path_num, start, goal, path
#---------------Graph creation-------------------
def made_graph(node_num, path_num, path):
Graph = [deque([]) for i in range(node_num+1)]
Graph_critical = [deque([]) for i in range(node_num+1)]
Graph_reverse = [deque([]) for i in range(node_num+1)]
#Graph[0], Graph_reverse[0]Is throwing away((Because the vertex number is taken from 1 when creating the graph structure)
need_day = np.zeros((node_num, node_num))
for i in range(path_num):
Graph[int(path[i][0])].append(int(path[i][1]))
Graph_critical[int(path[i][0])].append(int(path[i][1]))
Graph_reverse[int(path[i][1])].append(int(path[i][0])) #Creating a reverse directed graph
need_day[int(path[i][0]) - 1][int(path[i][1]) - 1] = int(path[i][2])
need_day[int(path[i][1]) - 1][int(path[i][0]) - 1] = int(path[i][2])
return Graph, Graph_critical, Graph_reverse, need_day
#------------BFS working part(Find the connection point time anymore)------------------
def bfs_normal(node_num, start, Graph, need_day):
stack = deque([start])
earliest_node_time = [0 for i in range(node_num)]
while stack: #Continue until stack is empty
tmp = stack[0]
if Graph[tmp]:
w = Graph[tmp].popleft()
if earliest_node_time[w-1] < earliest_node_time[tmp-1] + need_day[tmp-1][w-1]:
earliest_node_time[w-1] = int(earliest_node_time[tmp-1] + need_day[tmp-1][w-1])
stack.append(w)
else:
stack.popleft()
return earliest_node_time
#------------BFS working part(Find the latest connection point time)------------------
def bfs_reverse(node_num, goal, earliest_node_time, Graph_reverse, need_day):
stack = deque([goal])
latest_node_time = [float('inf') for i in range(node_num)]
latest_node_time[goal-1] = earliest_node_time[goal-1]
while stack: #Continue until stack is empty
tmp = stack[0]
if Graph_reverse[tmp]:
w = Graph_reverse[tmp].popleft()
if latest_node_time[w-1] > latest_node_time[tmp-1] - need_day[tmp-1][w-1]:
latest_node_time[w-1] = int(latest_node_time[tmp-1] - need_day[tmp-1][w-1])
stack.append(w)
else:
stack.popleft()
return latest_node_time
#--------------EL =Extract the points that will be TL-------------------
def check_critical_candidate(node_num, latest_node_time, earliest_node_time):
critical_candidate = [False for i in range(node_num)]
for i in range(node_num):
if latest_node_time[i] == earliest_node_time[i]:
critical_candidate[i] = True
return critical_candidate
#--------------DFS work part (following the candidate points of the critical path and searching to the goal)-----------
def check_critical_path(start, goal, candidate, Graph_critical):
stack = [start]
seen = deque([start])
while stack: #Continue until stack is empty
tmp = stack[-1]
if Graph_critical[tmp]:
w = Graph_critical[tmp].popleft()
if w == goal:
seen.append(w)
return seen
elif candidate[w - 1]:
seen.append(w)
stack.append(w)
else:
continue
else:
stack.pop()
seen.pop()
return seen
if __name__ == '__main__':
sys.exit(main())
input
The node at the start of work is 1, and the node at the end of work is the maximum value of the node (= node_num). $ N $: Number of states (node_num) $ P $: Number of works (path_num) $ {s_i, g_i, K_i} $: Pre-work state, post-work state, required days As
In the example image above
Will be.
output
------------------
Earliest_node_time
[0, 17, 11, 24, 21, 30, 41]
------------------
Latest_node_time
[0, 18, 11, 25, 21, 30, 41]
------------------
Critical path
1 -> 3 -> 5 -> 6 -> 7
This is explained below.
Problem decomposition
To find the critical path
- Find the earliest connection point time (hereinafter, EL)
- Find the latest connection point time (TL)
- Extract candidates that can be nodes in the critical path where EL = TL Find an example of the critical path by performing a depth-first search under the condition of passing through the candidate node of 4.3
I thought about the implementation in 4 stages. I will explain each of them.
Create a graph
When finding the latest connection point time, the number of days required for each work is calculated from the information of the earliest connection point time of each node. It was necessary to move the arrows in the graph in the opposite direction, as we would be pulling and comparing. Therefore, when creating a graph, create a graph with the opposite direction of travel at the same time. I decided to put the number of working days in a matrix and store the opposite information at the same time.
#---------------Graph creation-------------------
def made_graph(node_num, path_num, path):
Graph = [deque([]) for i in range(node_num+1)]
Graph_critical = [deque([]) for i in range(node_num+1)]
Graph_reverse = [deque([]) for i in range(node_num+1)]
#Graph[0], Graph_reverse[0]Is throwing away((Because the vertex number is taken from 1 when creating the graph structure)
need_day = np.zeros((node_num, node_num))
for i in range(path_num):
Graph[int(path[i][0])].append(int(path[i][1]))
Graph_critical[int(path[i][0])].append(int(path[i][1]))
Graph_reverse[int(path[i][1])].append(int(path[i][0])) #Creating a reverse directed graph
need_day[int(path[i][0]) - 1][int(path[i][1]) - 1] = int(path[i][2])
need_day[int(path[i][1]) - 1][int(path[i][0]) - 1] = int(path[i][2])
return Graph, Graph_critical, Graph_reverse, need_day
"""
------------------
Path
[deque([2, 3]), deque([4]), deque([4, 5]), deque([6]), deque([6]), deque([7]), deque([])]
------------------
Reverse_Path
[deque([]), deque([1]), deque([1]), deque([2, 3]), deque([3]), deque([4, 5]), deque([6])]
------------------
Need_time
[[ 0. 17. 11. 0. 0. 0. 0.]
[17. 0. 0. 7. 0. 0. 0.]
[11. 0. 0. 5. 10. 0. 0.]
[ 0. 7. 5. 0. 0. 5. 0.]
[ 0. 0. 10. 0. 0. 9. 0.]
[ 0. 0. 0. 5. 9. 0. 11.]
[ 0. 0. 0. 0. 0. 11. 0.]]
"""
Find the earliest connection point time (ET)
This problem is replaced by the ** choosing the route that maximizes the sum of the edge weights on each node **. On the contrary, the ** Dijkstra method ** is famous as a path problem that minimizes the weight. This time, I implemented it using that idea.
#------------BFS working part(Find the connection point time anymore)------------------
def bfs_normal(start):
stack = deque([start])
while stack: #Continue until stack is empty
tmp = stack[0]
if Graph[tmp]:
w = Graph[tmp].popleft()
if earliest_node_time[w-1] < earliest_node_time[tmp-1] + need_day[tmp-1][w-1]:
earliest_node_time[w-1] = int(earliest_node_time[tmp-1] + need_day[tmp-1][w-1])
stack.append(w)
else:
stack.popleft()
return earliest_node_time
Breadth-first search is performed, and Earliest_node_time is updated by adding the number of working days. If there are multiple routes, update if it is larger than the existing value.
Find the latest connection point time (LT)
Next, find the LT. At this time, the initial value is infinite as Latest_node_time. Prepare an array and update it to the smaller one.
#------------BFS working part(Find the latest connection point time)------------------
def bfs_reverse(goal):
stack = deque([goal])
latest_node_time[goal-1] = earliest_node_time[goal-1]
while stack: #Continue until stack is empty
tmp = stack[0]
if Graph_reverse[tmp]:
w = Graph_reverse[tmp].popleft()
if latest_node_time[w-1] > latest_node_time[tmp-1] - need_day[tmp-1][w-1]:
latest_node_time[w-1] = int(latest_node_time[tmp-1] - need_day[tmp-1][w-1])
stack.append(w)
else:
stack.popleft()
return latest_node_time
Find the critical path
The critical path is made by connecting the nodes where EL = TL. Conversely, you have to follow the node that satisfies EL = TL to reach the goal, so ** On condition that it passes through a candidate node with EL = TL Depth-first search was performed, and the goal arrival problem was found from start. ** **
Extract candidate nodes with EL = TL by comparing each element.
#--------------EL =Extract the points that will be TL-------------------
def check_critical_candidate(node_num, latest_node_time, earliest_node_time):
critical_candidate = [False for i in range(node_num)]
for i in range(node_num):
if latest_node_time[i] == earliest_node_time[i]:
critical_candidate[i] = True
return critical_candidate
Finally, DFS determines if you can reach the goal. Determine if you are among the candidates when you reach each node. It ends when you reach the goal, so if you have multiple critical paths Display one of them.
#--------------DFS work part (following the candidate points of the critical path and searching to the goal)-----------
def check_critical_path(start, goal, candidate, Graph_critical):
stack = [start]
seen = deque([start])
while stack: #Continue until stack is empty
tmp = stack[-1]
if Graph_critical[tmp]:
w = Graph_critical[tmp].popleft()
if w == goal:
seen.append(w)
return seen
elif candidate[w - 1]:
seen.append(w)
stack.append(w)
else:
continue
else:
stack.pop()
seen.pop()
return seen
Main function part
def main():
node_num, path_num, start, goal, path = input_data()
Graph, Graph_critical, Graph_reverse, need_day = made_graph(node_num, path_num, path)
earliest_node_time = bfs_normal(node_num, start, Graph, need_day)
latest_node_time = bfs_reverse(node_num, goal, earliest_node_time, Graph_reverse, need_day)
candidate = check_critical_candidate(node_num, latest_node_time, earliest_node_time)
temp = list(check_critical_path(start, goal, candidate, Graph_critical))
critical_path = [str(path) for path in temp]
#-------------Output part-----------------
print('------------------')
print('Earliest_node_time')
print(earliest_node_time)
print('------------------')
print('Latest_node_time')
print(latest_node_time)
print('------------------')
print('Critical path')
print(' -> '.join(critical_path))
return 0
I want to write it a little more beautifully I didn't have enough knowledge. I will improve it someday.
Summary
・ In order to get the connection point time anymore, the weighted directed graph It can be solved as a problem that maximizes the total weight value **. (Refer to the idea of Dijkstra's algorithm) -It is necessary to combine DFS and BFS to determine the critical path (?) → I would like to know if there is a better way.
Impressions
Since I wrote an article about Qiita for the first time, I'm glad that I was able to practice writing articles. I would like to write positively and organize my thoughts.
Other references
For the implementation of BFS using Python, I referred to the following article by @ takayg1. -Algorithm in Python (depth-first search, dfs)
Recommended Posts