Bota Control Toolkit Python

Definitions

BotaControlBlock

BotaControlBlock is defined as a parent class from which all control blocks inherit.

class BotaControlBlock(config)

Abstract base class providing the interface for all control processing blocks.

Constructor

__init__(config)

Unified constructor that dispatches based on the type of config.

Parameters:

config –

Configuration source — one of:

• str — path to a JSON configuration file (preferred)

• dict — a dictionary containing the configuration data

• BotaControlConfig — a configuration object populated elsewhere

Raises:

TypeError – If config is not one of the accepted types.

Runtime Update Methods

Note

All update methods operate on NumPy arrays (numpy.ndarray, dtype float64). Output arrays are written in-place — allocate them once before your control loop and reuse them every iteration to avoid unnecessary allocations.

update(input_signal, output_signal)

Reads input data and performs processing, writing the result into output_signal in-place.

Parameters:

• input_signal (numpy.ndarray) – Input signal data. Shape (N,), dtype float64.

• output_signal (numpy.ndarray) – Pre-allocated output array. Overwritten in-place with the processed result. Shape (M,), dtype float64.

Returns:

A return code indicating the status of the update operation.

Return type:

BotaControlReturnCode

update_inplace(signal)

Reads data and performs processing in place, overwriting signal with the result.

Parameters:

signal (numpy.ndarray) – Signal data to be processed in place. Shape (N,), dtype float64.

Returns:

A return code indicating the status of the update operation.

Return type:

BotaControlReturnCode

update_observer(measurement, state_estimate)

Observer update: reads measurement data and writes the state estimate in-place.

Parameters:

• measurement (numpy.ndarray) – Measurement data. Shape (N,), dtype float64.

• state_estimate (numpy.ndarray) – Pre-allocated output array. Overwritten in-place with the state estimate. Shape (M,), dtype float64.

Returns:

A return code indicating the status of the update operation.

Return type:

BotaControlReturnCode

update_controller(reference, state_estimate, control_action)

Controller update: reads reference and state estimate, writes the control action in-place.

Parameters:

• reference (numpy.ndarray) – Reference / setpoint data. Shape (N,), dtype float64.

• state_estimate (numpy.ndarray) – Current state estimate from the observer. Shape (M,), dtype float64.

• control_action (numpy.ndarray) – Pre-allocated output array. Overwritten in-place with the control action. Shape (P,), dtype float64.

Returns:

A return code indicating the status of the update operation.

Return type:

BotaControlReturnCode

In-line Configuration Methods

Specific implementations of control blocks may include additional in-line configuration methods that allow users to modify block parameters at configuration or runtime without needing to re-instantiate the block. These methods are specific to each block and are not part of the base class interface.

---

BotaControlReturnCode

class BotaControlReturnCode

Enum class representing the status of a control block update operation.

OK

Stale

Degraded

DeadlineMissed

NonFatalError

FatalError

---

Pipeline Wiring

Here’s how to connect multiple blocks to create a processing pipeline:

###############################################
# EXAMPLE OF A BUNDLED CONTROL PIPELINE USING THE BOTA CONTROL BLOCKS #
###############################################

import time
import numpy as np

# Define signals as pre-allocated numpy arrays (reused every iteration)
signal_1 = np.zeros(N, dtype=np.float64)
signal_2 = np.zeros(M, dtype=np.float64)
signal_3 = np.zeros(P, dtype=np.float64)

# Path to the JSON configuration file for each block
config_1 = "path/to/config_1.json"
config_2 = "path/to/config_2.json"
config_3 = "path/to/config_3.json"

# Instantiation & configuration at initialisation time
block_1 = Block1(config_1)
block_2 = Block2(config_2)
block_3 = Block3(config_3)

# Block in-line static configuration
block_1.set_parameter(value_1)
block_2.set_parameter(value_2)
block_3.set_parameter(value_3)

# Set the desired processing rate
rate_hz = 500.0  # 500 Hz
dt = 1.0 / rate_hz

running = True

# Loop reactor function to handle return codes and control flow
def loop_reactor(code: BotaControlReturnCode) -> None:
    global running
    match code:
        case BotaControlReturnCode.OK:
            pass  # Normal operation, continue processing
        # ... other cases ...
        case BotaControlReturnCode.FatalError:
            # Handle fatal errors (e.g. log the error, shut down safely, etc.)
            running = False  # Stop the loop on fatal error

while running:
    start_time = time.perf_counter()

    # Update signal_1 with new data (e.g. from sensors, reference generators, etc.)
    # Write directly into the pre-allocated array to avoid re-allocation
    signal_1[:] = get_new_data_for_signal_1()

    # Process data through the pipeline
    loop_reactor(block_1.update(signal_1, signal_2))
    # => signal_2 freshly updated by block_1 algorithm can be used HERE
    loop_reactor(block_2.update_inplace(signal_2))
    # => signal_2 freshly updated by block_2 algorithm can be used HERE
    loop_reactor(block_3.update(signal_2, signal_3))
    # => signal_3 freshly updated by block_3 algorithm can be used HERE

    #############
    # Control logic here #
    #############

    # Rate control — sleep to maintain the desired frequency
    elapsed = time.perf_counter() - start_time
    sleep_time = dt - elapsed
    if sleep_time > 0:
        time.sleep(sleep_time)

---

Distribution

The Bota Control Toolkit definitions are distributed as a Python package.

It can be installed using pip:

pip install bota-control

Important

The bota-control package is a dependency of all the other Bota Control Toolkit blocks.