ProtocolDataUnits Package

class ProtocolDataUnits.pdu.PDU[source]

Bases: object

class PDU[source]

Bases: object

The PDU (Protocol Data Unit) class represents a data structure used for encoding and decoding binary data.

It allows you to define the structure of a PDU with various field types, specify byte order, and encode data into binary format or decode binary data back into structured data.

Basic Example:

# Define a simple PDU structure with uint8 and float fields
pdu = PDU().length(16).order('big').uint8('type').float('value')
encoded = pdu.encode({'type': 1, 'value': 3.14})
decoded = pdu.decode(encoded)

Note

For more detailed examples, refer to the examples/pdu_examples.py file.

Attributes:

  • fields (list): A list of field definitions for the PDU.

  • byte_order (str): Byte order for encoding/decoding, ‘big’ or ‘little’.

  • pdu_length (int): Length of the PDU in bytes.

  • default_values (dict): Default values for fields.

Methods:

  • length(length): Sets the length of the PDU.

  • order(byte_order): Sets the byte order of the PDU.

  • encode(data, compress=False): Encodes structured data into a PDU.

  • decode(data, decompress=False): Decodes binary data back into structured data.

length(length=None)[source]

Sets the length of the PDU. If the length is not provided, it will be inferred from the defined fields.

Parameters:

  • length (int, optional): The total length of the PDU in bytes. If not provided, the length is inferred based on the fields.

Returns:

  • self: Returns the instance of the current PDU for method chaining.

order(byte_order)[source]

Sets the byte order for the PDU.

Parameters:

  • byte_order (str): The byte order to use. Should be ‘bigendian’ for big-endian or ‘littleendian’ for little-endian.

Returns:

  • self: Returns the instance of the current PDU for method chaining.

uint8(name=None, value=None, default=None)[source]

Adds a uint8 field to the PDU.

Parameters:

  • name (str, optional): The name of the field.

  • value (int, optional): A specific value for the field.

  • default (int, optional): A default value to use if none is provided during encoding.

Returns:

  • self: Returns the instance of the current PDU for method chaining.

uint16(name=None, value=None, default=None)[source]

Adds a uint16 field to the PDU.

Parameters:

  • name (str, optional): The name of the field.

  • value (int, optional): A specific value for the field.

  • default (int, optional): A default value to use if none is provided during encoding.

Returns:

  • self: Returns the instance of the current PDU for method chaining.

uint32(name=None, value=None, default=None)[source]

Adds a uint32 field to the PDU.

Parameters:

  • name (str, optional): The name of the field.

  • value (int, optional): A specific value for the field.

  • default (int, optional): A default value to use if none is provided during encoding.

Returns:

  • self: Returns the instance of the current PDU for method chaining.

float(name=None, value=None, default=None)[source]

Adds a float field to the PDU.

Parameters:

  • name (str, optional): The name of the field.

  • value (float, optional): A specific value for the field.

  • default (float, optional): A default value to use if none is provided during encoding.

Returns:

  • self: Returns the instance of the current PDU for method chaining.

double(name=None, value=None, default=None)[source]

Adds a double field to the PDU.

**Parameters:

  • name (str, optional): The name of the field.

  • value (float, optional): A specific value for the field.

  • default (float, optional): A default value to use if none is provided during encoding.

**Returns:

  • self: Returns the instance of the current PDU for method chaining.

int8(name=None, value=None, default=None)[source]

Adds an int8 field to the PDU.

Parameters:

  • name (str, optional): The name of the field.

  • value (int, optional): A specific value for the field.

  • default (int, optional): A default value to use if none is provided during encoding.

Returns:

  • self: Returns the instance of the current PDU for method chaining.

int16(name=None, value=None, default=None)[source]

Adds an int16 field to the PDU.

Parameters:

  • name (str, optional): The name of the field.

  • value (int, optional): A specific value for the field.

  • default (int, optional): A default value to use if none is provided during encoding.

Returns:

  • self: Returns the instance of the current PDU for method chaining.

int32(name=None, value=None, default=None)[source]

Adds an int32 field to the PDU.

Parameters:

  • name (str, optional): The name of the field.

  • value (int, optional): A specific value for the field.

  • default (int, optional): A default value to use if none is provided during encoding.

Returns:

  • self: Returns the instance of the current PDU for method chaining.

int64(name=None, value=None, default=None)[source]

Adds an int64 field to the PDU.

Parameter:

  • name (str, optional): The name of the field.

  • value (int, optional): A specific value for the field.

  • default (int, optional): A default value to use if none is provided during encoding.

Returns:

  • self: Returns the instance of the current PDU for method chaining.

uint64(name=None, value=None, default=None)[source]

Adds a uint64 field to the PDU.

Parameters:

  • name (str, optional): The name of the field.

  • value (int, optional): A specific value for the field.

  • default (int, optional): A default value to use if none is provided during encoding.

Returns:

  • self: Returns the instance of the current PDU for method chaining.

filler(count)[source]

Adds a filler field with a specified number of bytes.

Parameters:

  • count (int): The number of bytes for the filler.

Returns:

  • self: Returns the instance of the current PDU for method chaining.

padding(value)[source]

Adds padding bytes to the PDU.

Parameters:

  • value (int): The value to be used for the padding bytes.

Returns:

  • self: Returns the instance of the current PDU for method chaining.

fixed_string(name, length, default=None)[source]

Adds a fixed-length string field to the PDU.

The encoded format is a byte sequence of the specified length. If the provided string is shorter than the specified length, it will be padded with null bytes (``). If the string is longer, it will be truncated to fit the specified length.

Parameters:

  • name (str): The name of the field.

  • length (int): The length of the string in bytes.

  • default (str, optional): A default value to use if none is provided during encoding.

Example:

Returns:

  • self: Returns the instance of the current PDU for method chaining.

length_prefixed_string(name, default=None)[source]

Adds a length-prefixed string field to the PDU.

The encoded format is a 4-byte unsigned integer prefix (indicating the string length) followed by the string data itself.

Parameters:

  • name (str): The name of the field.

  • default (str, optional): A default value to use if none is provided during encoding.

Example:

Returns:

  • self: Returns the instance of the current PDU for method chaining.

variable_length_array(name, element_type, default=None)[source]

Adds a variable-length array field to the PDU.

The encoded format is a 4-byte unsigned integer prefix representing the number of elements in the array, followed by the serialized form of each element according to the specified element_type.

Parameters:

  • name (str): The name of the field.

  • element_type (str): The type of the elements in the array (e.g., ‘uint8’, ‘int16’).

  • default (list, optional): A default array to use if none is provided during encoding.

Example:

Returns:

  • self: Returns the instance of the current PDU for method chaining.

nested_pdu(name, pdu)[source]

Adds a nested PDU (sub-PDU) field to the current PDU structure. Keep this for backward compatible for now. This is same as the pdu_fragment.

Parameters:

  • name (str): The name of the field.

  • pdu (PDU): An instance of a PDU that represents the nested structure.

Returns:

  • self: Returns the instance of the current PDU for method chaining.

pdu_fragment(name, pdu)[source]

Adds a nested PDU (sub-PDU) field to the current PDU structure.

Parameters:

  • name (str): The name of the field.

  • pdu (PDU): An instance of a PDU that represents the nested structure.

Returns:

  • self: Returns the instance of the current PDU for method chaining.

static compute_crc(data)[source]
encode(data, compress=False)[source]

Encodes the provided data into a PDU (Protocol Data Unit) byte array.

The encode function serializes the data fields based on the specified PDU structure. It handles various data types including integers, floats, strings, arrays, and nested PDUs. Optionally, the encoded data can be compressed using zlib.

Parameters:

  • data (dict): A dictionary containing the field names and their corresponding values to be encoded.

  • compress (bool): If True, the encoded data will be compressed using zlib. Default is False.

Returns:

  • bytes: The encoded byte array representing the PDU.

Raises:

  • ValueError: If a required field is missing in the provided data.

Field Types Handled:

  • uint8, int8, uint16, int16, uint32, int32, uint64, int64

  • float, double

  • fixed_string: A string of fixed length.

  • length_prefixed_string: A string prefixed with its length.

  • variable_length_array: An array with a length prefix.

  • pdu_fragment: Another PDU structure embedded within the main PDU.

  • **filler: Padding bytes.

  • padding: Additional padding to achieve the specified PDU length.

CRC Computation:

  • A CRC32 checksum is computed and appended to the end of the encoded data to ensure data integrity.

decode(data, decompress=False)[source]

Decodes a PDU (Protocol Data Unit) byte array into its constituent fields.

The decode function parses the provided byte array based on the specified PDU structure, extracting the values of each field. It handles various data types including integers, floats, strings, arrays, and nested PDUs. Optionally, the input data can be decompressed using zlib.

Parameters:

  • data (bytes): The byte array representing the encoded PDU.

  • decompress (bool): If True, the input data will be decompressed using zlib. Default is False.

Returns:

  • dict: A dictionary containing the decoded field names and their corresponding values.

Raises:

  • ValueError: If the CRC checksum does not match, indicating data corruption.

Field Types Handled:

  • uint8, int8, uint16, int16, uint32, int32, uint64, int64

  • float, double

  • fixed_string: A string of fixed length.

  • length_prefixed_string: A string prefixed with its length.

  • variable_length_array: An array with a length prefix.

  • pdu_fragment: Another PDU structure embedded within the main PDU.

  • filler: Padding bytes.

  • padding: Additional padding to achieve the specified PDU length.

CRC Validation:

  • A CRC32 checksum is validated at the end of the decoded data to ensure data integrity.

to_json()[source]
static from_json(json_str)[source]
ProtocolDataUnits.pdu.create_pdu_format(length, byte_order, *fields)[source]

Create a PDU format using a simplified and user-friendly API.

Parameters:

  • length (int): The total length of the PDU.

  • byte_order (str): The byte order to use (‘big’ for big-endian or ‘little’ for little-endian).

  • *fields (tuples): A variable number of tuples, each representing a field in the PDU.

    Each tuple should contain the field type followed by the necessary arguments.

Returns:

  • PDU: An instance of the PDU class with the specified format.