2026 技术信号：工业设备预测性维护升温，传感器数据与故障预警成为制造业新入口

import json
import random
from datetime import datetime
from collections import defaultdict


class IndustrialDevice:
    def __init__(self, device_id, device_type):
        self.device_id = device_id
        self.device_type = device_type
        self.temperature = 0
        self.vibration = 0
        self.current = 0
        self.rpm = 0
        self.running_hours = 0
        self.status = "normal"
        self.updated_at = datetime.now().isoformat()

    def to_dict(self):
        return {
            "device_id": self.device_id,
            "device_type": self.device_type,
            "temperature": self.temperature,
            "vibration": self.vibration,
            "current": self.current,
            "rpm": self.rpm,
            "running_hours": self.running_hours,
            "status": self.status,
            "updated_at": self.updated_at
        }

def collect_device_sample(device: IndustrialDevice):
    device.temperature = round(
        random.uniform(35, 95),
        2
    )

    device.vibration = round(
        random.uniform(0.2, 8.0),
        2
    )

    device.current = round(
        random.uniform(5, 45),
        2
    )

    device.rpm = random.randint(800, 3500)

    device.running_hours = random.randint(100, 8000)

    if random.random() < 0.08:
        device.status = "warning"
    else:
        device.status = "normal"

    device.updated_at = datetime.now().isoformat()

    return device.to_dict()

def calculate_device_health(record):
    score = 100
    issues = []

    if record["temperature"] > 80:
        score -= 25
        issues.append("设备温度偏高。")

    if record["vibration"] > 6:
        score -= 25
        issues.append("设备振动异常，可能存在轴承或结构问题。")

    if record["current"] > 38:
        score -= 15
        issues.append("设备电流偏高，可能存在负载异常。")

    if record["running_hours"] > 6000:
        score -= 10
        issues.append("设备运行时长较高，建议关注老化风险。")

    if record["status"] != "normal":
        score -= 15
        issues.append("设备状态告警。")

    score = max(score, 0)

    if score >= 85:
        level = "healthy"
    elif score >= 65:
        level = "attention"
    elif score >= 40:
        level = "risk"
    else:
        level = "danger"

    return {
        "device_id": record["device_id"],
        "health_score": score,
        "health_level": level,
        "issues": issues
    }

def infer_possible_fault(record):
    faults = []

    if record["vibration"] > 6:
        faults.append({
            "fault_type": "bearing_wear",
            "message": "振动偏高，可能存在轴承磨损或安装松动。"
        })

    if record["temperature"] > 80 and record["current"] > 35:
        faults.append({
            "fault_type": "overload",
            "message": "温度和电流同时偏高，可能存在过载运行。"
        })

    if record["rpm"] < 1000 and record["current"] > 30:
        faults.append({
            "fault_type": "motor_blocking",
            "message": "转速偏低但电流偏高，可能存在电机阻滞。"
        })

    if record["running_hours"] > 7000:
        faults.append({
            "fault_type": "aging",
            "message": "累计运行时间较长，建议检查易损件。"
        })

    if not faults:
        faults.append({
            "fault_type": "normal",
            "message": "未发现明显故障特征。"
        })

    return faults

def generate_maintenance_plan(health, faults):
    level = health["health_level"]

    if level == "danger":
        action = "stop_and_inspect"
        message = "设备风险较高，建议停机检查。"
    elif level == "risk":
        action = "schedule_maintenance"
        message = "设备存在明显风险，建议安排近期维护。"
    elif level == "attention":
        action = "increase_monitoring"
        message = "设备状态需要关注，建议提高监测频率。"
    else:
        action = "normal_operation"
        message = "设备状态良好，可继续运行。"

    return {
        "device_id": health["device_id"],
        "action": 31225.t.kuaisou.com 
        "message": message,
        "faults": faults
    }

def run_predictive_maintenance():
    devices = [
        IndustrialDevice("dev_001", "motor"),
        IndustrialDevice("dev_002", "pump"),
        IndustrialDevice("dev_003", "compressor"),
        IndustrialDevice("dev_004", "fan")
    ]

    records = []
    health_results = []
    maintenance_plans = []

    for device in devices:
        record = collect_device_sample(device)
        health = calculate_device_health(record)
        faults = infer_possible_fault(record)
        plan = generate_maintenance_plan(
            health,
            faults
        )

        records.append(record)
        health_results.append(health)
        maintenance_plans.append(plan)

    level_count = defaultdict(int)

    for health in health_results:
        level_count[health["health_level"]] += 1

    report = {
        "report_name": "工业设备预测性维护报告",
        "device_records": 31224.t.kuaisou.com 
        "health_results": health_results,
        "level_count": dict(level_count),
        "maintenance_plans": maintenance_plans,
        "generate_time": datetime.now().isoformat()
    }

    return report


if __name__ == "__main__":
    report = run_predictive_maintenance()

    print(json.dumps(
        report,
        ensure_ascii=False,
        indent=2
    ))