# Copyright © 2025-26 l5yth & contributors # # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. """Build Meshtastic interface objects from caller-supplied target strings.""" from __future__ import annotations import sys from typing import TYPE_CHECKING from .. import config from ..connection import parse_ble_target from .targets import _DEFAULT_TCP_TARGET, _parse_network_target if TYPE_CHECKING: # pragma: no cover - import only used for type checking from meshtastic.ble_interface import BLEInterface as _BLEInterface # All cached interface classes live on the parent package # (``data.mesh_ingestor.interfaces``). Tests set them via # ``monkeypatch.setattr(mesh, "BLEInterface", ...)`` and the package proxy # routes those writes through to ``interfaces``; keeping a duplicate global on # this submodule would cache the wrong value across tests because # ``monkeypatch`` only restores attributes it set. The ``__init__.py`` # re-resolves ``SerialInterface``/``TCPInterface`` from ``meshtastic.*`` at # package-load time and assigns them to package-level attributes. class _DummySerialInterface: """In-memory replacement for ``meshtastic.serial_interface.SerialInterface``.""" def __init__(self) -> None: self.nodes: dict = {} def close(self) -> None: # pragma: no cover - nothing to close """No-op: the dummy interface holds no resources to release.""" pass class NoAvailableMeshInterface(RuntimeError): """Raised when no default mesh interface can be created.""" def _load_ble_interface(): """Return :class:`meshtastic.ble_interface.BLEInterface` when available. Returns: The resolved BLE interface class. Raises: RuntimeError: If the BLE dependencies are not installed. """ pkg = sys.modules.get("data.mesh_ingestor.interfaces") pkg_ble = getattr(pkg, "BLEInterface", None) if pkg is not None else None if pkg_ble is not None: return pkg_ble try: from meshtastic.ble_interface import BLEInterface as _resolved_interface except ImportError as exc: # pragma: no cover - exercised in non-BLE envs raise RuntimeError( "BLE interface requested but the Meshtastic BLE dependencies are not installed. " "Install the 'meshtastic[ble]' extra to enable BLE support." ) from exc if pkg is not None: setattr(pkg, "BLEInterface", _resolved_interface) for module_name in ("data.mesh_ingestor", "data.mesh"): mesh_module = sys.modules.get(module_name) if mesh_module is not None: setattr(mesh_module, "BLEInterface", _resolved_interface) return _resolved_interface def _create_serial_interface(port: str) -> tuple[object, str]: """Return an appropriate mesh interface for ``port``. Parameters: port: User-supplied port string which may represent serial, BLE or TCP. Returns: ``(interface, resolved_target)`` describing the created interface. """ pkg = sys.modules["data.mesh_ingestor.interfaces"] port_value = (port or "").strip() if port_value.lower() in {"", "mock", "none", "null", "disabled"}: config._debug_log( "Using dummy serial interface", context="interfaces.serial", port=port_value, ) return _DummySerialInterface(), "mock" ble_target = parse_ble_target(port_value) if ble_target: # Determine if it's a MAC address or UUID address_type = "MAC" if ":" in ble_target else "UUID" config._debug_log( "Using BLE interface", context="interfaces.ble", address=ble_target, address_type=address_type, ) return _load_ble_interface()(address=ble_target), ble_target network_target = _parse_network_target(port_value) if network_target: host, tcp_port = network_target config._debug_log( "Using TCP interface", context="interfaces.tcp", host=host, port=tcp_port, ) # Resolve via the package so test fakes installed via ``sys.modules`` # patches at ``meshtastic.tcp_interface`` propagate when interfaces # was imported earlier. tcp_cls = getattr(pkg, "TCPInterface", None) return ( tcp_cls(hostname=host, portNumber=tcp_port), f"tcp://{host}:{tcp_port}", ) config._debug_log( "Using serial interface", context="interfaces.serial", port=port_value, ) serial_cls = getattr(pkg, "SerialInterface", None) return serial_cls(devPath=port_value), port_value def _create_default_interface() -> tuple[object, str]: """Attempt to create the default mesh interface, raising on failure. Returns: ``(interface, resolved_target)`` for the discovered connection. Raises: NoAvailableMeshInterface: When no usable connection can be created. """ # Resolve via the package surface so that monkeypatches against the # backward-compat aliases (``mesh._default_serial_targets``, # ``mesh._create_serial_interface``) propagate at call time. pkg = sys.modules["data.mesh_ingestor.interfaces"] default_serial_targets = pkg._default_serial_targets create_serial = pkg._create_serial_interface errors: list[tuple[str, Exception]] = [] for candidate in default_serial_targets(): try: return create_serial(candidate) except Exception as exc: # pragma: no cover - hardware dependent errors.append((candidate, exc)) config._debug_log( "Failed to open serial candidate", context="interfaces.auto_discovery", target=candidate, error_class=exc.__class__.__name__, error_message=str(exc), ) try: return create_serial(_DEFAULT_TCP_TARGET) except Exception as exc: # pragma: no cover - network dependent errors.append((_DEFAULT_TCP_TARGET, exc)) config._debug_log( "Failed to open TCP fallback", context="interfaces.auto_discovery", target=_DEFAULT_TCP_TARGET, error_class=exc.__class__.__name__, error_message=str(exc), ) if errors: summary = "; ".join(f"{target}: {error}" for target, error in errors) raise NoAvailableMeshInterface( f"no mesh interface available ({summary})" ) from errors[-1][1] raise NoAvailableMeshInterface( # pragma: no cover - defensive only "no mesh interface available" )