IT技术社区_专业开发者平台_最新编程教程与解决方案 -沐辰社区

Python案例如何用Networkx做拓扑排序

wen python案例 163

本文目录导读：

Python案例如何用Networkx做拓扑排序

  1. 基础拓扑排序
  2. 课程安排系统案例
  3. 项目管理依赖排序
  4. 并行任务拓扑排序
  5. 高级功能：多条件拓扑排序
  6. 实用工具函数
  7. 关键要点

我来详细讲解如何使用NetworkX进行拓扑排序,包含多个实际案例。

基础拓扑排序

import networkx as nx
import matplotlib.pyplot as plt
# 创建一个有向无环图(DAG)
G = nx.DiGraph()
# 添加边（任务依赖关系）
edges = [
    ('A', 'B'),  # A → B：A完成后才能开始B
    ('A', 'C'),  # A → C：A完成后才能开始C
    ('B', 'D'),  # B → D：B完成后才能开始D
    ('C', 'D'),  # C → D：C完成后才能开始D
    ('D', 'E')   # D → E：D完成后才能开始E
]
G.add_edges_from(edges)
# 执行拓扑排序
try:
    topological_order = list(nx.topological_sort(G))
    print("拓扑排序结果:", topological_order)
    # 输出: ['A', 'B', 'C', 'D', 'E'] 或 ['A', 'C', 'B', 'D', 'E']
except nx.NetworkXUnfeasible:
    print("图中存在环，无法进行拓扑排序")

课程安排系统案例

import networkx as nx
class CourseScheduler:
    def __init__(self):
        self.graph = nx.DiGraph()
    def add_course_prerequisites(self, course, prerequisites):
        """添加课程及其先修课程"""
        for prereq in prerequisites:
            self.graph.add_edge(prereq, course)  # 先修课程 → 当前课程
    def get_study_order(self):
        """获取学习顺序"""
        try:
            schedule = list(nx.topological_sort(self.graph))
            return schedule
        except nx.NetworkXUnfeasible:
            print("课程依赖关系存在循环，无法安排学习顺序")
            return []
    def check_cycle(self):
        """检查是否有循环依赖"""
        try:
            nx.find_cycle(self.graph)
            return True
        except nx.NetworkXNoCycle:
            return False
    def visualize_schedule(self):
        """可视化课程依赖关系"""
        pos = nx.spring_layout(self.graph)
        nx.draw(self.graph, pos, with_labels=True, 
                node_color='lightblue', 
                node_size=2000,
                arrows=True,
                arrowstyle='->')
        plt.title("课程依赖关系图")
        plt.show()
# 使用案例
scheduler = CourseScheduler()
# 添加课程依赖
scheduler.add_course_prerequisites("Algorithms", ["Data Structures", "Mathematics"])
scheduler.add_course_prerequisites("Data Structures", ["Programming Basics"])
scheduler.add_course_prerequisites("Machine Learning", ["Algorithms", "Statistics"])
scheduler.add_course_prerequisites("Statistics", ["Mathematics"])
scheduler.add_course_prerequisites("Mathematics", ["Programming Basics"])
# 获取学习顺序
study_order = scheduler.get_study_order()
print("推荐学习顺序:", study_order)
# 输出类似: ['Programming Basics', 'Mathematics', 'Data Structures', 'Statistics', 'Algorithms', 'Machine Learning']

项目管理依赖排序

import networkx as nx
from datetime import datetime, timedelta
class ProjectTaskScheduler:
    def __init__(self):
        self.graph = nx.DiGraph()
        self.task_durations = {}
    def add_task(self, task_name, duration_days, dependencies=None):
        """添加任务及其依赖"""
        self.task_durations[task_name] = duration_days
        if dependencies:
            for dep in dependencies:
                self.graph.add_edge(dep, task_name)
        else:
            self.graph.add_node(task_name)
    def get_task_order(self):
        """获取任务执行顺序"""
        try:
            return list(nx.topological_sort(self.graph))
        except nx.NetworkXUnfeasible:
            print("任务依赖存在循环")
            return []
    def get_critical_path(self):
        """获取关键路径（最长路径）"""
        try:
            # 找到所有拓扑排序
            task_order = list(nx.topological_sort(self.graph))
            # 计算每个任务的最早开始时间
            earliest_start = {task: 0 for task in task_order}
            for task in task_order:
                for pred in self.graph.predecessors(task):
                    earliest_start[task] = max(
                        earliest_start[task],
                        earliest_start[pred] + self.task_durations[pred]
                    )
            # 计算每个任务的最晚开始时间
            latest_start = {task: float('inf') for task in task_order}
            for task in reversed(task_order):
                if self.graph.out_degree(task) == 0:
                    latest_start[task] = earliest_start[task]
                else:
                    for succ in self.graph.successors(task):
                        latest_start[task] = min(
                            latest_start[task],
                            latest_start[succ] - self.task_durations[task]
                        )
            # 找出关键路径上的任务（最早开始时间等于最晚开始时间）
            critical_tasks = [task for task in task_order 
                            if earliest_start[task] == latest_start[task]]
            return critical_tasks, earliest_start, latest_start
        except Exception as e:
            print(f"计算关键路径时出错: {e}")
            return [], {}, {}
    def get_project_duration(self):
        """获取项目总持续时间"""
        critical_tasks, _, _ = self.get_critical_path()
        return sum(self.task_durations[task] for task in critical_tasks)
# 使用案例
project = ProjectTaskScheduler()
# 添加项目任务
project.add_task("需求分析", 5)
project.add_task("系统设计", 3, ["需求分析"])
project.add_task("数据库设计", 2, ["系统设计"])
project.add_task("前端开发", 8, ["系统设计"])
project.add_task("后端开发", 10, ["系统设计", "数据库设计"])
project.add_task("测试", 4, ["前端开发", "后端开发"])
project.add_task("部署", 1, ["测试"])
project.add_task("用户培训", 2, ["部署"])
# 获取任务顺序
task_order = project.get_task_order()
print("任务执行顺序:", task_order)
# 获取关键路径
critical_tasks, earliest, latest = project.get_critical_path()
print("\n关键路径上的任务:", critical_tasks)
print("项目总工期:", project.get_project_duration(), "天")
# 显示任务时间安排
print("\n任务时间安排:")
for task in task_order:
    print(f"{task}: 最早开始={earliest[task]}天, 最晚开始={latest[task]}天, "
          f"工期={project.task_durations[task]}天")

并行任务拓扑排序

import networkx as nx
from collections import defaultdict
class ParallelTaskScheduler:
    def __init__(self):
        self.graph = nx.DiGraph()
    def add_dependency(self, from_task, to_task):
        """添加任务依赖"""
        self.graph.add_edge(from_task, to_task)
    def get_parallel_levels(self):
        """获取可以并行执行的任务层级"""
        try:
            topological_order = list(nx.topological_sort(self.graph))
            # 计算每个任务的层级
            levels = {}
            for task in topological_order:
                predecessors = list(self.graph.predecessors(task))
                if not predecessors:
                    levels[task] = 0
                else:
                    levels[task] = max(levels[pred] for pred in predecessors) + 1
            # 按层级分组
            level_groups = defaultdict(list)
            for task, level in levels.items():
                level_groups[level].append(task)
            return dict(level_groups)
        except nx.NetworkXUnfeasible:
            print("存在环，无法并行调度")
            return {}
    def visualize_parallel_tasks(self):
        """可视化并行任务"""
        levels = self.get_parallel_levels()
        fig, (ax1, ax2) = plt.subplots(1, 2, figsize=(12, 6))
        # 子图1：显示DAG
        pos = nx.spring_layout(self.graph)
        nx.draw(self.graph, pos, with_labels=True, 
                node_color='lightgreen', 
                node_size=2000,
                arrows=True,
                ax=ax1)
        ax1.set_title("任务依赖图")
        # 子图2：显示层级
        for level, tasks in levels.items():
            y_pos = [level] * len(tasks)
            x_pos = range(len(tasks))
            ax2.scatter(x_pos, y_pos, s=2000, alpha=0.6)
            for i, task in enumerate(tasks):
                ax2.annotate(task, (i, level), ha='center', va='center')
        ax2.set_xlabel("并行任务")
        ax2.set_ylabel("执行层级")
        ax2.set_title("并行任务层级")
        ax2.grid(True)
        plt.tight_layout()
        plt.show()
# 使用案例
parallel_scheduler = ParallelTaskScheduler()
# 添加依赖关系
parallel_scheduler.add_dependency("Task1", "Task4")
parallel_scheduler.add_dependency("Task1", "Task5")
parallel_scheduler.add_dependency("Task2", "Task5")
parallel_scheduler.add_dependency("Task3", "Task6")
parallel_scheduler.add_dependency("Task4", "Task7")
parallel_scheduler.add_dependency("Task5", "Task7")
parallel_scheduler.add_dependency("Task6", "Task8")
# 获取并行层级
parallel_levels = parallel_scheduler.get_parallel_levels()
print("并行执行计划:")
for level, tasks in sorted(parallel_levels.items()):
    print(f"层级 {level}: {', '.join(tasks)}")

高级功能：多条件拓扑排序

import networkx as nx
class AdvancedTopologicalScheduler:
    def __init__(self):
        self.graph = nx.DiGraph()
        self.priorities = {}  # 任务优先级
    def add_task(self, task, priority=0, dependencies=None):
        """添加带优先级的任务"""
        self.priorities[task] = priority
        if dependencies:
            for dep in dependencies:
                self.graph.add_edge(dep, task)
        else:
            self.graph.add_node(task)
    def topological_sort_with_priority(self):
        """按优先级排序的拓扑排序"""
        try:
            # 获取所有拓扑排序（生成器）
            all_topsorts = nx.all_topological_sorts(self.graph)
            # 根据优先级评分选择最优排序
            best_score = -1
            best_order = None
            for topsort in all_topsorts:
                # 计算当前排序的评分
                score = self._calculate_order_score(topsort)
                if score > best_score:
                    best_score = score
                    best_order = topsort
            return best_order, best_score
        except nx.NetworkXUnfeasible:
            print("存在环")
            return [], 0
    def _calculate_order_score(self, order):
        """计算排序评分"""
        score = 0
        for i, task in enumerate(order):
            # 优先级越高的任务越早完成得分越高
            priority = self.priorities.get(task, 0)
            score += priority * (len(order) - i)
        return score
    def get_possible_orders(self, max_results=5):
        """获取多种可能的排序"""
        try:
            all_topsorts = list(nx.all_topological_sorts(self.graph))
            return all_topsorts[:max_results]
        except nx.NetworkXUnfeasible:
            return []
# 使用案例
advanced_scheduler = AdvancedTopologicalScheduler()
# 添加带优先级的任务
advanced_scheduler.add_task("紧急Bug修复", priority=10)
advanced_scheduler.add_task("功能开发A", priority=5, dependencies=["数据库优化"])
advanced_scheduler.add_task("功能开发B", priority=3, dependencies=["功能开发A"])
advanced_scheduler.add_task("数据库优化", priority=8)
advanced_scheduler.add_task("代码审查", priority=6, dependencies=["功能开发A", "功能开发B"])
advanced_scheduler.add_task("测试", priority=4, dependencies=["代码审查"])
advanced_scheduler.add_task("文档更新", priority=1, dependencies=["功能开发A"])
# 获取最优排序
best_order, score = advanced_scheduler.topological_sort_with_priority()
print(f"最优排序 (评分: {score}): {best_order}")
# 获取所有可能的排序
print("\n所有可能的排序:")
for i, order in enumerate(advanced_scheduler.get_possible_orders(3), 1):
    print(f"方案{i}: {order}")

实用工具函数

def check_dag_health(graph):
    """检查DAG图健康状况"""
    results = {
        'is_dag': nx.is_directed_acyclic_graph(graph),
        'node_count': graph.number_of_nodes(),
        'edge_count': graph.number_of_edges(),
        'has_cycle': False,
        'roots': [],
        'leaves': []
    }
    if not results['is_dag']:
        results['has_cycle'] = True
        try:
            cycle = nx.find_cycle(graph)
            results['cycle'] = cycle
        except nx.NetworkXNoCycle:
            pass
    else:
        # 找出根节点（没有入边的节点）
        results['roots'] = [n for n in graph.nodes() if graph.in_degree(n) == 0]
        # 找出叶子节点（没有出边的节点）
        results['leaves'] = [n for n in graph.nodes() if graph.out_degree(n) == 0]
    return results
def layered_topological_sort(graph):
    """分层拓扑排序（用于并行执行）"""
    if not nx.is_directed_acyclic_graph(graph):
        return None
    # 复制图
    G = graph.copy()
    layers = []
    while G.nodes():
        # 找出当前层（入度为0的节点）
        current_layer = [n for n in G.nodes() if G.in_degree(n) == 0]
        if not current_layer:
            break
        layers.append(current_layer)
        # 移除当前层的节点
        G.remove_nodes_from(current_layer)
    return layers
# 完整的综合示例
def main():
    print("=== 拓扑排序综合示例 ===\n")
    # 1. 创建任务依赖图
    G = nx.DiGraph()
    tasks = [
        ('A', 'B'),
        ('A', 'C'),
        ('B', 'D'),
        ('C', 'D'),
        ('C', 'E'),
        ('D', 'F'),
        ('E', 'F'),
    ]
    G.add_edges_from(tasks)
    # 2. 检查DAG健康
    health = check_dag_health(G)
    print(f"图状态: {'正常' if health['is_dag'] else '有环'}")
    print(f"节点数: {health['node_count']}")
    print(f"边数: {health['edge_count']}")
    print(f"根节点: {health['roots']}")
    print(f"叶子节点: {health['leaves']}")
    # 3. 标准拓扑排序
    if health['is_dag']:
        standard_order = list(nx.topological_sort(G))
        print(f"\n标准拓扑排序: {standard_order}")
        # 4. 分层排序
        layers = layered_topological_sort(G)
        print("\n分层排序 (并行层级):")
        for i, layer in enumerate(layers):
            print(f"  层级 {i+1}: {layer}")
        # 5. 可视化
        pos = nx.spring_layout(G)
        plt.figure(figsize=(10, 6))
        # 为不同层级着色
        colors = ['red', 'blue', 'green', 'orange', 'purple']
        node_colors = []
        node_to_layer = {}
        for i, layer in enumerate(layers):
            for node in layer:
                node_to_layer[node] = i
        for node in G.nodes():
            if node in node_to_layer:
                node_colors.append(colors[node_to_layer[node] % len(colors)])
            else:
                node_colors.append('gray')
        nx.draw(G, pos, with_labels=True, 
                node_color=node_colors, 
                node_size=2000,
                arrows=True,
                font_size=12,
                font_weight='bold')
        plt.title("任务依赖图 (颜色表示执行层级)")
        plt.show()
if __name__ == "__main__":
    main()

关键要点

  1. 核心函数: nx.topological_sort() 是主要函数
  2. 错误处理: 始终使用 try-except 处理环检测
  3. 并行执行: 可以使用分层排序识别可并行任务
  4. 优先级: 可以通过自定义评分函数实现优先级排序
  5. 可视化: NetworkX 提供了丰富的可视化选项

这些案例覆盖了拓扑排序的常见应用场景,你可以根据具体需求选择合适的实现方式。

版权声明：NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 CBA直播 CBA直播 CBA直播 CBA直播 CBA直播 CBA直播 CBA直播 CBA直播 CBA直播 CBA直播 CBA直播 CBA直播 CBA直播 CBA直播 CBA直播 CBA直播 CBA直播 CBA直播 足球直播 足球直播 足球直播 足球直播 足球直播 足球直播 足球直播 足球直播 足球直播 足球直播 足球直播 足球直播 足球直播 足球直播 足球直播 足球直播 足球直播 足球直播 英超直播 英超直播 英超直播 英超直播 英超直播 英超直播 英超直播 英超直播 英超直播 英超直播 英超直播 英超直播 英超直播 英超直播 英超直播 英超直播 英超直播 英超直播

上一篇Python案例如何用Networkx做关键路径

下一篇Python案例如何用Networkx做计划评审

相关文章

抱歉，评论功能暂时关闭!

友情链接攻略窝 · 礼包窝 · 配置窝 · 换算窝 · 算法窝 · 查文窝 · 手册窝 · 问答窝 · 全文搜 · 手册库
相关站点礼包窝 · 查文窝