Communication

This module provides the SerialCommunication and MQTTCommunication classes that enable communication between the PC, Arduino / Teensy microcontrollers running Ataraxis software and the other infrastructure with MQTT bindings.

SerialCommunication supports the PC-MicroController communication over USB / UART interface, while MQTTCommunication supports the communication over the MQTT protocol (virtual / real TCP sockets). MQTTCommunication can be used to transmit data locally (within the same PC) or remotely via a network connection.

Additionally, this module exposes message and helper structures used to serialize and deserialize the transmitted data.

class ataraxis_communication_interface.communication.ControllerIdentification(transport_layer)

Bases: object

Identifies the connected microcontroller by communicating its unique byte id-code.

For the ID codes to be unique, they have to be manually assigned to the Kernel class of each concurrently used microcontroller.

This class initializes to nonsensical defaults and expects the SerialCommunication class that manages its lifetime to call update_message_data() method when necessary to parse valid incoming message data.

Parameters:: transport_layer (TransportLayer) – The reference to the TransportLayer class that is initialized and managed by the SerialCommunication class. This reference is used to read and parse the message data.

protocol_code: Stores the protocol code used by this type of messages.

message: Stores the serialized message payload.

controller_id: The unique ID of the microcontroller. This ID is hardcoded in the microcontroller firmware and helps track which AXMC firmware is running on the given controller.

__repr__()

Returns a string representation of the ControllerIdentification object.

Return type:: str

update_message_data()

Reads and parses the data stored in the reception buffer of the TransportLayer class, overwriting class attributes.

This method should be called by the SerialCommunication class whenever KernelData message is received and needs to be parsed (as indicated by the incoming message protocol). This method will then access the reception buffer and attempt to parse the data.

Return type:: None

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

Bases: object

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

Note, clearing the command queue does not terminate already executing commands, but it prevents recurrent commands from running again.

__post_init__()

Packs the data into the numpy array to optimize future transmission speed.

Return type:: None

__repr__()

Returns a string representation of the RepeatedModuleCommand object.

Return type:: str

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

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

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

protocol_code: uint8 = np.uint8(3): Stores the protocol code used by this type of messages.

return_code: uint8 = np.uint8(0): When this attribute is set to a value other than 0, the microcontroller will send this code back to the PC upon successfully receiving and decoding the command.

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

Bases: object

Instructs the Kernel to run the specified command exactly once.

Currently, the Kernel only supports blocking one-off commands.

__post_init__()

Packs the data into the numpy array to optimize future transmission speed.

Return type:: None

__repr__()

Returns a string representation of the KernelCommand object.

Return type:: str

command: uint8: The code of the command to execute. Valid command codes are in the range between 1 and 255.

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

protocol_code: uint8 = np.uint8(4): Stores the protocol code used by this type of messages.

return_code: uint8 = np.uint8(0): When this attribute is set to a value other than 0, the microcontroller will send this code back to the PC upon successfully receiving and decoding the command.

class ataraxis_communication_interface.communication.KernelData(transport_layer)

Bases: object

Communicates the event state-code of the Kernel and includes an additional data object.

This class initializes to nonsensical defaults and expects the SerialCommunication class that manages its lifetime to call update_message_data() method when necessary to parse valid incoming message data.

Parameters:: transport_layer (TransportLayer) – The reference to the TransportLayer class that is initialized and managed by the SerialCommunication class. This reference is used to read and parse the message data.

protocol_code: Stores the protocol code used by this type of messages.

message: Stores the serialized message payload.

command: The code of the command the Kernel was executing when it sent the message.

event: The code of the event that prompted sending the message.

data_object: The data object decoded from the received message. Note, data messages only support the objects whose prototypes are defined in the SerialPrototypes enumeration.

_transport_layer: Stores the reference to the TransportLayer class.

__repr__()

Returns a string representation of the KernelData object.

Return type:: str

update_message_data()

Reads and parses the data stored in the reception buffer of the TransportLayer class, overwriting class attributes.

This method should be called by the SerialCommunication class whenever KernelData message is received and needs to be parsed (as indicated by the incoming message protocol). This method will then access the reception buffer and attempt to parse the data.

Raises:: ValueError – If the prototype code transmitted with the message is not valid.
Return type:: None

class ataraxis_communication_interface.communication.KernelParameters(action_lock, ttl_lock, return_code=np.uint8(0))

Bases: object

Instructs the Kernel to update the microcontroller-wide parameters with the values included in the message.

These parameters are shared by the Kernel and all custom Modules, and the exact parameter layout is hardcoded. This is in contrast to Module parameters, that differ between module types.

__post_init__()

Packs the data into the numpy array to optimize future transmission speed.

Return type:: None

__repr__()

Returns a string representation of the KernelParameters object.

Return type:: str

action_lock: bool: Determines whether the controller allows non-ttl modules to change output pin states. When True, all hardware-connected pins are blocked from changing states. This has no effect on sensor and TTL pins.

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

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

protocol_code: uint8 = np.uint8(6): Stores the protocol code used by this type of messages.

return_code: uint8 = np.uint8(0): When this attribute is set to a value other than 0, the microcontroller will send this code back to the PC upon successfully receiving and decoding the command.

ttl_lock: bool: Same as action_lock, but specifically controls output TTL (Transistor-to-Transistor Logic) pin activity. This has no effect on sensor and non-ttl hardware-connected pins.

class ataraxis_communication_interface.communication.KernelState(transport_layer)

Bases: object

Communicates the event state-code of the Kernel.

This class initializes to nonsensical defaults and expects the SerialCommunication class that manages its lifetime to call update_message_data() method when necessary to parse valid incoming message data.

Parameters:: transport_layer (TransportLayer) – The reference to the TransportLayer class that is initialized and managed by the SerialCommunication class. This reference is used to read and parse the message data.

protocol_code: Stores the protocol code used by this type of messages.

message: Stores the serialized message payload.

command: The code of the command the Kernel was executing when it sent the message.

event: The code of the event that prompted sending the message.

__repr__()

Returns a string representation of the KernelState object.

Return type:: str

update_message_data()

Reads and parses the data stored in the reception buffer of the TransportLayer class, overwriting class attributes.

This method should be called by the SerialCommunication class whenever KernelData message is received and needs to be parsed (as indicated by the incoming message protocol). This method will then access the reception buffer and attempt to parse the data.

Return type:: None

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

Bases: object

Wraps an MQTT client and exposes methods for bidirectionally communicating with other clients connected to the same MQTT broker.

This class leverages MQTT protocol on Python side and to establish bidirectional communication between the Python process running this class and other MQTT clients. Primarily, the class is intended to be used together with SerialCommunication class to transfer data between microcontrollers and the rest of the infrastructure used during runtime. Usually, both communication classes will be managed by the same process (core) that handles the necessary transformations to bridge MQTT and Serial communication protocols used by this library. This class is not designed to be instantiated directly and should instead be used through the MicroControllerInterface class available through this library!

Notes

MQTT protocol requires a broker that facilitates communication, which this class does NOT provide. Make sure your infrastructure includes a working MQTT broker before using this class. See https://mqtt.org/ for more details.

Parameters:

ip (str, default: '127.0.0.1') – The IP address of the MQTT broker that facilitates the communication.
port (int, default: 1883) – The socket port used by the MQTT broker that facilitates the communication.
monitored_topics (None | tuple[str, ...], default: None) – The list of MQTT topics which the class instance should subscribe to and 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 the class should monitor for incoming messages sent by other MQTT clients.

_output_queue: A multithreading queue used to buffer incoming messages received from other MQTT clients before their data is requested via class methods.

_client: Stores the initialized mqtt client instance that carries out the communication.

__del__()

Ensures proper resource release when the class instance is garbage-collected.

Return type:: None

__repr__()

Returns a string representation of the MQTTCommunication object.

Return type:: str

_on_message(_client, _userdata, message)

The callback function used to receive data from MQTT broker.

When passed to the client, this function will be called each time a new message is received. This function will then record the message topic and payload and put them into the output_queue for the data to be consumed by external processes.

Parameters:

_client (Client) – The MQTT client that received the message. Currently not used.
_userdata (Any) – Custom user-defined data. Currently not used.
message (MQTTMessage) – The received MQTT message.

Return type:

None

connect()

Connects to the MQTT broker and subscribes to the requested input topics.

This method has to be called to initialize communication, both for incoming and outgoing messages. Any message sent to the MQTT broker from other clients before this method is called may not reach this class.

Return type:: None

Notes

If this class instance subscribes (listens) to any topics, it will start a perpetually active thread with a listener callback to monitor incoming traffic.

Raises:: RuntimeError – If the MQTT broker cannot be connected to 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 buffer queue.

Return type:: tuple[str, bytes | bytearray] | None
Returns:: A two-element tuple. The first element is a string that communicates the MQTT topic of the received message. The second element is the payload of the message, which is a bytes or bytearray object. If no buffered objects are stored in the queue (queue is empty), returns None.
Raises:: RuntimeError – If the instance is not connected to the MQTT broker.

property has_data: bool: Returns True if the instance received messages from other MQTT clients and can output received data via the get_dataq() method.

send_data(topic, payload=None)

Publishes the input payload to the specified MQTT topic.

This method should be used for sending data to MQTT via one of the input topics. This method does not verify the validity of the input topic or payload data.

Parameters:

topic (str) – The MQTT topic to publish the data to.
payload (str | bytes | bytearray | float | None, default: None) – The data to be published. When set to None, an empty message will be sent, which is often used as a boolean trigger.

Raises:

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

Return type:

None

class ataraxis_communication_interface.communication.ModuleData(transport_layer)

Bases: object

Communicates the event state-code of the sender Module and includes an additional data object.

This class initializes to nonsensical defaults and expects the SerialCommunication class that manages its lifetime to call update_message_data() method when necessary to parse valid incoming message data.

Parameters:: transport_layer (TransportLayer) – The reference to the TransportLayer class that is initialized and managed by the SerialCommunication class. This reference is used to read and parse the message data.

protocol_code: Stores the protocol code used by this type of messages.

message: Stores the serialized message payload.

module_type: The type (family) code of the module that sent the message.

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

command: The code of the command the module was executing when it sent the message.

event: The code of the event that prompted sending the message.

data_object: The data object decoded from the received message. Note, data messages only support the objects whose prototypes are defined in the SerialPrototypes enumeration.

_transport_layer: Stores the reference to the TransportLayer class.

__repr__()

Returns a string representation of the ModuleData object.

Return type:: str

update_message_data()

Reads and parses the data stored in the reception buffer of the TransportLayer class, overwriting class attributes.

This method should be called by the SerialCommunication class whenever ModuleData message is received and needs to be parsed (as indicated by the incoming message protocol). This method will then access the reception buffer and attempt to parse the data.

Raises:: ValueError – If the prototype code transmitted with the message is not valid.
Return type:: None

class ataraxis_communication_interface.communication.ModuleIdentification(transport_layer)

Bases: object

Identifies a hardware module instance by communicating its combined type + id 16-bit code.

It is expected that each hardware module instance will have a unique combination of type (family) code and instance (ID) code. The user assigns both type and ID codes at their discretion when writing the main .cpp file for each microcontroller.

This class initializes to nonsensical defaults and expects the SerialCommunication class that manages its lifetime to call update_message_data() method when necessary to parse valid incoming message data.

Parameters:: transport_layer (TransportLayer) – The reference to the TransportLayer class that is initialized and managed by the SerialCommunication class. This reference is used to read and parse the message data.

protocol_code: Stores the protocol code used by this type of messages.

message: Stores the serialized message payload.

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

__repr__()

Returns a string representation of the ModuleIdentification object.

Return type:: str

update_message_data()

Reads and parses the data stored in the reception buffer of the TransportLayer class, overwriting class attributes.

This method should be called by the SerialCommunication class whenever KernelData message is received and needs to be parsed (as indicated by the incoming message protocol). This method will then access the reception buffer and attempt to parse the data.

Return type:: None

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

Bases: object

Instructs the addressed Module to overwrite its custom parameters object with the included object data.

__post_init__()

Packs the data into the numpy array to optimize future transmission speed.

Return type:: None

__repr__()

Returns a string representation of the ModuleParameters object.

Return type:: str

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

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

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

parameter_data: tuple[signedinteger[Any] | unsignedinteger[Any] | floating[Any] | bool, ...]: A tuple of parameter values to send. Each value will be serialized into bytes and sequentially packed into the data object included with the message. Each parameter value has to use a scalar numpy type.

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

protocol_code: uint8 = np.uint8(5): Stores the protocol code used by this type of messages.

return_code: uint8 = np.uint8(0): When this attribute is set to a value other than 0, the microcontroller will send this code back to the PC upon successfully receiving and decoding the command.

class ataraxis_communication_interface.communication.ModuleState(transport_layer)

Bases: object

Communicates the event state-code of the sender Module.

This class initializes to nonsensical defaults and expects the SerialCommunication class that manages its lifetime to call update_message_data() method when necessary to parse valid incoming message data.

Parameters:: transport_layer (TransportLayer) – The reference to the TransportLayer class that is initialized and managed by the SerialCommunication class. This reference is used to read and parse the message data.

protocol_code: Stores the protocol code used by this type of messages.

message: Stores the serialized message payload.

module_type: The type (family) code of the module that sent the message.

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

command: The code of the command the module was executing when it sent the message.

event: The code of the event that prompted sending the message.

__repr__()

Returns a string representation of the ModuleState object.

Return type:: str

update_message_data()

Reads and parses the data stored in the reception buffer of the TransportLayer class, overwriting class attributes.

This method should be called by the SerialCommunication class whenever ModuleData message is received and needs to be parsed (as indicated by the incoming message protocol). This method will then access the reception buffer and attempt to parse the data.

Return type:: None

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 to run the specified command exactly once (non-recurrently).

__post_init__()

Packs the data into the numpy array to optimize future transmission speed.

Return type:: None

__repr__()

Returns a string representation of the OneOffModuleCommand object.

Return type:: str

command: uint8: The code of the command to execute. Valid command codes are in the range between 1 and 255.

module_id: uint8: The ID of the specific module 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 the command runs in blocking or non-blocking mode. If set to False, the controller will block in-place for any sensor- or time-waiting loops during command execution. Otherwise, the controller will run other commands while waiting for the block to complete.

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

protocol_code: uint8 = np.uint8(2): Stores the protocol code used by this type of messages.

return_code: uint8 = np.uint8(0): When this attribute is set to a value other than 0, the microcontroller will send this code back to the PC upon successfully receiving and decoding the command.

class ataraxis_communication_interface.communication.ReceptionCode(transport_layer)

Bases: object

Returns the reception_code originally received from the PC to indicate that the message with that code was received and parsed.

This class initializes to nonsensical defaults and expects the SerialCommunication class that manages its lifetime to call update_message_data() method when necessary to parse valid incoming message data.

Parameters:: transport_layer (TransportLayer) – The reference to the TransportLayer class that is initialized and managed by the SerialCommunication class. This reference is used to read and parse the message data.

protocol_code: Stores the protocol code used by this type of messages.

message: Stores the serialized message payload.

reception_code: The reception code originally sent as part of the outgoing Command or Parameters messages.

__repr__()

Returns a string representation of the ReceptionCode object.

Return type:: str

update_message_data()

Reads and parses the data stored in the reception buffer of the TransportLayer class, overwriting class attributes.

This method should be called by the SerialCommunication class whenever KernelData message is received and needs to be parsed (as indicated by the incoming message protocol). This method will then access the reception buffer and attempt to parse the data.

Return type:: None

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 to repeatedly (recurrently) run the specified command.

__post_init__()

Packs the data into the numpy array to optimize future transmission speed.

Return type:: None

__repr__()

Returns a string representation of the RepeatedModuleCommand object.

Return type:: str

command: uint8: The code of the command to execute. Valid command codes are in the range between 1 and 255.

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

module_id: uint8: The ID of the specific module 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 the command runs in blocking or non-blocking mode. If set to False, the controller will block in-place for any sensor- or time-waiting loops during command execution. Otherwise, the controller will run other commands while waiting for the block to complete.

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

protocol_code: uint8 = np.uint8(1): Stores the protocol code used by this type of messages.

return_code: uint8 = np.uint8(0): When this attribute is set to a value other than 0, the microcontroller will send this code back to the PC upon successfully receiving and decoding the command.

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

Bases: object

Wraps a TransportLayer class instance and exposes methods that allow communicating with a microcontroller running ataraxis-micro-controller library using the USB or UART protocol.

This class is built on top of the TransportLayer, designed to provide the microcontroller communication interface (API) for other Ataraxis libraries. This class is not designed to be instantiated directly and should instead be used through the MicroControllerInterface class available through this library!

Notes

This class is explicitly designed to use the same parameters as the Communication class used by the microcontroller. Do not modify this class unless you know what you are doing.

Due to the use of many non-pickleable classes, this class cannot be piped to a remote process and has to be initialized by the remote process directly.

This class is designed to integrate with DataLogger class available from the ataraxis_data_structures library. The DataLogger is used to write all incoming and outgoing messages to disk as serialized message payloads.

Parameters:

source_id (uint8) – The ID code to identify the source of the logged messages. This is used by the DataLogger to distinguish between log sources (classes that sent data to be logged) and, therefore, has to be unique for all Ataraxis classes that use DataLogger and are active at the same time.
microcontroller_serial_buffer_size (int) – The size, in bytes, of the buffer used by the target microcontroller’s Serial buffer. Usually, this information is available from the microcontroller’s manufacturer (UART / USB controller specification).
usb_port (str) – The name of the USB port to use for communication to, e.g.: ‘COM3’ or ‘/dev/ttyUSB0’. This has to be the port to which the target microcontroller is connected. Use the list_available_ports() function available from this library to get the list of discoverable serial port names.
logger_queue (Queue) – The multiprocessing Queue object exposed by the DataLogger class (via ‘input_queue’ property). This queue is used to buffer and pipe data to be logged to the logger cores.
baudrate (int, default: 115200) – The baudrate to use for the communication over the UART protocol. Should match the value used by the microcontrollers that only support UART protocol. This is ignored for microcontrollers that use the USB protocol.
test_mode (bool, default: False) – This parameter is only used during testing. When True, it initializes the underlying TransportLayer class in the test configuration. Make sure this is set to False during production runtime.

_transport_layer: The TransportLayer instance that handles the communication.

_module_data: Received ModuleData messages are unpacked into this structure.

_kernel_data: Received KernelData messages are unpacked into this structure.

_module_state: Received ModuleState messages are unpacked into this structure.

_kernel_state: Received KernelState messages are unpacked into this structure.

_controller_identification: Received ControllerIdentification messages are unpacked into this structure.

_module_identification: Received ModuleIdentification messages are unpacked into this structure.

_reception_code: Received ReceptionCode messages are unpacked into this structure.

_timestamp_timer: The PrecisionTimer instance used to stamp incoming and outgoing data as it is logged.

_source_id: Stores the unique integer-code that identifies the class instance in data logs.

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

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

__repr__()

Returns a string representation of the SerialCommunication object.

Return type:: str

_log_data(timestamp, data)

Packages and sends the input data to teh DataLogger instance that writes it to disk.

Parameters:

timestamp (int) – The value of the timestamp timer ‘elapsed’ property that communicates the number of elapsed microseconds relative to the ‘onset’ timestamp.
data (ndarray[Any, dtype[uint8]]) – The byte-serialized message payload that was sent or received.

Return type:

None

receive_message()

Receives the incoming message from the connected microcontroller and parses into the appropriate structure.

This method uses the protocol code, assumed to be stored in the first variable of each received payload, to determine how to parse the data. It then parses into a precreated message structure stored in class attributes.

Notes

To optimize overall runtime speed, this class creates message structures for all supported messages at initialization and overwrites the appropriate message attribute with the data extracted from each received message payload. This method than returns the reference to the overwritten class attribute. Therefore, it is advised to copy or finish working with the structure returned by this method before receiving another message. Otherwise, it is possible that the received message will be used to overwrite the data of the previously referenced structure, leading to the loss of unprocessed / unsaved data.

Return type:: ModuleData | ModuleState | KernelData | KernelState | ControllerIdentification | ModuleIdentification | ReceptionCode | None
Returns:: A reference the parsed message structure instance stored in class attributes, or None, if no message was received. Note, None return does not indicate an error, but rather indicates that the microcontroller did not send any data.
Raises:: ValueError – If the received message uses an invalid (unrecognized) message protocol code.

send_message(message)

Serializes the input command or parameters message and sends it to the connected microcontroller.

This method relies on every valid outgoing message structure exposing a packed_data attribute, that contains the serialized payload data to be sent. Functionally, this method is a wrapper around the TransportLayer’s write_data() and send_data() methods.

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

class ataraxis_communication_interface.communication.SerialProtocols(value, names=<not given>, *values, module=None, qualname=None, type=None, start=1, boundary=None)

Bases: IntEnum

Stores the protocol codes used in data transmission between the PC and the microcontroller over the serial port.

Each sent and received message starts with the specific protocol code from this enumeration that instructs the receiver on how to process the rest of the data payload. The codes available through this class have to match the contents of the kProtocols Enumeration available from the ataraxis-micro-controller library (axmc_communication_assets namespace).

Notes

The values available through this enumeration should be read through their ‘as_uint8’ property to enforce the type expected by other classes from ths library.

CONTROLLER_IDENTIFICATION = 12: Protocol used to identify the controller connected to a particular USB port. This service protocol is used by the controller that receives the ‘IdentifyController’ Kernel-addressed command and replies with it’s ID code. This protocol is automatically used by the MicrocontrollerInterface class during initialization and should not be used manually.

DEQUEUE_MODULE_COMMAND = 3: Protocol for sending Module-addressed commands that remove all queued commands (including recurrent commands).

KERNEL_COMMAND = 4: Protocol for sending Kernel-addressed commands. All Kernel commands are always non-repeatable (one-shot).

KERNEL_DATA = 8: Protocol for receiving Kernel-sent data or error messages that include an arbitrary data object in addition to event state-code.

KERNEL_PARAMETERS = 6: Protocol for sending Kernel-addressed parameters. The parameters transmitted via these messages will be used to overwrite the global parameters shared by the Kernel and all Modules of the microcontroller (global runtime parameters).

KERNEL_STATE = 10: Protocol for receiving Kernel-sent data or error messages that do not include additional data objects.

MODULE_DATA = 7: Protocol for receiving Module-sent data or error messages that include an arbitrary data object in addition to event state-code.

MODULE_IDENTIFICATION = 13: Protocol used to identify all hardware module instances managed by the connected microcontroller. This service protocol is used by the controller that receives the ‘IdentifyModules’ Kernel-addressed command and sequentially transmits the combined type_id uint16 code for each managed module instance. This protocol is automatically used by the MicrocontrollerInterface class during initialization and should not be used manually.

MODULE_PARAMETERS = 5: Protocol for sending Module-addressed parameters. This relies on transmitting arbitrary sized parameter objects likely to be unique for each module type (family).

MODULE_STATE = 9: Protocol for receiving Module-sent data or error messages that do not include additional data objects.

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

RECEPTION_CODE = 11: Protocol used to ensure that the microcontroller has received a previously sent command or parameter message. Specifically, when an outgoing message includes a reception_code, this code is transmitted back to the PC using this service protocol to acknowledge message reception. Currently, this protocol is only intended for testing purposes, as at this time the Communication class does not explicitly ensure message delivery.

REPEATED_MODULE_COMMAND = 1: Protocol for sending Module-addressed commands that should be repeated (executed recurrently).

UNDEFINED = 0: Not a valid protocol code. This is used to initialize the Communication class of the microcontroller.

as_uint8()

Convert the enum value to numpy.uint8 type.

Returns:: The enum value as a numpy unsigned 8-bit integer.
Return type:: np.uint8

class ataraxis_communication_interface.communication.SerialPrototypes(value, names=<not given>, *values, module=None, qualname=None, type=None, start=1, boundary=None)

Bases: IntEnum

Stores the prototype codes used in data transmission between the PC and the microcontroller over the serial port.

Prototype codes are used by Data messages (Kernel and Module) to inform the receiver about the structure (prototype) that can be used to deserialize the included data object. Transmitting these codes with the message ensures that the receiver has the necessary information to decode the data without doing any additional processing. In turn, this allows optimizing the reception procedure to efficiently decode the data objects.

Notes

While the use of 8-bit (byte) value limits the number of mapped prototypes to 255 (256 if 0 is made a valid value), this number should be enough to support many unique runtime configurations.

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()

Converts the enum value to numpy.uint8 type.

Return type:: uint8
Returns:: The enum value as a numpy unsigned 8-bit integer.

get_prototype()

Returns the prototype object associated with this prototype enum value.

The prototype object returned by this method can be passed to the reading method of the TransportLayer class to deserialize the received data object. This should be automatically done by the SerialCommunication class that uses this enum class.

Return type:: Union[bool, uint8, int8, uint16, int16, uint32, int32, uint64, int64, float32, float64, ndarray[Any, dtype[bool]], ndarray[Any, dtype[uint8]], ndarray[Any, dtype[int8]], ndarray[Any, dtype[uint16]], ndarray[Any, dtype[int16]], ndarray[Any, dtype[uint32]], ndarray[Any, dtype[int32]], ndarray[Any, dtype[uint64]], ndarray[Any, dtype[int64]], ndarray[Any, dtype[float32]], ndarray[Any, dtype[float64]]]
Returns:: The prototype object that is either a numpy scalar or shallow array type.

classmethod get_prototype_for_code(code)

Returns the prototype object associated with the input prototype code.

The prototype object returned by this method can be passed to the reading method of the TransportLayer class to deserialize the received data object. This should be automatically done by the SerialCommunication class that uses this enum.

Parameters:: code (uint8) – The prototype byte-code to retrieve the prototype for.
Return type:: Union[bool, uint8, int8, uint16, int16, uint32, int32, uint64, int64, float32, float64, ndarray[Any, dtype[bool]], ndarray[Any, dtype[uint8]], ndarray[Any, dtype[int8]], ndarray[Any, dtype[uint16]], ndarray[Any, dtype[int16]], ndarray[Any, dtype[uint32]], ndarray[Any, dtype[int32]], ndarray[Any, dtype[uint64]], ndarray[Any, dtype[int64]], ndarray[Any, dtype[float32]], ndarray[Any, dtype[float64]], 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.

MicroController Interface

This module provides the ModuleInterface and MicroControllerInterface classes. They aggregate the methods to enable Python PC clients to bidirectionally interface with custom hardware modules managed by an Arduino or Teensy microcontroller. Additionally, these classes contain the bindings to connect to the microcontrollers via the MQTT protocol, facilitating indirect communication with local and remote processes over MQTT broker.

Each microcontroller hardware module that manages physical hardware should be matched to a specialized interface derived from the base ModuleInterface class. Similarly, for each concurrently active microcontroller, there has to be a specific MicroControllerInterface instance that manages the ModuleInterface instances for the modules of that controller.

class ataraxis_communication_interface.microcontroller_interface.MicroControllerInterface(controller_id, microcontroller_serial_buffer_size, microcontroller_usb_port, data_logger, module_interfaces, baudrate=115200, mqtt_broker_ip='127.0.0.1', mqtt_broker_port=1883)

Bases: object

Interfaces with an Arduino or Teensy microcontroller running ataraxis-micro-controller library.

This class contains the logic that sets up a remote daemon process with SerialCommunication, MQTTCommunication, and DataLogger bindings to facilitate bidirectional communication and data logging between the microcontroller and concurrently active local (same PC) and remote (network) processes. Additionally, it exposes methods that send runtime parameters and commands to the Kernel and Module classes running on the connected microcontroller.

Notes

An instance of this class has to be instantiated for each microcontroller active at the same time. The communication will not be started until the start() method of the class instance is called.

This class uses SharedMemoryArray to control the runtime of the remote process, which makes it impossible to have more than one instance of this class with the same controller_id at a time.

Initializing MicroControllerInterface also completes the configuration of all ModuleInterface instances passed to the class constructor. It is essential to initialize both the interfaces and the MicroControllerInterface to have access to the full range of functionality provided by each ModuleInterface class.

Parameters:

controller_id (uint8) – The unique identifier code of the managed microcontroller. This information is hardcoded via the ataraxis-micro-controller (AXMC) library running on the microcontroller, and this class ensures that the code used by the connected microcontroller matches this argument when the connection is established. Critically, this code is also used as the source_id for the data sent from this class to the DataLogger. Therefore, it is important for this code to be unique across ALL concurrently active Ataraxis data producers, such as: microcontrollers and video systems. Valid codes are values between 1 and 255.
microcontroller_serial_buffer_size (int) – The size, in bytes, of the microcontroller’s serial interface (UART or USB) buffer. This size is used to calculate the maximum size of transmitted and received message payloads. This information is usually available from the microcontroller’s vendor.
microcontroller_usb_port (str) – The serial USB port to which the microcontroller is connected. This information is used to set up the bidirectional serial communication with the controller. You can use list_available_ports() function from ataraxis-transport-layer-pc library to discover addressable USB ports to pass to this argument. The function is also accessible through the CLI command: ‘axtl-ports’.
data_logger (DataLogger) – An initialized DataLogger instance used to log the data produced by this Interface instance. The DataLogger itself is NOT managed by this instance and will need to be activated separately. This instance only extracts the necessary information to pipe the data to the logger.
module_interfaces (tuple[ModuleInterface, ...]) – A tuple of classes that inherit from the ModuleInterface class that interface with specific hardware module instances managed by the connected microcontroller.
baudrate (int, default: 115200) – The baudrate at which the serial communication should be established. This argument is ignored for microcontrollers that use the USB communication protocol, such as most Teensy boards. The correct baudrate for microcontrollers using the UART communication protocol depends on the clock speed of the microcontroller’s CPU and the supported UART revision. Setting this to an unsupported value for microcontrollers that use UART will result in communication errors.
mqtt_broker_ip (str, default: '127.0.0.1') – The ip address of the MQTT broker used for MQTT communication. Typically, this would be a ‘virtual’ ip-address of the locally running MQTT broker, but the class can carry out cross-machine communication if necessary. MQTT communication will only be initialized if any of the input modules requires this functionality.
mqtt_broker_port (int, default: 1883) – The TCP port of the MQTT broker used for MQTT communication. This is used in conjunction with the mqtt_broker_ip argument to connect to the MQTT broker.

Raises:

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

_controller_id: Stores the id byte-code of the managed microcontroller.

_usb_port: Stores the USB port to which the controller is connected.

_baudrate: Stores the baudrate to use for serial communication with the controller.

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

_mqtt_ip: Stores the IP address of the MQTT broker used for MQTT communication.

_mqtt_port: Stores the port number of the MQTT broker used for MQTT communication.

_mp_manager: Stores the multiprocessing Manager used to initialize and manage input and output Queue objects.

_input_queue: Stores 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 runtime of the remote communication process.

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

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

_reset_command: Stores the pre-packaged Kernel-addressed command that resets the microcontroller’s hardware and software.

_disable_locks: Stores the pre-packaged Kernel parameters configuration that disables all pin locks. This allows writing to all microcontroller pins.

_enable_locks: Stores the pre-packaged Kernel parameters configuration that enables all pin locks. This prevents every Module managed by the Kernel from writing to any of the microcontroller pins.

_started: Tracks whether the communication process has been started. This is used to prevent calling the start() and stop() methods multiple times.

_start_mqtt_client: Determines whether to connect to MQTT broker during the main runtime cycle.

__del__()

Ensures that all class resources are properly released when the class instance is garbage-collected.

Return type:: None

__repr__()

Returns a string representation of the class instance.

Return type:: str

static _runtime_cycle(controller_id, module_interfaces, input_queue, logger_queue, terminator_array, usb_port, baudrate, microcontroller_buffer_size, mqtt_ip, mqtt_port, start_mqtt_client)

This method aggregates the communication runtime logic and is used as the target for the communication process.

This method is designed to run in a remote Process. It encapsulates the steps for sending and receiving the data from the connected microcontroller. Primarily, the method routes the data between the microcontroller, the multiprocessing queues (inpout and output) managed by the Interface instance, and the MQTT broker. Additionally, it manages data logging by interfacing with the DataLogger class via the logger_queue.

Notes

Each managed ModuleInterface may contain custom logic for processing and routing the data. This method calls the custom logic bindings for each interface on a need-based method.

Parameters:

controller_id (uint8) – The byte-code identifier of the target microcontroller. This is used to ensure that the instance interfaces with the correct controller and to source-stamp logged data.
module_interfaces (tuple[ModuleInterface, ...]) – A tuple that stores ModuleInterface classes managed by this MicroControllerInterface instance.
input_queue (Queue) – The multiprocessing queue used to issue commands to the microcontroller.
logger_queue (Queue) – The queue exposed by the DataLogger class that is used to buffer and pipe received and outgoing messages to be logged (saved) to disk.
terminator_array (SharedMemoryArray) – The shared memory array used to control the communication process runtime.
usb_port (str) – The serial port to which the target microcontroller is connected.
baudrate (int) – The communication baudrate to use. This option is ignored for controllers that use USB interface, but is essential for controllers that use the UART interface.
microcontroller_buffer_size (int) – The size of the microcontroller’s serial buffer. This is used to determine the maximum size of the incoming and outgoing message payloads.
mqtt_ip (str) – The IP-address of the MQTT broker to use for communication with other MQTT processes.
mqtt_port (int) – The port number of the MQTT broker to use for communication with other MQTT processes.
start_mqtt_client (bool) – Determines whether to start the MQTT client used by MQTTCommunication instance.

Return type:

None

_watchdog()

This method is used by the watchdog thread to ensure the communication process is alive during runtime.

This method will raise a RuntimeError if it detects that a process has prematurely shut down. It will verify process states every ~20 ms and will release the GIL between checking the states.

Return type:: None

Notes

If the method detects that the communication process is not alive, it will carry out the necessary resource cleanup before raising the error and terminating the class runtime.

lock_controller()

Configures connected MicroController parameters to prevent all modules from writing to any output pin.

Return type:: None

reset_controller()

Resets the connected MicroController to use default hardware and software parameters.

Return type:: None

send_message(message)

Sends the input message to the microcontroller managed by this interface instance.

This is the primary interface for communicating with the Microcontroller. It allows sending all valid outgoing message structures to the Microcontroller for further processing. This is the only interface explicitly designed to communicate both with hardware modules and the Kernel class that manages the runtime of the microcontroller.

Return type:: None

Notes

During initialization, the MicroControllerInterface provides each managed ModuleInterface with the reference to the input_queue object. Each ModuleInterface can use its own _input_queue attribute to send the data to the communication process, eliminating the need for the data to go through this method. If you are developing a custom interface, you have the option for using either queue interface for submitting data to be sent to the microcontroller.

Raises:: TypeError – If the input message is not a valid outgoing message structure.

start()

Initializes the communication with the target microcontroller and the MQTT broker.

The MicroControllerInterface class will not be able to carry out any communications until this method is called. After this method finishes its runtime, a watchdog thread is used to monitor the status of the process until stop() method is called, notifying the user if the process terminates prematurely.

Return type:: None

Notes

If send_message() was called before calling start(), all queued messages will be transmitted in one step. Multiple commands addressed to the same module sent in this fashion will likely interfere with each-other.

As part of this method runtime, the interface will verify the target microcontroller’s configuration to ensure compatibility.

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

stop()

Shuts down the communication process and frees all reserved resources.

Return type:: None

unlock_controller()

Configures connected MicroController parameters to allow all modules to write to any output pin.

Return type:: None

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

Bases: object

The base class from which all custom ModuleInterface classes should inherit.

Inheriting from this class grants all subclasses the static API that the MicroControllerInterface class uses to interface with specific module interfaces. It is essential that all abstract methods defined in this class are implemented for each custom module interface implementation that subclasses this class.

Notes

Similar to the ataraxis-micro-controller (AXMC) library, the interface class has to be implemented separately for each custom module. The (base) class exposes the static API used by the MicroControllerInterface class to integrate each custom interface implementation with the general communication runtime cycle. To make this integration possible, this class defines some abstract (pure virtual) methods that developers have to implement for their interfaces. Follow the implementation guidelines in the docstrings of each abstract method and check the examples for further guidelines on how to implement each abstract method.

When inheriting from this class, remember to call the parent’s init method in the child class init method by using ‘super().__init__()’! If this is not done, the MicroControllerInterface class will likely not be able to properly interact with your custom interface class!

All data received from or sent to the microcontroller is automatically logged as byte-serialized numpy arrays. If you do not need any additional processing steps, such as sending or receiving data over MQTT, do not enable any custom processing flags when initializing this superclass!

In addition to interfacing with the module, the class also contains methods to parse logged module data. It is expected that these modules will be used immediately after runtime to parse raw logged data and transform it into the desired format for further processing and analysis.

Some attributes of this class are assigned by the managing MicroControllerInterface class at its initialization. Therefore, to be fully functional, each ModuleInterface class has to be bound to an initialized MicroControllerInterface instance.

Parameters:

module_type (uint8) – The id-code that describes the broad type (family) of custom hardware modules managed by this interface class. This value has to match the code used by the custom module implementation on the microcontroller. Valid byte-codes range from 1 to 255.
module_id (uint8) – The code that identifies the specific custom hardware module instance managed by the interface class instance. This is used to identify unique modules in a broader module family, such as different rotary encoders if more than one is used at the same time. Valid byte-codes range from 1 to 255.
mqtt_communication (bool) – Determines whether this interface needs to communicate with MQTT. If your implementation of the process_received_data() method requires sending data to MQTT via MQTTCommunication, set this flag to True when implementing the class. Similarly, if your interface is configured to receive commands from MQTT, set this flag to True.
error_codes (set[uint8] | None, default: None) – A set that stores the numpy uint8 (byte) codes used by the interface module to communicate runtime errors. This set will be used during runtime to identify and raise error messages in response to managed module sending error State and Data messages to the PC. Note, status codes 0 through 50 are reserved for internal library use and should NOT be used as part of this set or custom hardware module class design. If the class does not produce runtime errors, set to None.
data_codes (set[uint8] | None, default: None) – A set that stores the numpy uint8 (byte) codes used by the interface module to communicate states and data that needs additional processing. All incoming messages from the module are automatically logged to disk during communication runtime. Messages with event-codes from this set would also be passed to the process_received_data() method for additional processing. If the class does not require additional processing for any incoming data, set to None.
mqtt_command_topics (set[str] | None, default: None) – A set of MQTT topics used by other MQTT clients to send commands to the module accessible through this interface instance. If the interface does not receive commands from mqtt, set this to None. The MicroControllerInterface set will use the set to initialize the MQTTCommunication class instance to monitor the requested topics and will use the use parse_mqtt_command() method to convert MQTT messages to module-addressed command structures.

_module_type: Stores the type (family) of the interfaced module.

_module_id: Stores the specific module instance ID within the broader type (family).

_type_id: Stores the type and id combined into a single uint16 value. This value should be unique for all possible type-id pairs and is used to ensure that each used module instance has a unique ID-type combination.

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

_mqtt_command_topics: Stores MQTT topics to monitor for incoming commands.

_error_codes: Stores all expected error-codes as a set.

_mqtt_communication: Determines whether this interface needs to communicate with MQTT.

_log_directory: Stores the path to the directory where the MicroControllerInterface that manages this class logs all received and transmitted messages related to this interface. The value for this attribute is assigned automatically by the managing MicroControllerInterface class during its initialization.

_microcontroller_id: Stores the unique ID byte-code of the microcontroller that controls the hardware module interfaced by this class instance. The value for this attribute is assigned automatically by the managing MicroControllerInterface class during its initialization.

_input_queue: Stores the multiprocessing queue that enables sending the data to the microcontroller via the managing MicroControllerInterface class. Putting messages into this queue is equivalent to submitting them to the send_data() method exposed by the managing MicroControllerInterface class. The value for this attribute is assigned automatically by the managing MicroControllerInterface class during its initialization.

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

__repr__()

Returns the string representation of the ModuleInterface instance.

Return type:: str

property data_codes: set[uint8]: Returns the set of message event-codes that are processed during runtime, in addition to logging them to disk.

property error_codes: set[uint8]: Returns the set of error event-codes used by the module instance.

extract_logged_data()

Extracts the data sent by the hardware module instance running on the microcontroller from the .npz log file generated during ModuleInterface runtime.

This method reads the compressed ‘.npz’ archives generated by the MicroControllerInterface class that works with this ModuleInterface during runtime and extracts all custom event-codes and data objects transmitted by the interfaced module instance from the microcontroller.

Notes

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

This method should be used as a convenience abstraction for the inner workings of the DataLogger class. For each ModuleInterface, it will decode and return the logged runtime data sent to the PC by the specific hardware module instance controlled by the interface. You need to manually implement further data processing steps as necessary for your specific use case and module implementation.

Return type:

dict[Any, list[dict[str, uint64 | Any]]]

Returns:

A dictionary that uses numpy uint8 event codes as keys and stores lists of dictionaries under each key. Each inner dictionary contains three elements. First, an uint64 timestamp, representing the number of microseconds since the UTC epoch onset. Second, the data object, transmitted with the message (or None, for state-only events). Third, the uint8 code of the command that the module was executing when it sent the message to the PC.

Raises:

RuntimeError – If this method is called before the ModuleInterface is used to initialize a MicroControllerInterface class.
ValueError – If the input path is not valid or does not point to an existing .npz archive.

abstract initialize_remote_assets()

Initializes custom interface assets to be used in the remote process.

This method is called at the beginning of the communication runtime by the managing MicroControllerInterface. Use this method to create and initialize any assets that cannot be pickled (to be transferred into the remote process).

Return type:: None

Notes

This method is called early in the preparation phase of the communication runtime, before any communication is actually carried out. Use this method to initialize unpickable assets, such as PrecisionTimer instances or connect to shared resources, such as SharedMemory buffers.

All assets managed or created by this method should be stored in the ModuleInterface instance’s attributes.

property module_id: uint8: Returns the code that identifies the specific Module instance managed by the Interface class instance.

property module_type: uint8: Returns the id-code that describes the broad type (family) of Modules managed by this interface class.

property mqtt_command_topics: set[str]: Returns the set of MQTT topics this instance monitors for incoming MQTT commands.

property mqtt_communication: bool: Returns True if the class instance is configured to communicate with MQTT during runtime.

abstract parse_mqtt_command(topic, payload)

Packages and returns a ModuleCommand message to send to the microcontroller, based on the input MQTT command message topic and payload.

This method is called by the MicroControllerInterface when other MQTT clients send command messages to one of the topics monitored by this ModuleInterface instance. This method resolves, packages, and returns the appropriate ModuleCommand message structure, based on the input message topic and payload.

Notes

This method is called only if ‘mqtt_command_topics’ class argument was used to set the monitored topics during class initialization. This method will never receive a message with a topic that is not inside the ‘mqtt_command_topics’ set.

Use this method to translate incoming MQTT messages into the appropriate command messages for the hardware module. While we currently do not explicitly support translating MQTT messages into parameter messages, this can be added in the future if enough interest is shown.

See the /examples folder included with the library for examples on how to implement this method.

Parameters:

topic (str) – The MQTT topic to which the other MQTT client sent the module-addressed command.
payload (bytes | bytearray) – The payload of the message.

Return type:

OneOffModuleCommand | RepeatedModuleCommand | DequeueModuleCommand | None

Returns:

A OneOffModuleCommand, RepeatedModuleCommand, or DequeueModuleCommand instance that stores the message to be sent to the microcontroller. None, if the class instance is not configured to receive commands from MQTT.

abstract process_received_data(message)

Processes the incoming message and executes user-defined logic.

This method is called by the MicroControllerInterface when the ModuleInterface instance receives a message from the microcontroller that uses an event code provided at class initialization as ‘data_codes’ argument. This method should be used to implement custom processing logic for the incoming data.

Notes

Primarily, this method is intended to execute custom data transmission logic. For example, it can be used to send a message over MQTT (via a custom implementation or our MQTTCommunication class), put the data into a multithreading or multiprocessing queue, or use it to set a SharedMemory object. Use this method as a gateway to inject custom data handling into the communication runtime.

Keep the logic inside this method as minimal as possible. All data from the microcontroller goes through the same communication process, so it helps to minimize real time processing of the data, as it allows for better communication throughput. Treat this method like you would treat a microcontroller hardware interrupt function.

If your communication / processing assets cannot be pickled (to be transferred into the remote process used for communication), implement their initialization via the initialize_remote_assets() method.

See the /examples folder included with the library for examples on how to implement this method.

Parameters:: message (ModuleData | ModuleState) – The ModuleState or ModuleData object that stores the message received from the module instance running on the microcontroller.
Return type:: None

reset_command_queue()

Instructs the microcontroller to clear all queued commands for the specific module instance managed by this ModuleInterface.

If the ModuleInterface has not been used to initialize the MicroControllerInterface, raises a RuntimeError.

Return type:: None

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