Where's the right place to define a calibration type?

Software
1

I’ve added a pH probe calibration, but when I want to call “load calibration” I get an error because the YAML decoder only recognises the subclasses hard-coded in AnyCalibration in structs. Is there a reason that calibration types outside of these might fail, and this error is correct? Or does this need changing to something like AnyCalibration = subclass_union(CalibrationBase) ?

2 
>>> ph_cal = load_active_calibration("ph_probe")
Traceback (most recent call last):
  File "/usr/local/lib/python3.11/dist-packages/pioreactor/calibrations/__init__.py", line 133, in load_calibration
    data = yaml_decode(target_file.read_bytes(), type=structs.subclass_union(structs.CalibrationBase))
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.11/dist-packages/msgspec/yaml.py", line 186, in decode
    return _convert(
           ^^^^^^^^^
msgspec.ValidationError: Invalid value 'ph' - at `$.calibration_type`

print(structs.subclass_union(CalibrationBase))
typing.Union[pioreactor.structs.SimpleStirringCalibration, pioreactor.structs.SimplePeristalticPumpCalibration, pioreactor.structs.ODCalibration]
3

Hm, I don’t think this should happen. When you define your pH calibration, it should be a subclass of CalibrationBase, and hence subclass_union(CalibrationBase) should include it.

  1. Can you confirm your ph calibration is a direct subclass of CalibrationBase?
  2. I can see you’re in a python shell, has your ph calibration been “hydrated” - that is, it’s been defined in that shell either via a direct or indirect import, or defined in the shell?
4

Sorry the shell was just to illustrate. This is where I define the calibration

from pioreactor.calibrations import CalibrationProtocol
from pioreactor.structs import CalibrationBase
from pioreactor.utils.timing import current_utc_datetime
from pioreactor.whoami import get_unit_name
import typing as t

class PHCalibration(CalibrationBase, kw_only=True, tag="ph"):

    electrode_type: str
    x: str = "pH"
    y: str = "Voltage"

    def voltage_to_ph(self, voltage: float) -> float:
        return self.y_to_x(voltage)
    def ph_to_voltage(self, ph: float) -> float:
        return self.x_to_y(ph)

class BufferBasedPHProtocol(CalibrationProtocol):
    target_device = "ph_probe"
    protocol_name = "buffer_based"
    description = "Calibrate the pH sensor using buffer solutions"

    def run(self, target_device: str) -> PHCalibration:
        return run_ph_calibration()

def calculate_poly_curve_of_best_fit(x: list[float], y: list[float], degree: int) -> list[float]:
    import numpy as np

    try:
        coefs = np.polyfit(x, y, deg=degree)
    except Exception:
        print("Unable to fit.")
        coefs = np.zeros(degree)

    return coefs.tolist()

def run_ph_calibration() -> PHCalibration:


    unit_name = get_unit_name()

    # run the calibration

   recorded_xs = [4,7,10,4,7,10]
    recorded_ys = [1.51, 1.12, 0.79, 1.49, 1.12, 0.79]

    curve_data = calculate_poly_curve_of_best_fit(recorded_xs, recorded_ys,  1)

    return PHCalibration(
        calibration_name=f"pH-{current_utc_datetime().strftime('%Y-%m-%d')}",
        created_at=current_utc_datetime(),
        curve_data_ = curve_data, # ax1, intercept
        curve_type="poly",
        x="pH",
        y="Voltage",
        recorded_data={"x":recorded_xs, "y": recorded_ys},
        calibrated_on_pioreactor_unit = unit_name,
        electrode_type="Vernier-FPH"
    )
5

Then this is how I’ve managed to get my script to load the calibration value (vernier-FPH-BTA/vernier_fph_bta/vernier_fph_bta.py at main · Change-Bio/vernier-FPH-BTA · GitHub)

Significant code for this example is:

from pioreactor.plugin_management import get_plugins
from pioreactor.calibrations import load_active_calibration

ph_calibration = get_plugins()["ph_calibration"].module.PHCalibration # we need to create this subclass here in order for load_active_calibration to work
self.ph_cal = load_active_calibration("ph_probe")
6

It’s all still pretty hacky but it’s working for now, I just hardcode the calibration values at the moment but will integrate a CLI borrowing from the OD calibration script at some point

7

So the PHCalibration isn’t seen until the plugin is loaded, which occurs when pioreactor.plugin_management.get_plugins runs. This is why your code only works after you run that (I don’t think you need the module bits after).

Now, get_plugins implicity runs when you invoke pioreactor with pio - are you doing that? OTOH, get_plugins doesn’t run in the shell unless you explicitly run it.

Are you running ReadPh via pio run vernier_fph_bta?

Oh! Also, plugins run in lexographic order of script name, so make sure PHCalibration is defined in a python file prior to ReadPh. You can use p01_x.py, p02_y.py file names to achieve this (as an example)

8

Thanks for the help! The reactor is running now so will leave it as it is but I can explore further next week. The Vernier plugin with ReadPh is installed through pip and run through the UI, guess this runs pio run in the background? Then PH calibration I run using pio calibrations run --device ph_probe.

The goal is to run pio calibrations run --device ph_probe before a run and then use the UI to start reading pH (it just reads the voltage from MQTT, converts using the calibration, and then republishes again)