# 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=, prox=<0..1>, CmaxN=, X=, 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)