前言#

软件工程的第一个作业——使用编程实现 PERT 图，在此分享我的实现思路。

代码开源地址：https://github.com/Aki-Hoyue/projects/tree/main/SE/PERT

PS：如果有能力，建议观看 GitHub 中的英文版报告，因为一开始写的就是英文版，本篇可能有些地方有翻译痕迹。 如果对代码部分有任何问题，欢迎留言讨论。

介绍与环境#

计划评审技术（英语：Program Evaluation and Review Technique，简称 “PERT”），主要目的是针对不确定性较高的工作项目，以网络图规划整个专案，以排定期望的专案时程。PERT 图把项目描绘成一个由编号结点（圆形或者方形）构成的网络图，编号节点代表着项目中的任务。每个结点都被编号，并且标注了任务、工期、开始时间和完成时间。线条上的箭头方向标明了任务次序，并且标识出在开始一个任务前必须完成那些任务。

该评审技术的主轴为“乐观时间”、“最有可能时间”及“悲观时间”。

最核心公式为 “专案期望时间” = $\dfrac{\text{乐观时间}+\text{悲观时间}+4\times\text{最可能时间}}{6}$ 。

我使用的是 Python 进行编程，需要的环境有：

Python 3.x 及更高版本
至少需要以下 Python 包：
- os：提供文件操作
- json：提供 JSON 文件操作
- math：提供一些数学函数
- scipy.stats：提供统计函数和分布
- networkx：创建、操作和分析复杂网络图
- matplotlib.pyplot：提供用于创建可视化效果的函数
- queue：提供用于实现队列的类
- tabulate：从数据创建格式化的表格，便于清晰展示数据

推荐使用 Anaconda，一个集成环境，安装具体可以看： https://hoyue.fun/algorithm_pyinstall.html

使用说明#

数据文件默认数据文件为 data.json，与代码在同一目录下。

它遵循 JSON 格式，是一个列表，其中每个节点都是一个字典，字典的示例格式是：

1
{
2
  "Activity Code": "A",
3
  "Immediate Predecessor": ["B", "C"],
4
  "Optimistic Estimate": 12,
5
  "Most Likely Estimate": 15,
6
  "Pessimistic Estimate": 18
7
}

对于每个活动的前置，如果它不存在则是一个空列表 []。除此之外，我在文件夹中还提供了两个测试数据：data1.json（与 data.json 数据相同），data2.json。

在终端中运行程序，会显示一个菜单。如图：

你在进行计算前，必须先加载数据文件。按 1 或 2 以加载文件。

Press 1: 加载默认文件 data.json
Press 2: 加载一个自定义文件，该文件位于程序的同一文件夹或给出绝对路径。例如：

成功加载文件后，你将看到通知：Data loaded successfully!（数据加载成功！），并回到菜单上。

如果输入了错误的文件，程序将报告错误并退出。

然后，你可以选择选项 3 - 9 来实现相应的功能。示例（使用数据文件 data2.json）：

Press 3: 计算每项活动的预计持续时间（Calculate the estimated duration for each activity.）
Press 4: 计算每个活动的时间，包括最早开始时间、最早完成时间、最晚开始时间、最晚完成时间、松弛时间以及是否处于关键路径中（Calculate the time for each activity including Earliest Start Time, Earliest Finish Time, Latest Start Time, Latest Finish Time, Slack Time and Whether it is in the critical path.）
Press 5: 找到所有的关键路径（Find the Critical Path.）
Press 6: 计算期望项目完成时间（Calculate the expected project completion time.）
Press 7: 计算在目标时间内完成项目的概率（Calculate the probability of completing the project by a targeted time.）

假设我们计算在大于 145 天内完成项目的概率。我们输入目标时间为 145，分布模式为 l（larger）。
Press 8: 计算所有项目（使用默认数据）

如果不加载文件，它将加载默认数据并在项之前显示。如果你已经加载了一个文件，它将根据你的数据进行计算。
Press 9: 展示活动拓扑图，将在一个新的窗口中打开（Draw an activity graph.）
Press 10: 退出 / Exit.

设计思路#

存储设计#

设计存储的时候，我设计了一个列表 PERT 去存储输入的数据。每一项都是一个字典，初始化为 PERT = []。

同时定义一个类 Graph，其构造函数将类 Graph 的每个对象初始化为一个空字典 graph，以获取有关此图的信息。

在类中的存储，我使用字典里嵌套字典的方式，便于遍历和存储信息。初始化代码如下：

1
# Define an array to store data and each item is a dictionary
2
PERT = []
3

4
class Graph:
5
    def __init__(self):
6
        self.graph = {}

在函数 file_load 中，它从 JSON 文件中读取数据并将其存储在 PERT 数组中。file_load 的参数是一个数据文件目录，可以通过用户输入传入，默认值为 data.json。在设计中，首先确定路径，如果路径不存在，则返回错误消息，否则将读取的数据存储在临时变量 data 中。

下一步是遍历整个 data，一次读入一个字典。Optimistic Estimate、Most Likely Estimate 和 Pessimistic Estimate 数据可用于使用公式直接计算 Estimate During 和方差。这些信息将作为 PERT 表的元素写入。

最后，返回 1，表示文件内容已成功读取。

1
# Load data from json file
2
def file_load(filename="data.json"):
3
    if os.path.exists(filename):
4
        with open(filename, "r") as data_table:
5
            data = json.load(data_table)
6
    else:
7
        print("Error, file not found!")
8
        print_separator()
9
        return 0
10

11
    # Iterate through data
12
    for item in data:
13
        activity_code = item['Activity Code']
14
        immidiate_predecessor = item['Immediate Predecessor']  # An array
15
        optimistic_estimate = item['Optimistic Estimate']
16
        most_likely_estimate = item['Most Likely Estimate']
17
        pessimistic_estimate = item['Pessimistic Estimate']
18
        estimate_during = (optimistic_estimate + 4 * most_likely_estimate + pessimistic_estimate) / 6
19
        variance = ((pessimistic_estimate - optimistic_estimate) / 6) ** 2
20

21
        PERT.append({
22
            "activity code": activity_code,
23
            "immidiate_predecessor": immidiate_predecessor,
24
            "estimate_during": estimate_during,
25
            "variance": variance
26
        })
27
    return 1

读入数据后，我们仍然需要将数据加载到 Graph 类中，我在 Graph 类中设计了成员函数 add_activity。它的参数是存储输入的表 PERT。

在添加数据到图中时，除了添加一些必要的信息外，我还设计了 next_activity 列表来存储活动的后驱节点，这样就可以很容易地找到 earliest start 和 earliest finish。

1
# Add a node to the graph
2
def add_activity(self, PERT):
3
    immidiate_predecessor = PERT['immidiate_predecessor']
4
    time = PERT['estimate_during']
5
    variance = PERT['variance']
6
    # Set predecessor to the new node. Next_activity is an array that stores the next activity.
7
    for i in immidiate_predecessor:
8
        if i in self.graph:
9
            self.graph[i]["next_activity"].append(PERT['activity code'])
10
    # Add the new node to the graph
11
    self.graph[PERT['activity code']] = {
12
        "activity_code": PERT['activity code'],
13
        "estimate_during": time,
14
        "variance": variance,
15
        "next_activity": [],
16
        "before_activity": immidiate_predecessor,
17
        "earliest_start_time": 0,
18
        "earliest_finish_time": 0,
19
        "latest_start_time": 0,
20
        "latest_finish_time": 0,
21
        "slack_time": -1,
22
        "critical_path": False
23
    }

计算最早开始和最早完成时间#

在函数 earliest_time 中，我们初始化没有前驱的节点作为起点，然后通过遍历后驱来依次计算 ES 和 EF。

在第一层循环中，逐个迭代对象 graph 字典中的活动名，在第二层循环中，循环遍历当前活动的后驱，如果后驱的 ES 小于当前节点的 EF，则后驱的 ES 被替换为当前节点的 EF，并重新计算后驱的 EF。

在循环结束时，所有活动的 ES 和 EF 都已经计算完毕。

1
# Calculate the earliest start time and earliest finish time
2
def earliest_time(self):
3
    # Initialize the earliest start time and earliest finish time
4
    for i in self.graph:
5
        if self.graph[i]["before_activity"] == []:
6
            self.graph[i]["earliest_finish_time"] = self.graph[i]["estimate_during"]
7
    # Caculate the earliest start time and earliest finish time
8
    for i in self.graph:
9
        for j in self.graph[i]["next_activity"]:
10
            if self.graph[j]["earliest_start_time"] < self.graph[i]["earliest_finish_time"]:
11
                self.graph[j]["earliest_start_time"] = self.graph[i]["earliest_finish_time"]
12
            self.graph[j]["earliest_finish_time"] = (
13
                self.graph[j]["earliest_start_time"] + self.graph[j]["estimate_during"]
14
            )

计算最晚开始和最晚完成时间#

在函数 latest_time 中，我们同样先将没有后驱的节点初始化，它们的 LF 都是最大的 LF。故我们需要先找到最大的 EF 才能找到最大的 LF。

在第一层循环中，倒序遍历活动名；第二层循环，循环遍历当前活动的前驱节点，如果前驱节点的 LF 大于当前节点的 LS，则前驱节点的 LF 会被当前节点的 LS 取代，并重新计算该节点的 LS。

最后，循环结束时会计算出所有活动的 LS 和 LF。

1
# Calculate the latest start time and latest finish time
2
def latest_time(self):
3
    # Initialize the latest start time and latest finish time
4
    estimate_time = 0
5
    for i in reversed(self.graph):
6
        if self.graph[i]["next_activity"] == []:
7
            estimate_time = max(estimate_time, self.graph[i]["earliest_finish_time"])
8
            self.graph[i]["latest_finish_time"] = estimate_time
9
            self.graph[i]["latest_start_time"] = (
10
                self.graph[i]["latest_finish_time"] - self.graph[i]["estimate_during"]
11
            )
12

13
    # Calculate the latest start time and latest finish time
14
    for i in reversed(self.graph):
15
        for j in self.graph[i]["before_activity"]:
16
            if self.graph[j]["latest_finish_time"] == 0 or \
17
               self.graph[j]["latest_finish_time"] > self.graph[i]["latest_start_time"]:
18
                self.graph[j]["latest_finish_time"] = self.graph[i]["latest_start_time"]
19
            self.graph[j]["latest_start_time"] = (
20
                self.graph[j]["latest_finish_time"] - self.graph[j]["estimate_during"]
21
            )

计算松弛时间#

计算松弛时间（slack time）非常简单，只需计算每个项目的 LS - ES 即可。

1
# Calculate the slack time LS-ES
2
def slack_time(self):
3
    # Calculate the slack time
4
    for i in self.graph:
5
        self.graph[i]["slack_time"] = (
6
            self.graph[i]["latest_start_time"] - self.graph[i]["earliest_start_time"]
7
        )

计算关键路径#

确定一项活动是否为关键路径（critical path）也非常简单：如果活动的松弛时间为 0，则它是关键路径，否则不是。

1
# Calculate the critical path that Slack = 0
2
def critical_path(self):
3
    # Calculate the critical path
4
    for i in self.graph:
5
        if self.graph[i]["slack_time"] == 0:
6
            self.graph[i]["critical_path"] = True

输出#

本部分是输出的集合，用于打印上述函数的结果。

首先，我们应该在输出之前执行那些计算函数，以便为输出提供足够的数据。在函数初始化中调用这些计算函数 initization。

1
# Initialization before printing
2
def initization(self):
3
    self.earliest_time()
4
    self.latest_time()
5
    self.slack_time()
6
    self.critical_path()

然后我们输出 Estimate duration，将其输出为表格，使用 tabulate 模块进行制表，将每个活动的数据写入 Estimate_table，并设置标题和表格格式。

1
# Print the Estimate duration
2
def print_estimate_duration(self):
3
    print("Estimate duration: ")
4
    Estimate_table = [["Activity", "Estimate Duration", "Variance"]]
5
    for i in self.graph:
6
        Estimate_table.append([
7
            self.graph[i]["activity_code"],
8
            self.graph[i]["estimate_during"],
9
            self.graph[i]["variance"]
10
        ])
11
    print(tabulate(Estimate_table, headers="firstrow", tablefmt='fancy_grid', numalign="center"))

同理，将必要的数据写入 table 中，输出 ES, EF, LS, LF 等内容，并设置标题和表格格式。

1
# Print the earliest/latest start time and finish time
2
def print_PERTtable(self):
3
    print("PERT table: ")
4
    table = [["Activity", "Ealiest Start", "Ealiest Finish",
5
              "Latest Start", "Latest Finish", "Slack Time", "Critical Path"]]
6
    for i in self.graph:
7
        table.append([
8
            self.graph[i]["activity_code"],
9
            self.graph[i]["earliest_start_time"],
10
            self.graph[i]["earliest_finish_time"],
11
            self.graph[i]["latest_start_time"],
12
            self.graph[i]["latest_finish_time"],
13
            self.graph[i]["slack_time"],
14
            self.graph[i]["critical_path"]
15
        ])
16
    print(tabulate(table, headers="firstrow", tablefmt='fancy_grid'))

输出关键路径，用 -> 连接每个关键路径：

1
# Print the critical path
2
def print_critical_path(self):
3
    print("Critical path: ")
4
    for i in self.graph:
5
        if self.graph[i]["critical_path"]:
6
            print(self.graph[i]["activity_code"], end=" ")
7
            if self.graph[i]["next_activity"] != []:
8
                print("->", end=" ")
9
    print()

在输出预计项目完成时间时，需要通过 keys() 属性找到最后一个最大的 LF/EF 并输出。

1
# Print the expected project completion time
2
def print_expected_project_completion_time(self):
3
    keys = list(self.graph.keys())
4
    print("Expected project completion time: %s" %
5
          self.graph[keys[-1]]["latest_finish_time"])

计算概率#

为了计算目标完成时间比实际完成时间 larger/smaller (l/s) 的概率，设计了函数 probability。它有两个参数，一个是目标完成时间 targeted，另一个是分布模式，可以是 larger/smaller (l/s)。

基于公式 Standard Deviation 和 Z-Score 可以计算，概率可以基于 scipy.stats 模块计算。

1
# Calculate the probability of completing the project. mode = larger/smaller
2
def probability(self, targeted, mode):
3
    if mode == "l":
4
        mode = "larger"
5
    elif mode == "s":
6
        mode = "smaller"
7

8
    var_sum = 0
9
    for i in self.graph:
10
        if self.graph[i]["critical_path"]:
11
            var_sum += self.graph[i]["variance"]
12

13
    standard_deviation = math.sqrt(var_sum)
14
    keys = list(self.graph.keys())
15
    expected = self.graph[keys[-1]]["latest_finish_time"]
16
    zscore = (targeted - expected) / standard_deviation
17

18
    if mode == "larger":
19
        probability = 1 - st.norm.cdf(zscore)
20
    else:
21
        probability = st.norm.cdf(zscore)
22

23
    print("Z-score: {}\nProbability of project completion {} than {} is: {:.2%} ({}). "
24
          .format(zscore, mode, targeted, probability, probability))

画图#

在函数 graph_table 中，G 用作网格图，graph_dict 作为对象的 graph，而队列 Q 则被设计用于使用 FIFO 中生成节点和边。pos 作为一个 map 来存储每个节点的坐标，postion 字典记录了每个节点的 row 和 col 以便生成之后节点的坐标。

首先，我先生成一条由多个关键路径组成的直线，用于代表 关键路径，初始节点坐标为 (0, 0)，依次加 1 并添加边，并附上边的标签（活动名称）。对于添加的每个节点，该节点都会进入队列，以便此后可以按照顺序生成其子节点。

接下来设置一个指向第一个节点的初始化节点（Start）。

接着，生成非关键路径的节点。一个节点根据其后驱节点的数量生成一个 range_col，它代表其后驱节点纵坐标的最大值，每个后驱节点的纵坐标相差 1。当只有一个后驱的情况下，其后驱节点的纵坐标与当前节点相同。

特殊情况：如果当前节点没有后驱并且不是最后一个关键路径节点，则它指向最后一个关键路径节点（说明它是最后一位节点并直接指向 END）。

接下来遍历当前节点的所有后驱，如果后驱不在已添加的节点列表中，则添加它，添加实线边并将后驱入队。

否则，判断当前节点的 EF 和后驱节点 ES 的大小，如果当前节点的 EF 较大，则需要将后代节点的原有的边设置为虚线，并添加新的实线边；如果后驱节点的 ES 很大，则直接添加虚线边。

这里有一个特别判断，如果当前节点的 EF 和后驱节点的 ES 大小相同，则需要将后代节点原来的非关键路径的边设置为虚线，并添加实线边。

最后通过 matplotlib.pyplot 和 networkx 模块分别绘制节点、边以及边的标签，并最终展示出来。

1
# Graph the table
2
def graph_table(self):
3
    graph_dict = self.graph
4
    G = nx.DiGraph()
5
    pos = {}
6
    Q = queue.Queue()
7
    postion = {}
8

9
    # Set the critical path
10
    row = 0
11
    for i in graph_dict:
12
        if graph_dict[i]["critical_path"]:
13
            Q.put(graph_dict[i]["activity_code"])
14
            G.add_node(graph_dict[i]["activity_code"])
15
            postion[graph_dict[i]["activity_code"]] = {"row": 0, "col": 0}
16
            pos[graph_dict[i]["activity_code"]] = (row, 0)
17
            row += 1
18
            postion[graph_dict[i]["activity_code"]]["row"] = row
19
            postion[graph_dict[i]["activity_code"]]["col"] = 0
20
            for j in graph_dict[i]["before_activity"]:
21
                if j in G.nodes():
22
                    G.add_edge(graph_dict[j]["activity_code"],
23
                               graph_dict[i]["activity_code"],
24
                               name=graph_dict[i]["activity_code"])
25

26
    # Set the startpoint and endpoint
27
    G.add_node("Start")
28
    pos["Start"] = (-1, 0)
29
    G.add_edge("Start", Q.queue[0], name=Q.queue[0])
30
    last_critical_point = Q.queue[Q.qsize() - 1]
31

32
    dash = []
33
    # Set the non-critical path
34
    while not Q.empty():
35
        cur = Q.get()
36

37
        if len(graph_dict[cur]["next_activity"]) > 1:
38
            range_col = math.floor(math.log2(len(graph_dict[cur]["next_activity"])))
39
        elif len(graph_dict[cur]["next_activity"]) == 1:
40
            range_col = postion[cur]["col"]
41

42
        if graph_dict[cur]["next_activity"] == [] and cur != last_critical_point:
43
            G.add_edge(cur, last_critical_point)
44
            dash.append((cur, last_critical_point))
45
            continue
46

47
        for next_node in graph_dict[cur]["next_activity"]:
48
            if next_node not in G.nodes():
49
                G.add_node(next_node)
50
                col = range_col
51
                row = (postion[cur]["row"]) * (1 - (col / 10.0)) + 1
52
                postion[next_node] = {"row": row, "col": col}
53
                pos[next_node] = (row, col)
54
                range_col -= 1
55
                if range_col == 0:
56
                    range_col -= 1
57
                G.add_edge(cur, next_node, name=next_node)
58
                Q.put(next_node)
59
            else:
60
                cur_EF = graph_dict[cur]["earliest_finish_time"]
61
                next_ES = graph_dict[next_node]["earliest_start_time"]
62
                if cur_EF > next_ES:
63
                    pred = list(G.predecessors(next_node))
64
                    for j in range(len(pred)):
65
                        G.remove_edge(pred[j], next_node)
66
                        G.add_edge(pred[j], next_node)
67
                        dash.append((pred[j], next_node))
68
                    G.add_edge(cur, next_node, name=next_node)
69
                elif cur_EF < next_ES:
70
                    for j in graph_dict[next_node]["before_activity"]:
71
                        if graph_dict[j]["critical_path"]:
72
                            G.add_edge(cur, j)
73
                            dash.append((cur, j))
74
                else:
75
                    pred = list(G.predecessors(next_node))
76
                    for j in range(len(pred)):
77
                        if graph_dict[pred[j]]["critical_path"] and \
78
                           not graph_dict[cur]["critical_path"]:
79
                            G.remove_edge(pred[j], next_node)
80
                            G.add_edge(pred[j], next_node)
81
                            dash.append((pred[j], next_node))
82
                            G.add_edge(cur, next_node, name=next_node)
83

84
    # Draw the graph
85
    nx.draw_networkx_nodes(G, pos)
86

87
    dashed_edges = [(u, v) for u, v in G.edges if (u, v) in dash]
88
    nx.draw_networkx_edges(G, pos, edgelist=dashed_edges, style='dashed')
89

90
    solid_edges = [(u, v) for u, v in G.edges if (u, v) not in dash]
91
    nx.draw_networkx_edges(G, pos, edgelist=solid_edges, style='solid')
92

93
    nx.draw_networkx_edge_labels(
94
        G, pos, edge_labels=nx.get_edge_attributes(G, 'name'),
95
        font_family="sans-serif", font_size=10
96
    )
97
    plt.show()

辅助函数#

程序中还有一些辅助函数，可以帮助生成和判断一些情况。

print_separator 函数用于生成分隔符以允许分隔重要的输出。

loading_test 函数用于测试数据是否已经加载，根据情况对其参数类型进行不同的分析。

当 Type = 0 时，函数会输出加载消息，返回 1 表示加载成功并输出成功消息，返回 0 表示已经加载。
当 Type = 2 时，情况与 Type = 0 非常相似，不同的是返回 1 时不执行任何操作。
当 Type = 1 时，会先检查是否需要加载。返回 1 表示已经加载，返回 0 表示需要先加载，并输出一条消息。

1
# Print separator
2
def print_separator():
3
    print("\n------------------------------------------------------------\n")
4

5

6
# Loading test.
7
def loading_test(type):
8
    if type == 0:
9
        # Type = 0 (load info -> return 1 (success) or 0 (already) for default file)
10
        if not PERT:
11
            print_separator()
12
            print("Data loaded successfully!")
13
            print_separator()
14
            return 1
15
        else:
16
            print_separator()
17
            print("Data has been loaded!")
18
            print_separator()
19
            return 0
20
    elif type == 1:
21
        # Type = 1 (check -> return 1 (ok) or 0 (need to load))
22
        if not PERT:
23
            print_separator()
24
            print("Please load data first!")
25
            print_separator()
26
            return 0
27
        else:
28
            print_separator()
29
            return 1
30
    elif type == 2:
31
        # Type = 2 (for custom file like type 0)
32
        if not PERT:
33
            return 1
34
        else:
35
            print_separator()
36
            print("Data has been loaded!")
37
            print_separator()
38
            return 0

主函数#

主要功能比较简单，主要是一个菜单，根据用户输入进行功能调用和反馈。

1
# Main function
2
# Initialization
3
PERT_graph = Graph()
4
print("Welcome to PERT project management system!\nAuthor: Hoyue")
5
while True:
6
    print("Menu: ")
7
    print("1. Load data from default file(data.json).")
8
    print("2. Load data from custom file.")
9
    print("3. Calculate the estimated duration for each activity.")
10
    print("4. Calculate the time for each activity including Earliest Start Time, Earliest Finish Time, Latest Start Time, Latest Finish Time, Slack Time and Whether it is in the critical path.")
11
    print("5. Find the Critical Path.")
12
    print("6. Calculate the expected project completion time.")
13
    print("7. Calculate the probability of completing the project by a targeted time.")
14
    print("8. Calculate all the items(By default data).")
15
    print("9. Draw an activity graph.")
16
    print("10. Exit.")
17
    choice = input("Please press the number to choose the function: ")
18

19
    if choice == "1":
20
        if loading_test(0):
21
            if file_load():
22
                for i in range(len(PERT)):
23
                    PERT_graph.add_activity(PERT[i])
24

25
    elif choice == "2":
26
        if loading_test(2):
27
            print_separator()
28
            filename = input("Please input your filename in the software folder(Format as data.json): ")
29
            if file_load(filename):
30
                for i in range(len(PERT)):
31
                    PERT_graph.add_activity(PERT[i])
32
                print("Data loaded successfully!")
33
                print_separator()
34
    elif choice == "3":
35
        if loading_test(1):
36
            PERT_graph.initization()
37
            PERT_graph.print_estimate_duration()
38
            print_separator()
39
    elif choice == "4":
40
        if loading_test(1):
41
            PERT_graph.initization()
42
            PERT_graph.print_PERTtable()
43
            print_separator()
44
    elif choice == "5":
45
        if loading_test(1):
46
            PERT_graph.initization()
47
            PERT_graph.print_critical_path()
48
            print_separator()
49
    elif choice == "6":
50
        if loading_test(1):
51
            PERT_graph.initization()
52
            PERT_graph.print_expected_project_completion_time()
53
            print_separator()
54
    elif choice == "7":
55
        if loading_test(1):
56
            PERT_graph.initization()
57
            print("For cauculating the probability of completing the project, please input the targeted time and distribution mode(larger/smaller).")
58
            targeted = float(input("Targeted time: "))
59
            mode = input("Distribution mode(larger/smaller or l/s): ")
60
            if mode not in ("larger", "smaller", "l", "s"):
61
                print_separator()
62
                print("Error input! Please input again!")
63
                print_separator()
64
                continue
65
            PERT_graph.probability(targeted, mode)
66
            print_separator()
67
    elif choice == "8":
68
        if not PERT:
69
            file_load()
70
            for i in range(len(PERT)):
71
                PERT_graph.add_activity(PERT[i])
72
        print_separator()
73
        PERT_graph.initization()
74
        PERT_graph.print_estimate_duration()
75
        PERT_graph.print_PERTtable()
76
        PERT_graph.print_critical_path()
77
        PERT_graph.print_expected_project_completion_time()
78
        print("For cauculating the probability of completing the project, please input the targeted time and distribution mode(larger/smaller).")
79
        targeted = float(input("Targeted time: "))
80
        mode = input("Distribution mode(larger/smaller or l/s): ")
81
        if mode not in ("larger", "smaller", "l", "s"):
82
            print_separator()
83
            print("Error input! Please input again!")
84
            print_separator()
85
            continue
86
        PERT_graph.probability(targeted, mode)
87
        print_separator()
88
    elif choice == "9":
89
        if loading_test(1):
90
            PERT_graph.initization()
91
            print("Activity graph:")
92
            print("Notice: If you don't have some fonts installed, you may get an error, just ignore it.")
93
            PERT_graph.graph_table()
94
            print_separator()
95
    elif choice == "10":
96
        print("Thank you! Goodbye!")
97
        break
98
    else:
99
        print_separator()
100
        print("Error input! Please input again!")
101
        print_separator()

限制#

我在设计的时候，只考虑了默认顺序下输入的情况，如果输入的数据存在前驱位于后驱后面的情况，可能会出现问题。以及输出的时候只考虑了末位，如果改为随机数据时可以试试增加变量存储。

还有就是在画图的时候，子节点的 range_col 和 row 的生成公式可以自定义，我设计并不是最优的情况，可能导致图看起来没这么好看。

后记#

这是我使用 Python 实现的 PERT 项目，练习 Python 非常合适。当然也可以使用 C++ / Java 等语言，核心思想都差不多。如果对代码部分有任何问题，欢迎留言讨论。