企业官网建设流程全解析

车间拓扑改造仿真：增删设备节点，评估网络连通性与物料输送效率

周三上午10点，工艺工程师老杨拿着一张车间布局图，在会议室白板前站了快半小时了。

"厂长说要提产30%，让我重新规划车间布局。"老杨转过身，指着白板上的图，"现在的问题是——我们的车间网络'太老'了。这条产线当初是按年产5万台设计的，现在要冲8万台，物流路径不够用了。"

"具体哪里不够？"生产总监问。

"你看，"老杨在图上画了几条红线，"从原材料入库到成品出库，物料要经过7个节点。其中3号AGV转运站是瓶颈——所有物料都要从它那里过。如果提产30%，这个节点直接堵死。"

"那就加一条线路，绕过它？"厂长说。

"加了之后，整条网络的连通性会不会受影响？物料输送路径会不会变长？"老杨摇头，"我需要的是先'在纸上'把改造方案跑一遍——删掉瓶颈节点、加几条新线路，看看改造后的网络还通不通、物料从入库到出库要多久、有没有新的瓶颈出现。"

"你需要的是车间拓扑改造仿真。"我打开编辑器，"用NetworkX建一个有向图——节点是设备（入库口、加工中心、AGV站、缓存区、出库口），边是物流路径，权重是输送时间或距离。然后你'删'几个节点、'加'几条边，程序自动计算改造前后的网络连通性、最短路径、瓶颈节点。"

import networkx as nx

# 1. 构建当前车间拓扑

G = nx.DiGraph()

G.add_edge("入库", "加工A", weight=5)

G.add_edge("加工A", "AGV3", weight=3) # 瓶颈节点

G.add_edge("AGV3", "加工B", weight=4)

G.add_edge("加工B", "出库", weight=6)

# 2. 模拟改造：新增一条路径绕过AGV3

G.add_edge("加工A", "加工B", weight=8) # 新路径

# 3. 评估连通性

print(f"网络连通: {nx.is_weakly_connected(G)}")

print(f"最短路径: {nx.shortest_path(G, '入库', '出库', weight='weight')}")

"就这些？"老杨瞪大了眼睛。

"核心逻辑就这些。"我运行了完整仿真，屏幕上跳出了改造前后的网络拓扑对比图和路径分析表：

改造方案: 新增路径"加工A→加工B"，移除瓶颈AGV3

─────────────────────────────────────────────

指标 改造前 改造后 变化

网络连通性 ✅ 连通 ✅ 连通 不变

最短路径长度 18 min 19 min +1 min（可接受）

瓶颈节点 AGV3 无 ✅ 消除

路径冗余度 1条 2条 ✅ 增加容错

节点平均度 1.6 2.0 ✅ 网络更健壮

"你看，"我指着对比图，"删掉AGV3、新增'加工A→加工B'的直接路径后，网络依然连通，最短路径只多了1分钟，但瓶颈消除了——而且现在从入库到出库有两条路径可选，一条断了还有另一条。"

老杨把拓扑图打印出来，贴在车间改造方案的封面上："这就是我要的——改造之前先'数字验证'一遍，确保不会改完之后更糟。"

那张网络对比图，帮我们把"车间要改造"这个模糊的指令，变成了"删掉AGV3、加一条直接路径、连通性不变、路径冗余度翻倍"的精确方案。

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

场景设定：制造企业需要对现有车间进行拓扑改造——新增设备（如增加一台加工中心）、删减设备（如淘汰老旧产线）、调整物流路径。改造前需要评估：改造后的车间网络是否仍然连通？物料输送路径是否变长？会不会出现新的瓶颈？ 如果直接施工后再发现问题，停工损失巨大。

现场原话（叙事化）：

"我们车间有句老话：'动线一动，全线震动'。"老杨说，"问题是，厂长说'加两台设备'，他不管加完之后物料怎么走。我得确保加完之后，从原材料到成品的每一条路径都还通、都还快。"

"那你们没有数字孪生平台吗？"我问。

"有，但那个平台是给管理层'看'的，不是给工程师'算'的。"老杨摇头，"它展示3D车间模型，但不能回答'如果我把3号AGV站拆了，最短路径会变成多少、瓶颈会转移到哪里'。我需要的是What-if仿真——'如果网络变成这样，会怎样'。"

"所以你要的是车间拓扑改造仿真与连通性评估程序——用NetworkX构建车间物流网络，模拟增删节点和边，评估改造前后的连通性和输送效率。"

核心矛盾："车间改造需要提前验证拓扑方案"与"现有系统只能展示静态布局，缺乏动态网络分析"之间的冲突。需要一个"车间拓扑改造仿真与评估程序"，用图论方法量化改造影响。

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

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

概述：车间布局与物流规划 车间拓扑：设备布局决定物流效率。

智能制造技术基础：物料输送系统 物流网络：AGV、传送带、缓存区构成的运输网络。

新一代支撑技术：数字孪生、图论算法 网络仿真：用图模型模拟拓扑变化。

智能工厂与智能生产：柔性产线、可重构制造 拓扑重构：根据产能需求动态调整车间布局。

演进范式：固定布局 → 可重构布局 → 自适应布局 从"建好就不动"到"按需调整"，用仿真验证调整方案。

一句话总结：我们需要构建一个"车间拓扑改造仿真与评估程序"，用NetworkX构建物流网络，模拟节点增删，评估连通性和输送效率。

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

3.1 问题本质：把车间想象成"地铁线路图"

把车间物流网络，想象成"城市的地铁线路图"：

* 设备节点 = 地铁站：每个站（入库口、加工中心、AGV站、缓存区、出库口）就是一个节点。

* 物流路径 = 地铁线路：连接两个站的轨道，有长度（输送时间/距离）。

* 拓扑改造 = 修新线/拆旧线/建新站/关旧站：

* 新增设备 = 建新地铁站：在图上加一个节点，再修几条连接线。

* 删减设备 = 关闭地铁站：把节点和所有连接线从图上删掉。

* 连通性 = 还能不能坐地铁到目的地：从入库口到出库口，还有没有路？

* 最短路径 = 最快坐地铁方案：从A到B，走哪条路最快？

* 瓶颈 = 换乘站太挤：某个节点经过的路径太多，流量超过容量。

工业应用：

* 构建拓扑：用NetworkX的

"DiGraph"（有向图）表示车间——节点是设备，边是物流路径，边的权重是输送时间（分钟）。

* 模拟改造：

"add_node()"新增设备、

"remove_node()"删减设备、

"add_edge()"新增路径。

* 连通性评估：

"nx.is_weakly_connected()"检查全网是否连通；

"nx.all_pairs_shortest_path_length()"计算所有节点对的最短路径。

* 效率评估：对比改造前后的平均最短路径长度、瓶颈节点（度中心性最高的节点）、路径冗余度（两个节点之间有多少条独立路径）。

3.2 业务逻辑 → 代码映射

构建车间拓扑

│

▼ ShopFloorNode

节点：

1. 设备ID、类型（入库/加工/AGV/缓存/出库）

2. 处理时间（在该节点的停留时间）

│

▼ ShopFloorNetwork

网络（NetworkX DiGraph）：

1. 节点：设备

2. 有向边：物流路径，权重=输送时间

│

▼ TopologyModifier

拓扑修改器：

1. add_node() / remove_node()

2. add_edge() / remove_edge()

│

▼ ConnectivityEvaluator

连通性评估器：

1. 全网连通性检查

2. 节点对可达性矩阵

3. 连通分量数量

│

▼ EfficiencyEvaluator

效率评估器：

1. 最短路径长度（Dijkstra）

2. 平均路径长度

3. 瓶颈节点识别（度中心性）

4. 路径冗余度

│

▼ Visualizer.plot()

可视化：

1. 改造前拓扑图

2. 改造后拓扑图

3. 路径长度对比柱状图

│

▼ ReportGenerator.generate()

生成报告：

1. 连通性对比

2. 效率指标对比

3. 改造建议

3.3 为什么用"有向图"而不是"列表"？

* 问题：车间物流有方向——物料从入库到出库是单向流动，但AGV可能双向行驶。列表只能记录"设备A旁边是设备B"，无法表达方向、权重和多条并行路径。

* 处理策略：NetworkX的有向图天然支持方向（

"A→B"和

"B→A"可以是不同的边）、权重（输送时间）、多路径（从A到B可以有多条边）。

* 工程合理性：图论是物流网络分析的标准方法，NetworkX提供了成熟的连通性、最短路径、中心性算法，无需自己实现。

3.4 连通性 vs 效率

* 连通性："能不能到"——布尔值。如果改造后从入库到出库不通了，这个方案直接否决。

* 效率："快不快"——数值。连通只是及格线，效率才是关键。用最短路径长度衡量。

四、OOP 代码实现

4.1 项目结构

shop_floor_topology/

├── shop_floor_topology.py # 核心代码

├── test_shop_floor_topology.py # 单元测试

├── results/ # 输出结果

│ ├── topology_before.png # 改造前拓扑图

│ ├── topology_after.png # 改造后拓扑图

│ ├── path_length_comparison.png # 路径长度对比

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

│ └── evaluation_results.csv # 评估指标数据

└── README.md

4.2 核心源码

<details>

<summary></summary>

"""

车间拓扑改造仿真：增删设备节点，评估网络连通性与物料输送效率

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

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

概述：车间布局与物流规划

技术基础：物料输送系统

支撑技术：数字孪生、图论算法

智能工厂：柔性产线、可重构制造

演进范式：固定布局 → 可重构布局 → 自适应布局

技术栈（严格）：

numpy # 数组运算

pandas # 结果统计

matplotlib # 可视化

networkx # 车间网络拓扑与路径计算

"""

from __future__ import annotations

import os

from dataclasses import dataclass

from pathlib import Path

from typing import List, Dict, Tuple, Set, Optional

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

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

# 1. 节点定义

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

class NodeType:

"""设备节点类型"""

ENTRY = "entry" # 入库口

PROCESS = "process" # 加工中心

AGV = "agv" # AGV转运站

BUFFER = "buffer" # 缓存区

EXIT = "exit" # 出库口

@dataclass

class ShopFloorNode:

"""车间设备节点"""

name: str

node_type: str

process_time: float = 0.0 # 在该节点的处理时间（分钟）

def __repr__(self):

return f"{self.name}({self.node_type})"

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

# 2. 车间网络

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

class ShopFloorNetwork:

"""车间物流网络（基于NetworkX有向图）"""

def __init__(self, name: str = "ShopFloor"):

self.name = name

self.graph = nx.DiGraph()

self.nodes: Dict[str, ShopFloorNode] = {}

def add_node(self, node: ShopFloorNode):

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

self.nodes[node.name] = node

self.graph.add_node(

node.name,

node_type=node.node_type,

process_time=node.process_time,

)

def add_edge(self, from_node: str, to_node: str,

travel_time: float):

"""添加物流路径（有向边，权重=输送时间）"""

if from_node not in self.nodes or to_node not in self.nodes:

raise ValueError(f"节点不存在: {from_node} 或 {to_node}")

self.graph.add_edge(

from_node, to_node,

weight=travel_time,

travel_time=travel_time,

)

def remove_node(self, node_name: str):

"""删除设备节点"""

if node_name in self.nodes:

self.graph.remove_node(node_name)

del self.nodes[node_name]

def remove_edge(self, from_node: str, to_node: str):

"""删除物流路径"""

if self.graph.has_edge(from_node, to_node):

self.graph.remove_edge(from_node, to_node)

def get_node_count(self) -> int:

return len(self.nodes)

def get_edge_count(self) -> int:

return self.graph.number_of_edges()

def copy(self) -> "ShopFloorNetwork":

"""深拷贝网络"""

new_net = ShopFloorNetwork(self.name + "_copy")

for node in self.nodes.values():

new_net.add_node(ShopFloorNode(

node.name, node.node_type, node.process_time

))

for u, v, data in self.graph.edges(data=True):

new_net.add_edge(u, v, data["travel_time"])

return new_net

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

# 3. 拓扑修改器

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

class TopologyModifier:

"""执行拓扑改造操作"""

def __init__(self, network: ShopFloorNetwork):

self.network = network

def add_equipment(self, node: ShopFloorNode,

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

"""

新增设备并连接

connections: [(from, to, travel_time), ...]

"""

print(f"[MODIFY] 新增设备: {node.name}")

self.network.add_node(node)

for u, v, t in connections:

self.network.add_edge(u, v, t)

def remove_equipment(self, node_name: str):

"""移除设备"""

print(f"[MODIFY] 移除设备: {node_name}")

self.network.remove_node(node_name)

def add_path(self, from_node: str, to_node: str, travel_time: float):

"""新增物流路径"""

print(f"[MODIFY] 新增路径: {from_node} → {to_node} ({travel_time}min)")

self.network.add_edge(from_node, to_node, travel_time)

def remove_path(self, from_node: str, to_node: str):

"""移除物流路径"""

print(f"[MODIFY] 移除路径: {from_node} → {to_node}")

self.network.remove_edge(from_node, to_node)

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

# 4. 连通性评估器

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

class ConnectivityEvaluator:

"""评估网络连通性"""

def __init__(self):

pass

def is_connected(self, network: ShopFloorNetwork) -> bool:

"""检查全网是否弱连通"""

return nx.is_weakly_connected(network.graph)

def connected_components_count(self, network: ShopFloorNetwork) -> int:

"""连通分量数量"""

return nx.number_weakly_connected_components(network.graph)

def reachability_matrix(self, network: ShopFloorNetwork) -> pd.DataFrame:

"""可达性矩阵（节点对之间是否可达）"""

node_names = list(network.nodes.keys())

n = len(node_names)

matrix = pd.DataFrame(

np.zeros((n, n), dtype=bool),

index=node_names, columns=node_names

)

for u in node_names:

if u not in network.graph:

continue

reachable = nx.descendants(network.graph, u)

reachable.add(u)

for v in reachable:

matrix.loc[u, v] = True

return matrix

def entry_to_exit_paths(self, network: ShopFloorNetwork) -> Dict:

"""检查所有入库口到出库口的路径"""

entry_nodes = [n for n, d in network.graph.nodes(data=True)

if d.get("node_type") == NodeType.ENTRY]

exit_nodes = [n for n, d in network.graph.nodes(data=True)

if d.get("node_type") == NodeType.EXIT]

results = {}

for entry in entry_nodes:

results[entry] = {}

for exit in exit_nodes:

if nx.has_path(network.graph, entry, exit):

path = nx.shortest_path(network.graph, entry, exit,

weight="weight")

length = nx.shortest_path_length(network.graph, entry, exit,

weight="weight")

results[entry][exit] = {

"reachable": True,

"path": path,

"length": length,

}

else:

results[entry][exit] = {

"reachable": False,

"path": None,

"length": float("inf"),

}

return results

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

# 5. 效率评估器

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

class EfficiencyEvaluator:

"""评估物料输送效率"""

def __init__(self):

pass

def average_shortest_path_length(self, network: ShopFloorNetwork) -> float:

"""计算全网平均最短路径长度（仅对可达节点对）"""

if not nx.is_weakly_connected(network.graph):

return float("inf")

total_length = 0.0

count = 0

for u in network.nodes:

lengths = nx.single_source_dijkstra_path_length(

network.graph, u, weight="weight"

)

for v, length in lengths.items():

if u < v: # 避免重复计算

total_length += length

count += 1

return total_length / count if count > 0 else float("inf")

def bottleneck_nodes(self, network: ShopFloorNetwork,

top_k: int = 3) -> List[Tuple[str, float]]:

"""识别瓶颈节点（基于度中心性）"""

in_degree = dict(network.graph.in_degree())

out_degree = dict(network.graph.out_degree())

# 总度数

total_degree = {}

for node in network.nodes:

total_degree[node] = in_degree.get(node, 0) + out_degree.get(node, 0)

# 排序

sorted_nodes = sorted(total_degree.items(), key=lambda x: x[1],

reverse=True)

return sorted_nodes[:top_k]

def path_redundancy(self, network: ShopFloorNetwork,

source: str, target: str) -> int:

"""计算两节点间的路径冗余度（独立路径数量）"""

if not nx.has_path(network.graph, source, target):

return 0

# 使用简单方法：枚举所有简单路径（对于小图可行）

try:

paths = list(nx.all_simple_paths(network.graph, source, target,

cutoff=10))

return len(paths)

except Exception:

return 0

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

# 6. 可视化器

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

class Visualizer:

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

def __init__(self):

self.results_dir = Path("results")

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

def plot_topology(self, network: ShopFloorNetwork, title: str,

filename: str):

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

print(f"[INFO] 绘制拓扑图: {title}")

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

# 分层布局

pos = nx.spring_layout(network.graph, seed=42, k=2, iterations=50)

# 节点颜色映射

color_map = {

NodeType.ENTRY: "#27AE60", # 绿色

NodeType.PROCESS: "#3498DB", # 蓝色

NodeType.AGV: "#E74C3C", # 红色

NodeType.BUFFER: "#F39C12", # 橙色

NodeType.EXIT: "#8E44AD", # 紫色

}

node_colors = [color_map.get(network.graph.nodes[n].get("node_type"),

"#95A5A6")

for n in network.graph.nodes]

# 节点大小（根据度中心性）

degrees = dict(network.graph.degree())

max_deg = max(degrees.values()) if degrees else 1

node_sizes = [500 + 1500 * degrees[n] / max_deg

for n in network.graph.nodes]

nx.draw_networkx_nodes(network.graph, pos, ax=ax,

node_color=node_colors, node_size=node_sizes,

alpha=0.85, edgecolors="white")

nx.draw_networkx_edges(network.graph, pos, ax=ax,

arrowstyle="->", arrowsize=15,

edge_color="#7F8C8D", width=1.5,

connectionstyle="arc3,rad=0.1")

nx.draw_networkx_labels(network.graph, pos, ax=ax,

font_size=9, font_weight="bold")

# 边权重标注

edge_labels = {(u, v): f"{d['weight']}min"

for u, v, d in network.graph.edges(data=True)}

nx.draw_networkx_edge_labels(network.graph, pos, ax=ax,

edge_labels=edge_labels,

font_size=7)

# 图例

patches = [mpatches.Patch(color=v, label=k)

for k, v in color_map.items()]

ax.legend(handles=patches, loc="upper right", fontsize=10)

ax.set_title(title, fontsize=14, fontweight="bold")

ax.axis("off")

plt.tight_layout()

plt.savefig(self.results_dir / filename, dpi=150, bbox_inches="tight")

plt.close()

print(f" 已保存: {self.results_dir / filename}")

def plot_path_comparison(self, before_data: Dict, after_data: Dict):

"""对比改造前后的路径长度"""

print("[INFO] 绘制路径长度对比...")

# 收集所有入库→出库的对比

comparisons = []

for entry, exits in before_data.items():

for exit, data in exits.items():

if data["reachable"] and exit in after_data.get(entry, {}):

after = after_data[entry][exit]

if after["reachable"]:

comparisons.append({

"route": f"{entry}→{exit}",

"before": data["length"],

"after": after["length"],

"change": after["length"] - data["length"],

})

if not comparisons:

print(" 无可达路径对比")

return

df = pd.DataFrame(comparisons)

df = df.sort_values("before")

fig, axes = plt.subplots(2, 1, figsize=(12, 8), sharex=True)

x = np.arange(len(df))

w = 0.35

# 绝对路径长度

axes[0].bar(x - w/2, df["before"], w, label="改造前",

color="#3498DB", alpha=0.8)

axes[0].bar(x + w/2, df["after"], w, label="改造后",

color="#E74C3C", alpha=0.8)

axes[0].set_ylabel("路径长度 (分钟)", fontsize=12)

axes[0].set_title("入库→出库 路径长度对比", fontsize=13, fontweight="bold")

axes[0].legend()

axes[0].set_xticks(x)

axes[0].set_xticklabels(df["route"], rotation=15)

axes[0].grid(True, alpha=0.3, axis="y")

# 变化量

colors = ["#E74C3C" if c > 0 else "#27AE60" for c in df["change"]]

axes[1].bar(df["route"], df["change"], color=colors, alpha=0.8)

axes[1].axhline(y=0, color="black", linewidth=0.8)

axes[1].set_ylabel("变化量 (分钟)", fontsize=12)

axes[1].set_title("路径长度变化（红色=变长，绿色=变短）",

fontsize=13, fontweight="bold")

axes[1].tick_params(axis="x", rotation=15)

axes[1].grid(True, alpha=0.3, axis="y")

plt.tight_layout()

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

dpi=150, bbox_inches="tight")

plt.close()

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

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

# 7. 报告生成器

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

class ReportGenerator:

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

def __init__(self):

self.results_dir = Path("results")

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

def generate(self, before_net: ShopFloorNetwork,

after_net: ShopFloorNetwork,

before_conn: Dict, after_conn: Dict,

before_eff: Dict, after_eff: Dict) -> str:

"""生成报告"""

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

report_lines = []

report_lines.append("=" * 80)

report_lines.append("车间拓扑改造仿真分析报告")

report_lines.append("=" * 80)

report_lines.append(f"\n网络规模:")

report_lines.append(f" {'指标':<20} {'改造前':>10} {'改造后':>10} {'变化':>10}")

report_lines.append("-" * 55)

report_lines.append(

f" {'节点数':<20} {before_net.get_node_count():>10} "

f"{after_net.get_node_count():>10} "

f"{after_net.get_node_count() - before_net.get_node_count():>+10}"

)

report_lines.append(

f" {'边数':<20} {before_net.get_edge_count():>10} "

f"{after_net.get_edge_count():>10} "

f"{after_net.get_edge_count() - before_net.get_edge_count():>+10}"

)

report_lines.append(f"\n连通性评估:")

report_lines.append(f" {'指标':<25} {'改造前':>10} {'改造后':>10}")

report_lines.append("-" * 50)

report_lines.append(

f" {'全网连通':<25} "

f"{'✅' if before_conn['connected'] else '❌':>10} "

f"{'✅' if after_conn['connected'] else '❌':>10}"

)

report_lines.append(

f" {'连通分量数':<25} "

f"{before_conn['components']:>10} "

f"{after_conn[

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