updates

Author: CamDavidsonPilon

CHANGELOG.md

@@ -4,10 +4,14 @@
 
  - new events in the dosing_automation_events table & export detailing when dosing starts and stops.
  - new query pattern for faster Experiment Overview chart loading. However, there is a random element to what data is displayed in a time series. Let me know if this is too distracting.
+ - OD calibrations now support multiple photodiode angles; `pio calibrations run --device od` can emit per-angle calibrations for 45/90/135.
+ - Calibration CLI now saves and can activate multiple OD calibration outputs in one run.
+ - Added an update helper to migrate legacy OD calibrations into per-angle devices.
 
 #### Breaking changes
 
  - Removed `/api/workers/<pioreactor_unit>/configuration`; use `/api/units/<pioreactor_unit>/configuration`.
+ - OD calibration devices are now per-angle (`od45`, `od90`, `od135`). Existing `od` calibration files and active calibration keys must be migrated.
 
 ### 25.12.10
 

config.dev.ini

@@ -52,6 +52,7 @@ Kd=0.0
 1=REF
 2=90
 
+
 [od_reading.config]
 # how many samples should the ADC publish per second?
 samples_per_second=0.2

core/pioreactor/calibrations/__init__.py

@@ -13,26 +13,25 @@
 from msgspec.yaml import decode as yaml_decode
 from msgspec.yaml import encode as yaml_encode
 from pioreactor import structs
-from pioreactor.types import PumpCalibrationDevices
+from pioreactor import types as pt
 from pioreactor.utils import local_persistent_storage
 from pioreactor.whoami import is_testing_env
+from typing import Generic, TypeVar, Literal
+
 
 if not is_testing_env():
     CALIBRATION_PATH = Path("/home/pioreactor/.pioreactor/storage/calibrations/")
 else:
     CALIBRATION_PATH = Path(os.environ["DOT_PIOREACTOR"]) / "storage" / "calibrations"
 
 # Lookup table for different calibration protocols
-Device = str
+Device = TypeVar("Device", bound=str)
 ProtocolName = str
 
 calibration_protocols: dict[Device, dict[ProtocolName, Type[CalibrationProtocol]]] = defaultdict(dict)
 
 
-OD_DEVICES = ["od", "od45", "od90", "od135"]
-
-
-class CalibrationProtocol:
+class CalibrationProtocol(Generic[Device]):
     protocol_name: ProtocolName
     target_device: Device | list[Device]
     description = ""
@@ -53,46 +52,46 @@ def run(
         raise NotImplementedError("Subclasses must implement this method.")
 
 
-class SingleVialODProtocol(CalibrationProtocol):
-    target_device = OD_DEVICES
+class SingleVialODProtocol(CalibrationProtocol[pt.ODCalibrationDevices]):
+    target_device = pt.OD_DEVICES
     protocol_name = "single_vial"
     description = "Calibrate OD using a single vial"
 
-    def run(self, target_device: str, **kwargs) -> structs.OD600Calibration:
+    def run(self, target_device: pt.ODCalibrationDevices, **kwargs) -> structs.OD600Calibration:
         from pioreactor.calibrations.od_calibration_single_vial import run_od_calibration
 
         return run_od_calibration(target_device)
 
 
-class StandardsODProtocol(CalibrationProtocol):
-    target_device = OD_DEVICES
+class StandardsODProtocol(CalibrationProtocol[pt.ODCalibrationDevices]):
+    target_device = pt.OD_DEVICES
     protocol_name = "standards"
     description = "Calibrate OD using standards. Requires multiple vials"
 
     def run(  # type: ignore
-        self, target_device: str, *args, **kwargs
+        self, target_device: pt.ODCalibrationDevices, *args, **kwargs
     ) -> structs.OD600Calibration | list[structs.OD600Calibration]:
         from pioreactor.calibrations.od_calibration_using_standards import run_od_calibration
 
         return run_od_calibration(target_device)
 
 
-class DurationBasedPumpProtocol(CalibrationProtocol):
-    target_device = ["media_pump", "alt_media_pump", "waste_pump"]
+class DurationBasedPumpProtocol(CalibrationProtocol[pt.PumpCalibrationDevices]):
+    target_device = pt.PUMP_DEVICES
     protocol_name = "duration_based"
 
-    def run(self, target_device: str, **kwargs) -> structs.SimplePeristalticPumpCalibration:
+    def run(self, target_device: pt.PumpCalibrationDevices, **kwargs) -> structs.SimplePeristalticPumpCalibration:
         from pioreactor.calibrations.pump_calibration import run_pump_calibration
 
         return run_pump_calibration(target_device)
 
 
-class DCBasedStirringProtocol(CalibrationProtocol):
+class DCBasedStirringProtocol(CalibrationProtocol[Device]):
     target_device = "stirring"
     protocol_name = "dc_based"
 
     def run(
-        self, target_device: str, min_dc: str | None = None, max_dc: str | None = None
+        self, target_device: Device, min_dc: str | None = None, max_dc: str | None = None
     ) -> structs.SimpleStirringCalibration:
         from pioreactor.calibrations.stirring_calibration import run_stirring_calibration
 
@@ -102,13 +101,13 @@ def run(
 
 
 @overload
-def load_active_calibration(device: Literal["od", "od45", "od90", "od135"]) -> structs.ODCalibration | None:
+def load_active_calibration(device: pt.ODCalibrationDevices) -> structs.ODCalibration | None:
     pass
 
 
 @overload
 def load_active_calibration(
-    device: PumpCalibrationDevices,
+    device: pt.PumpCalibrationDevices,
 ) -> structs.SimplePeristalticPumpCalibration | None:
     pass
 

core/pioreactor/calibrations/od_calibration_single_vial.py

@@ -82,7 +82,7 @@ def get_name_from_user() -> str:
 
 
 def get_metadata_from_user(
-    target_device: str,
+    target_device: pt.ODCalibrationDevices,
 ) -> tuple[pt.CalibratedOD, pt.CalibratedOD, pt.mL, pt.PdAngle, pt.PdChannel]:
     if config.get("od_reading.config", "ir_led_intensity") == "auto":
         echo(
@@ -421,7 +421,7 @@ def to_struct(
     return data_blob
 
 
-def run_od_calibration(target_device: str) -> structs.OD600Calibration:
+def run_od_calibration(target_device: pt.ODCalibrationDevices) -> structs.OD600Calibration:
     unit = get_unit_name()
     experiment = get_testing_experiment_name()
     curve_data_: list[float] = []

core/pioreactor/calibrations/od_calibration_using_standards.py

@@ -85,7 +85,7 @@ def get_name_from_user() -> str:
                 return name
 
 
-def get_metadata_from_user(target_device: str) -> dict[pt.PdChannel, pt.PdAngle]:
+def get_metadata_from_user(target_device: pt.ODCalibrationDevices) -> dict[pt.PdChannel, pt.PdAngle]:
     if config.get("od_reading.config", "ir_led_intensity") == "auto":
         echo(
             red(
@@ -317,7 +317,7 @@ def get_voltages_from_adc() -> dict[pt.PdChannel, pt.Voltage]:
     return od600_values, voltages_by_channel
 
 
-def run_od_calibration(target_device: str) -> list[structs.OD600Calibration]:
+def run_od_calibration(target_device: pt.ODCalibrationDevices) -> list[structs.OD600Calibration]:
     unit = get_unit_name()
     experiment = get_testing_experiment_name()
     curve_type = "poly"

core/pioreactor/types.py

@@ -158,5 +158,9 @@ class PublishableSetting(t.TypedDict, total=False):
 type mL = float
 type Seconds = float
 
-# calibration stuff
-type PumpCalibrationDevices = t.Literal["waste_pump", "media_pump", "alt_media_pump"]
+# calibration data
+OD_DEVICES = ["od", "od45", "od90", "od135"]
+PUMP_DEVICES = ["media_pump", "alt_media_pump", "waste_pump"]
+
+type ODCalibrationDevices = t.Literal["od", "od45", "od90", "od135"]
+type PumpCalibrationDevices = t.Literal["media_pump", "alt_media_pump", "waste_pump"]

core/tests/test_od_reading.py

@@ -856,6 +856,42 @@ def test_calibration_not_present() -> None:
         assert len(od.calibration_transformer.models) == 0, od.calibration_transformer.models
 
 
+def test_calibration_duplicate_channel_raises_value_error() -> None:
+    cal_1 = structs.OD600Calibration(
+        created_at=current_utc_datetime(),
+        curve_type="poly",
+        curve_data_=[2.0, 0.0],
+        calibration_name="linear_a",
+        ir_led_intensity=90.0,
+        angle="90",
+        recorded_data={"x": [0, 2], "y": [0, 4]},
+        pd_channel="2",
+        calibrated_on_pioreactor_unit=get_unit_name(),
+    )
+    cal_2 = structs.OD600Calibration(
+        created_at=current_utc_datetime(),
+        curve_type="poly",
+        curve_data_=[1.0, 0.0],
+        calibration_name="linear_b",
+        ir_led_intensity=90.0,
+        angle="90",
+        recorded_data={"x": [0, 1], "y": [0, 1]},
+        pd_channel="2",
+        calibrated_on_pioreactor_unit=get_unit_name(),
+    )
+
+    with pytest.raises(ValueError, match="already hydrated"):
+        start_od_reading(
+            make_channels("REF", "90"),
+            interval=None,
+            fake_data=True,
+            experiment="test_calibration_duplicate_channel_raises_value_error",
+            unit=get_unit_name(),
+            calibration=[cal_1, cal_2],
+            ir_led_intensity=90.0,
+        )
+
+
 def test_calibration_multi_angle_active_calibrations() -> None:
     experiment = "test_calibration_multi_angle_active_calibrations"