nokia.py

"""

"""
import contextlib
import logging
import time
import typing

from lxml import etree
import ncclient.manager

if typing.TYPE_CHECKING:
    from ncclient.lxml_ import NCElement

logger = logging.getLogger(__name__)

STATE_NS = 'urn:nokia.com:sros:ns:yang:sr:state'

# TODO: decide if these are really global
DEFAULT_NETCONF_PORT = 830
# cf. POL1-799
DEFAULT_NETCONF_CONNECT_TIMEOUT = 60  # seconds
# DEFAULT_NETCONF_CONNECT_TIMEOUT = 5  # seconds
ENDPOINT_TYPES = {'port', 'lag'}


@contextlib.contextmanager
def _connection(hostname: str, ssh_params: dict, port: int = DEFAULT_NETCONF_PORT):
    params = {
        'host': hostname,
        'port': port,
        'device_params': {'name': 'sros'},
        'nc_params': {'capabilities': ['urn:nokia.com:nc:pysros:pc']},
        'timeout': DEFAULT_NETCONF_CONNECT_TIMEOUT
    }
    params.update(ssh_params)

    conn = ncclient.manager.connect(**params)
    conn.async_mode = False
    yield conn


def _remove_ns(nc_doc: 'NCElement'):
    # convert to etree
    etree_doc = etree.fromstring(nc_doc.tostring.decode('utf-8'))
    for elem in etree_doc.getiterator():
        elem.tag = etree.QName(elem).localname
    etree.cleanup_namespaces(etree_doc)
    return etree_doc


def get_netconf_interface_info(router_name: str, ssh_params: dict):

    def _filter(endpoint_type: str):
        # https://github.com/nokia/pysros/blob/main/examples/show_port_counters.py
        # https://netdevops.me/2020/nokia-yang-tree-and-path-browser/
        # https://github.com/hellt/nokia-yangtree
        _f = etree.Element('filter')
        _state = etree.SubElement(
            _f,
            f'{{{STATE_NS}}}state',
            nsmap={'nokia-state': STATE_NS})
        _endpoint = etree.SubElement(_state, f'{{{STATE_NS}}}{endpoint_type}')
        # cf. POL1-799
        # _ = etree.SubElement(_endpoint, f'{{{STATE_NS}}}statistics')
        return _f

    start = time.time()
    with _connection(hostname=router_name, ssh_params=ssh_params) as router:
        # return {_ept: _remove_ns(router.get(filter=_filter(_ept))) for _ept in ENDPOINT_TYPES}
        now = time.time()
        logger.debug(f"seconds to connect ({router_name}: {now - start}")

        rsp = {}
        for _ept in ENDPOINT_TYPES:
            start = now
            logger.debug(f"Getting {router_name} {_ept} info")
            rsp[_ept] = _remove_ns(router.get(filter=_filter(_ept)))
            now = time.time()
            logger.debug(f"   seconds to get {_ept} info: {now - start}")

        return rsp


def interface_counters(raw_counter_docs):

    assert False, "TODO"


if __name__ == '__main__':
    from brian_polling_manager.interface_stats import config
    import os
    config_filename = os.path.join(
        os.path.dirname(__file__),
        '..', 'config-test.json')
    with open(config_filename) as f:
        params = config.load(f)

    ROUTERS = [
        'rt0.lon.uk.lab.office.geant.net',
        'rt0.ams.nl.lab.office.geant.net'
    ]

    # rsp = get_interface_info_ncrpc(
    raw_counter_docs = get_netconf_interface_info(
        router_name=ROUTERS[0],
        ssh_params=params['nokia'])

    print(interface_counters(raw_counter_docs))