Use the API

To interact with and control Devices in an Aqueduct Recipe, you'll make use of the aqueduct-py API. The API is a collection of Python Classes and Methods that you can leverage to control Recipe execution, generate User Inputs and Prompts, visualize data, and control Devices.

Each Device type has an associated Python class with the same name. So, our peristaltic pump device will use the PeristalticPump class in the Aqueduct API. You can see the source code for the different device types in the aqueduct-py repository here devices and the PeristalticPump specifically here: peristaltic.py.

Registering a Python Process and Loading Devices

To interact with the Aqueduct system and control the devices in your Setup, you need to register your Python process with the Aqueduct application. This registration is done using the following lines of code:

from aqueduct.core.aq import Aqueduct
from aqueduct.core.aq import InitParams

# parse initialization parameters and create Aqueduct instance
params = InitParams.parse()
aq = Aqueduct(params.user_id, params.ip_address, params.port)

aq.initialize(params.init)

After registering your recipe, you can access the devices in your Setup by their names. For example, to retrieve the peristaltic pump named peristaltic_pump_000001, use the following code:

from aqueduct.devices.pump import PeristalticPump
pump: PeristalticPump = aq.devices.get("peristaltic_pump_000001")
print(pump)

Copy the text into the editor. Now, let's queue our Recipe by using the Editor Widget menu (the ellipsis) to select:

> Queue Recipe

You'll receive a notification saying that the Recipe was successfully queued and the Recipe Status Indicator will display a blue Q. Click the Start/Resume Recipe Button on the left sidebar menu to start the Recipe.

The output in the Recipe Output Widget reads:

<aqueduct.devices.pump.peristaltic.PeristalticPump object at 0xb4b0fb90>

The printable representation shows that we have a aqueduct.devices.pump.peristaltic.PeristalticPump object at some location in memory 0xb4b0fb90.

You've run your first recipe! After the recipe is complete, the Recipe Status Indicator displays a blue C indicating that it's complete. This recipe didn't do anything useful, but these are the basics of scripting in the Aqueduct environment.

Enter your Recipe Script in the Editor Widget
Queue your Recipe
Press the Start/Resume Recipe Button to begin execution

Creating and Sending Device Commands

Now let's write a recipe that:

starts the pump at 2 mL/min
changes its speed in a loop up to 50 rpm by an increment of 0.1 mL/min
then decrements its speed to 0.1 mL/min
and then stops the pump

Let's use our previous snippet as a starting point:

from aqueduct.core.aq import Aqueduct
from aqueduct.core.aq import InitParams
from aqueduct.devices.pump import PeristalticPump

# parse initialization parameters and create Aqueduct instance
params = InitParams.parse()
aq = Aqueduct(params.user_id, params.ip_address, params.port)

# initialize the devices and set a command delay
aq.initialize(params.init)
aq.set_command_delay(0.05)

# get the peristaltic pump device and create a command object
pump: PeristalticPump = aq.devices.get("peristaltic_pump_000001")

# todo - start pump

# set the maximum speed and speed increment for the pump
MAX_SPEED: float = 50
INCREMENT: float = 0.1

# calculate the number of steps based on the maximum speed and increment
STEPS = int(MAX_SPEED / INCREMENT)

# loop through the speed increment steps
for i in range(0, STEPS):
    # todo
    _ = i

# loop through the speed decrement steps
for i in range(STEPS, 0, -1):
    # todo
    _ = i

# todo - stop pump

The first action we need to implement in the script is starting the pump.

To control a device in the Aqueduct system, you can create commands that specify the desired settings and actions for the device.

First, create an empty list to store the commands for the pump:

commands = pump.make_commands()

Next, create a specific command to start the pump. In this example, we want the pump to run continuously at a rate of 2 mL/min in the clockwise direction. The make_start_command() method is used to create the command with the desired settings:

command = pump.make_start_command(
    mode=pump.MODE.Continuous,
    rate_units=pump.RATE_UNITS.MlMin,
    rate_value=2,
    direction=pump.STATUS.Clockwise,
)

Once the command is created, you can set it for the pump. Since there may be multiple nodes or channels in a pump, you need to specify the index of the node for which you want to set the command. In this case, we are using the first (and only) node, which has an index of 0:

pump.set_command(commands, 0, command)

Finally, send the start command to the Aqueduct system to initiate the pump operation:

pump.start(commands)

These steps allow you to create and set commands for controlling devices in the Aqueduct system. By specifying the desired settings and actions in the commands, you can effectively operate and control the devices according to your requirements.

from aqueduct.core.aq import Aqueduct
from aqueduct.core.aq import InitParams
from aqueduct.devices.pump import PeristalticPump

# parse initialization parameters and create Aqueduct instance
params = InitParams.parse()
aq = Aqueduct(params.user_id, params.ip_address, params.port)

# initialize the devices and set a command delay
aq.initialize(params.init)
aq.set_command_delay(0.05)

# get the peristaltic pump device and create a command object
pump: PeristalticPump = aq.devices.get("peristaltic_pump_000001")

# make an empty commands list
commands = pump.make_commands()

command = pump.make_start_command(
    mode=pump.MODE.Continuous,
    rate_units=pump.RATE_UNITS.MlMin,
    rate_value=2,
    direction=pump.STATUS.Clockwise,
)

# set the command for the first (and only) node of the pump
pump.set_command(commands, 0, command)

# now send the start command
pump.start(commands)

# set the maximum speed and speed increment for the pump
MAX_SPEED: float = 50
INCREMENT: float = 0.1

# calculate the number of steps based on the maximum speed and increment
STEPS = int(MAX_SPEED / INCREMENT)

# loop through the speed increment steps
for i in range(0, STEPS):
    # todo
    _ = i

# loop through the speed decrement steps
for i in range(STEPS, 0, -1):
    # todo
    _ = i

# todo - stop pump

At this point, let's queue up our edited script and run it:

The pump starts and our recipe completes, as shown by the Recipe Status Indicator displaying a blue C.

Now we're ready to add logic to ramp up and then ramp down the pump's speed. We'll use two for loops - one loop to count up in a configurable increment to our target top speed (50 mL/min), and one loop to decrement from our top speed to our ending speed (2 mL/min).

To avoid duplicating parameters, we'll create the variables MAX_SPEED, INCREMENT, and STEPS to define these values in a single place in the code. Additionally, we'll add a print statement in each loop to confirm that we're hitting our targets, as the pump display may not update quickly enough to give us feedback.

commands = pump.make_commands()

command = pump.make_start_command(
    mode=pump.MODE.Continuous,
    rate_units=pump.RATE_UNITS.MlMin,
    rate_value=2,
    direction=pump.STATUS.Clockwise,
)

# set the command for the first (and only) node of the pump
pump.set_command(commands, 0, command)

# now send the start command
pump.start(commands)

# set the maximum speed and speed increment for the pump
MAX_SPEED: float = 50
INCREMENT: float = 0.1

# calculate the number of steps based on the maximum speed and increment
STEPS = int(MAX_SPEED / INCREMENT)

# ramp up the speed to `top_speed_rpm`
for i in range(0, STEPS):
    commands = pump.make_commands()

    # create a command to change the pump speed
    c = pump.make_change_speed_command(
        rate_value=i * INCREMENT, rate_units=pump.RATE_UNITS.MlMin
    )

    pump.set_command(commands, 0, c)

    pump.change_speed(commands)

    print(f"Ramping up to: {i * INCREMENT} mL/min")

# ramp down the speed to `2 mL/min`
for i in range(STEPS, 0, -1):
    commands = pump.make_commands()

    # create a command to change the pump speed
    c = pump.make_change_speed_command(
        rate_value=i * INCREMENT, rate_units=pump.RATE_UNITS.MlMin
    )

    pump.set_command(commands, 0, c)

    pump.change_speed(commands)

    print(f"Ramping down to: {i * INCREMENT} mL/min")

# todo

Our recipe logic is almost there; we just need to stop the pump after the ramp down to 2 mL/min has completed. Let's modify the script to call the PeristalticPump class's stop method at the end of the script:

from aqueduct.core.aq import Aqueduct
from aqueduct.core.aq import InitParams
from aqueduct.devices.pump import PeristalticPump

# parse initialization parameters and create Aqueduct instance
params = InitParams.parse()
aq = Aqueduct(params.user_id, params.ip_address, params.port)

# initialize the devices and set a command delay
aq.initialize(params.init)
aq.set_command_delay(0.05)

# get the peristaltic pump device and create a command object
pump: PeristalticPump = aq.devices.get("peristaltic_pump_000001")

commands = pump.make_commands()

command = pump.make_start_command(
    mode=pump.MODE.Continuous,
    rate_units=pump.RATE_UNITS.MlMin,
    rate_value=2,
    direction=pump.STATUS.Clockwise,
)

# set the command for the first (and only) node of the pump
pump.set_command(commands, 0, command)

# now send the start command
pump.start(commands)

# set the maximum speed and speed increment for the pump
MAX_SPEED: float = 50
INCREMENT: float = 0.1

# calculate the number of steps based on the maximum speed and increment
STEPS = int(MAX_SPEED / INCREMENT)

# ramp up the speed to `top_speed_rpm`
for i in range(0, STEPS):
    commands = pump.make_commands()

    # create a command to change the pump speed
    c = pump.make_change_speed_command(
        rate_value=i * INCREMENT, rate_units=pump.RATE_UNITS.MlMin
    )

    pump.set_command(commands, 0, c)

    pump.change_speed(commands)

    print(f"Ramping up to: {i * INCREMENT} mL/min")

# ramp down the speed to `2 mL/min`
for i in range(STEPS, 0, -1):
    commands = pump.make_commands()

    # create a command to change the pump speed
    c = pump.make_change_speed_command(
        rate_value=i * INCREMENT, rate_units=pump.RATE_UNITS.MlMin
    )

    pump.set_command(commands, 0, c)

    pump.change_speed(commands)

    print(f"Ramping down to: {i * INCREMENT} mL/min")

pump.stop() ## <---------- stop
print("Complete!")

Queue and rerun: (Rearrange your window if needed to make the Terminal output visible.)

Very nice. We've successfully implemented the steps we outlined earlier. Let's move on to see how to interact with an active Recipe.