move poly logic into struct, which reduces code elsewhere

CHANGELOG.md

@@ -52,7 +52,7 @@
   - GET `/api/workers/<pioreactor_unit>/calibrations/<device>`
   - GET `/unit_api/calibrations/<device>`
   - PATCH `/api/workers/<pioreactor_unit>/active_calibrations/<device>/<cal_name>`
-  - PATCH `/unit_api/calibrations/<device>/<cal_name>/active`
+  - PATCH `/unit_api/active_calibrations/<device>/<cal_name>`
   - DELETE `/api/workers/<pioreactor_unit>/active_calibrations/<device>/<cal_name>`
   - DELETE `/api/workers/<pioreactor_unit>/calibrations/<device>/<cal_name>`
   - DELETE `/unit_api/active_calibrations/<device>/<cal_name>`
@@ -62,26 +62,41 @@
   - PATCH `/api/units/<pioreactor_unit>/plugins/install`
   - PATCH `/api/units/<pioreactor_unit>/plugins/uninstall`
  - Changed the `settings` API (see docs).
- - New `/api/units` that returns a list of units (this is workers & leader)
+ - New `/api/units` that returns a list of units (this is workers & leader). If leader is also a worker, then it's identical to `/api/workers`
  - New `/api/experiments/<experiment>/historical_worker_assignments` that stores historical assignments to experiments
  - New Path API for getting the dir structure of `~/.pioreactor`:
   - `/unit_api/system/path/<path>`
 
 ### Enhancements
- - new SQL table for `historical_experiment_assignments`
+ - new SQL table for `historical_experiment_assignments` that stores historical assignments to experiments.
+ - UI performance improvements
 
 ### Breaking changes
- - any stirring calibrations needs to be redone
+ - removed Python library `diskcache`.
+ - any stirring calibrations needs to be redone. On the command line, run `pio calibration run --device stirring` to start the calibration assistant.
  - fixed typo `utils.local_persistant_storage` to `utils.local_persistent_storage`.
  - Kalman Filter database table is no longer populated. There is a way to re-add it, lmk.
- - moved intermittent cache location to `/tmp/pioreactor_cache/local_intermittent_pioreactor_metadata.sqlite`. This also determined by your configuration.
+ - moved intermittent cache location to `/tmp/pioreactor_cache/local_intermittent_pioreactor_metadata.sqlite`. This also determined by your configuration, see `[storage]`.
  - removed `calibrations` export dataset. Use the export option on the /Calibrations page instead.
  - persistent storage is now on single sqlite3 database in `/home/pioreactor/.pioreactor/storage/local_persistent_pioreactor_metadata.sqlite`. This is configurable in your configuration.
+ - When checking for calibrations in custom Dosing automations, users may have added:
+   ```python
+      with local_persistant_storage("current_pump_calibration") as cache:
+          if "media" not in cache:
+          ...
+   ```
+    This should be updated to:
+    ```python
+        with local_persistent_storage("active_calibrations") as cache:
+            if "media_pump" not in cache:
+            ...
+    ```
+  - removed `pioreactor.utils.gpio_helpers`
 
 ### Bug fixes
  - fix PWM3 not cleaning up correctly
+ - fixed Stirring not updating to best DC % when using a calibration after changing target RPM
 
-   - [ ] test self-test
 
 
 ### 24.12.10

pioreactor/actions/pump.py

@@ -421,8 +421,9 @@ def _get_pump_action(pump_device: PumpCalibrationDevices) -> str:
 
     if waste_calibration != DEFAULT_PWM_CALIBRATION and media_calibration != DEFAULT_PWM_CALIBRATION:
         # provided with calibrations, we can compute if media_rate > waste_rate, which is a danger zone!
-        if media_calibration.duration_ > waste_calibration.duration_:
-            ratio = min(waste_calibration.duration_ / media_calibration.duration_, ratio)
+        # `predict(1)` asks "how much lqd is moved in 1 second"
+        if media_calibration.predict(1) > waste_calibration.predict(1):
+            ratio = min(waste_calibration.predict(1) / media_calibration.predict(1), ratio)
     else:
         logger.warning(
             "Calibrations don't exist for pump(s). Keep an eye on the liquid level to avoid overflowing!"

pioreactor/background_jobs/od_reading.py

@@ -72,6 +72,7 @@
 from pioreactor.utils.streaming_calculations import ExponentialMovingAverage
 from pioreactor.utils.streaming_calculations import ExponentialMovingStd
 from pioreactor.utils.timing import catchtime
+from pioreactor.utils.math_helpers import closest_point_to_domain
 
 ALL_PD_CHANNELS: list[pt.PdChannel] = ["1", "2"]
 VALID_PD_ANGLES: list[pt.PdAngle] = ["45", "90", "135", "180"]
@@ -634,30 +635,6 @@ def hydate_models_from_disk(self) -> None:
         return
 
 
-def closest_point_to_domain(P: list[float], D: tuple[float, float]) -> float:
-    # Unpack the domain D into its lower and upper bounds
-    a, b = D
-
-    # Initialize the closest point and minimum distance
-    closest_point = None
-    min_distance = float("inf")
-
-    for p in P:
-        if a <= p <= b:  # Check if p is within the domain D
-            return p  # If p is within D, it's the closest point with distance 0
-
-        # Calculate the distance to the closest boundary of D
-        distance = min(abs(p - a), abs(p - b))
-
-        # Update the closest point if this distance is smaller than the current min_distance
-        if distance < min_distance:
-            min_distance = distance
-            closest_point = p
-
-    assert closest_point is not None
-    return closest_point
-
-
 class CachedCalibrationTransformer(CalibrationTransformer):
     def __init__(self) -> None:
         super().__init__()
@@ -700,49 +677,28 @@ def _hydrate_model(self, calibration_data: structs.ODCalibration) -> Callable[[f
 
             def calibration(observed_voltage: pt.Voltage) -> pt.OD:
                 poly = calibration_data.curve_data_
-                min_OD, max_OD = calibration_data.minimum_od600, calibration_data.maximum_od600
-                min_voltage, max_voltage = (
-                    calibration_data.minimum_voltage,
-                    calibration_data.maximum_voltage,
-                )
-
-                coef_shift = zeros_like(poly)
-                coef_shift[-1] = observed_voltage
-                solve_for_poly = poly - coef_shift
-                roots_ = roots(solve_for_poly)
-                plausible_ODs_ = sorted([real(r) for r in roots_ if (imag(r) == 0)])
+                min_OD, max_OD = min(calibration_data.recorded_data['y']), max(calibration_data.recorded_data['y'])
+                min_voltage, max_voltage = min(calibration_data.recorded_data['x']), max(calibration_data.recorded_data['x'])
 
-                if len(plausible_ODs_) == 0:
+                try:
+                    return calibration_data.ipredict(observed_voltage)
+                except exc.NoSolutionsFoundError as e:
                     if observed_voltage <= min_voltage:
                         return min_OD
                     elif observed_voltage > max_voltage:
                         return max_OD
-
-                # more than 0 possibilities...
-                # find the closest root to our OD domain (or in the OD domain)
-                ideal_OD = float(closest_point_to_domain(plausible_ODs_, (min_OD, max_OD)))
-
-                if ideal_OD < min_OD:
-                    # voltage less than the blank recorded during the calibration and the calibration curve doesn't have solutions (ex even-deg poly)
-                    # this isn't great, as there is nil noise in the signal.
-
-                    if not self.has_logged_warning:
-                        self.logger.warning(
-                            f"Signal outside suggested calibration range. Trimming signal. Calibrated for OD=[{min_OD:0.3g}, {max_OD:0.3g}], V=[{min_voltage:0.3g}, {max_voltage:0.3g}]. Observed {observed_voltage:0.3f}V."
-                        )
-                        self.has_logged_warning = True
+                except exc.SolutionBelowDomainError:
+                    self.logger.warning(
+                        f"Signal outside suggested calibration range. Trimming signal. Calibrated for OD=[{min_OD:0.3g}, {max_OD:0.3g}], V=[{min_voltage:0.3g}, {max_voltage:0.3g}]. Observed {observed_voltage:0.3f}V."
+                    )
+                    self.has_logged_warning = True
                     return min_OD
-
-                elif ideal_OD > max_OD:
-                    if not self.has_logged_warning:
-                        self.logger.warning(
-                            f"Signal outside suggested calibration range. Trimming signal. Calibrated for OD=[{min_OD:0.3g}, {max_OD:0.3g}], V=[{min_voltage:0.3g}, {max_voltage:0.3g}]. Observed {observed_voltage:0.3f}V."
-                        )
-                        self.has_logged_warning = True
+                except exc.SolutionAboveDomainError:
+                    self.logger.warning(
+                        f"Signal outside suggested calibration range. Trimming signal. Calibrated for OD=[{min_OD:0.3g}, {max_OD:0.3g}], V=[{min_voltage:0.3g}, {max_voltage:0.3g}]. Observed {observed_voltage:0.3f}V."
+                    )
+                    self.has_logged_warning = True
                     return max_OD
-                else:
-                    # happy path
-                    return ideal_OD
 
         else:
 

pioreactor/background_jobs/stirring.py

@@ -316,23 +316,21 @@ def initialize_rpm_to_dc_lookup(self) -> Callable:
         possible_calibration = load_active_calibration("stirring")
 
         if possible_calibration is not None:
-            self.logger.debug(f"Found stirring calibration: {possible_calibration.calibration_name}.")
+            calibration = possible_calibration
+            self.logger.debug(f"Found stirring calibration: {calibration.calibration_name}.")
 
-            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]
+            assert len(calibration.curve_data_) == 2
 
             # 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.
-            self._estimate_duty_cycle = coef * self.target_rpm + intercept
+            self._estimate_duty_cycle = calibration.ipredict(self.target_rpm)
 
             # we scale this by 90% to make sure the PID + prediction doesn't overshoot,
             # better to be conservative here.
             # equivalent to a weighted average: 0.1 * current + 0.9 * predicted
 
             return lambda rpm: self._estimate_duty_cycle - 0.90 * (
-                self._estimate_duty_cycle - (coef * rpm + intercept)
+                self._estimate_duty_cycle - (calibration.ipredict(rpm))
             )
         else:
             return lambda rpm: self._estimate_duty_cycle

pioreactor/calibrations/od_calibration.py

@@ -471,10 +471,6 @@ def save_results(
         calibrated_on_pioreactor_unit=unit,
         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,
         y="od600s",

pioreactor/calibrations/utils.py

@@ -2,8 +2,41 @@
 from __future__ import annotations
 
 from typing import Callable
+import click
 
 
+
+def green(string: str) -> str:
+    return click.style(string, fg="green")
+
+def red(string: str) -> str:
+    return click.style(string, fg="red")
+
+
+def bold(string: str) -> str:
+    return click.style(string, bold=True)
+
+
+def calculate_curve_of_best_fit(
+    x: list[float], y: list[float], degree: int
+) -> tuple[list[float], str]:
+    import numpy as np
+
+    # weigh the last point, the "blank measurement", more.
+    # 1. It's far away from the other points
+    # 2. We have prior knowledge that OD~0 when V~0.
+    n = len(voltages)
+    weights = np.ones_like(voltages)
+    weights[-1] = n / 2
+
+    try:
+        coefs = np.polyfit(inferred_od600s, voltages, deg=degree, w=weights).tolist()
+    except Exception:
+        echo("Unable to fit.")
+        coefs = np.zeros(degree).tolist()
+
+    return coefs, "poly"
+
 def curve_to_functional_form(curve_type: str, curve_data) -> str:
     if curve_type == "poly":
         d = len(curve_data)
@@ -63,3 +96,51 @@ def plot_data(
     plt.xfrequency(6)
 
     plt.show()
+
+
+
+def crunch_data_and_confirm_with_user(
+    calibration
+) -> bool:
+
+    click.clear()
+
+    y, x = calibration.recorded_data["y"], calibration.recorded_data["x"]
+    candidate_curve = calibration.curve_data_
+
+    while True:
+
+        if candidate_curve is not None:
+            degree = 1
+            candidate_curve = calculate_curve_of_best_fit(x, y, degree)
+
+
+        curve_callable = curve_to_callable("poly", candidate_curve)
+        plot_data(
+            x,
+            y,
+            interpolation_curve=curve_callable,
+            highlight_recent_point=False,
+        )
+        click.echo()
+        click.echo(f"Calibration curve: {curve_to_functional_form(curve_type, candidate_curve)}")
+        r = click.prompt(
+            green(
+                f"""
+    y: confirm and save to disk
+    n: exit completely
+    d: choose a new degree for polynomial fit (currently {len(candidate_curve)-1})
+
+    """
+            ),
+            type=click.Choice(["y", "n", "d"]),
+        )
+        if r == "y":
+            calibration.curve_data_ = candidate_curve
+            return True
+        elif r == "n":
+            return False
+        elif r == "d":
+            degree = click.prompt(green("Enter new degree"), type=click.IntRange(1, 5, clamp=True))
+        else:
+            return False

pioreactor/cli/calibrations.py

@@ -191,6 +191,31 @@ def delete_calibration(device: str, calibration_name: str) -> None:
         raise click.Abort()
 
     target_file.unlink()
+
+    with local_persistent_storage("active_calibrations") as c:
+        is_present = c.get(device) == calibration_name
+        if is_present:
+            c.pop(device)
+
+
     click.echo(f"Deleted calibration '{calibration_name}' of device '{device}'.")
 
-    # TODO: delete from leader and handle updating active?
+@calibration.command(name="analyze")
+@click.option("--device", required=True, help="Which calibration device to delete from.")
+@click.option("--name", "calibration_name", required=True, help="Which calibration name to delete.")
+def analyse_calibration(device: str, calibration_name: str) -> None:
+    """
+    Delete a calibration file from local storage.
+
+    Example usage:
+      calibration delete --device od --name my_od_cal_v1
+    """
+    target_file = CALIBRATION_PATH / device / f"{calibration_name}.yaml"
+    if not target_file.exists():
+        click.echo(f"No such calibration file: {target_file}")
+        raise click.Abort()
+
+
+    data = load_calibration(device, calibration_name)
+    finsished, degree = show_results_and_confirm_with_user(...)
+    # TODO finish this

pioreactor/exc.py

@@ -72,3 +72,19 @@ class RsyncError(OSError):
     """
     Syncing files failed
     """
+
+class NoSolutionsFoundError(ValueError):
+    """
+    No solutions found
+    """
+
+
+class SolutionBelowDomainError(ValueError):
+    """
+    Outside minimum range
+    """
+
+class SolutionAboveDomainError(ValueError):
+    """
+    Outside maximum range
+    """