2. 设备传感器数据模拟器
任务说明
本案例需要构建一个设备传感器数据模拟器，用于生成工业设备（以旋转机械为例）的振动、温度和转速三类传感器的时间序列数据。该模拟器需支持配置采样频率、设备健康状态（正常/退化/故障），并输出带时间戳的结构化数据，为后续PHM数据管道提供稳定、可重复的测试数据源。
实现方案
采用Python实现一个类SensorDataSimulator，核心逻辑包括：
- 基于
numpy生成确定性随机序列（固定随机种子保证可复现）； - 振动信号叠加正弦基频与高斯噪声，退化/故障状态额外引入幅值增长与冲击分量；
- 温度信号采用缓慢漂移模型，故障时叠加阶跃偏移；
- 转速信号在正常状态保持恒定，退化状态线性缓降，故障状态加入随机波动；
- 输出为
pandas.DataFrame，包含timestamp、vibration、temperature、rpm四列。
代码实现
import numpy as np
import pandas as pd
from datetime import datetime, timedelta
class SensorDataSimulator:
def __init__(self, sample_rate_hz=100, duration_sec=10, seed=42):
self.sample_rate = sample_rate_hz
self.duration = duration_sec
self.seed = seed
self.n_samples = sample_rate_hz * duration_sec
self.rng = np.random.default_rng(seed)
def _generate_vibration(self, health_state):
t = np.arange(self.n_samples) / self.sample_rate
base_freq = 30.0 # Hz
if health_state == 'normal':
amp = 1.0
noise = 0.1
signal = amp * np.sin(2 * np.pi * base_freq * t)
elif health_state == 'degraded':
amp = 1.5
noise = 0.2
# 幅值缓增 + 2倍频分量
trend = np.linspace(1.0, 1.8, self.n_samples)
signal = amp * trend * np.sin(2 * np.pi * base_freq * t) \
+ 0.5 * trend * np.sin(2 * np.pi * 2 * base_freq * t)
else: # fault
amp = 2.0
noise = 0.3
# 冲击分量：每0.5秒一次衰减振荡
impact = np.zeros_like(t)
impact_interval = 0.5
for start in np.arange(0, self.duration, impact_interval):
idx = int(start * self.sample_rate)
if idx < self.n_samples:
impact_len = min(int(0.05 * self.sample_rate), self.n_samples - idx)
tt = np.arange(impact_len) / self.sample_rate
impact[idx:idx+impact_len] = 3.0 * np.exp(-tt * 50) * np.sin(2 * np.pi * 200 * tt)
signal = amp * np.sin(2 * np.pi * base_freq * t) + impact
noise_arr = self.rng.normal(0, noise, self.n_samples)
return signal + noise_arr
def _generate_temperature(self, health_state):
base_temp = 65.0 # 摄氏度
drift = np.linspace(0, 2, self.n_samples) # 缓慢上升
noise = self.rng.normal(0, 0.3, self.n_samples)
temp = base_temp + drift + noise
if health_state == 'degraded':
temp += 5.0
elif health_state == 'fault':
temp += 12.0
return temp
def _generate_rpm(self, health_state):
base_rpm = 1500.0
if health_state == 'normal':
rpm = np.full(self.n_samples, base_rpm)
elif health_state == 'degraded':
rpm = np.linspace(base_rpm, base_rpm * 0.9, self.n_samples)
else: # fault
rpm = base_rpm + self.rng.normal(0, 30, self.n_samples)
return rpm
def generate(self, health_state='normal', start_time=None):
if health_state not in ('normal', 'degraded', 'fault'):
raise ValueError("health_state must be 'normal', 'degraded', or 'fault'")
if start_time is None:
start_time = datetime.now().replace(microsecond=0)
timestamps = [start_time + timedelta(seconds=i / self.sample_rate)
for i in range(self.n_samples)]
df = pd.DataFrame({
'timestamp': timestamps,
'vibration': self._generate_vibration(health_state),
'temperature': self._generate_temperature(health_state),
'rpm': self._generate_rpm(health_state)
})
return df
# 使用示例
if __name__ == '__main__':
sim = SensorDataSimulator(sample_rate_hz=100, duration_sec=5, seed=42)
df_normal = sim.generate('normal')
df_fault = sim.generate('fault')
print("正常状态数据形状:", df_normal.shape)
print("故障状态数据形状:", df_fault.shape)
print("\n正常状态前3行:")
print(df_normal.head(3).to_string(index=False))
print("\n故障状态前3行:")
print(df_fault.head(3).to_string(index=False))落地与验证
运行方式：将代码保存为sensor_simulator.py，在Python 3.8+环境中执行python sensor_simulator.py。依赖库为numpy和pandas，可通过pip install numpy pandas安装。
验证方法：
- 程序正常退出且无异常抛出；
- 输出DataFrame行数等于
sample_rate * duration（本例为500行）； - 正常状态振动幅值约在±1.5范围内，故障状态振动幅值明显增大且包含冲击尖峰；
- 正常状态温度约65~67°C，故障状态温度约77°C以上；
- 正常状态转速恒为1500，故障状态转速在1500±3σ范围内波动。
判定标准：满足上述5项观测条件即视为模拟器功能正确。由于固定了随机种子，多次运行结果完全一致，保证数据可复现性。