Adds new od calibration from standard action

pioreactor/actions/__init__.py

@@ -4,6 +4,7 @@
 from pioreactor.actions import led_intensity
 from pioreactor.actions import od_blank
 from pioreactor.actions import od_calibration
+from pioreactor.actions import od_calibration_from_standards
 from pioreactor.actions import pump
 from pioreactor.actions import pump_calibration
 from pioreactor.actions import self_test

pioreactor/actions/od_calibration.py

@@ -17,6 +17,7 @@
 from pioreactor import types as pt
 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.config import config
 from pioreactor.config import leader_address
 from pioreactor.mureq import patch
@@ -135,7 +136,13 @@ def start_stirring():
 
 
 def plot_data(
-    x, y, title, x_min=None, x_max=None, interpolation_curve=None, highlight_recent_point=True
+    x,
+    y,
+    title,
+    x_min=None,
+    x_max=None,
+    interpolation_curve=None,
+    highlight_recent_point=True,
 ):
     import plotext as plt  # type: ignore
 
@@ -159,8 +166,13 @@ def plot_data(
 
 
 def start_recording_and_diluting(
-    initial_od600: float, minimum_od600: float, dilution_amount: float, signal_channel
+    st: Stirrer,
+    initial_od600: float,
+    minimum_od600: float,
+    dilution_amount: float,
+    signal_channel,
 ):
+    target_rpm = st.target_rpm
     inferred_od600 = initial_od600
     voltages = []
     inferred_od600s = []
@@ -181,7 +193,10 @@ def start_recording_and_diluting(
         def get_voltage_from_adc() -> float:
             od_readings1 = od_reader.record_from_adc()
             od_readings2 = od_reader.record_from_adc()
-            return 0.5 * (od_readings1.ods[signal_channel].od + od_readings2.ods[signal_channel].od)
+            return 0.5 * (
+                od_readings1.ods[signal_channel].od
+                + od_readings2.ods[signal_channel].od
+            )
 
         for _ in range(4):
             # warm up
@@ -246,15 +261,22 @@ def get_voltage_from_adc() -> float:
                     x_min=minimum_od600,
                     x_max=initial_od600,
                 )
+                st.set_target_rpm(0)
                 click.echo()
                 click.echo(click.style("Stop❗", fg="red"))
                 click.echo("Carefully remove vial.")
-                click.echo("(Optional: take new OD600 reading with external instrument.)")
+                click.echo(
+                    "(Optional: take new OD600 reading with external instrument.)"
+                )
                 click.echo("Reduce volume in vial back to 10ml.")
-                click.echo("Confirm vial outside is dry and clean. Place back into Pioreactor.")
+                click.echo(
+                    "Confirm vial outside is dry and clean. Place back into Pioreactor."
+                )
                 while not click.confirm("Continue?", default=True):
                     pass
                 current_volume_in_vial = initial_volume_in_vial
+                st.set_target_rpm(target_rpm)
+                st.block_until_rpm_is_close_to_target(abs_tolerance=120)
                 sleep(1.0)
 
         click.clear()
@@ -299,7 +321,10 @@ def calculate_curve_of_best_fit(
 
 
 def show_results_and_confirm_with_user(
-    curve_data: list[float], curve_type: str, voltages: list[float], inferred_od600s: list[float]
+    curve_data: list[float],
+    curve_type: str,
+    voltages: list[float],
+    inferred_od600s: list[float],
 ) -> tuple[bool, int]:
     click.clear()
 
@@ -404,14 +429,16 @@ def od_calibration() -> None:
         ) = get_metadata_from_user()
         setup_HDC_instructions()
 
-        with start_stirring():
+        with start_stirring() as st:
             inferred_od600s, voltages = start_recording_and_diluting(
-                initial_od600, minimum_od600, dilution_amount, signal_channel
+                st, initial_od600, minimum_od600, dilution_amount, signal_channel
             )
 
         degree = 4
         while True:
-            curve_data_, curve_type = calculate_curve_of_best_fit(voltages, inferred_od600s, degree)
+            curve_data_, curve_type = calculate_curve_of_best_fit(
+                voltages, inferred_od600s, degree
+            )
             okay_with_result, degree = show_results_and_confirm_with_user(
                 curve_data_, curve_type, voltages, inferred_od600s
             )
@@ -433,7 +460,9 @@ def od_calibration() -> None:
         click.echo(click.style(f"Data for {name}", underline=True, bold=True))
         click.echo(data_blob)
         click.echo()
-        click.echo(click.style(f"Calibration curve for `{name}`", underline=True, bold=True))
+        click.echo(
+            click.style(f"Calibration curve for `{name}`", underline=True, bold=True)
+        )
         click.echo(curve_to_functional_form(curve_type, curve_data_))
         click.echo()
         click.echo(f"Finished calibration of {name} ✅")
@@ -494,8 +523,12 @@ def display_from_calibration_blob(data_blob) -> None:
             ),
         )
         click.echo()
-        click.echo(click.style(f"Calibration curve for `{name}`", underline=True, bold=True))
-        click.echo(curve_to_functional_form(data_blob["curve_type"], data_blob["curve_data_"]))
+        click.echo(
+            click.style(f"Calibration curve for `{name}`", underline=True, bold=True)
+        )
+        click.echo(
+            curve_to_functional_form(data_blob["curve_type"], data_blob["curve_data_"])
+        )
         click.echo()
         click.echo(click.style(f"Data for `{name}`", underline=True, bold=True))
         pprint(data_blob)
@@ -542,14 +575,16 @@ def change_current(name: str) -> None:
     try:
         with local_persistant_storage("od_calibrations") as all_calibrations:
             new_calibration = decode(
-                all_calibrations[name], type=structs.subclass_union(structs.ODCalibration)
+                all_calibrations[name],
+                type=structs.subclass_union(structs.ODCalibration),
             )
 
         angle = new_calibration.angle
         with local_persistant_storage("current_od_calibration") as current_calibrations:
             if angle in current_calibrations:
                 old_calibration = decode(
-                    current_calibrations[angle], type=structs.subclass_union(structs.ODCalibration)
+                    current_calibrations[angle],
+                    type=structs.subclass_union(structs.ODCalibration),
                 )
             else:
                 old_calibration = None
@@ -564,7 +599,9 @@ def change_current(name: str) -> None:
             click.echo("Could not update in database on leader ❌")
 
         if old_calibration:
-            click.echo(f"Replaced {old_calibration.name} with {new_calibration.name}   ✅")
+            click.echo(
+                f"Replaced {old_calibration.name} with {new_calibration.name}   ✅"
+            )
         else:
             click.echo(f"Set {new_calibration.name} to current calibration  ✅")
 
@@ -588,7 +625,9 @@ def list_() -> None:
     with local_persistant_storage("od_calibrations") as c:
         for name in c.iterkeys():
             try:
-                cal = decode(c[name], type=structs.subclass_union(structs.ODCalibration))
+                cal = decode(
+                    c[name], type=structs.subclass_union(structs.ODCalibration)
+                )
                 click.secho(
                     f"{cal.name:15s} {cal.created_at:%d %b, %Y}       {cal.angle:12s} {'✅' if cal.name in current else ''}",
                 )