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 |
|
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:
- Use the API for the
PeristalticPumpDevice type to automatically control the operation of the peristaltic pump - Use the Pause Recipe and E-Stop Recipe Buttons to control execution of an active Recipe
- Use the
recordargument andRecordableclass to capture timeseries data - Use the
Setpointclass to create real-time control parameters - Use the
logmethods 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.