docs

Author: CamDavidsonPilon

core/pioreactor/calibrations/protocols/README.md

@@ -12,6 +12,7 @@
 Device = TypeVar("Device", bound=str)
 ProtocolName = str
 
+# Registry of device -> protocol name -> protocol class (populated via CalibrationProtocol subclasses).
 calibration_protocols: dict[str, dict[ProtocolName, type["CalibrationProtocol[Any]"]]] = defaultdict(dict)
 
 

core/pioreactor/calibrations/registry.py

@@ -36,6 +36,7 @@ def advance(self, ctx: "SessionContext") -> "StepLike | None":
 
 
 StepLike = str | SessionStep | type[SessionStep]
+# Registry of step_id -> SessionStep class for a protocol's session flow.
 StepRegistry = dict[str, type[SessionStep]]
 
 

core/pioreactor/calibrations/session_flow.py

@@ -1,4 +1,11 @@
 # -*- coding: utf-8 -*-
+"""
+Huey background tasks used by the web API, including calibration actions invoked
+from `core/pioreactor/web/unit_calibration_sessions_api.py` via the session executor.
+This module also hosts the calibration action registry that maps action names
+to Huey tasks so unit API handlers can dispatch without hardcoded strings.
+"""
+# -*- coding: utf-8 -*-
 from __future__ import annotations
 
 import configparser
@@ -10,6 +17,7 @@
 import shutil
 import stat
 import zipfile
+from collections.abc import Callable
 from pathlib import Path
 from shlex import join
 from subprocess import check_call
@@ -39,6 +47,30 @@
 from pioreactor.web.config import huey
 from pioreactor.whoami import get_unit_name
 
+CalibrationActionHandler = tuple[
+    Any,
+    int,
+    str,
+    Callable[[Any], dict[str, Any]],
+]
+
+# Registry of calibration action -> handler that returns a Huey task, timeout, label, and normalizer.
+calibration_actions: dict[str, Callable[[dict[str, Any]], CalibrationActionHandler]] = {}
+
+
+def register_calibration_action(
+    action: str,
+    handler: Callable[[dict[str, Any]], CalibrationActionHandler],
+) -> None:
+    calibration_actions[action] = handler
+
+
+def get_calibration_action(action: str) -> Callable[[dict[str, Any]], CalibrationActionHandler]:
+    handler = calibration_actions.get(action)
+    if handler is None:
+        raise ValueError(f"Unknown calibration action: {action}")
+    return handler
+
 
 logger = create_logger(
     "background_tasks",
@@ -430,6 +462,104 @@ def calibration_read_voltage() -> float:
     return float(voltage_in_aux())
 
 
+def _default_normalizer(result: Any) -> dict[str, Any]:
+    if isinstance(result, dict):
+        return result
+    return {}
+
+
+def _voltage_normalizer(result: Any) -> dict[str, Any]:
+    return {"voltage": float(result)}
+
+
+def _voltages_normalizer(result: Any) -> dict[str, Any]:
+    return {"voltages": result}
+
+
+def _od_readings_normalizer(result: Any) -> dict[str, Any]:
+    return {"od_readings": result}
+
+
+def _stirring_normalizer(result: Any) -> dict[str, Any]:
+    if isinstance(result, dict):
+        return result
+    return {}
+
+
+def _register_core_calibration_actions() -> None:
+    register_calibration_action(
+        "pump",
+        lambda payload: (
+            calibration_execute_pump(
+                payload["pump_device"],
+                float(payload["duration_s"]),
+                float(payload["hz"]),
+                float(payload["dc"]),
+            ),
+            300,
+            "Pump action",
+            _default_normalizer,
+        ),
+    )
+    register_calibration_action(
+        "od_standards_measure",
+        lambda payload: (
+            calibration_measure_standard(
+                float(payload["rpm"]),
+                payload["channel_angle_map"],
+            ),
+            300,
+            "OD measurement",
+            _voltages_normalizer,
+        ),
+    )
+    register_calibration_action(
+        "od_reference_standard_read",
+        lambda payload: (
+            calibration_reference_standard_read(float(payload["ir_led_intensity"])),
+            300,
+            "Reference standard reading",
+            _od_readings_normalizer,
+        ),
+    )
+    register_calibration_action(
+        "stirring_calibration",
+        lambda payload: (
+            calibration_run_stirring(
+                float(payload["min_dc"]) if (payload.get("min_dc") is not None) else None,
+                float(payload["max_dc"]) if (payload.get("max_dc") is not None) else None,
+            ),
+            300,
+            "Stirring calibration",
+            _stirring_normalizer,
+        ),
+    )
+    register_calibration_action(
+        "read_voltage",
+        lambda payload: (
+            calibration_read_voltage(),
+            30,
+            "Voltage read",
+            _voltage_normalizer,
+        ),
+    )
+    register_calibration_action(
+        "save_calibration",
+        lambda payload: (
+            calibration_save_calibration(
+                payload["device"],
+                payload["calibration"],
+            ),
+            300,
+            "Saving calibration",
+            _default_normalizer,
+        ),
+    )
+
+
+_register_core_calibration_actions()
+
+
 @huey.task()
 @huey.lock_task("export-data-lock")
 def export_experiment_data_task(

core/pioreactor/web/tasks.py

@@ -36,9 +36,9 @@
 from pioreactor.web.app import publish_to_error_log
 from pioreactor.web.app import publish_to_log
 from pioreactor.web.app import query_temp_local_metadata_db
-from pioreactor.web.calibration_sessions_api import register_calibration_session_routes
 from pioreactor.web.config import huey
 from pioreactor.web.plugin_registry import registered_unit_api_routes
+from pioreactor.web.unit_calibration_sessions_api import register_calibration_session_routes
 from pioreactor.web.utils import abort_with
 from pioreactor.web.utils import attach_cache_control
 from pioreactor.web.utils import create_task_response

core/pioreactor/web/unit_api.py

@@ -1,4 +1,13 @@
 # -*- coding: utf-8 -*-
+"""
+Calibration session HTTP API.
+
+Uses step registries from calibration protocols and dispatches hardware actions
+through Huey tasks defined in `core/pioreactor/web/tasks.py`.
+The unit API routes here for session start/advance/abort and invokes the
+calibration action executor to perform privileged hardware work.
+"""
+# -*- coding: utf-8 -*-
 from __future__ import annotations
 
 from typing import Any
@@ -17,88 +26,20 @@
 from pioreactor.calibrations.structured_session import load_calibration_session
 from pioreactor.calibrations.structured_session import save_calibration_session
 from pioreactor.calibrations.structured_session import utc_iso_timestamp
-from pioreactor.web import tasks
+from pioreactor.web.tasks import get_calibration_action
 from pioreactor.web.utils import abort_with
 
 
 def _execute_calibration_action(action: str, payload: dict[str, Any]) -> dict[str, Any]:
-    if action == "pump":
-        task = tasks.calibration_execute_pump(
-            payload["pump_device"],
-            float(payload["duration_s"]),
-            float(payload["hz"]),
-            float(payload["dc"]),
-        )
-        try:
-            success = task(blocking=True, timeout=300)
-        except TaskException as exc:
-            raise ValueError(f"Pump action failed: {exc}") from exc
-        except HueyException as exc:
-            raise ValueError("Pump action timed out.") from exc
-        if not success:
-            raise ValueError("Pump action failed.")
-        return {}
-
-    if action == "od_standards_measure":
-        task = tasks.calibration_measure_standard(
-            float(payload["rpm"]),
-            payload["channel_angle_map"],
-        )
-        try:
-            voltages = task(blocking=True, timeout=300)
-        except TaskException as exc:
-            raise ValueError(f"OD measurement failed: {exc}") from exc
-        except HueyException as exc:
-            raise ValueError("OD measurement timed out.") from exc
-        return {"voltages": voltages}
-
-    if action == "od_reference_standard_read":
-        task = tasks.calibration_reference_standard_read(float(payload["ir_led_intensity"]))
-        try:
-            readings = task(blocking=True, timeout=300)
-        except TaskException as exc:
-            raise ValueError(f"Reference standard reading failed: {exc}") from exc
-        except HueyException as exc:
-            raise ValueError("Reference standard reading timed out.") from exc
-        return {"od_readings": readings}
-
-    if action == "stirring_calibration":
-        task = tasks.calibration_run_stirring(
-            float(payload["min_dc"]) if (payload.get("min_dc") is not None) else None,
-            float(payload["max_dc"]) if (payload.get("max_dc") is not None) else None,
-        )
-        try:
-            calibration_payload = task(blocking=True, timeout=300)
-        except TaskException as exc:
-            raise ValueError(f"Stirring calibration failed: {exc}") from exc
-        except HueyException as exc:
-            raise ValueError("Stirring calibration timed out.") from exc
-        return calibration_payload
-
-    if action == "read_voltage":
-        task = tasks.calibration_read_voltage()
-        try:
-            voltage = task(blocking=True, timeout=30)
-        except TaskException as exc:
-            raise ValueError(f"Voltage read failed: {exc}") from exc
-        except HueyException as exc:
-            raise ValueError("Voltage read timed out.") from exc
-        return {"voltage": float(voltage)}
-
-    if action == "save_calibration":
-        task = tasks.calibration_save_calibration(
-            payload["device"],
-            payload["calibration"],
-        )
-        try:
-            result = task(blocking=True, timeout=300)
-        except TaskException as exc:
-            raise ValueError(f"Saving calibration failed: {exc}") from exc
-        except HueyException as exc:
-            raise ValueError("Saving calibration timed out.") from exc
-        return result
-
-    raise ValueError("Unknown calibration action.")
+    handler = get_calibration_action(action)
+    task, timeout_s, error_label, normalize = handler(payload)
+    try:
+        result = task(blocking=True, timeout=timeout_s)
+    except TaskException as exc:
+        raise ValueError(f"{error_label} failed: {exc}") from exc
+    except HueyException as exc:
+        raise ValueError(f"{error_label} timed out.") from exc
+    return normalize(result)
 
 
 def _get_step_registry(protocol) -> Any: