diff --git a/pioreactor/calibrations/__init__.py b/pioreactor/calibrations/__init__.py index f5d05eef..fd513f05 100644 --- a/pioreactor/calibrations/__init__.py +++ b/pioreactor/calibrations/__init__.py @@ -61,7 +61,7 @@ class DurationBasedMediaPumpProtocol(CalibrationProtocol): def run(self) -> structs.SimplePeristalticPumpCalibration: from pioreactor.calibrations.pump_calibration import run_pump_calibration - return run_pump_calibration() + return run_pump_calibration(self.target_device) class DurationBasedAltMediaPumpProtocol(CalibrationProtocol): @@ -71,7 +71,7 @@ class DurationBasedAltMediaPumpProtocol(CalibrationProtocol): def run(self) -> structs.SimplePeristalticPumpCalibration: from pioreactor.calibrations.pump_calibration import run_pump_calibration - return run_pump_calibration() + return run_pump_calibration(self.target_device) class DurationBasedWasteMediaPumpProtocol(CalibrationProtocol): @@ -81,7 +81,7 @@ class DurationBasedWasteMediaPumpProtocol(CalibrationProtocol): def run(self) -> structs.SimplePeristalticPumpCalibration: from pioreactor.calibrations.pump_calibration import run_pump_calibration - return run_pump_calibration() + return run_pump_calibration(self.target_device) class DCBasedStirringProtocol(CalibrationProtocol): @@ -134,3 +134,11 @@ def load_calibration(device: str, calibration_name: str) -> structs.AnyCalibrati return data except ValidationError as e: raise ValidationError(f"Error reading {target_file.stem}: {e}") + + +def list_of_calibrations_by_device(device: str) -> list[str]: + calibration_dir = CALIBRATION_PATH / device + if not calibration_dir.exists(): + return [] + + return [file.stem for file in calibration_dir.glob("*.yaml")] diff --git a/pioreactor/calibrations/pump_calibration.py b/pioreactor/calibrations/pump_calibration.py index 9374fd4f..b2f8df48 100644 --- a/pioreactor/calibrations/pump_calibration.py +++ b/pioreactor/calibrations/pump_calibration.py @@ -24,13 +24,12 @@ from pioreactor.actions.pump import add_alt_media from pioreactor.actions.pump import add_media from pioreactor.actions.pump import remove_waste -from pioreactor.calibrations import load_active_calibration +from pioreactor.calibrations import list_of_calibrations_by_device from pioreactor.calibrations.utils import curve_to_callable from pioreactor.config import config from pioreactor.hardware import voltage_in_aux from pioreactor.logging import create_logger from pioreactor.types import PumpCalibrationDevices -from pioreactor.utils import local_persistent_storage from pioreactor.utils import managed_lifecycle from pioreactor.utils.math_helpers import correlation from pioreactor.utils.math_helpers import simple_linear_regression_with_forced_nil_intercept @@ -53,86 +52,55 @@ def bold(string: str) -> str: return style(string, bold=True) -def introduction() -> None: +def introduction(pump_device) -> None: import logging logging.disable(logging.WARNING) echo( - """This routine will calibrate the pumps on your current Pioreactor. You'll need: + f"""This routine will calibrate the {pump_device} on your current Pioreactor. You'll need: 1. A Pioreactor 2. A vial placed on a scale with accuracy at least 0.1g OR an accurate graduated cylinder. 3. A larger container filled with water - 4. A pump connected to the correct PWM channel (1, 2, 3, or 4) as determined in your Configuration. + 4. {pump_device} connected to the correct PWM channel (1, 2, 3, or 4) as determined in your configuration. We will dose for a set duration, you'll measure how much volume was expelled, and then record it back here. After doing this a few times, we can construct a calibration line for this pump. """ ) - confirm(green("Proceed?")) + confirm(green("Proceed?"), abort=True, default=True) clear() echo( "You don't need to place your vial in your Pioreactor. While performing this calibration, keep liquids away from the Pioreactor to keep it safe & dry" ) - confirm(green("Proceed?")) + confirm(green("Proceed?"), abort=True, default=True) clear() def get_metadata_from_user(pump_device: PumpCalibrationDevices) -> str: - with local_persistent_storage("pump_calibrations") as cache: - while True: - name = prompt( - style( - f"Optional: Provide a name for this calibration. [enter] to use default name `{pump_device}-{current_utc_datestamp()}`", - fg="green", - ), - type=str, - default=f"{pump_device}-{current_utc_datestamp()}", - show_default=False, - ).strip() - if name == "": - echo("Name cannot be empty") - continue - elif name in cache: - if confirm(green("❗️ Name already exists. Do you wish to overwrite?")): - break - elif name == "current": - echo("Name cannot be `current`.") - continue - else: + existing_calibrations = list_of_calibrations_by_device(pump_device) + while True: + name = prompt( + style( + f"Optional: Provide a name for this {pump_device} calibration. [enter] to use default name `{pump_device}-{current_utc_datestamp()}`", + fg="green", + ), + type=str, + default=f"{pump_device}-{current_utc_datestamp()}", + show_default=False, + ).strip() + if name == "": + echo("Name cannot be empty") + continue + elif name in existing_calibrations: + if confirm(green("❗️ Name already exists. Do you wish to overwrite?")): break + else: + break return name -def which_pump_are_you_calibrating() -> tuple[PumpCalibrationDevices, Callable]: - m = load_active_calibration("media_pump") - a = load_active_calibration("alt_media_pump") - w = load_active_calibration("waste_pump") - - echo(green(bold("Step 1"))) - r = prompt( - green( - f"""Which pump are you calibrating? -1. Media {f'[{m.calibration_name}, last ran {m.created_at:%d %b, %Y}]' if m else '[No calibration]'} -2. Alt-media {f'[{a.calibration_name}, last ran {a.created_at:%d %b, %Y}]' if a else '[No calibration]'} -3. Waste {f'[{w.calibration_name}, last ran {w.created_at:%d %b, %Y}]' if w else '[No calibration]'} -""", - ), - type=click.Choice(["1", "2", "3"]), - show_choices=True, - ) - - if r == "1": - return ("media_pump", add_media) - elif r == "2": - return ("alt_media_pump", add_alt_media) - elif r == "3": - return ("waste_pump", remove_waste) - else: - raise ValueError() - - def setup( pump_device: PumpCalibrationDevices, execute_pump: Callable, hz: float, dc: float, unit: str ) -> None: @@ -208,10 +176,10 @@ def choose_settings() -> tuple[float, float]: ) dc = prompt( green( - "Optional: Enter duty cycle percent as a whole number. [enter] for default 95%", + "Optional: Enter duty cycle percent as a whole number. [enter] for default 100%", ), type=click.IntRange(0, 100), - default=95, + default=100, show_default=False, ) @@ -361,7 +329,7 @@ def save_results( def run_pump_calibration( - min_duration: float = 0.40, max_duration: float = 1.5 + pump_device, min_duration: float = 0.40, max_duration: float = 1.5 ) -> structs.SimplePeristalticPumpCalibration: unit = get_unit_name() experiment = get_assigned_experiment_name(unit) @@ -371,9 +339,17 @@ def run_pump_calibration( with managed_lifecycle(unit, experiment, "pump_calibration"): clear() - introduction() + introduction(pump_device) + + if pump_device == "media_pump": + execute_pump = add_media + elif pump_device == "alt_media_pump": + execute_pump = add_alt_media + elif pump_device == "waste_pump": + execute_pump = remove_waste + else: + raise ValueError() - pump_device, execute_pump = which_pump_are_you_calibrating() name = get_metadata_from_user(pump_device) is_ready = True diff --git a/pioreactor/cli/calibrations.py b/pioreactor/cli/calibrations.py index 3d3fa2c5..327cc7d1 100644 --- a/pioreactor/cli/calibrations.py +++ b/pioreactor/cli/calibrations.py @@ -8,6 +8,7 @@ from pioreactor import structs from pioreactor.calibrations import CALIBRATION_PATH from pioreactor.calibrations import calibration_protocols +from pioreactor.calibrations import list_of_calibrations_by_device from pioreactor.calibrations import load_calibration from pioreactor.calibrations.utils import curve_to_callable from pioreactor.calibrations.utils import plot_data @@ -38,14 +39,17 @@ def list_calibrations(device: str) -> None: click.echo("-" * len(header)) with local_persistent_storage("active_calibrations") as c: - for file in calibration_dir.glob("*.yaml"): + for name in list_of_calibrations_by_device(device): try: - data = yaml_decode(file.read_bytes(), type=structs.subclass_union(structs.CalibrationBase)) + location = (calibration_dir / name).with_suffix(".yaml") + data = yaml_decode( + location.read_bytes(), type=structs.subclass_union(structs.CalibrationBase) + ) active = c.get(device) == data.calibration_name - row = f"{data.calibration_name:<50}{data.created_at.strftime('%Y-%m-%d %H:%M:%S'):<25}{'✅' if active else '':<10}{file}" + row = f"{data.calibration_name:<50}{data.created_at.strftime('%Y-%m-%d %H:%M:%S'):<25}{'✅' if active else '':<10}{location}" click.echo(row) except Exception as e: - error_message = f"Error reading {file.stem}: {e}" + error_message = f"Error reading {name}: {e}" click.echo(f"{error_message:<60}")