200 lines
8.5 KiB
Python
200 lines
8.5 KiB
Python
# Offline verdict for the 2026-09-22 primary-cause localization battery (v4).
|
|
# usage:
|
|
# python MDK-ARM/probe_classify.py A <runlog> <seriallog> <run_index> # boundary (one-shot or sustained)
|
|
# python MDK-ARM/probe_classify.py B <runlog> <seriallog> <run_index> # status (hold)
|
|
# python MDK-ARM/probe_classify.py C <runlog> <seriallog> <run_index> # partial drain
|
|
# python MDK-ARM/probe_classify.py F <runlog> <seriallog> <base_index> # fast-recon chain (10 pairs)
|
|
# runlog = the run's stdout capture (TS/STEP lines + HEX dump, ASCII)
|
|
# seriallog = tools/serial_tool/state/serial.log (the session's log)
|
|
# run_index = 1-based position of this run in the session (A/B/C)
|
|
# or 0-based first connection index (F).
|
|
# Sustained probe (MODE 2): the per-run eaten bytes = sum of POSITIVE probe-line
|
|
# n's in the run's device-time window (TS lines in the runlog, 8h PC->device
|
|
# offset + 4s margins; 15s run gaps keep windows disjoint).
|
|
import re, sys
|
|
from datetime import datetime, timezone, timedelta
|
|
|
|
CST = timezone(timedelta(hours=8)) # PC local; device logs in UTC
|
|
|
|
def pattern(base, n):
|
|
return bytes(((base + i) & 0xFF) for i in range(n))
|
|
|
|
def delete_block(data, off, n):
|
|
if n <= 0:
|
|
return data
|
|
return data[:off] + data[off + n:]
|
|
|
|
def run_timestamps(path):
|
|
ts = []
|
|
for raw in open(path, encoding="utf-8-sig", errors="replace"):
|
|
m = re.match(r"TS (\d+)", raw.strip())
|
|
if m:
|
|
ts.append(int(m.group(1)))
|
|
return ts
|
|
|
|
def read_run(path):
|
|
steps, streams = {}, {}
|
|
cur = None
|
|
for raw in open(path, encoding="utf-8-sig", errors="replace"):
|
|
line = raw.strip()
|
|
m = re.match(r"PAIR (\d+)", line) or re.match(r"CONN (\d+) canary=\d", line)
|
|
if m:
|
|
cur = int(m.group(1))
|
|
continue
|
|
if re.match(r"TS \d+", line):
|
|
continue
|
|
m = re.match(r"STEP (\S+) rx=(\d+)/(\d+) (OK|QUIET)", line)
|
|
if m:
|
|
steps[(cur, m.group(1))] = (int(m.group(2)), int(m.group(3)), m.group(4))
|
|
continue
|
|
m = re.match(r"(HEX\w+) (\d{5}) ([0-9a-f]*)", line)
|
|
if m:
|
|
streams.setdefault(m.group(1), []).append(
|
|
bytes.fromhex(m.group(3)) if m.group(3) else b"")
|
|
return steps, {k: b"".join(v) for k, v in streams.items()}
|
|
|
|
DEV_RE = re.compile(r"\[(\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2})\]\[")
|
|
def probe_events(seriallog):
|
|
"""(device_utc_epoch, n) for every probe line, in log order."""
|
|
hits = []
|
|
for raw in open(seriallog, encoding="utf-8", errors="replace"):
|
|
m = re.search(r"probe: recv64 n=(-?\d+)", raw)
|
|
if not m:
|
|
continue
|
|
t = DEV_RE.search(raw)
|
|
if not t:
|
|
continue
|
|
dt = datetime.strptime(t.group(1), "%Y-%m-%d %H:%M:%S").replace(tzinfo=timezone.utc)
|
|
hits.append((dt.timestamp(), int(m.group(1))))
|
|
return hits
|
|
|
|
def run_eaten(events, ts, ts_next):
|
|
"""Sum of positive probe n's in this run's device-time window."""
|
|
if not ts:
|
|
return None
|
|
lo = ts[0] - 4.0
|
|
hi = (ts_next - 4.0) if ts_next is not None else (lo + 14.0)
|
|
tot = 0
|
|
for t, n in events:
|
|
if lo <= t < hi and n > 0:
|
|
tot += n
|
|
return tot
|
|
|
|
def single_eaten(events, ts, ts_next):
|
|
"""One-shot probe: the (single) n in the run window; None if absent."""
|
|
if not ts:
|
|
return None
|
|
lo = ts[0] - 4.0
|
|
hi = (ts_next - 4.0) if ts_next is not None else (lo + 14.0)
|
|
for t, n in events:
|
|
if lo <= t < hi:
|
|
return n
|
|
return None
|
|
|
|
def main():
|
|
mode, runlog, seriallog = sys.argv[1], sys.argv[2], sys.argv[3]
|
|
base = int(sys.argv[4])
|
|
steps, streams = read_run(runlog)
|
|
ts_all = run_timestamps(runlog)
|
|
events = probe_events(seriallog)
|
|
|
|
if mode == "F":
|
|
for pair in range(10):
|
|
ci, cj = 2 * pair, 2 * pair + 1
|
|
nA = single_eaten(events, [ts_all[ci]] if ci < len(ts_all) else [],
|
|
[ts_all[cj + 1]] if cj + 1 < len(ts_all) else None) or 0
|
|
nB = single_eaten(events, [ts_all[cj]] if cj < len(ts_all) else [],
|
|
[ts_all[cj + 1]] if cj + 1 < len(ts_all) else None) or 0
|
|
gotA = streams.get("HEXc%02dF1" % ci, b"")
|
|
gotB1 = streams.get("HEXc%02dF1" % cj, b"")
|
|
gotG = streams.get("HEXc%02dG" % cj, b"")
|
|
wantA = 2048 - nA
|
|
wantB1 = 2048 - nB
|
|
a_ok = len(gotA) == wantA and gotA == delete_block(pattern(640, 2048), 0, nA)
|
|
b1_ok = len(gotB1) == wantB1 and gotB1 == delete_block(pattern(640, 2048), 0, nB)
|
|
expF2, expTail = pattern(2688, 2048), pattern(4736, 640)
|
|
if len(gotG) < 2048 or gotG[:2048] != expF2:
|
|
v = "WEDGE@F2"
|
|
elif len(gotG) < 2688 or gotG[2048:] != expTail:
|
|
v = "WEDGE@TAIL"
|
|
else:
|
|
v = "PASS"
|
|
if not a_ok:
|
|
v = "WEDGE@A" if len(gotA) < wantA else "MISMATCH@A"
|
|
elif not b1_ok:
|
|
v = "WEDGE@B1" if len(gotB1) < wantB1 else "MISMATCH@B1"
|
|
print("PAIR %d: %-12s A %d/%d B1 %d/%d F2 %d/2688 (nA=%s nB=%s)"
|
|
% (pair, v, len(gotA), wantA, len(gotB1), wantB1, len(gotG), nA, nB))
|
|
return
|
|
|
|
ts = [ts_all[base - 1]] if base - 1 < len(ts_all) else []
|
|
ts_next = ts_all[base] if (base < len(ts_all) and len(ts_all) > 1) else None
|
|
eaten = single_eaten(events, ts, ts_next)
|
|
eaten_sum = run_eaten(events, ts, ts_next) or 0
|
|
eaten_pos = eaten if (eaten is not None and eaten > 0) else 0
|
|
f = streams.get("HEX", b"")
|
|
|
|
if mode == "A":
|
|
# dual expectation: one-shot (first probe line) or sustained (window sum)
|
|
cands = {eaten_pos}
|
|
if eaten_sum != eaten_pos:
|
|
cands.add(eaten_sum)
|
|
verdict = None
|
|
for e in sorted(cands):
|
|
want_f = 4096 - e
|
|
exp_f = delete_block(pattern(2100, 4096), 2048, e)
|
|
got_f, got_tail = f[:want_f], f[want_f:]
|
|
if len(got_f) == want_f and got_f == exp_f and len(got_tail) == 640 \
|
|
and got_tail == pattern(6196, 640):
|
|
verdict = ("PASS", e)
|
|
break
|
|
if verdict:
|
|
print("VERDICT PASS | eaten=%d (probe n=%s, window sum=%d) | f %d/4096 + tail 640"
|
|
% (verdict[1], eaten, eaten_sum, 4096 - verdict[1]))
|
|
else:
|
|
want = 4096
|
|
if len(f) < 2048:
|
|
print("VERDICT STALL@F1 | probe n=%s sum=%d | stream %d bytes (fill1 echo lost)"
|
|
% (eaten, eaten_sum, len(f)))
|
|
elif len(f) < 4096 - eaten_sum:
|
|
print("VERDICT STALL@F2 | probe n=%s sum=%d | stream %d bytes (fill2 echo cut)"
|
|
% (eaten, eaten_sum, len(f)))
|
|
else:
|
|
print("VERDICT MISMATCH | probe n=%s sum=%d | stream %d bytes, bytes differ"
|
|
% (eaten, eaten_sum, len(f)))
|
|
|
|
elif mode == "B":
|
|
want_f2 = 2048 - eaten_pos
|
|
exp_f2 = delete_block(pattern(2688, 2048), 0, eaten_pos)
|
|
exp_tail = pattern(4736, 640)
|
|
got_f2, got_tail = f[:want_f2], f[want_f2:]
|
|
if len(got_f2) == want_f2 and got_f2 == exp_f2 and len(got_tail) == 640 \
|
|
and got_tail == exp_tail:
|
|
print("VERDICT PASS | probe n=%s (eaten=%d) | canary 2048 + tail 640" % (eaten, eaten_pos))
|
|
elif len(got_f2) < want_f2 or got_f2 != exp_f2[:len(got_f2)]:
|
|
print("VERDICT STALL@F2 | probe n=%s (eaten=%d) | canary %d/%d" % (eaten, eaten_pos, len(got_f2), want_f2))
|
|
else:
|
|
print("VERDICT STALL@TAIL | probe n=%s | tail %d/640" % (eaten, len(got_tail)))
|
|
|
|
elif mode == "C":
|
|
b1 = streams.get("HEX", b"")
|
|
k = streams.get("HEX2", b"")
|
|
want_b = 1460 - eaten_pos
|
|
exp_b = delete_block(pattern(640, 1460), 1024, eaten_pos)
|
|
if len(b1) < want_b or b1[:want_b] != exp_b:
|
|
print("VERDICT STALL@B1460 | probe n=%s (eaten=%d) | fill part %d/%d" % (eaten, eaten_pos, len(b1), want_b))
|
|
return
|
|
exp_k, exp_tail = pattern(2100, 1460), pattern(3560, 640)
|
|
got_k, got_tail = k[:1460], k[1460:]
|
|
if len(got_k) < 1460 or got_k != exp_k:
|
|
print("VERDICT STALL@K1460 | probe n=%s | canary %d/1460" % (eaten, len(got_k)))
|
|
elif len(got_tail) < 640 or got_tail != exp_tail:
|
|
print("VERDICT STALL@TAIL | probe n=%s | tail %d/640" % (eaten, len(got_tail)))
|
|
else:
|
|
print("VERDICT PASS | probe n=%s (eaten=%d) | fill 1460 + canary 1460 + tail 640" % (eaten, eaten_pos))
|
|
else:
|
|
print("unknown mode: %s" % mode)
|
|
sys.exit(3)
|
|
|
|
main()
|