api: implement abstraction for connecting to CCUJACK MQTT broker

api/src/dooris_api/__init__.py

@@ -52,6 +52,12 @@ def main():
         default=os.environ.get("DOORIS_CCUJACK_URL", "https://hmdooris-ccu.ccchh.net:2122"),
         help="The URL under which a CCUJACK instance is hosted that actually operates the locks",
     )
+    argp.add_argument(
+        "--ccujack-mqtt",
+        required=False,
+        default=os.environ.get("DOORIS_CCUJACK_MQTT", "hmdooris-ccu.ccchh.net:1883"),
+        help="The $HOSTNAME:$PORT of the CCUJack embedded MQTT server",
+    )
     argp.add_argument(
         "--ccujack-user",
         required="DOORIS_CCUJACK_USER" not in os.environ,

api/src/dooris_api/app.py

@@ -39,9 +39,11 @@ async def lifespan(app: FastAPI):
     # 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)
+        auth=BasicAuth(app_cfg.ccujack_user, app_cfg.ccujack_password),
+        mqtt_conn=app_cfg.ccujack_mqtt,
     )
     await app.extra["ccujack"].find_locks()
+    await app.extra["ccujack"].connect_mqtt()
 
     yield
 

api/src/dooris_api/ccujack.py

@@ -4,6 +4,8 @@ import logging
 import asyncio
 from pydantic import BaseModel, Field
 
+from dooris_api.mqtt_client import AsyncMqttClient
+
 
 logger = logging.getLogger(__name__)
 
@@ -69,7 +71,7 @@ class CCUJackClient:
     base_uri: str
     locks: LockData
 
-    def __init__(self, base_uri: str, auth: BasicAuth):
+    def __init__(self, base_uri: str, auth: BasicAuth, mqtt_conn: str):
         self.http = ClientSession(
             base_url=base_uri,
             auth=auth,
@@ -77,6 +79,10 @@ class CCUJackClient:
             connector=TCPConnector(ssl=False),
         )
         self.locks = None
+        self.mqtt = AsyncMqttClient(mqtt_conn, auth.login, auth.password)
+
+    async def connect_mqtt(self):
+        await self.mqtt.connect()
 
     async def find_locks(self):
         logger.debug("Inspecting lock devices present in CCUJack")

api/src/dooris_api/mqtt_client.py

@@ -3,6 +3,7 @@
 # https://github.com/eclipse-paho/paho.mqtt.python/blob/master/examples/loop_asyncio.py
 # 
 
+from typing import Any
 import logging
 import asyncio
 import socket
@@ -28,78 +29,64 @@ class AsyncLooper:
         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):
+    def __init__(self, connection_string: str, username: str, password: str):
+        self.connection_string = connection_string
         self.client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id="dooris")
+        self.client.username = username
+        self.client.password = password
         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_connect(self, client: mqtt.Client, userdata: Any, flags: mqtt.ConnectFlags, reason_code, properties: mqtt.Properties):
+        logger.debug(f"mqtt client connected with message '{reason_code}'")
 
-    def on_disconnect(self, client, userdata, flags, reason_code, properties):
+    def on_disconnect(self, client: mqtt.Client, userdata: Any, flags: mqtt.DisconnectFlags, reason_code, properties: mqtt.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):
+    def on_message(self, client: mqtt.Client, userdata: Any, 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):
+        server_host, server_port = self.connection_string.rsplit(":", maxsplit=1)
+        
         looper = AsyncLooper(asyncio.get_running_loop(), self.client)
-        self.client.connect("mqtt.eclipseprojects.io", 1883, 60)
+
+        logger.info("Connecting to mqtt server at %s:%s", server_host, server_port)
+        self.client.connect(server_host, int(server_port))
         self.client.socket().setsockopt(socket.SOL_SOCKET, socket.SO_SNDBUF, 2048)