introducing splines to calibrations

Author: CamDavidsonPilon

CHANGELOG.md

@@ -19,6 +19,10 @@
  - Self-test results now surface per-check pass/fail status in the Inventory page and support retrying failed checks.
  - Removed redundant `from __future__ import annotations` usage now that we run on Python 3.13.
  - When a Pioreactor model is changed, a (non-blocking) hardware check is performed.
+ - You can now restart the web server (lighttpd), and the background task queue, Huey, from the UI. Go to Leader -> "Long-running jobs", and see the "web server and queue" line.
+ - Added spline curve support for calibrations, including OD standards sessions and calibration charts.
+ - Added polynomial/spline curve utilities and JS curve helpers to keep UI plots in sync with backend curve types.
+ - `pio calibrations analyze` now supports `--fit poly|spline` (default poly).
 
 #### Breaking changes
  - Moved Self-test to the Inventory page. Pioreactors no longer need to be assigned to an experiment to run self-test.
@@ -36,6 +40,7 @@
  - Fix floating point error at the boundary of OD calibrations.
  - Fix runtime forward-reference errors in type annotations after dropping `__future__` imports.
  - Fix timeouts being too short on some UI export operations
+ - Re-save calibration files on `pio calibrations analyze` confirmation even when the curve is unchanged.
 
 ### 25.12.10
 

core/pioreactor/calibrations/protocols/od_reference_standard.py

@@ -256,8 +256,25 @@ def get_reference_standard_step(
     return get_session_step(_REFERENCE_STANDARD_STEPS, session, executor)
 
 
+def get_valid_od_devices_for_this_unit() -> list[str]:
+
+    pd_channels = config["od_config.photodiode_channel"]
+    valid_devices: list[pt.ODCalibrationDevices] = []
+
+    for _, angle in pd_channels.items():
+        if angle in (None, "", REF_keyword):
+            continue
+        device = f"od{angle}"
+        if device in pt.OD_DEVICES and device not in valid_devices:
+            valid_devices.append(cast(pt.ODCalibrationDevices, device))
+
+    if "od90" in valid_devices and "od45" in valid_devices and "od135" in valid_devices:
+        valid_devices.append("od")
+    return valid_devices
+
+
 class ODReferenceStandardProtocol(CalibrationProtocol[pt.ODCalibrationDevices]):
-    target_device = pt.OD_DEVICES
+    target_device = get_valid_od_devices_for_this_unit()
     protocol_name = "od_reference_standard"
     title = "Optics calibration jig"
     description = (

core/pioreactor/calibrations/protocols/od_single_vial.py

@@ -5,6 +5,7 @@
 from math import log2
 from time import sleep
 from typing import cast
+from typing import TYPE_CHECKING
 
 import click
 from click import clear
@@ -18,8 +19,6 @@
 from pioreactor.background_jobs.od_reading import average_over_od_readings
 from pioreactor.background_jobs.od_reading import REF_keyword
 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.calibrations import utils
 from pioreactor.calibrations.cli_helpers import action_block
 from pioreactor.calibrations.cli_helpers import green
@@ -37,6 +36,9 @@
 from pioreactor.whoami import get_unit_name
 from pioreactor.whoami import is_testing_env
 
+if TYPE_CHECKING:
+    from pioreactor.background_jobs.stirring import Stirrer
+
 
 def introduction() -> None:
     import logging
@@ -204,6 +206,8 @@ def setup_HDC_instructions() -> None:
 
 
 def start_stirring():
+    from pioreactor.background_jobs.stirring import start_stirring as stirring
+
     while not confirm(green("Ready to start stirring?"), default=True, abort=True, prompt_suffix=": "):
         pass
 
@@ -219,7 +223,7 @@ def start_stirring():
 
 
 def start_recording_and_diluting(
-    st: Stirrer,
+    st: "Stirrer",
     initial_od600: pt.OD,
     minimum_od600: pt.OD,
     dilution_amount: float,

core/pioreactor/calibrations/protocols/od_standards.py

@@ -20,7 +20,6 @@
 from pioreactor.background_jobs.od_reading import average_over_od_readings
 from pioreactor.background_jobs.od_reading import REF_keyword
 from pioreactor.background_jobs.od_reading import start_od_reading
-from pioreactor.background_jobs.stirring import start_stirring as stirring
 from pioreactor.calibrations import list_of_calibrations_by_device
 from pioreactor.calibrations import utils
 from pioreactor.calibrations.registry import CalibrationProtocol
@@ -137,6 +136,8 @@ def _measure_standard(
     rpm: float,
     channel_angle_map: dict[pt.PdChannel, pt.PdAngle],
 ) -> dict[pt.PdChannel, pt.Voltage]:
+    from pioreactor.background_jobs.stirring import start_stirring as stirring
+
     with stirring(
         target_rpm=rpm,
         unit=get_unit_name(),
@@ -180,11 +181,17 @@ def _devices_for_angles(channel_angle_map: dict[pt.PdChannel, pt.PdAngle]) -> li
 def _calculate_curve_data(
     od600_values: list[float],
     voltages: list[float],
-) -> list[float]:
-    degree = min(3, max(1, len(od600_values) - 1))
+) -> tuple[str, list]:
     weights = [1.0] * len(voltages)
     weights[0] = len(voltages) / 2
-    return utils.calculate_poly_curve_of_best_fit(od600_values, voltages, degree, weights)
+    if len(od600_values) >= 3:
+        from pioreactor.utils.splines import spline_fit
+
+        knots = min(4, len(od600_values))
+        return "spline", spline_fit(od600_values, voltages, knots=knots, weights=weights)
+
+    degree = min(3, max(1, len(od600_values) - 1))
+    return "poly", utils.calculate_poly_curve_of_best_fit(od600_values, voltages, degree, weights)
 
 
 def _build_standards_chart_metadata(
@@ -204,9 +211,10 @@ def _build_standards_chart_metadata(
         points = [{"x": float(od600_values[i]), "y": float(voltages[i])} for i in range(count)]
         curve = None
         if count > 1:
+            curve_type, curve_data = _calculate_curve_data(od600_values[:count], voltages[:count])
             curve = {
-                "type": "poly",
-                "coefficients": _calculate_curve_data(od600_values[:count], voltages[:count]),
+                "type": curve_type,
+                "coefficients": curve_data,
             }
         series.append(
             {
@@ -445,7 +453,7 @@ class AnotherStandard(SessionStep):
     def render(self, ctx: SessionContext) -> CalibrationStep:
         step = steps.form(
             "Next standard",
-            "Record another standard or redo the last measurement?",
+            "Record another standard, move on to the blank, or redo the last measurement?",
             [
                 fields.choice(
                     "next_action",
@@ -544,10 +552,10 @@ def advance(self, ctx: SessionContext) -> SessionStep | None:
         for pd_channel, angle in sorted(channel_angle_map.items(), key=lambda item: int(item[0])):
             voltages_list = ctx.data["voltages_by_channel"][pd_channel]
             od600_values = ctx.data["od600_values"]
-            curve_data_ = _calculate_curve_data(od600_values, voltages_list)
+            curve_type, curve_data_ = _calculate_curve_data(od600_values, voltages_list)
             cal = to_struct(
                 curve_data_,
-                "poly",
+                curve_type,
                 voltages_list,
                 od600_values,
                 angle,
@@ -590,8 +598,22 @@ def get_standards_step(
     return get_session_step(_OD_STANDARDS_STEPS, session, executor)
 
 
+def get_valid_od_devices_for_this_unit() -> list[str]:
+    pd_channels = config["od_config.photodiode_channel"]
+    valid_devices: list[pt.ODCalibrationDevices] = []
+
+    for _, angle in pd_channels.items():
+        if angle in (None, "", REF_keyword):
+            continue
+        device = f"od{angle}"
+        if device in pt.OD_DEVICES and device not in valid_devices:
+            valid_devices.append(cast(pt.ODCalibrationDevices, device))
+
+    return valid_devices
+
+
 class StandardsODProtocol(CalibrationProtocol[pt.ODCalibrationDevices]):
-    target_device = pt.OD_DEVICES
+    target_device = get_valid_od_devices_for_this_unit()
     protocol_name = "standards"
     title = "OD standards calibration"
     description = "Calibrate OD channels using a series of OD600 standards and a blank."

core/pioreactor/calibrations/protocols/stirring_dc_based.py

@@ -138,14 +138,17 @@ def _build_stirring_calibration_from_measurements(
         logger.warning("Something went wrong - detected negative correlation between RPM and stirring.")
         raise ValueError("Negative correlation between RPM and stirring.")
 
+    from pioreactor.utils.splines import spline_fit
+
+    knots = min(3, len(dcs))
     return SimpleStirringCalibration(
         pwm_hz=config.getfloat("stirring.config", "pwm_hz"),
         voltage=voltage,
         calibration_name=f"stirring-calibration-{current_utc_datetime().strftime('%Y-%m-%d_%H-%M')}",
         calibrated_on_pioreactor_unit=unit,
         created_at=current_utc_datetime(),
-        curve_data_=[alpha, beta],
-        curve_type="poly",
+        curve_data_=spline_fit(dcs, rpms, knots=knots),
+        curve_type="spline",
         recorded_data={"x": dcs, "y": rpms},
     )
 

core/pioreactor/calibrations/session_flow.py

@@ -449,7 +449,7 @@ def _render_chart_for_cli(chart: dict[str, object]) -> None:
         if isinstance(curve, dict):
             curve_type = curve.get("type")
             coeffs = curve.get("coefficients")
-            if curve_type == "poly" and isinstance(coeffs, list):
+            if curve_type in {"poly", "spline"} and isinstance(coeffs, list):
                 curve_callable = curve_to_callable(curve_type, coeffs)
         plot_data(
             x_vals,