Binary Data and Communication Protocols

Oser helps you to work with binary data structures. You can easily build and read binary data files, communication protocols, eeprom contents, etc.

More information can be found in the oser documentation.

Data structures and their dependencies are easily described. You do not need to count bits and bytes anymore.

Communication Protocol

Assume you have a binary communication protocol for a safety relevant system.

Each message consists of a header and one to many data objects. Each data object consists of a type and a type dependent payload. Every header and every data object’s consistency is assured with a CRC.

All data is big endian.

The header is specified in the following table.

Byte

Name

Type

Description

0

service_id

UInt8

The service id (Enum)

1..4

timestamp

UInt32

timestamp as unix epoch

5

number_of_data_objects

UInt8

the number of data objects

6..9

crc

UInt32

32 bit checksum with polynomial 0x04C11DB7

10..N

data_objects

Array

The array of data objects

The enumeration for the service_id is found in the following table.

Name

Value

Unicast

0x00

Broadcast

0xFF

The data object structure is specified in the following table.

Byte

Name

Type

Description

0

data_object_id

UInt8

The data object id (Enum)

1..N

payload

Object

The specific payload

N+1..N+4

crc

UInt32

32 bit checksum with polynomial 0x04C11DB7

The enumeration for the data_object_id can be found in the following table.

Name

Value

U32

0x00

Float[]

0x01

Status

0xFF

The U32 payload is described in the following table.

Byte

Name

Type

Description

1..4

value

UInt32

The value

The Float[] payload is described in the following table.

Byte

Name

Type

Description

1

number_of_values

UInt8

The number of float values

2..N

values

Float list

List of float values

The Status payload is described in the following table.

Byte

Name

Type

Description

0..3

uptime

UInt32

Uptime in seconds

4

status

UInt8

Status bits

The implementation of the protocol is located in binary_data/protocol.py.

# -*- coding: utf-8 -*-
#
# Copyright (c) 2020, HILSTER Testing Solutions GmbH, Germany - https://hilster.io
# All rights reserved.
#

import oser


class U32Payload(oser.ByteStruct):
    """
    +----------+------------------------+--------+----------------------------+
    | Byte     | Name                   | Type   | Description                |
    +==========+========================+========+============================+
    | 1..4     | value                  | UInt32 | The value                  |
    +----------+------------------------+--------+----------------------------+
    """
    def __init__(self):
        super(U32Payload, self).__init__()
        self.value = oser.UBInt32()


class FloatListPayload(oser.ByteStruct):
    """
    +----------+------------------------+--------+----------------------------+
    | Byte     | Name                   | Type   | Description                |
    +==========+========================+========+============================+
    | 1        | number_of_values       | UInt8  | The number of float values |
    +----------+------------------------+--------+----------------------------+
    | 2..N     | values                 | Float  | List of float values       |
    |          |                        | list   |                            |
    +----------+------------------------+--------+----------------------------+
    """
    def __init__(self):
        super(FloatListPayload, self).__init__()
        self.number_of_values = oser.UBInt8()
        self.values = oser.Array(
            length=lambda ctx: ctx.number_of_values.get(),
            prototype=oser.BFloat
        )


class StatusPayload(oser.ByteStruct):
    """
    +----------+------------------------+--------+----------------------------+
    | Byte     | Name                   | Type   | Description                |
    +==========+========================+========+============================+
    | 0..3     | uptime                 | UInt32 | Uptime in seconds          |
    +----------+------------------------+--------+----------------------------+
    | 4        | status                 | UInt8  | Status bits                |
    +----------+------------------------+--------+----------------------------+
    """
    def __init__(self):
        super(StatusPayload, self).__init__()
        self.uptime = oser.UBInt32()
        self.status = oser.UBInt8()


PAYLOAD = {
    "U32": (0x00, U32Payload),
    "Float[]": (0x01, FloatListPayload),
    "Status": (0xFF, StatusPayload),
}


def DataObjectId():
    """
    +-----------+------------------------+
    | Name      | Value                  |
    +===========+========================+
    | U32       | 0x00                   |
    +-----------+------------------------+
    | Float[]   | 0x01                   |
    +-----------+------------------------+
    | Status    | 0xFF                   |
    +-----------+------------------------+
    """
    return oser.Enum(
        prototype=oser.UBInt8,
        values={key: value[0] for key, value in PAYLOAD.items()}
    )


class DataObject(oser.ByteStruct):
    """
    +----------+------------------------+--------+----------------------------+
    | Byte     | Name                   | Type   | Description                |
    +==========+========================+========+============================+
    | 0        | data_object_id         | UInt8  | The data object id (Enum)  |
    +----------+------------------------+--------+----------------------------+
    | 1..N     | payload                | Object | The specific payload       |
    +----------+------------------------+--------+----------------------------+
    | N+1..N+4 | crc                    | UInt32 | 32 bit checksum with       |
    |          |                        |        | polynomial 0x04C11DB7      |
    +----------+------------------------+--------+----------------------------+
    """
    def __init__(self):
        super(DataObject, self).__init__()

        self.data_object_id = DataObjectId()
        self.payload = oser.Switch(
            condition=lambda ctx: self.data_object_id.get(),
            values={key: value[1]() for key, value in PAYLOAD.items()},
            default=oser.Null()
        )
        self.crc = oser.CRCB32(polynomial=0x04C11DB7, strict=True)


def ServiceId():
    """
    +-----------+------------------------+
    | Name      | Value                  |
    +===========+========================+
    | Unicast   | 0x00                   |
    +-----------+------------------------+
    | Broadcast | 0xFF                   |
    +-----------+------------------------+
    """
    return oser.Enum(
        prototype=oser.UBInt8,
        values={
            "Unicast": 0x00,
            "Broadcast": 0xFF,
        },
        value="Unicast"
    )


class Message(oser.ByteStruct):
    """
    +----------+------------------------+--------+----------------------------+
    | Byte     | Name                   | Type   | Description                |
    +==========+========================+========+============================+
    | 0        | service_id             | UInt8  | The service id (Enum)      |
    +----------+------------------------+--------+----------------------------+
    | 1..4     | timestamp              | UInt32 | timestamp as unix epoch    |
    +----------+------------------------+--------+----------------------------+
    | 5        | number_of_data_objects | UInt8  | the number of data objects |
    +----------+------------------------+--------+----------------------------+
    | 6..9     | crc                    | UInt32 | 32 bit checksum with       |
    |          |                        |        | polynomial 0x04C11DB7      |
    +----------+------------------------+--------+----------------------------+
    | 10..N    | data_objects           | Array  | The array of data objects  |
    +----------+------------------------+--------+----------------------------+
    """
    def __init__(self):
        super(Message, self).__init__()

        self.service_id = ServiceId()
        self.timestamp = oser.UBInt32()
        self.number_of_data_objects = oser.UBInt8()
        self.crc = oser.CRCB32(polynomial=0x04C11DB7, strict=True)
        
        self.data_objects = oser.Array(
            length=lambda ctx: ctx.number_of_data_objects.get(),
            prototype=DataObject
        )

An example on how to create a specific message is located in binary_data/create_message.py.

# -*- coding: utf-8 -*-
#
# Copyright (c) 2020, HILSTER Testing Solutions GmbH, Germany - https://hilster.io
# All rights reserved.
#

import time

import oser
from protocol import Message


if __name__ == '__main__':
    # instantiate a message
    message = Message()

    # set service id
    message.service_id.set("Broadcast")

    # set timestamp
    message.timestamp.set(int(time.time()))

    # add 3 data objects
    message.number_of_data_objects.set(3)

    # set data object 0 to U32
    data_object = message.data_objects[0]
    data_object.data_object_id.set("U32")
    data_object.payload.value.set(0x1234)

    # set data object 1 to Float[] with 4 floats
    data_object = message.data_objects[1]
    data_object.data_object_id.set("Float[]")
    data_object.payload.number_of_values.set(4)
    data_object.payload.values[:] = (1.5**i for i in range(4))

    # set data object 2 to Status
    data_object = message.data_objects[2]
    data_object.data_object_id.set("Status")
    data_object.payload.uptime.set(24*60*60)  # 1 day
    data_object.payload.status.set(0b11110000)

    # encode message to binary
    binary_message = message.encode()

    # print binary message
    print('>>> Message as hex <<<')
    print(oser.to_hex(binary_message))

    # print message
    print('\n>>> Message default print <<<')
    print(message)

    # print message with introspection
    print('\n>>> Message introspection <<<')
    print(message.introspect())

An example on how to parse a specific message and extract data is located in binary_data/parse_message.py.

# -*- coding: utf-8 -*-
#
# Copyright (c) 2020, HILSTER Testing Solutions GmbH, Germany - https://hilster.io
# All rights reserved.
#

from protocol import Message

binary_message = b"\x00\x5e\xc4\xf1\x91\x0a\xaf\x6b\x65\x56\x00\x00\x00\x00\x00\x00"\
                 b"\x00\x00\x00\x00\x00\x00\x00\x01\x04\xc1\x1d\xb7\x00\x00\x00\x00"\
                 b"\x04\x13\x04\x76\xdc\x00\x00\x00\x00\x09\x22\xc9\xf0\x0f\x00\x00"\
                 b"\x00\x00\x10\x4c\x11\xdb\x70\x00\x00\x00\x00\x19\x6e\xd8\x2b\x7f"\
                 b"\x00\x00\x00\x00\x24\x8b\x27\xc0\x3c\x00\x00\x00\x00\x31\xd0\xf3"\
                 b"\x70\x27\x00\x00\x00\x00\x40\x34\x86\x70\x77\x00\x00\x00\x00\x51"\
                 b"\x7c\x56\xb6\xb0"


if __name__ == '__main__':
    # the above binary_message can be decoded as a Message()
    message = Message()
    message.decode(binary_message)

    # what is the message type?
    print("Message type:", message.service_id.get())

    # how many data objects does the message have?
    print("Number of data objects:", message.number_of_data_objects.get())

    # what is the content of the 3rd data object?
    print("3rd value:", message.data_objects[2].payload.value.get())

    # change the crc at the end and try to decode the message again. What happens?
    broken_binary_message = binary_message[:-1] + b"\xb1"
    # message.decode(broken_binary_message)  # please comment out this line and run this script again

You will also find a example of a crc checksum error in the last example.