align it with the existing light/dark cycle automation

Author: CamDavidsonPilon

README.md

@@ -28,7 +28,8 @@ This plugin is also available on the Pioreactor web interface, in the _Plugins_
 Type into your command line:
 
 ```
-pio run led_calibration
+pio calibrations --device-name led_C --protocol-name led_calibration
+pio calibrations --device-name led_D --protocol-name led_calibration
 ```
 
 To perform this calibration, insert your vial containing media into the Pioreactor and submerge your light probe. Follow the prompts on the command line. The plugin will increase the light intensity, and prompt you to record the readings from your light probe. A calibration line of best fit will be generated based on your light probe readings.
@@ -39,32 +40,6 @@ An automation will become available on the web interface. To use this automation
 
 In the _Pioreactors_ tab, under _Manage_, you can _Start_ an _LED automation_. A new option becomes available in the drop-down menu called "Calibrated Light/Dark Cycle". Input your desired light intensity in AU (ex. 1000 AU). The automation will set the percent light intensity such that an output of 1000 AU occurs on **both** LEDs.
 
-## Subcommands
-
-Run a subcommand by typing the following into the command line:
-```
-pio run led_calibration <SUBCOMMAND>
-```
-The following subcommands are available:
-
-### **list**
-Prints a table with all existing calibrations stored on the leader. Headings include unique names, timestamps, and channels.
-
-| Name | Timestamp | Channel | Currently in use? |
-|------|----------|---------|-------------------|
-| Algae_C_2022 | 2022-08-29T20:12:00.400000Z | C | ✅ |
-| Algae_B_2022 | 2022-08-29T20:13:00.400000Z | B | ✅ |
-| Algae_B_2021 | 2021-08-29T20:15:00.400000Z | B |  |
-
-### **display**
-Displays the graph and data for the current calibration for each channel A, B, C, and D, if it exists. For example, for the data above, the current calibrations for Algae_C_2022 and Algae_B_2022 will be displayed.
-
-### **change_current**
-If you would like to change a current calibration to a previous one, use `change_current "<UNIQUE NAME>"`. These changes are based on the channel assigned to the calibration.
-
-For example:
-`pio run led_calibration change_current "Algae_B_2021"` would replace Algae_B_2022, since only one calibation is active per channel.
-
 ## When to perform an LED calibration
 
 Calibrations should be performed on a case-by-case basis. A new calibration must be performed per channel, and/or for new LED cables, and with any change in media that can alter the light intensity within the vial.

led_calibration_plugin/calibrated_light_dark_cycle.py

@@ -1,48 +1,48 @@
 # -*- coding: utf-8 -*-
 from __future__ import annotations
 
-from msgspec.json import decode
+from typing import Optional
+
 from pioreactor.automations import events
 from pioreactor.automations.led.base import LEDAutomationJobContrib
+from pioreactor.calibrations import load_active_calibration
 from pioreactor.exc import CalibrationError
 from pioreactor.types import LedChannel
 from pioreactor.utils import clamp
 from pioreactor.utils import local_persistent_storage
 
-from .led_calibration import LEDCalibration
-
 
 class CalibratedLightDarkCycle(LEDAutomationJobContrib):
     """
-    Follows as h light / h dark cycle. Starts light ON.
+    Follows as min light / min dark cycle. Starts light ON.
     """
 
-    automation_name = "calibrated_light_dark_cycle"
+    automation_name: str = "calibrated_light_dark_cycle"
     published_settings = {
         "duration": {
             "datatype": "float",
             "settable": False,
             "unit": "min",
         },
-        "light_intensity": {"datatype": "float", "settable": True, "unit": "AU"},
-        "light_duration_hours": {"datatype": "integer", "settable": True, "unit": "h"},
-        "dark_duration_hours": {"datatype": "integer", "settable": True, "unit": "h"},
+        "light_intensity": {"datatype": "float", "settable": True, "unit": "%"},
+        "light_duration_minutes": {"datatype": "integer", "settable": True, "unit": "min"},
+        "dark_duration_minutes": {"datatype": "integer", "settable": True, "unit": "min"},
     }
 
     def __init__(
         self,
-        light_intensity: float,
-        light_duration_hours: int,
-        dark_duration_hours: int,
+        light_intensity: float | str,
+        light_duration_minutes: int | str,
+        dark_duration_minutes: int | str,
         **kwargs,
     ):
         super().__init__(**kwargs)
-        self.hours_online: int = -1
+        self.minutes_online: int = -1
         self.light_active: bool = False
-        self.channels: list[LedChannel] = ["C", "D"]
+        self.channels: list[LedChannel] = ["D", "C"]
         self.set_light_intensity(light_intensity)
-        self.light_duration_hours = float(light_duration_hours)
-        self.dark_duration_hours = float(dark_duration_hours)
+        self.light_duration_minutes = float(light_duration_minutes)
+        self.dark_duration_minutes = float(dark_duration_minutes)
 
         with local_persistent_storage("active_calibrations") as cache:
             for channel in self.channels:
@@ -52,55 +52,66 @@ def __init__(
     def channel_to_led_device(self, channel: LedChannel) -> str:
         return f"led_{channel}"
 
-    def execute(self) -> events.AutomationEvent:
-        self.hours_online += 1
-        return self.trigger_leds(self.hours_online)
+    def execute(self) -> Optional[events.AutomationEvent]:
+        # runs every minute
+        self.minutes_online += 1
+        return self.trigger_leds(self.minutes_online)
+
+    def calculate_intensity_percent(self, channel: str) -> float:
 
-    def calculate_intensity_percent(self, channel):
-        with local_persistent_storage("current_led_calibration") as cache:
-            led_calibration = decode(
-                cache[self.channel_to_led_device(channel)], type=LEDCalibration
-            )
+        led_calibration = load_active_calibration(self.channel_to_led_device(channel))
 
-            intensity_percent = (
-                self.light_intensity - led_calibration.curve_data_[1]
-            ) / led_calibration.curve_data_[0]
+        intensity_percent = led_calibration.ipredict(self.light_intensity)
 
-            return clamp(0, intensity_percent, 100)
+        return clamp(0, intensity_percent, 100)
 
-    def trigger_leds(self, hours: int) -> events.AutomationEvent:
+    def trigger_leds(self, minutes: int) -> Optional[events.AutomationEvent]:
+        """
+        Changes the LED state based on the current minute in the cycle.
 
-        cycle_duration = self.light_duration_hours + self.dark_duration_hours
+        The light and dark periods are calculated as multiples of 60 minutes, forming a cycle.
+        Based on where in this cycle the current minute falls, the light is either turned ON or OFF.
 
-        if ((hours % cycle_duration) < self.light_duration_hours) and (not self.light_active):
+        Args:
+            minutes: The current minute of the cycle.
+
+        Returns:
+            An instance of AutomationEvent, indicating that LEDs' status might have changed.
+            Returns None if the LEDs' state didn't change.
+        """
+        cycle_duration_min = int(self.light_duration_minutes + self.dark_duration_minutes)
+        if ((minutes % cycle_duration_min) < (self.light_duration_minutes)) and (
+            not self.light_active
+        ):
             self.light_active = True
 
             for channel in self.channels:
                 intensity_percent = self.calculate_intensity_percent(channel)
                 self.set_led_intensity(channel, intensity_percent)
 
-            return events.ChangedLedIntensity(f"{hours}h: turned on LEDs.")
+            return events.ChangedLedIntensity(f"{minutes:.1f}min: turned on LEDs.")
 
-        elif ((hours % cycle_duration) >= self.light_duration_hours) and (self.light_active):
+        elif ((minutes % cycle_duration_min) >= (self.light_duration_minutes)) and (
+            self.light_active
+        ):
             self.light_active = False
             for channel in self.channels:
                 self.set_led_intensity(channel, 0)
-            return events.ChangedLedIntensity(f"{hours}h: turned off LEDs.")
+            return events.ChangedLedIntensity(f"{minutes:.1f}min: turned off LEDs.")
 
         else:
-            return events.NoEvent(f"{hours}h: no change.")
-
-    def set_dark_duration_hours(self, hours: int):
-
-        self.dark_duration_hours = hours
+            return None
 
-        self.trigger_leds(self.hours_online)
+    # minutes setters
+    def set_dark_duration_minutes(self, minutes: int):
+        self.dark_duration_minutes = minutes
 
-    def set_light_duration_hours(self, hours: int):
+        self.trigger_leds(self.minutes_online)
 
-        self.light_duration_hours = hours
+    def set_light_duration_minutes(self, minutes: int):
+        self.light_duration_minutes = minutes
 
-        self.trigger_leds(self.hours_online)
+        self.trigger_leds(self.minutes_online)
 
     def set_light_intensity(self, intensity_au):
         # this is the setter of light_intensity attribute, eg. called when updated over MQTT
@@ -116,6 +127,6 @@ def set_light_intensity(self, intensity_au):
             pass
 
     def set_duration(self, duration: float) -> None:
-        if duration != 60:
-            self.logger.warning("Duration should be set to 60.")
+        if duration != 1:
+            self.logger.warning("Duration should be set to 1.")
         super().set_duration(duration)

led_calibration_plugin/test_led_calibration.py

@@ -7,7 +7,7 @@
 from pioreactor.exc import CalibrationError
 from pioreactor.utils import local_intermittent_storage
 from pioreactor.utils import local_persistent_storage
-from pioreactor.utils.timing import current_utc_timestamp
+from pioreactor.utils.timing import current_utc_datetime
 from pioreactor.whoami import get_unit_name
 
 from .calibrated_light_dark_cycle import CalibratedLightDarkCycle
@@ -24,30 +24,30 @@ def test_led_fails_if_calibration_not_present():
 
     with local_persistent_storage("active_calibrations") as cache:
         if "led_C" in cache:
-            del cache["led_C"]
+            cache.pop("led_C")
         if "led_D" in cache:
-            del cache["led_D"]
+            cache.pop("led_D")
 
     with pytest.raises(CalibrationError):
 
         with CalibratedLightDarkCycle(
-            duration=0.01,
+            duration=1,
             light_intensity=-1,
-            light_duration_hours=16,
-            dark_duration_hours=8,
+            light_duration_minutes=16,
+            dark_duration_minutes=8,
             unit=unit,
             experiment=experiment,
         ):
 
-            pause(8)
+            pass
 
 
 def test_set_intensity_au_above_max() -> None:
     experiment = "test_set_intensity_au_above_max"
     unit = get_unit_name()
 
     cal = LEDCalibration(
-        created_at=current_utc_timestamp(),
+        created_at=current_utc_datetime(),
         calibrated_on_pioreactor_unit=unit,
         calibration_name=experiment,
         curve_data_=[1, 0],
@@ -60,14 +60,14 @@ def test_set_intensity_au_above_max() -> None:
     cal.set_as_active_calibration_for_device("led_D")
 
     with CalibratedLightDarkCycle(
-        duration=0.01,
+        duration=1,
         light_intensity=1500,
-        light_duration_hours=16,
-        dark_duration_hours=8,
+        light_duration_minutes=16,
+        dark_duration_minutes=8,
         unit=unit,
         experiment=experiment,
     ) as lc:
-
+        pause(10)
         assert lc.light_intensity == 1500  # test returns light_intensity (au)
 
         lc.set_light_intensity(2000)
@@ -80,7 +80,7 @@ def test_set_intensity_au_negative() -> None:
     unit = get_unit_name()
 
     cal = LEDCalibration(
-        created_at=current_utc_timestamp(),
+        created_at=current_utc_datetime(),
         calibrated_on_pioreactor_unit=unit,
         calibration_name=experiment,
         curve_data_=[1, 0],
@@ -93,10 +93,10 @@ def test_set_intensity_au_negative() -> None:
     cal.set_as_active_calibration_for_device("led_D")
 
     with CalibratedLightDarkCycle(
-        duration=0.01,
+        duration=1,
         light_intensity=-1,
-        light_duration_hours=16,
-        dark_duration_hours=8,
+        light_duration_minutes=16,
+        dark_duration_minutes=8,
         unit=unit,
         experiment=experiment,
     ) as lc: