path: root/stacks/daemon/device_backends.py

"""
Device backends for max25d — heterogeneous RF paths (TNC, BayCom, CRDOP).

Each enabled [devices] id gets one backend instance. Backends without hardware
validation log a startup warning but still wire real stack paths (not silent no-ops).
"""
from __future__ import annotations

import os
import select
import socket
import struct
import termios
import threading
import time
from abc import ABC, abstractmethod
from dataclasses import dataclass
from typing import Callable, Optional

from kiss_bridge import (
    MAX_PAYLOAD,
    KissDecoder,
    SerialProfile,
    ax25_build_ui,
    ax25_parse_ui,
    format_rx_line,
    kiss_data_frame,
    serial_profile_for_device,
)
from kiss_bridge import KissBridge  # noqa: E402 — re-exported wrapper target
from paths import normalize_max25_bcpr_path

LogFn = Callable[[str], None]
RxFn = Callable[[str], None]
InvalidFn = Callable[[], None]


def _spec_int(raw: str, default: int) -> int:
    try:
        return int(str(raw).strip())
    except (TypeError, ValueError):
        return default

# max25e0 host address defaults (overridable in max25d.ini [device.max25e0])
MAX25E0_DEFAULT_IPV4 = "127.0.0.25/8"
MAX25E0_DEFAULT_IPV6 = "::25/128"

# manifest.yaml device ids → default hardware + backend kind
DEVICE_REGISTRY: dict[str, dict[str, str | bool]] = {
    "tnc2c": {"hardware": "tncs", "backend": "kiss-serial", "tested": True},
    "pktnc2": {"hardware": "tncs", "backend": "kiss-serial", "tested": False},
    "tmodem": {"hardware": "tncs", "backend": "kiss-raw-serial", "tested": False},
    # Kernel baycom-ser12/par96 removed 2026-07-18 — use max25-bcpr → device max25e0
    "baycom-kiss": {"hardware": "modems", "backend": "kiss-raw-serial", "tested": False},
    "pccom-kiss": {"hardware": "modems", "backend": "kiss-raw-serial", "tested": False},
    "max25e0": {"hardware": "modems", "backend": "max25-bcpr-kiss", "tested": True},
    "max25e0:bc0": {"hardware": "modems", "backend": "max25-bcpr-kiss", "tested": True},
    "max25e0:bc1": {"hardware": "modems", "backend": "max25-bcpr-kiss", "tested": True},
    "soft-crdop": {"hardware": "soft-modems", "backend": "crdop-tcp", "tested": True},
    "audio-dummy": {"hardware": "acoustic-bench", "backend": "audio-dummy", "tested": True},
}


def registry_hardware(device_id: str, fallback: str = "tncs") -> str:
    entry = DEVICE_REGISTRY.get(device_id, {})
    return str(entry.get("hardware", fallback))


def registry_backend(device_id: str) -> str:
    entry = DEVICE_REGISTRY.get(device_id, {})
    return str(entry.get("backend", "kiss-serial"))


def registry_tested(device_id: str) -> bool:
    entry = DEVICE_REGISTRY.get(device_id, {})
    return bool(entry.get("tested", False))


def baycom_ctl_device_id(dev_cfg: DeviceBackendConfig) -> str:
    """Legacy helper: kernel BayCom ctl device id (stack removed — prefer max25-bcpr)."""
    entry = DEVICE_REGISTRY.get(dev_cfg.device_id, {})
    if entry.get("backend") == "baycom-kiss":
        return dev_cfg.device_id
    return "max25e0"


@dataclass
class DeviceBackendConfig:
    device_id: str
    hardware: str = ""
    backend_type: str = ""
    device_spec: str = ""
    enabled: bool = True
    # Serial (TNC / baycom-kiss USB)
    serial_device: str = ""
    serial_baud: int = 0
    serial_line: str = ""
    serial_dtr_rts: str = ""
    serial_kiss_entry: str = ""
    # BayCom kernel KISS PTY
    kiss_link: str = ""
    baycom_modem: str = "a"
    baycom_ini: str = ""
    # max25-bcpr userspace SER12 — max25e0 (+ forks max25e0:bcN)
    max25_bcpr_ini: str = ""
    max25_bcpr_device: str = ""  # bc0 | bc1
    # Host addresses (max25e0 family only; forks inherit from max25e0)
    ipv4: str = ""
    ipv6: str = ""
    # Legacy field aliases (read-only mirrors filled by parser)
    bcpr_ini: str = ""
    bcpr_device: str = ""
    # CRDOP TCP
    crdop_host: str = "127.0.0.1"
    crdop_port: int = 8515
    crdop_profile: str = "default"
    crdop_listen: bool = True
    # Acoustic bench / audio-dummy
    audio_mode: str = "loopback"  # loopback | alsa | host
    audio_capture: str = ""
    audio_playback: str = ""
    audio_sample_rate: int = 48000
    audio_host_port: int = 8520


class DeviceBackend(ABC):
    """Common RX/TX/PTT surface for max25d."""

    device_id: str
    status: str = "closed"
    backend_type: str = ""

    @abstractmethod
    def open(self) -> bool:
        ...

    @abstractmethod
    def close(self) -> None:
        ...

    @abstractmethod
    def attach_session(self, mycall: str) -> bool:
        ...

    @abstractmethod
    def detach_session(self) -> None:
        ...

    @abstractmethod
    def transmit(self, src: str, dst: str, text: str, ax25_ui: bool) -> tuple[bool, str]:
        ...


class KissSerialBackend(DeviceBackend):
    """TNC2C / PK-TNC2 — command-mode serial entry into KISS."""

    backend_type = "kiss-serial"

    def __init__(
        self,
        cfg: DeviceBackendConfig,
        root: str,
        on_rx: RxFn,
        log: Optional[LogFn] = None,
        prefix: Optional[str] = None,
        on_invalid: Optional[InvalidFn] = None,
    ) -> None:
        self.device_id = cfg.device_id
        self._cfg = cfg
        self._root = root
        self._prefix = prefix
        self._on_rx = on_rx
        self._on_invalid = on_invalid
        self._log = log or (lambda _m: None)
        self._bridge: Optional[KissBridge] = None
        self.status = "closed"

    def _bridge_log(self, msg: str) -> None:
        self._log(f"{self.device_id}: {msg}")

    def _ini_overrides(self) -> dict[str, str]:
        out: dict[str, str] = {}
        if self._cfg.serial_device:
            out["device"] = self._cfg.serial_device
        if self._cfg.serial_baud:
            out["baud"] = str(self._cfg.serial_baud)
        if self._cfg.serial_line:
            out["line"] = self._cfg.serial_line
        if self._cfg.serial_dtr_rts:
            out["dtr_rts"] = self._cfg.serial_dtr_rts
        if self._cfg.serial_kiss_entry:
            out["kiss_entry"] = self._cfg.serial_kiss_entry
        return out

    def open(self) -> bool:
        profile = serial_profile_for_device(
            self.device_id,
            self._root,
            self._ini_overrides(),
            prefix=self._prefix,
        )
        bridge = KissBridge(
            profile,
            self._on_rx,
            self._bridge_log,
            tree_root=self._root,
            install_prefix=self._prefix,
            on_invalid=self._on_invalid,
        )
        if not bridge.open():
            self._bridge = bridge
            self.status = bridge.status
            return False
        self._bridge = bridge
        self.status = bridge.status
        return True

    def close(self) -> None:
        if self._bridge is not None:
            self._bridge.close()
            self.status = self._bridge.status
            self._bridge = None
        else:
            self.status = "closed"

    def stabilize_session(self, mycall: str, *, force: bool = False) -> bool:
        if self._bridge is None:
            return False
        ok = self._bridge.stabilize_session(mycall, force=force)
        self.status = self._bridge.status
        return ok

    def attach_session(self, mycall: str) -> bool:
        if self._bridge is None:
            return False
        ok = self._bridge.attach_session(mycall)
        self.status = self._bridge.status
        return ok

    def detach_session(self) -> None:
        if self._bridge is None:
            return
        self._bridge.detach_session()
        self.status = self._bridge.status

    def transmit(self, src: str, dst: str, text: str, ax25_ui: bool) -> tuple[bool, str]:
        if self._bridge is None:
            return False, "serial not ready"
        ok, display = self._bridge.transmit(src, dst, text, ax25_ui)
        self.status = self._bridge.status
        return ok, display


class KissRawBackend(DeviceBackend):
    """Raw KISS on serial or BayCom KISS PTY (no command-mode entry)."""

    backend_type = "kiss-raw"

    def __init__(
        self,
        cfg: DeviceBackendConfig,
        path: str,
        profile: SerialProfile,
        on_rx: RxFn,
        log: Optional[LogFn] = None,
        *,
        is_pty: bool = False,
    ) -> None:
        self.device_id = cfg.device_id
        self._path = path
        self._profile = profile
        self._on_rx = on_rx
        self._log = log or (lambda _m: None)
        self._is_pty = is_pty
        self._fd: Optional[int] = None
        self._thread: Optional[threading.Thread] = None
        self._stop = threading.Event()
        self._lock = threading.Lock()
        self._mycall = ""
        self._kiss_active = False
        self._decoder = KissDecoder()
        self.status = "closed"

    def open(self) -> bool:
        path = self._path
        if not path:
            self.status = "error-no-path"
            self._log(f"{self.device_id}: no KISS path configured")
            return False
        if not os.path.exists(path):
            self.status = "error-no-device"
            self._log(f"{self.device_id}: path missing: {path}")
            return False
        try:
            fd = os.open(path, os.O_RDWR | os.O_NOCTTY | os.O_NONBLOCK)
            if not self._is_pty:
                self._configure_serial(fd)
            termios.tcflush(fd, termios.TCIOFLUSH)
        except OSError as exc:
            self.status = "error-open"
            self._log(f"{self.device_id}: open failed: {exc}")
            return False
        self._fd = fd
        self.status = "open"
        self._stop.clear()
        self._thread = threading.Thread(
            target=self._rx_loop,
            name=f"kiss-raw-{self.device_id}",
            daemon=True,
        )
        self._thread.start()
        self._log(f"{self.device_id}: raw KISS open {path}")
        return True

    def _configure_serial(self, fd: int) -> None:
        from kiss_bridge import _parse_baud, _parse_line

        speed = _parse_baud(self._profile.baud)
        databits, parity = _parse_line(self._profile.line)
        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)
        flags = struct.unpack("I", __import__("fcntl").ioctl(fd, 0x5415, struct.pack("I", 0)))[0]
        if self._profile.dtr_rts:
            flags |= 0x004 | 0x002
        __import__("fcntl").ioctl(fd, 0x5416, struct.pack("I", flags))

    def close(self) -> None:
        self._stop.set()
        if self._thread is not None:
            self._thread.join(timeout=2.0)
            self._thread = None
        with self._lock:
            if self._fd is not None:
                try:
                    os.close(self._fd)
                except OSError:
                    pass
                self._fd = None
            self._kiss_active = False
        self.status = "closed"
        self._decoder = KissDecoder()

    def attach_session(self, mycall: str) -> bool:
        if self._fd is None:
            return False
        self._mycall = mycall.upper()
        self._kiss_active = True
        self.status = "ready"
        return True

    def detach_session(self) -> None:
        self._kiss_active = False
        if self._fd is not None:
            self.status = "open"

    def stabilize_session(self, mycall: str, *, force: bool = False) -> bool:
        """Reopen KISS path after dead PTY / EIO (e.g. bcprd recycled outside max25d)."""
        path = self._path
        if (
            not force
            and self._kiss_active
            and self.status == "ready"
            and self._fd is not None
            and path
            and os.path.exists(path)
        ):
            if not self._is_pty:
                return True
            # PTY symlink may have been retargeted while our fd still points at a
            # deleted slave — force reopen when the live path inode differs.
            try:
                cur = os.stat(path)
                fd_st = os.fstat(self._fd)
                if cur.st_ino == fd_st.st_ino and cur.st_dev == fd_st.st_dev:
                    return True
            except OSError:
                pass
        was_active = self._kiss_active or force
        call = (mycall or self._mycall or "").upper()
        self.close()
        if not self.open():
            return False
        if was_active and call:
            return self.attach_session(call)
        return self._fd is not None

    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, "KISS not ready"
        if len(text.encode("utf-8")) > MAX_PAYLOAD:
            return False, "payload too long"
        info = text.encode("utf-8")
        frame = ax25_build_ui(src, dst, info)
        pkt = kiss_data_frame(0, frame)
        with self._lock:
            try:
                os.write(self._fd, pkt)
                # PTY: never tcdrain — if the master side dies, drain can hang forever.
                if not self._is_pty:
                    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 _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:
                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:
                    continue
                src, dst, info = parsed
                line = format_rx_line(src, dst, info, ax25_ui=True)
                self._on_rx(line)


class BayComKissBackend(KissRawBackend):
    """BayCom kernel modem (SER12 / PAR96) via baycom-pr KISS PTY."""

    backend_type = "baycom-kiss"

    def __init__(
        self,
        cfg: DeviceBackendConfig,
        on_rx: RxFn,
        log: Optional[LogFn] = None,
    ) -> None:
        modem = cfg.baycom_modem or "a"
        kiss = cfg.kiss_link or f"/var/run/baycom-pr/kiss-{modem}"
        if modem == "a" and not cfg.kiss_link:
            default = "/var/run/baycom-pr/kiss"
            if os.path.exists(default) or not os.path.exists(kiss):
                kiss = default
        profile = SerialProfile(baud=9600, line="8n1", dtr_rts=False)
        super().__init__(cfg, kiss, profile, on_rx, log, is_pty=True)


class Max25BcprKissBackend(KissRawBackend):
    """max25-bcpr userspace SER12 via KISS PTY (max25e0 / max25e0:bcN).

    Hardware is a TCM3105-class AFSK modem chip (bits↔AFSK + PTT) only — not a TNC.
    """

    backend_type = "max25-bcpr-kiss"

    def __init__(
        self,
        cfg: DeviceBackendConfig,
        on_rx: RxFn,
        log: Optional[LogFn] = None,
    ) -> None:
        tag = (cfg.max25_bcpr_device or cfg.bcpr_device or "bc0").strip() or "bc0"
        kiss = cfg.kiss_link or f"/tmp/max25-bcpr/kiss-{tag}"
        kiss = normalize_max25_bcpr_path(kiss)
        profile = SerialProfile(baud=9600, line="8n1", dtr_rts=False)
        super().__init__(cfg, kiss, profile, on_rx, log, is_pty=True)


class KissRawSerialBackend(KissRawBackend):
    """USB/async BayCom KISS serial (kiss-serial backend)."""

    backend_type = "kiss-raw-serial"

    def __init__(
        self,
        cfg: DeviceBackendConfig,
        root: str,
        on_rx: RxFn,
        log: Optional[LogFn] = None,
        prefix: Optional[str] = None,
    ) -> None:
        overrides: dict[str, str] = {}
        if cfg.serial_device:
            overrides["device"] = cfg.serial_device
        if cfg.serial_baud:
            overrides["baud"] = str(cfg.serial_baud)
        if cfg.serial_line:
            overrides["line"] = cfg.serial_line
        if cfg.serial_dtr_rts:
            overrides["dtr_rts"] = cfg.serial_dtr_rts
        prof = serial_profile_for_device(cfg.device_id, root, overrides, prefix=prefix)
        path = cfg.serial_device or prof.device
        super().__init__(cfg, path, prof, on_rx, log, is_pty=False)


class CrdopTcpBackend(DeviceBackend):
    """MAX25-SoftModem (CRDOP) via TCP host interface (:8515 / :8516).

    Native M25/KISS host protocol (MAX25-SoftModem) only.
    """

    backend_type = "crdop-tcp"

    def __init__(
        self,
        cfg: DeviceBackendConfig,
        on_rx: RxFn,
        log: Optional[LogFn] = None,
    ) -> None:
        self.device_id = cfg.device_id
        self._cfg = cfg
        self._on_rx = on_rx
        self._log = log or (lambda _m: None)
        self._ctrl: Optional[socket.socket] = None
        self._data: Optional[socket.socket] = None
        self._thread: Optional[threading.Thread] = None
        self._stop = threading.Event()
        self._lock = threading.RLock()
        self._mycall = ""
        self._connected = False
        self.status = "closed"

    def _line_term(self) -> str:
        return "\n"

    def open(self) -> bool:
        host = self._cfg.crdop_host
        port = self._cfg.crdop_port
        ctrl = None
        data = None
        try:
            ctrl = socket.create_connection((host, port), timeout=5.0)
            ctrl.settimeout(0.5)
            data = socket.create_connection((host, port + 1), timeout=5.0)
            data.settimeout(0.5)
        except OSError as exc:
            if ctrl is not None:
                try:
                    ctrl.close()
                except OSError:
                    pass
            self.status = "error-connect"
            self._log(f"{self.device_id}: CRDOP TCP connect failed ({host}:{port}): {exc}")
            return False
        self._ctrl = ctrl
        self._data = data
        self.status = "open"
        self._stop.clear()
        self._thread = threading.Thread(
            target=self._rx_loop,
            name=f"crdop-rx-{self.device_id}",
            daemon=True,
        )
        self._thread.start()
        self._log(f"{self.device_id}: CRDOP TCP open {host}:{port}")
        return True

    def close(self) -> None:
        self._stop.set()
        if self._thread is not None:
            self._thread.join(timeout=2.0)
            self._thread = None
        for sock in (self._ctrl, self._data):
            if sock is not None:
                try:
                    sock.close()
                except OSError:
                    pass
        self._ctrl = None
        self._data = None
        self._connected = False
        self.status = "closed"

    def _cmd(self, text: str) -> str:
        if self._ctrl is None:
            return ""
        term = self._line_term()
        payload = (text.rstrip(term) + term).encode("ascii", errors="replace")
        with self._lock:
            self._ctrl.sendall(payload)
            return self._read_line_unlocked()

    def _read_line_unlocked(self) -> str:
        if self._ctrl is None:
            return ""
        term = self._line_term()
        term_b = term.encode("ascii")
        buf = b""
        deadline = time.time() + 3.0
        while time.time() < deadline:
            try:
                chunk = self._ctrl.recv(4096)
            except socket.timeout:
                continue
            except OSError:
                break
            if not chunk:
                break
            buf += chunk
            while term_b in buf:
                raw, buf = buf.split(term_b, 1)
                line = raw.decode("ascii", errors="replace").strip()
                if line:
                    return line
        return ""

    def attach_session(self, mycall: str) -> bool:
        if self._ctrl is None:
            return False
        self._mycall = mycall.upper()
        cmds = [
            "INITIALIZE",
            "PROTOCOLMODE KISS",
            f"MYCALL {self._mycall}",
        ]
        if self._cfg.crdop_listen:
            cmds.append("LISTEN TRUE")
        for cmd in cmds:
            reply = self._cmd(cmd)
            self._log(f"{self.device_id}: {cmd} → {reply or '(no reply)'}")
        self._connected = True
        self.status = "ready"
        return True

    def detach_session(self) -> None:
        if self._ctrl is not None and self._connected:
            self._cmd("ABORT")
        self._connected = False
        if self._ctrl is not None:
            self.status = "open"

    def transmit(self, src: str, dst: str, text: str, ax25_ui: bool) -> tuple[bool, str]:
        if self._ctrl is None or self._data is None or not self._connected:
            return False, "CRDOP not ready"
        payload = text.encode("utf-8")
        if len(payload) > MAX_PAYLOAD:
            return False, "payload too long"
        with self._lock:
            try:
                body = ax25_build_ui(src, dst, payload)
                if len(body) >= 2:
                    body = body[:-2]
                self._data.sendall(body)
            except OSError as exc:
                self.status = "error-tx"
                return False, f"tx failed: {exc}"
        if ax25_ui:
            display = f"[CRDOP AX25 UI {src}>{dst}] {text}"
        else:
            display = text
        return True, display

    def _rx_loop(self) -> None:
        while not self._stop.is_set():
            ctrl = self._ctrl
            if ctrl is None:
                break
            try:
                ready, _, _ = select.select([ctrl], [], [], 0.5)
                if not ready:
                    continue
                chunk = ctrl.recv(4096)
            except (OSError, socket.timeout):
                continue
            if not chunk:
                time.sleep(0.05)
                continue
            term = self._line_term()
            for line in chunk.decode("ascii", errors="replace").split(term):
                line = line.strip()
                if not line:
                    continue
                if line.startswith("STATUS"):
                    self._on_rx(f"[CRDOP RX {self.device_id}] {line}")


class AudioDummyBackend(DeviceBackend):
    """Acoustic bench dummy — loopback, ALSA sniff, or M25 host TCP."""

    backend_type = "audio-dummy"

    def __init__(
        self,
        cfg: DeviceBackendConfig,
        on_rx: RxFn,
        log: Optional[LogFn] = None,
    ) -> None:
        self.device_id = cfg.device_id
        self._cfg = cfg
        self._on_rx = on_rx
        self._log = log or (lambda _m: None)
        self._ctrl: Optional[socket.socket] = None
        self._data: Optional[socket.socket] = None
        self._thread: Optional[threading.Thread] = None
        self._stop = threading.Event()
        self._lock = threading.RLock()
        self._mycall = ""
        self._connected = False
        self._engine = None
        self.status = "closed"

    def _import_engine(self):
        import sys
        from pathlib import Path

        lib = Path(__file__).resolve().parents[1] / "crdop" / "lib"
        if str(lib) not in sys.path:
            sys.path.insert(0, str(lib))
        from acoustic_engine import AcousticEngine  # noqa: WPS433
        from sound_proxy import SoundConfig  # noqa: WPS433

        sound = SoundConfig(
            capture=self._cfg.audio_capture or "default",
            playback=self._cfg.audio_playback or self._cfg.audio_capture or "default",
            sample_rate=self._cfg.audio_sample_rate,
        )
        return AcousticEngine(sample_rate=self._cfg.audio_sample_rate, sound=sound)

    def open(self) -> bool:
        mode = (self._cfg.audio_mode or "loopback").lower()
        if mode == "host":
            host = "127.0.0.1"
            port = self._cfg.audio_host_port
            ctrl = None
            data = None
            try:
                ctrl = socket.create_connection((host, port), timeout=3.0)
                ctrl.settimeout(0.5)
                data = socket.create_connection((host, port + 1), timeout=3.0)
                data.settimeout(0.5)
            except OSError as exc:
                if ctrl is not None:
                    try:
                        ctrl.close()
                    except OSError:
                        pass
                self.status = "error-connect"
                self._log(f"{self.device_id}: audio-dummy host connect failed: {exc}")
                return False
            self._ctrl = ctrl
            self._data = data
            self._stop.clear()
            self._thread = threading.Thread(
                target=self._host_rx_loop,
                name=f"audio-dummy-{self.device_id}",
                daemon=True,
            )
            self._thread.start()
            self.status = "open"
            self._log(f"{self.device_id}: audio-dummy host {host}:{port}")
            return True

        try:
            self._engine = self._import_engine()
        except Exception as exc:
            self.status = "error-engine"
            self._log(f"{self.device_id}: audio engine load failed: {exc}")
            return False

        if mode == "alsa" and self._cfg.audio_capture:
            self._stop.clear()
            self._thread = threading.Thread(
                target=self._alsa_sniff_loop,
                name=f"audio-sniff-{self.device_id}",
                daemon=True,
            )
            self._thread.start()
        self.status = "open"
        self._log(f"{self.device_id}: audio-dummy mode={mode}")
        return True

    def close(self) -> None:
        self._stop.set()
        if self._thread is not None:
            self._thread.join(timeout=2.0)
            self._thread = None
        for sock in (self._ctrl, self._data):
            if sock is not None:
                try:
                    sock.close()
                except OSError:
                    pass
        self._ctrl = None
        self._data = None
        self._connected = False
        self.status = "closed"

    def _host_cmd(self, text: str) -> str:
        if self._ctrl is None:
            return ""
        payload = (text.rstrip("\n") + "\n").encode("ascii", errors="replace")
        with self._lock:
            self._ctrl.sendall(payload)
            buf = b""
            deadline = time.time() + 2.0
            while time.time() < deadline:
                try:
                    chunk = self._ctrl.recv(4096)
                except socket.timeout:
                    continue
                if not chunk:
                    break
                buf += chunk
                if b"\n" in buf:
                    line, _ = buf.split(b"\n", 1)
                    return line.decode("ascii", errors="replace").strip()
        return ""

    def attach_session(self, mycall: str) -> bool:
        self._mycall = mycall.upper()
        if self._ctrl is not None:
            for cmd in (
                "INITIALIZE",
                "PROTOCOLMODE KISS",
                f"MYCALL {self._mycall}",
                "LISTEN TRUE",
            ):
                reply = self._host_cmd(cmd)
                self._log(f"{self.device_id}: {cmd} → {reply or '(no reply)'}")
        self._connected = True
        self.status = "ready"
        return True

    def detach_session(self) -> None:
        self._connected = False
        if self._ctrl is not None:
            self.status = "open"
        else:
            self.status = "closed"

    def transmit(self, src: str, dst: str, text: str, ax25_ui: bool) -> tuple[bool, str]:
        if not self._connected:
            return False, "audio-dummy not ready"
        payload = text.encode("utf-8")
        if len(payload) > MAX_PAYLOAD:
            return False, "payload too long"

        if self._engine is not None:
            pcm = self._engine.encode_ax25_ui(src, dst, text)
            rep = self._engine.analyze_pcm(pcm)
            for line in rep.decode_lines:
                self._on_rx(f"[AUDIO RX {self.device_id}] {line}")
            display = f"[AX25 UI {src}>{dst}] {text}" if ax25_ui else text
            return True, display

        if self._data is None:
            return False, "no data channel"
        import sys
        from pathlib import Path

        lib = Path(__file__).resolve().parents[1] / "crdop" / "lib"
        if str(lib) not in sys.path:
            sys.path.insert(0, str(lib))
        from ax25_codec import ax25_build_ui  # noqa: WPS433

        body = ax25_build_ui(src, dst, payload)
        if len(body) >= 2:
            body = body[:-2]
        try:
            with self._lock:
                self._data.sendall(body)
        except OSError as exc:
            return False, f"tx failed: {exc}"
        display = f"[AX25 UI {src}>{dst}] {text}" if ax25_ui else text
        return True, display

    def _alsa_sniff_loop(self) -> None:
        import sys
        from pathlib import Path

        lib = Path(__file__).resolve().parents[1] / "crdop" / "lib"
        if str(lib) not in sys.path:
            sys.path.insert(0, str(lib))
        from sound_proxy import SoundProxy  # noqa: WPS433

        if self._engine is None:
            return
        proxy = SoundProxy(self._engine.sound)

        def on_pcm(chunk: bytes) -> None:
            rep = self._engine.analyze_pcm(chunk)
            for line in rep.decode_lines:
                self._on_rx(f"[SNIFF {self.device_id}] {line}")

        try:
            proxy.sniff_loop(chunk_symbols=40, on_pcm=on_pcm, stop=self._stop)
        except Exception as exc:
            self._log(f"{self.device_id}: sniff error: {exc}")

    def _host_rx_loop(self) -> None:
        while not self._stop.is_set():
            if self._ctrl is None:
                break
            try:
                ready, _, _ = select.select([self._ctrl], [], [], 0.5)
                if not ready:
                    continue
                chunk = self._ctrl.recv(4096)
            except (OSError, socket.timeout):
                continue
            if not chunk:
                time.sleep(0.05)
                continue
            for line in chunk.decode("ascii", errors="replace").split("\n"):
                line = line.strip()
                if line.startswith("STATUS"):
                    self._on_rx(f"[AUDIO RX {self.device_id}] {line}")


def parse_device_spec(device_id: str, spec: str, cp, cfg_defaults: dict) -> DeviceBackendConfig:
    """Build backend config from [devices] value and optional [device.<id>]."""
    dev = DeviceBackendConfig(device_id=device_id)
    dev.hardware = registry_hardware(device_id, cfg_defaults.get("hardware", "tncs"))
    dev.backend_type = registry_backend(device_id)

    section = f"device.{device_id}"
    sec_opts: dict[str, str] = {}
    if cp.has_section(section):
        sec_opts = {k: cp.get(section, k) for k in cp.options(section)}

    if sec_opts.get("hardware"):
        dev.hardware = sec_opts["hardware"]
    if sec_opts.get("backend"):
        dev.backend_type = sec_opts["backend"]

    spec = (spec or "").strip()
    dev.device_spec = spec

    if spec.startswith("baycom:"):
        dev.backend_type = "baycom-kiss"
        dev.baycom_modem = spec.split(":", 1)[1].strip() or "a"
        dev.hardware = "modems"
    elif spec.startswith("max25-bcpr:") or spec.startswith("bcpr:"):
        # Userspace SER12 — product face max25-bcpr; device id remains max25e0
        dev.backend_type = "max25-bcpr-kiss"
        tag = spec.split(":", 1)[1].strip() or "bc0"
        dev.max25_bcpr_device = tag
        dev.bcpr_device = tag  # legacy alias
        dev.hardware = "modems"
        if not sec_opts.get("kiss_link"):
            dev.kiss_link = f"/tmp/max25-bcpr/kiss-{tag}"
    elif spec.startswith("crdop:"):
        dev.backend_type = "crdop-tcp"
        dev.crdop_profile = spec.split(":", 1)[1].strip() or "default"
        dev.hardware = "soft-modems"
    elif spec.startswith("audio:"):
        dev.backend_type = "audio-dummy"
        dev.audio_mode = spec.split(":", 1)[1].strip() or "loopback"
        dev.hardware = "acoustic-bench"
    elif spec.startswith("/") or spec.startswith("dev:"):
        if dev.backend_type in ("baycom-kiss",):
            dev.kiss_link = spec
        else:
            dev.serial_device = spec.removeprefix("dev:")
            if dev.backend_type == "crdop-tcp":
                dev.backend_type = "kiss-serial"
    elif spec:
        dev.serial_device = spec

    if sec_opts.get("kiss_link"):
        dev.kiss_link = sec_opts["kiss_link"]
    if dev.backend_type in ("max25-bcpr-kiss", "bcpr-kiss") and dev.kiss_link:
        dev.kiss_link = normalize_max25_bcpr_path(dev.kiss_link)
    if sec_opts.get("modem"):
        dev.baycom_modem = sec_opts["modem"]
    if sec_opts.get("baycom_ini"):
        dev.baycom_ini = sec_opts["baycom_ini"]
    if sec_opts.get("max25_bcpr_ini") or sec_opts.get("bcpr_ini"):
        ini_path = sec_opts.get("max25_bcpr_ini") or sec_opts.get("bcpr_ini") or ""
        dev.max25_bcpr_ini = ini_path
        dev.bcpr_ini = ini_path
    if sec_opts.get("max25_bcpr_device") or sec_opts.get("bcpr_device"):
        tag = sec_opts.get("max25_bcpr_device") or sec_opts.get("bcpr_device") or ""
        dev.max25_bcpr_device = tag
        dev.bcpr_device = tag
    if sec_opts.get("ipv4"):
        dev.ipv4 = sec_opts["ipv4"].strip()
    if sec_opts.get("ipv6"):
        dev.ipv6 = sec_opts["ipv6"].strip()
    if sec_opts.get("host"):
        dev.crdop_host = sec_opts["host"]
    if sec_opts.get("port"):
        dev.crdop_port = _spec_int(sec_opts["port"], dev.crdop_port)
    if sec_opts.get("listen"):
        dev.crdop_listen = sec_opts["listen"].lower() in ("1", "yes", "true", "on")
    if sec_opts.get("mode"):
        dev.audio_mode = sec_opts["mode"]
    if sec_opts.get("capture"):
        dev.audio_capture = sec_opts["capture"]
    if sec_opts.get("playback"):
        dev.audio_playback = sec_opts["playback"]
    if sec_opts.get("sample_rate"):
        dev.audio_sample_rate = _spec_int(sec_opts["sample_rate"], dev.audio_sample_rate)
    if sec_opts.get("host_port"):
        dev.audio_host_port = _spec_int(sec_opts["host_port"], dev.audio_host_port)

    serial_sec = f"serial.{device_id}"
    if cp.has_section(serial_sec):
        if cp.has_option(serial_sec, "device"):
            dev.serial_device = cp.get(serial_sec, "device")
        if cp.has_option(serial_sec, "baud"):
            dev.serial_baud = cp.getint(serial_sec, "baud")
        if cp.has_option(serial_sec, "line"):
            dev.serial_line = cp.get(serial_sec, "line")
        if cp.has_option(serial_sec, "dtr_rts"):
            dev.serial_dtr_rts = cp.get(serial_sec, "dtr_rts")
        if cp.has_option(serial_sec, "kiss_entry"):
            dev.serial_kiss_entry = cp.get(serial_sec, "kiss_entry")

    # max25e0 family: hardcoded host addresses (forks inherit from root max25e0)
    if device_id == "max25e0" or device_id.startswith("max25e0:"):
        root_opts: dict[str, str] = {}
        if device_id != "max25e0" and cp.has_section("device.max25e0"):
            root_opts = {k: cp.get("device.max25e0", k) for k in cp.options("device.max25e0")}
        if not dev.ipv4:
            dev.ipv4 = (root_opts.get("ipv4") or "").strip() or MAX25E0_DEFAULT_IPV4
        if not dev.ipv6:
            dev.ipv6 = (root_opts.get("ipv6") or "").strip() or MAX25E0_DEFAULT_IPV6

    return dev


def create_backend(
    dev_cfg: DeviceBackendConfig,
    root: str,
    on_rx: RxFn,
    log: Optional[LogFn] = None,
    prefix: Optional[str] = None,
    on_invalid: Optional[InvalidFn] = None,
) -> DeviceBackend:
    kind = dev_cfg.backend_type or registry_backend(dev_cfg.device_id)
    if kind == "kiss-serial":
        return KissSerialBackend(dev_cfg, root, on_rx, log, prefix=prefix, on_invalid=on_invalid)
    if kind == "baycom-kiss":
        return BayComKissBackend(dev_cfg, on_rx, log)
    if kind in ("max25-bcpr-kiss", "bcpr-kiss"):
        return Max25BcprKissBackend(dev_cfg, on_rx, log)
    if kind == "kiss-raw-serial":
        return KissRawSerialBackend(dev_cfg, root, on_rx, log, prefix=prefix)
    if kind == "crdop-tcp":
        return CrdopTcpBackend(dev_cfg, on_rx, log)
    if kind == "audio-dummy":
        return AudioDummyBackend(dev_cfg, on_rx, log)
    return KissSerialBackend(dev_cfg, root, on_rx, log, prefix=prefix)


def backend_needs_stack(kind: str) -> bool:
    return kind in (
        "kiss-serial",
        "baycom-kiss",
        "max25-bcpr-kiss", "bcpr-kiss",
        "kiss-raw-serial",
        "crdop-tcp",
        "audio-dummy",
    )


def backend_serial_label(backend: Optional[DeviceBackend]) -> str:
    if backend is None:
        return "n/a"
    return backend.status


# Legacy alias (tests / transitional)
BcprKissBackend = Max25BcprKissBackend