that was pretty easy

pioreactor/background_jobs/stirring.py

@@ -36,16 +36,6 @@ class RpmCalculator:
     """
     Super class for determining how to calculate the RPM from the hall sensor.
 
-    We do some funky things with RPi.GPIO here.
-
-    1) to minimize global imports, we import in init, and attach the module to self.
-    2) More egregious: we previously had this class call `add_event_detect` and afterwards `remove_event_detect`
-       in each __call__ - this made sure that we were saving CPU resources when we were not measuring the RPM.
-       This was causing `Bus error`, and crashing Python. What I think was happening was that the folder
-       `/sys/class/gpio/gpio25` was constantly being written and deleted in each __call__, causing problems with the
-       SD card. Anyways, what we do now is turn the pin from IN to OUT inbetween the calls to RPM measurement. This
-       is taken care of in `turn_{on,off}_collection`. Flipping this only writes to `/sys/class/gpio/gpio15/direction` once.
-
     Examples
     -----------
 
@@ -71,7 +61,7 @@ def setup(self) -> None:
 
     def turn_off_collection(self) -> None:
         self.collecting = False
-        self.hall_sensor_input_device.when_activated = None
+        self.hall_sensor_input_device.when_activated = None  # TODO: supress warning
 
     def turn_on_collection(self) -> None:
         self.collecting = True

pioreactor/cli/pio.py

@@ -29,10 +29,8 @@
 from pioreactor.config import get_leader_hostname
 from pioreactor.logging import create_logger
 from pioreactor.mureq import get
-from pioreactor.utils import is_pio_job_running
 from pioreactor.utils import local_intermittent_storage
 from pioreactor.utils import local_persistant_storage
-from pioreactor.utils.gpio_helpers import temporarily_set_gpio_unavailable
 from pioreactor.utils.networking import add_local
 
 
@@ -151,41 +149,13 @@ def log(message: str, level: str, name: str, local_only: bool):
 
 @pio.command(name="blink", short_help="blink LED")
 def blink() -> None:
-    monitor_running = is_pio_job_running("monitor")
-
-    if not monitor_running:
-        import RPi.GPIO as GPIO  # type: ignore
-
-        GPIO.setmode(GPIO.BCM)
-
-        from pioreactor.hardware import PCB_LED_PIN as LED_PIN
-
-        def led_on() -> None:
-            GPIO.output(LED_PIN, GPIO.HIGH)
-
-        def led_off() -> None:
-            GPIO.output(LED_PIN, GPIO.LOW)
-
-        with temporarily_set_gpio_unavailable(LED_PIN):
-            GPIO.setup(LED_PIN, GPIO.OUT)
-
-            for _ in range(4):
-                led_on()
-                sleep(0.14)
-                led_off()
-                sleep(0.14)
-                led_on()
-                sleep(0.14)
-                led_off()
-                sleep(0.45)
-
-            GPIO.cleanup(LED_PIN)
-
-    else:
-        pubsub.publish(
-            f"pioreactor/{whoami.get_unit_name()}/{whoami.UNIVERSAL_EXPERIMENT}/monitor/flicker_led_response_okay",
-            1,
-        )
+    """
+    monitor job is required to be running.
+    """
+    pubsub.publish(
+        f"pioreactor/{whoami.get_unit_name()}/{whoami.UNIVERSAL_EXPERIMENT}/monitor/flicker_led_response_okay",
+        1,
+    )
 
 
 @pio.command(name="kill", short_help="kill job(s)")

pioreactor/tests/conftest.py

@@ -8,12 +8,8 @@
 
 # Replace libraries by fake RPi ones
 
-sys.modules["RPi"] = fake_rpi.RPi  # Fake RPi
-sys.modules["RPi.GPIO"] = fake_rpi.RPi.GPIO  # Fake GPIO
 sys.modules["smbus"] = fake_rpi.smbus  # Fake smbus (I2C)
 
-fake_rpi.toggle_print(False)
-
 
 @pytest.fixture(autouse=True)
 def run_around_tests(request):

pioreactor/utils/pwm.py

@@ -14,6 +14,7 @@
 from pioreactor.pubsub import Client
 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.whoami import get_latest_experiment_name
@@ -29,6 +30,53 @@
         pass
 
 
+class HardwarePWMOutputDevice(HardwarePWM):
+    HARDWARE_PWM_CHANNELS: dict[GpioPin, int] = {12: 0, 13: 1}
+
+    def __init__(self, pin: GpioPin, active_high=True, initial_value: float = 0.0, frequency=100):
+        if (
+            pin not in self.HARDWARE_PWM_CHANNELS
+        ):  # Only GPIO pins 18 and 19 are supported for hardware PWM
+            raise ValueError(
+                "Only GPIO pins 12 (PWM channel 0) and 13 (PWM channel 1) are supported."
+            )
+
+        pwm_channel = self.HARDWARE_PWM_CHANNELS[pin]
+        super().__init__(pwm_channel, hz=frequency)
+        self._value = initial_value
+
+        if initial_value:
+            self.on()
+
+    def close(self):
+        self.stop()
+        super().close()
+
+    def on(self):
+        self.change_duty_cycle(1.0)
+        self._value = 1.0
+
+    def off(self):
+        self.change_duty_cycle(0.0)
+        self._value = 0.0
+
+    def toggle(self):
+        if self.value:
+            self.off()
+        else:
+            self.on()
+
+    @property
+    def value(self) -> float:
+        return self._value
+
+    @value.setter
+    def value(self, value: float) -> None:
+        value = clamp(0.0, value, 1.0)
+        self.change_duty_cycle(value * 100.0)
+        self._value = value
+
+
 class PWM:
     """
     This class abstracts out the Rpi's PWM library details
@@ -70,7 +118,7 @@ class PWM:
     >
     """
 
-    HARDWARE_PWM_CHANNELS: dict[GpioPin, int] = {12: 0, 13: 1}
+    HARDWARE_PWM_CHANNELS: set[GpioPin] = {12, 13}
 
     def __init__(
         self,
@@ -112,23 +160,18 @@ def __init__(
         gpio_helpers.set_gpio_availability(self.pin, False)
 
         if (not always_use_software) and (pin in self.HARDWARE_PWM_CHANNELS):
-            self.pwm = HardwarePWM(self.HARDWARE_PWM_CHANNELS[self.pin], self.hz)
-
+            self.pwm = HardwarePWMOutputDevice(self.pin, self.hz)
         else:
-            import RPi.GPIO as GPIO  # type: ignore
-
-            GPIO.setwarnings(
-                False
-            )  # we already "registered" this GPIO in the EEPROM, ignore GPIO telling us again.
-            GPIO.setmode(GPIO.BCM)
-            GPIO.setup(self.pin, GPIO.OUT, initial=GPIO.LOW)
+            from gpiozero import PWMOutputDevice
 
             if self.hz >= 1000:
                 self.logger.warning(
                     "Setting a PWM to a very high frequency with software. Did you mean to use a hardware PWM?"
                 )
 
-            self.pwm = GPIO.PWM(self.pin, self.hz)
+            self.pwm = PWMOutputDevice(
+                self.pin, initial_value=0, active_high=True, frequency=self.hz
+            )
 
         with local_intermittent_storage("pwm_hz") as cache:
             cache[self.pin] = self.hz
@@ -140,7 +183,7 @@ def __init__(
     @property
     def using_hardware(self) -> bool:
         try:
-            return isinstance(self.pwm, HardwarePWM)
+            return isinstance(self.pwm, HardwarePWMOutputDevice)
         except AttributeError:
             return False
 
@@ -164,16 +207,16 @@ def _serialize(self):
         )
 
     def start(self, initial_duty_cycle: float) -> None:
-        if not (0 <= initial_duty_cycle <= 100):
+        if not (0.0 <= initial_duty_cycle <= 100.0):
             raise PWMError("duty_cycle should be between 0 and 100, inclusive.")
 
-        self.duty_cycle = round(float(initial_duty_cycle), 5)
-        self.pwm.start(self.duty_cycle)
+        self.change_duty_cycle(initial_duty_cycle)
+        self.pwm.on()
         self._serialize()
 
     def stop(self) -> None:
-        self.pwm.stop()
-        self.duty_cycle = 0
+        self.pwm.off()
+        self.pwm.value = 0.0
         self._serialize()
 
     def change_duty_cycle(self, duty_cycle: float) -> None:
@@ -182,15 +225,14 @@ def change_duty_cycle(self, duty_cycle: float) -> None:
 
         self.duty_cycle = round(float(duty_cycle), 5)
 
-        if self.using_hardware:
-            self.pwm.change_duty_cycle(self.duty_cycle)
-        else:
-            self.pwm.ChangeDutyCycle(self.duty_cycle)  # type: ignore
+        self.pwm.value = self.duty_cycle / 100.0
 
         self._serialize()
 
     def cleanup(self) -> None:
         self.stop()
+        self.pwm.close()
+
         self.unlock()
 
         with local_intermittent_storage("pwm_dc") as cache:
@@ -201,16 +243,6 @@ def cleanup(self) -> None:
 
         gpio_helpers.set_gpio_availability(self.pin, True)
 
-        if self.using_hardware:
-            # `stop` handles cleanup.
-            pass
-        else:
-            import RPi.GPIO as GPIO
-
-            GPIO.setmode(GPIO.BCM)
-            GPIO.setup(self.pin, GPIO.OUT, initial=GPIO.LOW)
-            GPIO.cleanup(self.pin)
-
         self.logger.debug(f"Cleaned up GPIO-{self.pin}.")
 
         if not self._external_client:

pioreactor/whoami.py

@@ -145,16 +145,12 @@ def get_image_git_hash() -> str:
 
 
 if is_testing_env():
-    # import fake_rpi  # type: ignore
-    # fake_rpi.toggle_print(False)
-    # sys.modules["RPi"] = fake_rpi.RPi  # Fake RPi
-    # sys.modules["RPi.GPIO"] = fake_rpi.RPi.GPIO  # Fake GPIO
-
     # mock out gpiozero's pins
     from gpiozero import Device
     from gpiozero.pins.mock import MockFactory
+    from gpiozero.pins.mock import MockPWMPin
 
-    Device.pin_factory = MockFactory()
+    Device.pin_factory = MockFactory(pin_class=MockPWMPin)
 
     # allow Blinka to think we are an Rpi:
     # https://github.com/adafruit/Adafruit_Python_PlatformDetect/blob/75f69806222fbaf8535130ed2eacd07b06b1a298/adafruit_platformdetect/board.py