Commit 715d82d6 authored by Waleed Akbar's avatar Waleed Akbar
Browse files

feat(collectors): ECOC2026-Enhance NETCONF filter building and parsing for optical-channel metrics

parent 0328b023

src/telemetry/backend/service/collectors/netconf_oc/PathBuilder.py

+56 −10
Original line number Diff line number Diff line
@@ -24,6 +24,7 @@ Namespaces used: 
"""



import logging

from decimal import Decimal, InvalidOperation

from typing import Optional



import lxml.etree as ET
@@ -47,13 +48,21 @@ _FILTER_TEMPLATE = """\ 
      <name>{channel_name}</name>

      <optical-channel xmlns="{ns_terminal}">

        <state>

          <{metric_leaf}/>

            {metric_filter}

        </state>

      </optical-channel>

    </component>

  </components>

</filter>"""



_DEFAULT_METRIC_FILTER_TEMPLATE = "<{metric_leaf}/>"



_PRE_FEC_BER_FILTER = """\

          <digital-subcarriers-group>

            <pre-fec-ber/>

          </digital-subcarriers-group>

          <pre-fec-ber/>"""





def build(channel_name: str, metric_leaf: str) -> str:

    """Return a NETCONF subtree <filter> string for the given channel and metric.
@@ -62,12 +71,42 @@ def build(channel_name: str, metric_leaf: str) -> str: 
        channel_name: e.g. 'channel-1'

        metric_leaf:  e.g. 'pre-fec-ber' or 'input-power'

    """

    metric_filter = (

        _PRE_FEC_BER_FILTER

        if metric_leaf == 'pre-fec-ber'

        else _DEFAULT_METRIC_FILTER_TEMPLATE.format(metric_leaf=metric_leaf)

    )

    return _FILTER_TEMPLATE.format(

        ns_platform  = NS_PLATFORM,

        ns_terminal  = NS_TERMINAL,

        channel_name = channel_name,

        metric_leaf  = metric_leaf,

        metric_filter= metric_filter,

    )





def _to_float(raw: str, channel_name: str, metric_leaf: str) -> Optional[float]:

    raw = raw.strip()

    if not raw:

        return None

    try:

        return float(Decimal(raw))

    except (InvalidOperation, ValueError):

        LOGGER.warning(

            "PathBuilder.parse: cannot convert '%s' to float (channel=%s metric=%s)",

            raw, channel_name, metric_leaf,

        )

        return None





def _first_float(nodes, channel_name: str, metric_leaf: str) -> Optional[float]:

    for node in nodes:

        raw = node.text

        if raw is None:

            continue

        value = _to_float(raw, channel_name, metric_leaf)

        if value is not None:

            return value

    return None





def parse(data_ele: ET._Element, channel_name: str, metric_leaf: str) -> Optional[float]:
@@ -81,6 +120,20 @@ def parse(data_ele: ET._Element, channel_name: str, metric_leaf: str) -> Optiona 
    Returns:

        float value if found, None otherwise.

    """

    if metric_leaf == 'pre-fec-ber':

        dsc_group_xpath = (

            ".//ocp:components/ocp:component"

            "[ocp:name='{channel_name}']"

            "/oct:optical-channel/oct:state"

            "/oct:digital-subcarriers-group"

            "/oct:pre-fec-ber/oct:instant"

        ).format(channel_name=channel_name)



        dsc_group_nodes = data_ele.xpath(dsc_group_xpath, namespaces=_NAMESPACES)

        dsc_group_value = _first_float(dsc_group_nodes, channel_name, metric_leaf)

        if dsc_group_value is not None:

            return dsc_group_value



    xpath = (

        ".//ocp:components/ocp:component"

        "[ocp:name='{channel_name}']"
@@ -100,11 +153,4 @@ def parse(data_ele: ET._Element, channel_name: str, metric_leaf: str) -> Optiona 
    if raw is None:

        return None



    try:

        return float(raw)

    except ValueError:

        LOGGER.warning(

            "PathBuilder.parse: cannot convert '%s' to float (channel=%s metric=%s)",

            raw, channel_name, metric_leaf,

        )

        return None
 
    return _to_float(raw, channel_name, metric_leaf)