an event based turbidostat

CamDavidsonPilon · CamDavidsonPilon · commit c86961337419 · 2026-01-31T10:38:47.000-05:00
diff --git a/core/pioreactor/automations/dosing/__init__.py b/core/pioreactor/automations/dosing/__init__.py
@@ -8,4 +8,5 @@
 from .fed_batch import FedBatch
 from .pid_morbidostat import PIDMorbidostat
 from .silent import Silent as DosingSilent
+from .turbidostat import EventBasedTurbidostat
 from .turbidostat import Turbidostat
diff --git a/core/pioreactor/automations/dosing/turbidostat.py b/core/pioreactor/automations/dosing/turbidostat.py
@@ -1,4 +1,6 @@
 # -*- coding: utf-8 -*-
+from threading import Lock
+from threading import Thread
 from typing import cast
 from typing import Optional
 
@@ -138,3 +140,186 @@ def _execute_target_nod(self) -> Optional[events.DilutionEvent]:
             )
         else:
             return None
+
+
+class EventBasedTurbidostat(DosingAutomationJob):
+    """
+    Event-based turbidostat - react immediately to new OD messages instead of polling on a timer.
+
+    Experimental!
+
+    Algorithm:
+      1. Subscribe to OD updates via AutomationJob callbacks.
+      2. On each new OD (or nOD) message, update internal readings.
+      3. If targeting that signal, attempt to acquire a lock and run once.
+      4. If a run is already in progress, skip this trigger (no queuing).
+    """
+
+    automation_name = "turbidostat_event"
+    published_settings = {
+        "exchange_volume_ml": {"datatype": "float", "settable": True, "unit": "mL"},
+        "target_normalized_od": {"datatype": "float", "settable": True, "unit": "AU"},
+        "target_od": {"datatype": "float", "settable": True, "unit": "OD"},
+        "duration": {"datatype": "float", "settable": False, "unit": "min"},
+    }
+    target_od = None
+    target_normalized_od = None
+
+    def __init__(
+        self,
+        exchange_volume_ml: float | str,
+        target_normalized_od: Optional[float | str] = None,
+        target_od: Optional[float | str] = None,
+        **kwargs,
+    ) -> None:
+        self._run_lock = Lock()  # serialize event-triggered runs across MQTT callbacks
+        self._event_based_ready = False  # guard against callbacks firing before init completes
+        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.")
+
+        if target_normalized_od is not None and target_od is not None:
+            raise ValueError("Only provide target nOD or target OD, not both.")
+        elif target_normalized_od is None and target_od is None:
+            raise ValueError("Provide a target nOD or target OD.")
+
+        if target_normalized_od is not None:
+            self.target_normalized_od = float(target_normalized_od)
+        elif target_od is not None:
+            self.target_od = float(target_od)
+
+        self.exchange_volume_ml = float(exchange_volume_ml)
+        self.ema_od = ExponentialMovingAverage(
+            config.getfloat("turbidostat.config", "od_smoothing_ema", fallback=0.5)
+        )
+        self._event_based_ready = True
+
+    def set_duration(self, value: float | None) -> None:
+        # disable periodic polling in event-based mode
+        self.duration = 0.0
+        try:
+            self.run_thread.cancel()  # type: ignore[attr-defined]
+        except AttributeError:
+            pass
+
+    @property
+    def is_targeting_nOD(self) -> bool:
+        return self.target_normalized_od is not None
+
+    @property
+    def _od_channel(self) -> pt.PdChannel:
+        return cast(
+            pt.PdChannel,
+            config.get("turbidostat.config", "signal_channel", fallback="2"),
+        )
+
+    def execute(self) -> Optional[events.DilutionEvent]:
+        if self.is_targeting_nOD:
+            return self._execute_target_nod()
+        else:
+            return self._execute_target_od()
+
+    def set_target_normalized_od(self, new_target: float) -> None:
+        if not self.is_targeting_nOD:
+            self.logger.warning("You are currently targeting OD, and can only change that.")
+        else:
+            self.target_normalized_od = float(new_target)
+
+    def set_target_od(self, new_target: float) -> None:
+        if self.is_targeting_nOD:
+            self.logger.warning("You are currently targeting nOD, and can only change that.")
+        else:
+            self.target_od = float(new_target)
+
+    def _execute_target_od(self) -> Optional[events.DilutionEvent]:
+        assert self.target_od is not None
+        smoothed_od = self.ema_od.update(self.latest_od[self._od_channel])
+        if smoothed_od >= self.target_od:
+            self.ema_od.clear()
+            latest_od_before_dosing = smoothed_od
+            target_od_before_dosing = self.target_od
+
+            results = self.execute_io_action(
+                media_ml=self.exchange_volume_ml, waste_ml=self.exchange_volume_ml
+            )
+
+            data = {
+                "latest_od": latest_od_before_dosing,
+                "target_od": target_od_before_dosing,
+                "exchange_volume_ml": self.exchange_volume_ml,
+                "volume_actually_moved_ml": results["media_ml"],
+            }
+
+            return events.DilutionEvent(
+                f"Latest OD = {latest_od_before_dosing:.2f} ≥ Target OD = {target_od_before_dosing:.2f}; cycled {results['media_ml']:.2f} mL",
+                data,
+            )
+        else:
+            return None
+
+    def _execute_target_nod(self) -> Optional[events.DilutionEvent]:
+        assert self.target_normalized_od is not None
+        if self.latest_normalized_od >= self.target_normalized_od:
+            latest_normalized_od_before_dosing = self.latest_normalized_od
+            target_normalized_od_before_dosing = self.target_normalized_od
+            results = self.execute_io_action(
+                media_ml=self.exchange_volume_ml, waste_ml=self.exchange_volume_ml
+            )
+
+            data = {
+                "latest_normalized_od": latest_normalized_od_before_dosing,
+                "target_normalized_od": target_normalized_od_before_dosing,
+                "exchange_volume_ml": self.exchange_volume_ml,
+                "volume_actually_moved_ml": results["media_ml"],
+            }
+
+            return events.DilutionEvent(
+                f"Latest Normalized OD = {latest_normalized_od_before_dosing:.2f} ≥ Target  nOD = {target_normalized_od_before_dosing:.2f}; cycled {results['media_ml']:.2f} mL",
+                data,
+            )
+        else:
+            return None
+
+    def _queue_event_based_run(self) -> None:
+        if not self._event_based_ready:
+            return
+
+        if self.state != self.READY:
+            return
+
+        if not self._run_lock.acquire(blocking=False):
+            return
+
+        def _runner() -> None:
+            try:
+                self.run(timeout=5.0)
+            finally:
+                self._run_lock.release()
+
+        Thread(target=_runner, daemon=True).start()
+
+    def _set_ods(self, message: pt.MQTTMessage) -> None:
+        if not message.payload:
+            return
+
+        super()._set_ods(message)
+
+        if self.is_targeting_nOD or message.retain:
+            return
+
+        self._queue_event_based_run()
+
+    def _set_normalized_od(self, message: pt.MQTTMessage) -> None:
+        if not message.payload:
+            return
+
+        super()._set_normalized_od(message)
+
+        if (not self.is_targeting_nOD) or message.retain:
+            return
+
+        self._queue_event_based_run()
diff --git a/core/tests/test_turbidostat.py b/core/tests/test_turbidostat.py
@@ -0,0 +1,146 @@
+# -*- coding: utf-8 -*-
+import time
+from datetime import datetime
+from datetime import timezone
+from typing import Callable
+
+from msgspec.json import encode
+from pioreactor import structs
+from pioreactor.automations.dosing.turbidostat import EventBasedTurbidostat
+from pioreactor.utils import get_running_pio_job_id
+from pioreactor.utils.job_manager import JobManager
+from pioreactor.utils.timing import current_utc_datetime
+from pioreactor.whoami import get_unit_name
+
+
+unit = get_unit_name()
+
+
+def wait_for(predicate: Callable[[], bool], timeout: float = 5.0, check_interval: float = 0.01) -> bool:
+    deadline = time.time() + timeout
+    while time.time() < deadline:
+        try:
+            if predicate():
+                return True
+        except Exception:
+            pass
+        time.sleep(check_interval)
+    return False
+
+
+class DummyMessage:
+    def __init__(self, payload: bytes, retain: bool = False, topic: str = "") -> None:
+        self.payload = payload
+        self.retain = retain
+        self.topic = topic
+
+
+def setup_function() -> None:
+    job_id = get_running_pio_job_id("dosing_automation")
+    if job_id is not None:
+        with JobManager() as jm:
+            jm.set_not_running(job_id)
+
+    cal = structs.SimplePeristalticPumpCalibration(
+        calibration_name="setup_function",
+        curve_data_=structs.PolyFitCoefficients(coefficients=[1.0, 0.0]),
+        recorded_data={"x": [], "y": []},
+        dc=60,
+        hz=100,
+        created_at=datetime(2010, 1, 1, tzinfo=timezone.utc),
+        voltage=-1.0,
+        calibrated_on_pioreactor_unit=unit,
+    )
+    cal.set_as_active_calibration_for_device("media_pump")
+    cal.set_as_active_calibration_for_device("alt_media_pump")
+    cal.set_as_active_calibration_for_device("waste_pump")
+
+
+def test_event_based_turbidostat_targeting_nod() -> None:
+    experiment = "test_event_based_turbidostat_targeting_nod"
+    target_nod = 1.0
+    queued_runs = {"count": 0}
+
+    def queue_run() -> None:
+        queued_runs["count"] += 1
+
+    with EventBasedTurbidostat(
+        target_normalized_od=target_nod,
+        exchange_volume_ml=0.1,
+        unit=unit,
+        experiment=experiment,
+    ) as algo:
+        assert wait_for(lambda: algo.state == algo.READY, timeout=5.0)
+        algo._queue_event_based_run = queue_run  # type: ignore[method-assign]
+
+        algo._set_normalized_od(
+            DummyMessage(
+                payload=encode(structs.ODFiltered(od_filtered=1.05, timestamp=current_utc_datetime()))
+            )
+        )
+        assert queued_runs["count"] == 1
+
+        algo._set_ods(
+            DummyMessage(
+                payload=encode(
+                    structs.ODReadings(
+                        timestamp=current_utc_datetime(),
+                        ods={
+                            "2": structs.RawODReading(
+                                ir_led_intensity=80.0,
+                                timestamp=current_utc_datetime(),
+                                angle="45",
+                                od=0.5,
+                                channel="2",
+                            )
+                        },
+                    )
+                )
+            )
+        )
+        assert queued_runs["count"] == 1
+
+
+def test_event_based_turbidostat_targeting_od() -> None:
+    experiment = "test_event_based_turbidostat_targeting_od"
+    target_od = 0.1
+    queued_runs = {"count": 0}
+
+    def queue_run() -> None:
+        queued_runs["count"] += 1
+
+    with EventBasedTurbidostat(
+        target_od=target_od,
+        exchange_volume_ml=0.1,
+        unit=unit,
+        experiment=experiment,
+    ) as algo:
+        assert wait_for(lambda: algo.state == algo.READY, timeout=5.0)
+        algo._queue_event_based_run = queue_run  # type: ignore[method-assign]
+
+        algo._set_ods(
+            DummyMessage(
+                payload=encode(
+                    structs.ODReadings(
+                        timestamp=current_utc_datetime(),
+                        ods={
+                            "2": structs.RawODReading(
+                                ir_led_intensity=80.0,
+                                timestamp=current_utc_datetime(),
+                                angle="45",
+                                od=0.5,
+                                channel="2",
+                            )
+                        },
+                    )
+                )
+            )
+        )
+        assert queued_runs["count"] == 1
+
+        algo._set_normalized_od(
+            DummyMessage(
+                payload=encode(structs.ODFiltered(od_filtered=1.2, timestamp=current_utc_datetime()))
+            )
+        )
+        assert queued_runs["count"] == 1
