dead code?

Author: CamDavidsonPilon

core/pioreactor/calibrations/protocols/od_fusion_standards.py

@@ -75,7 +75,6 @@ def _aggregate_angles(readings: structs.ODReadings) -> dict[pt.PdAngle, float]:
 
 
 def _measure_fusion_standard(
-    od_value: float,
     rpm: float,
 ) -> dict[pt.PdAngle, float]:
     from pioreactor.background_jobs.stirring import start_stirring as stirring
@@ -127,7 +126,7 @@ def _measure_fusion_standard_for_session(
             for angle, value in raw_sample.items()
             if angle in FUSION_ANGLES
         }
-    return _measure_fusion_standard(od_value, rpm)
+    return _measure_fusion_standard(rpm)
 
 
 def _build_chart_metadata(records: list[tuple[pt.PdAngle, float, float]]) -> dict[str, object] | None:

core/pioreactor/calibrations/protocols/od_standards.py

@@ -125,7 +125,6 @@ def _read_voltages_from_adc(
 
 
 def _measure_standard(
-    od600_value: float,
     rpm: float,
     channel_angle_map: dict[pt.PdChannel, pt.PdAngle],
 ) -> dict[pt.PdChannel, pt.Voltage]:
@@ -143,23 +142,21 @@ def _measure_standard(
 
 def _measure_standard_for_session(
     ctx: SessionContext,
-    od600_value: float,
     rpm: float,
     channel_angle_map: dict[pt.PdChannel, pt.PdAngle],
 ) -> dict[pt.PdChannel, pt.Voltage]:
     if ctx.executor and ctx.mode == "ui":
         payload = ctx.executor(
             "od_standards_measure",
             {
-                "od600_value": od600_value,
                 "rpm": rpm,
                 "channel_angle_map": {str(k): str(v) for k, v in channel_angle_map.items()},
             },
         )
         raw = payload["voltages"]
         assert isinstance(raw, dict)
         return {cast(pt.PdChannel, channel): float(voltage) for channel, voltage in raw.items()}
-    return _measure_standard(od600_value, rpm, channel_angle_map)
+    return _measure_standard(rpm, channel_angle_map)
 
 
 def _default_calibration_name() -> str:
@@ -428,7 +425,7 @@ def render(self, ctx: SessionContext) -> CalibrationStep:
     def advance(self, ctx: SessionContext) -> SessionStep | None:
         channel_angle_map = _get_channel_angle_map(ctx)
         od600_value = ctx.inputs.float("od600_value", minimum=0)
-        voltages = _measure_standard_for_session(ctx, od600_value, ctx.data["rpm"], channel_angle_map)
+        voltages = _measure_standard_for_session(ctx, ctx.data["rpm"], channel_angle_map)
         ctx.data.setdefault("standard_index", 1)
         ctx.data["od600_values"].append(od600_value)
         for channel, voltage in voltages.items():
@@ -534,7 +531,7 @@ def render(self, ctx: SessionContext) -> CalibrationStep:
     def advance(self, ctx: SessionContext) -> SessionStep | None:
         channel_angle_map = _get_channel_angle_map(ctx)
         od600_blank = ctx.inputs.float("od600_blank", minimum=0)
-        voltages = _measure_standard_for_session(ctx, od600_blank, ctx.data["rpm"], channel_angle_map)
+        voltages = _measure_standard_for_session(ctx, ctx.data["rpm"], channel_angle_map)
         ctx.data["od600_values"].append(od600_blank)
         for channel, voltage in voltages.items():
             ctx.data["voltages_by_channel"][channel].append(voltage)

core/pioreactor/calibrations/session_flow.py

@@ -218,20 +218,6 @@ def step(self, value: StepLike) -> None:
     def data(self) -> dict[Str, Any]:
         return self.session.data
 
-    def ensure(self, condition: bool, message: Str) -> None:
-        if not condition:
-            raise ValueError(message)
-
-    def abort(self, message: Str) -> None:
-        self.session.status = "aborted"
-        self.session.error = message
-        self.session.step_id = "ended"
-
-    def fail(self, message: Str) -> None:
-        self.session.status = "failed"
-        self.session.error = message
-        self.session.step_id = "ended"
-
     def complete(self, result: dict[Str, Any]) -> None:
         self.session.status = "complete"
         self.session.result = result

core/pioreactor/calibrations/structured_session.py

@@ -9,7 +9,7 @@
 from msgspec.json import decode as json_decode
 from msgspec.json import encode as json_encode
 from pioreactor.utils import local_persistent_storage
-from pioreactor.utils.timing import current_utc_datetime
+from pioreactor.utils.timing import current_utc_timestamp
 
 SessionStatus = Literal["in_progress", "complete", "failed", "aborted"]
 StepType = Literal["info", "confirm", "form", "action", "result"]
@@ -51,7 +51,7 @@ class CalibrationSession(Struct, kw_only=True, frozen=False):
 
 
 def utc_iso_timestamp() -> str:
-    return current_utc_datetime().isoformat()
+    return current_utc_timestamp()
 
 
 def save_calibration_session(session: CalibrationSession) -> None:
@@ -66,9 +66,3 @@ def load_calibration_session(session_id: str) -> CalibrationSession | None:
     if payload is None:
         return None
     return json_decode(payload, type=CalibrationSession)
-
-
-def delete_calibration_session(session_id: str) -> None:
-    with local_persistent_storage("calibration_sessions") as store:
-        if session_id in store:
-            del store[session_id]

core/pioreactor/web/tasks.py

@@ -470,7 +470,6 @@ def calibration_measure_standard(
         cast(pt.PdChannel, channel): cast(pt.PdAngle, angle) for channel, angle in channel_angle_map.items()
     }
     voltages = _measure_standard(
-        od600_value=0.0,
         rpm=rpm,
         channel_angle_map=typed_map,
     )
@@ -491,7 +490,6 @@ def calibration_fusion_standard_observation(
         rpm,
     )
     sample = _measure_fusion_standard(
-        od_value=od_value,
         rpm=rpm,
     )
     serialized_sample = {str(angle): float(value) for angle, value in sample.items()}

core/tests/test_calibration_sessions.py

@@ -16,7 +16,6 @@
 from pioreactor.calibrations.session_flow import with_terminal_steps
 from pioreactor.calibrations.structured_session import CalibrationSession
 from pioreactor.calibrations.structured_session import CalibrationStep
-from pioreactor.calibrations.structured_session import delete_calibration_session
 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
@@ -70,7 +69,8 @@ def test_save_load_abort_delete_session() -> None:
     assert aborted.error == "Stop."
     assert aborted.updated_at != previous_updated_at
 
-    delete_calibration_session(session.session_id)
+    with local_persistent_storage("calibration_sessions") as store:
+        del store[session.session_id]
     assert load_calibration_session(session.session_id) is None