wip

pioreactor/actions/pump.py

@@ -30,13 +30,13 @@
 from pioreactor.whoami import get_assigned_experiment_name
 from pioreactor.whoami import get_unit_name
 
-DEFAULT_PWM_CALIBRATION = structs.PumpCalibration(
+DEFAULT_PWM_CALIBRATION: structs.AnyPumpCalibration = structs._PumpCalibration(
     pioreactor_unit=get_unit_name(),
     created_at=default_datetime_for_pioreactor(),
     hz=200.0,
     dc=100.0,
     voltage=-1,
-    calibration_name="default_pwm_calibration",
+    calibration_name="default_pump_calibration",
     curve_type="poly",
     curve_data_=[1.0, 0.0],
     recorded_data={"x": [], "y": []},

pioreactor/calibrations/__init__.py

@@ -34,38 +34,37 @@ class ODAssistant(CalibrationAssistant):
     target_calibration_type = "od"
     calibration_struct = structs.ODCalibration
 
-
     def run(self) -> structs.ODCalibration:
         from pioreactor.calibrations.od_calibration import run_od_calibration
 
-        return run_od_calibration(od_channel=od_channel)
+        return run_od_calibration()
 
 
 class MediaPumpAssistant(CalibrationAssistant):
     target_calibration_type = "media_pump"
     calibration_struct = structs.MediaPumpCalibration
 
-    def run(self) -> structs.PumpCalibration:
+    def run(self) -> structs.MediaPumpCalibration:
         from pioreactor.calibrations.pump_calibration import run_pump_calibration
 
         return run_pump_calibration()
 
+
 class AltMediaPumpAssistant(CalibrationAssistant):
     target_calibration_type = "alt_media_pump"
     calibration_struct = structs.AltMediaPumpCalibration
 
-
-    def run(self) -> structs.PumpCalibration:
+    def run(self) -> structs.AltMediaPumpCalibration:
         from pioreactor.calibrations.pump_calibration import run_pump_calibration
 
         return run_pump_calibration()
 
+
 class WastePumpAssistant(CalibrationAssistant):
     target_calibration_type = "waste_pump"
     calibration_struct = structs.WastePumpCalibration
 
-
-    def run(self) -> structs.PumpCalibration:
+    def run(self) -> structs.WastePumpCalibration:
         from pioreactor.calibrations.pump_calibration import run_pump_calibration
 
         return run_pump_calibration()
@@ -75,7 +74,6 @@ class StirringAssistant(CalibrationAssistant):
     target_calibration_type = "stirring"
     calibration_struct = structs.StirringCalibration
 
-
     def run(self, min_dc: str | None = None, max_dc: str | None = None) -> structs.StirringCalibration:
         from pioreactor.calibrations.stirring_calibration import run_stirring_calibration
 
@@ -108,4 +106,4 @@ def load_calibration(cal_type: str, calibration_name: str) -> structs.AnyCalibra
         data = yaml_decode(target_file.read_bytes(), type=assistant.calibration_struct)
         return data
     except ValidationError as e:
-        raise ValidationError(f"Error reading {target_file.stem()}: {e}")
+        raise ValidationError(f"Error reading {target_file.stem}: {e}")

pioreactor/calibrations/od_calibration.py

@@ -24,7 +24,7 @@
 from pioreactor.background_jobs.od_reading import start_od_reading
 from pioreactor.background_jobs.stirring import start_stirring as stirring
 from pioreactor.background_jobs.stirring import Stirrer
-from pioreactor.calibratiions import utils
+from pioreactor.calibrations import utils
 from pioreactor.config import config
 from pioreactor.config import leader_address
 from pioreactor.mureq import HTTPErrorStatus
@@ -499,11 +499,6 @@ def save_results(
         pd_channel=pd_channel,
     )
 
-    with local_persistant_storage("od_calibrations") as cache:
-        cache[name] = encode(data_blob)
-
-    publish_to_leader(name)
-    change_current(name)
 
     return data_blob
 
@@ -534,7 +529,7 @@ def get_data_from_data_file(
     return pd_channel, angle, ods, voltages, curve_data_, curve_type
 
 
-def run_od_calibration(data_file: str | None) -> None:
+def run_od_calibration(data_file: str | None) -> structs.ODCalibration:
     unit = get_unit_name()
     experiment = get_testing_experiment_name()
     curve_data_ = []  # type: ignore
@@ -605,4 +600,4 @@ def run_od_calibration(data_file: str | None) -> None:
                     "Currently [od_reading.config][use_calibration] is set to 0 in your config.ini. This should be set to 1 to use calibrations.",
                 )
             )
-        return
+        return data_blob

pioreactor/calibrations/pump_calibration.py

@@ -214,10 +214,10 @@ def setup(pump_type: str, execute_pump: Callable, hz: float, dc: float, unit: st
                 name="calibration",
                 created_at=current_utc_datetime(),
                 pump=pump_type,
-                duration_=1.0,
+                curve_type="poly",
+                curve_data=[1, 0],
                 hz=hz,
                 dc=dc,
-                bias_=0,
                 voltage=voltage_in_aux(),
                 pioreactor_unit=unit,
             ),
@@ -416,7 +416,7 @@ def publish_to_leader(name: str) -> bool:
 
     with local_persistant_storage("pump_calibrations") as all_calibrations:
         calibration_result = decode(
-            all_calibrations[name], type=structs.subclass_union(structs.PumpCalibration)
+            all_calibrations[name], type=structs.AnyPumpCalibration
         )
 
     try:
@@ -445,7 +445,7 @@ def get_data_from_data_file(data_file: str) -> tuple[list[float], list[float], f
     return durations, volumes, hz, dc
 
 
-def run_pump_calibration(min_duration: float, max_duration: float, json_file: str | None) -> None:
+def run_pump_calibration(min_duration: float, max_duration: float, json_file: str | None) -> structs.AnyPumpCalibration:
     unit = get_unit_name()
     experiment = get_assigned_experiment_name(unit)
 
@@ -523,3 +523,4 @@ def run_pump_calibration(min_duration: float, max_duration: float, json_file: st
             logger.warning("Too much uncertainty in slope - you probably want to rerun this calibration...")
 
         echo(f"Finished {pump_type} pump calibration `{name}`.")
+        return data_blob

pioreactor/cli/calibrations.py

@@ -35,7 +35,7 @@ def list_calibrations(cal_type: str):
         click.echo(f"No calibrations found for type '{cal_type}'. Directory does not exist.")
         return
 
-    assistant = calibration_assistants.get(cal_type)
+    assistant = calibration_assistants[cal_type]
 
     header = f"{'Name':<50}{'Created At':<25}{'Active?':<15}"
     click.echo(header)
@@ -131,7 +131,7 @@ def set_active_calibration(cal_type: str, calibration_name: str | None):
     if calibration_name is None:
         click.echo("No calibration name provided. Clearing active calibration.")
         with local_persistant_storage("active_calibrations") as c:
-            c.pop((cal_type, None))
+            c.pop(cal_type)
 
     else:
         data = load_calibration(cal_type, calibration_name)

pioreactor/structs.py

@@ -155,6 +155,14 @@ def calibration_type(self):
 
 class ODCalibration(CalibrationBase, kw_only=True):
     ir_led_intensity: float
+    angle: str
+    pd_channel: str
+    maximum_od600: float
+    minimum_od600: float
+    minimum_voltage: float
+    maximum_voltage: float
+
+
 
 class _PumpCalibration(CalibrationBase, kw_only=True):
     hz: t.Annotated[float, Meta(ge=0)]
@@ -163,6 +171,16 @@ class _PumpCalibration(CalibrationBase, kw_only=True):
     x: str = "duration"
     y: str = "volume"
 
+    @property
+    def duration_(self):
+        assert len(self.curve_data_) == 2
+        return self.curve_data_[1]
+
+    @property
+    def bias_(self):
+        assert len(self.curve_data_) == 2
+        return self.curve_data_[0]
+
     def ml_to_duration(self, ml: pt.mL) -> pt.Seconds:
         duration_ = self.duration_
         bias_ = self.bias_
@@ -193,7 +211,14 @@ class StirringCalibration(CalibrationBase, kw_only=True):
     y: str = "RPM"
 
 
-AnyCalibration = t.Union[StirringCalibration, MediaPumpCalibration, WastePumpCalibration, AltMediaPumpCalibration, ODCalibration]
+AnyCalibration = t.Union[
+    StirringCalibration, MediaPumpCalibration, WastePumpCalibration, AltMediaPumpCalibration, ODCalibration
+]
+
+
+AnyPumpCalibration = t.Union[
+    MediaPumpCalibration, WastePumpCalibration, AltMediaPumpCalibration
+]
 
 
 class Log(JSONPrintedStruct):

pioreactor/tests/test_dosing_automation.py

@@ -45,46 +45,40 @@ def setup_function() -> None:
     with local_persistant_storage("current_pump_calibration") as cache:
         cache["media"] = encode(
             structs.MediaPumpCalibration(
-                name="setup_function",
-                duration_=1.0,
-                bias_=0.0,
+                calibration_name="setup_function",
+                curve_data_=[1.0, 0.0],
+                curve_type="poly",
                 dc=60,
                 hz=100,
                 created_at=datetime(2010, 1, 1, tzinfo=timezone.utc),
                 voltage=-1.0,
-                pump="media",
-                durations=[0, 1],
-                volumes=[0, 1.5],
                 pioreactor_unit=unit,
+                recorded_data={"x": [], 'y': []}
             )
         )
         cache["alt_media"] = encode(
             structs.AltMediaPumpCalibration(
-                name="setup_function",
-                duration_=1.0,
-                bias_=0,
+                calibration_name="setup_function",
+                curve_data_=[1.0, 0.0],
+                curve_type="poly",
+                recorded_data={"x": [], 'y': []},
                 dc=60,
                 hz=100,
                 created_at=datetime(2010, 1, 1, tzinfo=timezone.utc),
                 voltage=-1.0,
-                pump="alt_media",
-                durations=[0, 1],
-                volumes=[0, 1.5],
                 pioreactor_unit=unit,
             )
         )
         cache["waste"] = encode(
             structs.WastePumpCalibration(
-                name="setup_function",
-                duration_=1.0,
-                bias_=0,
+                calibration_name="setup_function",
+                curve_data_=[1.0, 0.0],
+                curve_type="poly",
+                recorded_data={"x": [], 'y': []},
                 dc=60,
                 hz=100,
                 created_at=datetime(2010, 1, 1, tzinfo=timezone.utc),
                 voltage=-1.0,
-                pump="waste",
-                durations=[0, 1],
-                volumes=[0, 1.5],
                 pioreactor_unit=unit,
             )
         )

pioreactor/tests/test_mqtt_to_db_streaming.py

@@ -126,14 +126,14 @@ def test_dosing_events_land_in_db() -> None:
             exp,
             ml=1,
             calibration=structs.MediaPumpCalibration(
-                name="test",
-                duration_=1.0,
-                bias_=0.0,
+                calibration_name="test",
+                curve_data_=[1.0, 0.0],
+                curve_type="poly",
+                recorded_data={"x": [], 'y': []},
                 dc=60,
                 hz=100,
                 created_at=current_utc_datetime(),
                 voltage=-1.0,
-                pump="media",
                 pioreactor_unit=unit,
             ),
         )