工业一体机是面向工业环境设计的集成化计算终端，集成了处理器、显示屏和工业级接口（如RS232/RS485/CAN），支持在恶劣环境下长时间稳定运行。在选购工业一体机后，第一步往往是搭建数据采集系统——从传感器或PLC读取数据，处理后写入数据库，为后续监控和分析提供数据基础。本文用Python实现一套从串口读取到时序数据库写入的完整数据采集流程。

## 系统架构设计

数据采集系统的基本架构：传感器/PLC → 串口通信（RS485/Modbus） → Python采集程序 → 数据清洗/转换 → InfluxDB时序数据库 → Grafana可视化。

| 组件 | 技术选型 | 作用 |
|------|---------|------|
| 通信层 | pyserial + Modbus-RTU | 从PLC/传感器读取寄存器数据 |
| 处理层 | Python + pandas | 数据清洗、单位换算、阈值判断 |
| 存储层 | InfluxDB 2.x | 时序数据高效写入和查询 |
| 可视化 | Grafana | 实时仪表盘和历史趋势图 |
| 调度层 | APScheduler | 定时采集任务管理 |

## 串口数据采集实现

工业现场最常用的通信协议是Modbus-RTU，走RS485串口。先装依赖：

```bash
pip install pyserial pymodbus influxdb-client pandas apscheduler
```

采集模块的核心代码：

```python
import serial
from pymodbus.client import ModbusSerialClient
from datetime import datetime
import struct

class ModbusCollector:
    def __init__(self, port='/dev/ttyUSB0', baudrate=9600,
                 slave_id=1, timeout=1):
        self.client = ModbusSerialClient(
            port=port,
            baudrate=baudrate,
            bytesize=8,
            parity='N',
            stopbits=1,
            timeout=timeout
        )
        self.slave_id = slave_id

    def read_holding_registers(self, start_addr, count):
        """读取保持寄存器"""
        if not self.client.connect():
            raise ConnectionError("串口连接失败，检查端口和波特率")
        try:
            result = self.client.read_holding_registers(
                address=start_addr,
                count=count,
                slave=self.slave_id
            )
            if result.isError():
                raise Exception(f"Modbus通信错误: {result}")
            return result.registers
        finally:
            self.client.close()

    def read_float(self, start_addr):
        """读取32位浮点数（占2个寄存器）"""
        regs = self.read_holding_registers(start_addr, 2)
        # Modbus浮点数字节序：大端
        float_bytes = struct.pack('>HH', regs[0], regs[1])
        return struct.unpack('>f', float_bytes)[0]

    def read_all_sensors(self, register_map):
        """按寄存器映射表批量读取"""
        data = {}
        for name, (addr, reg_type) in register_map.items():
            if reg_type == 'float':
                data[name] = self.read_float(addr)
            elif reg_type == 'int16':
                regs = self.read_holding_registers(addr, 1)
                val = regs[0]
                # 处理有符号整数
                data[name] = val - 65536 if val > 32767 else val
            elif reg_type == 'uint16':
                regs = self.read_holding_registers(addr, 1)
                data[name] = regs[0]
        data['timestamp'] = datetime.utcnow()
        return data
```

使用时定义寄存器映射表：

```python
# 寄存器地址映射：{参数名: (起始地址, 数据类型)}
register_map = {
    'temperature': (0, 'float'),     # 温度 °C
    'pressure':    (2, 'float'),     # 压力 kPa
    'flow_rate':   (4, 'float'),     # 流量 L/min
    'motor_speed': (6, 'uint16'),    # 电机转速 rpm
    'error_code':  (8, 'int16'),     # 故障码
}
```

## InfluxDB时序数据库写入

InfluxDB适合工业时序数据存储，写入效率高，支持时间范围聚合查询。

```python
from influxdb_client import InfluxDBClient, Point, WritePrecision
from influxdb_client.client.write_api import SYNCHRONOUS

class InfluxDBWriter:
    def __init__(self, url, token, org, bucket):
        self.client = InfluxDBClient(
            url=url, token=token, org=org
        )
        self.write_api = self.client.write_api(
            write_options=SYNCHRONOUS
        )
        self.bucket = bucket

    def write_sensor_data(self, measurement, data):
        """写入传感器数据点"""
        point = Point(measurement)
        point.tag('device_id', data.get('device_id', 'default'))
        point.time(data['timestamp'], WritePrecision.NS)

        for key, value in data.items():
            if key in ('timestamp', 'device_id'):
                continue
            point.field(key, float(value))

        self.write_api.write(
            bucket=self.bucket,
            record=point
        )

    def close(self):
        self.client.close()
```

## 定时采集调度

用APScheduler实现定时采集，支持多设备轮询：

```python
from apscheduler.schedulers.background import BackgroundScheduler
import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

def collect_and_store(collector, writer, register_map, device_id):
    """单次采集并存储"""
    try:
        data = collector.read_all_sensors(register_map)
        data['device_id'] = device_id
        writer.write_sensor_data('sensor_data', data)
        logger.info(f"[{device_id}] 采集成功: {data}")
    except Exception as e:
        logger.error(f"[{device_id}] 采集失败: {e}")

# 初始化采集器和写入器
collector = ModbusCollector(port='/dev/ttyUSB0', baudrate=9600, slave_id=1)
writer = InfluxDBWriter(
    url='http://localhost:8086',
    token='your-token-here',
    org='your-org',
    bucket='industrial_data'
)

# 配置定时任务：每5秒采集一次
scheduler = BackgroundScheduler()
scheduler.add_job(
    collect_and_store,
    'interval', seconds=5,
    args=[collector, writer, register_map, 'PLC-001']
)
scheduler.start()

# 多设备轮询示例
devices = [
    {'port': '/dev/ttyUSB0', 'slave_id': 1, 'device_id': 'PLC-001'},
    {'port': '/dev/ttyUSB1', 'slave_id': 1, 'device_id': 'PLC-002'},
]

for dev in devices:
    dev_collector = ModbusCollector(
        port=dev['port'],
        slave_id=dev['slave_id']
    )
    scheduler.add_job(
        collect_and_store,
        'interval', seconds=5,
        args=[dev_collector, writer, register_map, dev['device_id']]
    )
```

## 常见问题（FAQ）

**Q：串口通信经常超时或读取失败怎么排查？**

A：按以下顺序排查。第一，检查物理连接——RS485的A/B线是否接反，终端电阻是否加装（120欧姆），线缆长度是否超过1200米。第二，检查波特率、数据位、停止位和校验位是否与PLC端配置一致。第三，用串口调试工具（如comtool）手动发送Modbus帧验证通信是否正常。第四，确认slave_id是否正确，同一总线上不能有重复的从站地址。第五，检查串口权限，Linux下需要将用户加入dialout组或用sudo运行。

**Q：InfluxDB写入频率过高导致内存占用大怎么办？**

A：开启批量写入模式，减少每次写入的网络开销。将write_api的写入选项从SYNCHRONOUS改为ASYNCHRONOUS或使用批处理模式。设置batch_size=500和flush_interval=1000ms，每500条或每1秒批量写入一次。另外，InfluxDB的数据保留策略（retention policy）要合理设置，工业数据通常保留30-90天，过期数据自动降采样或删除，避免磁盘膨胀。

**Q：如何处理采集过程中PLC临时断线的情况？**

A：实现重连机制和断线缓存。在采集函数中加入try-except捕获通信异常，失败时计数器加1，连续失败超过阈值（如5次）标记设备为离线状态。离线期间采集程序将数据缓存到本地SQLite或CSV文件，并记录时间戳。设备恢复后先写入断线期间的缓存数据（注意时间戳使用采集时刻而非恢复时刻），然后恢复实时采集。APScheduler可以设置misfire_grace_time参数处理任务积压。