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_retries3, retry_interval5): 连接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(hosthost, portport, timeout5) if client.connect(): print(fModbus连接成功 {host}:{port}) return client else: print(fModbus连接失败 {host}:{port}重试 {retry_count 1}/{max_retries}) retry_count 1 time.sleep(retry_interval) except Exception as e: print(fModbus连接异常: {e}重试 {retry_count 1}/{max_retries}) retry_count 1 time.sleep(retry_interval) print(fModbus连接失败达到最大重试次数 {max_retries}) return None def reconnect_modbus(modbus_client, host, port, max_retries3): 重新连接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(fSocket操作失败: {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(按 CtrlC 停止程序\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(timeout1.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) # CtrlC signal.signal(signal.SIGTERM, signal_handler) # 终止信号 # 加载配置文件 try: with open(RMS1000.json, r, encodingutf-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(fModbus配置: {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_size512) 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, durationNone) else: print(首次连接雷达失败开始自动重连流程...) # 直接进入自动重连流程 radar.continuous_receive_with_reconnect(modbus_client, durationNone) 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(程序已退出)