API Reference¶

This section provides detailed API documentation for pyMC_Core.

Core Modules¶

MeshNode¶

pymc_core.node.MeshNode ¶

MeshNode(radio, local_identity, config=None, *, contacts=None, channel_db=None, logger=None, event_service=None)

Thin transport layer for mesh radio communication.

Owns a radio interface and a :class:Dispatcher that handles raw packet I/O (TX lock, ACK management, handler dispatch). Application-layer concerns — contact lookup, message building, response waiting — belong in the companion layer (:class:CompanionBase and its subclasses).

Typical usage::

node = MeshNode(radio, identity, config={...})
await node.start()       # blocks in dispatcher.run_forever()
node.stop()

Initialise a mesh network node instance.

Parameters:

Name	Type	Description	Default
`radio`	`Optional[Any]`	Radio hardware interface for transmission/reception.	required
`local_identity`	`LocalIdentity`	Node's cryptographic identity for secure communication.	required
`config`	`Optional[dict]`	Optional configuration dictionary with node settings.	`None`
`contacts`	`Optional[Any]`	Optional contact storage for managing known nodes.	`None`
`channel_db`	`Optional[Any]`	Optional channel database for group communication.	`None`
`logger`	`Optional[Logger]`	Optional logger instance; defaults to module logger.	`None`
`event_service`	`Optional[Any]`	Optional event service for broadcasting mesh events.	`None`

send_packet `async` ¶

send_packet(pkt, *, wait_for_ack=False, **kwargs)

Send a raw packet via the dispatcher.

This is the single transport entry point. All message-building and response-waiting logic lives in the companion layer.

set_event_service ¶

set_event_service(event_service)

Set the event service for broadcasting mesh events.

start `async` ¶

start()

Start the mesh node and begin processing radio communications.

Enters the dispatcher's main event loop for handling incoming/outgoing messages. This method blocks until the node is stopped.

stop ¶

stop()

Stop the mesh node and clean up associated services.

Packet¶

pymc_core.protocol.Packet ¶

Packet()

Represents a mesh network packet with header, transport codes, path, and payload components.

This class handles serialization and deserialization of packets in the mesh protocol, providing methods for packet validation, hashing, and data extraction. It maintains compatibility with C++ packet formats for cross-platform interoperability.

Attributes:

Name	Type	Description
`header`	`int`	Single byte header containing packet type and flags.
`transport_codes`	`list`	Two 16-bit transport codes for TRANSPORT route types.
`path_len`	`int`	Encoded path length byte (bits 0-5 = hash count, bits 6-7 = hash size - 1).
`path`	`bytearray`	Variable-length path data for routing (hash_count × hash_size bytes).
`payload`	`bytearray`	Variable-length payload data.
`payload_len`	`int`	Actual length of payload data.
`_rssi`	`int`	Raw RSSI signal strength value from firmware.
`_snr`	`int`	Raw SNR value from firmware.

Example

packet = Packet()
packet.header = 0x01  # Flood routing
packet.set_path(b"\xAA\xBB\xCC")  # 3 hops, 1-byte hashes
packet.payload = b"Hello World"
packet.payload_len = len(packet.payload)
data = packet.write_to()
# data can be transmitted over the mesh network

Initialize a new empty packet with default values.

Sets up the packet structure with zero-initialized fields ready for population with actual packet data. All fields are initialized to safe default values to prevent undefined behavior.

rssi `property` ¶

rssi

Get the raw RSSI (Received Signal Strength Indicator) value.

Returns the signal strength measurement from the radio firmware in its native scale, typically used for relative signal comparisons.

Returns:

Name	Type	Description
`int`	`int`	Raw RSSI value from firmware. Higher values indicate stronger received signals.

snr `property` ¶

snr

Get the signal-to-noise ratio in decibels.

Provides convenient access to the calculated SNR value in dB, automatically converting from the raw firmware value.

Returns:

Name	Type	Description
`float`	`float`	SNR in decibels, where positive values indicate signal power above noise floor, negative values indicate signal below noise floor.

apply_path_hash_mode ¶

apply_path_hash_mode(mode, *, mark_applied=False)

Set path_len bits 6-7 from path_hash_mode for 0-hop packets (skip TRACE).

Used by companion and dispatcher so the rule lives in one place. TRACE packets are excluded because the repeater's trace handler uses path/path_len for SNR values, not routing hashes.

Parameters:

Name	Type	Description	Default
`mode`	`int`	Path hash mode: 0=1-byte, 1=2-byte, 2=3-byte per hop.	required
`mark_applied`	`bool`	If True, set _path_hash_mode_applied so dispatcher does not overwrite (used when companion applies its preference).	`False`

Raises:

Type	Description
`ValueError`	If mode not in (0, 1, 2).

calculate_packet_hash ¶

calculate_packet_hash()

Compute SHA256-based hash for ACK, deduplication, and validation.

Generates a cryptographic hash of the packet content for use in: - ACK packet generation and verification - Packet deduplication to prevent replay attacks - Message integrity validation

Returns:

Name	Type	Description
`bytes`	`bytes`	First MAX_HASH_SIZE bytes of SHA256 digest computed over the payload type, path length, and payload data.

Note

The hash includes payload type and path_len to ensure packets with different routing or content types produce different hashes.

get_crc ¶

get_crc()

Calculate a 4-byte CRC from SHA256 digest for ACK confirmation.

Generates a compact checksum derived from the packet's SHA256 hash, used specifically for ACK packet confirmation to ensure the ACK corresponds to the correct original packet.

Returns:

Name	Type	Description
`int`	`int`	32-bit CRC value extracted from the SHA256 digest, used for lightweight packet identification in ACKs.

Note

This CRC is more compact than the full hash but still provides sufficient uniqueness for ACK correlation in the mesh network.

get_packet_hash_hex ¶

get_packet_hash_hex(length=None)

Return upper-case hex string representation of this packet's hash.

Parameters:

Name	Type	Description	Default
`length`	`Optional[int]`	Maximum length of the returned hex string. Defaults to None (full hash string).	`None`

Returns:

Name	Type	Description
`str`	`str`	Upper-case hex string of the packet hash.

get_path_byte_len ¶

get_path_byte_len()

Calculate actual path byte length from the encoded path_len byte.

get_path_hash_count ¶

get_path_hash_count()

Extract hop count (0-63) from the encoded path_len byte.

get_path_hash_size ¶

get_path_hash_size()

Extract per-hop hash size (1, 2, or 3) from the encoded path_len byte.

get_path_hashes ¶

get_path_hashes()

Return path as a list of per-hop hash entries (1, 2, or 3 bytes each).

Groups the raw self.path bytearray using the hash size encoded in self.path_len. Each entry in the returned list is a bytes object whose length equals get_path_hash_size().

Returns:

Type	Description
`list`	list[bytes]: One entry per hop. Empty list when hop count is 0.

get_path_hashes_hex ¶

get_path_hashes_hex()

Return path as a list of uppercase hex strings, one per hop.

Examples::

1-byte hashes: ["B5", "A3", "F2"]
2-byte hashes: ["B5A3", "F2C1"]
3-byte hashes: ["B5A3F2", "C1D4E7"]

get_payload ¶

get_payload()

Get the packet payload as immutable bytes, truncated to declared length.

Returns:

Name	Type	Description
`bytes`	`bytes`	The actual payload data, limited to payload_len bytes. Returns empty bytes if payload_len is 0 or negative.

Note

This method ensures only the declared payload length is returned, preventing access to any extra data that might be in the buffer.

get_payload_app_data ¶

get_payload_app_data()

Extract application-specific data from the payload, skipping protocol headers.

Returns:

Name	Type	Description
`bytes`	`bytes`	Application data portion of the payload, excluding the protocol overhead (public key, timestamp, and signature). Returns empty bytes if the payload is too short to contain the full protocol header.

Note

The protocol header consists of: - Public key (PUB_KEY_SIZE bytes) - Timestamp (TIMESTAMP_SIZE bytes) - Signature (SIGNATURE_SIZE bytes)

get_payload_type ¶

get_payload_type()

Extract the 4-bit payload type from the packet header.

Returns:

Name	Type	Description
`int`	`int`	Payload type value indicating the type of data in the packet: - 0: Plain text message - 1: Encrypted message - 2: ACK packet - 3: Advertisement - 4: Login request/response - 5: Protocol control - 6-15: Reserved for future use

get_payload_ver ¶

get_payload_ver()

Extract the 2-bit payload version from the packet header.

Returns:

Name	Type	Description
`int`	`int`	Version number (0-3) indicating the packet format version. Higher versions may include additional features or format changes.

get_raw_length ¶

get_raw_length()

Calculate the total byte length of the packet on the wire.

Computes the exact size of the serialized packet as it would appear on the network, matching the C++ Packet::getRawLength() implementation.

Returns:

Name	Type	Description
`int`	`int`	Total packet size in bytes, calculated as: header(1) + [transport_codes(4)] + path_len(1) + path(N) + payload(M) Transport codes are only included if has_transport_codes() is True.

Note

This matches the wire format used by write_to() and expected by read_from().

get_route_type ¶

get_route_type()

Extract the 2-bit route type from the packet header.

Returns:

Name	Type	Description
`int`	`int`	Route type value (0-3) indicating routing method: - 0: Transport flood routing (with transport codes) - 1: Flood routing - 2: Direct routing - 3: Transport direct routing (with transport codes)

get_snr ¶

get_snr()

Calculate the signal-to-noise ratio in decibels.

Converts the raw SNR value from firmware into a standardized decibel representation for signal quality assessment.

Returns:

Name	Type	Description
`float`	`float`	SNR value in dB, where higher values indicate better signal quality relative to background noise.

has_transport_codes ¶

has_transport_codes()

Check if this packet includes transport codes in its format.

Returns:

Name	Type	Description
`bool`	`bool`	True if the packet uses transport flood or transport direct routing, which includes 4 bytes of transport codes after the header.

is_marked_do_not_retransmit ¶

is_marked_do_not_retransmit()

Check if this packet is marked to prevent retransmission.

Returns:

Name	Type	Description
`bool`	`bool`	True if the packet should not be retransmitted/forwarded. This indicates the packet has reached its destination or should remain local to the receiving node.

is_route_direct ¶

is_route_direct()

Check if this packet uses direct routing (with or without transport codes).

Returns:

Name	Type	Description
`bool`	`bool`	True if the packet uses any form of direct routing.

is_route_flood ¶

is_route_flood()

Check if this packet uses flood routing (with or without transport codes).

Returns:

Name	Type	Description
`bool`	`bool`	True if the packet uses any form of flood routing.

mark_do_not_retransmit ¶

mark_do_not_retransmit()

Mark this packet to prevent retransmission.

Sets a flag indicating this packet should not be forwarded by repeaters. This is typically set when a packet has been successfully delivered to its intended destination to prevent unnecessary network traffic.

Used by destination nodes after successfully decrypting and processing a message intended for them.

read_from ¶

read_from(data)

Deserialize a C++ wire-format packet from bytes.

Parses the binary packet data received over the network and populates the packet fields. The format must match the C++ Packet::readFrom() exactly.

Parameters:

Name	Type	Description	Default
`data`	`ByteString`	Raw packet data in wire format.	required

Returns:

Name	Type	Description
`bool`	`bool`	True if deserialization was successful.

Raises:

Type	Description
`ValueError`	If the packet format is invalid, truncated, or contains invalid values (e.g., path_len too large, invalid payload size).

set_path ¶

set_path(path_bytes, path_len_encoded=None)

Set the routing path with optional encoded path_len.

Parameters:

Name	Type	Description	Default
`path_bytes`	`bytes`	Raw path bytes to set.	required
`path_len_encoded`	`int`	Pre-encoded path_len byte. If None, assumes 1-byte hashes and encodes len(path_bytes) as the hop count.	`None`

write_to ¶

write_to()

Serialize the packet to a byte sequence compatible with C++ Packet::writeTo().

Creates a wire-format byte representation of the packet that can be transmitted over the mesh network. The format matches the C++ implementation exactly.

Returns:

Name	Type	Description
`bytes`	`bytes`	Serialized packet data in the format: `header(1) \| [transport_codes(4)] \| path_len(1) \| path(N) \| payload(M)` Transport codes are only included if has_transport_codes() is True.

Raises:

Type	Description
`ValueError`	If internal length values don't match actual buffer lengths, indicating data corruption or incorrect packet construction.

LocalIdentity¶

pymc_core.protocol.LocalIdentity ¶

LocalIdentity(seed=None)

Bases: Identity

Represents the local node's cryptographic identity with full key access.

Extends Identity with private key operations for signing and ECDH. Generates or derives Ed25519 and X25519 key pairs for secure communication.

Initialise a LocalIdentity instance with signing and encryption keys.

Creates a local identity with Ed25519 and X25519 key pairs for digital signatures and ECDH key agreement. If no seed is provided, generates a new random key pair.

Parameters:

Name	Type	Description	Default
`seed`	`Optional[bytes]`	Optional 32 or 64-byte seed. 32-byte for standard PyNaCl key generation, 64-byte for MeshCore firmware expanded key format [scalar\|\|nonce].	`None`

get_address_bytes ¶

get_address_bytes()

Get the address bytes derived from the public key.

Returns:

Type	Description
`bytes`	The first byte of SHA256 hash of the public key, used as address.

get_private_key ¶

get_private_key()

Get the X25519 private key for ECDH operations.

Returns:

Type	Description
`bytes`	The 32-byte X25519 private key.

get_shared_public_key ¶

get_shared_public_key()

Get the X25519 public key for ECDH operations.

Returns:

Type	Description
`bytes`	The 32-byte X25519 public key.

get_signing_key_bytes ¶

get_signing_key_bytes()

Get the signing key bytes for this identity.

For standard keys, returns the 32-byte Ed25519 seed. For firmware keys, returns the 64-byte expanded key format [scalar||nonce].

Returns:

Type	Description
`bytes`	The signing key bytes (32 or 64 bytes depending on key type).

sign ¶

sign(message)

Sign a message with the Ed25519 private key.

Parameters:

Name	Type	Description	Default
`message`	`bytes`	The message to sign.	required

Returns:

Type	Description
`bytes`	The 64-byte Ed25519 signature.

Protocol Modules¶

PacketBuilder¶

pymc_core.protocol.PacketBuilder ¶

Factory class for building mesh network packets with encryption and routing.

Provides static methods to construct various types of mesh network packets including text messages, advertisements, acknowledgements, and protocol requests. Handles encryption, authentication, and proper packet formatting for the mesh protocol.

All methods are static and thread-safe. Packets are constructed with proper headers, encryption, and routing information for reliable mesh communication.

create_ack `staticmethod` ¶

create_ack(pubkey, timestamp, attempt, text)

Create an acknowledgement packet for message delivery confirmation.

Generates a compact ACK packet that confirms receipt of a message with the specified timestamp and attempt number. The ACK includes a truncated hash for efficient validation.

Parameters:

Name	Type	Description	Default
`pubkey`	`bytes`	32-byte public key of the message sender.	required
`timestamp`	`int`	Unix timestamp from the original message.	required
`attempt`	`int`	Retry attempt number (0-3) from the original message.	required
`text`	`Union[str, bytes, memoryview]`	Confirmation text or additional ACK data.	required

Returns:

Name	Type	Description
`Packet`	`Packet`	ACK packet ready for transmission.

Raises:

Type	Description
`ValueError`	If pubkey is not exactly 32 bytes.

Example

pubkey = bytes(32)  # 32-byte public key
packet = PacketBuilder.create_ack(pubkey, 1234567890, 0, "delivered")
packet.get_payload_type()
# Returns: 2

create_advert `staticmethod` ¶

create_advert(local_identity, name, lat=0.0, lon=0.0, feature1=0, feature2=0, flags=ADVERT_FLAG_IS_CHAT_NODE, route_type='flood')

Create a user advertisement packet with location and feature information.

Generates a signed advertisement packet announcing the node's presence, location, and capabilities to the mesh network. The packet includes cryptographic signatures for authenticity.

Parameters:

Name	Type	Description	Default
`local_identity`	`Any`	Local node identity for signing the advertisement.	required
`name`	`str`	Display name for the node (max 31 characters).	required
`lat`	`float`	Latitude in decimal degrees (optional).	`0.0`
`lon`	`float`	Longitude in decimal degrees (optional).	`0.0`
`feature1`	`int`	First feature flag value (optional).	`0`
`feature2`	`int`	Second feature flag value (optional).	`0`
`flags`	`int`	Advertisement flags (default: chat node).	`ADVERT_FLAG_IS_CHAT_NODE`
`route_type`	`str`	Routing method ("flood" or "direct").	`'flood'`

Returns:

Name	Type	Description
`Packet`	`Packet`	Signed advertisement packet ready for broadcast.

Example

from pymc_core.protocol.identity import LocalIdentity
identity = LocalIdentity()
packet = PacketBuilder.create_advert(identity, "MyNode", 37.7749, -122.4194)
packet.get_payload_type()
# Returns: 3

create_anon_req `staticmethod` ¶

create_anon_req(dest, local_identity, shared_secret, plaintext, route_type='transport_flood')

Create an anonymous request packet for unauthenticated communication.

Generates a packet for anonymous requests that don't require full authentication, such as initial contact or public services.

Parameters:

Name	Type	Description	Default
`dest`	`Any`	Destination identity or contact.	required
`local_identity`	`LocalIdentity`	Local node identity.	required
`shared_secret`	`bytes`	Pre-computed shared secret for encryption.	required
`plaintext`	`bytes`	Unencrypted request data.	required
`route_type`	`str`	Routing method (default: transport_flood).	`'transport_flood'`

Returns:

Name	Type	Description
`Packet`	`Packet`	Anonymous request packet with encryption.

create_datagram `staticmethod` ¶

create_datagram(ptype, dest, local_identity, secret, plaintext, route_type='direct')

Create an encrypted datagram packet for secure communication.

Generates a generic encrypted packet for text messages, requests, or responses with end-to-end encryption using the provided secret.

Parameters:

Name	Type	Description	Default
`ptype`	`int`	Payload type (TXT_MSG, REQ, or RESPONSE).	required
`dest`	`Identity`	Destination identity for the packet.	required
`local_identity`	`LocalIdentity`	Local node identity for authentication.	required
`secret`	`bytes`	Shared secret for encryption.	required
`plaintext`	`bytes`	Unencrypted payload data.	required
`route_type`	`str`	Routing method ("direct" or "flood").	`'direct'`

Returns:

Name	Type	Description
`Packet`	`Packet`	Encrypted datagram packet ready for transmission.

Raises:

Type	Description
`ValueError`	If payload type is not supported.

Example

from pymc_core.protocol.identity import Identity, LocalIdentity
dest = Identity(bytes(32))
local = LocalIdentity()
secret = bytes(32)
packet = PacketBuilder.create_datagram(0, dest, local, secret, b"hello")
packet.get_payload_type()
# Returns: 0

create_direct_advert `staticmethod` ¶

create_direct_advert(*args, **kwargs)

Create an advertisement packet with direct routing.

Convenience method that creates an advertisement with route_type="direct". All other arguments are passed through to create_advert().

Returns:

Name	Type	Description
`Packet`	`Packet`	Advertisement packet configured for direct routing.

create_discovery_request `staticmethod` ¶

create_discovery_request(tag, filter_mask, since=0, prefix_only=False)

Create a node discovery request packet.

Generates a control packet to discover nearby nodes on the mesh network. This is a zero-hop broadcast packet that nearby nodes will respond to.

Parameters:

Name	Type	Description	Default
`tag`	`int`	Random identifier to match responses (uint32_t).	required
`filter_mask`	`int`	Bitmask of node types to discover; the bit at position `node_type` is set to select that type (e.g., for ADV_TYPE_REPEATER=2, use (1 << 2) == 0x04).	required
`since`	`int`	Optional timestamp - only nodes modified after this respond (uint32_t).	`0`
`prefix_only`	`bool`	Request 8-byte key prefix instead of full 32-byte key.	`False`

Returns:

Name	Type	Description
`Packet`	`Packet`	Discovery request packet ready to send as zero-hop.

Example

import random
tag = random.randint(0, 0xFFFFFFFF)
# Filter for repeaters: ADV_TYPE_REPEATER=2, so (1 << 2) = 0x04
packet = PacketBuilder.create_discovery_request(tag, filter_mask=0x04)
# Send as zero-hop broadcast

create_discovery_response `staticmethod` ¶

create_discovery_response(tag, node_type, inbound_snr, pub_key, prefix_only=False)

Create a node discovery response packet.

Generates a control packet in response to a discovery request. This is sent as a zero-hop packet to the requester.

Parameters:

Name	Type	Description	Default
`tag`	`int`	Tag from the discovery request to match.	required
`node_type`	`int`	Type of this node (0-15, e.g., 1 for repeater).	required
`inbound_snr`	`float`	SNR of the received request (will be multiplied by 4).	required
`pub_key`	`bytes`	Node's public key (32 bytes).	required
`prefix_only`	`bool`	Send only 8-byte key prefix instead of full key.	`False`

Returns:

Name	Type	Description
`Packet`	`Packet`	Discovery response packet ready to send as zero-hop.

Example

identity = LocalIdentity()
packet = PacketBuilder.create_discovery_response(
    tag=0x12345678,
    node_type=1,  # Repeater
    inbound_snr=8.5,
    pub_key=identity.get_public_key()
)

create_flood_advert `staticmethod` ¶

create_flood_advert(*args, **kwargs)

Create an advertisement packet with flood routing.

Convenience method that creates an advertisement with route_type="flood". All other arguments are passed through to create_advert().

Returns:

Name	Type	Description
`Packet`	`Packet`	Advertisement packet configured for flood routing.

create_group_data_packet `staticmethod` ¶

create_group_data_packet(ptype, channel_hash, channel_secret, plaintext, secret)

Create a group packet with generic encrypted data.

Generates a group packet for text messages or data with channel-specific encryption using the provided shared secret.

Parameters:

Name	Type	Description	Default
`ptype`	`int`	Payload type (GRP_TXT or GRP_DATA).	required
`channel_hash`	`int`	Single byte hash identifying the channel.	required
`channel_secret`	`bytes`	Channel-specific encryption secret.	required
`plaintext`	`bytes`	Unencrypted data to send.	required
`secret`	`bytes`	Additional secret for encryption.	required

Returns:

Name	Type	Description
`Packet`	`Packet`	Encrypted group data packet.

Raises:

Type	Description
`ValueError`	If payload type is not supported for groups.

create_group_datagram `staticmethod` ¶

create_group_datagram(group_name, local_identity, message, sender_name='Unknown', channels_config=None)

Create an encrypted group message for a specified channel.

Generates a group message packet encrypted with the channel's shared secret, allowing secure communication within a named group or channel.

Parameters:

Name	Type	Description	Default
`group_name`	`str`	Name of the channel to send the message to.	required
`local_identity`	`LocalIdentity`	Local node identity (unused in group messages).	required
`message`	`str`	Message text to send to the group.	required
`sender_name`	`str`	Display name of the sender (default: "Unknown").	`'Unknown'`
`channels_config`	`Optional[Any]`	List of channel configurations with secrets.	`None`

Returns:

Name	Type	Description
`Packet`	`Packet`	Encrypted group message packet.

Raises:

Type	Description
`ValueError`	If channels_config is None or channel not found.

Example

channels = [{"name": "general", "secret": "secret123"}]
from pymc_core.protocol.identity import LocalIdentity
identity = LocalIdentity()
packet = PacketBuilder.create_group_datagram(
    "general", identity, "Hello group!", "Alice", channels)
packet.get_payload_type()
# Returns: 6

create_login_packet `staticmethod` ¶

create_login_packet(contact, local_identity, password)

Create a login packet for repeater authentication.

Generates an encrypted login packet containing credentials for authenticating with a repeater node or room server.

Parameters:

Name	Type	Description	Default
`contact`	`Any`	Contact information for the repeater.	required
`local_identity`	`LocalIdentity`	Local node identity for encryption.	required
`password`	`str`	Authentication password (truncated to 15 chars).	required

Returns:

Name	Type	Description
`Packet`	`Packet`	Encrypted login packet ready for transmission.

create_logout_packet `staticmethod` ¶

create_logout_packet(contact, local_identity)

Create a logout packet for repeater authentication.

Generates a logout message to terminate an authenticated session with a repeater node.

Parameters:

Name	Type	Description	Default
`contact`	`Any`	The repeater contact to logout from.	required
`local_identity`	`LocalIdentity`	The local node identity for encryption.	required

Returns:

Name	Type	Description
`tuple`	`tuple[Packet, int]`	(packet, crc) - The logout packet and CRC for verification.

create_path_return `staticmethod` ¶

create_path_return(dest_hash, src_hash, secret, path, extra_type=255, extra=b'', path_len_encoded=None)

Create a secure return path packet with optional metadata.

Generates an encrypted packet containing a return path for secure two-way communication, with optional additional data.

The inner payload first byte is the encoded path_len (bits 6-7 = hash size - 1, bits 0-5 = hop count). When path_len_encoded is provided and valid, it is used so 2- and 3-byte path hashes match firmware. When path_len_encoded is None, the first byte is len(path) (1-byte hash semantics).

Parameters:

Name	Type	Description	Default
`dest_hash`	`int`	Destination node hash (1 byte).	required
`src_hash`	`int`	Source node hash (1 byte).	required
`secret`	`bytes`	Shared secret for encryption.	required
`path`	`Sequence[int]`	Sequence of node hashes for the return path.	required
`extra_type`	`int`	Type identifier for extra data (default: 0xFF).	`255`
`extra`	`bytes`	Additional binary data to include.	`b''`
`path_len_encoded`	`Optional[int]`	Encoded path_len byte from the original packet. If None, first byte of inner payload is len(path) (1-byte hashes).	`None`

Returns:

Name	Type	Description
`Packet`	`Packet`	Encrypted return path packet.

Raises:

Type	Description
`ValueError`	If combined path and extra data exceed packet limits, or if path_len_encoded is provided but path length does not match.

create_protocol_request `staticmethod` ¶

create_protocol_request(contact, local_identity, protocol_code, data=b'', timestamp=None)

Create a protocol request packet for repeater commands.

Generates an encrypted protocol request for administrative commands or special operations with repeaters and infrastructure nodes.

Parameters:

Name	Type	Description	Default
`contact`	`Any`	The repeater contact to send the request to.	required
`local_identity`	`LocalIdentity`	The local node identity for encryption.	required
`protocol_code`	`int`	The protocol command code.	required
`data`	`bytes`	Additional binary data for the request.	`b''`
`timestamp`	`Optional[int]`	Optional timestamp (uses current time if None).	`None`

Returns:

Name	Type	Description
`tuple`	`tuple[Packet, int]`	(packet, timestamp) - The created packet and the timestamp used.

Example

from pymc_core.protocol.identity import LocalIdentity
identity = LocalIdentity()
contact = type('Contact', (), {'public_key': '00'*32})()
packet, ts = PacketBuilder.create_protocol_request(
    contact, identity, 1, b"data")
packet.get_payload_type()
# Returns: 4

create_raw_data `staticmethod` ¶

create_raw_data(data)

Create a raw custom packet (PAYLOAD_TYPE_RAW_CUSTOM) with no encryption.

Route type is always DIRECT (consistent with firmware CMD_SEND_RAW_DATA). Caller must set pkt.path and pkt.path_len for direct routing.

create_self_advert `staticmethod` ¶

create_self_advert(local_identity, name, lat=0.0, lon=0.0, feature1=0, feature2=0, route_type='flood')

Create a self-advertisement packet for the local node.

Convenience method that creates an advertisement packet with the IS_CHAT_NODE flag set, announcing the local node's presence.

Parameters:

Name	Type	Description	Default
`local_identity`	`Any`	Local node identity for signing.	required
`name`	`str`	Display name for the node.	required
`lat`	`float`	Latitude in decimal degrees.	`0.0`
`lon`	`float`	Longitude in decimal degrees.	`0.0`
`feature1`	`int`	First feature flag value.	`0`
`feature2`	`int`	Second feature flag value.	`0`
`route_type`	`str`	Routing method ("flood" or "direct").	`'flood'`

Returns:

Name	Type	Description
`Packet`	`Packet`	Signed advertisement packet with chat node flag.

create_telem_request `staticmethod` ¶

create_telem_request(contact, local_identity, *, want_base=True, want_location=True, want_environment=True, include_entropy=True, route_type='direct')

Create a telemetry request packet for sensor data collection.

Generates a request for telemetry data from a node, allowing selective retrieval of base metrics, location data, and environmental sensors.

Parameters:

Name	Type	Description	Default
`contact`	`Any`	The node to request telemetry from.	required
`local_identity`	`LocalIdentity`	The local node identity for encryption.	required
`want_base`	`bool`	Include basic telemetry metrics.	`True`
`want_location`	`bool`	Include location/GPS data.	`True`
`want_environment`	`bool`	Include environmental sensors.	`True`
`include_entropy`	`bool`	Include entropy/randomness data.	`True`
`route_type`	`str`	Routing method ("direct" or "flood").	`'direct'`

Returns:

Name	Type	Description
`tuple`	`tuple[Packet, int]`	(packet, timestamp) - The telemetry request packet and timestamp.

Example

from pymc_core.protocol.identity import LocalIdentity
identity = LocalIdentity()
contact = type('Contact', (), {'public_key': '00'*32})()
packet, ts = PacketBuilder.create_telem_request(
    contact, identity, want_location=False)
packet.get_payload_type()
# Returns: 4

create_text_message `staticmethod` ¶

create_text_message(contact, local_identity, message, attempt=0, message_type='direct', out_path=None, txt_type=0)

Create a secure text message with encryption and CRC validation.

Generates an encrypted text message packet with proper authentication, CRC calculation for ACK verification, and optional routing path.

Parameters:

Name	Type	Description	Default
`contact`	`Any`	The contact to send the message to.	required
`local_identity`	`LocalIdentity`	The local node identity for encryption.	required
`message`	`str`	The message text to send.	required
`attempt`	`int`	The attempt number for retries (0-3).	`0`
`message_type`	`str`	The message routing type ("direct" or "flood").	`'direct'`
`out_path`	`Optional[list]`	The optional routing path for directed messages.	`None`
`txt_type`	`int`	Text type in upper 6 bits of the flags byte (0=PLAIN, 1=CLI_DATA, …), combined with attempt as `(txt_type << 2) \| (attempt & 3)`. Matches MeshCore `TXT_TYPE_*` so repeaters skip delivery ACK for CLI_DATA.	`0`

Returns:

Name	Type	Description
`tuple`	`tuple[Packet, int]`	(packet, crc) - The encrypted packet and CRC for ACK verification.

Example

from pymc_core.protocol.identity import LocalIdentity
identity = LocalIdentity()
contact = type('Contact', (), {'public_key': '00'*32, 'out_path': []})()
packet, crc = PacketBuilder.create_text_message(
    contact, identity, "Hello!", 0, "direct")
packet.get_payload_type()
# Returns: 0

create_trace `staticmethod` ¶

create_trace(tag, auth_code, flags, path=None)

Create a trace packet for network diagnostics and path discovery.

Generates a trace packet that can follow network paths for debugging and network topology discovery. Compatible with C++ implementation.

Parameters:

Name	Type	Description	Default
`tag`	`int`	Random identifier set by initiator (uint32_t).	required
`auth_code`	`int`	Optional authentication code (uint32_t).	required
`flags`	`int`	Control flags for trace behavior (uint8_t).	required
`path`	`Optional[Sequence[int]]`	Optional list of node IDs for the trace path.	`None`

Returns:

Name	Type	Description
`Packet`	`Packet`	Trace packet with proper wire format.

Example

packet = PacketBuilder.create_trace(12345, 0, 1, [1, 2, 3])
packet.get_payload_type()
# Returns: 7

PacketFilter¶

pymc_core.protocol.PacketFilter ¶

PacketFilter(window_seconds=30)

Lightweight packet filter for dispatcher routing decisions.

blacklist ¶

blacklist(packet_hash)

Add a packet hash to the blacklist.

cleanup_old_hashes ¶

cleanup_old_hashes()

Clean up old packet hashes beyond the deduplication window.

clear ¶

clear()

Clear all tracked data.

generate_hash ¶

generate_hash(data)

Generate a hash for packet data.

get_stats ¶

get_stats()

Get basic filter statistics.

is_blacklisted ¶

is_blacklisted(packet_hash)

Check if a packet hash is blacklisted.

is_duplicate ¶

is_duplicate(packet_hash)

Check if we've seen this packet recently.

track_packet ¶

track_packet(packet_hash)

Track a packet hash with current timestamp.

Hardware Modules¶

LoRaRadio (Base)¶

pymc_core.hardware.base.LoRaRadio ¶

Bases: ABC

begin `abstractmethod` ¶

begin()

Initialise the radio module.

get_last_rssi `abstractmethod` ¶

get_last_rssi()

Return last received RSSI in dBm.

get_last_snr `abstractmethod` ¶

get_last_snr()

Return last received SNR in dB.

send `abstractmethod` `async` ¶

send(data)

Send a packet asynchronously. Returns transmission metadata dict or None.

sleep `abstractmethod` ¶

sleep()

Put the radio into low-power mode.

wait_for_rx `abstractmethod` `async` ¶

wait_for_rx()

Wait for a packet to be received asynchronously.

SX1262Radio¶

pymc_core.hardware.sx1262_wrapper.SX1262Radio ¶

SX1262Radio(bus_id=0, cs_id=0, cs_pin=-1, gpio_chip=0, use_gpiod_backend=False, reset_pin=18, busy_pin=20, irq_pin=16, txen_pin=6, rxen_pin=-1, txled_pin=-1, rxled_pin=-1, en_pin=-1, frequency=868000000, tx_power=22, spreading_factor=7, bandwidth=125000, coding_rate=5, preamble_length=12, sync_word=13380, is_waveshare=False, use_dio3_tcxo=False, dio3_tcxo_voltage=1.8, use_dio2_rf=False)

Bases: LoRaRadio

SX1262 LoRa Radio implementation for Raspberry Pi

Initialize SX1262 radio

Parameters:

Name	Type	Description	Default
`bus_id`	`int`	SPI bus ID (default: 0)	`0`
`cs_id`	`int`	SPI chip select ID (default: 0)	`0`
`cs_pin`	`int`	Manual CS GPIO pin (-1 = use hardware CS, e.g. 21 for Waveshare HAT)	`-1`
`gpio_chip`	`int`	GPIO chip select ID (default: 0)	`0`
`use_gpiod_backend`	`bool`	Use alternative backend for GPIO support (default: False)	`False`
`reset_pin`	`int`	GPIO pin for reset (default: 18)	`18`
`busy_pin`	`int`	GPIO pin for busy signal (default: 20)	`20`
`irq_pin`	`int`	GPIO pin for interrupt (default: 16)	`16`
`txen_pin`	`int`	GPIO pin for TX enable (default: 6)	`6`
`rxen_pin`	`int`	GPIO pin for RX enable (default: -1 if not used)	`-1`
`txled_pin`	`int`	GPIO pin for TX LED (default: -1 if not used)	`-1`
`rxled_pin`	`int`	GPIO pin for RX LED (default: -1 if not used)	`-1`
`en_pin`	`int`	GPIO pin for powering up the radio goes high on init	`-1`
`frequency`	`int`	Operating frequency in Hz (default: 868MHz)	`868000000`
`tx_power`	`int`	TX power in dBm (default: 22)	`22`
`spreading_factor`	`int`	LoRa spreading factor (default: 7)	`7`
`bandwidth`	`int`	Bandwidth in Hz (default: 125kHz)	`125000`
`coding_rate`	`int`	Coding rate (default: 5 for 4/5)	`5`
`preamble_length`	`int`	Preamble length (default: 12)	`12`
`sync_word`	`int`	Sync word (default: 0x3444 for public network)	`13380`
`is_waveshare`	`bool`	Use alternate initialization needed for Waveshare HAT	`False`
`use_dio3_tcxo`	`bool`	Enable DIO3 TCXO control (default: False)	`False`
`dio3_tcxo_voltage`	`float`	TCXO reference voltage in volts (default: 1.8)	`1.8`
`use_dio2_rf`	`bool`	Enable DIO2 as RF switch control (default: False)	`False`

begin ¶

begin()

Initialize the SX1262 radio module. Returns True if successful, False otherwise.

check_radio_health ¶

check_radio_health()

Simple health check - restart RX task if it's dead.

cleanup ¶

cleanup()

Clean up radio resources

clear_custom_cad_thresholds ¶

clear_custom_cad_thresholds()

Clear custom CAD thresholds and revert to defaults.

get_instance `classmethod` ¶

get_instance(**kwargs)

Get the active instance or create a new one (singleton-like behavior)

get_last_rssi ¶

get_last_rssi()

Return last received RSSI in dBm

get_last_signal_rssi ¶

get_last_signal_rssi()

Return last received signal RSSI in dBm

get_last_snr ¶

get_last_snr()

Return last received SNR in dB

get_noise_floor ¶

get_noise_floor()

Get current noise floor in dBm. Returns properly sampled noise floor from background measurements.

get_status ¶

get_status()

Get radio status information

perform_cad `async` ¶

perform_cad(det_peak=None, det_min=None, timeout=1.0, calibration=False)

Perform Channel Activity Detection (CAD). If calibration=True, uses provided thresholds and returns info. If calibration=False, uses pre-calibrated/default thresholds.

Returns:

Name	Type	Description
`bool`	`Union[bool, dict]`	Channel activity detected (when calibration=False)
`dict`	`Union[bool, dict]`	Calibration data (when calibration=True)

send `async` ¶

send(data)

Send a packet asynchronously. Returns transmission metadata including LBT metrics.

set_bandwidth ¶

set_bandwidth(bw)

Set bandwidth in Hz

set_custom_cad_thresholds ¶

set_custom_cad_thresholds(peak, min_val)

Set custom CAD thresholds that override the defaults.

Parameters:

Name	Type	Description	Default
`peak`	`int`	CAD detection peak threshold (0-31)	required
`min_val`	`int`	CAD detection minimum threshold (0-31)	required

set_frequency ¶

set_frequency(frequency)

Set operating frequency

set_rx_callback ¶

set_rx_callback(callback)

Set a callback to be called with each received packet (bytes).

set_spreading_factor ¶

set_spreading_factor(sf)

Set spreading factor (6-12)

set_tx_power ¶

set_tx_power(power)

Set TX power in dBm

sleep ¶

sleep()

Put the radio into low-power sleep mode

wait_for_rx `async` ¶

wait_for_rx()

Not implemented: use set_rx_callback instead.

WsRadio¶

pymc_core.hardware.wsradio.WsRadio ¶

WsRadio(ip_address='192.168.0.33', port=81, timeout=30, radio_config=None)

Bases: LoRaRadio

begin ¶

begin()

Initialize the websocket radio (connection happens on first use)

send `async` ¶

send(data)

Send packet via WebSocket. Returns None (no metadata available).

Constants¶

pymc_core.protocol.constants ¶

Mesh protocol constants extracted from the C++ firmware.

API Reference¶

Core Modules¶

MeshNode¶

pymc_core.node.MeshNode ¶

send_packet async ¶

set_event_service ¶

start async ¶

stop ¶

Packet¶

pymc_core.protocol.Packet ¶

rssi property ¶

snr property ¶

apply_path_hash_mode ¶

calculate_packet_hash ¶

get_crc ¶

get_packet_hash_hex ¶

get_path_byte_len ¶

get_path_hash_count ¶

get_path_hash_size ¶

get_path_hashes ¶

get_path_hashes_hex ¶

get_payload ¶

get_payload_app_data ¶

get_payload_type ¶

get_payload_ver ¶

get_raw_length ¶

get_route_type ¶

get_snr ¶

has_transport_codes ¶

is_marked_do_not_retransmit ¶

is_route_direct ¶

is_route_flood ¶

mark_do_not_retransmit ¶

read_from ¶

set_path ¶

write_to ¶

LocalIdentity¶

pymc_core.protocol.LocalIdentity ¶

get_address_bytes ¶

get_private_key ¶

get_shared_public_key ¶

get_signing_key_bytes ¶

sign ¶

Protocol Modules¶

PacketBuilder¶

pymc_core.protocol.PacketBuilder ¶

create_ack staticmethod ¶

create_advert staticmethod ¶

create_anon_req staticmethod ¶

create_datagram staticmethod ¶

create_direct_advert staticmethod ¶

create_discovery_request staticmethod ¶

create_discovery_response staticmethod ¶

create_flood_advert staticmethod ¶

create_group_data_packet staticmethod ¶

create_group_datagram staticmethod ¶

create_login_packet staticmethod ¶

create_logout_packet staticmethod ¶

create_path_return staticmethod ¶

create_protocol_request staticmethod ¶

create_raw_data staticmethod ¶

create_self_advert staticmethod ¶

create_telem_request staticmethod ¶

create_text_message staticmethod ¶

create_trace staticmethod ¶

PacketFilter¶

pymc_core.protocol.PacketFilter ¶

blacklist ¶

cleanup_old_hashes ¶

clear ¶

generate_hash ¶

get_stats ¶

is_blacklisted ¶

is_duplicate ¶

track_packet ¶

Hardware Modules¶

LoRaRadio (Base)¶

pymc_core.hardware.base.LoRaRadio ¶

begin abstractmethod ¶

get_last_rssi abstractmethod ¶

send_packet `async` ¶

start `async` ¶

rssi `property` ¶

snr `property` ¶

create_ack `staticmethod` ¶

create_advert `staticmethod` ¶

create_anon_req `staticmethod` ¶

create_datagram `staticmethod` ¶

create_direct_advert `staticmethod` ¶

create_discovery_request `staticmethod` ¶

create_discovery_response `staticmethod` ¶

create_flood_advert `staticmethod` ¶

create_group_data_packet `staticmethod` ¶

create_group_datagram `staticmethod` ¶

create_login_packet `staticmethod` ¶

create_logout_packet `staticmethod` ¶

create_path_return `staticmethod` ¶

create_protocol_request `staticmethod` ¶

create_raw_data `staticmethod` ¶

create_self_advert `staticmethod` ¶

create_telem_request `staticmethod` ¶

create_text_message `staticmethod` ¶

create_trace `staticmethod` ¶

begin `abstractmethod` ¶

get_last_rssi `abstractmethod` ¶

get_last_snr `abstractmethod` ¶

send `abstractmethod` `async` ¶

sleep `abstractmethod` ¶

wait_for_rx `abstractmethod` `async` ¶

get_instance `classmethod` ¶

perform_cad `async` ¶

send `async` ¶

wait_for_rx `async` ¶

send `async` ¶