[Door] switch gpio out to mqtt

app/main.py

@@ -32,9 +32,9 @@ async def lifespan(app: FastAPI):
     create_first_user(db=get_db_session())
     init_controller(
         DoorController(
-            lock_pin=settings.lock_pin,
-            unlock_pin=settings.unlock_pin,
-            mock_factory=settings.mock_gpio,
+            mqtt_host=settings.mqtt_host,
+            mqtt_port=settings.mqtt_port,
+            mock_factory=settings.mock,
         ),
     )
     if not settings.disable_cards:

app/services/door.py

@@ -16,26 +16,25 @@ logger = logging.getLogger(__name__)
 class DoorController:
     def __init__(
         self,
-        lock_pin: int = 17,  # connected to 20 on the esp
-        unlock_pin: int = 18,  # connected to 21 on the esp
+        mqtt_host: str = "localhost",
+        mqtt_port: int = 1883,
         mock_factory: bool = False,
+
     ):
         self._is_open: bool = False
-        self._lock_pin = lock_pin
-        self._unlock_pin = unlock_pin
+        self._mqtt = None
+        self._mqtt_topic = "nukihub/lock/action"
         self._mock = mock_factory
-        self._chip = None
 
         if not mock_factory:
-            import lgpio
-            self._chip = lgpio.gpiochip_open(0)
-            lgpio.gpio_claim_output(self._chip, unlock_pin, 1)
-            lgpio.gpio_claim_output(self._chip, lock_pin, 1)
+            import paho.mqtt.client as mqtt
+            self._mqtt = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2)
+            self._mqtt.connect(mqtt_host, mqtt_port)
+            self._mqtt.loop_start()
+            logger.info("Mqtt client connect to %s:%s", mqtt_host, mqtt_port)
 
         logger.info(
-            "DoorController started. lock=%s unlock=%s mock=%s",
-            lock_pin,
-            unlock_pin,
+            "DoorController started. topic=%s mock=%s", self._mqtt_topic,
             mock_factory,
         )
 
@@ -45,9 +44,7 @@ class DoorController:
             logger.info("Door unlocked.[MOCK]")
             return
 
-        lgpio.gpio_write(self._chip, self._unlock_pin, 0)
-        sleep(0.4)
-        lgpio.gpio_write(self._chip, self._unlock_pin, 1)
+        self._mqtt.publish(self._mqtt_topic, "unlock")
         self._is_open = True
         logger.info("Door unlocked!")
 
@@ -57,9 +54,7 @@ class DoorController:
             logger.info("Door locked.[MOCK]")
             return
 
-        lgpio.gpio_write(self._chip, self._lock_pin, 0)
-        sleep(0.4)
-        lgpio.gpio_write(self._chip, self._lock_pin, 1)
+        self._mqtt.publish(self._mqtt_topic, "lock")
         self._is_open = False
         logger.info("Door locked!")
 

app/services/settings.py

@@ -11,9 +11,9 @@ class Settings(BaseSettings):
 
     secret_key: str
     sqlalchemy_database_url: str = "sqlite:///./gatekeeper.db"
-    lock_pin: int = 17
-    unlock_pin: int = 18
-    mock_gpio: bool = True
+    mqtt_host: str = "localhost"
+    mqtt_port: int = 1883
+    mock: bool = True
     alembic_config: str = "./alembic.ini"
 
     disable_cards: bool = False

module.nix

@@ -15,10 +15,10 @@ in
         description = "Where to save the database.";
         default = "/var/lib/gatekeeper";
       };
-      mockGpio = lib.mkOption {
+      mock = lib.mkOption {
         type = lib.types.str;
-        default = "True";
-        description = "Mock GPIO pins. Has to be a string!";
+        default = "False";
+        description = "Mock the doorcontroller. Has to be a string!";
       };
     };
   };
@@ -67,7 +67,7 @@ in
       environment = {
         SQLALCHEMY_DATABASE_URL = "sqlite:///${cfg.db}/gatekeeper.db";
         ALEMBIC_CONFIG = "${self}/alembic.ini";
-        MOCK_GPIO = cfg.mockGpio;
+        MOCK = cfg.mock;
       };
     };
   };