RMS1000通信

1、获取状态，通过modbus_tcp与kepserver通信 有重连机制

import socket import time import json import struct import threading from typing import Optional, List from pymodbus.client.tcp import ModbusTcpClient as ModbusClient from enum import Enum # 全局变量初始化 out_list = [0, 0, 0, 0] prev_out_list = [-1, -1, -1, -1] # 用于存储上一次的输出状态 running = True # 全局运行标志 class ConnectionState(Enum): """连接状态枚举""" DISCONNECTED = 0 CONNECTING = 1 CONNECTED = 2 ERROR = 3 def TCP_Connect(host, port, max_retries=3, retry_interval=5): """ 连接Modbus TCP服务器，支持自动重连 Args: host: Modbus服务器IP port: Modbus服务器端口 max_retries: 最大重试次数 retry_interval: 重试间隔(秒) Returns: ModbusClient: 连接对象 """ retry_count = 0 while retry_count < max_retries and running: try: client = ModbusClient(host=host, port=port, timeout=5) if client.connect(): print(f'Modbus连接成功 {host}:{port}') return client else: print(f'Modbus连接失败 {host}:{port}，重试 {retry_count + 1}/{max_retries}') retry_count += 1 time.sleep(retry_interval) except Exception as e: print(f'Modbus连接异常: {e}，重试 {retry_count + 1}/{max_retries}') retry_count += 1 time.sleep(retry_interval) print(f'Modbus连接失败，达到最大重试次数 {max_retries}') return None def reconnect_modbus(modbus_client, host, port, max_retries=3): """ 重新连接Modbus Args: modbus_client: 现有的Modbus客户端 host: Modbus服务器IP port: Modbus服务器端口 max_retries: 最大重试次数 Returns: ModbusClient: 新的连接对象 """ if modbus_client and modbus_client.connected: try: modbus_client.close() except: pass print(f'尝试重新连接Modbus {host}:{port}') new_client = TCP_Connect(host, port, max_retries, 10) return new_client class RadarTCPClient: def __init__(self, host: str, port: int, receive_buffer_size: int = 1024): """ 初始化雷达TCP客户端 Args: host: 雷达设备的IP地址 port: 雷达设备的端口号 receive_buffer_size: 每次接收的字节数量 """ self.host = host self.port = port self.receive_buffer_size = receive_buffer_size self.socket = None self.connected = False self.connection_state = ConnectionState.DISCONNECTED self.lock = threading.RLock() # 线程锁 self.reconnect_interval = 10 # 重连间隔(秒) self.max_reconnect_attempts = 0 # 0表示无限重试 self.reconnect_attempts = 0 self.command_hex = '02 73 45 4E 20 4C 4D 44 72 61 64 61 72 64 61 74 61 20 31 03' # 雷达命令 def _safe_socket_operation(self, operation, *args, **kwargs): """安全的socket操作""" with self.lock: if self.socket: try: return operation(*args, **kwargs) except (socket.error, socket.timeout, OSError) as e: self.connected = False self.connection_state = ConnectionState.ERROR print(f"Socket操作失败: {e}") return None return None def connect(self, timeout: float = 5.0) -> bool: """ 连接到雷达设备 Args: timeout: 连接超时时间（秒） Returns: bool: 连接是否成功 """ with self.lock: try: if self.socket: try: self.socket.close() except: pass self.socket = None self.connection_state = ConnectionState.CONNECTING print(f"正在连接雷达设备 {self.host}:{self.port}...") self.socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM) self.socket.settimeout(timeout) self.socket.connect((self.host, self.port)) self.socket.settimeout(None) # 恢复阻塞模式 self.connected = True self.connection_state = ConnectionState.CONNECTED self.reconnect_attempts = 0 print(f"成功连接到雷达设备 {self.host}:{self.port}") return True except socket.timeout: print(f"连接雷达设备超时 ({timeout}秒)") self.connection_state = ConnectionState.ERROR return False except socket.error as e: print(f"连接雷达设备失败: {e}") self.connection_state = ConnectionState.ERROR return False except Exception as e: print(f"连接雷达设备时发生未知错误: {e}") self.connection_state = ConnectionState.ERROR return False def auto_reconnect(self): """自动重连雷达""" while running and not self.connected: self.connection_state = ConnectionState.CONNECTING if self.max_reconnect_attempts > 0 and self.reconnect_attempts >= self.max_reconnect_attempts: print(f"达到最大重连次数 {self.max_reconnect_attempts}，停止重连") return False print(f"尝试重连雷达 {self.host}:{self.port} (尝试 {self.reconnect_attempts + 1})") if self.connect(): # 重新发送命令 if self.send_hex_bytes(self.command_hex): print("雷达重连成功并重新发送命令") return True else: print("雷达重连成功但发送命令失败，继续重连") self.connected = False self.reconnect_attempts += 1 # 等待重连间隔 for i in range(self.reconnect_interval): if not running: return False time.sleep(1) return self.connected def send_hex_bytes(self, hex_data: str) -> bool: """ 发送16进制字节流到雷达 Args: hex_data: 16进制字符串，例如 "01 02 03 04" 或 "01020304" Returns: bool: 发送是否成功 """ if not self.connected: print("未连接到雷达设备，请先调用connect()") return False try: # 移除空格和换行符 hex_data = hex_data.replace(" ", "").replace("\n", "").replace("\r", "") # 验证是否为有效的16进制字符串 if not all(c in "0123456789ABCDEFabcdef" for c in hex_data): print("错误：无效的16进制字符串") return False # 确保长度为偶数 if len(hex_data) % 2 != 0: print("警告：16进制字符串长度应为偶数，已自动补0") hex_data = "0" + hex_data # 转换为字节流 data_bytes = bytes.fromhex(hex_data) # 发送数据 with self.lock: if self.socket: self.socket.sendall(data_bytes) print(f"发送成功: {hex_data.upper()} ({len(data_bytes)} 字节)") return True else: return False except socket.error as e: print(f"发送失败: {e}") self.connected = False self.connection_state = ConnectionState.ERROR return False except Exception as e: print(f"发送时发生未知错误: {e}") return False def receive_data(self, timeout: Optional[float] = None) -> Optional[bytes]: """ 接收指定字节数量的数据 Args: timeout: 接收超时时间，None表示阻塞 Returns: Optional[bytes]: 接收到的字节数据，失败则返回None """ if not self.connected: return None try: if timeout: self.socket.settimeout(timeout) data = self.socket.recv(self.receive_buffer_size) if timeout: self.socket.settimeout(None) if not data: print("雷达连接已关闭") self.connected = False self.connection_state = ConnectionState.DISCONNECTED return None return data except socket.timeout: if timeout: return None else: print("接收超时") return None except socket.error as e: print(f"接收失败: {e}") self.connected = False self.connection_state = ConnectionState.DISCONNECTED return None except Exception as e: print(f"接收时发生未知错误: {e}") return None def process_radar_data(self, byte_list): """ 处理雷达数据，提取输出状态 Args: byte_list: 字节列表 Returns: List[bool]: 输出状态列表 """ global out_list, prev_out_list try: # 不同款式的雷达采用不同的雷达偏移 if byte_list[53] == 32: leida = 9 elif byte_list[52] == 32: leida = 8 elif byte_list[51] == 32: leida = 7 else: # 如果没有匹配到，返回当前状态 return out_list if byte_list[44 + leida + 9] == 32: lei_da_pian_yi_liang = 9 + leida + 5 elif byte_list[44 + leida + 8] == 32: lei_da_pian_yi_liang = 8 + leida + 5 elif byte_list[44 + leida + 7] == 32: lei_da_pian_yi_liang = 7 + leida + 5 else: return out_list index = 44 + lei_da_pian_yi_liang if index < len(byte_list): byte_val = byte_list[index] if 48 <= byte_val <= 64: # 0-9 和 :;<=>@ out = byte_val - 48 elif byte_val == 65: # A out = 0xA elif byte_val == 66: # B out = 0xB elif byte_val == 67: # C out = 0xC elif byte_val == 68: # D out = 0xD elif byte_val == 69: # E out = 0xE elif byte_val == 70: # F out = 0xF else: return out_list # 计算当前输出状态 current_out_list = [ bool(0x01 & out), bool(0x02 & out), bool(0x04 & out), bool(0x08 & out) ] return current_out_list else: return out_list except IndexError as e: print(f"处理雷达数据时索引错误: {e}") return out_list except Exception as e: print(f"处理雷达数据时发生错误: {e}") return out_list def continuous_receive_with_reconnect(self, modbus_client, duration: Optional[float] = None): """ 持续接收数据，支持自动重连 Args: modbus_client: Modbus客户端 duration: 持续接收的时间（秒），None表示持续接收直到手动停止 """ global out_list, prev_out_list, running print(f"开始持续接收雷达数据，接收缓冲区大小: {self.receive_buffer_size} 字节") print("按 Ctrl+C 停止程序\n") start_time = time.time() packet_count = 0 try: while running and (duration is None or (time.time() - start_time) < duration): # 检查雷达连接状态 if not self.connected: print("雷达连接断开，尝试自动重连...") if not self.auto_reconnect(): print("自动重连失败，继续尝试...") continue # 接收数据 data = self.receive_data(timeout=1.0) if data is None: # 接收超时或失败，检查连接 if not self.connected: continue else: # 正常超时，继续下一次接收 continue packet_count += 1 hex_str = data.hex().upper() byte_data = bytes.fromhex(hex_str) byte_list = list(byte_data) # 处理雷达数据 current_out_list = self.process_radar_data(byte_list) # 只有当输出状态发生变化时才执行打印和写入操作 if current_out_list != prev_out_list: out_list = current_out_list.copy() # 更新全局out_list print(f"输出状态变化: {out_list}") # 写入Modbus if modbus_client and modbus_client.connected: try: # 将布尔列表转换为寄存器值 register_values = [1 if val else 0 for val in out_list] modbus_client.write_registers(0, register_values) except Exception as e: print(f"写入Modbus失败: {e}") # 尝试重新连接Modbus modbus_client = reconnect_modbus(modbus_client, modbus_client.host, modbus_client.port) else: print("Modbus连接已断开，尝试重新连接...") modbus_client = reconnect_modbus(modbus_client, modbus_client.host, modbus_client.port) prev_out_list = out_list.copy() # 更新上一次的状态 # 每100个数据包打印一次状态 if packet_count % 100 == 0: print(f"已接收 {packet_count} 个数据包，当前状态: {out_list}") except KeyboardInterrupt: print("\n\n用户中断程序") except Exception as e: print(f"接收数据时发生错误: {e}") print("尝试重新连接...") finally: print("结束雷达数据接收") def close(self): """关闭连接""" with self.lock: self.connected = False self.connection_state = ConnectionState.DISCONNECTED if self.socket: try: self.socket.close() except: pass self.socket = None print("雷达连接已关闭") def signal_handler(signum, frame): """信号处理函数""" global running print("\n接收到停止信号，正在关闭程序...") running = False # 示例使用 if __name__ == "__main__": import signal # 注册信号处理 signal.signal(signal.SIGINT, signal_handler) # Ctrl+C signal.signal(signal.SIGTERM, signal_handler) # 终止信号 # 加载配置文件 try: with open('RMS1000.json', 'r', encoding='utf-8') as file: data = json.load(file) # 雷达设备配置 RADAR_HOST = data.get("RMS1000IP", "192.168.1.100") # 雷达IP地址 RADAR_PORT = data.get("RMS1000PORT", 3000) # 雷达端口号 IP = data.get("KEPSERVER_IP", "192.168.1.200") PORT = data.get("KEPSERVER_PORT", 502) print(f"雷达配置: {RADAR_HOST}:{RADAR_PORT}") print(f"Modbus配置: {IP}:{PORT}") except FileNotFoundError: print("配置文件 RMS1000.json 未找到，使用默认配置") RADAR_HOST = "192.168.1.100" RADAR_PORT = 3000 IP = "192.168.1.200" PORT = 502 except Exception as e: print(f"加载配置文件失败: {e}") exit(1) # 连接Modbus modbus_client = TCP_Connect(IP, PORT) if not modbus_client: print("Modbus连接失败，程序退出") exit(1) # 创建雷达客户端 radar = RadarTCPClient(RADAR_HOST, RADAR_PORT, receive_buffer_size=512) try: # 首次连接雷达 if radar.connect(): # 发送初始化命令 if radar.send_hex_bytes('02 73 45 4E 20 4C 4D 44 72 61 64 61 72 64 61 74 61 20 31 03'): print('雷达初始化命令发送成功') else: print('雷达初始化命令发送失败，继续尝试接收数据') # 开始持续接收（支持自动重连） radar.continuous_receive_with_reconnect(modbus_client, duration=None) else: print("首次连接雷达失败，开始自动重连流程...") # 直接进入自动重连流程 radar.continuous_receive_with_reconnect(modbus_client, duration=None) except Exception as e: print(f"程序运行发生错误: {e}") finally: running = False if radar.connected: radar.close() if modbus_client and modbus_client.connected: modbus_client.close() print("程序已退出")