summaryrefslogtreecommitdiff
path: root/stacks/tncs
diff options
context:
space:
mode:
authorinfo@mode42.com <info@mode42.com>2026-08-07 18:25:13 +0000
committerinfo@mode42.com <info@mode42.com>2026-08-07 18:25:13 +0000
commit04d965d67a7264a1c7c211494aebda1953df7603 (patch)
tree0ebd700a6e219f84a26a656f4bee8778bc75da7b /stacks/tncs
Initial push
Diffstat (limited to 'stacks/tncs')
-rw-r--r--stacks/tncs/CMakeLists.txt7
-rw-r--r--stacks/tncs/README.md40
-rw-r--r--stacks/tncs/RS232A259.jpgbin0 -> 375411 bytes
-rw-r--r--stacks/tncs/TNC2C.pdfbin0 -> 1519793 bytes
-rw-r--r--stacks/tncs/docs/HYBBX-TNC2C.md54
-rw-r--r--stacks/tncs/docs/TNC-RECOVERY.md108
-rw-r--r--stacks/tncs/docs/TNC2C-OPERATIONS.md66
-rw-r--r--stacks/tncs/docs/TNC2C-REFERENCE.md43
-rwxr-xr-xstacks/tncs/hybbx-ax25-beacon-once.py105
-rw-r--r--stacks/tncs/hybbx-dual.ini62
-rw-r--r--stacks/tncs/hybbx-tnc2c.ini38
-rwxr-xr-xstacks/tncs/load-serial-env.sh71
-rwxr-xr-xstacks/tncs/pktnc2-boot-wait.sh19
-rw-r--r--stacks/tncs/pktnc2-serial.env4
-rwxr-xr-xstacks/tncs/pktnc2-to-hybbx.sh24
-rw-r--r--stacks/tncs/test_tnc_serial_recovery.py128
-rw-r--r--stacks/tncs/tnc2c-2.jpgbin0 -> 395847 bytes
-rwxr-xr-xstacks/tncs/tnc2c-autotest.py925
-rwxr-xr-xstacks/tncs/tnc2c-autotest.sh11
-rw-r--r--stacks/tncs/tnc2c-boot-wait.py326
-rwxr-xr-xstacks/tncs/tnc2c-boot-wait.sh14
-rwxr-xr-xstacks/tncs/tnc2c-chat.sh52
-rwxr-xr-xstacks/tncs/tnc2c-check.sh62
-rwxr-xr-xstacks/tncs/tnc2c-expect.exp80
-rw-r--r--stacks/tncs/tnc2c-find.c306
-rw-r--r--stacks/tncs/tnc2c-health.py445
-rwxr-xr-xstacks/tncs/tnc2c-health.sh9
-rwxr-xr-xstacks/tncs/tnc2c-host-reset.py229
-rwxr-xr-xstacks/tncs/tnc2c-host-reset.sh21
-rwxr-xr-xstacks/tncs/tnc2c-hybbx-beacon-once.sh21
-rw-r--r--stacks/tncs/tnc2c-integration-test.py103
-rwxr-xr-xstacks/tncs/tnc2c-integration-test.sh4
-rwxr-xr-xstacks/tncs/tnc2c-listen.sh56
-rw-r--r--stacks/tncs/tnc2c-probe.c535
-rwxr-xr-xstacks/tncs/tnc2c-rf-tx-test.sh28
-rwxr-xr-xstacks/tncs/tnc2c-send-test.sh100
-rw-r--r--stacks/tncs/tnc2c-serial.env8
-rw-r--r--stacks/tncs/tnc2c-tx-rx-test.py184
-rw-r--r--stacks/tncs/tnc2c.jpgbin0 -> 739243 bytes
-rw-r--r--stacks/tncs/tnc_serial_recovery.py359
40 files changed, 4647 insertions, 0 deletions
diff --git a/stacks/tncs/CMakeLists.txt b/stacks/tncs/CMakeLists.txt
new file mode 100644
index 0000000..3bc43ce
--- /dev/null
+++ b/stacks/tncs/CMakeLists.txt
@@ -0,0 +1,7 @@
+add_executable(tnc2c-probe tnc2c-probe.c)
+target_compile_options(tnc2c-probe PRIVATE -std=c99 -Wall -Wextra -O2)
+
+add_executable(tnc2c-find tnc2c-find.c)
+target_compile_options(tnc2c-find PRIVATE -std=c99 -Wall -Wextra -O2)
+
+install(TARGETS tnc2c-probe tnc2c-find RUNTIME DESTINATION "${CMAKE_INSTALL_BINDIR}")
diff --git a/stacks/tncs/README.md b/stacks/tncs/README.md
new file mode 100644
index 0000000..e07a9ac
--- /dev/null
+++ b/stacks/tncs/README.md
@@ -0,0 +1,40 @@
+# stacks/tncs — Serial TNC tools
+
+Operator tools and docs for **Landolt TNC2C** and **PK-TNC2** (TheFirmware TNC-2 class). HyBBX attaches via `packet_radio` after MAX25 prep.
+
+**Recovery:** software-first (no power cycle in normal ops). Power-cycle boot-wait is rescue only. See [docs/TNC-RECOVERY.md](docs/TNC-RECOVERY.md).
+
+## Quick start
+
+```bash
+./scripts/build.sh
+./scripts/max25-ctl start --hardware tncs --device tnc2c
+```
+
+Adjust serial port in `tnc2c-serial.env` before first use.
+
+## Documentation
+
+| Doc | Content |
+|-----|---------|
+| [docs/TNC-RECOVERY.md](docs/TNC-RECOVERY.md) | Software ladder, rescue power-cycle |
+| [docs/TNC2C-OPERATIONS.md](docs/TNC2C-OPERATIONS.md) | Boot-wait, health, HyBBX handoff |
+| [docs/TNC2C-REFERENCE.md](docs/TNC2C-REFERENCE.md) | Parameters, tools, example profile |
+| [docs/HYBBX-TNC2C.md](docs/HYBBX-TNC2C.md) | HyBBX Secondary attach |
+
+HyBBX INI: [`hybbx-tnc2c.ini`](hybbx-tnc2c.ini) (onboard ttyS4 example) · shipped copy: `../../share/hybbx/tnc2c-host.ini.example` · dual TNC: [`hybbx-dual.ini`](hybbx-dual.ini)
+
+## Tools
+
+| Tool | Purpose |
+|------|---------|
+| `tnc2c-host-reset.sh` | Software recovery (no power cycle) |
+| `tnc2c-boot-wait.sh` | DTR during power-on — **rescue**; `--recover-only` for ladder only |
+| `pktnc2-boot-wait.sh` | PK-TNC2 wrapper (9600 8N1); supports `--recover-only` |
+| `tnc2c-integration-test.sh` | HyBBX-ready check |
+| `tnc2c-health.sh` | Serial/TNC check (no TX) |
+| `tnc2c-probe` | Scan serial ports for TNC response |
+
+## Safety
+
+KISS and TX tests can key the radio. Use a dummy load for transmit tests.
diff --git a/stacks/tncs/RS232A259.jpg b/stacks/tncs/RS232A259.jpg
new file mode 100644
index 0000000..c1dfe04
--- /dev/null
+++ b/stacks/tncs/RS232A259.jpg
Binary files differ
diff --git a/stacks/tncs/TNC2C.pdf b/stacks/tncs/TNC2C.pdf
new file mode 100644
index 0000000..d9f7b9d
--- /dev/null
+++ b/stacks/tncs/TNC2C.pdf
Binary files differ
diff --git a/stacks/tncs/docs/HYBBX-TNC2C.md b/stacks/tncs/docs/HYBBX-TNC2C.md
new file mode 100644
index 0000000..8d2207f
--- /dev/null
+++ b/stacks/tncs/docs/HYBBX-TNC2C.md
@@ -0,0 +1,54 @@
+# HyBBX + TNC2C
+
+MAX25 prepares the TNC (recovery + KISS entry); HyBBX attaches with **`kiss_entry = none`**.
+
+## Flow
+
+```
+max25d auto_start / max25-ctl start --hardware tncs --device tnc2c
+ ↓ (software recovery + kiss on)
+HyBBX Secondary starts packet_radio plugin (attach-only)
+ ↓
+KISS on serial → AX.25 sessions
+```
+
+Software recovery without power cycle: [TNC-RECOVERY.md](TNC-RECOVERY.md).
+
+## Preparation (MAX25)
+
+```bash
+./scripts/max25-ctl start --hardware tncs --device tnc2c
+# Production: max25d with [stack] auto_start = yes
+```
+
+One process per serial port — stop other userspace serial owners before HyBBX starts.
+
+## HyBBX INI
+
+Merge `share/hybbx/tnc2c-host.ini.example` or `stacks/tncs/hybbx-tnc2c.ini` into Secondary `hybbx.ini`:
+
+```ini
+[transport.packet_radio1]
+tnc = tnc2c
+device = /dev/ttyUSB0
+baud = 19200
+serial_line = 8n1
+rts_dtr = yes
+kiss_entry = none
+persist = 255
+
+[max25]
+check = yes
+```
+
+Set `device`, callsign, and frequency for your station.
+
+For HyBBX **without** max25d (standalone), use `kiss_entry = kiss_on` or `auto` instead — HyBBX runs its own recovery before KISS entry.
+
+## Checks
+
+- MAX25 prep returned host/KISS ready (boot-wait OK or max25d `ready`)
+- `fuser` on device is empty before HyBBX start (max25d holds port in production)
+- Log shows KISS active on the expected port
+
+Full contract: [docs/HYBBX.md](../../docs/HYBBX.md)
diff --git a/stacks/tncs/docs/TNC-RECOVERY.md b/stacks/tncs/docs/TNC-RECOVERY.md
new file mode 100644
index 0000000..2e36357
--- /dev/null
+++ b/stacks/tncs/docs/TNC-RECOVERY.md
@@ -0,0 +1,108 @@
+# TNC recovery (software-first)
+
+TheFirmware / TNC-2 class devices (Landolt TNC2C, PK-TNC2, TAPR TNC-2) can usually return to **host command mode** and **KISS** without a power cycle. Power-cycle with DTR held high is a **rescue fallback** only.
+
+Shared implementation: `stacks/tncs/tnc_serial_recovery.py` (used by `tnc2c-host-reset`, `tnc2c-boot-wait`, and `max25d` `kiss_bridge`).
+
+## When to use what
+
+| Situation | First action | Rescue (if software fails) |
+|-----------|--------------|----------------------------|
+| Echo-only (`INFO` → `INFO`, no banner) | `max25d` serial watch / `./tnc2c-host-reset.sh` | `./tnc2c-boot-wait.sh` only if cold-boot without DTR |
+| After `max25d` prep `error-host` | Auto boot-wait escalate (power OFF 10s → ON) | `./tnc2c-boot-wait.sh` if escalate disabled |
+| PK-TNC2 (9600 8N1) | `./pktnc2-boot-wait.sh --recover-only` | `./pktnc2-boot-wait.sh` + power cycle |
+| Cold boot, never saw `cmd:` | `./tnc2c-boot-wait.sh` (DTR before power-on) | Fix wiring (DTR/RTS, CTS bridge) |
+
+## Software recovery ladder (TheFirmware TF 2.7 native)
+
+Run with **DTR+RTS high** and the port open:
+
+1. Passive listen (~1.5 s) + **DTR settle 2 s** after open (max25d)
+2. KISS return `0xC0 0xFF 0xC0` (firmware reset → banner)
+3. Buffer flush `^Q^X` + JHOST 0 (leave host mode)
+4. **ESC V** — version / terminal probe
+5. **ESC QRES** — software cold boot (no mains power cycle; DTR must stay high)
+6. ESC E 0 + second ESC QRES
+7. Legacy TAPR `kiss off` + `INFO` (last resort only)
+
+Success markers: `TheFirmware`, `NORD`, `Version 2.7`, `Checksum`, `cmd:`.
+
+Then enter KISS: **ESC `@K`** (`0x1B 40 4B`). MYCALL: **ESC I** `<call>`.
+
+Native TF sequence: see [Software recovery ladder](#software-recovery-ladder-thefirmware-tf-27-native) above.
+
+## Operator commands
+
+```bash
+cd stacks/tncs
+
+# Software recovery (no power cycle)
+./tnc2c-host-reset.sh # terminal mode
+./tnc2c-host-reset.sh --kiss # terminal + KISS
+./tnc2c-boot-wait.sh --recover-only
+
+# PK-TNC2
+./pktnc2-boot-wait.sh --recover-only
+
+# Rescue: power cycle with DTR held (Landolt TNC2C cold-boot requirement)
+./tnc2c-boot-wait.sh # power OFF → ON while script runs
+```
+
+## MAX25 + HyBBX contract
+
+| Layer | Responsibility |
+|-------|----------------|
+| `max25d` / boot-wait | DTR sequencing, software recovery, `kiss on` / `auto` |
+| HyBBX `packet_radio` | Attach only — `kiss_entry = none` |
+| Power cycle | Rescue when DTR was low at cold boot or hardware hang |
+
+HyBBX INI (production): `kiss_entry = none`, `persist = 255` (CB CSMA), optional `[max25] check = yes`.
+
+max25d runs `recover_terminal()` in `kiss_bridge.stabilize_session()` before `MYCALL` and KISS entry. The KISS RX thread starts **only after** successful stabilization (avoids a race where RX consumes recovery replies).
+
+### max25d serial watch (automatic)
+
+When `[stack] serial_watch = yes` and `stack_recover_only = yes` (default), max25d **opens the TNC serial port itself** — no `boot-wait` subprocess (avoids port conflict). Recovery runs inline via `stabilize_session`.
+
+| INI key | Default | Role |
+|---------|---------|------|
+| `serial_watch` | yes | Enable periodic probe + auto-repair |
+| `serial_watch_interval` | 60 | Seconds between health probes |
+| `serial_repair_cooldown` | 20 | Minimum gap between repair attempts |
+| `stack_recover_only` | yes | Daemon start uses `--recover-only` (no power cycle) |
+| `stack_retry_interval` | 120 | Retry failed stack prep with recover-only |
+| `serial_bootwait_escalate` | yes | Escalate to boot-wait after repeated inline failures |
+| `serial_bootwait_escalate_after` | 3 | Inline `error-host` failures before boot-wait |
+| `serial_bootwait_escalate_cooldown` | 300 | Minimum seconds between boot-wait escalations |
+
+Triggers: initial prep `error-host` (immediate boot-wait escalate), `error-host` / `error-kiss` / `error-tx` / `error-io`, failed TX (one auto-retry). Serial watch still escalates after `serial_bootwait_escalate_after` consecutive inline failures if prep did not already escalate.
+
+**Healthy KISS:** when the backend is `ready`, serial watch does **not** tear down KISS on the periodic interval — repair runs only when status is in the error set above (not a periodic `kiss off` / probe on a working session).
+
+## Firmware RX diagnostics (max25d / boot-wait)
+
+On each recovery step, logs include byte count, matched firmware markers (`TheFirmware`, `cmd:`, …), and a printable preview. On failure:
+
+- `recovery: firmware assessment — …` — classified state (silent, echo-only, binary/KISS, non-banner text)
+- `recovery: RX capture — …` — full accumulated RX summary with hex snippet
+
+Use these lines to distinguish **DTR/cold-boot** (echo-only, 0 B passive) from **wrong baud/line** (garbage/binary) from **stuck KISS** (binary frames, no `cmd:`).
+
+Use when:
+
+- Software ladder ends in echo-only or silence
+- TNC was powered on **without** DTR high (Landolt TNC2C terminal detection)
+- Suspected hardware hang
+
+```bash
+./tnc2c-boot-wait.sh
+# While script listens: power OFF 10 s, power ON — keep script running (DTR stays high)
+```
+
+Do **not** close the serial port between boot-wait and HyBBX/max25d attach — DTR drop can return the TNC to echo mode.
+
+## See also
+
+- [TNC2C-OPERATIONS.md](TNC2C-OPERATIONS.md) — daily ops
+- [HYBBX-TNC2C.md](HYBBX-TNC2C.md) — HyBBX attach
+- [../../docs/PACKET-RADIO.md](../../docs/PACKET-RADIO.md) — serial profiles, CSMA
diff --git a/stacks/tncs/docs/TNC2C-OPERATIONS.md b/stacks/tncs/docs/TNC2C-OPERATIONS.md
new file mode 100644
index 0000000..aa9b8ef
--- /dev/null
+++ b/stacks/tncs/docs/TNC2C-OPERATIONS.md
@@ -0,0 +1,66 @@
+# TNC2C operations
+
+Landolt TNC2C on a serial port — software recovery, boot-wait rescue, verify, HyBBX attach.
+
+## Prerequisites
+
+- Port free: no HyBBX or other userspace serial owner on the TNC serial device
+- `tnc2c-serial.env` matches your wiring (device, baud, line format)
+- DTR+RTS asserted while talking to the TNC (boot-wait and host-reset do this)
+
+## Normal prep (max25d / HyBBX)
+
+Production path: `max25d` with `auto_start` runs boot-wait and software recovery before KISS attach.
+
+```bash
+./scripts/max25-ctl start --hardware tncs --device tnc2c
+# or: sudo max25d -c /etc/max25/max25d.ini
+```
+
+Manual recovery without power cycle:
+
+```bash
+cd stacks/tncs
+./tnc2c-host-reset.sh --kiss
+# or: ./tnc2c-boot-wait.sh --recover-only
+```
+
+Full ladder and rescue rules: [TNC-RECOVERY.md](TNC-RECOVERY.md).
+
+## Boot-wait (rescue — power cycle)
+
+Use when software recovery fails or the TNC was cold-started without DTR high:
+
+```bash
+cd stacks/tncs
+./tnc2c-boot-wait.sh
+```
+
+Power cycle the TNC while the script holds DTR+RTS. Expect `OK: HOST` when successful.
+
+## Verify
+
+```bash
+./tnc2c-integration-test.sh
+./tnc2c-health.sh
+```
+
+Integration test runs boot-wait + verify in one process so DTR does not drop between steps.
+
+## HyBBX
+
+1. MAX25 prep complete (boot-wait or `max25d` auto_start with recovery).
+2. Merge `share/hybbx/tnc2c-host.ini.example` into Secondary `hybbx.ini` — **`kiss_entry = none`** (MAX25 owns KISS entry).
+3. Do not run another userspace serial client on the same device while HyBBX is active.
+
+See [HYBBX-TNC2C.md](HYBBX-TNC2C.md) and [../../docs/HYBBX.md](../../docs/HYBBX.md).
+
+## Recovery
+
+| Symptom | Action |
+|---------|--------|
+| Echo only, no `cmd:` | `./tnc2c-host-reset.sh` then `./tnc2c-boot-wait.sh --recover-only` |
+| Still echo-only / silent | **Rescue:** `./tnc2c-boot-wait.sh` + power cycle (DTR high) |
+| `max25d` reports `error-host` | Stop other serial owners; run `--recover-only`; restart max25d |
+| Port busy | Stop HyBBX or other serial owner; check `fuser` on device |
+| Wrong baud | Fix `tnc2c-serial.env` and retry |
diff --git a/stacks/tncs/docs/TNC2C-REFERENCE.md b/stacks/tncs/docs/TNC2C-REFERENCE.md
new file mode 100644
index 0000000..774248c
--- /dev/null
+++ b/stacks/tncs/docs/TNC2C-REFERENCE.md
@@ -0,0 +1,43 @@
+# TNC2C reference (example profile)
+
+Example profile for a Landolt TNC2C with **19200 8N1** host link and **2400** radio AFSK. Adjust for your hardware.
+
+## Serial (host)
+
+| Parameter | Example |
+|-----------|---------|
+| Device | `/dev/ttyUSB0` |
+| Baud | 19200 |
+| Format | 8N1 |
+| RTS/DTR | Required for boot-wait |
+
+## Radio (typical CB packet)
+
+| Parameter | Example |
+|-----------|---------|
+| Modem | TCM3105 |
+| Radio baud | 2400 |
+| Band | CB |
+| Duplex | Half |
+
+## Config file
+
+`stacks/tncs/tnc2c-serial.env` — copy and edit device/baud for your station.
+
+## Tool matrix
+
+| Tool | Use |
+|------|-----|
+| `tnc2c-boot-wait.sh` | Host mode after power-on (rescue) |
+| `tnc2c-host-reset.sh` | Software recovery without power cycle |
+| `tnc2c-autotest.sh` | Quick 19200-8N1 probe |
+| `tnc2c-health.sh` | Full check, no TX |
+| `tnc2c-integration-test.sh` | Post boot-wait HyBBX gate |
+| `tnc2c-listen.sh` | Passive monitor |
+| `tnc2c-probe` | Port scan utility |
+
+## HyBBX INI snippet
+
+Use `share/hybbx/tnc2c-host.ini.example` — set `device`, `mycall`, and RF fields for your station.
+
+HyBBX TNC profiles: [hyBBX docs/TNCS.md](https://github.com/ngteq/hyBBX/blob/main/docs/TNCS.md)
diff --git a/stacks/tncs/hybbx-ax25-beacon-once.py b/stacks/tncs/hybbx-ax25-beacon-once.py
new file mode 100755
index 0000000..192d986
--- /dev/null
+++ b/stacks/tncs/hybbx-ax25-beacon-once.py
@@ -0,0 +1,105 @@
+#!/usr/bin/env python3
+"""One-shot AX.25 UI beacon matching HyBBX [broadcast] INI (HyBBX must stay running).
+
+Connects to the internal circuit hub and sends an HBX AX25_UI frame with TX flag
+the same way hybbx_broadcast_ax25() does. Requires packet_radio1 linked with
+link_password on the Main.
+"""
+import argparse
+import socket
+import struct
+import sys
+import time
+
+MAGIC = b"HBX"
+VERSION = 0x01
+HEADER_SIZE = 12
+PROTO_LINK_AUTH = 0x03
+PROTO_LINK_AUTH_ACK = 0x04
+PROTO_AX25_UI = 0x02
+FLAG_TX = 0x0002
+FLAG_PATH = 0x0004
+ADDR_LEN = 7
+
+
+def be16(v):
+ return struct.pack(">H", v)
+
+
+def be32(v):
+ return struct.pack(">I", v)
+
+
+def encode_frame(proto, flags, payload):
+ return bytes([ord("H"), ord("B"), ord("X"), VERSION, proto]) + be16(flags) + be32(len(payload)) + payload
+
+
+def encode_addr(call, ssid=0, last=False):
+ call = call.upper().ljust(6)[:6]
+ raw = bytes((ord(c) << 1) for c in call)
+ ssid_b = ((ssid & 0x0F) << 1) | (0x60 if not last else 0x61)
+ return raw + bytes([ssid_b])
+
+
+def pack_ax25_ui(dest, src, message):
+ body = bytes([0]) # digi count
+ body += encode_addr(dest, last=False)
+ body += encode_addr(src, last=True)
+ body += message.encode("ascii", errors="replace")
+ return encode_frame(PROTO_AX25_UI, FLAG_TX | FLAG_PATH, body)
+
+
+def link_auth_payload(password, link_id, role, baud=2400, duplex="half", bandwidth="low", mhz="27.235"):
+ lines = [
+ f"password={password}",
+ f"role={role}",
+ f"id={link_id}",
+ f"baud={baud}",
+ f"duplex={duplex}",
+ f"bandwidth={bandwidth}",
+ f"frequency_mhz={mhz}",
+ ]
+ return "\n".join(lines).encode("ascii")
+
+
+def recv_frames(sock, timeout=5.0):
+ sock.settimeout(timeout)
+ buf = b""
+ try:
+ while True:
+ chunk = sock.recv(4096)
+ if not chunk:
+ break
+ buf += chunk
+ except socket.timeout:
+ pass
+ return buf
+
+
+def main():
+ ap = argparse.ArgumentParser(description="HyBBX one-shot AX.25 RF beacon via circuit hub")
+ ap.add_argument("--host", default="127.0.0.1")
+ ap.add_argument("--port", type=int, default=7323)
+ ap.add_argument("--password", default="changeme")
+ ap.add_argument("--link-id", default="beacon-once")
+ ap.add_argument("--mycall", default="CB-0")
+ ap.add_argument("--dest", default="*")
+ ap.add_argument("--message", default="Broadcast: CB-0 online")
+ args = ap.parse_args()
+
+ sock = socket.create_connection((args.host, args.port), timeout=5)
+ auth = link_auth_payload(args.password, args.link_id, "gateway")
+ sock.sendall(encode_frame(PROTO_LINK_AUTH, 0, auth))
+ ack = recv_frames(sock, 3.0)
+ if b"ok=yes" not in ack:
+ print("LINK_AUTH failed:", ack[:200], file=sys.stderr)
+ sys.exit(1)
+ print(f"LINK_AUTH ok — TX {args.mycall} -> {args.dest}: {args.message!r}")
+ sock.sendall(pack_ax25_ui(args.dest, args.mycall, args.message))
+ time.sleep(2)
+ sock.close()
+ print("Beacon frame sent to circuit hub.")
+
+
+if __name__ == "__main__":
+ main()
diff --git a/stacks/tncs/hybbx-dual.ini b/stacks/tncs/hybbx-dual.ini
new file mode 100644
index 0000000..5e7e753
--- /dev/null
+++ b/stacks/tncs/hybbx-dual.ini
@@ -0,0 +1,62 @@
+; TNCs-Stack — HyBBX service mode dual TNC (packet_radio)
+; For two serial TNCs on unique UARTs. Do not use when BayCom already owns ttyS0.
+; Prerequisite: max25d [devices] with tnc2c + pktnc2, boot-wait / recovery per port.
+; BayCom dual-radio HyBBX: share/hybbx/service-dual.ini.example instead.
+;
+; Example (both TNCs active, no BayCom on ttyS0):
+; tnc2c → /dev/ttyS4
+; pktnc2 → /dev/ttyS5
+;
+; MAX25 owns KISS entry on both ports — kiss_entry = none on each transport.
+
+[networks]
+ax25 = yes
+
+[transport.packet_radio1]
+enabled = yes
+tnc = tnc2c
+protocol = kiss
+device_type = serial
+device = /dev/ttyS4
+baud = 19200
+serial_line = 8n1
+rts_dtr = yes
+kiss_entry = none
+modem = tcm3105
+radio_baud = 2400
+radio_band = cb
+frequency_mhz = 27.235
+txdelay = 50
+persist = 255
+slot = 10
+circuit_host = main.example.com
+circuit_port = 7323
+link_id = max25-tnc2c
+link_password = changeme
+mycall = CB-0
+
+[transport.packet_radio2]
+enabled = yes
+tnc = pktnc2
+protocol = kiss
+device_type = serial
+device = /dev/ttyS5
+baud = 9600
+serial_line = 8n1
+rts_dtr = no
+kiss_entry = none
+modem = tcm3105
+radio_baud = 1200
+radio_band = cb
+frequency_mhz = 27.205
+txdelay = 50
+persist = 255
+slot = 10
+circuit_host = main.example.com
+circuit_port = 7323
+link_id = max25-pktnc2
+link_password = changeme
+mycall = CB-1
+
+[max25]
+check = yes
diff --git a/stacks/tncs/hybbx-tnc2c.ini b/stacks/tncs/hybbx-tnc2c.ini
new file mode 100644
index 0000000..e1d0faa
--- /dev/null
+++ b/stacks/tncs/hybbx-tnc2c.ini
@@ -0,0 +1,38 @@
+; TNCs-Stack — HyBBX Secondary host (TNC2C / packet_radio)
+; Canonical shipped copy: share/hybbx/tnc2c-host.ini.example
+; Merge into HyBBX hybbx.ini on Secondary node.
+; Prerequisite: max25d auto_start or ./scripts/max25-ctl start --hardware tncs --device tnc2c
+;
+; Example onboard layout: TNC2C on /dev/ttyS4 (ttyS0 = BayCom, ttyS5 = PK-TNC2).
+; USB Linux hosts: set device = /dev/ttyUSB0 or /dev/serial/by-id/...
+
+[networks]
+ax25 = yes
+
+[transport.packet_radio1]
+enabled = yes
+tnc = tnc2c
+protocol = kiss
+device_type = serial
+device = /dev/ttyS4
+baud = 19200
+serial_line = 8n1
+rts_dtr = yes
+kiss_entry = none
+modem = tcm3105
+radio_baud = 2400
+clock_mhz = 4.9
+radio_band = cb
+radio_duplex = half
+frequency_mhz = 27.235
+txdelay = 50
+persist = 255
+slot = 10
+circuit_host = main.example.com
+circuit_port = 7323
+link_id = max25-tnc2c
+link_password = changeme
+mycall = CB-0
+
+[max25]
+check = yes
diff --git a/stacks/tncs/load-serial-env.sh b/stacks/tncs/load-serial-env.sh
new file mode 100755
index 0000000..a7f0e76
--- /dev/null
+++ b/stacks/tncs/load-serial-env.sh
@@ -0,0 +1,71 @@
+# shellcheck shell=bash
+# Load station serial env for TNC boot-wait (tnc2c | pktnc2).
+# Sourced by tnc2c-boot-wait.sh / pktnc2-boot-wait.sh — do not execute directly.
+#
+# Search order (first match wins):
+# /etc/max25/<id>-serial.env
+# $MAX25_ROOT/local/<id>-serial.env
+# $PWD/local/<id>-serial.env (cwd when start script runs)
+# <checkout>/local/<id>-serial.env (walk up from script dir)
+# $PREFIX/share/max25/stacks/tncs/<id>-serial.env
+
+load_serial_env() {
+ local device_id="${1:?device_id required}"
+ local script_dir="${2:?script_dir required}"
+ local env_name="${device_id}-serial.env"
+ local candidates=()
+ local p dir
+
+ candidates+=("/etc/max25/${env_name}")
+
+ if [[ -n "${MAX25_ROOT:-}" ]]; then
+ candidates+=("${MAX25_ROOT}/local/${env_name}")
+ candidates+=("${MAX25_ROOT}/stacks/tncs/${env_name}")
+ fi
+
+ if [[ -n "${PWD:-}" ]]; then
+ candidates+=("${PWD}/local/${env_name}")
+ candidates+=("${PWD}/stacks/tncs/${env_name}")
+ fi
+
+ dir="${script_dir}"
+ while [[ "${dir}" != "/" ]]; do
+ candidates+=("${dir}/local/${env_name}")
+ candidates+=("${dir}/stacks/tncs/${env_name}")
+ local parent
+ parent="$(dirname "${dir}")"
+ [[ "${parent}" == "${dir}" ]] && break
+ dir="${parent}"
+ done
+
+ candidates+=("${script_dir}/${env_name}")
+
+ local prefix="${script_dir}"
+ for _ in 1 2 3 4; do
+ prefix="$(dirname "${prefix}")"
+ candidates+=("${prefix}/share/max25/stacks/tncs/${env_name}")
+ done
+
+ for p in "${candidates[@]}"; do
+ if [[ -f "${p}" ]]; then
+ # shellcheck source=/dev/null
+ source "${p}"
+ SERIAL_ENV_FILE="${p}"
+ return 0
+ fi
+ done
+ SERIAL_ENV_FILE=""
+ return 1
+}
+
+apply_tnc2c_env() {
+ export TNC_DEV="${TNC2C_DEV:-${TNC_DEV:-/dev/ttyS4}}"
+ export TNC_BAUD="${TNC2C_BAUD:-${TNC_BAUD:-19200}}"
+ export TNC_LINE="${TNC2C_LINE:-${TNC_LINE:-8n1}}"
+}
+
+apply_pktnc2_env() {
+ export TNC_DEV="${PKTNC2_DEV:-${TNC_DEV:-/dev/ttyS5}}"
+ export TNC_BAUD="${PKTNC2_BAUD:-${TNC_BAUD:-9600}}"
+ export TNC_LINE="${PKTNC2_LINE:-${TNC_LINE:-8n1}}"
+}
diff --git a/stacks/tncs/pktnc2-boot-wait.sh b/stacks/tncs/pktnc2-boot-wait.sh
new file mode 100755
index 0000000..965fac8
--- /dev/null
+++ b/stacks/tncs/pktnc2-boot-wait.sh
@@ -0,0 +1,19 @@
+#!/bin/bash
+# PK-TNC2 boot-wait — TheFirmware TNC-2 class, typically 9600 8N1
+#
+# Software recovery (no power cycle):
+# ./pktnc2-boot-wait.sh --recover-only
+# Rescue (DTR + power cycle while script runs):
+# ./pktnc2-boot-wait.sh
+set -euo pipefail
+ROOT="$(cd "$(dirname "$0")" && pwd)"
+# shellcheck source=load-serial-env.sh
+source "${ROOT}/load-serial-env.sh"
+load_serial_env pktnc2 "${ROOT}" || true
+apply_pktnc2_env
+if [[ -n "${SERIAL_ENV_FILE:-}" ]]; then
+ echo "Using serial env: ${SERIAL_ENV_FILE}" >&2
+else
+ echo "WARN: no pktnc2-serial.env found — using ${TNC_DEV} ${TNC_BAUD} ${TNC_LINE}" >&2
+fi
+exec python3 "${ROOT}/tnc2c-boot-wait.py" "${TNC_DEV}" --baud "${TNC_BAUD}" --line "${TNC_LINE}" "$@"
diff --git a/stacks/tncs/pktnc2-serial.env b/stacks/tncs/pktnc2-serial.env
new file mode 100644
index 0000000..046e6d0
--- /dev/null
+++ b/stacks/tncs/pktnc2-serial.env
@@ -0,0 +1,4 @@
+# PK-TNC2 serial — planned; probe port after hardware delivery
+PKTNC2_DEV=/dev/ttyUSB1
+PKTNC2_BAUD=9600
+PKTNC2_LINE=8n1
diff --git a/stacks/tncs/pktnc2-to-hybbx.sh b/stacks/tncs/pktnc2-to-hybbx.sh
new file mode 100755
index 0000000..0444b5f
--- /dev/null
+++ b/stacks/tncs/pktnc2-to-hybbx.sh
@@ -0,0 +1,24 @@
+#!/bin/bash
+# AX.25 UI frame PK-TNC2 -> HyBBX/TNC2C over the air.
+# HyBBX must be running on Unit A (ttyS4). PK-TNC2: boot-wait first.
+#
+# Usage: ./pktnc2-to-hybbx.sh [message] [src] [dst]
+# Default: "CONNECT" CB-0-2 CB-0
+
+set -euo pipefail
+ROOT="$(cd "$(dirname "$0")" && pwd)"
+# shellcheck source=/dev/null
+source "$ROOT/pktnc2-serial.env"
+
+MSG="${1:-CONNECT}"
+SRC="${2:-CB-0-2}"
+DST="${3:-CB-0}"
+
+if pgrep -x hybbx >/dev/null && fuser -s "$TNC_DEV" 2>/dev/null; then
+ echo "WARN: HyBBX may hold ttyS5 - only ttyS4 should be active" >&2
+fi
+
+echo "ON-AIR TX: $SRC -> $DST: $MSG"
+echo "Radio: both on same MHz (e.g. 27.235 K24), SQ closed before TX"
+export TNC2C_DEV="$TNC_DEV" TNC2C_BAUD="$TNC_BAUD" TNC2C_LINE="$TNC_LINE"
+exec "$ROOT/tnc2c-send-test.sh" "$MSG" "$SRC" "$DST"
diff --git a/stacks/tncs/test_tnc_serial_recovery.py b/stacks/tncs/test_tnc_serial_recovery.py
new file mode 100644
index 0000000..99c9476
--- /dev/null
+++ b/stacks/tncs/test_tnc_serial_recovery.py
@@ -0,0 +1,128 @@
+#!/usr/bin/env python3
+"""Offline unit tests for tnc_serial_recovery (no serial hardware)."""
+from __future__ import annotations
+
+import importlib.util
+import sys
+from pathlib import Path
+
+ROOT = Path(__file__).resolve().parent
+spec = importlib.util.spec_from_file_location(
+ "tnc_serial_recovery", ROOT / "tnc_serial_recovery.py"
+)
+assert spec and spec.loader
+mod = importlib.util.module_from_spec(spec)
+spec.loader.exec_module(mod)
+
+
+class MockPort:
+ def __init__(self) -> None:
+ self.written: list[bytes] = []
+ self.responses: list[bytes] = []
+ self._read_idx = 0
+
+ def write_flush(self, data: bytes) -> None:
+ self.written.append(data)
+
+ def read_for(self, seconds: float) -> bytes:
+ del seconds
+ if self._read_idx < len(self.responses):
+ chunk = self.responses[self._read_idx]
+ self._read_idx += 1
+ return chunk
+ return b""
+
+
+def test_has_banner() -> None:
+ assert mod.has_banner(b"NORD><LINK TheFirmware Version 2.7")
+ assert mod.has_banner(b"cmd: ")
+ assert not mod.has_banner(b"INFO\r\nINFO")
+
+
+def test_format_rx_brief_and_classify() -> None:
+ assert mod.format_rx_brief(b"") == "0 B silent"
+ brief = mod.format_rx_brief(b"kiss off\rINFO\r")
+ assert "14 B" in brief
+ assert "markers=none" in brief
+ assert "echo-only" in mod.classify_recovery_failure(b"kiss off\rINFO\r")
+ assert "silent" in mod.classify_recovery_failure(b"")
+ diag = mod.classify_recovery_failure(b"TheFirmware cmd:")
+ assert "banner" in diag
+
+
+def test_probe_tf_terminal() -> None:
+ port = MockPort()
+ port.responses = [b"TheFirmware Version 2.7 DAMA\r\nChecksum"]
+ ok, combined, only_echo = mod.probe_tf_terminal(port.write_flush, port.read_for)
+ assert ok
+ assert not only_echo
+ assert port.written[0] == mod.TF_ESC_V
+
+
+def test_tf_mycall_frame() -> None:
+ assert mod.tf_mycall_frame("cb-0") == b"\x1bI CB-0\r"
+
+
+def test_recover_terminal_ok_on_passive_banner() -> None:
+ port = MockPort()
+ port.responses = [b"NORD TheFirmware Version 2.7\r\n"]
+ ok, rx = mod.recover_terminal(
+ port.write_flush, port.read_for, skip_kiss_frame=True, log=lambda _m: None
+ )
+ assert ok
+ assert b"TheFirmware" in rx
+ assert b"\xc0\xff\xc0" not in port.written
+
+
+def test_recover_terminal_ok_after_kiss_return() -> None:
+ port = MockPort()
+ port.responses = [
+ b"",
+ b"TheFirmware Version 2.7\r\nChecksum (0000) = DC0A\r\n",
+ ]
+ ok, rx = mod.recover_terminal(
+ port.write_flush, port.read_for, skip_kiss_frame=False, log=lambda _m: None
+ )
+ assert ok
+ assert b"TheFirmware" in rx
+ assert port.written[0] == b"\xc0\xff\xc0"
+
+
+def test_recover_terminal_sends_qres_on_failure() -> None:
+ port = MockPort()
+ port.responses = [b""] * 40
+ ok, _ = mod.recover_terminal(
+ port.write_flush, port.read_for, skip_kiss_frame=True, log=lambda _m: None
+ )
+ assert not ok
+ assert mod.TF_ESC_QRES in port.written
+ assert b"RESTART\r" not in port.written
+
+
+def test_enter_kiss_native() -> None:
+ port = MockPort()
+ port.responses = [b""]
+ mod.enter_kiss(port.write_flush, port.read_for, entry="kiss_on")
+ assert port.written[0] == mod.TF_ESC_AT_K
+
+
+def main() -> int:
+ tests = [
+ test_has_banner,
+ test_format_rx_brief_and_classify,
+ test_probe_tf_terminal,
+ test_tf_mycall_frame,
+ test_recover_terminal_ok_on_passive_banner,
+ test_recover_terminal_ok_after_kiss_return,
+ test_recover_terminal_sends_qres_on_failure,
+ test_enter_kiss_native,
+ ]
+ for fn in tests:
+ fn()
+ print(f"OK {fn.__name__}")
+ print(f"All {len(tests)} tests passed")
+ return 0
+
+
+if __name__ == "__main__":
+ sys.exit(main())
diff --git a/stacks/tncs/tnc2c-2.jpg b/stacks/tncs/tnc2c-2.jpg
new file mode 100644
index 0000000..29c70c4
--- /dev/null
+++ b/stacks/tncs/tnc2c-2.jpg
Binary files differ
diff --git a/stacks/tncs/tnc2c-autotest.py b/stacks/tncs/tnc2c-autotest.py
new file mode 100755
index 0000000..4470695
--- /dev/null
+++ b/stacks/tncs/tnc2c-autotest.py
@@ -0,0 +1,925 @@
+#!/usr/bin/env python3
+"""
+tnc2c-autotest - safe, exhaustive Landolt TNC2C serial discovery.
+
+Tests baud/parity combinations, modem lines, host vs echo vs garbage,
+firmware banner, and (optionally) KISS without RF frames.
+
+Does NOT send KISS UI frames or key PTT unless --tx is passed.
+
+Usage:
+ ./tnc2c-autotest.sh
+ ./tnc2c-autotest.sh /dev/ttyS4
+ ./tnc2c-autotest.sh --quick # likely bauds only (~45s)
+ ./tnc2c-autotest.sh --write-env # update tnc2c-serial.env if confident
+ ./tnc2c-autotest.sh --json out.json
+ ./tnc2c-autotest.sh --host-check # after power reset: 19200 8N1 only (~6s)
+ ./tnc2c-autotest.sh --host-check --write-env
+"""
+
+from __future__ import annotations
+
+import argparse
+import json
+import os
+import struct
+import subprocess
+import sys
+import time
+import termios
+import fcntl
+from dataclasses import asdict, dataclass, field
+from typing import Optional
+
+BAUD_MAP: dict[int, int] = {
+ 300: termios.B300,
+ 600: termios.B600,
+ 1200: termios.B1200,
+ 2400: termios.B2400,
+ 4800: termios.B4800,
+ 9600: termios.B9600,
+ 19200: termios.B19200,
+}
+
+BAUDS_FULL = (300, 600, 1200, 2400, 4800, 9600, 19200)
+BAUDS_QUICK = (19200, 2400, 4800, 9600)
+HOST_CHECK_LINES = ("8N1",)
+
+LINE_FORMATS = (
+ ("8N1", 8, "none", 1),
+ ("7E1", 7, "even", 1),
+ ("8E1", 8, "even", 1),
+)
+
+LINE_PREFERENCE = {"8N1": 3, "7E1": 2, "8E1": 1}
+
+FIRMWARE_MARKERS = (
+ b"TheFirmware",
+ b"NORD",
+ b"Version 2.7",
+ b"Checksum",
+ b"Copyright",
+ b"DAMA",
+ b"SMACK",
+)
+
+HOST_MARKERS = (
+ b"cmd:",
+ b"CMD:",
+ b"help",
+ b"HELP",
+ b"MYCALL",
+ b"KISS was",
+ b"KISS is",
+ b"KISS now",
+ b"Invalid",
+ b"Unknown",
+)
+
+
+@dataclass
+class ProfileResult:
+ name: str
+ baud: int
+ line: str
+ rts: bool = False
+ cts: bool = False
+ dsr: bool = False
+ dcd: bool = False
+ passive_bytes: int = 0
+ boot_bytes: int = 0
+ info_reply: bytes = b""
+ help_reply: bytes = b""
+ kiss_reply: bytes = b""
+ combined_reply: bytes = b""
+ printable_ratio: float = 0.0
+ host_score: int = 0
+ echo_only: bool = False
+ garbage: bool = False
+ has_cmd: bool = False
+ has_firmware: bool = False
+ notes: list[str] = field(default_factory=list)
+
+ def classify(self) -> str:
+ if (
+ self.has_firmware
+ or self.has_cmd
+ or has_firmware_banner(self.info_reply)
+ ):
+ return "HOST"
+ if self.garbage:
+ return "GARBAGE"
+ if self.echo_only or (
+ is_echo(b"INFO\r", self.info_reply) and is_echo(b"HELP\r", self.help_reply)
+ ):
+ return "ECHO"
+ if self.combined_reply:
+ return "PARTIAL"
+ return "SILENT"
+
+
+def has_firmware_banner(data: bytes) -> bool:
+ if not data:
+ return False
+ lower = data.lower()
+ return any(m.lower() in lower for m in FIRMWARE_MARKERS)
+
+
+def order_bauds(bauds: tuple[int, ...]) -> tuple[int, ...]:
+ """Test 19200 first so wrong bauds don't disturb a working host session."""
+ rest = tuple(b for b in bauds if b != 19200)
+ return (19200, *rest) if 19200 in bauds else bauds
+
+
+def line_format(line: str) -> tuple[int, str, int]:
+ for name, data_bits, parity, stop_bits in LINE_FORMATS:
+ if name == line:
+ return data_bits, parity, stop_bits
+ raise ValueError(f"unknown line format {line}")
+
+
+def timings_for(baud: int, gentle: bool) -> tuple[float, float, float]:
+ """Returns listen_s, cmd_wait, info_read."""
+ if gentle or baud == 19200:
+ return (2.0 if gentle else 1.5, 0.9 if gentle else 0.7, 3.0 if gentle else 3.0)
+ return (1.2, 0.55, 1.6)
+
+
+def finalize_profile(
+ res: ProfileResult,
+ info_cmd: bytes,
+ info: bytes,
+ help_cmd: bytes,
+ help: bytes,
+ kiss_on: bytes,
+ combined: bytes,
+) -> None:
+ res.info_reply = info
+ res.help_reply = help
+ res.kiss_reply = kiss_on
+ res.combined_reply = combined
+ res.printable_ratio = printable_ratio(combined)
+
+ s1, e1, c1, f1 = score_replies(info_cmd, info)
+ s2, e2, c2, f2 = score_replies(help_cmd, help)
+ s3, _, c3, f3 = score_replies(b"kiss on\r", kiss_on)
+ res.host_score = max(s1, s2, s3)
+ res.has_cmd = c1 or c2 or c3
+ res.has_firmware = (
+ f1 or f2 or f3
+ or has_firmware_banner(info)
+ or has_firmware_banner(help)
+ )
+ if res.has_firmware:
+ res.host_score = max(res.host_score, 55)
+ res.echo_only = False
+ elif is_echo(info_cmd, info) and (not help or is_echo(help_cmd, help)):
+ res.echo_only = True
+ else:
+ res.echo_only = (e1 and e2) and res.host_score < 15
+ res.garbage = (
+ bool(combined)
+ and res.printable_ratio < 0.55
+ and not res.has_firmware
+ and res.host_score < 20
+ )
+ if res.passive_bytes and res.has_firmware:
+ res.notes.append("boot_banner")
+ if kiss_on and not is_echo(b"kiss on\r", kiss_on):
+ res.notes.append("kiss_ack")
+
+
+def rank_key(res: ProfileResult) -> tuple:
+ """Higher is better. Prefer real host, then 8N1/7E1 over 8E1."""
+ return (
+ int(res.has_firmware),
+ int(res.has_cmd),
+ int(not res.echo_only),
+ res.host_score,
+ res.printable_ratio,
+ LINE_PREFERENCE.get(res.line, 0),
+ len(res.combined_reply),
+ )
+
+
+def load_env(path: str) -> dict[str, str]:
+ env: dict[str, str] = {}
+ if not os.path.isfile(path):
+ return env
+ with open(path, encoding="utf-8") as f:
+ for raw in f:
+ line = raw.strip()
+ if not line or line.startswith("#") or "=" not in line:
+ continue
+ k, v = line.split("=", 1)
+ env[k.strip()] = v.strip()
+ return env
+
+
+def printable_ratio(data: bytes) -> float:
+ if not data:
+ return 0.0
+ good = sum(1 for b in data if b in (9, 10, 13) or 32 <= b < 127)
+ return good / len(data)
+
+
+def open_serial(
+ dev: str,
+ baud: int,
+ data_bits: int,
+ parity: str,
+ stop_bits: int,
+ rts: bool = True,
+ dtr: bool = True,
+) -> int:
+ if baud not in BAUD_MAP:
+ raise ValueError(f"unsupported baud {baud}")
+ fd = os.open(dev, os.O_RDWR | os.O_NOCTTY | os.O_NONBLOCK)
+ t = termios.tcgetattr(fd)
+ t[0] = t[1] = 0
+ cflag = termios.CLOCAL | termios.CREAD
+ if data_bits == 7:
+ cflag |= termios.CS7
+ else:
+ cflag |= termios.CS8
+ if parity == "even":
+ cflag |= termios.PARENB
+ elif parity == "odd":
+ cflag |= termios.PARENB | termios.PARODD
+ t[2] = cflag
+ speed = BAUD_MAP[baud]
+ t[3] = t[4] = t[5] = speed
+ t[6][termios.VMIN] = 0
+ t[6][termios.VTIME] = 5
+ termios.tcsetattr(fd, termios.TCSANOW, t)
+ termios.tcflush(fd, termios.TCIOFLUSH)
+ flags = struct.unpack("I", fcntl.ioctl(fd, 0x5415, struct.pack("I", 0)))[0]
+ if rts:
+ flags |= 0x004
+ else:
+ flags &= ~0x004
+ if dtr:
+ flags |= 0x002
+ else:
+ flags &= ~0x002
+ fcntl.ioctl(fd, 0x5416, struct.pack("I", flags))
+ return fd
+
+
+def modem_lines(fd: int) -> tuple[bool, bool, bool, bool]:
+ flags = struct.unpack("I", fcntl.ioctl(fd, 0x5415, struct.pack("I", 0)))[0]
+ return (
+ bool(flags & 0x004),
+ bool(flags & 0x008),
+ bool(flags & 0x100),
+ bool(flags & 0x080),
+ )
+
+
+def read_for(fd: int, seconds: float) -> bytes:
+ end = time.time() + seconds
+ chunks: list[bytes] = []
+ while time.time() < end:
+ try:
+ b = os.read(fd, 4096)
+ if b:
+ chunks.append(b)
+ except BlockingIOError:
+ time.sleep(0.02)
+ return b"".join(chunks)
+
+
+def write_flush(fd: int, data: bytes) -> None:
+ os.write(fd, data)
+ termios.tcdrain(fd)
+
+
+def is_echo(cmd: bytes, reply: bytes) -> bool:
+ if not reply:
+ return False
+ c = cmd.strip(b"\r\n")
+ r = reply.strip()
+ return r in (c, c + b"\r", c + b"\n", c + b"\r\n")
+
+
+def score_replies(cmd: bytes, *chunks: bytes) -> tuple[int, bool, bool, bool]:
+ reply = b"".join(chunks)
+ if not reply:
+ return 0, False, False, False
+
+ score = 0
+ lower = reply.lower()
+ has_cmd = b"cmd:" in lower
+ has_fw = any(m.lower() in lower for m in FIRMWARE_MARKERS)
+
+ if has_cmd:
+ score += 60
+ if has_fw:
+ score += 50
+ for m in HOST_MARKERS:
+ if m.lower() in lower:
+ score += 8
+ if len(reply) > len(cmd) + 8 and not is_echo(cmd, reply):
+ score += 10
+ if b"\n" in reply and len(reply) > 30:
+ score += 5
+
+ echo = is_echo(cmd, reply) and score < 20
+ garbage = printable_ratio(reply) < 0.55 and len(reply) >= 4
+ return score, echo, has_cmd, has_fw
+
+
+def profile_name(baud: int, line: str) -> str:
+ return f"{baud}-{line}"
+
+
+def test_profile(
+ dev: str,
+ baud: int,
+ line: str,
+ data_bits: int,
+ parity: str,
+ stop_bits: int,
+ gentle: bool = False,
+) -> ProfileResult:
+ listen_s, cmd_wait, info_read = timings_for(baud, gentle)
+ res = ProfileResult(name=profile_name(baud, line), baud=baud, line=line)
+ fd = open_serial(dev, baud, data_bits, parity, stop_bits)
+ res.rts, res.cts, res.dsr, res.dcd = modem_lines(fd)
+
+ passive = read_for(fd, listen_s)
+ res.passive_bytes = len(passive)
+
+ if not gentle:
+ write_flush(fd, b"\x03\x1a")
+ time.sleep(0.15)
+ read_for(fd, 0.2)
+
+ write_flush(fd, b"\r\r")
+ time.sleep(cmd_wait)
+ wake = read_for(fd, listen_s)
+
+ write_flush(fd, b"kiss off\r")
+ time.sleep(cmd_wait)
+ kiss_off = read_for(fd, listen_s)
+
+ info_cmd = b"INFO\r"
+ write_flush(fd, info_cmd)
+ time.sleep(cmd_wait)
+ info = read_for(fd, info_read)
+
+ help_cmd = b"HELP\r"
+ write_flush(fd, help_cmd)
+ time.sleep(cmd_wait)
+ help = read_for(fd, listen_s + 0.3)
+
+ write_flush(fd, b"kiss on\r")
+ time.sleep(cmd_wait)
+ kiss_on = read_for(fd, listen_s)
+
+ write_flush(fd, b"kiss off\r")
+ time.sleep(0.2)
+ read_for(fd, 0.2)
+
+ finalize_profile(
+ res, info_cmd, info, help_cmd, help, kiss_on,
+ passive + wake + kiss_off + info + help + kiss_on,
+ )
+ os.close(fd)
+ return res
+
+
+def test_host_check(
+ dev: str,
+ baud: int,
+ line: str,
+ dtr_cycle: bool = False,
+ no_kiss: bool = False,
+) -> ProfileResult:
+ """Gentle single-profile check - DTR high, optional KISS-reset, kiss off, INFO."""
+ data_bits, parity, stop_bits = line_format(line)
+ res = ProfileResult(name=profile_name(baud, line), baud=baud, line=line)
+
+ prelude = b""
+ if dtr_cycle:
+ boot = boot_listen(dev, baud, line, data_bits, parity, stop_bits)
+ res.boot_bytes = len(boot)
+ prelude = boot
+ if has_firmware_banner(boot):
+ res.has_firmware = True
+ res.notes.append("dtr_boot_banner")
+
+ fd = open_serial(dev, baud, data_bits, parity, stop_bits)
+ res.rts, res.cts, res.dsr, res.dcd = modem_lines(fd)
+
+ passive = read_for(fd, 1.0)
+ res.passive_bytes = len(passive)
+ prelude += passive
+ if has_firmware_banner(passive):
+ res.has_firmware = True
+ res.notes.append("passive_boot_banner")
+
+ if not no_kiss:
+ write_flush(fd, b"\xc0\xff\xc0")
+ time.sleep(0.8)
+ prelude += read_for(fd, 0.8)
+ res.notes.append("kiss_reset")
+
+ write_flush(fd, b"kiss off\r")
+ time.sleep(0.4)
+ kiss_off = read_for(fd, 0.8)
+
+ info_cmd = b"INFO\r"
+ write_flush(fd, info_cmd)
+ time.sleep(0.3)
+ info = read_for(fd, 3.5)
+
+ finalize_profile(res, info_cmd, info, b"HELP\r", b"", b"", prelude + kiss_off + info)
+ os.close(fd)
+ return res
+
+
+def boot_listen(dev: str, baud: int, line: str, data_bits: int, parity: str, stop_bits: int) -> bytes:
+ fd = open_serial(dev, baud, data_bits, parity, stop_bits)
+ flags = struct.unpack("I", fcntl.ioctl(fd, 0x5415, struct.pack("I", 0)))[0]
+ flags &= ~0x002
+ fcntl.ioctl(fd, 0x5416, struct.pack("I", flags))
+ time.sleep(0.8)
+ flags |= 0x002 | 0x004
+ fcntl.ioctl(fd, 0x5416, struct.pack("I", flags))
+ data = read_for(fd, 4.0)
+ os.close(fd)
+ return data
+
+
+def deep_test(dev: str, res: ProfileResult) -> ProfileResult:
+ line = res.line
+ data_bits, parity, stop_bits = line_format(line)
+ gentle = res.baud == 19200
+
+ fd = open_serial(dev, res.baud, data_bits, parity, stop_bits)
+ res.rts, res.cts, res.dsr, res.dcd = modem_lines(fd)
+
+ write_flush(fd, b"kiss off\r")
+ time.sleep(0.6 if gentle else 0.4)
+ read_for(fd, 0.3)
+
+ for _ in range(4):
+ write_flush(fd, b"\r")
+ time.sleep(0.35)
+ enters = read_for(fd, 1.5)
+ if b"cmd:" in enters.lower():
+ res.has_cmd = True
+ res.host_score += 20
+ res.notes.append("cmd_after_enter")
+
+ write_flush(fd, b"MYCALL TEST-0\r")
+ time.sleep(0.6)
+ mycall = read_for(fd, 1.0)
+ if mycall and not is_echo(b"MYCALL TEST-0\r", mycall):
+ lower = mycall.lower()
+ if b"cmd:" in lower or has_firmware_banner(mycall):
+ res.host_score += 15
+ res.notes.append("mycall_ack")
+
+ if not res.has_firmware:
+ boot = boot_listen(dev, res.baud, line, data_bits, parity, stop_bits)
+ res.boot_bytes = len(boot)
+ if has_firmware_banner(boot):
+ res.has_firmware = True
+ res.host_score = max(res.host_score, 55)
+ res.echo_only = False
+ res.notes.append("dtr_boot_banner")
+
+ os.close(fd)
+ return res
+
+
+def port_holders(dev: str) -> list[str]:
+ holders: list[str] = []
+ try:
+ out = subprocess.run(
+ ["fuser", "-v", dev],
+ capture_output=True,
+ text=True,
+ timeout=3,
+ )
+ if out.returncode == 0 and out.stderr:
+ for line in out.stderr.splitlines():
+ line = line.strip()
+ if line and not line.startswith("USER"):
+ holders.append(line)
+ except (FileNotFoundError, subprocess.TimeoutExpired):
+ pass
+ return holders
+
+
+def preflight(dev: str) -> list[str]:
+ issues: list[str] = []
+ if os.system("pgrep -x hybbx >/dev/null 2>&1") == 0:
+ issues.append("hybbx is running - stop it first")
+ if not os.path.exists(dev):
+ issues.append(f"{dev} does not exist")
+ elif not os.access(dev, os.R_OK | os.W_OK):
+ issues.append(f"no access to {dev} (dialout group?)")
+ holders = port_holders(dev)
+ if holders:
+ issues.append(f"port in use: {', '.join(holders[:3])}")
+ return issues
+
+
+def quartz_hint(baud: int) -> str:
+ hints = []
+ term_98304 = {9600: "TERM Pos.1", 4800: "TERM Pos.2", 2400: "TERM Pos.3", 1200: "TERM Pos.4"}
+ term_49152 = {4800: "TERM Pos.1", 2400: "TERM Pos.2", 1200: "TERM Pos.3", 600: "TERM Pos.4"}
+ if baud in term_98304:
+ hints.append(f"with 9.8304 MHz crystal: {term_98304[baud]}")
+ if baud in term_49152:
+ hints.append(f"with 4.9152 MHz crystal: {term_49152[baud]}")
+ if baud == 19200:
+ hints.append("19200 is in the manual only with RADIO bridge (2.4576 MHz Pos.5), not TERM")
+ hints.append("-> possible mod, extended firmware, or different crystal")
+ return "; ".join(hints) if hints else "-"
+
+
+def suggest_env(best: ProfileResult) -> dict[str, str]:
+ line = best.line.lower()
+ return {
+ "TNC2C_DEV": "", # filled by caller
+ "TNC2C_BAUD": str(best.baud),
+ "TNC2C_LINE": line,
+ "TNC2C_RADIO_BAUD": "2400",
+ "TNC2C_MODEM": "tcm3105",
+ "HYBBX_SERIAL_LINE": line,
+ }
+
+
+def write_env_file(path: str, dev: str, best: ProfileResult) -> None:
+ vals = suggest_env(best)
+ vals["TNC2C_DEV"] = dev
+ lines = [
+ f"# Auto-generated by tnc2c-autotest {time.strftime('%Y-%m-%d %H:%M:%S')}",
+ f"# Best profile: {best.name} ({best.classify()}, score={best.host_score})",
+ ]
+ for k, v in vals.items():
+ lines.append(f"{k}={v}")
+ lines.append("")
+ with open(path, "w", encoding="utf-8") as f:
+ f.write("\n".join(lines))
+
+
+# --- optional TX (from health check) ---
+
+def kiss_escape(data: bytes) -> bytes:
+ out = bytearray()
+ for b in data:
+ if b == 0xC0:
+ out.extend((0xDB, 0xDC))
+ elif b == 0xDB:
+ out.extend((0xDB, 0xDD))
+ else:
+ out.append(b)
+ return bytes(out)
+
+
+def ax25_addr(call: str, ssid: int, last: bool) -> bytes:
+ call = call.upper().ljust(6)[:6]
+ raw = bytes((ord(c) << 1) for c in call)
+ ssid_b = ((ssid & 0x0F) << 1) | (0x01 if last else 0x00)
+ return raw + bytes([ssid_b])
+
+
+def ax25_crc(data: bytes) -> int:
+ crc = 0xFFFF
+ for b in data:
+ crc ^= b
+ for _ in range(8):
+ crc = (crc >> 1) ^ 0x8408 if crc & 1 else crc >> 1
+ return crc ^ 0xFFFF
+
+
+def ax25_ui(src: str, dst: str, info: bytes) -> bytes:
+ body = ax25_addr(dst, 0, False) + ax25_addr(src, 0, True) + b"\x03\xF0" + info
+ crc = ax25_crc(body)
+ return body + bytes((crc & 0xFF, crc >> 8))
+
+
+def run_tx_test(dev: str, best: ProfileResult) -> dict:
+ fmt = next(f for f in LINE_FORMATS if f[0] == best.line)
+ _, data_bits, parity, stop_bits = fmt
+ fd = open_serial(dev, best.baud, data_bits, parity, stop_bits)
+ write_flush(fd, b"kiss off\r")
+ time.sleep(0.4)
+ read_for(fd, 0.3)
+ write_flush(fd, b"kiss on\r")
+ time.sleep(0.5)
+ read_for(fd, 0.3)
+ payload = ax25_ui("TEST-0", "CQ", b"AUTOTEST")
+ pkt = b"\xC0" + bytes([0]) + kiss_escape(payload) + b"\xC0"
+ write_flush(fd, pkt)
+ time.sleep(3.0)
+ rx = read_for(fd, 1.0)
+ write_flush(fd, b"kiss off\r")
+ os.close(fd)
+ return {
+ "sent": len(pkt),
+ "rx_len": len(rx),
+ "echoed": pkt in rx or payload in rx,
+ }
+
+
+def preview(data: bytes, limit: int = 72) -> str:
+ if not data:
+ return "(empty)"
+ s = data[:limit].decode("ascii", errors="replace")
+ s = s.replace("\r", "\\r").replace("\n", "\\n")
+ if len(data) > limit:
+ s += "..."
+ return s
+
+
+def print_header(title: str) -> None:
+ print(f"\n{'=' * 64}")
+ print(title)
+ print("=" * 64)
+
+
+def print_result_summary(
+ best: ProfileResult,
+ dev: str,
+ ranked: list[ProfileResult],
+ root: str,
+ args: argparse.Namespace,
+ suggested: dict[str, str],
+ tx_result: Optional[dict],
+ t0: float,
+) -> int:
+ host_profiles = [r for r in ranked if r.classify() == "HOST"]
+ echo_profiles = [r for r in ranked if r.classify() == "ECHO"]
+ garbage_profiles = [r for r in ranked if r.classify() == "GARBAGE"]
+
+ print_header("Result")
+ print(f" Best profile: {best.name}")
+ print(f" Class: {best.classify()}")
+ print(f" Host score: {best.host_score}")
+ print(f" Modem: RTS={int(best.rts)} CTS={int(best.cts)} DSR={int(best.dsr)}")
+ print(f" INFO: {preview(best.info_reply, 120)}")
+ if best.has_firmware or has_firmware_banner(best.info_reply):
+ print(" Firmware banner: YES")
+ if best.has_cmd:
+ print(" cmd: prompt: YES")
+ print(f" Crystal/bridge: {quartz_hint(best.baud)}")
+
+ if host_profiles:
+ print(f"\n HOST profiles ({len(host_profiles)}): {', '.join(r.name for r in host_profiles[:5])}")
+ if echo_profiles and not host_profiles:
+ print(f"\n ECHO only ({len(echo_profiles)}) - TNC mirrors commands, does not process them.")
+ print(" -> Power-reset the TNC, then: ./tnc2c-autotest.sh --host-check --write-env")
+ if garbage_profiles:
+ top_g = sorted(garbage_profiles, key=lambda r: r.printable_ratio, reverse=True)[:3]
+ print(f" GARBAGE profiles: {', '.join(r.name for r in top_g)} (wrong baud/parity)")
+
+ print("\n Recommended settings:")
+ print(f" max25-terminal or tnc2c-host-reset @ {dev} -b {best.baud} (# {best.line})")
+ for k in ("TNC2C_BAUD", "TNC2C_LINE", "HYBBX_SERIAL_LINE"):
+ print(f" {k}={suggested[k]}")
+
+ if args.write_env and best.classify() == "HOST":
+ env_path = os.path.join(root, "tnc2c-serial.env")
+ write_env_file(env_path, dev, best)
+ print(f"\n Written: {env_path}")
+ elif args.write_env:
+ print("\n --write-env skipped (no reliable HOST profile)")
+
+ print_header("KISS-TX")
+ if not args.tx:
+ print(" Skipped (safe). Optional: --tx (PTT may key!)")
+ else:
+ print(" WARNING: sends AX.25 UI frame - check PTT LED!")
+ tx_result = run_tx_test(dev, best)
+ print(f" Sent {tx_result['sent']} B, RX {tx_result['rx_len']} B, Echo={tx_result['echoed']}")
+
+ elapsed = time.time() - t0
+ status = "OK" if best.classify() == "HOST" else (
+ "DEGRADED" if best.classify() in ("ECHO", "PARTIAL") else "FAIL"
+ )
+
+ print_header("SUMMARY")
+ print(f" Status: {status}")
+ print(f" Device: {dev}")
+ print(f" Profile: {best.name} ({best.classify()})")
+ print(f" Duration: {elapsed:.0f}s")
+
+ if args.json:
+ report = {
+ "device": dev,
+ "status": status,
+ "elapsed_s": round(elapsed, 1),
+ "best": asdict(best),
+ "ranking": [asdict(r) for r in ranked[:12]],
+ "suggested_env": suggested,
+ "tx": tx_result,
+ }
+
+ def fix(obj):
+ if isinstance(obj, dict):
+ return {k: fix(v) for k, v in obj.items()}
+ if isinstance(obj, list):
+ return [fix(v) for v in obj]
+ if isinstance(obj, bytes):
+ return obj.decode("ascii", errors="replace")
+ return obj
+
+ with open(args.json, "w", encoding="utf-8") as f:
+ json.dump(fix(report), f, indent=2, ensure_ascii=False)
+ print(f" JSON: {args.json}")
+
+ return 0 if status == "OK" else (1 if status == "DEGRADED" else 2)
+
+
+def run_host_check(
+ dev: str,
+ root: str,
+ env: dict[str, str],
+ args: argparse.Namespace,
+) -> int:
+ t0 = time.time()
+ print_header("TNC2C HOST-CHECK (DTR high - no DTR drop, no baud sweep)")
+ print(f"Device: {dev}")
+ print(f"Started: {time.strftime('%Y-%m-%d %H:%M:%S')}")
+ if args.dtr_cycle:
+ print("Note: --dtr-cycle active (DTR drop - may trigger echo mode)")
+
+ print_header("1) Preflight")
+ issues = preflight(dev)
+ if issues:
+ for i in issues:
+ print(f" FAIL: {i}")
+ print("\nAbort - clear port/background services and retry.")
+ return 2
+ print(" OK: port free, hybbx inactive, access OK")
+
+ kiss_step = "kiss off -> INFO" if args.no_kiss else "KISS-reset -> kiss off -> INFO"
+ print_header(f"2) Host-Check 19200-8N1 (passive -> {kiss_step})")
+ results: list[ProfileResult] = []
+ for line in HOST_CHECK_LINES:
+ name = profile_name(19200, line)
+ if not args.quiet:
+ print(f" {name:12} ...", end="", flush=True)
+ try:
+ r = test_host_check(
+ dev, 19200, line, dtr_cycle=args.dtr_cycle, no_kiss=args.no_kiss
+ )
+ results.append(r)
+ if not args.quiet:
+ print(
+ f" {r.classify():7} score={r.host_score:3} fw={int(r.has_firmware)} "
+ f"INFO={preview(r.info_reply, 50)!r}"
+ )
+ except OSError as e:
+ if not args.quiet:
+ print(f" ERROR: {e}")
+
+ if not results:
+ print(" No profiles tested.")
+ return 2
+
+ ranked = sorted(results, key=rank_key, reverse=True)
+ best = ranked[0]
+ suggested = suggest_env(best)
+ suggested["TNC2C_DEV"] = dev
+ return print_result_summary(best, dev, ranked, root, args, suggested, None, t0)
+
+
+def main() -> int:
+ parser = argparse.ArgumentParser(description="TNC2C safe exhaustive autotest")
+ parser.add_argument("device", nargs="?", default=None)
+ parser.add_argument(
+ "--host-check",
+ action="store_true",
+ help="gentle 19200-8N1 only - after power reset (~10s)",
+ )
+ parser.add_argument(
+ "--dtr-cycle",
+ action="store_true",
+ help="brief DTR drop (legacy; may trigger echo mode - not for host-check)",
+ )
+ parser.add_argument(
+ "--no-kiss",
+ action="store_true",
+ help="host-check: skip KISS-reset frame",
+ )
+ parser.add_argument("--quick", action="store_true", help="19200 first, then 2400/4800/9600")
+ parser.add_argument("--write-env", action="store_true", help="write tnc2c-serial.env if HOST found")
+ parser.add_argument("--json", metavar="FILE", help="write JSON report")
+ parser.add_argument("--tx", action="store_true", help="KISS TX test - keys PTT!")
+ parser.add_argument("-q", "--quiet", action="store_true")
+ args = parser.parse_args()
+
+ root = os.path.dirname(os.path.abspath(__file__))
+ env = load_env(os.path.join(root, "tnc2c-serial.env"))
+ dev = args.device or env.get("TNC2C_DEV", "/dev/ttyS4")
+
+ if args.host_check:
+ return run_host_check(dev, root, env, args)
+
+ bauds = order_bauds(BAUDS_QUICK if args.quick else BAUDS_FULL)
+ total = len(bauds) * len(LINE_FORMATS)
+
+ t0 = time.time()
+ print_header("TNC2C AUTOTEST (safe - no PTT unless --tx)")
+ print(f"Device: {dev}")
+ print(f"Mode: {'quick' if args.quick else 'full'} ({total} profiles, 19200 first)")
+ print(f"Started: {time.strftime('%Y-%m-%d %H:%M:%S')}")
+
+ print_header("1) Preflight")
+ issues = preflight(dev)
+ if issues:
+ for i in issues:
+ print(f" FAIL: {i}")
+ print("\nAbort - clear port/background services and retry.")
+ return 2
+ print(" OK: port free, hybbx inactive, access OK")
+
+ print_header("2) Host precheck 19200-8N1")
+ precheck = test_host_check(dev, 19200, "8N1", dtr_cycle=args.dtr_cycle)
+ print(
+ f" {precheck.name:12} {precheck.classify():7} fw={int(precheck.has_firmware)} "
+ f"INFO={preview(precheck.info_reply, 60)!r}"
+ )
+
+ print_header("3) Profile sweep (19200 first, no Ctrl+C at 19200)")
+ results: list[ProfileResult] = [precheck]
+ seen = {precheck.name}
+ n = 0
+ for baud in bauds:
+ for line, data_bits, parity, stop_bits in LINE_FORMATS:
+ name = profile_name(baud, line)
+ if name in seen:
+ continue
+ n += 1
+ if not args.quiet:
+ print(f" [{n:2}/{total - 1}] {name:12} ...", end="", flush=True)
+ try:
+ gentle = baud == 19200
+ r = test_profile(dev, baud, line, data_bits, parity, stop_bits, gentle=gentle)
+ results.append(r)
+ seen.add(name)
+ if not args.quiet:
+ print(
+ f" {r.classify():7} score={r.host_score:3} "
+ f"CTS={int(r.cts)} pr={r.printable_ratio:.0%} "
+ f"INFO={preview(r.info_reply, 40)!r}"
+ )
+ except OSError as e:
+ if not args.quiet:
+ print(f" ERROR: {e}")
+
+ ranked = sorted(results, key=rank_key, reverse=True)
+ best = ranked[0] if ranked else None
+
+ print_header("4) Ranking (Top 8)")
+ for r in ranked[:8]:
+ print(
+ f" {r.name:12} {r.classify():7} score={r.host_score:3} "
+ f"cmd={int(r.has_cmd)} fw={int(r.has_firmware)} echo={int(r.echo_only)} "
+ f"CTS={int(r.cts)}"
+ )
+
+ print_header("5) Deep test (Top 3)")
+ deep_targets = [r for r in ranked[:3] if r.host_score > 0 or r.combined_reply]
+ if not deep_targets and ranked:
+ deep_targets = ranked[:1]
+ for r in deep_targets:
+ if r.name == precheck.name and precheck.classify() == "HOST":
+ continue
+ print(f" -> {r.name} ...", flush=True)
+ deep_test(dev, r)
+
+ ranked = sorted(results, key=rank_key, reverse=True)
+ best = ranked[0]
+
+ if best and best.classify() in ("ECHO", "PARTIAL") and env.get("TNC2C_BAUD"):
+ saved_line = env.get("TNC2C_LINE", "8n1").upper()
+ if "7" in saved_line and "E" in saved_line:
+ saved_line = "7E1"
+ elif saved_line.startswith("8") and "E" in saved_line:
+ saved_line = "8E1"
+ else:
+ saved_line = "8N1"
+ saved_name = profile_name(int(env["TNC2C_BAUD"]), saved_line)
+ saved = next((r for r in results if r.name == saved_name), None)
+ if saved and saved.classify() != "HOST":
+ print(f" (Note: saved profile {saved_name} - use --host-check after power reset)")
+
+ if best is None:
+ print(" No response on any profile.")
+ return 2
+
+ suggested = suggest_env(best)
+ suggested["TNC2C_DEV"] = dev
+ return print_result_summary(best, dev, ranked, root, args, suggested, None, t0)
+
+
+if __name__ == "__main__":
+ sys.exit(main())
diff --git a/stacks/tncs/tnc2c-autotest.sh b/stacks/tncs/tnc2c-autotest.sh
new file mode 100755
index 0000000..d0f119d
--- /dev/null
+++ b/stacks/tncs/tnc2c-autotest.sh
@@ -0,0 +1,11 @@
+#!/bin/bash
+# Exhaustive safe TNC2C autotest — no PTT unless --tx.
+# Usage:
+# ./tnc2c-autotest.sh --host-check # after power reset (~6s, DTR stays high)
+# ./tnc2c-autotest.sh --host-check --write-env
+# ./tnc2c-autotest.sh --quick # full sweep, 19200 first (~2 min)
+# ./tnc2c-autotest.sh --write-env
+
+set -euo pipefail
+ROOT="$(cd "$(dirname "$0")" && pwd)"
+exec python3 "$ROOT/tnc2c-autotest.py" "$@"
diff --git a/stacks/tncs/tnc2c-boot-wait.py b/stacks/tncs/tnc2c-boot-wait.py
new file mode 100644
index 0000000..ba24e26
--- /dev/null
+++ b/stacks/tncs/tnc2c-boot-wait.py
@@ -0,0 +1,326 @@
+#!/usr/bin/env python3
+"""
+Hold /dev/ttyS4 open with DTR+RTS during TNC power-on, then native TF probe (ESC V).
+
+Usage:
+ 1. Run this FIRST (waits for port):
+ ./tnc2c-boot-wait.sh
+ 2. While it says "waiting", power-cycle the TNC (off 10s, on).
+ 3. Script detects banner or runs KISS return + ESC V automatically.
+"""
+
+from __future__ import annotations
+
+import argparse
+import importlib.util
+import fcntl
+import os
+import struct
+import sys
+import time
+import termios
+
+def load_recovery():
+ root = os.path.dirname(os.path.abspath(__file__))
+ path = os.path.join(root, "tnc_serial_recovery.py")
+ spec = importlib.util.spec_from_file_location("tnc_serial_recovery", path)
+ if spec is None or spec.loader is None:
+ return None
+ mod = importlib.util.module_from_spec(spec)
+ spec.loader.exec_module(mod)
+ return mod
+
+
+FIRMWARE_MARKERS = (
+ b"TheFirmware",
+ b"NORD",
+ b"Version 2.7",
+ b"Checksum",
+ b"Copyright",
+)
+
+
+def has_banner(data: bytes) -> bool:
+ lower = data.lower()
+ return any(m.lower() in lower for m in FIRMWARE_MARKERS)
+
+
+def parse_line(line: str) -> tuple[int, int]:
+ line = line.lower()
+ if line == "7e1":
+ return termios.CS7, termios.PARENB
+ if line == "8n1":
+ return termios.CS8, 0
+ raise ValueError(f"unsupported serial line: {line}")
+
+
+def parse_baud(baud: int) -> int:
+ table = {
+ 1200: termios.B1200,
+ 2400: termios.B2400,
+ 4800: termios.B4800,
+ 9600: termios.B9600,
+ 19200: termios.B19200,
+ }
+ if baud not in table:
+ raise ValueError(f"unsupported baud: {baud}")
+ return table[baud]
+
+
+def open_serial(dev: str, baud: int, line: str) -> int:
+ speed = parse_baud(baud)
+ databits, parity = parse_line(line)
+ fd = os.open(dev, os.O_RDWR | os.O_NOCTTY | os.O_NONBLOCK)
+ t = termios.tcgetattr(fd)
+ t[0] = t[1] = 0
+ t[2] = termios.CLOCAL | termios.CREAD | databits | parity
+ t[3] = t[4] = t[5] = speed
+ t[6][termios.VMIN] = 0
+ t[6][termios.VTIME] = 5
+ termios.tcsetattr(fd, termios.TCSANOW, t)
+ termios.tcflush(fd, termios.TCIOFLUSH)
+ flags = struct.unpack("I", fcntl.ioctl(fd, 0x5415, struct.pack("I", 0)))[0]
+ flags |= 0x004 | 0x002
+ fcntl.ioctl(fd, 0x5416, struct.pack("I", flags))
+ return fd
+
+
+def read_for(fd: int, seconds: float) -> bytes:
+ end = time.time() + seconds
+ chunks: list[bytes] = []
+ while time.time() < end:
+ try:
+ b = os.read(fd, 4096)
+ if b:
+ chunks.append(b)
+ except BlockingIOError:
+ time.sleep(0.02)
+ return b"".join(chunks)
+
+
+def write_flush(fd: int, data: bytes) -> None:
+ os.write(fd, data)
+ termios.tcdrain(fd)
+
+
+def show(data: bytes) -> None:
+ if data:
+ print(data.decode("ascii", errors="replace"))
+
+
+def is_cmd_echo(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 run_tx_rx(fd: int) -> None:
+ """TX KISS frame + passive RX on same open fd (before port close)."""
+ print("\n--- TX (KISS TEST-0 -> CQ, TXRX) ---")
+ write_flush(fd, b"\x1b@K")
+ time.sleep(0.5)
+ read_for(fd, 0.3)
+
+ def kiss_escape(data: bytes) -> bytes:
+ out = bytearray()
+ for b in data:
+ if b == 0xC0:
+ out.extend((0xDB, 0xDC))
+ elif b == 0xDB:
+ out.extend((0xDB, 0xDD))
+ else:
+ out.append(b)
+ return bytes(out)
+
+ def ax25_addr(call: str, last: bool) -> bytes:
+ call = call.upper().ljust(6)[:6]
+ raw = bytes((ord(c) << 1) for c in call)
+ ssid = ((0 & 0x0F) << 1) | (0x01 if last else 0x00)
+ return raw + bytes([ssid])
+
+ def ax25_crc(data: bytes) -> int:
+ crc = 0xFFFF
+ for b in data:
+ crc ^= b
+ for _ in range(8):
+ crc = (crc >> 1) ^ 0x8408 if crc & 1 else crc >> 1
+ return crc ^ 0xFFFF
+
+ body = ax25_addr("CQ", False) + ax25_addr("TEST-0", True) + b"\x03\xF0TXRX"
+ crc = ax25_crc(body)
+ payload = body + bytes((crc & 0xFF, crc >> 8))
+ pkt = b"\xC0\x00" + kiss_escape(payload) + b"\xC0"
+ write_flush(fd, pkt)
+ time.sleep(3.0)
+ tx_rx = read_for(fd, 1.0)
+ print(f" Sent {len(pkt)} B, serial RX {len(tx_rx)} B")
+ print(" -> CHECK: LED2 PTT + modem tone on 2m CB")
+
+ write_flush(fd, b"\xc0\xff\xc0")
+ time.sleep(1.0)
+ read_for(fd, 0.5)
+
+ print("\n--- RX (passive 10s) ---")
+ print(" Listening ... (empty band = 0 bytes OK; LED3 CD with noise?)")
+ rx = read_for(fd, 10.0)
+ print(f" Passive RX: {len(rx)} bytes")
+ if rx and b"\xc0" in rx:
+ print(" OK: KISS frame received")
+
+
+def verify_hybbx_ready(fd: int) -> tuple[bool, bytes, bool]:
+ """ESC V on open fd. Returns (banner_in_reply, data, only_echo)."""
+ write_flush(fd, b"\x1bV\r")
+ time.sleep(0.4)
+ info = read_for(fd, 5.0)
+ only_echo = is_cmd_echo(b"\x1bV\r", info)
+ return has_banner(info), info, only_echo
+
+
+def finish_host(fd: int, hybbx_ready: bool, do_tx_rx: bool) -> int:
+ if hybbx_ready:
+ print("\n--- HyBBX-Verify (ESC V) ---")
+ ok, verify, only_echo = verify_hybbx_ready(fd)
+ show(verify)
+ host_ok = ok or only_echo
+ if not host_ok:
+ if do_tx_rx:
+ run_tx_rx(fd)
+ os.close(fd)
+ print("\nWARN: unexpected ESC V response")
+ return 1
+ else:
+ host_ok = True
+
+ if do_tx_rx:
+ run_tx_rx(fd)
+
+ os.close(fd)
+ if host_ok:
+ print("\nOK: HOST - HyBBX-ready" + (" + TX/RX tested" if do_tx_rx else ""))
+ return 0
+ return 1
+
+
+def main() -> int:
+ parser = argparse.ArgumentParser(description="Hold DTR during TNC boot")
+ default_dev = (
+ os.environ.get("TNC_DEV")
+ or os.environ.get("TNC2C_DEV")
+ or os.environ.get("PKTNC2_DEV")
+ or "/dev/ttyS4"
+ )
+ default_baud = int(
+ os.environ.get("TNC_BAUD")
+ or os.environ.get("TNC2C_BAUD")
+ or os.environ.get("PKTNC2_BAUD")
+ or "19200"
+ )
+ default_line = (
+ os.environ.get("TNC_LINE")
+ or os.environ.get("TNC2C_LINE")
+ or os.environ.get("PKTNC2_LINE")
+ or "8n1"
+ )
+
+ parser.add_argument("device", nargs="?", default=default_dev)
+ parser.add_argument("--baud", type=int, default=default_baud)
+ parser.add_argument("--line", default=default_line, choices=("8n1", "7e1"))
+ parser.add_argument("--wait", type=int, default=45, help="seconds to listen for boot")
+ parser.add_argument(
+ "--no-hybbx-ready",
+ action="store_true",
+ help="only detect boot banner, skip ESC V verify",
+ )
+ parser.add_argument(
+ "--tx-rx",
+ action="store_true",
+ help="TX KISS + passive RX on same port before close (with power cycle)",
+ )
+ parser.add_argument(
+ "--recover-only",
+ action="store_true",
+ help="skip boot listen; run software recovery ladder only (no power cycle)",
+ )
+ args = parser.parse_args()
+ hybbx_ready = not args.no_hybbx_ready
+ do_tx_rx = args.tx_rx
+
+ if not os.access(args.device, os.R_OK | os.W_OK):
+ print(f"FAIL: no access to {args.device}", file=sys.stderr)
+ return 2
+
+ fd = open_serial(args.device, args.baud, args.line)
+
+ if args.recover_only:
+ print(f"=== TNC recover-only @ {args.device} {args.baud} {args.line.upper()} ===")
+ rec = load_recovery()
+ if rec is None:
+ print("FAIL: tnc_serial_recovery.py not found", file=sys.stderr)
+ os.close(fd)
+ return 2
+
+ def wf(data: bytes) -> None:
+ write_flush(fd, data)
+
+ def rf(seconds: float) -> bytes:
+ return read_for(fd, seconds)
+
+ ok, rx = rec.recover_terminal(wf, rf, log=lambda m: print(f" {m}"))
+ if ok:
+ show(rx[-800:] if len(rx) > 800 else rx)
+ return finish_host(fd, hybbx_ready, do_tx_rx)
+ os.close(fd)
+ print("\nFAIL: software recovery — try power-cycle with DTR high (re-run without --recover-only)")
+ return 1
+
+ print(f"=== TNC boot-wait @ {args.device} {args.baud} {args.line.upper()} ===")
+ print("DTR+RTS HIGH - power OFF the TNC now (10s), then power ON.")
+ print("CB: squelch CLOSED (CD off) - otherwise TX/boot may be disturbed.")
+ print(f"Listening {args.wait}s for boot banner ...\n")
+
+ buf = read_for(fd, args.wait)
+
+ if has_banner(buf):
+ print("--- Boot banner (passive) ---")
+ show(buf)
+ return finish_host(fd, hybbx_ready, do_tx_rx)
+
+ print(f"(no banner in {args.wait}s - trying KISS return + ESC V)\n")
+ write_flush(fd, b"\xc0\xff\xc0")
+ time.sleep(1.0)
+ buf += read_for(fd, 2.5)
+ write_flush(fd, b"\x1bV\r")
+ time.sleep(0.4)
+ info = read_for(fd, 4.0)
+ buf += info
+
+ print("--- KISS return + ESC V ---")
+ show(info if info else buf[-500:])
+
+ if has_banner(buf):
+ return finish_host(fd, hybbx_ready, do_tx_rx)
+
+ rec = load_recovery()
+ if rec is not None:
+ print("\n--- software recovery ladder ---")
+
+ def wf(data: bytes) -> None:
+ write_flush(fd, data)
+
+ def rf(seconds: float) -> bytes:
+ return read_for(fd, seconds)
+
+ ok, rx = rec.recover_terminal(wf, rf, log=lambda m: print(f" {m}"))
+ buf += rx
+ if ok:
+ show(rx[-800:] if len(rx) > 800 else rx)
+ return finish_host(fd, hybbx_ready, do_tx_rx)
+
+ os.close(fd)
+ print("\nDEGRADED: try ./tnc2c-host-reset.sh or power-cycle with this script running")
+ return 1
+
+
+if __name__ == "__main__":
+ sys.exit(main())
diff --git a/stacks/tncs/tnc2c-boot-wait.sh b/stacks/tncs/tnc2c-boot-wait.sh
new file mode 100755
index 0000000..ac101c1
--- /dev/null
+++ b/stacks/tncs/tnc2c-boot-wait.sh
@@ -0,0 +1,14 @@
+#!/bin/bash
+# TNC2C boot-wait — Landolt TNC2C, typically 19200 8N1, DTR during power-on
+set -euo pipefail
+ROOT="$(cd "$(dirname "$0")" && pwd)"
+# shellcheck source=load-serial-env.sh
+source "${ROOT}/load-serial-env.sh"
+load_serial_env tnc2c "${ROOT}" || true
+apply_tnc2c_env
+if [[ -n "${SERIAL_ENV_FILE:-}" ]]; then
+ echo "Using serial env: ${SERIAL_ENV_FILE}" >&2
+else
+ echo "WARN: no tnc2c-serial.env found — using ${TNC_DEV} ${TNC_BAUD} ${TNC_LINE}" >&2
+fi
+exec python3 "${ROOT}/tnc2c-boot-wait.py" "${TNC_DEV}" --baud "${TNC_BAUD}" --line "${TNC_LINE}" "$@"
diff --git a/stacks/tncs/tnc2c-chat.sh b/stacks/tncs/tnc2c-chat.sh
new file mode 100755
index 0000000..9323b5e
--- /dev/null
+++ b/stacks/tncs/tnc2c-chat.sh
@@ -0,0 +1,52 @@
+#!/bin/bash
+# Quick TNC2C terminal check on /dev/ttyS4 (no RF required).
+# WARNING: kiss on / MONitor can KEY PTT and transmit on the radio!
+# For passive RX only use: ./tnc2c-listen.sh
+# Usage: ./tnc2c-chat.sh [baud] [line]
+# line: 7e1 | 8n1 (default from $TNC2C_LINE or 9600/7e1)
+
+set -euo pipefail
+
+ROOT="$(cd "$(dirname "$0")" && pwd)"
+# shellcheck source=/dev/null
+[[ -f "$ROOT/tnc2c-serial.env" ]] && source "$ROOT/tnc2c-serial.env"
+
+DEV="${TNC2C_DEV:-/dev/ttyS4}"
+BAUD="${1:-${TNC2C_BAUD:-19200}}"
+LINE="${2:-${TNC2C_LINE:-7e1}}"
+
+case "$LINE" in
+7e1|7E1) STTY="cs7 parenb -cstopb" ;;
+8n1|8N1) STTY="cs8 -parenb -cstopb" ;;
+*) echo "Unknown line format: $LINE (use 7e1 or 8n1)" >&2; exit 1 ;;
+esac
+
+if [[ ! -r "$DEV" || ! -w "$DEV" ]]; then
+ echo "Cannot access $DEV (dialout group?)" >&2
+ exit 1
+fi
+
+echo "TNC2C chat: $DEV @ ${BAUD} ${LINE}"
+stty -F "$DEV" "$BAUD" $STTY raw -echo -icanon min 0 time 10
+
+exec 3<>"$DEV"
+
+printf '\r' >&3
+sleep 0.5
+timeout 2 cat <&3 | xxd -g1 | head -8 || true
+
+# Do NOT send kiss on — keys PTT. Only safe host echo probes:
+for cmd in '' INFO HELP; do
+ echo "--- $cmd ---"
+ if [[ -n "$cmd" ]]; then
+ printf '%s\r' "$cmd" >&3
+ else
+ printf '\r' >&3
+ fi
+ sleep 1
+ timeout 4 cat <&3 | tee /tmp/tnc2c-last.bin | xxd -g1 | head -12
+ strings -n 3 /tmp/tnc2c-last.bin || true
+done
+
+exec 3<&-
+exec 3>&-
diff --git a/stacks/tncs/tnc2c-check.sh b/stacks/tncs/tnc2c-check.sh
new file mode 100755
index 0000000..c0ab705
--- /dev/null
+++ b/stacks/tncs/tnc2c-check.sh
@@ -0,0 +1,62 @@
+#!/bin/bash
+# Preflight: HyBBX off? handshake OK? quick echo on TNC2C.
+# Usage: ./tnc2c-check.sh [device]
+
+set -euo pipefail
+
+ROOT="$(cd "$(dirname "$0")" && pwd)"
+# shellcheck source=/dev/null
+[[ -f "$ROOT/tnc2c-serial.env" ]] && source "$ROOT/tnc2c-serial.env"
+
+DEV="${1:-${TNC2C_DEV:-/dev/ttyS4}}"
+BAUD="${TNC2C_BAUD:-19200}"
+LINE="${TNC2C_LINE:-7e1}"
+
+echo "=== TNC2C check: $DEV (${BAUD} ${LINE}) ==="
+
+if pgrep -x hybbx >/dev/null 2>&1; then
+ echo "WARN: hybbx is running ($(pgrep -x hybbx | tr '\n' ' ')) - port may be in use"
+fi
+
+if fuser -s "$DEV" 2>/dev/null; then
+ echo "WARN: $DEV in use:"
+ fuser -v "$DEV" 2>&1 || true
+else
+ echo "OK: $DEV not in use"
+fi
+
+if [[ ! -r "$DEV" || ! -w "$DEV" ]]; then
+ echo "FAIL: no access to $DEV (dialout group?)" >&2
+ exit 1
+fi
+
+case "$LINE" in
+7e1|7E1) STTY="cs7 parenb -cstopb" ;;
+8n1|8N1) STTY="cs8 -parenb -cstopb" ;;
+*) echo "Unknown line: $LINE" >&2; exit 1 ;;
+esac
+
+stty -F "$DEV" "$BAUD" $STTY raw -echo min 0 time 5
+
+python3 <<PY
+import fcntl, os, struct, time
+
+TIOCMGET = 0x5415
+fd = os.open("$DEV", os.O_RDWR | os.O_NOCTTY)
+flags = struct.unpack("I", fcntl.ioctl(fd, TIOCMGET, struct.pack("I", 0)))[0]
+names = [(0x004, "RTS"), (0x008, "CTS"), (0x002, "DTR"), (0x100, "DSR")]
+for bit, name in names:
+ print(f" {name}={'1' if flags & bit else '0'}")
+os.write(fd, b"INFO\r")
+time.sleep(0.8)
+data = os.read(fd, 512)
+os.close(fd)
+print(f" RX {len(data)} bytes:", data[:80])
+PY
+
+if [[ -x "$ROOT/tnc2c-probe" ]]; then
+ echo "--- tnc2c-probe (single port) ---"
+ "$ROOT/tnc2c-probe" "$DEV" 2>&1 | tail -20
+fi
+
+echo "=== done ==="
diff --git a/stacks/tncs/tnc2c-expect.exp b/stacks/tncs/tnc2c-expect.exp
new file mode 100755
index 0000000..eaef9c0
--- /dev/null
+++ b/stacks/tncs/tnc2c-expect.exp
@@ -0,0 +1,80 @@
+#!/usr/bin/expect -f
+# Auto-detect TNC2C host line on a serial device.
+# Usage: ./tnc2c-expect.exp /dev/ttyS4
+
+set timeout 6
+if {[llength $argv] < 1} {
+ puts "Usage: $argv0 <device>"
+ exit 1
+}
+set dev [lindex $argv 0]
+
+proc try_line {dev baud parity} {
+ global timeout
+ if {$parity eq "7E1"} {
+ spawn stty -F $dev $baud cs7 parenb -cstopb raw -echo min 0 time 10
+ expect eof
+ } else {
+ spawn stty -F $dev $baud cs8 -parenb -cstopb raw -echo min 0 time 10
+ expect eof
+ }
+
+ spawn cu -l $dev -s $baud
+ if {[catch {expect {
+ -re "." { }
+ timeout { }
+ }}]} {
+ catch {close}
+ return 0
+ }
+
+ set score 0
+ set sample ""
+
+ send "\r\r"
+ sleep 1
+ send "INFO\r"
+ sleep 3
+
+ if {[catch {set sample $expect_out(buffer)}]} {
+ set sample ""
+ }
+
+ if {[string match -nocase *cmd:* $sample]} { incr score 300 }
+ if {[string match -nocase *MYCALL* $sample]} { incr score 150 }
+ if {[string match -nocase *TXDELAY* $sample]} { incr score 100 }
+ if {[string match -nocase *INFO* $sample]} { incr score 20 }
+ incr score [string length $sample]
+
+ puts " $baud $parity score=$score"
+ if {$sample ne ""} {
+ set short [string range $sample 0 180]
+ regsub -all {\r} $short {\\r} short
+ regsub -all {\n} $short {\\n} short
+ puts " $short"
+ }
+
+ send "~.\r"
+ catch {close}
+ return $score
+}
+
+puts "TNC2C expect scan on $dev"
+set best 0
+set bestcfg ""
+
+foreach baud {9600 2400 4800 1200 19200 600 300} {
+ foreach parity {7E1 8N1} {
+ set s [try_line $dev $baud $parity]
+ if {$s > $best} {
+ set best $s
+ set bestcfg "$baud $parity"
+ }
+ }
+}
+
+puts "\nBest: $bestcfg (score=$best)"
+if {$best >= 200} {
+ exit 0
+}
+exit 2
diff --git a/stacks/tncs/tnc2c-find.c b/stacks/tncs/tnc2c-find.c
new file mode 100644
index 0000000..fe26a1b
--- /dev/null
+++ b/stacks/tncs/tnc2c-find.c
@@ -0,0 +1,306 @@
+/*
+ * tnc2c-find — detect host baud/format for Landolt TNC2C on a serial port.
+ * Tries TERM jumper speeds (300..9600) with 7E1 and 8N1, scores cmd:/INFO replies.
+ *
+ * Build: make find
+ * Run: ./tnc2c-find /dev/ttyS4
+ */
+
+#if defined(__linux__) || defined(__GLIBC__)
+#define _DEFAULT_SOURCE 1
+#define _GNU_SOURCE 1
+#endif
+
+#include <errno.h>
+#include <fcntl.h>
+#include <stdio.h>
+#include <stdlib.h>
+#include <string.h>
+#include <sys/ioctl.h>
+#include <termios.h>
+#include <time.h>
+#include <unistd.h>
+
+typedef struct line_profile {
+ const char *label;
+ speed_t speed;
+ unsigned int baud;
+ tcflag_t databits;
+ tcflag_t parity;
+} line_profile_t;
+
+static const line_profile_t PROFILES[] = {
+ { "19200-8N1", B19200, 19200, CS8, 0 },
+ { "19200-7E1", B19200, 19200, CS7, PARENB },
+ { "9600-7E1", B9600, 9600, CS7, PARENB },
+ { "2400-7E1", B2400, 2400, CS7, PARENB },
+ { "4800-7E1", B4800, 4800, CS7, PARENB },
+ { "1200-7E1", B1200, 1200, CS7, PARENB },
+ { "9600-8N1", B9600, 9600, CS8, 0 },
+ { "2400-8N1", B2400, 2400, CS8, 0 },
+ { "4800-8N1", B4800, 4800, CS8, 0 },
+ { "1200-8N1", B1200, 1200, CS8, 0 },
+ { "600-7E1", B600, 600, CS7, PARENB },
+ { "300-7E1", B300, 300, CS7, PARENB },
+ { "600-8N1", B600, 600, CS8, 0 },
+ { "300-8N1", B300, 300, CS8, 0 },
+};
+
+static int sleep_ms(unsigned ms)
+{
+ struct timespec ts;
+
+ ts.tv_sec = (time_t)(ms / 1000u);
+ ts.tv_nsec = (long)(ms % 1000u) * 1000000L;
+ return nanosleep(&ts, NULL);
+}
+
+static int set_modem_lines(int fd)
+{
+#ifdef TIOCMGET
+ int flags;
+
+ if (ioctl(fd, TIOCMGET, &flags) != 0) {
+ return -1;
+ }
+ flags |= TIOCM_RTS | TIOCM_DTR;
+ return ioctl(fd, TIOCMSET, &flags);
+#else
+ (void)fd;
+ return 0;
+#endif
+}
+
+static int open_line(const char *device, const line_profile_t *profile)
+{
+ struct termios tty;
+ int fd;
+
+ fd = open(device, O_RDWR | O_NOCTTY | O_NONBLOCK);
+ if (fd < 0) {
+ return -1;
+ }
+
+ if (tcgetattr(fd, &tty) != 0) {
+ close(fd);
+ return -1;
+ }
+
+ cfmakeraw(&tty);
+ tty.c_cflag |= (CLOCAL | CREAD);
+ tty.c_cflag &= ~(CSIZE | PARENB | PARODD | CSTOPB);
+ tty.c_cflag |= profile->databits;
+ if (profile->parity != 0) {
+ tty.c_cflag |= profile->parity;
+ }
+ tty.c_cc[VMIN] = 0;
+ tty.c_cc[VTIME] = 10;
+
+#if defined(__linux__) || defined(__GLIBC__)
+ if (cfsetspeed(&tty, profile->speed) != 0) {
+ cfsetispeed(&tty, profile->speed);
+ cfsetospeed(&tty, profile->speed);
+ }
+#else
+ cfsetispeed(&tty, profile->speed);
+ cfsetospeed(&tty, profile->speed);
+#endif
+
+ if (tcsetattr(fd, TCSANOW, &tty) != 0) {
+ close(fd);
+ return -1;
+ }
+
+ tcflush(fd, TCIOFLUSH);
+ (void)set_modem_lines(fd);
+ return fd;
+}
+
+static size_t read_for(int fd, unsigned char *buf, size_t cap, unsigned ms)
+{
+ size_t total = 0;
+ unsigned elapsed = 0;
+
+ while (elapsed < ms && total < cap) {
+ ssize_t n = read(fd, buf + total, cap - total);
+
+ if (n > 0) {
+ total += (size_t)n;
+ elapsed = 0;
+ continue;
+ }
+ if (n < 0 && errno != EAGAIN && errno != EWOULDBLOCK) {
+ break;
+ }
+ sleep_ms(40);
+ elapsed += 40;
+ }
+
+ return total;
+}
+
+static int score_reply(const unsigned char *buf, size_t len)
+{
+ size_t i;
+ int score = 0;
+ int printable = 0;
+
+ if (len == 0) {
+ return 0;
+ }
+
+ for (i = 0; i < len; i++) {
+ unsigned char c = buf[i];
+
+ if (c >= 0x20 && c < 0x7f) {
+ printable++;
+ }
+ }
+
+ score += printable;
+ if (memmem(buf, len, "cmd:", 4) != NULL ||
+ memmem(buf, len, "CMD:", 4) != NULL) {
+ score += 200;
+ }
+ if (memmem(buf, len, "MYCALL", 6) != NULL ||
+ memmem(buf, len, "mycall", 6) != NULL) {
+ score += 120;
+ }
+ if (memmem(buf, len, "TXDELAY", 7) != NULL) {
+ score += 80;
+ }
+ if (memmem(buf, len, "KISS", 4) != NULL) {
+ score += 40;
+ }
+ if (len > 40) {
+ score += 60;
+ } else if (len > 15) {
+ score += 20;
+ }
+
+ /* Pure echo of a short command is weak. */
+ if (len <= 12 && printable > 0 && printable == (int)len) {
+ score -= 10;
+ }
+
+ return score;
+}
+
+static void dump_ascii(const unsigned char *buf, size_t len)
+{
+ size_t i;
+
+ for (i = 0; i < len; i++) {
+ unsigned char c = buf[i];
+
+ if (c == '\r') {
+ fputs("\\r", stdout);
+ } else if (c == '\n') {
+ fputs("\\n", stdout);
+ } else if (c >= 0x20 && c < 0x7f) {
+ putchar((int)c);
+ } else {
+ printf("\\x%02x", c);
+ }
+ }
+}
+
+static int try_profile(const char *device, const line_profile_t *profile,
+ int *best_score, char *best_label, size_t label_cap,
+ unsigned char *best_buf, size_t *best_len,
+ size_t best_cap)
+{
+ int fd;
+ unsigned char buf[4096];
+ size_t len;
+ int score;
+ const char *seq[] = { "\r\r\r", "INFO\r" };
+ size_t s;
+
+ fd = open_line(device, profile);
+ if (fd < 0) {
+ return -1;
+ }
+
+ sleep_ms(200);
+ len = 0;
+ for (s = 0; s < sizeof(seq) / sizeof(seq[0]); s++) {
+ (void)write(fd, seq[s], strlen(seq[s]));
+ sleep_ms(400);
+ len += read_for(fd, buf + len, sizeof(buf) - len, 3500);
+ }
+
+ close(fd);
+ score = score_reply(buf, len);
+
+ if (score > *best_score) {
+ *best_score = score;
+ snprintf(best_label, label_cap, "%s", profile->label);
+ if (len > best_cap) {
+ len = best_cap;
+ }
+ memcpy(best_buf, buf, len);
+ *best_len = len;
+ }
+
+ if (score >= 30) {
+ printf(" %-12s score=%4d len=%4zu ",
+ profile->label, score, len);
+ dump_ascii(buf, len < 120 ? len : 120);
+ putchar('\n');
+ }
+
+ return score;
+}
+
+int main(int argc, char **argv)
+{
+ const char *device;
+ size_t i;
+ int best_score = -1;
+ char best_label[32] = "";
+ unsigned char best_buf[4096];
+ size_t best_len = 0;
+
+ if (argc != 2) {
+ fprintf(stderr, "Usage: %s <serial-device>\n", argv[0]);
+ return 1;
+ }
+
+ device = argv[1];
+ printf("TNC2C auto-detect on %s\n", device);
+ printf("(RTS/DTR on, probe cmd:/INFO; best TERM jumper speeds)\n\n");
+
+ for (i = 0; i < sizeof(PROFILES) / sizeof(PROFILES[0]); i++) {
+ (void)try_profile(device, &PROFILES[i], &best_score, best_label,
+ sizeof(best_label), best_buf, &best_len,
+ sizeof(best_buf));
+ }
+
+ printf("\n");
+ if (best_score < 0) {
+ printf("No profile could open %s\n", device);
+ return 2;
+ }
+
+ printf("Best profile: %s (score=%d, %zu bytes)\n",
+ best_label[0] ? best_label : "?", best_score, best_len);
+ if (best_len > 0) {
+ printf("Sample: ");
+ dump_ascii(best_buf, best_len < 200 ? best_len : 200);
+ printf("\n");
+ }
+
+ if (best_score >= 200) {
+ printf("\nHost mode OK — use baud/serial_line from profile above in HyBBX.\n");
+ return 0;
+ }
+
+ if (best_score >= 30) {
+ printf("\nPartial link — check TERM jumper; try: tnc2c-host-reset %s\n", device);
+ return 3;
+ }
+
+ printf("\nNo usable TNC dialog detected.\n");
+ return 4;
+}
diff --git a/stacks/tncs/tnc2c-health.py b/stacks/tncs/tnc2c-health.py
new file mode 100644
index 0000000..642b6cf
--- /dev/null
+++ b/stacks/tncs/tnc2c-health.py
@@ -0,0 +1,445 @@
+#!/usr/bin/env python3
+"""
+tnc2c-health - full Landolt TNC2C health check (serial, host, KISS, optional TX).
+
+Does NOT replace visual LED checks (Power/PTT/CD/Connected/Status).
+
+Usage:
+ ./tnc2c-health.sh
+ ./tnc2c-health.sh /dev/ttyS4
+ ./tnc2c-health.sh --tx # include KISS TX attempt (keys PTT!)
+ ./tnc2c-health.sh --tx --quiet
+"""
+
+from __future__ import annotations
+
+import argparse
+import fcntl
+import os
+import struct
+import sys
+import time
+import termios
+from dataclasses import dataclass, field
+from typing import Optional
+
+BAUD_MAP = {
+ 300: termios.B300,
+ 600: termios.B600,
+ 1200: termios.B1200,
+ 2400: termios.B2400,
+ 4800: termios.B4800,
+ 9600: termios.B9600,
+ 19200: termios.B19200,
+}
+
+PROFILES = [
+ ("19200-8N1", 19200, False),
+ ("19200-7E1", 19200, True),
+ ("9600-7E1", 9600, True),
+ ("9600-8N1", 9600, False),
+ ("2400-7E1", 2400, True),
+ ("4800-7E1", 4800, True),
+]
+
+HOST_MARKERS = (
+ b"cmd:",
+ b"CMD:",
+ b"TNC",
+ b"***",
+ b"help",
+ b"HELP",
+ b"version",
+ b"Version",
+ b"MYCALL",
+ b"KISS was",
+ b"KISS is",
+)
+
+ECHO_ONLY_MARKERS = (
+ b"kiss off",
+ b"kiss on",
+ b"INFO",
+ b"HELP",
+ b"RESTART",
+)
+
+
+@dataclass
+class ProfileResult:
+ name: str
+ rts: bool
+ cts: bool
+ dsr: bool
+ info_reply: bytes = b""
+ host_score: int = 0
+ echo_only: bool = False
+ has_cmd_prompt: bool = False
+ kiss_ack: bool = False
+ kiss_echo_only: bool = False
+ notes: list[str] = field(default_factory=list)
+
+
+def load_env(path: str) -> dict[str, str]:
+ env: dict[str, str] = {}
+ if not os.path.isfile(path):
+ return env
+ with open(path, encoding="utf-8") as f:
+ for line in f:
+ line = line.strip()
+ if not line or line.startswith("#"):
+ continue
+ if "=" in line:
+ k, v = line.split("=", 1)
+ env[k.strip()] = v.strip()
+ return env
+
+
+def open_serial(dev: str, baud: int, seven_e1: bool) -> int:
+ fd = os.open(dev, os.O_RDWR | os.O_NOCTTY | os.O_NONBLOCK)
+ if baud not in BAUD_MAP:
+ raise ValueError(f"unsupported baud {baud}")
+ t = termios.tcgetattr(fd)
+ t[0] = t[1] = 0
+ t[2] = termios.CLOCAL | termios.CREAD | (termios.CS7 if seven_e1 else termios.CS8)
+ if seven_e1:
+ t[2] |= termios.PARENB
+ speed = BAUD_MAP[baud]
+ t[3] = t[4] = t[5] = speed
+ t[6][termios.VMIN] = 0
+ t[6][termios.VTIME] = 5
+ termios.tcsetattr(fd, termios.TCSANOW, t)
+ termios.tcflush(fd, termios.TCIOFLUSH)
+ flags = struct.unpack("I", fcntl.ioctl(fd, 0x5415, struct.pack("I", 0)))[0]
+ flags |= 0x004 | 0x002 # RTS DTR
+ fcntl.ioctl(fd, 0x5416, struct.pack("I", flags))
+ return fd
+
+
+def modem_lines(fd: int) -> tuple[bool, bool, bool]:
+ flags = struct.unpack("I", fcntl.ioctl(fd, 0x5415, struct.pack("I", 0)))[0]
+ return bool(flags & 0x004), bool(flags & 0x008), bool(flags & 0x100)
+
+
+def read_for(fd: int, seconds: float) -> bytes:
+ end = time.time() + seconds
+ chunks: list[bytes] = []
+ while time.time() < end:
+ try:
+ b = os.read(fd, 4096)
+ if b:
+ chunks.append(b)
+ except BlockingIOError:
+ time.sleep(0.03)
+ return b"".join(chunks)
+
+
+def write_flush(fd: int, data: bytes) -> None:
+ os.write(fd, data)
+ termios.tcdrain(fd)
+
+
+def score_host_reply(cmd: bytes, reply: bytes) -> tuple[int, bool, bool]:
+ if not reply:
+ return 0, False, False
+
+ stripped_cmd = cmd.strip(b"\r\n")
+ echo_only = reply.strip() == stripped_cmd or reply.strip() in (
+ stripped_cmd,
+ stripped_cmd + b"\r",
+ stripped_cmd + b"\n",
+ stripped_cmd + b"\r\n",
+ )
+
+ score = 0
+ if b"cmd:" in reply.lower() or b"cmd>" in reply.lower():
+ score += 50
+ for m in HOST_MARKERS:
+ if m.lower() in reply.lower():
+ score += 10
+ if len(reply) > len(cmd) + 4 and not echo_only:
+ score += 5
+ if b"\n" in reply and len(reply) > 20:
+ score += 5
+
+ has_cmd = b"cmd:" in reply.lower()
+ return score, echo_only, has_cmd
+
+
+def kiss_escape(data: bytes) -> bytes:
+ out = bytearray()
+ for b in data:
+ if b == 0xC0:
+ out.extend((0xDB, 0xDC))
+ elif b == 0xDB:
+ out.extend((0xDB, 0xDD))
+ else:
+ out.append(b)
+ return bytes(out)
+
+
+def ax25_addr(call: str, ssid: int, last: bool) -> bytes:
+ call = call.upper().ljust(6)[:6]
+ raw = bytes((ord(c) << 1) for c in call)
+ ssid_b = ((ssid & 0x0F) << 1) | (0x01 if last else 0x00)
+ return raw + bytes([ssid_b])
+
+
+def ax25_crc(data: bytes) -> int:
+ crc = 0xFFFF
+ for b in data:
+ crc ^= b
+ for _ in range(8):
+ crc = (crc >> 1) ^ 0x8408 if crc & 1 else crc >> 1
+ return crc ^ 0xFFFF
+
+
+def ax25_ui(src: str, dst: str, info: bytes) -> bytes:
+ body = ax25_addr(dst, 0, False) + ax25_addr(src, 0, True)
+ body += b"\x03\xF0" + info
+ crc = ax25_crc(body)
+ return body + bytes((crc & 0xFF, crc >> 8))
+
+
+def kiss_frame(port: int, payload: bytes) -> bytes:
+ return b"\xC0" + bytes([port & 0x0F]) + kiss_escape(payload) + b"\xC0"
+
+
+def test_profile(dev: str, name: str, baud: int, seven_e1: bool) -> ProfileResult:
+ res = ProfileResult(name=name, rts=False, cts=False, dsr=False)
+ fd = open_serial(dev, baud, seven_e1)
+ res.rts, res.cts, res.dsr = modem_lines(fd)
+
+ # break out of possible modes
+ write_flush(fd, b"\x03\x1a") # Ctrl+C, Ctrl+Z
+ time.sleep(0.2)
+ read_for(fd, 0.3)
+
+ write_flush(fd, b"kiss off\r")
+ time.sleep(0.4)
+ read_for(fd, 0.4)
+
+ write_flush(fd, b"\r")
+ time.sleep(0.2)
+ read_for(fd, 0.3)
+
+ cmd = b"INFO\r"
+ write_flush(fd, cmd)
+ time.sleep(0.6)
+ info = read_for(fd, 1.0)
+ res.info_reply = info
+ score, echo, has_cmd = score_host_reply(cmd, info)
+ res.host_score = score
+ res.echo_only = echo and score < 15
+ res.has_cmd_prompt = has_cmd
+
+ write_flush(fd, b"HELP\r")
+ time.sleep(0.5)
+ help_r = read_for(fd, 0.8)
+ hs, echo2, hc = score_host_reply(b"HELP\r", help_r)
+ res.host_score = max(res.host_score, hs)
+ if hc:
+ res.has_cmd_prompt = True
+ if echo2 and hs < 10:
+ res.echo_only = res.echo_only or True
+
+ write_flush(fd, b"kiss on\r")
+ time.sleep(0.5)
+ kiss_r = read_for(fd, 0.6)
+ if b"kiss" in kiss_r.lower() and len(kiss_r) < 32:
+ res.kiss_echo_only = True
+ if b"on" in kiss_r.lower() and len(kiss_r) > 8 and not res.kiss_echo_only:
+ res.kiss_ack = True
+
+ passive = read_for(fd, 1.0)
+ if passive:
+ res.notes.append(f"passive_after_kiss={len(passive)}B")
+
+ write_flush(fd, b"kiss off\r")
+ time.sleep(0.2)
+ os.close(fd)
+ return res
+
+
+def run_tx_test(dev: str, profile_name: str, baud: int, seven_e1: bool) -> dict:
+ fd = open_serial(dev, baud, seven_e1)
+ write_flush(fd, b"kiss off\r")
+ time.sleep(0.4)
+ read_for(fd, 0.3)
+ write_flush(fd, b"kiss on\r")
+ time.sleep(0.5)
+ read_for(fd, 0.3)
+
+ payload = ax25_ui("TEST-0", "CQ", b"HEALTH")
+ pkt = kiss_frame(0, payload)
+ write_flush(fd, pkt)
+ time.sleep(3.0)
+ rx = read_for(fd, 1.0)
+ write_flush(fd, b"kiss off\r")
+ os.close(fd)
+
+ echoed = pkt in rx or payload in rx
+ return {
+ "sent": len(pkt),
+ "rx_len": len(rx),
+ "frame_echoed_back": echoed,
+ "rx_sample": rx[:96],
+ }
+
+
+def print_header(title: str) -> None:
+ print(f"\n{'=' * 60}")
+ print(title)
+ print("=" * 60)
+
+
+def main() -> int:
+ parser = argparse.ArgumentParser(description="TNC2C full health check")
+ parser.add_argument("device", nargs="?", default=None)
+ parser.add_argument("--tx", action="store_true", help="attempt KISS TX (keys PTT)")
+ parser.add_argument("--quiet", action="store_true")
+ args = parser.parse_args()
+
+ root = os.path.dirname(os.path.abspath(__file__))
+ env = load_env(os.path.join(root, "tnc2c-serial.env"))
+ dev = args.device or env.get("TNC2C_DEV", "/dev/ttyS4")
+
+ print_header("TNC2C HEALTH CHECK")
+ print(f"Device: {dev}")
+ print(f"Time: {time.strftime('%Y-%m-%d %H:%M:%S')}")
+
+ # --- preflight ---
+ print_header("1) Preflight")
+ if os.system("pgrep -x hybbx >/dev/null 2>&1") == 0:
+ print("FAIL: hybbx is running - stop it first")
+ return 2
+ print("OK: hybbx not active")
+
+ if not os.access(dev, os.R_OK | os.W_OK):
+ print(f"FAIL: no access to {dev} (dialout group?)")
+ return 2
+ print(f"OK: access to {dev}")
+
+ # --- profile sweep ---
+ print_header("2) Serial profiles (host detection)")
+ results: list[ProfileResult] = []
+ for name, baud, seven in PROFILES:
+ try:
+ r = test_profile(dev, name, baud, seven)
+ results.append(r)
+ info_preview = r.info_reply[:40].replace(b"\r", b"\\r").replace(b"\n", b"\\n")
+ print(
+ f" {name:12} RTS={int(r.rts)} CTS={int(r.cts)} DSR={int(r.dsr)} "
+ f"score={r.host_score:2} cmd={int(r.has_cmd_prompt)} "
+ f"echo_only={int(r.echo_only)} INFO={info_preview!r}"
+ )
+ except OSError as e:
+ print(f" {name:12} ERROR: {e}")
+
+ best = max(results, key=lambda x: x.host_score, default=None)
+ if best is None:
+ print("FAIL: no profiles tested")
+ return 2
+
+ # pick working line: prefer env, else best score, else 19200-8N1 clean echo
+ preferred = env.get("TNC2C_LINE", "8n1").lower()
+ work_name = best.name
+ work_baud = 19200
+ work_7e1 = preferred.startswith("7e")
+ for name, baud, seven in PROFILES:
+ if name == best.name:
+ work_baud, work_7e1 = baud, seven
+ break
+ if best.host_score < 15:
+ for r in results:
+ if r.name == "19200-8N1" and r.info_reply.strip() in (
+ b"INFO\r\n",
+ b"INFO\r",
+ b"INFO",
+ ):
+ work_name = r.name
+ work_baud, work_7e1 = 19200, False
+ break
+
+ print(f"\nBest profile (host score): {best.name} (score={best.host_score})")
+ print(f"Working profile for tests: {work_name}")
+
+ # --- diagnosis ---
+ print_header("3) Serial / host mode diagnosis")
+ any_cmd = any(r.has_cmd_prompt for r in results)
+ mostly_echo = all(r.echo_only or r.host_score < 15 for r in results)
+
+ if any_cmd:
+ print("OK: real host mode (cmd: prompt) detected")
+ elif mostly_echo:
+ print("WARN: command echo only - no cmd: prompt, no INFO listing")
+ print(" -> TNC probably does NOT process commands")
+ print(" -> KISS/PTT via software probably ineffective")
+ print(" -> LEDs do not change (matches this picture)")
+ else:
+ print("PARTIAL: replies present but no clear host dialog")
+
+ if all(not r.cts for r in results):
+ print("WARN: CTS=0 on all profiles - handshake problem")
+ print(" -> bridge DB25 pins 4 and 5 in cable if needed (manual)")
+
+ # --- RF / LED manual ---
+ print_header("4) TNC LEDs (no radio / with radio)")
+ print("No radio (audio off): Pin4 open -> LED3 (CD) often ON continuously (normal)")
+ print("With radio: LED3 should be ON with noise, OFF with squelch closed")
+ print("LED 1 Power | LED 2 PTT | LED 3 CD | LED 4 Connected | LED 5 Status")
+
+ # --- optional TX ---
+ print_header("5) KISS send test")
+ if not args.tx:
+ print("Skipped (without --tx). To test: ./tnc2c-health.sh --tx")
+ else:
+ print("WARNING: sends KISS frame - PTT may key!")
+ tx = run_tx_test(dev, work_name, work_baud, work_7e1)
+ print(f" Profile: {work_name}")
+ print(f" Sent: {tx['sent']} bytes KISS")
+ print(f" RX: {tx['rx_len']} bytes")
+ print(f" Frame echoed back directly: {tx['frame_echoed_back']}")
+ if tx["frame_echoed_back"]:
+ print(" WARN: frame returns unchanged -> possible serial loopback,")
+ print(" not sent over radio")
+ print(" -> CHECK NOW: LED2 (PTT) and S-meter on radio")
+
+ # --- hardware checklist ---
+ print_header("6) Hardware checklist (if not transmitting)")
+ print(" [ ] PTT radio red -> TNC DIN Pin 3")
+ print(" [ ] GND radio black -> TNC DIN Pin 2")
+ print(" [ ] TX radio white -> TNC DIN Pin 1")
+ print(" [ ] RX 3.5mm tip -> TNC DIN Pin 4 (left center)")
+ print(" [ ] VOX on AE6110 -> OFF")
+ print(" [ ] Mike gain on TNC -> increase slowly (pot on modem)")
+ print(" [ ] RADIO bridge TNC -> 2400 baud (Pos. 2)")
+ print(" [ ] TERM bridge TNC -> 19200")
+
+ # --- software checklist ---
+ print_header("7) Software / next steps")
+ if mostly_echo and not any_cmd:
+ print(" 1. Software recovery (no power cycle):")
+ print(" ./tnc2c-host-reset.sh --kiss")
+ print(" ./tnc2c-boot-wait.sh --recover-only")
+ print(" 2. Expect cmd: or TheFirmware banner after recovery")
+ print(" 3. Rescue only if still echo-only: ./tnc2c-boot-wait.sh + power cycle (DTR high)")
+ print(" 4. When cmd: appears -> ./tnc2c-health.sh --tx")
+ print(" 5. HyBBX: kiss_entry=none when max25d owns KISS (see docs/TNC-RECOVERY.md)")
+ else:
+ print(f" Serial OK with {work_name}")
+ print(" ./tnc2c-send-test.sh for single send test")
+ print(" HyBBX: baud=19200 serial_line=8n1 (this device)")
+
+ print_header("SUMMARY")
+ status = "DEGRADED" if mostly_echo and not any_cmd else "OK"
+ print(f"Overall: {status}")
+ print(f"Device: {dev}")
+ print(f"Best host profile: {best.name} (score={best.host_score})")
+ print(f"CTS on best: {int(best.cts)} echo_only: {int(best.echo_only)} cmd: {int(best.has_cmd_prompt)}")
+
+ return 1 if status == "DEGRADED" else 0
+
+
+if __name__ == "__main__":
+ sys.exit(main())
diff --git a/stacks/tncs/tnc2c-health.sh b/stacks/tncs/tnc2c-health.sh
new file mode 100755
index 0000000..9d9ee10
--- /dev/null
+++ b/stacks/tncs/tnc2c-health.sh
@@ -0,0 +1,9 @@
+#!/bin/bash
+# Full TNC2C health check — serial, host mode, KISS, optional TX.
+# Usage:
+# ./tnc2c-health.sh # safe, no transmit
+# ./tnc2c-health.sh --tx # includes KISS send (PTT!)
+
+set -euo pipefail
+ROOT="$(cd "$(dirname "$0")" && pwd)"
+exec python3 "$ROOT/tnc2c-health.py" "$@"
diff --git a/stacks/tncs/tnc2c-host-reset.py b/stacks/tncs/tnc2c-host-reset.py
new file mode 100755
index 0000000..340642f
--- /dev/null
+++ b/stacks/tncs/tnc2c-host-reset.py
@@ -0,0 +1,229 @@
+#!/usr/bin/env python3
+"""
+tnc2c-host-reset - recover TNC-2 class terminal/host mode over serial.
+
+Uses full software recovery ladder (KISS-return, JHOST0, RESTART) — no power
+cycle required unless DTR was low at cold boot.
+
+Usage:
+ ./tnc2c-host-reset.sh
+ ./tnc2c-host-reset.sh /dev/ttyS4
+ ./tnc2c-host-reset.sh --kiss # recover + enter KISS
+ ./tnc2c-host-reset.sh --power-hint # remind to cycle TNC power first
+ ./tnc2c-host-reset.sh --no-kiss # skip KISS reset frame (step 2 only)
+"""
+
+from __future__ import annotations
+
+import argparse
+import importlib.util
+import os
+import sys
+import time
+
+BAUD_DEFAULT = 19200
+LINE_DEFAULT = "8n1"
+
+FIRMWARE_MARKERS = (
+ b"TheFirmware",
+ b"NORD",
+ b"Version 2.7",
+ b"Checksum",
+ b"Copyright",
+ b"DAMA",
+ b"SMACK",
+)
+
+
+def load_recovery():
+ root = os.path.dirname(os.path.abspath(__file__))
+ path = os.path.join(root, "tnc_serial_recovery.py")
+ spec = importlib.util.spec_from_file_location("tnc_serial_recovery", path)
+ if spec is None or spec.loader is None:
+ raise RuntimeError(f"cannot load {path}")
+ mod = importlib.util.module_from_spec(spec)
+ spec.loader.exec_module(mod)
+ return mod
+
+
+def load_env(path: str) -> dict[str, str]:
+ out: dict[str, str] = {}
+ if not os.path.isfile(path):
+ return out
+ with open(path, encoding="utf-8") as f:
+ for line in f:
+ line = line.strip()
+ if not line or line.startswith("#") or "=" not in line:
+ continue
+ k, v = line.split("=", 1)
+ out[k.strip()] = v.strip()
+ return out
+
+
+def parse_line(line: str) -> tuple[int, int]:
+ import termios
+
+ line = line.lower()
+ if line == "7e1":
+ return termios.CS7, termios.PARENB
+ if line == "8n1":
+ return termios.CS8, 0
+ raise ValueError(f"unsupported serial line: {line}")
+
+
+def parse_baud(baud: int) -> int:
+ import termios
+
+ table = {
+ 1200: termios.B1200,
+ 2400: termios.B2400,
+ 4800: termios.B4800,
+ 9600: termios.B9600,
+ 19200: termios.B19200,
+ }
+ if baud not in table:
+ raise ValueError(f"unsupported baud: {baud}")
+ return table[baud]
+
+
+def open_serial(dev: str, baud: int, line: str) -> int:
+ import fcntl
+ import struct
+ import termios
+
+ speed = parse_baud(baud)
+ databits, parity = parse_line(line)
+ fd = os.open(dev, os.O_RDWR | os.O_NOCTTY | os.O_NONBLOCK)
+ t = termios.tcgetattr(fd)
+ t[0] = t[1] = 0
+ t[2] = termios.CLOCAL | termios.CREAD | databits | parity
+ t[3] = t[4] = t[5] = speed
+ t[6][termios.VMIN] = 0
+ t[6][termios.VTIME] = 5
+ termios.tcsetattr(fd, termios.TCSANOW, t)
+ termios.tcflush(fd, termios.TCIOFLUSH)
+ flags = struct.unpack("I", fcntl.ioctl(fd, 0x5415, struct.pack("I", 0)))[0]
+ flags |= 0x004 | 0x002
+ fcntl.ioctl(fd, 0x5416, struct.pack("I", flags))
+ return fd
+
+
+def read_for(fd: int, seconds: float) -> bytes:
+ end = time.time() + seconds
+ chunks: list[bytes] = []
+ while time.time() < end:
+ try:
+ b = os.read(fd, 4096)
+ if b:
+ chunks.append(b)
+ except BlockingIOError:
+ time.sleep(0.02)
+ return b"".join(chunks)
+
+
+def write_flush(fd: int, data: bytes) -> None:
+ import termios
+
+ os.write(fd, data)
+ termios.tcdrain(fd)
+
+
+POWER_HINT = """\
+Note: If recovery fails (echo only), disconnect TNC power (10-15 s), reconnect
+while DTR+RTS stay high — run tnc2c-boot-wait.sh or this script immediately.
+"""
+
+
+def main() -> int:
+ parser = argparse.ArgumentParser(
+ description="TNC host-mode recovery with DTR+RTS (TheFirmware ladder)"
+ )
+ parser.add_argument("device", nargs="?", default=None)
+ parser.add_argument("--baud", type=int, default=None)
+ parser.add_argument("--line", default=None, choices=("8n1", "7e1"))
+ parser.add_argument("--kiss", action="store_true", help="enter KISS after recover")
+ parser.add_argument(
+ "--kiss-entry",
+ default="kiss_on",
+ choices=("kiss_on", "auto"),
+ help="KISS entry method with --kiss",
+ )
+ parser.add_argument("--power-hint", action="store_true")
+ parser.add_argument(
+ "--no-kiss",
+ action="store_true",
+ help="skip KISS return frame in recovery ladder",
+ )
+ args = parser.parse_args()
+
+ root = os.path.dirname(os.path.abspath(__file__))
+ env = load_env(os.path.join(root, "tnc2c-serial.env"))
+ dev = (
+ args.device
+ or env.get("TNC_DEV")
+ or env.get("TNC2C_DEV")
+ or env.get("PKTNC2_DEV")
+ or "/dev/ttyS4"
+ )
+ baud = args.baud or int(
+ env.get("TNC_BAUD") or env.get("TNC2C_BAUD") or env.get("PKTNC2_BAUD") or BAUD_DEFAULT
+ )
+ line = args.line or env.get("TNC_LINE") or env.get("TNC2C_LINE") or LINE_DEFAULT
+
+ if args.power_hint:
+ print(POWER_HINT)
+
+ if not os.path.exists(dev):
+ print(f"FAIL: {dev} does not exist", file=sys.stderr)
+ return 2
+ if not os.access(dev, os.R_OK | os.W_OK):
+ print(f"FAIL: no access to {dev}", file=sys.stderr)
+ return 2
+
+ recovery = load_recovery()
+
+ print(f"TNC host-reset @ {dev} {baud} {line.upper()} (DTR+RTS high)")
+ try:
+ fd = open_serial(dev, baud, line)
+ except OSError as e:
+ print(f"FAIL: {dev}: {e}", file=sys.stderr)
+ return 2
+
+ def wf(data: bytes) -> None:
+ write_flush(fd, data)
+
+ def rf(seconds: float) -> bytes:
+ return read_for(fd, seconds)
+
+ def log(msg: str) -> None:
+ print(f" {msg}")
+
+ ok, received = recovery.recover_terminal(
+ wf, rf, skip_kiss_frame=args.no_kiss, log=log
+ )
+
+ if ok and args.kiss:
+ recovery.enter_kiss(wf, rf, args.kiss_entry)
+ print(" KISS entry sent")
+
+ os.close(fd)
+
+ print()
+ if ok:
+ label = "HOST+KISS" if args.kiss else "HOST"
+ print(f"OK: {label} - terminal mode active (software recovery)")
+ if received:
+ text = received.decode("ascii", errors="replace")
+ if "NORD" in text or "TheFirmware" in text:
+ for line_out in text.splitlines():
+ if any(m.decode() in line_out for m in FIRMWARE_MARKERS[:5]):
+ print(f" | {line_out.strip()}")
+ return 0
+
+ print("FAIL: software recovery exhausted")
+ print(" -> Power-cycle with DTR high: ./tnc2c-boot-wait.sh")
+ return 1
+
+
+if __name__ == "__main__":
+ sys.exit(main())
diff --git a/stacks/tncs/tnc2c-host-reset.sh b/stacks/tncs/tnc2c-host-reset.sh
new file mode 100755
index 0000000..d0d1499
--- /dev/null
+++ b/stacks/tncs/tnc2c-host-reset.sh
@@ -0,0 +1,21 @@
+#!/bin/bash
+# TNC2C host-mode recovery — DTR+RTS high, software ladder (no power cycle usually).
+#
+# Usage:
+# ./tnc2c-host-reset.sh
+# ./tnc2c-host-reset.sh --kiss # recover + enter KISS
+# ./tnc2c-host-reset.sh --power-hint # if recovery fails: cycle power + boot-wait
+# ./tnc2c-host-reset.sh --no-kiss # skip KISS return frame in ladder
+
+set -euo pipefail
+ROOT="$(cd "$(dirname "$0")" && pwd)"
+# shellcheck source=load-serial-env.sh
+source "${ROOT}/load-serial-env.sh"
+load_serial_env tnc2c "${ROOT}" || true
+apply_tnc2c_env
+if [[ -n "${SERIAL_ENV_FILE:-}" ]]; then
+ echo "Using serial env: ${SERIAL_ENV_FILE}" >&2
+else
+ echo "WARN: no tnc2c-serial.env found — using ${TNC_DEV} ${TNC_BAUD} ${TNC_LINE}" >&2
+fi
+exec python3 "${ROOT}/tnc2c-host-reset.py" "${TNC_DEV}" --baud "${TNC_BAUD}" --line "${TNC_LINE}" "$@"
diff --git a/stacks/tncs/tnc2c-hybbx-beacon-once.sh b/stacks/tncs/tnc2c-hybbx-beacon-once.sh
new file mode 100755
index 0000000..7249191
--- /dev/null
+++ b/stacks/tncs/tnc2c-hybbx-beacon-once.sh
@@ -0,0 +1,21 @@
+#!/bin/bash
+# One-shot AX.25 beacon like INI [broadcast] - stop HyBBX briefly, then restart.
+# Auto-beacon (300s) runs from new HyBBX start.
+set -euo pipefail
+ROOT="$(cd "$(dirname "$0")" && pwd)"
+MSG="${1:-Broadcast: CB-0 online}"
+SRC="${2:-CB-0}"
+DST="${3:-*}"
+
+echo "Manual beacon: ${SRC} -> ${DST}: ${MSG}"
+if pgrep -x hybbx >/dev/null; then
+ echo "Stopping HyBBX ..."
+ pkill -x hybbx || sudo -u hybbx pkill -x hybbx
+ sleep 2
+fi
+"$ROOT/tnc2c-send-test.sh" "$MSG" "$SRC" "$DST"
+echo "Starting HyBBX ... (auto-beacon in ~300s)"
+sudo -u hybbx /etc/hybbx/hybbx-start &
+sleep 3
+pgrep -a hybbx || echo "WARN: hybbx not started - try boot-wait + hybbx-start"
+echo "Log: tail -f /etc/hybbx/logs/*.log - expect: [broadcast] ax25"
diff --git a/stacks/tncs/tnc2c-integration-test.py b/stacks/tncs/tnc2c-integration-test.py
new file mode 100644
index 0000000..930e0a6
--- /dev/null
+++ b/stacks/tncs/tnc2c-integration-test.py
@@ -0,0 +1,103 @@
+#!/usr/bin/env python3
+"""
+HyBBX-ready check - must use same open port as boot-wait (DTR must stay high).
+
+After boot-wait closes the port, DTR drops and the TNC often returns to echo mode.
+Use: ./tnc2c-boot-wait.sh (includes HyBBX verify by default)
+Or: ./tnc2c-integration-test.sh --boot (boot-wait + verify in one process)
+"""
+
+from __future__ import annotations
+
+import argparse
+import importlib.util
+import os
+import subprocess
+import sys
+import time
+
+ROOT = os.path.dirname(os.path.abspath(__file__))
+
+
+def load_module(name: str, filename: str):
+ path = os.path.join(ROOT, filename)
+ spec = importlib.util.spec_from_file_location(name, path)
+ mod = importlib.util.module_from_spec(spec)
+ spec.loader.exec_module(mod)
+ return mod
+
+
+def preflight() -> int | None:
+ if os.system("pgrep -x hybbx >/dev/null 2>&1") == 0:
+ print("FAIL: hybbx is running")
+ return 2
+ return None
+
+
+def run_quick_check(dev: str) -> int:
+ hr = load_module("host_reset", "tnc2c-host-reset.py")
+
+ print(f"TNC2C integration-test @ {dev}")
+ print("(Port was closed - DTR was low; echo is likely)\n")
+
+ try:
+ fd = hr.open_serial(dev)
+ except OSError as e:
+ print(f"FAIL: {e}")
+ return 2
+
+ received = b""
+ received += hr.read_for(fd, 1.0)
+ hr.write_flush(fd, b"kiss off\r")
+ time.sleep(0.6)
+ received += hr.read_for(fd, 1.0)
+ hr.write_flush(fd, b"INFO\r")
+ time.sleep(0.3)
+ received += hr.read_for(fd, 5.0)
+ os.close(fd)
+
+ if hr.has_banner(received):
+ print("OK: HOST - HyBBX may start")
+ print(received.decode("ascii", errors="replace")[:500])
+ return 0
+
+ print("FAIL: ECHO or no banner")
+ print(" -> Closing the port after boot-wait drops DTR.")
+ print(" -> Use ONE command:")
+ print(" ./tnc2c-boot-wait.sh")
+ print(" (power off/on while script runs - checks HyBBX-ready before close)")
+ print(" or:")
+ print(" ./tnc2c-integration-test.sh --boot")
+ return 1
+
+
+def run_with_boot(dev: str) -> int:
+ print("Starting boot-wait + HyBBX verify in one process ...\n")
+ script = os.path.join(ROOT, "tnc2c-boot-wait.sh")
+ return subprocess.call([script, dev])
+
+
+def main() -> int:
+ parser = argparse.ArgumentParser(description="TNC2C HyBBX-ready check")
+ parser.add_argument("device", nargs="?", default=None)
+ parser.add_argument(
+ "--boot",
+ action="store_true",
+ help="run boot-wait (power-cycle TNC while script runs)",
+ )
+ args = parser.parse_args()
+
+ fail = preflight()
+ if fail is not None:
+ return fail
+
+ hr = load_module("host_reset", "tnc2c-host-reset.py")
+ dev = args.device or hr.load_env(os.path.join(ROOT, "tnc2c-serial.env")) or "/dev/ttyS4"
+
+ if args.boot:
+ return run_with_boot(dev)
+ return run_quick_check(dev)
+
+
+if __name__ == "__main__":
+ sys.exit(main())
diff --git a/stacks/tncs/tnc2c-integration-test.sh b/stacks/tncs/tnc2c-integration-test.sh
new file mode 100755
index 0000000..e0523e4
--- /dev/null
+++ b/stacks/tncs/tnc2c-integration-test.sh
@@ -0,0 +1,4 @@
+#!/bin/bash
+set -euo pipefail
+ROOT="$(cd "$(dirname "$0")" && pwd)"
+exec python3 "$ROOT/tnc2c-integration-test.py" "$@"
diff --git a/stacks/tncs/tnc2c-listen.sh b/stacks/tncs/tnc2c-listen.sh
new file mode 100755
index 0000000..7020e17
--- /dev/null
+++ b/stacks/tncs/tnc2c-listen.sh
@@ -0,0 +1,56 @@
+#!/bin/bash
+# Passive listen on TNC2C - NO commands that can key PTT (no kiss on, no MONitor).
+# Usage: ./tnc2c-listen.sh [seconds] [device]
+
+set -euo pipefail
+
+ROOT="$(cd "$(dirname "$0")" && pwd)"
+# shellcheck source=/dev/null
+[[ -f "$ROOT/tnc2c-serial.env" ]] && source "$ROOT/tnc2c-serial.env"
+
+DEV="${2:-${TNC2C_DEV:-/dev/ttyS4}}"
+SECS="${1:-5}"
+BAUD="${TNC2C_BAUD:-19200}"
+LINE="${TNC2C_LINE:-7e1}"
+
+case "$LINE" in
+7e1|7E1) STTY="cs7 parenb -cstopb" ;;
+8n1|8N1) STTY="cs8 -parenb -cstopb" ;;
+*) echo "Unknown line: $LINE" >&2; exit 1 ;;
+esac
+
+if pgrep -x hybbx >/dev/null 2>&1; then
+ echo "WARN: hybbx is running - port may be in use" >&2
+fi
+
+echo "Passive listen: $DEV @ ${BAUD} ${LINE} for ${SECS}s (no TX commands)"
+stty -F "$DEV" "$BAUD" $STTY raw -echo min 0 time 5
+
+python3 <<PY
+import fcntl, os, struct, time, termios
+
+fd = os.open("$DEV", os.O_RDWR | os.O_NOCTTY | os.O_NONBLOCK)
+TIOCM_RTS, TIOCM_DTR = 0x004, 0x002
+TIOCMGET, TIOCMSET = 0x5415, 0x5416
+flags = struct.unpack("I", fcntl.ioctl(fd, TIOCMGET, struct.pack("I", 0)))[0]
+flags |= TIOCM_RTS | TIOCM_DTR
+fcntl.ioctl(fd, TIOCMSET, struct.pack("I", flags))
+termios.tcflush(fd, termios.TCIOFLUSH)
+
+end = time.time() + float("$SECS")
+chunks = []
+while time.time() < end:
+ try:
+ b = os.read(fd, 4096)
+ if b:
+ chunks.append(b)
+ print(f"+{len(b)} bytes")
+ except BlockingIOError:
+ time.sleep(0.05)
+
+data = b"".join(chunks)
+os.close(fd)
+print(f"Total: {len(data)} bytes")
+if data:
+ print(data[:256].hex(" "))
+PY
diff --git a/stacks/tncs/tnc2c-probe.c b/stacks/tncs/tnc2c-probe.c
new file mode 100644
index 0000000..8770409
--- /dev/null
+++ b/stacks/tncs/tnc2c-probe.c
@@ -0,0 +1,535 @@
+/*
+ * tnc2c-probe — scan /dev/ttyS* and /dev/ttyUSB* for a Landolt TNC2C.
+ *
+ * TNC2C host port (per manual): 7-bit, even parity, 1 stop bit.
+ * Terminal speed set by TERM jumper (this unit: 19200; manual default often 9600).
+ * RTS/CTS handshake; DTR asserted to wake some adapters.
+ *
+ * Build: make
+ * Run: ./tnc2c-probe
+ * ./tnc2c-probe /dev/ttyS4
+ */
+
+#if defined(__linux__) || defined(__GLIBC__)
+#define _DEFAULT_SOURCE 1
+#endif
+
+#include <dirent.h>
+#include <errno.h>
+#include <fcntl.h>
+#include <glob.h>
+#include <stdio.h>
+#include <stdlib.h>
+#include <string.h>
+#include <sys/ioctl.h>
+#include <termios.h>
+#include <time.h>
+#include <unistd.h>
+
+#define PROBE_READ_MS 900
+#define PROBE_GAP_MS 120
+#define RX_BUF_SIZE 4096
+#define MAX_DEVICES 64
+
+typedef struct serial_profile {
+ const char *label;
+ speed_t speed;
+ unsigned int baud_num;
+ tcflag_t databits;
+ tcflag_t parity;
+} serial_profile_t;
+
+typedef struct probe_result {
+ char device[64];
+ char profile[32];
+ char command[32];
+ int got_data;
+ size_t len;
+ char text[RX_BUF_SIZE];
+ unsigned char raw[RX_BUF_SIZE];
+} probe_result_t;
+
+static const serial_profile_t PROFILES[] = {
+ { "19200-7E1", B19200, 19200, CS7, PARENB },
+ { "19200-8N1", B19200, 19200, CS8, 0 },
+ { "9600-7E1", B9600, 9600, CS7, PARENB },
+ { "9600-8N1", B9600, 9600, CS8, 0 },
+ { "4800-7E1", B4800, 4800, CS7, PARENB },
+ { "2400-7E1", B2400, 2400, CS7, PARENB },
+ { "1200-7E1", B1200, 1200, CS7, PARENB },
+};
+
+static const char *PROBE_COMMANDS[] = {
+ "\r",
+ "INFO\r",
+ "HELP\r",
+ "?\r",
+};
+
+static int verbose = 0;
+
+static int sleep_ms(unsigned ms)
+{
+ struct timespec ts;
+
+ ts.tv_sec = (time_t)(ms / 1000u);
+ ts.tv_nsec = (long)(ms % 1000u) * 1000000L;
+ return nanosleep(&ts, NULL);
+}
+
+static int set_modem_lines(int fd, int rts_on, int dtr_on)
+{
+#ifdef TIOCMGET
+ int flags;
+
+ if (ioctl(fd, TIOCMGET, &flags) != 0) {
+ return -1;
+ }
+ if (rts_on) {
+ flags |= TIOCM_RTS;
+ } else {
+ flags &= ~TIOCM_RTS;
+ }
+ if (dtr_on) {
+ flags |= TIOCM_DTR;
+ } else {
+ flags &= ~TIOCM_DTR;
+ }
+ return ioctl(fd, TIOCMSET, &flags);
+#else
+ (void)fd;
+ (void)rts_on;
+ (void)dtr_on;
+ return 0;
+#endif
+}
+
+static int read_modem_lines(int fd, int *rts, int *cts, int *dtr, int *dsr)
+{
+#ifdef TIOCMGET
+ int flags;
+
+ if (ioctl(fd, TIOCMGET, &flags) != 0) {
+ return -1;
+ }
+ if (rts != NULL) {
+ *rts = (flags & TIOCM_RTS) != 0;
+ }
+ if (cts != NULL) {
+ *cts = (flags & TIOCM_CTS) != 0;
+ }
+ if (dtr != NULL) {
+ *dtr = (flags & TIOCM_DTR) != 0;
+ }
+ if (dsr != NULL) {
+ *dsr = (flags & TIOCM_DSR) != 0;
+ }
+ return 0;
+#else
+ (void)fd;
+ if (rts) {
+ *rts = -1;
+ }
+ if (cts) {
+ *cts = -1;
+ }
+ if (dtr) {
+ *dtr = -1;
+ }
+ if (dsr) {
+ *dsr = -1;
+ }
+ return 0;
+#endif
+}
+
+static int configure_port(int fd, const serial_profile_t *profile)
+{
+ struct termios tty;
+
+ if (tcgetattr(fd, &tty) != 0) {
+ return -1;
+ }
+
+ cfmakeraw(&tty);
+ tty.c_cflag |= (CLOCAL | CREAD);
+ tty.c_cflag &= ~(CSIZE | PARENB | PARODD | CSTOPB);
+ tty.c_cflag |= profile->databits;
+ if (profile->parity != 0) {
+ tty.c_cflag |= profile->parity;
+ }
+ tty.c_cc[VMIN] = 0;
+ tty.c_cc[VTIME] = 1;
+
+#if defined(__linux__) || defined(__GLIBC__)
+ if (cfsetspeed(&tty, profile->speed) != 0) {
+ cfsetispeed(&tty, profile->speed);
+ cfsetospeed(&tty, profile->speed);
+ }
+#else
+ cfsetispeed(&tty, profile->speed);
+ cfsetospeed(&tty, profile->speed);
+#endif
+
+ if (tcsetattr(fd, TCSANOW, &tty) != 0) {
+ return -1;
+ }
+
+ tcflush(fd, TCIOFLUSH);
+ return 0;
+}
+
+static int is_printable_ascii(unsigned char c)
+{
+ return c == '\r' || c == '\n' || c == '\t' || (c >= 0x20 && c <= 0x7e);
+}
+
+static void sanitize_text(const unsigned char *raw, size_t len, char *out, size_t out_cap)
+{
+ size_t i;
+ size_t o = 0;
+
+ if (out_cap == 0) {
+ return;
+ }
+
+ for (i = 0; i < len && o + 1 < out_cap; i++) {
+ unsigned char c = raw[i];
+
+ if (c == '\r') {
+ if (o + 2 < out_cap) {
+ out[o++] = '\\';
+ out[o++] = 'r';
+ }
+ } else if (c == '\n') {
+ if (o + 2 < out_cap) {
+ out[o++] = '\\';
+ out[o++] = 'n';
+ }
+ } else if (is_printable_ascii(c)) {
+ out[o++] = (char)c;
+ } else {
+ int n = snprintf(out + o, out_cap - o, "\\x%02x", c);
+ if (n <= 0 || (size_t)n >= out_cap - o) {
+ break;
+ }
+ o += (size_t)n;
+ }
+ }
+ out[o] = '\0';
+}
+
+static int looks_like_tnc(const unsigned char *raw, size_t len)
+{
+ static const char *needles[] = {
+ "cmd:", "CMD:", "TNC", "tnc", "INFO", "help", "HELP",
+ "***", "connected", "DISCONNECTED", "kiss", "KISS",
+ "MYCALL", "WA8DED", "NOCALL", "?"
+ };
+ char buf[RX_BUF_SIZE];
+ size_t i;
+
+ if (len == 0) {
+ return 0;
+ }
+
+ sanitize_text(raw, len, buf, sizeof(buf));
+ for (i = 0; i < sizeof(needles) / sizeof(needles[0]); i++) {
+ if (strstr(buf, needles[i]) != NULL) {
+ return 1;
+ }
+ }
+ return len >= 3;
+}
+
+static size_t read_for_ms(int fd, unsigned char *buf, size_t cap, unsigned ms)
+{
+ size_t total = 0;
+ unsigned elapsed = 0;
+
+ while (elapsed < ms && total < cap) {
+ ssize_t n = read(fd, buf + total, cap - total);
+
+ if (n > 0) {
+ total += (size_t)n;
+ elapsed = 0;
+ continue;
+ }
+ if (n < 0 && errno != EAGAIN && errno != EWOULDBLOCK) {
+ break;
+ }
+ sleep_ms(40);
+ elapsed += 40;
+ }
+
+ return total;
+}
+
+static int probe_once(const char *device, const serial_profile_t *profile,
+ const char *command, probe_result_t *result)
+{
+ int fd;
+ ssize_t wlen;
+
+ memset(result, 0, sizeof(*result));
+ snprintf(result->device, sizeof(result->device), "%s", device);
+ snprintf(result->profile, sizeof(result->profile), "%s", profile->label);
+ snprintf(result->command, sizeof(result->command), "%s",
+ command[0] == '\r' ? "<CR>" : command);
+
+ fd = open(device, O_RDWR | O_NOCTTY | O_NONBLOCK);
+ if (fd < 0) {
+ return -1;
+ }
+
+ if (configure_port(fd, profile) != 0) {
+ close(fd);
+ return -1;
+ }
+
+ (void)set_modem_lines(fd, 1, 1);
+ sleep_ms(80);
+ (void)read_for_ms(fd, result->raw, sizeof(result->raw), 150);
+
+ wlen = write(fd, command, strlen(command));
+ if (wlen < 0) {
+ close(fd);
+ return -1;
+ }
+
+ sleep_ms(PROBE_GAP_MS);
+ result->len = read_for_ms(fd, result->raw, sizeof(result->raw), PROBE_READ_MS);
+ close(fd);
+
+ result->got_data = result->len > 0;
+ sanitize_text(result->raw, result->len, result->text, sizeof(result->text));
+ return 0;
+}
+
+static int device_accessible(const char *path)
+{
+ return access(path, R_OK | W_OK) == 0;
+}
+
+static int collect_devices(char devices[][64], int max_devices)
+{
+ glob_t g;
+ size_t i;
+ int count = 0;
+ static const char *patterns[] = {
+ "/dev/ttyUSB[0-9]*",
+ "/dev/ttyS[0-9]*",
+ NULL
+ };
+ int p;
+
+ for (p = 0; patterns[p] != NULL; p++) {
+ if (glob(patterns[p], 0, NULL, &g) != 0) {
+ continue;
+ }
+ for (i = 0; i < g.gl_pathc && count < max_devices; i++) {
+ const char *path = g.gl_pathv[i];
+ int j;
+ int dup = 0;
+
+ for (j = 0; j < count; j++) {
+ if (strcmp(devices[j], path) == 0) {
+ dup = 1;
+ break;
+ }
+ }
+ if (!dup) {
+ snprintf(devices[count], 64, "%s", path);
+ count++;
+ }
+ }
+ globfree(&g);
+ }
+
+ return count;
+}
+
+static void print_device_header(const char *device)
+{
+ int fd;
+ int rts = -1, cts = -1, dtr = -1, dsr = -1;
+
+ printf("\n============================================================\n");
+ printf("Device: %s\n", device);
+ if (!device_accessible(device)) {
+ printf("Access: NO (need dialout group or root)\n");
+ return;
+ }
+ printf("Access: yes\n");
+
+ fd = open(device, O_RDWR | O_NOCTTY | O_NONBLOCK);
+ if (fd >= 0) {
+ if (read_modem_lines(fd, &rts, &cts, &dtr, &dsr) == 0) {
+ printf("Lines (before probe): RTS=%d CTS=%d DTR=%d DSR=%d\n",
+ rts, cts, dtr, dsr);
+ }
+ close(fd);
+ }
+}
+
+static void listen_quiet(const char *device, const serial_profile_t *profile)
+{
+ int fd;
+ unsigned char buf[512];
+ size_t n;
+
+ fd = open(device, O_RDWR | O_NOCTTY | O_NONBLOCK);
+ if (fd < 0) {
+ return;
+ }
+ if (configure_port(fd, profile) != 0) {
+ close(fd);
+ return;
+ }
+ (void)set_modem_lines(fd, 1, 1);
+ n = read_for_ms(fd, buf, sizeof(buf), 1200);
+ close(fd);
+
+ if (n > 0) {
+ char text[512];
+ sanitize_text(buf, n, text, sizeof(text));
+ printf(" Spontaneous traffic @ %s: %s\n", profile->label, text);
+ } else if (verbose) {
+ printf(" Spontaneous traffic @ %s: (none)\n", profile->label);
+ }
+}
+
+static int run_device_scan(const char *device)
+{
+ size_t pi;
+ size_t ci;
+ int hits = 0;
+ int best_score = 0;
+ char best_profile[32] = "";
+ char best_text[512] = "";
+
+ print_device_header(device);
+ if (!device_accessible(device)) {
+ return 0;
+ }
+
+ listen_quiet(device, &PROFILES[0]);
+ listen_quiet(device, &PROFILES[1]);
+
+ for (pi = 0; pi < sizeof(PROFILES) / sizeof(PROFILES[0]); pi++) {
+ for (ci = 0; ci < sizeof(PROBE_COMMANDS) / sizeof(PROBE_COMMANDS[0]); ci++) {
+ probe_result_t result;
+ int score;
+
+ if (probe_once(device, &PROFILES[pi], PROBE_COMMANDS[ci], &result) != 0) {
+ continue;
+ }
+ if (!result.got_data) {
+ if (verbose) {
+ printf(" %s + %s -> (no data)\n",
+ result.profile, result.command);
+ }
+ continue;
+ }
+
+ score = looks_like_tnc(result.raw, result.len) ? 2 : 1;
+ hits++;
+ printf(" HIT %s + %s (%zu bytes, score=%d)\n",
+ result.profile, result.command, result.len, score);
+ printf(" %s\n", result.text);
+
+ if (score > best_score) {
+ best_score = score;
+ snprintf(best_profile, sizeof(best_profile), "%s", result.profile);
+ strncpy(best_text, result.text, sizeof(best_text) - 1);
+ best_text[sizeof(best_text) - 1] = '\0';
+ }
+ }
+ }
+
+ if (hits == 0) {
+ printf(" Result: no response on any profile/command.\n");
+ printf(" Hint: check power, cable (DTE<->DCE modem cable), baud jumper.\n");
+ } else {
+ printf(" Result: %d response(s); best profile=%s (score=%d)\n",
+ hits, best_profile[0] ? best_profile : "?", best_score);
+ if (best_text[0] != '\0') {
+ printf(" Best: %s\n", best_text);
+ }
+ }
+
+ return hits;
+}
+
+static void usage(const char *argv0)
+{
+ fprintf(stderr,
+ "Usage: %s [-v] [device ...]\n"
+ "\n"
+ "Scan serial ports for a Landolt TNC2C.\n"
+ "Without device arguments, probes all /dev/ttyUSB* and /dev/ttyS*.\n"
+ "\n"
+ "Options:\n"
+ " -v verbose (show silent probes too)\n"
+ " -h this help\n",
+ argv0);
+}
+
+int main(int argc, char **argv)
+{
+ char devices[MAX_DEVICES][64];
+ int device_count = 0;
+ int argi;
+ int total_hits = 0;
+ int scanned = 0;
+
+ for (argi = 1; argi < argc; argi++) {
+ if (strcmp(argv[argi], "-h") == 0 || strcmp(argv[argi], "--help") == 0) {
+ usage(argv[0]);
+ return 0;
+ }
+ if (strcmp(argv[argi], "-v") == 0 || strcmp(argv[argi], "--verbose") == 0) {
+ verbose = 1;
+ continue;
+ }
+ if (device_count < MAX_DEVICES) {
+ snprintf(devices[device_count], 64, "%s", argv[argi]);
+ device_count++;
+ }
+ }
+
+ if (device_count == 0) {
+ device_count = collect_devices(devices, MAX_DEVICES);
+ }
+
+ printf("TNC2C serial probe\n");
+ printf("Profiles: 7E1 (manual) and 8N1 (HyBBX default) at common baud rates.\n");
+ printf("Found %d serial device(s).\n", device_count);
+
+ if (device_count == 0) {
+ fprintf(stderr, "No /dev/ttyUSB* or /dev/ttyS* devices found.\n");
+ return 1;
+ }
+
+ for (argi = 0; argi < device_count; argi++) {
+ total_hits += run_device_scan(devices[argi]);
+ scanned++;
+ }
+
+ printf("\n============================================================\n");
+ printf("Summary: scanned %d port(s), %d response(s) total.\n",
+ scanned, total_hits);
+
+ if (total_hits == 0) {
+ printf("\nCable note (TNC2C manual):\n");
+ printf(" - TNC2C DB25 is wired like a modem (DCE).\n");
+ printf(" - PC/USB-serial is DTE -> use a straight modem cable,\n");
+ printf(" not a null-modem crossover cable.\n");
+ printf(" - Some PCs need RTS-CTS bridged on the host side (pins 4+5).\n");
+ printf(" - Host settings: 7-bit, even parity (7E1), baud per TERM jumper.\n");
+ printf(" - LEDs after power-on: Power on; Status+Connected off after ~3 s.\n");
+ return 2;
+ }
+
+ return 0;
+}
diff --git a/stacks/tncs/tnc2c-rf-tx-test.sh b/stacks/tncs/tnc2c-rf-tx-test.sh
new file mode 100755
index 0000000..fc23eff
--- /dev/null
+++ b/stacks/tncs/tnc2c-rf-tx-test.sh
@@ -0,0 +1,28 @@
+#!/bin/bash
+# Send N AX.25 UI frames via KISS (HyBBX must be stopped).
+# Usage: ./tnc2c-rf-tx-test.sh [count] [message-prefix] [src] [dst]
+# Example (root): pkill hybbx; ./tnc2c-rf-tx-test.sh 3 "TX test" CB-0 QST
+
+set -euo pipefail
+
+COUNT="${1:-3}"
+PREFIX="${2:-TX test}"
+SRC="${3:-CB-0}"
+DST="${4:-QST}"
+GAP="${TNC2C_TX_GAP:-8}"
+ROOT="$(cd "$(dirname "$0")" && pwd)"
+
+if pgrep -x hybbx >/dev/null 2>&1; then
+ echo "ERROR: hybbx is running - as root: pkill hybbx" >&2
+ exit 1
+fi
+
+echo "RF TX: ${COUNT}x ${SRC} -> ${DST}, ${GAP}s gap (SQ closed, CD off)"
+for i in $(seq 1 "$COUNT"); do
+ echo "=== SEND ${i}/${COUNT} $(date +%H:%M:%S) ==="
+ "$ROOT/tnc2c-send-test.sh" "${PREFIX} ${i}/${COUNT}" "$SRC" "$DST"
+ if [[ "$i" -lt "$COUNT" ]]; then
+ sleep "$GAP"
+ fi
+done
+echo "Done - check PTT LED / modem tone on radio."
diff --git a/stacks/tncs/tnc2c-send-test.sh b/stacks/tncs/tnc2c-send-test.sh
new file mode 100755
index 0000000..4ac1b52
--- /dev/null
+++ b/stacks/tncs/tnc2c-send-test.sh
@@ -0,0 +1,100 @@
+#!/bin/bash
+# Send one AX.25 UI frame via KISS on TNC2C - keys PTT / transmits on radio.
+# Requires: antenna or dummyload, VOX off, HyBBX stopped.
+# Usage: ./tnc2c-send-test.sh [message] [src] [dst]
+
+set -euo pipefail
+
+MSG="${1:-HI}"
+SRC="${2:-TEST-0}"
+DST="${3:-CQ}"
+
+ROOT="$(cd "$(dirname "$0")" && pwd)"
+# shellcheck source=/dev/null
+[[ -f "$ROOT/tnc2c-serial.env" ]] && source "$ROOT/tnc2c-serial.env"
+
+if pgrep -x hybbx >/dev/null 2>&1; then
+ echo "ERROR: hybbx is running - stop it first" >&2
+ exit 1
+fi
+
+export TNC2C_DEV TNC2C_LINE
+python3 - "$MSG" "$SRC" "$DST" <<'PY'
+import fcntl, os, struct, sys, time, termios
+
+msg, src, dst = sys.argv[1:4]
+dev = os.environ.get("TNC2C_DEV", "/dev/ttyS4")
+line = os.environ.get("TNC2C_LINE", "8n1").lower()
+seven_e1 = line.startswith("7e")
+
+def kiss_escape(data: bytes) -> bytes:
+ out = bytearray()
+ for b in data:
+ if b == 0xC0:
+ out.extend((0xDB, 0xDC))
+ elif b == 0xDB:
+ out.extend((0xDB, 0xDD))
+ else:
+ out.append(b)
+ return bytes(out)
+
+def ax25_addr(call: str, ssid: int, last: bool) -> bytes:
+ call = call.upper().ljust(6)[:6]
+ raw = bytes((ord(c) << 1) for c in call)
+ ssid_b = ((ssid & 0x0F) << 1) | (0x01 if last else 0x00)
+ return raw + bytes([ssid_b])
+
+def ax25_crc(data: bytes) -> int:
+ crc = 0xFFFF
+ for b in data:
+ crc ^= b
+ for _ in range(8):
+ crc = (crc >> 1) ^ 0x8408 if crc & 1 else crc >> 1
+ return crc ^ 0xFFFF
+
+def ax25_ui(src, dst, info: bytes) -> bytes:
+ body = ax25_addr(dst, 0, False) + ax25_addr(src, 0, True)
+ body += b"\x03\xF0" + info
+ crc = ax25_crc(body)
+ return body + bytes((crc & 0xFF, crc >> 8))
+
+def kiss_frame(port: int, payload: bytes) -> bytes:
+ return b"\xC0" + bytes([port & 0x0F]) + kiss_escape(payload) + b"\xC0"
+
+print(f"Serial: {dev} 19200 {line}")
+
+fd = os.open(dev, os.O_RDWR | os.O_NOCTTY)
+t = termios.tcgetattr(fd)
+t[0] = t[1] = 0
+t[2] = termios.CLOCAL | termios.CREAD | (termios.CS7 if seven_e1 else termios.CS8)
+if seven_e1:
+ t[2] |= termios.PARENB
+t[3] = t[4] = t[5] = termios.B19200
+t[6][termios.VMIN] = 0
+t[6][termios.VTIME] = 5
+termios.tcsetattr(fd, termios.TCSANOW, t)
+termios.tcflush(fd, termios.TCIOFLUSH)
+flags = struct.unpack("I", fcntl.ioctl(fd, 0x5415, struct.pack("I", 0)))[0] | 0x006
+fcntl.ioctl(fd, 0x5416, struct.pack("I", flags))
+
+for cmd in (b"kiss off\r", b"kiss on\r"):
+ os.write(fd, cmd)
+ termios.tcdrain(fd)
+ time.sleep(0.5)
+
+pkt = kiss_frame(0, ax25_ui(src, dst, msg.encode("ascii", errors="replace")))
+print(f"TX {src} -> {dst}: {msg!r} ({len(pkt)} bytes KISS)")
+os.write(fd, pkt)
+termios.tcdrain(fd)
+time.sleep(3.0)
+try:
+ rx = os.read(fd, 4096)
+ if rx:
+ print(f"RX {len(rx)} bytes: {rx[:80]!r}")
+except BlockingIOError:
+ pass
+os.write(fd, b"kiss off\r")
+termios.tcdrain(fd)
+os.close(fd)
+print("Check LED 2 (PTT) on TNC and S-meter on radio.")
+PY
diff --git a/stacks/tncs/tnc2c-serial.env b/stacks/tncs/tnc2c-serial.env
new file mode 100644
index 0000000..89f111a
--- /dev/null
+++ b/stacks/tncs/tnc2c-serial.env
@@ -0,0 +1,8 @@
+# TNC2C serial — edit for your station
+# Example: USB-serial adapter; onboard UART is often /dev/ttyS0
+TNC2C_DEV=/dev/ttyUSB0
+TNC2C_BAUD=19200
+TNC2C_LINE=8n1
+TNC2C_RADIO_BAUD=2400
+TNC2C_MODEM=tcm3105
+HYBBX_SERIAL_LINE=8n1
diff --git a/stacks/tncs/tnc2c-tx-rx-test.py b/stacks/tncs/tnc2c-tx-rx-test.py
new file mode 100644
index 0000000..76e639d
--- /dev/null
+++ b/stacks/tncs/tnc2c-tx-rx-test.py
@@ -0,0 +1,184 @@
+#!/usr/bin/env python3
+"""TX then RX diagnostic for TNC2C + CB - no HyBBX."""
+
+from __future__ import annotations
+
+import fcntl
+import os
+import struct
+import sys
+import time
+import termios
+
+DEV = "/dev/ttyS4"
+BAUD = termios.B19200
+
+FIRMWARE = (b"TheFirmware", b"NORD", b"Version 2.7", b"Checksum")
+
+
+def open_port() -> int:
+ fd = os.open(DEV, os.O_RDWR | os.O_NOCTTY | os.O_NONBLOCK)
+ t = termios.tcgetattr(fd)
+ t[0] = t[1] = 0
+ t[2] = termios.CLOCAL | termios.CREAD | termios.CS8
+ t[3] = t[4] = t[5] = BAUD
+ t[6][termios.VMIN] = 0
+ t[6][termios.VTIME] = 5
+ termios.tcsetattr(fd, termios.TCSANOW, t)
+ termios.tcflush(fd, termios.TCIOFLUSH)
+ flags = struct.unpack("I", fcntl.ioctl(fd, 0x5415, struct.pack("I", 0)))[0]
+ flags |= 0x006
+ fcntl.ioctl(fd, 0x5416, struct.pack("I", flags))
+ return fd
+
+
+def modem(fd: int) -> str:
+ f = struct.unpack("I", fcntl.ioctl(fd, 0x5415, struct.pack("I", 0)))[0]
+ return f"RTS={int(bool(f&4))} CTS={int(bool(f&8))} DSR={int(bool(f&0x100))} DCD={int(bool(f&0x80))}"
+
+
+def read_sec(fd: int, sec: float) -> bytes:
+ end = time.time() + sec
+ chunks: list[bytes] = []
+ while time.time() < end:
+ try:
+ b = os.read(fd, 4096)
+ if b:
+ chunks.append(b)
+ except BlockingIOError:
+ time.sleep(0.02)
+ return b"".join(chunks)
+
+
+def write(fd: int, data: bytes) -> None:
+ os.write(fd, data)
+ termios.tcdrain(fd)
+
+
+def has_fw(data: bytes) -> bool:
+ lo = data.lower()
+ return any(m.lower() in lo for m in FIRMWARE)
+
+
+def kiss_escape(data: bytes) -> bytes:
+ out = bytearray()
+ for b in data:
+ if b == 0xC0:
+ out.extend((0xDB, 0xDC))
+ elif b == 0xDB:
+ out.extend((0xDB, 0xDD))
+ else:
+ out.append(b)
+ return bytes(out)
+
+
+def ax25_addr(call: str, last: bool) -> bytes:
+ call = call.upper().ljust(6)[:6]
+ raw = bytes((ord(c) << 1) for c in call)
+ return raw + bytes([((0 & 0x0F) << 1) | (0x01 if last else 0x00)])
+
+
+def ax25_crc(data: bytes) -> int:
+ crc = 0xFFFF
+ for b in data:
+ crc ^= b
+ for _ in range(8):
+ crc = (crc >> 1) ^ 0x8408 if crc & 1 else crc >> 1
+ return crc ^ 0xFFFF
+
+
+def ax25_ui(src: str, dst: str, info: bytes) -> bytes:
+ body = ax25_addr(dst, False) + ax25_addr(src, True) + b"\x03\xF0" + info
+ crc = ax25_crc(body)
+ return body + bytes((crc & 0xFF, crc >> 8))
+
+
+def kiss_frame(payload: bytes) -> bytes:
+ return b"\xC0\x00" + kiss_escape(payload) + b"\xC0"
+
+
+def main() -> int:
+ print("=" * 60)
+ print("TNC2C TX/RX Test")
+ print("=" * 60)
+
+ if os.system("pgrep -x hybbx >/dev/null 2>&1") == 0:
+ print("FAIL: hybbx is running - stop it first")
+ return 2
+
+ fd = open_port()
+ print(f"Device: {DEV} 19200 8N1 {modem(fd)}")
+
+ # --- Host check ---
+ print("\n[1] Host check")
+ passive = read_sec(fd, 2.0)
+ write(fd, b"kiss off\r")
+ time.sleep(0.5)
+ pre = read_sec(fd, 1.0)
+ write(fd, b"INFO\r")
+ time.sleep(0.3)
+ info = read_sec(fd, 4.0)
+ all_host = passive + pre + info
+ if has_fw(all_host):
+ print(" OK: firmware/host detected")
+ elif info.strip() in (b"INFO", b"INFO\r", b"INFO\r\n"):
+ print(" WARN: echo only - try ./tnc2c-host-reset.sh --kiss or ./tnc2c-boot-wait.sh --recover-only")
+ else:
+ print(f" INFO reply: {info[:120]!r}")
+
+ # --- TX ---
+ print("\n[2] TX (KISS UI TEST-0 -> CQ, Msg=TXRX)")
+ write(fd, b"kiss on\r")
+ time.sleep(0.5)
+ read_sec(fd, 0.3)
+ pkt = kiss_frame(ax25_ui("TEST-0", "CQ", b"TXRX"))
+ t0 = time.time()
+ write(fd, pkt)
+ time.sleep(3.0)
+ tx_rx = read_sec(fd, 1.0)
+ write(fd, b"kiss off\r")
+ time.sleep(0.3)
+
+ kiss_back = b"\xc0" in tx_rx or b"TXRX" in tx_rx
+ print(f" Sent: {len(pkt)} bytes KISS")
+ print(f" Serial RX: {len(tx_rx)} bytes")
+ if kiss_back:
+ print(" OK: KISS frame seen on serial")
+ else:
+ print(" WARN: no KISS echo on serial")
+ print(" -> CHECK NOW: LED2 PTT + modem tone on 2m CB radio")
+
+ # --- RX (passive, no TX) ---
+ print("\n[3] RX (passive 12s, kiss off - listening on radio/monitor)")
+ termios.tcflush(fd, termios.TCIOFLUSH)
+ write(fd, b"kiss off\r")
+ time.sleep(0.5)
+ read_sec(fd, 0.5)
+ print(" Listening 12s ... (RF noise/CD should keep LED3 on)")
+ rx_data = read_sec(fd, 12.0)
+ os.close(fd)
+
+ print(f" Passive RX: {len(rx_data)} bytes")
+ if rx_data:
+ printable = sum(1 for b in rx_data if 32 <= b < 127 or b in (9, 10, 13))
+ ratio = printable / len(rx_data)
+ print(f" Printable ratio: {ratio:.0%}")
+ if b"\xc0" in rx_data:
+ print(" OK: KISS frame(s) received (RF traffic or monitor)")
+ else:
+ print(f" Sample: {rx_data[:80]!r}")
+ else:
+ print(" (no serial traffic - normal on empty band)")
+ print(" -> Hardware RX: LED3 CD with noise/SQ open?")
+
+ print("\n" + "=" * 60)
+ print("Hardware checklist:")
+ print(" TX: LED2 (PTT) blinked? 2m CB heard modem tone?")
+ print(" RX: LED3 (CD) on with noise? SQ closed -> CD off?")
+ print(" DIN: Pin1=TX(white) Pin2=GND Pin3=PTT(red) Pin4=RX(tip)")
+ print("=" * 60)
+ return 0
+
+
+if __name__ == "__main__":
+ sys.exit(main())
diff --git a/stacks/tncs/tnc2c.jpg b/stacks/tncs/tnc2c.jpg
new file mode 100644
index 0000000..91a3116
--- /dev/null
+++ b/stacks/tncs/tnc2c.jpg
Binary files differ
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
git clone -b <branch> https://cgit.mode42.com/<repo>.git
git clone -b <branch> git://cgit.mode42.com/<repo>.git

info@mode42.com