diff options
Diffstat (limited to 'stacks/daemon/device_backends.py')
| -rw-r--r-- | stacks/daemon/device_backends.py | 1076 |
1 files changed, 1076 insertions, 0 deletions
diff --git a/stacks/daemon/device_backends.py b/stacks/daemon/device_backends.py new file mode 100644 index 0000000..b4c6c29 --- /dev/null +++ b/stacks/daemon/device_backends.py @@ -0,0 +1,1076 @@ +""" +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 |