mirror of
https://github.com/agessaman/meshcore-bot.git
synced 2026-11-04 11:57:16 +00:00
agessaman
fe737c9f5e
decode_meshcore_packet, the path processor, the path chunkers, format_path_string and the route/payload type-name tables are now plain functions in packet_decode.py. MessageHandler keeps every old method as a delegate that supplies the configured prefix width and its logger, so callers and tests that patch decode_meshcore_packet on a handler still work. The existing quirk is preserved: path_info uses the configured width while path/path_hex follow the packet's own bytes-per-hop. Parity: decoding a sample of 4,009 real packets from the dev database (plus malformed inputs) at prefix widths 2 and 4, with the path and format helpers, produces byte-identical output before and after. A handler built with no bot attribute now decodes path_info at width 2 instead of falling into the error branch; only bare test objects can hit that.
357 lines
14 KiB
Python
357 lines
14 KiB
Python
"""MeshCore packet decoding: header, transport codes, path and payload split.
|
||
|
||
Pure functions matching the firmware's Packet.cpp layout. MessageHandler keeps
|
||
its old methods as thin delegates that supply the configured node-prefix width
|
||
and its logger.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
from typing import Any
|
||
|
||
from .enums import PayloadType, PayloadVersion, RouteType
|
||
from .utils import decode_path_len_byte
|
||
|
||
|
||
def split_path_hex(path_hex: str, hex_chars: int) -> list[str]:
|
||
"""Split a hex path into lowercase node IDs of ``hex_chars`` each.
|
||
|
||
When the path does not divide evenly (or yields nothing), fall back to
|
||
one-byte (two-character) nodes, the legacy path encoding.
|
||
"""
|
||
nodes = [path_hex[i : i + hex_chars].lower() for i in range(0, len(path_hex), hex_chars)]
|
||
if (len(path_hex) % hex_chars) != 0 or not nodes:
|
||
nodes = [path_hex[i : i + 2].lower() for i in range(0, len(path_hex), 2)]
|
||
return nodes
|
||
|
||
|
||
def path_bytes_to_nodes(path_bytes: bytes, prefix_hex_chars: int) -> tuple[str, list[str]]:
|
||
"""Path bytes as ``(hex, uppercase node IDs)``, ``prefix_hex_chars`` per node.
|
||
|
||
A non-positive width means one byte per node; see :func:`split_path_hex`
|
||
for the fallback when the path does not divide evenly.
|
||
"""
|
||
n = prefix_hex_chars if prefix_hex_chars > 0 else 2
|
||
path_hex = path_bytes.hex()
|
||
return path_hex, [node.upper() for node in split_path_hex(path_hex, n)]
|
||
|
||
|
||
def decode_meshcore_packet(
|
||
raw_hex: str, payload_hex: str | None = None, *, prefix_hex_chars: int, logger: Any
|
||
) -> dict | None:
|
||
"""
|
||
Decode a MeshCore packet from raw hex data - matches Packet.cpp exactly
|
||
|
||
Args:
|
||
raw_hex: Raw packet data as hex string (may be RF data or direct MeshCore packet)
|
||
payload_hex: Optional extracted payload hex string (preferred over raw_hex)
|
||
|
||
Returns:
|
||
Decoded packet information or None if parsing fails
|
||
"""
|
||
# Ensure these are always defined for error logging (BUG-028)
|
||
byte_data: bytes = b""
|
||
hex_data: str = ""
|
||
try:
|
||
# Use payload_hex if provided (this is the actual MeshCore packet)
|
||
if payload_hex:
|
||
logger.debug("Using provided payload_hex for decoding")
|
||
hex_data = payload_hex
|
||
elif raw_hex:
|
||
logger.debug("Using raw_hex for decoding")
|
||
hex_data = raw_hex
|
||
else:
|
||
logger.debug("No packet data provided for decoding")
|
||
return None
|
||
|
||
# Remove 0x prefix if present (like in your other project)
|
||
if hex_data.startswith("0x"):
|
||
hex_data = hex_data[2:]
|
||
|
||
byte_data = bytes.fromhex(hex_data)
|
||
|
||
# Validate minimum packet size
|
||
if len(byte_data) < 2:
|
||
logger.error(f"Packet too short: {len(byte_data)} bytes")
|
||
return None
|
||
|
||
header = byte_data[0]
|
||
|
||
# Extract route type
|
||
route_type = RouteType(header & 0x03)
|
||
has_transport = route_type in [RouteType.TRANSPORT_FLOOD, RouteType.TRANSPORT_DIRECT]
|
||
|
||
# Calculate path length offset based on presence of transport codes
|
||
offset = 1
|
||
if has_transport:
|
||
offset += 4
|
||
|
||
# Check if we have enough data for path_len
|
||
if len(byte_data) <= offset:
|
||
logger.error(f"Packet too short for path_len at offset {offset}: {len(byte_data)} bytes")
|
||
return None
|
||
|
||
path_len_byte = byte_data[offset]
|
||
offset += 1
|
||
# Decode per firmware: low 6 bits = hop count, high 2 bits = size code (bytes_per_hop = code+1)
|
||
path_parts = decode_path_len_byte(path_len_byte)
|
||
if path_parts is None:
|
||
logger.debug("decode_meshcore_packet: invalid path_len byte (firmware would reject)")
|
||
return None
|
||
path_byte_length, bytes_per_hop = path_parts
|
||
|
||
# Check if we have enough data for the full path
|
||
if len(byte_data) < offset + path_byte_length:
|
||
logger.error(
|
||
f"Packet too short for path (need {offset + path_byte_length}, have {len(byte_data)})"
|
||
)
|
||
return None
|
||
|
||
# Extract path
|
||
path_bytes = byte_data[offset : offset + path_byte_length]
|
||
offset += path_byte_length
|
||
|
||
# Remaining data is payload
|
||
payload = byte_data[offset:]
|
||
|
||
# Extract payload version (bits 6-7)
|
||
payload_version = PayloadVersion((header >> 6) & 0x03)
|
||
|
||
# Only accept VER_1 (version 0)
|
||
if payload_version != PayloadVersion.VER_1:
|
||
logger.warning(
|
||
f"Encountered an unknown packet version. Version: {payload_version.value} RAW: {hex_data}"
|
||
)
|
||
return None
|
||
|
||
# Extract payload type (bits 2-5)
|
||
payload_type = PayloadType((header >> 2) & 0x0F)
|
||
|
||
# Chunk path by bytes_per_hop from packet (1, 2, or 3)
|
||
path_hex, path_values = path_bytes_to_nodes(path_bytes, bytes_per_hop * 2)
|
||
|
||
# Process path based on packet type
|
||
path_info = process_packet_path(
|
||
path_bytes, payload, route_type, payload_type, prefix_hex_chars=prefix_hex_chars, logger=logger
|
||
)
|
||
|
||
# Extract transport codes if present (only for TRANSPORT_FLOOD and TRANSPORT_DIRECT)
|
||
transport_codes = None
|
||
if has_transport and len(byte_data) >= 5: # header(1) + transport(4)
|
||
transport_bytes = byte_data[1:5]
|
||
transport_codes = {
|
||
"code1": int.from_bytes(transport_bytes[0:2], byteorder="little"),
|
||
"code2": int.from_bytes(transport_bytes[2:4], byteorder="little"),
|
||
"hex": transport_bytes.hex(),
|
||
}
|
||
|
||
packet_info = {
|
||
"header": f"0x{header:02x}",
|
||
# Raw values for backward compatibility
|
||
"route_type": route_type.value,
|
||
"route_type_name": route_type.name,
|
||
"payload_type": payload_type.value,
|
||
"payload_type_name": payload_type.name,
|
||
"payload_version": payload_version.value,
|
||
# Enum objects for improved type safety
|
||
"route_type_enum": route_type,
|
||
"payload_type_enum": payload_type,
|
||
"payload_version_enum": payload_version,
|
||
# Transport and path information
|
||
"has_transport_codes": has_transport,
|
||
"transport_codes": transport_codes,
|
||
"transport_size": 4 if has_transport else 0,
|
||
"path_len": len(path_values), # Hop count for display / routing_info
|
||
"path_len_byte": path_len_byte, # Raw wire byte (same as firmware Packet path_len)
|
||
"path_byte_length": path_byte_length, # Path bytes (for logs showing "X bytes")
|
||
"bytes_per_hop": bytes_per_hop, # For multi-byte path storage/retrieval
|
||
"path_info": path_info,
|
||
"path": path_values, # For backward compatibility
|
||
"path_hex": path_hex,
|
||
"payload_hex": payload.hex(),
|
||
"payload_bytes": len(payload),
|
||
}
|
||
|
||
logger.debug(
|
||
f"Successfully decoded: route={packet_info.get('route_type_name')}, type={packet_info.get('payload_type_name')}"
|
||
)
|
||
return packet_info
|
||
|
||
except Exception as e:
|
||
# Log as ERROR not DEBUG so we can see what's failing
|
||
logger.error(f"Error decoding packet (len={len(byte_data)}): {e}", exc_info=True)
|
||
logger.error(f"Failed packet hex: {hex_data}")
|
||
return None
|
||
|
||
|
||
def process_packet_path(
|
||
path_bytes: bytes,
|
||
payload: bytes,
|
||
route_type: RouteType,
|
||
payload_type: PayloadType,
|
||
*,
|
||
prefix_hex_chars: int,
|
||
logger: Any,
|
||
) -> dict:
|
||
"""
|
||
Process the path field based on packet and route type
|
||
|
||
Args:
|
||
path_bytes: Raw path bytes
|
||
payload: Payload bytes (needed for TRACE packets)
|
||
route_type: Route type from header
|
||
payload_type: Payload type from header
|
||
|
||
Returns:
|
||
dict: Processed path information
|
||
"""
|
||
try:
|
||
# Chunk path bytes into node IDs using configured prefix length (with legacy fallback)
|
||
_, path_nodes = path_bytes_to_nodes(path_bytes, prefix_hex_chars)
|
||
|
||
# Special handling for TRACE packets
|
||
if payload_type == PayloadType.TRACE:
|
||
# RF path bytes are per-hop SNR×4 (int8), not node hashes. The commanded route is
|
||
# in the payload after tag(4)+auth(4)+flags(1); use path_info / parse_trace_payload_route_hashes for display.
|
||
# In TRACE packets, path field contains SNR data
|
||
# Real routing path is in the payload as pathHashes (after tag(4) + auth(4) + flags(1))
|
||
snr_values = []
|
||
for b in path_bytes:
|
||
# Convert SNR byte to dB (signed value)
|
||
snr_db = (b - 256) / 4 if b > 127 else b / 4
|
||
snr_values.append(snr_db)
|
||
|
||
# Decode trace payload to extract pathHashes (routing path)
|
||
# path_hash_len from flags (bits 0-1): 1 << (flags & 3) = 1, 2, 4, or 8 bytes per hop
|
||
path_hashes = []
|
||
if len(payload) >= 9: # Minimum: tag(4) + auth(4) + flags(1)
|
||
try:
|
||
path_hashes_bytes = payload[9:]
|
||
flags = payload[8]
|
||
path_hash_len = 1 << (flags & 3) # 1, 2, 4, or 8 bytes per hop
|
||
if path_hash_len <= 0:
|
||
path_hash_len = 1
|
||
if len(path_hashes_bytes) % path_hash_len == 0:
|
||
path_hashes = [
|
||
path_hashes_bytes[i : i + path_hash_len].hex().upper()
|
||
for i in range(0, len(path_hashes_bytes), path_hash_len)
|
||
]
|
||
else:
|
||
# Fallback: 1 byte per hop (legacy)
|
||
path_hashes = [f"{b:02x}".upper() for b in path_hashes_bytes]
|
||
except Exception as e:
|
||
logger.debug(f"Error extracting pathHashes from trace payload: {e}")
|
||
path_hashes = [f"{b:02x}".upper() for b in payload[9:]]
|
||
|
||
return {
|
||
"type": "trace",
|
||
"snr_data": snr_values,
|
||
"snr_path": path_nodes, # SNR data as hex for reference
|
||
"path": path_hashes, # Actual routing path from payload pathHashes
|
||
"path_hashes": path_hashes, # Explicit field for pathHashes
|
||
"description": f"TRACE packet with {len(snr_values)} SNR readings and {len(path_hashes)} path nodes",
|
||
}
|
||
|
||
# Regular packets - determine path type based on route type
|
||
is_direct = route_type in [RouteType.DIRECT, RouteType.TRANSPORT_DIRECT]
|
||
|
||
if is_direct:
|
||
# Direct routing: path contains routing instructions
|
||
# Bytes are stripped at each hop
|
||
return {
|
||
"type": "routing_instructions",
|
||
"path": path_nodes,
|
||
"meaning": "bytes_stripped_at_each_hop",
|
||
"description": f"Direct route via {','.join(path_nodes)} ({len(path_nodes)} hops)",
|
||
}
|
||
else:
|
||
# Flood routing: path contains historical route
|
||
# Bytes are added as packet floods through network
|
||
return {
|
||
"type": "historical_route",
|
||
"path": path_nodes,
|
||
"meaning": "bytes_added_as_packet_floods",
|
||
"description": f"Flooded through {','.join(path_nodes)} ({len(path_nodes)} hops)",
|
||
}
|
||
|
||
except Exception as e:
|
||
logger.error(f"Error processing packet path: {e}")
|
||
# Return basic path info as fallback (legacy 1-byte-per-hop)
|
||
_, path_nodes = path_bytes_to_nodes(path_bytes, 2)
|
||
return {"type": "unknown", "path": path_nodes, "description": f"Path: {','.join(path_nodes)}"}
|
||
|
||
|
||
def format_path_string(hex_path: str, bytes_per_hop: int | None = None, *, logger: Any) -> str:
|
||
"""
|
||
Convert a hex path string to node prefix format.
|
||
|
||
Args:
|
||
hex_path: Hex string representing the path (e.g., "01025f7e" or "01025fab" for 2-byte hops).
|
||
bytes_per_hop: Optional bytes per hop (1, 2, or 3) for multi-byte paths; None = legacy 1 byte per node.
|
||
|
||
Returns:
|
||
str: Formatted path string (e.g., "01,02,5f,7e" or "0102,5fab")
|
||
"""
|
||
try:
|
||
if not hex_path:
|
||
return "Direct"
|
||
|
||
if bytes_per_hop is not None and bytes_per_hop > 0:
|
||
hex_chars = bytes_per_hop * 2
|
||
path_nodes = split_path_hex(hex_path, hex_chars)
|
||
if path_nodes:
|
||
return ",".join(path_nodes)
|
||
return "Direct"
|
||
|
||
# Legacy: one byte per node (two hex chars)
|
||
path_bytes = bytes.fromhex(hex_path)
|
||
path_nodes = []
|
||
for i in range(len(path_bytes)):
|
||
node_id = path_bytes[i]
|
||
path_nodes.append(f"{node_id:02x}")
|
||
|
||
if path_nodes:
|
||
return ",".join(path_nodes)
|
||
else:
|
||
return "Direct"
|
||
|
||
except Exception as e:
|
||
logger.debug(f"Error formatting path string: {e}")
|
||
truncated = hex_path[:16] if len(hex_path) > 16 else hex_path
|
||
return f"Raw: {truncated}{'...' if len(hex_path) > 16 else ''}"
|
||
|
||
|
||
def route_type_name(route_type: int) -> str:
|
||
"""Get human-readable name for route type"""
|
||
route_types = {
|
||
0x00: "ROUTE_TYPE_TRANSPORT_FLOOD",
|
||
0x01: "ROUTE_TYPE_FLOOD",
|
||
0x02: "ROUTE_TYPE_DIRECT",
|
||
0x03: "ROUTE_TYPE_TRANSPORT_DIRECT",
|
||
}
|
||
return route_types.get(route_type, f"UNKNOWN_ROUTE_{route_type:02x}")
|
||
|
||
|
||
def payload_type_name(payload_type: int) -> str:
|
||
"""Get human-readable name for payload type"""
|
||
payload_types = {
|
||
0x00: "REQ",
|
||
0x01: "RESPONSE",
|
||
0x02: "TXT_MSG",
|
||
0x03: "ACK",
|
||
0x04: "ADVERT",
|
||
0x05: "GRP_TXT",
|
||
0x06: "GRP_DATA",
|
||
0x07: "ANON_REQ",
|
||
0x08: "PATH",
|
||
0x09: "TRACE",
|
||
0x0A: "MULTIPART",
|
||
# Additional payload types found in meshcore library (may not be in official spec)
|
||
0x0B: "UNKNOWN_0b", # Not defined in official spec
|
||
0x0C: "UNKNOWN_0c", # Not defined in official spec
|
||
0x0D: "UNKNOWN_0d", # Not defined in official spec
|
||
0x0E: "UNKNOWN_0e", # Not defined in official spec
|
||
0x0F: "RAW_CUSTOM",
|
||
}
|
||
return payload_types.get(payload_type, f"UNKNOWN_{payload_type:02x}")
|