drain_refill_turbidostat.py -> ~/.pioreactor/plugins, drain_refill_turbidostat.yaml -> ~/.pioreactor/plugins/ui/automations/dosing/

drain_refill_turbidostat.py

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

from typing import Any

from pioreactor.automations import events
from pioreactor.automations.dosing.turbidostat import Turbidostat
from pioreactor.utils import SummableDict

"""
Suggested config:

[dosing_automation.drain_refill_turbidostat]
target_biomass=1.0
biomass_signal=auto

"""

__plugin_name__ = "drain_refill_turbidostat"
__plugin_summary__ = "Turbidostat variant that drains to N mL, then refills to M mL."
__plugin_version__ = "0.1.0"
__plugin_author__ = "Cam DO"


class DrainRefillTurbidostat(Turbidostat):
    automation_name = "drain_refill_turbidostat"

    baseline_volume_ml: float

    published_settings = {
        **Turbidostat.published_settings,
        "baseline_volume_ml": {"datatype": "float", "settable": False, "unit": "mL"},
    }

    def __init__(
        self,
        baseline_volume_ml: float | str,
        efflux_tube_volume_ml: float | str,
        current_volume_ml: float | str,
        target_biomass: float | str,
        biomass_signal: str,
        **kwargs: Any,
    ) -> None:
        resolved_baseline_volume_ml = float(baseline_volume_ml)
        resolved_current_volume_ml = float(current_volume_ml)

        super().__init__(
            exchange_volume_ml =  0.0,
            target_biomass= target_biomass,
            biomass_signal = biomass_signal,
            current_volume_ml = resolved_current_volume_ml,
            efflux_tube_volume_ml = float(efflux_tube_volume_ml),
            **kwargs,
        )
        self.baseline_volume_ml = resolved_baseline_volume_ml

    def execute(self) -> events.DilutionEvent | None:
        assert self.target_biomass is not None
        resolved_biomass_signal = self.resolved_biomass_signal
        latest_biomass = self.latest_biomass_value(resolved_biomass_signal, od_channel=self._od_channel)

        if latest_biomass < self.target_biomass:
            return None

        starting_volume_ml = self.current_volume_ml
        volumes_moved = self._drain_then_refill()

        data = {
            "latest_biomass": latest_biomass,
            "target_biomass": self.target_biomass,
            "resolved_biomass_signal": resolved_biomass_signal,
            "baseline_volume_ml": self.baseline_volume_ml,
            "efflux_tube_volume_ml": self.efflux_tube_volume_ml,
            "starting_volume_ml": starting_volume_ml,
            "waste_volume_actually_removed_ml": volumes_moved["waste_ml"],
            "media_volume_actually_added_ml": volumes_moved["media_ml"],
        }

        return events.DilutionEvent(
            (
                f"Latest biomass ({resolved_biomass_signal}) = {latest_biomass:.2f} ≥ "
                f"Target biomass = {self.target_biomass:.2f}; drained {volumes_moved['waste_ml']:.2f} mL "
                f"to {self.efflux_tube_volume_ml:.2f} mL, then refilled {volumes_moved['media_ml']:.2f} mL "
                f"to {self.baseline_volume_ml:.2f} mL"
            ),
            data,
        )

    def _drain_then_refill(self) -> SummableDict:
        source_of_event = f"{self.job_name}:{self.automation_name}"
        starting_volume_ml = self.current_volume_ml
        waste_ml = max(starting_volume_ml - self.efflux_tube_volume_ml, 0.0)
        volume_after_drain_ml = starting_volume_ml - waste_ml
        media_ml = max(self.baseline_volume_ml - volume_after_drain_ml, 0.0)

        volumes_moved = SummableDict(waste_ml=0.0, media_ml=0.0)

        if (
            waste_ml > 0
            and self.block_until_not_sleeping()
            and (self.state in (self.READY,))
            and not self._continue_pumping_event.is_set()
        ):
            volumes_moved["waste_ml"] = self.remove_waste_from_bioreactor(
                unit=self.unit,
                experiment=self.experiment,
                ml=waste_ml,
                source_of_event=source_of_event,
                mqtt_client=self.pub_client,
                logger=self.logger,
            )

        if (
            media_ml > 0
            and self.block_until_not_sleeping()
            and (self.state in (self.READY,))
            and not self._continue_pumping_event.is_set()
        ):
            volumes_moved["media_ml"] = self.add_media_to_bioreactor(
                unit=self.unit,
                experiment=self.experiment,
                ml=media_ml,
                source_of_event=source_of_event,
                mqtt_client=self.pub_client,
                logger=self.logger,
            )

        return volumes_moved

drain_refill_turbidostat.yaml

---
display_name: Drain/Refill Turbidostat
automation_name: drain_refill_turbidostat
source: drain_refill_turbidostat
description: >
  Every quarter-minute, check whether the biomass signal is above the target. If so,
  remove waste down to the configured efflux tube level and then refill with fresh
  media back to the configured baseline volume. The efflux tube must be physically
  set to that same level, and the culture should start at the baseline volume.

fields:
  - key: target_biomass
    default: null
    unit: OD/AU
    label: Target
    type: numeric
  - key: biomass_signal
    default: auto
    label: Biomass signal
    type: select
    options:
      - auto
      - od_fused
      - od
      - normalized_od
  - key: baseline_volume_ml
    default: 20.0
    unit: mL
    label: Baseline volume
    type: numeric