ARTICLE · INTELLIGENCE

战地情报 · 详情页

来自尧图项目组的一线实战观察与深度解析

Python实现PROFINET协议栈:从DCP设备发现到周期IO读写

Python实现PROFINET协议栈:从DCP设备发现到周期IO读写

简介:PROFINET是工业自动化领域广泛应用的实时以太网协议,这份Python实现项目包可直接运行,适合希望在Python生态下开发工业通信功能的自动化工程师与协议学习者。包内共11个文件,以6个Python脚本为核心,覆盖DCP发现、协议解析、RPC通信、主站服务等关键模块;4个HTML模板提供设备发现与监控的简单前端页面,另含.gitignore配置。整个压缩包仅11KB,结构紧凑,便于阅读和二次开发。目前已有81人浏览学习。借助这套实现,开发者可直接搭建PROFINET通信基础平台,省去从零编写协议栈的繁琐过程;同时可作为学习范本,帮助理解协议工作原理,并方便与数据分析、远程诊断等Python工具链结合,提升工业自动化项目的集成效率与创新能力。

1. 用 Python 碰 PROFINET:这个 zip 解压后能干什么、不能干什么

“Python 实现的 PROFINET 协议,直接可以使用”这个 zip 包,我拿到手先做了三件事:解压、跑通设备扫描、抓包逐帧对照。结论是它能直接用于产线调试前的仿真验证、上位机协议测试、批量设备配置,而不是拿去当实时控制核来用。搞工控的人听到 Python 做 PROFINET,第一反应是“实时性不行”——这话一半对。PROFINET 的 RT 实时通道是数据链路层直通,操作系统调度根本不是它的主场;但如果你要的是设备发现、连接建立、周期 IO 读写、批量读写参数这些不拼微秒抖动的活,Python 这套栈反而比 C 代码快出结果。适合三类人:做协议测试要大量抓包分析的、写脚本批量配置现场设备的、以及想把 PROFINET 每个字节都看明白的新手。

2. 拆包看协议栈:DCP、DCE-RPC、周期 IO 是怎么组织在一起的

2.1 先看懂 PROFINET 那四层“玄学”结构

PROFINET 和普通 TCP/IP 应用完全不是一个套路。它底层是标准以太网,EtherType 用 0x8892,所有 RT 实时数据直接旁路掉 TCP/UDP,不走 IP 层。这就是很多人第一次抓包时蒙掉的原因——Wireshark 里看不到 502 端口、看不到 TCP 握手,满屏全是 0x8892 的裸帧。

这个协议栈在功能上分四层:链路层负责以太网帧收发,全是 raw socket 操作;DCP(Discovery and Configuration Protocol)管设备发现和 IP 分配,类似 DHCP 但不走 UDP;CMPD(Context Management)跑在 DCE-RPC 之上,负责建立应用关系 AR,你可以理解成“在设备和管理器之间开一条会话”;最后是周期 IO 数据,以固定帧间隔在链路层上直接传,不再经过 CMPD。

理解了这层结构,再看这个 zip 包的代码就顺了。它不是基于 Scapy 那种封包框架写的,而是直接用 struct 手拼字节流,好处是每个字节都在自己控制下,坏处是字节序、对齐、填充位一旦弄错,设备就静默不理你。我拆包时先看了 DCP 模块,因为这是整套协议栈里最容易跑通、也最能验证环境是否正常的部分。

2.2 解压后的文件结构:每个 py 文件负责哪一层

把 zip 解压后,核心模块都在 profinet 目录下,文件职责从名字上就能对上号,我按依赖顺序理了一下:

文件负责层次关键作用
profinet/utils.py基础工具字节序转换、MAC 地址格式化、CRC 计算
profinet/dcp.pyDCP 协议设备发现、IP 参数读写、设备名称分配
profinet/cmpd.pyCMPD/RPC建立 AR 应用关系、连接管理、连接释放
profinet/rt_frame.py周期 IO输入输出数据帧的组包与解析
profinet/gsdml.py设备描述解析 GSDML 文件,提取槽号、子槽号、模块 ID
examples/scan_devices.py示例脚本扫描网段内所有 PROFINET 设备
examples/read_io.py示例脚本建立连接后周期读取设备输入数据

DCP 在最底层,因为它不需要建立连接,发广播就能得到响应,适合拿来验证环境。GSDML 解析排在最后,因为实际调试中只有当你需要按模块配置槽号时才用得上。这个分层顺序和调试顺序是一致的,我是先跑dcp.py再往上走的。

2.3 从一次 DCP 发现请求看帧是怎么拼出来的

设备发现是 PROFINET 栈的地基。DCP Identify 请求是一个以太网广播帧,目标 MAC 是ff:ff:ff:ff:ff:ff,EtherType 是 0x8892,FrameID 是 0xfefe。我一般会先手动拼一帧,确认自己理解协议后再用包内封装好的函数。核心组包代码大致长这样:

import socket import struct import uuid def build_dcp_identify_request(mac_local: bytes) -> bytes: # 以太网头:目标广播地址,源地址填本地网卡 MAC eth_header = struct.pack("!6s6sH", bytes.fromhex("ffffffffffff"), mac_local, 0x8892) # PROFINET 帧头:frame_id = 0xfefe 表示 DCP 协议帧 profinet_header = struct.pack("!H", 0xfefe) # DCP 头:ServiceID=0x03(Identify) ServiceType=0x00(Request) # XID 用随机数,设备响应时原样带回,用于匹配请求与响应 xid = uuid.uuid4().int & 0xFFFFFFFF dcp_header = struct.pack("!BBHI", 0x03, 0x00, xid, 0x0000) # Identify 请求的 block 是空的:只含 option/suboption/padding 三个 0 字节 # block_type 0x0002 表示 Identify 请求,block_length 是后续数据的字节数 block = struct.pack("!HH3s", 0x0002, 0x0003, b"\x00\x00\x00") dcp_header += struct.pack("!H", len(block)) return eth_header + profinet_header + dcp_header + block

注意struct.pack格式串里的!,这表示网络字节序(大端)。PROFINET 的 DCP 头长度是固定的 10 字节:ServiceID 1 字节、ServiceType 1 字节、XID 4 字节、ResponseDelay 2 字节、DataLength 2 字节。DataLength 只算 block 部分的长度,不算 DCP 头自身。block 里BlockLength字段的值是 3,指的是后面 option、suboption、padding 三个字节,不是 block 头自身的长度——这个细节很容易被误解。

组好帧以后直接用 raw socket 发出去:

sock = socket.socket(socket.AF_PACKET, socket.SOCK_RAW, socket.htons(0x8892)) sock.bind(("eth0", 0)) sock.send(build_dcp_identify_request(b"\x00\x1a\x2b\x3c\x4d\x5e"))

同一交换机下的 PROFINET 设备收到后,会回一帧 FrameID 为 0x0100 的响应,里面带设备名、IP 地址、MAC、设备角色这些信息。这里AF_PACKET是 Linux 下的 raw socket,Windows 上需要用AF_INET加抓包驱动,或者直接用包内封装好的接口——这也是为什么我建议先在 Linux 上跑这套代码。

3. 跑起来:环境搭建、网卡选择与第一台设备发现

3.1 环境准备:Python 版本、依赖与抓包工具

先说结论:Python 3.8 及以上就行,这个包依赖的是标准库socket、struct、threading,不需要 pip 装任何第三方协议库。这是这个 zip 最省心的地方,很多标着“Python 实现”的协议包上来就让你装一大串依赖,实际一跑发现核心逻辑全在别人库里,自己只是包了层壳。

两个环境上的建议。第一,Windows 下如果要用 raw socket 抓自己发的帧,需要装 Npcap 或者 WinPcap,否则你只能发不能收;我一般直接在 Wireshark 的安装包里勾选 Npcap 一并装掉。第二,Linux 下跑scan_devices.py时需要 root 权限,因为 raw socket 需要 CAP_NET_RAW。如果提示权限不足,用sudo跑,不要在 venv 里折腾半天最后发现是权限问题。

另外提一个新手高频问题:解压完 zip 后直接在 IDE 里跑,报ModuleNotFoundError,十有八九是profinet目录没在 Python 的搜索路径里。在项目根目录下运行脚本,或者把根目录加到PYTHONPATH,不要只双击examples里的文件,否则 import 路径全部断掉。

3.2 扫描网段:跑通第一台设备

包内自带的examples/scan_devices.py是整套代码的入口,我拿到任何一台 PROFINET 设备,第一件事就是跑它确认协议栈完好。命令很简单:

cd profinet-python sudo python examples/scan_devices.py --interface eth0 --timeout 3

--interface指定监听网口,注意不是 IP,是网卡的设备名;Linux 下用ip addr查看,Windows 下用wlan0或以太网适配器名。--timeout是发送 DCP 请求后等待响应的时间,单位秒,设 3 秒就够。脚本内部会向255.255.255.255发广播 DCP Identify,然后开始监听 0x0100 响应帧。

如果机器有多个网卡,尤其是笔记本,务必先确认当前网卡直连着目标设备。我踩过一台工控机有两个网口的坑,一个接着 PLC 一个接着办公网,扫描时报的还是办公网卡,自然什么都扫不到。跑通以后输出大致是这样的:

Device Found: MAC Address: 00:11:22:33:44:55 Name: plc-io-controller IP Address: 192.168.0.10 Role: IO Controller

这个输出比你自己抓包解析省事得多。如果显示No PROFINET device found,先别去改协议代码,99% 是网卡选错、广播被防火墙拦了、或者设备不在同一个二层网段。PROFINET 的 DCP 是链路层广播,跨三层路由是找不到的。

3.3 用 Wireshark 验证你发的帧长什么样

跑通扫描之后我强烈建议做一次抓包验证,确认你理解的和协议栈实际发的是一致的。Wireshark 过滤条件这么写:

ether proto 0x8892 && frame[14:2] == 0xfefe

这个过滤语句的含义:ether proto 0x8892只保留 PROFINET 以太网类型,frame[14:2]跳过以太网头后取 FrameID 字段,等于 0xfefe 的就是 DCP 请求帧。抓包命中后,展开 PROFINET IO 层,你会看到 ServiceID、ServiceType、XID 这些字段跟你代码里struct.pack的参数一一对应。

这里有个常见误用:很多人习惯用dcp作为 Wireshark 过滤关键字,但旧版 Wireshark 的 DCP 解析器只在 PROFINET 协议启用时才生效,直接用eth.proto == 0x8892反而更稳妥。另外,抓不到自己发的帧时,检查 Wireshark 的“捕获过滤器”,不要让它把广播帧过滤掉了。做完这一步,底层环境就算验证完毕,可以往上走连接建立了。

4. 建立连接与周期 IO 读写:从发现设备到读写数据

4.1 AR、CR、IOD:PROFINET 连接不是 TCP 握手

PROFINET 建立连接的过程比 TCP 三次握手复杂一个量级。两个设备之间先建立 AR(Application Relation,应用关系),相当于在这两个设备之间签一个“会话协议”;AR 里又包含至少两条 CR(Communication Relation,通信关系),一条是输入方向(设备到控制器),一条是输出方向(控制器到设备);每个 CR 里再细化成具体的 IOD(Input/Output Data object,输入输出数据对象),也就是每个槽位上的数据块。

这个过程是通过 DCE-RPC 调 CMPD 的Connect服务实现的。流程大致是:先做 RPC bind 建立 RPC 会话,然后发Connect请求,请求里带 AR 属性、CR 数量和每个 CR 对应的槽号信息;设备收到后校验这些参数是否与自己的 GSDML 配置匹配,匹配则返回Connect响应,里面带上实际建立的 CR 的帧 ID 和周期参数。从这时起,周期 IO 数据开始按约定帧间隔在链路上跑。

这就是为什么直接用 Python 循环读数据会踩坑——PROFINET 的周期 IO 不是“读一次返回一次”的请求响应模式,而是设备持续按固定周期往外发帧,你只是去监听并过滤出属于这个 CR 的帧。逻辑反过来了,要在接收侧做帧过滤和序列号校验。

4.2 用包内接口建立连接并周期读数据

包内把上述流程压缩成了一个ProfinetDevice类,我实际用起来三步就能接上:

from profinet import ProfinetDevice dev = ProfinetDevice( interface="eth0", ip="192.168.0.10", slot=1, subslot=1, timeout_ms=3000, ) dev.connect() while True: input_data = dev.read_inputs(timeout_ms=1000) if input_data: print("slot1 data:", input_data.hex())

ProfinetDevice构造器里的ip是目标设备的 IP,slot和subslot是你要读的模块位置,timeout_ms是连接阶段等待 RPC 响应的超时时间。connect()内部做了四件事:发起 DCE-RPC 的 bind 请求、构造并发送 CMPD Connect、等待设备回 Connect 响应、拿到响应里分配的周期帧 ID 后启动监听线程。这就是为什么connect()之后可以反复调read_inputs()而不用自己管 RPC 状态——监听线程会持续接收属于本 AR 的周期帧并存进缓冲区。

read_inputs(timeout_ms=1000)是从缓冲区里取最新一帧输入数据,1000表示最多等 1 秒,超时返回None。输出是原始字节,按你在 GSDML 里定义的格式去解析。需要注意,这里读到的数据是设备发给控制器的输入数据,如果你要写输出数据到设备,需要调用write_outputs(),参数和输入方向是对应的。

4.3 读写设备参数:Record Data 接口的用法

周期 IO 跑通之后,你会遇到另一个高频需求:读取或修改设备的参数,比如改设备名、改 IP、读诊断信息。这些用周期 IO 做不了,要走 Record Data 通道,本质还是 DCE-RPC 调 CMPD 的Read/Write服务。包内封装如下:

# 读取设备参数索引 0x0020(常用作模块标识) param = dev.read_record(slot=1, subslot=1, index=0x0020, length=128) print("record:", param.hex()) # 写设备 IP 参数:0x002E 是 IP 参数块索引 new_ip = bytes([192, 168, 0, 50, 255, 255, 255, 0]) dev.write_record(slot=1, subslot=1, index=0x002E, data=new_ip)

index是参数索引,不同设备含义不同,基础模块的标识读取是 0x0020,IP 配置是 0x002E 附近,具体的以 GSDML 文件里的<RecordParameter>节点为准。length是期望读回的最大字节数,按设备实际情况给,给小了可能报错。写操作里的data是原始字节,注意分隔符是255, 255, 255, 0这种点分十进制的十进制表示,不是字符串。

写参数之前一定要确认设备处于可写状态,有些设备在 IO Controller 占用连接时不允许第三方写参数。你读没问题,写直接被设备拒绝,这是正常的,不是代码 bug。我一般会先读一遍当前值,备份原始参数,再写新的,改错了也好退回去。

5. 避坑记录:字节序、防火墙、槽号与 XID 的五个翻车现场

5.1 字节序颠倒了,抓包没毛病就是读不对

现象:DCP 扫描能发现设备,设备名和 MAC 都对,但 IP 地址读出来是0.10.168.192这种倒序值。

原因:PROFINET 协议全程大端字节序,但 DCP 响应里的 IP 地址块是一次性pack成 32 位整型的,抓包软件会帮你解析成可读格式所以看不出来,你自己的解析代码如果用struct.unpack("!I")还好,一旦用了默认字节序的系统int.from_bytes()不带byteorder,4 字节就全部反过来了。

解决:所有多字节字段统一用struct.unpack("!…"系列,!强制网络字节序。如果是自己拼的 block,先写一个小助手函数把所有超过 1 字节的字段都过一遍,别心存侥幸觉得这里用不上大端。我基本每个项目的utils.py里都会放一个to_u16/to_u32,专门干这个。

5.2 扫描全是空:广播帧被系统防火墙拦在半路

现象:同一台机器上 Wireshark 能抓到设备的 DCP 响应,但scan_devices.py一个设备都扫不到,超时后报空结果。

原因:脚本里发送请求后,需要监听 0x8892 类型的链路层帧,Windows 防火墙默认拦截外部入站的 raw 帧,而 Wireshark 用的是 Npcap 的驱动层抓包接口,绕过了防火墙,所以一个能抓到、一个抓不到。这不是代码问题,是操作系统权限边界问题。

解决:Windows 下直接把专用调试网卡设为“专用网络”,并允许 Python 通过防火墙入站规则,更粗暴的方式是临时关闭防火墙跑测试,跑完再打开。Linux 下基本没有这个问题,但 Ubuntu 的ufw会拦,用sudo ufw disable或者把对应网卡设为信任。我现在的习惯是:调试环境一律用物理隔离的专用网段,不跟办公网混在一起。

5.3 连接成功但周期数据全是零:槽号配置和 GSDML 对不上

现象:connect()返回正常,read_inputs()也能取到数据,但数据永远是00 00 00…,且设备面板上一直报D_Mod_Missing模块缺失。

原因:PROFINET 的槽号从 0 开始,但每个设备 GSDML 里定义的模块位置可能完全不同。很多人默认槽 1 子槽 1 就是第一个数据块,结果设备的第一个实际数据模块在槽 2、子槽 1,或者槽 1 下根本没有已组态的模块。连接建立时设备会校验槽号,如果槽号不对,连接照常建立但数据路径是空的,于是读到全零。

解决:先用包内的profinet.gsdml模块解析设备 GSDML,把<Module ID="…">下的SlotNumber逐个列出来,确认实际数据槽位。另一个快速验证法是read_record(slot=1, subslot=1, index=0x0020)读模块标识,能读到说明这个槽上有模块,读不到就是空槽。我每次接新设备都先把槽位表拉出来,打印在工作笔记里,后续调参不用重复翻文档。

5.4 XID 用固定值,设备“间歇性失聪”

现象:DCP 请求第一次发出去设备秒回,连续跑第二次、第三次,设备完全没反应,重启脚本又好了。

原因:XID 字段在 PROFINET 协议里用于匹配请求和响应,同时起到防重放作用。代码里如果写死xid = 0x00000001,设备会认为这是重复的旧请求,直接丢弃不响应。Wireshark 看着包没问题,但你不改 XID,它就是翻车。

解决:每次构造请求时生成一个新的 XID,最简单的做法是用uuid.uuid4().int & 0xFFFFFFFF,或者维护一个全局计数器自增。这个坑不会导致所有请求都失败,而是“第一次好、第二次坏”这种极具迷惑性的表现,排查时最容易被忽略。

5.5 DCE-RPC 的长度对齐:连接请求永远超时

现象:发送 CMPD Connect 请求后一直等不到响应,Wireshark 里能看到请求帧,但设备就是不回包。

原因:DCE-RPC 的 NDR 编码要求所有对象按 4 字节对齐,很多手写 RPC 组包代码在拼接完参数字段后忘了补填充字节,导致整个请求的DataLength和后端解析出来的长度不一致。设备解析到一半发现长度字段对不上,直接当作非法报文丢弃。

解决:在拼完一个 RPC 参数块后,计算当前长度并补上(4 - len % 4) % 4个\x00字节,然后再填长度字段。还有一个更隐蔽的点:Connect请求里的 AR 标识 UUID 本身就是 16 字节,这个长度天然对齐,但后面的 CR 数量字段是 2 字节,接下来又跟一个 2 字节的填充字段,手拼时非常容易错位。我一般会让代码打印出整帧 hex,对照 Wireshark 里的解析结果逐字节数过去,哪个位置错了立刻就能看出来。

6. 进阶验证:抓包对照与软 PLC 端到端联调

6.1 抓包对照法:把包的时序和标准帧逐字段对齐

环境跑通、连接建立、数据能读之后,别急着收工。我每次接触新版本协议栈都会做一次“抓包金标对照”:先抓一帧标准设备之间的正常通信,再用自己的代码替代一端,抓同类型的帧放在 Wireshark 里逐字段对比。对照重点有三个:以太网头里的 FrameID 是否一致(DCP 是 0xfefe、周期数据各方向有各自段);CMPD Connect 请求里的 AR 类型字段和 CR 数量;周期帧的序列号是否连续、循环计数是否按周期递增。

Wireshark 的 PROFINET 解析器会帮你展开所有字段,你只需要盯着标红的部分看。如果自己的帧和标准帧之间只有 IP 地址、MAC 地址不同,其他完全一致,那这个协议栈基本可以放心用;如果有字段顺序或长度的差异,去改代码而不是去怀疑对端设备。这个习惯帮我省掉过无数次无意义的现场调试。

6.2 用软 PLC 做端到端联调:没有实体设备也能验证协议栈

如果你手边暂时没有实体 PROFINET 设备,可以在电脑上装一个软 PLC(比如 CODESYS 的仿真运行时)作为 IO Controller,然后把 Python 栈当成 IO Device 从站去连它。CODESYS 里组态一个 PROFINET 从站时,会要求你配置设备名和 IP,DCP 发现阶段用的就是这套信息。把ProfinetDevice启动为一个虚拟从站,控制器会尝试建立连接,你可以在控制器的诊断界面看连接状态和 IO 数据刷新。

这个联调的价值在于,它能验证你自己的协议栈在真实控制器眼中的行为是否符合预期,而不仅是“和标准帧对得上”。控制器对从站的要求比脚本严格得多:连接超时、序列号重复、周期抖动过大,都会触发诊断报警。我第一次用软 PLC 联调时,控制器报了一堆D_Mod_Missing,排查下来就是槽号配置和组态不一致,先在仿真环境里调完,再到现场一次通过——从那以后我每次接新设备,都会先用这套流程把协议栈和配置过一遍,再上真机,用抓包和仿真器双重验证后才敢动现场设备。希望帮到你。

本文还有配套的精品资源,点击获取

RELATED READING

延伸阅读

更多一线实战笔记与深度复盘,助您持续精进