工业一体机是面向工业环境设计的集成化计算设备，集成了处理器、显示屏和工业通信接口，具备宽温运行和抗电磁干扰能力，常作为工业现场的数据采集与控制终端使用。在数据采集应用中，工业一体机常需要通过Modbus TCP协议与PLC、变频器、仪表等设备进行通信。本文从开发环境搭建到完整数据采集系统实现，详细讲解基于Python的Modbus TCP客户端开发全流程。

## 开发环境

- 操作系统：Ubuntu 22.04 LTS / Debian 12
- Python版本：3.10+
- 核心依赖库：pymodbus 3.6+
- 测试工具：Modbus Slave模拟器

```bash
# 创建虚拟环境
python3 -m venv venv
source venv/bin/activate

# 安装依赖
pip install pymodbus==3.6.3
pip install pyinstaller # 打包用
```

## Modbus TCP协议要点

Modbus TCP基于以太网TCP/IP协议栈，使用502端口通信。与Modbus RTU不同，TCP版本去掉了CRC校验（由TCP层保证可靠性），增加了MBAP头部（Modbus Application Protocol Header）。

MBAP头部共7字节：

| 字段 | 长度 | 说明 |
|------|------|------|
| Transaction ID | 2字节 | 事务标识，用于请求/响应匹配 |
| Protocol ID | 2字节 | 协议标识，Modbus固定为0x0000 |
| Length | 2字节 | 后续数据长度 |
| Unit ID | 1字节 | 从站地址 |

常用的功能码：

```python
# 功能码定义
FUNCTION_CODES = {
'read_coils': 0x01, # 读线圈
'read_discrete_inputs': 0x02, # 读离散输入
'read_holding_regs': 0x03, # 读保持寄存器
'read_input_regs': 0x04, # 读输入寄存器
'write_single_coil': 0x05, # 写单个线圈
'write_single_reg': 0x06, # 写单个寄存器
'write_multi_regs': 0x10, # 写多个寄存器
}
```

## 完整客户端实现

```python
import logging
import time
from datetime import datetime
from pymodbus.client import ModbusTcpClient
from pymodbus.exceptions import ModbusException

class IndustrialDataCollector:
"""工业一体机Modbus TCP数据采集器"""

def __init__(self, host='192.168.1.100', port=502, timeout=3):
self.host = host
self.port = port
self.timeout = timeout
self.client = None
self.reconnect_attempts = 3
self.reconnect_delay = 2

# 配置日志
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s [%(levelname)s] %(message)s',
handlers=[
logging.FileHandler('/var/log/modbus_collector.log'),
logging.StreamHandler()
]
)
self.logger = logging.getLogger(__name__)

def connect(self):
"""建立Modbus TCP连接，支持自动重连"""
for attempt in range(self.reconnect_attempts):
try:
self.client = ModbusTcpClient(
host=self.host,
port=self.port,
timeout=self.timeout
)
if self.client.connect():
self.logger.info(f"已连接到 {self.host}:{self.port}")
return True
except Exception as e:
self.logger.warning(
f"连接失败(第{attempt+1}次): {e}"
)
time.sleep(self.reconnect_delay)

self.logger.error(f"无法连接到Modbus设备，已尝试{self.reconnect_attempts}次")
return False

def disconnect(self):
"""断开连接"""
if self.client and self.client.connected:
self.client.close()
self.logger.info("连接已断开")

def read_holding_registers(self, unit_id, start_addr, count):
"""
读取保持寄存器
:param unit_id: 从站地址
:param start_addr: 起始寄存器地址
:param count: 读取数量
:return: 寄存器值列表或None
"""
try:
result = self.client.read_holding_registers(
address=start_addr,
count=count,
slave=unit_id
)
if result.isError():
self.logger.error(
f"读取保持寄存器失败 - 从站:{unit_id} "
f"地址:{start_addr} 错误:{result}"
)
return None
return result.registers
except ModbusException as e:
self.logger.error(f"Modbus异常: {e}")
return None

def write_register(self, unit_id, address, value):
"""写入单个保持寄存器"""
try:
result = self.client.write_register(
address=address,
value=value,
slave=unit_id
)
if result.isError():
self.logger.error(f"写入失败 - 地址:{address} 错误:{result}")
return False
return True
except ModbusException as e:
self.logger.error(f"Modbus异常: {e}")
return False

def collect_data(self, config):
"""
按配置批量采集数据
:param config: 采集配置列表
[{'unit_id': 1, 'addr': 0, 'count': 10, 'name': '温度'},
{'unit_id': 1, 'addr': 10, 'count': 5, 'name': '压力'}]
:return: 采集结果字典
"""
results = {}
timestamp = datetime.now().strftime('%Y-%m-%d %H:%M:%S')

for item in config:
unit_id = item['unit_id']
addr = item['addr']
count = item['count']
name = item['name']

values = self.read_holding_registers(unit_id, addr, count)
if values is not None:
results[name] = {
'values': values,
'timestamp': timestamp,
'unit_id': unit_id,
'addr': addr
}
self.logger.info(
f"[{name}] 从站:{unit_id} 地址:{addr} "
f"值:{values}"
)
else:
results[name] = None
self.logger.warning(f"[{name}] 采集失败")

return results
```

## 数据格式转换

Modbus寄存器是16位的，但实际工业数据可能是32位浮点数（Float32）或32位整数（Int32），需要做字节拼接处理。

```python
import struct

class DataConverter:
"""Modbus数据格式转换工具"""

@staticmethod
def regs_to_float32(regs, byte_order='cdab'):
"""
将两个寄存器值转换为32位浮点数
:param regs: [reg_high, reg_low] 两个寄存器值
:param byte_order: 字节序
'abcd' - 大端序
'badc' - 字交换大端
'cdab' - 字交换小端
'dcba' - 小端序
:return: 浮点数值
"""
if len(regs) < 2:
return None

# 将寄存器值转为字节
b = regs[0].to_bytes(2, 'big') + regs[1].to_bytes(2, 'big')

if byte_order == 'abcd':
pass
elif byte_order == 'badc':
b = b[1:2] + b[0:1] + b[3:4] + b[2:3]
elif byte_order == 'cdab':
b = b[2:4] + b[0:2]
elif byte_order == 'dcba':
b = b[3:4] + b[2:3] + b[1:2] + b[0:1]

return struct.unpack('>f', b)[0]

@staticmethod
def regs_to_int32(regs, signed=True):
"""将两个寄存器值转换为32位整数"""
if len(regs) < 2:
return None
raw = (regs[0] << 16) | regs[1]
if signed and raw >= 0x80000000:
raw -= 0x100000000
return raw

@staticmethod
def regs_to_string(regs, encoding='utf-8'):
"""将寄存器值列表转为字符串"""
raw_bytes = b''
for reg in regs:
raw_bytes += reg.to_bytes(2, 'big')
try:
return raw_bytes.decode(encoding).strip('\x00')
except UnicodeDecodeError:
return None
```

## 看门狗与自动恢复

工业现场数据采集要求7×24小时不间断运行，必须实现看门狗机制。

```python
import signal
import os
import threading

class Watchdog:
"""进程看门狗，监控采集线程存活状态"""

def __init__(self, check_interval=30, max_misses=3):
self.check_interval = check_interval
self.max_misses = max_misses
self.miss_count = 0
self.last_heartbeat = time.time()
self._running = True

def heartbeat(self):
"""采集线程定期调用，表示存活"""
self.last_heartbeat = time.time()
self.miss_count = 0

def start(self, on_timeout_callback):
"""启动看门狗监控线程"""
def _monitor():
while self._running:
elapsed = time.time() - self.last_heartbeat
if elapsed > self.check_interval:
self.miss_count += 1
logging.warning(
f"采集线程心跳超时 "
f"(已等待{elapsed:.0f}s, 第{self.miss_count}次)"
)
if self.miss_count >= self.max_misses:
logging.error(
f"心跳连续超时{self.max_misses}次，"
f"触发恢复流程"
)
on_timeout_callback()
self.miss_count = 0
time.sleep(self.check_interval)

thread = threading.Thread(target=_monitor, daemon=True)
thread.start()

def stop(self):
self._running = False
```

## 主程序入口

```python
def main():
# 采集配置
config = [
{'unit_id': 1, 'addr': 0, 'count': 2, 'name': '反应釜温度'},
{'unit_id': 1, 'addr': 2, 'count': 2, 'name': '管道压力'},
{'unit_id': 2, 'addr': 0, 'count': 10, 'name': '产线计数'},
]

collector = IndustrialDataCollector(
host='192.168.1.100',
port=502,
timeout=3
)

converter = DataConverter()
watchdog = Watchdog(check_interval=30, max_misses=3)

def on_watchdog_timeout():
"""看门狗超时回调：重连设备"""
logging.info("正在重新连接Modbus设备...")
collector.disconnect()
time.sleep(5)
collector.connect()

watchdog.start(on_watchdog_timeout)

# 主采集循环
try:
if not collector.connect():
sys.exit(1)

while True:
results = collector.collect_data(config)

# 数据转换示例
if results.get('反应釜温度'):
regs = results['反应釜温度']['values']
temp = converter.regs_to_float32(regs, 'cdab')
logging.info(f"反应釜温度: {temp:.2f}℃")

watchdog.heartbeat()
time.sleep(1) # 1秒采集间隔

except KeyboardInterrupt:
logging.info("收到退出信号，正在关闭...")
finally:
collector.disconnect()
watchdog.stop()

if __name__ == '__main__':
main()
```

## 常见调试问题

**连接超时**：检查网络是否通，502端口是否开放。用telnet测试：`telnet 192.168.1.100 502`。

**读取返回非法数据地址（0x02）**：寄存器地址从0开始还是从1开始是常见坑。Modbus协议规范是0-based，但有些设备文档用1-based地址，需要减1。

**数据值不对**：99%是字节序问题。用Modbus Poll工具读取原始寄存器值，手动拼接验证。不同厂家的字节序不同，常见的有ABCD（大端）、CDAB（字交换）等。

**间歇性断连**：检查TCP KeepAlive设置，工业网络中交换机可能主动断开空闲连接。在创建client时加上keepalive参数。

## 常见问题（FAQ）

**Q：Modbus TCP和Modbus RTU的主要区别是什么？**
A：Modbus TCP基于以太网TCP/IP协议，使用502端口，去掉了CRC校验（由TCP层保证），增加了7字节MBAP头部；Modbus RTU基于串口（RS485/RS232），使用CRC16校验。TCP版本传输距离更远、速度更快、支持多主机访问，RTU版本成本低、布线简单，适合短距离点对多通信。

**Q：Python pymodbus库如何处理寄存器地址0-based和1-based的问题？**
A：pymodbus库使用0-based地址，即第一个寄存器地址为0。如果设备文档标注地址为1-based，在代码中需要将地址减1。建议先用Modbus Poll工具测试确认地址映射关系，再写入代码，避免因地址偏移导致读取到错误数据。

**Q：工业数据采集系统如何保证7×24小时不间断运行？**
A：需要实现三个层级的保障：进程级使用看门狗监控采集线程心跳，超时自动重启；连接级设置自动重连机制，检测到断连后延时重试；系统级可配合Linux的systemd服务管理，设置Restart=always实现进程崩溃后自动拉起。