58 lines
1.7 KiB
Python
58 lines
1.7 KiB
Python
# Shared helpers for the 2026-09-22 primary-cause localization battery.
|
|
# Pure-ASCII output (run logs are parsed by probe_classify.py).
|
|
import socket, sys, time
|
|
|
|
IP, PORT = "192.168.1.100", 8080
|
|
|
|
def pattern(base, n):
|
|
return bytes(((base + i) & 0xFF) for i in range(n))
|
|
|
|
def connect():
|
|
s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
|
|
s.settimeout(3.0)
|
|
s.connect((IP, PORT))
|
|
print("TS %d" % int(time.time()))
|
|
return s
|
|
|
|
def wait_step(s, want, label, timeout_s=3.0):
|
|
"""recv until `want` bytes arrive or `timeout_s` of quiet; (data, stalled)"""
|
|
rx = b""
|
|
last = time.time()
|
|
while len(rx) < want:
|
|
try:
|
|
chunk = s.recv(want - len(rx))
|
|
except socket.timeout:
|
|
chunk = b""
|
|
except OSError:
|
|
chunk = b""
|
|
break
|
|
if chunk:
|
|
rx += chunk
|
|
last = time.time()
|
|
elif time.time() - last >= timeout_s:
|
|
break
|
|
print("STEP %s rx=%d/%d %s" % (label, len(rx), want,
|
|
"QUIET" if len(rx) < want else "OK"))
|
|
return rx, len(rx) < want
|
|
|
|
def send_rounds64(s, count, base, label, timeout_s=3.0):
|
|
"""send `count` 64B rounds (global-increment pattern), verify each echo inline."""
|
|
for i in range(count):
|
|
data = pattern(base + i * 64, 64)
|
|
s.sendall(data)
|
|
rx, stalled = wait_step(s, 64, "%s r%02d" % (label, i), timeout_s)
|
|
if stalled or rx != data:
|
|
return False
|
|
return True
|
|
|
|
def hexdump(data, prefix="HEX"):
|
|
for i in range(0, len(data), 16):
|
|
print("%s %05d %s" % (prefix, i, data[i:i + 16].hex()))
|
|
|
|
def finish(s, code):
|
|
try:
|
|
s.close()
|
|
except OSError:
|
|
pass
|
|
sys.exit(code)
|