1 files changed, 359 insertions, 0 deletions
diff --git a/stacks/tncs/tnc_serial_recovery.py b/stacks/tncs/tnc_serial_recovery.py
new file mode 100644
index 0000000..f7b1370
--- /dev/null
+++ b/stacks/tncs/tnc_serial_recovery.py
@@ -0,0 +1,359 @@
+"""
+TheFirmware TF 2.7 — terminal recovery without power cycle.
+
+Native command set (NORD><LINK source / ALAS — see stacks/tncs/docs/TNC-RECOVERY.md).
+TAPR strings (kiss off, INFO, RESTART) are not primary; kept as last-resort legacy.
+
+Recovery ladder (stop on first banner):
+  0. DTR+RTS high + settle (caller)
+  1. Passive listen
+  2. KISS return 0xC0 0xFF 0xC0  (tfkiss.mac cmd 0xFF → firmware reset + banner)
+  3. Buffer flush ^Q^X + JHOST 0
+  4. ESC V  (version / terminal probe)
+  5. ESC QRES  (cold boot from EPROM — no mains power cycle)
+  6. ESC E 0 + ESC V
+  7. Legacy TAPR kiss off + INFO (optional last resort)
+"""
+
+from __future__ import annotations
+
+import time
+from typing import Callable
+
+FIRMWARE_MARKERS = (
+    b"TheFirmware",
+    b"NORD",
+    b"Version 2.7",
+    b"Checksum",
+    b"Copyright",
+    b"DAMA",
+    b"SMACK",
+    b"cmd:",
+    b"CMD:",
+)
+
+TF_ESC_V = b"\x1bV\r"
+TF_ESC_QRES = b"\x1bQRES\r"
+TF_ESC_AT_K = b"\x1b@K"
+TF_ESC_E0 = b"\x1bE0\r"
+
+LogFn = Callable[[str], None]
+
+
+def matched_markers(data: bytes) -> list[str]:
+    return [m.decode("ascii", errors="replace") for m in FIRMWARE_MARKERS if m.lower() in data.lower()]
+
+
+def escape_bytes(data: bytes, max_len: int = 96) -> str:
+    out: list[str] = []
+    for byte in data[:max_len]:
+        if 32 <= byte < 127:
+            out.append(chr(byte))
+        elif byte == 0x0D:
+            out.append("\\r")
+        elif byte == 0x0A:
+            out.append("\\n")
+        else:
+            out.append(f"\\x{byte:02x}")
+    if len(data) > max_len:
+        out.append("…")
+    return "".join(out)
+
+
+def format_rx_brief(data: bytes) -> str:
+    if not data:
+        return "0 B silent"
+    markers = matched_markers(data)
+    marker_txt = ",".join(markers) if markers else "none"
+    return f"{len(data)} B, markers={marker_txt}, text={escape_bytes(data)!r}"
+
+
+def format_rx_diag(data: bytes) -> str:
+    """One-line firmware/RX summary for operator logs."""
+    if not data:
+        return "silent (0 bytes) — check power, baud, CTS/DTR wiring"
+    markers = matched_markers(data)
+    hex_snip = data[:48].hex(" ")
+    if len(data) > 48:
+        hex_snip += " …"
+    parts = [
+        f"{len(data)} bytes",
+        f"markers={markers or 'none'}",
+        f"text={escape_bytes(data, 120)!r}",
+        f"hex={hex_snip}",
+    ]
+    return "; ".join(parts)
+
+
+def classify_recovery_failure(received: bytes) -> str:
+    if not received:
+        return (
+            "silent — TNC sent nothing (power, baud/line, DE-9 wiring, or boot without DTR)"
+        )
+    if has_banner(received):
+        return "banner seen in RX but terminal probe failed — check host mode or double-echo"
+    lower = received.lower()
+    if b"\x1bv" in lower or b"* " in received:
+        if not has_banner(received):
+            return (
+                "TF terminal alive (ESC V prompt) but no firmware banner — try ESC QRES with DTR high"
+            )
+    echo_hits = sum(
+        1
+        for token in (
+            b"kiss off",
+            b"info\r",
+            b"info\n",
+            b"restart\r",
+            b"jhost 0",
+        )
+        if token in lower
+    )
+    if echo_hits >= 2 or (
+        b"kiss off" in lower and (b"info" in lower or b"info\r" in lower)
+    ):
+        return (
+            "echo-only — TAPR-style or transparent echo, no TheFirmware banner "
+            "(Landolt: DTR high at power-on; try boot-wait or ESC QRES)"
+        )
+    if any(b < 0x20 and b not in (0x0D, 0x0A, 0x09) for b in received[:64]):
+        return (
+            f"binary/non-text ({len(received)} B) — likely KISS/stream mode, not terminal"
+        )
+    return f"non-banner text ({len(received)} B) — firmware state unclear, capture full RX"
+
+
+def has_banner(data: bytes) -> bool:
+    lower = data.lower()
+    return any(m.lower() in lower for m in FIRMWARE_MARKERS)
+
+
+def is_echo_only(cmd: bytes, reply: bytes) -> bool:
+    c = cmd.strip(b"\r\n")
+    r = reply.strip()
+    return r in (c, c + b"\r", c + b"\n", c + b"\r\n")
+
+
+def probe_tf_terminal(
+    write_flush: Callable[[bytes], None],
+    read_for: Callable[[float], bytes],
+    pause: float = 0.4,
+) -> tuple[bool, bytes, bool]:
+    """ESC V — native TheFirmware version / terminal probe."""
+    write_flush(TF_ESC_V)
+    time.sleep(pause)
+    reply = read_for(4.0)
+    only_echo = is_echo_only(TF_ESC_V, reply)
+    return has_banner(reply), reply, only_echo
+
+
+def probe_info(
+    write_flush: Callable[[bytes], None],
+    read_for: Callable[[float], bytes],
+    pause: float = 0.4,
+) -> tuple[bool, bytes, bool]:
+    """Alias: native TF probe (was TAPR kiss off + INFO)."""
+    return probe_tf_terminal(write_flush, read_for, pause=pause)
+
+
+def probe_combined(
+    write_flush: Callable[[bytes], None],
+    read_for: Callable[[float], bytes],
+) -> tuple[bool, bytes, bool]:
+    """Legacy TAPR kiss off + INFO — last resort only."""
+    write_flush(b"kiss off\rINFO\r")
+    time.sleep(0.8)
+    reply = read_for(5.0)
+    only_echo = is_echo_only(b"kiss off\rINFO\r", reply) or (
+        is_echo_only(b"kiss off\r", reply[:20]) and is_echo_only(b"INFO\r", reply[-20:])
+    )
+    return has_banner(reply), reply, only_echo
+
+
+def send_jhost0(write_flush: Callable[[bytes], None]) -> None:
+    """Leave WA8DED host mode → terminal (gettoweb / WA8DED guide)."""
+    send_buffer_flush(write_flush)
+    write_flush(b"\x00" * 300)
+    time.sleep(0.15)
+    write_flush(b"\x00\x01\x06JHOST 0\r")
+    time.sleep(0.8)
+
+
+def send_kiss_return(write_flush: Callable[[bytes], None]) -> None:
+    write_flush(b"\xc0\xff\xc0")
+    time.sleep(1.0)
+
+
+def send_esc_at_k(write_flush: Callable[[bytes], None]) -> None:
+    """Enter KISS from terminal (tfb.c @K) — not an exit path."""
+    write_flush(TF_ESC_AT_K)
+    time.sleep(0.8)
+
+
+def send_echo_off(write_flush: Callable[[bytes], None]) -> None:
+    """TheFirmware ESC E 0 — disable terminal echo."""
+    write_flush(TF_ESC_E0)
+    time.sleep(0.3)
+
+
+def send_qres(write_flush: Callable[[bytes], None]) -> None:
+    """Software cold boot from EPROM (tfd.c Qcmd) — DTR must stay high."""
+    write_flush(TF_ESC_QRES)
+    time.sleep(2.5)
+
+
+def send_buffer_flush(write_flush: Callable[[bytes], None]) -> None:
+    """WA8DED host guide: ^Q^X before commands if junk in buffer."""
+    write_flush(b"\x11\x18")
+    time.sleep(0.2)
+
+
+def tf_mycall_frame(call: str) -> bytes:
+    """ESC I — set own callsign (tfb.c Icmd)."""
+    return f"\x1bI {call.upper()}\r".encode("ascii", errors="replace")
+
+
+def _try_probe(
+    write_flush: Callable[[bytes], None],
+    read_for: Callable[[float], bytes],
+    out: LogFn,
+    label: str,
+) -> tuple[bool, bytes]:
+    ok, probe, only_echo = probe_tf_terminal(write_flush, read_for)
+    if ok:
+        out(f"recovery: OK after {label}")
+        return True, probe
+    out(
+        f"recovery: probe after {label} — {format_rx_brief(probe)}, "
+        f"echo_only={only_echo}, banner={has_banner(probe)}"
+    )
+    if not only_echo and probe:
+        out(f"recovery: partial reply after {label}")
+    return False, probe
+
+
+def recover_terminal(
+    write_flush: Callable[[bytes], None],
+    read_for: Callable[[float], bytes],
+    *,
+    skip_kiss_frame: bool = False,
+    log: LogFn | None = None,
+) -> tuple[bool, bytes]:
+    """Run full software recovery ladder. Returns (host_ok, all_rx_bytes)."""
+    out = log or (lambda _msg: None)
+    received = b""
+
+    chunk = read_for(1.5)
+    received += chunk
+    out(f"recovery: passive listen — {format_rx_brief(chunk)}")
+    if has_banner(chunk):
+        out("recovery: OK — banner on passive listen")
+        return True, received
+
+    if not skip_kiss_frame:
+        send_kiss_return(write_flush)
+        chunk = read_for(2.5)
+        received += chunk
+        out(f"recovery: KISS return frame — {format_rx_brief(chunk)}")
+        if has_banner(chunk):
+            out("recovery: OK after KISS return (firmware reset)")
+            return True, received
+
+    send_jhost0(write_flush)
+    chunk = read_for(1.0)
+    received += chunk
+    out(f"recovery: JHOST 0 — {format_rx_brief(chunk)}")
+    if has_banner(chunk):
+        out("recovery: OK after JHOST 0")
+        return True, received
+
+    ok, probe = _try_probe(write_flush, read_for, out, "ESC V")
+    received += probe
+    if ok:
+        return True, received
+
+    send_qres(write_flush)
+    chunk = read_for(3.0)
+    received += chunk
+    out(f"recovery: ESC QRES — {format_rx_brief(chunk)}")
+    if has_banner(chunk):
+        out("recovery: OK after ESC QRES")
+        return True, received
+
+    ok, probe = _try_probe(write_flush, read_for, out, "post-QRES ESC V")
+    received += probe
+    if ok:
+        return True, received
+
+    send_echo_off(write_flush)
+    received += read_for(0.3)
+    ok, probe = _try_probe(write_flush, read_for, out, "ESC E 0")
+    received += probe
+    if ok:
+        return True, received
+
+    send_qres(write_flush)
+    received += read_for(3.0)
+    out("recovery: second ESC QRES")
+    ok, probe = _try_probe(write_flush, read_for, out, "second QRES")
+    received += probe
+    if ok:
+        return True, received
+
+    ok, probe, _ = probe_combined(write_flush, read_for)
+    received += probe
+    if ok:
+        out("recovery: OK after legacy TAPR kiss off + INFO")
+        return True, received
+    if probe:
+        out(f"recovery: legacy TAPR probe — {format_rx_brief(probe)}")
+
+    out(
+        "recovery: FAILED — echo only or silent "
+        "(retry with DTR high; cold-boot without DTR may need boot-wait + power cycle)"
+    )
+    out(f"recovery: firmware assessment — {classify_recovery_failure(received)}")
+    out(f"recovery: RX capture — {format_rx_diag(received)}")
+    return False, received
+
+
+def enter_kiss(
+    write_flush: Callable[[bytes], None],
+    read_for: Callable[[float], bytes],
+    entry: str = "kiss_on",
+) -> None:
+    """Enter KISS after terminal recovery (native TF: ESC @K)."""
+    if entry == "tapr":
+        write_flush(b"kiss on\r")
+        time.sleep(0.5)
+        read_for(0.5)
+        return
+    if entry == "auto":
+        write_flush(TF_ESC_AT_K)
+        time.sleep(0.5)
+        reply = read_for(0.5)
+        if reply.strip() in (b"kiss on", b"kiss on\r", b"kiss on\n"):
+            write_flush(TF_ESC_AT_K)
+            time.sleep(0.5)
+            read_for(0.5)
+        return
+    write_flush(TF_ESC_AT_K)
+    time.sleep(0.5)
+    read_for(0.3)
+
+
+def recover_and_enter_kiss(
+    write_flush: Callable[[bytes], None],
+    read_for: Callable[[float], bytes],
+    *,
+    kiss_entry: str = "kiss_on",
+    skip_kiss_frame: bool = False,
+    log: LogFn | None = None,
+) -> bool:
+    ok, _ = recover_terminal(
+        write_flush, read_for, skip_kiss_frame=skip_kiss_frame, log=log
+    )
+    if not ok:
+        return False
+    enter_kiss(write_flush, read_for, kiss_entry)
+    return True