1. 健康管理PHM系统总体架构设计

任务说明

设计一个面向工业设备的健康管理PHM（Prognostics and Health Management）系统总体架构。本案例聚焦于架构层面的模块划分与数据流定义，明确从底层传感器数据采集到顶层决策支持的完整链路，为后续各模块的详细设计提供框架约束。

系统架构分层设计

本系统采用五层架构，自下而上依次为：感知采集层、数据传输层、数据处理层、分析诊断层、应用展示层。各层之间通过标准化接口交互，数据流单向向上传递，控制指令与配置参数反向下发。

模块职责与交互关系

各层模块职责划分如下：感知采集层负责通过振动、温度、电流等传感器以固定采样率（如振动10kHz、温度1Hz）获取设备运行原始信号，并在边缘侧完成时间戳对齐与初步滤波；数据传输层通过MQTT或OPC UA协议将数据打包上传，并实现断点续传与数据缓存；数据处理层对原始信号执行去噪、缺失值插补、特征提取（均方根值、峰值因子、频谱重心等）与标准化，形成可供算法直接消费的特征向量；分析诊断层基于特征向量完成故障模式识别（如轴承磨损、齿轮断齿）、健康指数（HI）计算与剩余寿命（RUL）回归预测；应用展示层将分析结果以仪表盘、趋势曲线和报警列表形式呈现给运维人员，并支持维护工单下发。

层间交互遵循以下约定：上层模块通过服务接口向下层请求数据，下层模块通过消息总线向上层推送结果；配置参数（如采样率、报警阈值）由上层向下层传递，实现系统运行策略的动态调整。各层之间通过标准化JSON数据格式进行解耦，确保任一层的算法替换不影响其他层功能。

接口定义与数据格式

为支撑上述架构落地，定义核心接口契约如下（以Python数据类形式给出）：

# 接口定义：data_classes.py
from dataclasses import dataclass, field
from typing import List, Dict, Optional
from datetime import datetime

@dataclass
class SensorReading:
    """感知层原始数据单元"""
    device_id: str
    timestamp: datetime
    sensor_type: str          # 'vibration' | 'temperature' | 'current' | 'speed'
    value: float
    unit: str

@dataclass
class FeatureVector:
    """处理层输出的特征向量"""
    device_id: str
    timestamp: datetime
    features: Dict[str, float]  # {'rms': 2.34, 'peak_factor': 3.1, 'temp': 65.2}
    sample_count: int

@dataclass
class HealthStatus:
    """分析层输出的健康评估结果"""
    device_id: str
    timestamp: datetime
    health_index: float        # 0~1，1为健康
    fault_type: Optional[str]  # None表示正常
    rul_hours: Optional[float] # 剩余寿命预测（小时）
    confidence: float          # 置信度 0~1
    alert_level: int           # 0正常 1警告 2严重

@dataclass
class SystemConfig:
    """配置下发数据结构"""
    device_id: str
    sampling_rate_hz: float
    alert_threshold: float
    model_params: Dict[str, float] = field(default_factory=dict)

架构落地与验证

本架构以最小可运行原型验证模块间数据流转。创建以下文件：

data_classes.py — 上述接口定义；
pipeline.py — 模拟从采集到展示的完整数据管道；
test_pipeline.py — 验证各层接口调用链。

# pipeline.py — 架构各层串联的最小实现
from datetime import datetime, timedelta
import random
from data_classes import SensorReading, FeatureVector, HealthStatus, SystemConfig

# ---------- 感知采集层 ----------
def collect_sensor_data(device_id: str, config: SystemConfig) -> List[SensorReading]:
    """模拟采集一个时间窗口内的传感器数据"""
    readings = []
    base_time = datetime.now()
    # 按配置的采样率生成模拟数据（此处为演示，仅生成10个点）
    for i in range(10):
        readings.append(SensorReading(
            device_id=device_id,
            timestamp=base_time + timedelta(milliseconds=i * 1000 / config.sampling_rate_hz),
            sensor_type='vibration',
            value=random.uniform(0.5, 3.0),  # 模拟振动加速度值 (g)
            unit='g'
        ))
    return readings

# ---------- 数据处理层 ----------
def process_features(readings: List[SensorReading]) -> FeatureVector:
    """从原始数据提取特征：RMS与峰值因子"""
    values = [r.value for r in readings]
    mean_val = sum(values) / len(values)
    rms = (sum(v**2 for v in values) / len(values)) ** 0.5
    peak = max(abs(v) for v in values)
    peak_factor = peak / rms if rms > 0 else 0.0
    return FeatureVector(
        device_id=readings[0].device_id,
        timestamp=readings[-1].timestamp,
        features={'rms': round(rms, 3), 'peak_factor': round(peak_factor, 3)},
        sample_count=len(readings)
    )

# ---------- 分析诊断层 ----------
def analyze_health(fv: FeatureVector) -> HealthStatus:
    """基于特征阈值进行健康评估与预警"""
    rms = fv.features['rms']
    # 简单阈值规则：RMS > 2.5 判定为轴承磨损预警
    if rms > 2.5:
        hi = max(0.0, 1.0 - (rms - 2.5) / 5.0)
        return HealthStatus(
            device_id=fv.device_id, timestamp=fv.timestamp,
            health_index=round(hi, 3), fault_type='bearing_wear',
            rul_hours=round(hi * 500, 1), confidence=0.85, alert_level=2
        )
    else:
        return HealthStatus(
            device_id=fv.device_id, timestamp=fv.timestamp,
            health_index=1.0, fault_type=None,
            rul_hours=2000.0, confidence=0.95, alert_level=0
        )

# ---------- 应用展示层 ----------
def display_status(status: HealthStatus) -> str:
    """格式化输出用于前端展示"""
    if status.alert_level == 0:
        return f"[正常] 设备{status.device_id} HI={status.health_index} RUL={status.rul_hours}h"
    else:
        return (f"[告警] 设备{status.device_id} 故障={status.fault_type} "
                f"HI={status.health_index} RUL={status.rul_hours}h 置信度={status.confidence}")

# ---------- 主流程：串联各层 ----------
def run_pipeline(device_id: str = "DEV-001"):
    config = SystemConfig(device_id=device_id, sampling_rate_hz=1000, alert_threshold=2.5)
    # 1. 采集
    raw_data = collect_sensor_data(device_id, config)
    # 2. 处理
    features = process_features(raw_data)
    # 3. 分析
    health = analyze_health(features)
    # 4. 展示
    output = display_status(health)
    return output, health

if __name__ == "__main__":
    # 运行两次演示不同健康状态
    for _ in range(2):
        msg, status = run_pipeline()
        print(msg)

# test_pipeline.py — 架构链路验证
from pipeline import run_pipeline, collect_sensor_data, process_features, analyze_health
from data_classes import SystemConfig

def test_full_pipeline():
    """验证从采集到展示的完整链路可执行"""
    output, status = run_pipeline("TEST-001")
    assert output is not None
    assert status.device_id == "TEST-001"
    assert 0.0 <= status.health_index <= 1.0
    assert status.alert_level in (0, 1, 2)
    print("链路验证通过")

def test_config_driven_sampling():
    """验证配置参数确实影响采集行为"""
    config = SystemConfig(device_id="CFG-001", sampling_rate_hz=500, alert_threshold=2.0)
    readings = collect_sensor_data("CFG-001", config)
    # 采样间隔应为 1/500 秒 = 2ms
    delta_ms = (readings[1].timestamp - readings[0].timestamp).total_seconds() * 1000
    assert abs(delta_ms - 2.0) < 0.01
    print("配置驱动验证通过")

if __name__ == "__main__":
    test_full_pipeline()
    test_config_driven_sampling()

运行与验证方法

在项目目录下执行以下命令验证架构可运行：

# 运行主流程（观察各层输出）
python pipeline.py

# 运行架构链路测试
python test_pipeline.py

判定标准：

pipeline.py 正常输出两行状态信息，每行包含设备ID、健康指数HI与剩余寿命RUL，且HI值在0~1范围内；
test_pipeline.py 中两个测试函数均打印“验证通过”且无异常抛出；
修改SystemConfig中的sampling_rate_hz后重新运行，采集时间戳间隔随之变化，证明配置下发链路生效。

以上验证仅确认架构各层接口贯通与数据流转正确，不涉及具体算法精度评估。实际部署时，各层模块可独立替换为高性能实现（如边缘侧C++采集、云端Spark处理、TensorFlow模型推理），接口契约保持不变。

📚 课程目录