0

我正在运行一段代码来连接到 Cisco 设备，建立 NETCONF 遥测会话并使用定期返回的数据。不过，我有点卡住和困惑，我可以建立一个 NETCONF 会话，建立一个带有回调的订阅，但是一旦数据进入回调，除了将它打印到屏幕。

我希望能够解析数据，将输出元素分配给变量，从回调函数中调用其他函数，引用我之前创建的字典等。但是，每当我尝试做这些事情时，我都会遇到错误. 正如您从代码中看到的那样，我创建了一个名为的 dict xpath_mapping，然后我想在回调中使用它，这样我就可以将订阅 ID 映射到 xpath，这样我就可以显示例如 CPU util。我需要能够映射数据，因为如果周期不同，某些数据会比其他数据更频繁地出现。

如何从回调中获取数据以便使用它？

我是否认为一旦应用程序运行，回调之外的所有内容都不再可访问？

这是代码：

#
# Import Required Modules
#
import signal
import sys
import time
import ast
from lxml import etree
from ncclient import manager


def get_info_list():

    with open('C:\\Users\\Dave\\PycharmProjects\\NetworkTelemetry\\info_csr1000v.txt', 'r') as info:
        info_list_temp = list(info.readlines())
        info_list = [z.strip() for z in info_list_temp]

    return info_list


def unknown_host_cb(host, fingerprint):
    return True


def connect(info_list):

    info = ast.literal_eval(info_list[0])

    m = manager.connect(host=info['HOST'],
                        port=info['PORT'],
                        username=info['USERNAME'],
                        password=info['PASSWORD'],
                        allow_agent=False,
                        look_for_keys=False,
                        hostkey_verify=False,
                        device_params={'name': 'iosxe'},
                        unknown_host_cb=unknown_host_cb)

    return m
#
# set up a ctrl+c handler to tear down the netconf session
# Subscriptions auto removed on NETCONF tear down
#
def sigint_handler(signal, frame):
    m.close_session()
    sys.exit(0)
signal.signal(signal.SIGINT, sigint_handler)


#
# A really simple callback, just spit out a header with some key
# details plus the XML payload pretty-printed.
#

"""Getting ready to parse info from XML File"""

def callback(notif):

    sub = notif.subscription_id
    event_time = notif.event_time
    print('Subscription ID: ', sub)
    print('Event Time:      ', event_time)
    print(etree.tostring(notif.datastore_ele, pretty_print=True).decode('utf-8'))


def errback(notif):
    pass


if __name__ == '__main__':
    #
    # Connect
    #
    info_list = get_info_list()
    m = connect(info_list)
    info_list.pop(0)
    #
    # iterate over the list of xpaths and create subscriptions
    #
    xpath_Dict = {}
    xpath_mapping = []
    subs = []
    for x in info_list:
        xpath = ast.literal_eval(x)
        s = m.establish_subscription(
            callback,
            errback,
            xpath=xpath['XPATH'],
            period=xpath['PERIOD'])
        Friendly = xpath['FRIENDLY_NAME']
        sub_id = s.subscription_id
        xpath_Dict = {'FRIENDLY_NAME': Friendly, 'SUB_ID': sub_id}
        xpath_mapping.append(xpath_Dict)
        print('Subscription Result : %s' % s.subscription_result)
        if s.subscription_result.endswith('ok'):
            print(Friendly)
            print(f'Subscription Id     : {sub_id}' )
            subs.append(s.subscription_id)
    if not len(subs):
        print('No active subscriptions, exiting.')
        sys.exit(1)
    #
    # simple forever loop
    #
    while True:
        time.sleep(1)

因为它是有效的，并且输出会定期打印，但是我需要做的不仅仅是打印它。任何人都可以解释这是如何工作的，或者我该如何做我需要做的事情？

4

0 回答 0