Setpoints and Logging

Currently, our Recipe runs forever. We've wrapped our pump ramp up/ramp down logic in an infinite while True loop. You'll likely want your Recipe to end or at least be able to terminate it on demand! More generally, it would be nice to have a way to adjust parameters in our Recipe logic from the user interface in real-time. To do this, we can use the Aqueduct Setpoint API.

The Setpoint API can be found here (Setpoint) and here (Aqueduct setpoint method):

The API methods are reproduced below for easier reference:

aqueduct/core/aq.py

class Aqueduct(object):

  # snip

  def setpoint(
        self,
        name: str,
        value: typing.Union[float, int, bool, str, datetime.datetime, list],
        dtype: str = None,
    ) -> Setpoint:
        """
        Creates a new Setpoint object.

        :param name: The name of the setpoint
        :type name: str
        :param value: The initial value of the setpoint
        :type value: typing.Union[float, int, bool, str, datetime.datetime, list]
        :param dtype: The data type of the setpoint value, defaults to None
        :type dtype: str, optional
        :return: A new Setpoint object
        :rtype: aqueduct.core.setpoint.Setpoint
        """
        s = Setpoint(name, value, dtype)
        self.register_setpoint(s)
        self.update_setpoint(s)
        return s

  # snip

aqueduct/core/setpoint.py

class Setpoint:
    """
    A class to provide simple interaction with Recipe values that
    appear as User Params on the Aqueduct Recipe Builder UI.

    Args:
        name (str): name of the Setpoint, will be displayed on the UI, should be unique
        value (Union[float, int, bool, str, datetime.datetime, list]): value to be assigned to the Setpoint on creating
        dtype ({'int', 'float', 'bool', 'list', 'datetime', 'str'}, optional): specify the type of value, used to ensure
            that Users cannot enter an invalid value

    Attributes:
        name (str): name of the Setpoint, will be displayed on the UI, should be unique
        value (Union[float, int, bool, str, datetime.datetime, list]): value of the Setpoint
        dtype (str): specify the type of value, used to ensure that Users cannot enter an invalid value
        timestamp (datetime.datetime): timestamp when the Setpoint is updated
        on_change (Callable): a function that is called when the Setpoint is updated
        args (list): arguments to pass to the `on_change` function
        kwargs (dict): keyword arguments to pass to the `on_change` function
        _aq (Aqueduct): reference to the Aqueduct object to allow for easy updating of the Setpoint

    """

    name: str = None
    value: Union[float, int, bool, str, datetime.datetime, list] = None
    dtype: str = None
    timestamp = None
    on_change: Callable = None
    args: list = []
    kwargs: dict = {}

    _aq: "Aqueduct" = None

    def __init__(
        self,
        name: str,
        value: Union[float, int, bool, str, datetime.datetime, list],
        dtype: str = None,
    ):
        """
        Constructor method.
        """

        if dtype is None:
            try:
                dtype = type(value).__name__
                if dtype not in ALLOWED_DTYPES:
                    raise ValueError(
                        "Object of type {} is not allowed as a Setpoint".format({dtype})
                    )
            except Exception:
                raise ValueError("Invalid Aqueduct Setpoint")

        self.name = name
        self.value = value
        self.dtype = dtype

    # not all methods present

    def update(self, value):
        """
        Update the value of this Setpoint object.

        :param value: The new value to set for the Setpoint.
        :type value: float, int, bool, str, datetime.datetime, list, required

        :return: None
        """
        self.value = value
        self._aq.update_setpoint(self)

Instead of using an always True condition for our while loop, let's use a boolean Setpoint instance that we can toggle from the user interface.

To implement this, we'll create an instance of a Setpoint named stop_setpoint prior to the entry to our while loop. Then, we'll use the value of this setpoint as the loop's controlling expression.

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")

volume_pumped = aq.recordable(        
  name="ml_pumped",                   
  dtype=float.__name__,               
  value=0.,                          
)

stop_setpoint = aq.setpoint(    ## <--- added setpoint creation
  name="stop",                        ## <--- name will be displayed on
                                      ##      the Setpoint menu
  dtype=bool.__name__,                ## <--- define the data type
  value=False,                        ## <--- set the initial value
)

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 = 1

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

while not stop_setpoint.value:        ## <--- modify loop controlling expression
    # 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)

        volume_pumped.update(pump.get_ml_done()[0]) ## <--- update the value

        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)

        volume_pumped.update(pump.get_ml_done()[0]) ## <--- update the value

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

# now this is reachable!
pump.stop()
print("Complete!")

Again, Queue and rerun the script. After starting, click the Setpoint Menu Toggle in the
left sidebar menu:

Setpoint Menu Toggle Setpoint

where you'll see two rows appear in the Setpoint Menu table.

One row will contain an entry with a button labelled stop and a bool indicator signifying that the type of the Setpoint is a boolean. In the right column, you'll find an input to change the value of the Setpoint. Select the field and change the value to True (true, True, and 1 are all acceptable values for True). After editing the value, the button's border will change to blue.

Click the button to finish updating the value.

Now, the expression for our while loop will evaluate to False, so you'll see the loop terminate and the output:

Complete!

appear in the Terminal Widget. So, we modifed our Setpoint from the user interface and used the modified value to control the logic of our recipe.

In addition to bool Setpoints, you can create Setpoints of types int, float, list, datetime, and str.

We have one last addition to make to our Recipe. As of now, we can plot data using Recordables and create real-time parameters using Setpoints. Often, you'll want to save certain data recorded during your Recipe permanently. To do this, we can use the Aqueduct class's set_log_file_name, log, and other logging methods, which can be found here and are duplicated below:

class Aqueduct(object):

    # snip

    def set_log_file_name(self, log_file_name: str):
        """Set the log file name."""
        self._logger.set_log_file_name(log_file_name)

    def log(self, message):
        """
        Logs a message.

        Args:
            message (str): The message to be logged.

        Returns:
            None
        """
        self._logger.log(message)

    def debug(self, message):
        """
        Logs a message with severity 'DEBUG'.

        Args:
            message (str): The message to be logged.

        Returns:
            None
        """
        self._logger.debug(message)

    def info(self, message):
        """
        Logs a message with severity 'INFO'.

        Args:
            message (str): The message to be logged.

        Returns:
            None
        """
        self._logger.info(message)

    def warning(self, message):
        """
        Logs a message with severity 'WARNING'.

        Args:
            message (str): The message to be logged.

        Returns:
            None
        """
        self._logger.warning(message)

    def error(self, message):
        """
        Logs a message with severity 'ERROR'.

        Args:
            message (str): The message to be logged.

        Returns:
            None
        """
        self._logger.error(message)

    def critical(self, message):
        """
        Logs a message with severity 'CRITICAL'.

        Args:
            message (str): The message to be logged.

        Returns:
            None
        """
        self._logger.critical(message)

We've already used the pump's ml_pumped method to make a Recordable; now, let's add the code needed to save the value to a log file.

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")

volume_pumped = aq.recordable(        
  name="ml_pumped",                   
  dtype=float.__name__,               
  value=0.,                          
)

stop_setpoint = aq.setpoint(   
  name="stop",                        
  dtype=bool.__name__,                
  value=False,                        
)

aq.set_log_file_name("first_log")     ## <--- set the name of the log file

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 = 1

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

while not stop_setpoint.value:        
    # 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)

        ml_pumped = pump.get_ml_done()[0]

        aq.log(f"ml_pumped: {ml_pumped},") ## <--- update the log data

        volume_pumped.update(ml_pumped) 

        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)

        ml_pumped = pump.get_ml_done()[0]

        volume_pumped.update(pump.get_ml_done()[0])

        aq.log(f"ml_pumped: {ml_pumped},") ## <--- update the log data

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

# now this is reachable!
pump.stop()
print("Complete!")

With these changes, the ml_pumped value will be captured and appended to the log file in each loop iteration.

Let's rerun the Recipe. Let the loop complete a few iterations, and then change the stop Setpoint to True. Navigate to the Logs section of the Dashboard and select the Tab for Sim Mode logs:

That completes our introduction to Recipe Scripting! We've seen how to:

  1. Use the API for the PeristalticPump Device type to automatically control the operation of the peristaltic pump
  2. Use the Pause Recipe and E-Stop Recipe Buttons to control execution of an active Recipe
  3. Use the record argument and Recordable class to capture timeseries data
  4. Use the Setpoint class to create real-time control parameters
  5. Use the log methods to save data permanently

With that introduction to the core API Classes and Methods, let's move on to a more complex real-world application to control the pH of a reaction. The complexity of the code needed to execute our real-world example will provide an opportunity to create a Library, which will allow us to cleanly separate parts of our code, enable easier reuse of logic, and leverage the capabilities of Integrated Development Environments to accelerate and simplify development.