mirror of
https://github.com/meshcore-dev/meshcore_py.git
synced 2026-04-20 22:13:49 +00:00
159 lines
4.6 KiB
Python
159 lines
4.6 KiB
Python
import asyncio
|
|
import logging
|
|
from typing import Any
|
|
|
|
from meshcore import MeshCore, EventType
|
|
|
|
SERIAL_PORT = "COM4" # change this to your serial port
|
|
CHANNEL_IDX = 1 # change this to the index of your "#ping" channel
|
|
|
|
logging.basicConfig(level=logging.INFO)
|
|
_LOGGER = logging.getLogger("serial_pingbot")
|
|
|
|
latest_pathinfo_str = "(? hops, ?)"
|
|
|
|
|
|
def parse_rx_log_data(payload: Any) -> dict[str, Any]:
|
|
"""Parse RX_LOG event payload to extract LoRa packet details.
|
|
|
|
Expected format (hex):
|
|
byte0: header
|
|
byte1: path_len
|
|
next path_len bytes: path nodes
|
|
next byte: channel_hash (optional)
|
|
"""
|
|
result: dict[str, Any] = {}
|
|
|
|
try:
|
|
hex_str = None
|
|
|
|
if isinstance(payload, dict):
|
|
hex_str = payload.get("payload") or payload.get("raw_hex")
|
|
elif isinstance(payload, (str, bytes)):
|
|
hex_str = payload
|
|
|
|
if not hex_str:
|
|
return result
|
|
|
|
if isinstance(hex_str, bytes):
|
|
hex_str = hex_str.hex()
|
|
|
|
hex_str = str(hex_str).lower().replace(" ", "").replace("\n", "").replace("\r", "")
|
|
|
|
if len(hex_str) < 4:
|
|
return result
|
|
|
|
result["header"] = hex_str[0:2]
|
|
|
|
try:
|
|
path_len = int(hex_str[2:4], 16)
|
|
result["path_len"] = path_len
|
|
except ValueError:
|
|
return {}
|
|
|
|
path_start = 4
|
|
path_end = path_start + (path_len * 2)
|
|
|
|
if len(hex_str) < path_end:
|
|
return {}
|
|
|
|
path_hex = hex_str[path_start:path_end]
|
|
result["path"] = path_hex
|
|
result["path_nodes"] = [path_hex[i:i + 2] for i in range(0, len(path_hex), 2)]
|
|
|
|
if len(hex_str) >= path_end + 2:
|
|
result["channel_hash"] = hex_str[path_end:path_end + 2]
|
|
|
|
except Exception as ex:
|
|
_LOGGER.debug(f"Error parsing RX_LOG data: {ex}")
|
|
|
|
return result
|
|
|
|
|
|
def format_pathinfo(parsed: dict[str, Any]) -> str:
|
|
"""Return string in format: '(<path_len> hops, <aa:bb:cc>)'."""
|
|
path_len = parsed.get("path_len")
|
|
nodes = parsed.get("path_nodes") or []
|
|
|
|
if path_len is None:
|
|
return "(? hops, ?)"
|
|
|
|
if path_len == 0:
|
|
return "(0 hops, direct)"
|
|
|
|
path_str = ":".join(nodes) if nodes else "?"
|
|
return f"({path_len} hops, {path_str})"
|
|
|
|
|
|
async def main():
|
|
global latest_pathinfo_str
|
|
|
|
meshcore = await MeshCore.create_serial(SERIAL_PORT, debug=True)
|
|
print(f"Connected on {SERIAL_PORT}")
|
|
|
|
await meshcore.start_auto_message_fetching()
|
|
|
|
async def handle_rx_log_data(event):
|
|
global latest_pathinfo_str
|
|
|
|
rx = event.payload or {}
|
|
raw = rx.get("payload") # use 'payload' (not 'raw_hex') for this parser
|
|
if not raw:
|
|
return
|
|
|
|
parsed = parse_rx_log_data(raw)
|
|
if parsed:
|
|
latest_pathinfo_str = format_pathinfo(parsed)
|
|
|
|
async def handle_channel_message(event):
|
|
msg = event.payload or {}
|
|
|
|
pathinfo = latest_pathinfo_str
|
|
|
|
chan = msg.get("channel_idx")
|
|
text = msg.get("text", "")
|
|
path_len = msg.get("path_len")
|
|
sender = text.split(":", 1)[0].strip()
|
|
|
|
print("~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~")
|
|
print(pathinfo)
|
|
print("~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~")
|
|
print(f"Received on channel {chan} from {sender}: {text} | path_len={path_len}")
|
|
|
|
if chan == CHANNEL_IDX and "ping" in text.lower():
|
|
reply = f"@[{sender}] Pong 🏓{pathinfo}"
|
|
print(f"Detected Ping. Replying in channel {CHANNEL_IDX} with:\n{reply}")
|
|
|
|
result = await meshcore.commands.send_chan_msg(CHANNEL_IDX, reply)
|
|
if result.type == EventType.ERROR:
|
|
print(f"Error sending reply: {result.payload}")
|
|
else:
|
|
print("Reply sent")
|
|
|
|
sub_chan = meshcore.subscribe(
|
|
EventType.CHANNEL_MSG_RECV,
|
|
handle_channel_message,
|
|
attribute_filters={"channel_idx": CHANNEL_IDX},
|
|
)
|
|
|
|
sub_rx = meshcore.subscribe(
|
|
EventType.RX_LOG_DATA,
|
|
handle_rx_log_data,
|
|
)
|
|
|
|
try:
|
|
print(f"Listening for 'Ping' on channel {CHANNEL_IDX} and RX_LOG_DATA...")
|
|
while True:
|
|
await asyncio.sleep(3600)
|
|
except KeyboardInterrupt:
|
|
print("Stopping listener...")
|
|
finally:
|
|
meshcore.unsubscribe(sub_chan)
|
|
meshcore.unsubscribe(sub_rx)
|
|
await meshcore.stop_auto_message_fetching()
|
|
await meshcore.disconnect()
|
|
print("Disconnected")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
asyncio.run(main())
|