d04e4c8370
close() stopped the asyncio loop without disconnecting first, leaving the
EventDispatcher._process_events() task and the serial transport pending
("Task was destroyed but it is pending!" / "Event loop is closed").
Now await MeshCore.disconnect() and let the transport's connection_lost
callback settle before stopping the loop, then join the thread.
Found by running the adapter against a live MeshCore-flashed Heltec V3.
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
104 lines
4.1 KiB
Python
104 lines
4.1 KiB
Python
"""MeshCore adapter — wraps the official `meshcore` (meshcore_py) async client.
|
|
|
|
meshcore_py is asyncio-native (MeshCore.create_serial/create_tcp/create_ble,
|
|
event subscription via meshcore.subscribe(EventType.X, coroutine_handler)).
|
|
The rest of this bridge is thread/callback-based (Meshtastic's pubsub,
|
|
Reticulum's RNS transport thread), so this adapter owns a dedicated event
|
|
loop on a background thread and exposes plain synchronous connect()/send()/
|
|
close() methods, matching the shape of the other two adapters.
|
|
"""
|
|
|
|
import asyncio
|
|
import threading
|
|
from collections.abc import Callable
|
|
|
|
|
|
class MeshCoreAdapter:
|
|
NETWORK = "meshcore"
|
|
|
|
def __init__(
|
|
self,
|
|
on_message: Callable[[str, str, str], None],
|
|
port: str | None = None,
|
|
host: str | None = None,
|
|
tcp_port: int = 4000,
|
|
channel_idx: int = 0,
|
|
):
|
|
"""port: serial device path. host: TCP hostname/IP (mutually exclusive
|
|
with port). channel_idx: which MeshCore public channel to bridge —
|
|
channel 0 is the default public channel on stock firmware.
|
|
"""
|
|
self._on_message = on_message
|
|
self._port = port
|
|
self._host = host
|
|
self._tcp_port = tcp_port
|
|
self._channel_idx = channel_idx
|
|
self._mc = None
|
|
self._loop: asyncio.AbstractEventLoop | None = None
|
|
self._thread: threading.Thread | None = None
|
|
|
|
def connect(self) -> None:
|
|
ready = threading.Event()
|
|
self._thread = threading.Thread(target=self._run_loop, args=(ready,), daemon=True)
|
|
self._thread.start()
|
|
ready.wait(timeout=15)
|
|
|
|
def _run_loop(self, ready: threading.Event) -> None:
|
|
self._loop = asyncio.new_event_loop()
|
|
asyncio.set_event_loop(self._loop)
|
|
self._loop.run_until_complete(self._connect_async())
|
|
ready.set()
|
|
self._loop.run_forever()
|
|
|
|
async def _connect_async(self) -> None:
|
|
from meshcore import EventType, MeshCore
|
|
|
|
if self._host:
|
|
self._mc = await MeshCore.create_tcp(self._host, self._tcp_port)
|
|
else:
|
|
self._mc = await MeshCore.create_serial(self._port or "/dev/ttyUSB0")
|
|
|
|
self._mc.subscribe(EventType.CHANNEL_MSG_RECV, self._handle_channel_msg)
|
|
|
|
async def _handle_channel_msg(self, event) -> None: # noqa: ANN001
|
|
data = event.payload
|
|
if data.get("channel_idx") != self._channel_idx:
|
|
return
|
|
text = data.get("text", "")
|
|
if text:
|
|
# CHANNEL_MSG_RECV carries no sender identity at the protocol
|
|
# level (verified against meshcore/reader.py) — MeshCore apps
|
|
# convey the sender by convention inside the text itself.
|
|
self._on_message(self.NETWORK, "?", text)
|
|
|
|
def send(self, text: str) -> None:
|
|
if self._mc is None or self._loop is None:
|
|
raise RuntimeError("MeshCoreAdapter.send() called before connect()")
|
|
asyncio.run_coroutine_threadsafe(
|
|
self._mc.commands.send_chan_msg(self._channel_idx, text), self._loop
|
|
)
|
|
|
|
def close(self) -> None:
|
|
if self._loop is None:
|
|
return
|
|
# Disconnect on the loop thread first: MeshCore.disconnect() stops the
|
|
# EventDispatcher task and closes the serial transport. Skipping this
|
|
# and stopping the loop directly leaves _process_events() pending
|
|
# ("Task was destroyed but it is pending!" / "Event loop is closed").
|
|
if self._mc is not None:
|
|
async def _shutdown() -> None:
|
|
await self._mc.disconnect()
|
|
# Let the serial transport's connection_lost callback run to
|
|
# completion before the loop stops, otherwise it lingers as a
|
|
# pending SerialTransport._call_connection_lost task.
|
|
await asyncio.sleep(0.25)
|
|
|
|
try:
|
|
fut = asyncio.run_coroutine_threadsafe(_shutdown(), self._loop)
|
|
fut.result(timeout=5)
|
|
except Exception:
|
|
pass
|
|
self._loop.call_soon_threadsafe(self._loop.stop)
|
|
if self._thread is not None:
|
|
self._thread.join(timeout=5)
|