产线换型等待时长分析用Python算清工单之间的隐形浪费周一早晨的生产例会生产主管老赵把一摞工单记录甩在桌上脸色不太好看。上周我们产线的OEE是78%老赵指着投影看起来还行但你们知道损失在哪吗我扫了一眼报表设备停机时间不多啊故障才停了四十分钟。问题不在故障停机。老赵调出另一张表你们看工单切换——周五下午工单A做完是14:30工单B开始是15:45。这中间75分钟设备在干嘛换型。我说换模具、调参数、首件检验这些都要时间。对但75分钟是不是太长了老赵说标准换型时间应该是30分钟。这多出来的45分钟就是纯浪费——设备开着但没在产出。那你们MES系统没有换型时长统计吗我问。MES有每个工单的开始时间和结束时间。老赵摇头但它不会自动算工单A结束到工单B开始之间隔了多久。我只能自己拿两个时间相减手动一条条算。上周有四十多个工单切换我算了半天眼睛都花了还不确定有没有算错。这就是一个时间差计算的问题。我打开电脑你把工单数据导出来我写个程序按时间排序后用当前工单的开始时间减去上一条工单的结束时间差值就是换型等待时长。再自动和标准时间对比超标的标红。我敲了几行代码import pandas as pd# 读取工单数据df pd.read_csv(work_orders.csv, parse_dates[start_time, end_time])# 按结束时间排序df df.sort_values(end_time).reset_index(dropTrue)# 计算换型时长 当前工单开始时间 - 上一条工单结束时间分钟df[prev_end] df[end_time].shift(1)df[changeover_min] (df[start_time] - df[prev_end]).dt.total_seconds() / 60# 标记异常df[is_abnormal] df[changeover_min] 30就这些老赵瞪大了眼睛。核心就这么多。我运行代码屏幕上跳出了结果工单 上条结束 本单开始 换型时长 是否超标WO-1023 14:30 15:45 75.0 min 超标WO-1024 16:30 16:50 20.0 min 正常WO-1025 18:10 19:30 80.0 min 超标...你看我指着屏幕40次换型里有12次超过了30分钟的标准平均超标时长42分钟。光上周就浪费了8.4小时的产能。老赵沉默了几秒8.4小时……如果把这些时间省下来能多跑两个小批量订单。而且这还是上周的数据。我说如果把过去三个月的数据都跑一遍你就能看出换型时长的趋势——是越来越好还是越来越差。还能按产品类型分组看哪些产品切换特别慢。老赵把结果拿给车间主任看。主任看完当场说下周开始换型超30分钟的班长要写原因。数据你每周给我一份。以前我只知道换型花了时间现在我知道每次换型花了多少、超了多少、浪费了多少产能。老赵说这一个时间差帮我们看见了看不见的浪费。一、实际应用场景真实痛点场景设定工厂产线按工单Work Order组织生产每个工单有产品型号、计划数量、实际产出、开始时间、结束时间。不同产品之间需要换型更换模具/夹具、调整设备参数、首件检验等这段时间设备不产出是OEE损失的重要来源。生产管理人员需要量化每次换型的等待时长识别超标换型分析换型效率趋势。但MES系统只记录工单的开始和结束时间不自动计算工单之间的间隔工程师需要手动逐条计算面对数十甚至上百次换型时效率低下且容易出错。现场原话叙事化我们车间有句老话设备不转就是在烧钱。老赵说但换型的时候设备确实在转——空转、调试、试生产。它消耗了电、气和人工但没有产出合格品。这段时间到底有多长没人能准确说出来。那你们SMED快速换模项目没有测量换型时间吗我问。有标准时间——30分钟。但那是目标值不是实际值。老赵说实际每次换型花了多少取决于当班的熟练度、物料是否到位、设备有没有异常。我想知道的是实际值和标准值之间差了多少差在哪所以你需要的是换型等待时长的自动计算与对比分析。对。而且不只是算出来。老赵补充我想按产品组合来看——从A产品切到B产品要多久从B切到C又要多久。有些组合可能特别慢那就值得专门优化。核心矛盾生产管理需要量化换型损失以驱动效率改善与MES数据只记录工单边界时间缺乏自动计算工单间隔的能力之间的冲突。需要一个换型等待时长分析程序自动排序工单、计算相邻工单时间差、对比标准值标记异常并支持按产品组合分析。二、痛点分析映射到长安大学《智能制造导论》课程模型《智能制造导论》模块 本篇痛点对应概述生产计划与排程、OEE设备综合效率 OEE换型等待是性能损失和计划停机的重要组成部分直接影响OEE。智能制造技术基础MES制造执行系统、工单管理 MES数据工单起止时间是MES的基础数据但原始数据需要二次加工才能产生管理洞察。新一代支撑技术工业大数据生产数据分析、过程挖掘 过程挖掘从工单时序数据中挖掘换型模式发现效率瓶颈。智能工厂与智能生产精益生产SMED快速换模、生产透明化 精益生产量化换型损失是实施SMED的前提——不能测量就不能改善。演进范式凭经验估算 → 手动Excel计算 → 自动时间差分析 → 实时换型监控 从大概半小时吧到精确75分钟超标45分钟实现换型过程的数字化透明。一句话总结我们需要构建一个换型等待时长分析程序基于工单起止时间数据自动计算相邻工单间的时间间隔对比标准换型时间识别异常为SMED改善提供数据基础。三、核心逻辑讲解大白话3.1 问题本质把换型等待看成火车换乘把换型等待和火车换乘的关系想象成赶火车* 每个工单 一趟列车列车A在14:30到站工单A结束列车B在15:45发车工单B开始。* 换型等待 换乘时间你下了列车A走到另一个站台等列车B发车。这段时间里你不在任何一趟列车上——这就是换乘等待。* 标准换型时间 计划换乘时间正常情况下走到另一个站台需要10分钟标准换型30分钟。* 超标 换乘花了太久如果走到站台发现列车晚点了、或者走错了站台又绕回来换乘时间就变成了45分钟——超标了。* 时间差计算 用发车时间减去到站时间15:45 - 14:30 75分钟。这就是换乘等待。工业应用* 输入工单CSV包含字段工单号、产品型号、开始时间、结束时间。* 排序按结束时间升序排列因为换型是从上一条工单结束开始的。* 时间差当前工单的start_time - 上一条工单的end_time 换型等待时长。* 对比标准如果换型时长 标准值如30分钟标记为超标。* 产品组合分析看从产品A切到产品B的换型时长是否特别长。3.2 业务逻辑 → 代码映射读取工单CSV│▼ DataLoader.load()数据加载1. pd.read_csv() 读取工单2. parse_dates 解析开始/结束时间│▼ DataPreprocessor.preprocess()数据预处理1. 按结束时间排序2. 去除异常数据结束时间晚于开始时间等│▼ ChangeoverCalculator.calculate()换型时长计算1. 用 shift(1) 获取上一条工单的结束时间2. 当前开始 - 上条结束 换型时长分钟3. 第一条工单无前置换型时长设为 NaN│▼ ChangeoverAnalyzer.analyze()分析1. 对比标准换型时间标记超标2. 按产品组合前产品→后产品分组统计3. 计算超标率和平均超标时长│▼ Visualizer.plot()可视化1. 换型时长趋势图按时间顺序标注超标线2. 产品组合换型时长热力图3. 超标分布饼图│▼ ReportGenerator.generate_report()生成报告1. 总体换型统计2. 超标清单3. SMED改善建议3.3 为什么用shift(1)而不是自己写循环*shift(1)pandas的向量化操作把整列向下移动一行瞬间完成取上一条记录的操作。不需要写for循环代码简洁且执行效率高。* 循环方式for i in range(1, len(df)): diff df.iloc[i][start] - df.iloc[i-1][end]。功能一样但代码啰嗦大数据量时慢。* 工程选择本例用shift(1)一行代码搞定时间差计算。同时处理第一条工单无前置工单的情况——shift后第一条的prev_end是NaT对应的换型时长自然为NaN不参与超标判断。3.4 如何处理同一产线多台设备的情况* 问题如果数据来自多条产线或多台设备工单混在一起排序会出错——设备A的工单结束时间不应该和设备B的工单开始时间计算换型。* 处理策略增加equipment_id字段按设备分组后分别计算换型时长。* 本例处理在配置中支持group_by参数如果指定了设备列则先groupby再计算。四、OOP 代码实现4.1 项目结构changeover_analysis/├── data/│ └── work_orders.csv # 工单数据├── results/ # 输出结果│ ├── changeover_summary.csv│ ├── changeover_trend.png│ ├── product_matrix_heatmap.png│ ├── abnormal_pie.png│ └── changeover_report.txt├── changeover_analysis.py # 核心代码├── test_changeover_analysis.py # 单元测试├── README.md└── requirements.txt4.2 核心源码detailssummary/summary产线换型等待时长分析课程映射长安大学《智能制造导论》概述生产计划与排程、OEE设备综合效率技术基础MES制造执行系统、工单管理支撑技术工业大数据生产数据分析、过程挖掘智能工厂精益生产SMED快速换模、生产透明化演进范式凭经验估算 → 手动Excel计算 → 自动时间差分析 → 实时换型监控技术栈严格numpy # 数值计算pandas # 数据加载、时间差计算matplotlib # 可视化networkx # 无scikit-learn # 无预留聚类scipy # 无torch # 无from __future__ import annotationsimport osfrom dataclasses import dataclass, fieldfrom pathlib import Pathfrom typing import List, Dict, Optionalfrom datetime import datetime, timedeltaimport randomimport numpy as npimport pandas as pdimport matplotlib.pyplot as pltplt.rcParams[font.sans-serif] [SimHei, DejaVu Sans]plt.rcParams[axes.unicode_minus] False# ----------------------------------------------------------------------# 1. 配置# ----------------------------------------------------------------------dataclassclass ChangeoverConfig:换型分析配置data_dir: str dataresults_dir: str resultsdata_file: str work_orders.csv# 列名wo_col: str work_orderproduct_col: str product_codestart_col: str start_timeend_col: str end_timeequipment_col: str equipment_id # 可选# 标准换型时间分钟standard_changeover_min: float 30.0# 分析参数group_by_equipment: bool Falserandom_seed: int 42# ----------------------------------------------------------------------# 2. 数据加载器# ----------------------------------------------------------------------class DataLoader:工单数据加载器def __init__(self, config: ChangeoverConfig):self.config configself.data_dir Path(config.data_dir)os.makedirs(self.data_dir, exist_okTrue)def generate_synthetic_data(self, n_orders: int 50):生成模拟工单数据print(f[INFO] 生成模拟工单数据{n_orders}条...)random.seed(self.config.random_seed)np.random.seed(self.config.random_seed)products [P001, P002, P003, P004, P005, P006]equipments [EQ-A, EQ-B]base_time datetime.now() - timedelta(days7)data []current_time base_timefor i in range(1, n_orders 1):product random.choice(products)equipment random.choice(equipments) if self.config.group_by_equipment else EQ-A# 生产时长2~8小时production_min random.randint(120, 480)start current_timeend start timedelta(minutesproduction_min)# 换型等待第一条无前置if i 1:# 标准30分钟但有一定概率超标if random.random() 0.3: # 30%概率超标wait_min random.uniform(35, 90)else:wait_min random.uniform(15, 30)start end timedelta(minuteswait_min)end start timedelta(minutesproduction_min)current_time end timedelta(minutesrandom.uniform(5, 15)) # 微小间隔data.append({work_order: fWO-{2024000 i},product_code: product,equipment_id: equipment,start_time: start,end_time: end})df pd.DataFrame(data)df.to_csv(self.data_dir / self.config.data_file, indexFalse)print(f 生成数据: {len(df)}条)return dfdef load_data(self) - pd.DataFrame:加载工单数据print(f[INFO] 加载工单数据...)csv_path self.data_dir / self.config.data_fileif not csv_path.exists():self.generate_synthetic_data()df pd.read_csv(csv_path, parse_dates[self.config.start_col, self.config.end_col])print(f 加载数据: {len(df)}条)return df# ----------------------------------------------------------------------# 3. 数据预处理器# ----------------------------------------------------------------------class DataPreprocessor:数据预处理def __init__(self, config: ChangeoverConfig):self.config configdef preprocess(self, df: pd.DataFrame) - pd.DataFrame:预处理print(f[INFO] 预处理数据...)df df.copy()# 按结束时间排序df df.sort_values(self.config.end_col).reset_index(dropTrue)# 去除异常结束时间早于开始时间valid df[self.config.end_col] df[self.config.start_col]if (~valid).sum() 0:print(f 去除异常记录: {(~valid).sum()}条)df df[valid]print(f 预处理完成: {len(df)}条)return df# ----------------------------------------------------------------------# 4. 换型计算器# ----------------------------------------------------------------------class ChangeoverCalculator:计算换型等待时长def __init__(self, config: ChangeoverConfig):self.config configdef calculate(self, df: pd.DataFrame) - pd.DataFrame:计算换型时长print(f[INFO] 计算换型等待时长...)df df.copy()if self.config.group_by_equipment:# 按设备分组计算df[prev_end] df.groupby(self.config.equipment_col)[self.config.end_col].shift(1)else:df[prev_end] df[self.config.end_col].shift(1)# 计算换型时长分钟df[changeover_min] (df[self.config.start_col] - df[prev_end]).dt.total_seconds() / 60# 第一条的换型时长为NaNvalid_count df[changeover_min].notna().sum()print(f 计算完成: {valid_count}次换型)return df# ----------------------------------------------------------------------# 5. 换型分析器# ----------------------------------------------------------------------class ChangeoverAnalyzer:换型分析def __init__(self, config: ChangeoverConfig):self.config configdef analyze(self, df: pd.DataFrame) - pd.DataFrame:分析换型数据print(f[INFO] 分析换型数据...)df df.copy()# 标记超标df[is_abnormal] df[changeover_min] self.config.standard_changeover_mindf[is_abnormal] df[is_abnormal].fillna(False)# 产品组合df[from_product] df[product_code].shift(1)df[to_product] df[product_code]df[product_pair] df[from_product] → df[to_product]# 统计total df[changeover_min].notna().sum()abnormal df[is_abnormal].sum()if total 0:print(f 总换型次数: {total})print(f 超标次数: {abnormal} ({abnormal/total*100:.1f}%))print(f 平均换型时长: {df[changeover_min].mean():.1f}分钟)return dfdef product_pair_stats(self, df: pd.DataFrame) - pd.DataFrame:按产品组合统计valid df[df[changeover_min].notna()].copy()if len(valid) 0:return pd.DataFrame()stats valid.groupby(product_pair).agg(count(changeover_min, count),avg_min(changeover_min, mean),max_min(changeover_min, max),abnormal_count(is_abnormal, sum)).reset_index()stats[abnormal_rate] stats[abnormal_count] / stats[count] * 100stats stats.sort_values(avg_min, ascendingFalse)return stats# ----------------------------------------------------------------------# 6. 可视化器# ----------------------------------------------------------------------class Visualizer:可视化结果def __init__(self, config: ChangeoverConfig):self.config configself.results_dir Path(config.results_dir)os.makedirs(self.results_dir, exist_okTrue)def plot_trend(self, df: pd.DataFrame):换型时长趋势图print(f[INFO] 绘制换型时长趋势图...)valid df[df[changeover_min].notna()].reset_index(dropTrue)fig, ax plt.subplots(figsize(14, 6))colors [red if x else steelblue for x in valid[is_abnormal]]ax.bar(range(len(valid)), valid[changeover_min], colorcolors, alpha0.8)ax.axhline(yself.config.standard_changeover_min, colororange, linestyle--,linewidth2, labelf标准换型时间({self.config.standard_changeover_min}min))ax.set_xlabel(换型序号, fontsize12)ax.set_ylabel(换型时长分钟, fontsize12)ax.set_title(换型等待时长趋势红色超标, fontsize14, fontweightbold)ax.legend(fontsize11)ax.grid(True, alpha0.3, axisy)plt.tight_layout()plt.savefig(self.results_dir / changeover_trend.png, dpi150, bbox_inchestight)plt.close()print(f 已保存: {self.results_dir / changeover_trend.png})def plot_product_heatmap(self, stats: pd.DataFrame):产品组合换型时长热力图print(f[INFO] 绘制产品组合热力图...)if len(stats) 0:return# 构建矩阵pairs stats[product_pair].str.split( → , expandTrue)from_products pairs[0].unique()to_products pairs[1].unique()matrix np.full((len(from_products), len(to_products)), np.nan)for _, row in stats.iterrows():i list(from_products).index(row[product_pair].split( → )[0])j list(to_products).index(row[product_pair].split( → )[1])matrix[i, j] row[avg_min]fig, ax plt.subplots(figsize(10, 8))im ax.imshow(matrix, cmapYlOrRd, aspectauto)ax.set_xticks(range(len(to_products)))ax.set_xticklabels(to_products, rotation45, haright)ax.set_yticks(range(len(from_products)))ax.set_yticklabels(from_products)ax.set_xlabel(目标产品, fontsize12)ax.set_ylabel(来源产品, fontsize12)ax.set_title(产品组合换型时长热力图分钟, fontsize14, fontweightbold)# 标注数值for i in range(len(from_products)):for j in range(len(to_products)):if not np.isnan(matrix[i, j]):ax.text(j, i, f{matrix[i, j]:.0f}, hacenter, vacenter, fontsize9)plt.colorbar(im, axax, label平均换型时长分钟)plt.tight_layout()plt.savefig(self.results_dir / product_matrix_heatmap.png, dpi150, bbox_inchestight)plt.close()print(f 已保存: {self.results_dir / product_matrix_heatmap.png})def plot_pie(self, df: pd.DataFrame):超标分布饼图print(f[INFO] 绘制超标分布饼图...)valid df[df[changeover_min].notna()]counts valid[is_abnormal].value_counts()fig, ax plt.subplots(figsize(8, 8))labels [正常, 超标] if True in counts.index else [正常]sizes [counts.get(False, 0), counts.get(True, 0)]colors [steelblue, red]ax.pie(sizes, labelslabels, colorscolors, autopct%1.1f%%,startangle90, textprops{fontsize: 12})ax.set_title(换型超标分布, fontsize14, fontweightbold)plt.tight_layout()plt.savefig(self.results_dir / abnormal_pie.png, dpi150, bbox_inchestight)plt.close()print(f 已保存: {self.results_dir / abnormal_pie.png})# ----------------------------------------------------------------------# 7. 报告生成器# ----------------------------------------------------------------------class ReportGenerator:分析报告生成器def __init__(self, config: ChangeoverConfig):self.config configself.results_dir Path(config.results_dir)os.makedirs(self.results_dir, exist_okTrue)def generate(self, df: pd.DataFrame, pair_stats: pd.DataFrame) - str:生成报告print(f[INFO] 生成分析报告...)report_lines []report_lines.append( * 80)report_lines.append(产线换型等待时长分析报告)report_lines.append( * 80)report_lines.append(f\n分析时间: {datetime.now().strftime(%Y-%m-%d %H:%M:%S)})valid df[df[changeover_min].notna()]total len(valid)abnormal valid[is_abnormal].sum()# 概况report_lines.append(f\n数据概况:)report_lines.append(f 工单总数: {len(df)})report_lines.append(f 有效换型次数: {total})report_lines.append(f 超标次数: {abnormal} ({abnormal/total*100:.1f}%) if total 0 else 无有效换型数据)if total 0:report_lines.append(f 平均换型时长: {valid[changeover_min].mean():.1f}分钟)report_lines.append(f 最长换型: {valid[changeover_min].max():.1f}分钟)report_lines.append(f 最短换型: {valid[changeover_min].min():.1f}分钟)# 超标清单abnormal_list valid[valid[is_abnormal]]if len(abnormal_list) 0:report_lines.append(f\n超标换型清单:)report_lines.append(- * 70)for _, row in abnormal_list.head(10).iterrows():report_lines.append(f {row[work_order]:15s} f{row[from_product]}→{row[to_product]:5s} f换型:{row[changeover_min]:6.1f}min f(超标{row[changeover_min] - self.config.standard_changeover_min:.1f}min))# 产品组合分析if len(pair_stats) 0:report_lines.append(f\n产品组合换型时长排名:)report_lines.append(- * 50)for _, row in pair_stats.head(5).iterrows():report_lines.append(f {row[product_pair]:15s} f平均:{row[avg_min]:5.1f}min f次数:{row[count]:3d} f超标率:{row[abnormal_rate]:.0f}%)# 建议report_lines.append(f\nSMED改善建议:)report_lines.append(- * 40)if len(pair_stats) 0:worst pair_stats.iloc[0]report_lines.append(f 1. 优先优化 {worst[product_pair]} 组合平均{worst[avg_min]:.0f}分钟)report_lines.append(f 2. 将换型时间纳入班组KPI考核每周回顾超标情况)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 / changeover_report.txtwith open(report_path, w, encodingutf-8) as f:f.write(report_text)print(f 报告已保存: {report_path})return report_text# ----------------------------------------------------------------------# 8. 主程序演示# ----------------------------------------------------------------------def demo():完整演示流程print( * 80)print(产线换型等待时长分析)print( * 80)config ChangeoverConfig(data_dirdata,results_dirresults,group_by_equipmentFalse)# 1. 加载print(\n[INFO] 步骤1: 加载数据...)loader DataLoader(config)df loader.load_data()# 2. 预处理print(\n[INFO] 步骤2: 预处理...)preprocessor DataPreprocessor(config)df preprocessor.preprocess(df)# 3. 计算换型时长print(\n[INFO] 步骤3: 计算换型时长...)calculator ChangeoverCalculator(config)df calculator.calculate(df)# 4.利用AI解决实际问题如果你觉得这个工具好用欢迎关注长安牧笛