122 lines
4.8 KiB
Python
122 lines
4.8 KiB
Python
# Simulates the VZNLSDK Xilinx discovery: sends the exact 35-byte search
|
|
# request, then parses the virtual camera's response exactly the way
|
|
# _ParseXilinxDeviceInfo does, printing each validation step.
|
|
import socket
|
|
import struct
|
|
import sys
|
|
import time
|
|
|
|
UDP_PORT = 6789
|
|
TARGET = ("127.0.0.1", UDP_PORT)
|
|
|
|
# Exact 35-byte search request from VzXilinxDeviceEnumrator.cpp s_szRequest[]
|
|
search_req = bytes([
|
|
0xe1, 0xe2, 0xe3, 0xe4, # preamble
|
|
0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, # serial (broadcast)
|
|
0x53, 0x42, # "SB" start
|
|
0x00, 0x02, # htons(SearchDev) = 2
|
|
0x00, 0x00, 0x00, 0x00, # reserve
|
|
0x00, 0x17, # htons(23) group length
|
|
0x00, 0x00, 0x00, 0x00, # CRC
|
|
0x00, 0x00, # command type
|
|
0x00, 0x00, # command length
|
|
0x00, 0x00, # command seq
|
|
0x01, # data byte
|
|
0x45, 0x42, # "EB" end
|
|
])
|
|
print(f"Search request: {len(search_req)} bytes")
|
|
|
|
sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
|
|
sock.settimeout(3.0)
|
|
sock.bind(("127.0.0.1", 0)) # recv port 0 (OS-assigned), like the SDK
|
|
local_port = sock.getsockname()[1]
|
|
print(f"Bound local port {local_port}")
|
|
|
|
sock.sendto(search_req, TARGET)
|
|
print(f"Sent search to {TARGET}")
|
|
|
|
try:
|
|
data, addr = sock.recvfrom(4096)
|
|
except socket.timeout:
|
|
print("TIMEOUT: no response received")
|
|
sys.exit(1)
|
|
|
|
print(f"Received {len(data)} bytes from {addr}")
|
|
print(f"Raw hex: {data.hex()}\n")
|
|
fail = 0
|
|
|
|
def ok(cond, label):
|
|
global fail
|
|
mark = "OK" if cond else "*** FAIL ***"
|
|
if not cond:
|
|
fail += 1
|
|
print(f" {mark}: {label}")
|
|
|
|
# ---- Parse exactly like _ParseXilinxDeviceInfo ----
|
|
preamble = struct.unpack("<I", data[0:4])[0]
|
|
ok(preamble == 0xE4E3E2E1, f"preamble == 0xe4e3e2e1 (got 0x{preamble:08x})")
|
|
|
|
sn = data[4:12]
|
|
print(f" INFO: SN = {sn.hex()}")
|
|
|
|
shCmdType = socket.ntohs(struct.unpack("<H", data[12:14])[0])
|
|
ok(shCmdType == 0x40, f"shCmdType(ntohs) == 0x40 (got 0x{shCmdType:04x})")
|
|
|
|
shCommand = socket.ntohs(struct.unpack("<H", data[14:16])[0])
|
|
print(f" INFO: shCommand = 0x{shCommand:04x}")
|
|
shCommandLength = struct.unpack("<H", data[16:18])[0]
|
|
print(f" INFO: shCommandLength = {shCommandLength}")
|
|
shCommandSequen = struct.unpack("<H", data[18:20])[0]
|
|
print(f" INFO: shCommandSequen = {shCommandSequen}")
|
|
nReserve = struct.unpack("<I", data[20:24])[0]
|
|
print(f" INFO: nReserve = {nReserve}")
|
|
shPackageHead = struct.unpack("<H", data[24:26])[0]
|
|
print(f" INFO: shPackageHead = 0x{shPackageHead:04x}")
|
|
shAckType = socket.ntohs(struct.unpack("<H", data[26:28])[0])
|
|
print(f" INFO: shAckType = 0x{shAckType:04x}")
|
|
shCommandLen = socket.ntohs(struct.unpack("<H", data[28:30])[0])
|
|
checksum = struct.unpack("<I", data[30:34])[0]
|
|
shCmdSeq = struct.unpack("<H", data[34:36])[0]
|
|
print(f" INFO: shCommandLen = {shCommandLen}, checksum = {checksum}, seq = {shCmdSeq}")
|
|
|
|
remain = shCommandLen - 14
|
|
ok(remain >= 68, f"shCommandLen-14 = {remain} >= sizeof(SVzXilinxDeviceInfo)=68")
|
|
|
|
if remain >= 68:
|
|
# Try to read 68-byte SVzXilinxDeviceInfo
|
|
if len(data) >= 36 + 68:
|
|
dev = data[36:104]
|
|
print(f" INFO: devInfo size = {len(dev)}")
|
|
productType = dev[0]
|
|
verCode = struct.unpack("<I", dev[4:8])[0]
|
|
ver = dev[8:40].split(b"\0")[0].decode("ascii", "replace")
|
|
w = struct.unpack("<H", dev[40:42])[0]
|
|
h = struct.unpack("<H", dev[42:44])[0]
|
|
ipType = dev[44]
|
|
ip = ".".join(str(b) for b in dev[45:49])
|
|
mac = ":".join(f"{b:02x}" for b in dev[49:55])
|
|
gw = ".".join(str(b) for b in dev[55:59])
|
|
dev_sn = dev[59:67].hex()
|
|
print(f" INFO: productType={productType} verCode={verCode} ver='{ver}'")
|
|
print(f" INFO: res={w}x{h} ipType={ipType}")
|
|
print(f" INFO: IP={ip} MAC={mac} GW={gw} SN={dev_sn}")
|
|
ok(len(data) >= 104, "buffer has all 68 devInfo bytes")
|
|
else:
|
|
ok(False, f"buffer too short for devInfo ({len(data)} < 104)")
|
|
|
|
# tail handling
|
|
pos = 36 + 68
|
|
remain2 = len(data) - pos
|
|
print(f" INFO: bytes after devInfo = {remain2}")
|
|
if remain2 == 2:
|
|
# SDK reads shPackageEnd via native LE but NEVER validates its value
|
|
# (_ParseXilinxDeviceInfo returns true once the devInfo is parsed).
|
|
pkgEnd = struct.unpack("<H", data[pos:pos+2])[0]
|
|
print(f" INFO: pkgEnd LE value 0x{pkgEnd:04x} (SDK does not validate this)")
|
|
ok(True, "exactly 2 bytes after devInfo (SDK takes the ==2 branch)")
|
|
else:
|
|
ok(False, f"expected exactly 2 bytes after devInfo, got {remain2}")
|
|
print(f" INFO: extra bytes hex = {data[pos:].hex()}")
|
|
|
|
print(f"\n=== RESULT: {'PASS - SDK would discover this device' if fail==0 else f'{fail} FAILURE(S) - device would NOT be discovered'} ===")
|