implement door status synchronization towards club-assistant

api/src/dooris_api/__init__.py

@@ -123,6 +123,24 @@ def main():
         default=os.environ.get("DOORIS_MAIN_LOCK", None),
         help="Name or ID of the door which will be considered the main door for spaceapi. If none is set, integration to spaceapi is disabled",
     )
+    argp.add_argument(
+        "--home-assistant-mqtt",
+        required=False,
+        default=os.environ.get("DOORIS_HOME_ASSISTANT_MQTT", "club-assistant.ccchh.net:1883"),
+        help="$HOSTNAME:$PORT of an MQTT server connected to our Home-Assistant",
+    )
+    argp.add_argument(
+        "--home-assistant-mqtt-user",
+        required=False,
+        default=os.environ.get("DOORIS_HOME_ASSISTANT_MQTT_USER", "dooris"),
+        help="Username to authenticate towards our Home-Assistant MQTT server",
+    )
+    argp.add_argument(
+        "--home-assistant-mqtt-password",
+        required="DOORIS_HOME_ASSISTANT_MQTT_PASSWORD" not in os.environ,
+        default=os.environ.get("DOORIS_HOME_ASSISTANT_MQTT_PASSWORD", None),
+        help="Password to authenticate towards our Home-Assistant MQTT server",
+    )
 
     args = argp.parse_args()
 

api/src/dooris_api/app.py

@@ -15,6 +15,7 @@ from dooris_api import deps, models, exceptions, app_config
 from dooris_api.ccujack import CCUJackClient
 from dooris_api.keycloak import KeycloakClient
 from dooris_api.spaceapi import SpaceApiClient
+from dooris_api.home_assistant import HomeAssistantClient
 
 
 logger = logging.getLogger(__name__)
@@ -32,6 +33,14 @@ async def lifespan(app: FastAPI):
         scope=app_cfg.openid_scope,
     )
 
+    app.extra["home_assistant"] = HomeAssistantClient(
+        auth=BasicAuth(
+            app_cfg.home_assistant_mqtt_user, app_cfg.home_assistant_mqtt_password
+        ),
+        mqtt_conn=app_cfg.home_assistant_mqtt,
+    )
+    await app.extra["home_assistant"].start()
+
     app.extra["spaceapi"] = SpaceApiClient(
         base_uri=app_cfg.spaceapid_base_uri,
         auth=BasicAuth(app_cfg.spaceapid_user, app_cfg.spaceapid_password),
@@ -43,6 +52,7 @@ async def lifespan(app: FastAPI):
         mqtt_conn=app_cfg.ccujack_mqtt,
         spaceapi_client=app.extra["spaceapi"],
         main_lock=app_cfg.main_lock,
+        home_assistant=app.extra["home_assistant"],
     )
     await app.extra["ccujack"].start()
 
@@ -56,8 +66,11 @@ async def lifespan(app: FastAPI):
 
     yield
 
-    await app.extra["ccujack"].close_connections()
-    await app.extra["keycloak"].stop()
+    await asyncio.gather(
+        app.extra["ccujack"].close_connections(),
+        app.extra["home_assistant"].close_connections(),
+        app.extra["keycloak"].stop(),
+    )
 
 
 app = FastAPI(

api/src/dooris_api/ccujack.py

@@ -6,6 +6,7 @@ from pydantic import BaseModel, Field
 
 from dooris_api.mqtt_client import AsyncMqttClient
 from dooris_api.spaceapi import SpaceApiClient
+from dooris_api.home_assistant import HomeAssistantClient
 
 
 logger = logging.getLogger(__name__)
@@ -76,9 +77,18 @@ class CCUJackClient:
     task_find_locks: asyncio.Task
     data_updated: asyncio.Event
     spaceapi_client: SpaceApiClient
+    home_assistant_client: HomeAssistantClient
     main_lock: Optional[str]
 
-    def __init__(self, base_uri: str, auth: BasicAuth, mqtt_conn: str, spaceapi_client: SpaceApiClient, main_lock: Optional[str]):
+    def __init__(
+        self,
+        base_uri: str,
+        auth: BasicAuth,
+        mqtt_conn: str,
+        spaceapi_client: SpaceApiClient,
+        main_lock: Optional[str],
+        home_assistant: HomeAssistantClient,
+    ):
         self.http = ClientSession(
             base_url=base_uri,
             auth=auth,
@@ -92,13 +102,14 @@ class CCUJackClient:
         self.task_find_locks = None
         self.data_updated = asyncio.Event()
         self.spaceapi_client = spaceapi_client
+        self.home_assistant_client = home_assistant
         self.main_lock = main_lock
 
     async def start(self):
         self.task_process_messages = asyncio.get_running_loop().create_task(
             self.process_mqt_messages(), name="process-mqtt-messages"
         )
-        self.task_cron= asyncio.get_running_loop().create_task(
+        self.task_cron = asyncio.get_running_loop().create_task(
             self.cron(), name="ccujack-cron"
         )
 
@@ -131,7 +142,7 @@ class CCUJackClient:
                 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:
@@ -141,7 +152,7 @@ class CCUJackClient:
                                 f"device/status/{i_lock.address}/{i_channel.index}/{i_param.id}"
                             )
                 await self.mqtt.update_subscriptions(mqtt_topics)
-                await self._update_spaceapi()
+                await self._update_federated_status()
 
     async def process_mqt_messages(self):
         while True:
@@ -157,7 +168,7 @@ class CCUJackClient:
                 self.data_updated.set()
 
                 if param_name.endswith("/LOCK_STATE"):
-                    await self._update_spaceapi()
+                    await self._update_federated_status()
 
             except Exception as e:
                 logger.exception(f"could not process incoming mqtt message: {e}")
@@ -167,10 +178,11 @@ class CCUJackClient:
     async def cron(self):
         while True:
             try:
-
                 logger.info("Running CCUJack cron")
                 if not self.mqtt.is_connected:
-                    logger.warning("MQTT client was discovered to be disconnected; reconnecting now")
+                    logger.warning(
+                        "MQTT client was discovered to be disconnected; reconnecting now"
+                    )
                     await self.mqtt.connect()
 
                 await self.find_locks()
@@ -181,8 +193,7 @@ class CCUJackClient:
             except Exception as e:
                 logger.exception(f"Error in CCUJack cron task: {e}")
             finally:
-                await asyncio.sleep(15 * 60) # 15 minutes
-            
+                await asyncio.sleep(15 * 60)  # 15 minutes
 
     async def query_param_value(self, address: str) -> CCUValue:
         if address in self.param_values:
@@ -196,45 +207,51 @@ class CCUJackClient:
         logger.debug("Writing parameter value '%s' to '%s'", value, address)
         await self.http.put(f"/device/{address}/~pv", json={"v": value})
 
-    async def _update_spaceapi(self):
+    async def _update_federated_status(self):
         """
-        Write the current lock state into spaceapid  
+        Write the current lock states into other systems suchs as SpaceApi or Club-Assistant
         """
         if not self.main_lock:
-            logger.debug("No main lock is configured, skipping spaceapid update")
-            return
+            logger.info("No main lock is configured, skipping spaceapid update")
 
-        logger.info(f"Pushing current state of lock {self.main_lock} to spaceapid")
-        for (i_lock, lock_channels) in self.locks:
-            if i_lock.identifier.lower() == self.main_lock.lower() or i_lock.title.lower() == self.main_lock.lower():
-                is_currently_open = False
+        for i_lock, lock_channels in self.locks:
+            # figure out the current lock state
+            binary_is_currently_open = False
+            detailed_status = "unknown"
+            for i_channel, params in lock_channels:
+                for i_param in params:
+                    if i_param.id == "LOCK_STATE":
+                        # do a proper lookup and not just directly access the cache because the method
+                        # could have been triggered by something other than MQTT in which case the value
+                        # would not necessarily be available yet
+                        value = await self.query_param_value(
+                            f"{i_lock.address}/{i_channel.index}/{i_param.id}"
+                        )
+                        match value.v:
+                            case 0:  # unknown
+                                detailed_status = "unknown"
+                                binary_is_currently_open = False
+                            case 1:  # locked
+                                detailed_status = "locked"
+                                binary_is_currently_open = False
+                            case 2:  # unlocked
+                                detailed_status = "unlocked"
+                                binary_is_currently_open = True
 
-                # find the correct parameter and extract it into a usable variable
-                for (i_channel, params) in lock_channels:
-                    for i_param in params:
-                        if i_param.id == "LOCK_STATE":
+            # post updates to home-assistant every time
+            await self.home_assistant_client.set_status(
+                i_lock.title.lower(), detailed_status
+            )
 
-                            # do a proper lookup and not just directly access the cache because the method
-                            # could have been triggered by something other than MQTT in which case the value
-                            # would not necessarily be available yet
-                            value = await self.query_param_value(
-                                f"{i_lock.address}/{i_channel.index}/{i_param.id}"
-                            )
-                            match value.v:
-                                case 0:  # unknown
-                                    is_currently_open = False
-                                case 1:  # locked
-                                    is_currently_open = False
-                                case 2:  # unlocked
-                                    is_currently_open = True
-
-                # actually post the update
-                await self.spaceapi_client.set_status(is_currently_open)
-
-                break
-
-        else:
-            logger.error(f"Could not push current lock state of lock {self.main_lock} to spaceapid because no such lock is known")
+            # post update to spaceapi only if thie is the main lock
+            if self.main_lock and (
+                i_lock.identifier.lower() == self.main_lock.lower()
+                or i_lock.title.lower() == self.main_lock.lower()
+            ):
+                logger.info(
+                    f"Pushing current state of lock {self.main_lock} to spaceapid"
+                )
+                await self.spaceapi_client.set_status(binary_is_currently_open)
 
     async def _inspect_ccu_device(
         self, device_ref: CCURef

api/src/dooris_api/home_assistant.py

@@ -0,0 +1,50 @@
+"""
+Integration into our clubs Home-Assistant for lock state synchonization
+"""
+
+import logging
+import asyncio
+from dooris_api.mqtt_client import AsyncMqttClient
+from aiohttp import BasicAuth
+
+
+logger = logging.getLogger(__name__)
+
+
+class HomeAssistantClient:
+    def __init__(self, auth: BasicAuth, mqtt_conn: str):
+        self.mqtt = AsyncMqttClient(mqtt_conn, auth.login, auth.password)
+
+    async def start(self):
+        self.task_cron = asyncio.get_running_loop().create_task(
+            self.cron(), name="home-assistant-cron"
+        )
+
+    async def set_status(self, lock_title: str, status: str):
+        logger.info(
+            f"Pushing status of lock '{lock_title}' -> '{status}' to Home-Assistant"
+        )
+        await self.mqtt.publish(
+            f"z9/dooris/{lock_title}/status", status.encode("UTF-8"), qos=1
+        )
+
+    async def cron(self):
+        while True:
+            try:
+                logger.info("Running Home-Assistant cron")
+                if not self.mqtt.is_connected:
+                    logger.warning(
+                        "MQTT client was discovered to be disconnected; reconnecting now"
+                    )
+                    await self.mqtt.connect()
+
+            except asyncio.CancelledError:
+                logger.info("Home-Assistant task cron stopped")
+                raise
+            except Exception as e:
+                logger.exception(f"Error in Home-Assistant cron task: {e}")
+            finally:
+                await asyncio.sleep(15 * 60)  # 15 minutes
+
+    async def close_connections(self):
+        await self.mqtt.disconnect()

api/src/dooris_api/mqtt_client.py

@@ -143,7 +143,9 @@ class AsyncMqttClient:
         if to_remove:
             self.active_subscriptions.difference_update(to_remove)
             if self.is_connected:
-                logger.info(f"mqtt client unsubscribing from topics {','.join(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
@@ -170,7 +172,7 @@ class AsyncMqttClient:
     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()
@@ -179,3 +181,7 @@ class AsyncMqttClient:
     @property
     def is_connected(self) -> bool:
         return self.client.is_connected()
+
+    async def publish(self, topic: str, msg: bytes, qos: int):
+        logger.debug(f"Publishing mqtt message on topic {topic}: {msg}")
+        self.client.publish(topic, msg, qos)