2. 传感器数据模拟与采集

任务说明

本案例将构建一个轻量级传感器数据模拟与采集系统：模拟生成温度、湿度、压力三类传感器的时间序列数据，通过统一采集接口读取，并存储为CSV文件与JSON文件，便于后续数字孪生场景回放或分析。

核心实现

import csv
import json
import random
import time
from datetime import datetime, timedelta

class SensorSimulator:
    """模拟三类传感器，生成带噪声的物理量数据"""
    def __init__(self, seed=42):
        random.seed(seed)
        self.base_temp = 25.0
        self.base_humidity = 60.0
        self.base_pressure = 1013.0

    def read_temp(self):
        # 温度：基础值 + 缓慢漂移 + 随机噪声
        self.base_temp += random.uniform(-0.1, 0.1)
        return round(self.base_temp + random.gauss(0, 0.3), 2)

    def read_humidity(self):
        # 湿度：基础值 + 随机噪声
        self.base_humidity += random.uniform(-0.5, 0.5)
        self.base_humidity = max(0, min(100, self.base_humidity))
        return round(self.base_humidity + random.gauss(0, 0.8), 2)

    def read_pressure(self):
        # 压力：基础值 + 随机噪声
        self.base_pressure += random.uniform(-0.3, 0.3)
        return round(self.base_pressure + random.gauss(0, 0.5), 2)

class DataCollector:
    """按固定间隔采集所有传感器数据并存储"""
    def __init__(self, simulator, interval_sec=1.0):
        self.sim = simulator
        self.interval = interval_sec
        self.records = []

    def collect_once(self, timestamp=None):
        """采集一次所有传感器数据"""
        ts = timestamp or datetime.now()
        record = {
            "timestamp": ts.strftime("%Y-%m-%d %H:%M:%S.%f")[:-3],
            "temperature": self.sim.read_temp(),
            "humidity": self.sim.read_humidity(),
            "pressure": self.sim.read_pressure()
        }
        self.records.append(record)
        return record

    def collect_for_duration(self, seconds):
        """持续采集指定时长（秒）"""
        start = datetime.now()
        while (datetime.now() - start).total_seconds() < seconds:
            self.collect_once()
            time.sleep(self.interval)

    def save_csv(self, filename):
        """存储为CSV文件"""
        if not self.records:
            return
        with open(filename, "w", newline="", encoding="utf-8") as f:
            writer = csv.DictWriter(f, fieldnames=list(self.records[0].keys()))
            writer.writeheader()
            writer.writerows(self.records)

    def save_json(self, filename):
        """存储为JSON文件（含元信息）"""
        payload = {
            "sensor_count": 3,
            "sensor_types": ["temperature", "humidity", "pressure"],
            "sample_count": len(self.records),
            "data": self.records
        }
        with open(filename, "w", encoding="utf-8") as f:
            json.dump(payload, f, indent=2, ensure_ascii=False)

if __name__ == "__main__":
    # 使用固定种子保证可复现性
    sim = SensorSimulator(seed=2024)
    collector = DataCollector(sim, interval_sec=0.2)

    # 采集5秒数据（约25个样本）
    collector.collect_for_duration(seconds=5)

    # 存储结果
    collector.save_csv("sensor_data.csv")
    collector.save_json("sensor_data.json")

    # 打印前3条记录预览
    print("采集完成，共", len(collector.records), "条记录")
    for r in collector.records[:3]:
        print(r)

数据采集时序

运行与验证

运行方式：将代码保存为 sensor_simulator.py，在Python 3.8+环境执行 python sensor_simulator.py。程序会采集约5秒数据（间隔0.2秒，约25条记录），生成两个输出文件。

验证步骤：

1. 观察控制台输出：应显示“采集完成，共N条记录”（N约为25），并打印前3条记录，每条包含时间戳、温度、湿度、压力四个字段。
2. 检查当前目录下是否生成 sensor_data.csv 与 sensor_data.json 两个文件。
3. 打开CSV文件，确认首行为表头 timestamp,temperature,humidity,pressure，数据行数与控制台报告一致。
4. 打开JSON文件，确认包含 sensor_count、sensor_types、sample_count 元信息及完整数据数组。

判定标准：CSV与JSON中记录数一致且均大于0；温度值在20~30℃范围、湿度在40%~80%范围、压力在1000~1025 hPa范围（因随机漂移可能轻微越界，属正常现象）；时间戳按采集间隔递增。