Multiple instances of RepeatedTimer

Software
1

I have a couple questions.

  1. Can I run multiple instances of RepeatedTimer at the same time (as shown at bottom of code)?
  2. Also, within a function for object obj_1, can I call a function from an unrelated object obj_2? (ex: sc.get_latest_growth_rate() makes a call to dc.latest_growth_rate)
  3. How can I update the dosing controller to use new values? I’m thinking like dc.set_new_values( new_values) or something.

In this code, I want a main loop to run every 24 hours and do some logic based on how my yeast are growing. This main loop will usually be making a change to settings like dosing volume/rate, alt_media dosing volume/rate, and temperature. I need to pull the latest growth rate from the dosing automation and send changes to dosing and temperature automations.

I want a background loop to run every hour and do some logic based on how my yeast are growing. The background loop will only take action if the growth rate drops too low. I will also need to pull the latest growth rate from the dosing automation and sometimes send changes to dosing and temperature automations.

from pioreactor.background_jobs.dosing_control import DosingController
from pioreactor.background_jobs.temperature_control import TemperatureController
from pioreactor.utils.timing import RepeatedTimer
from pioreactor.whoami import get_unit_name, get_latest_experiment_name

Class Schedule():
	def __init__(self, **kwargs):
		pass
		
    def main_loop(self):
        self.current_time = perf_counter()
        growth_rate = self.get_latest_growth_rates()
        if growth_rate > self.min_growth_rate:
			# main loop logic here
			# I will want to change both dc and tc automations
			
    def background_loop(self):
        self.current_time = perf_counter()
        if self.time_since_main_loop() - 3600 >= 0:
            growth_rate = self.get_latest_growth_rate()
			if growth_rate >= min_growth_rate:
				# logic case 1
			if growth_rate < min_growth_rate:
				# logic case 2
			
	def get_latest_growth_rate(self):
		# I want to call dc.latest_growth_rate, is this an acceptable method of calling it? (i.e., no issues with scope?)
		self.latest_growth_rate = dc.latest_growth_rate
		return self.latest_growth_rate

HOURSTOSECONDS = 3600

sc = Schedule()
dc = DosingController(
        "turbidostat",
        duration = 1,
        unit=get_unit_name(),
        experiment=get_latest_experiment_name()
        )
tc = TemperatureController(
        "thermostat",
        target_temperature=30,
        unit=get_unit_name(),
        experiment=get_latest_experiment_name()
        )

main_loop = RepeatedTimer(24*HOURS_TO_SECONDS, sc.main_loop).start()
background_loop = RepeatedTimer(1*HOURS_TO_SECONDS, sc.background_loop).start()

dc.block_until_disconnected()
2

I’ll try to help as best I can.

  1. Yup, multiple RepeatedTimers can run. The way you’ve set this up makes sense. sc.main_loop will run every 24 hours, and sc.background_loop will run every hour.

  2. Yea that should work. By the time get_latest_growth_rate runs, dc is defined in the global namespace, and will be accessible. One note: latest_growth_rate is a property of dc.automation_job. So your code should look like self.latest_growth_rate = dc.automation_job.latest_growth_rate.

  3. Do you mean you want to turbidostat to use new values¹? Like changing the target OD or volume? You should be able to do something like:

    dc.automation_job.volume =  <whatever>
dc.automation_job.target_normalized_od =  <whatever>

    Sometimes instead you need to use a different method (like this, but not for turbidostat.

    So your code might look something like:

        def background_loop(self):
        self.current_time = perf_counter()
        if self.time_since_main_loop() - 3600 >= 0:
            growth_rate = self.get_latest_growth_rate()
			if growth_rate >= min_growth_rate:
				dc.automation_job.target_normalized_od = growth_rate / 2 # I'm making this up.

¹ The dosing controller just controls which automation (turbidostat, chemostat, etc.) is running and how to start and end them - it doesn’t do much handling of data or settings.

3

Yea that should work. By the time get_latest_growth_rate runs, dc is defined in the global namespace, and will be accessible. One note: latest_growth_rate is a property of dc.automation_job . So your code should look like self.latest_growth_rate = dc.automation_job.latest_growth_rate

Thanks. I thought that there might be an issue if I tried issuing the commands to the DosingController but wasn’t sure what else it should be.

Do you mean you want to turbidostat to use new values¹? Like changing the target OD or volume?

Yeah, I want to give new values to the automation. My new turbidostat automation has a media_per_dose and an alt_media_per_dose that I want to control.

Thanks. Btw I am planning on setting up the two new pioreactors I just got in the next day or two. Looking forward to trying to control two of them at once.

4

I want create a DosingController for my custom turbidostat, but I’m not sure how I should be putting in the variables.

What I currently have:

dc = DosingController(
    "turbidostat",
    duration = 1,
    unit=get_unit_name(),
    experiment=get_experiment_name(),
     )

Here is what I call when I start my new turbidostat through the UI:

alt_tracking_turbidostat(skip_first_run=0, media_volume=1, alt_media_volume=1, alt_media_interval=4, alt_media_dose_increase=0.1, duration=0.01, target_normalized_od=2.6)

Can I just copy/paste these values straight into my call to DosingController?

dc = DosingController(
    "alt_tracking_turbidostat",
    skip_first_run=0, media_volume=1, alt_media_volume=1, alt_media_interval=4, 
    alt_media_dose_increase=0.1, duration=0.01, target_normalized_od=2.6,
    unit=get_unit_name(),
    experiment=get_experiment_name(),
     )
5

Yea, that looks right. Let me know if it doesn’t work

6

Yeah, I don’t think it is check inside ~/.pioreactor/plugins/… for the files. I’m getting this error:

Traceback (most recent call last):
File “/home/pioreactor/poet_alg_a.py”, line 82, in
dc = DosingController(
File “/usr/local/lib/python3.9/dist-packages/pioreactor/background_jobs/base.py”, line 87, in call
obj = type.call(cls, *args, **kwargs)
File “/usr/local/lib/python3.9/dist-packages/pioreactor/background_jobs/dosing_control.py”, line 69, in init
raise KeyError(
KeyError: “Unable to find automation alt_turbidostat. Available automations are [‘chemostat’, ‘fed_batch’, ‘morbidostat’, ‘pid_morbidostat’, ‘silent’, ‘turbidostat’]”

I am trying to call DosingController(“alt_turbidostat”,…) which is a currently-working automation that I can run through the pioreactor1.local/pioreactors Dosing Automation UI.

7

Oh this is a new one! I’m guessing your automation AltTrackingTurbidostat is defined in a different python file in ~/.pioreactor/plugins? The way plugins are loaded is file-by-file, so we need to make sure that AltTrackingTurbidostat is defined in a previous file (or the same file…).

Buuuut I don’t force an order for how they are loaded. I should order them by filename, so a user can enumerate them as

01_my_automation.py
02_script.py
...

I’ll fix this in the next version. For now, can you try something ^ above (50% chance of working), or put the AltTrackingTurbidostat automation in the same file?

8

My automation is at ~/.pioreactor/plugins/alt_turbidostat.py and my python script is at ~/my_script.py

How should I handle this?

9

Ah, okay. Hm, this is a common pattern that I’ll need to address, but for now you can do this: at the top of ~/my_script.py, add the following lines:

from pioreactor.plugin_management import get_plugins
get_plugins()

That will force load the contents of ~/.pioreactor/plugins