diff --git a/api/src/dooris_api/__init__.py b/api/src/dooris_api/__init__.py index 8ee8809..cd488a6 100644 --- a/api/src/dooris_api/__init__.py +++ b/api/src/dooris_api/__init__.py @@ -52,12 +52,6 @@ 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, diff --git a/api/src/dooris_api/app.py b/api/src/dooris_api/app.py index f148f53..fa3d75e 100644 --- a/api/src/dooris_api/app.py +++ b/api/src/dooris_api/app.py @@ -39,11 +39,9 @@ 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), - mqtt_conn=app_cfg.ccujack_mqtt, + auth=BasicAuth(app_cfg.ccujack_user, app_cfg.ccujack_password) ) await app.extra["ccujack"].find_locks() - await app.extra["ccujack"].connect_mqtt() yield diff --git a/api/src/dooris_api/ccujack.py b/api/src/dooris_api/ccujack.py index 78c9e00..5d5eca0 100644 --- a/api/src/dooris_api/ccujack.py +++ b/api/src/dooris_api/ccujack.py @@ -4,8 +4,6 @@ import logging import asyncio from pydantic import BaseModel, Field -from dooris_api.mqtt_client import AsyncMqttClient - logger = logging.getLogger(__name__) @@ -71,7 +69,7 @@ class CCUJackClient: base_uri: str locks: LockData - def __init__(self, base_uri: str, auth: BasicAuth, mqtt_conn: str): + def __init__(self, base_uri: str, auth: BasicAuth): self.http = ClientSession( base_url=base_uri, auth=auth, @@ -79,10 +77,6 @@ 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") diff --git a/api/src/dooris_api/mqtt_client.py b/api/src/dooris_api/mqtt_client.py index f7cb22b..d11ce39 100644 --- a/api/src/dooris_api/mqtt_client.py +++ b/api/src/dooris_api/mqtt_client.py @@ -3,7 +3,6 @@ # https://github.com/eclipse-paho/paho.mqtt.python/blob/master/examples/loop_asyncio.py # -from typing import Any import logging import asyncio import socket @@ -29,64 +28,78 @@ 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, connection_string: str, username: str, password: str): - self.connection_string = connection_string + def __init__(self): 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: 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: mqtt.Client, userdata: Any, flags: mqtt.DisconnectFlags, reason_code, properties: mqtt.Properties): - logger.debug("mqtt client disconnected") + 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_message(self, client: mqtt.Client, userdata: Any, msg): + 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): - server_host, server_port = self.connection_string.rsplit(":", maxsplit=1) - looper = AsyncLooper(asyncio.get_running_loop(), self.client) - - logger.info("Connecting to mqtt server at %s:%s", server_host, server_port) - self.client.connect(server_host, int(server_port)) + self.client.connect("mqtt.eclipseprojects.io", 1883, 60) self.client.socket().setsockopt(socket.SOL_SOCKET, socket.SO_SNDBUF, 2048)