58 lines
1.9 KiB
Python
58 lines
1.9 KiB
Python
"""Meshtastic adapter — wraps the official `meshtastic` Python client.
|
|
|
|
Uses the same pubsub pattern as meshtastic/python's own examples
|
|
(examples/replymessage.py, examples/tcp_pubsub_send_and_receive.py):
|
|
subscribe to the "meshtastic.receive" topic, filter for text packets,
|
|
send via MeshInterface.sendText().
|
|
"""
|
|
|
|
from collections.abc import Callable
|
|
|
|
from pubsub import pub
|
|
|
|
|
|
class MeshtasticAdapter:
|
|
NETWORK = "meshtastic"
|
|
|
|
def __init__(
|
|
self,
|
|
on_message: Callable[[str, str, str], None],
|
|
port: str | None = None,
|
|
host: str | None = None,
|
|
):
|
|
"""port: serial device path (e.g. /dev/ttyUSB0). host: TCP hostname/IP.
|
|
Exactly one of port/host should be set; if neither is set, the
|
|
underlying library auto-detects a serial device.
|
|
"""
|
|
self._on_message = on_message
|
|
self._port = port
|
|
self._host = host
|
|
self._iface = None
|
|
|
|
def connect(self) -> None:
|
|
import meshtastic.serial_interface
|
|
import meshtastic.tcp_interface
|
|
|
|
pub.subscribe(self._handle_receive, "meshtastic.receive")
|
|
|
|
if self._host:
|
|
self._iface = meshtastic.tcp_interface.TCPInterface(hostname=self._host, timeout=10)
|
|
else:
|
|
self._iface = meshtastic.serial_interface.SerialInterface(devPath=self._port, timeout=10)
|
|
|
|
def _handle_receive(self, packet: dict, interface) -> None: # noqa: ANN001
|
|
text = packet.get("decoded", {}).get("text")
|
|
if not text:
|
|
return
|
|
sender = packet.get("fromId") or str(packet.get("from"))
|
|
self._on_message(self.NETWORK, sender, text)
|
|
|
|
def send(self, text: str) -> None:
|
|
if self._iface is None:
|
|
raise RuntimeError("MeshtasticAdapter.send() called before connect()")
|
|
self._iface.sendText(text)
|
|
|
|
def close(self) -> None:
|
|
if self._iface is not None:
|
|
self._iface.close()
|