1. 传感器数据采集与存储

任务目标

本案例实现一个模拟传感器数据采集程序：以固定采样周期生成振动（加速度）与温度数据，并将每条记录实时追加写入CSV文件。程序支持设置采样时长与采样频率，运行结束后输出文件保存路径与总记录数，用于验证采集与存储流程的完整性。

实现方案

使用Python标准库csv与time完成数据写入和时间控制，不依赖第三方库。模拟振动数据采用正弦波叠加随机噪声，温度数据采用缓慢漂移的随机游走模型，以贴近真实传感器特征。程序通过命令行参数控制采样时长与采样频率，便于不同场景下复用。

完整代码

import csv
import math
import random
import time
from datetime import datetime
from pathlib import Path

def generate_vibration(t: float, base_freq: float = 50.0) -> float:
    """模拟振动加速度信号(m/s²)：50Hz主频 + 谐波 + 噪声"""
    signal = 2.0 * math.sin(2 * math.pi * base_freq * t)
    signal += 0.8 * math.sin(2 * math.pi * base_freq * 2 * t)
    signal += random.gauss(0, 0.3)
    return round(signal, 4)

def generate_temperature(last_temp: float) -> float:
    """模拟温度(℃)：随机游走，限制在合理工业范围内"""
    new_temp = last_temp + random.gauss(0, 0.05)
    new_temp = max(20.0, min(90.0, new_temp))
    return round(new_temp, 2)

def run_acquisition(duration_s: float, sample_rate_hz: float, output_path: str) -> None:
    """按指定时长和采样率采集数据并写入CSV"""
    interval = 1.0 / sample_rate_hz
    total_samples = int(duration_s * sample_rate_hz)
    file_path = Path(output_path)
    file_path.parent.mkdir(parents=True, exist_ok=True)

    temp = 45.0  # 初始温度
    start_time = time.time()
    next_sample_time = start_time

    with open(file_path, 'w', newline='', encoding='utf-8') as f:
        writer = csv.writer(f)
        writer.writerow(['timestamp', 'vibration_mps2', 'temperature_c'])

        for i in range(total_samples):
            # 等待至下一个采样时刻，保证采样率准确
            now = time.time()
            if now < next_sample_time:
                time.sleep(next_sample_time - now)

            t = i * interval  # 逻辑时间轴
            vib = generate_vibration(t)
            temp = generate_temperature(temp)
            ts = datetime.now().isoformat(timespec='milliseconds')

            writer.writerow([ts, vib, temp])
            next_sample_time = start_time + (i + 1) * interval

    print(f"采集完成，共 {total_samples} 条记录")
    print(f"数据已保存至: {file_path.resolve()}")

if __name__ == "__main__":
    import argparse
    parser = argparse.ArgumentParser(description="模拟传感器数据采集")
    parser.add_argument("--duration", type=float, default=5.0, help="采样时长（秒）")
    parser.add_argument("--rate", type=float, default=10.0, help="采样频率（Hz）")
    parser.add_argument("--output", type=str, default="sensor_data.csv", help="输出CSV文件路径")
    args = parser.parse_args()

    run_acquisition(args.duration, args.rate, args.output)

运行方式

# 基本运行：采集5秒，10Hz采样率，输出到默认文件
python sensor_acquisition.py

# 自定义参数：采集30秒，100Hz采样率，指定输出路径
python sensor_acquisition.py --duration 30 --rate 100 --output data/run1.csv

验证方法

1. 运行上述命令，程序应输出“采集完成，共N条记录”及文件保存路径。
2. 打开生成的CSV文件，确认包含表头timestamp,vibration_mps2,temperature_c，且数据行数等于duration × rate。
3. 检查数据合理性：振动值大致在-3到3 m/s²范围内波动；温度值在20~90℃之间缓慢变化，相邻记录温差通常小于0.5℃。
4. 用--duration 2 --rate 5快速测试，应生成10条记录，验证参数控制生效。

关键说明

• 时间戳使用系统当前时间，采样间隔由time.sleep保证，实际运行时间约等于设定的采样时长。
• 振动信号叠加了随机噪声，因此每次运行生成的数据不完全相同，但统计特征一致，属正常现象。
• 若需接入真实传感器，只需将generate_vibration和generate_temperature函数体替换为对应硬件读取代码，其余存储逻辑无需改动。