1. 工业设备传感器数据采集与PHM数据管道搭建
任务目标
用 Python 搭建一条最小可复用的 PHM 数据采集管道：从模拟的多源传感器（振动、温度、电流）按固定采样周期读取数据，做时间对齐与标准化，写入本地 CSV 存储，并支持后续按批次读取。完成后应能持续采集、落盘，并验证数据完整性与字段一致性。
环境与依赖
- Python 3.9+，仅用标准库（
csv、math、random、time、datetime、pathlib），无需第三方包。 - 输出目录：
./phm_data，自动创建。
数据管道结构
管道分四步：采集（模拟三路传感器）→ 对齐（统一时间戳）→ 标准化（统一字段与单位）→ 存储（按批次写 CSV）。下图说明数据流与模块关系。
完整实现
import csv, math, random, time
from datetime import datetime, timezone
from pathlib import Path
DATA_DIR = Path("./phm_data")
DATA_DIR.mkdir(exist_ok=True)
FIELDS = ["timestamp", "device_id", "vib_g", "temp_c", "curr_a"]
def read_vibration(t):
# 模拟：基础振动 + 周期性扰动 + 噪声
return round(0.5 + 0.3 * math.sin(2 * math.pi * 5 * t) + random.uniform(-0.05, 0.05), 4)
def read_temperature(t):
# 模拟：缓慢升温 + 噪声
return round(25.0 + 0.01 * t + random.uniform(-0.2, 0.2), 3)
def read_current(t):
# 模拟：负载电流 + 噪声
return round(3.0 + 0.5 * math.sin(2 * math.pi * 0.5 * t) + random.uniform(-0.1, 0.1), 4)
def collect_sample(device_id, t):
"""采集一次多源信号并对齐到同一时间戳。"""
return {
"timestamp": datetime.now(timezone.utc).isoformat(timespec="milliseconds"),
"device_id": device_id,
"vib_g": read_vibration(t),
"temp_c": read_temperature(t),
"curr_a": read_current(t),
}
def write_batch(rows, batch_index):
"""按批次写入 CSV，返回文件路径。"""
path = DATA_DIR / f"batch_{batch_index:04d}.csv"
with path.open("w", newline="", encoding="utf-8") as f:
writer = csv.DictWriter(f, fieldnames=FIELDS)
writer.writeheader()
writer.writerows(rows)
return path
def run_pipeline(device_id="motor_01", samples=50, interval=0.02, batch_size=10):
"""主入口：采集 -> 对齐 -> 标准化 -> 分批存储。"""
buffer, batch_index, written = [], 0, []
for i in range(samples):
buffer.append(collect_sample(device_id, i * interval))
if len(buffer) == batch_size:
written.append(write_batch(buffer, batch_index))
batch_index += 1
buffer = []
time.sleep(interval)
if buffer:
written.append(write_batch(buffer, batch_index))
return written
def load_batch(path):
"""读取一个批次，验证字段一致性。"""
with open(path, newline="", encoding="utf-8") as f:
reader = csv.DictReader(f)
rows = list(reader)
assert reader.fieldnames == FIELDS, f"字段不一致: {reader.fieldnames}"
return rows
if __name__ == "__main__":
files = run_pipeline(samples=50, batch_size=10)
print("生成文件:", [str(p) for p in files])
total = 0
for p in files:
rows = load_batch(p)
total += len(rows)
print(f"{p.name}: {len(rows)} 行, 首行={rows[0]}")
print("总样本数:", total)
运行与验证
- 运行：
python pipeline.py。 - 观察输出：应打印 5 个批次文件（每批 10 行），总样本数为 50。
- 检查
./phm_data下生成的batch_0000.csv等文件，表头应为timestamp,device_id,vib_g,temp_c,curr_a。 - 通过条件：
load_batch的字段断言不报错，且每行三路信号均为数值、时间戳为 ISO 格式。
扩展建议
将 read_* 函数替换为真实采集接口（如串口、Modbus、公开数据集读取）即可复用整条管道；write_batch 可改为追加模式或写入数据库，字段顺序保持不变即可保证下游故障预测模型输入一致。