226 lines
8.2 KiB
Python
226 lines
8.2 KiB
Python
from queue import Full
|
||
import socket
|
||
import threading
|
||
import time
|
||
from typing import Dict, Optional
|
||
from loguru import logger
|
||
|
||
import telnetlib #2026/6/2
|
||
|
||
# 全局：TCP客户端管理（每个UART一个实例）
|
||
tcp_clients: Dict[str, "TCPUARTClient"] = {}
|
||
# HEAD = b"\xeb\x90" #地测帧头
|
||
# TAIL = b"\x57\x16" #地测帧尾
|
||
# MIN_FRAME = 23 ##地测长度最小值
|
||
# MAX_FRAME = 1000 ##地测帧长度最大值
|
||
|
||
class TCPUARTClient:
|
||
def __init__(self, dev_name: str, local_port: int, remote_port: int, remote_ip="127.0.0.1"):
|
||
self.dev_name = dev_name
|
||
self.local_port = local_port
|
||
self.remote_port = remote_port
|
||
self.remote_ip = remote_ip
|
||
self.socket = None
|
||
self.running = False
|
||
self.thread = None
|
||
|
||
self.HEAD = b"\xeb\x90" #地测帧头
|
||
self.TAIL = b"\x57\x16" #地测帧尾
|
||
self.MIN_LEN = 23 ##地测长度最小值
|
||
self.MAX_LEN = 1000 ##地测帧长度最大值
|
||
self.tcp_buf = b''
|
||
|
||
#self.tn = None
|
||
|
||
|
||
def connect(self):
|
||
try:
|
||
|
||
self.socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
|
||
# 关键点：禁止Nagle算法，模仿网络调试助手，强制使用纯raw socket连接，但是导致TCP收不到数据 2026/6/2
|
||
self.socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) #停用
|
||
#self.socket.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1) #telnet
|
||
self.socket.bind(("127.0.0.1", self.local_port))
|
||
self.socket.connect((self.remote_ip, self.remote_port))
|
||
self.socket.setblocking(False)
|
||
|
||
self.running = True
|
||
print(f"[TCP-UART] {self.dev_name} 连接成功: local={self.local_port} remote={self.remote_port}")
|
||
|
||
#telnet
|
||
#self.tn = telnetlib.Telnet("127.0.0.1",{self.remote_port},timeout=5)
|
||
|
||
|
||
return True
|
||
except Exception as e:
|
||
print(f"[TCP-UART] {self.dev_name} 连接失败: {e}")
|
||
return False
|
||
'''
|
||
# 去除转义FF的函数 2026/6/2,可能有安全问题
|
||
def unescape_ff(raw:bytes)->bytes:
|
||
buf = bytearray()
|
||
i = 0
|
||
n = len(raw)
|
||
while i<n:
|
||
buf.append(raw[i])
|
||
|
||
#当前字节FF，下一个字节FF：跳过下一个字节，只保留一个
|
||
if raw[i] == 0xFF and i+1<n and raw[i+1]==0xFF:
|
||
i +=1
|
||
i +=1
|
||
return bytes(buf)
|
||
'''
|
||
# 去除转义FF的函数 2026/6/2
|
||
def unescape_ff(self, raw:bytes)->bytes:
|
||
out = bytearray()
|
||
idx=0
|
||
length = len(raw)
|
||
while idx<length:
|
||
cur = raw[idx]
|
||
out.append(cur)
|
||
if cur == 0xff and (idx + 1 < length) and raw[idx+1]==0xff:
|
||
idx+=1
|
||
idx+=1
|
||
return bytes(out)
|
||
|
||
def receive_loop(self):
|
||
from midware.core.bus_udp_decoupler import bus_recv_queues
|
||
from midware.config.base_config import DEVICE_CONFIG_DICT
|
||
|
||
while self.running:
|
||
try:
|
||
|
||
#data = self.tn.read_some()
|
||
|
||
data = self.socket.recv(8192)#4096修改为16384 2026/5/12 tmy 65536->8192 2026/5/15
|
||
if not data:
|
||
time.sleep(0.001)
|
||
continue
|
||
self.tcp_buf += data #新数据追加到缓存尾部2026/6/2
|
||
|
||
dev_name = None
|
||
protocol = None
|
||
for name, cfg in DEVICE_CONFIG_DICT.items():
|
||
if cfg.get("tcp_bus_port") == self.remote_port:
|
||
dev_name = name
|
||
protocol = cfg.get("protocol")
|
||
break
|
||
|
||
if not dev_name or not protocol:
|
||
continue
|
||
|
||
#循环拆包，大多数情况是只跑0~1次循环 2026/6/2
|
||
while True:
|
||
if len(self.tcp_buf)>self.MAX_LEN * 2:
|
||
self.tcp_buf = self.tcp_buf[-1000:]
|
||
h_pos = self.tcp_buf.find(self.HEAD)
|
||
if h_pos ==-1:
|
||
break
|
||
#截断帧头前面垃圾
|
||
if h_pos>0:
|
||
self.tcp_buf = self.tcp_buf[h_pos:]
|
||
#缓存不足最小帧，直接退出
|
||
if(len(self.tcp_buf)<self.MIN_LEN):
|
||
break
|
||
#从帧头向后查找结束符号
|
||
t_pos = self.tcp_buf.find(self.TAIL,2)
|
||
if t_pos == -1:
|
||
break
|
||
frame = self.tcp_buf[:t_pos+2]
|
||
#logger.info(f'去除转义FF帧前的有问题帧内容：{frame.hex()}')
|
||
|
||
if(self.MIN_LEN<= len(frame) <= self.MAX_LEN):
|
||
#去除转义FF
|
||
#logger.info(f'去除转义FF帧内容：')
|
||
real_data = self.unescape_ff(frame)
|
||
|
||
#logger.info(f'去除转义FF帧内容：{real_data.hex()}')
|
||
#入队
|
||
try:
|
||
q = bus_recv_queues.get(protocol)
|
||
if q:
|
||
try:
|
||
q.put_nowait({
|
||
"dev_name": dev_name,
|
||
"raw_data": real_data,
|
||
"protocol": protocol,
|
||
})
|
||
logger.debug(f'接受TCP重定向数据，存入bus_recv_queues，内容：{real_data.hex()}')
|
||
except Full as e:
|
||
print("队列已满",e)
|
||
|
||
except Exception as e:
|
||
print("其他异常",repr(e))
|
||
except:
|
||
pass
|
||
#切掉已经解析的帧
|
||
self.tcp_buf = self.tcp_buf[(t_pos+2) :]
|
||
|
||
|
||
|
||
'''
|
||
# 匹配设备名
|
||
dev_name = None
|
||
protocol = None
|
||
for name, cfg in DEVICE_CONFIG_DICT.items():
|
||
if cfg.get("tcp_bus_port") == self.remote_port:
|
||
dev_name = name
|
||
protocol = cfg.get("protocol")
|
||
break
|
||
|
||
if not dev_name or not protocol:
|
||
continue
|
||
|
||
logger.info(f'接受TCP重定向数据，存入bus_recv_queues，内容：{data.hex()}')
|
||
|
||
|
||
# 推入现有接收队列（完全兼容你原有架构）
|
||
q = bus_recv_queues.get(protocol)
|
||
if q:
|
||
try:
|
||
q.put_nowait({
|
||
"dev_name": dev_name,
|
||
"raw_data": data,
|
||
"protocol": protocol,
|
||
})
|
||
except:
|
||
pass
|
||
'''
|
||
except BlockingIOError:
|
||
time.sleep(0.001)
|
||
except Exception as e:
|
||
print(f"[TCP-UART] {self.dev_name} 接收异常: {e}")
|
||
break
|
||
|
||
self.running = False
|
||
|
||
def start(self):
|
||
if self.connect():
|
||
self.thread = threading.Thread(target=self.receive_loop, daemon=True)
|
||
self.thread.start()
|
||
tcp_clients[self.dev_name] = self
|
||
|
||
def stop(self):
|
||
self.running = False
|
||
if self.socket:
|
||
self.socket.close()
|
||
|
||
# ======================
|
||
# 批量启动所有UART TCP客户端
|
||
# ======================
|
||
def start_all_tcp_uart_clients():
|
||
from midware.config.base_config import DEVICE_CONFIG_DICT
|
||
|
||
for dev_name, cfg in DEVICE_CONFIG_DICT.items():
|
||
tcp_local = cfg.get("tcp_local_port", 1)
|
||
tcp_bus = cfg.get("tcp_bus_port", 1)
|
||
|
||
# 只启动端口 !=1 的UART设备
|
||
if tcp_local != 1 and tcp_bus != 1:
|
||
client = TCPUARTClient(
|
||
dev_name=dev_name,
|
||
local_port=tcp_local,
|
||
remote_port=tcp_bus
|
||
)
|
||
client.start()
|
||
time.sleep(0.1) |