""" 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.].""" 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