62 lines
2.2 KiB
Python
62 lines
2.2 KiB
Python
import socket
|
|
import time
|
|
from binascii import unhexlify
|
|
|
|
# 目标服务器地址和端口
|
|
TARGET_HOST = 'localhost' # 请根据实际情况修改
|
|
TARGET_PORT = 8888 # 请根据实际情况修改
|
|
|
|
# 你的注册指令数据
|
|
REGISTER_DATA_HEX = "7e80D2DA725711FB7e"
|
|
|
|
# 你的Modbus指令数据 01034040fb6f6141bd5555bfc82e00bd7d2000c2b0e1063e90c9983c9df308be91110041d9e3753fedf7df44789610c1d60000bf555555c477900041e1058342e1b9874f50
|
|
MODBUS_DATA_HEX = "01034040fb6f6141bd5555bfc82e00bd7d2000c2b0e1063e90c9983c9df308be91110041d9e3753fedf7df44789610c1d60000bf555555c477900041e1058342e1b9874f50"
|
|
|
|
def main():
|
|
# 将十六进制字符串转换为字节
|
|
register_bytes = unhexlify(REGISTER_DATA_HEX)
|
|
modbus_bytes = unhexlify(MODBUS_DATA_HEX)
|
|
|
|
try:
|
|
# 创建TCP套接字
|
|
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
|
|
# 设置连接超时
|
|
sock.settimeout(5)
|
|
print(f"正在连接到 {TARGET_HOST}:{TARGET_PORT}...")
|
|
sock.connect((TARGET_HOST, TARGET_PORT))
|
|
print("已连接。")
|
|
|
|
# 1. 发送注册包
|
|
print("发送注册包...")
|
|
sock.sendall(register_bytes)
|
|
print("注册包已发送。")
|
|
|
|
# 2. 等待0.5秒
|
|
print("等待 0.5 秒...")
|
|
time.sleep(0.5)
|
|
|
|
# 3. 发送Modbus数据包
|
|
print("发送Modbus数据包...")
|
|
sock.sendall(modbus_bytes)
|
|
print("Modbus数据包已发送。")
|
|
|
|
# （可选）接收最终响应
|
|
try:
|
|
final_response = sock.recv(1024)
|
|
if final_response:
|
|
print(f"收到最终响应: {final_response.hex()}")
|
|
else:
|
|
print("收到空的最终响应。")
|
|
except socket.timeout:
|
|
print("等待最终响应超时。")
|
|
# 忽略其他接收错误，直接结束
|
|
|
|
except ConnectionRefusedError:
|
|
print(f"错误: 连接被拒绝，请确认服务器在 {TARGET_HOST}:{TARGET_PORT} 上运行。")
|
|
except ValueError as e:
|
|
print(f"错误: 无效的十六进制字符串. {e}")
|
|
except Exception as e:
|
|
print(f"发生错误: {e}")
|
|
|
|
if __name__ == "__main__":
|
|
main() |