1. 为什么“旧上位机不肯改”是工业现场最真实的困境你手头有一套运行了八年的上位机系统界面还是XP风格的ActiveX控件数据库用的是Access MDB通信协议硬编码在Delphi写的DLL里——它不崩溃、不报警、不掉线就是死活不支持新买的声光语音终端。供应商说“加个Modbus TCP接口可以加钱三万起工期两个月还得停机三天。”车间主任叼着烟说“停机产线一小时损失两万你算算账。”最后这句话不是推脱是现实。这不是技术落后的问题是存量系统与增量设备之间的物理性割裂。旧上位机不是“不能改”而是“改不起”没有源码、没有文档、没有维护人哪怕有改一处可能触发二十年前埋下的内存泄漏逻辑更关键的是它早已嵌入MES报工、ERP工单、DCS联锁逻辑中牵一发而动全身。这时候任何要求“重写上位机”或“说服甲方升级”的方案本质上都是在说“请先推倒整栋楼再盖新厨房”。而声光语音终端——比如某国产型号NX-CIF105或是进口品牌如Honeywell XPS系列——它们出厂只认标准协议Modbus TCP、TCP字节帧非结构化二进制流、或HTTP REST API。它们不理解你Delphi DLL里那个叫SendCmd(0x1A, 0x03, 0x01)的私有函数也不认识你Access数据库里存着的“报警等级映射表”。它们只等一个IP端口然后收一串字节、回一串字节。所以“原生TCP字节帧接入”不是炫技是唯一可行的缝合术不碰旧系统一根线不改一行代码不申请一次停机在旧上位机和新终端之间架一座字节级的翻译桥。这座桥不处理业务逻辑只做三件事监听旧上位机发出的原始TCP指令、按终端能懂的格式重组字节帧、把终端返回的响应原样塞回旧上位机的socket缓冲区。它像一个哑巴翻译——听不懂双方语言但能把A说的“开门”准确转成B能执行的0x01 0x05 0x00 0x00 0xFF 0x00 0x8C 0x3A再把B回的0x01 0x05 0x00 0x00 0xFF 0x00 0x8C 0x3A原样传回去。关键词里的“TCP”“字节帧”“声光语音终端”“Modbus TCP”“Python”不是随意堆砌的技术标签而是这个缝合术的四根承重柱TCP是底层传输载体必须处理粘包、半包、连接保活、异常断连重试字节帧是数据形态意味着没有JSON/XML的容错性一个bit错整帧废必须严格按终端手册定义的起始符、长度域、校验方式解析声光语音终端是目标设备它的协议文档往往只有一页PDF字段说明用中文夹杂英文缩写比如“0x02命令字启动语音播报含音量0~100”但没写音量值是放在第3字节还是第4字节Modbus TCP是常见替代路径但很多终端尤其是国产中低端型号为降低成本只实现裸TCP字节帧不走Modbus应用层封装——这就排除了直接用pymodbus库的捷径Python是桥接程序的实现语言不是因为它多快而是因为它的socket控制粒度够细、字节操作够直观、调试反馈够快且能打包成Windows服务静默运行不惊动旧上位机进程。我做过17个类似项目最短的一次从接到需求到上线只用了38小时——不是靠黑科技而是靠对这四根柱子的肌肉记忆。下面我就带你把这座桥一块砖一块砖垒出来。2. 字节帧协议逆向没有文档时如何从Wireshark抓包中“读出”终端语言旧上位机不改新终端协议又模糊第一步不是写代码是破译终端的字节语言。你手里可能只有一份《NX-CIF105通讯协议V1.2.pdf》但翻到最后发现它只写了“支持TCP透传模式”连帧格式示例都没有。这时候Wireshark不是可选项是开工许可证。2.1 抓包环境搭建让旧上位机“开口说话”别急着连终端先让旧上位机和一台测试PC通信。假设旧上位机IP是192.168.1.10它通过TCP向某个IP比如192.168.1.100的502端口发指令——这很可能是它原本对接PLC的Modbus TCP地址。现在你需要把这台测试PC变成“中间人”在测试PC上安装Wireshark开启混杂模式配置Windows防火墙允许入站TCP 502端口仅限测试网段修改旧上位机的配置文件通常是.ini或注册表把目标IP从192.168.1.100改成测试PC的IP192.168.1.200启动旧上位机触发一次典型操作比如点击界面上的“启动声光报警”按钮Wireshark过滤条件输入ip.addr 192.168.1.10 and tcp.port 502开始抓包。提示如果旧上位机用的是短连接每次操作建连-发包-断连抓包会看到大量SYN/FIN包如果是长连接保持socket一直打开则要关注TCP Stream的连续数据流。务必确认旧上位机的连接模式——这直接影响桥接程序的socket管理策略。2.2 从原始字节流中定位有效帧三步剥离法Wireshark里看到的是一堆十六进制比如0000 00 00 00 00 00 06 00 01 00 05 00 00 00 00 00 00 ................ 0010 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 ................ 0020 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 ................ 0030 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 ................这显然不是终端协议。真正的有效帧往往藏在Modbus TCP头后面。Modbus TCP帧固定7字节头事务标识符2字节协议标识符2字节长度2字节单元标识符1字节后面才是功能码和数据。所以你要找的是起始特征看第7字节即Modbus头后第一个字节如果是0x01、0x02、0x03等标准功能码说明这是Modbus帧但如果你的目标终端不走Modbus这一字节可能是自定义命令字比如0x10启动报警、0x11停止报警、0x20播放语音ID 1长度线索观察多组抓包看每帧总长度是否固定。比如所有“启动报警”帧都是12字节所有“查询状态”帧都是8字节变化字节定位对比两次相同操作的抓包找出唯一变化的字节位置——那很可能就是参数域。比如第一次发00 00 00 01 00 00 00 00 00 00 00 00第二次发00 00 00 01 00 00 00 00 00 00 00 01变的是最后一个字节那它大概率是“报警等级”或“通道号”。我遇到过一个案例终端协议文档写“音量值0~100”但抓包发现当上位机设音量50时对应字节是0x32ASCII 2设音量100时是0x31 0x30 0x30ASCII 100。原来它传的是ASCII字符串不是二进制数值这种坑不抓包永远不知道。2.3 校验算法还原CRC16、XOR还是无校验字节帧的灵魂是校验。没有校验终端拒收校验错终端静默。常见校验方式有三种校验类型特征还原方法无校验帧尾无额外字节长度固定且稳定抓包比对多帧确认末尾无变化字节XOR校验帧尾1字节值等于前面所有字节异或取一帧前N-1字节用Pythonfunctools.reduce(lambda x,y: x^y, bytes_list)计算看是否等于最后一字节CRC16-Modbus帧尾2字节大端序多项式0x8005用crcmod.predefined.mkCrcFun(modbus)计算比对结果实操中我优先试XOR因为简单且国产终端常用。写一段Python快速验证def xor_check(frame): if len(frame) 2: return False calc 0 for b in frame[:-1]: # 去掉最后一个字节 calc ^ b return calc frame[-1] # 从Wireshark导出原始字节去掉TCP/IP头只留payload test_frame bytes.fromhex(01 10 00 00 00 01 02 00 01 91 9A) # 示例帧 print(xor_check(test_frame)) # True or False如果XOR失败再试CRC16。注意有些终端用CRC16-IBM多项式0x8005初始值0xFFFF无反转有些用CRC16-CCITT初始值0x0000必须对照终端实际返回的校验值反推。最笨但最可靠的办法用终端厂商提供的调试工具发一帧已知内容抓包看校验值再暴力穷举常见CRC参数组合。注意千万别在没确认校验方式前就写发送逻辑。我见过团队因默认用CRC16-Modbus导致终端持续返回“校验错误”报文排查三天才发现厂商文档小字写着“本型号使用XOR校验”。3. 桥接程序核心设计双Socket隧道与字节帧状态机确认了终端协议下一步是写桥接程序。它的本质是一个TCP双向代理但比普通代理复杂得多它要理解字节帧的边界要处理粘包/半包要维持连接状态还要在异常时优雅降级。Python是理想选择但必须避开socketserver这类高层封装——你需要直接操控socket的recv buffer和send buffer。3.1 架构选型为什么不用现成的TCP代理工具有人会问为什么不用socat或haproxy答案很直接它们不理解字节帧语义。socat TCP4:192.168.1.100:502 TCP4:192.168.1.200:502只能做透明转发但旧上位机发的是Modbus TCP帧终端要的是裸字节帧中间差了一个协议转换层。socat无法把00 00 00 00 00 06 00 01 00 05 00 00Modbus TCP头读线圈拆解成01 05 00 00 FF 00纯功能码数据更无法插入CRC校验。所以桥接程序必须是有状态的字节处理器核心模块有三个上位机监听器Upstream Listener绑定本地端口如502接收旧上位机连接解析其发送的原始字节流终端通信器Downstream Handler维护与声光语音终端的TCP长连接将解析后的指令帧发给终端并接收响应帧状态机Frame State Machine定义字节帧的生命周期——从收到第一个字节到识别起始符到累积满长度到校验到交付下游——每一步都可能失败需有明确的错误分支。3.2 上位机监听器解决粘包与半包的实战方案旧上位机发包不会按“一帧一包”发送。TCP是流协议send()调用可能被内核合并也可能被IP层分片。你收到的可能是半包只收到一帧的前10字节共12字节粘包一次recv()收到两帧如[帧1][帧2]混合包[帧1前半][帧2全][帧3前半]。解决方案不是“等足够字节再处理”而是基于协议特征的流式解析。假设你的终端协议规定帧以0xAA开头第2字节是长度含头尾第3字节是命令字最后2字节是CRC16。那么解析逻辑如下class UpstreamListener: def __init__(self, host0.0.0.0, port502): self.sock socket.socket(socket.AF_INET, socket.SOCK_STREAM) self.sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) self.sock.bind((host, port)) self.sock.listen(5) self.buffer b # 接收缓冲区 def recv_frame(self, conn): 从conn接收完整一帧返回bytes或None while True: # 1. 查找起始符0xAA start_idx self.buffer.find(b\xAA) if start_idx -1: # 未找到起始符接收更多数据 data conn.recv(4096) if not data: return None self.buffer data continue # 2. 起始符后至少要有长度字节第2字节和CRC2字节共4字节 if len(self.buffer) start_idx 4: data conn.recv(4096) if not data: return None self.buffer data continue # 3. 读取长度字节第2字节 frame_len self.buffer[start_idx 1] total_len frame_len 2 # 长度字节CRC2字节需根据协议确认 if len(self.buffer) start_idx total_len: # 数据不足继续接收 data conn.recv(4096) if not data: return None self.buffer data continue # 4. 截取完整帧 frame self.buffer[start_idx:start_idx total_len] self.buffer self.buffer[start_idx total_len:] # 清除已处理部分 return frame这个recv_frame方法的关键在于它不假设recv()返回整帧而是把socket当作字节流用buffer缓存未处理数据用while循环不断补充、查找、截取。start_idx定位起始符frame_len决定需要多少字节self.buffer保存跨recv()调用的残留数据。这才是工业现场真正可靠的粘包处理。3.3 终端通信器长连接保活与异常熔断声光语音终端不是服务器它资源有限。你必须主动维护连接否则它可能因超时断开。但频繁重连又增加终端负担。平衡点是心跳保活 熔断机制。心跳每30秒发一个空帧如0xAA 0x03 0x00 0x00 0x00或专用心跳命令如0xAA 0x01 0x00熔断如果连续3次心跳无响应或发送指令后5秒无返回则关闭连接等待10秒后重连重连退避首次失败后等1秒第二次等2秒第三次等4秒避免雪崩。Python实现要点import time import threading class DownstreamHandler: def __init__(self, terminal_ip, terminal_port): self.ip terminal_ip self.port terminal_port self.sock None self.connected False self.last_heartbeat 0 self.fail_count 0 def connect(self): try: self.sock socket.socket(socket.AF_INET, socket.SOCK_STREAM) self.sock.settimeout(5) self.sock.connect((self.ip, self.port)) self.connected True self.fail_count 0 print(fConnected to {self.ip}:{self.port}) except Exception as e: self.connected False self.fail_count 1 print(fConnect failed: {e}, fail count: {self.fail_count}) def send_and_recv(self, frame): if not self.connected: self.connect() if not self.connected: return None try: self.sock.sendall(frame) # 设置recv超时避免卡死 self.sock.settimeout(5) response self.sock.recv(1024) self.last_heartbeat time.time() return response except socket.timeout: print(Recv timeout) self._handle_disconnect() except ConnectionResetError: print(Connection reset by peer) self._handle_disconnect() except Exception as e: print(fSend/recv error: {e}) self._handle_disconnect() return None def _handle_disconnect(self): if self.sock: self.sock.close() self.connected False # 指数退避重连 wait_time min(2 ** self.fail_count, 60) time.sleep(wait_time)实测心得很多声光终端的TCP栈很脆弱settimeout(5)必须设否则recv()可能永久阻塞sendall()比send()可靠它确保所有字节发出ConnectionResetError捕获比OSError更精准专指对方主动断连。4. 字节帧转换引擎从Modbus TCP到裸字节的精准映射旧上位机发的是Modbus TCP帧终端要的是裸字节帧。转换不是简单删头去尾而是语义级映射。比如旧上位机用Modbus功能码0x05写单个线圈控制报警启停终端却用命令字0x10参数0x01表示“启动”0x100x00表示“停止”。转换引擎就是这个翻译官。4.1 映射规则表用字典而非if-else提升可维护性硬编码if func_code 0x05: ... elif func_code 0x03: ...会导致代码臃肿、难扩展。正确做法是定义映射规则表每个规则包含匹配条件、转换逻辑、错误处理。# mapping_rules.py MAPPING_RULES { alarm_control: { match: lambda frame: (len(frame) 12 and frame[7] 0x05 and # Modbus功能码 frame[9] 0xFF), # 线圈值FF00 convert: lambda frame: bytes([0xAA, 0x04, 0x10, 0x01, 0x00, 0x00]) calc_crc16(bytes([0xAA, 0x04, 0x10, 0x01, 0x00, 0x00])), description: Modbus 0x05 FF00 - Terminal Alarm Start }, alarm_stop: { match: lambda frame: (len(frame) 12 and frame[7] 0x05 and frame[9] 0x00), convert: lambda frame: bytes([0xAA, 0x04, 0x10, 0x00, 0x00, 0x00]) calc_crc16(bytes([0xAA, 0x04, 0x10, 0x00, 0x00, 0x00])), description: Modbus 0x05 0000 - Terminal Alarm Stop }, play_voice: { match: lambda frame: (len(frame) 12 and frame[7] 0x10 and # 功能码0x10写多个寄存器 frame[11] 0x01), # 寄存器值0x01 convert: lambda frame: build_voice_frame(frame[11]), # 自定义函数 description: Modbus 0x10 reg0x01 - Play Voice ID 1 } }主转换函数遍历规则表def convert_modbus_to_terminal(modbus_frame): for rule_name, rule in MAPPING_RULES.items(): if rule[match](modbus_frame): try: terminal_frame rule[convert](modbus_frame) print(fConverted: {rule_name} - {terminal_frame.hex()}) return terminal_frame except Exception as e: print(fConvert error in {rule_name}: {e}) return None print(fNo matching rule for frame: {modbus_frame.hex()}) return None这样做的好处是新增一种控制逻辑只需在MAPPING_RULES里加一个dict不用改主逻辑规则可单独单元测试运维人员甚至能看懂规则描述自己微调。4.2 CRC校验注入动态计算与字节拼接终端帧的CRC必须实时计算不能写死。以CRC16-Modbus为例Python标准库不直接提供需用crcmodpip install crcmodimport crcmod # 创建CRC16-Modbus函数 crc16_func crcmod.predefined.mkCrcFun(modbus) def calc_crc16(data): 计算data的CRC16-Modbus返回2字节bytes crc crc16_func(data) # Modbus CRC是低字节在前高字节在后 return crc.to_bytes(2, little) # 构建完整帧头数据CRC def build_full_frame(header, payload): frame_without_crc header payload crc_bytes calc_crc16(frame_without_crc) return frame_without_crc crc_bytes # 示例构建启动报警帧 header bytes([0xAA, 0x04, 0x10, 0x01]) payload bytes([0x00, 0x00]) full_frame build_full_frame(header, payload) # full_frame b\xaa\x04\x10\x01\x00\x00\x91\x9a关键细节CRC16-Modbus的字节序是小端low byte first即低位字节在前。很多初学者用to_bytes(2, big)导致校验错终端拒收。Wireshark抓包时直接看终端返回的正确帧CRC字段的两个字节顺序就是标准。4.3 响应帧回传保持上下位机会话一致性终端返回的响应帧必须原样、及时地回传给旧上位机。但要注意两点时序一致性旧上位机发完帧A期待在几毫秒内收到响应A。如果桥接程序把响应A缓存起来等响应B一起发上位机就会超时重发造成重复指令格式伪装旧上位机 expecting Modbus TCP响应如00 00 00 00 00 03 00 05 00 00但终端返回的是裸字节如AA 04 10 01 00 00 91 9A。你必须把裸字节“包装”成Modbus TCP响应否则上位机解析失败。包装逻辑很简单提取终端响应的有效载荷去掉头尾校验填入Modbus TCP响应模板def wrap_terminal_response(terminal_resp, original_modbus_req): 将终端裸响应包装成Modbus TCP响应 original_modbus_req: 原始Modbus请求帧用于提取事务ID等 if len(original_modbus_req) 7: return None # Modbus TCP头事务ID(2)协议ID(2)长度(2)单元ID(1) trans_id original_modbus_req[0:2] proto_id b\x00\x00 # 终端响应的有效载荷假设去掉前2字节头和后2字节CRC payload terminal_resp[2:-2] if len(terminal_resp) 4 else b # 构建Modbus响应功能码请求的功能码数据有效载荷 func_code original_modbus_req[7] if len(original_modbus_req) 7 else 0x00 modbus_payload bytes([func_code]) payload # 计算长度字段长度 1功能码 len(payload) length 1 len(payload) length_bytes length.to_bytes(2, big) # 完整Modbus响应帧 modbus_resp trans_id proto_id length_bytes b\x01 modbus_payload return modbus_resp # 使用示例 modbus_req bytes.fromhex(00 00 00 00 00 06 00 01 00 05 00 00) terminal_resp bytes.fromhex(AA 04 10 01 00 00 91 9A) wrapped wrap_terminal_response(terminal_resp, modbus_req) # wrapped b\x00\x00\x00\x00\x00\x03\x00\x01\x00\x00这个wrap_terminal_response函数保证了旧上位机看到的永远是它熟悉的Modbus TCP格式只是数据内容被桥接程序悄悄替换了。它感知不到中间有座桥这就是改造成功的标志。5. 部署与运维Windows服务化、日志追踪与零停机切换程序写完了但工业现场不接受“双击运行”的脚本。它必须像Windows服务一样后台静默运行开机自启异常自动恢复且切换过程不能影响产线。5.1 打包为Windows服务pywin32的正确用法pyinstaller打包exe是基础但要成为服务需用pywin32。关键不是win32serviceutil.InstallService而是服务主循环的健壮性。import win32serviceutil import win32service import win32event import servicemanager import socket import sys import time from bridge_core import BridgeEngine # 你的主类 class TCPTerminalBridgeService(win32serviceutil.ServiceFramework): _svc_name_ TCPTerminalBridge _svc_display_name_ TCP Terminal Bridge Service _svc_description_ Bridges legacy SCADA to sound-light-voice terminals via raw TCP frames def __init__(self, args): win32serviceutil.ServiceFramework.__init__(self, args) self.hWaitStop win32event.CreateEvent(None, 0, 0, None) self.is_alive True self.bridge None def SvcDoRun(self): servicemanager.LogMsg(servicemanager.EVENTLOG_INFORMATION_TYPE, servicemanager.PYS_SERVICE_STARTED, (self._svc_name_, )) self.main() def SvcStop(self): self.ReportServiceStatus(win32service.SERVICE_STOP_PENDING) win32event.SetEvent(self.hWaitStop) self.is_alive False if self.bridge: self.bridge.stop() # 调用你的bridge.stop()方法 def main(self): # 初始化桥接引擎 self.bridge BridgeEngine( upstream_port502, downstream_ip192.168.1.100, downstream_port502 ) # 主循环检查服务状态运行桥接逻辑 while self.is_alive: try: self.bridge.run_once() # 你的单次运行逻辑 time.sleep(0.1) # 避免CPU空转 except Exception as e: servicemanager.LogErrorMsg(fBridge error: {e}) time.sleep(1) if __name__ __main__: if len(sys.argv) 1: servicemanager.Initialize() servicemanager.PrepareToHostSingle(TCPTerminalBridgeService) servicemanager.StartServiceCtrlDispatcher() else: win32serviceutil.HandleCommandLine(TCPTerminalBridgeService)编译命令pyinstaller --onefile --windowed --hidden-importwin32timezone service_script.py注意--windowed防止弹窗--hidden-importwin32timezone是pywin32常见缺失依赖生成的exe需用管理员权限安装服务service_script.exe install。5.2 日志体系分级记录与故障定位工业系统日志不是为了好看是为了5分钟内定位问题。我采用三级日志INFO级正常连接、帧转换成功、心跳成功——每小时一条即可避免刷屏WARNING级终端连接失败、帧校验错误、响应超时——这些是潜在风险需人工关注ERROR级socket异常、主线程崩溃、服务停止——必须立即告警。日志格式强制包含时间戳、线程ID、操作类型、关键参数import logging from logging.handlers import RotatingFileHandler def setup_logger(): logger logging.getLogger(TCPTerminalBridge) logger.setLevel(logging.DEBUG) # 文件处理器按大小轮转保留10个历史文件 file_handler RotatingFileHandler( bridge.log, maxBytes10*1024*1024, # 10MB backupCount10 ) file_formatter logging.Formatter( %(asctime)s | %(threadName)s | %(levelname)-8s | %(message)s, datefmt%Y-%m-%d %H:%M:%S ) file_handler.setFormatter(file_formatter) logger.addHandler(file_handler) # 控制台处理器仅DEBUG时 if DEBUG_MODE: console_handler logging.StreamHandler() console_handler.setFormatter(file_formatter) logger.addHandler(console_handler) return logger # 使用示例 logger setup_logger() logger.info(Upstream connected from 192.168.1.10:54321) logger.warning(CRC check failed on frame AA041001... (len8)) logger.error(Downstream socket closed unexpectedly)5.3 零停机切换三步法平滑接管旧流量最后一步也是最关键的一步如何把旧上位机的流量从直连PLC切换到桥接程序而不中断生产Step 1并行运行验证桥接逻辑不修改旧上位机配置让它继续连PLC桥接程序监听另一个端口如503用测试工具模拟上位机发帧验证终端响应是否正确此阶段只验证不接管真实流量。Step 2流量镜像比对输出一致性修改上位机配置目标IP指向桥接程序192.168.1.200端口502但桥接程序不转发给终端而是把收到的帧和终端模拟响应同时写入日志用另一台PC抓包对比旧上位机发出的帧与桥接程序日志记录的帧是否完全一致确认无误后再启用真实转发。Step 3热切换一键回滚切换前备份旧上位机配置文件切换时只需改一个IP地址30秒内完成桥接程序内置回滚开关如果检测到连续10次终端无响应自动切换到“直通模式”把上位机帧原样转发给原PLC IP并发送邮件告警。实战教训某次切换后发现终端响应延迟比PLC高
