diff options
| author | info@mode42.com <info@mode42.com> | 2026-08-07 18:25:13 +0000 |
|---|---|---|
| committer | info@mode42.com <info@mode42.com> | 2026-08-07 18:25:13 +0000 |
| commit | 04d965d67a7264a1c7c211494aebda1953df7603 (patch) | |
| tree | 0ebd700a6e219f84a26a656f4bee8778bc75da7b /stacks/tncs | |
Initial push
Diffstat (limited to 'stacks/tncs')
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 Binary files differnew file mode 100644 index 0000000..c1dfe04 --- /dev/null +++ b/stacks/tncs/RS232A259.jpg diff --git a/stacks/tncs/TNC2C.pdf b/stacks/tncs/TNC2C.pdf Binary files differnew file mode 100644 index 0000000..d9f7b9d --- /dev/null +++ b/stacks/tncs/TNC2C.pdf 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 Binary files differnew file mode 100644 index 0000000..29c70c4 --- /dev/null +++ b/stacks/tncs/tnc2c-2.jpg 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 Binary files differnew file mode 100644 index 0000000..91a3116 --- /dev/null +++ b/stacks/tncs/tnc2c.jpg 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 |
