重頭戲部分來了,寫到這里我感覺得仔細認真點了,可能在NetworkX中,實現某些算法就一句話的事,但是這個算法是做什么的,用在什么地方,原理是怎么樣的,不清除,所以,我決定先把圖論
中常用算法弄個明白在寫這部分.
圖論常用算法看我的博客:
下面我將使用NetworkX實現上面的算法,建議不清楚的部分打開兩篇博客對照理解.
我將圖論的經典問題及常用算法的總結寫在下面兩篇博客中:
圖論---問題篇
圖論---算法篇
目錄:
* 11.3關鍵路徑算法(CPA)
注意:如果代碼出現找不庫,請返回第一個教程,把庫文件導入.
11.3關鍵路徑算法(CPA)
以下代碼從這里復制,由於版本問題,將代碼中的:nx.topological_sort(self, reverse=True)
改為list(reversed(list(nx.topological_sort(self))))
- import networkx as nx
- import matplotlib.pyplot as plt
- from matplotlib.font_manager import *
- #定義自定義字體,文件名從1.b查看系統中文字體中來
- myfont = FontProperties(fname='/usr/share/fonts/truetype/wqy/wqy-zenhei.ttc')
- #解決負號'-'顯示為方塊的問題
- matplotlib.rcParams['axes.unicode_minus']=False
- class CPM(nx.DiGraph):
- def __init__(self):
- super().__init__()
- self._dirty = True
- self._critical_path_length = -1
- self._criticalPath = None
- def add_node(self, *args, **kwargs):
- self._dirty = True
- super().add_node(*args, **kwargs)
- def add_nodes_from(self, *args, **kwargs):
- self._dirty = True
- super().add_nodes_from(*args, **kwargs)
- def add_edge(self, *args): # , **kwargs):
- self._dirty = True
- super().add_edge(*args) # , **kwargs)
- def add_edges_from(self, *args, **kwargs):
- self._dirty = True
- super().add_edges_from(*args, **kwargs)
- def remove_node(self, *args, **kwargs):
- self._dirty = True
- super().remove_node(*args, **kwargs)
- def remove_nodes_from(self, *args, **kwargs):
- self._dirty = True
- super().remove_nodes_from(*args, **kwargs)
- def remove_edge(self, *args): # , **kwargs):
- self._dirty = True
- super().remove_edge(*args) # , **kwargs)
- def remove_edges_from(self, *args, **kwargs):
- self._dirty = True
- super().remove_edges_from(*args, **kwargs)
-
- #根據前向拓撲排序算弧的最早發生時間
- def _forward(self):
- for n in nx.topological_sort(self):
- es = max([self.node[j]['EF'] for j in self.predecessors(n)], default=0)
- self.add_node(n, ES=es, EF=es + self.node[n]['duration'])
-
- #根據前向拓撲排序算弧的最遲發生時間
- def _backward(self):
- #for n in nx.topological_sort(self, reverse=True):
- for n in list(reversed(list(nx.topological_sort(self)))):
- lf = min([self.node[j]['LS'] for j in self.successors(n)], default=self._critical_path_length)
- self.add_node(n, LS=lf - self.node[n]['duration'], LF=lf)
-
- #最早發生時間=最遲發生時間,則判斷該節點為關鍵路徑上的關鍵活動
- def _compute_critical_path(self):
- graph = set()
- for n in self:
- if self.node[n]['EF'] == self.node[n]['LF']:
- graph.add(n)
- self._criticalPath = self.subgraph(graph)
- def critical_path_length(self):
- if self._dirty:
- self._update()
- return self._critical_path_length
- def critical_path(self):
- if self._dirty:
- self._update()
- return sorted(self._criticalPath, key=lambda x: self.node[x]['ES'])
- def _update(self):
- self._forward()
- self._critical_path_length = max(nx.get_node_attributes(self, 'EF').values())
- self._backward()
- self._compute_critical_path()
- self._dirty = False
- if __name__ == "__main__":
-
- #構建graph
- G = CPM()
- G.add_node('A', duration=5)
- G.add_node('B', duration=2)
- G.add_node('C', duration=4)
- G.add_node('D', duration=4)
- G.add_node('E', duration=3)
- G.add_node('F', duration=7)
- G.add_node('G', duration=4)
- G.add_edges_from([
- ('A', 'B'),
- ('A', 'C'),
- ('C','D'),
- ('C','E'),
- ('C','G'),
- ('B','D'),
- ('D','F'),
- ('E','F'),
- ('G','F'),
- ])
-
- #顯示graph
- nx.draw_spring(G,with_labels=True)
- plt.title('AOE網絡',fontproperties=myfont)
- plt.axis('on')
- plt.xticks([])
- plt.yticks([])
- plt.show()
-
-
- print('關鍵活動為:')
- print(G.critical_path_length, G.critical_path)
- G.add_node('D', duration=2)
- print('\n修改D活動持續時間4為2后的關鍵活動為:')
- print(G.critical_path_length, G.critical_path)

關鍵路徑示例(該圖非黑色線為手工繪制,數字手工添加)
從graph中可以知道,有兩條關鍵路徑,分別是:A->C->G->F
和A->C->D->F
,長度都是20.
輸出:
關鍵活動為: 20 ['A', 'C', 'D', 'G', 'F']
修改D活動持續時間4為2后的關鍵活動為: 20 ['A', 'C', 'G', 'F']
關鍵活動為: ['A', 'C', 'D', 'G', 'F'],可以構成兩條邊.D活動持續時間4為2后,關鍵路徑變化.