adding support for negative dosing events

Author: CamDavidsonPilon

CHANGELOG.md

@@ -16,10 +16,11 @@ rate_std=0.25
  - New image installs only:
    - updated base OS to the latest 25-05-06 Raspberry Pi OS. The big change is using Linux kernel 6.12.
  - Added the column `hours_since_experiment_created` to dataset exports that details hours since experiment was created.
+ - When a pump runs, it _first_ fires off a dosing_event, which stores information about how much liquid is moved. However, if the pump is stopped early, there was no correction issued to the amount of liquid actually moved. Now, when a pump is stopped early, a _negative_ volume is sent s.t. the delta between the initial amount and new amount is equal to the dosed amount (so when you sum up the volume changes, you get the actual change, as expected).
  - Performance optimizations
 
 #### Bug fixes
- - fixed stirrer not spinning on Pioreactor page (UI) in some cases
+ - fixed stir bar not spinning on Pioreactor page (UI) in some cases
  - alert user if their OD reading is constant before starting the growth-rate calculator, which would break things.
  - alert user if their software is installed in a non-standard location. If so, try `pio uninstall pioreactor -y`.
  - Added a warning if the OD calibration is invalid (ex: a constant line)

pioreactor/actions/pump.py

@@ -299,7 +299,7 @@ def _get_pump_action(pump_device: PumpCalibrationDevices) -> str:
 
             logger.info(_to_human_readable_action(None, duration, pump_device))
         elif continuously:
-            duration = 2.5
+            duration = 1.0
             ml = calibration.duration_to_ml(duration)  # can be wrong if calibration is not defined
 
             logger.info(f"Running {pump_device} continuously.")
@@ -328,11 +328,13 @@ def _get_pump_action(pump_device: PumpCalibrationDevices) -> str:
             if not continuously:
                 pump.by_duration(duration, block=False)
                 # how does this work? What's up with the (or True)?
-                # exit_event.wait returns True iff the event is set, i.e by an interrupt. If we timeout (good path)
-                # then we eval (False or True), hence we break out of this while loop.
+                # exit_event.wait returns True iff the event is set, i.e by an interrupt.
+                # If we timeout (good path), then we eval (False or True), hence we break out of this while loop.
+
                 while not (state.exit_event.wait(duration) or True):
                     pump.interrupt.set()
             else:
+                # continously path
                 pump.continuously(block=False)
 
                 # we only break out of this while loop via a interrupt or MQTT signal => event.set()
@@ -348,9 +350,16 @@ def _get_pump_action(pump_device: PumpCalibrationDevices) -> str:
                 logger.info(f"Stopped {pump_device}.")
 
         if state.exit_event.is_set():
-            # ended early
+            # ended early. We should calculate how much _wasnt_ added, and update that.
             shortened_duration = time.monotonic() - pump_start_time
-            return pump.duration_to_ml(shortened_duration)
+            actually_dosed = pump.duration_to_ml(shortened_duration)
+            logger.info(f"Stopped {pump_device} early.")
+
+            # need to reverse the data that was fired, dosing _can_ be negative, so we publish actually_dosed - ml s.t. ml + (actually_dosed - ml) = actually_dosed.
+            _publish_pump_action(
+                action_name, actually_dosed - ml, unit, experiment, mqtt_client, source_of_event
+            )
+            return actually_dosed
         return ml
 
 

pioreactor/background_jobs/dosing_automation.py

@@ -22,6 +22,7 @@
 from pioreactor.automations.base import AutomationJob
 from pioreactor.config import config
 from pioreactor.logging import create_logger
+from pioreactor.utils import clamp
 from pioreactor.utils import is_pio_job_running
 from pioreactor.utils import local_persistent_storage
 from pioreactor.utils import SummableDict
@@ -92,23 +93,24 @@ def update(
         cls, new_dosing_event: structs.DosingEvent, current_liquid_volume: float, max_volume: float
     ) -> float:
         assert current_liquid_volume >= 0
+        assert max_volume >= 0
         volume, event = float(new_dosing_event.volume_change), new_dosing_event.event
         if event == "add_media":
-            return current_liquid_volume + volume
+            return max(current_liquid_volume + volume, 0)
         elif event == "add_alt_media":
-            return current_liquid_volume + volume
+            return max(current_liquid_volume + volume, 0)
         elif event == "remove_waste":
             if new_dosing_event.source_of_event == "manually":
                 # we assume the user has extracted what they want, regardless of level or tube height.
                 return max(current_liquid_volume - volume, 0.0)
             elif current_liquid_volume <= max_volume:
                 # if the current volume is less than the outflow tube, no liquid is removed
-                return current_liquid_volume
+                return max(current_liquid_volume, 0)
             else:
                 # since we do some additional "removing" after adding, we don't want to
                 # count that as being removed (total volume is limited by position of outflow tube).
                 # hence we keep an lowerbound here.
-                return max(current_liquid_volume - volume, max_volume)
+                return max(current_liquid_volume - volume, max_volume, 0)
         else:
             raise ValueError("Unknown event type")
 
@@ -127,6 +129,7 @@ def update(
         current_liquid_volume: float,
     ) -> float:
         assert 0.0 <= current_alt_media_fraction <= 1.0
+        assert current_liquid_volume >= 0
         volume, event = float(new_dosing_event.volume_change), new_dosing_event.event
 
         if event == "add_media":
@@ -153,14 +156,16 @@ def _update_alt_media_fraction(
         alt_media_delta: float,
         current_liquid_volume: float,
     ) -> float:
-        assert media_delta >= 0
-        assert alt_media_delta >= 0
         total_addition = media_delta + alt_media_delta
 
-        return round(
-            (current_alt_media_fraction * current_liquid_volume + alt_media_delta)
-            / (current_liquid_volume + total_addition),
-            10,
+        return clamp(
+            0.0,
+            round(
+                (current_alt_media_fraction * current_liquid_volume + alt_media_delta)
+                / (current_liquid_volume + total_addition),
+                10,
+            ),
+            1.0,
         )
 
 

pioreactor/logging.py

@@ -99,14 +99,14 @@ class MQTTHandler(logging.Handler):
 
     def __init__(
         self,
-        topic: str,
+        topic_prefix: str,
         client: Client,
         qos: int = 0,
         retain: bool = False,
         **mqtt_kwargs,
     ) -> None:
         logging.Handler.__init__(self)
-        self.topic = topic
+        self.topic_prefix = topic_prefix
         self.qos = qos
         self.retain = retain
         self.mqtt_kwargs = mqtt_kwargs
@@ -124,7 +124,7 @@ def emit(self, record) -> None:
                 return
 
         mqtt_msg = self.client.publish(
-            f"{self.topic}/{record.levelname.lower()}",
+            f"{self.topic_prefix}/{record.levelname.lower()}",
             payload,
             qos=self.qos,
             retain=self.retain,
@@ -198,7 +198,7 @@ def create_logger(
                 "NOTICE": "green",
                 "WARNING": "yellow",
                 "ERROR": "red",
-                "CRITICAL": "red,bg_white",
+                "CRITICAL": "red",
             },
         )
     )
@@ -217,10 +217,10 @@ def create_logger(
         assert pub_client is not None
 
         # create MQTT handlers for logs table
-        topic = (
+        topic_prefix = (
             f"pioreactor/{unit}/{experiment}/logs/{source}"  # NOTE: we later append the log-level, ex: /debug
         )
-        mqtt_to_db_handler = MQTTHandler(topic, pub_client)
+        mqtt_to_db_handler = MQTTHandler(topic_prefix, pub_client)
         mqtt_to_db_handler.setLevel(logging.DEBUG)
         mqtt_to_db_handler.setFormatter(CustomisedJSONFormatter())
 

pioreactor/plugin_management/__init__.py

@@ -34,7 +34,7 @@
 
 Adding to ~/.pioreactor/plugins
 
-   1. Scripts placed in ~/.pioreactor/plugins are automagically loaded.
+   1. Scripts placed in ~/.pioreactor/plugins are automagically loaded in lexicographical order.
      The authors can add metadata to their file with the following variables at the
      highest level in the file:
 

pioreactor/tests/test_dosing_automation.py

@@ -1214,6 +1214,53 @@ def test_liquid_volume_calculator() -> None:
     assert current_volume == 5
 
 
+def test_liquid_volume_calculator_with_negative_values() -> None:
+    # let's start from 0 volume, and start adding.
+    vc = LiquidVolumeCalculator
+    current_volume = 0.0
+    max_volume = 14
+
+    # adding 6ml of media
+    event = DosingEvent(
+        volume_change=6,
+        event="add_media",
+        timestamp=default_datetime_for_pioreactor(0),
+        source_of_event="test",
+    )
+    current_volume = vc.update(event, current_volume, max_volume)
+    assert current_volume == 6
+
+    # testing if we can "back track" the volume if we underdose /  stop early.
+    event = DosingEvent(
+        volume_change=-3,  # test stopping early half-way through, should we report -2?
+        event="add_media",
+        timestamp=default_datetime_for_pioreactor(1),
+        source_of_event="test",
+    )
+    current_volume = vc.update(event, current_volume, max_volume)
+    assert current_volume == 3
+
+    #  okay keep removing?
+    event = DosingEvent(
+        volume_change=-3,
+        event="add_media",
+        timestamp=default_datetime_for_pioreactor(2),
+        source_of_event="test",
+    )
+    current_volume = vc.update(event, current_volume, max_volume)
+    assert current_volume == 0
+
+    #  okay keep removing? Shouldn't go negative!
+    event = DosingEvent(
+        volume_change=-3,
+        event="add_media",
+        timestamp=default_datetime_for_pioreactor(3),
+        source_of_event="test",
+    )
+    current_volume = vc.update(event, current_volume, max_volume)
+    assert current_volume == 0
+
+
 def test_alt_media_calculator_from_0_volume() -> None:
     # let's start from 0, and start adding.
     ac = AltMediaFractionCalculator
@@ -1268,6 +1315,76 @@ def test_alt_media_calculator_from_0_volume() -> None:
     assert current_alt_media_fraction == 0.6
 
 
+def test_alt_media_calculator_from_0_volume_with_negative_doses() -> None:
+    # let's start from 0, and start adding.
+    ac = AltMediaFractionCalculator
+    vc = LiquidVolumeCalculator
+
+    current_volume = 0.0
+    max_volume = 14
+    current_alt_media_fraction = 0.0  # this value doesn't matter, could be anything since volume = 0.
+
+    # adding 6ml of media
+    event = DosingEvent(
+        volume_change=6,
+        event="add_media",
+        timestamp=default_datetime_for_pioreactor(0),
+        source_of_event="test",
+    )
+    current_alt_media_fraction = ac.update(event, current_alt_media_fraction, current_volume)
+    current_volume = vc.update(event, current_volume, max_volume)
+    assert current_alt_media_fraction == 0.0
+
+    # add 6ml alt_media
+    event = DosingEvent(
+        volume_change=6,
+        event="add_alt_media",
+        timestamp=default_datetime_for_pioreactor(1),
+        source_of_event="test",
+    )
+    current_alt_media_fraction = ac.update(event, current_alt_media_fraction, current_volume)
+    current_volume = vc.update(event, current_volume, max_volume)
+    assert current_alt_media_fraction == 0.5
+
+    # two counterfactuals:
+    # A. We added 3ml more alt-media.
+    # B. We added 6ml more alt-media, but stop half way through. => +6ml + -3ml
+
+    # A. add 3ml alt_media
+    event = DosingEvent(
+        volume_change=3,
+        event="add_alt_media",
+        timestamp=default_datetime_for_pioreactor(2),
+        source_of_event="test",
+    )
+    A_current_alt_media_fraction = ac.update(event, current_alt_media_fraction, current_volume)
+    A_current_volume = vc.update(event, current_volume, max_volume)
+    assert A_current_alt_media_fraction == 0.6
+
+    # B. We added 6ml more alt-media, but stop half way through. => +6ml + -3ml
+    event = DosingEvent(
+        volume_change=6,
+        event="add_alt_media",
+        timestamp=default_datetime_for_pioreactor(2),
+        source_of_event="test",
+    )
+    B_current_alt_media_fraction = ac.update(event, current_alt_media_fraction, current_volume)
+    B_current_volume = vc.update(event, current_volume, max_volume)
+
+    event = DosingEvent(
+        volume_change=-3,
+        event="add_alt_media",
+        timestamp=default_datetime_for_pioreactor(2),
+        source_of_event="test",
+    )
+    B_current_alt_media_fraction = ac.update(event, B_current_alt_media_fraction, B_current_volume)
+    B_current_volume = vc.update(event, B_current_volume, max_volume)
+    assert B_current_alt_media_fraction == 0.6
+
+    assert B_current_alt_media_fraction == A_current_alt_media_fraction
+    assert B_current_volume == A_current_volume
+
+
 def test_adding_pumps_and_calling_them_from_execute_io_action() -> None:
     experiment = "test_adding_pumps_and_calling_them_from_execute_io_action"
     unit = get_unit_name()

pioreactor/tests/test_pumps.py

@@ -1,6 +1,7 @@
 # -*- coding: utf-8 -*-
 from __future__ import annotations
 
+import json
 import threading
 import time
 from datetime import datetime
@@ -126,6 +127,13 @@ def join(self):
             threading.Thread.join(self)
             return self._return
 
+    volume_updates = []
+
+    def collect_updates(msg):
+        volume_updates.append(json.loads(msg.payload.decode()))
+
+    subscribe_and_callback(collect_updates, f"pioreactor/{unit}/{exp}/dosing_events", allow_retained=False)
+
     expected_ml = 20
     t = ThreadWithReturnValue(target=add_media, args=(unit, exp, expected_ml), daemon=True)
     t.start()
@@ -141,6 +149,10 @@ def join(self):
     resulting_ml = t.join()
 
     assert resulting_ml < expected_ml
+    assert len(volume_updates) == 2
+    print(volume_updates[0])
+    assert volume_updates[0]["volume_change"] == expected_ml
+    assert -expected_ml < volume_updates[1]["volume_change"] < 0  # fire off a negative volume change
 
 
 def test_continuously_running_pump_will_disconnect_via_mqtt() -> None:

pioreactor/utils/__init__.py

@@ -238,6 +238,7 @@ def __enter__(self) -> Self:
 
     def __exit__(self, *args) -> None:
         self.state = "disconnected"
+        self._exit()
         self.publish_setting("$state", self.state)
         if not self._externally_provided_client:
             assert self.mqtt_client is not None

pioreactor/utils/adcs.py

@@ -113,9 +113,14 @@ def read_from_channel(self, channel: pt.AdcChannel) -> pt.AnalogValue:
             )
 
     def get_firmware_version(self) -> tuple[int, int]:
-        result = bytearray(2)
-        self.i2c.writeto_then_readfrom(hardware.ADC, bytes([0x08]), result)
-        return (result[1], result[0])
+        try:
+            result = bytearray(2)
+            self.i2c.writeto_then_readfrom(hardware.ADC, bytes([0x08]), result)
+            return (result[1], result[0])
+        except OSError:
+            raise exc.HardwareNotFoundError(
+                f"Unable to find i2c channel {hardware.ADC}. Is the HAT attached? Is the firmware loaded?"
+            )
 
     def from_voltage_to_raw(self, voltage: pt.Voltage) -> pt.AnalogValue:
         return int((voltage / 3.3) * 4095 * self.scale)