summaryrefslogtreecommitdiff
path: root/stacks/daemon/kiss_bridge.py
diff options
context:
space:
mode:
Diffstat (limited to 'stacks/daemon/kiss_bridge.py')
-rw-r--r--stacks/daemon/kiss_bridge.py574
1 files changed, 574 insertions, 0 deletions
diff --git a/stacks/daemon/kiss_bridge.py b/stacks/daemon/kiss_bridge.py
new file mode 100644
index 0000000..b433d0e
--- /dev/null
+++ b/stacks/daemon/kiss_bridge.py
@@ -0,0 +1,574 @@
+"""
+KISS serial bridge for max25d — AX.25 UI over TNC2C / PK-TNC2.
+
+PTT: TNC firmware keys on KISS DATA (requires MYCALL); kernel BayCom (baycom_ser_fdx)
+keys RTS in the driver when the KISS bridge accepts a DATA frame — no max25d PTT command.
+"""
+from __future__ import annotations
+
+import fcntl
+import os
+import struct
+import termios
+import threading
+import time
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Callable, Optional
+
+from ax25_codec import ( # noqa: E402
+ ax25_build_ui,
+ ax25_crc,
+ ax25_crc_valid,
+ ax25_parse_ui,
+ format_callsign,
+ parse_callsign,
+ validate_callsign,
+)
+from tx_pace import tx_pace_before_send # noqa: E402
+
+FEND = 0xC0
+FESC = 0xDB
+TFEND = 0xDC
+TFESC = 0xDD
+
+# Native KISS return (TheFirmware TF 2.7) — firmware reset to banner, not TAPR kiss off
+KISS_RETURN_FRAME = b"\xc0\xff\xc0"
+
+KISS_CMD_DATA = 0x00
+MAX_FRAME = 1024
+MAX_PAYLOAD = 256
+
+
+@dataclass
+class SerialProfile:
+ device: str = "/dev/ttyS4"
+ baud: int = 19200
+ line: str = "8n1"
+ dtr_rts: bool = True
+ kiss_entry: str = "kiss_on" # kiss_on | auto
+
+
+class KissDecoder:
+ def __init__(self) -> None:
+ self._buf = bytearray()
+ self._in_frame = False
+ self._escape = False
+
+ def feed(self, data: bytes) -> list[tuple[int, bytes]]:
+ frames: list[tuple[int, bytes]] = []
+ for byte in data:
+ if byte == FEND:
+ if self._in_frame and self._buf:
+ parsed = self._deliver()
+ if parsed is not None:
+ frames.append(parsed)
+ self._in_frame = True
+ self._escape = False
+ self._buf.clear()
+ continue
+ if not self._in_frame:
+ continue
+ if self._escape:
+ if byte == TFEND:
+ byte = FEND
+ elif byte == TFESC:
+ byte = FESC
+ self._escape = False
+ elif byte == FESC:
+ self._escape = True
+ continue
+ if len(self._buf) >= MAX_FRAME:
+ self._buf.clear()
+ self._in_frame = False
+ continue
+ self._buf.append(byte)
+ return frames
+
+ def _deliver(self) -> Optional[tuple[int, bytes]]:
+ if len(self._buf) < 1:
+ return None
+ cmd_byte = self._buf[0]
+ if (cmd_byte & 0x0F) != KISS_CMD_DATA:
+ return None
+ port = (cmd_byte >> 4) & 0x0F
+ payload = bytes(self._buf[1:])
+ return port, payload
+
+
+def kiss_escape(data: bytes) -> bytes:
+ out = bytearray()
+ for b in data:
+ if b == FEND:
+ out.extend((FESC, TFEND))
+ elif b == FESC:
+ out.extend((FESC, TFESC))
+ else:
+ out.append(b)
+ return bytes(out)
+
+
+def kiss_encode(port: int, cmd: int, payload: bytes) -> bytes:
+ cmd_byte = ((port & 0x0F) << 4) | (cmd & 0x0F)
+ return b"\xC0" + kiss_escape(bytes([cmd_byte]) + payload) + b"\xC0"
+
+
+def kiss_data_frame(port: int, ax25_frame: bytes) -> bytes:
+ """Build KISS DATA; strip FCS when CRC validates (ax25ipd / KISS convention)."""
+ if ax25_crc_valid(ax25_frame):
+ ax25_frame = ax25_frame[:-2]
+ return kiss_encode(port, KISS_CMD_DATA, ax25_frame)
+
+
+def format_rx_line(src: str, dst: str, payload: bytes, ax25_ui: bool) -> str:
+ try:
+ text = payload.decode("utf-8")
+ except UnicodeDecodeError:
+ text = payload.decode("utf-8", errors="replace")
+ if ax25_ui:
+ return f"[AX25 UI {src}>{dst}] {text}"
+ return text
+
+
+def load_env_file(path: str) -> dict[str, str]:
+ out: dict[str, str] = {}
+ if not path or not os.path.isfile(path):
+ return out
+ with open(path, encoding="utf-8") as fh:
+ for line in fh:
+ line = line.strip()
+ if not line or line.startswith("#"):
+ continue
+ if "=" in line:
+ key, val = line.split("=", 1)
+ out[key.strip()] = val.strip()
+ return out
+
+
+def _load_serial_env(device_id: str, root: str, prefix: Optional[str] = None) -> dict[str, str]:
+ from paths import serial_env_candidates
+
+ tree = Path(root)
+ pref = Path(prefix) if prefix else None
+ for path in serial_env_candidates(device_id, tree, pref):
+ env = load_env_file(str(path))
+ if env:
+ return env
+ return {}
+
+
+def serial_profile_for_device(
+ device_id: str,
+ root: str,
+ ini: dict[str, str],
+ prefix: Optional[str] = None,
+) -> SerialProfile:
+ prof = SerialProfile()
+ if ini.get("device"):
+ prof.device = ini["device"]
+ if ini.get("baud"):
+ prof.baud = int(ini["baud"])
+ if ini.get("line"):
+ prof.line = ini["line"].lower()
+ if ini.get("dtr_rts"):
+ prof.dtr_rts = ini["dtr_rts"].lower() in ("1", "yes", "true", "on")
+ if ini.get("kiss_entry"):
+ prof.kiss_entry = ini["kiss_entry"].lower()
+
+ env = _load_serial_env(device_id, root, prefix)
+
+ if device_id == "tnc2c":
+ prof.device = ini.get("device") or env.get("TNC2C_DEV", prof.device)
+ prof.baud = int(ini.get("baud") or env.get("TNC2C_BAUD", prof.baud))
+ prof.line = (ini.get("line") or env.get("TNC2C_LINE", prof.line)).lower()
+ if "dtr_rts" not in ini:
+ prof.dtr_rts = True
+ if "kiss_entry" not in ini:
+ prof.kiss_entry = "kiss_on"
+ elif device_id == "pktnc2":
+ prof.device = ini.get("device") or env.get("PKTNC2_DEV") or env.get("TNC_DEV", prof.device)
+ prof.baud = int(ini.get("baud") or env.get("PKTNC2_BAUD") or env.get("TNC_BAUD", "9600"))
+ prof.line = (ini.get("line") or env.get("PKTNC2_LINE") or env.get("TNC_LINE", "8n1")).lower()
+ if "dtr_rts" not in ini:
+ prof.dtr_rts = False
+ if "kiss_entry" not in ini:
+ prof.kiss_entry = "auto"
+ elif device_id in ("pccom-kiss", "baycom-kiss"):
+ prof.device = ini.get("device") or env.get("PCCOM_KISS_DEV", prof.device or "/dev/ttyUSB0")
+ prof.baud = int(ini.get("baud") or env.get("PCCOM_KISS_BAUD", "9600"))
+ prof.line = (ini.get("line") or env.get("PCCOM_KISS_LINE", "8n1")).lower()
+ dtr = ini.get("dtr_rts") or env.get("PCCOM_KISS_DTR_RTS", "no")
+ prof.dtr_rts = str(dtr).lower() in ("1", "yes", "true", "on")
+ if "kiss_entry" not in ini:
+ prof.kiss_entry = "none"
+ elif device_id == "tmodem":
+ default_dev = "/dev/cuaU0" if sys.platform.startswith("freebsd") else "/dev/ttyACM0"
+ prof.device = ini.get("device") or env.get("TMODEM_DEV", prof.device or default_dev)
+ prof.baud = int(ini.get("baud") or env.get("TMODEM_BAUD", "115200"))
+ prof.line = (ini.get("line") or env.get("TMODEM_LINE", "8n1")).lower()
+ dtr = ini.get("dtr_rts") or env.get("TMODEM_DTR_RTS", "no")
+ prof.dtr_rts = str(dtr).lower() in ("1", "yes", "true", "on")
+ if "kiss_entry" not in ini:
+ prof.kiss_entry = "none"
+ return prof
+
+
+def _parse_line(line: str) -> tuple[int, int]:
+ line = line.lower()
+ if line == "7e1":
+ return termios.CS7, termios.PARENB
+ return termios.CS8, 0
+
+
+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]
+
+
+class KissBridge:
+ """Thread-safe KISS bridge on a serial TNC port."""
+
+ def __init__(
+ self,
+ profile: SerialProfile,
+ on_rx: Callable[[str], None],
+ log: Optional[Callable[[str], None]] = None,
+ *,
+ tree_root: str = "",
+ install_prefix: Optional[str] = None,
+ on_invalid: Optional[Callable[[], None]] = None,
+ ) -> None:
+ self.profile = profile
+ self._on_rx = on_rx
+ self._on_invalid = on_invalid
+ self._log = log or (lambda _m: None)
+ self._tree_root = tree_root
+ self._install_prefix = install_prefix
+ self._fd: Optional[int] = None
+ self._thread: Optional[threading.Thread] = None
+ self._stop = threading.Event()
+ self._lock = threading.Lock()
+ self._kiss_active = False
+ self._mycall = ""
+ self.status = "closed"
+ self._decoder = KissDecoder()
+
+ def open(self) -> bool:
+ dev = self.profile.device
+ if not os.access(dev, os.R_OK | os.W_OK):
+ self.status = "error-no-device"
+ self._log(f"serial: no access to {dev}")
+ return False
+ try:
+ speed = _parse_baud(self.profile.baud)
+ databits, parity = _parse_line(self.profile.line)
+ except ValueError as exc:
+ self.status = "error-config"
+ self._log(f"serial: {exc}")
+ return False
+ try:
+ 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]
+ if self.profile.dtr_rts:
+ flags |= 0x004 | 0x002
+ fcntl.ioctl(fd, 0x5416, struct.pack("I", flags))
+ except OSError as exc:
+ self.status = "error-open"
+ self._log(f"serial open failed: {exc}")
+ return False
+ self._fd = fd
+ self._log(f"serial open {dev} {self.profile.baud} {self.profile.line.upper()}")
+ if self.profile.dtr_rts:
+ time.sleep(2.0)
+ self._log("serial: DTR settle (2s)")
+ self.status = "open"
+ return True
+
+ def close(self) -> None:
+ self._stop_rx_thread()
+ with self._lock:
+ if self._fd is not None:
+ if self._kiss_active:
+ self._write_unlocked(KISS_RETURN_FRAME)
+ try:
+ os.close(self._fd)
+ except OSError:
+ pass
+ self._fd = None
+ self._kiss_active = False
+ self.status = "closed"
+ self._decoder = KissDecoder()
+
+ def _stop_rx_thread(self) -> None:
+ """Stop KISS RX thread so recovery owns the serial FD exclusively."""
+ self._stop.set()
+ if self._thread is not None:
+ self._thread.join(timeout=2.0)
+ self._thread = None
+ self._stop.clear()
+
+ def _start_rx_thread(self) -> None:
+ """Start KISS RX after terminal recovery and KISS entry succeed."""
+ if self._fd is None or self._thread is not None or not self._kiss_active:
+ return
+ self._stop.clear()
+ self._thread = threading.Thread(target=self._rx_loop, name="kiss-rx", daemon=True)
+ self._thread.start()
+
+ def attach_session(self, mycall: str) -> bool:
+ if self._fd is None:
+ return False
+ self._mycall = mycall.upper()
+ self._stop_rx_thread()
+ with self._lock:
+ ok = self._stabilize_unlocked(self._mycall, force_ladder=False)
+ if ok:
+ self._start_rx_thread()
+ return ok
+
+ def stabilize_session(self, mycall: str, *, force: bool = False) -> bool:
+ """Probe terminal/KISS health and repair without closing the port (keeps DTR)."""
+ if self._fd is None:
+ self.status = "error-open"
+ return False
+ if not force and self._kiss_active and self.status == "ready":
+ return True
+ self._mycall = mycall.upper()
+ self._stop_rx_thread()
+ with self._lock:
+ ok = self._stabilize_unlocked(self._mycall, force_ladder=force)
+ if ok:
+ self._start_rx_thread()
+ return ok
+
+ def _load_recovery_mod(self):
+ import importlib.util
+
+ from paths import tnc_serial_recovery_path
+
+ path = None
+ if self._tree_root:
+ prefix = Path(self._install_prefix) if self._install_prefix else None
+ path = tnc_serial_recovery_path(Path(self._tree_root), prefix)
+ if path is None:
+ path = Path(__file__).resolve().parents[1] / "tncs" / "tnc_serial_recovery.py"
+ if not path.is_file():
+ return None
+ 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
+
+ def _recovery_io(self) -> tuple[Callable[[bytes], None], Callable[[float], bytes]]:
+ def wf(data: bytes) -> None:
+ self._write_unlocked(data)
+
+ def rf(seconds: float) -> bytes:
+ return self._drain_unlocked(seconds)
+
+ return wf, rf
+
+ def _leave_kiss_unlocked(self) -> None:
+ if not self._kiss_active:
+ return
+ self._write_unlocked(KISS_RETURN_FRAME)
+ time.sleep(0.3)
+ self._drain_unlocked(0.2)
+ self._kiss_active = False
+
+ def _enter_kiss_session_unlocked(self) -> bool:
+ if not self._set_mycall_unlocked(self._mycall):
+ self._log("serial: MYCALL may have failed")
+ if not self._enter_kiss_unlocked():
+ self.status = "error-kiss"
+ self._kiss_active = False
+ return False
+ self._kiss_active = True
+ self.status = "ready"
+ return True
+
+ def _stabilize_unlocked(self, mycall: str, *, force_ladder: bool) -> bool:
+ """Host probe, optional recovery ladder, MYCALL + KISS — port stays open."""
+ self._mycall = mycall.upper()
+ try:
+ mod = self._load_recovery_mod()
+ wf, rf = self._recovery_io()
+ self._leave_kiss_unlocked()
+
+ if mod is None:
+ self._log("serial: tnc_serial_recovery.py not found")
+ self.status = "error-config"
+ self._kiss_active = False
+ return False
+
+ ok, probe_data, only_echo = mod.probe_info(wf, rf, pause=0.25)
+ if not ok or only_echo or force_ladder:
+ self._log(
+ f"serial: initial probe — {mod.format_rx_brief(probe_data)}, "
+ f"echo_only={only_echo}, banner={mod.has_banner(probe_data)}"
+ )
+ if ok and not only_echo and not force_ladder:
+ return self._enter_kiss_session_unlocked()
+ if only_echo or not ok or force_ladder:
+ label = "auto-repair" if force_ladder else "recovery ladder"
+ self._log(f"serial: {label}")
+ ok, _ = mod.recover_terminal(wf, rf, log=self._log)
+ if not ok:
+ self.status = "error-host"
+ self._kiss_active = False
+ return False
+ return self._enter_kiss_session_unlocked()
+ self.status = "error-host"
+ self._kiss_active = False
+ return False
+ except OSError as exc:
+ self.status = "error-io"
+ self._kiss_active = False
+ self._log(f"serial: stabilize I/O error ({exc})")
+ return False
+
+ def _recover_terminal_unlocked(self) -> bool:
+ """Legacy hook — full ladder when probe fails."""
+ try:
+ mod = self._load_recovery_mod()
+ if mod is None:
+ return True
+ wf, rf = self._recovery_io()
+ ok, _, only_echo = mod.probe_info(wf, rf)
+ if ok and not only_echo:
+ self._log("serial: terminal mode OK")
+ return True
+ self._log("serial: software recovery ladder")
+ ok, _ = mod.recover_terminal(wf, rf, log=self._log)
+ if ok:
+ self._log("serial: recovery OK")
+ return ok
+ except Exception as exc:
+ self._log(f"serial: recovery skipped ({exc})")
+ return True
+
+ def detach_session(self) -> None:
+ self._stop_rx_thread()
+ with self._lock:
+ if self._fd is not None and self._kiss_active:
+ self._write_unlocked(KISS_RETURN_FRAME)
+ time.sleep(0.2)
+ self._kiss_active = False
+ if self._fd is not None:
+ self.status = "open"
+
+ def transmit(self, src: str, dst: str, text: str, ax25_ui: bool) -> tuple[bool, str]:
+ if self._fd is None or not self._kiss_active:
+ return False, "serial not ready"
+ if len(text.encode("utf-8")) > MAX_PAYLOAD:
+ return False, "payload too long"
+ try:
+ validate_callsign(src)
+ validate_callsign(dst)
+ except ValueError as exc:
+ return False, str(exc)
+ info = text.encode("utf-8")
+ try:
+ frame = ax25_build_ui(src, dst, info)
+ except ValueError as exc:
+ return False, str(exc)
+ pkt = kiss_data_frame(0, frame)
+ with self._lock:
+ try:
+ tx_pace_before_send()
+ self._write_unlocked(pkt)
+ termios.tcdrain(self._fd)
+ except OSError as exc:
+ self.status = "error-tx"
+ return False, f"tx failed: {exc}"
+ display = format_rx_line(src, dst, info, ax25_ui)
+ return True, display
+
+ def _write_unlocked(self, data: bytes) -> None:
+ if self._fd is None:
+ return
+ os.write(self._fd, data)
+
+ def _drain_unlocked(self, seconds: float) -> bytes:
+ if self._fd is None:
+ return b""
+ end = time.time() + seconds
+ chunks: list[bytes] = []
+ while time.time() < end:
+ try:
+ chunk = os.read(self._fd, 4096)
+ if chunk:
+ chunks.append(chunk)
+ except BlockingIOError:
+ time.sleep(0.02)
+ return b"".join(chunks)
+
+ def _set_mycall_unlocked(self, call: str) -> bool:
+ mod = self._load_recovery_mod()
+ if mod is not None and hasattr(mod, "tf_mycall_frame"):
+ cmd = mod.tf_mycall_frame(call)
+ else:
+ cmd = f"\x1bI {call.upper()}\r".encode("ascii", errors="replace")
+ self._write_unlocked(cmd)
+ time.sleep(0.4)
+ reply = self._drain_unlocked(0.6)
+ return b"?" not in reply[:32]
+
+ def _enter_kiss_unlocked(self) -> bool:
+ entry = self.profile.kiss_entry
+ if entry == "tapr":
+ self._write_unlocked(b"kiss on\r")
+ time.sleep(0.5)
+ self._drain_unlocked(0.3)
+ return True
+ self._write_unlocked(b"\x1b@K")
+ time.sleep(0.5)
+ self._drain_unlocked(0.3)
+ return True
+
+ def _rx_loop(self) -> None:
+ while not self._stop.is_set():
+ fd = self._fd
+ if fd is None:
+ break
+ try:
+ chunk = os.read(fd, 4096)
+ except BlockingIOError:
+ time.sleep(0.05)
+ continue
+ except OSError:
+ self._log("serial: rx I/O error — watch will repair")
+ self.status = "error-io"
+ break
+ if not chunk:
+ time.sleep(0.05)
+ continue
+ for _port, payload in self._decoder.feed(chunk):
+ if not payload:
+ continue
+ parsed = ax25_parse_ui(payload)
+ if parsed is None:
+ if self._on_invalid is not None and len(payload) >= 16:
+ self._on_invalid()
+ continue
+ src, dst, info = parsed
+ line = format_rx_line(src, dst, info, ax25_ui=True)
+ self._on_rx(line)
git clone -b <branch> https://cgit.mode42.com/<repo>.git
git clone -b <branch> git://cgit.mode42.com/<repo>.git


info@mode42.com