Communication

Provides the SerialCommunication and MQTTCommunication classes that jointly support the communication between and within host-machines (PCs) and Arduino / Teensy microcontrollers.

class ataraxis_communication_interface.communication.ControllerIdentification(message=<factory>)

Bases: object

Communicates the unique identifier code of the microcontroller.

property controller_id: uint8: Returns the unique identifier of the microcontroller.

message: ndarray[tuple[Any, ...], dtype[uint8]]: The parsed message header data.

class ataraxis_communication_interface.communication.DequeueModuleCommand(module_type, module_id, return_code=np.uint8(0))

Bases: object

Instructs the addressed Module instance to clear (empty) its command queue.

module_id: uint8: The ID of the specific module instance within the broader module family.

module_type: uint8: The type (family) code of the module to which the command is addressed.

packed_data: ndarray[tuple[Any, ...], dtype[uint8]] | None = None: Stores the serialized message data.

protocol_code: uint8 = np.uint8(3): Stores the message protocol code.

return_code: uint8 = np.uint8(0): The code to use for acknowledging the reception of the message, if set to a non-zero value.

class ataraxis_communication_interface.communication.KernelCommand(command, return_code=np.uint8(0))

Bases: object

Instructs the Kernel to run the specified command exactly once.

command: uint8: The code of the command to execute.

packed_data: ndarray[tuple[Any, ...], dtype[uint8]] | None = None: Stores the serialized message data.

protocol_code: uint8 = np.uint8(4): Stores the message protocol code.

return_code: uint8 = np.uint8(0): The code to use for acknowledging the reception of the message, if set to a non-zero value.

class ataraxis_communication_interface.communication.KernelData(message=<factory>, data_object=np.uint8(0))

Bases: object

Communicates that the Kernel has encountered a notable event and includes an additional data object.

property command: uint8: Returns the code of the command executed by the Kernel when it sent the message.

data_object: number[Any] | ndarray[tuple[Any, ...], dtype[Any]] = np.uint8(0): The parsed data object transmitted with the message.

property event: uint8: Returns the code of the event that prompted sending the message.

message: ndarray[tuple[Any, ...], dtype[uint8]]: The parsed message header data.

property prototype_code: uint8: Returns the code that specifies the type of the data object transmitted with the message.

class ataraxis_communication_interface.communication.KernelState(message=<factory>)

Bases: object

Communicates that the Kernel has encountered a notable event.

property command: uint8: Returns the code of the command executed by the Kernel when it sent the message.

property event: uint8: Returns the code of the event that prompted sending the message.

message: ndarray[tuple[Any, ...], dtype[uint8]]: The parsed message header data.

class ataraxis_communication_interface.communication.MQTTCommunication(ip='127.0.0.1', port=1883, monitored_topics=None)

Bases: object

Provides methods for bidirectionally communicating with other clients connected to the same MQTT broker using the MQTT protocol over the TCP interface.

Notes

Primarily, the class is intended to be used alongside the SerialCommunication class to transfer the data between microcontrollers and the rest of the runtime infrastructure.

The MQTT protocol requires a broker that facilitates the communication, which has to be available to this class at initialization. See https://mqtt.org/ for more details.

Parameters:

ip (str, default: '127.0.0.1') – The IP address of the MQTT broker.
port (int, default: 1883) – The socket port used by the MQTT broker.
monitored_topics (None | tuple[str, ...], default: None) – The list of MQTT topics to monitor for incoming messages.

_ip: Stores the IP address of the MQTT broker.

_port: Stores the port used by the broker’s TCP socket.

_connected: Tracks whether the class instance is currently connected to the MQTT broker.

_monitored_topics: Stores the topics monitored by the instance for incoming messages.

_output_queue: Buffers incoming messages received from other MQTT clients before their data is accessed via class methods.

_client: The initialized MQTT client instance that carries out the communication.

connect()

Connects to the MQTT broker and subscribes to the requested list of monitored topics.

Return type:: None

Notes

This method has to be called after class initialization to start the communication process. Any message sent to the MQTT broker from other clients before this method is called may not reach this instance.

If this instance is configured to subscribe (listen) to any topics, it starts a perpetually active thread with a listener callback to monitor the incoming traffic.

Raises:: ConnectionError – If the MQTT broker cannot be connected using the provided IP and Port.

disconnect()

Disconnects the client from the MQTT broker.

Return type:: None

get_data()

Extracts and returns the first available message stored inside the instance’s buffer queue.

Return type:: tuple[str, bytes | bytearray] | None
Returns:: A two-element tuple if there is data to retrieve. The first element is the MQTT topic of the received message. The second element is the payload of the message. If there is no data to retrieve, returns None.
Raises:: ConnectionError – If the instance is not connected to the MQTT broker.

property has_data: bool: Returns True if the instance’s get_data() method can be used to retrieve a message received from another MQTT client.

send_data(topic, payload=None)

Publishes the input payload to the specified MQTT topic.

Parameters:

topic (str) – The MQTT topic to publish the data to.
payload (str | bytes | bytearray | float | None, default: None) – The data to be published. Setting this to None sends an empty message.

Raises:

ConnectionError – If the instance is not connected to the MQTT broker.

Return type:

None

class ataraxis_communication_interface.communication.ModuleData(message=<factory>, data_object=np.uint8(0))

Bases: object

Communicates that the Module has encountered a notable event and includes an additional data object.

property command: uint8: Returns the code of the command executed by the module that sent the message.

data_object: number[Any] | ndarray[tuple[Any, ...], dtype[Any]] = np.uint8(0): The parsed data object transmitted with the message.

property event: uint8: Returns the code of the event that prompted sending the message.

message: ndarray[tuple[Any, ...], dtype[uint8]]: The parsed message header data.

property module_id: uint8: Returns the unique identifier code of the module instance that sent the message.

property module_type: uint8: Returns the type (family) code of the module that sent the message.

property prototype_code: uint8: Returns the code that specifies the type of the data object transmitted with the message.

class ataraxis_communication_interface.communication.ModuleIdentification(module_type_id=np.uint16(0))

Bases: object

Identifies a hardware module instance by communicating its combined type and id code.

module_type_id: uint16 = np.uint16(0): The unique uint16 code that results from combining the type and ID codes of the module instance.

class ataraxis_communication_interface.communication.ModuleParameters(module_type, module_id, parameter_data, return_code=np.uint8(0))

Bases: object

Instructs the addressed Module instance to update its parameters with the included data.

module_id: uint8: The ID of the specific module instance within the broader module family.

module_type: uint8: The type (family) code of the module to which the command is addressed.

packed_data: ndarray[tuple[Any, ...], dtype[uint8]] | None = None: Stores the serialized message data.

parameter_data: tuple[number[Any] | bool, ...]: A tuple of parameter values to send. The values inside the tuple must match the type and format of the values used in the addressed module’s parameter structure on the microcontroller.

parameters_size: ndarray[tuple[Any, ...], dtype[uint8]] | None = None: Stores the total size of the serialized parameters in bytes.

protocol_code: uint8 = np.uint8(5): Stores the message protocol code.

return_code: uint8 = np.uint8(0): The code to use for acknowledging the reception of the message, if set to a non-zero value.

class ataraxis_communication_interface.communication.ModuleState(message=<factory>)

Bases: object

Communicates that the Module has encountered a notable event.

property command: uint8: Returns the code of the command executed by the module that sent the message.

property event: uint8: Returns the code of the event that prompted sending the message.

message: ndarray[tuple[Any, ...], dtype[uint8]]: The parsed message header data.

property module_id: uint8: Returns the ID of the specific module instance within the broader module family.

property module_type: uint8: Returns the type (family) code of the module that sent the message.

class ataraxis_communication_interface.communication.OneOffModuleCommand(module_type, module_id, command, return_code=np.uint8(0), noblock=np.True_)

Bases: object

Instructs the addressed Module instance to run the specified command exactly once (non-recurrently).

command: uint8: The code of the command to execute.

module_id: uint8: The ID of the specific module instance within the broader module family.

module_type: uint8: The type (family) code of the module to which the command is addressed.

noblock: bool = np.True_: Determines whether to allow concurrent execution of other commands while waiting for the requested command to complete.

packed_data: ndarray[tuple[Any, ...], dtype[uint8]] | None = None: Stores the serialized message data.

protocol_code: uint8 = np.uint8(2): Stores the message protocol code.

return_code: uint8 = np.uint8(0): The code to use for acknowledging the reception of the message, if set to a non-zero value.

class ataraxis_communication_interface.communication.ReceptionCode(message=<factory>)

Bases: object

Communicates the reception code originally received with the message sent by the PC to indicate that the message was received and parsed by the microcontroller.

message: ndarray[tuple[Any, ...], dtype[uint8]]: The parsed message header data.

property reception_code: uint8: Returns the reception code originally sent as part of the outgoing Command or Parameters message.

class ataraxis_communication_interface.communication.RepeatedModuleCommand(module_type, module_id, command, return_code=np.uint8(0), noblock=np.True_, cycle_delay=np.uint32(0))

Bases: object

Instructs the addressed Module instance to run the specified command repeatedly (recurrently).

command: uint8: The code of the command to execute.

cycle_delay: uint32 = np.uint32(0): The delay, in microseconds, before repeating (cycling) the command.

module_id: uint8: The ID of the specific module instance within the broader module family.

module_type: uint8: The type (family) code of the module to which the command is addressed.

noblock: bool = np.True_: Determines whether to allow concurrent execution of other commands while waiting for the requested command to complete.

packed_data: ndarray[tuple[Any, ...], dtype[uint8]] | None = None: Stores the serialized message data.

protocol_code: uint8 = np.uint8(1): Stores the message protocol code.

return_code: uint8 = np.uint8(0): The code to use for acknowledging the reception of the message, if set to a non-zero value.

class ataraxis_communication_interface.communication.SerialCommunication(controller_id, microcontroller_serial_buffer_size, port, logger_queue, baudrate=115200, *, test_mode=False)

Bases: object

Provides methods for bidirectionally communicating with a microcontroller running the ataraxis-micro-controller library over the USB or UART serial interface.

Notes

This class is explicitly designed to be used by other library assets and should not be used directly by end users. An instance of this class is initialized and managed by the MicrocontrollerInterface class.

Parameters:

controller_id (uint8) – The identifier code of the microcontroller to communicate with.
microcontroller_serial_buffer_size (int) – The size, in bytes, of the buffer used by the communicated microcontroller’s serial communication interface. Usually, this information is available from the microcontroller’s manufacturer (UART / USB controller specification).
port (str) – The name of the serial port to connect to, e.g.: ‘COM3’ or ‘/dev/ttyUSB0’.
logger_queue (Queue) – The multiprocessing Queue object exposed by the DataLogger instance used to pipe the data to be logged to the logger process.
baudrate (int, default: 115200) – The baudrate to use for communication if the microcontroller uses the UART interface. Must match the value used by the microcontroller. This parameter is ignored when using the USB interface.
test_mode (bool, default: False) – Determines whether the instance uses a pySerial (real) or a StreamMock (mocked) communication interface. This flag is used during testing and should be disabled for all production runtimes.

_transport_layer: The TransportLayer instance that handles the communication.

_module_data: Stores the data of the last received ModuleData message.

_kernel_data: Stores the data of the last received KernelData message.

_module_state: Stores the data of the last received ModuleState message.

_kernel_state: Stores the data of the last received KernelState message.

_controller_identification: Stores the data of the last received ControllerIdentification message.

_module_identification: Stores the data of the last received ModuleIdentification message.

_reception_code: Stores the data of the last received ReceptionCode message.

_timestamp_timer: Stores the PrecisionTimer instance used to timestamp incoming and outgoing data as it is being saved (logged) to disk.

_source_id: Stores the unique identifier of the microcontroller with which the instance communicates at runtime.

_logger_queue: Stores the multiprocessing Queue that buffers and pipes the data to the DataLogger process(es).

_usb_port: Stores the ID of the USB port used for communication.

receive_message()

Receives a message sent by the microcontroller and parses its contents into the appropriate instance attribute.

Notes

Each call to this method overwrites the previously received message data stored in the instance’s attributes. It is advised to finish working with the received message data before receiving another message.

Return type:: ModuleData | ModuleState | KernelData | KernelState | ControllerIdentification | ModuleIdentification | ReceptionCode | None
Returns:: A reference to the parsed message data stored as an instance’s attribute, or None, if no message was received.
Raises:: ValueError – If the received message uses an invalid (unrecognized) message protocol code. If the received data message uses an unsupported data object prototype code.

send_message(message)

Serializes the input message and sends it to the connected microcontroller.

Parameters:: message (RepeatedModuleCommand | OneOffModuleCommand | DequeueModuleCommand | KernelCommand | ModuleParameters) – The message to send to the microcontroller.
Return type:: None

class ataraxis_communication_interface.communication.SerialProtocols(*values)

Bases: IntEnum

Defines the protocol codes used to specify incoming and outgoing message layouts during PC-microcontroller communication.

Notes

The elements in this enumeration should be accessed through their ‘as_uint8’ property to enforce the type expected by other classes from this library.

CONTROLLER_IDENTIFICATION = 11: Used to identify the host-microcontroller to the PC.

DEQUEUE_MODULE_COMMAND = 3: Used by Module-addressed commands that remove all queued commands, including recurrent commands.

KERNEL_COMMAND = 4: Used by Kernel-addressed commands. All Kernel commands are always non-repeatable (one-shot).

KERNEL_DATA = 7: Used by Kernel data or error messages that include an arbitrary data object in addition to event state-code.

KERNEL_STATE = 9: Used by Kernel data or error messages that only include the state-code.

MODULE_DATA = 6: Used by Module data or error messages that include an arbitrary data object in addition to the event state-code.

MODULE_IDENTIFICATION = 12: Used to identify the hardware module instances managed by the microcontroller’s Kernel instance to the PC.

MODULE_PARAMETERS = 5: Used by Module-addressed parameter messages.

MODULE_STATE = 8: Used by Module data or error messages that only include the state-code.

ONE_OFF_MODULE_COMMAND = 2: Used by Module-addressed commands that should not be repeated (executed only once).

RECEPTION_CODE = 10: Used to acknowledge the reception of command and parameter messages from the PC.

REPEATED_MODULE_COMMAND = 1: Used by Module-addressed commands that should be repeated (executed recurrently).

UNDEFINED = 0: Not a valid protocol code. Used to initialize the SerialCommunication class.

as_uint8()

Returns the specified enumeration element as a numpy uint8 type.

Return type:: uint8

class ataraxis_communication_interface.communication.SerialPrototypes(*values)

Bases: IntEnum

Defines the prototype codes used during data transmission to specify the layout of additional data objects transmitted by KernelData and ModuleData messages.

EIGHT_BOOLS = 31: An array of 8 8-bit booleans

EIGHT_FLOAT32S = 108: An array of 8 single-precision 32-bit floating-point numbers

EIGHT_FLOAT64S = 144: An array of 8 double-precision 64-bit floating-point numbers

EIGHT_INT16S = 73: An array of 8 signed 16-bit integers

EIGHT_INT32S = 107: An array of 8 signed 32-bit integers

EIGHT_INT64S = 143: An array of 8 signed 64-bit integers

EIGHT_INT8S = 33: An array of 8 signed 8-bit integers

EIGHT_UINT16S = 72: An array of 8 unsigned 16-bit integers

EIGHT_UINT32S = 106: An array of 8 unsigned 32-bit integers

EIGHT_UINT64S = 142: An array of 8 unsigned 64-bit integers

EIGHT_UINT8S = 32: An array of 8 unsigned 8-bit integers

ELEVEN_BOOLS = 50: An array of 11 8-bit booleans

ELEVEN_FLOAT32S = 123: An array of 11 single-precision 32-bit floating-point numbers

ELEVEN_FLOAT64S = 153: An array of 11 double-precision 64-bit floating-point numbers

ELEVEN_INT16S = 88: An array of 11 signed 16-bit integers

ELEVEN_INT32S = 122: An array of 11 signed 32-bit integers

ELEVEN_INT64S = 152: An array of 11 signed 64-bit integers

ELEVEN_INT8S = 52: An array of 11 signed 8-bit integers

ELEVEN_UINT16S = 87: An array of 11 unsigned 16-bit integers

ELEVEN_UINT32S = 121: An array of 11 unsigned 32-bit integers

ELEVEN_UINT64S = 151: An array of 11 unsigned 64-bit integers

ELEVEN_UINT8S = 51: An array of 11 unsigned 8-bit integers

FIFTEEN_BOOLS = 69: An array of 15 8-bit booleans

FIFTEEN_FLOAT32S = 141: An array of 15 single-precision 32-bit floating-point numbers

FIFTEEN_FLOAT64S = 165: An array of 15 double-precision 64-bit floating-point numbers

FIFTEEN_INT16S = 105: An array of 15 signed 16-bit integers

FIFTEEN_INT32S = 140: An array of 15 signed 32-bit integers

FIFTEEN_INT64S = 164: An array of 15 signed 64-bit integers

FIFTEEN_INT8S = 71: An array of 15 signed 8-bit integers

FIFTEEN_UINT16S = 104: An array of 15 unsigned 16-bit integers

FIFTEEN_UINT32S = 139: An array of 15 unsigned 32-bit integers

FIFTEEN_UINT64S = 163: An array of 15 unsigned 64-bit integers

FIFTEEN_UINT8S = 70: An array of 15 unsigned 8-bit integers

FIVE_BOOLS = 20: An array of 5 8-bit booleans

FIVE_FLOAT32S = 86: An array of 5 single-precision 32-bit floating-point numbers

FIVE_FLOAT64S = 120: An array of 5 double-precision 64-bit floating-point numbers

FIVE_INT16S = 49: An array of 5 signed 16-bit integers

FIVE_INT32S = 85: An array of 5 signed 32-bit integers

FIVE_INT64S = 119: An array of 5 signed 64-bit integers

FIVE_INT8S = 22: An array of 5 signed 8-bit integers

FIVE_UINT16S = 48: An array of 5 unsigned 16-bit integers

FIVE_UINT32S = 84: An array of 5 unsigned 32-bit integers

FIVE_UINT64S = 118: An array of 5 unsigned 64-bit integers

FIVE_UINT8S = 21: An array of 5 unsigned 8-bit integers

FOURTEEN_BOOLS = 64: An array of 14 8-bit booleans

FOURTEEN_FLOAT32S = 135: An array of 14 single-precision 32-bit floating-point numbers

FOURTEEN_FLOAT64S = 162: An array of 14 double-precision 64-bit floating-point numbers

FOURTEEN_INT16S = 100: An array of 14 signed 16-bit integers

FOURTEEN_INT32S = 134: An array of 14 signed 32-bit integers

FOURTEEN_INT64S = 161: An array of 14 signed 64-bit integers

FOURTEEN_INT8S = 66: An array of 14 signed 8-bit integers

FOURTEEN_UINT16S = 99: An array of 14 unsigned 16-bit integers

FOURTEEN_UINT32S = 133: An array of 14 unsigned 32-bit integers

FOURTEEN_UINT64S = 160: An array of 14 unsigned 64-bit integers

FOURTEEN_UINT8S = 65: An array of 14 unsigned 8-bit integers

FOUR_BOOLS = 12: An array of 4 8-bit booleans

FOUR_FLOAT32S = 76: An array of 4 single-precision 32-bit floating-point numbers

FOUR_FLOAT64S = 111: An array of 4 double-precision 64-bit floating-point numbers

FOUR_INT16S = 35: An array of 4 signed 16-bit integers

FOUR_INT32S = 75: An array of 4 signed 32-bit integers

FOUR_INT64S = 110: An array of 4 signed 64-bit integers

FOUR_INT8S = 14: An array of 4 signed 8-bit integers

FOUR_UINT16S = 34: An array of 4 unsigned 16-bit integers

FOUR_UINT32S = 74: An array of 4 unsigned 32-bit integers

FOUR_UINT64S = 109: An array of 4 unsigned 64-bit integers

FOUR_UINT8S = 13: An array of 4 unsigned 8-bit integers

NINE_BOOLS = 42: An array of 9 8-bit booleans

NINE_FLOAT32S = 114: An array of 9 single-precision 32-bit floating-point numbers

NINE_FLOAT64S = 147: An array of 9 double-precision 64-bit floating-point numbers

NINE_INT16S = 81: An array of 9 signed 16-bit integers

NINE_INT32S = 113: An array of 9 signed 32-bit integers

NINE_INT64S = 146: An array of 9 signed 64-bit integers

NINE_INT8S = 44: An array of 9 signed 8-bit integers

NINE_UINT16S = 80: An array of 9 unsigned 16-bit integers

NINE_UINT32S = 112: An array of 9 unsigned 32-bit integers

NINE_UINT64S = 145: An array of 9 unsigned 64-bit integers

NINE_UINT8S = 43: An array of 9 unsigned 8-bit integers

ONE_BOOL = 1: 1 8-bit boolean

ONE_FLOAT32 = 19: 1 single-precision 32-bit floating-point number

ONE_FLOAT64 = 41: 1 double-precision 64-bit floating-point number

ONE_INT16 = 8: 1 signed 16-bit integer

ONE_INT32 = 18: 1 signed 32-bit integer

ONE_INT64 = 40: 1 signed 64-bit integer

ONE_INT8 = 3: 1 signed 8-bit integer

ONE_UINT16 = 7: 1 unsigned 16-bit integer

ONE_UINT32 = 17: 1 unsigned 32-bit integer

ONE_UINT64 = 39: 1 unsigned 64-bit integer

ONE_UINT8 = 2: 1 unsigned 8-bit integer

SEVEN_BOOLS = 28: An array of 7 8-bit booleans

SEVEN_FLOAT32S = 103: An array of 7 single-precision 32-bit floating-point numbers

SEVEN_FLOAT64S = 138: An array of 7 double-precision 64-bit floating-point numbers

SEVEN_INT16S = 68: An array of 7 signed 16-bit integers

SEVEN_INT32S = 102: An array of 7 signed 32-bit integers

SEVEN_INT64S = 137: An array of 7 signed 64-bit integers

SEVEN_INT8S = 30: An array of 7 signed 8-bit integers

SEVEN_UINT16S = 67: An array of 7 unsigned 16-bit integers

SEVEN_UINT32S = 101: An array of 7 unsigned 32-bit integers

SEVEN_UINT64S = 136: An array of 7 unsigned 64-bit integers

SEVEN_UINT8S = 29: An array of 7 unsigned 8-bit integers

SIX_BOOLS = 23: An array of 6 8-bit booleans

SIX_FLOAT32S = 93: An array of 6 single-precision 32-bit floating-point numbers

SIX_FLOAT64S = 129: An array of 6 double-precision 64-bit floating-point numbers

SIX_INT16S = 57: An array of 6 signed 16-bit integers

SIX_INT32S = 92: An array of 6 signed 32-bit integers

SIX_INT64S = 128: An array of 6 signed 64-bit integers

SIX_INT8S = 25: An array of 6 signed 8-bit integers

SIX_UINT16S = 56: An array of 6 unsigned 16-bit integers

SIX_UINT32S = 91: An array of 6 unsigned 32-bit integers

SIX_UINT64S = 127: An array of 6 unsigned 64-bit integers

SIX_UINT8S = 24: An array of 6 unsigned 8-bit integers

TEN_BOOLS = 45: An array of 10 8-bit booleans

TEN_FLOAT32S = 117: An array of 10 single-precision 32-bit floating-point numbers

TEN_FLOAT64S = 150: An array of 10 double-precision 64-bit floating-point numbers

TEN_INT16S = 83: An array of 10 signed 16-bit integers

TEN_INT32S = 116: An array of 10 signed 32-bit integers

TEN_INT64S = 149: An array of 10 signed 64-bit integers

TEN_INT8S = 47: An array of 10 signed 8-bit integers

TEN_UINT16S = 82: An array of 10 unsigned 16-bit integers

TEN_UINT32S = 115: An array of 10 unsigned 32-bit integers

TEN_UINT64S = 148: An array of 10 unsigned 64-bit integers

TEN_UINT8S = 46: An array of 10 unsigned 8-bit integers

THIRTEEN_BOOLS = 61: An array of 13 8-bit booleans

THIRTEEN_FLOAT32S = 132: An array of 13 single-precision 32-bit floating-point numbers

THIRTEEN_FLOAT64S = 159: An array of 13 double-precision 64-bit floating-point numbers

THIRTEEN_INT16S = 98: An array of 13 signed 16-bit integers

THIRTEEN_INT32S = 131: An array of 13 signed 32-bit integers

THIRTEEN_INT64S = 158: An array of 13 signed 64-bit integers

THIRTEEN_INT8S = 63: An array of 13 signed 8-bit integers

THIRTEEN_UINT16S = 97: An array of 13 unsigned 16-bit integers

THIRTEEN_UINT32S = 130: An array of 13 unsigned 32-bit integers

THIRTEEN_UINT64S = 157: An array of 13 unsigned 64-bit integers

THIRTEEN_UINT8S = 62: An array of 13 unsigned 8-bit integers

THREE_BOOLS = 9: An array of 3 8-bit booleans

THREE_FLOAT32S = 60: An array of 3 single-precision 32-bit floating-point numbers

THREE_FLOAT64S = 96: An array of 3 double-precision 64-bit floating-point numbers

THREE_INT16S = 27: An array of 3 signed 16-bit integers

THREE_INT32S = 59: An array of 3 signed 32-bit integers

THREE_INT64S = 95: An array of 3 signed 64-bit integers

THREE_INT8S = 11: An array of 3 signed 8-bit integers

THREE_UINT16S = 26: An array of 3 unsigned 16-bit integers

THREE_UINT32S = 58: An array of 3 unsigned 32-bit integers

THREE_UINT64S = 94: An array of 3 unsigned 64-bit integers

THREE_UINT8S = 10: An array of 3 unsigned 8-bit integers

TWELVE_BOOLS = 53: An array of 12 8-bit booleans

TWELVE_FLOAT32S = 126: An array of 12 single-precision 32-bit floating-point numbers

TWELVE_FLOAT64S = 156: An array of 12 double-precision 64-bit floating-point numbers

TWELVE_INT16S = 90: An array of 12 signed 16-bit integers

TWELVE_INT32S = 125: An array of 12 signed 32-bit integers

TWELVE_INT64S = 155: An array of 12 signed 64-bit integers

TWELVE_INT8S = 55: An array of 12 signed 8-bit integers

TWELVE_UINT16S = 89: An array of 12 unsigned 16-bit integers

TWELVE_UINT32S = 124: An array of 12 unsigned 32-bit integers

TWELVE_UINT64S = 154: An array of 12 unsigned 64-bit integers

TWELVE_UINT8S = 54: An array of 12 unsigned 8-bit integers

TWO_BOOLS = 4: An array of 2 8-bit booleans

TWO_FLOAT32S = 38: An array of 2 single-precision 32-bit floating-point numbers

TWO_FLOAT64S = 79: An array of 2 double-precision 64-bit floating-point numbers

TWO_INT16S = 16: An array of 2 signed 16-bit integers

TWO_INT32S = 37: An array of 2 signed 32-bit integers

TWO_INT64S = 78: An array of 2 signed 64-bit integers

TWO_INT8S = 6: An array of 2 signed 8-bit integers

TWO_UINT16S = 15: An array of 2 unsigned 16-bit integers

TWO_UINT32S = 36: An array of 2 unsigned 32-bit integers

TWO_UINT64S = 77: An array of 2 unsigned 64-bit integers

TWO_UINT8S = 5: An array of 2 unsigned 8-bit integers

as_uint8()

Returns the specified the enumeration value as a numpy uint8 type.

Return type:: uint8

get_prototype()

Returns the prototype object associated with the prototype enumeration value.

Return type:: TypeAliasType

classmethod get_prototype_for_code(code)

Returns the prototype object associated with the input prototype code.

Parameters:: code (uint8) – The prototype byte-code for which to retrieve the prototype object.
Return type:: TypeAliasType | None
Returns:: The prototype object that is either a numpy scalar or shallow array type. If the input code is not one of the supported codes, returns None to indicate a matching error.

ataraxis_communication_interface.communication.check_mqtt_connectivity(host='127.0.0.1', port=1883)

Checks whether an MQTT broker is reachable at the specified host and port.

This function attempts to connect to the MQTT broker and reports the result. It is intended to be used as a CLI command to verify MQTT broker availability before running code that depends on MQTT communication.

Parameters:

host (str, default: '127.0.0.1') – The IP address or hostname of the MQTT broker.
port (int, default: 1883) – The socket port used by the MQTT broker.

Return type:

None

MicroController Interface

Provides the ModuleInterface and MicroControllerInterface classes that aggregate the methods allowing Python PC clients to bidirectionally interface with custom hardware modules managed by Arduino or Teensy microcontrollers.

class ataraxis_communication_interface.microcontroller_interface.ExtractedMessageData(timestamp, command, data)

Bases: object

Contains the data parsed from a message sent to the PC by a hardware module instance during runtime.

command: uint8: The code of the command that the module was executing when it sent the message to the PC.

data: None | number | ndarray[tuple[Any, ...], dtype[number]]: The parsed data object transmitted with the message (or None, for state-only messages).

timestamp: uint64: The number of microseconds elapsed since the UTC epoch onset when the message was received by the PC.

class ataraxis_communication_interface.microcontroller_interface.ExtractedModuleData(module_type, module_id, event_data)

Bases: object

Contains the data parsed from all non-service messages sent to the PC by a hardware module instance during runtime.

event_data: dict[uint8, tuple[ExtractedMessageData, ...]]: A dictionary that uses message event codes as keys and tuples of ExtractedMessageData instances as values.

module_id: int: Stores the unique identifier code of the hardware module instance whose data is stored in the ‘data’ attribute.

module_type: int: Stores the type (family) code of the hardware module instance whose data is stored in the ‘data’ attribute.

class ataraxis_communication_interface.microcontroller_interface.MicroControllerInterface(controller_id, data_logger, module_interfaces, buffer_size, port, baudrate=115200, keepalive_interval=0)

Bases: object

Interfaces with the hardware module instances managed by the Arduino or Teensy microcontroller running the ataraxis-micro-controller library.

This class binds each hardware module managed by the microcontroller to its user-facing interface implemented via this library. It abstracts all necessary steps to bidirectionally communicate with the microcontroller and log the incoming and outgoing message data to disk.

Notes

An instance of this class has to be instantiated for each microcontroller active at the same time.

Initializing this class does not automatically start the communication. Call the start() method of an initialized class instance to start the communication with the microcontroller.

Initializing MicroControllerInterface also completes the configuration of all ModuleInterface instances passed to the instance during initialization.

Parameters:

controller_id (uint8) – The unique identifier code of the managed microcontroller. This value is used to identify the microcontroller in all output streams (e.g., log files and terminal messages).
data_logger (DataLogger) – An initialized DataLogger instance used to log all incoming and outgoing messages handled by this MicroControllerInterface instance.
module_interfaces (tuple[ModuleInterface, ...]) – The custom hardware module interfaces for the hardware module instance managed by the microcontroller. Note, each module instance requires a unique interface instance.
buffer_size (int) – The size, in bytes, of the buffer used by the microcontroller’s serial communication interface. Usually, this information is available from the microcontroller’s manufacturer (UART / USB controller specification).
port (str) – The name of the serial port to connect to, e.g.: ‘COM3’ or ‘/dev/ttyUSB0’. Use the ‘axci-id’ CLI command to discover the available microcontrollers and their respective communication port names.
baudrate (int, default: 115200) – The baudrate to use for communication if the microcontroller uses the UART interface. Must match the value used by the microcontroller. This parameter is ignored when using the USB interface.
keepalive_interval (int, default: 0) – The interval, in milliseconds, at which to send the keepalive messages to the microcontroller. Setting this argument to 0 disables keepalive messaging functionality.

Raises:

TypeError – If any of the input arguments are not of the expected type.

_started: Tracks whether the communication process has been started.

_controller_id: Stores the id of the managed microcontroller.

_port: Stores the USB port used for microcontroller communication.

_baudrate: Stores the baudrate used during communication over the UART serial interface.

_buffer_size: Stores the microcontroller’s serial buffer size, in bytes.

_modules: Stores ModuleInterface instances managed by this MicroControllerInterface.

_logger_queue: The Multiprocessing Queue object used to pipe log data to the DataLogger core(s).

_log_directory: Stores the output directory used by the DataLogger to save temporary log entries and the final .npz log archive.

_mp_manager: The multiprocessing Manager used to initialize and manage the Queue instance that pipes command and parameter messages to the communication process.

_input_queue: The multiprocessing Queue used to pipe the data to be sent to the microcontroller to the remote communication process.

_terminator_array: Stores the SharedMemoryArray instance used to control the remote communication process.

_communication_process: Stores the Process instance that runs the communication cycle.

_watchdog_thread: Stores the thread used to monitor the runtime status of the remote communication process.

_reset_command: Stores the pre-packaged Kernel-addressed command that resets the managed microcontroller to the default state.

_keepalive_interval: Stores the keepalive interval in milliseconds.

reset_controller()

Resets the managed microcontroller to use the default hardware and software parameters.

Return type:: None

start()

Starts the instance’s communication process and begins interfacing with the microcontroller.

Return type:: None

Notes

As part of this method runtime, the interface verifies the target microcontroller’s configuration to ensure it matches the interface’s configuration.

Raises:: RuntimeError – If the instance fails to initialize the communication process.

stop()

Stops the instance’s communication process and releases all reserved resources.

Return type:: None

class ataraxis_communication_interface.microcontroller_interface.ModuleInterface(module_type, module_id, error_codes=None, data_codes=None)

Bases: ABC

Provides the API used by other library components to interface with the custom hardware module controlled by the companion Arduino or Teensy microcontroller.

Any class that inherits from this class gains the API used by the MicroControllerInterface class to bind the interface to the managed hardware module instance running on the companion microcontroller. Also, inheriting from this class provides the user-facing API for sending commands and parameters to the managed hardware module.

Notes

Every custom hardware module interface has to inherit from this base class. When inheriting from this class, initialize the superclass by calling the ‘super().__init__()’ during the subclass initialization.

All data received from or sent to the microcontroller is automatically logged to disk. Only provide additional data and error codes if the interface must carry out ‘online’ error detection or data processing.

Some attributes of this (base) class are assigned by the managing MicroControllerInterface during its initialization. Each module interface that inherits from the base ModuleInterface class has to be bound to an initialized MicroControllerInterface instance to be fully functional.

Use the utility methods inherited from the base ModuleInterface to send command and parameter messages to the managed hardware module instance. These methods are heavily optimized for runtime efficiency and performance.

Parameters:

module_type (uint8) – The code that identifies the type (family) of the interfaced module.
module_id (uint8) – The code that identifies the specific interfaced module instance.
error_codes (set[uint8] | None, default: None) – An optional set of codes used by the module to communicate runtime errors. Receiving a message with an event-code from this set raises a RuntimeError and aborts the runtime.
data_codes (set[uint8] | None, default: None) – An optional set of codes used by the module to communicate data messages that required online processing. Received messages with an event-code from this set are passed to the interface instance’s process_received_data() method for further processing.

_module_type: Stores the id-code of the managed hardware module’s type (family).

_module_id: Stores the specific instance ID of the managed hardware module.

_type_id: Stores the type and id codes combined into a single uint16 value.

_data_codes: Stores all message event-codes that require additional processing.

_error_codes: Stores all message error-codes that warrant runtime interruption.

_input_queue: The multiprocessing queue used to send command and parameter messages to the microcontroller communication process.

_dequeue_command: Stores the instance’s DequeueModuleCommand object.

_create_command_message: LRU-cached method (maxsize=32) for creating command message objects. Caches up to 32 unique command message configurations to avoid redundant object creation and serialization during repeated command operations.

_create_parameters_message: LRU-cached method (maxsize=16) for creating parameter message objects. Caches up to 16 unique parameter configurations to avoid redundant object creation and serialization when repeatedly sending the same parameter presets.

Raises:: TypeError – If input arguments are not of the expected type.

property data_codes: set[uint8]: Returns the set of message event-codes that require online processing during runtime.

property error_codes: set[uint8]: Returns the set of message event-codes event codes that trigger runtime errors.

abstractmethod initialize_remote_assets()

Initializes the interface instance assets used in the remote microcontroller communication process.

This method is called during the initial setup sequence of the remote microcontroller communication process, before the PC-microcontroller communication cycle.

Return type:: None

Notes

This method should instantiate all interface assets that do not support pickling, such as PrecisionTimer or SharedMemoryArray instances. All assets initialized by this method must be destroyed by the terminate_remote_assets() method.

property module_id: uint8: Returns the id-code of the specific module instance managed by this interface instance.

property module_type: uint8: Returns the id-code of the type (family) of modules managed by this interface instance.

abstractmethod process_received_data(message)

Processes the input message.

This method is called during the communication cycle’s runtime when the interface instance receives a message from the microcontroller that uses an event code provided at class initialization as ‘data_codes’ argument.

Notes

This method should implement the custom online data-processing logic associated with each message whose event code is specified in the ‘data_codes’ argument.

All incoming message data is automatically cached (saved) to disk at runtime, so this method should NOT be used for data saving purposes.

The data processing logic implemented via this method should be optimized for runtime speed, as processing the data hogs the communication process, reducing its throughput.

Parameters:: message (ModuleData | ModuleState) – The ModuleState or ModuleData instance that stores the message data received from the interfaced hardware module instance.
Return type:: None

reset_command_queue()

Instructs the microcontroller to clear the managed hardware module’s command queue.

Return type:: None

send_command(command, noblock, repetition_delay=np.uint32(0))

Packages the input command data into the appropriate message structure and sends it to the managed hardware module.

Notes

This method caches up to 32 unique command messages in the instance-specific LRU cache to speed up sending previously created command messages.

Parameters:

command (uint8) – The id-code of the command to execute.
noblock (bool) – Determines whether the microcontroller managing the hardware module is allowed to concurrently execute other commands while executing the requested command.
repetition_delay (uint32, default: np.uint32(0)) – The time, in microseconds, to wait before repeating the command. If set to 0, the command is only executed once.

Return type:

None

send_parameters(parameter_data)

Packages the input parameter tuple into the appropriate message structure and sends it to the managed hardware module.

Notes

This method caches up to 16 unique parameter messages in the instance-specific LRU cache to speed up sending previously created command messages.

Parameters:: parameter_data (tuple[unsignedinteger[Any] | signedinteger[Any] | bool | floating[Any], ...]) – A tuple that contains the values for the PC-addressable parameters of the target hardware module. Note, the parameters must appear in the same order and use the same data-types as the module’s parameter structure on the microcontroller.
Return type:: None

set_input_queue(input_queue)

Overwrites the ‘_input_queue’ instance attribute with the reference to the provided Queue object.

This service method is used during the MicroControllerInterface initialization to finalize the instance’s configuration and should not be called directly by end users.

Return type:: None

abstractmethod terminate_remote_assets()

Terminates the interface instance assets used in the remote microcontroller communication process.

This method is the opposite of the initialize_remote_assets() method. It is called as part of the remote communication process shutdown routine to ensure any resources claimed by the interface are properly released before the communication process terminates.

Return type:: None

property type_id: uint16: Returns the unique 16-bit unsigned integer value that results from combining the bits of the type-code and the id-code of the managed module instance.

ataraxis_communication_interface.microcontroller_interface.extract_logged_hardware_module_data(log_path, module_type_id, n_workers=-1)

Extracts the received message data for the requested hardware module instances from the .npz log file generated by a MicroControllerInterface instance during runtime.

This function reads the ‘.npz’ archive generated by the DataLogger’s assemble_log_archives() method for a MicroControllerInterface instance and extracts the data for all non-system messages transmitted by the requested hardware module instances from the microcontroller to the PC.

Notes

At this time, the function exclusively works with the data sent by the microcontroller to the PC.

The extracted data does not contain library-reserved events and messages. This includes all Kernel messages and module messages with event codes 0 through 50. The only exceptions to this rule are messages with event code 2, which report command completion. These messages are parsed in addition to custom messages sent by each hardware module.

If the target .npz archive contains fewer than 2000 messages, the processing is carried out sequentially regardless of the specified worker-count.

Parameters:

log_path (Path) – The path to the .npz archive file that stores the logged data generated by the MicroControllerInterface instance during runtime.
module_type_id (tuple[tuple[int, int], ...]) – A tuple of tuples, where each inner tuple stores the type and ID codes of a specific hardware module, whose data should be extracted from the archive, e.g.: ((3, 1), (4, 2)).
n_workers (int, default: -1) – The number of parallel worker processes (CPU cores) to use for processing. Setting this to a value below 1 uses all available CPU cores. Setting this to a value of 1 conducts the processing sequentially.

Return type:

tuple[ExtractedModuleData, ...]

Returns:

A tuple of ExtractedModuleData instances. Each instance stores all data extracted from the log archive for one specific hardware module instance.

Raises:

ValueError – If the target .npz archive does not exist.

ataraxis_communication_interface.microcontroller_interface.print_microcontroller_ids(baudrate=115200)

Prints all available serial ports and identifies which ones are connected to Arduino or Teensy microcontrollers running the ataraxis-micro-controller library.

Uses parallel processing to simultaneously query all ports for microcontroller identification.

Parameters:: baudrate (int, default: 115200) – The baudrate to use for communication during identification. Note, the same baudrate value is used to evaluate all available microcontrollers. The baudrate is only used by the microcontrollers that communicate via the UART serial interface and is ignored by microcontrollers that use the USB interface.
Return type:: None

MCP Server

Provides a Model Context Protocol (MCP) server for agentic interaction with the library.

This module exposes microcontroller discovery and MQTT broker connectivity checking functionality through the MCP protocol, enabling AI agents to programmatically interact with the library’s core features.

ataraxis_communication_interface.mcp_server.check_mqtt_broker(host='127.0.0.1', port=1883)

Checks whether an MQTT broker is reachable at the specified host and port.

Attempts to connect to the MQTT broker and reports the result. Use this tool to verify MQTT broker availability before running code that depends on MQTT communication.

Parameters:

host (str, default: '127.0.0.1') – The IP address or hostname of the MQTT broker.
port (int, default: 1883) – The socket port used by the MQTT broker.

Return type:

str

ataraxis_communication_interface.mcp_server.list_microcontrollers(baudrate=115200)

Discovers all available serial ports and identifies which ones are connected to Arduino or Teensy microcontrollers running the ataraxis-micro-controller library.

Uses parallel processing to simultaneously query all ports for microcontroller identification.

Parameters:: baudrate (int, default: 115200) – The baudrate to use for communication during identification. Note, the same baudrate value is used to evaluate all available microcontrollers. The baudrate is only used by microcontrollers that communicate via the UART serial interface and is ignored by microcontrollers that use the USB interface.
Return type:: str

ataraxis_communication_interface.mcp_server.run_mcp_server()

Starts the MCP server with stdio transport.

This function is intended to be used as a CLI entry point. It starts the MCP server using the stdio transport protocol, which is the recommended transport for Claude Desktop integration.

Return type:: None

ataraxis_communication_interface.mcp_server.run_server(transport='stdio')

Starts the MCP server with the specified transport.

Parameters:: transport (Literal['stdio', 'sse', 'streamable-http'], default: 'stdio') – The transport protocol to use. Supported values are ‘stdio’ for standard input/output communication (recommended for Claude Desktop integration), ‘sse’ for Server-Sent Events, and ‘streamable-http’ for HTTP-based communication.
Return type:: None