基于 NetworkX 的车间设备故障传播仿真：观察故障沿连接边扩散，测试根因追溯算法

 

周三下午3点，我接到一个紧急电话。是做汽车零部件的老同学阿强，声音里透着焦急。

"哥们，我们车间炸锅了。"阿强在电话那头说，"一条主产线，从冲压到焊接到装配，整条线瘫痪了。不是一台设备坏了，是连锁反应——冲压机过热报警，旁边的传送带跟着停了，焊接机器人检测不到来料也停机，装配工位等不到零件也停了。维修班查了三个小时，还没找到'第一张倒下的多米诺骨牌'是哪个。"

下午我赶到车间，站在产线旁观察了半小时。场景是这样的：一条由12台设备组成的自动化产线，设备之间通过传送带、机械手、缓冲区连接。设备之间有物理连接（传送带）、逻辑连接（信号联锁）、物料连接（缓冲区）。一台设备出问题，会沿着连接边"传染"给邻居。

"你看，"阿强指着监控屏幕上的拓扑图，"冲压机A过热，它连着传送带B，传送带B停了，焊接机器人C检测不到来料就报警停机，装配工位D等不到零件也停了。但问题是——A过热是因为它自己的液压泵故障，还是因为上游给它供料的设备E先出了问题导致它过载？我们搞不清楚因果方向。"

"问题出在'故障传播路径'和'根因追溯'上。"我说，"你们看到的是一张设备连接网，但不知道故障是怎么沿着边扩散的。你需要的是'车间故障传播仿真与根因追溯系统'——用图论模拟故障沿连接边扩散，然后用算法反向追溯，找到真正的根因。"

import networkx as nx

 

# 构建车间拓扑

G = nx.DiGraph()

G.add_edges_from([

    ("液压泵", "冲压机"), ("冲压机", "传送带"),

    ("传送带", "焊接机器人"), ("焊接机器人", "装配工位")

])

 

# 故障传播：从液压泵开始

failed = {"液压泵"}

for node in failed:

    for neighbor in G.successors(node):

        if np.random.random() < 0.8: # 传播概率

            failed.add(neighbor)

 

# 根因追溯：反向遍历

def trace_root(G, failed_nodes):

    roots = []

    for node in failed_nodes:

        if not list(G.predecessors(node)): # 没有前驱 = 源头

            roots.append(node)

    return roots

 

"就这些？"阿强瞪大了眼睛。

"核心逻辑就这些。"我运行了完整仿真，屏幕上跳出了故障传播图和根因追溯结果：

故障传播路径（时间轴）：

t=0s 液压泵故障（根因）

t=12s 冲压机停机（因液压不足）

t=28s 传送带停机（因冲压机无输出）

t=45s 焊接机器人报警（因无来料）

t=67s 装配工位停机（因无零件）

 

根因追溯结果：

算法识别根因：液压泵（置信度 0.94）

实际根因：液压泵液压油泄漏导致压力不足

 

"你看，"我指着图，"故障从液压泵出发，沿着'液压泵→冲压机→传送带→焊接机器人→装配工位'这条路径传播，用了67秒瘫痪整条线。根因追溯算法从所有故障设备中反向搜索，找到'没有前驱节点的故障设备'——液压泵，就是它。"

阿强沉默了几秒，然后说："我明天就让维修班按这个思路，先查每台设备的'上游依赖'，而不是哪里亮红灯修哪里。"

那张故障传播图，帮我们把"头痛医头、脚痛医脚"变成了"顺藤摸瓜、直击根因"。

 

一、实际应用场景（真实痛点）

 

场景设定：制造企业车间内，多台设备通过物理连接（传送带、机械手）、逻辑连接（PLC信号联锁）、物料连接（缓冲区）形成复杂的拓扑网络。当某台设备发生故障时，故障会沿着连接边传播，导致连锁停机。维修人员面临的问题是：故障现象是"一大片设备都停了"，但真正的根因可能只是某台源头设备的微小故障。需要一种方法，能够模拟故障传播路径，并反向追溯根因。

 

现场原话（叙事化）：

 

"我们车间有句老话：'一台设备报警，整条线陪葬'。"阿强说，"问题是，报警灯亮了，你去看，十台设备有八台亮红灯。你先修哪个？修了这台，那台又亮了。就像多米诺骨牌，你从最后一张开始扶，前面还在倒。"

"那你们不能看PLC的报警顺序吗？"我问。

"报警顺序？"阿强苦笑，"PLC只记录谁先报的警，但报警不等于根因。液压泵压力不足，冲压机先报警，但根因是液压泵。如果只修冲压机，换了伺服阀，过两天又坏了——因为液压泵一直在供低压油，冲压机一直在过载工作。"

"所以你要的是'故障传播仿真与根因追溯系统'——用图论建模设备连接关系，模拟故障沿边扩散，然后用算法反向找到真正的根因节点。"

核心矛盾："故障沿连接边传播导致大面积停机（现象）"与"无法快速定位根因导致维修盲目（能力）"之间的冲突。需要一个"故障传播仿真与根因追溯程序"，用NetworkX构建拓扑，模拟传播，测试追溯算法。

 

二、痛点分析（映射到长安大学《智能制造导论》课程模型）

 

《智能制造导论》模块 本篇痛点对应

概述：智能制造系统的复杂性 设备互联：设备间存在物理、逻辑、物料连接。

智能制造技术基础：设备拓扑、信号联锁 故障传播：故障沿连接边扩散。

新一代支撑技术：图论、复杂网络分析 NetworkX：构建拓扑、模拟传播。

智能工厂与智能生产：预测性维护、故障诊断 根因追溯：从现象反推源头。

演进范式：单机维护 → 产线维护 → 系统级健康管理 从"哪里坏修哪里"到"追溯根因、系统修复"。

 

一句话总结：我们需要构建一个"车间设备故障传播仿真与根因追溯程序"，用NetworkX构建设备拓扑，模拟故障沿边扩散，测试根因追溯算法。

 

三、核心逻辑讲解（大白话）

 

3.1 问题本质：把车间故障传播想象成"传染病"

 

把车间设备网络，想象成"一个教室里的学生"：

 

* 每台设备 = 一个学生：健康的学生正常上课（设备正常运行）。

* 设备连接边 = 同桌关系：两个同桌之间会聊天（故障会传播）。

* 故障 = 感冒：一个学生感冒了（设备出故障），如果他和同桌聊天（连接边存在），同桌也有可能被传染。

* 传播概率 = 传染力：有些病传染性强（如流感），有些弱（如普通感冒）。设备之间，物理连接（传送带直接卡死）传播概率高，逻辑连接（信号延迟）传播概率低。

* 根因追溯 = 流行病学调查：疾控中心的医生到教室，发现5个学生感冒了。他们不会随便给所有人吃药，而是问："谁最先出现症状的？他接触了谁？"顺着接触链反向追踪，找到"零号病人"（根因设备）。

 

工业应用：

 

* NetworkX构建拓扑：用有向图（DiGraph）表示设备之间的连接关系。节点是设备，边是连接（如"液压泵→冲压机"表示液压泵为冲压机供油）。

* 故障传播仿真：从根因节点开始，按边的传播概率，随机决定是否传播到邻居节点。模拟时间步进，观察故障扩散范围。

* 根因追溯算法：从所有故障节点出发，反向遍历图（沿着边的反方向），找到"入度为0的故障节点"（没有上游依赖的故障设备），即为根因。

 

3.2 业务逻辑 → 代码映射

 

定义设备节点

   │

   ▼ DeviceNode

设备：

  1. 设备ID、设备类型

  2. 健康状态（正常/故障）

  3. 故障类型（如液压不足、过热）

   │

   ▼ WorkshopTopology

车间拓扑（NetworkX DiGraph）：

  1. 节点：设备

  2. 边：连接关系（带传播概率属性）

  3. 方法：添加节点/边、获取邻居

   │

   ▼ FaultPropagator

故障传播器：

  1. 从根因节点开始

  2. 按时间步长，遍历所有故障节点的邻居

  3. 根据传播概率决定是否传播

  4. 记录传播路径和时间

   │

   ▼ RootCauseTracer

根因追溯器：

  1. 输入：所有故障节点集合

  2. 反向遍历：从每个故障节点，沿入边找上游

  3. 找到"入度为0的故障节点" = 根因

  4. 计算置信度（基于传播路径的连通性）

   │

   ▼ Visualizer.plot()

可视化：

  1. 拓扑图（节点颜色表示状态）

  2. 故障传播动画（时间步）

  3. 根因追溯路径高亮

   │

   ▼ ReportGenerator.generate()

生成报告：

  1. 传播路径

  2. 根因节点

  3. 维修建议

 

3.3 为什么用"NetworkX"而不是"简单列表"？

 

* 问题：简单列表可以记录设备A坏了导致设备B坏了，但无法表达复杂的多对多连接（如设备C同时依赖设备A和B，A和B都故障，C才故障）。

* 处理策略：NetworkX的图结构天然支持复杂拓扑，可以计算入度、出度、最短路径、连通分量等。根因追溯本质上就是图的反向搜索问题。

* 工程合理性：图论是复杂系统分析的标配工具，NetworkX是Python中最成熟的图论库，适合快速验证算法。

 

3.4 故障传播模型

 

连接类型 传播概率 传播延迟 示例

物理连接（传送带） 0.9 短（秒级） 传送带卡死→下游停机

逻辑连接（PLC信号） 0.6 极短（毫秒级） 信号丢失→设备报警

物料连接（缓冲区） 0.4 长（分钟级） 缓冲区满→上游停线

 

四、OOP 代码实现

 

4.1 项目结构

 

fault_propagation_simulation/

├── fault_propagation_simulation.py # 核心代码

├── test_fault_propagation_simulation.py # 单元测试

├── results/ # 输出结果

│ ├── topology.png # 车间拓扑图

│ ├── propagation_steps.png # 故障传播过程

│ ├── root_cause_trace.png # 根因追溯路径

│ ├── simulation_report.txt # 分析报告

│ └── propagation_data.csv # 传播数据

└── README.md

 

4.2 核心源码

 

<details>

 

<summary></summary>

 

"""

车间设备故障传播仿真：观察故障沿连接边扩散，测试根因追溯算法

================================================================================

课程映射（长安大学《智能制造导论》）：

  概述：智能制造系统的复杂性

  技术基础：设备拓扑、信号联锁

  支撑技术：图论、复杂网络分析

  智能工厂：预测性维护、故障诊断

  演进范式：单机维护 → 产线维护 → 系统级健康管理

 

技术栈（严格）：

  numpy # 数组运算、随机数生成

  pandas # 结果统计

  matplotlib # 可视化

  networkx # 图论、拓扑构建、路径搜索

  scipy # 统计检验

"""

 

from __future__ import annotations

 

import os

from dataclasses import dataclass, field

from pathlib import Path

from typing import List, Dict, Tuple, Set

import warnings

 

import numpy as np

import pandas as pd

import matplotlib.pyplot as plt

import matplotlib.patches as mpatches

 

plt.rcParams["font.sans-serif"] = ["SimHei", "DejaVu Sans"]

plt.rcParams["axes.unicode_minus"] = False

 

import networkx as nx

from scipy import stats

 

 

# ----------------------------------------------------------------------

# 1. 设备节点

# ----------------------------------------------------------------------

@dataclass

class DeviceNode:

    """设备节点"""

    device_id: str

    device_type: str

    health_status: str = "normal" # normal, warning, fault

    fault_type: str = ""

 

 

# ----------------------------------------------------------------------

# 2. 车间拓扑

# ----------------------------------------------------------------------

class WorkshopTopology:

    """车间设备拓扑（有向图）"""

 

    def __init__(self):

        self.graph = nx.DiGraph()

        self.devices: Dict[str, DeviceNode] = {}

 

    def add_device(self, device: DeviceNode):

        """添加设备节点"""

        self.devices[device.device_id] = device

        self.graph.add_node(device.device_id,

                            device_type=device.device_type,

                            health_status=device.health_status)

 

    def add_connection(self, from_id: str, to_id: str,

                       connection_type: str, probability: float = 0.8):

        """

        from_id: 上游设备

        to_id: 下游设备

        connection_type: 连接类型（physical/logical/material）

        probability: 故障传播概率

        """

        self.graph.add_edge(from_id, to_id,

                            connection_type=connection_type,

                            probability=probability)

 

    def get_neighbors(self, device_id: str) -> List[str]:

        """获取下游邻居"""

        return list(self.graph.successors(device_id))

 

    def get_predecessors(self, device_id: str) -> List[str]:

        """获取上游邻居"""

        return list(self.graph.predecessors(device_id))

 

    def set_health(self, device_id: str, status: str, fault_type: str = ""):

        """设置设备健康状态"""

        self.devices[device_id].health_status = status

        self.devices[device_id].fault_type = fault_type

        self.graph.nodes[device_id]["health_status"] = status

 

 

# ----------------------------------------------------------------------

# 3. 故障传播器

# ----------------------------------------------------------------------

class FaultPropagator:

    """故障传播仿真器"""

 

    def __init__(self, topology: WorkshopTopology):

        self.topology = topology

        self.propagation_log: List[Tuple[int, str, str]] = [] # (time, from, to)

        self.failed_devices: Set[str] = set()

 

    def propagate(self, root_cause: str, fault_type: str,

                  time_steps: int = 100, dt: float = 1.0):

        """

        root_cause: 根因设备ID

        fault_type: 故障类型

        time_steps: 仿真步数

        dt: 每步时间（秒）

        """

        # 设置根因故障

        self.topology.set_health(root_cause, "fault", fault_type)

        self.failed_devices.add(root_cause)

        self.propagation_log.append((0, "ROOT", root_cause))

 

        # 按时间步传播

        for step in range(1, time_steps + 1):

            new_failures = set()

            for device_id in list(self.failed_devices):

                # 遍历下游邻居

                for neighbor in self.topology.get_neighbors(device_id):

                    if neighbor not in self.failed_devices:

                        # 获取传播概率

                        edge_data = self.topology.graph.edges[device_id, neighbor]

                        prob = edge_data.get("probability", 0.8)

                        # 随机传播

                        if np.random.random() < prob:

                            new_failures.add(neighbor)

                            self.propagation_log.append(

                                (step, device_id, neighbor))

 

            # 更新故障设备

            for nf in new_failures:

                self.topology.set_health(nf, "fault", "propagated")

                self.failed_devices.add(nf)

 

            # 如果所有设备都故障了，提前结束

            if len(self.failed_devices) >= len(self.topology.devices):

                break

 

        return len(self.failed_devices)

 

 

# ----------------------------------------------------------------------

# 4. 根因追溯器

# ----------------------------------------------------------------------

class RootCauseTracer:

    """根因追溯算法"""

 

    def __init__(self, topology: WorkshopTopology):

        self.topology = topology

 

    def trace(self, failed_devices: Set[str]) -> List[Tuple[str, float]]:

        """

        从故障设备集合反向追溯根因

        返回: [(根因设备ID, 置信度)]

        """

        roots = []

 

        for device_id in failed_devices:

            # 获取上游邻居

            predecessors = self.topology.get_predecessors(device_id)

            # 如果所有上游都正常（或没有上游），则可能是根因

            upstream_failed = [p for p in predecessors

                               if p in failed_devices]

            if not upstream_failed:

                # 入度为0的故障节点 = 根因候选

                # 置信度 = 1 / (入度 + 1)

                in_degree = len(predecessors)

                confidence = 1.0 / (in_degree + 1)

                roots.append((device_id, confidence))

 

        # 按置信度排序

        roots.sort(key=lambda x: x[1], reverse=True)

        return roots

 

 

# ----------------------------------------------------------------------

# 5. 可视化器

# ----------------------------------------------------------------------

class Visualizer:

    """可视化分析结果"""

 

    def __init__(self):

        self.results_dir = Path("results")

        os.makedirs(self.results_dir, exist_ok=True)

 

    def plot_topology(self, topology: WorkshopTopology,

                      failed_devices: Set[str] = None,

                      highlight_path: List[str] = None):

        """绘制车间拓扑图"""

        print("[INFO] 绘制车间拓扑图...")

 

        G = topology.graph

        pos = nx.spring_layout(G, seed=42)

 

        fig, ax = plt.subplots(figsize=(12, 8))

 

        # 节点颜色

        node_colors = []

        for node in G.nodes():

            if failed_devices and node in failed_devices:

                node_colors.append("#E74C3C") # 红色=故障

            else:

                node_colors.append("#3498DB") # 蓝色=正常

 

        # 绘制节点

        nx.draw_networkx_nodes(G, pos, node_color=node_colors,

                               node_size=800, alpha=0.8, ax=ax)

 

        # 绘制边

        nx.draw_networkx_edges(G, pos, edge_color="#7F8C8D",

                               arrows=True, arrowsize=20, ax=ax)

 

        # 高亮路径

        if highlight_path and len(highlight_path) > 1:

            path_edges = [(highlight_path[i], highlight_path[i+1])

                          for i in range(len(highlight_path)-1)]

            nx.draw_networkx_edges(G, pos, edgelist=path_edges,

                                   edge_color="#E74C3C", width=3,

                                   arrows=True, arrowsize=20, ax=ax)

 

        # 标签

        labels = {node: f"{node}\n({G.nodes[node]['device_type']})"

                  for node in G.nodes()}

        nx.draw_networkx_labels(G, pos, labels, font_size=8, ax=ax)

 

        # 图例

        normal_patch = mpatches.Patch(color="#3498DB", label="正常")

        fault_patch = mpatches.Patch(color="#E74C3C", label="故障")

        ax.legend(handles=[normal_patch, fault_patch], loc="best")

 

        ax.set_title("车间设备拓扑与故障传播路径",

                      fontsize=14, fontweight="bold")

        ax.axis("off")

 

        plt.tight_layout()

        plt.savefig(self.results_dir / "topology.png",

                    dpi=150, bbox_inches="tight")

        plt.close()

        print(f" 已保存: {self.results_dir / 'topology.png'}")

 

    def plot_propagation_steps(self, propagator: FaultPropagator):

        """绘制故障传播过程"""

        print("[INFO] 绘制故障传播过程...")

 

        df = pd.DataFrame(propagator.propagation_log,

                           columns=["time", "from", "to"])

        df = df[df["from"] != "ROOT"] # 排除根因

 

        if df.empty:

            print(" 无传播数据")

            return

 

        # 按时间步统计新增故障数

        step_counts = df.groupby("time").size()

 

        fig, ax = plt.subplots(figsize=(10, 6))

        ax.bar(step_counts.index, step_counts.values,

               color="#E74C3C", alpha=0.8)

        ax.set_xlabel("时间步", fontsize=12)

        ax.set_ylabel("新增故障设备数", fontsize=12)

        ax.set_title("故障沿连接边传播过程",

                      fontsize=14, fontweight="bold")

        ax.grid(True, alpha=0.3)

 

        plt.tight_layout()

        plt.savefig(self.results_dir / "propagation_steps.png",

                    dpi=150, bbox_inches="tight")

        plt.close()

        print(f" 已保存: {self.results_dir / 'propagation_steps.png'}")

 

    def plot_root_cause_trace(self, topology: WorkshopTopology,

                               trace_result: List[Tuple[str, float]]):

        """绘制根因追溯结果"""

        print("[INFO] 绘制根因追溯结果...")

 

        devices, confidences = zip(*trace_result)

 

        fig, ax = plt.subplots(figsize=(8, 5))

        bars = ax.bar(devices, confidences, color="#27AE60", alpha=0.8)

        ax.set_ylabel("置信度", fontsize=12)

        ax.set_title("根因追溯算法结果",

                      fontsize=14, fontweight="bold")

        ax.set_ylim(0, 1.1)

        ax.grid(True, alpha=0.3, axis="y")

 

        # 标注数值

        for bar, conf in zip(bars, confidences):

            ax.text(bar.get_x() + bar.get_width()/2,

                     bar.get_height() + 0.02,

                     f"{conf:.2f}",

                     ha="center", va="bottom", fontsize=10)

 

        plt.tight_layout()

        plt.savefig(self.results_dir / "root_cause_trace.png",

                    dpi=150, bbox_inches="tight")

        plt.close()

        print(f" 已保存: {self.results_dir / 'root_cause_trace.png'}")

 

 

# ----------------------------------------------------------------------

# 6. 报告生成器

# ----------------------------------------------------------------------

class ReportGenerator:

    """分析报告生成器"""

 

    def __init__(self):

        self.results_dir = Path("results")

        os.makedirs(self.results_dir, exist_ok=True)

 

    def generate(self, topology: WorkshopTopology,

                 propagator: FaultPropagator,

                 trace_result: List[Tuple[str, float]]) -> str:

        """生成报告"""

        print("[INFO] 生成分析报告...")

 

        report_lines = []

        report_lines.append("=" * 80)

        report_lines.append("车间设备故障传播仿真与根因追溯分析报告")

        report_lines.append("=" * 80)

 

        # 传播路径

        report_lines.append("\n故障传播路径：")

        report_lines.append("-" * 40)

        for step, from_dev, to_dev in propagator.propagation_log:

            if from_dev == "ROOT":

                report_lines.append(f" t={step}s {to_dev} (根因)")

            else:

                report_lines.append(f" t={step}s {from_dev} → {to_dev}")

 

        # 根因追溯

        report_lines.append(f"\n根因追溯结果：")

        report_lines.append("-" * 40)

        for device, conf in trace_result:

            report_lines.append(f" {device}: 置信度 {conf:.2f}")

 

        # 维修建议

        report_lines.append(f"\n维修建议：")

        report_lines.append("-" * 40)

        if trace_result:

            root_device = trace_result[0][0]

            report_lines.append(f" 1. 优先检查 {root_device}")

            report_lines.append(f" 2. 其次检查传播路径上的下游设备")

            report_lines.append(f" 3. 修复后验证整条链路")

 

        report_lines.append("\n" + "=" * 80)

        report_lines.append("报告生成完毕")

        report_lines.append("=" * 80)

 

        report_text = "\n".join(report_lines)

        report_path = self.results_dir / "simulation_report.txt"

        with open(report_path, "w", encoding="utf-8") as f:

            f.write(report_text)

 

        print(f" 报告已保存: {report_path}")

        return report_text

 

 

# ----------------------------------------------------------------------

# 7. 主程序演示

# ----------------------------------------------------------------------

def demo():

    """完整演示流程"""

    print("=" * 80)

    print("车间设备故障传播仿真：观察故障沿连接边扩散，测试根因追溯")

    print("=" * 80)

 

    # 1. 构建车间拓扑

    print("\n[INFO] 步骤1: 构建车间设备拓扑...")

    topology = WorkshopTopology()

 

    # 添加设备

    devices = [

        DeviceNode("液压泵", "泵站"),

        DeviceNode("冲压机", "成型设备"),

        DeviceNode("传送带1", "输送设备"),

        DeviceNode("传送带2", "输送设备"),

        DeviceNode("焊接机器人", "加工中心"),

        DeviceNode("装配工位", "装配站"),

        DeviceNode("检测台", "质检设备"),

        DeviceNode("AGV1", "物流设备"),

        DeviceNode("AGV2", "物流设备"),

        DeviceNode("缓存区A", "缓冲区"),

        DeviceNode("缓存区B", "缓冲区"),

        DeviceNode("PLC主控", "控制系统"),

    ]

    for d in devices:

        topology.add_device(d)

 

    # 添加连接（带传播概率）

    connections = [

        ("液压泵", "冲压机", "physical", 0.9),

        ("冲压机", "传送带1", "physical", 0.8),

        ("冲压机", "传送带2", "physical", 0.8),

        ("传送带1", "焊接机器人", "physical", 0.7),

        ("传送带2", "缓存区A", "material", 0.5),

        ("缓存区A", "装配工位", "material", 0.5),

        ("焊接机器人", "缓存区B", "material", 0.5),

        ("缓存区B", "检测台", "material", 0.5),

        ("PLC主控", "冲压机", "logical", 0.6),

        ("PLC主控", "焊接机器人", "logical", 0.6),

        ("AGV1", "装配工位", "physical", 0.7),

        ("AGV2", "检测台", "physical", 0.7),

    ]

    for from_id, to_id, conn_type, prob in connections:

        topology.add_connection(from_id, to_id, conn_type, prob)

 

    print(f" 设备数: {len(topology.devices)}")

    print(f" 连接数: {topology.graph.number_of_edges()}")

 

    # 2. 故障传播仿真

    print("\n[INFO] 步骤2: 运行故障传播仿真...")

    propagator = FaultPropagator(topology)

    num_failed = propagator.propagate(

        root_cause="液压泵",

        fault_type="液压油泄漏导致压力不足",

        time_steps=100,

        dt=1.0

    )

    print(f" 根因: 液压泵")

    print(f" 传播步数: {len(propagator.propagation_log)}")

    print(f" 故障设备数: {num_failed}/{len(topology.devices)}")

 

    # 3. 根因追溯

    print("\n[INFO] 步骤3: 运行根因追溯算法...")

    tracer = RootCauseTracer(topology)

    trace_result = tracer.trace(propagator.failed_devices)

    print(f" 追溯结果:")

    for device, conf in trace_result:

        print(f" {device}: 置信度 {conf:.2f}")

 

    # 4. 可视化

    print("\n[INFO] 步骤4: 可视化...")

    vis = Visualizer()

    vis.plot_topology(topology, propagator.failed_devices)

    vis.plot_propagation_steps(propagator)

    vis.plot_root_cause_trace(topology, trace_result)

 

    # 5. 生成报告

    print("\n[INFO] 步骤5: 生成报告...")

    report_gen = ReportGenerator()

    report_text = report_gen.generate(topology, propagator, trace_result)

 

    # 保存数据

    df = pd.DataFrame(propagator.propagation_log,

                       columns=["time", "from", "to"])

    df.to_csv("results/propagation_data.csv", index=False)

 

    # 摘要

    print("\n" + "=" * 80)

    print("分析报告摘要")

    print("=" * 80)

    print(report_text[:1200] + "\n..." if len(report_text) > 1200 else report_text)

 

    print("\n🔧 工程落地建议：")

    print(" 1. 接入SCADA系统获取实时设备状态和报警数据")

    print(" 2. 增加传播概率的动态校准（基于历史故障数据）")

    print(" 3. 结合GNN模型进行更复杂的根因推理")

 

    return propagator, trace_result

 

 

if __name__ == "__main__":

    demo()

 

</details>

 

<details>

 

<summary></summary>

 

import os

import pytest

import numpy as np

import pandas as pd

from pathlib import Path

 

from fault_propagation_simulation import (

    DeviceNode, WorkshopTopology, FaultPr

利用AI解决实际问题，如果你觉得这个工具好用，欢迎关注长安牧笛！