path: root/stacks/daemon/kiss_bridge.py

"""
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)