From f45348f8dfd0bfe38f05e8419feaa1e31652a16f Mon Sep 17 00:00:00 2001
From: lilly <li@lly.sh>
Date: Tue, 19 May 2026 09:34:51 +0200
Subject: [PATCH] api: implement basic mqtt client

---
 api/pyproject.toml                |   1 +
 api/src/dooris_api/app.py         |   3 +-
 api/src/dooris_api/mqtt_client.py | 105 ++++++++++++++++++++++++++++++
 api/uv.lock                       |  11 ++++
 4 files changed, 119 insertions(+), 1 deletion(-)
 create mode 100644 api/src/dooris_api/mqtt_client.py

diff --git a/api/pyproject.toml b/api/pyproject.toml
index 1b0cc5f..4e42dd1 100644
--- a/api/pyproject.toml
+++ b/api/pyproject.toml
@@ -6,6 +6,7 @@ requires-python = ">=3.12"
 dependencies = [
     "aiohttp>=3.13.5",
     "fastapi>=0.136.1",
+    "paho-mqtt>=2.1.0",
     "simple-openid-connect>=2.4.0",
     "uvicorn>=0.46.0",
 ]
diff --git a/api/src/dooris_api/app.py b/api/src/dooris_api/app.py
index 3bcd848..fa3d75e 100644
--- a/api/src/dooris_api/app.py
+++ b/api/src/dooris_api/app.py
@@ -11,7 +11,7 @@ from simple_openid_connect.client import OpenidClient
 from simple_openid_connect.data import TokenSuccessResponse, RpInitiatedLogoutRequest
 from aiohttp import BasicAuth
 
-from dooris_api import deps, models, exceptions, app_config
+from dooris_api import deps, models, exceptions, app_config, mqtt_client
 from dooris_api.ccujack import CCUJackClient
 
 
@@ -36,6 +36,7 @@ async def lifespan(app: FastAPI):
         scope=app_cfg.openid_scope,
     )
 
+    # TODO: regularly re-query CCUJACK to discover new locks
     app.extra["ccujack"] = CCUJackClient(
         base_uri=app_cfg.ccujack_url,
         auth=BasicAuth(app_cfg.ccujack_user, app_cfg.ccujack_password)
diff --git a/api/src/dooris_api/mqtt_client.py b/api/src/dooris_api/mqtt_client.py
new file mode 100644
index 0000000..d11ce39
--- /dev/null
+++ b/api/src/dooris_api/mqtt_client.py
@@ -0,0 +1,105 @@
+#
+# This whole implementation is adapted from the upstream GitHub example
+# https://github.com/eclipse-paho/paho.mqtt.python/blob/master/examples/loop_asyncio.py
+# 
+
+import logging
+import asyncio
+import socket
+
+import paho.mqtt.client as mqtt
+
+
+logger = logging.getLogger(__name__)
+
+
+class AsyncLooper:
+    """
+    Helper class to implement loopgin with asyncio for the underlying mqtt IO  
+    """
+
+    def __init__(self, loop: asyncio.AbstractEventLoop, client: mqtt.Client):
+        self.loop = loop
+
+        self.client = client
+        self.client.on_socket_open = self.on_socket_open
+        self.client.on_socket_close = self.on_socket_close
+        self.client.on_socket_register_write = self.on_socket_register_write
+        self.client.on_socket_unregister_write = self.on_socket_unregister_write
+
+    def on_socket_open(self, client, userdata, sock):
+        logger.debug("mqtt socket opened")
+
+        def cb():
+            logger.debug("mqtt socket is readable, calling loop_read()")
+            client.loop_read()
+
+        self.loop.add_reader(sock, cb)
+        self.task_misc = self.loop.create_task(self.misc_loop())
+
+    def on_socket_close(self, client, userdata, sock):
+        logger.debug("mqtt socket closed")
+        self.loop.remove_reader(sock)
+        self.task_misc.cancel()
+
+    def on_socket_register_write(self, client, userdata, sock):
+        logger.debug("watching mqtt socket for writability")
+
+        def cb():
+            logger.debug("mqtt socket ist writable, calling loop_write()")
+            client.loop_write()
+
+        self.loop.add_writer(sock, cb)
+
+    def on_socket_unregister_write(self, client, userdata, sock):
+        logger.debug("stopping to watch mqtt socket for writability")
+        self.loop.remove_writer(sock)
+
+    async def misc_loop(self):
+        logger.debug("mqtt misc_loop() started")
+
+        while self.client.loop_misc() == mqtt.MQTT_ERR_SUCCESS:
+            try:
+                await asyncio.sleep(1)
+            except asyncio.CancelledError:
+                break
+
+        logger.debug("mqtt exiting misc_loop()")
+
+
+class AsyncMqttClient:
+    loop: asyncio.AbstractEventLoop
+    
+    def __init__(self):
+        self.client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id="dooris")
+        self.client.on_connect = self.on_connect
+        self.client.on_message = self.on_message
+        self.client.on_disconnect = self.on_disconnect
+
+    def on_connect(self, client, userdata, flags, reason_code, properties):
+        logger.debug("mqtt client connected")
+        print("client", type(client), client)
+        print("userdata", type(userdata), userdata)
+        print("flags", type(flags), flags)
+        print("reason_code", type(reason_code), reason_code)
+        print("properties", type(properties), properties)
+
+    def on_disconnect(self, client, userdata, flags, reason_code, properties):
+        logger.debug("mqtt client disconnected")
+        print("client", type(client), client)
+        print("userdata", type(userdata), userdata)
+        print("flags", type(flags), flags)
+        print("reason_code", type(reason_code), reason_code)
+        print("properties", type(properties), properties)
+
+    def on_message(self, client, userdata, msg):
+        logger.debug("mqtt client got message")
+        print("client", type(client), client)
+        print("userdata", type(userdata), userdata)
+        print("msg", type(msg), msg)
+
+    async def connect(self):
+        looper = AsyncLooper(asyncio.get_running_loop(), self.client)
+        self.client.connect("mqtt.eclipseprojects.io", 1883, 60)
+        self.client.socket().setsockopt(socket.SOL_SOCKET, socket.SO_SNDBUF, 2048)    
+
diff --git a/api/uv.lock b/api/uv.lock
index 9e7f71a..6b6a649 100644
--- a/api/uv.lock
+++ b/api/uv.lock
@@ -400,6 +400,7 @@ source = { editable = "." }
 dependencies = [
     { name = "aiohttp" },
     { name = "fastapi" },
+    { name = "paho-mqtt" },
     { name = "simple-openid-connect" },
     { name = "uvicorn" },
 ]
@@ -413,6 +414,7 @@ dev = [
 requires-dist = [
     { name = "aiohttp", specifier = ">=3.13.5" },
     { name = "fastapi", specifier = ">=0.136.1" },
+    { name = "paho-mqtt", specifier = ">=2.1.0" },
     { name = "simple-openid-connect", specifier = ">=2.4.0" },
     { name = "uvicorn", specifier = ">=0.46.0" },
 ]
@@ -734,6 +736,15 @@ wheels = [
     { url = "https://files.pythonhosted.org/packages/b2/6c/d8a02ffb24876b5f51fbd781f479fc6525a518553a4196bd0433dae9ff8e/orderedmultidict-1.0.2-py2.py3-none-any.whl", hash = "sha256:ab5044c1dca4226ae4c28524cfc5cc4c939f0b49e978efa46a6ad6468049f79b", size = 11897, upload-time = "2025-11-18T08:00:41.44Z" },
 ]
 
+[[package]]
+name = "paho-mqtt"
+version = "2.1.0"
+source = { registry = "https://pypi.org/simple" }
+sdist = { url = "https://files.pythonhosted.org/packages/39/15/0a6214e76d4d32e7f663b109cf71fb22561c2be0f701d67f93950cd40542/paho_mqtt-2.1.0.tar.gz", hash = "sha256:12d6e7511d4137555a3f6ea167ae846af2c7357b10bc6fa4f7c3968fc1723834", size = 148848, upload-time = "2024-04-29T19:52:55.591Z" }
+wheels = [
+    { url = "https://files.pythonhosted.org/packages/c4/cb/00451c3cf31790287768bb12c6bec834f5d292eaf3022afc88e14b8afc94/paho_mqtt-2.1.0-py3-none-any.whl", hash = "sha256:6db9ba9b34ed5bc6b6e3812718c7e06e2fd7444540df2455d2c51bd58808feee", size = 67219, upload-time = "2024-04-29T19:52:48.345Z" },
+]
+
 [[package]]
 name = "parso"
 version = "0.8.7"