nicer

Author: CamDavidsonPilon

core/pioreactor/calibrations/protocols/README.md

@@ -1,47 +1,3 @@
-# Calibration CLI UX Spec
+# Calibrations
 
-This spec defines the CLI UX conventions for calibration flows. Apply these patterns consistently across all calibration scripts.
-
-Use pioreactor.calibrations.cli_helpers to assist.
-
-## Goals
-
-- Keep calibration outputs unchanged (same computed results, same data structures).
-- Make the CLI flow readable and consistent.
-- Separate user input, information, and actions visually.
-
-## Colors
-
-Use pioreactor.calibrations.cli_helpers to get functions for these.
-
-- User input prompts: green.
-- Informational messages and headings: white.
-- User actions (physical steps): cyan.
-- Errors / aborts: red.
-
-## Message Types
-
-- **Info:** single-line status or explanation (white).
-- **Heading:** key sections, white + bold + underline.
-- **Action:** physical steps to perform, cyan.
-- **Prompt:** questions or inputs, green.
-- **Error:** aborts or invalid states, red.
-
-## Spacing Rules
-
-- Use an action block for physical steps, with a blank line before and after.
-- Avoid stacking info, prompt, and action on the same line.
-- When a prompt follows an action block, keep the blank line between them.
-
-## Prompt Style
-
-- Don't use a trailing colon for input prompts, e.g. `Enter OD600 measurement` .
-- Use a question mark for confirmations, e.g. `Continue?`.
-- Keep prompts short and specific.
-- If the information is just pausing to start something, add `...` at the end. Example: "Warming up OD...", "Waiting for X..."
-
-## Consistency Notes
-
-- Prefer simple, direct language and consistent verbs (Enter, Confirm, Record).
-- Avoid changing calibration behavior or output format.
-- Reuse shared helpers for styling and spacing where possible.
+Use `scratch/generate_calibration_flow_graphs.py` to generate directed graphs of the flows.

core/pioreactor/calibrations/protocols/od_reference_standard.py

@@ -4,6 +4,7 @@
 import uuid
 from time import sleep
 from typing import cast
+from typing import ClassVar
 
 from pioreactor import structs
 from pioreactor import types as pt
@@ -12,10 +13,14 @@
 from pioreactor.background_jobs.od_reading import start_od_reading
 from pioreactor.calibrations import utils as calibration_utils
 from pioreactor.calibrations.registry import CalibrationProtocol
+from pioreactor.calibrations.session_flow import advance_session
+from pioreactor.calibrations.session_flow import CalibrationComplete
+from pioreactor.calibrations.session_flow import get_session_step
 from pioreactor.calibrations.session_flow import run_session_in_cli
 from pioreactor.calibrations.session_flow import SessionContext
-from pioreactor.calibrations.session_flow import SessionEngine
 from pioreactor.calibrations.session_flow import SessionExecutor
+from pioreactor.calibrations.session_flow import SessionStep
+from pioreactor.calibrations.session_flow import StepRegistry
 from pioreactor.calibrations.session_flow import steps
 from pioreactor.calibrations.structured_session import CalibrationSession
 from pioreactor.calibrations.structured_session import CalibrationStep
@@ -138,15 +143,10 @@ def start_reference_standard_session(
     )
 
 
-def reference_standard_flow(ctx: SessionContext) -> CalibrationStep:
-    if ctx.session.status != "in_progress":
-        if ctx.session.result is not None:
-            return steps.result(ctx.session.result)
-        return steps.info("Calibration ended", "This calibration session has ended.")
+class Intro(SessionStep):
+    step_id = "intro"
 
-    if ctx.step == "intro":
-        if ctx.inputs.has_inputs:
-            ctx.step = "record_readings"
+    def render(self, ctx: SessionContext) -> CalibrationStep:
         return steps.info(
             "OD reference standard calibration",
             (
@@ -155,106 +155,121 @@ def reference_standard_flow(ctx: SessionContext) -> CalibrationStep:
             ),
         )
 
-    if ctx.step == "record_readings":
+    def advance(self, ctx: SessionContext) -> SessionStep | None:
         if ctx.inputs.has_inputs:
-            if is_pio_job_running("od_reading"):
-                raise ValueError("OD reading should be turned off.")
-
-            ir_led_intensity = get_ir_led_intensity()
-            channel_angle_map = get_channel_angle_map(
-                cast(pt.ODCalibrationDevices, ctx.session.target_device)
-            )
-
-            od_readings = _record_reference_standard_for_session(ctx, ir_led_intensity)
-            recorded_ods = [0.0, 1000 * STANDARD_OD]
-            timestamp = current_utc_datetime().strftime("%Y-%m-%d_%H-%M")
-
-            calibration_links: list[dict[str, str | None]] = []
-            if isinstance(od_readings, dict):
-                for raw_channel, od_reading_payload in od_readings.items():
-                    pd_channel = cast(pt.PdChannel, raw_channel)
-                    if pd_channel not in channel_angle_map:
-                        continue
-                    angle = channel_angle_map[pd_channel]
-                    od_value = float(od_reading_payload["od"])
-
-                    recorded_voltages = [0.0, 1000 * od_value]
-                    curve_data_ = calibration_utils.calculate_poly_curve_of_best_fit(
-                        recorded_ods, recorded_voltages, degree=1
-                    )
-                    calibration = structs.ODCalibration(
-                        created_at=current_utc_datetime(),
-                        calibrated_on_pioreactor_unit=get_unit_name(),
-                        calibration_name=f"od{angle}-optics-calibration-jig-{timestamp}",
-                        angle=angle,
-                        curve_data_=curve_data_,
-                        curve_type="poly",
-                        recorded_data={"x": recorded_ods, "y": recorded_voltages},
-                        ir_led_intensity=ir_led_intensity,
-                        pd_channel=pd_channel,
-                    )
-                    calibration_links.append(ctx.store_calibration(calibration, f"od{angle}"))
-            else:
-                for raw_channel, od_reading_struct in od_readings.ods.items():
-                    pd_channel = cast(pt.PdChannel, raw_channel)
-                    if pd_channel not in channel_angle_map:
-                        continue
-                    angle = channel_angle_map[pd_channel]
-                    od_value = float(od_reading_struct.od)
-
-                    recorded_voltages = [0.0, 1000 * od_value]
-                    curve_data_ = calibration_utils.calculate_poly_curve_of_best_fit(
-                        recorded_ods, recorded_voltages, degree=1
-                    )
-                    calibration = structs.ODCalibration(
-                        created_at=current_utc_datetime(),
-                        calibrated_on_pioreactor_unit=get_unit_name(),
-                        calibration_name=f"od{angle}-optical-reference-standard-{timestamp}",
-                        angle=angle,
-                        curve_data_=curve_data_,
-                        curve_type="poly",
-                        recorded_data={"x": recorded_ods, "y": recorded_voltages},
-                        ir_led_intensity=ir_led_intensity,
-                        pd_channel=pd_channel,
-                    )
-                    calibration_links.append(ctx.store_calibration(calibration, f"od{angle}"))
-
-            if not calibration_links:
-                raise ValueError("No matching channels were recorded for this calibration.")
-
-            ctx.complete({"calibrations": calibration_links})
+            return RecordReadings()
+        return None
+
+
+class RecordReadings(SessionStep):
+    step_id = "record_readings"
+
+    def render(self, ctx: SessionContext) -> CalibrationStep:
         return steps.action(
             "Record reference standard",
             "Continue to record OD readings against the optical reference standard.",
         )
 
-    return steps.info("Unknown step", "This step is not recognized.")
+    def advance(self, ctx: SessionContext) -> SessionStep | None:
+        if is_pio_job_running("od_reading"):
+            raise ValueError("OD reading should be turned off.")
+
+        ir_led_intensity = get_ir_led_intensity()
+        channel_angle_map = get_channel_angle_map(cast(pt.ODCalibrationDevices, ctx.session.target_device))
+
+        od_readings = _record_reference_standard_for_session(ctx, ir_led_intensity)
+        recorded_ods = [0.0, 1000 * STANDARD_OD]
+        timestamp = current_utc_datetime().strftime("%Y-%m-%d_%H-%M")
+
+        calibration_links: list[dict[str, str | None]] = []
+        if isinstance(od_readings, dict):
+            for raw_channel, od_reading_payload in od_readings.items():
+                pd_channel = cast(pt.PdChannel, raw_channel)
+                if pd_channel not in channel_angle_map:
+                    continue
+                angle = channel_angle_map[pd_channel]
+                od_value = float(od_reading_payload["od"])
+
+                recorded_voltages = [0.0, 1000 * od_value]
+                curve_data_ = calibration_utils.calculate_poly_curve_of_best_fit(
+                    recorded_ods, recorded_voltages, degree=1
+                )
+                calibration = structs.ODCalibration(
+                    created_at=current_utc_datetime(),
+                    calibrated_on_pioreactor_unit=get_unit_name(),
+                    calibration_name=f"od{angle}-optics-calibration-jig-{timestamp}",
+                    angle=angle,
+                    curve_data_=curve_data_,
+                    curve_type="poly",
+                    recorded_data={"x": recorded_ods, "y": recorded_voltages},
+                    ir_led_intensity=ir_led_intensity,
+                    pd_channel=pd_channel,
+                )
+                calibration_links.append(ctx.store_calibration(calibration, f"od{angle}"))
+        else:
+            for raw_channel, od_reading_struct in od_readings.ods.items():
+                pd_channel = cast(pt.PdChannel, raw_channel)
+                if pd_channel not in channel_angle_map:
+                    continue
+                angle = channel_angle_map[pd_channel]
+                od_value = float(od_reading_struct.od)
+
+                recorded_voltages = [0.0, 1000 * od_value]
+                curve_data_ = calibration_utils.calculate_poly_curve_of_best_fit(
+                    recorded_ods, recorded_voltages, degree=1
+                )
+                calibration = structs.ODCalibration(
+                    created_at=current_utc_datetime(),
+                    calibrated_on_pioreactor_unit=get_unit_name(),
+                    calibration_name=f"od{angle}-optical-reference-standard-{timestamp}",
+                    angle=angle,
+                    curve_data_=curve_data_,
+                    curve_type="poly",
+                    recorded_data={"x": recorded_ods, "y": recorded_voltages},
+                    ir_led_intensity=ir_led_intensity,
+                    pd_channel=pd_channel,
+                )
+                calibration_links.append(ctx.store_calibration(calibration, f"od{angle}"))
+
+        if not calibration_links:
+            raise ValueError("No matching channels were recorded for this calibration.")
+
+        ctx.complete({"calibrations": calibration_links})
+        return CalibrationComplete()
+
+
+_REFERENCE_STANDARD_STEPS: StepRegistry = {
+    Intro.step_id: Intro,
+    RecordReadings.step_id: RecordReadings,
+}
 
 
 def advance_reference_standard_session(
     session: CalibrationSession,
     inputs: dict[str, object],
     executor: SessionExecutor | None = None,
 ) -> CalibrationSession:
-    engine = SessionEngine(flow=reference_standard_flow, session=session, mode="ui", executor=executor)
-    engine.advance(inputs)
-    return engine.session
+    return advance_session(_REFERENCE_STANDARD_STEPS, session, inputs, executor)
 
 
 def get_reference_standard_step(
     session: CalibrationSession, executor: SessionExecutor | None = None
 ) -> CalibrationStep | None:
-    engine = SessionEngine(flow=reference_standard_flow, session=session, mode="ui", executor=executor)
-    return engine.get_step()
+    return get_session_step(_REFERENCE_STANDARD_STEPS, session, executor)
 
 
 class ODReferenceStandardProtocol(CalibrationProtocol[pt.ODCalibrationDevices]):
     target_device = pt.OD_DEVICES
     protocol_name = "od_reference_standard"
     description = "Calibrate OD using the Pioreactor Optical Reference Standard."
+    step_registry: ClassVar[StepRegistry] = _REFERENCE_STANDARD_STEPS
+
+    @classmethod
+    def start_session(cls, target_device: pt.ODCalibrationDevices) -> CalibrationSession:
+        return start_reference_standard_session(target_device)
 
     def run(  # type: ignore
         self, target_device: pt.ODCalibrationDevices, *args, **kwargs
     ) -> list[structs.ODCalibration]:
         session = start_reference_standard_session(target_device)
-        return run_session_in_cli(reference_standard_flow, session)
+        return run_session_in_cli(_REFERENCE_STANDARD_STEPS, session)