#!/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()