Add Support for Arista CMIS Enhanced LPO Debug Registers

setup.py

@@ -58,6 +58,8 @@
         'sonic_platform_base.sonic_xcvr.codes.public',
         'sonic_platform_base.sonic_xcvr.codes.bailly',
         'sonic_platform_base.sonic_xcvr.utils',
+        'sonic_platform_base.sonic_xcvr.api.arista',
+        'sonic_platform_base.sonic_xcvr.mem_maps.arista',
         'sonic_platform_base.sonic_xcvr.api.credo',
         'sonic_platform_base.sonic_xcvr.mem_maps.credo',
         'sonic_platform_base.sonic_xcvr.codes.credo',

sonic_platform_base/sonic_xcvr/api/arista/cmis_enhanced_lpo.py

@@ -0,0 +1,198 @@
+"""
+    cmis_enhanced_lpo.py
+
+    API for Arista Enhanced LPO modules.
+"""
+
+from ..public.cmis import CmisApi
+from ...fields import arista_lpo_consts as lpo
+
+# LPO_CAPABILITY is C1h:128. Bits 2, 3, and 4 advertise OER, VMA, and OMA support.
+_OER_CAPABILITY_MASK = 0x04
+_VMA_CAPABILITY_MASK = 0x08
+_OMA_CAPABILITY_MASK = 0x10
+
+_VMA_FLAG_FIELDS = [
+    lpo.LPO_TX_HOST_INPUT_VMA_HIGH_ALARM_FLAG,
+    lpo.LPO_TX_HOST_INPUT_VMA_LOW_ALARM_FLAG,
+    lpo.LPO_TX_HOST_INPUT_VMA_HIGH_WARNING_FLAG,
+    lpo.LPO_TX_HOST_INPUT_VMA_LOW_WARNING_FLAG,
+]
+
+_OMA_FLAG_FIELDS = [
+    lpo.LPO_RX_INPUT_OMA_HIGH_ALARM_FLAG,
+    lpo.LPO_RX_INPUT_OMA_LOW_ALARM_FLAG,
+    lpo.LPO_RX_INPUT_OMA_HIGH_WARNING_FLAG,
+    lpo.LPO_RX_INPUT_OMA_LOW_WARNING_FLAG,
+]
+
+_VMA_TX_FIELDS = [
+    lpo.LPO_HOST_INPUT_VMA_TX1,
+    lpo.LPO_HOST_INPUT_VMA_TX2,
+    lpo.LPO_HOST_INPUT_VMA_TX3,
+    lpo.LPO_HOST_INPUT_VMA_TX4,
+    lpo.LPO_HOST_INPUT_VMA_TX5,
+    lpo.LPO_HOST_INPUT_VMA_TX6,
+    lpo.LPO_HOST_INPUT_VMA_TX7,
+    lpo.LPO_HOST_INPUT_VMA_TX8,
+]
+
+_OMA_RX_FIELDS = [
+    lpo.LPO_INPUT_OMA_RX1,
+    lpo.LPO_INPUT_OMA_RX2,
+    lpo.LPO_INPUT_OMA_RX3,
+    lpo.LPO_INPUT_OMA_RX4,
+    lpo.LPO_INPUT_OMA_RX5,
+    lpo.LPO_INPUT_OMA_RX6,
+    lpo.LPO_INPUT_OMA_RX7,
+    lpo.LPO_INPUT_OMA_RX8,
+]
+
+
+_VMA_THRESHOLD_FIELDS = [
+    lpo.LPO_TX_HOST_INPUT_VMA_HIGH_ALARM_THRESHOLD,
+    lpo.LPO_TX_HOST_INPUT_VMA_LOW_ALARM_THRESHOLD,
+    lpo.LPO_TX_HOST_INPUT_VMA_HIGH_WARNING_THRESHOLD,
+    lpo.LPO_TX_HOST_INPUT_VMA_LOW_WARNING_THRESHOLD,
+]
+
+_OMA_THRESHOLD_FIELDS = [
+    lpo.LPO_RX_INPUT_OMA_HIGH_ALARM_THRESHOLD,
+    lpo.LPO_RX_INPUT_OMA_LOW_ALARM_THRESHOLD,
+    lpo.LPO_RX_INPUT_OMA_HIGH_WARNING_THRESHOLD,
+    lpo.LPO_RX_INPUT_OMA_LOW_WARNING_THRESHOLD,
+]
+
+
+def _value_or_na(value):
+    """Return the API fallback for decoded fields that failed to read."""
+    return value if value is not None else 'N/A'
+
+
+def _read_value_or_na(xcvr_eeprom, field):
+    """Read an already-decoded EEPROM field and normalize read failures."""
+    return _value_or_na(xcvr_eeprom.read(field))
+
+
+def _set_na(output, fields):
+    """Populate unsupported scalar fields without reading optional registers."""
+    for field in fields:
+        output[field] = 'N/A'
+
+
+def _set_lane_flag_na(output, fields):
+    """Populate unsupported per-lane flag fields with the API fallback value."""
+    for field in fields:
+        for lane in range(1, 9):
+            output["{}{}".format(field, lane)] = 'N/A'
+
+
+def _lane_flag_na_dict(field):
+    """Build the per-lane fallback shape expected by decoded flag fields."""
+    flags = {}
+    _set_lane_flag_na(flags, [field])
+    return flags
+
+
+def _is_capability_supported(capabilities, mask):
+    """Check whether an optional LPO capability bit is advertised."""
+    return bool(capabilities & mask)
+
+
+class CmisEnhancedLpoApi(CmisApi):
+    def _read_if_supported(self, field, capabilities, mask):
+        """Read optional info fields only when the capability bit is set."""
+        if not _is_capability_supported(capabilities, mask):
+            return 'N/A'
+        return _read_value_or_na(self.xcvr_eeprom, field)
+
+    def get_transceiver_info(self):
+        xcvr_info = super().get_transceiver_info()
+        if xcvr_info is None:
+            return None
+        capabilities = self.xcvr_eeprom.read(lpo.LPO_CAPABILITY)
+        if capabilities is None:
+            return None
+
+        xcvr_info.update({
+            lpo.LPO_TX_POLARITY_INVERTED: _read_value_or_na(self.xcvr_eeprom, lpo.LPO_TX_POLARITY_INVERTED),
+            lpo.LPO_RX_POLARITY_INVERTED: _read_value_or_na(self.xcvr_eeprom, lpo.LPO_RX_POLARITY_INVERTED),
+            lpo.LPO_TX_HOST_INPUT_VMA_MON_ACCURACY: self._read_if_supported(
+                lpo.LPO_TX_HOST_INPUT_VMA_MON_ACCURACY, capabilities, _VMA_CAPABILITY_MASK
+            ),
+            lpo.LPO_RX_INPUT_OMA_MON_ACCURACY: self._read_if_supported(
+                lpo.LPO_RX_INPUT_OMA_MON_ACCURACY, capabilities, _OMA_CAPABILITY_MASK
+            ),
+            lpo.LPO_TX_OUTER_EXTINCTION_RATIO_MAX: self._read_if_supported(
+                lpo.LPO_TX_OUTER_EXTINCTION_RATIO_MAX, capabilities, _OER_CAPABILITY_MASK
+            ),
+        })
+        return xcvr_info
+
+    def get_transceiver_dom_real_value(self):
+        trans_dom = super().get_transceiver_dom_real_value()
+        if trans_dom is None:
+            return None
+        capabilities = self.xcvr_eeprom.read(lpo.LPO_CAPABILITY)
+        if capabilities is None:
+            return None
+
+        if _is_capability_supported(capabilities, _VMA_CAPABILITY_MASK):
+            for field in _VMA_TX_FIELDS:
+                trans_dom[field] = _read_value_or_na(self.xcvr_eeprom, field)
+        else:
+            _set_na(trans_dom, _VMA_TX_FIELDS)
+
+        if _is_capability_supported(capabilities, _OMA_CAPABILITY_MASK):
+            for field in _OMA_RX_FIELDS:
+                trans_dom[field] = _read_value_or_na(self.xcvr_eeprom, field)
+        else:
+            _set_na(trans_dom, _OMA_RX_FIELDS)
+
+        return trans_dom
+
+    def get_transceiver_threshold_info(self):
+        threshold_dict = super().get_transceiver_threshold_info()
+        if threshold_dict is None:
+            return None
+        capabilities = self.xcvr_eeprom.read(lpo.LPO_CAPABILITY)
+        if capabilities is None:
+            return None
+
+        if _is_capability_supported(capabilities, _VMA_CAPABILITY_MASK):
+            for field in _VMA_THRESHOLD_FIELDS:
+                threshold_dict[field] = _read_value_or_na(self.xcvr_eeprom, field)
+        else:
+            _set_na(threshold_dict, _VMA_THRESHOLD_FIELDS)
+
+        if _is_capability_supported(capabilities, _OMA_CAPABILITY_MASK):
+            for field in _OMA_THRESHOLD_FIELDS:
+                threshold_dict[field] = _read_value_or_na(self.xcvr_eeprom, field)
+        else:
+            _set_na(threshold_dict, _OMA_THRESHOLD_FIELDS)
+
+        return threshold_dict
+
+    def get_transceiver_dom_flags(self):
+        dom_flag_dict = super().get_transceiver_dom_flags()
+        if dom_flag_dict is None:
+            return None
+        capabilities = self.xcvr_eeprom.read(lpo.LPO_CAPABILITY)
+        if capabilities is None:
+            return None
+
+        if _is_capability_supported(capabilities, _VMA_CAPABILITY_MASK):
+            for field in _VMA_FLAG_FIELDS:
+                flags = self.xcvr_eeprom.read(field)
+                dom_flag_dict.update(flags if flags is not None else _lane_flag_na_dict(field))
+        else:
+            _set_lane_flag_na(dom_flag_dict, _VMA_FLAG_FIELDS)
+
+        if _is_capability_supported(capabilities, _OMA_CAPABILITY_MASK):
+            for field in _OMA_FLAG_FIELDS:
+                flags = self.xcvr_eeprom.read(field)
+                dom_flag_dict.update(flags if flags is not None else _lane_flag_na_dict(field))
+        else:
+            _set_lane_flag_na(dom_flag_dict, _OMA_FLAG_FIELDS)
+
+        return dom_flag_dict

sonic_platform_base/sonic_xcvr/fields/arista_lpo_consts.py

@@ -0,0 +1,79 @@
+# Enhanced LPO (Arista) - field groups
+LPO_EEPROM_FIELD = "LpoEepromField"
+LPO_INFO_FIELD = "LpoInfoField"
+LPO_VMA_THRESHOLDS_FIELD = "LpoVmaThresholdsField"
+LPO_OMA_THRESHOLDS_FIELD = "LpoOmaThresholdsField"
+LPO_VMA_FLAGS_FIELD = "LpoVmaFlagsField"
+LPO_VMA_DOM_FIELD = "LpoVmaDomField"
+LPO_VMA_MASKS_FIELD = "LpoVmaMasksField"
+LPO_OMA_FLAGS_FIELD = "LpoOmaFlagsField"
+LPO_OMA_DOM_FIELD = "LpoOmaDomField"
+LPO_OMA_MASKS_FIELD = "LpoOmaMasksField"
+
+# Enhanced LPO - page 01h advertisement
+LPO_EEPROM_COMPLIANCE = "LPOEepromCompliance"
+LPO_ENHANCED_SPEC_VERSION = "LPOEnhancedSpecVersion"
+
+# Enhanced LPO - page C1h info / capability
+LPO_CAPABILITY = "LPOCapability"
+LPO_TX_OUTER_EXTINCTION_RATIO_MAX = "LPOTxOuterExtinctionRatioMax"
+LPO_TX_POLARITY_INVERTED = "LPOTxPolarityInverted"
+LPO_RX_POLARITY_INVERTED = "LPORxPolarityInverted"
+LPO_TX_HOST_INPUT_VMA_MON_ACCURACY = "LPOTxHostInputVMAMonAccuracySupported"
+LPO_RX_INPUT_OMA_MON_ACCURACY = "LPORxInputOMAMonAccuracySupported"
+
+# Enhanced LPO - page C1h VMA thresholds
+LPO_TX_HOST_INPUT_VMA_HIGH_ALARM_THRESHOLD = "LPOTxHostInputVMAHighAlarmThreshold"
+LPO_TX_HOST_INPUT_VMA_LOW_ALARM_THRESHOLD = "LPOTxHostInputVMALowAlarmThreshold"
+LPO_TX_HOST_INPUT_VMA_HIGH_WARNING_THRESHOLD = "LPOTxHostInputVMAHighWarningThreshold"
+LPO_TX_HOST_INPUT_VMA_LOW_WARNING_THRESHOLD = "LPOTxHostInputVMALowWarningThreshold"
+
+# Enhanced LPO - page C1h OMA thresholds
+LPO_RX_INPUT_OMA_HIGH_ALARM_THRESHOLD = "LPORxInputOMAHighAlarmThreshold"
+LPO_RX_INPUT_OMA_LOW_ALARM_THRESHOLD = "LPORxInputOMALowAlarmThreshold"
+LPO_RX_INPUT_OMA_HIGH_WARNING_THRESHOLD = "LPORxInputOMAHighWarningThreshold"
+LPO_RX_INPUT_OMA_LOW_WARNING_THRESHOLD = "LPORxInputOMALowWarningThreshold"
+
+# Enhanced LPO - page C2h VMA flags
+LPO_TX_HOST_INPUT_VMA_HIGH_ALARM_FLAG = "LPOTxHostInputVMAHighAlarmFlag"
+LPO_TX_HOST_INPUT_VMA_LOW_ALARM_FLAG = "LPOTxHostInputVMALowAlarmFlag"
+LPO_TX_HOST_INPUT_VMA_HIGH_WARNING_FLAG = "LPOTxHostInputVMAHighWarningFlag"
+LPO_TX_HOST_INPUT_VMA_LOW_WARNING_FLAG = "LPOTxHostInputVMALowWarningFlag"
+
+# Enhanced LPO - page C2h per-lane VMA measurements
+LPO_HOST_INPUT_VMA_TX1 = "LPOHostInputVMATx1"
+LPO_HOST_INPUT_VMA_TX2 = "LPOHostInputVMATx2"
+LPO_HOST_INPUT_VMA_TX3 = "LPOHostInputVMATx3"
+LPO_HOST_INPUT_VMA_TX4 = "LPOHostInputVMATx4"
+LPO_HOST_INPUT_VMA_TX5 = "LPOHostInputVMATx5"
+LPO_HOST_INPUT_VMA_TX6 = "LPOHostInputVMATx6"
+LPO_HOST_INPUT_VMA_TX7 = "LPOHostInputVMATx7"
+LPO_HOST_INPUT_VMA_TX8 = "LPOHostInputVMATx8"
+
+# Enhanced LPO - page C2h VMA masks
+LPO_TX_HOST_INPUT_VMA_HIGH_ALARM_MASK = "LPOTxHostInputVMAHighAlarmMask"
+LPO_TX_HOST_INPUT_VMA_LOW_ALARM_MASK = "LPOTxHostInputVMALowAlarmMask"
+LPO_TX_HOST_INPUT_VMA_HIGH_WARNING_MASK = "LPOTxHostInputVMAHighWarningMask"
+LPO_TX_HOST_INPUT_VMA_LOW_WARNING_MASK = "LPOTxHostInputVMALowWarningMask"
+
+# Enhanced LPO - page C2h OMA flags
+LPO_RX_INPUT_OMA_HIGH_ALARM_FLAG = "LPORxInputOMAHighAlarmFlag"
+LPO_RX_INPUT_OMA_LOW_ALARM_FLAG = "LPORxInputOMALowAlarmFlag"
+LPO_RX_INPUT_OMA_HIGH_WARNING_FLAG = "LPORxInputOMAHighWarningFlag"
+LPO_RX_INPUT_OMA_LOW_WARNING_FLAG = "LPORxInputOMALowWarningFlag"
+
+# Enhanced LPO - page C2h per-lane OMA measurements
+LPO_INPUT_OMA_RX1 = "LPOInputOMARx1"
+LPO_INPUT_OMA_RX2 = "LPOInputOMARx2"
+LPO_INPUT_OMA_RX3 = "LPOInputOMARx3"
+LPO_INPUT_OMA_RX4 = "LPOInputOMARx4"
+LPO_INPUT_OMA_RX5 = "LPOInputOMARx5"
+LPO_INPUT_OMA_RX6 = "LPOInputOMARx6"
+LPO_INPUT_OMA_RX7 = "LPOInputOMARx7"
+LPO_INPUT_OMA_RX8 = "LPOInputOMARx8"
+
+# Enhanced LPO - page C2h OMA masks
+LPO_RX_INPUT_OMA_HIGH_ALARM_MASK = "LPORxInputOMAHighAlarmMask"
+LPO_RX_INPUT_OMA_LOW_ALARM_MASK = "LPORxInputOMALowAlarmMask"
+LPO_RX_INPUT_OMA_HIGH_WARNING_MASK = "LPORxInputOMAHighWarningMask"
+LPO_RX_INPUT_OMA_LOW_WARNING_MASK = "LPORxInputOMALowWarningMask"

sonic_platform_base/sonic_xcvr/fields/cmis_enhanced_lpo.py

@@ -0,0 +1,50 @@
+import struct
+
+from .xcvr_field import NumberRegField
+
+
+class _LpoScaledNumberRegField(NumberRegField):
+    factor = 1
+
+    def __init__(self, name, offset, *fields, **kwargs):
+        super(_LpoScaledNumberRegField, self).__init__(name, offset, *fields, **kwargs)
+        self.precision = kwargs.get("precision", 3)
+
+    def decode(self, raw_data, **decoded_deps):
+        value = struct.unpack(self.format, raw_data)[0] * self.factor
+        return float("{:.{}f}".format(value, self.precision))
+
+
+class _LpoLowerNibbleScaledRegField(_LpoScaledNumberRegField):
+    def decode(self, raw_data, **decoded_deps):
+        value = (struct.unpack(self.format, raw_data)[0] & 0x0F) * self.factor
+        return float("{:.{}f}".format(value, self.precision))
+
+
+class LpoVmaField(_LpoScaledNumberRegField):
+    factor = 5
+
+
+class LpoOmaField(_LpoScaledNumberRegField):
+    factor = 0.0001
+
+
+class LpoOerField(_LpoScaledNumberRegField):
+    factor = 0.1
+
+
+class LpoVmaAccuracyField(_LpoLowerNibbleScaledRegField):
+    factor = 5
+
+
+class LpoOmaAccuracyField(_LpoLowerNibbleScaledRegField):
+    factor = 0.2
+
+
+class LpoLaneFlagRegField(NumberRegField):
+    def decode(self, raw_data, **decoded_deps):
+        value = struct.unpack(self.format, raw_data)[0]
+        decoded = {}
+        for lane in range(1, 9):
+            decoded["{}{}".format(self.name, lane)] = bool((value >> (lane - 1)) & 1)
+        return decoded