[sfp-refactor] Add initial support for CMIS in sonic_xcvr (sonic-net#220

)

sonic_platform_base/sonic_xcvr/api/public/cmis.py

@@ -1,7 +1,7 @@
 """
     cmis.py
 
-    Implementation of XcvrApi that corresponds to CMIS
+    Implementation of XcvrApi that corresponds to the CMIS specification.
 """
 
 from ...fields import consts
@@ -16,5 +16,270 @@ def __init__(self, xcvr_eeprom):
     def get_model(self):
         return self.xcvr_eeprom.read(consts.VENDOR_PART_NO_FIELD)
 
-    # TODO: other XcvrApi methods
+    def get_serial(self):
+        return self.xcvr_eeprom.read(consts.VENDOR_SERIAL_NO_FIELD)
 
+    def get_transceiver_info(self):
+        admin_info = self.xcvr_eeprom.read(consts.ADMIN_INFO_FIELD)
+        media_type = self.xcvr_eeprom.read(consts.MEDIA_TYPE_FIELD)
+        if admin_info is None or media_type is None:
+            return None
+
+        ext_id = admin_info[consts.EXT_ID_FIELD]
+        power_class = ext_id[consts.POWER_CLASS_FIELD]
+        max_power = ext_id[consts.MAX_POWER_FIELD]
+
+        xcvr_info = {
+            "type": admin_info[consts.ID_FIELD],
+            "type_abbrv_name": admin_info[consts.ID_ABBRV_FIELD],
+            "hardware_rev": admin_info[consts.VENDOR_REV_FIELD],
+            "serial": admin_info[consts.VENDOR_SERIAL_NO_FIELD],
+            "manufacturer": admin_info[consts.VENDOR_NAME_FIELD],
+            "model": admin_info[consts.VENDOR_PART_NO_FIELD],
+            "connector": admin_info[consts.CONNECTOR_FIELD],
+            "encoding": "N/A", # Not supported
+            "ext_identifier": "%s (%sW Max)" % (power_class, max_power),
+            "ext_rateselect_compliance": "N/A", # Not supported
+            "cable_type": "Length cable Assembly(m)",
+            "cable_length": float(admin_info[consts.LENGTH_ASSEMBLY_FIELD]),
+            "nominal_bit_rate": 0, # Not supported
+            "specification_compliance": media_type,
+            "vendor_date": admin_info[consts.VENDOR_DATE_FIELD],
+            "vendor_oui": admin_info[consts.VENDOR_OUI_FIELD],
+            # TODO
+            "application_advertisement": "N/A",
+        }
+        return xcvr_info
+
+    def get_transceiver_bulk_status(self):
+        rx_los = self.get_rx_los()
+        tx_fault = self.get_tx_fault()
+        tx_disable = self.get_tx_disable()
+        tx_disabled_channel = self.get_tx_disable_channel()
+        temp = self.get_module_temperature()
+        voltage = self.get_voltage()
+        tx_bias = self.get_tx_bias()
+        rx_power = self.get_rx_power()
+        tx_power = self.get_tx_power()
+        read_failed = rx_los is None or \
+                      tx_fault is None or \
+                      tx_disable is None or \
+                      tx_disabled_channel is None or \
+                      temp is None or \
+                      voltage is None or \
+                      tx_bias is None or \
+                      rx_power is None or \
+                      tx_power is None
+        if read_failed:
+            return None
+
+        bulk_status = {
+            "rx_los": all(rx_los) if self.get_rx_los_support() else 'N/A',
+            "tx_fault": all(tx_fault) if self.get_tx_fault_support() else 'N/A',
+            "tx_disable": all(tx_disable),
+            "tx_disabled_channel": tx_disabled_channel,
+            "temperature": temp,
+            "voltage": voltage
+        }
+
+        for i in range(1, self.NUM_CHANNELS + 1):
+            bulk_status["tx%dbias" % i] = tx_bias[i - 1]
+            bulk_status["rx%dpower" % i] = rx_power[i - 1]
+            bulk_status["tx%dpower" % i] = tx_power[i - 1]
+
+        return bulk_status
+
+    def get_transceiver_threshold_info(self):
+        threshold_info_keys = ['temphighalarm',    'temphighwarning',
+                               'templowalarm',     'templowwarning',
+                               'vcchighalarm',     'vcchighwarning',
+                               'vcclowalarm',      'vcclowwarning',
+                               'rxpowerhighalarm', 'rxpowerhighwarning',
+                               'rxpowerlowalarm',  'rxpowerlowwarning',
+                               'txpowerhighalarm', 'txpowerhighwarning',
+                               'txpowerlowalarm',  'txpowerlowwarning',
+                               'txbiashighalarm',  'txbiashighwarning',
+                               'txbiaslowalarm',   'txbiaslowwarning'
+                              ]
+        threshold_info_dict = dict.fromkeys(threshold_info_keys, 'N/A')
+
+        thresh_support = self.get_transceiver_thresholds_support()
+        if thresh_support is None:
+            return None
+        if not thresh_support:
+            return threshold_info_dict
+        thresh = self.xcvr_eeprom.read(consts.THRESHOLDS_FIELD)
+        if thresh is None:
+            return None
+
+        return {
+            "temphighalarm": float("{:.3f}".format(thresh[consts.TEMP_HIGH_ALARM_FIELD])),
+            "templowalarm": float("{:.3f}".format(thresh[consts.TEMP_LOW_ALARM_FIELD])),
+            "temphighwarning": float("{:.3f}".format(thresh[consts.TEMP_HIGH_WARNING_FIELD])),
+            "templowwarning": float("{:.3f}".format(thresh[consts.TEMP_LOW_WARNING_FIELD])),
+            "vcchighalarm": float("{:.3f}".format(thresh[consts.VOLTAGE_HIGH_ALARM_FIELD])),
+            "vcclowalarm": float("{:.3f}".format(thresh[consts.VOLTAGE_LOW_ALARM_FIELD])),
+            "vcchighwarning": float("{:.3f}".format(thresh[consts.VOLTAGE_HIGH_WARNING_FIELD])),
+            "vcclowwarning": float("{:.3f}".format(thresh[consts.VOLTAGE_LOW_WARNING_FIELD])),
+            "rxpowerhighalarm": float("{:.3f}".format(thresh[consts.RX_POWER_HIGH_ALARM_FIELD])),
+            "rxpowerlowalarm": float("{:.3f}".format(self.mw_to_dbm(thresh[consts.RX_POWER_LOW_ALARM_FIELD]))),
+            "rxpowerhighwarning": float("{:.3f}".format(self.mw_to_dbm(thresh[consts.RX_POWER_HIGH_WARNING_FIELD]))),
+            "rxpowerlowwarning": float("{:.3f}".format(self.mw_to_dbm(thresh[consts.RX_POWER_LOW_WARNING_FIELD]))),
+            "txpowerhighalarm": float("{:.3f}".format(self.mw_to_dbm(thresh[consts.TX_POWER_HIGH_ALARM_FIELD]))),
+            "txpowerlowalarm": float("{:.3f}".format(self.mw_to_dbm(thresh[consts.TX_POWER_LOW_ALARM_FIELD]))),
+            "txpowerhighwarning": float("{:.3f}".format(self.mw_to_dbm(thresh[consts.TX_POWER_HIGH_WARNING_FIELD]))),
+            "txpowerlowwarning": float("{:.3f}".format(self.mw_to_dbm(thresh[consts.TX_POWER_LOW_WARNING_FIELD]))),
+            "txbiashighalarm": float("{:.3f}".format(thresh[consts.TX_BIAS_HIGH_ALARM_FIELD])),
+            "txbiaslowalarm": float("{:.3f}".format(thresh[consts.TX_BIAS_LOW_ALARM_FIELD])),
+            "txbiashighwarning": float("{:.3f}".format(thresh[consts.TX_BIAS_HIGH_WARNING_FIELD])),
+            "txbiaslowwarning": float("{:.3f}".format(thresh[consts.TX_BIAS_LOW_WARNING_FIELD]))
+        }
+
+    def get_module_temperature(self):
+        if not self.get_temperature_support():
+            return 'N/A'
+        temp = self.xcvr_eeprom.read(consts.TEMPERATURE_FIELD)
+        if temp is None:
+            return None
+        return float("{:.3f}".format(temp))
+
+    def get_voltage(self):
+        if not self.get_voltage_support():
+            return 'N/A'
+        voltage = self.xcvr_eeprom.read(consts.VOLTAGE_FIELD)
+        if voltage is None:
+            return None
+        return float("{:.3f}".format(voltage))
+
+    def is_flat_memory(self):
+        return self.xcvr_eeprom.read(consts.FLAT_MEM_FIELD)
+
+    def get_temperature_support(self):
+        return not self.is_flat_memory()
+
+    def get_voltage_support(self):
+        return not self.is_flat_memory()
+
+    def get_rx_los_support(self):
+        return not self.is_flat_memory()
+
+    def get_rx_los(self):
+        rx_los_support = self.get_rx_los_support()
+        if rx_los_support is None:
+            return None
+        if not rx_los_support:
+            return ["N/A" for _ in range(self.NUM_CHANNELS)]
+        rx_los = self.xcvr_eeprom.read(consts.RX_LOS_FIELD)
+        if rx_los is None:
+            return None
+        return [bool(rx_los & (1 << i)) for i in range(self.NUM_CHANNELS)]
+
+    def get_tx_bias_support(self):
+        return not self.is_flat_memory()
+
+    def get_tx_bias(self):
+        tx_bias_support = self.get_tx_bias_support()
+        if tx_bias_support is None:
+            return None
+        if not tx_bias_support:
+            return ["N/A" for _ in range(self.NUM_CHANNELS)]
+        tx_bias = self.xcvr_eeprom.read(consts.TX_BIAS_FIELD)
+        if tx_bias is None:
+            return None
+        return [channel_bias for channel_bias in tx_bias.values()]
+
+    def get_tx_power(self):
+        tx_power_support = self.get_tx_power_support()
+        if tx_power_support is None:
+            return None
+        if not tx_power_support:
+            return ["N/A" for _ in range(self.NUM_CHANNELS)]
+        tx_power = self.xcvr_eeprom.read(consts.TX_POWER_FIELD)
+        if tx_power is None:
+            return None
+        return [float("{:.3f}".format(channel_power)) for channel_power in tx_power.values()]
+
+    def get_tx_power_support(self):
+        return not self.is_flat_memory()
+
+    def get_rx_power(self):
+        rx_power_support = self.get_rx_power_support()
+        if rx_power_support is None:
+            return None
+        if not rx_power_support:
+            return ["N/A" for _ in range(self.NUM_CHANNELS)]
+        rx_power = self.xcvr_eeprom.read(consts.RX_POWER_FIELD)
+        if rx_power is None:
+            return None
+        return [float("{:.3f}".format(channel_power)) for channel_power in rx_power.values()]
+
+    def get_rx_power_support(self):
+        return not self.is_flat_memory()
+
+    def get_tx_fault_support(self):
+        return not self.is_flat_memory() and self.xcvr_eeprom.read(consts.TX_FAULT_SUPPORT_FIELD)
+
+    def get_tx_fault(self):
+        tx_fault_support = self.get_tx_fault_support()
+        if tx_fault_support is None:
+            return None
+        if not tx_fault_support:
+            return ["N/A" for _ in range(self.NUM_CHANNELS)]
+        tx_fault = self.xcvr_eeprom.read(consts.TX_FAULT_FIELD)
+        if tx_fault is None:
+            return None
+        return [bool(tx_fault & (1 << i)) for i in range(self.NUM_CHANNELS)]
+
+    def get_tx_disable_support(self):
+        return not self.is_flat_memory() and self.xcvr_eeprom.read(consts.TX_DISABLE_SUPPORT_FIELD)
+
+    def get_tx_disable(self):
+        tx_disable_support = self.get_tx_disable_support()
+        if tx_disable_support is None:
+            return None
+        if not tx_disable_support:
+            return ["N/A" for _ in range(self.NUM_CHANNELS)]
+        tx_disable = self.xcvr_eeprom.read(consts.TX_DISABLE_FIELD)
+        if tx_disable is None:
+            return None
+        return [bool(tx_disable & (1 << i)) for i in range(self.NUM_CHANNELS)]
+
+    def tx_disable(self, tx_disable):
+        val = 0xFF if tx_disable else 0x0
+        return self.xcvr_eeprom.write(consts.TX_DISABLE_FIELD, val)
+
+    def get_tx_disable_channel(self):
+        tx_disable_support = self.get_tx_disable_support()
+        if tx_disable_support is None:
+            return None
+        if not tx_disable_support:
+            return 'N/A'
+        return self.xcvr_eeprom.read(consts.TX_DISABLE_FIELD)
+
+    def tx_disable_channel(self, channel, disable):
+        channel_state = self.get_tx_disable_channel()
+        if channel_state is None or channel_state == 'N/A':
+            return False
+
+        for i in range(self.NUM_CHANNELS):
+            mask = (1 << i)
+            if not (channel & mask):
+                continue
+            if disable:
+                channel_state |= mask
+            else:
+                channel_state &= ~mask
+
+        return self.xcvr_eeprom.write(consts.TX_DISABLE_FIELD, channel_state)
+
+    def get_transceiver_thresholds_support(self):
+        return not self.is_flat_memory()
+
+    def get_lpmode_support(self):
+        power_class = self.xcvr_eeprom.read(consts.POWER_CLASS_FIELD)
+        if power_class is None:
+            return False
+        return "Power Class 1" not in power_class
+
+    def get_power_override_support(self):
+        return False

sonic_platform_base/sonic_xcvr/api/public/sff8436.py

@@ -1,3 +1,10 @@
+"""
+    sff8436.py
+
+    Implementation of XcvrApi that corresponds to the SFF-8436 specification for
+    QSFP+ pluggable transceivers.
+"""
+
 from ...fields import consts
 from ..xcvr_api import XcvrApi
 
@@ -50,7 +57,7 @@ def get_transceiver_info(self):
             "ext_identifier": ", ".join([power_class, clei_code, cdr_tx, cdr_rx]),
             "ext_rateselect_compliance": serial_id[consts.EXT_RATE_SELECT_COMPLIANCE_FIELD],
             "cable_type": cable_type,
-            "cable_length": cable_len,
+            "cable_length": float(cable_len),
             "nominal_bit_rate": serial_id[consts.NOMINAL_BR_FIELD],
             "specification_compliance": str(serial_id[consts.SPEC_COMPLIANCE_FIELD]),
             "vendor_date": serial_id[consts.VENDOR_DATE_FIELD],

sonic_platform_base/sonic_xcvr/api/public/sff8636.py

@@ -1,3 +1,10 @@
+"""
+    sff8636.py
+
+    Implementation of XcvrApi that corresponds to the SFF-8636 specification for
+    QSFP28 pluggable transceivers.
+"""
+
 from ...fields import consts
 from ..xcvr_api import XcvrApi
 
@@ -57,7 +64,7 @@ def get_transceiver_info(self):
             "ext_identifier": ", ".join([power_class, clei_code, cdr_tx, cdr_rx]),
             "ext_rateselect_compliance": serial_id[consts.EXT_RATE_SELECT_COMPLIANCE_FIELD],
             "cable_type": cable_type,
-            "cable_length": cable_len,
+            "cable_length": float(cable_len),
             "nominal_bit_rate": serial_id[consts.NOMINAL_BR_FIELD],
             "specification_compliance": str(spec_compliance),
             "vendor_date": serial_id[consts.VENDOR_DATE_FIELD],

sonic_platform_base/sonic_xcvr/api/xcvr_api.py

@@ -58,7 +58,7 @@ def get_transceiver_info(self):
         encoding                   |string         |encoding information
         ext_identifier             |string         |extend identifier
         ext_rateselect_compliance  |string         |extended rateSelect compliance
-        cable_length               |int            |cable length in m
+        cable_length               |float          |cable length in m
         nominal_bit_rate           |int            |nominal bit rate by 100Mbs
         specification_compliance   |string         |specification compliance
         vendor_date                |string         |vendor date

sonic_platform_base/sonic_xcvr/codes/public/cmis.py

@@ -0,0 +1,22 @@
+from .sff8024 import Sff8024
+
+class CmisCodes(Sff8024):
+    POWER_CLASSES = {
+        0: "Power Class 1",
+        1: "Power Class 2",
+        2: "Power Class 3",
+        3: "Power Class 4",
+        4: "Power Class 5",
+        5: "Power Class 6",
+        6: "Power Class 7",
+        7: "Power Class 8"
+    }
+
+    MEDIA_TYPES = {
+        0: "Undefined",
+        1: "nm_850_media_interface",
+        2: "sm_media_interface",
+        3: "passive_copper_media_interface",
+        4: "active_cable_media_interface",
+        5: "base_t_media_interface",
+    }

sonic_platform_base/sonic_xcvr/fields/consts.py

@@ -109,13 +109,26 @@
 
 ADMIN_INFO_FIELD = "AdminInfo"
 
+CTRLS_ADVT_FIELD = "Supported Controls Advertisement"
+FLAGS_ADVT_FIELD = "Supported Flags Advertisement"
+
+LANE_DATAPATH_CTRL_FIELD = "Lane Control and Data Path Control"
+LANE_DATAPATH_STATUS_FIELD = "Lane Status and Data Path Status"
+LEN_MULT_FIELD = "LengthMultiplier"
+MAX_POWER_FIELD = "MaxPower"
+MEDIA_TYPE_FIELD = "Media Type"
+MGMT_CHAR_FIELD = "Management Characteristics"
+MGMT_CHAR_MISC_FIELD = "Management Characteristics (Misc)"
+
+MODULE_CHAR_ADVT_FIELD = "Module Characteristics Advertising"
+
 # C-CMIS
 
 ## Media Lane FEC Performance Monitoring
-MEDIA_LANE_FEC_PM_FIELD = "Media Lane FEC Performance Monitoring"
-MEDIA_LANE_LINK_PM_FIELD = "Media Lane Link Performance Monitoring"
 RX_BITS_PM_FIELD = "rxBitsPm"
 RX_BITS_SUB_INT_PM_FIELD = "rxBitsSubIntPm"
+MEDIA_LANE_FEC_PM_FIELD = "Media Lane FEC Performance Monitoring"
+MEDIA_LANE_LINK_PM_FIELD = "Media Lane Link Performance Monitoring"
 
 ## Media Lane Link Performance Monitoring
 RX_AVG_CD_PM_FIELD = "rxAvgCdPm"

sonic_platform_base/sonic_xcvr/fields/public/cmis.py

@@ -0,0 +1,13 @@
+from ..xcvr_field import NumberRegField
+from .. import consts
+
+class CableLenField(NumberRegField):
+    def __init__(self, name, offset, *fields, **kwargs):
+        super(CableLenField, self).__init__(name, offset, *fields, **kwargs)
+
+    def decode(self, raw_data, **decoded_deps):
+        base_len = super(CableLenField, self).decode(raw_data, **decoded_deps)
+        len_mult = decoded_deps.get(consts.LEN_MULT_FIELD)
+
+        mult = 10 ** (len_mult - 1)
+        return base_len * mult