🚀 5.0.0

.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:

.github/workflows/release.yaml

@@ -68,7 +68,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:

.pylintrc

@@ -90,7 +90,7 @@ persistent=yes
 
 # Minimum Python version to use for version dependent checks. Will default to
 # the version used to run pylint.
-py-version=3.7
+py-version=3.9
 
 # Discover python modules and packages in the file system subtree.
 recursive=no

CHANGELOG.md

@@ -1,5 +1,18 @@
 # Changelog
 
+## Version 5
+
+### 5.0.0
+
+- **Breaking changes**:
+  - Minimum version of Python supported: 3.9;
+- Feature: Supporting Raspberry Pi 5:
+  - Using `gpiozero` with two `lgpio` and `RPi.GPIO` factories depending on your hardware;
+  - The feature implemented by [Mattia Vidoni](https://github.com/ch3p4ll3).
+- How to migrate:
+  - Ensure having Python version 3.9 or higher;
+  - If you can not upgrade Python to at least 3.9 — continue using version 4.
+
 ## Version 4
 
 ### 4.2.1

README.md

@@ -5,7 +5,7 @@
 [![Coverage Status](https://coveralls.io/repos/github/borisbu/OctoRelay/badge.svg?branch=master)](https://coveralls.io/github/borisbu/OctoRelay?branch=master)
 [![Downloads of latest release](https://img.shields.io/github/downloads/borisbu/octorelay/latest/release.zip?color=blue)](https://github.com/borisbu/OctoRelay/releases/latest)
 
-The plugin adds buttons to control GPIO pins of Raspberry Pi for switching relays and indicating their states.
+The plugin adds buttons to control GPIO pins of Raspberry Pi (incl. Pi 5) for switching relays and indicating their states.
 
 ![WebUI interface](img/controls.png)
 
@@ -26,8 +26,9 @@ The plugin allows you to set your own icons and flexibly customize the way the r
 
 ## Requirements
 
-- Python: at least `3.7`,
-- OctoPrint: at least `1.5.3`.
+- Python: at least `3.9`:
+  - Earlier versions of Python are supported by the plugin version `4.x` (however it does not support Raspberry Pi 5);
+- OctoPrint: at least `1.5.3`:
   - For [AutoConnect feature](https://github.com/borisbu/OctoRelay/blob/master/CHANGELOG.md#330): at least `1.9.0`.
 
 ## Setup

mypy.ini

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

octoprint_octorelay/__init__.py

@@ -17,7 +17,7 @@
     STARTUP, PRINTING_STOPPED, PRINTING_STARTED, PRIORITIES, FALLBACK_PRIORITY, PREEMPTIVE_CANCELLATION_CUTOFF,
     CANCEL_TASK_COMMAND, USER_ACTION, TURNED_ON
 )
-from .driver import Relay
+from .driver import Driver
 from .task import Task
 from .listing import Listing
 from .migrations import migrate
@@ -108,7 +108,7 @@ def handle_list_all_command(self) -> Listing:
         settings = self._settings.get([], merged=True) # expensive
         for index in RELAY_INDEXES:
             if bool(settings[index]["active"]):
-                relay = Relay(
+                relay = Driver.ensure(
                     int(settings[index]["relay_pin"] or 0),
                     bool(settings[index]["inverted_output"])
                 )
@@ -124,7 +124,7 @@ def handle_get_status_command(self, index: str) -> bool:
         settings = self._settings.get([index], merged=True) # expensive
         if not bool(settings["active"]):
             raise HandlingException(400)
-        return Relay(
+        return Driver.ensure(
             int(settings["relay_pin"] or 0),
             bool(settings["inverted_output"])
         ).is_closed()
@@ -241,7 +241,7 @@ def toggle_relay(self, index, target: Optional[bool] = None) -> bool:
                 self._printer.disconnect()
         pin = int(settings["relay_pin"] or 0)
         inverted = bool(settings["inverted_output"])
-        relay = Relay(pin, inverted)
+        relay = Driver.ensure(pin, inverted)
         self._logger.debug(
             f"Toggling the relay {index} on pin {pin}" if target is None else
             f"Turning the relay {index} {'ON' if target else 'OFF'} (pin {pin})"
@@ -313,7 +313,7 @@ def update_ui(self):
         ))
         for index in RELAY_INDEXES:
             active = bool(settings[index]["active"])
-            relay = Relay(
+            relay = Driver.ensure(
                 int(settings[index]["relay_pin"] or 0),
                 bool(settings[index]["inverted_output"])
             )
@@ -362,7 +362,7 @@ def input_polling(self):
         for index in RELAY_INDEXES:
             active = self.model[index]["active"]
             model_state = self.model[index]["relay_state"] # bool since v3.1
-            actual_state = Relay(
+            actual_state = Driver.ensure(
                 self.model[index]["relay_pin"],
                 self.model[index]["inverted_output"]
             ).is_closed() if active else False
@@ -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,17 +1,17 @@
 # -*- coding: utf-8 -*-
-from typing import Optional
-from RPi import GPIO
-
-# The driver operates BCM mode of pins enumeration
-GPIO.setmode(GPIO.BCM)
+from typing import Optional, List
+from gpiozero import LED
 
 def xor(left: bool, right: bool) -> bool:
     return left is not right
 
-class Relay():
-    def __init__(self, pin: int, inverted: bool):
+class Driver():
+    cache: List["Driver"] = []
+
+    def __init__(self, pin: int, inverted: bool, pin_factory=None):
         self.pin = pin # GPIO pin
         self.inverted = inverted # marks the relay as normally closed
+        self.handle = LED(pin, pin_factory=pin_factory)
 
     def __repr__(self) -> str:
         return f"{type(self).__name__}(pin={self.pin},inverted={self.inverted},closed={self.is_closed()})"
@@ -26,11 +26,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)
-        return xor(self.inverted, pin_state)
+        return xor(self.inverted, self.handle.is_lit)
 
     def toggle(self, desired_state: Optional[bool] = None) -> bool:
         """
@@ -40,8 +36,16 @@ 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.handle.on if xor(self.inverted, desired_state) else self.handle.off)()
         return desired_state
+
+    @classmethod
+    def ensure(cls, pin: int, inverted: bool, pin_factory=None):
+        for relay in cls.cache:
+            if relay.pin == pin:
+                if xor(relay.inverted, inverted):
+                    relay.inverted = inverted
+                return relay
+        relay = cls(pin, inverted, pin_factory)
+        cls.cache.append(relay)
+        return relay

octoprint_octorelay/listing.py

@@ -1,7 +1,5 @@
 # -*- coding: utf-8 -*-
-from typing import List
-# todo after dropping 3.7 take TypedDict from typing instead
-from typing_extensions import TypedDict
+from typing import List, TypedDict
 
 # false positive, see https://github.com/pylint-dev/pylint/issues/4166
 # pylint: disable=too-many-ancestors

octoprint_octorelay/model.py

@@ -1,8 +1,5 @@
 # -*- coding: utf-8 -*-
-from typing import Optional, Dict
-# todo after dropping 3.7 take TypedDict from typing instead
-from typing_extensions import TypedDict
-
+from typing import Optional, Dict, TypedDict
 from .const import RELAY_INDEXES
 
 # pylint: disable=too-few-public-methods

setup.py

@@ -50,8 +50,7 @@ def get_version_and_cmdclass(pkg_path):
 plugin_license = "AGPLv3"
 
 # 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 = ["gpiozero", "lgpio", "RPi.GPIO"]
 
 # --------------------------------------------------------------------------------------------------------------------
 # More advanced options that you usually shouldn't have to touch follow after this point

tests/test_driver.py

@@ -1,58 +1,46 @@
 # -*- coding: utf-8 -*-
 import unittest
 import sys
-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
+from gpiozero.pins.mock import MockFactory
 
 # pylint: disable=wrong-import-position
-from octoprint_octorelay.driver import Relay
+from octoprint_octorelay.driver import Driver
 
 # avoid keeping other modules automatically imported by this test
 del sys.modules["octoprint_octorelay"]
 del sys.modules["octoprint_octorelay.migrations"]
 del sys.modules["octoprint_octorelay.task"]
 
-class TestRelayDriver(unittest.TestCase):
+class TestDriver(unittest.TestCase):
     def test_constructor(self):
-        GPIO_mock.setmode.assert_called_with("MockedBCM")
-        relay = Relay(18, True)
-        self.assertIsInstance(relay, Relay)
+        relay = Driver(18, True, MockFactory())
+        self.assertIsInstance(relay, Driver)
         self.assertEqual(relay.pin, 18)
         self.assertTrue(relay.inverted)
 
     def test_serialization(self):
-        relay = Relay(18, True)
+        relay = Driver(18, True, MockFactory())
         serialization = f"{relay}"
-        self.assertEqual(serialization, "Relay(pin=18,inverted=True,closed=True)")
+        self.assertEqual(serialization, "Driver(pin=18,inverted=True,closed=True)")
 
     def test_close(self):
         cases = [
-            { "relay": Relay(18, False), "expected_pin_state": True },
-            { "relay": Relay(18, True), "expected_pin_state": False }
+            { "relay": Driver(18, False, MockFactory()), "expected_pin_state": True },
+            { "relay": Driver(18, True, MockFactory()), "expected_pin_state": False }
         ]
         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)
+            self.assertEqual(case["relay"].handle.is_lit, 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": Driver(18, False, MockFactory()), "expected_pin_state": False },
+            { "relay": Driver(18, True, MockFactory()), "expected_pin_state": True }
         ]
         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)
+            self.assertEqual(case["relay"].handle.is_lit, case["expected_pin_state"])
 
     def test_is_closed(self):
         cases = [
@@ -62,12 +50,9 @@ def test_is_closed(self):
             { "mocked_state": 0, "inverted": True, "expected_relay_state": True },
         ]
         for case in cases:
-            GPIO_mock.input = Mock(return_value=case["mocked_state"])
-            relay = Relay(18, case["inverted"])
+            relay = Driver(18, case["inverted"], MockFactory())
+            relay.handle.value = case["mocked_state"]
             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)
 
     def test_toggle__no_argument(self):
         cases = [
@@ -77,14 +62,32 @@ def test_toggle__no_argument(self):
             { "mocked_state": 0, "inverted": True, "expected_pin_state": True, "expected_relay_state": False },
         ]
         for case in cases:
-            GPIO_mock.input = Mock(return_value=case["mocked_state"])
-            relay = Relay(18, case["inverted"])
+            relay = Driver(18, case["inverted"], MockFactory())
+            relay.handle.value = case["mocked_state"]
             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.handle.is_lit, case["expected_pin_state"])
+
+    def test_ensure(self):
+        # Test creating a new relay
+        relay1 = Driver.ensure(17, False, MockFactory())
+        self.assertEqual(len(Driver.cache), 1)
+        self.assertEqual(relay1.pin, 17)
+        self.assertFalse(relay1.inverted)
+
+        # Test retrieving the existing relay with the same pin and inversion
+        relay2 = Driver.ensure(17, True, MockFactory())
+        self.assertIs(relay1, relay2)
+        self.assertEqual(len(Driver.cache), 1)  # Should still be 1
+
+        # Test retrieving the existing relay with the same pin but different inversion
+        relay3 = Driver.ensure(17, True, MockFactory())
+        self.assertEqual(len(Driver.cache), 1)  # Should still be 1
+        self.assertIs(relay1, relay3)
+        self.assertTrue(relay1.inverted)  # Inversion should be updated
+
+    def tearDown(self):
+        # Clear cache after each test
+        Driver.cache = []
 
 if __name__ == "__main__":
     unittest.main()