fix tests that were using the wrong od min/maxs

Author: CamDavidsonPilon

pioreactor/background_jobs/od_reading.py

@@ -665,55 +665,43 @@ def hydate_models(self, calibration_data: structs.ODCalibration | None) -> None:
         )
 
     def _hydrate_model(self, calibration_data: structs.ODCalibration) -> Callable[[pt.Voltage], pt.OD]:
-        if calibration_data.curve_type == "poly":
-            """
-            Finds the smallest root in the range [minOD, maxOD] calibrated against.
-
-            Note: if the calibration curve is non-monotonic, ie. has a maximum not on the boundary of the range,
-            this procedure effectively ignores it.
-
-            """
+        if (calibration_data.x != "OD600") or (calibration_data.y != "Voltage"):
+            self.logger.error(f"Calibration {calibration_data.calibration_name} is not for OD600.")
+            raise exc.CalibrationError(f"Calibration {calibration_data.calibration_name} is not for OD600.")
 
-            def calibration(observed_voltage: pt.Voltage) -> pt.OD:
-                assert calibration_data.x == "OD600", "calibration is wrong"
-                assert calibration_data.y == "Voltage", "calibration is wrong"
-                min_OD, max_OD = min(calibration_data.recorded_data["x"]), max(
-                    calibration_data.recorded_data["x"]
-                )
-                min_voltage, max_voltage = min(calibration_data.recorded_data["y"]), max(
-                    calibration_data.recorded_data["y"]
-                )
+        def _calibrate_signal(observed_voltage: pt.Voltage) -> pt.OD:
+            min_OD, max_OD = min(calibration_data.recorded_data["x"]), max(
+                calibration_data.recorded_data["x"]
+            )
+            min_voltage, max_voltage = min(calibration_data.recorded_data["y"]), max(
+                calibration_data.recorded_data["y"]
+            )
 
-                try:
-                    return calibration_data.y_to_x(observed_voltage, enforce_bounds=True)
-                except exc.NoSolutionsFoundError:
-                    if observed_voltage <= min_voltage:
-                        return min_OD
-                    elif observed_voltage > max_voltage:
-                        return max_OD
-                    else:
-                        raise exc.NoSolutionsFoundError(
-                            f"No solution found for calibrated signal. Calibrated for OD=[{min_OD:0.3g}, {max_OD:0.3g}], V=[{min_voltage:0.3g}, {max_voltage:0.3g}]. Observed {observed_voltage:0.3f}V, which would map outside the allowed values."
-                        )
-                except exc.SolutionBelowDomainError:
-                    self.logger.warning(
-                        f"Signal below suggested calibration range. Trimming signal. Calibrated for OD=[{min_OD:0.3g}, {max_OD:0.3g}], V=[{min_voltage:0.3g}, {max_voltage:0.3g}]. Observed {observed_voltage:0.3f}V, which would map outside the allowed values."
-                    )
-                    self.has_logged_warning = True
+            try:
+                return calibration_data.y_to_x(observed_voltage, enforce_bounds=True)
+            except exc.NoSolutionsFoundError:
+                if observed_voltage <= min_voltage:
                     return min_OD
-                except exc.SolutionAboveDomainError:
-                    self.logger.warning(
-                        f"Signal above suggested calibration range. Trimming signal. Calibrated for OD=[{min_OD:0.3g}, {max_OD:0.3g}], V=[{min_voltage:0.3g}, {max_voltage:0.3g}]. Observed {observed_voltage:0.3f}V."
-                    )
-                    self.has_logged_warning = True
+                elif observed_voltage > max_voltage:
                     return max_OD
+                else:
+                    raise exc.NoSolutionsFoundError(
+                        f"No solution found for calibrated signal. Calibrated for OD=[{min_OD:0.3g}, {max_OD:0.3g}], V=[{min_voltage:0.3g}, {max_voltage:0.3g}]. Observed {observed_voltage:0.3f}V, which would map outside the allowed values."
+                    )
+            except exc.SolutionBelowDomainError:
+                self.logger.warning(
+                    f"Signal below suggested calibration range. Trimming signal. Calibrated for OD=[{min_OD:0.3g}, {max_OD:0.3g}], V=[{min_voltage:0.3g}, {max_voltage:0.3g}]. Observed {observed_voltage:0.3f}V, which would map outside the allowed values."
+                )
+                self.has_logged_warning = True
+                return min_OD
+            except exc.SolutionAboveDomainError:
+                self.logger.warning(
+                    f"Signal above suggested calibration range. Trimming signal. Calibrated for OD=[{min_OD:0.3g}, {max_OD:0.3g}], V=[{min_voltage:0.3g}, {max_voltage:0.3g}]. Observed {observed_voltage:0.3f}V."
+                )
+                self.has_logged_warning = True
+                return max_OD
 
-        else:
-
-            def calibration(observed_voltage: pt.Voltage) -> pt.OD:
-                return observed_voltage
-
-        return calibration
+        return _calibrate_signal
 
     def __call__(self, od_readings: structs.ODReadings) -> structs.ODReadings:
         od_readings = copy(od_readings)
@@ -1045,7 +1033,7 @@ def record_from_adc(self) -> structs.ODReadings | None:
 
         try:
             od_readings = self.calibration_transformer(raw_od_readings)
-        except exc.NoSolutionsFoundError:
+        except (exc.NoSolutionsFoundError, exc.CalibrationError):
             # some calibration error occurred
             od_readings = None
 

pioreactor/tests/test_od_reading.py

@@ -876,7 +876,7 @@ def test_calibration_simple_linear_calibration_positive_slope() -> None:
             pause()
             pause()
             voltage = 10.0
-            assert od.calibration_transformer.models["2"](voltage) == 1.0
+            assert od.calibration_transformer.models["2"](voltage) == max(cal.recorded_data["x"])
             pause()
             pause()
             pause()
@@ -1055,7 +1055,7 @@ def float_to_od_readings_struct(ch: pt.PdChannel, v: float) -> structs.ODReading
             ods={ch: structs.RawODReading(od=v, angle="90", channel=ch, timestamp=current_utc_datetime())},
         )
 
-    assert cc(float_to_od_readings_struct("2", 0.001)).ods["2"].od == 0
+    assert cc(float_to_od_readings_struct("2", 0.001)).ods["2"].od == min(cal.recorded_data["x"])
     assert cc(float_to_od_readings_struct("2", 0.002)).ods["2"].od == 0
     assert abs(cc(float_to_od_readings_struct("2", 0.004)).ods["2"].od - 0.0032975807375385234) < 1e-5
     assert abs(cc(float_to_od_readings_struct("2", 0.02)).ods["2"].od - 0.03639585015289039) < 1e-5

pioreactor/whoami.py

@@ -153,7 +153,7 @@ def am_I_a_worker() -> bool:
 
 
 @cache
-def get_pioreactor_version() -> tuple[int, int]:
+def get_pioreactor_version() -> tuple[int, int, *tuple[int, ...]]:
     # pioreactor model version
     if os.environ.get("MODEL_VERSION"):
         return version_text_to_tuple(os.environ["MODEL_VERSION"])