影刀RPA 制造业自动化MES系统生产数据采集与统计作者林焱什么情况用制造业的车间里最常见的场景是车间主任每天拿着纸质报表把每条产线的产量、不良品数、设备运行时长抄下来再录入到MES系统里。录完数据还要做日报、周报、月报。这个过程至少占了一个人半天的工作量。更麻烦的是——纸质记录容易出错、数据有延迟今天的产量明天才能看到、历史数据追溯困难。影刀RPA可以从MES系统中自动拉数据、做统计、生成报表把车间数据流变成实时的。核心场景从MES/WMS等制造系统自动采集生产数据、生成统计报表。怎么做第一步理清制造数据流┌─────────┐ ┌─────────┐ ┌─────────┐ ┌─────────┐ │ PLC/设备 │ → │ MES系统 │ → │ 数据采集 │ → │ 统计报表 │ │ 实时数据 │ │ (中间层) │ │ (RPA) │ │ (Excel) │ └─────────┘ └─────────┘ └─────────┘ └─────────┘不要试图直接从PLC/传感器读数据——那是工业自动化的事。RPA该做的是操作MES系统的界面或API从数据库/报表中提取数据。第二步MES系统数据采集大多数MES系统是B/S架构的Web系统如西门子SIMATIC IT、达索Apriso、国内用友MES可以用影刀的网页操作节点操控。importrequestsimportpandasaspdfromdatetimeimportdatetime,timedeltaclassMESDataCollector:MES系统数据采集器def__init__(self,mes_url,username,password):self.mes_urlmes_url self.sessionrequests.Session()self._login(username,password)def_login(self,username,password):登录MES系统login_urlf{self.mes_url}/api/auth/loginrespself.session.post(login_url,json{username:username,password:password},timeout30)resp.raise_for_status()print(MES系统登录成功)defget_production_data(self,work_center,date_from,date_to): 按工位和日期范围获取生产数据 返回工单号、产品、计划数、实际产出、不良数、设备运行时间 urlf{self.mes_url}/api/production/reportparams{workCenter:work_center,dateFrom:date_from,dateTo:date_to}respself.session.get(url,paramsparams,timeout30)dataresp.json()# 转换为DataFramerecords[]foritemindata.get(records,[]):records.append({工单号:item.get(workOrder),产品编码:item.get(productCode),产品名称:item.get(productName),计划产量:item.get(planQty),实际产量:item.get(actualQty),不良品数:item.get(defectQty),良率:round(1-item.get(defectQty,0)/max(item.get(actualQty,1),1),3),设备运行时长(min):item.get(runtime),设备停机时长(min):item.get(downtime),产线:item.get(line),日期:item.get(prodDate)})returnpd.DataFrame(records)defget_device_status(self):获取设备实时状态运行/待机/故障/维护urlf{self.mes_url}/api/device/statusrespself.session.get(url,timeout30)devicesresp.json().get(devices,[])dfpd.DataFrame(devices)df[数据时间]datetime.now().isoformat()returndfdefget_oee_data(self,work_center,date_str): 获取OEE数据设备综合效率 OEE 时间开动率 × 性能开动率 × 合格品率 urlf{self.mes_url}/api/oee/calculaterespself.session.get(url,params{workCenter:work_center,date:date_str},timeout30)oeeresp.json()print(fOEE报告 -{work_center}-{date_str})print(f 时间开动率:{oee[availability]:.1%})print(f 性能开动率:{oee[performance]:.1%})print(f 合格品率:{oee[quality]:.1%})print(f OEE:{oee[oee]:.1%})returnoee第三步生产日报自动生成classProductionReporter:生产日报生成器def__init__(self,collector):self.collectorcollectordefgenerate_daily_report(self,report_date):生成单日生产日报# 获取所有工位的生产数据work_centers[WC01,WC02,WC03,WC04]all_data[]forwcinwork_centers:try:dfself.collector.get_production_data(wc,report_date,report_date)all_data.append(df)exceptExceptionase:print(f获取{WC}数据失败{e})ifnotall_data:returnNonecombinedpd.concat(all_data,ignore_indexTrue)# 汇总统计summary{日期:report_date,总计划产量:combined[计划产量].sum(),总实际产量:combined[实际产量].sum(),总不良品:combined[不良品数].sum(),综合良率:round(1-combined[不良品数].sum()/max(combined[实际产量].sum(),1),3),达成率:round(combined[实际产量].sum()/max(combined[计划产量].sum(),1),3),总运行时长(h):round(combined[设备运行时长(min)].sum()/60,1),总停机时长(h):round(combined[设备停机时长(min)].sum()/60,1),}# 生成Excel报表withpd.ExcelWriter(f生产日报_{report_date}.xlsx)aswriter:# Sheet1: 汇总pd.DataFrame([summary]).to_excel(writer,sheet_name汇总,indexFalse)# Sheet2: 明细combined.to_excel(writer,sheet_name明细,indexFalse)# Sheet3: 按产品统计product_summarycombined.groupby(产品名称).agg({实际产量:sum,不良品数:sum,良率:mean}).reset_index()product_summary.to_excel(writer,sheet_name按产品,indexFalse)returnsummary,combineddefgenerate_weekly_report(self,week_start_date):生成周报最近7天fromdatetimeimporttimedelta all_weeks[]foriinrange(7):dateweek_start_datetimedelta(daysi)try:summary,detailself.generate_daily_report(date.strftime(%Y-%m-%d))all_weeks.append(summary)exceptExceptionase:print(f跳过{date}{e})weekly_dfpd.DataFrame(all_weeks)# 趋势图数据trend_data{产量趋势:weekly_df[总实际产量].tolist(),良率趋势:[round(r*100,1)forrinweekly_df[综合良率].tolist()],达成率趋势:[round(r*100,1)forrinweekly_df[达成率].tolist()]}returnweekly_df,trend_data第四步设备异常告警classDeviceAlertMonitor:设备异常监控def__init__(self,collector):self.collectorcollectordefcheck_device_health(self):检查所有设备状态发现异常推送告警dfself.collector.get_device_status()alerts[]# 检查故障设备fault_devicesdf[df[status]故障]for_,deviceinfault_devices.iterrows():alerts.append({级别:严重,设备:device[deviceName],问题:f设备故障当前状态{device[status]},持续时间:device.get(statusDuration,未知)})# 检查停机超过阈值的设备DOWN_THRESHOLD60# 分钟for_,deviceindf.iterrows():downtimedevice.get(downtimeMin,0)ifdowntimeDOWN_THRESHOLD:alerts.append({级别:警告,设备:device[deviceName],问题:f停机超过{DOWN_THRESHOLD}分钟已停机{downtime}分钟,})# 检查OEE低于标准的工位forwcin[WC01,WC02,WC03,WC04]:try:oeeself.collector.get_oee_data(wc,datetime.now().strftime(%Y-%m-%d))ifoee[oee]0.6:# OEE低于60%alerts.append({级别:警告,工位:wc,问题:fOEE偏低{oee[oee]:.1%}阈值60%,详情:f开动率{oee[availability]:.1%}| 性能{oee[performance]:.1%}| 质量{oee[quality]:.1%}})except:passreturnalerts有什么坑坑1MES系统的「伪实时」很多MES号称实时实际上数据刷新周期是5-15分钟。如果你以为数据是实时的写了个每分钟拉一次的流程拉到的可能都是重复数据。解决方法先确认MES的实际数据刷新频率RPA采集频率设得比MES刷新频率慢一倍。坑2车间网络不稳定车间的网络环境和办公室不能比——WiFi信号弱、带宽有限、偶尔断网。如果RPA流程没有健壮的网络处理一天可能挂好几次。解决方法所有网络请求加timeout和重试机制。关键数据采集失败时要有本地缓存兜底。坑3工位编号变化车间调整产线工位编号从WC05变成了WC05-A。你的RPA代码里硬编码了工位列表直接漏掉数据。解决方法工位列表从MES的配置API动态获取不要硬编码。坑4报表格式的「灵活性」厂长说周报要加一列「与上周对比」月底又说要加上「累计达成率」。如果你每次都在代码里改一周改三次。解决方法报表模板用Excel模板文件管理代码负责填数据格式和指标交给业务人员在模板里设。总结制造业RPA的关键词是「稳定」和「标准化」。MES数据采集要关注数据刷新频率和网络容错报表生成要把模板和逻辑分离。别追求花哨的图表——车间主任要的是准、快、不丢数据。