Как получить свойства тегов (адрес, описание) из kepware?

Question

У меня проблема с автоматизацией конвейера приема. Я читаю все теги с сервера kepware и принимаю их с помощью nifi. Я использую клиент opcua, чтобы посмотреть, есть ли тег, который был добавлен на сервер kepware, а также чтобы получить описание и адрес этого тега. Но этот путь очень медленный и заставляет сервер kepware перестать работать.

Есть ли способ обнаружить добавленные теги и свойства (адрес, описание) на сервере kepware без использования метода опроса, который считывает данные с сервера каждые 10 секунд?

Я использую следующее, но это не так эффективно, так как мы читаем до 200000 тегов из kepware: /

from opcua import Client, ua
from opcua.ua import status_codes
from threading import Thread
import threading
import logging
import time
import csv
import re

NUM_SUBSCRIPTIONS = 10


output_files_name_machines_tags = []
for i in range(NUM_SUBSCRIPTIONS):
    output_file = "../resources/machine_tags_"+str(i)+".txt"
    output_files_name_machines_tags.append(output_file)


output_file_name_special_tags = "../resources/advanced_Tags.txt"
tags_description = '../resources/tags_description.csv'

special_tag = "ns=2;s=_AdvancedTags"
Infos = ['Address','Description']


#logging.basicConfig(level=logging.WARN)
client = Client("opc.tcp://10.129.140.215:49320/",timeout=3600000)


class Industrial_Tags_Updater():
    list_machine_tags_changed = [False]*NUM_SUBSCRIPTIONS
    list_special_tags_changed = False
    list_machine_tags = [[] for i in range(NUM_SUBSCRIPTIONS)]
    list_special_tags = []
    output_description_tags = []
    output_description_tags_changed = False
    distribute_integer = 0
    nodes_to_be_described = []


    def process_tag_nodes(self, node_child_level_1):
        nodes_children_level_2 = node_child_level_1.get_children()

        #print(nodes_children_level_2)
        for node_child_level_2 in nodes_children_level_2:
            if not "._" in get_node_string(node_child_level_2):
                #print(node_child_level_2)
                nodes_children_level_3 = node_child_level_2.get_children()
                for node_child_level_3 in nodes_children_level_3:
                    tag_name = get_node_string(node_child_level_3)
                    if special_tag in tag_name:
                        nodes_children_level_4 = node_child_level_3.get_children()
                        for node_child_level_4 in nodes_children_level_4:
                            #print(get_node_string(node_child_level_4))
                            if not get_node_string(node_child_level_4) in self.list_special_tags:
                                self.list_special_tags_changed = True
                                self.list_special_tags.append(get_node_string(node_child_level_4))
                                self.output_description_tags.append([get_node_string(node_child_level_4)])

                    elif not "._" in tag_name:
                        if not any(tag_name in list for list in self.list_machine_tags):
                            for info in Infos:
                                self.nodes_to_be_described.append(tag_name+"._"+info)


        params = ua.ReadParameters()
        if self.nodes_to_be_described:
            for node_id_str in self.nodes_to_be_described:
                nodeid = ua.NodeId.from_string(node_id_str)
                attr = ua.ReadValueId()
                attr.NodeId = nodeid
                attr.AttributeId = ua.AttributeIds.Value
                params.NodesToRead.append(attr)

            results = client.uaclient.read(params)


            for i in range(0,len(results),len(Infos)):
                tag_str = params.NodesToRead[i].NodeId.to_string().replace('._'+Infos[0],'')
                if results[i].StatusCode.name != 'BadNodeIdUnknown':
                    tag_description = []
                    tag_description.append(tag_str)
                    for j in range(len(Infos)):
                        tag_description.append(results[i+j].Value.Value)
                    if tag_description not in self.output_description_tags:
                        self.output_description_tags.append(tag_description)
                        self.output_description_tags_changed = True
                        #print(tag_description)

                    if not any(tag_str in list for list in self.list_machine_tags):
                        self.list_machine_tags[int(self.distribute_integer/10)%NUM_SUBSCRIPTIONS].append(tag_str)
                        self.list_machine_tags_changed[int(self.distribute_integer/10)%NUM_SUBSCRIPTIONS] = True
                        self.distribute_integer += 1

                else :
                    node_child_level_3 = client.get_node(tag_str)   
                    nodes_children_level_4 = node_child_level_3.get_children()
                    for node_child_level_4 in nodes_children_level_4:
                        node_string = get_node_string(node_child_level_4)
                        if not any(node_string in list for list in self.list_machine_tags):
                            self.list_machine_tags[int(self.distribute_integer/10)%NUM_SUBSCRIPTIONS].append(node_string)
                            self.list_machine_tags_changed[int(self.distribute_integer/10)%NUM_SUBSCRIPTIONS] = True
                            self.output_description_tags.append([node_string])
                            self.distribute_integer += 1



    #Write the list of tags to be consumed by nifi processor
    def write_machine_tags_file(self, file, num_subscription):
        output_file = open(file, "w")
        output_file.write("\n".join(self.list_machine_tags[num_subscription]))
        output_file.close()
        print("List machine tags file updated pour la soubscription"+str(num_subscription))

    #Write the list of tags to be consumed by nifi processor
    def write_special_tags_file(self, ):
        output_file = open(output_file_name_special_tags, "w")
        output_file.write("\n".join(self.list_special_tags))
        output_file.close()
        print("List special tags file updated")

    def write_description_file(self, ):
        #write to a csv file the description of the tags
        infos_data = self.output_description_tags
        for i in range(len(infos_data)):
             infos_data[i] = (infos_data[i] + (len(Infos)+1) * [""])[:len(Infos)+1]

        fieldnames = ["Tag_name"]
        fieldnames.extend(Infos)
        infos_data.insert(0,fieldnames)

        with open(tags_description, 'w') as csvFile:
            writer = csv.writer(csvFile)
            writer.writerows(infos_data)
        print("Description tags file updated")
        csvFile.close()

    def reset_update_description_change(self, ):
        self.output_description_tags_changed = False
        self.output_description_tags = []

    def reset_update_machine_change(self, ):
        self.list_machine_tags_changed = [False]*NUM_SUBSCRIPTIONS

    def reset_update_special_change(self, ):
        self.list_special_tags_changed = False



if __name__ == "__main__":

    def get_node_string(node):
        return node.nodeid.to_string()

    try:
        client.connect()
        objects = client.get_objects_node()
        nodes_children_level_1 = objects.get_children()
        #print(nodes_children_level_1)
        nodes_level_1_filtred = []
        nodes_level_1_filtred.append(client.get_node(special_tag))
        for node_child_level_1 in nodes_children_level_1:
            pattern = r'ns=2;s=[^_]'
            if re.match(pattern, get_node_string(node_child_level_1)):
                nodes_level_1_filtred.append(node_child_level_1)

        tags_updater = Industrial_Tags_Updater()

        #print(nodes_level_1_filtred)
        while True:
            #Creation of threads
            threads_list = []
            for node_child_level_1 in nodes_level_1_filtred:
                threads_list.append(Thread(target=tags_updater.process_tag_nodes, args=(node_child_level_1, )))

            #launching threads
            for thread in threads_list:
                thread.start()
            #waiting the threds to be finished
            for thread in threads_list:
                thread.join()

            #print("finished")

            if tags_updater.list_special_tags_changed or True in tags_updater.list_machine_tags_changed or tags_updater.output_description_tags_changed:
                tags_updater.write_description_file()
                tags_updater.reset_update_description_change()

            for i in range(NUM_SUBSCRIPTIONS):
                if tags_updater.list_machine_tags_changed[i]:
                    tags_updater.write_machine_tags_file(output_files_name_machines_tags[i], i)

            tags_updater.reset_update_machine_change()

            if tags_updater.list_special_tags_changed:
                tags_updater.write_special_tags_file()
                tags_updater.reset_update_special_change()


            #Schedule the the threads to run evry minutes
            time.sleep(10)

    finally:
        client.disconnect()

Можем ли мы добавить события в kepware, чтобы сообщить клиенту, что тег был добавлен, или использовать API kepware, где люди добавляют тег в этот API ?? !!! Я не знаю, есть ли решение для решения этой проблемы, пожалуйста, помогите !!