Transport Layer¶

Provides the TransportLayer class used to establish and maintain bidirectional serial communication with Arduino and Teensy microcontrollers running the ataraxis-transport-layer-mc library over USB / UART interface.

class ataraxis_transport_layer_pc.transport_layer.TransportLayer(port, microcontroller_serial_buffer_size, baudrate, polynomial=np.uint8(7), initial_crc_value=np.uint8(0), final_crc_xor_value=np.uint8(0), *, reflected=False, test_mode=False)¶

Bases: object

Provides methods for sending and receiving serialized data over the USB and UART communication interfaces.

This class instantiates and manages all library assets used to transcode, validate, and bidirectionally transfer serial data over the target communication interface. Critically, this includes the transmission and reception buffers that are used to temporarily store the outgoing and incoming data payloads. All user-facing class methods interact with the data stored in one of these buffers.

Parameters:

port (str) – The name of the serial port to connect to, e.g.: ‘COM3’ or ‘/dev/ttyUSB0’. Use the ‘axtl-ports’ CLI command to discover available port names.
microcontroller_serial_buffer_size (int) – The size, in bytes, of the buffer used by the connected microcontroller’s serial communication interface. Usually, this information is available from the microcontroller’s manufacturer (UART / USB controller specification). Must be at least 9 bytes, as 8 bytes are consumed by the packet metadata and at least one byte has to remain available for the payload. This value bounds the transmitted payload size alone, as reception is capped at the 254-byte COBS limit.
baudrate (int) – The baudrate to use for communication if the microcontroller uses the UART interface. Should match the value used by the microcontroller. This parameter is ignored when using the USB interface.
polynomial (uint8 | uint16 | uint32, default: np.uint8(7)) – The polynomial to use for the generation of the CRC lookup table. The polynomial must be expressed in the standard non-reflected, MSB-aligned form used by published CRC parameter catalogues.
initial_crc_value (uint8 | uint16 | uint32, default: np.uint8(0)) – The value to which the CRC checksum is initialized before calculation.
final_crc_xor_value (uint8 | uint16 | uint32, default: np.uint8(0)) – The value with which the CRC checksum is XORed after calculation.
reflected (bool, default: False) – Determines whether the CRC checksum is computed least significant bit first and written to the packet postamble least significant byte first. This must match the setting used by the microcontroller.
test_mode (bool, default: False) – Determines whether the instance uses a pySerial (real) or a SerialMock (mocked) communication interface. This flag is used during testing and should be disabled for all production runtimes.

_opened¶: Tracks whether the serial communication has been opened (the port has been connected).

_port¶: Depending on the test_mode flag, stores either a SerialMock or Serial object that provides the serial communication interface.

_crc_processor¶: Stores the CRCProcessor instance that provides methods for working CRC checksums.

_cobs_processor¶: Stores the COBSProcessor instance that provides methods for encoding and decoding transmitted payloads.

_start_byte¶: Stores the byte-value that marks the beginning of transmitted and received packets.

_delimiter_byte¶: Stores the byte-value that marks the end of transmitted and received packets.

_timeout¶: Stores the number of microseconds to wait between receiving any two consecutive bytes of a packet.

_max_tx_payload_size¶: Stores the maximum number of bytes that can be transmitted as a single payload.

_max_rx_payload_size¶: Stores the maximum number of bytes that can be received from the microcontroller as a single payload.

_min_rx_payload_size¶: Stores the minimum number of bytes that can be received from the Microcontroller as a single payload.

_postamble_size¶: Stores the byte-size of the CRC checksum.

_transmission_buffer¶: The buffer used to stage the data to be sent to the Microcontroller.

_reception_buffer¶: The buffer used to store the decoded data received from the Microcontroller.

_bytes_in_transmission_buffer¶: Tracks how many bytes (relative to index 0) of the transmission buffer are currently used to store the payload to be transmitted.

_bytes_in_reception_buffer¶: Same as _bytes_in_transmission_buffer, but for the reception buffer.

_consumed_bytes¶: Tracks the number of the last received payload bytes that have been consumed by the read_data() method calls.

_leftover_bytes¶: A buffer used to preserve any ‘unconsumed’ bytes that were read from the serial port but not used to reconstruct the payload sent from the Microcontroller. This is used to minimize the number of calls to pySerial methods, as they are costly to run.

_timeout_guard¶: Stores the Timeout instance that bounds the wait for the bytes that are missing from a partially received packet.

_accepted_numpy_scalars¶: Stores numpy types (classes) that can be used as scalar inputs or as ‘dtype’ fields of the numpy arrays that are provided to class methods.

_minimum_packet_size¶: Stores the minimum number of bytes that can represent a valid packet. This value is used to optimize packet reception logic.

Raises:

TypeError – If any of the input arguments are not of the expected type.
ValueError – If any of the input arguments have invalid values.

property available: bool¶: Returns True if enough bytes are available from the serial port to justify attempting to receive a packet.

property bytes_in_reception_buffer: int¶: Returns the number of payload bytes stored inside the instance’s reception buffer.

property bytes_in_transmission_buffer: int¶: Returns the number of payload bytes stored inside the instance’s transmission buffer.

read_data(data_object)¶

Reconstructs an object matching the input prototype from the data stored in the instance’s reception buffer, consuming (discarding) all read bytes.

This method deserializes the objects stored in the reception buffer as a sequence of bytes. Calling this method consumes the read bytes, making it impossible to retrieve the same data from the reception buffer again.

Notes

The input object serves as a prototype that determines the type and the number of bytes to read. Numpy scalar and array prototypes retain their original values, so the deserialized data is accessed through the returned object. Dataclass prototypes additionally have their fields overwritten in place.

At this time, the method only works with valid numpy scalars and arrays as well as python dataclasses entirely made out of valid numpy types.

The maximum runtime speed of this method is achieved when reading data as numpy arrays, which is optimized to a single read operation. The minimum runtime speed is achieved by reading dataclasses, as it involves looping over dataclass attributes.

Parameters:

data_object (Any) – An initialized numpy scalar or array object or a python dataclass made entirely out of valid numpy objects. Supported numpy types are: uint8, uint16, uint32, uint64, int8, int16, int32, int64, float32, float64, and bool. Array prototypes have to be 1-dimensional and not empty to be supported.

Return type:

Any

Returns:

The deserialized data object extracted from the instance’s reception buffer.

Raises:

TypeError – If the input object is not a supported numpy scalar, numpy array, or python dataclass.
ValueError – If the payload stored inside the reception buffer does not have enough unconsumed bytes available to reconstruct the requested object. If the input object is a multidimensional or empty numpy array.

receive_data()¶

Receives a data packet from the communication interface, verifies its integrity, and decodes its payload into the instance’s reception buffer.

Notes

Before attempting to receive the packet, the method verifies that the communication interface holds enough bytes to justify parsing. It is safe to call this method cyclically (as part of a loop) until a packet is received.

This method resets the instance’s reception buffer before attempting to receive the data, discarding any potentially unprocessed data.

Return type:: bool
Returns:: True if the packet was successfully received and unpacked. False if the communication interface does not contain enough bytes to justify processing the packet or if the available bytes carry no start byte, which indicates communication line noise.
Raises:: RuntimeError – If the method runs into an error while receiving or processing the packet’s data.

property reception_buffer: ndarray[tuple[Any, ...], dtype[uint8]]¶: Returns a copy of the reception buffer array, which stores the decoded data received from the Microcontroller.

property reception_payload: ndarray[tuple[Any, ...], dtype[uint8]]¶: Returns a copy of the payload bytes stored inside the instance’s reception buffer, which is the leading region of that buffer the last successful reception filled.

reset_reception_buffer()¶

Resets the instance’s reception buffer, discarding any stored data.

Return type:: None

reset_transmission_buffer()¶

Resets the instance’s transmission buffer, discarding any stored data.

Return type:: None

send_data()¶

Packages the data inside the instance’s transmission buffer into a serialized packet and transmits it over the communication interface.

Return type:: None

Notes

This method resets the instance’s transmission buffer after transmitting the data, discarding any data stored inside the buffer.

Raises:: ValueError – If the instance’s transmission buffer does not store any payload data.

property transmission_buffer: ndarray[tuple[Any, ...], dtype[uint8]]¶: Returns a copy of the transmission buffer array, which stores the data staged to be sent to the Microcontroller.

write_data(data_object)¶

Serializes and writes the input object’s data to the end of the payload stored in the instance’s transmission buffer.

Notes

At this time, the method only works with numpy scalars and arrays, as well as python dataclasses entirely made out of valid numpy types.

The maximum runtime speed for this method is achieved when writing data as numpy arrays, which is optimized to a single write operation. The minimum runtime speed is achieved by writing dataclasses, as it involves looping over dataclass attributes. When writing dataclasses, all attributes are serialized and written as a consecutive data block.

Parameters:

data_object (Any) – A numpy scalar or array object or a python dataclass made entirely out of valid numpy objects. Supported numpy types are: uint8, uint16, uint32, uint64, int8, int16, int32, int64, float32, float64, and bool. Arrays have to be 1-dimensional, contiguous, and not empty to be supported.

Raises:

TypeError – If the input object is not a supported numpy scalar, numpy array, or python dataclass.
ValueError – If writing the object’s data would grow the payload past the maximum transmittable payload size. If the input object is a numpy array that is not one-dimensional, is empty, or is not stored contiguously in memory.

Return type:

None

class ataraxis_transport_layer_pc.transport_layer.TransportLayerStatus(*values)¶

Bases: IntEnum

Defines the status codes used by the TransportLayer class to communicate the state of various processing steps between the JIT-compiled methods and the user-facing API methods.

DELIMITER_FOUND_TOO_EARLY = 6¶: Delimiter byte value encountered before reaching the end of the encoded payload data block. It is expected that the last byte of the encoded payload is set to the delimiter value and that the value is not present anywhere else inside the encoded payload region. Encountering the delimiter early indicates packet corruption.

DELIMITER_NOT_FOUND = 7¶: Delimiter byte value not encountered at the end of the encoded payload data block. See the DELIMITER_FOUND_TOO_EARLY description for more details, but this code also indicates packet corruption.

EMPTY_ARRAY_ERROR = -3¶: The data to be written or the prototype to be read is an empty NumPy array.

INSUFFICIENT_BUFFER_SPACE_ERROR = -1¶: The reception or transmission buffer does not have enough space for the requested operation.

MULTIDIMENSIONAL_ARRAY_ERROR = -2¶: The data to be written or the prototype to be read are not a one-dimensional NumPy array.

NOT_ENOUGH_CRC_BYTES = 3¶: Not enough bytes read to fully parse the packet. The packet payload was successfully parsed, but there were not enough bytes to fully parse the CRC postamble.

NOT_ENOUGH_PACKET_BYTES = 2¶: Not enough bytes read to fully parse the packet. The packet size was resolved, but there were not enough bytes to fully parse the packet (encoded payload + crc postamble).

NO_BYTES_TO_READ = 4¶: No start byte found, which is interpreted as ‘no bytes to read.’ Usually, this situation is caused by communication line noise generating ‘noise bytes’.

PACKET_PARSED = 1¶: Packet fully parsed.

PACKET_SIZE_UNKNOWN = 0¶: Not enough bytes read to fully parse the packet. The start byte was found, but the payload_size byte has not yet been read, so the packet size cannot be resolved.

PAYLOAD_SIZE_MISMATCH = 5¶: Parsed payload_size value does not match the expected value. This likely indicates packet corruption or communication parameter mismatch between the TransportLayer instance and the connected Microcontroller.

ataraxis_transport_layer_pc.transport_layer.list_available_ports()¶

Provides the information about each serial port addressable through the pySerial library.

This function is intended to be used for discovering and selecting the serial port ‘names’ to use with TransportLayer instances.

Return type:: tuple[ListPortInfo, ...]
Returns:: A tuple of ListPortInfo instances, each storing ID and descriptive information about each discovered serial port.

ataraxis_transport_layer_pc.transport_layer.print_available_ports()¶

Prints all serial ports active on the host-system with descriptive information about the device connected to that port to the terminal.

This command is intended to be used for discovering the USB ports that can be connected to by a TransportLayer class instance.

Return type:: None

Helper Modules¶

Provides the low-level helper classes that support the runtime of TransportLayer class methods.

class ataraxis_transport_layer_pc.helper_modules.COBSProcessor¶

Bases: object

Exposes the API for encoding and decoding data using the Consistent Overhead Byte Stuffing (COBS) scheme.

This class wraps a JIT-compiled COBS processor implementation, combining the convenience of a pure-python API with the speed of the C-compiled processing code.

Notes

This class is intended to be used by the TransportLayer class and should not be used directly by the end-users. It makes specific assumptions about the layout and contents of the processed data buffers that are not verified during runtime and must be enforced through the use of the TransportLayer class.

_processor¶: Stores the CompiledCOBSProcessor instance, which carries out all computations.

decode_payload(packet)¶

Decodes the COBS-encoded payload from the input packet.

Expects the input packets to adhere to the following structure: [Overhead] … [COBS Encoded Payload] … [Delimiter].

Parameters:: packet (ndarray[tuple[Any, ...], dtype[uint8]]) – The COBS-encoded packet from which to decode the payload.
Return type:: ndarray[tuple[Any, ...], dtype[uint8]]
Returns:: The payload decoded from the packet.
Raises:: ValueError – If the decoding fails, indicating uncaught packet corruption.

encode_payload(payload)¶

Encodes the input payload into a transmittable packet using COBS scheme.

The encoding produces the following packet structure: [Overhead] … [COBS Encoded Payload] … [Delimiter].

Parameters:: payload (ndarray[tuple[Any, ...], dtype[uint8]]) – The payload to be encoded using the COBS scheme.
Return type:: ndarray[tuple[Any, ...], dtype[uint8]]
Returns:: The serialized packet encoded using the COBS scheme.

property processor: CompiledCOBSProcessor¶: Returns the jit-compiled COBS processor instance, which external code can use to interface with the compiled class directly and bypass the Python wrapper.

class ataraxis_transport_layer_pc.helper_modules.CRCProcessor(polynomial, initial_crc_value, final_xor_value, *, reflected=False)¶

Bases: object

Exposes the API for working with Cyclic Redundancy Check (CRC) checksums used to verify the integrity of transferred data packets.

This class wraps a JIT-compiled CRC processor implementation, combining the convenience of a pure-python API with the speed of the C-compiled processing code.

Notes

Parameters:

polynomial (uint8 | uint16 | uint32) – The polynomial to use for the generation of the CRC lookup table. The polynomial must be expressed in the standard non-reflected, MSB-aligned form used by published CRC parameter catalogues.
initial_crc_value (uint8 | uint16 | uint32) – The value to which the CRC checksum is initialized before calculation.
final_xor_value (uint8 | uint16 | uint32) – The value with which the CRC checksum is XORed after calculation.
reflected (bool, default: False) – Determines whether the processor consumes each data byte least significant bit first and writes the checksum postamble least significant byte first.

_processor¶: Stores the CompiledCRCProcessor instance, which carries out all computations.

Raises:

TypeError – If class initialization arguments are not of the valid type.
ValueError – If the initial CRC value or the final XOR value is wider than the polynomial.

calculate_checksum(buffer, check)¶

Calculates the checksum for the data stored in the input buffer.

Depending on configuration, this method can be used to either generate and write the CRC checksum to the end of the packet or to verify the integrity of the incoming packet using its checksum postamble.

Parameters:

buffer (ndarray[tuple[Any, ...], dtype[uint8]]) – The buffer that contains the COBS-encoded packet for which to resolve the checksum. The buffer must include the space for the CRC checksum at the end of the packet.
check (bool) – Determines whether the method is called to verify the incoming packet’s data integrity or to generate and write the CRC checksum to the outgoing packet’s postamble section.

Return type:

uint16

Returns:

The total size of the buffer, including the appended CRC checksum, when generating a new checksum. When verifying data integrity, returns the value 1 to indicate the data is intact.

Raises:

ValueError – If the method is unable to verify the incoming packet’s data integrity.

property crc_byte_length: uint8¶: Returns the byte-size used by the CRC checksums.

property crc_table: ndarray[tuple[Any, ...], dtype[CRCType]]¶: Returns a copy of the CRC checksum lookup table.

property final_xor_value: CRCType¶: Returns the final XOR value used for checksum calculation.

property initial_crc_value: CRCType¶: Returns the initial value used for checksum calculation.

property polynomial: CRCType¶: Returns the polynomial used for checksum calculation.

property processor: CompiledCRCProcessor¶: Returns the jit-compiled CRC processor instance, which external code can use to interface with the compiled class directly and bypass the Python wrapper.

property reflected: bool¶: Returns whether the processor consumes data and writes the checksum postamble least significant end first.

type ataraxis_transport_layer_pc.helper_modules.CRCType = uint8 | uint16 | uint32¶

class ataraxis_transport_layer_pc.helper_modules.CompiledCOBSProcessor¶

Bases: object

Provides methods for encoding and decoding data using the Consistent Overhead Byte Stuffing (COBS) scheme.

Notes

This class is intended to be initialized through Numba’s ‘jitclass’ function.

See the original paper for the details on COBS methodology and specific data packet layouts: S. Cheshire and M. Baker, “Consistent overhead byte stuffing,” in IEEE/ACM Transactions on Networking, vol. 7, no. 2, pp. 159-172, April 1999, doi: 10.1109/90.769765.

maximum_payload_size¶: The maximum size of the payload, in bytes. Due to COBS, cannot exceed 254 bytes.

minimum_payload_size¶: The minimum size of the payload, in bytes.

maximum_packet_size¶: The maximum size of the packet, in bytes. Due to COBS, it cannot exceed 256 bytes (254 payload bytes + 1 overhead + 1 delimiter byte).

minimum_packet_size¶: The minimum size of the packet, in bytes. Due to COBS cannot be below 3 bytes.

delimiter¶: The byte value used as the packet delimiter.

decode_payload(packet)¶

Decodes the COBS-encoded payload from the input packet.

Parameters:: packet (ndarray[tuple[Any, ...], dtype[uint8]]) – The COBS-encoded packet from which to decode the payload.
Return type:: ndarray[tuple[Any, ...], dtype[uint8]]
Returns:: The payload decoded from the packet or an empty uninitialized numpy array if the method fails to decode the payload.

encode_payload(payload)¶

Encodes the input payload into a transmittable packet using the COBS scheme.

Parameters:: payload (ndarray[tuple[Any, ...], dtype[uint8]]) – The payload to be encoded using the COBS scheme.
Return type:: ndarray[tuple[Any, ...], dtype[uint8]]
Returns:: The packet encoded using the COBS scheme.

class ataraxis_transport_layer_pc.helper_modules.CompiledCRCProcessor(polynomial, initial_crc_value, final_xor_value, reflected)¶

Bases: object

Provides methods for working with Cyclic Redundancy Check (CRC) checksums used to verify the integrity of transferred data packets.

Notes

This class is intended to be initialized through Numba’s ‘jitclass’ function.

For more information on how the CRC checksum works, see the original paper: W. W. Peterson and D. T. Brown, “Cyclic Codes for Error Detection,” in Proceedings of the IRE, vol. 49, no. 1, pp. 228-235, Jan. 1961, doi: 10.1109/JRPROC.1961.287814.

To increase runtime speed, this class generates a static CRC lookup table using the input polynomial, which is subsequently used to calculate CRC checksums.

Parameters:

polynomial (uint8 | uint16 | uint32) – The polynomial used to generate the CRC lookup table, expressed in the standard non-reflected, MSB-aligned form used by published CRC parameter catalogues.
initial_crc_value (uint8 | uint16 | uint32) – The initial value to which the CRC checksum variable is initialized during calculation.
final_xor_value (uint8 | uint16 | uint32) – The final XOR value to be applied to the calculated CRC checksum value.
reflected (bool) – Determines whether the class consumes each data byte least significant bit first and writes the checksum postamble least significant byte first.

polynomial¶: Stores the polynomial used for the CRC checksum calculation.

initial_crc_value¶: Stores the initial value used for the CRC checksum calculation.

final_xor_value¶: Stores the final XOR value used for the CRC checksum calculation.

reflected¶: Determines whether the class processes data least significant end first.

initial_register¶: Stores the initial value in the form the checksum register consumes, which is the bit-reversed initial value for reflected configurations.

crc_byte_length¶: Stores the length of the CRC polynomial in bytes.

crc_table¶: The array that stores the CRC lookup table.

expected_residue¶: Stores the checksum value that verifying an intact packet produces.

calculate_checksum(buffer, check=False)¶

Calculates the checksum for the data stored in the input buffer.

Parameters:

buffer (ndarray[tuple[Any, ...], dtype[uint8]]) – The buffer that contains the COBS-encoded packet for which to resolve the checksum. The buffer must include the space for the CRC checksum at the end of the packet.
check (bool, default: False) – Determines whether the method is called to verify the incoming packet’s data integrity or to generate and write the CRC checksum to the outgoing packet’s postamble section.

Return type:

uint16

Returns:

The size of the buffer occupied by the packet’s data and the appended CRC checksum if the method is called to calculate the new CRC checksum. The value ‘1’ if the method is configured to verify the packet’s data integrity and the data is intact and ‘0’ otherwise.

class ataraxis_transport_layer_pc.helper_modules.SerialMock¶

Bases: object

Mocks the behavior of the PySerial’s Serial class for testing purposes.

This class provides a mock implementation of the Serial class, enabling unit tests for the TransportLayer class without a hardware connection. It replicates the core functionalities of the PySerial’s Serial class that are relevant for testing, such as reading and writing data.

is_open¶: Determines whether the mock serial port is open.

tx_buffer¶: A byte buffer that stores transmitted data.

rx_buffer¶: A byte buffer that stores received data.

in_waiting¶: A read-only property returning the number of bytes available for reading from the rx_buffer.

out_waiting¶: A read-only property returning the number of bytes pending transmission in the tx_buffer.

close()¶

Closes the mock serial port, setting is_open to False.

Return type:: None

property in_waiting: int¶: Returns the number of bytes stored in the rx_buffer.

open()¶

Opens the mock serial port, setting is_open to True.

Return type:: None

property out_waiting: int¶: Returns the number of bytes stored in the tx_buffer.

read(size=1)¶

Reads a specified number of bytes from the rx_buffer.

Parameters:: size (int, default: 1) – The number of bytes to read from the input buffer.
Return type:: bytes
Returns:: A bytes’ object containing the requested data from the rx_buffer.
Raises:: RuntimeError – If the mock serial port is not open.

reset_input_buffer()¶

Clears the rx_buffer attribute.

Raises:: RuntimeError – If the mock serial port is not open.
Return type:: None

reset_output_buffer()¶

Clears the tx_buffer attribute.

Raises:: RuntimeError – If the mock serial port is not open.
Return type:: None

write(data)¶

Writes data to the tx_buffer.

Parameters:

data (bytes) – The serialized data to be written to the output buffer.

Raises:

TypeError – If data is not a bytes’ object.
RuntimeError – If the mock serial port is not open.

Return type:

None