Hi. I want to try controlling both temperature and dosing from a single automation, but this is new territory for me and I am not sure how to approach this or what conflicts might arise and why. If you have any advice/suggestions, or possible relevant terminology that I could google, that would be great! So far, I have been using the /pioreactor/automations/dosing/chemostat.py as a template and building everything off there, so I am feeling a lot more confident working with dosing and inheritance from DosingAutomationJob.
I am hoping to create an automation which can schedule when to update and run and update dosing (nutrient media, salt solution alternate media) and temperature control.
I am not really sure what questions I should be asking, so I tried writing out my thought process. Here is what I think I should generally do:
1. My class ChemostatAltMedia should be inheriting from both DosingAutomationJobContrib and Thermostat, or possibly TemperatureController (this would be inherited through Thermostat regardless, right?)
I would define my class like this: class ChemostatAltMedia(DosingAutomationJobContrib, Thermostat):. I see that there is an execute(self) function defined in both parent classes DosingAutomationJobContrib and Thermostat, and I am also defining execute(self) in child class ChemostatAltMedia. How are these conflicts resolved, and should I be concerned about these types of conflicts? I suspect that an instance attribute/function will overwrite inherited class attributes from a
parent, but what about conflicts when inheriting class attributes or functions from two different parents? Moreover, can I still call these inherited attributes/functions via something like self.DosingAutomationContrib.execute(), self.Thermostat.execute(), and self.execute()?
2. I will need to update how I instantiate my class.
So far, I have only been working with dosing, so I have been instantiating my classes via dc = DosingController("scheduled_chemostat_alt_media",...). I suspect I might need to expand this somehow, but I am not sure what that would look like. Maybe something like this?
if __name__ == "__main__":
from pioreactor.background_jobs.dosing_control import DosingController
from pioreactor.background_jobs.temperature_control import TemperatureController
dc = DosingControl("scheduled_chemostat_alt_media",...)
dc = TemperatureController(dc,...)
3. Here is my code that I have written so far.
Edit: I cleaned up and removed a lot of code in a reply to this post.
Currently, it is giving me an error saying __init__() missing 1 required positional argument: 'target_temperature', but this goes away when I remove Thermostat from class ChemostatAltMedia(...). I am passing in a target_temperature argument to DosingController, but that is not resolving the issue. I have tried a couple different changes to the code without success, but I think I need to get a better idea of what is going on to troubleshoot this more effectively. I have been looking through the source code but havenāt had luck so far.
# -*- coding: utf-8 -*-
from __future__ import annotations
from pioreactor.automations import events
from pioreactor.automations.dosing.base import DosingAutomationJobContrib
from pioreactor.automations.temperature.thermostat import Thermostat # Question: Is this what I should be importing? Or is another automation more ideal?
from pioreactor.exc import CalibrationError # [TODO] Remove this line and see if anything breaks
from pioreactor.utils import local_persistant_storage
class ChemostatAltMedia(DosingAutomationJobContrib, Thermostat):
"""
Chemostat mode - try to keep [nutrient] constant.
"""
automation_name = "scheduled_chemostat_alt_media"
published_settings = {
"volume": {"datatype": "float", "settable": True, "unit": "mL"},
"fraction_alt_media": {"datatype": "float", "settable": True, "unit": "%"},
"duration": {"datatype": "float", "settable": True, "unit": "min"},
"array": {"datatype": "array", "settable": True, "unit": "N/A, mL"},
}
# [TODO] Remove array as a function argument.
def __init__(self, media_ml: float, fraction_alt_media: float, target_temperature, salt_value_array, dose_value_array, array, **kwargs):
super(ChemostatAltMedia, self).__init__(**kwargs)
self.volume = float(media_ml)
self.counter = 0
self.alt_media_ml = 0 # [TODO] Remove this line and see if anything breaks
self.media_ml = 0 # Remove this line and see if anything breaks
self.alt_dosing_schedule = array # [TODO] Remove this line.
self.salt_value_array = salt_value_array # xxxx_value_array stores the future scheduled values
self.dose_value_array = dose_value_array
# xxxx_state_array keeps track of when different features should be run or updated
self.dose_state_array = {
'run_flag': True, 'run_interval': 20, 'run_counter': 0,
'update_flag': True, 'update_interval': 30, 'update_counter': 0,
'run_amount': 0.5, 'previous_run_amount': 0, 'name': 'nutrient_media',
'interval_counter': 0}
self.salt_state_array = {
'run_flag': True, 'run_interval': 8, 'run_counter': 0,
'update_flag': True, 'update_interval': 720, 'update_counter': 0,
'run_amount': 0, 'previous_run_amount': 0, 'name': 'salt_media',
'interval_counter': 0} # 720 update interval -> 12 hours
# self.temp_state_array will control the temperature schedule. Current values are copied over from self.salt_state_array
self.temp_state_array = {
'run_flag': True, 'run_interval': 8, 'run_counter': 0,
'update_flag': True, 'update_interval': 720, 'update_counter': 0,
'run_amount': 0, 'previous_run_amount': 0, 'name': 'temp_control',
'interval_counter': 0} # 720 update interval -> 12 hours
def check_for_update(self, state_array):
# This function takes in a state array and checks whether an update or dosing is required
# [TODO] Add in a check for the type of state array. Temperature or LEDs may 'run' differently than dosings
if self.counter - state_array['run_counter'] >= state_array['run_interval']:
state_array['run_flag'] = True
state_array['run_counter'] = self.counter
if self.counter - state_array['update_counter'] >= state_array['update_interval']:
state_array['update_flag'] = True
state_array['update_counter'] = self.counter
return state_array
def run_dosing(self):
# By default, no media should be added on any given run_dosing()
media_ml = 0
alt_media_ml = 0
waste_ml = 0
# Check what media is scheduled to be added and set the appropriate amount
if self.dose_state_array['run_flag'] == True: # When media is required, set the amount to add and then turn off the flag
media_ml = self.dose_state_array['run_amount']
self.dose_state_array['run_flag'] = False
if self.salt_state_array['run_flag'] == True: # When salt is required, set the amount to add and then turn off the flag
alt_media_ml = self.salt_state_array['run_amount']
self.salt_state_array['run_flag'] = False
# Calculate waste amount and execute dosing
waste_ml = media_ml + alt_media_ml
volume_actually_cycled = self.execute_io_action(alt_media_ml, media_ml, waste_ml)
return volume_actually_cycled
def update_dosing_values(self, state_array):
if state_array['update_flag'] == True:
if state_array['name'] == 'salt_media':
state_array['run_amount'] = self.salt_value_array[0][state_array['interval_counter']]
state_array['run_interval'] = self.salt_value_array[1][state_array['interval_counter']]
state_array['interval_counter'] += 1
elif state_array['name'] == 'nutrient_media':
state_array['run_amount'] = state_array['run_amount'] # Not currently in use.
state_array['update_flag'] = False
return state_array
def update_temperature_values(self, state_array):
# [TODO] Write code to update temperature values here
return
def execute(self) -> events.DilutionEvent:
self.dose_state_array = self.check_for_update(self.dose_state_array) # Flag any values that are due to update
self.dose_state_array = self.update_dosing_values(self.dose_state_array) # Update any values that are flagged for an update
self.salt_state_array = self.check_for_update(self.salt_state_array)
self.salt_state_array = self.update_dosing_values(self.salt_state_array)
## Eventually control temperature schedule ##
# self.temp_state_array = self.check_for_update(self.temp_state_array)
# self.temp_state_array = self.update_temperature_values(self.temp_state_array)
volume_actually_cycled = self.run_dosing() # Run the dosing with updated values
self.counter += 1 # Counter increments every 'duration' (i.e., 1 minute) and is used to track time elapsed since start
return events.DilutionEvent(
f"exchanged {volume_actually_cycled[0]}mL",
data={"volume_actually_cycled": volume_actually_cycled[0]},
)
if __name__ == "__main__":
from pioreactor.background_jobs.dosing_control import DosingController
dc = DosingController(
"scheduled_chemostat_alt_media",
duration=1,
fraction_alt_media=0,
media_ml = 0.6, # currently not in use.
volume=1.0,
target_temperature = 25,
array=[[5, 10, 20, 40, 60, 80], [0.5, 0, 1.0, 0, 1.5, 0]], # [TODO] Remove array as an argument
salt_value_array = [[0, 0.1, 0, 0.1, 0, 0.1, 0, 0.1, 0, 0.1, 0, 0.1, 0, 0.1, 0, 0.1], [60, 55, 55, 25, 25, 15, 15, 10, 10, 7, 7, 5, 5, 3.6, 3.6, 2.5]],
dose_value_array = [],
unit="test_unit",
experiment="test_experiment"
)
dc.block_until_disconnected()