back to calibrations

pioreactor/actions/__init__.py

@@ -1,10 +1,2 @@
 # -*- coding: utf-8 -*-
 from __future__ import annotations
-
-from pioreactor.actions import led_intensity
-from pioreactor.actions import od_blank
-from pioreactor.actions import pump
-from pioreactor.actions import self_test
-from pioreactor.actions.leader import backup_database
-from pioreactor.actions.leader import experiment_profile
-from pioreactor.actions.leader import export_experiment_data

pioreactor/background_jobs/od_reading.py

@@ -63,6 +63,7 @@
 from pioreactor import whoami
 from pioreactor.background_jobs.base import BackgroundJob
 from pioreactor.background_jobs.base import LoggerMixin
+from pioreactor.calibrations import load_active_calibration
 from pioreactor.config import config
 from pioreactor.hardware import ADC_CHANNEL_FUNCS
 from pioreactor.pubsub import publish
@@ -630,7 +631,7 @@ class NullCalibrationTransformer(CalibrationTransformer):
     def __init__(self) -> None:
         super().__init__()
 
-    def hydate_models_from_disk(self, channel_angle_map: dict[pt.PdChannel, pt.PdAngle]) -> None:
+    def hydate_models_from_disk(self) -> None:
         self.models: dict[pt.PdChannel, Callable] = {}
         return
 
@@ -664,38 +665,29 @@ def __init__(self) -> None:
         super().__init__()
         self.has_logged_warning = False
 
-    def hydate_models_from_disk(self, channel_angle_map: dict[pt.PdChannel, pt.PdAngle]) -> None:
+    def hydate_models_from_disk(self) -> None:
         self.models: dict[pt.PdChannel, Callable] = {}
 
-        with local_persistent_storage("current_od_calibration") as c:
-            for channel, angle in channel_angle_map.items():
-                if angle in c:
-                    calibration_data = decode(c[angle], type=structs.AnyODCalibration)  # type: ignore
-                    name = calibration_data.name
-
-                    if config.get("od_reading.config", "ir_led_intensity") != "auto" and (
-                        calibration_data.ir_led_intensity
-                        != config.getfloat("od_reading.config", "ir_led_intensity")
-                    ):
-                        msg = f"The calibration `{name}` was calibrated with a different IR LED intensity ({calibration_data.ir_led_intensity} vs current: {config.getfloat('od_reading.config', 'ir_led_intensity')}). Either re-calibrate, turn off calibration, or change the ir_led_intensity in the config.ini."
-                        self.logger.error(msg)
-                        raise exc.CalibrationError(msg)
-                    # confirm that PD channel is the same as when calibration was performed
-                    elif calibration_data.pd_channel != channel:
-                        msg = f"The calibration `{name}` was calibrated with a different PD channel ({calibration_data.pd_channel} vs current: {channel})."
-                        self.logger.error(msg)
-                        raise exc.CalibrationError(msg)
-
-                    self.models[channel] = self._hydrate_model(calibration_data)
-                    self.logger.info(f"Using OD calibration `{name}` for channel {channel}.")
-                    self.logger.debug(
-                        f"Using OD calibration `{name}` for channel {channel}, {calibration_data.curve_type=}, {calibration_data.curve_data_=}"
-                    )
+        calibration_data: structs.ODCalibration = load_active_calibration("od")
+        if calibration_data is None:
+            self.logger.debug(f"No calibration available for OD, skipping.")
+            return
 
-                else:
-                    self.logger.debug(
-                        f"No calibration available for channel {channel}, angle {angle}, skipping."
-                    )
+        name = calibration_data.calibration_name
+        channel = calibration_data.pd_channel
+
+        if config.get("od_reading.config", "ir_led_intensity") != "auto" and (
+            calibration_data.ir_led_intensity != config.getfloat("od_reading.config", "ir_led_intensity")
+        ):
+            msg = f"The calibration `{name}` was calibrated with a different IR LED intensity ({calibration_data.ir_led_intensity} vs current: {config.getfloat('od_reading.config', 'ir_led_intensity')}). Either re-calibrate, turn off calibration, or change the ir_led_intensity in the config.ini."
+            self.logger.error(msg)
+            raise exc.CalibrationError(msg)
+
+        self.models[channel] = self._hydrate_model(calibration_data)
+        self.logger.info(f"Using OD calibration `{name}` for channel {channel}.")
+        self.logger.debug(
+            f"Using OD calibration `{name}` for channel {channel}, {calibration_data.curve_type=}, {calibration_data.curve_data_=}"
+        )
 
     def _hydrate_model(self, calibration_data: structs.ODCalibration) -> Callable[[float], float]:
         if calibration_data.curve_type == "poly":
@@ -858,7 +850,7 @@ def __init__(
         self.calibration_transformer.add_external_logger(self.logger)
         self.ir_led_reference_tracker.add_external_logger(self.logger)
 
-        self.calibration_transformer.hydate_models_from_disk(channel_angle_map)
+        self.calibration_transformer.hydate_models_from_disk()
 
         self.channel_angle_map = channel_angle_map
         self.interval = interval

pioreactor/calibrations/__init__.py

@@ -3,6 +3,8 @@
 
 from pathlib import Path
 from typing import Callable
+from typing import Literal
+from typing import overload
 
 from msgspec import ValidationError
 from msgspec.yaml import decode as yaml_decode
@@ -44,7 +46,7 @@ class MediaPumpAssistant(CalibrationAssistant):
     target_calibration_type = "media_pump"
     calibration_struct = structs.MediaPumpCalibration
 
-    def run(self) -> structs.MediaPumpCalibration:
+    def run(self) -> structs.AnyPumpCalibration:  # structs.MediaPumpCalibration:
         from pioreactor.calibrations.pump_calibration import run_pump_calibration
 
         return run_pump_calibration()
@@ -54,7 +56,7 @@ class AltMediaPumpAssistant(CalibrationAssistant):
     target_calibration_type = "alt_media_pump"
     calibration_struct = structs.AltMediaPumpCalibration
 
-    def run(self) -> structs.AltMediaPumpCalibration:
+    def run(self) -> structs.AnyPumpCalibration:  # structs.AltMediaPumpCalibration:
         from pioreactor.calibrations.pump_calibration import run_pump_calibration
 
         return run_pump_calibration()
@@ -64,7 +66,7 @@ class WastePumpAssistant(CalibrationAssistant):
     target_calibration_type = "waste_pump"
     calibration_struct = structs.WastePumpCalibration
 
-    def run(self) -> structs.WastePumpCalibration:
+    def run(self) -> structs.AnyPumpCalibration:  # structs.WastePumpCalibration:
         from pioreactor.calibrations.pump_calibration import run_pump_calibration
 
         return run_pump_calibration()
@@ -82,6 +84,31 @@ def run(self, min_dc: str | None = None, max_dc: str | None = None) -> structs.S
         )
 
 
+@overload
+def load_active_calibration(cal_type: Literal["od"]) -> structs.ODCalibration:
+    pass
+
+
+@overload
+def load_active_calibration(cal_type: Literal["media"]) -> structs.MediaPumpCalibration:
+    pass
+
+
+@overload
+def load_active_calibration(cal_type: Literal["waste"]) -> structs.WastePumpCalibration:
+    pass
+
+
+@overload
+def load_active_calibration(cal_type: Literal["alt_media"]) -> structs.AltMediaPumpCalibration:
+    pass
+
+
+@overload
+def load_active_calibration(cal_type: Literal["stirring"]) -> structs.StirringCalibration:
+    pass
+
+
 def load_active_calibration(cal_type: str) -> None | structs.AnyCalibration:
     with local_persistent_storage("active_calibrations") as c:
         active_cal_name = c.get(cal_type)

pioreactor/calibrations/od_calibration.py

@@ -25,13 +25,14 @@
 from pioreactor.background_jobs.stirring import start_stirring as stirring
 from pioreactor.background_jobs.stirring import Stirrer
 from pioreactor.calibrations import utils
+from pioreactor.calibrations.utils import curve_to_callable
 from pioreactor.config import config
 from pioreactor.config import leader_address
 from pioreactor.mureq import HTTPErrorStatus
 from pioreactor.pubsub import patch_into_leader
 from pioreactor.pubsub import put_into_leader
 from pioreactor.utils import is_pio_job_running
-from pioreactor.utils import local_persistant_storage
+from pioreactor.utils import local_persistent_storage
 from pioreactor.utils import managed_lifecycle
 from pioreactor.utils.timing import current_utc_datestamp
 from pioreactor.utils.timing import current_utc_datetime
@@ -70,7 +71,7 @@ def introduction() -> None:
 
 
 def get_name_from_user() -> str:
-    with local_persistant_storage("od_calibrations") as cache:
+    with local_persistent_storage("od_calibrations") as cache:
         while True:
             name = prompt(
                 green(
@@ -471,65 +472,30 @@ def save_results(
     pd_channel: pt.PdChannel,
     unit: str,
 ) -> structs.ODCalibration:
-    if angle == "45":
-        struct: Type[structs.ODCalibration] = structs.OD45Calibration
-    elif angle == "90":
-        struct = structs.OD90Calibration
-    elif angle == "135":
-        struct = structs.OD135Calibration
-    elif angle == "180":
-        struct = structs.OD180Calibration
-    else:
-        raise ValueError()
-
-    data_blob = struct(
+    data_blob = structs.ODCalibration(
         created_at=current_utc_datetime(),
         pioreactor_unit=unit,
-        name=name,
+        calibration_name=name,
         angle=angle,
         maximum_od600=max(od600s),
         minimum_od600=min(od600s),
         minimum_voltage=min(voltages),
         maximum_voltage=max(voltages),
         curve_data_=curve_data_,
         curve_type=curve_type,
-        voltages=voltages,
-        od600s=od600s,
+        y="od600s",
+        x="voltages",
+        recorded_data={"x": voltages, "y": od600s},
         ir_led_intensity=float(config["od_reading.config"]["ir_led_intensity"]),
         pd_channel=pd_channel,
     )
 
+    data_blob.save_to_disk()
 
     return data_blob
 
 
-def get_data_from_data_file(
-    data_file: str,
-) -> tuple[pt.PdChannel, pt.PdAngle, list[float], list[float], list[float] | None, str | None]:
-    import json
-
-    click.echo(f"Pulling data from {data_file}...")
-
-    with open(data_file, "r") as f:
-        data = json.loads(f.read())
-
-    curve_data_ = data.get("curve_data_", [])
-    curve_type = data.get("curve_type", None)
-
-    ods, voltages = data["od600s"], data["voltages"]
-
-    assert len(ods) == len(voltages), "data must be the same length."
-
-    pd_channel = data.get(
-        "pd_channel",
-        "1" if config["od_config.photodiode_channel_reverse"]["REF"] == "2" else "2",
-    )
-    angle = data.get("angle", str(config["od_config.photodiode_channel"][pd_channel]))
-
-    return pd_channel, angle, ods, voltages, curve_data_, curve_type
-
-
-def run_od_calibration(data_file: str | None) -> structs.ODCalibration:
+def run_od_calibration() -> structs.ODCalibration:
     unit = get_unit_name()
     experiment = get_testing_experiment_name()
     curve_data_ = []  # type: ignore
@@ -539,27 +505,22 @@ def run_od_calibration(data_file: str | None) -> structs.ODCalibration:
         introduction()
         name = get_name_from_user()
 
-        if data_file is None:
-            if any(is_pio_job_running(["stirring", "od_reading"])):
-                echo(red("Both Stirring and OD reading should be turned off."))
-                raise click.Abort()
-
-            (
-                initial_od600,
-                minimum_od600,
-                dilution_amount,
-                angle,
-                pd_channel,
-            ) = get_metadata_from_user()
-            setup_HDC_instructions()
-
-            with start_stirring() as st:
-                inferred_od600s, voltages = start_recording_and_diluting(
-                    st, initial_od600, minimum_od600, dilution_amount, pd_channel
-                )
-        else:
-            pd_channel, angle, inferred_od600s, voltages, curve_data_, curve_type = get_data_from_data_file(  # type: ignore
-                data_file
+        if any(is_pio_job_running(["stirring", "od_reading"])):
+            echo(red("Both Stirring and OD reading should be turned off."))
+            raise click.Abort()
+
+        (
+            initial_od600,
+            minimum_od600,
+            dilution_amount,
+            angle,
+            pd_channel,
+        ) = get_metadata_from_user()
+        setup_HDC_instructions()
+
+        with start_stirring() as st:
+            inferred_od600s, voltages = start_recording_and_diluting(
+                st, initial_od600, minimum_od600, dilution_amount, pd_channel
             )
 
         degree = 5 if len(voltages) > 10 else 3
@@ -574,7 +535,7 @@ def run_od_calibration(data_file: str | None) -> structs.ODCalibration:
 
             curve_data_, curve_type = calculate_curve_of_best_fit(voltages, inferred_od600s, degree)
 
-        echo("Saving results...")
+        echo("Saving results to disk...")
         data_blob = save_results(
             curve_data_,  # type: ignore
             curve_type,  # type: ignore
@@ -585,6 +546,10 @@ def run_od_calibration(data_file: str | None) -> structs.ODCalibration:
             pd_channel,
             unit,
         )
+
+        echo("Setting as the active calibration...")
+        data_blob.set_as_active_calibration()
+
         echo(style(f"Calibration curve for `{name}`", underline=True, bold=True))
         echo(utils.curve_to_functional_form(curve_type, curve_data_))
         echo()