Chemostat + Alt Pump automation

Software
1

Hello everyone,

I was wondering if anyone had already written the following dosing automation or plugin before we attempted to write it ourselves:

We want to run our pioreactors as chemostats using the media pump and in addition, run the alt_pump to add an inducer after each of the media exchanges in order to keep the concentration of the inducer constant. Note: (We cannot keep the inducer in the media being fed into the pioreactor).

Seems like something that should be scriptable but was hoping someone in the community had already done the work before having to write it ourselves.

Thank you very much,
Chris

2

:wave: hi @cwilsonPrime!

This is an interesting automation, and I couldn’t help but attempt it myself. Here’s my implementation of the automation.

It looks a lot like a chemostat, but there’s an additional parameter target_inducer_fraction. This is how much alt-media you want in the vial (which is proportional to your concentration of the inducer).

The idea is to perform the usual chemostat operation with media, and then calculate an additional amount of alt-media to dose to adjust the alt_media_fraction (which gets updated in real-time) to equal target_inducer_fraction

# -*- coding: utf-8 -*-
from __future__ import annotations

from pioreactor.automations import events
from pioreactor.automations.dosing.base import DosingAutomationJob
from pioreactor.exc import CalibrationError
from pioreactor.utils import local_persistant_storage


class ChemostatWithInducer(DosingAutomationJob):

    automation_name = "chemostat_with_inducer"
    published_settings = {
        "volume": {"datatype": "float", "settable": True, "unit": "mL"},
        "target_inducer_fraction": {"datatype": "float", "settable": True, "unit": "%"},
    }

    def __init__(self, volume: float | str, target_inducer_fraction: float | str, **kwargs) -> None:
        super().__init__(**kwargs)

        with local_persistant_storage("current_pump_calibration") as cache:
            if "media" not in cache:
                raise CalibrationError("Media pump calibration must be performed first.")
            elif "waste" not in cache:
                raise CalibrationError("Waste pump calibration must be performed first.")
            elif "alt_media" not in cache:
                raise CalibrationError("Alt-media pump calibration must be performed first.")

        self.volume = float(volume)
        self.target_inducer_fraction = float(target_inducer_fraction)

    def execute(self) -> events.DilutionEvent:
        media_exchanged = self.execute_io_action(media_ml=self.volume, waste_ml=self.volume)

        # after this occurs, our alt_media_fraction has been reduced. We need to get it back up to target_inducer_fraction.
        # the math of this:
        # target_inducer_fraction = (alt_media_fraction * vial_volume + delta_alt_media) / (vial_volume + delta_alt_media)
        # solving for delta_alt_media:
        # delta_alt_media = vial_volume * (target_inducer_fraction - alt_media_fraction) / (1 - target_inducer_fraction)
        delta_alt_media = max(self.vial_volume * (self.target_inducer_fraction - self.alt_media_fraction) / (1 - self.target_inducer_fraction), 0)

        # now add that much alt media
        alt_media_exchanged = self.execute_io_action(alt_media_ml=delta_alt_media, waste_ml=delta_alt_media)

        # I _think_ there is a case where if alt_media_exchanged > 0.75, this might fail, however, it would correct on the next run.
        # assert abs(self.alt_media_fraction - self.target_inducer_fraction) < 0.01 # less than 1% error

        return events.DilutionEvent(
            f"exchanged {media_exchanged['media_ml']:.2f}ml of media and {alt_media_exchanged['alt_media_ml']:.2f}mL of alt media",
        )

You can try this out locally. Copy and paste this code into a file in ~/.pioreactor/plugins/chemostat_with_inducer.py,

then run (this won’t activate any pumps, since we are setting TESTING=1).

GLOBAL_CONFIG=~/.pioreactor/config.ini TESTING=1 pio run dosing_control --automation-name chemostat_with_inducer --volume 0.5 --duration 0.75 --target_inducer_fraction 0.05

You may need to install sudo pip install fake-rpi, too…
In a separate SSH shell, you can watch the updates with:

pio mqtt -t pioreactor/+/+/dosing_automation/alt_media_fraction
3

Thanks so much, CamDavidsonPilon, this helps a bunch. I’ll test it and let you know how it goes.

4

Hi Chris (and Cameron),

Have you managed to create this automation and gotten it to work fine? I would be interested in using it if so.

Sorry to bring up such an old post, however an alt_media chemostat function would be great for me :slight_smile:

Best,

Dylan

5

Here’s a more modern dosing automation that attemps to keep inducer concentration constant after each exchange, even when exchange_volume_ml and target inducer concentration change during the run, assuming the alt-media stock is more concentrated than the target.

  1. Add the following code as .py file in ~/.pioreactor/plugins on all the Pioreactors in your cluster you want to run this automation on (let me know if you need a hand with this)
# -*- coding: utf-8 -*-
from pioreactor.automations import events
from pioreactor.automations.dosing.base import DosingAutomationJobContrib
from pioreactor.exc import CalibrationError
from pioreactor.utils import local_persistent_storage

__plugin_name__ = "chemostat_with_inducer"
__plugin_version__ = "0.1.0"
__plugin_summary__ = "Chemostat automation that doses inducer after each media exchange."
__plugin_author__ = "pioreactor-team"

# Add to config if you want defaults:
#
# [dosing_automation.chemostat_with_inducer]
# exchange_volume_ml=1.0
# inducer_volume_ml=0.1


class ChemostatWithInducer(DosingAutomationJobContrib):
    """
    Chemostat mode with an inducer bolus after each media exchange.

    Keeps inducer concentration constant by dosing an inducer bolus after each
    media exchange. This assumes inducer is not present in the media reservoir.
    """

    automation_name = "chemostat_with_inducer"
    published_settings = {
        "exchange_volume_ml": {"datatype": "float", "settable": True, "unit": "mL"},
        "target_inducer_concentration": {"datatype": "float", "settable": True, "unit": "mM"},
        "stock_inducer_concentration": {"datatype": "float", "settable": True, "unit": "mM"},
    }

    def __init__(
        self,
        exchange_volume_ml: float | str,
        target_inducer_concentration: float | str,
        stock_inducer_concentration: float | str,
        **kwargs,
    ) -> None:
        super().__init__(**kwargs)

        with local_persistent_storage("active_calibrations") as cache:
            if "media_pump" not in cache:
                raise CalibrationError("Media pump calibration must be performed first.")
            elif "waste_pump" not in cache:
                raise CalibrationError("Waste pump calibration must be performed first.")
            elif "alt_media_pump" not in cache:
                raise CalibrationError("Alt-media pump calibration must be performed first.")

        self.set_exchange_volume_ml(exchange_volume_ml)
        self.set_target_inducer_concentration(target_inducer_concentration)
        self.set_stock_inducer_concentration(stock_inducer_concentration)

    def set_exchange_volume_ml(self, value: float | str) -> None:
        self.exchange_volume_ml = float(value)
        if self.exchange_volume_ml < 0:
            raise ValueError("exchange_volume_ml must be >= 0.")

    def set_target_inducer_concentration(self, value: float | str) -> None:
        self.target_inducer_concentration = float(value)
        if self.target_inducer_concentration < 0:
            raise ValueError("target_inducer_concentration must be >= 0.")

    def set_stock_inducer_concentration(self, value: float | str) -> None:
        self.stock_inducer_concentration = float(value)
        if self.stock_inducer_concentration <= 0:
            raise ValueError("stock_inducer_concentration must be > 0.")

    def _calculate_inducer_volume_ml(self) -> float:
        if self.target_inducer_concentration == 0:
            return 0.0

        exchange_fraction = self.exchange_volume_ml / self.current_volume_ml
        concentration_ratio = self.stock_inducer_concentration / self.target_inducer_concentration
        denominator = concentration_ratio + exchange_fraction - 1.0
        if denominator <= 0:
            raise ValueError(
                "stock_inducer_concentration is too low to restore target after exchange."
            )

        inducer_volume_ml = self.exchange_volume_ml / denominator
        if inducer_volume_ml < 0:
            raise ValueError("calculated inducer_volume_ml is negative.")

        return inducer_volume_ml

    def execute(self) -> events.DilutionEvent:
        exchange_results = self.execute_io_action(
            media_ml=self.exchange_volume_ml,
            waste_ml=self.exchange_volume_ml,
        )

        inducer_volume_ml = self._calculate_inducer_volume_ml()
        inducer_results = {"alt_media_ml": 0.0}
        if inducer_volume_ml > 0:
            inducer_results = self.execute_io_action(
                alt_media_ml=inducer_volume_ml,
                waste_ml=inducer_volume_ml,
            )

        data = {
            "exchange_volume_ml": self.exchange_volume_ml,
            "target_inducer_concentration": self.target_inducer_concentration,
            "stock_inducer_concentration": self.stock_inducer_concentration,
            "inducer_volume_ml": inducer_volume_ml,
            "current_volume_ml": self.current_volume_ml,
            "volume_actually_cycled_ml": exchange_results["media_ml"],
            "inducer_volume_actually_added_ml": inducer_results["alt_media_ml"],
        }

        return events.DilutionEvent(
            (
                f"exchanged {exchange_results['media_ml']:.2f}mL, "
                f"added {inducer_results['alt_media_ml']:.2f}mL inducer"
            ),
            data=data,
        )
  1. On your leader, add the following .yaml file to ~.pioreactor/plugins/ui/automations/dosing
display_name: Chemostat with inducer bolus
automation_name: chemostat_with_inducer
description: >
  Runs a chemostat using the media pump, then doses inducer via the alt-media pump
  after each exchange to maintain a constant inducer concentration (stock concentration
  assumed constant).
source: chemostat_with_inducer
fields:
  - key: exchange_volume_ml
    default: 1.0
    label: Exchange volume
    unit: mL
    type: numeric
  - key: duration
    default: 20
    label: Time between exchanges
    unit: min
    type: numeric
  - key: target_inducer_concentration
    default: 1.0
    label: Target inducer concentration
    unit: mM
    type: numeric
  - key: stock_inducer_concentration
    default: 100.0
    label: Stock inducer concentration
    unit: mM
    type: numeric

If done correctly, you should see the following in your UI:

Screenshot 2026-02-02 at 9.57.24 AM
Screenshot 2026-02-02 at 9.57.24 AM1622×1444 228 KB