1. 项目缘起与整体设计思路1.1 为什么偏偏是FINS协议搞工业自动化的朋友对FINS协议应该不陌生它是欧姆龙系列PLC上位机通信的经典协议全称是Factory Interface Network Service。很多做设备数据采集、MES对接、产线监控的项目绕不开要和欧姆龙PLC打交道。市面上现成的FINS通信库不是没有但要么是C#写的、要么封装得太重、要么在Linux环境下跑不起来真正想用Python做一套轻量、可控、方便二次开发的TCP服务端还是得自己动手。这个项目的目标很明确用Python实现一个FINS协议的TCP服务端让上位机系统能够通过标准FINS指令读写PLC的DM区、CIO区、定时器、计数器等数据。说白了就是自己造一个翻译官把Python世界的数据请求翻译成PLC能听懂的FINS报文再把PLC返回的报文解析成Python能处理的结构。适合谁来参考如果你正在做工业数据采集、设备联网、产线数字化改造手上有欧姆龙PLC又希望用Python技术栈来搭建服务端那这篇内容就是写给你的。不需要你是协议专家但至少要懂一点TCP通信和字节操作的基础。1.2 整体架构怎么搭我一开始的思路就很清晰服务端要能同时处理多个客户端的连接请求每个连接独立维护会话状态收到FINS报文后解析命令、执行对应操作、组装响应报文返回。整体分成四层网络层基于socketserver或asyncio实现TCP监听和连接管理负责字节流的收发。协议层处理FINS报文的头部解析、命令码识别、地址解析、数据编解码。业务层根据解析出的命令执行具体操作比如读DM区、写CIO区、读定时器当前值等。数据层维护一个模拟的PLC内存区域或者对接真实的PLC设备。为什么选TCP而不是UDPFINS协议本身支持UDP和TCP两种承载方式。UDP速度快、开销小但不可靠丢包了就得靠上层重传TCP有连接保障、顺序保障对于数据采集这种要求准确性的场景更合适。而且TCP服务端可以维持长连接减少反复建连的开销。实测下来在局域网环境下TCP的延迟完全够用单次读写基本在10ms以内。1.3 协议帧结构拆解FINS/TCP的报文结构分两层外层是FINS/TCP头内层是FINS帧。外层头固定16字节结构如下偏移长度字段说明04Magic固定为FINS44Length后续数据长度84Command命令码0x00000000表示FINS帧收发124ErrorCode错误码164Reserved保留内层FINS帧的头部是10字节的ICF、RSV、GCT、DNA、DA1、DA2、SNA、SA1、SA2、SID后面跟命令码2字节和具体数据。这个结构必须吃透否则解析的时候字节偏移一错后面全乱。注意FINS/TCP头的Length字段是网络字节序大端而FINS帧内部有些字段是的小端有些是大端这个坑我后面会专门讲。2. 核心细节解析与实操要点2.1 字节序处理最容易翻车的地方FINS协议里字节序不统一这是新手最容易踩的坑。FINS/TCP头的Length、Command、ErrorCode都是大端序网络字节序但FINS帧内部的命令码、地址、数据长度等字段有的是大端有的是小端具体取决于字段定义。我的做法是所有网络传输的整数字段统一用struct.pack(I, value)处理确保大端。FINS帧内部则根据具体字段查手册确认。比如读DM区的命令码是0x0101在FINS帧里就是b\x01\x01直接按字节写就行不用转换。但数据长度字段是2字节大端就得用struct.pack(H, length)。import struct def pack_fins_tcp_header(length, command0, error_code0): 打包FINS/TCP外层头16字节 return struct.pack(4sIII, bFINS, length, command, error_code) def pack_fins_frame(icf, rsv, gct, dna, da1, da2, sna, sa1, sa2, sid, command, data): 打包FINS帧 header struct.pack(BBBBBBBBBB, icf, rsv, gct, dna, da1, da2, sna, sa1, sa2, sid) cmd struct.pack(H, command) return header cmd data实测下来只要字节序对了报文拼装基本不会出问题。建议每拼一个字段就打印一下十六进制对照手册核对比事后调试省事得多。2.2 地址解析DM区、CIO区的编码规则FINS协议里访问PLC内存区域需要把区域类型和地址编码成特定格式。以DM区为例区域代码是0x82地址是2字节大端。CIO区代码是0x30定时器是0x09计数器是0x09和定时器共用区域代码通过命令码区分。读DM区从D100开始读10个字的报文数据部分是这样的def build_read_dm_command(start_addr, count): 构建读DM区命令的数据部分 area_code 0x82 # DM区 addr_bytes struct.pack(H, start_addr) count_bytes struct.pack(H, count) return struct.pack(B, area_code) addr_bytes count_bytes这里有个细节地址是从0开始还是从1开始欧姆龙PLC的DM区地址通常从D0开始但有些文档写的是从D1开始。我的经验是实际通信时按0基地址处理读D100就传100读D0就传0。如果发现读出来的数据偏移了一位那就是基地址搞错了改成1基再试。提示不同型号的PLC区域代码可能略有差异建议先查对应型号的通信手册确认。CP系列和CJ系列的DM区代码都是0x82但CS系列有些型号不一样。2.3 会话管理与连接保持FINS/TCP支持两种连接模式短连接和长连接。短连接是每次请求都新建TCP连接用完就关长连接是建立一次连接后持续收发。工业现场更推荐长连接因为反复建连的开销在高速采集场景下很可观。服务端这边我用socketserver.ThreadingTCPServer来实现每个客户端连接起一个独立线程处理。每个连接维护一个会话对象记录客户端地址、连接时间、最后活动时间、已处理的请求数等。如果超过一定时间没有活动就主动断开释放资源。import socketserver import threading import time class FinsSession: def __init__(self, client_address): self.client_address client_address self.connect_time time.time() self.last_active time.time() self.request_count 0 def touch(self): self.last_active time.time() self.request_count 1 class FinsHandler(socketserver.BaseRequestHandler): def handle(self): session FinsSession(self.client_address) while True: try: data self.request.recv(4096) if not data: break session.touch() response self.process_fins_packet(data) if response: self.request.sendall(response) except ConnectionResetError: break这里有个坑recv不保证一次收到完整报文。FINS/TCP头里虽然有Length字段但TCP是流式协议可能分多次到达。稳妥的做法是先收16字节头解析出Length再按Length收剩余部分。我一开始图省事直接recv(4096)结果在数据量大或者网络抖动时经常收到半截报文解析直接崩。2.4 模拟PLC内存的设计如果手头没有真实PLC或者想在开发阶段先跑通逻辑可以做一个模拟的PLC内存。用一个字典维护各区域的数据class MockPlcMemory: def __init__(self): self.dm_area [0] * 32768 # DM区32768个字 self.cio_area [0] * 6144 # CIO区 self.timer_area [0] * 4096 self.counter_area [0] * 4096 def read_dm(self, start, count): return self.dm_area[start:startcount] def write_dm(self, start, values): for i, v in enumerate(values): self.dm_area[start i] v模拟内存的好处是开发调试阶段不依赖硬件单元测试也好写。等逻辑跑通了再把数据层换成真实PLC的通信接口就行。这种分层设计让整个项目可测试性大大提升。3. 实操过程与核心环节实现3.1 环境准备与依赖安装Python版本建议3.8以上太老的版本对asyncio和类型注解支持不好。依赖方面核心只用标准库的socket、struct、threading、socketserver不需要额外装第三方包。如果要做异步版本可以用asyncio也是标准库。python --version # Python 3.8.10 或更高项目目录结构建议这样组织fins_server/ ├── main.py # 入口启动服务 ├── fins_protocol.py # 协议解析与打包 ├── fins_handler.py # 业务处理 ├── plc_memory.py # 模拟内存 └── tests/ └── test_protocol.py # 单元测试为什么这么分协议解析和业务处理分开方便单独测试协议层。模拟内存独立出来将来换真实PLC只改这一个文件。这种模块化设计在后期维护时优势明显。3.2 服务端启动与监听主入口很简单绑定端口、启动服务、等待连接import socketserver from fins_handler import FinsHandler HOST 0.0.0.0 PORT 9600 if __name__ __main__: server socketserver.ThreadingTCPServer((HOST, PORT), FinsHandler) server.allow_reuse_address True print(fFINS TCP Server listening on {HOST}:{PORT}) try: server.serve_forever() except KeyboardInterrupt: server.shutdown()端口选9600是FINS/TCP的常用默认端口但实际项目里可以根据需要改。allow_reuse_address True这个设置很重要否则服务重启时可能报Address already in use得等系统释放端口才能再启动调试阶段特别烦人。3.3 完整报文处理流程收到一个FINS/TCP报文后处理流程分五步解析外层头读前16字节校验Magic是否为FINS取出Length。读取完整FINS帧根据Length读取剩余字节。解析FINS帧头取出ICF到SID这10字节再取命令码。分发命令根据命令码调用对应的处理函数比如0x0101走读DM区0x0102走写DM区。组装响应把处理结果按FINS格式打包加上外层头返回。def process_fins_packet(self, raw_data): if len(raw_data) 16: return None magic, length, command, error_code struct.unpack(4sIII, raw_data[:16]) if magic ! bFINS: return None fins_frame raw_data[16:16length] if len(fins_frame) 12: return None # 解析FINS帧头 icf, rsv, gct, dna, da1, da2, sna, sa1, sa2, sid struct.unpack(BBBBBBBBBB, fins_frame[:10]) fins_command struct.unpack(H, fins_frame[10:12])[0] data fins_frame[12:] # 分发处理 if fins_command 0x0101: result self.handle_read_dm(data) elif fins_command 0x0102: result self.handle_write_dm(data) else: result self.build_error_response(0x0001) # 不支持的命令 # 组装响应 resp_frame struct.pack(BBBBBBBBBB, icf, rsv, gct, dna, da1, da2, sna, sa1, sa2, sid) resp_frame struct.pack(H, fins_command | 0x0100) # 响应命令码请求命令码|0x0100 resp_frame result resp_header struct.pack(4sIII, bFINS, len(resp_frame), 0, 0) return resp_header resp_frame这里有个关键点FINS响应帧的命令码是请求命令码加上0x0100。比如读DM区请求是0x0101响应就是0x0201。这个规则必须记住否则客户端收到响应会认不出来。3.4 读DM区的完整实现读DM区的请求数据部分是区域代码1字节 起始地址2字节 读取字数2字节。响应数据部分是结束代码2字节 读取到的数据每字2字节。def handle_read_dm(self, data): if len(data) 5: return struct.pack(H, 0x0002) # 数据长度错误 area_code data[0] start_addr struct.unpack(H, data[1:3])[0] count struct.unpack(H, data[3:5])[0] if area_code ! 0x82: return struct.pack(H, 0x0003) # 区域代码错误 values self.plc_memory.read_dm(start_addr, count) result struct.pack(H, 0x0000) # 正常结束 for v in values: result struct.pack(H, v) return result实测下来读100个字的数据响应报文大概200多字节一次TCP往返就能完成。如果客户端要读的数据量很大比如几千个字建议分批读每次不超过500个字避免报文过大导致分片。3.5 写DM区的完整实现写DM区的请求数据部分是区域代码1字节 起始地址2字节 写入字数2字节 写入数据每字2字节。响应数据部分只有结束代码2字节。def handle_write_dm(self, data): if len(data) 5: return struct.pack(H, 0x0002) area_code data[0] start_addr struct.unpack(H, data[1:3])[0] count struct.unpack(H, data[3:5])[0] if area_code ! 0x82: return struct.pack(H, 0x0003) values [] for i in range(count): offset 5 i * 2 if offset 2 len(data): return struct.pack(H, 0x0002) values.append(struct.unpack(H, data[offset:offset2])[0]) self.plc_memory.write_dm(start_addr, values) return struct.pack(H, 0x0000)写操作有个注意事项一定要先校验数据长度是否和声明的字数匹配否则可能写入不完整的数据。我踩过一次坑客户端发的报文里声明写10个字但实际只带了8个字的数据服务端没校验就直接写结果后两个地址被写入了脏数据。3.6 客户端测试脚本服务端写好了得有个客户端来验证。用Python写个简单的测试客户端import socket import struct def build_read_dm_request(start_addr, count, sid1): fins_frame struct.pack(BBBBBBBBBB, 0x80, 0x00, 0x02, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, sid) fins_frame struct.pack(H, 0x0101) fins_frame struct.pack(B, 0x82) fins_frame struct.pack(H, start_addr) fins_frame struct.pack(H, count) header struct.pack(4sIII, bFINS, len(fins_frame), 0, 0) return header fins_frame def parse_read_response(raw_data): fins_frame raw_data[16:] end_code struct.unpack(H, fins_frame[12:14])[0] if end_code ! 0: return None values [] data_part fins_frame[14:] for i in range(0, len(data_part), 2): values.append(struct.unpack(H, data_part[i:i2])[0]) return values client socket.socket(socket.AF_INET, socket.SOCK_STREAM) client.connect((127.0.0.1, 9600)) client.sendall(build_read_dm_request(100, 10)) response client.recv(4096) values parse_read_response(response) print(DM100-DM109:, values) client.close()这个测试脚本能跑通说明服务端的基本收发逻辑没问题。建议先写测试脚本再写服务端这样开发过程中随时能验证比等服务端全写完再联调效率高得多。4. 常见问题与排查技巧实录4.1 报文解析错位问题速查现象可能原因排查方法解析出的命令码是乱码字节序搞反了打印原始字节对照手册确认大小端读到的数据偏移一位地址基址搞错尝试0基和1基两种方式响应客户端不识别响应命令码没加0x0100检查响应命令码计算逻辑收到半截报文recv没按Length收完整先收16字节头再按Length收剩余连接频繁断开超时设置太短调整会话超时时间4.2 连接超时与心跳处理工业现场网络环境复杂有时候客户端会异常断开但服务端不知道导致连接资源泄漏。我的做法是给每个会话加一个最后活动时间戳起一个后台线程定期扫描超过5分钟没活动的连接主动关闭。def cleanup_sessions(self): while True: time.sleep(60) now time.time() for session in list(self.sessions): if now - session.last_active 300: self.sessions.remove(session) # 关闭对应连接5分钟这个值可以根据实际场景调整。如果客户端采集频率高可以设短一点如果只是偶尔查询可以设长一点。关键是别让死连接一直占着资源。4.3 大数据量读写的分片策略FINS协议单次读写的数据量有限制一般建议不超过500个字。如果要读几千个字得在应用层做分片。我的做法是封装一个read_dm_bulk函数内部循环调用单次读每次读500个字拼起来返回。def read_dm_bulk(self, start_addr, total_count, chunk_size500): result [] remaining total_count addr start_addr while remaining 0: count min(chunk_size, remaining) values self.read_dm_once(addr, count) result.extend(values) addr count remaining - count return result分片大小500是我实测下来比较稳妥的值太小了往返次数多效率低太大了报文容易分片。当然具体值可以根据网络MTU调整一般保证单次报文不超过1400字节比较安全。4.4 错误码处理与日志记录FINS协议有一套标准的结束代码常见的有0x0000正常结束0x0001命令不支持0x0002数据长度错误0x0003区域代码错误0x0004地址越界0x0005数据错误服务端每个请求处理完都要返回对应的结束代码客户端根据结束代码判断操作是否成功。日志方面建议记录每个请求的客户端地址、命令码、地址范围、处理结果、耗时。出问题时翻日志能快速定位。import logging logging.basicConfig(levellogging.INFO, format%(asctime)s - %(levelname)s - %(message)s) def log_request(self, session, command, addr, count, end_code, elapsed): logging.info(fClient{session.client_address} CMD0x{command:04X} fAddr{addr} Count{count} EndCode0x{end_code:04X} fElapsed{elapsed*1000:.2f}ms)4.5 性能优化经验单线程处理请求在低并发场景下够用但如果同时有几十个客户端连接就得考虑并发优化。我用ThreadingTCPServer实测下来20个客户端并发读平均响应时间在15ms左右完全能满足大部分采集场景。如果追求更高性能可以改用asyncio做异步IO单线程就能处理大量并发连接。但异步代码调试起来比多线程麻烦建议先用多线程版本跑通确有性能瓶颈再改异步。提示Python的GIL在多线程IO密集型场景下影响不大因为大部分时间花在等待网络IO上真正CPU计算的时间很少。所以多线程方案在FINS服务端这种IO密集型场景下是够用的。4.6 真实PLC对接注意事项如果要把模拟内存换成真实PLC有几个点要注意网络配置确保服务端和PLC在同一网段PLC的FINS/TCP端口默认9600但有些型号可以改。节点地址FINS帧里的DNA、DA1、DA2、SNA、SA1、SA2这些节点地址字段要和PLC的实际配置匹配。一般DNA0、DA1PLC节点号、DA20、SNA0、SA1服务端节点号、SA20。超时设置真实PLC的响应时间比模拟内存慢socket超时要设长一点建议3秒以上。重试机制网络抖动时请求可能失败建议加2-3次重试重试间隔100ms。我踩过的一个坑是节点地址配错了报文发出去PLC根本不回排查了半天才发现是DA1设成了0而实际PLC节点号是1。所以对接真实设备前一定先把PLC的通信参数确认清楚。4.7 单元测试怎么写协议解析这种逻辑单元测试特别重要。我一般用unittest写几组测试用例覆盖正常读写、边界地址、错误命令等场景import unittest from fins_protocol import build_read_dm_command, parse_fins_frame class TestFinsProtocol(unittest.TestCase): def test_read_dm_command(self): cmd build_read_dm_command(100, 10) self.assertEqual(cmd[0], 0x82) self.assertEqual(struct.unpack(H, cmd[1:3])[0], 100) self.assertEqual(struct.unpack(H, cmd[3:5])[0], 10) def test_invalid_magic(self): raw bXXXX b\x00 * 12 result parse_fins_frame(raw) self.assertIsNone(result)测试用例不用多但关键路径一定要覆盖。每次改完协议代码跑一遍测试能避免很多低级错误。5. 后续扩展方向与个人体会这个服务端目前实现了基础的DM区读写后续可以扩展的方向不少。比如支持CIO区、定时器、计数器的读写支持强制置位/复位操作支持多客户端并发的高级调度甚至可以把服务端做成一个网关同时对接多台PLC对外提供统一的REST接口。我个人在实际操作中的体会是FINS协议本身不复杂难的是细节。字节序、地址基址、命令码规则、响应格式每一个细节错了都会导致通信失败。最好的办法就是边写边用测试脚本验证每实现一个命令就立刻测通别攒到最后一起调。另外把协议解析和业务处理分开将来换PLC型号或者加新命令时改动范围可控不会牵一发动全身。最后分享一个小技巧调试FINS报文时用Wireshark抓包过滤条件设成tcp.port 9600能看到完整的请求和响应报文。对照抓到的字节和你的代码输出哪里对不上一眼就能看出来比盲猜高效得多。