Merge branch 'master' into tests/write-unit-tests-rpc-attribute_updates-bacnet-connector

Author: Andry925

requirements.txt

@@ -22,3 +22,4 @@ tb-mqtt-client==1.13.9
 service-identity
 psutil
 cryptography
+PySocks

tests/requirements.txt

@@ -22,7 +22,7 @@ pyjwt==2.6.0
 tb-rest-client
 pyopenssl
 opcua
-asyncua
+asyncua==1.1.5
 twisted
 pymodbus==3.9.2
 pyserial

tests/unit/service/test_storage.py

@@ -146,14 +146,20 @@ class TestSQLiteEventStorageRotation(TestCase):
     def setUp(self):
         self.directory = path.join("storage", "data")
         self.db_path = path.join(self.directory, "data.db")
+        if not path.exists(self.directory):
+            try:
+                from os import makedirs
+                makedirs(self.directory, exist_ok=True)
+            except Exception as e:
+                LOG.warning(f"Failed to create directory {self.directory}: {e}")
         self.config = {
             "data_file_path": self.db_path,
             "messages_ttl_check_in_hours": 1,
             "messages_ttl_in_days": 7,
             "max_read_records_count": 1000,
-            "size_limit": 0.025,
+            "size_limit": 0.05,
             "max_db_amount": 3,
-            "oversize_check_period": 1 / 60,
+            "oversize_check_period": 1 / 20,
             "writing_batch_size": 1000,
         }
         self.settings = StorageSettings(self.config, enable_validation=False)
@@ -163,8 +169,9 @@ def setUp(self):
     def tearDown(self):
         self.stop_event.set()
         self.sqlite_storage.stop()
-        rmtree(self.directory)
-        sleep(2)
+        sleep(1)
+        rmtree(self.directory, ignore_errors=True)
+        sleep(1)
 
     def _drain_storage(self, storage: SQLiteEventStorage):
         out = []
@@ -187,16 +194,16 @@ def _fill_storage(self, storage, count, delay=0.0):
                 sleep(delay)
 
     def _db_files(self):
-        return sorted(f for f in listdir(self.directory) if f.endswith(".db"))
+        try:
+            return sorted(f for f in listdir(self.directory)
+                         if f.endswith(".db") and not (f.endswith(".db-shm") or f.endswith(".db-wal")))
+        except FileNotFoundError:
+            return []
 
     def test_write_read_without_rotation(self):
         self._fill_storage(self.sqlite_storage, 20)
-        fat_msg = "X" * 32768
         sleep(1)
         self.assertListEqual(self._db_files(), ["data.db"])
-        all_messages = self._drain_storage(self.sqlite_storage)
-        self.assertEqual(len(all_messages), 20)
-        self.assertListEqual(all_messages, [f"{i}:{fat_msg}" for i in range(20)])
 
         self.sqlite_storage.stop()
 
@@ -205,21 +212,28 @@ def test_rotation_creates_new_db(self):
         self._fill_storage(self.sqlite_storage, DATA_RANGE, delay=0.1)
         sleep(2.0)
         dbs = self._db_files()
-        self.assertEqual(len(dbs), 2)
+        self.assertEqual(len(dbs), 2, f"Expected 2 database files after rotation, got {len(dbs)}: {dbs}")
         self.assertLessEqual(len(dbs), self.config["max_db_amount"])
-
         self.sqlite_storage.stop()
+        sleep(0.5)
 
     def test_rotation_persists_across_restart(self):
         DATA_RANGE = 150
-        fat_msg = "X" * 32768
         self._fill_storage(self.sqlite_storage, DATA_RANGE, delay=0.1)
         sleep(2.0)
+        dbs_before = self._db_files()
+        self.assertGreater(len(dbs_before), 1, "Expected rotation to create at least one new database file")
         self.sqlite_storage.stop()
         storage2 = SQLiteEventStorage(self.settings, LOG, self.stop_event)
-        all_messages = self._drain_storage(storage2)
-        self.assertEqual(len(all_messages), DATA_RANGE)
-        self.assertListEqual(all_messages, [f"{i}:{fat_msg}" for i in range(DATA_RANGE)])
+        dbs_after = self._db_files()
+        self.assertEqual(len(dbs_before), len(dbs_after),
+                         f"Database file count changed after restart: {len(dbs_before)} -> {len(dbs_after)}")
+        self.assertListEqual(dbs_before, dbs_after,
+                             "Database files changed after restart")
+        messages = self._drain_storage(storage2)
+        self.assertEqual(len(messages), DATA_RANGE,
+                         f"Expected to read {DATA_RANGE} messages, got {len(messages)}")
+
         storage2.stop()
 
     def test_no_new_database_appear_after_max_db_amount_reached(self):
@@ -246,11 +260,6 @@ def test_no_new_database_appear_after_max_db_amount_reached(self):
             put_results,
             "Expected self.sqlite_storage.put(...) to eventually return False once max_db_amount was reached",
         )
-        all_messages = self._drain_storage(self.sqlite_storage)
-        self.assertEqual(len(all_messages), len(messages_before_db_amount_reached))
-        self.assertListEqual(
-            all_messages, messages_before_db_amount_reached
-        )
         self.sqlite_storage.stop()
 
     def test_sqlite_storage_is_operational_after_max_db_amount_reached_and_storage_restart(
@@ -281,9 +290,4 @@ def test_sqlite_storage_is_operational_after_max_db_amount_reached_and_storage_r
             put_results,
             "Expected storage.put(...) eventually to return False once max_db_amount was reached",
         )
-        all_messages = list(self._drain_storage(storage2))
-        self.assertEqual(len(all_messages), len(messages_before_db_amount_reached))
-        self.assertListEqual(
-            all_messages, messages_before_db_amount_reached
-        )
-        storage2.stop()
+        storage2.stop()

thingsboard_gateway/connectors/bacnet/bacnet_connector.py

@@ -16,6 +16,7 @@
 import asyncio
 from asyncio import Queue, CancelledError, QueueEmpty
 from copy import deepcopy
+from datetime import time
 from threading import Thread
 from string import ascii_lowercase
 from random import choice
@@ -24,6 +25,8 @@
 from concurrent.futures import TimeoutError
 
 from thingsboard_gateway.connectors.bacnet.constants import SUPPORTED_OBJECTS_TYPES
+from typing import TYPE_CHECKING
+from ast import literal_eval
 
 from thingsboard_gateway.connectors.bacnet.ede_parser import EDEParser
 from thingsboard_gateway.connectors.bacnet.entities.routers import Routers
@@ -42,11 +45,12 @@
     from bacpypes3.apdu import ErrorRejectAbortNack
 
 from bacpypes3.pdu import Address, IPv4Address
+from bacpypes3.primitivedata import Null, Real
+from bacpypes3.basetypes import DailySchedule, TimeValue, DeviceObjectPropertyReference
 from thingsboard_gateway.connectors.bacnet.device import Device, Devices
 from thingsboard_gateway.connectors.bacnet.entities.device_object_config import DeviceObjectConfig
 from thingsboard_gateway.connectors.bacnet.application import Application
 from thingsboard_gateway.connectors.bacnet.backward_compatibility_adapter import BackwardCompatibilityAdapter
-from bacpypes3.primitivedata import Null
 
 if TYPE_CHECKING:
     from thingsboard_gateway.gateway.tb_gateway_service import TBGatewayService
@@ -548,6 +552,12 @@ async def __write_property(self, address, object_id, property_id, value, priorit
             if value is None:
                 value = Null(())
 
+            if property_id == "weeklySchedule":
+                value = self.__prepare_weekly_schedule_value(value)
+
+            if property_id == "listOfObjectPropertyReferences":
+               value = self.__prepare_list_of_object_property_references_value(value, property_id)
+
             await self.__application.write_property(address, object_id, property_id, value, priority=priority)
             result['value'] = value
             return result
@@ -561,6 +571,32 @@ async def __write_property(self, address, object_id, property_id, value, priorit
             self.__log.error('Error writing property %s:%s to device %s: %s', object_id, property_id, address, err)
             return {'error': str(err)}
 
+    async def __prepare_weekly_schedule_value(self, value):
+        schedule = []
+        raw_schedule = literal_eval(value)
+        for idx, day in enumerate(raw_schedule):
+            schedule.append(DailySchedule(daySchedule=[]))
+            for sched in day:
+                schedule[idx].daySchedule.append(
+                    TimeValue(
+                        time=time(int(sched[0].split(":")[0]), int(sched[0].split(":")[1])),
+                        value=Real(sched[1])
+                    )
+                )
+        return schedule
+
+    async def __prepare_list_of_object_property_references_value(self, value, property_id):
+        props = []
+        raw_props = literal_eval(value)
+        for prop in raw_props:
+            props.append(
+                DeviceObjectPropertyReference(
+                    objectIdentifier = prop,
+                    propertyIdentifier = property_id
+                )
+            )
+        return props
+
     async def __convert_data(self):
         while not self.__stopped:
             try:

thingsboard_gateway/connectors/bacnet/bacnet_uplink_converter.py

@@ -15,7 +15,7 @@
 
 from bacpypes3.basetypes import DateTime
 from bacpypes3.constructeddata import AnyAtomic, Array
-from bacpypes3.basetypes import ErrorType, PriorityValue, ObjectPropertyReference
+from bacpypes3.basetypes import ErrorType, PriorityValue, ObjectPropertyReference, DailySchedule, DeviceObjectPropertyReference
 
 from thingsboard_gateway.connectors.bacnet.bacnet_converter import AsyncBACnetConverter
 from thingsboard_gateway.connectors.bacnet.entities.uplink_converter_config import UplinkConverterConfig
@@ -109,9 +109,19 @@ def __convert_data(self, data):
                     for item in value:
                         if isinstance(item, PriorityValue):
                             result.append(str(getattr(item, item._choice)))
+                        if isinstance(item, DailySchedule):
+                            schedular_array = []
+                            for sched in item.daySchedule:
+                                schedular_array.append((str(sched.time), str(sched.value.get_value())))
+                            result.append(schedular_array)
                         else:
                             result.append(item)
-
+                    value = result
+                elif isinstance(value, list):
+                    result = []
+                    for item in value:
+                        if isinstance(item, DeviceObjectPropertyReference):
+                            result.append(str(item.objectIdentifier))
                     value = result
 
                 converted_data.append({'propName': str(value_prop_id), 'value': value})

thingsboard_gateway/connectors/opcua/opcua_connector.py

@@ -27,9 +27,8 @@
 from cachetools import TTLCache
 
 from thingsboard_gateway.connectors.connector import Connector
-from thingsboard_gateway.connectors.opcua.entities.rpc_request import OpcUaRpcRequest, OpcUaRpcType
 from thingsboard_gateway.gateway.constants import CONNECTOR_PARAMETER, RECEIVED_TS_PARAMETER, CONVERTED_TS_PARAMETER, \
-    DATA_RETRIEVING_STARTED, REPORT_STRATEGY_PARAMETER, RPC_DEFAULT_TIMEOUT, ON_ATTRIBUTE_UPDATE_DEFAULT_TIMEOUT
+    DATA_RETRIEVING_STARTED, REPORT_STRATEGY_PARAMETER, ON_ATTRIBUTE_UPDATE_DEFAULT_TIMEOUT
 from thingsboard_gateway.gateway.entities.converted_data import ConvertedData
 from thingsboard_gateway.gateway.entities.report_strategy_config import ReportStrategyConfig
 from thingsboard_gateway.gateway.statistics.statistics_service import StatisticsService
@@ -38,13 +37,30 @@
 from thingsboard_gateway.tb_utility.tb_logger import init_logger
 from thingsboard_gateway.tb_utility.tb_utility import TBUtility
 
+# Try import asyncua library or install it and import
+installation_required = False
+required_version = '1.1.5'
+force_install = False
+
+from packaging import version
 try:
-    import asyncua
-except (ImportError, ModuleNotFoundError):
+    from asyncua import __version__ as asyncua_version
+
+    if version.parse(asyncua_version) < version.parse(required_version):
+        installation_required = True
+
+    if version.parse(asyncua_version) > version.parse(required_version):
+        installation_required = True
+        force_install = True
+
+except ImportError:
+    installation_required = True
+
+if installation_required:
     print("OPC-UA library not found")
-    TBUtility.install_package("asyncua")
-    import asyncua
+    TBUtility.install_package("asyncua", required_version, force_install=force_install)
 
+import asyncua
 from asyncua import ua, Node
 from asyncua.ua import NodeId, UaStringParsingError
 from asyncua.common.ua_utils import value_to_datavalue
@@ -54,6 +70,7 @@
 from asyncua.ua.uaerrors import UaStatusCodeError, BadNodeIdUnknown, BadConnectionClosed, \
     BadInvalidState, BadSessionClosed, BadAttributeIdInvalid, BadCommunicationError, BadOutOfService, BadNoMatch, \
     BadUnexpectedError, UaStatusCodeErrors, BadWaitingForInitialData, BadSessionIdInvalid, BadSubscriptionIdInvalid
+from thingsboard_gateway.connectors.opcua.entities.rpc_request import OpcUaRpcRequest, OpcUaRpcType
 from thingsboard_gateway.connectors.opcua.device import Device
 from thingsboard_gateway.connectors.opcua.backward_compatibility_adapter import BackwardCompatibilityAdapter
 
@@ -448,7 +465,7 @@ async def retry_connect_with_backoff(self, max_retries=8, initial_delay=1, backo
             except Exception as e:
                 base_time = self.__client.session_timeout / 1000 if (
                         last_contact_delta > 0 and last_contact_delta < self.__client.session_timeout / 1000) else 0
-                time_to_wait = base_time / 1000 + delay
+                time_to_wait = base_time + delay
                 time_to_wait = min(self.__server_conf.get('sessionTimeoutInMillis', 120000) / 1000 * 0.8, time_to_wait)
                 self.__log.error('Encountered error: %r. Next connection try in %i second(s)...', e, time_to_wait)
                 await asyncio.sleep(time_to_wait)

thingsboard_gateway/gateway/constants.py

@@ -149,4 +149,8 @@ def from_string(cls, value: str):
     'enable': False
 }
 
-CUSTOM_RPC_DIR = "/etc/thingsboard-gateway/rpc"
+CUSTOM_RPC_DIR = "/etc/thingsboard-gateway/rpc"
+
+# Provisioning constants
+
+PROVISIONED_CREDENTIALS_FILENAME = ".credentials"