generalize to allow multiple od cals

Author: CamDavidsonPilon

AGENTS.md

@@ -71,12 +71,6 @@ make frontend-dev   # Run React dev server on 127.0.0.1:3000
   pytest core/tests/test_cli.py
   ```
 
- * To run just the web backend tests, use
-
-  ```bash
-  pytest core/tests/web/
-  ```
-
 ---
 
 ## Logging

core/pioreactor/background_jobs/od_reading.py

@@ -706,6 +706,10 @@ def hydate_models(self, calibration_data: structs.ODCalibration | None) -> None:
             return
 
         channel = calibration_data.pd_channel
+
+        if channel in self.models:
+            raise ValueError(f"Calibration model for channel {channel} already hydrated.")
+
         cal_type = calibration_data.calibration_type
         calibration_name = calibration_data.calibration_name
 
@@ -1337,7 +1341,7 @@ def start_od_reading(
     fake_data: bool = False,
     unit: pt.Unit | None = None,
     experiment: pt.Experiment | None = None,
-    calibration: bool | structs.ODCalibration | None = True,
+    calibration: bool | structs.ODCalibration | list[structs.ODCalibration] | None = True,
     ir_led_intensity: float | None = None,
 ) -> ODReader:
     """
@@ -1386,7 +1390,14 @@ def start_od_reading(
     calibration_transformer: CalibrationTransformerProtocol
     if calibration is True:
         calibration_transformer = CachedCalibrationTransformer()
-        calibration_transformer.hydate_models(load_active_calibration("od"))
+        for channel, angle in channel_angle_map.items():
+            active_calibration = load_active_calibration(f"od{angle}")  # type: ignore
+            if active_calibration is not None:
+                calibration_transformer.hydate_models(active_calibration)
+    elif isinstance(calibration, list):
+        calibration_transformer = CachedCalibrationTransformer()
+        for calibration_item in calibration:
+            calibration_transformer.hydate_models(calibration_item)
     elif isinstance(calibration, structs.CalibrationBase):
         calibration_transformer = CachedCalibrationTransformer()
         calibration_transformer.hydate_models(calibration)

core/pioreactor/calibrations/__init__.py

@@ -29,6 +29,9 @@
 calibration_protocols: dict[Device, dict[ProtocolName, Type[CalibrationProtocol]]] = defaultdict(dict)
 
 
+OD_DEVICES = ["od", "od45", "od90", "od135"]
+
+
 class CalibrationProtocol:
     protocol_name: ProtocolName
     target_device: Device | list[Device]
@@ -44,30 +47,34 @@ def __init_subclass__(cls, **kwargs):
         else:
             raise ValueError("target_device must be a string or a list of strings")
 
-    def run(self, *args, **kwargs) -> structs.CalibrationBase:
+    def run(
+        self, target_device: str, *args, **kwargs
+    ) -> structs.CalibrationBase | list[structs.CalibrationBase]:
         raise NotImplementedError("Subclasses must implement this method.")
 
 
 class SingleVialODProtocol(CalibrationProtocol):
-    target_device = "od"
+    target_device = OD_DEVICES
     protocol_name = "single_vial"
     description = "Calibrate OD using a single vial"
 
-    def run(self, *args, **kwargs) -> structs.OD600Calibration:
+    def run(self, target_device: str, **kwargs) -> structs.OD600Calibration:
         from pioreactor.calibrations.od_calibration_single_vial import run_od_calibration
 
-        return run_od_calibration()
+        return run_od_calibration(target_device)
 
 
 class StandardsODProtocol(CalibrationProtocol):
-    target_device = "od"
+    target_device = OD_DEVICES
     protocol_name = "standards"
     description = "Calibrate OD using standards. Requires multiple vials"
 
-    def run(self, *args, **kwargs) -> structs.OD600Calibration:
+    def run(  # type: ignore
+        self, target_device: str, *args, **kwargs
+    ) -> structs.OD600Calibration | list[structs.OD600Calibration]:
         from pioreactor.calibrations.od_calibration_using_standards import run_od_calibration
 
-        return run_od_calibration()
+        return run_od_calibration(target_device)
 
 
 class DurationBasedPumpProtocol(CalibrationProtocol):
@@ -95,7 +102,7 @@ def run(
 
 
 @overload
-def load_active_calibration(device: Literal["od"]) -> structs.ODCalibration | None:
+def load_active_calibration(device: Literal["od", "od45", "od90", "od135"]) -> structs.ODCalibration | None:
     pass
 
 

core/pioreactor/calibrations/od_calibration_single_vial.py

@@ -81,7 +81,9 @@ def get_name_from_user() -> str:
                 return name
 
 
-def get_metadata_from_user() -> tuple[pt.CalibratedOD, pt.CalibratedOD, pt.mL, pt.PdAngle, pt.PdChannel]:
+def get_metadata_from_user(
+    target_device: str,
+) -> tuple[pt.CalibratedOD, pt.CalibratedOD, pt.mL, pt.PdAngle, pt.PdChannel]:
     if config.get("od_reading.config", "ir_led_intensity") == "auto":
         echo(
             red(
@@ -131,9 +133,9 @@ def get_metadata_from_user() -> tuple[pt.CalibratedOD, pt.CalibratedOD, pt.mL, p
     confirm(green("Continue?"), abort=True, default=True, prompt_suffix=": ")
 
     pd_channels = config["od_config.photodiode_channel"]
-    ref_channel = next((k for k, v in pd_channels.items() if v == REF_keyword), None)
+    ref_channels = [k for k, v in pd_channels.items() if v == REF_keyword]
 
-    if ref_channel is None:
+    if not ref_channels:
         echo(
             red(
                 "REF required for OD calibration. Set an input to REF in [od_config.photodiode_channel] in your config."
@@ -142,7 +144,41 @@ def get_metadata_from_user() -> tuple[pt.CalibratedOD, pt.CalibratedOD, pt.mL, p
         raise click.Abort()
         # technically it's not required? we just need a specific PD channel to calibrate from.
 
-    pd_channel = cast(pt.PdChannel, "1" if ref_channel == "2" else "2")  # TODO: fix for dr
+    channel_angle_map: dict[pt.PdChannel, pt.PdAngle] = {}
+    for channel, angle in pd_channels.items():
+        if angle in (None, "", REF_keyword):
+            continue
+        channel_angle_map[cast(pt.PdChannel, channel)] = cast(pt.PdAngle, angle)
+
+    if not channel_angle_map:
+        echo(red("Need at least one non-REF PD channel for this calibration."))
+        raise click.Abort()
+
+    channel_choices = sorted(channel_angle_map.keys(), key=int)
+    if target_device != "od":
+        target_angle = target_device.removeprefix("od")
+        channel_choices = [
+            channel for channel in channel_choices if channel_angle_map[channel] == target_angle
+        ]
+        if not channel_choices:
+            echo(
+                red(
+                    f"No channels configured for angle {target_angle}°. Check [od_config.photodiode_channel]."
+                )
+            )
+            raise click.Abort()
+
+    if len(channel_choices) == 1:
+        pd_channel = channel_choices[0]
+    else:
+        pd_channel = cast(
+            pt.PdChannel,
+            prompt(
+                green("Select the PD channel to calibrate"),
+                type=click.Choice(channel_choices),
+                prompt_suffix=": ",
+            ),
+        )
 
     confirm(
         green(
@@ -385,7 +421,7 @@ def to_struct(
     return data_blob
 
 
-def run_od_calibration() -> structs.OD600Calibration:
+def run_od_calibration(target_device: str) -> structs.OD600Calibration:
     unit = get_unit_name()
     experiment = get_testing_experiment_name()
     curve_data_: list[float] = []
@@ -405,7 +441,7 @@ def run_od_calibration() -> structs.OD600Calibration:
             dilution_amount,
             angle,
             pd_channel,
-        ) = get_metadata_from_user()
+        ) = get_metadata_from_user(target_device)
         setup_HDC_instructions()
 
         with start_stirring() as st: