new od calibration

CHANGELOG.md

@@ -2,7 +2,8 @@
 
 
 #### Enhancements
- - new RP2040 firmware improvements => less noise in OD readings
+ - new OD calibration using standards (requires multiple vials). Try `pio calibrations run --device od`. This was inspired by the plugin by @odcambc.
+ - new RP2040 firmware improvements (v0.4) => faster response over i2c.
  - Improved chart colors in the UI
  - The OD reading CLI has a new option, `--snapshot`, that will start the job, take a single reading, and exit. This is useful for scripting purposes.
  - A new CLI for pumps: `pio run pumps`. Add pumps as options:
@@ -25,7 +26,8 @@
  - GET `/api/experiment_profiles/running/experiments/<experiment>` introduced
 
 #### Breaking changes
- - plugins should migrate from `click_some_name` to autodiscover plugins, to importing `run`. Example:
+
+ - (Eventually) plugins should migrate from `click_some_name` to autodiscover plugins, to importing `run`. Example:
    ```
    import click
    from pioreactor.cli.run import run
@@ -38,7 +40,6 @@
        ...
    ```
 
-
 #### Bug fixes
  - fixed ui not showing 3p calibrations
  - experiment profiles start now use the unit_api directly. This may mitigate the issue where huey workers stampeding on each other when try to start many jobs.

pioreactor/calibrations/__init__.py

@@ -1,6 +1,7 @@
 # -*- coding: utf-8 -*-
 from __future__ import annotations
 
+from collections import defaultdict
 from pathlib import Path
 from typing import Callable
 from typing import Literal
@@ -22,20 +23,24 @@
     CALIBRATION_PATH = Path(".pioreactor/storage/calibrations/")
 
 # Lookup table for different calibration protocols
-calibration_protocols: dict[tuple[str, str], Type[CalibrationProtocol]] = {}
+Device = str
+ProtocolName = str
+
+calibration_protocols: dict[Device, dict[ProtocolName, Type[CalibrationProtocol]]] = defaultdict(dict)
 
 
 class CalibrationProtocol:
-    protocol_name: str
-    target_device: str | list[str]
+    protocol_name: ProtocolName
+    target_device: Device | list[Device]
+    description = ""
 
     def __init_subclass__(cls, **kwargs):
         super().__init_subclass__(**kwargs)
         if isinstance(cls.target_device, str):
-            calibration_protocols[(cls.target_device, cls.protocol_name)] = cls
+            calibration_protocols[cls.target_device][cls.protocol_name] = cls
         elif isinstance(cls.target_device, list):
             for device in cls.target_device:
-                calibration_protocols[(device, cls.protocol_name)] = cls
+                calibration_protocols[device][cls.protocol_name] = cls
         else:
             raise ValueError("target_device must be a string or a list of strings")
 
@@ -46,19 +51,23 @@ def run(self, *args, **kwargs):
 class SingleVialODProtocol(CalibrationProtocol):
     target_device = "od"
     protocol_name = "single_vial"
+    description = "Calibrate OD using a single vial"
 
     def run(self, *args, **kwargs) -> structs.ODCalibration:
-        from pioreactor.calibrations.od_calibration import run_od_calibration
+        from pioreactor.calibrations.od_calibration_single_vial import run_od_calibration
 
         return run_od_calibration()
 
 
-class BatchVialODProtocol(CalibrationProtocol):
+class StandardsODProtocol(CalibrationProtocol):
     target_device = "od"
-    protocol_name = "batch_vial"
+    protocol_name = "standards"
+    description = "Calibrate OD using standards. Requires multiple vials"
 
     def run(self, *args, **kwargs) -> structs.ODCalibration:
-        raise NotImplementedError("Not implemented yet")
+        from pioreactor.calibrations.od_calibration_using_standards import run_od_calibration
+
+        return run_od_calibration()
 
 
 class DurationBasedPumpProtocol(CalibrationProtocol):
@@ -102,7 +111,7 @@ def load_active_calibration(device: Literal["stirring"]) -> structs.SimpleStirri
     pass
 
 
-def load_active_calibration(device: str) -> structs.AnyCalibration | None:
+def load_active_calibration(device: Device) -> structs.AnyCalibration | None:
     with local_persistent_storage("active_calibrations") as c:
         active_cal_name = c.get(device)
 
@@ -112,7 +121,7 @@ def load_active_calibration(device: str) -> structs.AnyCalibration | None:
     return load_calibration(device, active_cal_name)
 
 
-def load_calibration(device: str, calibration_name: str) -> structs.AnyCalibration:
+def load_calibration(device: Device, calibration_name: str) -> structs.AnyCalibration:
     target_file = CALIBRATION_PATH / device / f"{calibration_name}.yaml"
 
     if not target_file.exists():
@@ -127,7 +136,7 @@ def load_calibration(device: str, calibration_name: str) -> structs.AnyCalibrati
         raise ValidationError(f"Error reading {target_file.stem}: {e}")
 
 
-def list_of_calibrations_by_device(device: str) -> list[str]:
+def list_of_calibrations_by_device(device: Device) -> list[str]:
     device_dir = CALIBRATION_PATH / device
     if not device_dir.exists():
         return []