Python中间件实现旧上位机日志转TCP字节帧对接声光报警终端

1. 项目背景与核心痛点拆解1.1 旧上位机为什么成了“钉子户”做过工业现场改造的人大概都遇到过这种局面产线上跑了好几年的上位机软件功能稳定、逻辑复杂、跟MES和数据库都有耦合唯一的问题就是它“不会说话”——它只会通过自己那套私有协议或者干脆就是串口打印把数据吐给原来的老终端。现在产线要升级新增了声光语音报警终端要求实时接收设备状态、报警等级、工位号这些信息但旧上位机源码找不到、原厂不维护、开发人员早离职了。你让它改它改不了你把它换掉风险太大牵一发动全身。这就是典型的“旧上位机不肯改”场景。核心矛盾在于新终端只认标准TCP字节帧旧上位机只会按自己的方式发数据。两边协议对不上中间又没有缓冲层项目就卡住了。我这次接到的现场就是这样一台运行了六年的工控机上面跑着一套C#写的上位机负责采集四台设备的运行状态。现在要在车间顶部装四个声光语音终端分别对应四条产线哪个工位报警对应的终端就要亮红灯、播语音、显示文字。旧上位机没有对外接口只有一个本地日志文件在不断追加状态记录。1.2 为什么选择原生TCP字节帧方案摆在面前的路其实有好几条一是找原厂改上位机报价高、周期长而且对方明确说“老版本不再维护”二是用OPC UA或者Modbus TCP做中间层但旧上位机根本不支持这些协议三是直接读日志文件用脚本解析后转发。第三条路听起来最“土”但恰恰是最快、最可控的。我最终确定的方案是写一个Python中间件实时监控旧上位机的日志文件解析出状态变化然后以原生TCP字节帧的形式主动推送给声光语音终端。这里的关键词是“原生TCP字节帧”——不走Modbus TCP不走HTTP就是最朴素的Socket连接自己定义帧结构自己处理粘包和断线重连。为什么不用Modbus TCP因为声光语音终端虽然支持Modbus TCP但它的寄存器映射是固定的能写的寄存器数量有限而且语音播报的内容需要预先烧录或者通过特定功能码触发灵活性太差。相比之下原生TCP字节帧可以让我自由定义协议帧头、设备ID、报警类型、文字内容长度、文字内容、校验码想怎么排就怎么排。终端那边只需要一个简单的Socket服务端收到帧后解析并执行对应动作。这个方案还有一个好处对旧上位机零侵入。我不需要动它一行代码不需要它开放任何端口只需要它继续写日志就行。日志文件是它本来就有的功能我只是“偷看”这个文件而已。风险最低回滚最容易——万一中间件出问题把进程杀掉旧上位机照常运行产线不受影响。1.3 适用场景与读者定位这套方案适合以下几类人一是做工业现场改造的自动化工程师手头有旧设备要接入新系统二是做物联网网关开发的程序员需要快速实现协议转换三是刚入门的Python学习者想找一个有实际价值的练手项目。哪怕你之前没接触过TCP编程只要跟着思路走也能把整套逻辑跑通。需要提前说明的是本文不会涉及任何特定品牌的商业机密所有代码和配置都是通用思路。你拿过去之后只需要根据自己现场的日志格式和终端协议做少量修改就能直接复用。2. 核心细节解析与实操要点2.1 日志文件监控的三种方式对比监控日志文件变化听起来简单做起来坑不少。我试过三种方式这里直接给结论方式原理优点缺点适用场景轮询读取每隔N秒读一次文件末尾实现简单兼容性好实时性差频繁IO日志量小、实时性要求低watchdog库监听文件系统事件实时性高资源占用低跨平台行为不一致Linux/Windows均可推荐tail -f模拟持续读取文件新增内容实时性高逻辑直观文件被切割时需处理日志会滚动切割的场景我最终选择了watchdog 文件指针记录的组合方案。watchdog负责监听文件修改事件一旦文件有变化就触发回调回调里用记录的文件指针位置读取新增内容避免重复读取。这个方案在Windows和Linux上都跑得很稳而且CPU占用几乎可以忽略。注意如果旧上位机的日志文件会被定期清空或重命名一定要处理文件句柄失效的情况。我的做法是每次回调都重新检查文件是否存在如果不存在就等待文件重新创建然后重置指针。2.2 TCP字节帧的帧结构设计帧结构是整个方案的核心。设计得好解析起来轻松设计得不好后面全是坑。我定的帧格式如下| 帧头(2B) | 设备ID(1B) | 命令类型(1B) | 数据长度(2B) | 数据内容(NB) | 校验和(1B) |逐字段说明帧头固定为0xAA 0x55用来做帧同步。收到数据后先找这两个字节找不到就丢弃避免解析错位。设备ID1字节范围0x01到0x04对应四个声光语音终端。0x00作为广播地址保留。命令类型1字节0x01表示报警触发0x02表示报警清除0x03表示文字更新0x04表示心跳。数据长度2字节大端序表示后面数据内容的字节数。最大支持65535字节实际使用中文字内容不会超过200字节。数据内容变长具体格式由命令类型决定。比如报警触发时数据内容就是报警文字UTF-8编码。校验和1字节从帧头到数据内容所有字节的累加和取低8位。简单但够用能挡住大部分传输错误。为什么用大端序因为网络字节序就是大端Python的struct模块打包时用!H就能直接生成大端序的2字节整数省得自己转换。为什么校验和不用CRC因为现场环境电磁干扰不算严重累加和足够而且计算速度快终端那边用单片机也能轻松实现。2.3 粘包与断线重连的处理策略TCP是流式协议没有消息边界。你发两次对方可能一次收到你发一次对方可能分两次收到。这就是粘包和拆包问题。我的处理策略是接收端按帧头长度字段来切分。具体做法维护一个缓冲区每次收到数据就追加到缓冲区末尾。然后循环检查缓冲区如果缓冲区长度小于最小帧长7字节直接返回如果前两字节不是0xAA 0x55就把第一个字节丢掉继续检查如果找到了帧头就读出数据长度字段判断缓冲区里是否有完整的一帧如果有就切出来处理然后继续循环如果没有就等待更多数据。断线重连方面我用了心跳超时检测的机制。中间件每隔5秒向终端发一个心跳帧命令类型0x04数据内容为空。如果连续3次发送失败或者Socket的send方法抛出异常就关闭当前连接等待2秒后重新连接。重连时不需要终端做任何特殊处理因为终端那边也是按帧解析的收到心跳就回复一个心跳确认收不到就认为连接断了。实操心得在Windows上如果终端是突然断电而不是正常关闭Socket的send方法可能不会立刻报错而是会阻塞一段时间。这时候需要设置settimeout我设的是3秒。超过3秒没发出去就强制关闭重连。3. 实操过程与核心环节实现3.1 环境准备与依赖安装先说一下我的运行环境Windows 10专业版Python 3.9.7终端是四个支持TCP的声光语音报警器。Python的安装过程我就不赘述了网上教程很多注意安装时勾选“Add Python to PATH”就行。需要安装的第三方库只有两个pip install watchdogwatchdog用于文件监控。为什么不用pyinotify因为那是Linux专用的Windows上跑不了。watchdog跨平台API也简单。如果你需要打包成exe方便部署再装一个pyinstallerpip install pyinstaller打包命令后面会讲。这里先提醒一句不要用系统自带的Python环境直接跑生产脚本建议用virtualenv创建一个独立环境避免依赖冲突。我吃过这个亏现场工控机上装了一堆乱七八糟的库结果watchdog的版本跟某个库不兼容排查了半天。3.2 日志解析模块的编写旧上位机的日志格式是这样的2024-01-15 08:23:45.123 [INFO] Line1 StatusRunning Alarm0 2024-01-15 08:23:46.456 [WARN] Line2 StatusRunning Alarm1 CodeE101 2024-01-15 08:23:47.789 [INFO] Line3 StatusIdle Alarm0每行一条记录包含时间戳、日志级别、产线编号、状态、报警标志和报警代码。我需要提取的是产线编号、报警标志和报警代码。报警标志为1时触发对应终端的声光报警报警标志从1变为0时清除报警。解析代码的核心逻辑import re LOG_PATTERN re.compile( r\[(\w)\]\sLine(\d)\sStatus(\w)\sAlarm(\d)(?:\sCode(\w))? ) def parse_log_line(line): match LOG_PATTERN.search(line) if not match: return None level, line_no, status, alarm, code match.groups() return { line: int(line_no), status: status, alarm: int(alarm), code: code or }这里用正则而不是split是因为日志格式可能会微调正则的容错性更好。比如某一行没有Code字段正则里的(?:...)?就能处理。另外Line(\d)只匹配单个数字因为现场只有四条产线。如果你有十几条产线改成Line(\d)就行。注意事项日志文件可能是GBK编码而不是UTF-8尤其是老旧的C#程序。我一开始用UTF-8读结果中文乱码正则匹配失败。后来改成open(file, r, encodinggbk, errorsignore)才正常。你可以先用记事本打开日志文件看右下角显示的编码格式。3.3 TCP客户端与帧发送实现中间件作为TCP客户端主动连接终端的IP和端口。每个终端一个独立的Socket连接用字典管理import socket import struct import time class TerminalClient: def __init__(self, device_id, host, port): self.device_id device_id self.host host self.port port self.sock None self.last_heartbeat 0 self.connect() def connect(self): try: self.sock socket.socket(socket.AF_INET, socket.SOCK_STREAM) self.sock.settimeout(3) self.sock.connect((self.host, self.port)) print(fDevice {self.device_id} connected to {self.host}:{self.port}) except Exception as e: print(fDevice {self.device_id} connect failed: {e}) self.sock None def build_frame(self, cmd_type, datab): header b\xAA\x55 device_id struct.pack(!B, self.device_id) cmd struct.pack(!B, cmd_type) length struct.pack(!H, len(data)) payload header device_id cmd length data checksum sum(payload) 0xFF return payload struct.pack(!B, checksum) def send_frame(self, cmd_type, datab): if self.sock is None: self.connect() if self.sock is None: return False try: frame self.build_frame(cmd_type, data) self.sock.sendall(frame) return True except Exception as e: print(fDevice {self.device_id} send failed: {e}) self.sock.close() self.sock None return False def send_heartbeat(self): now time.time() if now - self.last_heartbeat 5: self.send_frame(0x04) self.last_heartbeat now这段代码有几个关键点sendall而不是send确保数据全部发出去settimeout(3)防止阻塞发送失败后立即关闭Socket并置空下次发送时会自动重连。3.4 主循环与状态管理主循环的逻辑是watchdog监听日志文件变化解析新增行对比上一次的状态如果报警状态发生变化就向对应终端发送帧。import os from watchdog.observers import Observer from watchdog.events import FileSystemEventHandler class LogHandler(FileSystemEventHandler): def __init__(self, log_path, clients): self.log_path log_path self.clients clients self.last_pos os.path.getsize(log_path) if os.path.exists(log_path) else 0 self.last_alarm {1: 0, 2: 0, 3: 0, 4: 0} def on_modified(self, event): if event.src_path ! self.log_path: return self.process_new_lines() def process_new_lines(self): if not os.path.exists(self.log_path): self.last_pos 0 return with open(self.log_path, r, encodinggbk, errorsignore) as f: f.seek(self.last_pos) for line in f: result parse_log_line(line) if result is None: continue line_no result[line] alarm result[alarm] if line_no not in self.clients: continue if alarm ! self.last_alarm.get(line_no, 0): self.last_alarm[line_no] alarm if alarm 1: text f产线{line_no}报警 {result[code]} self.clients[line_no].send_frame(0x01, text.encode(utf-8)) else: self.clients[line_no].send_frame(0x02) self.last_pos f.tell()这里用last_alarm字典记录每条产线上一次的报警状态只有状态变化时才发帧避免重复报警。文字内容用UTF-8编码终端那边按UTF-8解码就行。实操心得日志文件如果被旧上位机以“追加”模式写入f.seek(self.last_pos)之后f.tell()返回的位置是准确的。但如果旧上位机是“重写”模式每次清空再写文件大小会变小seek到超出文件末尾的位置会读不到数据。我的处理是每次回调都检查文件大小如果比上次记录的位置小就重置last_pos为0。3.5 打包部署与开机自启开发完成后用pyinstaller打包成单个exepyinstaller --onefile --name tcp_bridge main.py打包出来的exe放在dist目录下。然后创建一个批处理文件start.batecho off cd /d %~dp0 start /min tcp_bridge.exe把start.bat放到Windows的启动目录shell:startup开机就会自动运行。为什么用start /min因为现场工控机的屏幕还要显示旧上位机界面中间件窗口最小化就行不需要用户看到。如果你希望更稳妥可以用Windows的任务计划程序设置“计算机启动时”触发这样即使有人误关了窗口重启后也能自动恢复。4. 常见问题与排查技巧实录4.1 连接建立失败排查表现象可能原因排查方法解决方式connect timeout终端IP不对或不在同一网段ping终端IP检查网络配置确保同网段connection refused终端端口未监听telnet终端IP 端口确认终端TCP服务已开启连接成功但收不到数据帧格式不匹配用网络调试助手抓包核对帧头、长度、校验和频繁断线重连网络不稳定或终端处理慢查看中间件日志增加心跳间隔降低发送频率中文乱码编码不一致对比发送和接收的字节统一用UTF-8编码这个表是我在现场排查时总结的基本上覆盖了90%的问题。其中“连接成功但收不到数据”最常见多半是帧结构没对齐。我的建议是先用网络调试助手比如NetAssist模拟终端手动发一帧过去看能不能解析。如果能说明中间件没问题问题在终端如果不能说明帧结构设计有误。4.2 日志文件被锁定的处理旧上位机在写日志时可能会以独占方式打开文件导致Python读不了。我遇到过这种情况open的时候报PermissionError。解决办法有两个一是用shared模式打开但Python的open不支持二是复制一份日志文件再读。我采用的是第二种每次回调时先把日志文件复制到临时目录然后读临时文件。虽然多了一次IO但胜在稳定。复制的时候用shutil.copy2保留时间戳。读完之后删除临时文件。import shutil import tempfile def safe_read_log(log_path): tmp os.path.join(tempfile.gettempdir(), log_copy.txt) try: shutil.copy2(log_path, tmp) with open(tmp, r, encodinggbk, errorsignore) as f: return f.readlines() except Exception as e: print(fRead log failed: {e}) return []这个方案有个小缺点如果日志文件很大比如几百MB复制会很慢。但现场日志每天滚动一次单个文件不超过10MB所以影响不大。4.3 终端响应慢导致积压四个终端如果同时报警中间件会连续发送四帧。如果某个终端处理慢Socket的发送缓冲区满了sendall就会阻塞。虽然设了3秒超时但阻塞期间其他终端的帧也发不出去。我的优化方案是每个终端一个独立的发送线程。主线程解析日志后把帧放入对应终端的队列发送线程从队列里取帧发送。这样即使某个终端阻塞也不影响其他终端。import queue import threading class TerminalClient: def __init__(self, device_id, host, port): # ... 前面的初始化代码 ... self.send_queue queue.Queue(maxsize100) self.thread threading.Thread(targetself._send_loop, daemonTrue) self.thread.start() def _send_loop(self): while True: cmd_type, data self.send_queue.get() if cmd_type is None: break self.send_frame(cmd_type, data) def enqueue_frame(self, cmd_type, datab): try: self.send_queue.put_nowait((cmd_type, data)) except queue.Full: print(fDevice {self.device_id} queue full, drop frame)队列设了最大长度100满了就丢弃。为什么丢弃而不是阻塞因为报警信息有时效性积压的旧报警不如不发。这个策略在现场跑了一年多没出过问题。4.4 独家避坑技巧汇总最后分享几个文档里不会写、但实际很管用的技巧日志文件路径不要用相对路径。现场部署时工作目录可能跟开发时不一样。用os.path.abspath转成绝对路径或者直接在配置文件里写死。终端IP建议用静态IP。DHCP租约到期后IP变了中间件就连不上了。如果必须用DHCP就在终端上配DDNS中间件用域名连接。中间件加一个简单的HTTP状态页。用Python自带的http.server起一个端口返回当前连接状态和最后发送时间。现场排查时浏览器一打开就能看到比翻日志快多了。日志切割时注意文件句柄。如果旧上位机每天生成一个新日志文件比如log_20240115.txtwatchdog的on_modified可能不会触发新文件的创建事件。需要同时监听on_created并在回调里更新log_path。测试时先用localhost。在开发机上用两个Python脚本一个模拟终端一个跑中间件先把逻辑跑通再去现场。现场调试时间宝贵不要把基础问题留到现场解决。这套方案从立项到上线我用了大约一周时间其中编码三天现场调试两天。上线后连续运行了六个月中间只因为一次交换机断电重启过。旧上位机一行代码没改新终端完美接入。如果你也遇到类似场景希望这篇实录能帮你少走弯路。