okay that wasn't so bad

pioreactor/actions/pump.py

@@ -40,7 +40,6 @@
     curve_type="poly",
     curve_data_=[1.0, 0.0],
     recorded_data={"x": [], "y": []},
-    calibration_subtype="generic",
 )
 
 

pioreactor/background_jobs/stirring.py

@@ -18,6 +18,7 @@
 from pioreactor import hardware
 from pioreactor import structs
 from pioreactor.background_jobs.base import BackgroundJobWithDodging
+from pioreactor.calibrations import load_active_calibration
 from pioreactor.config import config
 from pioreactor.utils import clamp
 from pioreactor.utils import is_pio_job_running
@@ -33,7 +34,6 @@
 from pioreactor.whoami import get_assigned_experiment_name
 from pioreactor.whoami import get_unit_name
 from pioreactor.whoami import is_testing_env
-from pioreactor.calibrations import load_active_calibration
 
 if is_testing_env():
     from pioreactor.utils.mock import MockRpmCalculator
@@ -324,10 +324,10 @@ def initialize_rpm_to_dc_lookup(self) -> Callable:
         if possible_calibration is not None:
             self.logger.debug(f"Found stirring calibration: {possible_calibration.calibration_name}.")
 
-            assert len(possible_calibration.curve_data_)  == 2
+            assert len(possible_calibration.curve_data_) == 2
             # invert the linear function.
-            coef = 1.0/possible_calibration.curve_data_[0]
-            intercept = -possible_calibration.curve_data_[1]/possible_calibration.curve_data_[0]
+            coef = 1.0 / possible_calibration.curve_data_[0]
+            intercept = -possible_calibration.curve_data_[1] / possible_calibration.curve_data_[0]
 
             # since we have calibration data, and the initial_duty_cycle could be
             # far off, giving the below equation a bad "first step". We set it here.

pioreactor/calibrations/__init__.py

@@ -1,13 +1,16 @@
 # -*- coding: utf-8 -*-
 from __future__ import annotations
+
 from pathlib import Path
 from typing import Callable
+
 from msgspec import ValidationError
 from msgspec.yaml import decode as yaml_decode
 from msgspec.yaml import encode as yaml_encode
+
 from pioreactor import structs
-from pioreactor.whoami import is_testing_env
 from pioreactor.utils import local_persistant_storage
+from pioreactor.whoami import is_testing_env
 
 if not is_testing_env():
     CALIBRATION_PATH = Path("/home/pioreactor/.pioreactor/storage/calibrations/")
@@ -31,20 +34,36 @@ class ODAssistant(CalibrationAssistant):
     target_calibration_type = "od"
     calibration_struct = structs.ODCalibration
 
-    def __init__(self):
-        pass
 
     def run(self) -> structs.ODCalibration:
         from pioreactor.calibrations.od_calibration import run_od_calibration
 
         return run_od_calibration(od_channel=od_channel)
 
-class PumpAssistant(CalibrationAssistant):
-    target_calibration_type = "pump"
-    calibration_struct = structs.PumpCalibration
 
-    def __init__(self):
-        pass
+class MediaPumpAssistant(CalibrationAssistant):
+    target_calibration_type = "media_pump"
+    calibration_struct = structs.MediaPumpCalibration
+
+    def run(self) -> structs.PumpCalibration:
+        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:
+        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:
         from pioreactor.calibrations.pump_calibration import run_pump_calibration
@@ -56,8 +75,6 @@ class StirringAssistant(CalibrationAssistant):
     target_calibration_type = "stirring"
     calibration_struct = structs.StirringCalibration
 
-    def __init__(self):
-        pass
 
     def run(self, min_dc: str | None = None, max_dc: str | None = None) -> structs.StirringCalibration:
         from pioreactor.calibrations.stirring_calibration import run_stirring_calibration
@@ -67,22 +84,23 @@ def run(self, min_dc: str | None = None, max_dc: str | None = None) -> structs.S
         )
 
 
-def load_active_calibration(cal_type: str, cal_subtype: str | None=None) -> None | structs.AnyCalibration:
-
+def load_active_calibration(cal_type: str) -> None | structs.AnyCalibration:
     with local_persistant_storage("active_calibrations") as c:
-        active_cal_name = c.get((cal_type, cal_subtype))
+        active_cal_name = c.get(cal_type)
 
     if active_cal_name is None:
         return None
 
     return load_calibration(cal_type, active_cal_name)
 
-def load_calibration(cal_type: str, calibration_name: str) -> structs.AnyCalibration:
 
+def load_calibration(cal_type: str, calibration_name: str) -> structs.AnyCalibration:
     target_file = CALIBRATION_PATH / cal_type / f"{calibration_name}.yaml"
 
     if not target_file.exists():
-        raise FileNotFoundError(f"Calibration {calibration_name} was not found in {CALIBRATION_PATH / cal_type}")
+        raise FileNotFoundError(
+            f"Calibration {calibration_name} was not found in {CALIBRATION_PATH / cal_type}"
+        )
 
     assistant = calibration_assistants[cal_type]
 
@@ -91,4 +109,3 @@ def load_calibration(cal_type: str, calibration_name: str) -> structs.AnyCalibra
         return data
     except ValidationError as e:
         raise ValidationError(f"Error reading {target_file.stem()}: {e}")
-

pioreactor/calibrations/od_calibration.py

@@ -24,6 +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.config import config
 from pioreactor.config import leader_address
 from pioreactor.mureq import HTTPErrorStatus
@@ -37,7 +38,7 @@
 from pioreactor.whoami import get_testing_experiment_name
 from pioreactor.whoami import get_unit_name
 from pioreactor.whoami import is_testing_env
-from pioreactor.calibratiions import utils
+
 
 def green(string: str) -> str:
     return style(string, fg="green")
@@ -605,6 +606,3 @@ def run_od_calibration(data_file: str | None) -> None:
                 )
             )
         return
-
-
-

pioreactor/calibrations/pump_calibration.py

@@ -523,5 +523,3 @@ 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}`.")
-
-

pioreactor/cli/calibrations.py

@@ -7,11 +7,11 @@
 from msgspec.yaml import decode as yaml_decode
 from msgspec.yaml import encode as yaml_encode
 
-from pioreactor.calibrations.utils import curve_to_callable
-from pioreactor.calibrations.utils import plot_data
-from pioreactor.calibrations import CALIBRATION_PATH
 from pioreactor.calibrations import calibration_assistants
+from pioreactor.calibrations import CALIBRATION_PATH
 from pioreactor.calibrations import load_calibration
+from pioreactor.calibrations.utils import curve_to_callable
+from pioreactor.calibrations.utils import plot_data
 from pioreactor.utils import local_persistant_storage
 from pioreactor.whoami import is_testing_env
 
@@ -37,16 +37,16 @@ def list_calibrations(cal_type: str):
 
     assistant = calibration_assistants.get(cal_type)
 
-    header = f"{'Name':<50}{'Created At':<25}{'Subtype':<15}{'Active?':<15}"
+    header = f"{'Name':<50}{'Created At':<25}{'Active?':<15}"
     click.echo(header)
     click.echo("-" * len(header))
 
     with local_persistant_storage("active_calibrations") as c:
         for file in calibration_dir.glob("*.yaml"):
             try:
                 data = yaml_decode(file.read_bytes(), type=assistant.calibration_struct)
-                active = c.get((cal_type, data.calibration_subtype)) == data.calibration_name
-                row = f"{data.calibration_name:<50}{data.created_at.strftime('%Y-%m-%d %H:%M:%S'):<25}{data.calibration_subtype or '':<15}{'✅' if active else '':<15}"
+                active = c.get(cal_type) == data.calibration_name
+                row = f"{data.calibration_name:<50}{data.created_at.strftime('%Y-%m-%d %H:%M:%S'):<25}{'✅' if active else '':<15}"
                 click.echo(row)
             except Exception as e:
                 error_message = f"Error reading {file.stem}: {e}"
@@ -65,7 +65,9 @@ def run_calibration(ctx, cal_type: str):
     # Dispatch to the assistant function for that type
     assistant = calibration_assistants.get(cal_type)
     if assistant is None:
-        click.echo(f"No assistant found for calibration type '{cal_type}'. Available types: {list(calibration_assistants.keys())}")
+        click.echo(
+            f"No assistant found for calibration type '{cal_type}'. Available types: {list(calibration_assistants.keys())}"
+        )
         raise click.Abort()
 
     # Run the assistant function to get the final calibration data
@@ -86,8 +88,7 @@ def run_calibration(ctx, cal_type: str):
 
     # make active
     with local_persistant_storage("active_calibrations") as c:
-        c[(cal_type, calibration_data.calibration_subtype)] = calibration_name
-
+        c[cal_type] = calibration_name
 
     click.echo(f"Calibration '{calibration_name}' of type '{cal_type}' saved to {out_file}")
 
@@ -136,7 +137,7 @@ def set_active_calibration(cal_type: str, calibration_name: str | None):
         data = load_calibration(cal_type, calibration_name)
 
         with local_persistant_storage("active_calibrations") as c:
-            c[(data.calibration_type, data.calibration_subtype)] = data.calibration_name
+            c[data.calibration_type] = data.calibration_name
 
 
 @calibration.command(name="delete")
@@ -159,4 +160,3 @@ def delete_calibration(cal_type: str, calibration_name: str):
     click.echo(f"Deleted calibration '{calibration_name}' of type '{cal_type}'.")
 
     # TODO: delete from leader and handle updating active?
-

pioreactor/structs.py

@@ -147,7 +147,6 @@ class CalibrationBase(Struct, tag_field="calibration_type", tag=to_calibration_t
     x: str  # ex: voltage
     y: str  # ex: od600
     recorded_data: dict[t.Literal["x", "y"], list[float]]
-    calibration_subtype: str | None = None
 
     @property
     def calibration_type(self):
@@ -157,8 +156,7 @@ def calibration_type(self):
 class ODCalibration(CalibrationBase, kw_only=True):
     ir_led_intensity: float
 
-
-class PumpCalibration(CalibrationBase, kw_only=True):
+class _PumpCalibration(CalibrationBase, kw_only=True):
     hz: t.Annotated[float, Meta(ge=0)]
     dc: t.Annotated[float, Meta(ge=0)]
     voltage: float
@@ -176,23 +174,27 @@ def duration_to_ml(self, duration: pt.Seconds) -> pt.mL:
         return t.cast(pt.mL, duration * duration_ + bias_)
 
 
-class MediaPumpCalibration(PumpCalibration, kw_only=True):
+class MediaPumpCalibration(_PumpCalibration, kw_only=True):
     pass
 
-class AltMediaPumpCalibration(PumpCalibration, kw_only=True):
+
+class AltMediaPumpCalibration(_PumpCalibration, kw_only=True):
     pass
 
-class MediaPumpCalibration(PumpCalibration, kw_only=True):
+
+class WastePumpCalibration(_PumpCalibration, kw_only=True):
     pass
 
+
 class StirringCalibration(CalibrationBase, kw_only=True):
     pwm_hz: t.Annotated[float, Meta(ge=0)]
     voltage: float
     x: str = "DC %"
     y: str = "RPM"
 
 
-AnyCalibration = t.Union[StirringCalibration, PumpCalibration, ODCalibration]
+AnyCalibration = t.Union[StirringCalibration, MediaPumpCalibration, WastePumpCalibration, AltMediaPumpCalibration, ODCalibration]
+
 
 class Log(JSONPrintedStruct):
     message: str