1. 安川Memobus PLC通讯基础解析
在工业自动化系统中，PLC（可编程逻辑控制器）作为核心控制设备，其数据采集的稳定性和效率直接影响整个生产系统的运行质量。安川Memobus协议是安川电机为其PLC设备开发的专用通讯协议，基于Modbus RTU协议进行了功能扩展和优化。
1.1 协议特点与技术规范
Memobus协议采用主从式通讯架构，具有以下核心特性：
- 物理层支持RS-232C和RS-485两种接口标准
- 默认波特率9600bps（可配置为19200/38400等）
- 数据帧格式：1位起始位 + 8位数据位 + 1位停止位（无奇偶校验）
- 采用CRC-16校验机制确保数据完整性
- 单次通讯最大支持125个连续寄存器读取
与标准Modbus RTU相比，Memobus协议在功能码和地址映射方面有以下差异：
- 功能码03（读取保持寄存器）对应Memobus的"R"命令
- 功能码06（写单个寄存器）对应Memobus的"W"命令
- 寄存器地址采用5位十进制数表示（00000-09999对应不同存储区）
1.2 硬件连接准备
典型接线方案（以RS-485接口为例）：
- PLC端接线端子：SDA(+)接A线，SDB(-)接B线
- 终端电阻：在总线两端各接120Ω电阻
- 接地要求：屏蔽层单点接地，接地电阻<100Ω
重要提示：RS-485网络必须采用手拉手拓扑结构，避免星型或T型连接。通讯距离超过50米时建议使用屏蔽双绞线。
2. 通讯框架搭建与核心实现
2.1 Python环境配置
推荐使用Python 3.8+版本，核心依赖库：
bash复制pip install pyserial==3.5 crcmod==1.7
基础通讯类结构设计：
python复制import serial
import time
from crcmod import mkCrcFun
class MemobusPLC:
def __init__(self, port, baudrate=9600, timeout=0.5):
self.ser = serial.Serial(
port=port,
baudrate=baudrate,
bytesize=8,
parity='N',
stopbits=1,
timeout=timeout
)
self.crc16 = mkCrcFun(0x18005, rev=True, initCrc=0xFFFF)
def __enter__(self):
return self
def __exit__(self, exc_type, exc_val, exc_tb):
self.close()
def close(self):
if self.ser and self.ser.is_open:
self.ser.close()
2.2 命令帧构造原理
标准读寄存器命令帧结构：
| 字节位置 | 内容 | 说明 |
|---|---|---|
| 0 | 0x01 | 从站地址（默认1） |
| 1 | 0x03 | 功能码（读保持寄存器） |
| 2-3 | 起始地址 | 大端格式 |
| 4-5 | 寄存器数量 | 大端格式 |
| 6-7 | CRC16 | 小端格式 |
命令构造方法实现：
python复制def _build_read_command(self, address, count):
if not 0 <= address <= 9999 or not 1 <= count <= 125:
raise ValueError("Invalid address or count")
cmd = bytearray([
0x01, # 从站地址
0x03, # 功能码
(address >> 8) & 0xFF, # 地址高字节
address & 0xFF, # 地址低字节
(count >> 8) & 0xFF, # 数量高字节
count & 0xFF # 数量低字节
])
crc = self.crc16(cmd).to_bytes(2, 'little')
cmd.extend(crc)
return cmd
3. 数据类型采集实现详解
3.1 整型数据采集
3.1.1 INT16采集实现
python复制def read_int16(self, address):
"""读取有符号16位整数
Args:
address: 寄存器地址(0-9999)
Returns:
解码后的整数值
Raises:
IOError: 通讯错误时抛出
"""
cmd = self._build_read_command(address, 1)
self.ser.write(cmd)
# 预期响应：地址(1)+功能码(1)+字节数(1)+数据(2)+CRC(2)
response = self.ser.read(7)
if len(response) != 7:
raise IOError("Incomplete response")
# 校验CRC
if self.crc16(response[:-2]) != int.from_bytes(response[-2:], 'little'):
raise IOError("CRC check failed")
return int.from_bytes(response[3:5], 'big', signed=True)
3.1.2 UINT16采集优化
python复制def read_uint16(self, address):
"""读取无符号16位整数
优化点：
- 增加响应超时处理
- 添加重试机制
"""
max_retry = 3
for attempt in range(max_retry):
try:
cmd = self._build_read_command(address, 1)
self.ser.flushInput()
self.ser.write(cmd)
response = self.ser.read(7)
if len(response) == 7:
if self.crc16(response[:-2]) == int.from_bytes(response[-2:], 'little'):
return int.from_bytes(response[3:5], 'big', signed=False)
if attempt == max_retry - 1:
raise IOError("Max retry reached")
time.sleep(0.1)
except serial.SerialTimeoutException:
continue
3.2 32位数据采集方案
3.2.1 INT32采集实现
32位整数占用两个连续寄存器，需特殊处理：
python复制def read_int32(self, address):
"""读取有符号32位整数
注意：地址必须为偶数，否则可能引发对齐错误
"""
if address % 2 != 0:
raise ValueError("Address must be even for 32-bit read")
cmd = self._build_read_command(address, 2)
self.ser.write(cmd)
# 预期响应：地址(1)+功能码(1)+字节数(1)+数据(4)+CRC(2)
response = self.ser.read(9)
if len(response) != 9:
raise IOError("Incomplete response for 32-bit read")
if self.crc16(response[:-2]) != int.from_bytes(response[-2:], 'little'):
raise IOError("CRC check failed")
return int.from_bytes(response[3:7], 'big', signed=True)
3.2.2 高低字节序处理
不同PLC型号可能使用不同字节序，需添加配置选项：
python复制def __init__(self, port, byteorder='big', wordorder='big', **kwargs):
self.byteorder = byteorder # 字节序：'big'或'little'
self.wordorder = wordorder # 字序：'big'或'little'
# ...其他初始化代码...
def read_int32(self, address):
# ...前置校验代码...
data = response[3:7]
if self.wordorder == 'little':
# 交换寄存器顺序
data = data[2:4] + data[0:2]
return int.from_bytes(data, self.byteorder, signed=True)
3.3 布尔量采集技术
3.3.1 单点读取实现
python复制def read_bool(self, address):
"""读取单个布尔量
Args:
address: 位地址格式为"寄存器地址.位索引"(如"100.3")
"""
if isinstance(address, str):
reg_addr, bit_pos = map(int, address.split('.'))
if not 0 <= bit_pos <= 15:
raise ValueError("Bit position must be 0-15")
else:
reg_addr = address
bit_pos = 0
cmd = self._build_read_command(reg_addr, 1)
self.ser.write(cmd)
response = self.ser.read(7)
if len(response) != 7:
raise IOError("Incomplete response")
if self.crc16(response[:-2]) != int.from_bytes(response[-2:], 'little'):
raise IOError("CRC check failed")
reg_value = int.from_bytes(response[3:5], 'big')
return (reg_value & (1 << bit_pos)) != 0
3.3.2 批量读取优化
python复制def read_multiple_bools(self, start_address, count):
"""批量读取布尔量
优化：单次通讯读取多个位状态
"""
reg_count = (count + 15) // 16
cmd = self._build_read_command(start_address, reg_count)
self.ser.write(cmd)
expected_len = 3 + 2 * reg_count + 2
response = self.ser.read(expected_len)
if len(response) != expected_len:
raise IOError("Incomplete response")
if self.crc16(response[:-2]) != int.from_bytes(response[-2:], 'little'):
raise IOError("CRC check failed")
results = []
for i in range(count):
reg_index = i // 16
bit_pos = i % 16
reg_value = int.from_bytes(response[3+2*reg_index : 5+2*reg_index], 'big')
results.append((reg_value & (1 << bit_pos)) != 0)
return results
4. 工业现场应用实践
4.1 性能优化策略
4.1.1 通讯超时配置
根据网络状况调整超时参数：
python复制# 良好网络环境（响应快）
plc = MemobusPLC('/dev/ttyUSB0', timeout=0.1)
# 复杂工业环境（响应慢）
plc = MemobusPLC('COM3', timeout=1.0)
4.1.2 批量读取优化
减少通讯次数提升效率：
python复制def read_multiple_registers(self, start_address, count):
"""批量读取多个寄存器
最大支持125个寄存器单次读取
"""
if count > 125:
raise ValueError("Max 125 registers per read")
cmd = self._build_read_command(start_address, count)
self.ser.write(cmd)
expected_len = 3 + 2 * count + 2
response = self.ser.read(expected_len)
# ...校验处理...
return [int.from_bytes(response[3+2*i:5+2*i], 'big')
for i in range(count)]
4.2 异常处理机制
4.2.1 错误代码解析
python复制ERROR_CODES = {
0x01: "非法功能码",
0x02: "非法数据地址",
0x03: "非法数据值",
0x04: "从站设备故障",
0x05: "确认",
0x06: "从站设备忙"
}
def _check_error_response(self, response):
"""检查异常响应"""
if len(response) < 5:
return None
if response[1] & 0x80: # 错误响应标志
error_code = response[2]
return ERROR_CODES.get(error_code, f"未知错误({error_code})")
return None
4.2.2 自动重试机制
python复制def safe_read(self, address, count, retries=3):
"""带自动重试的读取方法"""
for attempt in range(retries):
try:
return self.read_multiple_registers(address, count)
except IOError as e:
if attempt == retries - 1:
raise
time.sleep(0.5 * (attempt + 1))
4.3 实际应用案例
4.3.1 温度监控系统
python复制def read_temperature(plc, sensor_config):
"""读取温度传感器数据
sensor_config示例：
{
'address': 100, # 起始寄存器地址
'type': 'int16', # 数据类型
'scale': 0.1, # 缩放系数
'unit': '℃' # 单位
}
"""
raw_value = {
'int16': plc.read_int16,
'uint16': plc.read_uint16,
'int32': plc.read_int32
}[sensor_config['type']](sensor_config['address'])
return raw_value * sensor_config['scale']
4.3.2 设备状态监控
python复制def monitor_machine_status(plc):
"""监控设备运行状态"""
status_bits = plc.read_multiple_bools(200, 8)
return {
'power_on': status_bits[0],
'running': status_bits[1],
'alarm': status_bits[2],
'emergency_stop': status_bits[3],
# ...其他状态位...
}
5. 高级功能扩展
5.1 数据写入实现
5.1.1 单寄存器写入
python复制def write_register(self, address, value):
"""写入单个寄存器"""
cmd = bytearray([
0x01, # 从站地址
0x06, # 功能码(写单个寄存器)
(address >> 8) & 0xFF,
address & 0xFF,
(value >> 8) & 0xFF,
value & 0xFF
])
crc = self.crc16(cmd).to_bytes(2, 'little')
cmd.extend(crc)
self.ser.write(cmd)
response = self.ser.read(8)
if len(response) != 8 or response[:6] != cmd[:6]:
raise IOError("Write verification failed")
5.1.2 多寄存器写入
python复制def write_multiple_registers(self, start_address, values):
"""写入多个连续寄存器"""
if len(values) > 120:
raise ValueError("Too many registers for single write")
cmd = bytearray([
0x01, # 从站地址
0x10, # 功能码(写多个寄存器)
(start_address >> 8) & 0xFF,
start_address & 0xFF,
(len(values) >> 8) & 0xFF,
len(values) & 0xFF,
len(values) * 2 # 字节数
])
for value in values:
cmd.extend([(value >> 8) & 0xFF, value & 0xFF])
crc = self.crc16(cmd).to_bytes(2, 'little')
cmd.extend(crc)
self.ser.write(cmd)
response = self.ser.read(8)
if len(response) != 8 or response[:6] != cmd[:6]:
raise IOError("Multi-write verification failed")
5.2 异步通讯支持
5.2.1 基于线程的异步读取
python复制from threading import Thread
class AsyncPLCReader:
def __init__(self, plc, callback):
self.plc = plc
self.callback = callback
self.running = False
def start(self, address, interval=1.0):
self.running = True
self.thread = Thread(target=self._read_loop,
args=(address, interval))
self.thread.start()
def stop(self):
self.running = False
self.thread.join()
def _read_loop(self, address, interval):
while self.running:
try:
value = self.plc.read_int16(address)
self.callback(value)
except IOError as e:
print(f"Read error: {e}")
time.sleep(interval)
5.2.2 使用asyncio实现
python复制import asyncio
class AsyncMemobusPLC:
def __init__(self, port, loop=None):
self.ser = serial.Serial(port)
self.loop = loop or asyncio.get_event_loop()
self.reader = None
self.writer = None
async def connect(self):
self.reader, self.writer = await serial_asyncio.open_serial_connection(
url=self.ser.port,
baudrate=self.ser.baudrate,
loop=self.loop
)
async def read_int16(self, address):
cmd = self._build_read_command(address, 1)
self.writer.write(cmd)
response = await self.reader.read(7)
if len(response) != 7:
raise IOError("Incomplete response")
return int.from_bytes(response[3:5], 'big', signed=True)
5.3 数据日志与监控
5.3.1 实时数据记录
python复制class DataLogger:
def __init__(self, plc, config):
self.plc = plc
self.config = config # 数据点配置列表
self.data = {item['name']: [] for item in config}
def start_logging(self, interval=1.0):
self.running = True
while self.running:
timestamp = time.time()
for item in self.config:
try:
value = getattr(self.plc, f"read_{item['type']}")(item['address'])
self.data[item['name']].append((timestamp, value))
except IOError:
continue
time.sleep(interval)
def save_to_csv(self, filename):
with open(filename, 'w') as f:
f.write("timestamp," + ",".join(item['name'] for item in self.config) + "\n")
# 获取最小长度避免索引越界
min_len = min(len(self.data[name]) for name in self.data)
for i in range(min_len):
line = []
for item in self.config:
name = item['name']
ts, val = self.data[name][i]
line.append(str(val))
f.write(f"{ts}," + ",".join(line) + "\n")
5.3.2 异常状态监测
python复制class AlarmMonitor:
def __init__(self, plc, alarm_config):
self.plc = plc
self.alarm_config = alarm_config # 报警配置字典
def check_alarms(self):
active_alarms = []
for name, config in self.alarm_config.items():
value = getattr(self.plc, f"read_{config['type']}")(config['address'])
if config.get('inverted', False):
triggered = value < config['threshold']
else:
triggered = value > config['threshold']
if triggered:
active_alarms.append({
'name': name,
'value': value,
'timestamp': time.time()
})
return active_alarms