I am trying to implement additional pumps to the Pioreactor using an external motor driver. I have had success with controlling a third pump using separate test programs and through the command line, but am encountering errors when implementing it through custom automations following the Pioreactor docs.
I am using the Adafruit DC & Stepper Motor HAT and have it mounted to an RPi 3A+. The Pioreactor HAT sits on top of that.
For now I simply want to test the functionality of the third pump and Motor HAT by semi-recreating the fed-batch and chemostat automations. In both of which, I am replacing the default media pump on PWM 2 of the Pioreactor HAT with M1 on the Motor HAT. I have both automations successfully showing up in the UI so the .yaml files have no issues and I believe it is only the Python files. Below are the Python programs I’ve written utilizing the external pump instead of the built-in media pump.
three_pumps.py
# -*- coding: utf-8 -*-
# Exchange X mL every period
import time
from adafruit_motorkit import MotorKit
from pioreactor.automations import events
from pioreactor.automations.dosing.base import DosingAutomationJobContrib
class ThreePumps(DosingAutomationJobContrib):
automation_name = "three_pumps"
published_settings = {
"exchange_volume_ml": {"datatype": "float", "settable": True, "unit": "mL"},
}
def __init__(self, exchange_volume_ml, **kwargs):
super().__init__(**kwargs)
# Initialize Motor HAT
self.kit = MotorKit(address=0x60) # TODO: Check I2C address
self.exchange_volume_ml = float(exchange_volume_ml)
def add_third_media_to_bioreactor(self, ml: float, unit: str, experiment: str, source_of_event: str, mqtt_client) -> float:
seconds_per_ml = 3.0 # TODO: Calibration
duration = ml * seconds_per_ml
self.kit.motor1.throttle = 1.0
time.sleep(duration)
self.kit.motor1.throttle = 0.0
self.logger.info(f"Pumped {ml}")
return ml
def execute(self) -> None:
results = self.execute_io_action(third_media_ml=self.exchange_volume_ml, waste_ml=self.exchange_volume_ml)
return
This chemostat-style automation starts and runs when starting from the UI, but I encounter the following error everytime the pump is supposed to run: ThreePumps.add_third_media_to_bioreactor() got an unexpected keyword argument 'logger'.
fed_batch_ext.py
# -*- coding: utf-8 -*-
# Add X mL every period
import time
from adafruit_motorkit import MotorKit
from pioreactor.automations import events
from pioreactor.automations.dosing.base import DosingAutomationJobContrib
class FedBatchExt(DosingAutomationJobContrib):
automation_name = "fed_batch_ext"
published_settings = {
"dosing_volume_ml": {"datatype": "float", "unit": "mL", "settable": True},
}
def __init__(self, dosing_volume_ml, **kwargs) -> None:
super().__init__(**kwargs)
# Initialize Motor HAT
self.kit = MotorKit(address=0x60) # TODO: Check I2C address
self.logger.warning("When using the fed-batch automation, no liquid is removed. Carefully monitor the level of liquid to avoid overflow!")
self.dosing_volume_ml = float(dosing_volume_ml)
def add_third_media_to_bioreactor(self, ml: float, unit: str, experiment: str, source_of_event: str, mqtt_client) -> float:
seconds_per_ml = 3.0 # TODO: Calibration
duration = ml * seconds_per_ml
self.kit.motor1.throttle = 1.0
time.sleep(duration)
self.kit.motor1.throttle = 0.0
return ml
def execute(self):
vol = self.add_third_media_to_bioreactor(ml=self.dosing_volume_ml, unit=self.unit, experiment=self.experiment, source_of_event=f"{self.job_name}:{self.automation_name}")
if vol != self.dosing_volume_ml:
self.logger.warning("Under-dosed!")
return events.AddMediaEvent(f"Added {vol} mL")
With this automation to mimic a fed-batch operation except using an external pump, I get this error: FedBatchExt.add_third_media_to_bioreactor() missing 1 required positional argument: 'mqtt_client'
The error codes are pretty self explanatory in that there are unexpected/missing arguments, but I am confused on why. For the first program I am not passing logger so I am unsure why it is getting a logger argument. Similarly for the second program I am unsure why it is saying mqtt_client is missing when it is being passed as an argument in the functions.
Thank you for any assistance you can provide!