周三下午FMS 柔性制造单元调度室。这批订单又卡在工装上了工艺员小赵指着 MES 屏幕五台加工中心十二个托盘三十多套夹具今天第三批活儿F03 气动虎钳和 F07 零点快换同时被三个工单抢结果 4 号机空转等夹具2 号机干完没地方卸整条线节拍全乱了。我接上他导出的 FMS 工单与夹具绑定表。这表里有什么小赵问。每个工单用哪些夹具、在哪台机床、加工多长时间、前后工序依赖我指着屏幕但它就是一张静态 BOM 表没把夹具当成会在机床之间流转的共享资源来建模。现在排产靠经验哪个夹具是瓶颈、什么时候会冲突全靠调度员拍脑袋。我就想干一件事小赵说把夹具流转画成一张网节点是夹具边是从工单 A 到工单 B 的传递关系然后算每个夹具的介数中心性、被争用次数、冲突窗口提前告诉我哪个夹具会卡脖子、几点几分开始抢。比如 F03 虎钳的介数中心性排第一被 6 个工单跨 4 台机床争用冲突窗口集中在 10:20~11:40我接话用 networkx 建夹具流转有向网算度中心性/介数/紧密度pandas 做工单-夹具矩阵numpy 算冲突时间窗matplotlib 画拓扑图热力图甘特图scipy 做冲突概率密度估计。对小赵点头别给我黑盒要能说清楚是F03 在 4 号机和 2 号机之间来回倒还是零点快换接口只有 3 个但 7 个工单要抢导致的瓶颈调度员看得懂能提前调。用 pandas 读工单-夹具绑定表networkx 建流转网计算中心性指标numpy 算冲突时间窗scipy 做密度估计matplotlib 出 6 图报告存 results/我开工程数据自包含合成一批含多品种小批量混流的数据下载就能跑。敲了行原型# 夹具 i 在时间窗 [t1, t2] 被工单 A 占用# 工单 B 需要同一夹具且时间窗重叠 → 冲突conflict (t_start_B t_end_A) and (t_start_A t_end_B)risk_score centrality(fixture) * conflict_count完整版 OOP 封好我说加载器、流转网构建器、中心性分析器、冲突窗口计算器、风险评分器、可视化器输出瓶颈排名冲突热力图6图报告。小赵凑近看那以后看报告F03 虎钳风险分 87.3 排第一介数中心性 0.42冲突高峰 10:20~11:40拓扑图里 F03 节点最大热力图红区集中在 4 号机和 2 号机之间建议加一套 F03 或把工单 12 推迟 20 分钟。对我接话夹具不是静态 BOM是动态流转的共享资源。数字孪生里挂夹具资源节点这套就是 FMS 的夹具大脑。一、实际应用场景真实痛点场景设定FMS 柔性制造单元多品种小批量混流生产夹具虎钳、零点快换、专用工装在机床间共享流转。现排产仅按工单 BOM 绑定未建模夹具作为有限共享资源的流转冲突。需基于工单-夹具数据构建夹具流转网络分析 瓶颈识别 冲突风险量化程序。现场原话叙事化不是我们不会排产小赵说是会排但夹具不够用。三十多套夹具挂在十二个托盘上五台机床抢。F03 虎钳只有两套今天七个工单要用4 号机干一半要等夹具从 2 号机转过来一等就是 25 分钟。调度员看屏幕干着急因为系统不告诉他夹具流转路径长什么样。还有那种隐性冲突小赵补充工单之间时间窗没碰上但夹具周转时间算少了实际上卸夹清洗上料要 8 分钟结果下一个工单卡在等夹具整条线节拍掉 15%。这种靠经验根本算不准。核心矛盾静态 BOM 绑定 经验排产 与 夹具流转网络建模 中心性瓶颈识别 冲突时间窗量化 风险预警 之间的断层。二、痛点分析映射到滨州职业学院《先进制造技术》课程模型《先进制造技术》模块 本篇痛点对应柔性制造系统FMS与先进生产管理夹具资源、混流排产、资源冲突、在制品流转 夹具共享资源建模 冲突分析先进制造技术基础系统可靠性、瓶颈分析 瓶颈夹具识别智能制造与数字孪生资源状态镜像、物流节点 夹具流转网络作数字孪生资源层先进制造新模式数据驱动生产调度 从经验排产→网络分析辅助决策一句话总结我们需要一个FMS 工单-夹具数据→夹具流转网络分析瓶颈识别冲突风险评估程序用pandas 读工单-夹具绑定networkx 建流转有向网中心性计算numpy 算冲突时间窗scipy 做冲突密度估计matplotlib 画拓扑/热力/甘特实现从经验排产到网络分析量化风险可解释瓶颈。三、核心逻辑讲解大白话3.1 问题本质把夹具想成共享单车把 FMS 里的夹具想成共享单车* 夹具 一辆单车* 工单 一个骑车人* 机床 停车桩* 流转 人骑完停到下一个桩下一个人骑走* 只有 2 辆单车但 7 个人要骑 夹具争用* 两个人同时到桩但只有一辆车 时间窗冲突* 某辆车被最多人骑过 瓶颈夹具中心性最高* 流转网络 画一张谁把车停到哪、谁又骑走的路线图* 介数中心性 这辆车是不是必经之路绕不开* 冲突风险分 被抢次数 × 流转关键度 × 时间重叠密度3.2 业务逻辑 → 代码映射导入工单-夹具绑定数据│▼ FixtureDataLoader (pandas)读取表:工单ID, 夹具ID, 机床ID, 开始时间, 结束时间, 工序序校验时间合法性, 按工单工序排序│▼ FixtureFlowNetwork (networkx)流转有向网构建:节点 夹具ID边 夹具从工单A流转到工单B (同夹具跨工单)边权 流转次数 / 平均周转时间机床-夹具二分图(辅助)│▼ CentralityAnalyzer (networkx)中心性计算:度中心性: 被多少工单使用介数中心性: 是否关键流转路径紧密度: 流转效率特征向量中心性: 是否连接其他关键夹具│▼ ConflictWindowCalculator (numpy)冲突时间窗:对每个夹具, 找所有工单占用时段两两比较时间重叠 → 冲突对冲突密度 重叠分钟数 / 总窗口│▼ RiskScorer (numpy scipy)风险评分:risk centrality_weight × conflict_density × shortage_factorshortage_factor 需求工单数 / 可用套数输出排序│▼ FixtureVisualizer (matplotlib)可视化:1. 夹具流转有向网(节点大小中心性)2. 机床-夹具二分图3. 冲突热力图(夹具×时间窗)4. 瓶颈排名柱图5. 风险评分雷达/散点6. 工单-夹具甘特叠加│▼ SyntheticFMSGenerator (numpy)合成数据:5台机床, 30夹具, 20工单含故意制造的夹具争用(F03/F07)可复现3.3 为什么不能只看夹具数量够不够视角 问题数套数 2 套够 7 个工单时间不重叠就够看 BOM 静态不知道流转路径建网络 看清谁绕不开、谁被反复倒算时间窗 量化冲突分钟数风险评分 综合中心性冲突短缺3.4 优化前后对比维度 经验排产 本程序瓶颈识别 靠感觉 介数中心性量化冲突发现 发生后才知道 时间窗预计算风险排序 无 综合评分调度建议 人工调 可量化加一套或推迟X分钟输出 口头交代 拓扑图热力图报告四、OOP 代码实现4.1 项目结构fms_fixture_network/├── fms_fixture_network/│ ├── __init__.py│ ├── fixture_data_loader.py # 工单-夹具数据加载│ ├── fixture_flow_network.py # 流转有向网(networkx)│ ├── centrality_analyzer.py # 中心性分析(networkx)│ ├── conflict_window_calc.py # 冲突时间窗(numpy)│ ├── risk_scorer.py # 风险评分│ ├── visualizer.py # 可视化│ └── synthetic_fms_data.py # 合成FMS数据├── tests/│ ├── __init__.py│ └── test_fixture_network.py├── results/│ ├── flow_network.png│ ├── bipartite_graph.png│ ├── conflict_heatmap.png│ ├── bottleneck_rank.png│ ├── risk_scatter.png│ ├── gantt_overlay.png│ ├── centrality_table.csv│ ├── conflict_report.csv│ ├── risk_ranking.csv│ └ fixture_report.txt└── run_fixture_network.py4.2 核心源码detailssummary/summaryFMS 工单-夹具数据加载器import pandas as pdfrom pathlib import Pathfrom typing import Optionalclass FixtureDataLoader:读取工单-夹具-机床绑定表def __init__(self, filepath: str fms_fixture_jobs.csv,encoding: str utf-8):self.filepath Path(filepath)self.encoding encodingdef load(self) - pd.DataFrame:if not self.filepath.exists():raise FileNotFoundError(self.filepath)df pd.read_csv(self.filepath, encodingself.encoding)req [job_id, op_seq, fixture_id, machine_id,t_start_min, t_end_min]miss [c for c in req if c not in df.columns]if miss:raise ValueError(f缺列: {miss})for c in req [fixture_count]:if c in df.columns:df[c] pd.to_numeric(df[c], errorscoerce)df df.dropna(subsetreq).sort_values([job_id, op_seq]).reset_index(dropTrue)# 校验时间bad df[df[t_end_min] df[t_start_min]]if len(bad):raise ValueError(f{len(bad)}行时间不合法)return dfdef fixture_inventory(self) - pd.DataFrame:从数据推断可用套数(默认每夹具2套, 可按实际改)df self.load()fix_list df[fixture_id].unique()return pd.DataFrame({fixture_id: fix_list,available: 2, # 默认2套})/detailsdetailssummary/summary夹具流转有向网 (networkx)import networkx as nximport pandas as pdfrom collections import defaultdictfrom typing import Dict, List, Tupleclass FixtureFlowNetwork:节点 夹具ID边 夹具从工单A的工序流转到工单B的工序边权 流转次数(同夹具跨工单对)def __init__(self):self.G nx.DiGraph()self.bipartite nx.Graph() # 机床-夹具二分图def build(self, df: pd.DataFrame) - nx.DiGraph:self.G.clear()# 节点for fid in df[fixture_id].unique():self.G.add_node(fid, ntypefixture)# 边: 按时间排序, 同夹具连续工单间建边df_sorted df.sort_values(t_start_min)prev_row {}edge_count defaultdict(int)for _, row in df_sorted.iterrows():fid row[fixture_id]key (prev_row.get(fid, None), row[job_id])if prev_row.get(fid) is not None:edge_count[(prev_row[fid], row[job_id])] 1prev_row[fid] row[job_id]for (src, dst), cnt in edge_count.items():if src ! dst:self.G.add_edge(src, dst, fixture_flowcnt, weightcnt)return self.Gdef build_bipartite(self, df: pd.DataFrame) - nx.Graph:self.bipartite.clear()machines df[machine_id].unique()fixtures df[fixture_id].unique()for m in machines:self.bipartite.add_node(m, bipartite0)for f in fixtures:self.bipartite.add_node(f, bipartite1)for _, row in df.iterrows():self.bipartite.add_edge(row[machine_id], row[fixture_id])return self.bipartitedef summary(self) - Dict:return {nodes: self.G.number_of_nodes(),edges: self.G.number_of_edges(),density: nx.density(self.G),is_weakly_connected: nx.is_weakly_connected(self.G),}/detailsdetailssummary/summary中心性分析 (networkx)import networkx as nximport pandas as pdfrom typing import Dictclass CentralityAnalyzer:计算夹具节点的多种中心性指标def __init__(self, G: nx.DiGraph):self.G Gdef compute_all(self) - pd.DataFrame:deg dict(self.G.degree())in_deg dict(self.G.in_degree())out_deg dict(self.G.out_degree())try:betw nx.betweenness_centrality(self.G, weightweight)except Exception:betw {n: 0.0 for n in self.G.nodes()}try:close nx.closeness_centrality(self.G)except Exception:close {n: 0.0 for n in self.G.nodes()}try:eig nx.eigenvector_centrality(self.G, max_iter1000, weightweight)except Exception:eig {n: 0.0 for n in self.G.nodes()}rows []for n in self.G.nodes():rows.append({fixture_id: n,degree: deg.get(n, 0),in_degree: in_deg.get(n, 0),out_degree: out_deg.get(n, 0),betweenness: round(betw.get(n, 0), 4),closeness: round(close.get(n, 0), 4),eigenvector: round(eig.get(n, 0), 4),})return pd.DataFrame(rows).sort_values(betweenness, ascendingFalse).reset_index(dropTrue)/detailsdetailssummary/summary冲突时间窗计算 (numpy)import numpy as npimport pandas as pdfrom typing import Dict, List, Tupleclass ConflictWindowCalculator:对每个夹具, 找所有工单占用时段两两比较时间重叠 → 冲突对def __init__(self, time_granularity_min: float 1.0):self.gran time_granularity_mindef calc_conflicts(self, df: pd.DataFrame) - pd.DataFrame:records []for fid in df[fixture_id].unique():sub df[df[fixture_id] fid].copy()sub sub.sort_values(t_start_min)n len(sub)conflict_pairs []total_overlap_min 0.0for i in range(n):for j in range(i1, n):a_start sub.iloc[i][t_start_min]a_end sub.iloc[i][t_end_min]b_start sub.iloc[j][t_start_min]b_end sub.iloc[j][t_end_min]# 重叠检测overlap_start max(a_start, b_start)overlap_end min(a_end, b_end)if overlap_start overlap_end:overlap_min overlap_end - overlap_starttotal_overlap_min overlap_minconflict_pairs.append({job_a: sub.iloc[i][job_id],job_b: sub.iloc[j][job_id],overlap_min: round(overlap_min, 1),})# 需求工单数demand_jobs sub[job_id].nunique()records.append({fixture_id: fid,demand_jobs: demand_jobs,total_overlap_min: round(total_overlap_min, 1),conflict_pairs: len(conflict_pairs),conflict_detail: conflict_pairs,})return pd.DataFrame(records)/detailsdetailssummary/summary风险评分 (numpy scipy)import numpy as npimport pandas as pdfrom scipy.stats import gaussian_kdefrom typing import Dictclass RiskScorer:risk centrality_weight × conflict_density × shortage_factordef __init__(self, available_df: pd.DataFrame):self.avail available_df.set_index(fixture_id)[available].to_dict()def score(self, cent_df: pd.DataFrame,conf_df: pd.DataFrame) - pd.DataFrame:merged cent_df.merge(conf_df, onfixture_id)# shortage_factor demand_jobs / max(available,1)merged[shortage] merged[demand_jobs].apply(lambda d: d / max(1, self.avail.get(merged[merged.indexd.index].index[0] if False else merged.iloc[0][fixture_id], 2)))# 修正shortage计算sf []for _, r in merged.iterrows():fid r[fixture_id]av self.avail.get(fid, 2)sf.append(r[demand_jobs] / max(1, av))merged[shortage_factor] sf# conflict_density total_overlap_min / max(demand_jobs*10, 1)merged[conflict_density] merged[total_overlap_min] / (merged[demand_jobs] * 10).clip(lower1)# 综合风险分 (0-100)merged[risk_score] (merged[betweenness] * 50 merged[conflict_density] * 30 merged[shortage_factor] * 20).clip(0, 100).round(1)return merged.sort_values(risk_score, ascendingFalse).reset_index(dropTrue)/detailsdetailssummary/summary可视化 (matplotlib networkx)import numpy as npimport pandas as pdimport matplotlib.pyplot as pltfrom pathlib import Pathimport networkx as nxplt.rcParams[font.sans-serif] [SimHei, DejaVu Sans]plt.rcParams[axes.unicode_minus] Falseclass FixtureVisualizer:def __init__(self, results_dir: str results):self.results_dir Path(results_dir)self.results_dir.mkdir(exist_okTrue)def flow_network(self, G, cent_df):fig, ax plt.subplots(figsize(13,10))pos nx.spring_layout(G, seed42, k0.8)sizes []for n in G.nodes():r cent_df[cent_df[fixture_id]n]if len(r):sizes.append(300 r.iloc[0][betweenness]*5000)else:sizes.append(300)nx.draw_networkx_nodes(G, pos, node_sizesizes,node_color#E74C3C, alpha0.85,edgecolorsblack, linewidths0.5, axax)nx.draw_networkx_edges(G, pos, width1.2, alpha0.4,arrowsTrue, arrowsize12, axax)labels {n: n for n in G.nodes()}nx.draw_networkx_labels(G, pos, labels, font_size8, axax)ax.set_title(夹具流转有向网(节点大小介数中心性),fontsize13, fontweightbold)ax.axis(off)plt.tight_layout()plt.savefig(self.results_dir/flow_network.png, dpi150, bbox_inchestight)plt.close()def bipartite(self, B):fig, ax plt.subplots(figsize(12,8))pos nx.bipartite_layout(B,nodes[n for n,d in B.nodes(dataTrue) if d.get(bipartite)0])colors [#3498DB if d.get(bipartite)0 else #E67E22for n,d in B.nodes(dataTrue)]nx.draw_networkx_nodes(B, pos, node_colorcolors, node_size400,edgecolorsblack, linewidths0.5, axax)nx.draw_networkx_edges(B, pos, alpha0.3, width0.8, axax)nx.draw_networkx_labels(B, pos, font_size7, axax)ax.set_title(机床-夹具二分图, fontsize13, fontweightbold)ax.axis(off)plt.tight_layout()plt.savefig(self.results_dir/bipartite_graph.png, dpi150, bbox_inchestight)plt.close()def conflict_heatmap(self, df, conf_df):夹具×小时冲突热力hours list(range(8, 18))fix_ids conf_df[fixture_id].tolist()[:15] # 最多15个heat np.zeros((len(fix_ids), len(hours)))for i, fid in enumerate(fix_ids):sub df[df[fixture_id]fid]for j, h in enumerate(hours):h_start h * 60h_end (h1)*60overlap 0for _, row in sub.iterrows():o_start max(row[t_start_min], h_start)o_end min(row[t_end_min], h_end)if o_start o_end:overlap o_end - o_startheat[i,j] overlapfig, ax plt.subplots(figsize(12, max(4, len(fix_ids)*0.4)))im ax.imshow(heat, aspectauto, cmapYlOrRd)ax.set_xticks(range(len(hours)))ax.set_xticklabels([f{h}:00 for h in hours])ax.set_yticks(range(len(fix_ids)))ax.set_yticklabels(fix_ids)ax.set_xlabel(时间); ax.set_ylabel(夹具)ax.set_title(夹具-时间冲突热力图(分钟), fontsize13, fontweightbold)plt.colorbar(im, axax)plt.tight_layout()plt.savefig(self.results_dir/conflict_heatmap.png, dpi150, bbox_inchestight)plt.close()def bottleneck_rank(self, risk_df):top risk_df.head(10)fig, ax plt.subplots(figsize(10,6))ax.barh(top[fixture_id][::-1], top[risk_score][::-1],color#E74C3C)ax.set_xlabel(风险评分)ax.set_title(瓶颈夹具风险排名(Top10), fontsize13, fontweightbold)ax.grid(axisx, alpha0.3)plt.tight_layout()plt.savefig(self.results_dir/bottleneck_rank.png, dpi150, bbox_inchestight)plt.close()def risk_scatter(self, risk_df):fig, ax plt.subplots(figsize(8,7))sc ax.scatter(risk_df[betweenness], risk_df[conflict_density],srisk_df[shortage_factor]*100,crisk_df[risk_score], cmapRdYlGn_r, alpha0.7)for _, r in risk_df.head(5).iterrows():ax.annotate(r[fixture_id],(r[betweenness], r[conflict_density]),fontsize8)ax.set_xlabel(介数中心性); ax.set_ylabel(冲突密度)ax.set_title(风险散点(气泡短缺因子, 颜色风险分),fontsize13, fontweightbold)plt.colorbar(sc, axax)ax.grid(alpha0.3)plt.tight_layout()plt.savefig(self.results_dir/risk_scatter.png, dpi150, bbox_inchestight)plt.close()def gantt_overlay(self, df, risk_df):工单-夹具甘特叠加top_fixtures risk_df.head(3)[fixture_id].tolist()sub df[df[fixture_id].isin(top_fixtures)]fig, ax plt.subplots(figsize(14,5))colors {F03:#E74C3C,F07:#F39C12,F11:#3498DB}y_pos 0labels []for fid in top_fixtures:for _, row in sub[sub[fixture_id]fid].iterrows():ax.barh(y_pos, row[t_end_min]-row[t_start_min],leftrow[t_start_min],colorcolors.get(fid,#95A5A6), alpha0.7)labels.append(f{fid}-{row[job_id]})y_pos 1ax.set_yticks(range(y_pos))ax.set_yticklabels(labels, fontsize7)ax.set_xlabel(时间 (min)); ax.set_title(Top3风险夹具 工单占用甘特,fontsize13, fontweightbold)ax.grid(axisx, alpha0.3)plt.tight_layout()plt.savefig(self.results_dir/gantt_overlay.png, dpi150, bbox_inchestight)plt.close()/detailsdetailssummary/summary合成FMS工单-夹具数据import numpy as npimport pandas as pdfrom pathlib import Pathfrom typing import Optionalclass SyntheticFMSGenerator:5台机床, 30夹具, 20工单故意制造 F03/F07 争用def __init__(self, rng: Optional[np.random.RandomState] None):self.rng rng or np.random.RandomState(42)def generate(self, out_path: str fms_fixture_jobs.csv,n_jobs: int 20, n_machines: int 5) - pd.DataFrame:machines [fM{i1} for i in range(n_machines)]fixtures [fF{i1:02d} for i in range(30)]# 故意让F03,F07被多用hot_fixtures [F03,F07,F11,F15]records []t_base 8 * 60 # 8:00开始for jid in range(1, n_jobs1):n_ops self.rng.randint(2, 5)t t_base self.rng.randint(0, 120)for op in range(1, n_ops1):# 选夹具: 70%概率从hot中选if self.rng.rand() 0.7:fix self.rng.choice(hot_fixtures)else:fix self.rng.choice(fixtures)mach self.rng.choice(machines)dur self.rng.randint(15, 45)records.append({job_id: fJ{jid:02d},op_seq: op,fixture_id: fix,machin利用AI解决实际问题如果你觉得这个工具好用欢迎关注长安牧笛