demo what gpiozero looks like

.pre-commit-config.yaml

@@ -36,7 +36,7 @@ repos:
     hooks:
     -   id: mypy
         additional_dependencies: [
-            msgspec==0.18.1,
+            msgspec==0.18.2,
             types-pkg_resources==0.1.3,
             types-paho-mqtt==1.6.0.6
         ]

pioreactor/actions/self_test.py

@@ -63,8 +63,6 @@ def test_REF_is_in_correct_position(
     # The idea is to trigger stirring on and off and the REF should not see a change in signal / variance, but the other PD should.
     from statistics import variance
 
-    assert is_HAT_present()
-
     reference_channel = cast(PdChannel, config["od_config.photodiode_channel_reverse"][REF_keyword])
     signal_channel = "2" if reference_channel == "1" else "1"
 
@@ -122,8 +120,6 @@ def test_all_positive_correlations_between_pds_and_leds(
     from pprint import pformat
     from random import shuffle
 
-    assert is_HAT_present()
-
     # better to err on the side of MORE samples than less - it's only a few extra seconds...
     # we randomize to reduce effects of temperature
     INTENSITIES = list(range(20, 85, 5))
@@ -224,8 +220,6 @@ def test_ambient_light_interference(
 ) -> None:
     # test ambient light IR interference. With all LEDs off, and the Pioreactor not in a sunny room, we should see near 0 light.
 
-    assert is_HAT_present()
-
     adc_reader = ADCReader(
         channels=ALL_PD_CHANNELS,
         dynamic_gain=False,
@@ -281,8 +275,6 @@ def test_REF_is_lower_than_0_dot_256_volts(
 
 
 def test_PD_is_near_0_volts_for_blank(client, logger: Logger, unit: str, experiment: str) -> None:
-    assert is_HAT_present()
-
     reference_channel = cast(PdChannel, config["od_config.photodiode_channel_reverse"][REF_keyword])
     signal_channel = "2" if reference_channel == "1" else "1"
     assert config.get("od_config.photodiode_channel", signal_channel, fallback=None) in [
@@ -341,7 +333,6 @@ def test_positive_correlation_between_temperature_and_heating(
 def test_aux_power_is_not_too_high(
     client: Client, logger: Logger, unit: str, experiment: str
 ) -> None:
-    assert is_HAT_present()
     assert voltage_in_aux() <= 18.0
 
 
@@ -382,7 +373,7 @@ def test_positive_correlation_between_rpm_and_stirring(
 
             st.set_duty_cycle(dc)
             sleep(0.75)
-            measured_rpms.append(rpm_calc(3.0))
+            measured_rpms.append(rpm_calc.estimate(3.0))
             dcs.append(dc)
 
         measured_correlation = round(correlation(dcs, measured_rpms), 2)

pioreactor/actions/stirring_calibration.py

@@ -66,7 +66,7 @@ def stirring_calibration(min_dc: int, max_dc: int) -> None:
             for count, dc in enumerate(dcs, start=1):
                 st.set_duty_cycle(dc)
                 sleep(8)
-                rpm = rpm_calc(4)
+                rpm = rpm_calc.estimate(4)
                 measured_rpms.append(rpm)
                 logger.debug(f"Detected {rpm=:.1f} RPM @ {dc=}%")
 

pioreactor/background_jobs/monitor.py

@@ -137,9 +137,7 @@ def pretty_version(info: tuple) -> str:
             run_immediately=False,
         ).start()
 
-        self.add_pre_button_callback(self.led_on)
         self.add_pre_button_callback(self._republish_state)
-        self.add_post_button_callback(self.led_off)
 
         self.start_passive_listeners()
 
@@ -155,26 +153,13 @@ def _setup_GPIO(self) -> None:
         set_gpio_availability(BUTTON_PIN, False)
         set_gpio_availability(LED_PIN, False)
 
-        import RPi.GPIO as GPIO  # type: ignore
+        from gpiozero import LED, Button
 
-        # I am hiding all the slow imports, but in this case, I need GPIO module
-        # in many functions.
-        self.GPIO = GPIO
+        self.led = LED(LED_PIN)
+        self.button = Button(BUTTON_PIN, pull_up=False)
 
-        self.GPIO.setmode(GPIO.BCM)
-        self.GPIO.setup(BUTTON_PIN, GPIO.IN, pull_up_down=self.GPIO.PUD_DOWN)
-        self.GPIO.setup(LED_PIN, GPIO.OUT)
-
-        try:
-            retry(
-                self.GPIO.add_event_detect,
-                retries=3,
-                delay=1.0,
-                args=(BUTTON_PIN, self.GPIO.RISING),
-                kwargs={"callback": self.button_down_and_up, "bouncetime": 100},
-            )
-        except RuntimeError:
-            self.logger.warning("Failed to add button detect.")
+        self.led.source = self.button
+        self.button.when_pressed = self.button_down_and_up
 
     def check_for_network(self) -> None:
         if whoami.is_testing_env():
@@ -407,18 +392,18 @@ def on_ready(self) -> None:
         self.check_state_of_jobs_on_machine()
 
     def on_disconnected(self) -> None:
-        self.GPIO.cleanup(LED_PIN)
-        self.GPIO.cleanup(BUTTON_PIN)
+        self.led.close()
+        self.button.close()
         set_gpio_availability(BUTTON_PIN, True)
         set_gpio_availability(LED_PIN, True)
 
     def led_on(self) -> None:
-        self.GPIO.output(LED_PIN, self.GPIO.HIGH)
+        self.led.on()
 
     def led_off(self) -> None:
-        self.GPIO.output(LED_PIN, self.GPIO.LOW)
+        self.led.off()
 
-    def button_down_and_up(self, *args) -> None:
+    def button_down_and_up(self, *args) -> None:  # TODO: why args??
         # Warning: this might be called twice: See "Switch debounce" in https://sourceforge.net/p/raspberry-gpio-python/wiki/Inputs/
         # don't put anything that is not idempotent in here.
 
@@ -430,7 +415,7 @@ def button_down_and_up(self, *args) -> None:
             except Exception:
                 self.logger.debug(f"Error in pre_function={pre_function.__name__}.", exc_info=True)
 
-        while self.GPIO.input(BUTTON_PIN) == self.GPIO.HIGH:
+        while self.button.is_pressed:
             sleep(0.02)
 
         for post_function in self._post_button:

pioreactor/background_jobs/stirring.py

@@ -52,7 +52,7 @@ class RpmCalculator:
 
     > rpm_calc = RpmCalculator()
     > rpm_calc.setup()
-    > rpm_calc(seconds_to_observe=1.5)
+    > rpm_estimate = rpm_calc.estimate(seconds_to_observe=1.5)
 
     """
 
@@ -66,10 +66,6 @@ def setup(self) -> None:
         # use the GPIO for this.
         set_gpio_availability(self.hall_sensor_pin, False)
 
-        import RPi.GPIO as GPIO  # type: ignore
-
-        self.GPIO = GPIO
-        self.GPIO.setmode(self.GPIO.BCM)
         self.GPIO.setup(self.hall_sensor_pin, self.GPIO.IN, pull_up_down=self.GPIO.PUD_UP)
 
         # ignore any changes that occur within 15ms - at 1000rpm (very fast), the
@@ -95,7 +91,7 @@ def clean_up(self) -> None:
             self.GPIO.cleanup(self.hall_sensor_pin)
         set_gpio_availability(self.hall_sensor_pin, True)
 
-    def __call__(self, seconds_to_observe: float) -> float:
+    def estimate(self, seconds_to_observe: float) -> float:
         return 0.0
 
     def callback(self, *args) -> None:
@@ -139,7 +135,7 @@ def clear_aggregates(self) -> None:
         self._running_count = 0
         self._start_time = None
 
-    def __call__(self, seconds_to_observe: float) -> float:
+    def estimate(self, seconds_to_observe: float) -> float:
         self.clear_aggregates()
         self.turn_on_collection()
         self.sleep_for(seconds_to_observe)
@@ -356,7 +352,7 @@ def poll(self, poll_for_seconds: float) -> Optional[structs.MeasuredRPM]:
         if self.rpm_calculator is None:
             return None
 
-        recent_rpm = round(self.rpm_calculator(poll_for_seconds), 2)
+        recent_rpm = round(self.rpm_calculator.estimate(poll_for_seconds), 2)
 
         self._measured_rpm = recent_rpm
         self.measured_rpm = structs.MeasuredRPM(

pioreactor/whoami.py

@@ -145,11 +145,16 @@ def get_image_git_hash() -> str:
 
 
 if is_testing_env():
-    import fake_rpi  # type: ignore
+    # 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
 
-    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
+
+    Device.pin_factory = MockFactory()
 
     # allow Blinka to think we are an Rpi:
     # https://github.com/adafruit/Adafruit_Python_PlatformDetect/blob/75f69806222fbaf8535130ed2eacd07b06b1a298/adafruit_platformdetect/board.py

requirements/requirements_worker.txt

@@ -1,5 +1,5 @@
 -r requirements.txt
-RPi.GPIO==0.7.1; platform_machine == "armv7l"
+gpiozero==2.0; platform_machine == "armv7l"
 adafruit-circuitpython-ads1x15==2.2.12
 DAC43608==0.2.6
 TMP1075==0.2.1

setup.py

@@ -34,7 +34,7 @@
 
 
 WORKER_REQUIREMENTS = [
-    "RPi.GPIO==0.7.1",
+    "gpiozero==2.0",
     "adafruit-circuitpython-ads1x15==2.2.12",
     "DAC43608==0.2.7",
     "TMP1075==0.2.1",