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

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

什么情况用

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

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

怎么做

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

# 影刀Python节点：从IoT平台API获取数据
import requests
from datetime import datetime, timedelta

IOT_PLATFORM = {
    "api_url": "https://iot.yourplatform.com/api/v1",
    "api_key": "your_api_key",  # 从影刀凭证管理读取
}

def get_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()
    
    return data.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"]},
]

for device in devices_to_monitor:
    for metric in device["metrics"]:
        values = get_device_data(device["id"], metric, hours=1)
        if values:
            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"},
    },
}

def check_alerts(device_id, metric, current_value):
    """检查传感器值是否触发告警"""
    rules = ALERT_RULES.get(device_id, {}).get(metric)
    if not rules:
        return None
    
    if current_value > rules.get("max") or current_value < rules.get("min"):
        return {
            "level": "🔴 严重",
            "message": f"{metric} = {current_value}{rules['unit']}，超出正常范围[{rules['min']}-{rules['max']}]",
        }
    elif current_value > rules.get("warning") or (rules.get("min") and current_value < rules.get("min") + 5):
        return {
            "level": "🟡 预警",
            "message": f"{metric} = {current_value}{rules['unit']}，接近阈值{rules.get('warning')}",
        }
    
    return None

第三步：趋势分析与预测

# 影刀Python节点：异常趋势检测
import numpy as np

def detect_trend(values, window=12):
    """检测是否呈上升趋势（可能预示故障）"""
    if len(values) < window:
        return None
    
    recent = values[-window:]
    
    # 简单线性回归判断趋势
    x = np.arange(len(recent))
    y = np.array([v["value"] for v in recent])
    
    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_val
    
    if trend_strength > 0.3:
        return f"📈 持续上升（7天趋势值：{trend_strength:.2f}），建议提前检查"
    elif trend_strength < -0.3:
        return f"📉 持续下降（7天趋势值：{trend_strength:.2f}），可能存在泄漏"
    
    return None

# 对每个设备的每个指标做趋势分析
for device in devices_to_monitor:
    for metric in device["metrics"]:
        values = get_device_data(device["id"], metric, hours=168)  # 一周数据
        trend = detect_trend(values)
        if trend:
            print(f"{device['name']} - {metric}: {trend}")

第四步：生成巡检日报

def generate_inspection_report(device_data, alerts, trends):
    """生成设备巡检日报"""
    report = f"## 设备巡检日报 ({datetime.now().strftime('%Y-%m-%d')})\n\n"
    
    # 告警汇总
    report += "### 🚨 告警汇总\n"
    if alerts:
        for alert in alerts:
            report += f"- {alert['level']} {alert['device']}: {alert['message']}\n"
    else:
        report += "- ✅ 所有设备运行正常\n"
    
    # 趋势预警
    report += "\n### 📊 趋势分析\n"
    if trends:
        for trend in trends:
            report += f"- {trend}\n"
    
    # 各设备状态表
    report += "\n### 📋 设备状态一览\n"
    report += "| 设备 | 温度 | 压力 | 振动 | 状态 |\n"
    report += "|------|------|------|------|------|\n"
    for device in device_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"
    
    return report

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

第五步：自动生成工单

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

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：设备停机期间的数据

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

---

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