remove gpio_helpers; it was never used, fix pop bug

CHANGELOG.md

@@ -26,7 +26,7 @@
 
     For example, to run a pump calibration, use `pio calibrations run --device media_pump`. View all your media pump calibrations with:  `pio calibrations list --device media_pump`. And to duplicate a
 
-   - For now, the actual calibrations are the same protocol as before, but in the near future, we'll be updating them with new features. Adding this CLI and YAML format was the first step.
+   - For now, the actual calibrations are the same protocol as before, but in the near future, we'll be updating them with new features. Adding this unified CLI and YAML format was the first step.
 
 
 #### Web API changes

pioreactor/actions/pump.py

@@ -293,7 +293,7 @@ def _get_pump_action(pump_device: PumpCalibrationDevices) -> str:
                 ml = float(ml)
                 if calibration is None:
                     logger.error(
-                        f"Active calibration not found. Run {pump_device} calibration first: pio calibrations run --device {pump_device} or set active with `pio run set-active`"
+                        f"Active calibration not found. Run {pump_device} calibration first: `pio calibrations run --device {pump_device}` or set active with `pio run set-active`"
                     )
                     raise exc.CalibrationError(
                         f"Active calibration not found. Run {pump_device} calibration: `pio calibrations run --device {pump_device}`, or set active with `pio run set-active`"

pioreactor/background_jobs/monitor.py

@@ -30,7 +30,6 @@
 from pioreactor.pubsub import QOS
 from pioreactor.structs import Voltage
 from pioreactor.types import MQTTMessage
-from pioreactor.utils.gpio_helpers import set_gpio_availability
 from pioreactor.utils.networking import discover_workers_on_network
 from pioreactor.utils.networking import get_ip
 from pioreactor.utils.timing import current_utc_datetime
@@ -159,9 +158,6 @@ def add_post_button_callback(cls, function: Callable) -> None:
     def _setup_GPIO(self) -> None:
         import lgpio
 
-        set_gpio_availability(BUTTON_PIN, False)
-        set_gpio_availability(LED_PIN, False)
-
         if not whoami.is_testing_env():
             self._handle = lgpio.gpiochip_open(GPIOCHIP)
 
@@ -420,9 +416,6 @@ def on_disconnected(self) -> None:
             self._button_callback.cancel()
             lgpio.gpiochip_close(self._handle)
 
-        set_gpio_availability(BUTTON_PIN, True)
-        set_gpio_availability(LED_PIN, True)
-
     def led_on(self) -> None:
         import lgpio
 

pioreactor/background_jobs/stirring.py

@@ -22,7 +22,6 @@
 from pioreactor.utils import clamp
 from pioreactor.utils import is_pio_job_running
 from pioreactor.utils import JobManager
-from pioreactor.utils.gpio_helpers import set_gpio_availability
 from pioreactor.utils.pwm import PWM
 from pioreactor.utils.streaming_calculations import PID
 from pioreactor.utils.timing import catchtime
@@ -60,7 +59,6 @@ def setup(self) -> None:
 
         # we delay the setup so that when all other checks are done (like in stirring's uniqueness), we can start to
         # use the GPIO for this.
-        set_gpio_availability(hardware.HALL_SENSOR_PIN, False)
 
         if not is_testing_env():
             self._handle = lgpio.gpiochip_open(hardware.GPIOCHIP)
@@ -97,8 +95,6 @@ def clean_up(self) -> None:
             self._edge_callback.cancel()
             lgpio.gpiochip_close(self._handle)
 
-        set_gpio_availability(hardware.HALL_SENSOR_PIN, True)
-
     def estimate(self, seconds_to_observe: float) -> float:
         return 0.0
 

pioreactor/calibrations/stirring_calibration.py

@@ -53,7 +53,7 @@ def run_stirring_calibration(
         dcs = (
             list(range(round(max_dc), round(min_dc), -3))
             + list(range(round(min_dc), round(max_dc), 3))
-            + list(range(round(max_dc), round(min_dc) - 3, -3))
+            + list(range(round(max_dc), round(min_dc), -3))
         )
         n_samples = len(dcs)
 

pioreactor/cli/calibrations.py

@@ -69,8 +69,9 @@ def _display_calibrations_by_device(device: str) -> None:
     "--device", "device", required=True, help="Target device of calibration (e.g. od, pump, stirring)."
 )
 @click.option("--protocol-name", required=False, help="name of protocol, defaults to basic builtin protocol")
+@click.option("-y", is_flag=True, help="Skip asking for confirmation for active.")
 @click.pass_context
-def run_calibration(ctx, device: str, protocol_name: str | None) -> None:
+def run_calibration(ctx, device: str, protocol_name: str | None, y: bool) -> None:
     """
     Run an interactive calibration assistant for a specific protocol.
     On completion, stores a YAML file in: /home/pioreactor/.pioreactor/storage/calibrations/<device>/<calibration_name>.yaml
@@ -107,15 +108,16 @@ def run_calibration(ctx, device: str, protocol_name: str | None) -> None:
 
     out_file = calibration_struct.save_to_disk_for_device(device)
 
-    if click.confirm(
-        f"Do you want to set this calibration as the Active Calibration for {device}?", default=True
-    ):
-        calibration_struct.set_as_active_calibration_for_device(device)
-        click.echo(f"Set{calibration_struct.calibration_name} as the active calibration for {device}.")
-    else:
-        click.echo(
-            f"Okay. You can use 'pio calibration set-active --device {device} --name {calibration_struct.calibration_name}' to set this calibration as the active one."
-        )
+    if not y:
+        if click.confirm(
+            f"Do you want to set this calibration as the Active Calibration for {device}?", default=True
+        ):
+            calibration_struct.set_as_active_calibration_for_device(device)
+            click.echo(f"Set{calibration_struct.calibration_name} as the active calibration for {device}.")
+        else:
+            click.echo(
+                f"Okay. You can use 'pio calibration set-active --device {device} --name {calibration_struct.calibration_name}' to set this calibration as the active one."
+            )
 
     click.echo(
         f"Calibration '{calibration_struct.calibration_name}' of device '{device}' saved to {out_file} ✅"
@@ -158,9 +160,14 @@ def set_active_calibration(device: str, calibration_name: str | None) -> None:
     """
 
     if calibration_name is None:
-        click.echo("No calibration name provided. Clearing active calibration.")
         with local_persistent_storage("active_calibrations") as c:
-            c.pop(device)
+            is_present = c.pop(device)
+        if is_present:
+            click.echo(f"No calibration name provided. Clearing active calibration for {device}.")
+        else:
+            click.echo(
+                f"No calibration name provided. Tried clearing active calibration for {device}, but didn't find one."
+            )
 
     else:
         data = load_calibration(device, calibration_name)

pioreactor/cli/pio.py

@@ -196,62 +196,24 @@ def cache():
 @cache.command(name="view", short_help="print out the contents of a cache")
 @click.argument("cache")
 def view_cache(cache: str) -> None:
-    from pathlib import Path
-    import tempfile
-
-    tmp_dir = tempfile.gettempdir()
-
-    persistant_dir = (
-        "/home/pioreactor/.pioreactor/storage/" if not whoami.is_testing_env() else ".pioreactor/storage"
-    )
-
-    # is it a temp cache or persistant cache?
-    if Path(f"{tmp_dir}/{cache}").is_dir():
-        cacher = local_intermittent_storage
-
-    elif Path(f"{persistant_dir}/{cache}").is_dir():
-        cacher = local_persistent_storage
-
-    else:
-        click.echo(f"cache {cache} not found.")
-        return
-
-    with cacher(cache) as c:
-        for key in sorted(list(c.iterkeys())):
-            click.echo(f"{click.style(key, bold=True)} = {c[key]}")
+    for cacher in [local_intermittent_storage, local_persistent_storage]:  # TODO: this sucks
+        with cacher(cache) as c:
+            for key in sorted(list(c.iterkeys())):
+                click.echo(f"{click.style(key, bold=True)} = {c[key]}")
 
 
 @cache.command(name="clear", short_help="clear out the contents of a cache")
 @click.argument("cache")
 @click.argument("key")
 @click.option("--as-int", is_flag=True, help="evict after casting key to int, useful for gpio pins.")
 def clear_cache(cache: str, key: str, as_int: bool) -> None:
-    from pathlib import Path
-    import tempfile
-
-    tmp_dir = tempfile.gettempdir()
-
-    persistant_dir = (
-        "/home/pioreactor/.pioreactor/storage/" if not whoami.is_testing_env() else ".pioreactor/storage"
-    )
-
-    # is it a temp cache or persistant cache?
-    if Path(f"{tmp_dir}/{cache}").is_dir():
-        cacher = local_intermittent_storage
-
-    elif Path(f"{persistant_dir}/{cache}").is_dir():
-        cacher = local_persistent_storage
-
-    else:
-        click.echo(f"cache {cache} not found.")
-        return
-
-    with cacher(cache) as c:
-        if as_int:
-            key = int(key)  # type: ignore
+    for cacher in [local_intermittent_storage, local_persistent_storage]:
+        with cacher(cache) as c:
+            if as_int:
+                key = int(key)  # type: ignore
 
-        if key in c:
-            del c[key]
+            if key in c:
+                del c[key]
 
 
 @pio.command(

pioreactor/structs.py

@@ -191,14 +191,16 @@ class ODCalibration(CalibrationBase, kw_only=True, tag="od"):
     minimum_od600: float
     minimum_voltage: float
     maximum_voltage: float
+    x: str = "Voltage"
+    y: str = "OD600"
 
 
 class SimplePeristalticPumpCalibration(CalibrationBase, kw_only=True, tag="simple_peristaltic_pump"):
     hz: t.Annotated[float, Meta(ge=0)]
     dc: t.Annotated[float, Meta(ge=0)]
     voltage: float
-    x: str = "duration"
-    y: str = "volume"
+    x: str = "Duration"
+    y: str = "Volume"
 
     @property
     def duration_(self):

pioreactor/utils/__init__.py

@@ -334,10 +334,12 @@ def iterkeys(self):
     def pop(self, key, default=None):
         self.cursor.execute(f"SELECT value FROM {self.table_name} WHERE key = ?", (key,))
         result = self.cursor.fetchone()
+        self.cursor.execute(f"DELETE FROM {self.table_name} WHERE key = ?", (key,))
+
         if result is None:
             return default
-        self.cursor.execute(f"DELETE FROM {self.table_name} WHERE key = ?", (key,))
-        return result[0]
+        else:
+            return result[0]
 
     def __contains__(self, key):
         self.cursor.execute(f"SELECT 1 FROM {self.table_name} WHERE key = ?", (key,))

pioreactor/utils/pwm.py

@@ -18,7 +18,6 @@
 from pioreactor.pubsub import create_client
 from pioreactor.types import GpioPin
 from pioreactor.utils import clamp
-from pioreactor.utils import gpio_helpers
 from pioreactor.utils import local_intermittent_storage
 from pioreactor.version import rpi_version_info
 from pioreactor.whoami import get_assigned_experiment_name
@@ -199,8 +198,6 @@ def __init__(
             self.logger.error(msg)
             raise PWMError(msg)
 
-        gpio_helpers.set_gpio_availability(self.pin, False)
-
         self._pwm: HardwarePWMOutputDevice | SoftwarePWMOutputDevice | MockPWMOutputDevice
 
         if is_testing_env():
@@ -287,8 +284,6 @@ def clean_up(self) -> None:
         with local_intermittent_storage("pwm_dc") as cache:
             cache.pop(self.pin)
 
-        gpio_helpers.set_gpio_availability(self.pin, True)
-
         self.logger.debug(f"Cleaned up GPIO-{self.pin}.")
 
         if not self._external_client: