Python — Header & Packet

Header and packet types are defined in the nnrp.core module.

Import

from nnrp.core import (
    NnrpHeader, NnrpPacket, TensorSectionData,
    HEADER_MAGIC, HEADER_LENGTH,
)

---

Constants

Name	Type	Value	Description
HEADER_MAGIC	bytes	b"NNRP"	Fixed 4-byte magic prefix for every packet
HEADER_LENGTH	int	40	Fixed header byte length

---

NnrpHeader

Fixed 40-byte packet header (@dataclass(frozen=True, slots=True)). Wire format: struct.Struct("<4sBBBBIIIIIHHQ").

Fields

Field	Type	Description
wire_format	WireFormat	Wire format version (currently CURRENT=0)
version_major	int	Protocol major version (currently 1)
msg_type	MessageType	Message type
flags	HeaderFlags	Header flags
meta_len	int	Metadata byte count
body_len	int	Body byte count
session_id	int	Session ID
frame_id	int	Frame ID
view_id	int	View/viewport ID
route_id	int	Route ID (uint16)
trace_id	int	Trace ID (uint16)
header_len	int	Fixed HEADER_LENGTH=40

Methods

def pack(self) -> bytes:
    """Serialize header to 40 bytes (including 4-byte magic prefix)."""

@classmethod
def unpack(
    cls,
    payload: bytes,
    *,
    expected_wire_format: WireFormat | None = None,
) -> "NnrpHeader":
    """Parse header from bytes. Raises ValueError if expected_wire_format doesn't match."""

---

NnrpPacket

Complete NNRP packet containing header, metadata, and body (@dataclass(frozen=True, slots=True)).

Fields

Field	Type	Description
header	NnrpHeader	Packet header
metadata	bytes	Raw metadata bytes
body	bytes	Raw body bytes

Methods

def pack(self) -> bytes:
    """Serialize complete packet to bytes."""

@classmethod
def build(
    cls,
    msg_type: MessageType,
    *,
    wire_format: WireFormat = WireFormat.CURRENT,
    version_major: int = 1,
    flags: HeaderFlags = HeaderFlags.NONE,
    session_id: int = 0,
    frame_id: int = 0,
    view_id: int = 0,
    route_id: int = 0,
    trace_id: int = 0,
    metadata: bytes = b"",
    body: bytes = b"",
) -> "NnrpPacket":
    """Construct and return a new NnrpPacket."""

@classmethod
def unpack(cls, payload: bytes) -> "NnrpPacket":
    """Parse complete packet (header + metadata + body) from bytes."""

---

TensorSectionData

Describes a single tensor section (@dataclass(frozen=True, slots=True)).

Fields

Field	Type	Description
role_id	int	Tensor role ID
default_codec_id	int	Default codec ID
dtype_id	int	Data type (see TensorDType)
tile_payloads	tuple[bytes, ...]	Raw payload bytes per tile
codec_ids	tuple[int, ...]	Per-tile codec IDs (may be empty)
layout_id	int	Memory layout (see TensorLayout)
scale_policy	int	Quantization scale policy (see ScalePolicy)
payload_stride_bytes	int	Fixed stride in bytes; 0 means variable
element_count_per_tile	int	Element count per tile

Methods

def normalized_tile_payloads(self) -> tuple[bytes, ...]: ...
def normalized_codec_ids(self) -> tuple[int, ...]: ...

---

Example: Build and parse a PING packet

from nnrp.core import NnrpPacket
from nnrp.core.enums import MessageType

packet = NnrpPacket.build(MessageType.PING, session_id=42, frame_id=1)
raw = packet.pack()

restored = NnrpPacket.unpack(raw)
assert restored.header.msg_type is MessageType.PING
assert restored.header.session_id == 42

---

Typical Use Cases

When to use the low-level packet API

Most application code does not need NnrpPacket or NnrpHeader directly — ClientSession / ServerSession covers all common operations. Use the low-level API when:

• Implementing a custom transport adapter
• Writing protocol conformance tests or fuzz tooling
• Debugging: capturing and replaying raw packet bytes
• Adding custom ControlExtensionEntry fields to the handshake

Measuring RTT with a raw PING

import time
from nnrp.core import NnrpPacket
from nnrp.core.enums import MessageType

async def measure_rtt(connection) -> float:
    seq = int(time.monotonic() * 1000) & 0xFFFF
    ping = NnrpPacket.build(MessageType.PING, session_id=connection.session_id,
                             frame_id=seq)
    t0 = time.monotonic()
    await connection.send_packet(ping)
    pong = await connection.receive_packet(timeout=2.0)
    assert pong.header.msg_type is MessageType.PONG
    return time.monotonic() - t0

Parsing an incoming raw packet

from nnrp.core import NnrpPacket
from nnrp.core.messages import unpack_body
from nnrp.core.enums import MessageType

raw_bytes = await raw_transport.recv()
packet = NnrpPacket.unpack(raw_bytes)
match packet.header.msg_type:
    case MessageType.FRAME_SUBMIT:
        body = unpack_body(packet)
        process_submit(body)
    case MessageType.RESULT_PUSH:
        body = unpack_body(packet)
        process_result(body)

---

Common Pitfalls

WARNING

1. Header length is fixed by the protocol version. Do not append custom bytes to the header; put extension fields in metadata via ControlExtensionEntry.

2. Magic bytes are validated by unpack(). An off-by-one read offset from a TCP length-prefix framing bug will cause ValueError: bad magic.

3. pack() returns immutable bytes. Do not attempt in-place mutation; reconstruct via NnrpPacket.build() instead.

4. session_id and frame_id are unsigned integers. Passing a negative value does not raise an error in Python, but the serialized result is silently truncated, causing parse errors on the receiver.