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)

Represents a node in a mesh network for radio communication.

Manages radio communication, message routing, and protocol handling within a mesh network. Provides high-level APIs for sending messages, telemetry requests, and commands to other nodes and repeaters.

The node integrates with various components like contact storage, channel databases, and event services for comprehensive mesh functionality.

Initialise a mesh network node instance.

Sets up the node's core components including radio interface, identity management, and communication handlers.

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_group_text async

send_group_text(group_name, message)

Broadcast a text message to all members of a group.

Sends a group datagram that will be received by all nodes configured for the specified group. Group messages are fire-and-forget with no acknowledgements expected.

Parameters:

Name	Type	Description	Default
group_name	str	Name of the group to broadcast to.	required
message	str	Text content to broadcast.	required

Returns:

Name	Type	Description
	dict	Dictionary with transmission results and signal metrics.
Note	dict	Group messages don't wait for acknowledgements.

Example

result = await node.send_group_text("team_alpha", "Meeting at 15:00")
print(f"Broadcast to {result['group']}: {result['success']}")

send_login async

send_login(repeater_name, password)

Authenticate with a repeater node.

Sends login credentials to a repeater and waits for authentication response. Successful login may grant administrative privileges.

Parameters:

Name	Type	Description	Default
repeater_name	str	Name of the repeater to authenticate with.	required
password	str	Authentication password for the repeater.	required

Returns:

Type	Description
dict	Dictionary with login results including success status,
dict	admin privileges, and keep-alive intervals.

Raises:

Type	Description
RuntimeError	If repeater contact not found.

Example

result = await node.send_login("repeater_01", "secret123")
if result["success"] and result["is_admin"]:
    print("Admin access granted")

send_logout async

send_logout(repeater_name)

Terminate authentication session with a repeater.

Sends a logout command to end the current session with a repeater. This should be called when finished with repeater operations.

Parameters:

Name	Type	Description	Default
repeater_name	str	Name of the repeater to logout from.	required

Returns:

Type	Description
dict	Dictionary with logout results and performance metrics.

Raises:

Type	Description
RuntimeError	If repeater contact not found.

Example

result = await node.send_logout("repeater_01")
print(f"Logout {'successful' if result['success'] else 'failed'}")

send_protocol_request async

send_protocol_request(repeater_name, protocol_code, data=b'')

Send a protocol-specific request to a repeater.

Transmits a custom protocol request with optional data payload and waits for the repeater's response.

Parameters:

Name	Type	Description	Default
repeater_name	str	Name of the repeater to send request to.	required
protocol_code	int	Protocol operation code (0-255).	required
data	bytes	Optional binary data payload for the request.	b''

Returns:

Type	Description
dict	Dictionary with protocol response, parsed data, and timing metrics.

Raises:

Type	Description
RuntimeError	If repeater contact or protocol handler not found.

Example

result = await node.send_protocol_request("repeater_01", 0x10, b"config")
if result["success"]:
    print(f"Response: {result['response']}")

send_repeater_command async

send_repeater_command(repeater_name, command, parameters=None)

Send a text-based command to a repeater and await response.

Transmits a command string to a repeater using the text message protocol and waits for a response. Useful for administrative operations and status queries.

Parameters:

Name	Type	Description	Default
repeater_name	str	Name of the repeater to send command to.	required
command	str	Command string to execute on the repeater.	required
parameters	Optional[str]	Optional parameters for the command.	None

Returns:

Type	Description
dict	Dictionary with command results, response text, and timing data.

Raises:

Type	Description
RuntimeError	If repeater contact not found.

Example

result = await node.send_repeater_command("repeater_01", "status")
if result["success"]:
    print(f"Response: {result['response']}")

send_status_request async

send_status_request(repeater_name)

Request status information from a repeater.

Queries a repeater for its current operational status and configuration. This is a convenience method that uses the text command interface.

Parameters:

Name	Type	Description	Default
repeater_name	str	Name of the repeater to query.	required

Returns:

Type	Description
dict	Dictionary with status information and response metrics.

Raises:

Type	Description
RuntimeError	If repeater contact not found.

Example

status = await node.send_status_request("repeater_01")
if status["success"]:
    print(f"Status: {status['response']}")

send_telemetry_request async

send_telemetry_request(contact_name, want_base=True, want_location=True, want_environment=True, timeout=10.0)

Request telemetry data from a contact node.

Sends a telemetry request and waits for the target node to respond with requested sensor data including base metrics, location, and environmental readings.

Parameters:

Name	Type	Description	Default
contact_name	str	Name of the contact to query.	required
want_base	bool	Include basic telemetry metrics in request.	True
want_location	bool	Include GPS/location data in request.	True
want_environment	bool	Include environmental sensors in request.	True
timeout	float	Maximum time to wait for response in seconds.	10.0

Returns:

Type	Description
dict	Dictionary with request results, telemetry data, and performance
dict	metrics including round-trip time.

Raises:

Type	Description
RuntimeError	If contact not found or protocol handler unavailable.

Example

result = await node.send_telemetry_request("sensor_node")
if result["success"]:
    print(f"Temperature: {result['telemetry_data'].get('temp')}")

send_text async

send_text(contact_name, message, attempt=1, message_type='direct', out_path=None)

Send a text message to a specified contact.

Transmits a text message to another node in the mesh network, with optional routing and retry configuration.

Parameters:

Name	Type	Description	Default
contact_name	str	Name of the target contact in the contact book.	required
message	str	Text content to send.	required
attempt	int	Message attempt number for retry logic (default: 1).	1
message_type	str	Routing type - "direct" or other supported types.	'direct'
out_path	Optional[list]	Optional list of intermediate nodes for routing.	None

Returns:

Type	Description
dict	Dictionary containing transmission results including success status,
dict	signal strength metrics (SNR/RSSI), and routing information.

Raises:

Type	Description
RuntimeError	If the specified contact is not found.

Example

result = await node.send_text("alice", "Hello, world!")
print(result["success"])  # True if message sent successfully

send_trace_packet async

send_trace_packet(contact_name, tag, auth_code, flags=0, path=None, timeout=5.0)

Send a diagnostic trace packet for network analysis.

Transmits a trace packet to analyse routing paths and network performance. Always expects a response with trace data, signal metrics, and routing information.

Parameters:

Name	Type	Description	Default
contact_name	str	Name of the target contact for tracing.	required
tag	int	Unique identifier for this trace operation.	required
auth_code	int	Authentication code for the trace request.	required
flags	int	Optional flags to modify trace behaviour.	0
path	Optional[list]	Optional custom routing path for the trace.	None
timeout	float	Maximum time to wait for trace response.	5.0

Returns:

Type	Description
dict	Dictionary with trace results, routing data, and signal metrics.

Raises:

Type	Description
RuntimeError	If contact not found or trace handler unavailable.

Example

trace = await node.send_trace_packet("target_node", 0x12345678, 0xABCD)
if trace["success"]:
    print(f"RTT: {trace['rtt_ms']}ms")

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.

Initialises the radio interface and dispatcher, then enters the main event loop for handling incoming/outgoing messages. This method blocks until the node is stopped.

Note

This is an asynchronous operation that runs indefinitely until cancelled or the node is stopped.

stop

stop()

Stop the mesh node and clean up associated services.

Terminates radio communications and shuts down all active handlers. This method is synchronous and should be called to gracefully shut down the node.

Packet

pymc_core.protocol.Packet

Packet()

Represents a mesh network packet with header, 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.
path_len	int	Length of the path component in bytes.
path	bytearray	Variable-length path data for routing.
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
packet.path = b"node1->node2"
packet.path_len = len(packet.path)
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.

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_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) + path_len(1) + path(N) + payload(M)

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: Flood routing - 1: Direct routing - 2: Transport flood routing - 3: Reserved

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.

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).

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) | path_len(1) | path(N) | payload(M)

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-byte seed for deterministic key generation.	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.

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_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'')

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.

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''

Returns:

Name	Type	Description
Packet	Packet	Encrypted return path packet.

Raises:

Type	Description
ValueError	If combined path and extra data exceed packet limits.

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_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)

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

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.

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, reset_pin=18, busy_pin=20, irq_pin=16, txen_pin=6, rxen_pin=-1, frequency=868000000, tx_power=22, spreading_factor=7, bandwidth=125000, coding_rate=5, preamble_length=12, sync_word=13380)

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
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
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

begin

begin()

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

cleanup

cleanup()

Clean up radio resources

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_snr

get_last_snr()

Return last received SNR in dB

get_status

get_status()

Get radio status information

send async

send(data)

Send a packet asynchronously

set_bandwidth

set_bandwidth(bw)

Set bandwidth in Hz

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)

Constants

pymc_core.protocol.constants

Mesh protocol constants extracted from the C++ firmware.