Skip to content

Commit

Permalink
more
Browse files Browse the repository at this point in the history
  • Loading branch information
@CamDavidsonPilon
CamDavidsonPilon committed Jan 4, 2025
1 parent 2ca3956 commit 7003e2a
Show file tree
Hide file tree
Showing 3 changed files with 56 additions and 68 deletions.
14 changes: 11 additions & 3 deletions pioreactor/calibrations/__init__.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -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):
Expand All @@ -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):
Expand All @@ -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):
Expand Down Expand Up @@ -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")]
98 changes: 37 additions & 61 deletions pioreactor/calibrations/pump_calibration.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -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:
Expand Down Expand Up @@ -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,
)

Expand Down Expand Up @@ -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)
Expand All @@ -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
Expand Down
12 changes: 8 additions & 4 deletions pioreactor/cli/calibrations.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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}")


Expand Down

0 comments on commit 7003e2a

Please sign in to comment.