58 lines
1.5 KiB
Python
58 lines
1.5 KiB
Python
import socket
|
|
import time
|
|
import struct
|
|
from udpmsg import *
|
|
|
|
# 目标IP地址和端口
|
|
target_ip = '127.0.0.1'
|
|
target_port = 5000
|
|
|
|
|
|
mTest = msg_CcuStateFbMsg()
|
|
mTest2 = msg_CcuSetParamFbMsg()
|
|
mTest3 = msg_disHighVolBusFbMsg()
|
|
|
|
valid_msg = mTest.pack()
|
|
valid_msg2 = mTest2.pack()
|
|
valid_msg3 = mTest3.pack()
|
|
print(valid_msg)
|
|
|
|
|
|
# 创建UDP套接字
|
|
sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
|
|
server_address = ('localhost', 5001)
|
|
sock.bind(server_address)
|
|
|
|
# 要发送的信息
|
|
# message = "Hello, this is a test message."
|
|
message = valid_msg3
|
|
# 发送信息的次数
|
|
num_messages = 100
|
|
|
|
# 发送信息的时间间隔（秒）
|
|
interval = 1
|
|
|
|
for i in range(num_messages):
|
|
# 发送信息
|
|
mTest.fcuState += 1
|
|
sock.sendto(mTest.pack(), (target_ip, target_port))
|
|
time.sleep(interval)
|
|
sock.sendto(msg_CcuSetParamFbMsg().pack(), (target_ip, target_port))
|
|
time.sleep(interval)
|
|
mTest3.busbarAVoltage += 1
|
|
sock.sendto(mTest3.pack(), (target_ip, target_port))
|
|
time.sleep(interval)
|
|
sock.sendto(msg_disHighAVolBusFbMsg().pack(), (target_ip, target_port))
|
|
time.sleep(interval)
|
|
sock.sendto(msg_disHighBVolBusFbMsg().pack(), (target_ip, target_port))
|
|
time.sleep(interval)
|
|
sock.sendto(msg_disLowBusFbMsg().pack(), (target_ip, target_port))
|
|
# 打印发送的信息和目标地址
|
|
print(f"Sent message to {target_ip}:{target_port}")
|
|
# 等待一段时间
|
|
# sock.sendto(valid_msg2, (target_ip, target_port))
|
|
time.sleep(interval)
|
|
|
|
# 关闭套接字
|
|
sock.close()
|