影刀RPA 工业设备巡检：传感器数据自动记录与异常预警

作者：林焱 | 分类：影刀RPA新手教程 | 难度：★★★

什么情况用

工厂设备巡检目前很多还是纸质表格：巡检员拿着本子走到每台设备前，看温度表、压力表、振动值，手写记录。回来后录入电脑做报表。不仅效率低，而且数据滞后——上午发现异常可能下午才报给维修部门。

用影刀RPA从工业物联网平台（如树根互联、海尔COSMOPlat、阿里云IoT）自动拉取传感器数据，定时生成设备健康报告，异常自动推送。

怎么做

第一步：从IoT平台获取设备数据

# 影刀Python节点：从IoT平台API获取数据importrequestsfromdatetimeimportdatetime,timedelta IOT_PLATFORM={"api_url":"https://iot.yourplatform.com/api/v1","api_key":"your_api_key",# 从影刀凭证管理读取}defget_device_data(device_id,metric,hours=24):"""从IoT平台获取设备传感器历史数据"""url=f"{IOT_PLATFORM['api_url']}/devices/{device_id}/telemetry"headers={"Authorization":f"Bearer{IOT_PLATFORM['api_key']}"}end_time=datetime.now()start_time=end_time-timedelta(hours=hours)params={"metric":metric,# temperature, vibration, pressure 等"start":start_time.isoformat(),"end":end_time.isoformat(),"interval":"5m",# 5分钟采样间隔![在这里插入图片描述](https://i-blog.csdnimg.cn/direct/1784da20d26f4eb99c4b278029fca001.png#pic_center)}resp=requests.get(url,headers=headers,params=params)data=resp.json()returndata.get("values",[])# 获取关键设备数据devices_to_monitor=[{"id":"PUMP-A001","name":"主循环泵A","metrics":["temperature","vibration","pressure"]},{"id":"MOTOR-B001","name":"主电机B","metrics":["temperature","rpm","current"]},{"id":"COMP-C001","name":"空压机C","metrics":["temperature","pressure","runtime"]},]fordeviceindevices_to_monitor:formetricindevice["metrics"]:values=get_device_data(device["id"],metric,hours=1)ifvalues:latest=values[-1]print(f"{device['name']}-{metric}:{latest['value']}（{latest['timestamp']}）")

第二步：定义告警阈值

# 影刀Python节点：异常检测ALERT_RULES={"PUMP-A001":{"temperature":{"min":10,"max":80,"warning":70,"unit":"°C"},"vibration":{"min":0,"max":7.0,"warning":5.0,"unit":"mm/s"},"pressure":{"min":0.1,"max":2.5,"warning":2.2,"unit":"MPa"},},"MOTOR-B001":{"temperature":{"min":0,"max":95,"warning":85,"unit":"°C"},"rpm":{"min":100,"max":3000,"warning":2950,"unit":"RPM"},"current":{"min":0,"max":150,"warning":135,"unit":"A"},},}defcheck_alerts(device_id,metric,current_value):"""检查传感器值是否触发告警"""rules=ALERT_RULES.get(device_id,{}).get(metric)ifnotrules:returnNoneifcurrent_value>rules.get("max")orcurrent_value<rules.get("min"):return{"level":"🔴 严重","message":f"{metric}={current_value}{rules['unit']}，超出正常范围[{rules['min']}-{rules['max']}]",}elifcurrent_value>rules.get("warning")or(rules.get("min")andcurrent_value<rules.get("min")+5):return{"level":"🟡 预警","message":f"{metric}={current_value}{rules['unit']}，接近阈值{rules.get('warning')}",}returnNone

第三步：趋势分析与预测

# 影刀Python节点：异常趋势检测importnumpyasnpdefdetect_trend(values,window=12):"""检测是否呈上升趋势（可能预示故障）"""iflen(values)<window:returnNonerecent=values[-window:]# 简单线性回归判断趋势x=np.arange(len(recent))y=np.array([v["value"]forvinrecent])slope=np.polyfit(x,y,1)[0]mean_val=np.mean(y)# 趋势强度（斜率/均值）![在这里插入图片描述](https://i-blog.csdnimg.cn/direct/56ee6ca259134f1c84485609d37c06f6.png#pic_center)trend_strength=slope*window/mean_valiftrend_strength>0.3:returnf"📈 持续上升（7天趋势值：{trend_strength:.2f}），建议提前检查"eliftrend_strength<-0.3:returnf"📉 持续下降（7天趋势值：{trend_strength:.2f}），可能存在泄漏"returnNone# 对每个设备的每个指标做趋势分析fordeviceindevices_to_monitor:formetricindevice["metrics"]:values=get_device_data(device["id"],metric,hours=168)# 一周数据trend=detect_trend(values)iftrend:print(f"{device['name']}-{metric}:{trend}")

第四步：生成巡检日报

defgenerate_inspection_report(device_data,alerts,trends):"""生成设备巡检日报"""report=f"## 设备巡检日报 ({datetime.now().strftime('%Y-%m-%d')})\n\n"# 告警汇总report+="### 🚨 告警汇总\n"ifalerts:foralertinalerts:report+=f"-{alert['level']}{alert['device']}:{alert['message']}\n"else:report+="- ✅ 所有设备运行正常\n"# 趋势预警report+="\n### 📊 趋势分析\n"iftrends:fortrendintrends:report+=f"-{trend}\n"# 各设备状态表report+="\n### 📋 设备状态一览\n"report+="| 设备 | 温度 | 压力 | 振动 | 状态 |\n"report+="|------|------|------|------|------|\n"fordeviceindevice_data:report+=f"|{device['name']}|{device.get('temperature','N/A')}|{device.get('pressure','N/A')}|{device.get('vibration','N/A')}|{device.get('status','N/A')}|\n"returnreport

影刀【企微推送】给维修班组，【邮件发送】给设备经理。

第五步：自动生成工单

当检测到严重告警时，自动在维修系统创建工单：

1. 【判断】→ 告警等级 = “严重”
2. 【Python节点】→ 生成工单内容（设备编号、异常指标、建议处理方式）
3. 【打开网页】→ 维修管理系统
4. 【填写】→ 工单信息
5. 【点击】→ 「提交」

有什么坑

坑1：不同IoT平台API差异很大

树根互联、阿里云IoT、华为IoT的API格式完全不同。需要为每个平台单独写适配代码。建议封装统一的"数据获取接口"，上层业务逻辑不变，只换底层适配。

坑2：告警阈值难以一次设准

阈值设太低→天天误报告警疲劳，设太高→真有问题发现不了。建议先跑一个月数据，用"均值±3倍标准差"作为初始阈值，运行一段时间后让设备工程师微调。

坑3：传感器本身可能故障

传感器读数突然变成0或99999，不一定是设备坏了，可能是传感器坏了。RPA无法区分，但可以通过规则辅助判断：如果读数跳变幅度超过物理可能（比如温度1秒内从50°C跳到500°C），大概率是传感器故障而非设备故障。

坑4：数据采集延迟

IoT平台的数据可能有1-5分钟延迟。如果是实时性要求高的场景（如反应釜温控），延迟可能导致反应不及时。了解清楚数据链路延迟再设定检查频率。

坑5：设备停机期间的数据

设备停机期间温度下降、压力归零是正常的，不应触发"温度过低""压力为零"告警。需要在告警逻辑中增加"设备是否在运行"的状态判断。

总结：工业设备巡检自动化的价值在于从定时巡检转变为状态监测——不用等巡检员发现异常，而是系统主动告诉你哪台设备"不对劲"，把事后维修变成事前预防。