Writing a new automation
An automation is a hands-off way to adjust the environment for the microbes. We currently support three types of automations: dosing, LED, and temperature.
In this section, we'll develop a simple dosing automation.
Creating our first custom automation
Writing an automation involves creating a Python class and overriding specific methods. It would be helpful to be somewhat familiar with Python classes before beginning. Here's an example of a (naive) turbidostat automation, i.e. it will add fresh media and remove old media when an optical density threshold is exceeded. The full code is below, and we'll go through each line of code after:
from pioreactor.automations.dosing.base import DosingAutomationJobContrib
class NaiveTurbidostat(DosingAutomationJobContrib):
automation_name = "naive_turbidostat"
published_settings = {
"target_od": {"datatype": "float", "settable": True, "unit": "od"},
}
def __init__(self, target_od, **kwargs):
super().__init__(**kwargs)
self.target_od = float(target_od)
def execute(self):
if self.latest_od > self.target_od:
self.execute_io_action(media_ml=1.0, waste_ml=1.0)
First important thing is to subclass from DosingAutomationJobContrib:
from pioreactor.automations.dosing.base import DosingAutomationJobContrib
class NaiveTurbidostat(DosingAutomationJobContrib):
...
The DosingAutomationJobContrib is a subclass of a BackgroundJob. The -Contrib part is a small detail to specify that it's a third-party automation (i.e. you are developing it, not us.)
We need a "key" to i) distinguish this from other automations, and ii) be able to be communicate between systems (think: the web UI in JavaScript to Python, and back). The automation_name attribute does this for us. Normally, it's the snakecase of the class name.
automation_name = "naive_turbidostat"
The published_settings tells the Pioreactor software what class attributes are published to MQTT, and if they are editable via MQTT (we will try editing over MQTT later). This is important if you wish to dynamically change attributes of an automation during an experiment, for example: from the web interface. Our class has the following:
...
published_settings = {
"target_od": {"datatype": "float", "settable": True, "unit": "od"},
}
...
The associated metadata says that the class attribute target_od is a float, is editable via MQTT (so it can be changed using the web interface), and it has units od.
Next, we define how to initialize our automation class. Here we can add settings we want to accept from the user: what is our initial target optical density. Note the boilerplate **kwargs, and super() are important.
def __init__(self, target_od, **kwargs):
super().__init__(**kwargs)
self.target_od = float(target_od)
Finally, every duration (specified in the controller, later in this section) minutes, the function execute will run. duration can be kept low (ex: to often check for some condition), or used to set some periodic task (ex: a chemostat doses every X minutes). The execute contains the core logic of the automation. In our simple case, we want to dilute the vial if we have exceed the latest_od:
def execute(self):
if self.latest_od > self.target_od:
self.execute_io_action(media_ml=1.0, waste_ml=1.0)
Since we are working with a fixed volume, media_ml must equal waste_ml, else an error will be thrown. What is latest_od attribute? Our class, when active, is listening to new optical densities being recorded. Hence when execute runs, we'll have access to the most up-to-date value of optical density. Likewise, there are also latest_normalized_od and latest_growth_rate attributes that update when a new growth-rate value is calculated. All three attributes are defined and maintained in the parent class.
Running the automation
How do we run this automation now? Let's copy the code into a file called naive_turbidostat.py and place it into the folder ~/.pioreactor/plugins.
You can create this file on your Pioreactor's Raspberry Pi: after accessing the Raspberry Pi's command line, typing nano ~/.pioreactor/plugins/naive_turbidostat.py, and pasting in the code below.
# -*- coding: utf-8 -*-
from pioreactor.automations.dosing.base import DosingAutomationJobContrib
class NaiveTurbidostat(DosingAutomationJobContrib):
automation_name = "naive_turbidostat"
published_settings = {
"target_od": {"datatype": "float", "settable": True, "unit": "od"},
}
def __init__(self, target_od, **kwargs):
super().__init__(**kwargs)
self.target_od = float(target_od)
def execute(self):
if self.latest_od > self.target_od:
self.execute_io_action(media_ml=1.0, waste_ml=1.0)
Run the script with pio run dosing_atomation --automation-name naive_turbidostat --target-od 0.5. This will start the job. After a moment, you may notice that warnings are thrown - that's because there's no optical density measurements being produced! You can use crtl-c to stop the job.
Editing attributes over MQTT (optional)
We'll demonstrate the ability to dynamically change the target_od attribute using MQTT. For each member of published_settings, the DosingAutomationJobContrib class listens to the MQTT topic:
pioreactor/<unit name>/<experiment>/dosing_automation/<attribute>/set
We'll use mosquitto_pub to publish a message to this topic. So, with the Python script running, open a new command line, and enter the following:
mosquitto_pub -t "pioreactor/test_unit/test_experiment/dosing_automation/target_od/set" -m 5.0 -u pioreactor -P raspberry
You should see some logs in the Python console report that the target_od was changed. Also, a the value of 5.0 is published and retained to the topic pioreactor/test_unit/test_experiment/dosing_automation/target_od
Why is this useful?
- This is how the web interface updates settings in running activities.
- Other Pioreactor activities can update each other's settings.
- External programs or apps can monitor and update settings this way, too.
Adding the automation to the UI
To add your automation to the UI so it appears in the automation drop-down, follow the the steps here.
Extensions of our custom automation
Below are some extensions, with additions highlighted.
Dynamic volume exchanged
Exchanging 1ml each time may not be enough, so we add volume to the published_settings. Now, from the UI, we can dynamically adjust the volume.
from pioreactor.automations.dosing.base import DosingAutomationJobContrib
class NaiveTurbidostat(DosingAutomationJobContrib):
automation_name = "naive_turbidostat"
published_settings = {
"target_od": {"datatype": "float", "settable": True, "unit": "od"},
"volume": {"datatype": "float", "settable": True, "unit": "mL"},
}
def __init__(self, target_od, volume, **kwargs):
super().__init__(**kwargs)
self.target_od = float(target_od)
self.volume = float(volume)
def execute(self):
if self.latest_od > self.target_od:
self.execute_io_action(media_ml=self.volume, waste_ml=self.volume)
Using latest_growth_rate
If our growth rate is high, we may want to modify the volume exchanged to keep up. A naive solution: we can bump up the exchanged volume if the growth rate is high. Much better would be a dynamic solution, or a feedback loop.
from pioreactor.automations.dosing.base import DosingAutomationJobContrib
class NaiveTurbidostat(DosingAutomationJobContrib):
automation_name = "naive_turbidostat"
published_settings = {
"target_od": {"datatype": "float", "settable": True, "unit": "od"},
"volume": {"datatype": "float", "settable": True, "unit": "mL"},
}
def __init__(self, target_od, volume, **kwargs):
super().__init__(**kwargs)
self.target_od = float(target_od)
self.volume = float(volume)
def execute(self):
if self.latest_od > self.target_od:
if self.latest_growth_rate > 0.2:
self.execute_io_action(media_ml=2 * self.volume, waste_ml=2 * self.volume)
else:
self.execute_io_action(media_ml=self.volume, waste_ml=self.volume)