diff --git a/api/src/dooris_api/__init__.py b/api/src/dooris_api/__init__.py index 203d379..cdf4d67 100644 --- a/api/src/dooris_api/__init__.py +++ b/api/src/dooris_api/__init__.py @@ -81,6 +81,7 @@ def main(): default=[i for i in os.environ.get("DOORIS_STATIC_API_TOKENS", "").split(",") if bool(i)], ) args = argp.parse_args() + print(args.static_api_tokens) # setup logging logging.basicConfig( diff --git a/api/src/dooris_api/ccujack.py b/api/src/dooris_api/ccujack.py index de08bbf..03d0480 100644 --- a/api/src/dooris_api/ccujack.py +++ b/api/src/dooris_api/ccujack.py @@ -90,12 +90,15 @@ class CCUJackClient: self.data_updated = asyncio.Event() async def start(self): - self.task_process_messages = asyncio.get_running_loop().create_task( - self.process_mqt_messages(), name="process-mqtt-messages" - ) + await self.mqtt.connect() + await self.find_locks() + self.task_cron= asyncio.get_running_loop().create_task( self.cron(), name="ccujack-cron" ) + 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()) @@ -159,18 +162,15 @@ class CCUJackClient: while True: try: + await asyncio.sleep(15 * 60) # 15 minutes logger.info("Running CCUJack cron") - if not self.mqtt.is_connected: + await self.find_locks() + if not self.mqtt.is_connected(): logger.warning("MQTT client was discovered to be disconnected; reconnecting now") await self.mqtt.connect() - await self.find_locks() - except Exception as e: logger.exception(f"Error in CCUJack cron task: {e}") - finally: - await asyncio.sleep(15 * 60) # 15 minutes - async def query_param_value(self, address: str) -> CCUValue: if address in self.param_values: diff --git a/api/src/dooris_api/mqtt_client.py b/api/src/dooris_api/mqtt_client.py index a090c5f..3df1f19 100644 --- a/api/src/dooris_api/mqtt_client.py +++ b/api/src/dooris_api/mqtt_client.py @@ -128,21 +128,19 @@ class AsyncMqttClient: 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 self.active_subscriptions.update(to_add) - if self.is_connected: - 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 self.active_subscriptions.difference_update(to_remove) - if self.is_connected: - 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) @@ -159,19 +157,13 @@ class AsyncMqttClient: # re-establish all supposed mqtt subscriptions if len(self.active_subscriptions) > 0: qos = 1 - self.fut_subscribe = asyncio.get_running_loop().create_future() - self.client.subscribe([(i, qos) for i in self.active_subscriptions]) - await self.fut_subscribe + await self.client.subscribe((i, qos) for i in self.active_subscriptions) async def disconnect(self): - if not self.is_connected: - return - logger.info("Disconnecting mqtt client from broker") self.fut_disconnect = asyncio.get_running_loop().create_future() self.client.disconnect() await self.fut_disconnect - @property def is_connected(self) -> bool: return self.client.is_connected()