stirring calibration, and removing the old diskcache for persistant s…

…torages

pioreactor/actions/od_calibration.py

@@ -6,7 +6,6 @@
 
 from math import log2
 from time import sleep
-from typing import Callable
 from typing import cast
 from typing import Type
 
@@ -618,18 +617,6 @@ def curve_to_functional_form(curve_type: str, curve_data) -> str:
         raise ValueError()
 
 
-def curve_to_callable(curve_type: str, curve_data) -> Callable:
-    if curve_type == "poly":
-        import numpy as np
-
-        def curve_callable(x):
-            return np.polyval(curve_data, x)
-
-        return curve_callable
-
-    else:
-        raise NotImplementedError
-
 
 def display(name: str | None) -> None:
     def display_from_calibration_blob(data_blob: dict) -> None:

pioreactor/actions/pump.py

@@ -38,9 +38,9 @@
     voltage=-1,
     calibration_name="default_pwm_calibration",
     curve_type="poly",
-    curve_data_ = [1., 0.],
-    recorded_data = {'x': [], 'y': []},
-    calibration_subtype="generic"
+    curve_data_=[1.0, 0.0],
+    recorded_data={"x": [], "y": []},
+    calibration_subtype="generic",
 )
 
 

pioreactor/actions/self_test.py

@@ -397,16 +397,7 @@ def test_positive_correlation_between_rpm_and_stirring(
     assert is_heating_pcb_present(), "Heating PCB was not detected."
     assert voltage_in_aux() <= 18.0, f"Voltage measured {voltage_in_aux()} > 18.0V"
 
-    with local_persistant_storage("stirring_calibration") as cache:
-        if "linear_v1" in cache:
-            parameters = loads(cache["linear_v1"])
-            rpm_coef = parameters["rpm_coef"]
-            intercept = parameters["intercept"]
-
-            initial_dc = rpm_coef * 700 + intercept
-
-        else:
-            initial_dc = config.getfloat("stirring.config", "initial_duty_cycle")
+    initial_dc = config.getfloat("stirring.config", "initial_duty_cycle")
 
     dcs = []
     measured_rpms = []

pioreactor/background_jobs/base.py

@@ -1086,7 +1086,8 @@ def set_enable_dodging_od(self, value: bool):
                 self.logger.debug("Will attempt to dodge later OD readings.")
                 self.set_currently_dodging_od(False)
         else:
-            self.logger.debug("Running continuously through OD readings.")
+            if is_pio_job_running("od_reading"):
+                self.logger.debug("Running continuously through OD readings.")
             self.set_currently_dodging_od(False)
 
     def action_to_do_after_od_reading(self) -> None:

pioreactor/background_jobs/stirring.py

@@ -317,6 +317,7 @@ def initialize_rpm_to_dc_lookup(self) -> Callable:
             return lambda rpm: self._estimate_duty_cycle
 
         assert isinstance(self.target_rpm, float)
+
         with local_persistant_storage("stirring_calibration") as cache:
             if "linear_v1" in cache:
                 self.logger.debug("Found stirring calibration `linear_v1`.")

pioreactor/calibrations/stirring.py → pioreactor/calibrations/stirring_calibration.py

@@ -11,32 +11,30 @@
 
 from pioreactor.background_jobs import stirring
 from pioreactor.config import config
+from pioreactor.exc import JobPresentError
+from pioreactor.hardware import voltage_in_aux
 from pioreactor.logging import create_logger
+from pioreactor.structs import StirringCalibration
 from pioreactor.utils import is_pio_job_running
 from pioreactor.utils import local_persistant_storage
 from pioreactor.utils import managed_lifecycle
 from pioreactor.utils.math_helpers import simple_linear_regression
-from pioreactor.utils.timing import current_utc_timestamp
+from pioreactor.utils.timing import current_utc_datetime
 from pioreactor.whoami import get_assigned_experiment_name
 from pioreactor.whoami import get_testing_experiment_name
 from pioreactor.whoami import get_unit_name
-from pioreactor.structs import StirringCalibration
-from pioreactor.hardware import voltage_in_aux
-from pioreactor.exc import JobPresentError
-
-def run_stirring_calibration(min_dc: int | None = None, max_dc: int | None = None) -> StirringCalibration:
 
 
+def run_stirring_calibration(min_dc: float | None = None, max_dc: float | None = None) -> StirringCalibration:
     if max_dc is None and min_dc is None:
         # seed with initial_duty_cycle
-        config_initial_duty_cycle = config.getfloat("stirring.config", "initial_duty_cycle")
+        config_initial_duty_cycle = config.getfloat("stirring.config", "initial_duty_cycle", fallback=30)
         min_dc, max_dc = round(config_initial_duty_cycle * 0.75), round(config_initial_duty_cycle * 1.33)
     elif (max_dc is not None) and (min_dc is not None):
         assert min_dc < max_dc, "min_dc >= max_dc"
     else:
         raise ValueError("min_dc and max_dc must both be set.")
 
-
     unit = get_unit_name()
     experiment = get_testing_experiment_name()
     action_name = "stirring_calibration"
@@ -53,9 +51,9 @@ def run_stirring_calibration(min_dc: int | None = None, max_dc: int | None = Non
 
         # go up and down to observe any hysteresis.
         dcs = (
-            list(range(max_dc, min_dc, -3))
-            + list(range(min_dc, max_dc, 3))
-            + list(range(max_dc, min_dc - 3, -3))
+              list(range(round(max_dc), round(min_dc), -3))
+            + list(range(round(min_dc), round(max_dc), 3))
+            + list(range(round(max_dc), round(min_dc) - 3, -3))
         )
         n_samples = len(dcs)
 
@@ -74,9 +72,9 @@ def run_stirring_calibration(min_dc: int | None = None, max_dc: int | None = Non
 
             for count, dc in enumerate(dcs, start=1):
                 st.set_duty_cycle(dc)
-                sleep(8)
-                rpm = rpm_calc.estimate(4)
-                measured_rpms.append(rpm)
+                sleep(1)
+                rpm = rpm_calc.estimate(2)
+                measured_rpms.append(dc + 1)
                 logger.debug(f"Detected {rpm=:.1f} RPM @ {dc=}%")
 
                 # log progress
@@ -104,7 +102,7 @@ def run_stirring_calibration(min_dc: int | None = None, max_dc: int | None = Non
         # since in practice, we want a look up from RPM -> required DC, we
         # set x=measure_rpms, y=dcs
         (rpm_coef, rpm_coef_std), (intercept, intercept_std) = simple_linear_regression(
-            filtered_measured_rpms, filtered_dcs
+            filtered_dcs, filtered_measured_rpms
         )
         logger.debug(f"{rpm_coef=}, {rpm_coef_std=}, {intercept=}, {intercept_std=}")
 
@@ -117,12 +115,12 @@ def run_stirring_calibration(min_dc: int | None = None, max_dc: int | None = Non
             raise ValueError("Intercept should be greater than 0.")
 
         return StirringCalibration(
-            hz=config.getfloat("stirring.config", "hz"),
+            pwm_hz=config.getfloat("stirring.config", "pwm_hz"),
             voltage=voltage_in_aux(),
-            calibration_name = f"stirring-calibration-{current_utc_timestamp()}",
+            calibration_name=f"stirring-calibration-{current_utc_datetime().strftime('%Y-%m-%d_%H-%M-%S')}",
             pioreactor_unit=unit,
-            created_at=current_utc_timestamp(),
-            curve_data_= [rpm_coef, intercept],
-            curve_type = "poly",
-            recorded_data={"x": filtered_dcs, "y": filtered_measured_rpms}
-        )
+            created_at=current_utc_datetime(),
+            curve_data_=[rpm_coef, intercept],
+            curve_type="poly",
+            recorded_data={"x": filtered_dcs, "y": filtered_measured_rpms},
+        )

pioreactor/calibrations/utils.py

@@ -0,0 +1,54 @@
+from typing import Callable
+
+
+
+def curve_to_callable(curve_type: str, curve_data: list[float]) -> Callable:
+    if curve_type == "poly":
+        import numpy as np
+
+        def curve_callable(x):
+            return np.polyval(curve_data, x)
+
+        return curve_callable
+
+    else:
+        raise NotImplementedError
+
+
+def plot_data(
+    x: list[float],
+    y: list[float],
+    title: str,
+    x_label: str,
+    y_label: str,
+    x_min=None,
+    x_max=None,
+    interpolation_curve=None,
+    highlight_recent_point=False,
+):
+    import plotext as plt  # type: ignore
+
+    plt.clf()
+
+    if interpolation_curve:
+        plt.plot(sorted(x), [interpolation_curve(x_) for x_ in sorted(x)], color=204)
+        plt.plot_size(145, 26)
+
+    plt.scatter(x, y, marker="hd")
+
+    if highlight_recent_point:
+        plt.scatter([x[-1]], [y[-1]], color=204, marker="hd")
+
+    plt.theme("pro")
+    plt.title(title)
+    plt.xlabel(x_label)
+    plt.ylabel(y_label)
+
+    plt.plot_size(105, 22)
+
+
+    plt.xlim(x_min, x_max)
+    plt.yfrequency(6)
+    plt.xfrequency(6)
+
+    plt.show()