Experiment: Using gpiod for driving pins

.github/workflows/CI.yaml

@@ -51,7 +51,7 @@ jobs:
     runs-on: ubuntu-latest
     strategy:
       matrix:
-        python: [ "3.7", "3.8", "3.9", "3.10", "3.11", "3.12" ]
+        python: [ "3.9", "3.10", "3.11", "3.12" ]
         octoprint: [ "1.5", "1.6", "1.7", "1.8", "1.9", "1.10" ]
         exclude:
           # These versions are not compatible to each other:

mypy.ini

@@ -7,7 +7,8 @@ exclude = (?x)(
 [mypy-octoprint.*]
 ignore_missing_imports = True
 
-[mypy-RPi.*]
+
+[mypy-gpiod.*]
 ignore_missing_imports = True
 
 [mypy-flask.*]

octoprint_octorelay/__init__.py

@@ -371,7 +371,7 @@ def input_polling(self):
                 self.update_ui()
                 break
 
-__plugin_pythoncompat__ = ">=3.7,<4"
+__plugin_pythoncompat__ = ">=3.9,<4"
 __plugin_implementation__ = OctoRelayPlugin()
 
 __plugin_hooks__ = {

octoprint_octorelay/driver.py

@@ -1,15 +1,20 @@
 # -*- coding: utf-8 -*-
 from typing import Optional
-from RPi import GPIO
-
-# The driver operates BCM mode of pins enumeration
-GPIO.setmode(GPIO.BCM)
+from gpiod import request_lines, LineSettings
+from gpiod.line import Direction, Value
 
 def xor(left: bool, right: bool) -> bool:
     return left is not right
 
 class Relay():
     def __init__(self, pin: int, inverted: bool):
+        self.request = request_lines(
+            "/dev/gpiochip0",
+            consumer = "OctoRelay",
+            config = {
+                pin: LineSettings(direction=Direction.OUTPUT)
+            }
+        )
         self.pin = pin # GPIO pin
         self.inverted = inverted # marks the relay as normally closed
 
@@ -26,10 +31,7 @@ def open(self):
 
     def is_closed(self) -> bool:
         """Returns the logical state of the relay."""
-        GPIO.setwarnings(False)
-        GPIO.setup(self.pin, GPIO.OUT)
-        pin_state = bool(GPIO.input(self.pin))
-        GPIO.setwarnings(True)
+        pin_state = self.request.get_value(self.pin) == Value.ACTIVE
         return xor(self.inverted, pin_state)
 
     def toggle(self, desired_state: Optional[bool] = None) -> bool:
@@ -40,8 +42,5 @@ def toggle(self, desired_state: Optional[bool] = None) -> bool:
         """
         if desired_state is None:
             desired_state = not self.is_closed()
-        GPIO.setwarnings(False)
-        GPIO.setup(self.pin, GPIO.OUT)
-        GPIO.output(self.pin, xor(self.inverted, desired_state))
-        GPIO.setwarnings(True)
+        self.request.set_value(self.pin, Value.ACTIVE if xor(self.inverted, desired_state) else Value.INACTIVE)
         return desired_state

setup.py

@@ -35,7 +35,7 @@ def get_version_and_cmdclass(pkg_path):
 
 # The plugin's description. Can be overwritten within OctoPrint's internal data via __plugin_description__ in the plugin
 # module
-plugin_description = """A plugin that adds buttons to the navbar for toggling GPIO on the RPi. It can be used for turning relays on and off."""
+plugin_description = """A plugin that adds buttons to the navbar for toggling GPIO on Raspberry Pi. It can be used for turning relays on and off."""
 
 # The plugin's author. Can be overwritten within OctoPrint's internal data via __plugin_author__ in the plugin module
 plugin_author = "Boris Burgstaller"
@@ -51,7 +51,7 @@ def get_version_and_cmdclass(pkg_path):
 
 # Any additional requirements besides OctoPrint should be listed here
 # todo after dropping 3.7 remove typing-extensions (used by model.py and listing.py)
-plugin_requires = ["RPi.GPIO", "typing-extensions"]
+plugin_requires = ["gpiod", "typing-extensions"]
 
 # --------------------------------------------------------------------------------------------------------------------
 # More advanced options that you usually shouldn't have to touch follow after this point

tests/test_driver.py

@@ -4,10 +4,17 @@
 from unittest.mock import Mock
 
 # Mocks used for assertions
-GPIO_mock = Mock()
-GPIO_mock.BCM = "MockedBCM"
-GPIO_mock.OUT = "MockedOUT"
-sys.modules["RPi.GPIO"] = GPIO_mock
+gpiod_mock = Mock()
+line_mock = Mock()
+gpiod_mock.request_lines = Mock(return_value=line_mock)
+gpiod_mock.LineSettings = Mock(return_value="LineSettingsMock")
+sys.modules["gpiod"] = gpiod_mock
+line_mock.Direction = Mock()
+line_mock.Direction.OUTPUT = "OutputMock"
+line_mock.Value = Mock()
+line_mock.Value.ACTIVE = "ActiveMock"
+line_mock.Value.INACTIVE = "InactiveMock"
+sys.modules["gpiod.line"] = line_mock
 
 # pylint: disable=wrong-import-position
 from octoprint_octorelay.driver import Relay
@@ -19,11 +26,16 @@
 
 class TestRelayDriver(unittest.TestCase):
     def test_constructor(self):
-        GPIO_mock.setmode.assert_called_with("MockedBCM")
         relay = Relay(18, True)
         self.assertIsInstance(relay, Relay)
         self.assertEqual(relay.pin, 18)
         self.assertTrue(relay.inverted)
+        gpiod_mock.LineSettings.assert_called_with(direction="OutputMock")
+        gpiod_mock.request_lines.assert_called_with(
+            "/dev/gpiochip0",
+            consumer = "OctoRelay",
+            config = { 18: "LineSettingsMock" }
+        )
 
     def test_serialization(self):
         relay = Relay(18, True)
@@ -32,59 +44,48 @@ def test_serialization(self):
 
     def test_close(self):
         cases = [
-            { "relay": Relay(18, False), "expected_pin_state": True },
-            { "relay": Relay(18, True), "expected_pin_state": False }
+            { "relay": Relay(18, False), "expected_pin_state": "ActiveMock" },
+            { "relay": Relay(18, True), "expected_pin_state": "InactiveMock" }
         ]
         for case in cases:
             case["relay"].close()
-            GPIO_mock.setup.assert_called_with(18, "MockedOUT")
-            GPIO_mock.output.assert_called_with(18, case["expected_pin_state"])
-            GPIO_mock.setwarnings.assert_any_call(False)
-            GPIO_mock.setwarnings.assert_called_with(True)
+            line_mock.set_value.assert_called_with(18, case["expected_pin_state"])
 
     def test_open(self):
         cases = [
-            { "relay": Relay(18, False), "expected_pin_state": False },
-            { "relay": Relay(18, True), "expected_pin_state": True }
+            { "relay": Relay(18, False), "expected_pin_state": "InactiveMock" },
+            { "relay": Relay(18, True), "expected_pin_state": "ActiveMock" }
         ]
         for case in cases:
             case["relay"].open()
-            GPIO_mock.setup.assert_called_with(18, "MockedOUT")
-            GPIO_mock.output.assert_called_with(18, case["expected_pin_state"])
-            GPIO_mock.setwarnings.assert_any_call(False)
-            GPIO_mock.setwarnings.assert_called_with(True)
+            line_mock.set_value.assert_called_with(18, case["expected_pin_state"])
 
     def test_is_closed(self):
         cases = [
-            { "mocked_state": 1, "inverted": False, "expected_relay_state": True },
-            { "mocked_state": 0, "inverted": False, "expected_relay_state": False },
-            { "mocked_state": 1, "inverted": True, "expected_relay_state": False },
-            { "mocked_state": 0, "inverted": True, "expected_relay_state": True },
+            { "mocked_state": "ActiveMock", "inverted": False, "expected_relay_state": True },
+            { "mocked_state": "InactiveMock", "inverted": False, "expected_relay_state": False },
+            { "mocked_state": "ActiveMock", "inverted": True, "expected_relay_state": False },
+            { "mocked_state": "InactiveMock", "inverted": True, "expected_relay_state": True },
         ]
         for case in cases:
-            GPIO_mock.input = Mock(return_value=case["mocked_state"])
+            line_mock.get_value = Mock(return_value=case["mocked_state"])
             relay = Relay(18, case["inverted"])
             self.assertEqual(relay.is_closed(), case["expected_relay_state"])
-            GPIO_mock.setwarnings.assert_any_call(False)
-            GPIO_mock.setwarnings.assert_called_with(True)
-            GPIO_mock.input.assert_called_with(18)
+            line_mock.get_value.assert_called_with(18)
 
     def test_toggle__no_argument(self):
         cases = [
-            { "mocked_state": 1, "inverted": False, "expected_pin_state": False, "expected_relay_state": False },
-            { "mocked_state": 0, "inverted": False, "expected_pin_state": True, "expected_relay_state": True },
-            { "mocked_state": 1, "inverted": True, "expected_pin_state": False, "expected_relay_state": True },
-            { "mocked_state": 0, "inverted": True, "expected_pin_state": True, "expected_relay_state": False },
+            { "mocked_state": "ActiveMock", "inverted": False, "exp_pin": "InactiveMock", "exp_relay": False },
+            { "mocked_state": "InactiveMock", "inverted": False, "exp_pin": "ActiveMock", "exp_relay": True },
+            { "mocked_state": "ActiveMock", "inverted": True, "exp_pin": "InactiveMock", "exp_relay": True },
+            { "mocked_state": "InactiveMock", "inverted": True, "exp_pin": "ActiveMock", "exp_relay": False },
         ]
         for case in cases:
-            GPIO_mock.input = Mock(return_value=case["mocked_state"])
+            line_mock.get_value = Mock(return_value=case["mocked_state"])
             relay = Relay(18, case["inverted"])
-            self.assertEqual(relay.toggle(), case["expected_relay_state"])
-            GPIO_mock.setwarnings.assert_any_call(False)
-            GPIO_mock.setwarnings.assert_called_with(True)
-            GPIO_mock.input.assert_called_with(18)
-            GPIO_mock.setup.assert_called_with(18, "MockedOUT")
-            GPIO_mock.output.assert_called_with(18, case["expected_pin_state"])
+            self.assertEqual(relay.toggle(), case["exp_relay"])
+            line_mock.get_value.assert_called_with(18)
+            line_mock.set_value.assert_called_with(18, case["exp_pin"])
 
 if __name__ == "__main__":
     unittest.main()

tests/test_init.py

@@ -10,7 +10,7 @@
 from octoprint.access import ADMIN_GROUP, USER_GROUP
 
 # Patching required before importing OctoRelayPlugin class
-sys.modules["RPi.GPIO"] = Mock()
+sys.modules["gpiod"] = Mock()
 
 # Mocks used for assertions
 timerMock = Mock()
@@ -432,7 +432,7 @@ def test_get_update_information(self):
 
     def test_python_compatibility(self):
         # Should be the current Python compability string
-        self.assertEqual(__plugin_pythoncompat__, ">=3.7,<4")
+        self.assertEqual(__plugin_pythoncompat__, ">=3.9,<4")
 
     def test_exposed_implementation(self):
         # Should be an instance of the plugin class

tests/test_migrations.py

@@ -9,7 +9,7 @@
     del sys.modules["octoprint_octorelay.migrations"]
 
 # Mocking required before the further import
-sys.modules["RPi.GPIO"] = Mock()
+sys.modules["gpiod"] = Mock()
 
 # pylint: disable=wrong-import-position
 from octoprint_octorelay.const import SETTINGS_VERSION