from typing import List, Tuple, Optional, Any, Dict from aiohttp import ClientSession, BasicAuth, TCPConnector import logging import asyncio from pydantic import BaseModel, Field from dooris_api.mqtt_client import AsyncMqttClient logger = logging.getLogger(__name__) DEVICE_TYPE_LOCK = "HmIP-DLD" CHANNEL_TYPES_RELEVANT = ["DOOR_LOCK_STATE_TRANSMITTER", "MAINTENANCE"] PARAM_ADDRESSES_RELEVANT = [ "ACTIVITY_STATE", "LOCK_STATE", "LOCK_TARGET_LEVEL", "ERROR_JAMMED", "LOW_BAT", "UNREACH", ] class CCURef(BaseModel): href: str rel: str title: str class CCUDeviceList(BaseModel): description: str identifier: str title: str links: List[CCURef] = Field(alias="~links") class CCUDeviceInfo(BaseModel): address: str identifier: str title: str type: str links: List[CCURef] = Field(alias="~links") class CCUChannelInfo(BaseModel): address: Optional[str] = None index: Optional[int] = None title: str description: str = Field(default="") type: Optional[str] = Field(default=None) links: List[CCURef] = Field(alias="~links") class CCUParamInfo(BaseModel): id: str mqttStatusTopic: str links: List[CCURef] = Field(alias="~links") class CCUValue(BaseModel): s: int ts: int v: Any LockData = List[Tuple[CCUDeviceInfo, List[Tuple[CCUChannelInfo, List[CCUParamInfo]]]]] class CCUJackClient: base_uri: str locks: LockData param_values: Dict[str, Any] task_process_messages: asyncio.Task task_find_locks: asyncio.Task data_updated: asyncio.Event def __init__(self, base_uri: str, auth: BasicAuth, mqtt_conn: str): self.http = ClientSession( base_url=base_uri, auth=auth, 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.task_find_locks = None 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" ) self.task_cron= asyncio.get_running_loop().create_task( self.cron(), name="ccujack-cron" ) async def close_connections(self): await asyncio.gather(self.mqtt.disconnect(), self.http.close()) self.task_process_messages.cancel() self.task_process_messages = None self.task_cron.cancel() self.task_cron = 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) for i in devices.links if i.rel == "device" ] ) # save the result new_locks = [i for i in device_infos if i[0].type == DEVICE_TYPE_LOCK] if new_locks != self.locks: logger.info("Found new locks, updating state") 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 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") 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: return self.param_values[address] 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()) async def set_param_value(self, address: str, value: Any): logger.debug("Writing parameter value '%s' to '%s'", value, address) await self.http.put(f"/device/{address}/~pv", json={"v": value}) async def _inspect_ccu_device( self, device_ref: CCURef ) -> Tuple[CCUDeviceInfo, List[Tuple[CCUChannelInfo, List[CCUParamInfo]]]]: logger.debug("Inspecting device '%s' (%s)", device_ref.href, device_ref.title) async with self.http.get(f"/device/{device_ref.href}") as resp: device_info = CCUDeviceInfo.model_validate(await resp.json()) if device_info.type != DEVICE_TYPE_LOCK: return device_info, [] channel_infos = await asyncio.gather( *[ self._inspect_ccu_channel(device_ref, i) for i in device_info.links if i.rel == "channel" ] ) return device_info, [ i for i in channel_infos if i[0].type in CHANNEL_TYPES_RELEVANT ] async def _inspect_ccu_channel( self, device_ref: CCURef, channel_ref: CCURef ) -> Tuple[CCUChannelInfo, List[CCUParamInfo]]: logger.debug( "Inspecting device channel '%s/%s'", device_ref.href, channel_ref.href ) async with self.http.get( f"/device/{device_ref.href}/{channel_ref.href}" ) as resp: channel_info = CCUChannelInfo.model_validate(await resp.json()) param_infos = await asyncio.gather( *[ self._inspect_ccu_param(device_ref, channel_ref, i) for i in channel_info.links if i.rel == "parameter" and i.href in PARAM_ADDRESSES_RELEVANT ] ) return channel_info, param_infos async def _inspect_ccu_param( self, device_ref: CCURef, channel_ref: CCURef, param_ref: CCURef ) -> CCUParamInfo: logger.debug( "Inspecting device parameter '%s/%s/%s'", device_ref.href, channel_ref.href, param_ref.href, ) async with self.http.get( f"/device/{device_ref.href}/{channel_ref.href}/{param_ref.href}" ) as resp: return CCUParamInfo.model_validate(await resp.json())