259 lines
9.3 KiB
Python
259 lines
9.3 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
数据中心 → 信号管理页面 (out) 添加信号功能测试 (v2)
|
||
使用原始帧接收方式（已验证可用）
|
||
"""
|
||
import socket, base64, os, struct, json, time, sys
|
||
|
||
IP = '198.120.0.100'
|
||
PORT = 8000
|
||
|
||
def ws_handshake(sock):
|
||
key = base64.b64encode(os.urandom(16)).decode()
|
||
req = (f"GET /ws HTTP/1.1\r\nHost: {IP}:{PORT}\r\nUpgrade: websocket\r\n"
|
||
f"Connection: Upgrade\r\nSec-WebSocket-Key: {key}\r\n"
|
||
f"Sec-WebSocket-Version: 13\r\n\r\n")
|
||
sock.sendall(req.encode())
|
||
resp = b""
|
||
while b"\r\n\r\n" not in resp:
|
||
c = sock.recv(4096)
|
||
if not c: break
|
||
resp += c
|
||
if b"101" not in resp.split(b"\r\n")[0]:
|
||
raise Exception(f"Handshake failed: {resp[:200]}")
|
||
return resp
|
||
|
||
def ws_send_text(sock, msg):
|
||
"""发送文本帧"""
|
||
if isinstance(msg, dict):
|
||
msg = json.dumps(msg, ensure_ascii=False)
|
||
payload = msg.encode()
|
||
L = len(payload)
|
||
mask = os.urandom(4)
|
||
masked = bytes(payload[i] ^ mask[i % 4] for i in range(L))
|
||
frame = bytearray([0x81])
|
||
if L < 126:
|
||
frame.append(0x80 | L)
|
||
elif L < 65536:
|
||
frame.append(0x80 | 126)
|
||
frame.extend(struct.pack('>H', L))
|
||
else:
|
||
frame.append(0x80 | 127)
|
||
frame.extend(struct.pack('>Q', L))
|
||
frame.extend(mask)
|
||
frame.extend(masked)
|
||
sock.sendall(bytes(frame))
|
||
|
||
def ws_recv_all(sock, timeout=2.0):
|
||
"""接收所有可用的 WebSocket 帧，解析后返回消息列表"""
|
||
sock.settimeout(timeout)
|
||
all_data = b""
|
||
try:
|
||
while True:
|
||
chunk = sock.recv(65536)
|
||
if not chunk:
|
||
break
|
||
all_data += chunk
|
||
except socket.timeout:
|
||
pass
|
||
|
||
msgs = []
|
||
offset = 0
|
||
while offset + 2 <= len(all_data):
|
||
b0 = all_data[offset]
|
||
b1 = all_data[offset + 1]
|
||
opcode = b0 & 0x0F
|
||
length = b1 & 0x7F
|
||
offset += 2
|
||
if length == 126:
|
||
if offset + 2 > len(all_data): break
|
||
length = struct.unpack('>H', all_data[offset:offset+2])[0]
|
||
offset += 2
|
||
elif length == 127:
|
||
if offset + 8 > len(all_data): break
|
||
length = struct.unpack('>Q', all_data[offset:offset+8])[0]
|
||
offset += 8
|
||
if offset + length > len(all_data):
|
||
break
|
||
payload = all_data[offset:offset+length]
|
||
offset += length
|
||
|
||
if opcode == 0x01: # text
|
||
try:
|
||
msgs.append(json.loads(payload.decode('utf-8')))
|
||
except:
|
||
msgs.append(payload.decode('utf-8', errors='replace'))
|
||
elif opcode == 0x02: # binary (dc_data 在这里)
|
||
try:
|
||
msgs.append(json.loads(payload.decode('utf-8')))
|
||
except:
|
||
msgs.append(payload)
|
||
elif opcode == 0x08: # close
|
||
pass
|
||
return msgs
|
||
|
||
|
||
def main():
|
||
print("=" * 70)
|
||
print(" RTU 数据中心 → 信号管理页(out) 添加信号功能测试")
|
||
print(f" 目标: {IP}:{PORT}")
|
||
print("=" * 70)
|
||
|
||
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
|
||
sock.settimeout(10)
|
||
sock.connect((IP, PORT))
|
||
print(f"[✓] TCP 连接成功")
|
||
|
||
ws_handshake(sock)
|
||
print(f"[✓] WebSocket 握手成功")
|
||
|
||
# ---- 0. 初始握手：获取 cmd_list ----
|
||
ws_send_text(sock, {"type": "get_cmds"})
|
||
time.sleep(0.5)
|
||
msgs = ws_recv_all(sock, timeout=2.0)
|
||
for msg in msgs:
|
||
if isinstance(msg, dict) and msg.get('type') == 'cmd_list':
|
||
print(f"[✓] cmd_list: {len(msg.get('cmds', []))} 个命令")
|
||
for cmd in msg.get('cmds', []):
|
||
print(f" - {cmd.get('name')}: {cmd.get('desc')}")
|
||
|
||
# ---- 1. 注册 dc_data 信号 ----
|
||
print("\n" + "=" * 70)
|
||
print(" 步骤 1: 通过 datacenter 命令注册所有信号")
|
||
print("=" * 70)
|
||
|
||
all_dc_data = {}
|
||
for st in ['out', 'in', 'yk', 'ao', 'param']:
|
||
print(f"\n [→] datacenter {st}")
|
||
ws_send_text(sock, {"type": "cmd", "cmd": f"datacenter {st}"})
|
||
time.sleep(1.5)
|
||
msgs = ws_recv_all(sock, timeout=3.0)
|
||
for msg in msgs:
|
||
if isinstance(msg, dict):
|
||
if msg.get('type') == 'dc_data':
|
||
all_dc_data[st] = msg
|
||
cnt = len(msg.get(st, []))
|
||
print(f" [✓] dc_data: {st} = {cnt} 个信号")
|
||
if cnt > 0:
|
||
for item in msg.get(st, [])[:5]:
|
||
print(f" saddr={item.get('saddr','?')} desc={item.get('desc','?')} ctrl={item.get('ctrl_type','?')}")
|
||
if cnt > 5:
|
||
print(f" ... 还有 {cnt-5} 个")
|
||
elif msg.get('type'):
|
||
pass # 忽略
|
||
elif isinstance(msg, str) and len(msg) < 300:
|
||
print(f" [cmd输出] {msg.strip()[:200]}")
|
||
|
||
# ---- 2. 选择 out 信号 ----
|
||
print("\n" + "=" * 70)
|
||
print(" 步骤 2: 选择 out 信号并添加到配置页")
|
||
print("=" * 70)
|
||
|
||
dc_out = all_dc_data.get('out', {})
|
||
out_signals = dc_out.get('out', [])
|
||
|
||
if not out_signals:
|
||
print(" [!] 没有可用的 out 信号！")
|
||
print(" 尝试直接添加一个已知信号测试...")
|
||
# 尝试添加一个可能存在的信号
|
||
test_saddr = "iec.self_check"
|
||
print(f" [→] 尝试添加: {test_saddr}")
|
||
ws_send_text(sock, {"saddr": test_saddr, "signal_type": "out", "curd": "add", "setting_zone": "0", "signal_data": ""})
|
||
time.sleep(3.0)
|
||
msgs = ws_recv_all(sock, timeout=3.0)
|
||
print(f" 收到 {len(msgs)} 条消息")
|
||
for msg in msgs:
|
||
if isinstance(msg, dict):
|
||
print(f" type={msg.get('type','?')}")
|
||
out_data = msg.get('out')
|
||
if out_data is not None:
|
||
print(f" out 信号数: {len(out_data) if isinstance(out_data, list) else '?'}")
|
||
if isinstance(out_data, list):
|
||
for o in out_data:
|
||
print(f" saddr={o.get('saddr','?')} val={o.get('val','?')}")
|
||
else:
|
||
# 选择前3个信号
|
||
test_signals = out_signals[:3]
|
||
test_saddrs = {s.get('saddr') for s in test_signals}
|
||
print(f" 从 {len(out_signals)} 个信号中选择 {len(test_signals)} 个:")
|
||
for i, sig in enumerate(test_signals):
|
||
print(f" [{i+1}] saddr={sig.get('saddr')} desc={sig.get('desc','?')}")
|
||
|
||
# ---- 3. 执行 curd:add ----
|
||
print("\n" + "=" * 70)
|
||
print(" 步骤 3: 执行 curd:add")
|
||
print("=" * 70)
|
||
|
||
for i, sig in enumerate(test_signals):
|
||
saddr = sig.get('saddr')
|
||
print(f"\n [{i+1}] curd:add saddr={saddr}")
|
||
ws_send_text(sock, {"saddr": saddr, "signal_type": "out", "curd": "add", "setting_zone": "0", "signal_data": ""})
|
||
time.sleep(2.0)
|
||
|
||
# ---- 4. 等待 ws_task 推送 out 数据 ----
|
||
print("\n" + "=" * 70)
|
||
print(" 步骤 4: 等待 ws_task 推送 out 数据（最多15秒）")
|
||
print("=" * 70)
|
||
|
||
out_received = False
|
||
out_data = []
|
||
start = time.time()
|
||
while time.time() - start < 15.0:
|
||
msgs = ws_recv_all(sock, timeout=3.0)
|
||
for msg in msgs:
|
||
if isinstance(msg, dict):
|
||
out_arr = msg.get('out')
|
||
if isinstance(out_arr, list) and len(out_arr) > 0:
|
||
out_data = out_arr
|
||
out_received = True
|
||
break
|
||
if out_received:
|
||
break
|
||
if msgs:
|
||
print(f" [i] 等待中... ({time.time()-start:.0f}s)")
|
||
|
||
# ---- 5. 验证 ----
|
||
print("\n" + "=" * 70)
|
||
print(" 步骤 5: 验证结果")
|
||
print("=" * 70)
|
||
|
||
if out_received:
|
||
print(f" [✓] 收到 out 数据: {len(out_data)} 个信号")
|
||
registered = {o.get('saddr') for o in out_data}
|
||
for o in out_data:
|
||
marker = " ← 新添加" if o.get('saddr') in test_saddrs else ""
|
||
print(f" saddr={o.get('saddr','?')} val={o.get('val','?')} desc={o.get('desc','?')}{marker}")
|
||
|
||
all_found = test_saddrs.issubset(registered)
|
||
missing = test_saddrs - registered
|
||
if all_found:
|
||
print(f"\n [✓] 所有 {len(test_saddrs)} 个信号已成功注册到 out!")
|
||
else:
|
||
print(f"\n [✗] 缺少 {len(missing)} 个: {missing}")
|
||
print(f" 当前已注册: {registered}")
|
||
|
||
# ---- 6. 清理 ----
|
||
print("\n" + "=" * 70)
|
||
print(" 步骤 6: 清理 — 删除测试信号")
|
||
print("=" * 70)
|
||
for sig in test_signals:
|
||
saddr = sig.get('saddr')
|
||
print(f" [→] del saddr={saddr}")
|
||
ws_send_text(sock, {"saddr": saddr, "signal_type": "out", "curd": "del", "setting_zone": "0", "signal_data": ""})
|
||
time.sleep(0.3)
|
||
else:
|
||
print(" [✗] 未收到 out 数据推送")
|
||
print(" 可能原因: add_signal 失败，或 ws_task 未推送变化")
|
||
|
||
# 清理 - 发送关闭帧
|
||
try:
|
||
sock.sendall(b'\x88\x80' + os.urandom(4)) # close frame
|
||
except:
|
||
pass
|
||
sock.close()
|
||
print("\n[✓] 测试完成")
|
||
|
||
|
||
if __name__ == '__main__':
|
||
main()
|