<commit_before>#     Copyright 2022. ThingsBoard
#
#     Licensed under the Apache License, Version 2.0 (the "License");
#     you may not use this file except in compliance with the License.
#     You may obtain a copy of the License at
#
#         http://www.apache.org/licenses/LICENSE-2.0
#
#     Unless required by applicable law or agreed to in writing, software
#     distributed under the License is distributed on an "AS IS" BASIS,
#     WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
#     See the License for the specific language governing permissions and
#     limitations under the License.

import asyncio
import re
from random import choice
from string import ascii_lowercase
from threading import Thread
from time import sleep, time
from queue import Queue

from thingsboard_gateway.connectors.connector import Connector, log
from thingsboard_gateway.connectors.opcua_asyncio.device import Device
from thingsboard_gateway.tb_utility.tb_loader import TBModuleLoader
from thingsboard_gateway.tb_utility.tb_utility import TBUtility

try:
    import asyncua
except ImportError:
    print("OPC-UA library not found")
    TBUtility.install_package("asyncua")
    import asyncua

from asyncua.crypto.security_policies import SecurityPolicyBasic256Sha256, SecurityPolicyBasic256, \
    SecurityPolicyBasic128Rsa15
from asyncua.ua.uaerrors import UaStatusCodeError, BadNodeIdUnknown, BadConnectionClosed, \
    BadInvalidState, BadSessionClosed

DEFAULT_UPLINK_CONVERTER = 'OpcUaUplinkConverter'

SECURITY_POLICIES = {
    "Basic128Rsa15": SecurityPolicyBasic128Rsa15,
    "Basic256": SecurityPolicyBasic256,
    "Basic256Sha256": SecurityPolicyBasic256Sha256,
}


class OpcUaConnectorAsyncIO(Connector, Thread):
    DATA_TO_SEND = Queue(-1)
    SUB_DATA_TO_CONVERT = Queue(-1)

    def __init__(self, gateway, config, connector_type):
        self.statistics = {'MessagesReceived': 0,
                           'MessagesSent': 0}
        self.__log = log
        super().__init__()
        self._connector_type = connector_type
        self.__gateway = gateway
        self.__server_conf = config['server']

        self.setName(
            self.__server_conf.get("name", 'OPC-UA Connector ' + ''.join(choice(ascii_lowercase) for _ in range(5))))

        if "opc.tcp" not in self.__server_conf.get("url"):
            self.__opcua_url = "opc.tcp://" + self.__server_conf.get("url")
        else:
            self.__opcua_url = self.__server_conf.get("url")

        self.__loop = asyncio.new_event_loop()

        self.__client = None

        self.__connected = False
        self.__stopped = False
        self.daemon = True

        self.__device_nodes = []
        self.__last_poll = 0

    def open(self):
        self.__stopped = False
        self.start()
        log.info("Starting OPC-UA Connector (Async IO)")

    def close(self):
        task = self.__loop.create_task(self.__reset_nodes())

        while not task.done():
            sleep(.2)

        self.__stopped = True
        self.__connected = False
        self.__log.info('%s has been stopped.', self.get_name())

    async def __reset_nodes(self, device_name=None):
        for device in self.__device_nodes:
            if device_name is None or device.name == device_name:
                for section in ('attributes', 'timeseries'):
                    for node in device.values.get(section, []):
                        node['valid'] = False
                        if node.get('sub_on', False):
                            try:
                                sub, handle = node['subscription']
                                await sub.unsubscribe(handle)
                                await sub.delete()
                            except:
                                pass
                            node['subscription'] = None
                            node['sub_on'] = False

    def get_name(self):
        return self.name

    def is_connected(self):
        return self.__connected

    def run(self):
        data_send_thread = Thread(name='Send Data Thread', target=self.__send_data, daemon=True)
        data_send_thread.start()

        if not self.__server_conf.get('disableSubscriptions', False):
            sub_data_convert_thread = Thread(name='Sub Data Convert Thread', target=self.__convert_sub_data,
                                             daemon=True)
            sub_data_convert_thread.start()

        self.__loop.run_until_complete(self.start_client())

    async def start_client(self):
        while not self.__stopped:
            self.__client = asyncua.Client(url=self.__opcua_url,
                                           timeout=self.__server_conf.get('timeoutInMillis', 4000) / 1000)

            if self.__server_conf.get("type") == "cert.PEM":<commit_msg>opcua_asyncio: fixes cert.PEM auth_settings call<commit_after>#     Copyright 2022. ThingsBoard
#
#     Licensed under the Apache License, Version 2.0 (the "License");
#     you may not use this file except in compliance with the License.
#     You may obtain a copy of the License at
#
#         http://www.apache.org/licenses/LICENSE-2.0
#
#     Unless required by applicable law or agreed to in writing, software
#     distributed under the License is distributed on an "AS IS" BASIS,
#     WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
#     See the License for the specific language governing permissions and
#     limitations under the License.

import asyncio
import re
from random import choice
from string import ascii_lowercase
from threading import Thread
from time import sleep, time
from queue import Queue

from thingsboard_gateway.connectors.connector import Connector, log
from thingsboard_gateway.connectors.opcua_asyncio.device import Device
from thingsboard_gateway.tb_utility.tb_loader import TBModuleLoader
from thingsboard_gateway.tb_utility.tb_utility import TBUtility

try:
    import asyncua
except ImportError:
    print("OPC-UA library not found")
    TBUtility.install_package("asyncua")
    import asyncua

from asyncua.crypto.security_policies import SecurityPolicyBasic256Sha256, SecurityPolicyBasic256, \
    SecurityPolicyBasic128Rsa15
from asyncua.ua.uaerrors import UaStatusCodeError, BadNodeIdUnknown, BadConnectionClosed, \
    BadInvalidState, BadSessionClosed

DEFAULT_UPLINK_CONVERTER = 'OpcUaUplinkConverter'

SECURITY_POLICIES = {
    "Basic128Rsa15": SecurityPolicyBasic128Rsa15,
    "Basic256": SecurityPolicyBasic256,
    "Basic256Sha256": SecurityPolicyBasic256Sha256,
}


class OpcUaConnectorAsyncIO(Connector, Thread):
    DATA_TO_SEND = Queue(-1)
    SUB_DATA_TO_CONVERT = Queue(-1)

    def __init__(self, gateway, config, connector_type):
        self.statistics = {'MessagesReceived': 0,
                           'MessagesSent': 0}
        self.__log = log
        super().__init__()
        self._connector_type = connector_type
        self.__gateway = gateway
        self.__server_conf = config['server']

        self.setName(
            self.__server_conf.get("name", 'OPC-UA Connector ' + ''.join(choice(ascii_lowercase) for _ in range(5))))

        if "opc.tcp" not in self.__server_conf.get("url"):
            self.__opcua_url = "opc.tcp://" + self.__server_conf.get("url")
        else:
            self.__opcua_url = self.__server_conf.get("url")

        self.__loop = asyncio.new_event_loop()

        self.__client = None

        self.__connected = False
        self.__stopped = False
        self.daemon = True

        self.__device_nodes = []
        self.__last_poll = 0

    def open(self):
        self.__stopped = False
        self.start()
        log.info("Starting OPC-UA Connector (Async IO)")

    def close(self):
        task = self.__loop.create_task(self.__reset_nodes())

        while not task.done():
            sleep(.2)

        self.__stopped = True
        self.__connected = False
        self.__log.info('%s has been stopped.', self.get_name())

    async def __reset_nodes(self, device_name=None):
        for device in self.__device_nodes:
            if device_name is None or device.name == device_name:
                for section in ('attributes', 'timeseries'):
                    for node in device.values.get(section, []):
                        node['valid'] = False
                        if node.get('sub_on', False):
                            try:
                                sub, handle = node['subscription']
                                await sub.unsubscribe(handle)
                                await sub.delete()
                            except:
                                pass
                            node['subscription'] = None
                            node['sub_on'] = False

    def get_name(self):
        return self.name

    def is_connected(self):
        return self.__connected

    def run(self):
        data_send_thread = Thread(name='Send Data Thread', target=self.__send_data, daemon=True)
        data_send_thread.start()

        if not self.__server_conf.get('disableSubscriptions', False):
            sub_data_convert_thread = Thread(name='Sub Data Convert Thread', target=self.__convert_sub_data,
                                             daemon=True)
            sub_data_convert_thread.start()

        self.__loop.run_until_complete(self.start_client())

    async def start_client(self):
        while not self.__stopped:
            self.__client = asyncua.Client(url=self.__opcua_url,
                                           timeout=self.__server_conf.get('timeoutInMillis', 4000) / 1000)

            if self.__server_conf["identity"].get("type") == "cert.PEM":
{'fraction_non_alphanumeric': 0.08666047663262148, 'fraction_numerical': 0.013411740431239038, 'mean_word_length': 2.358974358974359, 'pattern_counts': {'":': 8, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 8, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3435393', 'n_tokens_mistral': 2763, 'n_tokens_neox': 2627, 'n_words': 665}