new pumps api

CHANGELOG.md

@@ -3,6 +3,7 @@
  - experiment profiles start now use the unit_api directly. I think this mitigates the huey workers stampeding on each other when try to start many jobs.
  - improved chart colors in the UI
  - /jobs/running/<job>
+ - od_reading snapshot
 
 ### 25.1.21
 

pioreactor/actions/pumps.py

@@ -0,0 +1,49 @@
+# -*- coding: utf-8 -*-
+# pumps.py
+# a higher-level CLI API than the `pio run add_*` api
+from __future__ import annotations
+
+from typing import Callable
+
+import click
+
+from pioreactor.actions.pump import add_alt_media
+from pioreactor.actions.pump import add_media
+from pioreactor.actions.pump import remove_waste
+from pioreactor.whoami import get_assigned_experiment_name
+from pioreactor.whoami import get_unit_name
+
+registered_pumps: dict[str, Callable[..., float]] = {
+    "media": add_media,
+    "alt_media": add_alt_media,
+    "waste": remove_waste,
+}
+
+
+@click.command(
+    name="pumps",
+    context_settings=dict(ignore_unknown_options=True, allow_extra_args=True),
+)
+@click.pass_context
+def click_pumps(ctx):
+    """
+    example: pio run pump --waste 2 --media 1 --waste 2
+    """
+
+    unit = get_unit_name()
+    experiment = get_assigned_experiment_name(unit)
+
+    pump_script = [(ctx.args[i][2:], ctx.args[i + 1]) for i in range(0, len(ctx.args), 2)]
+
+    for pump, volume in pump_script:
+        volume = float(volume)
+        pump = pump.rstrip("-").rstrip(
+            "_"
+        )  # why? users might be passing in the options via a key-value, and this way they can specify the same pump multiple times. Ex: experiment profiles.
+
+        pump_func = registered_pumps.get(pump.replace("-", "_"))
+
+        if pump_func:
+            pump_func(ml=volume, unit=unit, experiment=experiment)
+        else:
+            raise ValueError(pump)

pioreactor/cli/run.py

@@ -14,6 +14,7 @@
 from pioreactor.actions.pump import click_circulate_alt_media
 from pioreactor.actions.pump import click_circulate_media
 from pioreactor.actions.pump import click_remove_waste
+from pioreactor.actions.pumps import click_pumps
 from pioreactor.actions.self_test import click_self_test
 from pioreactor.automations.dosing import *  # noqa: F403, F401
 from pioreactor.automations.led import *  # noqa: F403, F401
@@ -50,6 +51,7 @@ def run() -> None:
 
 run.add_command(click_led_intensity)
 run.add_command(click_add_alt_media)
+run.add_command(click_pumps)
 run.add_command(click_add_media)
 run.add_command(click_remove_waste)
 run.add_command(click_circulate_media)