v1.5.1b1 resubscribing to CoV subscriptions

bacnetinterface/CHANGELOG.md

@@ -3,9 +3,6 @@
 # 1.5.0
 13/08/2024
 
-If there are any issues, please report on GitHub!
-Changed from v1.4.1 to 1.5.0 as there's a lot of changes!
-
 ## Added
 - Reading resolution property now. Integration will use it once it's updated as well. [#46](https://github.com/Bepacom-Raalte/bepacom-HA-Addons/issues/46)
 - Added devices_setup configuration option to allow the user to configure behaviour. See Documentation for usage. [#43](https://github.com/Bepacom-Raalte/bepacom-HA-Addons/discussions/43)

bacnetinterface_dev/CHANGELOG.md

@@ -1,11 +1,21 @@
 <!-- https://developers.home-assistant.io/docs/add-ons/presentation#keeping-a-changelog -->
 
+# 1.5.1b1
+18/08/2024
+
+## Fixed
+- Fixed monitoring of CoV subscriptions failing due to loss of connection. CoV tasks will now get removed after trying to resub without response when the CoV lifetime has passed.
+- If address changed since last I Am request, it'll get updated internally now.
+
+## Added
+- Added resubscribing CoV to an object after the subscription had been timed out due to no response. [#52](https://github.com/Bepacom-Raalte/bepacom-HA-Addons/issues/52)
+
+## Dependencies
+- ⬆️ Bumped base-python image to version v14.0.1.
 
-# 1.5.0b11
-10/08/2024
 
-If there are any issues, please report on GitHub!
-Changed from v1.4.1 to 1.5.0 as there's a lot of changes!
+# 1.5.0
+13/08/2024
 
 ## Added
 - Reading resolution property now. Integration will use it once it's updated as well. [#46](https://github.com/Bepacom-Raalte/bepacom-HA-Addons/issues/46)

bacnetinterface_dev/build.yaml

@@ -1,7 +1,7 @@
 ---
 build_from:
-  aarch64: ghcr.io/hassio-addons/base-python:14.0.0
-  armhf: ghcr.io/hassio-addons/base-python:14.0.0
-  armv7: ghcr.io/hassio-addons/base-python:14.0.0
-  amd64: ghcr.io/hassio-addons/base-python:14.0.0
-  i386: ghcr.io/hassio-addons/base-python:14.0.0
+  aarch64: ghcr.io/hassio-addons/base-python:14.0.1
+  armhf: ghcr.io/hassio-addons/base-python:14.0.1
+  armv7: ghcr.io/hassio-addons/base-python:14.0.1
+  amd64: ghcr.io/hassio-addons/base-python:14.0.1
+  i386: ghcr.io/hassio-addons/base-python:14.0.1

bacnetinterface_dev/config.yaml

@@ -1,6 +1,6 @@
 # https://developers.home-assistant.io/docs/add-ons/configuration#add-on-config
 name: Bepacom BACnet/IP Interface Development Version
-version: "1.5.0b11"
+version: "1.5.1b1"
 slug: bacnetinterface_dev
 description: Bepacom BACnet/IP interface for the Bepacom EcoPanel. Allows BACnet/IP devices to be available to Home Assistant through an API
 url: "https://github.com/Bepacom-Raalte/bepacom-HA-Addons/tree/main/bacnetinterface"

bacnetinterface_dev/rootfs/usr/bin/BACnetIOHandler.py

@@ -1,7 +1,9 @@
 """BACnet handler classes for BACnet add-on."""
+
 import asyncio
 import json
 from ast import List
+from logging import config
 from math import e, isinf, isnan
 from re import A
 from typing import Any, Dict, TypeVar
@@ -29,6 +31,7 @@
 from bacpypes3.object import CharacterStringValueObject, get_vendor_info
 from bacpypes3.pdu import Address
 from bacpypes3.primitivedata import ObjectIdentifier, ObjectType, OctetString
+from bacpypes3.service.cov import SubscriptionContextManager
 from const import (LOGGER, device_properties_to_read,
                    object_properties_to_read_once,
                    object_properties_to_read_periodically,
@@ -38,6 +41,34 @@
 _debug = 0
 
 
+def custom_init(
+    self,
+    app: "Application",  # noqa: F821
+    address: Address,
+    monitored_object_identifier: ObjectIdentifier,
+    subscriber_process_identifier: int,
+    issue_confirmed_notifications: bool,
+    lifetime: int,
+):
+    original_init(
+        self,
+        app,
+        address,
+        monitored_object_identifier,
+        subscriber_process_identifier,
+        issue_confirmed_notifications,
+        lifetime,
+    )
+
+    # result of refresh task to check if exception occurred
+    self.refresh_subscription_task = None
+
+
+original_init = SubscriptionContextManager.__init__
+
+SubscriptionContextManager.__init__ = custom_init
+
+
 class BACnetIOHandler(NormalApplication, ForeignApplication):
     bacnet_device_dict: dict = {}
     subscription_tasks: list = []
@@ -422,9 +453,11 @@ def deep_update(
         return mapping
 
     def dev_to_addr(self, dev: ObjectIdentifier) -> Address | None:
-        for instance in self.device_info_cache.instance_cache:
-            if instance == dev[1]:
-                return self.device_info_cache.instance_cache[instance].address
+
+        for address, device_info in self.device_info_cache.address_cache.items():
+            if device_info.device_instance == dev[1]:
+                return address
+
         return None
 
     def addr_to_dev(self, addr: Address) -> ObjectIdentifier | None:
@@ -479,6 +512,7 @@ async def do_IAmRequest(self, apdu) -> None:
 
         if device_id in self.device_info_cache.instance_cache:
             LOGGER.debug(f"Device {apdu.iAmDeviceIdentifier} already in cache!")
+            await self.device_info_cache.set_device_info(apdu)
             in_cache = True
         else:
             await self.device_info_cache.set_device_info(apdu)
@@ -488,6 +522,111 @@ async def do_IAmRequest(self, apdu) -> None:
 
         if not in_cache:
             await self.i_am_queue.put(apdu)
+        else:
+            # Check if object list is still the same, otherwise read entire dict again
+            await self.handle_object_list_check(apdu)
+
+            # Check if CoV tasks are still active, otherwise resub.
+            await self.handle_cov_check(apdu.iAmDeviceIdentifier)
+
+    async def handle_object_list_check(self, apdu) -> None:
+
+        device_id = apdu.iAmDeviceIdentifier[1]
+
+        object_list = self.bacnet_device_dict[f"device:{apdu.iAmDeviceIdentifier[1]}"][
+            f"device:{apdu.iAmDeviceIdentifier[1]}"
+        ].get("objectList")
+
+        if not await self.read_multiple_device_props(apdu=apdu):
+            LOGGER.warning(f"Failed to get: {device_id}, {device_id}")
+            if self.bacnet_device_dict.get(f"device:{device_id}"):
+                self.bacnet_device_dict.pop(f"device:{device_id}")
+
+        if object_list != self.bacnet_device_dict[
+            f"device:{apdu.iAmDeviceIdentifier[1]}"
+        ][f"device:{apdu.iAmDeviceIdentifier[1]}"].get("objectList"):
+            LOGGER.warning(f"Not implemented yet: object lists aren't equal!")
+
+    def identifier_to_string(self, object_identifier) -> str:
+        return f"{object_identifier[0].attr}:{object_identifier[1]}"
+
+    def task_in_tasklist(self, task_name) -> bool:
+        return any(task_name in task.get_name() for task in self.subscription_tasks)
+
+    async def handle_cov_check(self, device_identifier) -> None:
+
+        device_string = self.identifier_to_string(device_identifier)
+
+        if self.addon_device_config is None:
+            return
+
+        specific_config = [
+            config
+            for config in self.addon_device_config
+            if config.get("deviceID")
+            == f"{device_identifier[0]}:{device_identifier[1]}"
+        ]
+
+        if not specific_config:
+            specific_config = [
+                config
+                for config in self.addon_device_config
+                if config.get("deviceID") == "all"
+            ]
+            if not specific_config:
+                return
+
+        index = self.addon_device_config.index(specific_config[0])
+
+        config = self.addon_device_config[index]
+
+        if "all" in config.get("CoV_list", []):
+            object_list = self.bacnet_device_dict[f"device:{device_identifier[1]}"][
+                f"device:{device_identifier[1]}"
+            ].get("objectList")
+
+            if device_identifier in object_list:
+                object_list.remove(device_identifier)
+
+            object_list = [
+                object_identifier
+                for object_identifier in object_list
+                if object_identifier[0] in subscribable_objects
+            ]
+
+            for object_identifier in object_list:
+
+                task_name = f"{self.identifier_to_string(device_identifier)},{self.identifier_to_string(object_identifier)},confirmed"
+
+                if self.task_in_tasklist(task_name):
+                    continue
+
+                await self.create_subscription_task(
+                    device_identifier=device_identifier,
+                    object_identifier=object_identifier,
+                    confirmed_notifications=True,
+                    lifetime=config.get(
+                        "CoV_lifetime", self.default_subscription_lifetime
+                    ),
+                )
+                await asyncio.sleep(0)
+
+        elif config.get("CoV_list", []):
+
+            for object_identifier in config.get("CoV_list"):
+
+                task_name = f"{self.identifier_to_string(device_identifier)},{self.identifier_to_string(object_identifier)},confirmed"
+
+                if self.task_in_tasklist(task_name):
+                    continue
+
+                await self.create_subscription_task(
+                    device_identifier=device_identifier,
+                    object_identifier=object_identifier,
+                    confirmed_notifications=True,
+                    lifetime=config.get("CoV_lifetime"),
+                )
+                await asyncio.sleep(0)
 
     async def IAm_handler(self):
         """Do the things when receiving I Am requests"""
@@ -1094,7 +1233,45 @@ async def subscription_task(
                 LOGGER.debug(f"Created {task_name} subscription task successfully")
 
                 while True:
-                    property_identifier, property_value = await subscription.get_value()
+                    try:
+                        property_identifier, property_value = await asyncio.wait_for(
+                            subscription.get_value(), 10
+                        )
+                    except asyncio.TimeoutError:
+                        # check if address has changes
+                        if subscription.address != self.dev_to_addr(
+                            dev=device_identifier
+                        ):
+                            old_key = (
+                                subscription.address,
+                                subscription.subscriber_process_identifier,
+                            )
+                            self._cov_contexts.pop(old_key)
+
+                            subscription.address = self.dev_to_addr(
+                                dev=device_identifier
+                            )
+
+                            new_key = (
+                                subscription.address,
+                                subscription.subscriber_process_identifier,
+                            )
+
+                            self._cov_contexts[new_key] = subscription
+
+                        if not isinstance(
+                            subscription.refresh_subscription_task, asyncio.Task
+                        ):
+                            continue
+
+                        if subscription.refresh_subscription_task.done():
+                            # check for exceptions (gets raised by result if there is)
+                            subscription.refresh_subscription_task.result()
+
+                        continue
+
+                    except Exception:
+                        raise
 
                     object_class = self.vendor_info.get_object_class(
                         subscription.monitored_object_identifier[0]

bacnetinterface_dev/rootfs/usr/bin/main.py

@@ -28,14 +28,6 @@
 KeyType = TypeVar("KeyType")
 
 
-def exception_handler(loop, context):
-    """Handle uncaught exceptions"""
-    try:
-        LOGGER.exception(f'An uncaught error occurred: {context["exception"]}')
-    except:
-        LOGGER.error("Tried to log error, but something went horribly wrong!!!")
-
-
 def exception_handler(loop, context):
     """Handle uncaught exceptions"""
     try:

bacnetinterface_dev/rootfs/usr/bin/webAPI.py

@@ -1,4 +1,5 @@
 """API script for BACnet add-on."""
+
 import asyncio
 import codecs
 import csv
@@ -633,13 +634,12 @@ async def write_property(
     deviceid: str = Path(description="device:instance"),
     objectid: str = Path(description="object:instance"),
     property: str = Path(description="property, for example presentValue"),
-    value: str
-    | int
-    | float
-    | bool
-    | None = Query(default=None, description="Property value"),
-    array_index: int
-    | None = Query(default=None, description="Array index, usually left empty"),
+    value: str | int | float | bool | None = Query(
+        default=None, description="Property value"
+    ),
+    array_index: int | None = Query(
+        default=None, description="Array index, usually left empty"
+    ),
     priority: int | None = Query(default=None, description="Write priority"),
 ):
     """Write to a property of an object from a device."""