zjk
54c32a180a
- 新增 pCCU 复合管控器程序（src/pCCU）：FC/PM 双链路(UDP)、协议编解码器 (FcProtocol/PmProtocol/Frame/Message/MessageRegistry)、快照与落库(DbStore)、 内置 WebServer；挂入 src/CMakeLists.txt 与板卡 mission（h100.moos） - 惯导/深度数据：新增 uDevice_insData 消息解析，接入 UpperCommManager， pPowerMangerHost 注册监视，UpmsgTest 覆盖 - pPowerMangerHost 设备控制页增强（web/index.html） - 部署/运维脚本改为 4 个 systemd 服务（moosdb/pPowerManger/pPowerMangerHost/pCCU）： deploy/clean-data/fetch-data 同步适配 - 协议文档：归档 0824 控制主机与复合管控器通信协议 xlsx 及关联 docx，移除旧 md - 测试：新增 pccu 单元/集成测试与 disSysTest Python 配电仿真 - 清理：移除未使用的 CCU 状态调试打印 coutMsg；.gitignore 补充编译产物/临时文件忽略
96 lines
3.5 KiB
Python
96 lines
3.5 KiB
Python
#!/usr/bin/env python3
|
|
"""验证 pCCU WebSocket 推送的 JSON 快照格式与分组完整性。"""
|
|
import socket, base64, os, struct, time, json, sys
|
|
|
|
HOST, PORT = "127.0.0.1", 18081
|
|
|
|
class WsClient:
|
|
def __init__(self):
|
|
self.s = socket.create_connection((HOST, PORT), timeout=5)
|
|
key = base64.b64encode(os.urandom(16)).decode()
|
|
req = (f"GET /ws HTTP/1.1\r\nHost: {HOST}:{PORT}\r\n"
|
|
f"Upgrade: websocket\r\nConnection: Upgrade\r\n"
|
|
f"Sec-WebSocket-Key: {key}\r\nSec-WebSocket-Version: 13\r\n\r\n")
|
|
self.s.sendall(req.encode())
|
|
self.buf = b""
|
|
# 读取 HTTP 101 响应头
|
|
while b"\r\n\r\n" not in self.buf:
|
|
chunk = self.s.recv(4096)
|
|
if not chunk:
|
|
raise RuntimeError("连接关闭")
|
|
self.buf += chunk
|
|
head, self.buf = self.buf.split(b"\r\n\r\n", 1)
|
|
if b"101" not in head.split(b"\r\n")[0]:
|
|
raise RuntimeError("WebSocket 升级失败: " + head.decode(errors="replace"))
|
|
|
|
def _read(self, n):
|
|
while len(self.buf) < n:
|
|
chunk = self.s.recv(4096)
|
|
if not chunk:
|
|
raise RuntimeError("连接关闭")
|
|
self.buf += chunk
|
|
out, self.buf = self.buf[:n], self.buf[n:]
|
|
return out
|
|
|
|
def recv_text(self, timeout=6):
|
|
self.s.settimeout(timeout)
|
|
end = time.time() + timeout
|
|
while time.time() < end:
|
|
if len(self.buf) < 2:
|
|
try:
|
|
chunk = self.s.recv(4096)
|
|
except socket.timeout:
|
|
return None
|
|
if not chunk:
|
|
return None
|
|
self.buf += chunk
|
|
continue
|
|
b1, b2 = self.buf[0], self.buf[1]
|
|
opcode = b1 & 0x0F
|
|
length = b2 & 0x7F
|
|
hdr_len = 2
|
|
self.buf = self.buf[2:] # 消费帧头前2字节
|
|
if length == 126:
|
|
length = struct.unpack(">H", self._read(2))[0]
|
|
hdr_len += 2
|
|
elif length == 127:
|
|
length = struct.unpack(">Q", self._read(8))[0]
|
|
hdr_len += 8
|
|
masked = (b2 & 0x80) != 0
|
|
mask = self._read(4) if masked else b""
|
|
payload = self._read(length)
|
|
if opcode == 1: # text
|
|
if masked:
|
|
payload = bytes(b ^ mask[i % 4] for i, b in enumerate(payload))
|
|
return payload.decode("utf-8", "replace")
|
|
# ping/pong/close 跳过，继续
|
|
return None
|
|
|
|
def main():
|
|
try:
|
|
ws = WsClient()
|
|
except Exception as e:
|
|
print("FAIL: WS 连接失败:", e)
|
|
return 1
|
|
text = ws.recv_text()
|
|
if not text:
|
|
print("FAIL: 未收到快照")
|
|
return 1
|
|
try:
|
|
data = json.loads(text)
|
|
except Exception as e:
|
|
print("FAIL: 快照非合法JSON:", text[:200])
|
|
return 1
|
|
ok = True
|
|
for key in ("fc", "pmCmd", "links", "fcStatusCount", "pmControlCount", "logs"):
|
|
ok = ok and key in data
|
|
print(f"PASS: 快照 JSON 有效, 分组keys={sorted(data.keys())}")
|
|
print(f" fc.fc_status={data['fc']['fc_status']}, fc.fc_mode={data['fc']['fc_mode']}, "
|
|
f"heartbeat={data['fc']['heartbeat']}")
|
|
print(f" links={json.dumps(data['links'], ensure_ascii=False)}")
|
|
print(f" logs条数={len(data['logs'])}, fcStatusCount={data['fcStatusCount']}")
|
|
ws.s.close()
|
|
return 0 if ok else 1
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main()) |