需求：python发送udp数据包数据，支持host、port、valueFile等启动参数，其中valueFile是json格式的文件，要解析编码完成转换。
分模块实现：
发送端：
导入所需的模块和类，例如sys、logging、argparse、json、socket以及datetime和timedelta。
定义一个Sender类，变量MULTI_TABLE_HEADER和SINGLE_TABLE_HEADER分别表示多表模式和单表模式下的消息头；变量interval表示发送消息的时间间隔，默认为5秒；变量is_multi_table表示是否使用多表模式，默认为True；变量udp_host和udp_port分别表示UDP服务器的主机和端口，默认为'localhost'和12345；变量value_file表示JSON值文件的路径，默认为'value.json'。
MULTI_TABLE_HEADER = "80000000" SINGLE_TABLE_HEADER = "80000001" interval = 5 is_multi_table = True udp_host = 'localhost' udp_port = 12345 value_file = 'resources/value.json'
方法send_message_to_udp_server用于将消息发送到UDP服务器，使用socketAPI。
def send_message_to_udp_server(self,host, port, message): sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) # message = b'Hello, UDP!' sock.sendto(message, (host, port)) sock.close()
方法run是主要的运行逻辑。它读取指定的JSON文件，并构造一个DqMessage对象。然后循环发送消息到UDP服务器，发送完成后等待一定时间再次发送，直到程序停止运行,同时保存日志信息。
def run(self): # 读取valueFile指定的JSON文件 with open(self.value_file) as file: dq_params = json.load(file) # 构造DqMessage对象 dq_message = DqMessage() dq_message.header = self.MULTI_TABLE_HEADER if self.is_multi_table else self.SINGLE_TABLE_HEADER dq_message.dataList = [ DqParam(dq_param['tableCode'], dq_param['seq'], dq_param['type'], dq_param['value']) for dq_param in dq_params ] print(dq_message.dataList) # 设置日志输出格式和保存路径 logging.basicConfig(filename='send_msg.log', format='%(asctime)s - %(levelname)s - %(message)s', level=logging.INFO) while True: # 获取当前时间 start_time = datetime.now() # 构造并发送数据包 message = dq_message.to_byte_buf(self.is_multi_table) self.send_message_to_udp_server(self.udp_host, self.udp_port, message) # 输出日志信息 logging.info(f"发送内容16进制：{message}") logging.info(f"UDP消息已发送至 {self.udp_host}:{self.udp_port}") # 计算下一次发送的时间点 end_time = datetime.now() elapsed_time = (end_time - start_time).total_seconds() sleep_time = max(self.interval - elapsed_time, 0) sleep_time = min(sleep_time, self.interval) next_send_time = end_time + timedelta(seconds=sleep_time) # 输出日志信息 logging.info(f"下一次发送时间：{next_send_time.strftime('%Y-%m-%d %H:%M:%S')}") # 等待到下一次发送的时间点 while datetime.now() < next_send_time: pass
数据封装模块：Dqmessage和Dqparam是实现编码转换和数据封装的模块，不与展示，根据数据具体实现。
接收端验证主要部分：
def receive_udp_message(self): sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) sock.bind((self.host, self.port)) while True: data, addr = sock.recvfrom(1024) # 接收数据包和发送方地址 print('Received message:', data) # 输出日志信息 logging.info(f"接收消息:{data}") hex_str = binascii.hexlify(data).decode('utf-8') print('verified message:', hex_str) logging.info(f"验证消息:{hex_str}")
