185 lines
8.6 KiB
Python
Executable File
185 lines
8.6 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""phtst_report.py — lee el historial crudo de una unidad y dice hasta qué largo de línea pasa el módem (tecla X, fw 0.6.2+).
|
|
|
|
Uso:
|
|
python3 phtst_report.py H0027.txt # el .txt de «⬇ Historial crudo» del monitor (o /api/unit_history/H0027)
|
|
python3 phtst_report.py captura1.txt captura2.txt
|
|
cat H0027.txt | python3 phtst_report.py -
|
|
|
|
Qué manda la unidad con la tecla X (ver hbox_v5_bringup.ino, phtstService):
|
|
$PHTST,<largo>,<pasada>,<n>,<dígitos>*CS con EXACTAMENTE <largo> caracteres desde '$' hasta el checksum (sin CR LF)
|
|
pasadas 1 y 2: control de 60, 100..109, control, 110..119, ..., 140, control (control = línea de 60)
|
|
luego 6 pares PEGADOS (pasada 'P', las dos líneas en una sola escritura, sin hueco): 60+60, 100+60, 100+100, dos veces
|
|
$PHEV ... PHTST,ini ... / PHTST,fin n=105 enmarcan la corrida.
|
|
Una línea cuenta como «llegó» solo si está ENTERA: checksum bien y largo real = largo declarado (un módem que recorta en
|
|
vez de descartar dejaría líneas cortas con checksum malo; esas salen como «dañadas»).
|
|
El servidor vuelca el historial cada ~10 s y, pasada la primera hora, lo reescribe en anillo (no queda en orden): el
|
|
script ordena por la hora de cada línea y solo da veredicto si llegó el cierre ($PHEV ... PHTST,fin). Bajar el historial
|
|
~15 s después de «PRUEBA DE LARGO fin» en la consola.
|
|
"""
|
|
import re
|
|
import sys
|
|
|
|
MIN, MAX, CTRL, PASSES = 100, 140, 60, 2
|
|
PAIRS = [(60, 60), (100, 60), (100, 100)] * 2
|
|
|
|
|
|
def expected():
|
|
"""[(seq, largo, pasada)] tal como las manda el firmware: 105 líneas."""
|
|
out, seq = [], 0
|
|
for p in range(1, PASSES + 1):
|
|
if p == 1:
|
|
seq += 1; out.append((seq, CTRL, str(p)))
|
|
for L in range(MIN, MAX + 1):
|
|
seq += 1; out.append((seq, L, str(p)))
|
|
if (L + 1 - MIN) % 10 == 0 or L == MAX:
|
|
seq += 1; out.append((seq, CTRL, str(p)))
|
|
for a, b in PAIRS:
|
|
seq += 1; out.append((seq, a, 'P'))
|
|
seq += 1; out.append((seq, b, 'P'))
|
|
return out
|
|
|
|
|
|
def checksum_ok(s):
|
|
if not s.startswith('$') or '*' not in s:
|
|
return False
|
|
body, _, cs = s[1:].partition('*')
|
|
try:
|
|
want = int(cs[:2], 16)
|
|
except ValueError:
|
|
return False
|
|
got = 0
|
|
for ch in body:
|
|
got ^= ord(ch)
|
|
return got == want and len(cs) == 2
|
|
|
|
|
|
TS = re.compile(r'^(\d{4}-\d{2}-\d{2}[ T]\d{2}:\d{2}:\d{2})')
|
|
|
|
|
|
def read_runs(texts):
|
|
"""Ordena por hora (historial en anillo) y separa corridas por PHTST,ini. Devuelve
|
|
[(estado, {seq: (ok, largo_real, largo_declarado, pasada)})], estado = 'fin' | 'cancelada' | 'abierta'."""
|
|
rows = []
|
|
for text in texts:
|
|
for raw in text.splitlines():
|
|
m = TS.match(raw)
|
|
rows.append((m.group(1) if m else '', len(rows), raw))
|
|
if any(r[0] for r in rows):
|
|
rows.sort(key=lambda r: (r[0], r[1])) # estable: dentro del mismo segundo, el orden del archivo
|
|
runs, cur, state = [], None, 'abierta'
|
|
for _, _, raw in rows:
|
|
if 'PHTST,ini' in raw:
|
|
if cur is not None:
|
|
runs.append((state, cur))
|
|
cur, state = {}, 'abierta'
|
|
continue
|
|
if cur is not None and ('PHTST,fin' in raw or 'PHTST,cancelada' in raw):
|
|
state = 'fin' if 'PHTST,fin' in raw else 'cancelada'
|
|
continue
|
|
i = raw.find('$PHTST,')
|
|
if i < 0:
|
|
continue
|
|
if cur is None: # sin marca de inicio (historial cortado): corrida abierta
|
|
cur = {}
|
|
s = raw[i:].strip()
|
|
f = s.split(',')
|
|
try:
|
|
declared, pas, seq = int(f[1]), f[2], int(f[3])
|
|
except (IndexError, ValueError):
|
|
continue
|
|
ok = checksum_ok(s) and len(s) == declared
|
|
if seq not in cur or ok:
|
|
cur[seq] = (ok, len(s), declared, pas)
|
|
if cur is not None:
|
|
runs.append((state, cur))
|
|
return runs
|
|
|
|
|
|
def report(got, state='fin'):
|
|
exp = expected()
|
|
lines = []
|
|
say = lines.append
|
|
intact = {seq for seq, v in got.items() if v[0]}
|
|
damaged = sorted(seq for seq, v in got.items() if not v[0])
|
|
say(f"Líneas $PHTST esperadas {len(exp)}; llegaron enteras {len(intact)}; dañadas {len(damaged)}.")
|
|
if damaged:
|
|
decl = sorted({got[q][2] for q in damaged}); real = sorted({got[q][1] for q in damaged})
|
|
say(f" dañadas (checksum malo o largo real distinto del declarado): largos declarados {decl[0]}..{decl[-1]}, "
|
|
f"llegaron con {real[0]}..{real[-1]} caracteres => el módem RECORTA las largas en vez de descartarlas")
|
|
if state != 'fin':
|
|
say("CORRIDA INCOMPLETA: " + ("se canceló (X otra vez o Modo B off)." if state == 'cancelada' else
|
|
"no llegó el cierre PHTST,fin. Baja el historial otra vez ~15 s después de «PRUEBA DE LARGO fin».") +
|
|
" Sin veredicto; lo que llegó hasta ahora:")
|
|
ctrl = [e for e in exp if e[1] == CTRL and e[2] != 'P']
|
|
ctrl_ok = sum(1 for e in ctrl if e[0] in intact)
|
|
say(f"Líneas de control (60): {ctrl_ok}/{len(ctrl)}.")
|
|
singles = [e for e in exp if e[2] != 'P' and e[1] != CTRL]
|
|
arrived = {L: [] for L in range(MIN, MAX + 1)}
|
|
for seq, L, p in singles:
|
|
arrived[L].append(seq in intact)
|
|
row = ''.join('#' if all(arrived[L]) else ('+' if any(arrived[L]) else '.') for L in range(MIN, MAX + 1))
|
|
say(f"Largo 100..140 (# = llegó en las 2 pasadas, + = en una, . = en ninguna):")
|
|
say(f" 100 {row[:10]} {row[10:20]} {row[20:30]} {row[30:40]} {row[40:]} 140")
|
|
all_ok = [L for L in range(MIN, MAX + 1) if all(arrived[L])]
|
|
none_ok = [L for L in range(MIN, MAX + 1) if not any(arrived[L])]
|
|
verdict_line = None
|
|
if not any(any(v) for v in arrived.values()):
|
|
verdict_line = "No llegó ninguna línea de 100 o más: límite por línea < 100 caracteres (o la corrida no llegó)."
|
|
else:
|
|
lim = max(L for L in range(MIN, MAX + 1) if any(arrived[L]))
|
|
clean = all(all(arrived[L]) for L in range(MIN, lim + 1)) and all(not any(arrived[L]) for L in range(lim + 1, MAX + 1))
|
|
if clean and lim == MAX:
|
|
verdict_line = f"Llegaron TODAS hasta {MAX}: el límite por línea es > {MAX} caracteres (o no hay límite en este rango)."
|
|
elif clean:
|
|
verdict_line = f"LÍMITE POR LÍNEA: {lim} caracteres (sin CR LF) — de {lim + 1} para arriba no llegó ninguna."
|
|
else:
|
|
holes = [L for L in range(MIN, lim + 1) if not all(arrived[L])]
|
|
verdict_line = (f"Resultado no limpio: la más larga que llegó es {lim}, pero faltan {holes} por debajo "
|
|
f"(pérdidas sueltas: cobertura / UDP). Repetir X; el límite probable es {lim}.")
|
|
if state == 'fin':
|
|
say(verdict_line)
|
|
if ctrl_ok < len(ctrl):
|
|
say(" OJO: se perdieron líneas de control (cortas): hubo pérdidas que no son por largo. Repetir la prueba.")
|
|
pair_res = []
|
|
pair_seqs = [e for e in exp if e[2] == 'P']
|
|
for k in range(len(PAIRS)):
|
|
(s1, a, _), (s2, b, _) = pair_seqs[2 * k], pair_seqs[2 * k + 1]
|
|
pair_res.append(((a, b), s1 in intact, s2 in intact))
|
|
summary = {}
|
|
for (a, b), o1, o2 in pair_res:
|
|
t = summary.setdefault((a, b), [0, 0])
|
|
t[0] += int(o1 and o2); t[1] += 1
|
|
say("Pares pegados (las dos líneas en una sola ráfaga, sin hueco): " +
|
|
', '.join(f"{a}+{b} {ok}/{n}" for (a, b), (ok, n) in summary.items()))
|
|
if state != 'fin':
|
|
return lines
|
|
single100 = all(arrived[100])
|
|
if single100 and summary[(100, 100)][0] == summary[(100, 100)][1] and summary[(100, 60)][0] == summary[(100, 60)][1]:
|
|
say(" => el módem corta por LÍNEA, no por ráfaga: líneas pegadas pasan si cada una es corta.")
|
|
elif single100 and summary[(100, 100)][0] < summary[(100, 100)][1]:
|
|
say(" => los pares de 100 no pasan aunque la línea de 100 sola sí: el módem corta por RÁFAGA/paquete. Cada línea "
|
|
"del MCU tiene que salir sola (0.6.2 ya lo hace: la siguiente espera a que la anterior salga + 8 ms).")
|
|
return lines
|
|
|
|
|
|
def main(argv):
|
|
if len(argv) < 2:
|
|
print(__doc__); return 2
|
|
texts = []
|
|
for a in argv[1:]:
|
|
texts.append(sys.stdin.read() if a == '-' else open(a, encoding='utf-8', errors='replace').read())
|
|
runs = read_runs(texts)
|
|
if not runs:
|
|
print("No hay líneas $PHTST en la entrada: ¿se corrió X con Modo B activo? ¿es el historial de esa unidad y de esa hora?")
|
|
return 1
|
|
for i, (state, got) in enumerate(runs, 1):
|
|
if len(runs) > 1:
|
|
print(f"== corrida {i} de {len(runs)}")
|
|
print('\n'.join(report(got, state)))
|
|
return 0
|
|
|
|
|
|
if __name__ == '__main__':
|
|
sys.exit(main(sys.argv))
|