101 lines
3.1 KiB
Python
101 lines
3.1 KiB
Python
# Simulates the SDK heartbeat: connects to the virtual camera's TCP 6679,
|
|
# sends UpdateAck(203) heartbeats periodically, and verifies each response.
|
|
# If the connection drops or a response is wrong, the SDK would mark the
|
|
# device offline after 3 consecutive failures.
|
|
import socket
|
|
import struct
|
|
import time
|
|
import sys
|
|
|
|
HOST = "127.0.0.1"
|
|
PORT = 6679
|
|
|
|
def pack_frame(op, cmd, seq, payload=b""):
|
|
total = 24 + len(payload)
|
|
return (b"VZEB" + struct.pack("<IIII", total, op, cmd, seq) +
|
|
payload + b"VZEE")
|
|
|
|
def unpack(buf):
|
|
# returns (op, cmd, seq, payload, consumed) or None if incomplete
|
|
if len(buf) < 20:
|
|
return None
|
|
# find VZEB head
|
|
pos = buf.find(b"VZEB")
|
|
if pos < 0:
|
|
return None
|
|
total = struct.unpack("<I", buf[pos+4:pos+8])[0]
|
|
if len(buf) < pos + total:
|
|
return None
|
|
op, cmd, seq = struct.unpack("<III", buf[pos+8:pos+20])
|
|
tail = buf[pos+total-4:pos+total]
|
|
payload = buf[pos+20:pos+total-4]
|
|
if tail != b"VZEE":
|
|
return None
|
|
return (op, cmd, seq, payload, pos + total)
|
|
|
|
def main():
|
|
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
|
|
sock.settimeout(3.0)
|
|
try:
|
|
sock.connect((HOST, PORT))
|
|
except Exception as e:
|
|
print(f"[FAIL] connect failed: {e}")
|
|
sys.exit(1)
|
|
print("connected")
|
|
|
|
buf = b""
|
|
seq = 0
|
|
ok = 0
|
|
fail = 0
|
|
start = time.time()
|
|
while time.time() - start < 15:
|
|
seq += 1
|
|
# Send heartbeat (UpdateAck = 203, empty payload, Request)
|
|
sock.sendall(pack_frame(1, 203, seq))
|
|
t0 = time.time()
|
|
# Wait for the matching response
|
|
got = None
|
|
deadline = time.time() + 4.0
|
|
while time.time() < deadline:
|
|
try:
|
|
chunk = sock.recv(4096)
|
|
except socket.timeout:
|
|
break
|
|
if not chunk:
|
|
print("[FAIL] connection closed by server")
|
|
fail += 1
|
|
sock.close()
|
|
print(f"result: ok={ok} fail={fail}")
|
|
sys.exit(1)
|
|
buf += chunk
|
|
r = unpack(buf)
|
|
if r:
|
|
got = r
|
|
# Consume exactly this frame so a stale response isn't re-read.
|
|
buf = buf[r[4]:]
|
|
break
|
|
if not got:
|
|
print(f"[FAIL] heartbeat #{seq}: no response")
|
|
fail += 1
|
|
else:
|
|
op, cmd, rseq, payload, _ = got
|
|
if op == 2 and cmd == 203 and rseq == seq:
|
|
err = struct.unpack("<I", payload[:4])[0] if len(payload) >= 4 else -1
|
|
if err == 0:
|
|
ok += 1
|
|
print(f" OK heartbeat #{seq} err=0 ({time.time()-t0:.1f}s)")
|
|
else:
|
|
print(f"[FAIL] heartbeat #{seq} errCode={err}")
|
|
fail += 1
|
|
else:
|
|
print(f"[FAIL] heartbeat #{seq} unexpected resp op={op} cmd={cmd} seq={rseq}")
|
|
fail += 1
|
|
time.sleep(2.0)
|
|
|
|
sock.close()
|
|
print(f"\nresult: ok={ok} fail={fail}")
|
|
print("PASS" if fail == 0 else "FAIL")
|
|
|
|
if __name__ == "__main__":
|
|
main()
|