目录
1.主程序
2.活动类
3.任务列表JSON文件
4.输出模板文件
首页 后端开发 Python教程 Python如何实现关键路径和七格图计算

Python如何实现关键路径和七格图计算

Apr 26, 2023 pm 05:28 PM
python

1.主程序

主程序主要实现了一个Project类,其中包含了计算关键路径和七格图的方法。具体实现方式如下:

1.  定义了一个Activity类,包含了活动的id、名称、持续时间和紧前任务列表等属性。 

2.  定义了一个Project类,包含了活动列表、项目持续时间、日志等属性,以及计算关键路径、计算七格图、计算总浮动时间、计算自由浮动时间等方法。

3.  从JSON文件中读取活动信息,并创建Project对象并添加活动。

4.  调用Project对象的calculate方法,计算每个活动的最早开始时间、最晚开始时间等数据。

5.  调用Project对象的calculate_critical_path方法,计算关键路径。

6.  调用Project对象的calculate_project_duration方法,计算项目总工期。

7.  使用Jinja2模板引擎生成项目的活动清单,并将关键路径

import json  
from datetime import datetime  
from typing import List  
  
import graphviz  
from jinja2 import Template  
  
from activity import Activity  
  
  
class Project:  
    def __init__(self):  
        self.activities: List[Activity] = []  
        self.duration = 0  
        self.logger = []  
  
    def log(self, log: str) -> None:  
        self.logger.append(log)  
  
    def add_activity(self, activity: Activity) -> None:  
        """  
        添加一个活动到项目中  
        :param  
        activity: 待添加的活动  
        """        # 将活动添加到项目中  
        self.activities.append(activity)  
  
    def calculate(self) -> None:  
        """ 计算整个项目的关键信息  
        :return: None  
        """        self.calculate_successor()  
        self._calculate_forward_pass()  # 计算正推法  
        self._calculate_backward_pass()  # 计算倒推法  
        self._calculate_total_floats()  # 计算总浮动时间  
        self._calculate_free_floats()  # 计算自由浮动时间  
  
    def calculate_successor(self) -> None:  
        self.log("开始计算紧后活动")  
        for act in self.activities:  
            for pred in act.predecessors:  
                for act_inner in self.activities:  
                    if act_inner.id == pred:  
                        act_inner.successors.append(act.id)  
  
    def _calculate_forward_pass(self) -> None:  
        self.log("## 开始正推法计算")  
        # 进入 while 循环,只有当所有活动的最早开始时间和最早完成时间都已经计算出来时,才会退出循环  
        while not self._is_forward_pass_calculated():  
            # 遍历每个活动  
            for activity in self.activities:  
                # 如果活动的最早开始时间已经被计算过,则跳过  
                if activity.est is not None:  
                    continue  
                # 如果活动没有前置活动, 则从1开始计算最早开始时间和最早结束时间  
                if not activity.predecessors:  
                    activity.est = 1  
                    activity.eft = activity.est + activity.duration - 1  
                    self.log(  
                        f"活动 {activity.name} 没有紧前活动,设定最早开始时间为1, 并根据工期计算最早结束时间为{activity.eft}")  
                else:  
                    # 计算当前活动的所有前置活动的最早完成时间  
                    predecessors_eft = [act.eft for act in self.activities if  
                                        act.id in activity.predecessors and act.eft is not None]  
                    # 如果当前活动的所有前置活动的最早完成时间都已经计算出来,则计算当前活动的最早开始时间和最早完成时间  
                    if len(predecessors_eft) == len(activity.predecessors):  
                        activity.est = max(predecessors_eft) + 1  
                        activity.eft = activity.est + activity.duration - 1  
                        self.log(  
                            f"活动 {activity.name} 紧前活动已完成正推法计算, 开始日期按最早开始时间里面最大的," +  
                            f"设定为{activity.est}并根据工期计算最早结束时间为{activity.eft}")  
  
                        # 更新项目总持续时间为最大最早完成时间  
        self.duration = max([act.eft for act in self.activities])  
  
    def _calculate_backward_pass(self) -> None:  
        """ 计算倒推法  
        :return: None  
        """  
        self.log("## 开始倒推法计算")  # 输出提示信息  
        # 进入 while 循环,只有当所有活动的最晚开始时间和最晚完成时间都已经计算出来时,才会退出循环  
        while not self._is_backward_pass_calculated():  
            # 遍历每个活动  
            for act in reversed(self.activities):  
                # 如果活动的最晚开始时间已经被计算过,则跳过  
                if act.lft is not None:  
                    continue  
                # 如果活动没有后继活动, 则从总持续时间开始计算最晚开始时间和最晚结束时间  
                if not act.successors:  
                    act.lft = self.duration  
                    act.lst = act.lft - act.duration + 1  
                    self.log(f"活动 {act.name} 没有紧后活动,按照正推工期设定最晚结束时间为{act.lft}," +  
                             f"并根据工期计算最晚开始时间为{act.lst}")  
                else:  
                    # 计算当前活动的所有后继活动的最晚开始时间  
                    successors_lst = self._calculate_lst(act)  
                    # 如果当前活动的所有后继活动的最晚开始时间都已经计算出来,则计算当前活动的最晚开始时间和最晚完成时间  
                    if len(successors_lst) == len(act.successors):  
                        act.lft = min(successors_lst) - 1  
                        act.lst = act.lft - act.duration + 1  
                        self.log(f"活动 {act.name} 紧后活动计算完成,按照倒推工期设定最晚结束时间为{act.lft}," +  
                                 f"并根据工期计算最晚开始时间为{act.lst}")  
        # 更新项目总持续时间为最大最晚完成时间  
        self.duration = max([act.lft for act in self.activities])  
  
    def _calculate_lst(self, activity: Activity) -> List[int]:  
        """计算某一活动的所有最晚开始时间  
        :param activity: 活动对象  
        :return: 最晚开始时间列表  
        """        rst = []  # 初始化结果列表  
        for act in activity.successors:  # 遍历该活动的后继活动  
            for act2 in self.activities:  # 遍历所有活动  
                if act2.id == act and act2.lst is not None:  # 如果找到了该后继活动且其最晚开始时间不为空  
                    rst.append(act2.lst)  # 将最晚开始时间加入结果列表  
        return rst  # 返回结果列表  
  
    def _is_forward_pass_calculated(self) -> bool:  
        """ 判断整个项目正推法计算已经完成  
        :return: 若已计算正向传递则返回True,否则返回False  
        """  
        for act in self.activities:  # 遍历所有活动  
            if act.est is None or act.eft is None:  # 如果该活动的最早开始时间或最早完成时间为空  
                return False  # 则返回False,表示还未计算正向传递  
        return True  # 如果所有活动的最早开始时间和最早完成时间都已计算,则返回True,表示已计算正向传递  
  
    def _is_backward_pass_calculated(self) -> bool:  
        """ 判断整个项目倒推法计算已经完成  
        :return: 若已计算倒推法则返回True,否则返回False  
        """        for act in self.activities:  # 遍历所有活动  
            if act.lst is None or act.lft is None:  # 如果该活动的最晚开始时间或最晚完成时间为空  
                return False  # 则返回False,表示还未计算倒推法  
        return True  # 如果所有活动的最晚开始时间和最晚完成时间都已计算,则返回True,表示已计算倒推法  
  
    def _calculate_total_floats(self) -> None:  
        """ 计算所有活动的总浮动时间  
        :return: None  
         """  
        self.log(f"## 开始计算项目所有活动的总浮动时间")  
        for act in self.activities:  # 遍历所有活动  
            if act.est is not None and act.lst is not None:  # 如果该活动的最早开始时间和最晚开始时间都已计算  
                act.tf = act.lst - act.est  # 则计算该活动的总浮动时间  
                self.log(f"计算{act.name}的总浮动时间" + f"最晚开始时间{act.lst} - 最早开始时间{act.est} = {act.tf}", )  
            else:  # 如果该活动的最早开始时间或最晚开始时间为空  
                act.tf = None  # 则将该活动的总浮动时间设为None  
  
    def _calculate_free_floats(self) -> None:  
        """ 计算所有活动的自由浮动时间  
        :return: None  
        """        self.log(f"## 开始计算项目所有活动的自由浮动时间")  # 输出提示信息  
        for act in self.activities:  # 遍历所有活动  
            if act.tf == 0:  # 如果该活动的总浮动时间为0  
                self.log(f"计算{act.name}的自由浮动时间" + f"因为{act.name}的总浮动时间为0,自由浮动时间为0")  # 输出提示信息  
                act.ff = 0  # 则将该活动的自由浮动时间设为0  
            elif act.tf > 0:  # 如果该活动的总浮动时间大于0  
                self.log(f"计算{act.name}的自由浮动时间")  # 输出提示信息  
                self.log(f"- {act.name}的总浮动时间{act.tf} > 0,")  # 输出提示信息  
                tmp = []  # 初始化临时列表  
                for act2 in self.activities:  # 遍历所有活动  
                    if act2.id in act.successors:  # 如果该活动是该活动的紧后活动  
                        self.log(f"- {act.name}的紧后活动{act2.name}的自由浮动动时间为{act2.tf}")  # 输出提示信息  
                        tmp.append(act2.tf)  # 将该紧后活动的自由浮动时间加入临时列表  
                if len(tmp) != 0:  # 如果临时列表不为空  
                    act.ff = act.tf - max(tmp)  # 则计算该活动的自由浮动时间  
                    if act.ff < 0:  
                        act.ff = 0  
                    self.log(f"- 用活动自己的总浮动{act.tf}减去多个紧后活动总浮动的最大值{max(tmp)} = {act.ff}")  
                else:  # 如果临时列表为空  
                    act.ff = act.tf  # 则将该活动的自由浮动时间设为总浮动时间  
  
    def calculate_critical_path(self) -> List[Activity]:  
        """ 计算整个项目的关键路径  
        :return: 整个项目的关键路径  
        """        ctc_path = []  # 初始化关键路径列表  
        for act in self.activities:  # 遍历所有活动  
            if act.tf == 0:  # 如果该活动的总浮动时间为0  
                ctc_path.append(act)  # 则将该活动加入关键路径列表  
        return ctc_path  # 返回关键路径列表  
  
    def calculate_project_duration(self) -> int:  
        """ 计算整个项目的持续时间  
        :return: 整个项目的持续时间  
        """        return max(activity.eft for activity in self.activities)  # 返回所有活动的最早完成时间中的最大值,即整个项目的持续时间  
  
  
# 从JSON文件中读取活动信息  
with open(&#39;activities.json&#39;, &#39;r&#39;, encoding=&#39;utf-8&#39;) as f:  
    activities_data = json.load(f)  
  
# 创建Project对象并添加活动  
project = Project()  
for activity_data in activities_data:  
    activity = Activity(  
        activity_data[&#39;id&#39;],  
        activity_data[&#39;name&#39;],  
        activity_data[&#39;duration&#39;],  
        activity_data[&#39;predecessors&#39;]  
    )    project.add_activity(activity)  
  
# 计算每个活动的最早开始时间、最晚开始时间等数据  
project.calculate()  
  
# 计算关键路径和项目总工期  
critical_path = project.calculate_critical_path()  
project_duration = project.calculate_project_duration()  
  
# 生成项目的活动清单  
with open(&#39;template.html&#39;, &#39;r&#39;, encoding=&#39;utf-8&#39;) as f:  
    template = Template(f.read())  
html = template.render(  
    activities=project.activities,  
    critical_path=critical_path,  
    project_duration=project_duration,  
    log=project.logger  
)  
  
# 生成项目进度网络图  
aon_graph = graphviz.Digraph(format=&#39;png&#39;, graph_attr={&#39;rankdir&#39;: &#39;LR&#39;})  
for activity in project.activities:  
    aon_graph.node(str(activity.id), activity.name)  
    for predecessor in activity.predecessors:  
        aon_graph.edge(str(predecessor), str(activity.id))  
  
timestamp = datetime.now().strftime(&#39;%Y%m%d%H%M%S&#39;)  
  
aon_filename = f"aon_{timestamp}"  
  
aon_graph.render(aon_filename)  
  
# 将项目进度网络图插入到HTML文件中  
aon_image = f&#39;<img src="{aon_filename}.png" alt="Precedence Diagramming Method: AON">&#39;  
html = html.replace(&#39;<p>Precedence Diagramming Method: AON: <br/>[image]</p>&#39;,  
                    &#39;<p>紧前关系绘图法: AON: <br/>&#39; + aon_image + &#39;</p>&#39;)  
  
filename = datetime.now().strftime(&#39;%Y%m%d%H%M%S&#39;) + &#39;.html&#39;  
with open(filename, &#39;w&#39;, encoding=&#39;utf-8&#39;) as f:  
    f.write(html)
登录后复制

2.活动类

程序名:activity.py

class Activity:  
    """  
    活动类,用于表示项目中的一个活动。  
  
    Attributes:        id (int): 活动的唯一标识符。  
        name (str): 活动的名称。  
        duration (int): 活动的持续时间。  
        predecessors (List[int]): 活动的前置活动列表,存储前置活动的id。  
        est (int): 活动的最早开始时间。  
        lst (int): 活动的最晚开始时间。  
        eft (int): 活动的最早完成时间。  
        lft (int): 活动的最晚完成时间。  
        tf (int): 活动的总浮动时间。  
        ff (int): 活动的自由浮动时间。  
        successors (List[int]): 活动的后继活动列表,存储后继活动的Activity对象。  
    """  
    def __init__(self, id: int, name: str, duration: int, predecessors: List[int]):  
        """  
        初始化活动对象。  
  
        Args:            id (int): 活动的唯一标识符。  
            name (str): 活动的名称。  
            duration (int): 活动的持续时间。  
            predecessors (List[int]): 活动的前置活动列表,存储前置活动的id。  
        """        self.id = id  
        self.name = name  
        self.duration = duration  
        self.predecessors = predecessors  
        self.est = None  
        self.lst = None  
        self.eft = None  
        self.lft = None  
        self.tf = None  
        self.ff = None  
        self.successors = []  
  
    def __str__(self):  
        return f"id: {self.id}, name: {self.name}, est: {self.est}, lst: {self.lst}, eft: {self.eft}, lft: {self.lft},"  
        + f"successors: {self.successors}"
登录后复制

3.任务列表JSON文件

文件名:activities.json

[  
  {    "id": 1,  
    "name": "A",  
    "duration": 2,  
    "predecessors": []  
  },  
  {  
    "id": 9,  
    "name": "A2",  
    "duration": 3,  
    "predecessors": []  
  },  
    {  
    "id": 10,  
    "name": "A3",  
    "duration": 2,  
    "predecessors": []  
  },  
  {  
    "id": 2,  
    "name": "B",  
    "duration": 3,  
    "predecessors": [  
      1,  
      9  
    ]  
  },  
  {  
    "id": 3,  
    "name": "C",  
    "duration": 4,  
    "predecessors": [  
      1  
    ]  
  },  
  {  
    "id": 4,  
    "name": "D",  
    "duration": 2,  
    "predecessors": [  
      2,10  
    ]  
  },  
  {  
    "id": 5,  
    "name": "E",  
    "duration": 3,  
    "predecessors": [  
      2  
    ]  
  },  
  {  
    "id": 6,  
    "name": "F",  
    "duration": 2,  
    "predecessors": [  
      3  
    ]  
  },  
  {  
    "id": 7,  
    "name": "G",  
    "duration": 3,  
    "predecessors": [  
      4,  
      5  
    ]  
  },  
  {  
    "id": 8,  
    "name": "H",  
    "duration": 2,  
    "predecessors": [  
      6,  
      7  
    ]  
  },  
  {  
    "id": 11,  
    "name": "H2",  
    "duration": 4,  
    "predecessors": [  
      6,  
      7  
    ]  
  }  
]
登录后复制

4.输出模板文件

<!DOCTYPE html>  
<html>  
<head>  
    <meta charset="UTF-8">  
    <title>PMP关键路径计算</title>  
    <style>        table {  
            border-collapse: collapse;  
            width: 100%;  
        }  
  
        th, td {  
            border: 1px solid black;  
            padding: 8px;  
            text-align: center;  
        }  
  
        th {  
            background-color: #4CAF50;  
            color: white;  
        }  
  
        .critical {  
            background-color: #ffcccc;  
        }  
    </style>  
</head>  
<body>  
<h2>活动清单</h2>  
<table>  
    <tr>  
        <th>ID</th>  
        <th>活动名</th>  
        <th>持续时间</th>  
        <th>紧前活动</th>  
        <th>紧后活动</th>  
        <th>最早开始时间EST</th>  
        <th>最早结束时间EFT</th>  
        <th>最晚开始时间LST</th>  
        <th>最晚结束时间LFT</th>  
        <th>总浮动时间TF</th>  
        <th>自由浮动时间FF</th>  
    </tr>  
    {% for activity in activities %}  
    <tr {% if activity in critical_path %}class="critical" {% endif %}>  
        <td>{{ activity.id }}</td>  
        <td>{{ activity.name }}</td>  
        <td>{{ activity.duration }}</td>  
        <td>            {% for predecessor in activity.predecessors %}  
            {% for act in activities %}  
            {% if act.id == predecessor %}  
            {{ act.name }}  
            {% endif %}  
            {% endfor %}  
            {% if not loop.last %}, {% endif %}  
            {% endfor %}  
        </td>  
        <td>            {% for successor in activity.successors %}  
            {% for act in activities %}  
  
            {% if act.id == successor %}  
            {{ act.name }}  
            {% endif %}  
            {% endfor %}  
            {% if not loop.last %}, {% endif %}  
            {% endfor %}  
        </td>  
        <td>{{ activity.est }}</td>  
        <td>{{ activity.eft }}</td>  
        <td>{{ activity.lst }}</td>  
        <td>{{ activity.lft }}</td>  
        <td>{{ activity.tf }}</td>  
        <td>{{ activity.ff }}</td>  
    </tr>  
    {% endfor %}  
</table>  
<p>关键路径是: {% for activity in critical_path %}{{ activity.name }}{% if not loop.last %} -> {% endif %}{% endfor  
    %}</p>  
<p>项目总工期: {{ project_duration }}</p>  
<p>Precedence Diagramming Method: AON: <br/>[image]</p>  
<p>  
<table>  
    <tr>  
        <th>执行过程</th>  
    </tr>  
    {% for i in log %}  
    <tr>  
        <td >{{i}}</td>  
    </tr>  
    {% endfor %}  
</table>  
</p>  
</body>  
</html>
登录后复制

以上是Python如何实现关键路径和七格图计算的详细内容。更多信息请关注PHP中文网其他相关文章!

本站声明
本文内容由网友自发贡献,版权归原作者所有,本站不承担相应法律责任。如您发现有涉嫌抄袭侵权的内容,请联系admin@php.cn

热AI工具

Undresser.AI Undress

Undresser.AI Undress

人工智能驱动的应用程序,用于创建逼真的裸体照片

AI Clothes Remover

AI Clothes Remover

用于从照片中去除衣服的在线人工智能工具。

Undress AI Tool

Undress AI Tool

免费脱衣服图片

Clothoff.io

Clothoff.io

AI脱衣机

AI Hentai Generator

AI Hentai Generator

免费生成ai无尽的。

热门文章

R.E.P.O.能量晶体解释及其做什么(黄色晶体)
1 个月前 By 尊渡假赌尊渡假赌尊渡假赌
R.E.P.O.最佳图形设置
1 个月前 By 尊渡假赌尊渡假赌尊渡假赌
威尔R.E.P.O.有交叉游戏吗?
1 个月前 By 尊渡假赌尊渡假赌尊渡假赌

热工具

记事本++7.3.1

记事本++7.3.1

好用且免费的代码编辑器

SublimeText3汉化版

SublimeText3汉化版

中文版,非常好用

禅工作室 13.0.1

禅工作室 13.0.1

功能强大的PHP集成开发环境

Dreamweaver CS6

Dreamweaver CS6

视觉化网页开发工具

SublimeText3 Mac版

SublimeText3 Mac版

神级代码编辑软件(SublimeText3)

PHP和Python:代码示例和比较 PHP和Python:代码示例和比较 Apr 15, 2025 am 12:07 AM

PHP和Python各有优劣,选择取决于项目需求和个人偏好。1.PHP适合快速开发和维护大型Web应用。2.Python在数据科学和机器学习领域占据主导地位。

CentOS上PyTorch的GPU支持情况如何 CentOS上PyTorch的GPU支持情况如何 Apr 14, 2025 pm 06:48 PM

在CentOS系统上启用PyTorchGPU加速,需要安装CUDA、cuDNN以及PyTorch的GPU版本。以下步骤将引导您完成这一过程:CUDA和cuDNN安装确定CUDA版本兼容性:使用nvidia-smi命令查看您的NVIDIA显卡支持的CUDA版本。例如,您的MX450显卡可能支持CUDA11.1或更高版本。下载并安装CUDAToolkit:访问NVIDIACUDAToolkit官网,根据您显卡支持的最高CUDA版本下载并安装相应的版本。安装cuDNN库:前

docker原理详解 docker原理详解 Apr 14, 2025 pm 11:57 PM

Docker利用Linux内核特性,提供高效、隔离的应用运行环境。其工作原理如下:1. 镜像作为只读模板,包含运行应用所需的一切;2. 联合文件系统(UnionFS)层叠多个文件系统,只存储差异部分,节省空间并加快速度;3. 守护进程管理镜像和容器,客户端用于交互;4. Namespaces和cgroups实现容器隔离和资源限制;5. 多种网络模式支持容器互联。理解这些核心概念,才能更好地利用Docker。

Python vs. JavaScript:社区,图书馆和资源 Python vs. JavaScript:社区,图书馆和资源 Apr 15, 2025 am 12:16 AM

Python和JavaScript在社区、库和资源方面的对比各有优劣。1)Python社区友好,适合初学者,但前端开发资源不如JavaScript丰富。2)Python在数据科学和机器学习库方面强大,JavaScript则在前端开发库和框架上更胜一筹。3)两者的学习资源都丰富,但Python适合从官方文档开始,JavaScript则以MDNWebDocs为佳。选择应基于项目需求和个人兴趣。

minio安装centos兼容性 minio安装centos兼容性 Apr 14, 2025 pm 05:45 PM

MinIO对象存储:CentOS系统下的高性能部署MinIO是一款基于Go语言开发的高性能、分布式对象存储系统,与AmazonS3兼容。它支持多种客户端语言,包括Java、Python、JavaScript和Go。本文将简要介绍MinIO在CentOS系统上的安装和兼容性。CentOS版本兼容性MinIO已在多个CentOS版本上得到验证,包括但不限于:CentOS7.9:提供完整的安装指南,涵盖集群配置、环境准备、配置文件设置、磁盘分区以及MinI

CentOS上PyTorch的分布式训练如何操作 CentOS上PyTorch的分布式训练如何操作 Apr 14, 2025 pm 06:36 PM

在CentOS系统上进行PyTorch分布式训练,需要按照以下步骤操作:PyTorch安装:前提是CentOS系统已安装Python和pip。根据您的CUDA版本,从PyTorch官网获取合适的安装命令。对于仅需CPU的训练,可以使用以下命令:pipinstalltorchtorchvisiontorchaudio如需GPU支持,请确保已安装对应版本的CUDA和cuDNN,并使用相应的PyTorch版本进行安装。分布式环境配置:分布式训练通常需要多台机器或单机多GPU。所

CentOS上PyTorch版本怎么选 CentOS上PyTorch版本怎么选 Apr 14, 2025 pm 06:51 PM

在CentOS系统上安装PyTorch,需要仔细选择合适的版本,并考虑以下几个关键因素:一、系统环境兼容性:操作系统:建议使用CentOS7或更高版本。CUDA与cuDNN:PyTorch版本与CUDA版本密切相关。例如,PyTorch1.9.0需要CUDA11.1,而PyTorch2.0.1则需要CUDA11.3。cuDNN版本也必须与CUDA版本匹配。选择PyTorch版本前,务必确认已安装兼容的CUDA和cuDNN版本。Python版本:PyTorch官方支

centos如何安装nginx centos如何安装nginx Apr 14, 2025 pm 08:06 PM

CentOS 安装 Nginx 需要遵循以下步骤:安装依赖包,如开发工具、pcre-devel 和 openssl-devel。下载 Nginx 源码包,解压后编译安装,并指定安装路径为 /usr/local/nginx。创建 Nginx 用户和用户组,并设置权限。修改配置文件 nginx.conf,配置监听端口和域名/IP 地址。启动 Nginx 服务。需要注意常见的错误,如依赖问题、端口冲突和配置文件错误。性能优化需要根据具体情况调整,如开启缓存和调整 worker 进程数量。

See all articles