3、数据采集入门：使用Python读取模拟传感器数据（CSV/JSON格式）

各位同学好，我是你们的老朋友。今天咱们聊聊数据采集这个环节。

说实话，我在水质监测这行干了快十年，见过太多因为数据读取出错导致后续分析全盘皆输的案例。你想想看，传感器数据都读不对，后面的异常检测、报警还有什么意义？所以这一章，我带你从最基础的CSV和JSON格式入手，把数据采集这个基本功练扎实。

3.1 为什么是CSV和JSON？

很多刚入行的朋友会问我：「老师，为什么不用数据库？用Excel不也行吗？」

我的回答是：在实际的水质监测现场，传感器设备往往资源有限。它们输出的数据格式，要么是CSV，要么是JSON。CSV简单直接，一行一条记录；JSON结构灵活，适合嵌套数据。我当年在太湖做项目时，那批老式传感器只支持CSV输出，后来换了新设备才支持JSON。说白了，这两种格式就是水质监测数据采集的「通用语言」。

核心要点：

  • CSV：适合表格型数据，每行一个采样点，每列一个指标
  • JSON：适合带元数据、时间戳、设备ID等复杂结构的数据

3.2 读取CSV格式的模拟传感器数据

先来看CSV。假设我们有一个模拟的pH传感器数据文件 ph_sensor.csv，内容大概是这样的：

timestamp,ph_value,temperature,device_id
2024-01-15 08:00:00,7.2,22.5,SENSOR_001
2024-01-15 08:05:00,7.3,22.6,SENSOR_001
2024-01-15 08:10:00,7.1,22.4,SENSOR_001
2024-01-15 08:15:00,7.4,22.7,SENSOR_001

用Python读取它，我推荐用 pandas 库。为什么？因为pandas处理时间序列数据太方便了。我在项目中遇到过用纯Python的 csv 模块逐行解析，结果遇到缺失值、数据类型转换时，代码写得又臭又长。后来全换成pandas，效率翻倍。

import pandas as pd

# 读取CSV文件
df = pd.read_csv('ph_sensor.csv')

# 看一眼数据长什么样
print(df.head())

# 把时间列转成datetime类型，方便后续分析
df['timestamp'] = pd.to_datetime(df['timestamp'])

# 设置时间为索引
df.set_index('timestamp', inplace=True)

print(df.info())

我的小技巧： 读取CSV时，记得加上 parse_dates=['timestamp'] 参数，pandas会自动帮你解析时间列，省去手动转换的麻烦。

3.3 读取JSON格式的模拟传感器数据

JSON格式稍微复杂一点，但更灵活。比如一个多参数水质监测仪，可能同时输出pH、溶解氧、浊度等多个指标，还附带设备状态信息：

{
  "device_id": "SENSOR_002",
  "timestamp": "2024-01-15 08:00:00",
  "measurements": {
    "ph": 7.2,
    "do": 6.8,
    "turbidity": 3.5
  },
  "status": "normal"
}

读取JSON，我习惯用 json 模块先解析，再转成DataFrame。嗯，这里要注意：JSON的嵌套结构需要展平，否则pandas读不懂。

import json
import pandas as pd

# 读取JSON文件
with open('multi_sensor.json', 'r') as f:
    data = json.load(f)

# 如果是单条记录，直接展平
flat_data = {
    'device_id': data['device_id'],
    'timestamp': data['timestamp'],
    'ph': data['measurements']['ph'],
    'do': data['measurements']['do'],
    'turbidity': data['measurements']['turbidity'],
    'status': data['status']
}

df = pd.DataFrame([flat_data])
print(df)

但实际项目中，JSON文件往往包含多条记录（比如一个数组）。我曾经在长江某断面项目中，一天要处理上千条JSON记录。那时候我写了个通用函数，专门处理这种批量JSON数据：

def parse_sensor_json(file_path):
    with open(file_path, 'r') as f:
        records = json.load(f)  # 假设是个列表
    
    flat_records = []
    for record in records:
        flat = {
            'device_id': record['device_id'],
            'timestamp': record['timestamp'],
            'ph': record['measurements']['ph'],
            'do': record['measurements']['do'],
            'turbidity': record['measurements']['turbidity'],
            'status': record['status']
        }
        flat_records.append(flat)
    
    return pd.DataFrame(flat_records)

df = parse_sensor_json('multi_sensor_batch.json')
print(df.head())

避坑指南： 我曾经遇到过JSON文件中某个字段缺失，导致整个解析报错。建议在展平时使用 .get() 方法，并设置默认值，比如 record.get('measurements', {}).get('ph', None)。这样即使数据不完整，程序也不会崩溃。

3.4 实战：模拟传感器数据流

光说不练假把式。咱们模拟一个真实场景：假设你正在调试一套水质监测系统，传感器每5分钟上报一次数据。你需要写一个脚本，持续读取CSV文件，并实时打印最新数据。

import pandas as pd
import time

def read_latest_sensor_data(file_path, last_position=0):
    """读取CSV文件中新增的行"""
    with open(file_path, 'r') as f:
        f.seek(last_position)
        new_lines = f.readlines()
        last_position = f.tell()
    
    if not new_lines:
        return None, last_position
    
    # 跳过表头
    if last_position == 0:
        new_lines = new_lines[1:]
    
    if not new_lines:
        return None, last_position
    
    # 解析新增行
    data = []
    for line in new_lines:
        parts = line.strip().split(',')
        if len(parts) == 4:
            data.append({
                'timestamp': parts[0],
                'ph_value': float(parts[1]),
                'temperature': float(parts[2]),
                'device_id': parts[3]
            })
    
    return pd.DataFrame(data), last_position

# 模拟持续读取
last_pos = 0
while True:
    df_new, last_pos = read_latest_sensor_data('ph_sensor.csv', last_pos)
    if df_new is not None and not df_new.empty:
        print(f"新数据到达: {df_new['timestamp'].values[0]} - pH: {df_new['ph_value'].values[0]}")
    time.sleep(10)  # 每10秒检查一次

这个脚本我在多个项目中用过，比如在太湖蓝藻预警系统中，就是靠这种「增量读取」的方式，实时捕获传感器数据，然后触发后续的异常检测流程。

3.5 数据质量检查：读进来只是第一步

数据读进来了，不代表就能直接用。我见过太多人，数据读进来就急着做分析，结果发现pH值出现-999这种异常值，或者时间戳格式不统一。所以，每次读取完数据，我建议你立刻做三件事：

  1. 检查缺失值：df.isnull().sum()
  2. 检查数据类型：df.dtypes，确保数值列是float/int，时间列是datetime
  3. 检查取值范围：比如pH值应该在0-14之间，超出这个范围的数据要标记出来
# 快速数据质量检查
def quick_data_check(df):
    print("=== 数据质量报告 ===")
    print(f"总记录数: {len(df)}")
    print(f"缺失值统计:\n{df.isnull().sum()}")
    print(f"数据类型:\n{df.dtypes}")
    
    # pH值范围检查
    if 'ph_value' in df.columns:
        invalid_ph = df[(df['ph_value'] < 0) | (df['ph_value'] > 14)]
        if len(invalid_ph) > 0:
            print(f"警告: 发现 {len(invalid_ph)} 条pH值异常数据")
    
    print("===================")

quick_data_check(df)

我的习惯： 每次读取数据后，我都会把原始数据备份一份，然后才做清洗。这样万一处理错了，还能找回原始数据。你想想看，要是把原始数据覆盖了，那可就真抓瞎了。

3.6 小结

这一章咱们聊了怎么用Python读取CSV和JSON格式的模拟传感器数据。说白了，核心就三点：

  • CSV用pandas的 read_csv()，记得处理时间列
  • JSON用 json.load() 解析，然后展平嵌套结构
  • 读取后立刻做数据质量检查，别等分析时才发现问题

下一章，咱们会把这些数据用起来，做真正的异常检测。到时候你会发现，今天打下的数据采集基础，有多重要。

好，今天就到这儿。有问题随时找我，咱们下节课见。

← 上一章 📖 返回目录 下一章 →