ARTICLE DETAIL

资讯详情

深耕编程入门与网站建设的一线实战洞察。

python的先进制造技术工业场景模拟第四篇:加载FMS车间工单数据表,按产品类型分组,计算各类工单平均加工时长。

python的先进制造技术工业场景模拟第四篇:加载FMS车间工单数据表,按产品类型分组,计算各类工单平均加工时长。 周四下午FMS 车间调度室。这个月第三批订单又延期了。生产主管老刘把工单 Excel 表甩到桌上转头冲着调度系统喊小王你那个排产软件不是说能算出来吗结果还是拖了两天我拉过椅子坐下打开那份工单表。几百行数据字段包括工单号、产品类型A型法兰/B型壳体/C型轴套、计划加工时长分钟、实际加工时长分钟、设备编号、操作员、完成状态。问题一眼就看到了——小王你排产的时候A 型法兰的计划时长是按什么给的按经验啊小王挠挠头之前做的几批差不多都是 90 分钟一个。差不多我指着屏幕你看这组数据——A 型法兰实际加工时长从 72 分钟到 156 分钟波动快一倍了。你用 90 分钟去排产遇到实际要 150 分钟的批次整条线就堵死了。老刘凑过来那到底该用多少分钟排产得按产品类型分组算每类工单的实际平均加工时长我说不是拍脑袋差不多是用历史数据说话。我敲了几行代码import pandas as pdimport matplotlib.pyplot as plt# 加载工单数据df pd.read_excel(fms_workorders.xlsx)# 按产品类型分组计算平均实际加工时长avg_times df.groupby(product_type)[actual_time].mean()# 画柱状图avg_times.plot(kindbar, color#3498DB)plt.ylabel(平均加工时长 (分钟))plt.title(各类产品平均加工时长)plt.show()print(avg_times)# A型法兰 112.3# B型壳体 87.6# C型轴套 54.2这只是核心逻辑我说完整版会用 OOP 封装一个类管工单数据加载一个类管数据清洗处理空值、异常值一个类管分组统计按产品类型算平均/最大/最小加工时长一个类管可视化柱状图箱线图。数据自包含——用 numpy 合成一份工单 CSV读者下载就能跑。老刘盯着屏幕所以你这东西就是把几百个工单按产品类型分堆每堆算一个靠谱的平均加工时间——以后排产就用这个数不用再差不多了。对。而且不光是平均我补充箱线图还能看出每类产品的加工时长波动范围——波动大的产品排产时要多留缓冲时间。小王点点头那我明天就把这个数嵌进排产逻辑里。一、实际应用场景真实痛点场景设定柔性制造系统FMS车间承接多品种小批量订单每种产品A型法兰、B型壳体、C型轴套等的加工工艺路径相似但参数不同历史工单记录了计划加工时长和实际加工时长。现场常见痛点排产时凭经验估算加工时长未利用历史工单数据计算各类产品的实际平均加工时长导致排产偏差大、订单延期、设备利用率低。现场原话叙事化我不是没有数据小王说工单系统里几百条历史记录计划时间和实际时间都有。但我从来没认真分析过——每次排产都是凭感觉给个数差不多就行。差不多就是差很多我说A 型法兰你给 90 分钟实际平均要 112 分钟每批 10 个就是多出来将近 4 个小时。一天排三批直接拖到第二天。核心矛盾经验估算的加工时长与基于历史数据的实际平均加工时长之间的偏差。需要一个FMS 工单加工时长分析程序用pandas 加载工单数据按产品类型分组计算平均/统计加工时长用matplotlib 可视化对比为排产提供数据支撑。二、痛点分析映射到滨州职业学院《先进制造技术》课程模型《先进制造技术》模块 本篇痛点对应柔性制造系统FMS与先进生产管理生产计划与排程 工单加工时长统计排产的基础输入是准确的工序时间经验估算导致计划偏差。先进制造技术基础制造系统性能评估 加工效率分析通过历史数据统计各类产品的实际加工时长评估制造系统效率。智能制造与数字孪生制造大数据驱动决策 数据驱动的排产优化从工单历史数据中挖掘规律替代经验判断。数控加工与CAD/CAM技术数控加工工艺规划 工艺参数与加工时间关联不同产品的数控加工程序复杂度不同加工时长存在差异。一句话总结我们需要构建一个FMS 工单加工时长分析程序用pandas 加载和分组统计工单数据用numpy 计算统计指标用matplotlib 绘制柱状图和箱线图输出各类产品的平均加工时长及波动范围为排产决策提供数据依据。三、核心逻辑讲解大白话3.1 问题本质把工单分析想象成快递分拣计时把 FMS 车间的工单数据想象成快递分拣中心的历史记录* 工单数据表 快递分拣记录每次分拣一批包裹记录包裹类型小件/中件/大件、计划分拣时间、实际分拣时间。* 产品类型分组 按包裹大小分堆小件放一堆、中件放一堆、大件放一堆。* 计算平均加工时长 算每堆的平均分拣时间小件平均 5 分钟、中件平均 12 分钟、大件平均 25 分钟。* 排产用平均时长 按平均时间预估下次来 100 个小件按 5×100500 分钟排计划而不是凭感觉说大概两小时。* 箱线图 看波动范围小件大部分在 4~6 分钟偶尔有 10 分钟的异常排产时多留点余量。工业应用* pandas 加载read_csv() /read_excel() 读取工单表。* 分组统计groupby(product_type)[actual_time].agg([mean, std, min, max]) 一行代码算出每类产品的平均、标准差、最小、最大加工时长。* 异常值处理用numpy 结合箱线图法则IQR识别异常工单如设备故障导致的异常长加工时间避免拉偏平均值。* matplotlib 可视化柱状图对比各类产品平均时长箱线图展示每类产品的时长分布和波动。3.2 业务逻辑 → 代码映射定义工单数据模型│▼ WorkorderLoader (pandas)加载工单 CSV/Excelpd.read_csv(fms_workorders.csv)解析产品类型、计划时长、实际时长│▼ DataCleaner (pandas/numpy)数据清洗处理空值删除或标记异常值检测IQR 法则过滤未完成工单│▼ DurationAnalyzer (pandas/numpy)加工时长分析groupby(product_type).agg(mean(actual_time, mean),std(actual_time, std),min(actual_time, min),max(actual_time, max),count(actual_time, count),)计算计划vs实际偏差│▼ Visualizer (matplotlib)可视化1. 柱状图各类产品平均加工时长2. 箱线图每类产品的时长分布3. 对比图计划 vs 实际按产品类型│▼ SyntheticDataGenerator (numpy)合成数据生成用 numpy 按不同分布生成各类产品的加工时长包含正常波动和少量异常值确保数据自包含、可复现3.3 为什么用分组统计而不是逐个看* 问题几百条工单逐行翻看无法形成这类产品一般要多久的整体认知。* 处理策略groupby 按产品类型分堆一次性算出每堆的统计指标效率高且信息密度大。* 工程合理性FMS 车间产品种类可能多达几十种分组统计是生产数据分析的标准方法。3.4 分析前后对比维度 经验估算 数据驱动分析加工时长 差不多 90 分钟 A型法兰 112.3±18.5 分钟排产准确性 偏差 20%~50% 偏差 10%异常处理 不知道有异常 IQR 识别异常工单排除后重算可视化 无 柱状图箱线图一目了然四、OOP 代码实现4.1 项目结构fms_workorder_analysis/├── fms_workorder_analysis/│ ├── __init__.py│ ├── workorder_loader.py # 工单数据加载│ ├── data_cleaner.py # 数据清洗与异常检测│ ├── duration_analyzer.py # 加工时长分析│ ├── visualizer.py # 可视化│ └── synthetic_data.py # 合成数据生成├── tests/│ ├── __init__.py│ └── test_workorder_analysis.py # 单元测试├── results/│ ├── avg_duration_bar.png # 平均加工时长柱状图│ ├── duration_boxplot.png # 加工时长箱线图│ ├── plan_vs_actual.png # 计划vs实际对比图│ ├── duration_summary.csv # 加工时长汇总│ └── analysis_report.txt # 分析报告└── run_analysis.py # 主程序入口4.2 核心源码detailssummary/summaryFMS 工单数据加载器import pandas as pdfrom pathlib import Pathfrom typing import Optionalclass WorkorderLoader:FMS 车间工单数据加载器支持 CSV 和 Excel 格式的工单数据。def __init__(self, filepath: str, encoding: str utf-8):self.filepath Path(filepath)self.encoding encodingself._raw_df: Optional[pd.DataFrame] Nonedef load(self,product_col: str product_type,plan_col: str plan_time,actual_col: str actual_time,status_col: str status) - pd.DataFrame:加载工单数据Parameters----------product_col : str产品类型列名plan_col : str计划加工时长列名actual_col : str实际加工时长列名status_col : str工单状态列名Returns-------pd.DataFrameif not self.filepath.exists():raise FileNotFoundError(f文件不存在: {self.filepath})suffix self.filepath.suffix.lower()if suffix .csv:self._raw_df pd.read_csv(self.filepath, encodingself.encoding)elif suffix in (.xlsx, .xls):self._raw_df pd.read_excel(self.filepath)else:raise ValueError(f不支持的文件格式: {suffix})# 列名标准化col_aliases {product_col: [产品类型, product, item_type, 型号],plan_col: [计划时长, plan_duration, planned_time, 标准工时],actual_col: [实际时长, actual_duration, real_time, 实际工时],status_col: [状态, status, 完成状态, order_status],}rename_map {}for target, aliases in 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)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 Tuple, Optionalclass DataCleaner:FMS 工单数据清洗器处理空值、过滤未完成工单、检测异常加工时长。def __init__(self, iqr_multiplier: float 1.5):self.iqr_multiplier iqr_multiplierdef clean(self,df: pd.DataFrame,product_col: str product_type,plan_col: str plan_time,actual_col: str actual_time,status_col: str status) - pd.DataFrame:清洗工单数据Parameters----------df : pd.DataFrame原始工单数据product_col : str产品类型列名plan_col : str计划时长列名actual_col : str实际时长列名status_col : str状态列名Returns-------pd.DataFrame清洗后的数据result df.copy()# 确保数值列是 floatresult[plan_col] pd.to_numeric(result[plan_col], errorscoerce)result[actual_col] pd.to_numeric(result[actual_col], errorscoerce)# 过滤未完成工单if status_col in result.columns:completed_mask result[status_col].astype(str).str.contains(完成|done|complete|closed, caseFalse, naFalse)result result[completed_mask]# 删除关键字段为空的行result result.dropna(subset[product_col, actual_col])# 过滤非正数result result[(result[actual_col] 0) (result[plan_col] 0)]return result.reset_index(dropTrue)def detect_outliers(self,df: pd.DataFrame,group_col: str product_type,value_col: str actual_time) - pd.DataFrame:使用 IQR 法则检测异常加工时长Parameters----------df : pd.DataFrame清洗后的数据group_col : str分组列名value_col : str数值列名Returns-------pd.DataFrame新增 is_outlier 列result df.copy()result[is_outlier] Falsefor group_name, group_df in result.groupby(group_col):values group_df[value_col].valuesq1, q3 np.percentile(values, [25, 75])iqr q3 - q1lower q1 - self.iqr_multiplier * iqrupper q3 self.iqr_multiplier * iqrmask (group_df.index.isin(group_df.index)) ((group_df[value_col] lower) |(group_df[value_col] upper))result.loc[group_df.index, is_outlier] maskreturn result/detailsdetailssummary/summary加工时长分析器import numpy as npimport pandas as pdfrom typing import Dict, Optionalclass DurationAnalyzer:FMS 工单加工时长分析按产品类型分组计算统计指标。def __init__(self,product_col: str product_type,plan_col: str plan_time,actual_col: str actual_time):self.product_col product_colself.plan_col plan_colself.actual_col actual_coldef analyze(self, df: pd.DataFrame) - pd.DataFrame:按产品类型分析加工时长Parameters----------df : pd.DataFrame清洗后的数据Returns-------pd.DataFrame每种产品的统计指标summary df.groupby(self.product_col).agg(count(self.actual_col, count),avg_actual(self.actual_col, mean),std_actual(self.actual_col, std),min_actual(self.actual_col, min),max_actual(self.actual_col, max),avg_plan(self.plan_col, mean),).reset_index()# 计算偏差summary[deviation] (summary[avg_actual] - summary[avg_plan])summary[deviation_pct] (summary[deviation] / summary[avg_plan] * 100)# 填充空值summary[std_actual] summary[std_actual].fillna(0)# 排序summary summary.sort_values(avg_actual, ascendingFalse)return summarydef analyze_excluding_outliers(self,df: pd.DataFrame,outlier_col: str is_outlier) - pd.DataFrame:排除异常值后重新分析Parameters----------df : pd.DataFrame含异常标记的数据outlier_col : str异常标记列名Returns-------pd.DataFrame排除异常后的统计结果clean_df df[~df[outlier_col]].copy()return self.analyze(clean_df)def get_recommendation(self, summary: pd.DataFrame) - pd.DataFrame:生成排产建议使用平均时长 1 个标准差作为安全时长Parameters----------summary : pd.DataFrame分析结果Returns-------pd.DataFrame含排产建议时长的数据rec summary.copy()rec[suggested_time] rec[avg_actual] rec[std_actual]return rec[[self.product_col, avg_actual, std_actual,suggested_time, count]]/detailsdetailssummary/summary可视化器import numpy as npimport pandas as pdimport matplotlib.pyplot as pltfrom pathlib import Pathfrom typing import Optionalplt.rcParams[font.sans-serif] [SimHei, DejaVu Sans]plt.rcParams[axes.unicode_minus] Falseclass Visualizer:FMS 工单加工时长可视化def __init__(self, results_dir: str results):self.results_dir Path(results_dir)self.results_dir.mkdir(exist_okTrue)def plot_avg_duration_bar(self,summary: pd.DataFrame,product_col: str product_type,value_col: str avg_actual,title: str 各类产品平均加工时长) - None:绘制平均加工时长柱状图fig, ax plt.subplots(figsize(10, 5))products summary[product_col].tolist()values summary[value_col].tolist()colors plt.cm.Set3(np.linspace(0, 1, len(products)))bars ax.bar(products, values, colorcolors)ax.set_ylabel(平均加工时长 (分钟), fontsize12)ax.set_title(title, fontsize14, fontweightbold)ax.tick_params(axisx, rotation30)# 标注数值for bar, val in zip(bars, values):ax.text(bar.get_x() bar.get_width() / 2,bar.get_height() 1,f{val:.1f},hacenter,vabottom,fontsize9,)plt.tight_layout()plt.savefig(self.results_dir / avg_duration_bar.png,dpi150, bbox_inchestight)plt.close()def plot_duration_boxplot(self,df: pd.DataFrame,product_col: str product_type,value_col: str actual_time) - None:绘制加工时长箱线图fig, ax plt.subplots(figsize(10, 5))products df[product_col].unique()data [df[df[product_col] p][value_col].valuesfor p in products]bp ax.boxplot(data, labelsproducts, patch_artistTrue,showfliersTrue)ax.set_ylabel(加工时长 (分钟), fontsize12)ax.set_title(各类产品加工时长分布, fontsize14, fontweightbold)ax.tick_params(axisx, rotation30)# 着色colors plt.cm.Set3(np.linspace(0, 1, len(products)))for patch, color in zip(bp[boxes], colors):patch.set_facecolor(color)plt.tight_layout()plt.savefig(self.results_dir / duration_boxplot.png,dpi150, bbox_inchestight)plt.close()def plot_plan_vs_actual(self,summary: pd.DataFrame,product_col: str product_type) - None:绘制计划 vs 实际对比图fig, ax plt.subplots(figsize(10, 5))products summary[product_col].tolist()x np.arange(len(products))width 0.35ax.bar(x - width/2, summary[avg_plan], width,label计划时长, color#3498DB, alpha0.8)ax.bar(x width/2, summary[avg_actual], width,label实际时长, color#E74C3C, alpha0.8)ax.set_ylabel(时长 (分钟), fontsize12)ax.set_title(计划 vs 实际加工时长对比, fontsize14, fontweightbold)ax.set_xticks(x)ax.set_xticklabels(products, rotation30)ax.legend()plt.tight_layout()plt.savefig(self.results_dir / plan_vs_actual.png,dpi150, bbox_inchestight)plt.close()/detailsdetailssummary/summary合成数据生成器import numpy as npimport pandas as pdfrom pathlib import Pathfrom typing import Optionalfrom datetime import datetime, timedeltaclass SyntheticDataGenerator:FMS 工单合成数据生成器生成包含不同产品类型、计划/实际时长、异常值的模拟工单数据。def __init__(self, rng: Optional[np.random.RandomState] None):self.rng rng or np.random.RandomState(42)def generate(self,n_records: int 400,start_date: str 2025-01-01,output_path: str fms_workorders.csv) - pd.DataFrame:生成合成工单数据Parameters----------n_records : int工单条数start_date : str起始日期output_path : str输出路径Returns-------pd.DataFrame# 产品类型配置(名称, 计划时长均值, 实际时长均值, 标准差)products [(A型法兰, 90, 110, 18),(B型壳体, 70, 85, 12),(C型轴套, 45, 55, 8),(D型端盖, 60, 72, 10),(E型支架, 100, 120, 22),]start datetime.strptime(start_date, %Y-%m-%d)records []for _ in range(n_records):prod_name, plan_mean, actual_mean, std products[self.rng.randint(0, len(products))]# 计划时长略有波动plan_time max(10, plan_mean self.rng.normal(0, 5))# 实际时长正态分布 少量异常值if self.rng.random() 0.05: # 5% 异常actual_time actual_mean self.rng.uniform(2, 4) * stdelse:actual_time max(5, self.rng.normal(actual_mean, std))# 状态95% 完成status 完成 if self.rng.random() 0.95 else 进行中# 日期date start timedelta(daysint(self.rng.randint(0, 60)))records.append({workorder_id: fWO{_1:05d},product_type: prod_name,plan_time: round(plan_time, 1),actual_time: round(actual_time, 1),status: status,operator: fOP{self.rng.randint(1, 8):02d},date: date.strftime(%Y-%m-%d),})df pd.DataFrame(records)output_path Path(output_path)output_path.parent.mkdir(parentsTrue, exist_okTrue)df.to_csv(output_path, indexFalse, encodingutf-8)return df/detailsdetailssummary/summaryFMS 车间工单加工时长分析课程映射滨州职业学院《先进制造技术》柔性制造系统FMS与先进生产管理生产计划与排程工单加工时长统计先进制造技术基础制造系统性能评估加工效率分析智能制造与数字孪生制造大数据驱动决策数据驱动的排产优化数控加工与CAD/CAM技术数控加工工艺规划工艺参数与加工时间关联技术栈严格pandas # 工单加载、分组统计numpy # 统计计算与合成数据生成matplotlib # 柱状图、箱线图、对比图可视化import sysimport ossys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))import numpy as npimport pandas as pdfrom pathlib import Pathfrom fms_workorder_analysis.workorder_loader import WorkorderLoaderfrom fms_workorder_analysis.data_cleaner import DataCleanerfrom fms_workorder_analysis.duration_analyzer import DurationAnalyzerfrom fms_workorder_analysis.visualizer import Visualizerfrom fms_workorder_analysis.synthetic_data import SyntheticDataGeneratordef main():主程序print( * 70)print(FMS 车间工单加工时长分析)print( * 70)results_dir Path(results)results_dir.mkdir(exist_okTrue)# 1. 生成合成数据print(\n[1/6] 生成合成工单数据...)generator SyntheticDataGenerator(rngnp.random.RandomState(42))csv_path fms_workorders.csvdf_raw generator.generate(n_records400,start_date2025-01-01,output_pathcsv_path,)print(f 生成 {len(df_raw)} 条工单记录)# 2. 加载数据print(\n[2/6] 加载工单数据...)loader WorkorderLoader(csv_path)df loader.load()print(f 加载 {len(df)} 条记录)print(f 产品类型: {df[product_type].unique().tolist()})# 3. 数据清洗print(\n[3/6] 清洗数据...)cleaner DataCleaner(iqr_multiplier1.5)df_clean cleaner.clean(df)print(f 清洗后: {len(df_clean)} 条有效记录)print(f 各产品数量:)for prod, count in df_clean[product_type].value_counts().items():print(f {prod}: {count})# 4. 异常检测print(\n[4/6] 检测异常加工时长...)df_with_outliers cleaner.detect_outliers(df_clean, group_colproduct_type, valu利用AI解决实际问题如果你觉得这个工具好用欢迎关注长安牧笛
返回列表