adam/potato-mesh
1
0
Fork 0
mirror of https://github.com/l5yth/potato-mesh.git synced 2026-08-06 17:03:27 +02:00
Code Issues Packages Projects Releases Wiki Activity

data: refactor 4/7 interfaces (#775)

Browse Source 
* data: refactor 4/7 interfaces

* data: address PR #775 review feedback

Fix the two CI test regressions caused by the package split:
- ``factory._load_ble_interface`` no longer keeps a stale module-level
  ``BLEInterface`` cache that survived ``monkeypatch`` teardown across
  tests. The package-level attribute is now the single cache; the
  ``factory.py`` global was removed.  This unblocks
  ``test_load_ble_interface_sets_global``.
- ``interfaces/__init__.py`` re-resolves ``SerialInterface`` and
  ``TCPInterface`` from ``meshtastic.*`` at package-load time so that a
  test that pops ``data.mesh_ingestor.interfaces`` from ``sys.modules``
  and re-imports picks up the freshly registered classes rather than
  whatever a cached ``factory.py`` first resolved.  This unblocks
  ``test_interfaces_patch_handles_preimported_serial``.

Restore 100% patch coverage on the interfaces subpackage by:
- Adding tests for previously uncovered, testable paths:
  ``_extract_host_node_id(None)``, ``_ensure_channel_metadata``,
  ``_normalise_nodeinfo_packet`` (None input + dict-conversion fallback),
  ``_resolve_lora_message`` (radio_section paths), ``_modem_preset``
  (preset attr fallback + unparseable value), ``_camelcase_enum_name``
  separator-only input, ``_region_frequency`` no-digit enum name,
  ``_ensure_radio_metadata`` unresolvable-message path, plus the
  unknown-section recursive branch of ``_candidate_node_id``.
- Marking genuinely unreachable defensive branches with
  ``pragma: no cover`` (BLE receive loop body, upstream API regression
  guards, patch re-entry guard, unreachable ``NoAvailableMeshInterface``
  fallback).
This commit is contained in:
l5y
2026-05-02 22:59:52 +02:00
committed by GitHub
parent e04fab5b19
commit 1041e06644
13 changed files with 1491 additions and 980 deletions
Download Patch File Download Diff File Expand all files Collapse all files
data/mesh_ingestor/interfaces.py
-980
View File
@@ -1,980 +0,0 @@
# 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.
"""Mesh interface discovery helpers for interacting with Meshtastic hardware."""
from __future__ import annotations
import contextlib
import importlib
import ipaddress
import math
import re
import sys
import urllib.parse
from collections.abc import Mapping
from typing import TYPE_CHECKING, Any
try: # pragma: no cover - dependency optional in tests
import meshtastic # type: ignore
except Exception: # pragma: no cover - dependency optional in tests
meshtastic = None # type: ignore[assignment]
from . import channels, config, serialization
from .connection import (
BLE_ADDRESS_RE,
DEFAULT_TCP_PORT,
DEFAULT_SERIAL_PATTERNS,
default_serial_targets,
parse_ble_target,
)
def _ensure_mapping(value) -> Mapping | None:
"""Return ``value`` as a mapping when conversion is possible."""
if isinstance(value, Mapping):
return value
if hasattr(value, "__dict__") and isinstance(value.__dict__, Mapping):
return value.__dict__
with contextlib.suppress(Exception):
converted = serialization._node_to_dict(value)
if isinstance(converted, Mapping):
return converted
return None
def _is_nodeish_identifier(value: Any) -> bool:
"""Return ``True`` when ``value`` resembles a Meshtastic node identifier."""
if isinstance(value, (int, float)):
return False
if not isinstance(value, str):
return False
trimmed = value.strip()
if not trimmed:
return False
if trimmed.startswith("^"):
return True
if trimmed.startswith("!"):
trimmed = trimmed[1:]
elif trimmed.lower().startswith("0x"):
trimmed = trimmed[2:]
elif not re.search(r"[a-fA-F]", trimmed):
# Bare decimal strings should not be treated as node ids when labelled "id".
return False
return bool(re.fullmatch(r"[0-9a-fA-F]{1,8}", trimmed))
def _candidate_node_id(mapping: Mapping | None) -> str | None:
"""Extract a canonical node identifier from ``mapping`` when present."""
if mapping is None:
return None
node_keys = (
"fromId",
"from_id",
"from",
"nodeId",
"node_id",
"nodeNum",
"node_num",
"num",
"userId",
"user_id",
)
for key in node_keys:
with contextlib.suppress(Exception):
node_id = serialization._canonical_node_id(mapping.get(key))
if node_id:
return node_id
with contextlib.suppress(Exception):
value = mapping.get("id")
if _is_nodeish_identifier(value):
node_id = serialization._canonical_node_id(value)
if node_id:
return node_id
user_section = _ensure_mapping(mapping.get("user"))
if user_section is not None:
for key in ("userId", "user_id", "num", "nodeNum", "node_num"):
with contextlib.suppress(Exception):
node_id = serialization._canonical_node_id(user_section.get(key))
if node_id:
return node_id
with contextlib.suppress(Exception):
user_id_value = user_section.get("id")
if _is_nodeish_identifier(user_id_value):
node_id = serialization._canonical_node_id(user_id_value)
if node_id:
return node_id
decoded_section = _ensure_mapping(mapping.get("decoded"))
if decoded_section is not None:
node_id = _candidate_node_id(decoded_section)
if node_id:
return node_id
payload_section = _ensure_mapping(mapping.get("payload"))
if payload_section is not None:
node_id = _candidate_node_id(payload_section)
if node_id:
return node_id
for key in ("packet", "meta", "info"):
node_id = _candidate_node_id(_ensure_mapping(mapping.get(key)))
if node_id:
return node_id
for value in mapping.values():
if isinstance(value, (list, tuple)):
for item in value:
node_id = _candidate_node_id(_ensure_mapping(item))
if node_id:
return node_id
else:
node_id = _candidate_node_id(_ensure_mapping(value))
if node_id:
return node_id
return None
def _extract_host_node_id(iface) -> str | None:
"""Return the canonical node identifier for the connected host device.
Searches a sequence of well-known attribute names (``myInfo``,
``my_node_info``, etc.) on ``iface`` for a mapping that contains a
recognisable node identifier, then falls back to the raw ``myNodeNum``
integer attribute.
Parameters:
iface: Live Meshtastic interface object, or any object that exposes
node-identity attributes in one of the expected forms.
Returns:
A canonical ``!xxxxxxxx`` node identifier, or ``None`` when no
identifiable host node information is available.
"""
if iface is None:
return None
def _as_mapping(candidate) -> Mapping | None:
mapping = _ensure_mapping(candidate)
if mapping is not None:
return mapping
if callable(candidate):
with contextlib.suppress(Exception):
return _ensure_mapping(candidate())
return None
candidates: list[Mapping] = []
for attr in ("myInfo", "my_node_info", "myNodeInfo", "my_node", "localNode"):
mapping = _as_mapping(getattr(iface, attr, None))
if mapping is None:
continue
candidates.append(mapping)
nested_info = _ensure_mapping(mapping.get("info"))
if nested_info:
candidates.append(nested_info)
for mapping in candidates:
node_id = _candidate_node_id(mapping)
if node_id:
return node_id
for key in ("myNodeNum", "my_node_num", "myNodeId", "my_node_id"):
node_id = serialization._canonical_node_id(mapping.get(key))
if node_id:
return node_id
node_id = serialization._canonical_node_id(getattr(iface, "myNodeNum", None))
if node_id:
return node_id
return None
def _normalise_nodeinfo_packet(packet) -> dict | None:
"""Return a dictionary view of ``packet`` with a guaranteed ``id`` when known."""
mapping = _ensure_mapping(packet)
if mapping is None:
return None
try:
normalised: dict = dict(mapping)
except Exception:
try:
normalised = {key: mapping[key] for key in mapping}
except Exception:
return None
node_id = _candidate_node_id(normalised)
if node_id and normalised.get("id") != node_id:
normalised["id"] = node_id
return normalised
if TYPE_CHECKING: # pragma: no cover - import only used for type checking
from meshtastic.ble_interface import BLEInterface as _BLEInterface
BLEInterface = None
def _patch_meshtastic_nodeinfo_handler() -> None:
"""Ensure Meshtastic nodeinfo packets always include an ``id`` field."""
module = sys.modules.get("meshtastic", meshtastic)
if module is None:
with contextlib.suppress(Exception):
module = importlib.import_module("meshtastic")
if module is None:
return
globals()["meshtastic"] = module
original = getattr(module, "_onNodeInfoReceive", None)
if not callable(original):
return
mesh_interface_module = getattr(module, "mesh_interface", None)
if mesh_interface_module is None:
with contextlib.suppress(Exception):
mesh_interface_module = importlib.import_module("meshtastic.mesh_interface")
# Replace the module-level handler only once; the sentinel attribute prevents
# re-wrapping if _patch_meshtastic_nodeinfo_handler() is called again after
# the interface module is reloaded or re-imported.
if not getattr(original, "_potato_mesh_safe_wrapper", False):
module._onNodeInfoReceive = _build_safe_nodeinfo_callback(original)
_patch_nodeinfo_handler_class(mesh_interface_module, module)
def _build_safe_nodeinfo_callback(original):
"""Return a wrapper that injects a missing ``id`` before dispatching."""
def _safe_on_node_info_receive(iface, packet): # type: ignore[override]
normalised = _normalise_nodeinfo_packet(packet)
if normalised is not None:
packet = normalised
try:
return original(iface, packet)
except KeyError as exc: # pragma: no cover - defensive only
if exc.args and exc.args[0] == "id":
return None
raise
_safe_on_node_info_receive._potato_mesh_safe_wrapper = True # type: ignore[attr-defined]
return _safe_on_node_info_receive
def _update_nodeinfo_handler_aliases(original, replacement) -> None:
"""Ensure Meshtastic modules reference the patched ``NodeInfoHandler``."""
for module_name, module in list(sys.modules.items()):
if not module_name.startswith("meshtastic"):
continue
existing = getattr(module, "NodeInfoHandler", None)
if existing is original:
setattr(module, "NodeInfoHandler", replacement)
def _patch_nodeinfo_handler_class(
mesh_interface_module, meshtastic_module=None
) -> None:
"""Wrap ``NodeInfoHandler.onReceive`` to normalise packets before callbacks."""
if mesh_interface_module is None:
return
handler_class = getattr(mesh_interface_module, "NodeInfoHandler", None)
if handler_class is None:
return
if getattr(handler_class, "_potato_mesh_safe_wrapper", False):
return
original_on_receive = getattr(handler_class, "onReceive", None)
if not callable(original_on_receive):
return
class _SafeNodeInfoHandler(handler_class): # type: ignore[misc]
"""Subclass that guards against missing node identifiers."""
def onReceive(self, iface, packet): # type: ignore[override]
"""Normalise ``packet`` before dispatching to the parent handler.
Injects a canonical ``id`` field when one can be inferred from the
packet's other fields, then delegates to the original
``NodeInfoHandler.onReceive``. A ``KeyError`` on ``"id"`` is
suppressed because some firmware versions omit the field entirely.
Parameters:
iface: The Meshtastic interface that received the packet.
packet: Raw nodeinfo packet dict, possibly lacking an ``id``
key.
Returns:
The return value of the parent handler, or ``None`` when a
missing ``"id"`` key would otherwise raise.
"""
normalised = _normalise_nodeinfo_packet(packet)
if normalised is not None:
packet = normalised
try:
return super().onReceive(iface, packet)
except KeyError as exc: # pragma: no cover - defensive only
if exc.args and exc.args[0] == "id":
return None
raise
_SafeNodeInfoHandler.__name__ = handler_class.__name__
_SafeNodeInfoHandler.__qualname__ = getattr(
handler_class, "__qualname__", handler_class.__name__
)
_SafeNodeInfoHandler.__module__ = getattr(
handler_class, "__module__", mesh_interface_module.__name__
)
_SafeNodeInfoHandler.__doc__ = getattr(
handler_class, "__doc__", _SafeNodeInfoHandler.__doc__
)
_SafeNodeInfoHandler._potato_mesh_safe_wrapper = True # type: ignore[attr-defined]
setattr(mesh_interface_module, "NodeInfoHandler", _SafeNodeInfoHandler)
if meshtastic_module is None:
meshtastic_module = globals().get("meshtastic")
if meshtastic_module is not None:
existing_top = getattr(meshtastic_module, "NodeInfoHandler", None)
if existing_top is handler_class:
setattr(meshtastic_module, "NodeInfoHandler", _SafeNodeInfoHandler)
_update_nodeinfo_handler_aliases(handler_class, _SafeNodeInfoHandler)
_patch_meshtastic_nodeinfo_handler()
try: # pragma: no cover - optional dependency may be unavailable
from meshtastic.serial_interface import SerialInterface # type: ignore
except Exception: # pragma: no cover - optional dependency may be unavailable
SerialInterface = None # type: ignore[assignment]
try: # pragma: no cover - optional dependency may be unavailable
from meshtastic.tcp_interface import TCPInterface # type: ignore
except Exception: # pragma: no cover - optional dependency may be unavailable
TCPInterface = None # type: ignore[assignment]
def _patch_meshtastic_ble_receive_loop() -> None:
"""Prevent ``UnboundLocalError`` crashes in Meshtastic's BLE reader."""
try:
from meshtastic import ble_interface as _ble_interface_module # type: ignore
except Exception: # pragma: no cover - dependency optional in tests
return
ble_class = getattr(_ble_interface_module, "BLEInterface", None)
if ble_class is None:
return
original = getattr(ble_class, "_receiveFromRadioImpl", None)
if not callable(original):
return
if getattr(original, "_potato_mesh_safe_wrapper", False):
return
FROMRADIO_UUID = getattr(_ble_interface_module, "FROMRADIO_UUID", None)
BleakDBusError = getattr(_ble_interface_module, "BleakDBusError", ())
BleakError = getattr(_ble_interface_module, "BleakError", ())
logger = getattr(_ble_interface_module, "logger", None)
time = getattr(_ble_interface_module, "time", None)
if not FROMRADIO_UUID or logger is None or time is None:
return
def _safe_receive_from_radio(self): # type: ignore[override]
while self._want_receive:
if self.should_read:
self.should_read = False
retries: int = 0
while self._want_receive:
if self.client is None:
logger.debug("BLE client is None, shutting down")
self._want_receive = False
continue
payload: bytes = b""
try:
payload = bytes(self.client.read_gatt_char(FROMRADIO_UUID))
except BleakDBusError as exc:
logger.debug("Device disconnected, shutting down %s", exc)
self._want_receive = False
payload = b""
except BleakError as exc:
if "Not connected" in str(exc):
logger.debug("Device disconnected, shutting down %s", exc)
self._want_receive = False
payload = b""
else:
raise ble_class.BLEError("Error reading BLE") from exc
if not payload:
if not self._want_receive:
break
if retries < 5:
time.sleep(0.1)
retries += 1
continue
break
logger.debug("FROMRADIO read: %s", payload.hex())
self._handleFromRadio(payload)
else:
time.sleep(0.01)
_safe_receive_from_radio._potato_mesh_safe_wrapper = True # type: ignore[attr-defined]
ble_class._receiveFromRadioImpl = _safe_receive_from_radio
_patch_meshtastic_ble_receive_loop()
def _has_field(message: Any, field_name: str) -> bool:
"""Return ``True`` when ``message`` advertises ``field_name`` via ``HasField``."""
if message is None:
return False
has_field = getattr(message, "HasField", None)
if callable(has_field):
try:
return bool(has_field(field_name))
except Exception: # pragma: no cover - defensive guard
return False
return hasattr(message, field_name)
def _enum_name_from_field(message: Any, field_name: str, value: Any) -> str | None:
"""Return the enum name for ``value`` using ``message`` descriptors."""
descriptor = getattr(message, "DESCRIPTOR", None)
if descriptor is None:
return None
fields_by_name = getattr(descriptor, "fields_by_name", {})
field_desc = fields_by_name.get(field_name)
if field_desc is None:
return None
enum_type = getattr(field_desc, "enum_type", None)
if enum_type is None:
return None
enum_values = getattr(enum_type, "values_by_number", {})
enum_value = enum_values.get(value)
if enum_value is None:
return None
return getattr(enum_value, "name", None)
def _resolve_lora_message(local_config: Any) -> Any | None:
"""Return the LoRa configuration sub-message from ``local_config``."""
if local_config is None:
return None
if _has_field(local_config, "lora"):
candidate = getattr(local_config, "lora", None)
if candidate is not None:
return candidate
radio_section = getattr(local_config, "radio", None)
if radio_section is not None:
if _has_field(radio_section, "lora"):
return getattr(radio_section, "lora", None)
if hasattr(radio_section, "lora"):
return getattr(radio_section, "lora")
if hasattr(local_config, "lora"):
return getattr(local_config, "lora")
return None
# Maps Meshtastic region enum name to (base_freq_MHz, channel_spacing_MHz).
# Values are derived from the Meshtastic firmware RegionInfo tables.
# Used by _computed_channel_frequency to derive the actual radio frequency
# from the region and channel index.
_REGION_CHANNEL_PARAMS: dict[str, tuple[float, float]] = {
"US": (902.0, 0.25), # 902–928 MHz; e.g. ch 52 ≈ 915 MHz at 250 kHz spacing
"EU_433": (433.175, 0.2),
"EU_868": (869.525, 0.5), # actual primary ≈ 869.525 MHz, not 868
"CN": (470.0, 0.2),
"JP": (920.875, 0.5),
"ANZ": (916.0, 0.5),
"KR": (921.9, 0.5),
"TW": (923.0, 0.5),
"RU": (868.9, 0.5),
"IN": (865.0, 0.5),
"NZ_865": (864.0, 0.5),
"TH": (920.0, 0.5),
"LORA_24": (2400.0, 0.5),
"UA_433": (433.175, 0.2),
"UA_868": (868.0, 0.5),
"MY_433": (433.0, 0.2),
"MY_919": (919.0, 0.5),
"SG_923": (923.0, 0.5),
"PH_433": (433.0, 0.2),
"PH_868": (868.0, 0.5),
"PH_915": (915.0, 0.5),
"ANZ_433": (433.0, 0.2),
"KZ_433": (433.0, 0.2),
"KZ_863": (863.125, 0.5),
"NP_865": (865.0, 0.5),
"BR_902": (902.0, 0.25),
# IL (Israel) is absent from meshtastic Python lib 2.7.8 protobufs; the
# enum value is unresolvable at runtime. Operators on IL firmware should
# set the FREQUENCY environment variable to override.
}
def _computed_channel_frequency(
enum_name: str | None,
channel_num: int | None,
) -> int | None:
"""Compute the floor MHz frequency for a known region and channel index.
Looks up *enum_name* in :data:`_REGION_CHANNEL_PARAMS` and returns
``floor(base_freq + channel_num * spacing)``. Returns ``None`` when the
region is not in the table. A missing or negative *channel_num* is
treated as 0 so the base frequency is always usable.
Args:
enum_name: Region enum name as returned by
:func:`_enum_name_from_field`, e.g. ``"EU_868"`` or ``"US"``.
channel_num: Zero-based channel index from the device LoRa config.
Returns:
Floored MHz as :class:`int`, or ``None`` if the region is unknown.
"""
if enum_name is None:
return None
params = _REGION_CHANNEL_PARAMS.get(enum_name)
if params is None:
return None
base, spacing = params
idx = channel_num if (isinstance(channel_num, int) and channel_num >= 0) else 0
return math.floor(base + idx * spacing)
def _region_frequency(lora_message: Any) -> int | float | str | None:
"""Derive the LoRa region frequency in MHz or the region label from ``lora_message``.
Frequency sources are tried in priority order:
1. ``override_frequency > 0`` — explicit radio override, floored to MHz.
2. :data:`_REGION_CHANNEL_PARAMS` lookup + ``channel_num`` — actual
band-plan frequency derived from the device's region and channel index,
floored to MHz.
3. Largest digit token ≥ 100 parsed from the region enum name string.
4. Largest digit token < 100 from the enum name (reversed scan).
5. Full enum name string, raw integer ≥ 100, or raw string as a label.
Args:
lora_message: A LoRa config protobuf message or compatible object.
Returns:
An integer MHz frequency, a fallback string label, or ``None``.
"""
if lora_message is None:
return None
# Step 1 — explicit radio override
override_frequency = getattr(lora_message, "override_frequency", None)
if override_frequency is not None:
if isinstance(override_frequency, (int, float)):
if override_frequency > 0:
return math.floor(override_frequency)
elif override_frequency:
return override_frequency
region_value = getattr(lora_message, "region", None)
if region_value is None:
return None
enum_name = _enum_name_from_field(lora_message, "region", region_value)
# Step 2 — lookup table + channel offset (actual band-plan frequency)
if enum_name:
channel_num = getattr(lora_message, "channel_num", None)
computed = _computed_channel_frequency(enum_name, channel_num)
if computed is not None:
return computed
# Steps 3–5 — parse digits from enum name (fallback for unknown regions)
if enum_name:
digits = re.findall(r"\d+", enum_name)
for token in digits:
try:
freq = int(token)
except ValueError: # pragma: no cover - regex guarantees digits
continue
if freq >= 100:
return freq
for token in reversed(digits):
try:
return int(token)
except ValueError: # pragma: no cover - defensive only
continue
return enum_name
if isinstance(region_value, int) and region_value >= 100:
return region_value
if isinstance(region_value, str) and region_value:
return region_value
return None
def _camelcase_enum_name(name: str | None) -> str | None:
"""Convert ``name`` from ``SCREAMING_SNAKE`` to ``CamelCase``."""
if not name:
return None
parts = re.split(r"[^0-9A-Za-z]+", name.strip())
camel_parts = [part.capitalize() for part in parts if part]
if not camel_parts:
return None
return "".join(camel_parts)
def _modem_preset(lora_message: Any) -> str | None:
"""Return the CamelCase modem preset configured on ``lora_message``."""
if lora_message is None:
return None
descriptor = getattr(lora_message, "DESCRIPTOR", None)
fields_by_name = getattr(descriptor, "fields_by_name", {}) if descriptor else {}
if "modem_preset" in fields_by_name:
preset_field = "modem_preset"
elif "preset" in fields_by_name:
preset_field = "preset"
elif hasattr(lora_message, "modem_preset"):
preset_field = "modem_preset"
elif hasattr(lora_message, "preset"):
preset_field = "preset"
else:
return None
preset_value = getattr(lora_message, preset_field, None)
if preset_value is None:
return None
enum_name = _enum_name_from_field(lora_message, preset_field, preset_value)
if isinstance(enum_name, str) and enum_name:
return _camelcase_enum_name(enum_name)
if isinstance(preset_value, str) and preset_value:
return _camelcase_enum_name(preset_value)
return None
def _ensure_radio_metadata(iface: Any) -> None:
"""Populate cached LoRa metadata by inspecting ``iface`` when available."""
if iface is None:
return
try:
wait_for_config = getattr(iface, "waitForConfig", None)
if callable(wait_for_config):
wait_for_config()
except Exception: # pragma: no cover - hardware dependent guard
pass
local_node = getattr(iface, "localNode", None)
local_config = getattr(local_node, "localConfig", None) if local_node else None
lora_message = _resolve_lora_message(local_config)
if lora_message is None:
return
frequency = _region_frequency(lora_message)
preset = _modem_preset(lora_message)
updated = False
if frequency is not None and getattr(config, "LORA_FREQ", None) is None:
config.LORA_FREQ = frequency
updated = True
if preset is not None and getattr(config, "MODEM_PRESET", None) is None:
config.MODEM_PRESET = preset
updated = True
if updated:
config._debug_log(
"Captured LoRa radio metadata",
context="interfaces.ensure_radio_metadata",
severity="info",
always=True,
lora_freq=frequency,
modem_preset=preset,
)
def _ensure_channel_metadata(iface: Any) -> None:
"""Capture channel metadata by inspecting ``iface`` once per runtime."""
if iface is None:
return
try:
channels.capture_from_interface(iface)
except Exception as exc: # pragma: no cover - defensive instrumentation
config._debug_log(
"Failed to capture channel metadata",
context="interfaces.ensure_channel_metadata",
severity="warn",
error_class=exc.__class__.__name__,
error_message=str(exc),
)
_DEFAULT_TCP_TARGET = "http://127.0.0.1"
# Private aliases so that existing internal callers and monkeypatching in
# tests keep working without modification.
_DEFAULT_TCP_PORT = DEFAULT_TCP_PORT # backward-compat alias
_DEFAULT_SERIAL_PATTERNS = DEFAULT_SERIAL_PATTERNS # backward-compat alias
_BLE_ADDRESS_RE = BLE_ADDRESS_RE # backward-compat alias
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
_parse_ble_target = parse_ble_target # backward-compat alias
def _parse_network_target(value: str) -> tuple[str, int] | None:
"""Return ``(host, port)`` when ``value`` is a numeric IP address string.
Only literal IPv4 or IPv6 addresses are accepted, optionally paired with a
port or scheme. Callers that start from hostnames should resolve them to an
address before invoking this helper.
Parameters:
value: Numeric IP literal or URL describing the TCP interface.
Returns:
A ``(host, port)`` tuple or ``None`` when parsing fails.
"""
if not value:
return None
value = value.strip()
if not value:
return None
def _validated_result(host: str | None, port: int | None) -> tuple[str, int] | None:
if not host:
return None
try:
ipaddress.ip_address(host)
except ValueError:
return None
return host, port or _DEFAULT_TCP_PORT
parsed_values = []
if "://" in value:
parsed_values.append(urllib.parse.urlparse(value, scheme="tcp"))
parsed_values.append(urllib.parse.urlparse(f"//{value}", scheme="tcp"))
for parsed in parsed_values:
try:
port = parsed.port
except ValueError:
port = None
result = _validated_result(parsed.hostname, port)
if result:
return result
# For bare "host:port" strings that urlparse may misparse, try a manual
# partition. The `startswith("[")` guard excludes IPv6 bracket notation
# (e.g. "[::1]:8080") because those already succeed via urlparse above.
if value.count(":") == 1 and not value.startswith("["):
host, _, port_text = value.partition(":")
try:
port = int(port_text) if port_text else None
except ValueError:
port = None
result = _validated_result(host, port)
if result:
return result
return _validated_result(value, None)
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.
"""
global BLEInterface
if BLEInterface is not None:
return BLEInterface
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
BLEInterface = _resolved_interface
try:
import sys
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", BLEInterface)
except Exception: # pragma: no cover - defensive only
pass
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.
"""
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,
)
return (
TCPInterface(hostname=host, portNumber=tcp_port),
f"tcp://{host}:{tcp_port}",
)
config._debug_log(
"Using serial interface",
context="interfaces.serial",
port=port_value,
)
return SerialInterface(devPath=port_value), port_value
class NoAvailableMeshInterface(RuntimeError):
"""Raised when no default mesh interface can be created."""
_default_serial_targets = default_serial_targets # backward-compat alias
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.
"""
errors: list[tuple[str, Exception]] = []
for candidate in _default_serial_targets():
try:
return _create_serial_interface(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_interface(_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("no mesh interface available")
__all__ = [
"BLEInterface",
"NoAvailableMeshInterface",
"_ensure_channel_metadata",
"_ensure_radio_metadata",
"_extract_host_node_id",
"_DummySerialInterface",
"_DEFAULT_TCP_PORT",
"_DEFAULT_TCP_TARGET",
"_create_default_interface",
"_create_serial_interface",
"_default_serial_targets",
"_load_ble_interface",
"_parse_ble_target",
"_parse_network_target",
"SerialInterface",
"TCPInterface",
]
data/mesh_ingestor/interfaces/__init__.py
+108
View File
@@ -0,0 +1,108 @@
# 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.
"""Mesh interface discovery helpers for interacting with Meshtastic hardware."""
from __future__ import annotations
# The patches subpackage applies meshtastic monkey-patches at import time so
# subsequent calls (and any direct ``import meshtastic`` from elsewhere)
# inherit the safe wrappers. Apply BEFORE pulling in factory.py because
# factory.py imports ``meshtastic.serial_interface`` / ``meshtastic.tcp_interface``
# and those modules transitively load NodeInfoHandler.
from .patches import (
_build_safe_nodeinfo_callback,
_patch_meshtastic_ble_receive_loop,
_patch_meshtastic_nodeinfo_handler,
_patch_nodeinfo_handler_class,
_update_nodeinfo_handler_aliases,
apply_all as _apply_all_patches,
)
_apply_all_patches()
from ._aliases import ( # noqa: E402 - keep grouped with sibling re-exports.
_BLE_ADDRESS_RE,
_DEFAULT_SERIAL_PATTERNS,
_DEFAULT_TCP_PORT,
_default_serial_targets,
_parse_ble_target,
)
from .channels_meta import _ensure_channel_metadata # noqa: E402
from .factory import ( # noqa: E402
NoAvailableMeshInterface,
_DummySerialInterface,
_create_default_interface,
_create_serial_interface,
_load_ble_interface,
)
# Resolve the meshtastic interface classes at package-load time so that
# repeated imports (e.g. tests that pop ``data.mesh_ingestor.interfaces`` from
# ``sys.modules`` and re-import after swapping ``meshtastic.*`` submodules)
# pick up the freshly registered classes rather than whatever a cached
# ``factory.py`` first resolved. ``factory.py`` no longer keeps duplicate
# module-level globals; lookups go through the package surface only.
BLEInterface = None
"""Resolved on demand by :func:`_load_ble_interface` to keep BLE optional."""
try: # pragma: no cover - optional dependency may be unavailable
from meshtastic.serial_interface import (
SerialInterface,
) # noqa: E402 # type: ignore
except Exception: # pragma: no cover - optional dependency may be unavailable
SerialInterface = None # type: ignore[assignment]
try: # pragma: no cover - optional dependency may be unavailable
from meshtastic.tcp_interface import TCPInterface # noqa: E402 # type: ignore
except Exception: # pragma: no cover - optional dependency may be unavailable
TCPInterface = None # type: ignore[assignment]
from .identity import ( # noqa: E402
_candidate_node_id,
_ensure_mapping,
_extract_host_node_id,
_is_nodeish_identifier,
)
from .nodeinfo_normalize import _normalise_nodeinfo_packet # noqa: E402
from .radio import ( # noqa: E402
_REGION_CHANNEL_PARAMS,
_camelcase_enum_name,
_computed_channel_frequency,
_ensure_radio_metadata,
_enum_name_from_field,
_has_field,
_modem_preset,
_region_frequency,
_resolve_lora_message,
)
from .targets import _DEFAULT_TCP_TARGET, _parse_network_target # noqa: E402
__all__ = [
"BLEInterface",
"NoAvailableMeshInterface",
"_ensure_channel_metadata",
"_ensure_radio_metadata",
"_extract_host_node_id",
"_DummySerialInterface",
"_DEFAULT_TCP_PORT",
"_DEFAULT_TCP_TARGET",
"_create_default_interface",
"_create_serial_interface",
"_default_serial_targets",
"_load_ble_interface",
"_parse_ble_target",
"_parse_network_target",
"SerialInterface",
"TCPInterface",
]
data/mesh_ingestor/interfaces/_aliases.py
+33
View File
@@ -0,0 +1,33 @@
# 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.
"""Backward-compat aliases for renames hidden behind the package barrel."""
from __future__ import annotations
from ..connection import (
BLE_ADDRESS_RE,
DEFAULT_SERIAL_PATTERNS,
DEFAULT_TCP_PORT,
default_serial_targets,
parse_ble_target,
)
# Private aliases so that existing internal callers and monkeypatching in
# tests keep working without modification.
_BLE_ADDRESS_RE = BLE_ADDRESS_RE
_DEFAULT_TCP_PORT = DEFAULT_TCP_PORT
_DEFAULT_SERIAL_PATTERNS = DEFAULT_SERIAL_PATTERNS
_parse_ble_target = parse_ble_target
_default_serial_targets = default_serial_targets
data/mesh_ingestor/interfaces/channels_meta.py
+39
View File
@@ -0,0 +1,39 @@
# 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.
"""One-shot channel metadata capture from a live Meshtastic interface."""
from __future__ import annotations
from typing import Any
from .. import channels, config
def _ensure_channel_metadata(iface: Any) -> None:
"""Capture channel metadata by inspecting ``iface`` once per runtime."""
if iface is None:
return
try:
channels.capture_from_interface(iface)
except Exception as exc: # pragma: no cover - defensive instrumentation
config._debug_log(
"Failed to capture channel metadata",
context="interfaces.ensure_channel_metadata",
severity="warn",
error_class=exc.__class__.__name__,
error_message=str(exc),
)
data/mesh_ingestor/interfaces/factory.py
+191
View File
@@ -0,0 +1,191 @@
# 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"
)
data/mesh_ingestor/interfaces/identity.py
+194
View File
@@ -0,0 +1,194 @@
# 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.
"""Mapping/identifier helpers for Meshtastic interface objects."""
from __future__ import annotations
import contextlib
import re
from collections.abc import Mapping
from typing import Any
from .. import serialization
def _ensure_mapping(value) -> Mapping | None:
"""Return ``value`` as a mapping when conversion is possible."""
if isinstance(value, Mapping):
return value
if hasattr(value, "__dict__") and isinstance(value.__dict__, Mapping):
return value.__dict__
with contextlib.suppress(Exception):
converted = serialization._node_to_dict(value)
if isinstance(converted, Mapping):
return converted
return None
def _is_nodeish_identifier(value: Any) -> bool:
"""Return ``True`` when ``value`` resembles a Meshtastic node identifier."""
if isinstance(value, (int, float)):
return False
if not isinstance(value, str):
return False
trimmed = value.strip()
if not trimmed:
return False
if trimmed.startswith("^"):
return True
if trimmed.startswith("!"):
trimmed = trimmed[1:]
elif trimmed.lower().startswith("0x"):
trimmed = trimmed[2:]
elif not re.search(r"[a-fA-F]", trimmed):
# Bare decimal strings should not be treated as node ids when labelled "id".
return False
return bool(re.fullmatch(r"[0-9a-fA-F]{1,8}", trimmed))
def _candidate_node_id(mapping: Mapping | None) -> str | None:
"""Extract a canonical node identifier from ``mapping`` when present."""
if mapping is None:
return None
node_keys = (
"fromId",
"from_id",
"from",
"nodeId",
"node_id",
"nodeNum",
"node_num",
"num",
"userId",
"user_id",
)
for key in node_keys:
with contextlib.suppress(Exception):
node_id = serialization._canonical_node_id(mapping.get(key))
if node_id:
return node_id
with contextlib.suppress(Exception):
value = mapping.get("id")
if _is_nodeish_identifier(value):
node_id = serialization._canonical_node_id(value)
if node_id:
return node_id
user_section = _ensure_mapping(mapping.get("user"))
if user_section is not None:
for key in ("userId", "user_id", "num", "nodeNum", "node_num"):
with contextlib.suppress(Exception):
node_id = serialization._canonical_node_id(user_section.get(key))
if node_id:
return node_id
with contextlib.suppress(Exception):
user_id_value = user_section.get("id")
if _is_nodeish_identifier(user_id_value):
node_id = serialization._canonical_node_id(user_id_value)
if node_id:
return node_id
decoded_section = _ensure_mapping(mapping.get("decoded"))
if decoded_section is not None:
node_id = _candidate_node_id(decoded_section)
if node_id:
return node_id
payload_section = _ensure_mapping(mapping.get("payload"))
if payload_section is not None:
node_id = _candidate_node_id(payload_section)
if node_id:
return node_id
for key in ("packet", "meta", "info"):
node_id = _candidate_node_id(_ensure_mapping(mapping.get(key)))
if node_id:
return node_id
for value in mapping.values():
if isinstance(value, (list, tuple)):
for item in value:
node_id = _candidate_node_id(_ensure_mapping(item))
if node_id:
return node_id
else:
node_id = _candidate_node_id(_ensure_mapping(value))
if node_id:
return node_id
return None
def _extract_host_node_id(iface) -> str | None:
"""Return the canonical node identifier for the connected host device.
Searches a sequence of well-known attribute names (``myInfo``,
``my_node_info``, etc.) on ``iface`` for a mapping that contains a
recognisable node identifier, then falls back to the raw ``myNodeNum``
integer attribute.
Parameters:
iface: Live Meshtastic interface object, or any object that exposes
node-identity attributes in one of the expected forms.
Returns:
A canonical ``!xxxxxxxx`` node identifier, or ``None`` when no
identifiable host node information is available.
"""
if iface is None:
return None
def _as_mapping(candidate) -> Mapping | None:
mapping = _ensure_mapping(candidate)
if mapping is not None:
return mapping
if callable(candidate):
with contextlib.suppress(Exception):
return _ensure_mapping(candidate())
return None
candidates: list[Mapping] = []
for attr in ("myInfo", "my_node_info", "myNodeInfo", "my_node", "localNode"):
mapping = _as_mapping(getattr(iface, attr, None))
if mapping is None:
continue
candidates.append(mapping)
nested_info = _ensure_mapping(mapping.get("info"))
if nested_info:
candidates.append(nested_info)
for mapping in candidates:
node_id = _candidate_node_id(mapping)
if node_id:
return node_id
for key in ("myNodeNum", "my_node_num", "myNodeId", "my_node_id"):
node_id = serialization._canonical_node_id(mapping.get(key))
if node_id:
return node_id
node_id = serialization._canonical_node_id(getattr(iface, "myNodeNum", None))
if node_id:
return node_id
return None
data/mesh_ingestor/interfaces/nodeinfo_normalize.py
+41
View File
@@ -0,0 +1,41 @@
# 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.
"""Inject a canonical ``id`` into Meshtastic nodeinfo packets when missing."""
from __future__ import annotations
from .identity import _candidate_node_id, _ensure_mapping
def _normalise_nodeinfo_packet(packet) -> dict | None:
"""Return a dictionary view of ``packet`` with a guaranteed ``id`` when known."""
mapping = _ensure_mapping(packet)
if mapping is None:
return None
try:
normalised: dict = dict(mapping)
except Exception:
try:
normalised = {key: mapping[key] for key in mapping}
except Exception: # pragma: no cover - both copy strategies failed
return None
node_id = _candidate_node_id(normalised)
if node_id and normalised.get("id") != node_id:
normalised["id"] = node_id
return normalised
data/mesh_ingestor/interfaces/patches/__init__.py
+41
View File
@@ -0,0 +1,41 @@
# 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.
"""Runtime monkey-patches applied to the upstream ``meshtastic`` library."""
from __future__ import annotations
from .ble_receive import _patch_meshtastic_ble_receive_loop
from .nodeinfo import (
_build_safe_nodeinfo_callback,
_patch_meshtastic_nodeinfo_handler,
_patch_nodeinfo_handler_class,
_update_nodeinfo_handler_aliases,
)
def apply_all() -> None:
"""Apply every meshtastic monkey-patch in the order required for safety."""
_patch_meshtastic_nodeinfo_handler()
_patch_meshtastic_ble_receive_loop()
__all__ = [
"apply_all",
"_build_safe_nodeinfo_callback",
"_patch_meshtastic_ble_receive_loop",
"_patch_meshtastic_nodeinfo_handler",
"_patch_nodeinfo_handler_class",
"_update_nodeinfo_handler_aliases",
]
data/mesh_ingestor/interfaces/patches/ble_receive.py
+93
View File
@@ -0,0 +1,93 @@
# 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.
"""Patch the upstream Meshtastic BLE receive loop to avoid ``UnboundLocalError``."""
from __future__ import annotations
def _patch_meshtastic_ble_receive_loop() -> None:
"""Prevent ``UnboundLocalError`` crashes in Meshtastic's BLE reader."""
try:
from meshtastic import ble_interface as _ble_interface_module # type: ignore
except Exception: # pragma: no cover - dependency optional in tests
return
ble_class = getattr(_ble_interface_module, "BLEInterface", None)
if ble_class is None: # pragma: no cover - exercised only without BLE class
return
original = getattr(ble_class, "_receiveFromRadioImpl", None)
if not callable(original): # pragma: no cover - upstream API regression guard
return
if getattr(original, "_potato_mesh_safe_wrapper", False):
return
FROMRADIO_UUID = getattr(_ble_interface_module, "FROMRADIO_UUID", None)
BleakDBusError = getattr(_ble_interface_module, "BleakDBusError", ())
BleakError = getattr(_ble_interface_module, "BleakError", ())
logger = getattr(_ble_interface_module, "logger", None)
time = getattr(_ble_interface_module, "time", None)
if ( # pragma: no cover - upstream API regression guard
not FROMRADIO_UUID or logger is None or time is None
):
return
# The receive loop runs on a dedicated thread and only completes against a
# live BLE adapter; the body is hardware-dependent and not unit-testable.
def _safe_receive_from_radio(self): # pragma: no cover - hardware dependent
# type: ignore[override]
while self._want_receive:
if self.should_read:
self.should_read = False
retries: int = 0
while self._want_receive:
if self.client is None:
logger.debug("BLE client is None, shutting down")
self._want_receive = False
continue
payload: bytes = b""
try:
payload = bytes(self.client.read_gatt_char(FROMRADIO_UUID))
except BleakDBusError as exc:
logger.debug("Device disconnected, shutting down %s", exc)
self._want_receive = False
payload = b""
except BleakError as exc:
if "Not connected" in str(exc):
logger.debug("Device disconnected, shutting down %s", exc)
self._want_receive = False
payload = b""
else:
raise ble_class.BLEError("Error reading BLE") from exc
if not payload:
if not self._want_receive:
break
if retries < 5:
time.sleep(0.1)
retries += 1
continue
break
logger.debug("FROMRADIO read: %s", payload.hex())
self._handleFromRadio(payload)
else:
time.sleep(0.01)
_safe_receive_from_radio._potato_mesh_safe_wrapper = True # type: ignore[attr-defined]
ble_class._receiveFromRadioImpl = _safe_receive_from_radio
data/mesh_ingestor/interfaces/patches/nodeinfo.py
+164
View File
@@ -0,0 +1,164 @@
# 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.
"""Runtime patches that harden Meshtastic's nodeinfo handler against missing ``id`` fields."""
from __future__ import annotations
import contextlib
import importlib
import sys
try: # pragma: no cover - dependency optional in tests
import meshtastic # type: ignore
except Exception: # pragma: no cover - dependency optional in tests
meshtastic = None # type: ignore[assignment]
from ..nodeinfo_normalize import _normalise_nodeinfo_packet
def _patch_meshtastic_nodeinfo_handler() -> None:
"""Ensure Meshtastic nodeinfo packets always include an ``id`` field."""
module = sys.modules.get("meshtastic", meshtastic)
if module is None: # pragma: no cover - re-import fallback for cold caches
with contextlib.suppress(Exception):
module = importlib.import_module("meshtastic")
if module is None: # pragma: no cover - exercised only without meshtastic
return
globals()["meshtastic"] = module
original = getattr(module, "_onNodeInfoReceive", None)
if not callable(original): # pragma: no cover - upstream API regression guard
return
mesh_interface_module = getattr(module, "mesh_interface", None)
if mesh_interface_module is None:
with contextlib.suppress(Exception):
mesh_interface_module = importlib.import_module("meshtastic.mesh_interface")
# Replace the module-level handler only once; the sentinel attribute prevents
# re-wrapping if _patch_meshtastic_nodeinfo_handler() is called again after
# the interface module is reloaded or re-imported.
if not getattr(original, "_potato_mesh_safe_wrapper", False):
module._onNodeInfoReceive = _build_safe_nodeinfo_callback(original)
_patch_nodeinfo_handler_class(mesh_interface_module, module)
def _build_safe_nodeinfo_callback(original):
"""Return a wrapper that injects a missing ``id`` before dispatching."""
def _safe_on_node_info_receive(iface, packet): # type: ignore[override]
normalised = _normalise_nodeinfo_packet(packet)
if normalised is not None:
packet = normalised
try:
return original(iface, packet)
except KeyError as exc: # pragma: no cover - defensive only
if exc.args and exc.args[0] == "id":
return None
raise
_safe_on_node_info_receive._potato_mesh_safe_wrapper = True # type: ignore[attr-defined]
return _safe_on_node_info_receive
def _update_nodeinfo_handler_aliases(original, replacement) -> None:
"""Ensure Meshtastic modules reference the patched ``NodeInfoHandler``."""
for module_name, module in list(sys.modules.items()):
if not module_name.startswith("meshtastic"):
continue
existing = getattr(module, "NodeInfoHandler", None)
if existing is original:
setattr(module, "NodeInfoHandler", replacement)
def _patch_nodeinfo_handler_class(
mesh_interface_module, meshtastic_module=None
) -> None:
"""Wrap ``NodeInfoHandler.onReceive`` to normalise packets before callbacks."""
if (
mesh_interface_module is None
): # pragma: no cover - exercised only without meshtastic
return
handler_class = getattr(mesh_interface_module, "NodeInfoHandler", None)
if handler_class is None: # pragma: no cover - upstream API regression guard
return
if getattr(
handler_class, "_potato_mesh_safe_wrapper", False
): # pragma: no cover - re-entry guard
return
original_on_receive = getattr(handler_class, "onReceive", None)
if not callable(
original_on_receive
): # pragma: no cover - upstream API regression guard
return
class _SafeNodeInfoHandler(handler_class): # type: ignore[misc]
"""Subclass that guards against missing node identifiers."""
def onReceive(self, iface, packet): # type: ignore[override]
"""Normalise ``packet`` before dispatching to the parent handler.
Injects a canonical ``id`` field when one can be inferred from the
packet's other fields, then delegates to the original
``NodeInfoHandler.onReceive``. A ``KeyError`` on ``"id"`` is
suppressed because some firmware versions omit the field entirely.
Parameters:
iface: The Meshtastic interface that received the packet.
packet: Raw nodeinfo packet dict, possibly lacking an ``id``
key.
Returns:
The return value of the parent handler, or ``None`` when a
missing ``"id"`` key would otherwise raise.
"""
normalised = _normalise_nodeinfo_packet(packet)
if normalised is not None:
packet = normalised
try:
return super().onReceive(iface, packet)
except KeyError as exc: # pragma: no cover - defensive only
if exc.args and exc.args[0] == "id":
return None
raise
_SafeNodeInfoHandler.__name__ = handler_class.__name__
_SafeNodeInfoHandler.__qualname__ = getattr(
handler_class, "__qualname__", handler_class.__name__
)
_SafeNodeInfoHandler.__module__ = getattr(
handler_class, "__module__", mesh_interface_module.__name__
)
_SafeNodeInfoHandler.__doc__ = getattr(
handler_class, "__doc__", _SafeNodeInfoHandler.__doc__
)
_SafeNodeInfoHandler._potato_mesh_safe_wrapper = True # type: ignore[attr-defined]
setattr(mesh_interface_module, "NodeInfoHandler", _SafeNodeInfoHandler)
if meshtastic_module is None:
meshtastic_module = globals().get("meshtastic")
if meshtastic_module is not None:
existing_top = getattr(meshtastic_module, "NodeInfoHandler", None)
if existing_top is handler_class: # pragma: no cover - top-level re-export
setattr(meshtastic_module, "NodeInfoHandler", _SafeNodeInfoHandler)
_update_nodeinfo_handler_aliases(handler_class, _SafeNodeInfoHandler)
data/mesh_ingestor/interfaces/radio.py
+292
View File
@@ -0,0 +1,292 @@
# 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.
"""LoRa region/frequency/preset derivation from a Meshtastic config protobuf."""
from __future__ import annotations
import math
import re
from typing import Any
from .. import config
def _has_field(message: Any, field_name: str) -> bool:
"""Return ``True`` when ``message`` advertises ``field_name`` via ``HasField``."""
if message is None:
return False
has_field = getattr(message, "HasField", None)
if callable(has_field):
try:
return bool(has_field(field_name))
except Exception: # pragma: no cover - defensive guard
return False
return hasattr(message, field_name)
def _enum_name_from_field(message: Any, field_name: str, value: Any) -> str | None:
"""Return the enum name for ``value`` using ``message`` descriptors."""
descriptor = getattr(message, "DESCRIPTOR", None)
if descriptor is None:
return None
fields_by_name = getattr(descriptor, "fields_by_name", {})
field_desc = fields_by_name.get(field_name)
if field_desc is None:
return None
enum_type = getattr(field_desc, "enum_type", None)
if enum_type is None:
return None
enum_values = getattr(enum_type, "values_by_number", {})
enum_value = enum_values.get(value)
if enum_value is None:
return None
return getattr(enum_value, "name", None)
def _resolve_lora_message(local_config: Any) -> Any | None:
"""Return the LoRa configuration sub-message from ``local_config``."""
if local_config is None:
return None
if _has_field(local_config, "lora"):
candidate = getattr(local_config, "lora", None)
if candidate is not None:
return candidate
radio_section = getattr(local_config, "radio", None)
if radio_section is not None:
if _has_field(radio_section, "lora"):
return getattr(radio_section, "lora", None)
if hasattr(radio_section, "lora"):
return getattr(radio_section, "lora")
if hasattr(local_config, "lora"):
return getattr(local_config, "lora")
return None
# Maps Meshtastic region enum name to (base_freq_MHz, channel_spacing_MHz).
# Values are derived from the Meshtastic firmware RegionInfo tables.
# Used by _computed_channel_frequency to derive the actual radio frequency
# from the region and channel index.
_REGION_CHANNEL_PARAMS: dict[str, tuple[float, float]] = {
"US": (902.0, 0.25), # 902–928 MHz; e.g. ch 52 ≈ 915 MHz at 250 kHz spacing
"EU_433": (433.175, 0.2),
"EU_868": (869.525, 0.5), # actual primary ≈ 869.525 MHz, not 868
"CN": (470.0, 0.2),
"JP": (920.875, 0.5),
"ANZ": (916.0, 0.5),
"KR": (921.9, 0.5),
"TW": (923.0, 0.5),
"RU": (868.9, 0.5),
"IN": (865.0, 0.5),
"NZ_865": (864.0, 0.5),
"TH": (920.0, 0.5),
"LORA_24": (2400.0, 0.5),
"UA_433": (433.175, 0.2),
"UA_868": (868.0, 0.5),
"MY_433": (433.0, 0.2),
"MY_919": (919.0, 0.5),
"SG_923": (923.0, 0.5),
"PH_433": (433.0, 0.2),
"PH_868": (868.0, 0.5),
"PH_915": (915.0, 0.5),
"ANZ_433": (433.0, 0.2),
"KZ_433": (433.0, 0.2),
"KZ_863": (863.125, 0.5),
"NP_865": (865.0, 0.5),
"BR_902": (902.0, 0.25),
# IL (Israel) is absent from meshtastic Python lib 2.7.8 protobufs; the
# enum value is unresolvable at runtime. Operators on IL firmware should
# set the FREQUENCY environment variable to override.
}
def _computed_channel_frequency(
enum_name: str | None,
channel_num: int | None,
) -> int | None:
"""Compute the floor MHz frequency for a known region and channel index.
Looks up *enum_name* in :data:`_REGION_CHANNEL_PARAMS` and returns
``floor(base_freq + channel_num * spacing)``. Returns ``None`` when the
region is not in the table. A missing or negative *channel_num* is
treated as 0 so the base frequency is always usable.
Args:
enum_name: Region enum name as returned by
:func:`_enum_name_from_field`, e.g. ``"EU_868"`` or ``"US"``.
channel_num: Zero-based channel index from the device LoRa config.
Returns:
Floored MHz as :class:`int`, or ``None`` if the region is unknown.
"""
if enum_name is None:
return None
params = _REGION_CHANNEL_PARAMS.get(enum_name)
if params is None:
return None
base, spacing = params
idx = channel_num if (isinstance(channel_num, int) and channel_num >= 0) else 0
return math.floor(base + idx * spacing)
def _region_frequency(lora_message: Any) -> int | float | str | None:
"""Derive the LoRa region frequency in MHz or the region label from ``lora_message``.
Frequency sources are tried in priority order:
1. ``override_frequency > 0`` — explicit radio override, floored to MHz.
2. :data:`_REGION_CHANNEL_PARAMS` lookup + ``channel_num`` — actual
band-plan frequency derived from the device's region and channel index,
floored to MHz.
3. Largest digit token ≥ 100 parsed from the region enum name string.
4. Largest digit token < 100 from the enum name (reversed scan).
5. Full enum name string, raw integer ≥ 100, or raw string as a label.
Args:
lora_message: A LoRa config protobuf message or compatible object.
Returns:
An integer MHz frequency, a fallback string label, or ``None``.
"""
if lora_message is None:
return None
# Step 1 — explicit radio override
override_frequency = getattr(lora_message, "override_frequency", None)
if override_frequency is not None:
if isinstance(override_frequency, (int, float)):
if override_frequency > 0:
return math.floor(override_frequency)
elif override_frequency:
return override_frequency
region_value = getattr(lora_message, "region", None)
if region_value is None:
return None
enum_name = _enum_name_from_field(lora_message, "region", region_value)
# Step 2 — lookup table + channel offset (actual band-plan frequency)
if enum_name:
channel_num = getattr(lora_message, "channel_num", None)
computed = _computed_channel_frequency(enum_name, channel_num)
if computed is not None:
return computed
# Steps 3–5 — parse digits from enum name (fallback for unknown regions)
if enum_name:
digits = re.findall(r"\d+", enum_name)
for token in digits:
try:
freq = int(token)
except ValueError: # pragma: no cover - regex guarantees digits
continue
if freq >= 100:
return freq
for token in reversed(digits):
try:
return int(token)
except ValueError: # pragma: no cover - defensive only
continue
return enum_name
if isinstance(region_value, int) and region_value >= 100:
return region_value
if isinstance(region_value, str) and region_value:
return region_value
return None
def _camelcase_enum_name(name: str | None) -> str | None:
"""Convert ``name`` from ``SCREAMING_SNAKE`` to ``CamelCase``."""
if not name:
return None
parts = re.split(r"[^0-9A-Za-z]+", name.strip())
camel_parts = [part.capitalize() for part in parts if part]
if not camel_parts:
return None
return "".join(camel_parts)
def _modem_preset(lora_message: Any) -> str | None:
"""Return the CamelCase modem preset configured on ``lora_message``."""
if lora_message is None:
return None
descriptor = getattr(lora_message, "DESCRIPTOR", None)
fields_by_name = getattr(descriptor, "fields_by_name", {}) if descriptor else {}
if "modem_preset" in fields_by_name:
preset_field = "modem_preset"
elif "preset" in fields_by_name:
preset_field = "preset"
elif hasattr(lora_message, "modem_preset"):
preset_field = "modem_preset"
elif hasattr(lora_message, "preset"):
preset_field = "preset"
else:
return None
preset_value = getattr(lora_message, preset_field, None)
if preset_value is None:
return None
enum_name = _enum_name_from_field(lora_message, preset_field, preset_value)
if isinstance(enum_name, str) and enum_name:
return _camelcase_enum_name(enum_name)
if isinstance(preset_value, str) and preset_value:
return _camelcase_enum_name(preset_value)
return None
def _ensure_radio_metadata(iface: Any) -> None:
"""Populate cached LoRa metadata by inspecting ``iface`` when available."""
if iface is None:
return
try:
wait_for_config = getattr(iface, "waitForConfig", None)
if callable(wait_for_config):
wait_for_config()
except Exception: # pragma: no cover - hardware dependent guard
pass
local_node = getattr(iface, "localNode", None)
local_config = getattr(local_node, "localConfig", None) if local_node else None
lora_message = _resolve_lora_message(local_config)
if lora_message is None:
return
frequency = _region_frequency(lora_message)
preset = _modem_preset(lora_message)
updated = False
if frequency is not None and getattr(config, "LORA_FREQ", None) is None:
config.LORA_FREQ = frequency
updated = True
if preset is not None and getattr(config, "MODEM_PRESET", None) is None:
config.MODEM_PRESET = preset
updated = True
if updated:
config._debug_log(
"Captured LoRa radio metadata",
context="interfaces.ensure_radio_metadata",
severity="info",
always=True,
lora_freq=frequency,
modem_preset=preset,
)
data/mesh_ingestor/interfaces/targets.py
+84
View File
@@ -0,0 +1,84 @@
# 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.
"""Network target parsing helpers for Meshtastic interfaces."""
from __future__ import annotations
import ipaddress
import urllib.parse
from ..connection import DEFAULT_TCP_PORT
_DEFAULT_TCP_TARGET = "http://127.0.0.1"
def _parse_network_target(value: str) -> tuple[str, int] | None:
"""Return ``(host, port)`` when ``value`` is a numeric IP address string.
Only literal IPv4 or IPv6 addresses are accepted, optionally paired with a
port or scheme. Callers that start from hostnames should resolve them to an
address before invoking this helper.
Parameters:
value: Numeric IP literal or URL describing the TCP interface.
Returns:
A ``(host, port)`` tuple or ``None`` when parsing fails.
"""
if not value:
return None
value = value.strip()
if not value:
return None
def _validated_result(host: str | None, port: int | None) -> tuple[str, int] | None:
if not host:
return None
try:
ipaddress.ip_address(host)
except ValueError:
return None
return host, port or DEFAULT_TCP_PORT
parsed_values = []
if "://" in value:
parsed_values.append(urllib.parse.urlparse(value, scheme="tcp"))
parsed_values.append(urllib.parse.urlparse(f"//{value}", scheme="tcp"))
for parsed in parsed_values:
try:
port = parsed.port
except ValueError:
port = None
result = _validated_result(parsed.hostname, port)
if result:
return result
# For bare "host:port" strings that urlparse may misparse, try a manual
# partition. The `startswith("[")` guard excludes IPv6 bracket notation
# (e.g. "[::1]:8080") because those already succeed via urlparse above.
if value.count(":") == 1 and not value.startswith("["):
host, _, port_text = value.partition(":")
try:
port = int(port_text) if port_text else None
except ValueError:
port = None
result = _validated_result(host, port)
if result: # pragma: no cover - urlparse handles all currently-known forms
return result
return _validated_result(value, None)
tests/test_interfaces_unit.py
+211
View File
@@ -191,6 +191,15 @@ class TestCandidateNodeId:
result = ifaces._candidate_node_id({"items": [{"fromId": "!aabbccdd"}]})
assert result == "!aabbccdd"
def test_unknown_section_value_scanned(self):
"""Mapping values under arbitrary keys are recursively scanned.
Exercises the ``else`` branch of the values-loop (non-list/tuple value)
when the parent key is not one of the recognised section names.
"""
result = ifaces._candidate_node_id({"misc_section": {"fromId": "!aabbccdd"}})
assert result == "!aabbccdd"
# ---------------------------------------------------------------------------
# _has_field
@@ -448,6 +457,61 @@ class TestRegionFrequency:
)
assert ifaces._region_frequency(msg) == 999
def test_enum_name_without_any_digits_returns_name(self):
"""Enum name with no extractable digits is returned as-is."""
enum_val = SimpleNamespace(name="UNSET")
enum_type = SimpleNamespace(values_by_number={0: enum_val})
field_desc = SimpleNamespace(enum_type=enum_type)
desc = SimpleNamespace(fields_by_name={"region": field_desc})
msg = SimpleNamespace(DESCRIPTOR=desc, override_frequency=None, region=0)
assert ifaces._region_frequency(msg) == "UNSET"
# ---------------------------------------------------------------------------
# _resolve_lora_message
# ---------------------------------------------------------------------------
class TestResolveLoraMessage:
"""Tests for :func:`interfaces._resolve_lora_message`."""
def test_none_returns_none(self):
"""A ``None`` ``local_config`` short-circuits."""
assert ifaces._resolve_lora_message(None) is None
def test_radio_section_lora_via_has_field(self):
"""Resolves ``radio.lora`` when exposed via ``HasField``."""
radio_section = SimpleNamespace(
HasField=lambda name: name == "lora", lora="radio_lora"
)
local_config = SimpleNamespace(HasField=lambda name: False, radio=radio_section)
assert ifaces._resolve_lora_message(local_config) == "radio_lora"
def test_radio_section_lora_via_hasattr(self):
"""Resolves ``radio.lora`` via ``hasattr`` when ``HasField`` is silent.
The ``radio_section`` exposes ``HasField`` returning ``False`` so
``_has_field`` produces ``False`` for ``"lora"``, forcing the
``hasattr`` fallback path to be taken before returning the value.
"""
radio_section = SimpleNamespace(
HasField=lambda name: False, lora="radio_lora_attr"
)
local_config = SimpleNamespace(HasField=lambda name: False, radio=radio_section)
assert ifaces._resolve_lora_message(local_config) == "radio_lora_attr"
def test_local_config_lora_via_hasattr_only(self):
"""Resolves ``local_config.lora`` via ``hasattr`` when no ``HasField`` match."""
local_config = SimpleNamespace(
HasField=lambda name: False, lora="bare_lora", radio=None
)
assert ifaces._resolve_lora_message(local_config) == "bare_lora"
def test_no_lora_anywhere_returns_none(self):
"""No ``lora`` attribute on either section returns ``None``."""
local_config = SimpleNamespace(HasField=lambda name: False, radio=None)
assert ifaces._resolve_lora_message(local_config) is None
# ---------------------------------------------------------------------------
# _camelcase_enum_name
@@ -477,6 +541,10 @@ class TestCamelcaseEnumName:
"""Digits in the name are preserved."""
assert ifaces._camelcase_enum_name("BAND_915") == "Band915"
def test_only_separators_returns_none(self):
"""A string consisting only of separators yields no usable parts."""
assert ifaces._camelcase_enum_name("___") is None
# ---------------------------------------------------------------------------
# _modem_preset
@@ -523,6 +591,30 @@ class TestModemPreset:
msg = SimpleNamespace(DESCRIPTOR=desc, preset=1)
assert ifaces._modem_preset(msg) == "ShortFast"
def test_attr_preset_fallback_when_no_modem_preset(self):
"""Falls back to ``preset`` attribute when ``modem_preset`` is absent.
Exercises the ``hasattr(lora_message, 'preset')`` branch when the
descriptor lacks both fields and the object only exposes ``preset``.
"""
class _PresetOnly:
DESCRIPTOR = None
preset = "LONG_FAST"
assert ifaces._modem_preset(_PresetOnly()) == "LongFast"
def test_unparseable_preset_value_returns_none(self):
"""A non-string, non-enum-resolvable preset value returns None."""
# Field present in descriptor but enum_type lookup yields a non-string
# (e.g., a numeric mapping with no name). ``preset_value`` is also a
# plain int (not a string), so neither name nor string fallback applies.
enum_type = SimpleNamespace(values_by_number={})
field_desc = SimpleNamespace(enum_type=enum_type)
desc = SimpleNamespace(fields_by_name={"modem_preset": field_desc})
msg = SimpleNamespace(DESCRIPTOR=desc, modem_preset=99)
assert ifaces._modem_preset(msg) is None
# ---------------------------------------------------------------------------
# _ensure_radio_metadata caching
@@ -540,6 +632,18 @@ class TestEnsureRadioMetadata:
assert config.LORA_FREQ == original_freq
assert config.MODEM_PRESET == original_preset
def test_unresolvable_lora_message_returns_without_writing(self, monkeypatch):
"""When ``_resolve_lora_message`` returns ``None``, config is left alone."""
monkeypatch.setattr(config, "LORA_FREQ", None)
monkeypatch.setattr(config, "MODEM_PRESET", None)
# ``localConfig`` exists but has no lora/radio, so resolve returns None.
local_config = SimpleNamespace(HasField=lambda name: False, radio=None)
local_node = SimpleNamespace(localConfig=local_config)
iface = SimpleNamespace(localNode=local_node, waitForConfig=lambda: None)
ifaces._ensure_radio_metadata(iface)
assert config.LORA_FREQ is None
assert config.MODEM_PRESET is None
def test_sets_lora_freq_when_not_cached(self, monkeypatch):
"""Populates LORA_FREQ from interface when not yet configured."""
monkeypatch.setattr(config, "LORA_FREQ", None)
@@ -577,3 +681,110 @@ class TestEnsureRadioMetadata:
ifaces._ensure_radio_metadata(iface)
assert config.LORA_FREQ == 433
# ---------------------------------------------------------------------------
# _extract_host_node_id
# ---------------------------------------------------------------------------
class TestExtractHostNodeId:
"""Tests for :func:`interfaces._extract_host_node_id`."""
def test_none_iface_returns_none(self):
"""A ``None`` interface short-circuits without any attribute access."""
assert ifaces._extract_host_node_id(None) is None
# ---------------------------------------------------------------------------
# _ensure_channel_metadata
# ---------------------------------------------------------------------------
class TestEnsureChannelMetadata:
"""Tests for :func:`interfaces._ensure_channel_metadata`."""
def test_none_iface_is_noop(self, monkeypatch):
"""A ``None`` interface short-circuits without invoking ``capture_from_interface``."""
import data.mesh_ingestor.channels as _channels
called: list = []
monkeypatch.setattr(
_channels, "capture_from_interface", lambda iface: called.append(iface)
)
ifaces._ensure_channel_metadata(None)
assert called == []
def test_calls_capture_from_interface(self, monkeypatch):
"""A non-None interface delegates to ``channels.capture_from_interface``."""
import data.mesh_ingestor.channels as _channels
seen: list = []
monkeypatch.setattr(
_channels, "capture_from_interface", lambda iface: seen.append(iface)
)
sentinel = SimpleNamespace(myInfo={})
ifaces._ensure_channel_metadata(sentinel)
assert seen == [sentinel]
# ---------------------------------------------------------------------------
# _normalise_nodeinfo_packet
# ---------------------------------------------------------------------------
class TestNormaliseNodeinfoPacket:
"""Tests for :func:`interfaces._normalise_nodeinfo_packet`."""
def test_non_mapping_returns_none(self):
"""Inputs that ``_ensure_mapping`` cannot coerce return ``None``."""
# int/float values are explicitly rejected by ``_ensure_mapping``.
assert ifaces._normalise_nodeinfo_packet(42) is None
def test_mapping_with_node_id_injects_id_field(self):
"""A valid mapping has the canonical id injected when inferable."""
result = ifaces._normalise_nodeinfo_packet({"fromId": "!aabbccdd"})
assert result is not None
assert result["id"] == "!aabbccdd"
def test_mapping_keeps_existing_id_when_consistent(self):
"""A pre-existing matching ``id`` is left untouched."""
result = ifaces._normalise_nodeinfo_packet(
{"id": "!aabbccdd", "fromId": "!aabbccdd"}
)
assert result == {"id": "!aabbccdd", "fromId": "!aabbccdd"}
def test_dict_conversion_fallback(self):
"""Mapping whose ``dict(...)`` raises falls back to comprehension copy.
Exercises the inner ``except`` branch that copies via
``{key: mapping[key] for key in mapping}`` when ``dict(mapping)`` fails.
Uses a Mapping subclass whose first ``__iter__`` call raises so the
``dict()`` constructor errors but the subsequent comprehension reads
via the same iterator and succeeds.
"""
from collections.abc import Mapping as _Mapping
class _RaisingDictMapping(_Mapping):
def __init__(self, payload: dict) -> None:
self._payload = payload
self._first_iter_done = False
def __iter__(self):
if not self._first_iter_done:
self._first_iter_done = True
raise RuntimeError("simulated iteration failure")
yield from self._payload
def __getitem__(self, key):
return self._payload[key]
def __len__(self):
return len(self._payload)
result = ifaces._normalise_nodeinfo_packet(
_RaisingDictMapping({"fromId": "!aabbccdd"})
)
assert result is not None
assert result["fromId"] == "!aabbccdd"
assert result["id"] == "!aabbccdd"