Python程序员跨界玩转PLC：用Snap7实现工业控制的数据读写

第一次看到PLC程序员讨论"I0.1"、"M10.2"、"DB5.DBX10.0"这样的地址时，作为Python开发者的你可能会感到一头雾水。这就像突然被扔进了一个全是专业术语的会议室，周围的人都在说着你听不懂的黑话。但别担心，本文将带你用Python开发者熟悉的视角，一步步破解这些PLC地址的密码。

1. PLC存储区的Python式解读

在开始操作之前，我们需要先理解PLC中几种核心存储区的概念。对于Python开发者来说，可以把PLC的存储区想象成几种不同的变量类型：

• 输入区(I区) ：相当于只读变量，PLC从外部设备(如传感器)获取的数据存储在这里
• 输出区(Q区) ：相当于只写变量，PLC发送给外部设备(如执行器)的数据存储在这里
• 位存储区(M区) ：相当于全局变量，程序运行过程中使用的中间状态标志
• 数据块(DB区) ：相当于字典或类实例，结构化数据的存储区域

# Python类比
I区 = 传感器输入  # 只读
Q区 = 执行器输出  # 只写 
M区 = global_vars  # 可读写
DB区 = data_objects  # 结构化数据

1.1 地址解码：从PLC黑话到Python能理解的格式

PLC程序员使用的地址格式看似复杂，但其实有规律可循。让我们用表格来解析这些地址：

PLC地址格式	含义	Python类比
I0.1	输入区字节0的第1位	input_buffer[0][1]
Q2.5	输出区字节2的第5位	output_buffer[2][5]
M10.2	位存储区字节10的第2位	flags[10][2]
DB5.DBX10.0	数据块5中字节10的第0位	db[5]['data'][10][0]
MW201	位存储区从字节201开始的字(2字节)	struct.unpack('h', flags[201:203])

2. Snap7库的核心武器：read_area和write_area

Snap7库提供了两个核心方法来实现PLC的读写操作。理解这两个方法，就掌握了与PLC通信的钥匙。

2.1 read_area方法详解

def read_area(area, dbnumber, start, size):
    """
    从PLC读取数据
    :param area: 区域代码(如输入区、输出区等)
    :param dbnumber: 数据块编号(仅当area为DB区时使用)
    :param start: 起始字节偏移量
    :param size: 要读取的字节数
    :return: 包含读取数据的bytearray
    """

注意：size参数的单位是字节，不是位。读取单个位时也需要读取整个字节。

2.2 write_area方法详解

def write_area(area, dbnumber, start, data):
    """
    向PLC写入数据
    :param area: 区域代码
    :param dbnumber: 数据块编号
    :param start: 起始字节偏移量
    :param data: 要写入的数据(bytearray格式)
    """

2.3 区域代码对照表

PLC中的每个存储区在Snap7中都有对应的区域代码：

存储区	Snap7常量	十六进制值	Python类比
输入区(I)	snap7.snap7types.areas.PE	0x81	只读缓冲区
输出区(Q)	snap7.snap7types.areas.PA	0x82	只写缓冲区
位存储区(M)	snap7.snap7types.areas.MK	0x83	全局变量
数据块(DB)	snap7.snap7types.areas.DB	0x84	字典/对象
定时器	snap7.snap7types.areas.TM	0x1D	计时器对象
计数器	snap7.snap7types.areas.CT	0x1C	计数器对象

3. 实战演练：从理论到代码

现在让我们通过几个实际例子，演示如何读写不同类型的PLC地址。

3.1 读写单个位(M10.1)

import snap7
import struct

def read_write_single_bit(client):
    """读写M10.1位"""
    area = snap7.snap7types.areas.MK  # M区
    dbnumber = 0  # 不适用于M区
    start = 10  # 字节偏移量
    size = 1  # 读取1个字节
    
    # 读取当前值
    data = client.read_area(area, dbnumber, start, size)
    print(f"M10.1当前值: {bool(data[0] & 0x02)}")  # 检查第1位(0x02)
    
    # 将M10.1置1
    new_value = bytes([data[0] | 0x02])  # 设置第1位
    client.write_area(area, dbnumber, start, new_value)
    
    # 验证写入结果
    updated_data = client.read_area(area, dbnumber, start, size)
    print(f"M10.1更新后值: {bool(updated_data[0] & 0x02)}")

3.2 读写字(MW201)

def read_write_word(client):
    """读写MW201(2字节)"""
    area = snap7.snap7types.areas.MK
    dbnumber = 0
    start = 201  # MW201的字节偏移量
    size = 2  # 字=2字节
    
    # 读取当前值
    data = client.read_area(area, dbnumber, start, size)
    value = struct.unpack('>h', data)[0]  # 大端序有符号短整型
    print(f"MW201当前值: {value}")
    
    # 写入新值123
    new_value = struct.pack('>h', 123)
    client.write_area(area, dbnumber, start, new_value)
    
    # 验证写入结果
    updated_data = client.read_area(area, dbnumber, start, size)
    print(f"MW201更新后值: {struct.unpack('>h', updated_data)[0]}")

3.3 读写数据块(DB5.DBW10)

def read_write_data_block(client):
    """读写DB5.DBW10(数据块5中的字，偏移10)"""
    area = snap7.snap7types.areas.DB
    dbnumber = 5  # 数据块编号
    start = 10  # 偏移量
    size = 2  # 字=2字节
    
    # 读取当前值
    data = client.read_area(area, dbnumber, start, size)
    value = struct.unpack('>h', data)[0]
    print(f"DB5.DBW10当前值: {value}")
    
    # 写入新值456
    new_value = struct.pack('>h', 456)
    client.write_area(area, dbnumber, start, new_value)
    
    # 验证写入结果
    updated_data = client.read_area(area, dbnumber, start, size)
    print(f"DB5.DBW10更新后值: {struct.unpack('>h', updated_data)[0]}")

4. 高级技巧与最佳实践

掌握了基础读写操作后，让我们来看一些提高效率和可靠性的技巧。

4.1 批量读写优化

频繁的小数据读写会影响性能，应该尽量批量操作：

def bulk_read_write(client):
    """批量读写多个数据"""
    # 批量读取M区从字节200开始的10个字节
    bulk_data = client.read_area(snap7.snap7types.areas.MK, 0, 200, 10)
    
    # 解析多个值
    mw200 = struct.unpack('>h', bulk_data[0:2])[0]  # MW200
    mw202 = struct.unpack('>h', bulk_data[2:4])[0]  # MW202
    m201_byte = bulk_data[4]  # MB201
    m201_5 = bool(m201_byte & 0x20)  # M201.5
    
    # 批量写入
    new_values = bytearray()
    new_values.extend(struct.pack('>h', 100))  # MW200
    new_values.extend(struct.pack('>h', 200))  # MW202
    new_values.append(0xFF)  # MB201=255
    client.write_area(snap7.snap7types.areas.MK, 0, 200, new_values)

4.2 错误处理与重试机制

工业环境中网络可能不稳定，需要健壮的错误处理：

from time import sleep

def robust_read(client, area, dbnumber, start, size, max_retries=3):
    """带重试的读取操作"""
    for attempt in range(max_retries):
        try:
            return client.read_area(area, dbnumber, start, size)
        except Exception as e:
            if attempt == max_retries - 1:
                raise
            print(f"读取失败，重试 {attempt + 1}/{max_retries}: {e}")
            sleep(1)

4.3 数据类型转换工具函数

创建一些辅助函数简化常见数据类型的转换：

def plc_read_int(client, area, dbnumber, start):
    """读取有符号整型"""
    data = client.read_area(area, dbnumber, start, 2)
    return struct.unpack('>h', data)[0]

def plc_write_int(client, area, dbnumber, start, value):
    """写入有符号整型"""
    data = struct.pack('>h', value)
    client.write_area(area, dbnumber, start, data)

def plc_read_bool(client, area, dbnumber, byte_offset, bit_offset):
    """读取单个布尔值"""
    data = client.read_area(area, dbnumber, byte_offset, 1)
    return bool(data[0] & (1 << bit_offset))

def plc_write_bool(client, area, dbnumber, byte_offset, bit_offset, value):
    """写入单个布尔值"""
    data = client.read_area(area, dbnumber, byte_offset, 1)
    if value:
        new_byte = data[0] | (1 << bit_offset)
    else:
        new_byte = data[0] & ~(1 << bit_offset)
    client.write_area(area, dbnumber, byte_offset, bytes([new_byte]))

4.4 监控数据变化

实现一个简单的数据变化监控器：

def monitor_plc_data(client, area, dbnumber, start, size, interval=1.0):
    """监控PLC数据变化"""
    last_value = None
    try:
        while True:
            current_value = client.read_area(area, dbnumber, start, size)
            if current_value != last_value:
                print(f"数据变化: {current_value.hex()}")
                last_value = current_value
            sleep(interval)
    except KeyboardInterrupt:
        print("监控停止")

在工业4.0和物联网的大背景下，掌握Python与PLC的通信能力为开发者打开了一个全新的领域。通过本文介绍的技术，你可以轻松地将现代软件开发实践引入传统的工业控制领域，构建更智能、更高效的工业自动化解决方案。