产线“体检单”用 Python 揪出那个拖后腿的瓶颈工位*“上个月底生产例会开成了批斗会。厂长把一张 OEE 报表甩在桌上‘这条线设计节拍 90 秒实际跑下来 120 秒那 30 秒被谁吃了’生产班长老李支支吾吾‘可能是焊接工位最近换模频繁。’质量主管插嘴‘不对是检测工位新来的质检员手生。’大家吵成一团谁也拿不出实锤。我回工位翻了翻 MES 导出的 Excel那是 7 个工位 3 天的 5000 多条节拍记录。数据很全但全是‘流水账’只有工单号、工位号、开始时间、结束时间。我想算平均加工时长却发现数据里有‘脏东西’夜班交接的空档、设备报警的停机、甚至还有测试用的空跑数据。以前我们找瓶颈全靠‘拍脑袋’加‘甩锅大会’。老李说焊接慢是因为给他的坯料来晚了质检说慢是因为前面流过来的次品多。谁也不服谁。那天晚上我写了个脚本。核心逻辑很简单先‘洗菜’清洗数据把异常值剔掉再‘算账’按工位分组聚合算出每个工位的平均加工时长最后‘排座次’排序找出耗时最长的那个。结果出来大家都沉默了。不是焊接也不是质检而是‘缓存台’——那个大家眼中最不起眼的过渡工位。因为它堆料太多导致后续工位频繁待机。老李拍大腿‘我就说焊接老等料嘛’厂长盯着那张工位耗时柱状图看了很久说了一句‘以后开会拿数据说话。’现在这个脚本成了我们产线优化的‘标配’。每次换型我都跑一遍。老李说‘以前是瞎子摸象现在是给产线做 CT。’”*—— 对应长安大学《智能制造导论》“智能生产—生产过程优化”通过数据驱动识别生产瓶颈消除浪费实现节拍平衡与效率提升。一、实际应用场景真实痛点场景设定离散制造车间如汽车零部件、电子产品组装一条由 7 个工位组成的自动化产线上料→加工→装配→焊接→检测→缓存→下料。MES 系统记录了每个工位每次作业的开始与结束时间。生产主管发现产线整体节拍慢于设计值但无法定位具体是哪个工位“拖后腿”。现场原话叙事化“我们这条线名义上 90 秒出一个产品实际跑起来总在 120 秒左右晃悠。厂长急天天追产量。问题出在‘看不见’。数据有但散。MES 里能查到每个工位的作业记录但那是‘流水账’。你想知道哪个工位最慢得自己一条条加再除以数量。几千条记录Excel 卡得不行算出来的数还不敢信因为里面混着夜班、停机、调试。上个月为了找瓶颈我们开了三次会。焊接工位老张说‘我这边没问题是前面上料慢我老等料。’上料的小王反驳‘我这边堆了一堆料是焊接跟不上把缓存台塞满了。’检测工位的小李更委屈‘我检得再快前面流不过来我也得干坐着。’大家都在甩锅因为没人能拿出一个全量的、清洗过的统计结果。老李甚至想了个‘土办法’拿秒表站在线头盯半天。但这只能看瞬时状态不能反映全天、全周的趋势而且人一走数据就没了。最头疼的是‘缓存台’。它只是个过渡工位大家都不把它当回事。但数据告诉我它在高峰期堆积了 15 件在制品导致下游工位平均等待时间高达 25 秒。以前我们以为它是‘缓冲’数据证明它成了‘堰塞湖’。后来我写了这个 Python 工具。它干的第一件事就是‘清洗’。比如剔除掉时长超过 10 分钟的异常记录那是换模或维修不是正常加工剔除掉凌晨 2 点到 4 点的数据那是无人值守的空跑测试。然后它按工位分组计算平均加工时长、标准差看稳定性、最大最小值。结果一目了然缓存台的平均滞留时间最长。它不是不干活而是‘干得太慢’积压。我们调整了缓存台的传送逻辑把滞留时间压下去后整线节拍立刻缩短了 15 秒。现在老李他们再也不吵了。每次优化前我先跑脚本把‘体检单’打出来。数据摆在桌面上谁慢谁快一清二楚。厂长说‘这比开十个会都管用。’”痛点分析映射到课程模型《智能制造导论》模块 本篇痛点对应概述优质、高效、低耗 低效产线节拍不达标不透明瓶颈位置模糊缺乏数据支撑。智能制造技术基础工业大数据、数据清洗 数据脏乱原始节拍数据含异常值、空跑数据缺乏分析数据未转化为生产洞察。新一代支撑技术数据分析、可视化 统计分析计算平均时长、标准差可视化柱状图/热力图直观展示瓶颈。智能工厂/智能生产生产过程管控、精益生产 瓶颈识别定位制约产能的关键工位节拍平衡为产线优化提供依据。演进范式经验驱动拍脑袋→ 数据驱动定量分析 从“甩锅大会”到“数据说话”实现生产管理的精细化、透明化。核心矛盾“产线物理运行的连续性”与“管理视角的碎片化”之间的矛盾。产线每时每刻都在运行但管理决策依赖人工统计和主观经验。需要一种工具将物理世界的“时间流”转化为管理世界的“数据洞察”。二、核心逻辑讲解大白话2.1 节拍分析 给产线做“心电图”- 原始数据就像病人的心率记录每跳一下记一笔工位开始、结束。- 加工时长两次心跳之间的间隔。间隔太长说明心脏工位跳得慢间隔太短可能是早搏异常数据。- 平均加工时长这个工位的平均“心跳速度”。速度最慢的就是“心脏病”最严重的——瓶颈。- 数据清洗剔除掉“假心跳”空跑、调试和“乱心跳”设备报警导致的超长记录。- 分组统计不是看单次心跳而是看这个工位一整天、一整周的平均表现。这就好比体检不是看一次血压而是看平均血压。2.2 业务逻辑 → 代码映射MES导出的Excel/CSV原始节拍记录│▼ DataLoader.load_cycle_data()数据加载1. 读取工单号、工位号、开始时间、结束时间2. 转换时间格式字符串→datetime3. 计算原始加工时长结束-开始│▼ DataCleaner.clean_data()数据清洗关键步骤1. 剔除异常时长1秒误触发或600秒换模/维修2. 剔除非生产时段如凌晨2:00-4:00空跑测试3. 剔除无效工位如TEST、MAINTENANCE│▼ CycleAnalyzer.calculate_statistics()统计分析核心1. 按工位分组groupby2. 计算各工位- 平均加工时长mean- 标准差std看稳定性- 加工次数count- 最小/最大时长│▼ BottleneckIdentifier.find_bottleneck()瓶颈定位1. 按平均加工时长降序排序2. 找出Top N如Top 1作为瓶颈3. 计算瓶颈对整线的影响如瓶颈时长/总平均时长│▼ Visualizer.plot_station_duration()可视化1. 柱状图各工位平均加工时长瓶颈高亮2. 箱线图各工位时长分布看离散程度3. 热力图工位×时段的平均时长看时间趋势2.3 为什么不能直接用原始数据的平均值- 异常值干扰一次 10 分钟的换模会拉高平均时长让你误以为工位很慢。- 非生产数据混入空跑测试、调试数据会稀释真实加工时长。- 样本量不均有的工位一天干 500 次有的只干 50 次。直接平均会掩盖低频工位的真实问题。- 清洗的意义就像体检前空腹排除干扰因素才能得到真实的“健康指标”。三、OOP 代码实现3.1 项目结构production_cycle_analysis/├── production_cycle_analysis.py # 核心数据加载、清洗、分析、可视化├── test_production_cycle_analysis.py # 单元测试 9/9├── visualize.py # 可视化入口├── README.md├── cycle_raw_data.csv # 原始节拍数据示例├── cycle_cleaned_data.csv # 清洗后数据├── station_stats.csv # 工位统计结果├── bottleneck_report.csv # 瓶颈分析报告├── station_duration_bar.png # 工位平均时长柱状图├── station_duration_box.png # 工位时长分布箱线图└── station_heatmap.png # 工位-时段热力图3.2 核心源码detailssummary/summary产线节拍分析定位瓶颈工位课程映射长安大学《智能制造导论》概述高效、低耗 → 识别瓶颈提升产线平衡率技术基础工业大数据节拍数据、数据清洗剔除异常支撑技术统计分析均值/标准差、可视化瓶颈定位智能生产生产过程优化节拍平衡、精益生产消除浪费演进范式经验驱动 → 数据驱动量化分析技术栈严格pandas / numpy # 数据加载、清洗、分组统计、计算matplotlib # 可视化柱状图、箱线图、热力图scipy # 统计检验正态性检验、异常值检测scikit-learn # 可选聚类分析识别异常模式networkx # 可选构建工位拓扑图分析上下游影响torch # 可选时序预测预测未来瓶颈from __future__ import annotationsimport osimport warningsfrom dataclasses import dataclass, fieldfrom typing import List, Dict, Tuple, Optional, Anyfrom enum import Enumfrom pathlib import Pathfrom datetime import datetime, timeimport numpy as npimport pandas as pdimport matplotlib.pyplot as pltimport matplotlib.dates as mdatesfrom scipy import statswarnings.filterwarnings(ignore)plt.rcParams[font.sans-serif] [SimHei, DejaVu Sans]plt.rcParams[axes.unicode_minus] False# ----------------------------------------------------------------------# 1. 数据模型与枚举# ----------------------------------------------------------------------class StationType(Enum):LOAD 上料PROCESS 加工ASSEMBLE 装配WELD 焊接INSPECT 检测BUFFER 缓存UNLOAD 下料MAINTENANCE 维护TEST 测试dataclassclass CycleRecord:单条节拍记录work_order: strstation_id: strstation_type: StationTypestart_time: datetimeend_time: datetimeduration_seconds: float field(initFalse)is_valid: bool Truemetadata: Dict field(default_factorydict)def __post_init__(self):self.duration_seconds (self.end_time - self.start_time).total_seconds()if self.duration_seconds 0:raise ValueError(结束时间不能早于开始时间)dataclassclass StationStats:单个工位统计结果station_id: strstation_type: StationTypecount: int # 加工次数mean_duration: float # 平均时长秒std_duration: float # 标准差秒min_duration: float # 最小时长秒max_duration: float # 最大时长秒cv: float # 变异系数std/mean稳定性指标total_duration: float # 总时长秒utilization_rate: float 0.0 # 利用率需结合工作时间计算def __post_init__(self):if self.mean_duration 0:self.cv self.std_duration / self.mean_durationelse:self.cv float(inf)dataclassclass BottleneckReport:瓶颈分析报告bottleneck_station_id: strbottleneck_station_type: StationTypebottleneck_mean_duration: floatall_station_stats: List[StationStats]bottleneck_impact_ratio: float # 瓶颈时长 / 总平均时长bottleneck_rank: intanalysis_time: datetime field(default_factorydatetime.now)# ----------------------------------------------------------------------# 2. 数据加载器# ----------------------------------------------------------------------class DataLoader:加载产线节拍数据。def __init__(self):self.raw_df: Optional[pd.DataFrame] Noneself.station_type_mapping: Dict[str, StationType] {LOAD: StationType.LOAD,PROC: StationType.PROCESS,ASSY: StationType.ASSEMBLE,WELD: StationType.WELD,INSP: StationType.INSPECT,BUFF: StationType.BUFFER,UNLD: StationType.UNLOAD,MAIN: StationType.MAINTENANCE,TEST: StationType.TEST}def load_from_csv(self, csv_path: str) - pd.DataFrame:从CSV加载数据。print(f[INFO] 加载节拍数据: {csv_path})df pd.read_csv(csv_path)# 验证必要列required_cols [work_order, station_id, start_time, end_time]for col in required_cols:if col not in df.columns:raise ValueError(fCSV缺少必要列: {col})# 转换时间格式df[start_time] pd.to_datetime(df[start_time], errorscoerce)df[end_time] pd.to_datetime(df[end_time], errorscoerce)# 删除无效时间df df.dropna(subset[start_time, end_time])# 计算原始时长df[duration_seconds] (df[end_time] - df[start_time]).dt.total_seconds()# 映射工位类型如果CSV中没有则根据ID推断if station_type not in df.columns:df[station_type] df[station_id].str[:4].map(self.station_type_mapping)else:df[station_type] df[station_type].map(self.station_type_mapping)# 按时间排序df.sort_values(start_time, inplaceTrue)df.reset_index(dropTrue, inplaceTrue)self.raw_df dfprint(f 加载 {len(df)} 条记录)print(f 时间范围: {df[start_time].min()} 至 {df[start_time].max()})print(f 工位数量: {df[station_id].nunique()})print(f 工单数量: {df[work_order].nunique()})return dfdef generate_synthetic_data(self,num_stations: int 7,num_orders: int 100,base_cycle_time: float 90.0,noise_level: float 0.2,anomaly_rate: float 0.05,start_date: str 2024-06-01) - pd.DataFrame:生成模拟节拍数据。print(f[INFO] 生成模拟节拍数据...)print(f 工位数量: {num_stations}, 工单数量: {num_orders})np.random.seed(42)station_ids [fST{i:03d} for i in range(1, num_stations 1)]station_types list(StationType)[:num_stations]records []current_time pd.Timestamp(start_date)for order_idx in range(num_orders):order_id fWO{order_idx:06d}order_start_time current_timefor station_idx, (station_id, station_type) in enumerate(zip(station_ids, station_types)):# 模拟不同工位的加工时长制造瓶颈缓存台故意设慢if station_type StationType.BUFFER:mean_duration base_cycle_time * 1.5 # 缓存台慢50%elif station_type StationType.WELD:mean_duration base_cycle_time * 1.2 # 焊接慢20%else:mean_duration base_cycle_time * np.random.uniform(0.8, 1.1)# 添加噪声duration max(1, np.random.normal(mean_duration, mean_duration * noise_level))# 注入异常值换模、维修等if np.random.rand() anomaly_rate:duration np.random.uniform(300, 600) # 5-10分钟start_time current_timeend_time start_time pd.Timedelta(secondsduration)records.append({work_order: order_id,station_id: station_id,station_type: station_type.name,start_time: start_time,end_time: end_time,duration_seconds: duration})current_time end_time# 工单间隔current_time pd.Timedelta(secondsnp.random.uniform(10, 30))df pd.DataFrame(records)df[station_type] df[station_type].map(lambda x: StationType[x] if x in StationType.__members__ else StationType.PROCESS)self.raw_df dfprint(f 生成 {len(df)} 条记录)print(f 平均节拍: {df[duration_seconds].mean():.2f}秒)return df# ----------------------------------------------------------------------# 3. 数据清洗器# ----------------------------------------------------------------------class DataCleaner:清洗节拍数据剔除异常值。def __init__(self,min_duration: float 1.0,max_duration: float 600.0,exclude_night_start: time time(2, 0),exclude_night_end: time time(4, 0),exclude_station_types: List[StationType] None):self.min_duration min_duration # 最小有效时长秒self.max_duration max_duration # 最大有效时长秒self.exclude_night_start exclude_night_startself.exclude_night_end exclude_night_endself.exclude_station_types exclude_station_types or [StationType.MAINTENANCE, StationType.TEST]def clean(self, df: pd.DataFrame) - pd.DataFrame:执行清洗流程。print([INFO] 清洗节拍数据...)original_count len(df)cleaned_df df.copy()# 1. 剔除异常时长cleaned_df self._filter_duration(cleaned_df)print(f 剔除异常时长后: {len(cleaned_df)} 条)# 2. 剔除夜间非生产时段cleaned_df self._filter_night_time(cleaned_df)print(f 剔除夜间时段后: {len(cleaned_df)} 条)# 3. 剔除无效工位类型cleaned_df self._filter_station_types(cleaned_df)print(f 剔除无效工位后: {len(cleaned_df)} 条)# 4. 剔除逻辑错误结束时间早于开始时间cleaned_df self._filter_logic_errors(cleaned_df)print(f 剔除逻辑错误后: {len(cleaned_df)} 条)cleaned_count len(cleaned_df)removed_count original_count - cleaned_countprint(f 总计剔除 {removed_count} 条 ({removed_count/original_count:.2%}))print(f 保留 {cleaned_count} 条有效记录)return cleaned_dfdef _filter_duration(self, df: pd.DataFrame) - pd.DataFrame:剔除时长异常记录。mask (df[duration_seconds] self.min_duration) \(df[duration_seconds] self.max_duration)return df[mask].copy()def _filter_night_time(self, df: pd.DataFrame) - pd.DataFrame:剔除夜间非生产时段。# 提取时间部分time_of_day df[start_time].dt.time# 处理跨夜情况如02:00-04:00if self.exclude_night_start self.exclude_night_end:mask ~((time_of_day self.exclude_night_start) \(time_of_day self.exclude_night_end))else:# 跨午夜如22:00-02:00mask ~((time_of_day self.exclude_night_start) | \(time_of_day self.exclude_night_end))return df[mask].copy()def _filter_station_types(self, df: pd.DataFrame) - pd.DataFrame:剔除无效工位类型。if not self.exclude_station_types:return dfmask ~df[station_type].isin(self.exclude_station_types)return df[mask].copy()def _filter_logic_errors(self, df: pd.DataFrame) - pd.DataFrame:剔除逻辑错误记录。mask df[end_time] df[start_time]return df[mask].copy()# ----------------------------------------------------------------------# 4. 节拍分析器# ----------------------------------------------------------------------class CycleAnalyzer:分析节拍数据计算工位统计指标。def __init__(self, df: pd.DataFrame):self.df df.copy()self.station_stats: List[StationStats] []def calculate_station_statistics(self) - List[StationStats]:按工位计算统计指标。print([INFO] 计算工位统计指标...)# 按工位分组grouped self.df.groupby([station_id, station_type])for (station_id, station_type), group in grouped:count len(group)mean_duration group[duration_seconds].mean()std_duration group[duration_seconds].std()min_duration group[duration_seconds].min()max_duration group[duration_seconds].max()total_duration group[duration_seconds].sum()stats StationStats(station_idstation_id,station_typestation_type,countcount,mean_durationmean_duration,std_durationstd_duration,min_durationmin_duration,max_durationmax_duration,cvstd_duration / mean_duration if mean_duration 0 else float(inf),total_durationtotal_duration)self.station_stats.append(stats)print(f {station_id} ({station_type.value}): f平均{mean_duration:.2f}s, 次数{count}, 稳定性(CV){stats.cv:.3f})print(f 完成 {len(self.station_stats)} 个工位的统计)return self.station_statsdef calculate_overall_statistics(self) - Dict[str, float]:计算整体统计指标。if self.df.empty:return {}overall_mean self.df[duration_seconds].mean()overall_std self.df[duration_seconds].std()overall_cv overall_std / overall_mean if overall_mean 0 else float(inf)# 计算产线平衡率Line Balance Rate, LBR# LBR (各工位平均时间之和 / (瓶颈工位时间 * 工位数)) * 100%if self.station_stats:sum_avg_times sum(stat.mean_duration for stat in self.station_stats)bottleneck_time max(stat.mean_duration for stat in self.station_stats)num_stations len(self.station_stats)lbr (sum_avg_times / (bottleneck_time * num_stations)) * 100 if bottleneck_time 0 else 0else:lbr 0return {overall_mean_duration: overall_mean,overall_std_duration: overall_std,overall_cv: overall_cv,line_balance_rate: lbr,total_records: len(self.df)}# ----------------------------------------------------------------------# 5. 瓶颈识别器# ----------------------------------------------------------------------class BottleneckIdentifier:识别瓶颈工位。def __init__(self, station_stats: List[StationStats]):self.station_stats station_statsdef find_bottleneck(self, top_n: int 1) - BottleneckReport:定位瓶颈工位。print(f[INFO] 定位瓶颈工位 (Top {top_n})...)if not self.station_stats:raise ValueError(无工位统计数据)# 按平均加工时长降序排序sorted_stats sorted(self.station_stats, keylambda x: x.mean_duration, reverseTrue)bottleneck sorted_stats[0]all_mean_duration np.mean([stat.mean_duration for stat in self.station_stats])# 计算瓶颈影响比bottleneck_impact_ratio bottleneck.mean_duration / all_mean_duration if all_mean_duration 0 else 0report BottleneckReport(bottleneck_station_idbottleneck.station_id,bottleneck_station_typebottleneck.station_type,bottleneck_mean_durationbottleneck.mean_duration,all_station_statssorted_stats,bottleneck_impact_ratiobottleneck_impact_ratio,bottleneck_rank1)print(f 瓶颈工位: {bottleneck.station_id} ({bottleneck.station_type.value}))print(f 平均加工时长: {bottleneck.mean_duration:.2f}秒)print(f 瓶颈影响比: {bottleneck_impact_ratio:.2f})print(f 加工次数: {bottleneck.count})print(f 稳定性(CV): {bottleneck.cv:.3f})return reportdef find_bottleneck_by_utilization(self, working_time_seconds: float) - BottleneckReport:基于利用率定位瓶颈需提供总工作时间。print(f[INFO] 基于利用率定位瓶颈...)# 计算各工位利用率for stat in self.station_stats:stat.utilization_rate stat.total_duration / working_time_seconds if working_time_seconds 0 else 0# 按利用率降序排序sorted_stats sorted(self.station_stats, keylambda x: x.utilization_rate, reverseTrue)bottleneck sorted_stats[0]all_utilization np.mean([stat.utilization_rate for stat in self.station_stats])report BottleneckReport(bottleneck_station_idbottleneck.station_id,bottleneck_station_typebottleneck.station_type,bottleneck_mean_durationbottleneck.mean_duration,all_station_statssorted_stats,bottleneck_impact_ratiobottleneck.utilization_rate / all_utilization if all_utilization 0 else 0,bottleneck_rank1)print(f 瓶颈工位(利用率): {bottleneck.station_id} ({bottleneck.station_type.value}))print(f 利用率: {bottleneck.utilization_rate:.2%})return report# ----------------------------------------------------------------------# 6. 可视化# ----------------------------------------------------------------------class Visualizer:可视化分析结果。staticmethoddef plot_station_duration_bar(station_stats: List[StationStats],b利用AI解决实际问题如果你觉得这个工具好用欢迎关注长安牧笛