adding estimators to api and UI

Author: CamDavidsonPilon

core/pioreactor/background_jobs/od_reading.py

@@ -810,7 +810,7 @@ def __init__(self) -> None:
         super().__init__()
         self.models: dict[pt.PdChannel, Callable] = {}
         self.verifiers: dict[pt.PdChannel, Callable] = {}
-        self.has_logged_warning = False
+        self._last_bound_warning_at: float | None = None
 
     def hydate_models(self, calibration_data: structs.ODCalibration | None) -> None:
         if calibration_data is None:
@@ -889,31 +889,35 @@ def clamp_to_recorded_extrema() -> pt.OD:
             try:
                 return calibration_data.y_to_x(observed_voltage, enforce_bounds=True)
             except exc.NoSolutionsFoundError:
-                if not self.has_logged_warning:
+                if self._should_warn_about_bounds():
                     self.logger.warning(
                         f"No solution found for calibrated signal. 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, which would map outside the allowed values."
                     )
-                    self.has_logged_warning = True
                 return clamp_to_recorded_extrema()
             except exc.SolutionBelowDomainError:
-                if not self.has_logged_warning:
+                if self._should_warn_about_bounds():
                     below_or_above = "below" if observed_voltage <= min_voltage else "outside of"
                     self.logger.warning(
                         f"Signal {below_or_above} 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, which would map outside the allowed values."
                     )
-                    self.has_logged_warning = True
                 return clamp_to_recorded_extrema()
             except exc.SolutionAboveDomainError:
-                if not self.has_logged_warning:
+                if self._should_warn_about_bounds():
                     above_or_outside = "above" if observed_voltage >= max_voltage else "outside of"
                     self.logger.warning(
                         f"Signal {above_or_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, which would map outside the allowed values."
                     )
-                    self.has_logged_warning = True
                 return clamp_to_recorded_extrema()
 
         return _calibrate_signal, _verify
 
+    def _should_warn_about_bounds(self) -> bool:
+        now = time()
+        if self._last_bound_warning_at is None or (now - self._last_bound_warning_at) > 300:
+            self._last_bound_warning_at = now
+            return True
+        return False
+
     def __call__(self, od_readings: structs.ODReadings) -> structs.ODReadings:
         calibrated_od_readings = structs.ODReadings(ods={}, timestamp=od_readings.timestamp)
         for channel in od_readings.ods:
@@ -957,6 +961,7 @@ class CachedEstimatorTransformer(LoggerMixin, EstimatorTransformerProtocol):
     def __init__(self) -> None:
         super().__init__()
         self.estimator: structs.ODFusionEstimator | None = None
+        self._last_bound_warning_at: float | None = None
 
     def hydrate_estimator(self, estimator: structs.ODFusionEstimator | None) -> None:
         if estimator is None:
@@ -973,12 +978,34 @@ def __call__(self, raw_od_readings: structs.ODReadings) -> structs.ODFused | Non
         }
         try:
             od_fused_value = compute_fused_od(self.estimator, fused_inputs)
+            if self._should_warn_about_bounds(od_fused_value):
+                self.logger.warning(
+                    "Fused OD estimate hit estimator bounds: estimator=%s min_logc=%s max_logc=%s od_fused=%s ref_normalization=%s",
+                    self.estimator.estimator_name,
+                    self.estimator.min_logc,
+                    self.estimator.max_logc,
+                    od_fused_value,
+                    config.get("od_reading.config", "ref_normalization", fallback="classic"),
+                )
         except Exception as e:
             self.logger.debug(f"Failed to compute fused OD: {e}", exc_info=True)
             return None
 
         return structs.ODFused(od_fused=od_fused_value, timestamp=raw_od_readings.timestamp)
 
+    def _should_warn_about_bounds(self, od_fused_value: float) -> bool:
+        if self.estimator is None:
+            return False
+
+        lower_bound = 10**self.estimator.min_logc
+        upper_bound = 10**self.estimator.max_logc
+        if (od_fused_value <= lower_bound * 1.001) or (od_fused_value >= upper_bound * 0.999):
+            now = time()
+            if self._last_bound_warning_at is None or (now - self._last_bound_warning_at) > 300:
+                self._last_bound_warning_at = now
+                return True
+        return False
+
 
 class ODReader(BackgroundJob):
     """

core/pioreactor/calibrations/protocols/od_fusion_standards.py

@@ -300,7 +300,7 @@ def render(self, ctx: SessionContext) -> CalibrationStep:
             f"Place standard vial {standard_index}",
             (
                 f"Place standard vial {standard_index} with a stir bar into the Pioreactor. "
-                "We will take four readings, then you will remove it."
+                "We will take OD readings, then you will remove it."
             ),
         )
         step.metadata = {

core/pioreactor/web/api.py

@@ -1427,6 +1427,15 @@ def get_all_active_calibrations(pioreactor_unit: str) -> DelayedResponseReturnVa
     return create_task_response(task)
 
 
+@api_bp.route("/workers/<pioreactor_unit>/active_estimators", methods=["GET"])
+def get_all_active_estimators(pioreactor_unit: str) -> DelayedResponseReturnValue:
+    if pioreactor_unit == UNIVERSAL_IDENTIFIER:
+        task = broadcast_get_across_workers("/unit_api/active_estimators")
+    else:
+        task = tasks.multicast_get("/unit_api/active_estimators", [pioreactor_unit])
+    return create_task_response(task)
+
+
 @api_bp.route("/workers/<pioreactor_unit>/zipped_calibrations", methods=["GET"])
 def get_all_calibrations_as_yamls(pioreactor_unit: str) -> ResponseReturnValue:
     if pioreactor_unit == UNIVERSAL_IDENTIFIER:

core/pioreactor/web/static/asset-manifest.json

@@ -1,7 +1,7 @@
 {
   "files": {
     "main.css": "/static/static/css/main.9c7a48b7.css",
-    "main.js": "/static/static/js/main.ab57c232.js",
+    "main.js": "/static/static/js/main.12bd0d12.js",
     "static/media/pioreactor_cloud.webp": "/static/static/media/pioreactor_cloud.b15b29e435797dc69d76.webp",
     "static/media/roboto-all-500-normal.woff": "/static/static/media/roboto-all-500-normal.0ab669b7a0d19b178f57.woff",
     "static/media/roboto-all-700-normal.woff": "/static/static/media/roboto-all-700-normal.a457fde362a540fcadff.woff",
@@ -30,10 +30,10 @@
     "static/media/roboto-greek-ext-700-normal.woff2": "/static/static/media/roboto-greek-ext-700-normal.bd9854c751441ccc1a70.woff2",
     "index.html": "/static/index.html",
     "main.9c7a48b7.css.map": "/static/static/css/main.9c7a48b7.css.map",
-    "main.ab57c232.js.map": "/static/static/js/main.ab57c232.js.map"
+    "main.12bd0d12.js.map": "/static/static/js/main.12bd0d12.js.map"
   },
   "entrypoints": [
     "static/css/main.9c7a48b7.css",
-    "static/js/main.ab57c232.js"
+    "static/js/main.12bd0d12.js"
   ]
 }

core/pioreactor/web/static/index.html

@@ -1 +1 @@
-<!doctype html><html lang="en"><head><meta charset="utf-8"/><link rel="icon" href="/static/favicon.ico"/><meta name="viewport" content="width=device-width,initial-scale=1"/><meta name="theme-color" content="#000000"/><meta name="description" content="Pioreactor"/><link rel="apple-touch-icon" href="/static/logo192.png"/><link rel="manifest" href="/static/manifest.json"/><script defer="defer" src="/static/static/js/main.ab57c232.js"></script><link href="/static/static/css/main.9c7a48b7.css" rel="stylesheet"></head><body><div id="root"></div></body></html>
+<!doctype html><html lang="en"><head><meta charset="utf-8"/><link rel="icon" href="/static/favicon.ico"/><meta name="viewport" content="width=device-width,initial-scale=1"/><meta name="theme-color" content="#000000"/><meta name="description" content="Pioreactor"/><link rel="apple-touch-icon" href="/static/logo192.png"/><link rel="manifest" href="/static/manifest.json"/><script defer="defer" src="/static/static/js/main.12bd0d12.js"></script><link href="/static/static/css/main.9c7a48b7.css" rel="stylesheet"></head><body><div id="root"></div></body></html>

…or/web/static/static/js/main.ab57c232.js …or/web/static/static/js/main.12bd0d12.jscore/pioreactor/web/static/static/js/main.ab57c232.js renamed to core/pioreactor/web/static/static/js/main.12bd0d12.js core/pioreactor/web/static/static/js/main.ab57c232.js renamed to core/pioreactor/web/static/static/js/main.12bd0d12.js

@@ -839,6 +839,34 @@ def get_all_active_calibrations() -> ResponseReturnValue:
     return attach_cache_control(jsonify(all_calibrations), max_age=10)
 
 
+@unit_api_bp.route("/active_estimators", methods=["GET"])
+def get_all_active_estimators() -> ResponseReturnValue:
+    estimator_dir = ESTIMATOR_PATH
+
+    if not estimator_dir.exists():
+        return attach_cache_control(jsonify({}), max_age=10)
+
+    all_estimators: dict[str, dict] = {}
+
+    with local_persistent_storage("active_estimators") as cache:
+        for device in cache.iterkeys():
+            estimator_name = cache[device]
+            estimator_file_path = estimator_dir / device / f"{estimator_name}.yaml"
+            if not estimator_file_path.exists():
+                continue
+            try:
+                estimator = to_builtins(yaml_decode(estimator_file_path.read_bytes(), type=AllEstimators))
+                estimator["is_active"] = True
+                estimator["pioreactor_unit"] = HOSTNAME
+                all_estimators[device] = estimator
+            except Exception as e:
+                publish_to_error_log(
+                    f"Error reading {estimator_file_path.stem}: {e}", "get_all_active_estimators"
+                )
+
+    return attach_cache_control(jsonify(all_estimators), max_age=10)
+
+
 @unit_api_bp.route("/zipped_calibrations", methods=["GET"])
 def get_all_calibrations_as_zipped_yaml() -> ResponseReturnValue:
     calibration_dir = CALIBRATION_PATH