增加了消息读取的功能,并且添加了python测试脚本
This commit is contained in:
@@ -0,0 +1,44 @@
|
||||
import socket
|
||||
import time
|
||||
import struct
|
||||
from udpmsg import msg_CcuStateFbMsg
|
||||
from udpmsg import msg_CcuSetParamFbMsg
|
||||
# 目标IP地址和端口
|
||||
target_ip = '127.0.0.1'
|
||||
target_port = 5000
|
||||
|
||||
|
||||
mTest = msg_CcuStateFbMsg()
|
||||
mTest2 = msg_CcuSetParamFbMsg()
|
||||
|
||||
valid_msg = mTest.pack()
|
||||
valid_msg2 = mTest2.pack()
|
||||
print(valid_msg)
|
||||
|
||||
|
||||
# 创建UDP套接字
|
||||
sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
|
||||
|
||||
# 要发送的信息
|
||||
# message = "Hello, this is a test message."
|
||||
message = valid_msg
|
||||
# 发送信息的次数
|
||||
num_messages = 10
|
||||
|
||||
# 发送信息的时间间隔(秒)
|
||||
interval = 1
|
||||
|
||||
for i in range(num_messages):
|
||||
# 发送信息
|
||||
sock.sendto(message, (target_ip, target_port))
|
||||
|
||||
# 打印发送的信息和目标地址
|
||||
print(f"Sent message to {target_ip}:{target_port}: {message}")
|
||||
|
||||
# 等待一段时间
|
||||
time.sleep(interval)
|
||||
sock.sendto(valid_msg2, (target_ip, target_port))
|
||||
time.sleep(interval)
|
||||
|
||||
# 关闭套接字
|
||||
sock.close()
|
||||
Reference in New Issue
Block a user