Python MeshCore

February 14, 2026 · View on GitHub

Python library for interacting with MeshCore companion radio nodes.

Installation

pip install meshcore

Quick Start

Connect to your device and send a message:

import asyncio
from meshcore import MeshCore, EventType

async def main():
    # Connect to your device
    meshcore = await MeshCore.create_serial("/dev/ttyUSB0")
    
    # Get your contacts
    result = await meshcore.commands.get_contacts()
    if result.type == EventType.ERROR:
        print(f"Error getting contacts: {result.payload}")
        return
        
    contacts = result.payload
    print(f"Found {len(contacts)} contacts")
    
    # Send a message to the first contact
    if contacts:
        # Get the first contact
        contact = next(iter(contacts.items()))[1]
        
        # Pass the contact object directly to send_msg
        result = await meshcore.commands.send_msg(contact, "Hello from Python!")
        
        if result.type == EventType.ERROR:
            print(f"Error sending message: {result.payload}")
        else:
            print("Message sent successfully!")
    
    await meshcore.disconnect()

asyncio.run(main())

Development Setup

To set up for development:

# Create and activate virtual environment
python -m venv venv
source venv/bin/activate  # On Windows: venv\Scripts\activate

# Install in development mode
pip install -e .

# Run examples
python examples/pubsub_example.py -p /dev/ttyUSB0

Usage Guide

Command Return Values

All command methods in MeshCore return an Event object that contains both the event type and its payload. This allows for consistent error handling and type checking:

# Command result structure
result = await meshcore.commands.some_command()

# Check if the command was successful or resulted in an error
if result.type == EventType.ERROR:
    # Handle error case
    print(f"Command failed: {result.payload}")
else:
    # Handle success case - the event type will be specific to the command
    # (e.g., EventType.DEVICE_INFO, EventType.CONTACTS, EventType.MSG_SENT)
    print(f"Command succeeded with event type: {result.type}")
    # Access the payload data
    data = result.payload

Common error handling pattern:

result = await meshcore.commands.send_msg(contact, "Hello!")
if result.type == EventType.ERROR:
    print(f"Error sending message: {result.payload}")
else:
    # For send_msg, a successful result will have type EventType.MSG_SENT
    print(f"Message sent with expected ack: {result.payload['expected_ack'].hex()}")

Connecting to Your Device

Connect via Serial, BLE, or TCP:

# Serial connection
meshcore = await MeshCore.create_serial("/dev/ttyUSB0", 115200, debug=True)

# BLE connection (scans for devices if address not provided)
meshcore = await MeshCore.create_ble("12:34:56:78:90:AB")

# BLE connection with PIN pairing for enhanced security
meshcore = await MeshCore.create_ble("12:34:56:78:90:AB", pin="123456")

# TCP connection
meshcore = await MeshCore.create_tcp("192.168.1.100", 4000)

BLE PIN Pairing

For enhanced security, MeshCore supports BLE PIN pairing. This requires the device to be configured with a PIN and the client to provide the matching PIN during connection:

# First configure the device PIN (if not already set)
meshcore = await MeshCore.create_ble("12:34:56:78:90:AB")
await meshcore.commands.set_devicepin(123456)

# Then connect with PIN pairing
meshcore = await MeshCore.create_ble("12:34:56:78:90:AB", pin="123456")

PIN Pairing Features:

Automatic pairing initiation when PIN is provided
Graceful fallback if pairing fails (connection continues if device is already paired)
Compatible with all BLE connection methods (address, scanning, pre-configured client)
Logging of pairing success/failure for debugging

Note: BLE pairing behavior may vary by platform:

Linux/Windows: PIN pairing is fully supported
macOS: Pairing may be handled automatically by the system UI

Auto-Reconnect and Connection Events

Enable automatic reconnection when connections are lost:

# Enable auto-reconnect with custom retry limits
meshcore = await MeshCore.create_tcp(
    "192.168.1.100", 4000,
    auto_reconnect=True,
    max_reconnect_attempts=5
)

# Subscribe to connection events
async def on_connected(event):
    print(f"Connected: {event.payload}")
    if event.payload.get('reconnected'):
        print("Successfully reconnected!")

async def on_disconnected(event):
    print(f"Disconnected: {event.payload['reason']}")
    if event.payload.get('max_attempts_exceeded'):
        print("Max reconnection attempts exceeded")

meshcore.subscribe(EventType.CONNECTED, on_connected)
meshcore.subscribe(EventType.DISCONNECTED, on_disconnected)

# Check connection status
if meshcore.is_connected:
    print("Device is currently connected")

Auto-reconnect features:

Exponential backoff (1s, 2s, 4s, 8s max delay)
Configurable retry limits (default: 3 attempts)
Automatic disconnect detection (especially useful for TCP connections)
Connection events with detailed information

Using Commands (Synchronous Style)

Send commands and wait for responses:

# Get device information
result = await meshcore.commands.send_device_query()
if result.type == EventType.ERROR:
    print(f"Error getting device info: {result.payload}")
else:
    print(f"Device model: {result.payload['model']}")

# Get list of contacts
result = await meshcore.commands.get_contacts()
if result.type == EventType.ERROR:
    print(f"Error getting contacts: {result.payload}")
else:
    contacts = result.payload
    for contact_id, contact in contacts.items():
        print(f"Contact: {contact['adv_name']} ({contact_id})")

# Send a message (destination key in bytes)
result = await meshcore.commands.send_msg(dst_key, "Hello!")
if result.type == EventType.ERROR:
    print(f"Error sending message: {result.payload}")

# Setting device parameters
result = await meshcore.commands.set_name("My Device")
if result.type == EventType.ERROR:
    print(f"Error setting name: {result.payload}")
    
result = await meshcore.commands.set_tx_power(20)  # Set transmit power
if result.type == EventType.ERROR:
    print(f"Error setting TX power: {result.payload}")

Finding Contacts

Easily find contacts by name or key:

# Find a contact by name
contact = meshcore.get_contact_by_name("Bob's Radio")
if contact:
    print(f"Found Bob at: {contact['adv_lat']}, {contact['adv_lon']}")
    
# Find by partial key prefix
contact = meshcore.get_contact_by_key_prefix("a1b2c3")

Event-Based Programming (Asynchronous Style)

Subscribe to events to handle them asynchronously:

# Subscribe to incoming messages
async def handle_message(event):
    data = event.payload
    print(f"Message from {data['pubkey_prefix']}: {data['text']}")
    
subscription = meshcore.subscribe(EventType.CONTACT_MSG_RECV, handle_message)

# Subscribe to advertisements
async def handle_advert(event):
    print("Advertisement detected!")
    
meshcore.subscribe(EventType.ADVERTISEMENT, handle_advert)

# When done, unsubscribe
meshcore.unsubscribe(subscription)

Filtering Events by Attributes

Filter events based on their attributes to handle only specific ones:

# Subscribe only to messages from a specific contact
async def handle_specific_contact_messages(event):
    print(f"Message from Alice: {event.payload['text']}")
    
contact = meshcore.get_contact_by_name("Alice")
if contact:
    alice_subscription = meshcore.subscribe(
        EventType.CONTACT_MSG_RECV,
        handle_specific_contact_messages,
        attribute_filters={"pubkey_prefix": contact["public_key"][:12]}
    )

# Send a message and wait for its specific acknowledgment
async def send_and_confirm_message(meshcore, dst_key, message):
    # Send the message and get information about the sent message
    sent_result = await meshcore.commands.send_msg(dst_key, message)
    
    # Extract the expected acknowledgment code from the message sent event
    if sent_result.type == EventType.ERROR:
        print(f"Error sending message: {sent_result.payload}")
        return False
        
    expected_ack = sent_result.payload["expected_ack"].hex()
    print(f"Message sent, waiting for ack with code: {expected_ack}")
    
    # Wait specifically for this acknowledgment
    result = await meshcore.wait_for_event(
        EventType.ACK,
        attribute_filters={"code": expected_ack},
        timeout=10.0
    )
    
    if result:
        print("Message confirmed delivered!")
        return True
    else:
        print("Message delivery confirmation timed out")
        return False

Hybrid Approach (Commands + Events)

Combine command-based and event-based styles:

import asyncio

async def main():
    # Connect to device
    meshcore = await MeshCore.create_serial("/dev/ttyUSB0")
    
    # Set up event handlers
    async def handle_ack(event):
        print("Message acknowledged!")
    
    async def handle_battery(event):
        print(f"Battery level: {event.payload}%")
    
    # Subscribe to events
    meshcore.subscribe(EventType.ACK, handle_ack)
    meshcore.subscribe(EventType.BATTERY, handle_battery)
    
    # Create background task for battery checking
    async def check_battery_periodically():
        while True:
            # Send command (returns battery level)
            result = await meshcore.commands.get_bat()
            if result.type == EventType.ERROR:
                print(f"Error checking battery: {result.payload}")
            else:
                print(f"Battery level: {result.payload.get('level', 'unknown')}%")
            await asyncio.sleep(60)  # Wait 60 seconds between checks
    
    # Start the background task
    battery_task = asyncio.create_task(check_battery_periodically())
    
    # Send manual command and wait for response
    await meshcore.commands.send_advert(flood=True)
    
    try:
        # Keep the main program running
        await asyncio.sleep(float('inf'))
    except asyncio.CancelledError:
        # Clean up when program ends
        battery_task.cancel()
        await meshcore.disconnect()

# Run the program
asyncio.run(main())

Auto-Fetching Messages

Let the library automatically fetch incoming messages:

# Start auto-fetching messages
await meshcore.start_auto_message_fetching()

# Just subscribe to message events - the library handles fetching
async def on_message(event):
    print(f"New message: {event.payload['text']}")
    
meshcore.subscribe(EventType.CONTACT_MSG_RECV, on_message)

# When done
await meshcore.stop_auto_message_fetching()

Debug Mode

Enable debug logging for troubleshooting:

# Enable debug mode when creating the connection
meshcore = await MeshCore.create_serial("/dev/ttyUSB0", debug=True)

This logs detailed information about commands sent and events received.

Common Examples

Sending Messages to Contacts

Commands that require a destination (send_msg, send_login, send_statusreq, etc.) now accept either:

A string with the hex representation of a public key
A contact object with a "public_key" field
Bytes object (for backward compatibility)

# Get contacts and send to a specific one
result = await meshcore.commands.get_contacts()
if result.type == EventType.ERROR:
    print(f"Error getting contacts: {result.payload}")
else:
    contacts = result.payload
    for key, contact in contacts.items():
        if contact["adv_name"] == "Alice":
            # Option 1: Pass the contact object directly
            result = await meshcore.commands.send_msg(contact, "Hello Alice!")
            if result.type == EventType.ERROR:
                print(f"Error sending message: {result.payload}")
            
            # Option 2: Use the public key string
            result = await meshcore.commands.send_msg(contact["public_key"], "Hello again Alice!")
            if result.type == EventType.ERROR:
                print(f"Error sending message: {result.payload}")
            
            # Option 3 (backward compatible): Convert the hex key to bytes
            dst_key = bytes.fromhex(contact["public_key"])
            result = await meshcore.commands.send_msg(dst_key, "Hello once more Alice!")
            if result.type == EventType.ERROR:
                print(f"Error sending message: {result.payload}")
            break

# You can also directly use a contact found by name
contact = meshcore.get_contact_by_name("Bob")
if contact:
    result = await meshcore.commands.send_msg(contact, "Hello Bob!")
    if result.type == EventType.ERROR:
        print(f"Error sending message: {result.payload}")

Monitoring Channel Messages

# Subscribe to channel messages
async def channel_handler(event):
    msg = event.payload
    print(f"Channel {msg['channel_idx']}: {msg['text']}")
    
meshcore.subscribe(EventType.CHANNEL_MSG_RECV, channel_handler)

API Reference

Event Types

All events in MeshCore are represented by the EventType enum. These events are dispatched by the library and can be subscribed to:

Event Type	String Value	Description	Typical Payload
Device & Status Events
`SELF_INFO`	`"self_info"`	Device's own information after appstart	Device configuration, public key, coordinates
`DEVICE_INFO`	`"device_info"`	Device capabilities and firmware info	Firmware version, model, max contacts/channels
`BATTERY`	`"battery_info"`	Battery level and storage info	Battery level, used/total storage
`CURRENT_TIME`	`"time_update"`	Device time response	Current timestamp
`STATUS_RESPONSE`	`"status_response"`	Device status statistics	Battery, TX queue, noise floor, packet counts
`CUSTOM_VARS`	`"custom_vars"`	Custom variable responses	Key-value pairs of custom variables
Contact Events
`CONTACTS`	`"contacts"`	Contact list response	Dictionary of contacts by public key
`NEW_CONTACT`	`"new_contact"`	New contact discovered	Contact information
`CONTACT_URI`	`"contact_uri"`	Contact export URI	Shareable contact URI
Messaging Events
`CONTACT_MSG_RECV`	`"contact_message"`	Direct message received	Message text, sender prefix, timestamp
`CHANNEL_MSG_RECV`	`"channel_message"`	Channel message received	Message text, channel index, timestamp
`MSG_SENT`	`"message_sent"`	Message send confirmation	Expected ACK code, suggested timeout
`NO_MORE_MSGS`	`"no_more_messages"`	No pending messages	Empty payload
`MESSAGES_WAITING`	`"messages_waiting"`	Messages available notification	Empty payload
Network Events
`ADVERTISEMENT`	`"advertisement"`	Node advertisement detected	Public key of advertising node
`PATH_UPDATE`	`"path_update"`	Routing path update	Public key and path information
`ACK`	`"acknowledgement"`	Message acknowledgment	ACK code
`PATH_RESPONSE`	`"path_response"`	Path discovery response	Inbound/outbound path data
`TRACE_DATA`	`"trace_data"`	Route trace information	Path with SNR data for each hop
Telemetry Events
`TELEMETRY_RESPONSE`	`"telemetry_response"`	Telemetry data response	LPP-formatted sensor data
`MMA_RESPONSE`	`"mma_response"`	Memory Management Area data	Min/max/avg telemetry over time range
`ACL_RESPONSE`	`"acl_response"`	Access Control List data	List of keys and permissions
Channel Events
`CHANNEL_INFO`	`"channel_info"`	Channel configuration	Channel name, secret, index
Raw Data Events
`RAW_DATA`	`"raw_data"`	Raw radio data	SNR, RSSI, payload hex
`RX_LOG_DATA`	`"rx_log_data"`	RF log data	SNR, RSSI, raw payload
`LOG_DATA`	`"log_data"`	Generic log data	Various log information
Binary Protocol Events
`BINARY_RESPONSE`	`"binary_response"`	Generic binary response	Tag and hex data
`SIGN_START`	`"sign_start"`	Start of an on-device signing session	Maximum buffer size (bytes) for data to sign
`SIGNATURE`	`"signature"`	Resulting on-device signature	Raw signature bytes
Authentication Events
`LOGIN_SUCCESS`	`"login_success"`	Successful login	Permissions, admin status, pubkey prefix
`LOGIN_FAILED`	`"login_failed"`	Failed login attempt	Pubkey prefix
Command Response Events
`OK`	`"command_ok"`	Command successful	Success confirmation, optional value
`ERROR`	`"command_error"`	Command failed	Error reason or code
Connection Events
`CONNECTED`	`"connected"`	Connection established	Connection details, reconnection status
`DISCONNECTED`	`"disconnected"`	Connection lost	Disconnection reason

Available Commands

All commands are async methods that return Event objects. Commands are organized into functional groups:

Device Commands (`meshcore.commands.*`)

Command	Parameters	Returns	Description
Device Information
`send_appstart()`	None	`SELF_INFO`	Get device self-information and configuration
`send_device_query()`	None	`DEVICE_INFO`	Query device capabilities and firmware info
`get_bat()`	None	`BATTERY`	Get battery level and storage information
`get_time()`	None	`CURRENT_TIME`	Get current device time
`get_self_telemetry()`	None	`TELEMETRY_RESPONSE`	Get device's own telemetry data
`get_custom_vars()`	None	`CUSTOM_VARS`	Retrieve all custom variables
`get_allowed_repeat_freq()`	None	`ALLOWED_FREQ`	Retreive allowed repeat freqs from device
Device Configuration
`set_name(name)`	`name: str`	`OK`	Set device name/identifier
`set_coords(lat, lon)`	`lat: float, lon: float`	`OK`	Set device GPS coordinates
`set_time(val)`	`val: int`	`OK`	Set device time (Unix timestamp)
`set_tx_power(val)`	`val: int`	`OK`	Set radio transmission power level
`set_devicepin(pin)`	`pin: int`	`OK`	Set device PIN for security
`set_custom_var(key, value)`	`key: str, value: str`	`OK`	Set custom variable
Radio Configuration
`set_radio(freq, bw, sf, cr)`	`freq: float, bw: float, sf: int, cr: int`	`OK`	Configure radio (freq MHz, bandwidth kHz, spreading factor, coding rate 5-8)
`set_tuning(rx_dly, af)`	`rx_dly: int, af: int`	`OK`	Set radio tuning parameters
Telemetry Configuration
`set_telemetry_mode_base(mode)`	`mode: int`	`OK`	Set base telemetry mode
`set_telemetry_mode_loc(mode)`	`mode: int`	`OK`	Set location telemetry mode
`set_telemetry_mode_env(mode)`	`mode: int`	`OK`	Set environmental telemetry mode
`set_manual_add_contacts(enabled)`	`enabled: bool`	`OK`	Enable/disable manual contact addition
`set_advert_loc_policy(policy)`	`policy: int`	`OK`	Set location advertisement policy
Channel Management
`get_channel(channel_idx)`	`channel_idx: int`	`CHANNEL_INFO`	Get channel configuration
`set_channel(channel_idx, name, secret)`	`channel_idx: int, name: str, secret: bytes`	`OK`	Configure channel (secret must be 16 bytes)
Device Actions
`send_advert(flood=False)`	`flood: bool`	`OK`	Send advertisement (optionally flood network)
`reboot()`	None	None	Reboot device (no response expected)
Security
`export_private_key()`	None	`PRIVATE_KEY/DISABLED`	Export device private key (requires PIN auth & enabled firmware)
`import_private_key(key)`	`key: bytes`	`OK`	Import private key to device
Statistics
`get_stats_core()`	None	`STATS_CORE`	Get core statistics (voltage, uptime, errors, queue length)
`get_stats_radio()`	None	`STATS_RADIO`	Get radio statistics (noise floor, last RSSI/SNR, tx/rx time stats)
`get_stats_packets()`	None	`STATS_PACKETS`	Get packet statistics (rx/tx totals, flood vs. direct, recv_errors when present)
Advanced Configuration
`set_multi_acks(multi_acks)`	`multi_acks: int`	`OK`	Set multi-acks mode (experimental ack repeats)

Contact Commands (`meshcore.commands.*`)

Command	Parameters	Returns	Description
Contact Management
`get_contacts(lastmod=0)`	`lastmod: int`	`CONTACTS`	Get contact list (filter by last modification time)
`add_contact(contact)`	`contact: dict`	`OK`	Add new contact to device
`update_contact(contact, path, flags)`	`contact: dict, path: bytes, flags: int`	`OK`	Update existing contact
`remove_contact(key)`	`key: str/bytes`	`OK`	Remove contact by public key
Contact Operations
`reset_path(key)`	`key: str/bytes`	`OK`	Reset routing path for contact
`share_contact(key)`	`key: str/bytes`	`OK`	Share contact with network
`export_contact(key=None)`	`key: str/bytes/None`	`CONTACT_URI`	Export contact as URI (None exports node)
`import_contact(card_data)`	`card_data: bytes`	`OK`	Import contact from card data
Contact Modification
`change_contact_path(contact, path)`	`contact: dict, path: bytes`	`OK`	Change routing path for contact
`change_contact_flags(contact, flags)`	`contact: dict, flags: int`	`OK`	Change contact flags/settings

Messaging Commands (`meshcore.commands.*`)

Command	Parameters	Returns	Description
Message Handling
`get_msg(timeout=None)`	`timeout: float`	`CONTACT_MSG_RECV/CHANNEL_MSG_RECV/NO_MORE_MSGS`	Get next pending message
`send_msg(dst, msg, timestamp=None)`	`dst: contact/str/bytes, msg: str, timestamp: int`	`MSG_SENT`	Send direct message
`send_cmd(dst, cmd, timestamp=None)`	`dst: contact/str/bytes, cmd: str, timestamp: int`	`MSG_SENT`	Send command message
`send_chan_msg(chan, msg, timestamp=None)`	`chan: int, msg: str, timestamp: int`	`MSG_OK`	Send channel message
Authentication
`send_login(dst, pwd)`	`dst: contact/str/bytes, pwd: str`	`MSG_SENT`	Send login request
`send_logout(dst)`	`dst: contact/str/bytes`	`MSG_SENT`	Send logout request
Information Requests
`send_statusreq(dst)`	`dst: contact/str/bytes`	`MSG_SENT`	Request status from contact
`send_telemetry_req(dst)`	`dst: contact/str/bytes`	`MSG_SENT`	Request telemetry from contact
Advanced Messaging
`send_binary_req(dst, bin_data)`	`dst: contact/str/bytes, bin_data: bytes`	`MSG_SENT`	Send binary data request
`send_path_discovery(dst)`	`dst: contact/str/bytes`	`MSG_SENT`	Initiate path discovery
`send_trace(auth_code, tag, flags, path=None)`	`auth_code: int, tag: int, flags: int, path: list`	`MSG_SENT`	Send route trace packet
Message Retry & Scope
`send_msg_with_retry(dst, msg, ...)`	`dst, msg, timestamp, max_attempts, max_flood_attempts, flood_after, timeout, min_timeout`	`MSG_SENT/None`	Send message with automatic retry and ACK waiting
`set_flood_scope(scope)`	`scope: str`	`OK`	Set flood scope (hash like "#name", "0"/""/"*" to disable, or raw key)

Binary Protocol Commands (`meshcore.commands.*`)

Command	Parameters	Returns	Description
`req_status(contact, timeout=0)`	`contact: dict, timeout: float`	`STATUS_RESPONSE`	Get detailed status via binary protocol
`req_telemetry(contact, timeout=0)`	`contact: dict, timeout: float`	`TELEMETRY_RESPONSE`	Get telemetry via binary protocol
`req_mma(contact, start, end, timeout=0)`	`contact: dict, start: int, end: int, timeout: float`	`MMA_RESPONSE`	Get historical telemetry data
`req_acl(contact, timeout=0)`	`contact: dict, timeout: float`	`ACL_RESPONSE`	Get access control list
`sign_start()`	None	`SIGN_START`	Begin a signing session; returns maximum buffer size for data to sign
`sign_data(chunk)`	`chunk: bytes`	`OK`	Append a data chunk to the current signing session (can be called multiple times)
`sign_finish()`	None	`SIGNATURE`	Finalize signing and return the signature for all accumulated data

Helper Methods

Method	Returns	Description
`get_contact_by_name(name)`	`dict/None`	Find contact by advertisement name
`get_contact_by_key_prefix(prefix)`	`dict/None`	Find contact by partial public key
`sign(data, chunk_size=512)`	`Event` (`SIGNATURE`/`ERROR`)	High-level helper to sign arbitrary data on-device, handling chunking for you
`is_connected`	`bool`	Check if device is currently connected
`subscribe(event_type, callback, filters=None)`	`Subscription`	Subscribe to events with optional filtering
`unsubscribe(subscription)`	None	Remove event subscription
`wait_for_event(event_type, filters=None, timeout=None)`	`Event/None`	Wait for specific event

Event Filtering

Events can be filtered by their attributes when subscribing:

# Filter by public key prefix
meshcore.subscribe(
    EventType.CONTACT_MSG_RECV,
    handler,
    attribute_filters={"pubkey_prefix": "a1b2c3d4e5f6"}
)

# Filter by channel index
meshcore.subscribe(
    EventType.CHANNEL_MSG_RECV,
    handler,
    attribute_filters={"channel_idx": 0}
)

# Filter acknowledgments by code
meshcore.subscribe(
    EventType.ACK,
    handler,
    attribute_filters={"code": "12345678"}
)

Examples in the Repo

Check the examples/ directory for more:

pubsub_example.py: Event subscription system with auto-fetching
serial_infos.py: Quick device info retrieval
serial_msg.py: Message sending and receiving
serial_pingbot.py: Ping bot which can be run on a channel
serial_rss_bot.py: A RSS feed to Meshcore channel example, which broadcasts emergency bushfire warnings in VIC, AU
serial_meshcore_ollama.py: Simple Ollama to Meshcore gateway, a simple chat box
ble_pin_pairing_example.py: BLE connection with PIN pairing
ble_private_key_export.py: BLE private key export with PIN authentication
ble_t1000_infos.py: BLE connections