185 lines
8.3 KiB
Python
185 lines
8.3 KiB
Python
# Reproducible regression test for the "TEL:" telemetry parser in app/hub.py.
|
|
#
|
|
# History: the firmware 1.1.1 build (F6) added two diagnostic fields between
|
|
# LPF and X in the telemetry line:
|
|
# ... LPF=<lpf>, prox=<0..1>, CmaxN=<counts>, X=<x>, Y=<y>
|
|
# The original RE_TEL expected X immediately after LPF, so the new line stopped
|
|
# parsing (the "TEL ya no se parsea" regression). This test locks the contract:
|
|
#
|
|
# * parses the NEW line (with prox/CmaxN) -- incl. the real 2026-10-30 line
|
|
# * still parses the OLD line (without prox/CmaxN) -> backward compatible
|
|
# * handles signed (negative) X/Y coordinates
|
|
# * captures prox/cmaxn into on_tel() points (None on old-format lines)
|
|
# * negates X/Y in on_tel() so telemetry overlays the sample trajectory
|
|
# * does NOT cross-match (sample / FineTune / SunPos lines -> no TEL)
|
|
# * feeds the tel_samples buffer and the TEL_HISTORY replay on connect
|
|
# * dispatches handle_esp32_message -> "TEL:" (new fw) / "DATA:" (old fw)
|
|
#
|
|
# It mirrors _smoke_test.py: fastapi is stubbed before the import so this runs
|
|
# on a machine where fastapi is NOT installed.
|
|
import sys, types, json, asyncio, tempfile, shutil
|
|
|
|
def _install_stub():
|
|
try:
|
|
import fastapi # noqa: F401 (real one present -> nothing to do)
|
|
return
|
|
except Exception:
|
|
pass
|
|
fa = types.ModuleType("fastapi")
|
|
class WebSocket: pass
|
|
fa.WebSocket = WebSocket
|
|
sys.modules["fastapi"] = fa
|
|
|
|
_install_stub()
|
|
import app.hub as hubmod # noqa: E402 (imported after stubbing)
|
|
from app.hub import HubManager, RE_TEL, RE_SAMPLE # noqa: E402
|
|
|
|
# Redirect log writes to a temp dir so the test never touches data_logs/.
|
|
_tmp = tempfile.mkdtemp(prefix="teltest_")
|
|
hubmod.LOGS_DIR = _tmp
|
|
hub = HubManager()
|
|
|
|
class FakeClient:
|
|
def __init__(self):
|
|
self.sent = []
|
|
async def accept(self): pass
|
|
async def send_text(self, msg):
|
|
self.sent.append(msg)
|
|
|
|
# --- Fixtures -----------------------------------------------------------------
|
|
# Real line captured from the field device (2026-10-30 10:34:03), new firmware.
|
|
TEL_REAL = ("LOG:[2026-10-30 10:34:03] TEL: t=246s, steps=29825(+1000), "
|
|
"avgC=5469365, avgV=5160269, avgP=679686, LPF=4338597, "
|
|
"prox=0.00, CmaxN=53643680, X=-9827, Y=-65788")
|
|
# Same layout, signed coordinates and a non-trivial proximity value.
|
|
TEL_NEG = ("LOG:[2026-10-02 09:34:05] TEL: t=125s, steps=5100(-20), "
|
|
"avgC=-100, avgV=18000, avgP=1400, LPF=-50, "
|
|
"prox=0.85, CmaxN=123456, X=-54810, Y=-48890")
|
|
# OLD firmware (pre-F6): no prox/CmaxN. Must keep parsing.
|
|
TEL_OLD = ("LOG:[2026-10-02 09:34:03] TEL: t=123s, steps=5000(+100), "
|
|
"avgC=9900, avgV=18380, avgP=1490, LPF=9900, X=54810, Y=48890")
|
|
# Legacy periodic sample (RE_SAMPLE) and a plain event line.
|
|
OLD_SMP = "LOG:[2026-10-02 09:34:07] X:54810,Y:48890,C:9898,V:18263,P:1450,S:17,er:1500"
|
|
FINE = "LOG:[2026-10-02 09:34:09] FineTune: inicio"
|
|
|
|
def test_parsers():
|
|
# NEW format (real line) -- every field present.
|
|
m = RE_TEL.match(TEL_REAL)
|
|
assert m, "RE_TEL failed to match the real new-format TEL line"
|
|
g = m.groupdict()
|
|
assert g["ts"] == "2026-10-30 10:34:03"
|
|
assert g["t"] == "246" and g["steps"] == "29825" and g["stepsDelta"] == "+1000"
|
|
assert g["avgC"] == "5469365" and g["avgV"] == "5160269" and g["avgP"] == "679686"
|
|
assert g["lpf"] == "4338597"
|
|
assert g["prox"] == "0.00" and g["cmaxN"] == "53643680"
|
|
assert g["x"] == "-9827" and g["y"] == "-65788"
|
|
|
|
# Signed coords + non-trivial prox.
|
|
mn = RE_TEL.match(TEL_NEG)
|
|
assert mn, "RE_TEL failed to match a TEL line with signed X/Y"
|
|
assert mn.group("x") == "-54810" and mn.group("y") == "-48890"
|
|
assert mn.group("avgC") == "-100" and mn.group("stepsDelta") == "-20"
|
|
assert mn.group("prox") == "0.85" and mn.group("cmaxN") == "123456"
|
|
|
|
# OLD format (backward compatibility): prox/cmaxN optional -> None.
|
|
mo = RE_TEL.match(TEL_OLD)
|
|
assert mo, "RE_TEL failed to match the old (pre-F6) TEL line"
|
|
assert mo.group("x") == "54810" and mo.group("y") == "48890"
|
|
assert mo.group("prox") is None and mo.group("cmaxN") is None
|
|
|
|
# Bare line without the leading LOG: prefix must also match.
|
|
assert RE_TEL.match(TEL_OLD[4:]), "RE_TEL must match a bare (no LOG:) line"
|
|
|
|
# No cross-matching with sample / event lines.
|
|
assert RE_TEL.match(OLD_SMP) is None, "legacy sample line must NOT match RE_TEL"
|
|
assert RE_TEL.match(FINE) is None, "event line must NOT match RE_TEL"
|
|
assert RE_SAMPLE.match(OLD_SMP), "legacy sample line must still match RE_SAMPLE"
|
|
|
|
def test_on_tel():
|
|
hub.tel_samples.clear()
|
|
# New format: negation of X/Y + prox/cmaxn captured.
|
|
pt = hub.on_tel(RE_TEL.match(TEL_REAL).groupdict())
|
|
assert pt["x"] == 9827 and pt["y"] == 65788, "on_tel must negate X/Y"
|
|
assert pt["t"] == 246 and pt["steps"] == 29825 and pt["steps_delta"] == 1000
|
|
assert pt["lpf"] == 4338597
|
|
assert pt["prox"] == 0.0 and pt["cmaxn"] == 53643680
|
|
for k in ("gpoa", "voc", "isc_ma", "pcpv_mw", "az", "el"):
|
|
assert k in pt, "on_tel point missing field %s" % k
|
|
|
|
# Old format: prox/cmaxn absent -> None, no crash.
|
|
pt_old = hub.on_tel(RE_TEL.match(TEL_OLD).groupdict())
|
|
assert pt_old["x"] == -54810 and pt_old["y"] == -48890
|
|
assert pt_old["prox"] is None and pt_old["cmaxn"] is None
|
|
assert len(hub.tel_samples) == 2
|
|
|
|
def test_buffer_and_history():
|
|
hub.tel_samples.clear()
|
|
for line in (TEL_REAL, TEL_NEG, TEL_OLD):
|
|
hub.on_tel(RE_TEL.match(line).groupdict())
|
|
assert len(hub.tel_samples) == 3, "buffer must hold all three TEL points"
|
|
|
|
# TEL_HISTORY replay must deliver the buffered points to a new client.
|
|
async def _run():
|
|
client = FakeClient()
|
|
await hub.connect_client(client)
|
|
hist = [x for x in client.sent if x.startswith("TEL_HISTORY:")]
|
|
assert len(hist) == 1, "connect_client must send exactly one TEL_HISTORY"
|
|
payload = json.loads(hist[0][len("TEL_HISTORY:"):])
|
|
assert isinstance(payload, list) and len(payload) == 3
|
|
# The real line's negated coords must be in the replayed history.
|
|
assert any(p.get("x") == 9827 and p.get("y") == 65788 for p in payload), \
|
|
"TEL_HISTORY must include the negated real-line coords"
|
|
# Old-format point carried prox=None through the round trip.
|
|
assert any(p.get("prox") is None for p in payload), \
|
|
"old-format point must keep prox=None in history"
|
|
hub.disconnect_client(client)
|
|
asyncio.run(_run())
|
|
|
|
def test_dispatch():
|
|
async def _run():
|
|
client = FakeClient()
|
|
hub.client_sockets.append(client)
|
|
await hub.handle_esp32_message(TEL_REAL)
|
|
tel_msgs = [x for x in client.sent if x.startswith("TEL:")]
|
|
assert len(tel_msgs) == 1, "expected one TEL: broadcast, got %d" % len(tel_msgs)
|
|
assert json.loads(tel_msgs[0][4:])["x"] == 9827
|
|
assert json.loads(tel_msgs[0][4:])["cmaxn"] == 53643680
|
|
client.sent.clear()
|
|
await hub.handle_esp32_message(OLD_SMP)
|
|
assert any(x.startswith("DATA:") for x in client.sent), "old line must emit DATA:"
|
|
assert not any(x.startswith("TEL:") for x in client.sent), "old line must NOT emit TEL:"
|
|
client.sent.clear()
|
|
await hub.handle_esp32_message(FINE)
|
|
assert not any(x.startswith(("DATA:", "TEL:")) for x in client.sent), "event line must only log"
|
|
hub.disconnect_client(client)
|
|
asyncio.run(_run())
|
|
|
|
def test_position_mapping():
|
|
previous = hub.azdev
|
|
try:
|
|
hub.azdev = 18.0
|
|
# Firmware mapping: X is azimuth; Y is elevation plus X coupling.
|
|
x, y = 39750, 87450
|
|
az, el = hub._pos_deg(x, y)
|
|
assert abs(az - 160.0) < 1e-9
|
|
assert abs(el - 50.0) < 1e-9
|
|
sample = RE_SAMPLE.match(OLD_SMP).groupdict()
|
|
sample.update(x=str(x), y=str(y))
|
|
tel = RE_TEL.match(TEL_REAL).groupdict()
|
|
tel.update(x=str(-x), y=str(-y))
|
|
for point in (hub.on_sample(sample), hub.on_tel(tel)):
|
|
assert point['az'] == 160.0 and point['el'] == 50.0
|
|
finally:
|
|
hub.azdev = previous
|
|
|
|
def _run():
|
|
test_parsers()
|
|
test_on_tel()
|
|
test_buffer_and_history()
|
|
test_dispatch()
|
|
test_position_mapping()
|
|
print("OK: RE_TEL (new+old format), on_tel conversion, buffer, TEL_HISTORY and dispatch are all consistent.")
|
|
|
|
_run()
|
|
shutil.rmtree(_tmp, ignore_errors=True)
|