周五下午FMS 柔性制造系统控制室。这排产排了个寂寞生产调度老周指着监控屏幕AGV 把毛坯送到加工中心 A结果 A 前面的工装还没拆——上一件还在夹具上卡着。工件在托盘上干等了八分钟。然后 A 加工完AGV 又花了三分钟才把成品运走。一件活儿总共在 A 上停了二十三分钟真正切削才十一分钟剩下十二分钟全在等工装和等物流。我看了眼 MES 系统里的任务记录表任务ID | 加工单元 | 开始时间 | 结束时间 | 工装准备开始 | 工装准备结束 | 实际切削开始 | 实际切削结束 | AGV到达 | AGV离开T001 | MC-A | 08:00 | 08:23 | 08:00 | 08:05 | 08:05 | 08:16 | 08:16 | 08:23系统里这些时间戳都有我问。都有老周点头但系统只记录不分析。它不会告诉你工装等待占总工时的比例是多少、哪个加工单元的等待最严重、是工装准备慢还是 AGV 调度慢。我想看一眼就知道哪个环节在拖后腿但 MES 只给你一张流水账表几千行数据人眼根本看不出规律。所以你需要的不是数据是归因对老周眼睛亮了我需要把每一件活儿的工时拆成几段工装准备等待、实际切削、物流等待。然后按加工单元分组算每段占总工时的比例。如果 MC-A 的工装等待占 35%MC-B 只占 8%——那我就知道瓶颈在 A 的工装管理不是 AGV。但 MES 不给我这个视图我得自己从几千行时间戳里算。明白了我打开 VS Code你需要一个程序导入 FMS 加工任务记录解析每个任务的多段时间戳计算工装等待时长、切削时长、物流等待时长按加工单元分组统计各段占总工时的比例找出瓶颈单元。用 pandas 做时间差计算用 numpy 算比例用 matplotlib 画堆叠柱状图和饼图用 networkx 画单元间物流流向图。我敲了几行代码import pandas as pddf pd.read_csv(fms_tasks.csv)df[工装等待] (df[工装准备结束] - df[工装准备开始]).dt.total_seconds() / 60df[切削时长] (df[实际切削结束] - df[实际切削开始]).dt.total_seconds() / 60df[总工时] (df[结束时间] - df[开始时间]).dt.total_seconds() / 60df[工装等待比例] df[工装等待] / df[总工时] * 100# 按单元分组unit_stats df.groupby(加工单元).agg({工装等待比例: mean, 总工时: mean})这只是核心逻辑我说完整版会用 OOP 封装一个类管数据加载解析多段时间戳一个类管工时拆解计算工装等待、切削、物流等待、缓冲等待一个类管按单元分组统计各段占比、均值、标准差一个类管瓶颈识别等待比例最高的单元一个类管可视化堆叠柱状图、单元工时饼图、物流流向网络图。数据自包含——用 numpy 合成一份含 300 条 FMS 任务记录的时间数据模拟不同单元的工装等待和物流延迟读者下载就能跑。老周凑近屏幕所以你这东西就是把每件活儿的工时拆成几块按单元统计哪块占比最大——我一看就知道 MC-A 的工装等待占了 35%是该加人还是该换快换夹具数据说话。对。而且还能做趋势分析我补充用 scipy 做各单元等待比例的假设检验——MC-A 和 MC-B 的工装等待是否有显著差异。用 scikit-learn 做聚类把任务按等待模式分组——是工装瓶颈型、物流瓶颈型还是均衡型。FMS 的调度优化不是凭感觉加 AGV是用数据拆解每一分钟的去向让等待无处藏身。一、实际应用场景真实痛点场景设定FMS柔性制造系统中一个加工任务从毛坯上线到成品下线经历多个阶段工装准备等待夹具被占用/需更换、实际切削加工、物流等待AGV 调度/缓冲站排队、检测等待等。每个阶段都有独立时间戳记录。实际生产中调度员需要量化各阶段耗时占比识别瓶颈加工单元但 MES 系统仅提供流水账式的时间记录缺乏自动化的工时拆解和归因分析。现场原话叙事化我不是不知道有等待老周说每件活儿我都能看到开始时间和结束时间中间也有各阶段的时间戳。但几千件活儿、六台加工中心我怎么知道哪个单元的工装等待最严重MES 给我的是明细表——一行一行的时间戳不是汇总。我想看的是MC-A 上周所有任务里工装等待平均占了多少比例和 MC-B 比差多少是周一特别堵还是每天都堵这些 MES 不告诉我我得自己拿 Excel 拉公式——拉完一次下周数据又更新了再拉一次。没完没了。核心矛盾MES 流水账时间戳与调度员需要的工时归因与瓶颈识别之间的分析断层。需要一个FMS 工装等待占比统计与瓶颈分析程序用pandas 做时间差计算和分组聚合用numpy 计算各段占比用matplotlib 绘制堆叠柱状图和饼图用networkx 绘制单元间物流流向图用scipy 做统计检验输出结构化的工时归因报告。二、痛点分析映射到滨州职业学院《先进制造技术》课程模型《先进制造技术》模块 本篇痛点对应柔性制造系统FMS与先进生产管理FMS 调度与单元控制 工装等待量化拆解任务工时识别工装瓶颈。先进制造技术基础先进制造系统性能评价 OEE 细分将总工时拆解为增值/非增值时间。智能制造与数字孪生FMS 数字孪生 时间线可视化用堆叠图映射物理任务流。先进制造新模式精益生产消除浪费 等待时间消除数据驱动的非增值活动压缩。一句话总结我们需要构建一个FMS 工装等待占比统计与瓶颈分析程序用pandas 拆解多段时间戳计算各阶段耗时占比用numpy 做分组统计用matplotlib 绘制工时归因图用networkx 绘制单元物流流向用scipy 做瓶颈显著性检验实现从时间戳流水账到调度决策支持的转化。三、核心逻辑讲解大白话3.1 问题本质把 FMS 任务想象成过安检把一件工件在 FMS 中的流转想象成过机场安检* 总工时 从排队到拿回行李的总时间你 08:00 站到队伍里08:23 拿到行李离开。* 工装准备等待 等前面的旅客收拾托盘前面的人把行李放上去、收拾好、走开——你才能把行李放上去。对应工件等夹具空闲/换型。* 实际切削 行李过 X 光机真正干活的时间X 光扫一遍就完事。* 物流等待 等行李从传送带另一端出来X 光完了行李还在里面转你得等。对应 AGV 调度延迟或缓冲站排队。* 占比 时间都去哪儿了如果过安检总共 23 分钟X 光只花了 3 分钟剩下 20 分钟都在等——那安检口就是瓶颈。工业应用* pandas 时间差计算pd.to_datetime() 解析时间戳(end - start).dt.total_seconds() / 60 算分钟数。* numpy 向量化占比wait_ratio wait_minutes / total_minutes * 100一行算完全部任务。* groupby 分组归因df.groupby(unit)[wait_ratio].mean() 找出哪个单元等待最严重。3.2 业务逻辑 → 代码映射定义 FMS 任务时间数据模型│▼ TaskDataLoader (pandas)导入任务记录 CSVpd.read_csv(fms_tasks.csv)解析任务ID、加工单元、各阶段时间戳│▼ TimeCalculator (pandas/numpy)工时拆解工装等待 工装准备结束 - 工装准备开始切削时长 切削结束 - 切削开始物流等待 AGV离开 - AGV到达 - 切削时长缓冲等待 总工时 - 各段之和│▼ UnitStatistics (pandas/numpy)按单元分组统计groupby(unit).agg(avg_wait_ratio(工装等待比例, mean),avg_cut_ratio(切削比例, mean),avg_logistics_ratio(物流等待比例, mean),task_count(task_id, count))│▼ BottleneckAnalyzer (numpy/scipy)瓶颈识别等待比例最高的单元scipy.stats t-test 检验单元间差异显著性│▼ TaskVisualizer (matplotlib networkx)可视化1. 单元工时堆叠柱状图各段占比2. 单单元工时饼图3. 单元间物流流向网络图4. 等待比例趋势折线图│▼ SyntheticDataGenerator (numpy)合成数据生成模拟 6 个加工单元 × 50 任务的工时数据不同单元设置不同等待水平3.3 为什么用时间差向量化而不是逐行计算* 问题300 条任务 × 6 个时间戳 1800 个时间值逐行减算代码冗长且慢。* 处理策略pandas 的datetime 列直接相减返回Timedelta 对象.dt.total_seconds() 批量转数值。* 工程合理性实际 FMS 日任务量可达数千条向量化是唯一合理的选择。3.4 分析前后对比维度 MES 流水账 Python 程序分析时间拆解 无只有开始/结束 工装/切削/物流/缓冲四段拆解归因分析 人工目测 按单元自动计算占比瓶颈识别 凭感觉 数据量化 统计检验趋势分析 无 按天/班次聚合趋势决策支持 好像 A 比较慢 A 工装等待占 35%需加快换夹具四、OOP 代码实现4.1 项目结构fms_fixture_wait/├── fms_fixture_wait/│ ├── __init__.py│ ├── data_loader.py # 任务数据加载│ ├── time_calculator.py # 工时拆解计算│ ├── unit_statistics.py # 按单元统计│ ├── bottleneck_analyzer.py # 瓶颈分析│ ├── visualizer.py # 可视化│ └── synthetic_data.py # 合成数据生成├── tests/│ ├── __init__.py│ └── test_fms_wait.py # 单元测试├── results/│ ├── unit_stack_bar.png # 单元工时堆叠柱图│ ├── unit_pie_MC-A.png # 单单元工时饼图│ ├── logistics_network.png # 物流流向网络图│ ├── wait_trend.png # 等待比例趋势图│ ├── unit_statistics.csv # 按单元统计结果│ ├── bottleneck_report.txt # 瓶颈分析报告│ └── fms_tasks.csv # 合成任务数据└── run_fms_analysis.py # 主程序入口4.2 核心源码detailssummary/summaryFMS 加工任务数据加载器import pandas as pdfrom pathlib import Pathfrom typing import Optionalclass TaskDataLoader:FMS 加工任务记录加载器支持 CSV 格式的多阶段时间戳数据。TIME_COLUMNS [start_time, end_time,fixture_prep_start, fixture_prep_end,cutting_start, cutting_end,agv_arrive, agv_leave,]def __init__(self, filepath: str fms_tasks.csv,encoding: str utf-8):self.filepath Path(filepath)self.encoding encodingself._raw_df: Optional[pd.DataFrame] Nonedef load(self) - pd.DataFrame:加载并解析任务数据Returns-------pd.DataFrameif not self.filepath.exists():raise FileNotFoundError(f文件不存在: {self.filepath})self._raw_df pd.read_csv(self.filepath, encodingself.encoding)# 时间列解析time_col_aliases {start_time: [开始时间, task_start, start_time],end_time: [结束时间, task_end, end_time],fixture_prep_start: [工装准备开始, fixture_start, fp_start],fixture_prep_end: [工装准备结束, fixture_end, fp_end],cutting_start: [切削开始, cut_start, machining_start],cutting_end: [切削结束, cut_end, machining_end],agv_arrive: [AGV到达, agv_arrive, transport_arrive],agv_leave: [AGV离开, agv_leave, transport_leave],}rename_map {}for target, aliases in time_col_aliases.items():if target not in self._raw_df.columns:for alias in aliases:if alias in self._raw_df.columns:rename_map[alias] targetbreakif rename_map:self._raw_df self._raw_df.rename(columnsrename_map)# 解析为 datetimefor col in self._raw_df.columns:if time in col or start in col or end in col or arrive in col or leave in col:self._raw_df[col] pd.to_datetime(self._raw_df[col], errorscoerce)return self._raw_df.copy()def get_raw_data(self) - Optional[pd.DataFrame]:return self._raw_df.copy() if self._raw_df is not None else None/detailsdetailssummary/summary工时拆解计算器import numpy as npimport pandas as pdfrom typing import Dict, Optionalclass TimeCalculator:FMS 任务工时拆解计算器将总工时拆解为工装等待、切削加工、物流等待、缓冲等待。def __init__(self):passdef calculate(self, df: pd.DataFrame) - pd.DataFrame:计算各阶段时长分钟和占比Parameters----------df : pd.DataFrame含时间戳列的数据Returns-------pd.DataFrame新增各阶段时长和占比列result df.copy()# 总工时分钟result[total_minutes] ((result[end_time] - result[start_time]).dt.total_seconds() / 60)# 工装准备时长result[fixture_minutes] ((result[fixture_prep_end] - result[fixture_prep_start]).dt.total_seconds() / 60)# 切削时长result[cutting_minutes] ((result[cutting_end] - result[cutting_start]).dt.total_seconds() / 60)# 物流等待 AGV 相关非切削非工装时间# AGV 运输总耗时agv_minutes ((result[agv_leave] - result[agv_arrive]).dt.total_seconds() / 60)# 物流等待 AGV 总耗时 - 切削时长切削时 AGV 不在result[logistics_minutes] np.maximum(0, agv_minutes - result[cutting_minutes])# 缓冲等待 总工时 - 工装 - 切削 - 物流result[buffer_minutes] np.maximum(0,result[total_minutes]- result[fixture_minutes]- result[cutting_minutes]- result[logistics_minutes],)# 占比%result[fixture_ratio] (result[fixture_minutes] / result[total_minutes] * 100).round(2)result[cutting_ratio] (result[cutting_minutes] / result[total_minutes] * 100).round(2)result[logistics_ratio] (result[logistics_minutes] / result[total_minutes] * 100).round(2)result[buffer_ratio] (result[buffer_minutes] / result[total_minutes] * 100).round(2)# 清理异常值占比为负或超 100 的标记ratio_cols [fixture_ratio, cutting_ratio, logistics_ratio, buffer_ratio]for col in ratio_cols:result[col] result[col].clip(0, 100)return resultdef get_summary(self, df: pd.DataFrame) - Dict:整体工时汇总return {avg_total_minutes: round(df[total_minutes].mean(), 2),avg_fixture_minutes: round(df[fixture_minutes].mean(), 2),avg_cutting_minutes: round(df[cutting_minutes].mean(), 2),avg_logistics_minutes: round(df[logistics_minutes].mean(), 2),avg_buffer_minutes: round(df[buffer_minutes].mean(), 2),avg_fixture_ratio: round(df[fixture_ratio].mean(), 2),avg_cutting_ratio: round(df[cutting_ratio].mean(), 2),avg_logistics_ratio: round(df[logistics_ratio].mean(), 2),}/detailsdetailssummary/summary按加工单元统计import numpy as npimport pandas as pdfrom typing import Dict, Optionalclass UnitStatistics:FMS 加工单元工时统计器按加工单元分组计算各阶段平均占比和分布。def __init__(self):passdef calculate(self,df: pd.DataFrame,unit_col: str unit_id) - pd.DataFrame:按加工单元分组统计Parameters----------df : pd.DataFrame含工时拆解结果的数据unit_col : str加工单元列名Returns-------pd.DataFrameif df.empty or unit_col not in df.columns:return pd.DataFrame()stats df.groupby(unit_col).agg(task_count(total_minutes, count),avg_total_minutes(total_minutes, mean),avg_fixture_minutes(fixture_minutes, mean),avg_cutting_minutes(cutting_minutes, mean),avg_logistics_minutes(logistics_minutes, mean),avg_buffer_minutes(buffer_minutes, mean),avg_fixture_ratio(fixture_ratio, mean),avg_cutting_ratio(cutting_ratio, mean),avg_logistics_ratio(logistics_ratio, mean),avg_buffer_ratio(buffer_ratio, mean),std_fixture_ratio(fixture_ratio, std),max_fixture_ratio(fixture_ratio, max),min_fixture_ratio(fixture_ratio, min),).reset_index()# 四舍五入numeric_cols stats.select_dtypes(include[np.number]).columnsstats[numeric_cols] stats[numeric_cols].round(2)# 按工装等待占比降序stats stats.sort_values(avg_fixture_ratio, ascendingFalse).reset_index(dropTrue)return statsdef get_unit_detail(self,df: pd.DataFrame,unit_id: str,unit_col: str unit_id) - Dict:获取单个单元的详细统计unit_df df[df[unit_col] unit_id]if unit_df.empty:return {}return {unit_id: unit_id,task_count: len(unit_df),avg_total: round(unit_df[total_minutes].mean(), 2),avg_fixture: round(unit_df[fixture_minutes].mean(), 2),avg_cutting: round(unit_df[cutting_minutes].mean(), 2),fixture_ratio_mean: round(unit_df[fixture_ratio].mean(), 2),fixture_ratio_std: round(unit_df[fixture_ratio].std(), 2),fixture_ratio_max: round(unit_df[fixture_ratio].max(), 2),cutting_ratio_mean: round(unit_df[cutting_ratio].mean(), 2),}/detailsdetailssummary/summary瓶颈分析器import numpy as npimport pandas as pdfrom scipy import statsfrom typing import Dict, List, Optionalclass BottleneckAnalyzer:FMS 加工单元瓶颈分析器基于工装等待占比识别瓶颈提供统计检验。def __init__(self, top_k: int 3):self.top_k top_kdef identify_bottlenecks(self,unit_stats_df: pd.DataFrame,unit_col: str unit_id) - pd.DataFrame:识别工装等待瓶颈单元Returns-------pd.DataFrameTop K 瓶颈单元if unit_stats_df.empty:return pd.DataFrame()sorted_df unit_stats_df.sort_values(avg_fixture_ratio, ascendingFalse).reset_index(dropTrue)return sorted_df.head(self.top_k)def compare_units(self,df: pd.DataFrame,unit_a: str,unit_b: str,unit_col: str unit_id) - Dict:对比两个单元的工装等待比例t-testReturns-------dicta_data df[df[unit_col] unit_a][fixture_ratio].dropna()b_data df[df[unit_col] unit_b][fixture_ratio].dropna()if len(a_data) 2 or len(b_data) 2:return {error: 样本量不足}t_stat, p_value stats.ttest_ind(a_data, b_data)return {unit_a: unit_a,unit_b: unit_b,unit_a_mean: round(a_data.mean(), 2),unit_b_mean: round(b_data.mean(), 2),unit_a_std: round(a_data.std(), 2),unit_b_std: round(b_data.std(), 2),t_statistic: round(t_stat, 4),p_value: round(p_value, 6),significant: p_value 0.05,interpretation: (f{unit_a} vs {unit_b}: (差异显著 if p_value 0.05 else 差异不显著) f (p{p_value:.4f})),}def all_pairwise_tests(self,df: pd.DataFrame,unit_col: str unit_id) - List[Dict]:所有单元两两对比units df[unit_col].unique()results []for i, ua in enumerate(units):for ub in units[i 1:]:result self.compare_units(df, ua, ub, unit_col)if error not in result:results.append(result)return resultsdef fixture_vs_cutting_correlation(self,df: pd.DataFrame) - Dict:工装等待与切削时长的相关性if df.empty:return {}corr df[fixture_ratio].corr(df[cutting_ratio])return {correlation: round(corr, 4) if not np.isnan(corr) else None,interpretation: (负相关工装等待越高切削占比越低if corr and corr -0.3else 弱相关或无相关),}/detailsdetailssummary/summaryFMS 工时数据可视化import numpy as npimport pandas as pdimport matplotlib.pyplot as pltimport networkx as nxfrom pathlib import Pathfrom typing import Optional, Listplt.rcParams[font.sans-serif] [SimHei, DejaVu Sans]plt.rcParams[axes.unicode_minus] Falseclass TaskVisualizer:FMS 加工任务工时可视化def __init__(self, results_dir: str results):self.results_dir Path(results_dir)self.results_dir.mkdir(exist_okTrue)self.stage_colors {fixture: #E74C3C, # 红 - 工装等待cutting: #2ECC71, # 绿 - 切削logistics: #3498DB, # 蓝 - 物流buffer: #F39C12, # 橙 - 缓冲}def plot_unit_stack_bar(self,unit_stats_df: pd.DataFrame,unit_col: str unit_id) - None:单元工时堆叠柱状图fig, ax plt.subplots(figsize(14, 7))units unit_stats_df[unit_col].valuesfixture unit_stats_df[avg_fixture_ratio].valuescutting unit_stats_df[avg_cutting_ratio].valueslogistics unit_stats_df[avg_logistics_ratio].valuesbuffer unit_stats_df[avg_buffer_ratio].valuesx np.arange(len(units))width 0.6ax.bar(x, fixture, width, label工装等待,colorself.stage_colors[fixture], bottom0)ax.bar(x, cutting, width, label切削加工,colorself.stage_colors[cutting], bottomfixture)ax.bar(x, logistics, width, label物流等待,colorself.stage_colors[logistics],bottomfixture cutting)ax.bar(x, buffer, width, label缓冲等待,colorself.stage_colors[buffer],bottomfixture cutting logistics)ax.set_xlabel(加工单元, fontsize12)ax.set_ylabel(占比 (%), fontsize12)ax.set_title(各加工单元工时占比分布, fontsize14, fontweightbold)ax.set_xticks(x)ax.set_xticklabels(units, rotation0)ax.legend(fontsize10, locupper right)ax.set_ylim(0, 100)ax.grid(axisy, alpha0.3)# 标注工装等待数值for i, v in enumerate(fixture):ax.text(i, v / 2, f{v:.1f}%, hacenter, vacenter,fontsize9, fontweightbold, colorwhite)plt.tight_layout()plt.savefig(self.results_dir / unit_stack_bar.png,dpi150, bbox_inchestight)plt.close()def plot_unit_pie(self,df: pd.DataFrame,unit_id: str,unit_col: str unit_id) - None:单单元工时饼图unit_df df[df[unit_col] unit_id]if unit_df.empty:returnavg_fixture unit_df[fixture_ratio].mean()avg_cutting unit_df[cutting_ratio].mean()avg_logistics unit_df[logistics_ratio].mean()avg_buffer unit_df[buffer_ratio].mean()sizes [avg_fixture, avg_cutting, avg_logistics, avg_buffer]labels [工装等待, 切削加工, 物流等待, 缓冲等待]colors [self.stage_colors[fixture], self.stage_colors[cutting],self.stage_colors[logistics], self.stage_colors[buffer]]fig, ax plt.subplots(figsize(8, 8))ax.pie(sizes, labelslabels, colorscolors, autopct%1.1f%%,shadowFalse, startangle90)ax.set_title(f{unit_id} 工时分布, fontsize14, fontweightbold)利用AI解决实际问题如果你觉得这个工具好用欢迎关注长安牧笛 
长河君 · 2026/9/30 10:36:50 · 阅读 1.2k python的先进制造技术工业场景模拟第十七篇:导入FMS单元加工任务记录，统计每个加工单元的工装等待时长占总工时比例。
周五下午FMS 柔性制造系统控制室。这排产排了个寂寞生产调度老周指着监控屏幕AGV 把毛坯送到加工中心 A结果 A 前面的工装还没拆——上一件还在夹具上卡着。工件在托盘上干等了八分钟。然后 A 加工完AGV 又花了三分钟才把成品运走。一件活儿总共在 A 上停了二十三分钟真正切削才十一分钟剩下十二分钟全在等工装和等物流。我看了眼 MES 系统里的任务记录表任务ID | 加工单元 | 开始时间 | 结束时间 | 工装准备开始 | 工装准备结束 | 实际切削开始 | 实际切削结束 | AGV到达 | AGV离开T001 | MC-A | 08:00 | 08:23 | 08:00 | 08:05 | 08:05 | 08:16 | 08:16 | 08:23系统里这些时间戳都有我问。都有老周点头但系统只记录不分析。它不会告诉你工装等待占总工时的比例是多少、哪个加工单元的等待最严重、是工装准备慢还是 AGV 调度慢。我想看一眼就知道哪个环节在拖后腿但 MES 只给你一张流水账表几千行数据人眼根本看不出规律。所以你需要的不是数据是归因对老周眼睛亮了我需要把每一件活儿的工时拆成几段工装准备等待、实际切削、物流等待。然后按加工单元分组算每段占总工时的比例。如果 MC-A 的工装等待占 35%MC-B 只占 8%——那我就知道瓶颈在 A 的工装管理不是 AGV。但 MES 不给我这个视图我得自己从几千行时间戳里算。明白了我打开 VS Code你需要一个程序导入 FMS 加工任务记录解析每个任务的多段时间戳计算工装等待时长、切削时长、物流等待时长按加工单元分组统计各段占总工时的比例找出瓶颈单元。用 pandas 做时间差计算用 numpy 算比例用 matplotlib 画堆叠柱状图和饼图用 networkx 画单元间物流流向图。我敲了几行代码import pandas as pddf pd.read_csv(fms_tasks.csv)df[工装等待] (df[工装准备结束] - df[工装准备开始]).dt.total_seconds() / 60df[切削时长] (df[实际切削结束] - df[实际切削开始]).dt.total_seconds() / 60df[总工时] (df[结束时间] - df[开始时间]).dt.total_seconds() / 60df[工装等待比例] df[工装等待] / df[总工时] * 100# 按单元分组unit_stats df.groupby(加工单元).agg({工装等待比例: mean, 总工时: mean})这只是核心逻辑我说完整版会用 OOP 封装一个类管数据加载解析多段时间戳一个类管工时拆解计算工装等待、切削、物流等待、缓冲等待一个类管按单元分组统计各段占比、均值、标准差一个类管瓶颈识别等待比例最高的单元一个类管可视化堆叠柱状图、单元工时饼图、物流流向网络图。数据自包含——用 numpy 合成一份含 300 条 FMS 任务记录的时间数据模拟不同单元的工装等待和物流延迟读者下载就能跑。老周凑近屏幕所以你这东西就是把每件活儿的工时拆成几块按单元统计哪块占比最大——我一看就知道 MC-A 的工装等待占了 35%是该加人还是该换快换夹具数据说话。对。而且还能做趋势分析我补充用 scipy 做各单元等待比例的假设检验——MC-A 和 MC-B 的工装等待是否有显著差异。用 scikit-learn 做聚类把任务按等待模式分组——是工装瓶颈型、物流瓶颈型还是均衡型。FMS 的调度优化不是凭感觉加 AGV是用数据拆解每一分钟的去向让等待无处藏身。一、实际应用场景真实痛点场景设定FMS柔性制造系统中一个加工任务从毛坯上线到成品下线经历多个阶段工装准备等待夹具被占用/需更换、实际切削加工、物流等待AGV 调度/缓冲站排队、检测等待等。每个阶段都有独立时间戳记录。实际生产中调度员需要量化各阶段耗时占比识别瓶颈加工单元但 MES 系统仅提供流水账式的时间记录缺乏自动化的工时拆解和归因分析。现场原话叙事化我不是不知道有等待老周说每件活儿我都能看到开始时间和结束时间中间也有各阶段的时间戳。但几千件活儿、六台加工中心我怎么知道哪个单元的工装等待最严重MES 给我的是明细表——一行一行的时间戳不是汇总。我想看的是MC-A 上周所有任务里工装等待平均占了多少比例和 MC-B 比差多少是周一特别堵还是每天都堵这些 MES 不告诉我我得自己拿 Excel 拉公式——拉完一次下周数据又更新了再拉一次。没完没了。核心矛盾MES 流水账时间戳与调度员需要的工时归因与瓶颈识别之间的分析断层。需要一个FMS 工装等待占比统计与瓶颈分析程序用pandas 做时间差计算和分组聚合用numpy 计算各段占比用matplotlib 绘制堆叠柱状图和饼图用networkx 绘制单元间物流流向图用scipy 做统计检验输出结构化的工时归因报告。二、痛点分析映射到滨州职业学院《先进制造技术》课程模型《先进制造技术》模块 本篇痛点对应柔性制造系统FMS与先进生产管理FMS 调度与单元控制 工装等待量化拆解任务工时识别工装瓶颈。先进制造技术基础先进制造系统性能评价 OEE 细分将总工时拆解为增值/非增值时间。智能制造与数字孪生FMS 数字孪生 时间线可视化用堆叠图映射物理任务流。先进制造新模式精益生产消除浪费 等待时间消除数据驱动的非增值活动压缩。一句话总结我们需要构建一个FMS 工装等待占比统计与瓶颈分析程序用pandas 拆解多段时间戳计算各阶段耗时占比用numpy 做分组统计用matplotlib 绘制工时归因图用networkx 绘制单元物流流向用scipy 做瓶颈显著性检验实现从时间戳流水账到调度决策支持的转化。三、核心逻辑讲解大白话3.1 问题本质把 FMS 任务想象成过安检把一件工件在 FMS 中的流转想象成过机场安检* 总工时 从排队到拿回行李的总时间你 08:00 站到队伍里08:23 拿到行李离开。* 工装准备等待 等前面的旅客收拾托盘前面的人把行李放上去、收拾好、走开——你才能把行李放上去。对应工件等夹具空闲/换型。* 实际切削 行李过 X 光机真正干活的时间X 光扫一遍就完事。* 物流等待 等行李从传送带另一端出来X 光完了行李还在里面转你得等。对应 AGV 调度延迟或缓冲站排队。* 占比 时间都去哪儿了如果过安检总共 23 分钟X 光只花了 3 分钟剩下 20 分钟都在等——那安检口就是瓶颈。工业应用* pandas 时间差计算pd.to_datetime() 解析时间戳(end - start).dt.total_seconds() / 60 算分钟数。* numpy 向量化占比wait_ratio wait_minutes / total_minutes * 100一行算完全部任务。* groupby 分组归因df.groupby(unit)[wait_ratio].mean() 找出哪个单元等待最严重。3.2 业务逻辑 → 代码映射定义 FMS 任务时间数据模型│▼ TaskDataLoader (pandas)导入任务记录 CSVpd.read_csv(fms_tasks.csv)解析任务ID、加工单元、各阶段时间戳│▼ TimeCalculator (pandas/numpy)工时拆解工装等待 工装准备结束 - 工装准备开始切削时长 切削结束 - 切削开始物流等待 AGV离开 - AGV到达 - 切削时长缓冲等待 总工时 - 各段之和│▼ UnitStatistics (pandas/numpy)按单元分组统计groupby(unit).agg(avg_wait_ratio(工装等待比例, mean),avg_cut_ratio(切削比例, mean),avg_logistics_ratio(物流等待比例, mean),task_count(task_id, count))│▼ BottleneckAnalyzer (numpy/scipy)瓶颈识别等待比例最高的单元scipy.stats t-test 检验单元间差异显著性│▼ TaskVisualizer (matplotlib networkx)可视化1. 单元工时堆叠柱状图各段占比2. 单单元工时饼图3. 单元间物流流向网络图4. 等待比例趋势折线图│▼ SyntheticDataGenerator (numpy)合成数据生成模拟 6 个加工单元 × 50 任务的工时数据不同单元设置不同等待水平3.3 为什么用时间差向量化而不是逐行计算* 问题300 条任务 × 6 个时间戳 1800 个时间值逐行减算代码冗长且慢。* 处理策略pandas 的datetime 列直接相减返回Timedelta 对象.dt.total_seconds() 批量转数值。* 工程合理性实际 FMS 日任务量可达数千条向量化是唯一合理的选择。3.4 分析前后对比维度 MES 流水账 Python 程序分析时间拆解 无只有开始/结束 工装/切削/物流/缓冲四段拆解归因分析 人工目测 按单元自动计算占比瓶颈识别 凭感觉 数据量化 统计检验趋势分析 无 按天/班次聚合趋势决策支持 好像 A 比较慢 A 工装等待占 35%需加快换夹具四、OOP 代码实现4.1 项目结构fms_fixture_wait/├── fms_fixture_wait/│ ├── __init__.py│ ├── data_loader.py # 任务数据加载│ ├── time_calculator.py # 工时拆解计算│ ├── unit_statistics.py # 按单元统计│ ├── bottleneck_analyzer.py # 瓶颈分析│ ├── visualizer.py # 可视化│ └── synthetic_data.py # 合成数据生成├── tests/│ ├── __init__.py│ └── test_fms_wait.py # 单元测试├── results/│ ├── unit_stack_bar.png # 单元工时堆叠柱图│ ├── unit_pie_MC-A.png # 单单元工时饼图│ ├── logistics_network.png # 物流流向网络图│ ├── wait_trend.png # 等待比例趋势图│ ├── unit_statistics.csv # 按单元统计结果│ ├── bottleneck_report.txt # 瓶颈分析报告│ └── fms_tasks.csv # 合成任务数据└── run_fms_analysis.py # 主程序入口4.2 核心源码detailssummary/summaryFMS 加工任务数据加载器import pandas as pdfrom pathlib import Pathfrom typing import Optionalclass TaskDataLoader:FMS 加工任务记录加载器支持 CSV 格式的多阶段时间戳数据。TIME_COLUMNS [start_time, end_time,fixture_prep_start, fixture_prep_end,cutting_start, cutting_end,agv_arrive, agv_leave,]def __init__(self, filepath: str fms_tasks.csv,encoding: str utf-8):self.filepath Path(filepath)self.encoding encodingself._raw_df: Optional[pd.DataFrame] Nonedef load(self) - pd.DataFrame:加载并解析任务数据Returns-------pd.DataFrameif not self.filepath.exists():raise FileNotFoundError(f文件不存在: {self.filepath})self._raw_df pd.read_csv(self.filepath, encodingself.encoding)# 时间列解析time_col_aliases {start_time: [开始时间, task_start, start_time],end_time: [结束时间, task_end, end_time],fixture_prep_start: [工装准备开始, fixture_start, fp_start],fixture_prep_end: [工装准备结束, fixture_end, fp_end],cutting_start: [切削开始, cut_start, machining_start],cutting_end: [切削结束, cut_end, machining_end],agv_arrive: [AGV到达, agv_arrive, transport_arrive],agv_leave: [AGV离开, agv_leave, transport_leave],}rename_map {}for target, aliases in time_col_aliases.items():if target not in self._raw_df.columns:for alias in aliases:if alias in self._raw_df.columns:rename_map[alias] targetbreakif rename_map:self._raw_df self._raw_df.rename(columnsrename_map)# 解析为 datetimefor col in self._raw_df.columns:if time in col or start in col or end in col or arrive in col or leave in col:self._raw_df[col] pd.to_datetime(self._raw_df[col], errorscoerce)return self._raw_df.copy()def get_raw_data(self) - Optional[pd.DataFrame]:return self._raw_df.copy() if self._raw_df is not None else None/detailsdetailssummary/summary工时拆解计算器import numpy as npimport pandas as pdfrom typing import Dict, Optionalclass TimeCalculator:FMS 任务工时拆解计算器将总工时拆解为工装等待、切削加工、物流等待、缓冲等待。def __init__(self):passdef calculate(self, df: pd.DataFrame) - pd.DataFrame:计算各阶段时长分钟和占比Parameters----------df : pd.DataFrame含时间戳列的数据Returns-------pd.DataFrame新增各阶段时长和占比列result df.copy()# 总工时分钟result[total_minutes] ((result[end_time] - result[start_time]).dt.total_seconds() / 60)# 工装准备时长result[fixture_minutes] ((result[fixture_prep_end] - result[fixture_prep_start]).dt.total_seconds() / 60)# 切削时长result[cutting_minutes] ((result[cutting_end] - result[cutting_start]).dt.total_seconds() / 60)# 物流等待 AGV 相关非切削非工装时间# AGV 运输总耗时agv_minutes ((result[agv_leave] - result[agv_arrive]).dt.total_seconds() / 60)# 物流等待 AGV 总耗时 - 切削时长切削时 AGV 不在result[logistics_minutes] np.maximum(0, agv_minutes - result[cutting_minutes])# 缓冲等待 总工时 - 工装 - 切削 - 物流result[buffer_minutes] np.maximum(0,result[total_minutes]- result[fixture_minutes]- result[cutting_minutes]- result[logistics_minutes],)# 占比%result[fixture_ratio] (result[fixture_minutes] / result[total_minutes] * 100).round(2)result[cutting_ratio] (result[cutting_minutes] / result[total_minutes] * 100).round(2)result[logistics_ratio] (result[logistics_minutes] / result[total_minutes] * 100).round(2)result[buffer_ratio] (result[buffer_minutes] / result[total_minutes] * 100).round(2)# 清理异常值占比为负或超 100 的标记ratio_cols [fixture_ratio, cutting_ratio, logistics_ratio, buffer_ratio]for col in ratio_cols:result[col] result[col].clip(0, 100)return resultdef get_summary(self, df: pd.DataFrame) - Dict:整体工时汇总return {avg_total_minutes: round(df[total_minutes].mean(), 2),avg_fixture_minutes: round(df[fixture_minutes].mean(), 2),avg_cutting_minutes: round(df[cutting_minutes].mean(), 2),avg_logistics_minutes: round(df[logistics_minutes].mean(), 2),avg_buffer_minutes: round(df[buffer_minutes].mean(), 2),avg_fixture_ratio: round(df[fixture_ratio].mean(), 2),avg_cutting_ratio: round(df[cutting_ratio].mean(), 2),avg_logistics_ratio: round(df[logistics_ratio].mean(), 2),}/detailsdetailssummary/summary按加工单元统计import numpy as npimport pandas as pdfrom typing import Dict, Optionalclass UnitStatistics:FMS 加工单元工时统计器按加工单元分组计算各阶段平均占比和分布。def __init__(self):passdef calculate(self,df: pd.DataFrame,unit_col: str unit_id) - pd.DataFrame:按加工单元分组统计Parameters----------df : pd.DataFrame含工时拆解结果的数据unit_col : str加工单元列名Returns-------pd.DataFrameif df.empty or unit_col not in df.columns:return pd.DataFrame()stats df.groupby(unit_col).agg(task_count(total_minutes, count),avg_total_minutes(total_minutes, mean),avg_fixture_minutes(fixture_minutes, mean),avg_cutting_minutes(cutting_minutes, mean),avg_logistics_minutes(logistics_minutes, mean),avg_buffer_minutes(buffer_minutes, mean),avg_fixture_ratio(fixture_ratio, mean),avg_cutting_ratio(cutting_ratio, mean),avg_logistics_ratio(logistics_ratio, mean),avg_buffer_ratio(buffer_ratio, mean),std_fixture_ratio(fixture_ratio, std),max_fixture_ratio(fixture_ratio, max),min_fixture_ratio(fixture_ratio, min),).reset_index()# 四舍五入numeric_cols stats.select_dtypes(include[np.number]).columnsstats[numeric_cols] stats[numeric_cols].round(2)# 按工装等待占比降序stats stats.sort_values(avg_fixture_ratio, ascendingFalse).reset_index(dropTrue)return statsdef get_unit_detail(self,df: pd.DataFrame,unit_id: str,unit_col: str unit_id) - Dict:获取单个单元的详细统计unit_df df[df[unit_col] unit_id]if unit_df.empty:return {}return {unit_id: unit_id,task_count: len(unit_df),avg_total: round(unit_df[total_minutes].mean(), 2),avg_fixture: round(unit_df[fixture_minutes].mean(), 2),avg_cutting: round(unit_df[cutting_minutes].mean(), 2),fixture_ratio_mean: round(unit_df[fixture_ratio].mean(), 2),fixture_ratio_std: round(unit_df[fixture_ratio].std(), 2),fixture_ratio_max: round(unit_df[fixture_ratio].max(), 2),cutting_ratio_mean: round(unit_df[cutting_ratio].mean(), 2),}/detailsdetailssummary/summary瓶颈分析器import numpy as npimport pandas as pdfrom scipy import statsfrom typing import Dict, List, Optionalclass BottleneckAnalyzer:FMS 加工单元瓶颈分析器基于工装等待占比识别瓶颈提供统计检验。def __init__(self, top_k: int 3):self.top_k top_kdef identify_bottlenecks(self,unit_stats_df: pd.DataFrame,unit_col: str unit_id) - pd.DataFrame:识别工装等待瓶颈单元Returns-------pd.DataFrameTop K 瓶颈单元if unit_stats_df.empty:return pd.DataFrame()sorted_df unit_stats_df.sort_values(avg_fixture_ratio, ascendingFalse).reset_index(dropTrue)return sorted_df.head(self.top_k)def compare_units(self,df: pd.DataFrame,unit_a: str,unit_b: str,unit_col: str unit_id) - Dict:对比两个单元的工装等待比例t-testReturns-------dicta_data df[df[unit_col] unit_a][fixture_ratio].dropna()b_data df[df[unit_col] unit_b][fixture_ratio].dropna()if len(a_data) 2 or len(b_data) 2:return {error: 样本量不足}t_stat, p_value stats.ttest_ind(a_data, b_data)return {unit_a: unit_a,unit_b: unit_b,unit_a_mean: round(a_data.mean(), 2),unit_b_mean: round(b_data.mean(), 2),unit_a_std: round(a_data.std(), 2),unit_b_std: round(b_data.std(), 2),t_statistic: round(t_stat, 4),p_value: round(p_value, 6),significant: p_value 0.05,interpretation: (f{unit_a} vs {unit_b}: (差异显著 if p_value 0.05 else 差异不显著) f (p{p_value:.4f})),}def all_pairwise_tests(self,df: pd.DataFrame,unit_col: str unit_id) - List[Dict]:所有单元两两对比units df[unit_col].unique()results []for i, ua in enumerate(units):for ub in units[i 1:]:result self.compare_units(df, ua, ub, unit_col)if error not in result:results.append(result)return resultsdef fixture_vs_cutting_correlation(self,df: pd.DataFrame) - Dict:工装等待与切削时长的相关性if df.empty:return {}corr df[fixture_ratio].corr(df[cutting_ratio])return {correlation: round(corr, 4) if not np.isnan(corr) else None,interpretation: (负相关工装等待越高切削占比越低if corr and corr -0.3else 弱相关或无相关),}/detailsdetailssummary/summaryFMS 工时数据可视化import numpy as npimport pandas as pdimport matplotlib.pyplot as pltimport networkx as nxfrom pathlib import Pathfrom typing import Optional, Listplt.rcParams[font.sans-serif] [SimHei, DejaVu Sans]plt.rcParams[axes.unicode_minus] Falseclass TaskVisualizer:FMS 加工任务工时可视化def __init__(self, results_dir: str results):self.results_dir Path(results_dir)self.results_dir.mkdir(exist_okTrue)self.stage_colors {fixture: #E74C3C, # 红 - 工装等待cutting: #2ECC71, # 绿 - 切削logistics: #3498DB, # 蓝 - 物流buffer: #F39C12, # 橙 - 缓冲}def plot_unit_stack_bar(self,unit_stats_df: pd.DataFrame,unit_col: str unit_id) - None:单元工时堆叠柱状图fig, ax plt.subplots(figsize(14, 7))units unit_stats_df[unit_col].valuesfixture unit_stats_df[avg_fixture_ratio].valuescutting unit_stats_df[avg_cutting_ratio].valueslogistics unit_stats_df[avg_logistics_ratio].valuesbuffer unit_stats_df[avg_buffer_ratio].valuesx np.arange(len(units))width 0.6ax.bar(x, fixture, width, label工装等待,colorself.stage_colors[fixture], bottom0)ax.bar(x, cutting, width, label切削加工,colorself.stage_colors[cutting], bottomfixture)ax.bar(x, logistics, width, label物流等待,colorself.stage_colors[logistics],bottomfixture cutting)ax.bar(x, buffer, width, label缓冲等待,colorself.stage_colors[buffer],bottomfixture cutting logistics)ax.set_xlabel(加工单元, fontsize12)ax.set_ylabel(占比 (%), fontsize12)ax.set_title(各加工单元工时占比分布, fontsize14, fontweightbold)ax.set_xticks(x)ax.set_xticklabels(units, rotation0)ax.legend(fontsize10, locupper right)ax.set_ylim(0, 100)ax.grid(axisy, alpha0.3)# 标注工装等待数值for i, v in enumerate(fixture):ax.text(i, v / 2, f{v:.1f}%, hacenter, vacenter,fontsize9, fontweightbold, colorwhite)plt.tight_layout()plt.savefig(self.results_dir / unit_stack_bar.png,dpi150, bbox_inchestight)plt.close()def plot_unit_pie(self,df: pd.DataFrame,unit_id: str,unit_col: str unit_id) - None:单单元工时饼图unit_df df[df[unit_col] unit_id]if unit_df.empty:returnavg_fixture unit_df[fixture_ratio].mean()avg_cutting unit_df[cutting_ratio].mean()avg_logistics unit_df[logistics_ratio].mean()avg_buffer unit_df[buffer_ratio].mean()sizes [avg_fixture, avg_cutting, avg_logistics, avg_buffer]labels [工装等待, 切削加工, 物流等待, 缓冲等待]colors [self.stage_colors[fixture], self.stage_colors[cutting],self.stage_colors[logistics], self.stage_colors[buffer]]fig, ax plt.subplots(figsize(8, 8))ax.pie(sizes, labelslabels, colorscolors, autopct%1.1f%%,shadowFalse, startangle90)ax.set_title(f{unit_id} 工时分布, fontsize14, fontweightbold)利用AI解决实际问题如果你觉得这个工具好用欢迎关注长安牧笛