diff --git a/api/src/dooris_api/app.py b/api/src/dooris_api/app.py index edb338f..3b53d35 100644 --- a/api/src/dooris_api/app.py +++ b/api/src/dooris_api/app.py @@ -1,11 +1,11 @@ -from typing import Optional, List, AsyncIterable +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 -from fastapi.sse import EventSourceResponse from contextlib import asynccontextmanager from simple_openid_connect.client import OpenidClient from simple_openid_connect.data import TokenSuccessResponse, RpInitiatedLogoutRequest @@ -42,13 +42,11 @@ 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"].connect_mqtt() await app.extra["ccujack"].find_locks() + await app.extra["ccujack"].connect_mqtt() yield - await app.extra["ccujack"].close_connections() - app = FastAPI( title="Dooris", @@ -73,7 +71,9 @@ 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,9 +122,7 @@ 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 ( @@ -242,18 +240,6 @@ 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 1132e63..78c9e00 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, Dict +from typing import List, Tuple, Optional, Any from aiohttp import ClientSession, BasicAuth, TCPConnector import logging import asyncio @@ -70,9 +70,6 @@ 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 - data_updated: asyncio.Event def __init__(self, base_uri: str, auth: BasicAuth, mqtt_conn: str): self.http = ClientSession( @@ -81,30 +78,17 @@ class CCUJackClient: raise_for_status=True, connector=TCPConnector(ssl=False), ) - self.mqtt = AsyncMqttClient(mqtt_conn, auth.login, auth.password) self.locks = None - self.param_values = dict() - self.task_process_messages = None - self.data_updated = asyncio.Event() + self.mqtt = AsyncMqttClient(mqtt_conn, auth.login, auth.password) 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()) - self.task_process_messages.cancel() - self.task_process_messages = None async def find_locks(self): logger.debug("Inspecting lock devices present in CCUJack") - 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) @@ -113,45 +97,9 @@ class CCUJackClient: ] ) - # 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: - 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 - 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: - return self.param_values[address] + self.locks = [i for i in device_infos if i[0].type == DEVICE_TYPE_LOCK] + 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 1548e66..f7cb22b 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, List, Set, Iterable +from typing import Any 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,102 +58,35 @@ class AsyncLooper: class AsyncMqttClient: loop: asyncio.AbstractEventLoop - connection_string: str - 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, - 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_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, 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: mqtt.MQTTMessage): - 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}'" - ) - self.fut_unsubscribe.set_result(None) - - 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 - """ - - to_add = topics.difference(self.active_subscriptions) - if to_add: - logger.info(f"mqtt client subscribing to topics {', '.join(to_add)}") - 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 + 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) - - self.looper = AsyncLooper(asyncio.get_running_loop(), self.client) + + 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