Recipes

Choose a task and start with known bytes. These examples need no network connection; all packets are already in memory.

Read a device packet from TCP

A device sent 10 00 04 01 02 4f 4b. Its protocol defines a one-byte command, a two-byte payload length, a version, and data with a one-byte length prefix. Describe a reply to command 0x10:

from bytespec import Constructor, PayloadLength, ProtoModel, field
from bytespec.types import UInt8

class Reply(ProtoModel):
    __constructor__ = 0x10
    __header__ = (Constructor(1), PayloadLength(2))
    version: UInt8
    data: bytes = field(prefix_length=1)

received = bytes.fromhex("10 00 04 01 02 4f 4b")
reply = Reply.decode(received)
print(reply.version, reply.data)
assert reply.encode() == received

1 b'OK'

① Command · 10

The value of __constructor__ = 0x10. A message for a different command will fail validation by Reply.

② Payload length · 00 04

Four bytes after the header: the version, the data length, and two bytes for OK.

③ Version · 01

UInt8 reads one byte and returns the ordinary integer 1.

④ Data · 02 4f 4b

Length 2, followed by the bytes for OK. data remains bytes, not a string.

Here, received already contains a complete packet. A single TCP recv() call may return part of a packet or several packets at once: accumulate the required bytes first. bytespec parses a buffer; it does not manage the socket.

Read multiple messages from one buffer

A buffer contains two replies in the previous format. Continue with the Reply class:

buffer = received + received
first, offset = Reply.decode_from(buffer, 0)
print(first.data, offset)
second, offset = Reply.decode_from(buffer, offset)
print(second.data, offset)
assert offset == len(buffer)

b'OK' 7
b'OK' 14

The first reply occupies bytes 0–6. The returned offset 7 points to the start of the second reply. After the second, it is 14: the end of the buffer. decode() accepts exactly one message; use decode_from() for multiple messages. Handling the remainder of a buffer: Files and multiple messages.

Read a format without the bytespec header

A sensor sends exactly two bytes: temperature 23 and battery level 87. No metadata bytes may be added:

from bytespec import ProtoModel
from bytespec.types import UInt8

class SensorReading(ProtoModel):
    __header__ = ()
    temperature: UInt8
    battery: UInt8

received = bytes.fromhex("17 57")
reading = SensorReading.decode(received)
print(reading.temperature, reading.battery)
assert reading.encode() == received

23 87

__header__ = () removes all header elements. The first byte belongs directly to temperature, the second to battery. Field sizes determine the boundaries in this fixed format. Other options: Headers and framing.

Read a little-endian number

A device stores the counter value 258 as two bytes, 02 01, least significant byte first. Set this byte order:

from bytespec import ByteOrder, ProtoModel
from bytespec.types import UInt16

class DeviceCounter(ProtoModel):
    __header__ = ()
    __byte_order__ = ByteOrder.LITTLE
    value: UInt16

received = bytes.fromhex("02 01")
counter = DeviceCounter.decode(received)
print(counter.value)
assert counter.encode() == received

258

The number is 2 + 1 * 256. With the default big-endian order, the same bytes would mean 513. The setting applies to fixed-width numbers and prefixes in this model; it does not reorder text bytes.

Store a custom application type

Your application represents an identifier as a UserId object, while the device expects a four-byte integer. Codecs: custom formats and types has the complete example: the UserId class, its codec, and using it through field(codec=...). A codec only converts an individual field.

Build a nested message

Now let us combine the features we have covered: an author model, a list of tags, and an optional note. You can run this example as a complete script.

from bytespec import ProtoModel, field
from bytespec.types import UInt32

class User(ProtoModel):
    id: UInt32
    name: str

class Message(ProtoModel):
    author: User
    text: str
    tags: list[str] = field(default_factory=list)
    note: str | None = field(flag=0)

message = Message(
    author=User(id=42, name="Anna"),
    text="Hello!",
    tags=["news"],
)
encoded = message.encode()
decoded = Message.decode(encoded)

print(decoded.author.name)  # Anna
print(decoded.text)         # Hello!
print(decoded.tags)         # ['news']
print(decoded.note)         # None

Creation and decoding work just as they did for the first model. Tags are restored as a list, the author as a User, and the omitted note as None. These structures needed no separate serialization calls.

Let us change the note and encode the message again:

decoded.note = "Updated"
updated = Message.decode(decoded.encode())
print(updated.note)  # Updated

Try this example with empty tags and with note="": both values are preserved. Only note=None removes the optional field from the message.

You can now try a model on a single packet from your own protocol. Headers and framing shows how to configure framing bytes; Validating model values explains how to check relationships between fields; Handling errors covers error handling. To read several messages from one buffer, open Files and multiple messages.