From 1a50d67df68e62b3b018aeddd4980a4dc839137b Mon Sep 17 00:00:00 2001 From: lilly Date: Tue, 19 May 2026 09:34:51 +0200 Subject: [PATCH 1/5] api: implement abstraction for connecting to CCUJACK MQTT broker --- api/pyproject.toml | 1 + api/src/dooris_api/mqtt_client.py | 92 +++++++++++++++++++++++++++++++ api/uv.lock | 11 ++++ 3 files changed, 104 insertions(+) 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/mqtt_client.py b/api/src/dooris_api/mqtt_client.py new file mode 100644 index 0000000..f7cb22b --- /dev/null +++ b/api/src/dooris_api/mqtt_client.py @@ -0,0 +1,92 @@ +# +# This whole implementation is adapted from the upstream GitHub example +# https://github.com/eclipse-paho/paho.mqtt.python/blob/master/examples/loop_asyncio.py +# + +from typing import Any +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): + def cb(): + 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): + self.loop.remove_reader(sock) + self.task_misc.cancel() + + def on_socket_register_write(self, client, userdata, sock): + def cb(): + client.loop_write() + + self.loop.add_writer(sock, cb) + + def on_socket_unregister_write(self, client, userdata, sock): + self.loop.remove_writer(sock) + + async def misc_loop(self): + while self.client.loop_misc() == mqtt.MQTT_ERR_SUCCESS: + try: + await asyncio.sleep(1) + except asyncio.CancelledError: + break + + +class AsyncMqttClient: + loop: asyncio.AbstractEventLoop + + 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: 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") + 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): + logger.debug("mqtt client got message") + 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.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" From 0331dd6406f9e6f06d1fcb741e89fdc2862a3f70 Mon Sep 17 00:00:00 2001 From: lilly Date: Tue, 19 May 2026 09:34:51 +0200 Subject: [PATCH 2/5] api: connect to CCUJACK MQTT broker on startup --- api/src/dooris_api/__init__.py | 6 ++++++ api/src/dooris_api/app.py | 5 ++++- api/src/dooris_api/ccujack.py | 8 +++++++- 3 files changed, 17 insertions(+), 2 deletions(-) diff --git a/api/src/dooris_api/__init__.py b/api/src/dooris_api/__init__.py index cd488a6..8ee8809 100644 --- a/api/src/dooris_api/__init__.py +++ b/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, diff --git a/api/src/dooris_api/app.py b/api/src/dooris_api/app.py index 3bcd848..3b53d35 100644 --- a/api/src/dooris_api/app.py +++ b/api/src/dooris_api/app.py @@ -36,11 +36,14 @@ 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) + 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 diff --git a/api/src/dooris_api/ccujack.py b/api/src/dooris_api/ccujack.py index 5d5eca0..78c9e00 100644 --- a/api/src/dooris_api/ccujack.py +++ b/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") From 44d484cfc196058bca136efbd38d874fba094327 Mon Sep 17 00:00:00 2001 From: lilly Date: Tue, 19 May 2026 13:09:00 +0200 Subject: [PATCH 3/5] api: use proper connection shutdown for downstream services --- api/src/dooris_api/app.py | 13 ++--- api/src/dooris_api/ccujack.py | 16 ++++++ api/src/dooris_api/mqtt_client.py | 83 +++++++++++++++++++++++++------ 3 files changed, 92 insertions(+), 20 deletions(-) diff --git a/api/src/dooris_api/app.py b/api/src/dooris_api/app.py index 3b53d35..32d626d 100644 --- a/api/src/dooris_api/app.py +++ b/api/src/dooris_api/app.py @@ -2,7 +2,6 @@ from typing import Optional, List import logging import secrets import sys -import os from datetime import datetime, UTC from fastapi import FastAPI, Request, Response, status from fastapi.responses import RedirectResponse @@ -42,11 +41,13 @@ async def lifespan(app: FastAPI): 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() + await app.extra["ccujack"].find_locks() yield + await app.extra["ccujack"].close_connections() + app = FastAPI( title="Dooris", @@ -71,9 +72,7 @@ async def get_user_info( ) -> models.UserStatus: return models.UserStatus( is_authorized=current_user.may_operate_locks, - guaranteed_session_until=datetime.fromtimestamp( - current_user.id_token.exp, UTC - ), + guaranteed_session_until=datetime.fromtimestamp(current_user.id_token.exp, UTC), username=current_user.id_token.preferred_username, ccchh_roles=current_user.ccchh_roles, ) @@ -122,7 +121,9 @@ async def login_init( response_class=RedirectResponse, status_code=302, ) -async def login_callback(req: Request, resp: Response, oidc_client: deps.OpenidClient) -> str: +async def login_callback( + req: Request, resp: Response, oidc_client: deps.OpenidClient +) -> str: # check that the user is currently in an authenticating state # these cookies are set by the login_init() view if ( diff --git a/api/src/dooris_api/ccujack.py b/api/src/dooris_api/ccujack.py index 78c9e00..ed1e038 100644 --- a/api/src/dooris_api/ccujack.py +++ b/api/src/dooris_api/ccujack.py @@ -84,8 +84,16 @@ class CCUJackClient: async def connect_mqtt(self): await self.mqtt.connect() + async def close_connections(self): + await asyncio.gather( + self.mqtt.disconnect(), + self.http.close() + ) + async def find_locks(self): logger.debug("Inspecting lock devices present in CCUJack") + + # iterate through the CCUJACK API to find all devices async with self.http.get("/device") as resp: devices = CCUDeviceList.model_validate(await resp.json()) @@ -99,6 +107,14 @@ class CCUJackClient: self.locks = [i for i in device_infos if i[0].type == DEVICE_TYPE_LOCK] + # update active mqtt subscriptions + mqtt_topics = set() + for i_lock, lock_channels in self.locks: + for i_channel, channel_params in lock_channels: + for i_param in channel_params: + mqtt_topics.add(f"device/status/{i_lock.address}/{i_channel.index}/{i_param.id}") + # await self.mqtt.update_subscriptions(mqtt_topics) + async def query_param_value(self, address: str): logger.debug("Querying parameter value from '%s'", address) async with self.http.get(f"/device/{address}/~pv") as resp: diff --git a/api/src/dooris_api/mqtt_client.py b/api/src/dooris_api/mqtt_client.py index f7cb22b..3f6de6f 100644 --- a/api/src/dooris_api/mqtt_client.py +++ b/api/src/dooris_api/mqtt_client.py @@ -1,9 +1,9 @@ # # This whole implementation is adapted from the upstream GitHub example # https://github.com/eclipse-paho/paho.mqtt.python/blob/master/examples/loop_asyncio.py -# +# -from typing import Any +from typing import Any, List, Set, Iterable import logging import asyncio import socket @@ -16,7 +16,7 @@ logger = logging.getLogger(__name__) class AsyncLooper: """ - Helper class to implement loopgin with asyncio for the underlying mqtt IO + Helper class to implement loopgin with asyncio for the underlying mqtt IO """ def __init__(self, loop: asyncio.AbstractEventLoop, client: mqtt.Client): @@ -58,9 +58,14 @@ class AsyncLooper: class AsyncMqttClient: loop: asyncio.AbstractEventLoop - + connection_string: str + looper: AsyncLooper + client: mqtt.Client + active_subscriptions: Set[str] + def __init__(self, connection_string: str, username: str, password: str): self.connection_string = connection_string + self.active_subscriptions = set() self.client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id="dooris") self.client.username = username self.client.password = password @@ -68,25 +73,75 @@ class AsyncMqttClient: 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): + 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}'") + self.fut_connected.set_result(None) - def on_disconnect(self, client: mqtt.Client, userdata: Any, flags: mqtt.DisconnectFlags, reason_code, properties: mqtt.Properties): - logger.debug("mqtt client disconnected") - print("flags", type(flags), flags) - print("reason_code", type(reason_code), reason_code) - print("properties", type(properties), properties) + def on_disconnect( + self, + client: mqtt.Client, + userdata: Any, + flags: mqtt.DisconnectFlags, + reason_code, + properties: mqtt.Properties, + ): + logger.debug(f"mqtt client disconnected with message '{reason_code}'") + if self.fut_disconnect: + self.fut_disconnect.set_result(None) - def on_message(self, client: mqtt.Client, userdata: Any, msg): + def on_message(self, client: mqtt.Client, userdata: Any, msg: mqtt.MQTTMessage): logger.debug("mqtt client got message") print("msg", type(msg), msg) + def on_subscribe(self, client, userdata, mid, reason_code, properties): + logger.debug(f"mqtt client subscribed to topics with message '{reason_code}'") + self.fut_subscribe.set_result(None) + + def on_unsubscribe(self, client, userdata, mid, reason_code, properties): + logger.debug(f"mqtt client unsubscribed from topics with message '{reason_code}'") + self.fut_unsubscribe.set_result(None) + + async def update_subscriptions(self, topics: Iterable[str]): + """ + Update MQTT subscriptions so that the client is subscribed to exactly the given list of topics + """ + + to_add = topics.difference(self.active_subscriptions) + if to_add: + logger.info(f"mqtt client subscribing to topics {', '.join(to_add)}") + qos = 2 + self.fut_subscribe = asyncio.get_running_loop().create_future() + self.client.subscribe([(i, qos) for i in to_add]) + await self.fut_subscribe + + to_remove = self.active_subscriptions.difference(topics) + if to_remove: + logger.info(f"mqtt client unsubscribing from topics {','.join(to_remove)}") + self.fut_unsubscribe = asyncio.get_running_loop().create_future() + self.client.unsubscribe(list(to_remove)) + await self.fut_unsubscribe + async def connect(self): server_host, server_port = self.connection_string.rsplit(":", maxsplit=1) - - looper = AsyncLooper(asyncio.get_running_loop(), self.client) + + self.looper = AsyncLooper(asyncio.get_running_loop(), self.client) logger.info("Connecting to mqtt server at %s:%s", server_host, server_port) + self.fut_connected = asyncio.get_running_loop().create_future() self.client.connect(server_host, int(server_port)) - self.client.socket().setsockopt(socket.SOL_SOCKET, socket.SO_SNDBUF, 2048) + self.client.socket().setsockopt(socket.SOL_SOCKET, socket.SO_SNDBUF, 2048) + await self.fut_connected + + async def disconnect(self): + logger.info("Disconnecting mqtt client from broker") + self.fut_disconnect = asyncio.get_running_loop().create_future() + self.client.disconnect() + await self.fut_disconnect From 4103c0ca5f6c12518bbd8efb43eb975e508de837 Mon Sep 17 00:00:00 2001 From: lilly Date: Tue, 19 May 2026 14:10:47 +0200 Subject: [PATCH 4/5] api: process mqtt messages to keep a current local lock state --- api/src/dooris_api/ccujack.py | 44 ++++++++++++++++++++++++------- api/src/dooris_api/mqtt_client.py | 24 ++++++++++++----- 2 files changed, 53 insertions(+), 15 deletions(-) diff --git a/api/src/dooris_api/ccujack.py b/api/src/dooris_api/ccujack.py index ed1e038..112fbc1 100644 --- a/api/src/dooris_api/ccujack.py +++ b/api/src/dooris_api/ccujack.py @@ -1,4 +1,4 @@ -from typing import List, Tuple, Optional, Any +from typing import List, Tuple, Optional, Any, Dict from aiohttp import ClientSession, BasicAuth, TCPConnector import logging import asyncio @@ -70,6 +70,8 @@ LockData = List[Tuple[CCUDeviceInfo, List[Tuple[CCUChannelInfo, List[CCUParamInf class CCUJackClient: base_uri: str locks: LockData + param_values: Dict[str, Any] + task_process_messages: asyncio.Task def __init__(self, base_uri: str, auth: BasicAuth, mqtt_conn: str): self.http = ClientSession( @@ -78,17 +80,21 @@ class CCUJackClient: raise_for_status=True, connector=TCPConnector(ssl=False), ) - self.locks = None self.mqtt = AsyncMqttClient(mqtt_conn, auth.login, auth.password) + self.locks = None + self.param_values = dict() + self.task_process_messages = None async def connect_mqtt(self): await self.mqtt.connect() + self.task_process_messages = asyncio.get_running_loop().create_task( + self.process_mqt_messages(), name="process-mqtt-messages" + ) async def close_connections(self): - await asyncio.gather( - self.mqtt.disconnect(), - self.http.close() - ) + await asyncio.gather(self.mqtt.disconnect(), self.http.close()) + self.task_process_messages.cancel() + self.task_process_messages = None async def find_locks(self): logger.debug("Inspecting lock devices present in CCUJack") @@ -112,10 +118,30 @@ class CCUJackClient: for i_lock, lock_channels in self.locks: for i_channel, channel_params in lock_channels: for i_param in channel_params: - mqtt_topics.add(f"device/status/{i_lock.address}/{i_channel.index}/{i_param.id}") - # await self.mqtt.update_subscriptions(mqtt_topics) + mqtt_topics.add( + f"device/status/{i_lock.address}/{i_channel.index}/{i_param.id}" + ) + await self.mqtt.update_subscriptions(mqtt_topics) + + async def process_mqt_messages(self): + while True: + try: + msg = await self.mqtt.messages.get() + + param_name = msg.topic.removeprefix("device/status/") + param_value = CCUValue.model_validate_json(msg.payload) + logger.debug( + f"Got new value from MQTT for parameter {param_name}: {param_value}" + ) + self.param_values[param_name] = param_value + + except Exception as e: + logger.exception(f"could not process incoming mqtt message: {e}") + + async def query_param_value(self, address: str) -> CCUValue: + if address in self.param_values: + return self.param_values[address] - async def query_param_value(self, address: str): logger.debug("Querying parameter value from '%s'", address) async with self.http.get(f"/device/{address}/~pv") as resp: return CCUValue.model_validate(await resp.json()) diff --git a/api/src/dooris_api/mqtt_client.py b/api/src/dooris_api/mqtt_client.py index 3f6de6f..1548e66 100644 --- a/api/src/dooris_api/mqtt_client.py +++ b/api/src/dooris_api/mqtt_client.py @@ -62,16 +62,22 @@ class AsyncMqttClient: looper: AsyncLooper client: mqtt.Client active_subscriptions: Set[str] + messages: asyncio.Queue def __init__(self, connection_string: str, username: str, password: str): self.connection_string = connection_string self.active_subscriptions = set() + self.messages = asyncio.Queue() 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_connect_fail = self.on_connect_fail self.client.on_message = self.on_message self.client.on_disconnect = self.on_disconnect + self.client.on_subscribe = self.on_subscribe + self.client.on_unsubscribe = self.on_unsubscribe + self.client.on_disconnect = self.on_disconnect def on_connect( self, @@ -84,6 +90,12 @@ class AsyncMqttClient: logger.debug(f"mqtt client connected with message '{reason_code}'") self.fut_connected.set_result(None) + def on_connect_fail(self, client, userdata): + logger.error("mqtt client could not connect to broker") + self.fut_connected.set_exception( + Exception("mqtt client could not connect to broker") + ) + def on_disconnect( self, client: mqtt.Client, @@ -97,26 +109,26 @@ class AsyncMqttClient: self.fut_disconnect.set_result(None) def on_message(self, client: mqtt.Client, userdata: Any, msg: mqtt.MQTTMessage): - logger.debug("mqtt client got message") - print("msg", type(msg), msg) + self.messages.put_nowait(msg) def on_subscribe(self, client, userdata, mid, reason_code, properties): logger.debug(f"mqtt client subscribed to topics with message '{reason_code}'") self.fut_subscribe.set_result(None) def on_unsubscribe(self, client, userdata, mid, reason_code, properties): - logger.debug(f"mqtt client unsubscribed from topics with message '{reason_code}'") + logger.debug( + f"mqtt client unsubscribed from topics with message '{reason_code}'" + ) self.fut_unsubscribe.set_result(None) - async def update_subscriptions(self, topics: Iterable[str]): + async def update_subscriptions(self, topics: Iterable[str], qos: int = 1): """ - Update MQTT subscriptions so that the client is subscribed to exactly the given list of topics + Update MQTT subscriptions so that the client is subscribed to exactly the given list of topics """ to_add = topics.difference(self.active_subscriptions) if to_add: logger.info(f"mqtt client subscribing to topics {', '.join(to_add)}") - qos = 2 self.fut_subscribe = asyncio.get_running_loop().create_future() self.client.subscribe([(i, qos) for i in to_add]) await self.fut_subscribe From 7ac0a4106cdc621f7b62dab7317d7c0c6317fb9b Mon Sep 17 00:00:00 2001 From: lilly Date: Tue, 19 May 2026 14:37:55 +0200 Subject: [PATCH 5/5] api: implement Server-Sent-Events on top of mqtt parameters --- api/src/dooris_api/app.py | 15 ++++++++++++++- api/src/dooris_api/ccujack.py | 34 ++++++++++++++++++++++------------ 2 files changed, 36 insertions(+), 13 deletions(-) diff --git a/api/src/dooris_api/app.py b/api/src/dooris_api/app.py index 32d626d..edb338f 100644 --- a/api/src/dooris_api/app.py +++ b/api/src/dooris_api/app.py @@ -1,10 +1,11 @@ -from typing import Optional, List +from typing import Optional, List, AsyncIterable import logging import secrets import sys from datetime import datetime, UTC from fastapi import FastAPI, Request, Response, status from fastapi.responses import RedirectResponse +from fastapi.sse import EventSourceResponse from contextlib import asynccontextmanager from simple_openid_connect.client import OpenidClient from simple_openid_connect.data import TokenSuccessResponse, RpInitiatedLogoutRequest @@ -241,6 +242,18 @@ async def list_locks(ccujack: deps.CCUJackClient) -> List[models.Lock]: return result +@app.get( + "/api/locks/stream", + tags=["locks"], + responses={status.HTTP_401_UNAUTHORIZED: {"model": models.HttpProblemDetail}}, + response_class=EventSourceResponse, +) +async def watch_locks(ccujack: deps.CCUJackClient) -> AsyncIterable[List[models.Lock]]: + while True: + yield await list_locks(ccujack) + await ccujack.data_updated.wait() + + @app.patch( "/api/locks/{lock_id}", tags=["locks"], diff --git a/api/src/dooris_api/ccujack.py b/api/src/dooris_api/ccujack.py index 112fbc1..1132e63 100644 --- a/api/src/dooris_api/ccujack.py +++ b/api/src/dooris_api/ccujack.py @@ -72,6 +72,7 @@ class CCUJackClient: locks: LockData param_values: Dict[str, Any] task_process_messages: asyncio.Task + data_updated: asyncio.Event def __init__(self, base_uri: str, auth: BasicAuth, mqtt_conn: str): self.http = ClientSession( @@ -84,6 +85,7 @@ class CCUJackClient: self.locks = None self.param_values = dict() self.task_process_messages = None + self.data_updated = asyncio.Event() async def connect_mqtt(self): await self.mqtt.connect() @@ -99,10 +101,10 @@ class CCUJackClient: async def find_locks(self): logger.debug("Inspecting lock devices present in CCUJack") - # iterate through the CCUJACK API to find all devices async with self.http.get("/device") as resp: devices = CCUDeviceList.model_validate(await resp.json()) + # inspect CCUJACK for locks device_infos = await asyncio.gather( *[ self._inspect_ccu_device(i) @@ -111,17 +113,22 @@ class CCUJackClient: ] ) - self.locks = [i for i in device_infos if i[0].type == DEVICE_TYPE_LOCK] - - # update active mqtt subscriptions - mqtt_topics = set() - for i_lock, lock_channels in self.locks: - for i_channel, channel_params in lock_channels: - for i_param in channel_params: - mqtt_topics.add( - f"device/status/{i_lock.address}/{i_channel.index}/{i_param.id}" - ) - await self.mqtt.update_subscriptions(mqtt_topics) + # save the result + new_locks = [i for i in device_infos if i[0].type == DEVICE_TYPE_LOCK] + if new_locks != self.locks: + self.locks = new_locks + self.data_updated.set() + self.data_updated.clear() + + # update active mqtt subscriptions based on newly discovered devices + mqtt_topics = set() + for i_lock, lock_channels in self.locks: + for i_channel, channel_params in lock_channels: + for i_param in channel_params: + mqtt_topics.add( + f"device/status/{i_lock.address}/{i_channel.index}/{i_param.id}" + ) + await self.mqtt.update_subscriptions(mqtt_topics) async def process_mqt_messages(self): while True: @@ -134,9 +141,12 @@ class CCUJackClient: f"Got new value from MQTT for parameter {param_name}: {param_value}" ) self.param_values[param_name] = param_value + self.data_updated.set() except Exception as e: logger.exception(f"could not process incoming mqtt message: {e}") + finally: + self.data_updated.clear() async def query_param_value(self, address: str) -> CCUValue: if address in self.param_values: