从零开发AB PLC通信协议:CIP/PCCC协议解析与Python实战
2026/8/13 8:52:48 网站建设 项目流程

1. 项目缘起:为什么需要自己动手搞AB PLC协议?

在工业自动化这个行当里,AB(Allen-Bradley)的PLC(可编程逻辑控制器)就像车间里的“大脑”,地位举足轻重。但很多时候,我们这些做系统集成、数据采集或者MES(制造执行系统)开发的,会遇到一个挺头疼的问题:怎么让上位机、SCADA(数据采集与监控系统)或者我们自己写的应用,跟这些AB PLC“说上话”?官方当然有解决方案,比如罗克韦尔自家的FactoryTalk、KEPServerEX OPC服务器,或者一些第三方的驱动库。但用过的都知道,要么是授权费用不菲,要么是灵活性受限,要么是部署起来一堆依赖,在特定场景下(比如轻量级边缘计算、定制化协议网关、老旧系统改造)显得特别笨重。

这时候,自己动手开发一套AB PLC的通信协议栈,就从“可选项”变成了“必选项”。这活儿听起来挺硬核,像是资深工程师的专属领域,但实际拆解下来,核心就是理解AB那套封闭但又自成体系的通信规则,然后用代码把它“翻译”成我们能用的数据。我最近刚完成一个项目,需要从多台不同系列的AB PLC(包括CompactLogix、Micro800系列)里实时读取生产数据,并推送到云端数据库。受限于成本和架构,没法上全套的OPC方案,于是硬着头皮把AB的几个主流协议(主要是CIP和PCCC的变种)摸了一遍,攒了不少经验和教训。

这篇文章,我就把自己从零开始折腾AB PLC协议开发的过程、核心原理、踩过的坑以及一些实用的代码片段,做个系统的梳理。目标不是给你一个能直接拷贝的万能库(那也不现实,协议细节和PLC型号、固件版本强相关),而是给你一张“地图”和一套“工具”,让你知道这条路该怎么走,遇到障碍该怎么绕过去。无论你是想做个简单的数据采集工具,还是开发一个专用的协议转换网关,希望这些内容都能帮到你。

2. 协议家族概览:AB PLC的“语言体系”

在跟AB PLC对话之前,得先搞清楚它到底会说哪几种“方言”。AB的通信协议不是一个单一的东西,而是一个随着产品线演进形成的家族。主要可以分为两大类:较新的、面向对象的CIP协议族,和较老的、基于消息的PCCC协议族。理解它们的适用场景是选型的第一步。

2.1 CIP:现代AB设备的通用语

CIP(Common Industrial Protocol,通用工业协议)是罗克韦尔自动化提出的,基于对象模型的工业通信协议。你可以把它理解成工业领域的“HTTP+SOAP/REST”,它定义了一套标准的服务、对象和数据类型。在AB的世界里,EtherNet/IP和ControlNet等网络都是基于CIP的应用层协议。

核心特点:

  • 面向对象:PLC中的一切(程序、标签、模块)都被抽象为对象(Object),每个对象有类(Class)、实例(Instance)和属性(Attribute)。比如,一个Program对象类下,有多个程序实例,每个实例有状态、大小等属性。
  • 显式消息(Explicit Messaging):这是最常用的方式,用于非实时、一对一的请求/响应通信,例如读取/写入标签值。它使用TCP/IP(通常端口44818)作为传输层,协议格式相对规整。
  • 隐式消息(Implicit Messaging):用于实时性要求高的I/O数据交换,通常通过UDP和组播实现,是预配置好的周期性数据交换。我们做数据采集,大部分时候打交道的是显式消息。
  • 服务(Services):对对象执行的操作,比如Get_Attribute_Single(读取单个属性)、Set_Attribute_Single(写入单个属性)、Read_TagWrite_Tag等。

开发重点:实现CIP显式消息协议,关键在于构建正确的CIP封装报文。报文是分层封装的:最外层是以太网帧/IP包/TCP段,里面是封装头(Encapsulation Header),再里面才是CIP数据。封装头包含了会话句柄、命令代码(如0x6F代表发送RRData,用于CIP服务)等信息。CIP数据部分则包含了服务请求路径(Path)和具体服务数据。

2.2 PCCC/DF1:传统PLC的“方言”

PCCC(Programmable Controller Communication Commands)是一套更早的、用于与SLC 500、MicroLogix、PLC-5等老型号PLC通信的命令集。DF1是其在串行链路(RS-232/485)上的物理层和数据链路层协议。后来,为了在以太网上传输PCCC命令,又衍生出了CIP PCCC模式,即把PCCC命令作为数据负载,封装在CIP显式消息中进行传输。

核心特点:

  • 基于命令/响应:协议由一系列固定的命令码(Command Code)构成,例如0x0F是读取数据文件,0xAA是诊断状态。每个命令有固定的报文格式。
  • 文件号寻址:数据不按标签名寻址,而是按“文件类型”(如N-整型,F-浮点,B-位)和“文件号”、“元素号”来定位。比如,N7:0表示整型文件7的第0个元素。
  • 常用于较老型号:Micro800系列的部分型号、SLC 500等主要使用这种方式。

开发重点:实现PCCC协议,需要准确构造命令帧。如果是串口DF1,还要处理数据链路层的帧头、帧尾、校验以及可能的纠错重发机制。如果是基于以太网的CIP PCCC,则需要先建立CIP会话,然后将构造好的PCCC命令帧作为CIP服务的特定数据发送。

注意:选择哪种协议,首要取决于目标PLC的型号和支持的通信方式。新型的ControlLogix、CompactLogix通常首选纯CIP(EtherNet/IP)。而Micro800或一些兼容性场景下,可能需要使用CIP PCCC。最稳妥的方法是查阅PLC的具体手册或通过Wireshark抓包分析其与官方软件(如Connected Components Workbench, Studio 5000 Logix Designer)的通信过程。

3. 核心实战:从抓包分析到报文构造

理论说再多,不如动手抓个包看得明白。协议开发,Wireshark是你的第一导师。下面我以读取一个ControlLogix PLC的标签值为例,拆解整个流程。

3.1 环境搭建与抓包准备

  1. 硬件:一台AB PLC(如CompactLogix 5380),一台安装有Studio 5000和Wireshark的工程师站电脑,通过交换机连接在同一局域网。
  2. 软件配置:在Studio 5000中,为PLC配置好IP地址,并创建一个简单的测试程序,里面定义几个不同类型的标签,例如DINT_Test(双整数)、REAL_Test(浮点数)、BOOL_Test(布尔量)。
  3. 抓包:在Wireshark中,选择正确的网卡,开始抓包。然后在Studio 5000的“通信”窗口中,在线连接PLC,并打开“监视标签”视图,观察标签值。此时,Wireshark会捕获到所有的通信报文。

3.2 解密一个典型的CIP读取标签请求

过滤Wireshark的显示,只关注与PLC IP地址相关的TCP流量(例如tcp and ip.addr == 192.168.1.10)。你会看到类似下图的会话:

No. Time Source Destination Protocol Length Info 100 10.123456 192.168.1.100 192.168.1.10 TCP 66 49252 → 44818 [PSH, ACK] Seq=1 Ack=1 Win=xxx Len=0 101 10.123457 192.168.1.10 192.168.1.100 TCP 66 44818 → 49252 [ACK] Seq=1 Ack=45 Win=xxx Len=0 102 10.123458 192.168.1.100 192.168.1.10 TCP 1514 [TCP segment of a reassembled PDU] 103 10.123459 192.168.1.10 192.168.1.100 TCP 66 [ACK] 104 10.123460 192.168.1.10 192.168.1.100 TCP 1514 [TCP segment of a reassembled PDU]

我们需要关注的是携带实际数据的数据包。右键点击一个看起来是请求的包(比如No.102),选择“追踪流” -> “TCP流”。你会看到一个完整的请求-响应字节流。

请求报文解剖(简化版):

一个完整的CIP读取标签请求,从外到内分为三层封装:

  1. TCP/IP层:标准的以太网帧、IP包头、TCP包头。目标端口是44818
  2. CIP封装层(Encapsulation Header)
    • 命令(Command)0x6F。这代表“SendRRData”,用于发送请求并要求回复数据。
    • 长度(Length):后面数据部分的长度。
    • 会话句柄(Session Handle):在会话建立时由PLC分配的一个4字节ID,后续所有通信都要带上它。建立会话的命令是0x65(RegisterSession)。
    • 状态(Status)0x0000,表示成功。
    • 发送者上下文(Sender Context):一个8字节的标识,用于匹配请求和响应,通常由客户端生成一个随机值。
    • 选项(Options)0x0000
  3. CIP数据层(Interface: CIP Data)
    • 接口句柄(Interface Handle)0x00000000,表示CIP。
    • 超时(Timeout)0x0000,表示无限等待。
    • 项数(Item Count)0x0002,表示后面跟了两个地址项(Item)。
    • 地址项1(Item 1):类型0x0000(Null Address),长度0x0000
    • 地址项2(Item 2):类型0x00B2(Connected Data Item),长度后面跟实际数据长度。
    • CIP服务数据(Service Data)
      • 服务路径(Path):指向目标对象。例如,读取标签的路径可能是0x20, 0x06, 0x24, 0x01。这需要解析:0x20(逻辑段,端口号),0x06(背板),0x24(Class ID, 这里是0x24=36,代表Symbol Class?不完全是,更常见的是直接指定标签名)。实际上,对于Logix标签,路径通常包含“Symbol Instance”等信息,更通用的方式是使用“ANSI Ext. Symbolic”片段类型(0x91)后跟编码的标签名。
      • 服务代码(Service Code)0x4C,这是Read Tag服务的代码。
      • 请求数据:包含要读取的标签名称(如“MyTag”的ASCII编码,可能带结构体成员访问MyTag.SubElement)、元素数量、数据类型等信息。

看到这里你可能有点晕,确实,手动构造这个路径是最复杂的一环。一个取巧的方法是:先用官方软件正常通信,抓取读取你目标标签的报文,然后直接模仿它的路径结构。路径的解析规则(CIP Segment)本身就是一个专题。

3.3 响应报文解析与错误处理

PLC的响应报文结构类似,在CIP封装层,命令会是0x6F(SendRRData的回复),状态码需要检查。重点在CIP数据层的响应部分:

  • 服务代码:请求的代码+0x80(表示响应),例如0xCC0x4C+0x80)。
  • 回复状态(Reply Status):一个字节,0x00表示成功。其他值代表错误,如0x05(路径错误)、0x08(资源不可用)、0x15(数据类型不匹配)等。必须处理这些状态码,这是调试时最重要的信息。
  • 附加状态(Extended Status):可选的更详细错误信息。
  • 响应数据:如果成功,这里就是读取到的原始字节数据。你需要根据请求中指定的数据类型(DINT, REAL, BOOL等)来解析这些字节(注意AB PLC通常使用小端字节序)。

实操心得:不要试图一次性理解所有路径规则。先从模仿一个成功的抓包报文开始。用Python或C#写个小程序,把抓到的请求报文字节数组原封不动地发出去,如果能收到正确响应,就成功了一大半。然后逐步替换其中的标签名字段,观察变化。

4. 关键难点与避坑指南

自己开发协议,肯定会遇到各种坑。下面是我总结的几个最典型的难点和解决方案。

4.1 标签路径的编码:最大的“黑盒”

如前所述,如何将“Program:MainProgram.MyDINTTag”这样的标签名,转换成CIP报文里那一串十六进制路径,是首要难题。

解决方案:

  1. 抓包逆向:最可靠的方法。用Studio 5000对不同类型标签(基本类型、数组、结构体成员)进行读写操作,抓包对比路径部分的差异。你会发现,对于全局标签和程序作用域标签,路径是不同的。
  2. 理解CIP路径段格式:一个路径由多个段(Segment)组成。每个段第一个字节是类型/格式。常见的有:
    • 0x91:ANSI扩展符号段(ANSI Extended Symbolic)。后面跟一个字节的长度(L),再跟L个字节的标签名ASCII码。这是用于指定标签名的主要方式。
    • 0x28:逻辑段(Logical Segment),用于指定端口、桥路等。
    • 0x20:端口段(Port Segment)。
    • 段可以嵌套,例如访问结构体成员:路径 =[标签名段] + [成员名段]
  3. 使用已知的Class/Instance ID:对于一些固定对象,如Identity Object(Class 0x01),可以直接使用数字路径。但对于用户自定义标签,必须用符号名。
  4. 参考开源实现:研究像pycomm3cpppo这样的Python库,或者libplctagC库的源代码,看它们是如何构造路径的。这是快速学习的捷径。

踩坑记录:我曾试图直接根据文档手动计算路径,结果屡屡失败。后来发现,对于数组元素的访问(如MyArray[5]),路径中不仅需要标签名段,还需要一个特殊的“索引段”,其编码方式也有特定规则。最终,通过对比抓包数据,才确定了正确的格式。

4.2 数据类型与字节序的“陷阱”

AB PLC内部的数据表示和我们的PC可能不同。

  • 字节序(Endianness):大多数现代AB PLC(基于Logix平台)使用小端字节序(Little-Endian)。也就是说,一个DINT(32位整数)值0x12345678,在网络上传输的字节顺序是0x78, 0x56, 0x34, 0x12。你在解析响应数据时,必须进行反转。
  • 数据类型代码(Data Type Codes):CIP协议为每种数据类型分配了一个16位的代码。例如,0xC1代表BOOL0xC2代表SINT0xC3代表INT0xC4代表DINT0xCA代表REAL。在读写标签时,有时需要在报文中指定这些类型代码。
  • 字符串类型:AB的字符串是带有长度字节和最大长度字节的结构。例如,一个STRING[40]类型,在内存中可能占用42个字节(2字节头部+40字节数据)。处理时需要特别小心。

提示:在构造写标签请求时,数据部分必须严格按照PLC期望的格式和字节序排列。一个有效的测试方法是,先用你的代码读取一个已知值的标签,看看解析出来的字节数组是什么样子,然后以此为模板构造写入请求。

4.3 会话管理与连接保活

CIP通信不是无状态的。它需要先建立一个会话(Session)。

  1. 注册会话(RegisterSession):发送命令码为0x65的封装报文,数据部分通常为空。PLC回复的报文中会包含一个4字节的会话句柄(Session Handle)。这个句柄在后续所有通信的封装头中都必须携带。
  2. 保活(Keep-Alive):TCP连接本身有保活机制,但CIP会话也可能超时。有些简单的客户端实现会忽略这一点,但对于需要长期稳定连接的场景,需要定期(例如每分钟)发送一个空的SendRRData命令(或特定的“未连接列表”请求)来保持会话活跃。更规范的做法是处理PLC可能发送的“连接超时”错误,并实现重连机制。
  3. 注销会话(UnRegisterSession):通信结束时,发送命令码为0x66的报文,并携带会话句柄,以礼貌地释放PLC端的资源。

4.4 性能优化:多标签读取与异步处理

频繁地单个读取标签效率极低,尤其是需要监控几十上百个标签时。AB CIP协议支持多标签读取(Read Tag Fragmented/Read Tag Service for multiple tags)

  • 原理:在一个Read Tag服务请求中,可以串联多个“请求路径”。每个路径指向一个标签。PLC会在一个响应报文中,按顺序返回所有标签的数据和状态。
  • 好处:极大减少了网络往返次数和报文开销,提升采集效率数倍。
  • 限制:单个请求的总大小是有限制的(受PLC型号和固件限制,通常几KB到几十KB),如果标签太多或数据太大,需要分批次读取。
  • 实现:在构造请求时,服务代码后的请求数据部分,不再是单个标签的信息,而是一个列表。列表中的每一项包含:该标签请求的数据长度、路径信息。响应部分也会是一个对应的列表。

对于高性能应用,还需要考虑使用异步I/O和非阻塞Socket,避免通信线程被阻塞,影响整体程序响应。

5. 从零搭建一个简易的AB PLC读取客户端(Python示例)

理论说了这么多,我们动手写一个最核心的片段,用Python的socket库实现一个读取单个DINT标签的客户端。这里省略了错误处理的完整代码,聚焦于核心流程。

import socket import struct import time import random class SimpleABClient: def __init__(self, plc_ip, plc_port=44818): self.plc_ip = plc_ip self.plc_port = plc_port self.session_handle = 0 self.sock = None def connect(self): """建立TCP连接并注册CIP会话""" self.sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) self.sock.settimeout(5.0) # 设置超时 self.sock.connect((self.plc_ip, self.plc_port)) self._register_session() def _register_session(self): """发送注册会话命令 (Command: 0x65)""" # CIP封装头 command = 0x65 length = 0 session_handle = 0 status = 0 sender_context = random.randbytes(8) # 随机生成8字节发送上下文 options = 0 # 构建封装报文 encap_header = struct.pack('<HHII8sI', command, length, session_handle, status, sender_context, options) self.sock.send(encap_header) response = self.sock.recv(1024) # 解析响应,获取会话句柄 (响应中的第5-8字节) if len(response) >= 24: resp_session_handle = struct.unpack_from('<I', response, 4)[0] self.session_handle = resp_session_handle print(f"Session registered. Handle: 0x{self.session_handle:08X}") else: raise Exception("Failed to register session.") def read_tag(self, tag_name): """读取一个标签(假设为DINT类型)""" # 1. 构建CIP服务路径 (这里是一个简化示例,实际路径更复杂) # 假设路径为: 端口 1, 然后标签名 # 实际中需要通过抓包分析确定准确的路径字节 path_segments = b'' # 示例:一个可能的路径构造(非真实,仅示意) # path_segments += struct.pack('BB', 0x20, 0x01) # Port 1 segment # path_segments += struct.pack('BB', 0x91, len(tag_name)) + tag_name.encode('ascii') # Symbolic segment # 这里我们简化,用一个从抓包中提取的固定路径字节数组代替 # 假设抓包得到读取`MyDINT`标签的路径数据是以下字节: # (这需要你从实际抓包中获取并替换) path_data = bytes.fromhex('20 06 24 01 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 91 07 4D 79 44 49 4E 54') # 示意 # 2. 构建CIP服务数据 (Read Tag Service) service_code = 0x4C # Read Tag request_path_size = len(path_data) // 2 # 路径长度以16-bit字为单位 # 请求数据:路径长度(字) + 路径数据 request_data = struct.pack('<H', request_path_size) + path_data # 完整的CIP数据项 cip_service = struct.pack('<B', service_code) + request_data cip_service_length = len(cip_service) # 3. 构建CIP连接项和数据项 # 连接地址项 (Null) item1_type = 0x0000 item1_length = 0x0000 item1 = struct.pack('<HH', item1_type, item1_length) # 连接数据项 (Connected Data) item2_type = 0x00B2 item2_length = cip_service_length item2 = struct.pack('<HH', item2_type, item2_length) + cip_service # 4. 构建完整的CIP数据部分 interface_handle = 0x00000000 timeout = 0x0000 item_count = 0x0002 cip_data = struct.pack('<IIHH', interface_handle, timeout, item_count, 0) + item1 + item2 cip_data_length = len(cip_data) # 5. 构建CIP封装头 encap_command = 0x6F # SendRRData encap_length = cip_data_length encap_status = 0x0000 sender_context = random.randbytes(8) encap_options = 0x00000000 encap_header = struct.pack('<HHII8sI', encap_command, encap_length, self.session_handle, encap_status, sender_context, encap_options) # 6. 组装并发送完整报文 full_packet = encap_header + cip_data self.sock.send(full_packet) # 7. 接收并解析响应 response = self.sock.recv(4096) # 简化解析:跳过封装头,找到CIP响应数据 # 响应封装头长度固定24字节 cip_resp_data = response[24:] # 解析CIP数据项... # 找到服务响应码 (应该是 0xCC = 0x4C+0x80) # 检查状态码 # 提取数据部分... # 这里省略详细的字节级解析过程 # 假设我们解析出数据部分是4字节的DINT (小端) data_bytes = cip_resp_data[-4:] # 简化,实际位置需计算 value = struct.unpack('<i', data_bytes)[0] # 小端有符号32位整数 return value def close(self): """注销会话并关闭连接""" if self.session_handle: # 发送UnRegisterSession命令 (0x66) unreg_packet = struct.pack('<HHII8sI', 0x66, 0, self.session_handle, 0, random.randbytes(8), 0) self.sock.send(unreg_packet) self.sock.recv(1024) if self.sock: self.sock.close() # 使用示例 if __name__ == "__main__": client = SimpleABClient("192.168.1.10") try: client.connect() tag_value = client.read_tag("MyDINT") # 需要替换为真实的路径构造逻辑 print(f"Tag value read: {tag_value}") except Exception as e: print(f"Error: {e}") finally: client.close()

重要说明:上面的path_data是硬编码的示例,绝对无法直接运行。你需要用自己抓包分析得到的真实路径字节串替换它。这个代码的价值在于展示了从建立会话、构造封装报文、发送请求到接收响应的完整框架。真正的开发工作,大部分都花在如何根据不同的标签名和结构,动态生成正确的path_data上。

6. 进阶话题与工具推荐

当你掌握了基础的单标签读写后,可能会需要更高级的功能。

6.1 处理数组和结构体

  • 数组:读取数组需要指定元素数量。在请求中,除了标签路径,还需要包含请求的元素个数。对于大型数组,可能需要进行分片读取(Fragmented Read)。
  • 结构体(UDT):读取整个结构体,可以将其视为一个字节块。读取结构体成员,则需要在标签路径后追加成员的偏移量或符号名路径。这通常通过添加额外的路径段来实现,例如标签名.成员名。抓包分析是理解其编码规则的不二法门。

6.2 写入操作与类型转换

写标签(Service Code0x4D)的报文结构与读标签类似,但在请求数据部分需要包含要写入的数据值。你必须确保数据值的字节序列与PLC中标签的数据类型完全匹配,包括字节序和填充字节。对于BOOL类型写入单个位,可能需要使用Write Tag Service的特定格式。

6.3 协议开发辅助工具

  1. Wireshark:必备,配合CIP/EtherNet/IP解析插件(默认可能已安装)效果更佳。
  2. Python库
    • pycomm3:一个功能相对丰富的第三方库,支持Logix PLC的CIP通信和部分PCCC。阅读其源码是很好的学习材料。
    • cpppo:另一个强大的库,专注于CIP协议,支持更底层的操作。
    • python-snap7:虽然主要针对西门子S7协议,但其设计思路和异步处理方式值得借鉴。
  3. C/C++库
    • libplctag:一个用C写的、支持多种PLC协议(包括AB)的库,性能很好,有各种语言的绑定。研究它的AB协议实现部分受益匪浅。
  4. 串口/网络调试助手:当开发DF1串口协议时,串口调试助手是验证帧格式和校验码的利器。

6.4 安全性考量

工业协议开发,安全往往被忽视,但至关重要。

  • 隔离:确保你的协议网关或采集程序运行在独立的、防火墙保护的网络区域。
  • 认证:新型的AB PLC支持基于角色的访问控制。你的客户端可能需要支持密码认证(CIP Security)。这涉及到更复杂的密钥交换和报文加密。
  • 输入验证:防止非法的标签名或数据导致PLC异常。
  • 资源管理:避免过快的请求频率拖垮PLC的通信处理能力。

自己动手开发AB PLC协议,是一条充满挑战但回报丰厚的路。它让你摆脱了对商业OPC服务器或特定驱动库的依赖,获得了对数据流的完全控制权,特别适合嵌入式网关、定制化数据平台或对成本敏感的项目。整个过程的核心在于耐心地抓包分析、严谨地对照文档(虽然AB的开放文档不多)、以及大胆地编码测试。从最简单的读取一个整数开始,逐步扩展到处理数组、结构体、多标签读写和错误重试,每一步的突破都会带来巨大的成就感。记住,你遇到的绝大多数问题,答案都藏在Wireshark捕获的那些绿色和红色的数据包里。

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询