Data Visualization and Recording
Often, you'll want access to historical data generated by Devices in addition to seeing data about their current states. Most Devices in the Aqueduct Library have the ability to record their state at each update interval. These recorded timeseries can be visualized in the user interface with a Chart Widget.
Let's add a new Tab and Tab Container to our window and then add a Chart Widget:
A Chart Widget allows you to view Recordable timeseries for both Device and User data. We'll see how to create a User Recordable later, but for now let's add a Device Recordable to the Chart by selecting:
> Series > Devices > peristaltic_pump_000001 > pump 0 disp. (mL)
Use the controls section in the left menu to start peristaltic_pump_000001 in the clockwise direction at 10 mL/min. Make sure that the checkbox in the record toggle is selected and then press the start button on the control panel:
As the pump starts operating, you'll see a line appear on the chart that increases with time. We're plotting the milliliters displaced by the pump.
Enter a larger rate value (20 mL/min, for instance) in the controls panel and
press the start button again. The milliliters displaced value will reset to 0 and then start
increasing at a higher rate.
You can add multiple Chart Widgets to your window layout. Add a second Chart Widget to the Tab Container and then add the pump 0 rate (mL/min) timeseries.
Let's revisit our infinite loop recipe from the last step and modify it to
record the changes in the pump's rate. The only change that we need to make is
the addition of setting the record argument to True in the pump's start method:
pump.start(commands, record = True)
You'll see a record argument in many Device methods throughout the
Aqueduct API. The functionality of this argument is the same:
setting record to True will instruct the platform to record
the device's state in a timeseries and make it accessible in Chart Widgets.
Make the above change to your code in the Editor Widget, queue the Recipe, and begin execution. Now, in addition to seeing the pump cycling between 0 and 50 mL/min, you'll see a sawtooth-like timeseries of the pump's rpm as our Recipe continuously changes its speed.
You can create your own timeseries
using the Aqueduct API's Recordable class. For instance, you may wish to calculate a
derived value using device data as an input.
Let's make a Recordable to display the total volume displaced by the pump during our Recipe. Go
back to the PeristalticPump class in the Aqueduct API (peristaltic.py) and take a look at the get_ml_done method:
aqueduct/devices/pump/peristaltic.py
class PeristalticPump(devices.base.obj.Device):
# snip
def get_ml_done(self) -> Tuple[float]:
"""
Get the volume dispensed by the peristaltic pump in milliliters.
:return: Volume dispensed in milliliters.
:rtype: tuple
"""
return self.extract_live_as_tuple(PeristalticPumpLiveKeys.ml_done.value)
## snip
We can use the pump's get_ml_done method to calculate volume displaced in milliliters.
The other tool we'll need to implement our volume counter is the Recordable API, which is found here (Recordable) and here (Aqueduct recordable method):
The API methods are reproduced below for easier reference:
aqueduct/core/aq.py
class Aqueduct(object):
# snip
def recordable(
self,
name: str,
value: typing.Union[float, int, bool, str, datetime.datetime, list],
dtype: str = None,
) -> Recordable:
"""
Creates a new Recordable object and registers it with Aqueduct.
:param name: The name of the Recordable.
:type name: str
:param value: The initial value of the Recordable.
:type value: Union[float, int, bool, str, datetime.datetime, list]
:param dtype: The data type of the Recordable value, defaults to None.
:type dtype: str, optional
:return: The newly created Recordable object.
:rtype: Recordable
"""
recordable = Recordable(name, value, dtype)
self.register_recordable(recordable)
self.update_recordable(recordable)
return recordable
# snip
aqueduct/core/recordable.py
class Recordable:
"""
The `Recordable` class allows you to log timestamped data.
:param name: Name of the recordable, will be displayed on the UI, should be unique
:type name: str, required
:param value: Initial value to be assigned to the recordable on creation
:type value: float, int, bool, str, datetime.datetime, list, required
:param dtype: Specify the type of value, used to ensure that users cannot
enter an invalid value
:type dtype: {'int', 'float', 'bool', 'list', 'datetime', 'str'}, optional
"""
name: str = None
value: Union[float, int, bool, str, datetime.datetime, list] = None
dtype: str = None
timestamp = None
_aq: "Aqueduct" = None
def __init__(
self,
name: str,
value: Union[float, int, bool, str, datetime.datetime, list],
dtype: str = None,
):
"""
Constructor method.
:param name: name of the Recordable, will be displayed on the UI, should be unique
:type name: str, required
:param value: value to be assigned to the Recordable on creation
:type value: float, int, bool, str, datetime.datetime, list, required
:param dtype: specify the type of value, used to ensure that users cannot enter an invalid value
:type dtype: {'int', 'float', 'bool', 'list', 'datetime', 'str'}, optional
"""
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 Recordable".format(
{dtype}
)
)
except Exception:
raise ValueError("Invalid Aqueduct Recordable")
self.name = name
self.value = value
self.dtype = dtype
# not all methods present
def assign(self, aqueduct: "Aqueduct"):
"""
Assigns the Recordable to an `Aqueduct` instance.
:param aqueduct: The `Aqueduct` instance to assign the Recordable to.
:type aqueduct: Aqueduct, required
"""
self._aq = aqueduct
def update(self, value):
"""
Updates the value of the Recordable.
:param value: The new value for the Recordable.
:type value: Union[float, int, bool, str, datetime.datetime, list], required
"""
self.value = value
self._aq.update_recordable(self)
def clear(self):
"""
Clears the value of the Recordable.
"""
self._aq.clear_recordable(self)
So, our strategy will be to create a Recordable instance using the aqueduct object's
recordable method, assign it a name of our choosing, and then update it as needed with the
value returned from the pump's ml_pumped method.
Let's modify our code to create the Recordable before the entry to our infinite loop and
update the value at the end of each loop iteration:
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( ## <--- added recordable creation
name="ml_pumped", ## <--- name will be displayed on chart Widget
dtype=float.__name__, ## <--- define the data type
value=0., ## <--- 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 True:
# 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")
pump.stop()
print("Complete!")
Let's Queue and rerun the script. After starting, navigate to the tab with your
Chart Widgets and add the ml_pumped series by selecting:
> Series > User > ml_pumped
from the Chart context menu. You should see a second timeseries with a line that continues to increase as the pump traverses more revolutions.
Looks good!
This Chapter has introduced us to Recordables and recording data, which make timeseries
data available to plot in a Chart Widget. Many device API methods accept a record argument
that will instruct the application to record that device's state at each update interval. User Recordables
can be used to extend the built-in device recording parameters with derived or calculated values
that you may wish to record over time.
Now, let's move on to the next Chapter, which will introduce Setpoints and Logging.