Support for Custom Amphenol 800G Backplane catridge (sonic-net#555)

* Support for Custom Amphenol 800G Backplane catridge

* Address comment

* Fix xcvrd crash

* Fix the crash

* Improve code coverage

Author: prgeor

sonic_platform_base/sonic_xcvr/api/amphenol/__init__.py

@@ -0,0 +1,81 @@
+"""
+    Custom API for Amphenol 800G Backplane Cartridge
+"""
+import copy
+from ...fields import consts
+from ..public.cmis import CmisApi
+
+AMPH_BACKPLANE_INFO_DICT = {
+        "type": "N/A",
+        "type_abbrv_name": "N/A",
+        "hardware_rev": "N/A",
+        "serial": "N/A",
+        "cable_length": "N/A",
+        "manufacturer": "N/A", # Vendor Name
+        "model": "N/A", # Vendor Part Number
+        "vendor_date": "N/A",
+        "vendor_oui": "N/A",
+        "vendor_rev": "N/A",
+        "application_advertisement": "N/A",
+        "host_electrical_interface": "N/A",
+        "host_lane_count": "N/A",
+        "cable_type": "N/A",
+        "cmis_rev": "N/A",
+        "specification_compliance": "N/A",
+        "slot_id": "unknown"
+        }
+
+class AmphBackplaneImpl(CmisApi):
+    """
+    Custom API for Amphenol 800G Backplane Catridge
+    """
+    def __init__(self, xcvr_eeprom):
+        super(AmphBackplaneImpl, self).__init__(xcvr_eeprom)
+
+    def get_slot_id(self):
+        """
+        Get the slot id
+        """
+        slot = self.xcvr_eeprom.read(consts.CARTRDIGE_SLOT_ID)
+        if slot is None:
+            return "N/A"
+        return slot
+
+    def get_transceiver_info(self):
+        admin_info = self.xcvr_eeprom.read(consts.ADMIN_INFO_FIELD)
+        if admin_info is None:
+            return None
+
+        xcvr_info = copy.deepcopy(AMPH_BACKPLANE_INFO_DICT)
+        xcvr_info.update({
+            "type": admin_info[consts.ID_FIELD],
+            "type_abbrv_name": admin_info[consts.ID_ABBRV_FIELD],
+            "hardware_rev": self.get_module_hardware_revision(),
+            "cable_length": float(admin_info[consts.LENGTH_ASSEMBLY_FIELD]),
+            "application_advertisement": str(self.get_application_advertisement()) \
+                        if len(self.get_application_advertisement()) > 0 else 'N/A',
+            "host_electrical_interface": self.get_host_electrical_interface(),
+            "host_lane_count": self.get_host_lane_count(),
+            "host_lane_assignment_option": self.get_host_lane_assignment_option(),
+            "cable_type": self.get_cable_length_type(),
+            "cmis_rev": self.get_cmis_rev(),
+            "specification_compliance": self.get_module_media_type(),
+            "vdm_supported": self.is_transceiver_vdm_supported()
+        })
+
+        appl_adv = self.get_application_advertisement()
+        if len(appl_adv) > 0:
+            xcvr_info["application_advertisement"] = str(appl_adv)
+
+        # Vendor specific fields
+        xcvr_info.update({
+            "serial": admin_info[consts.VENDOR_SERIAL_NO_FIELD],
+            "manufacturer": admin_info[consts.VENDOR_NAME_FIELD],
+            "model": admin_info[consts.VENDOR_PART_NO_FIELD],
+            "vendor_date": admin_info[consts.VENDOR_DATE_FIELD],
+            "vendor_oui": admin_info[consts.VENDOR_OUI_FIELD],
+            "vendor_rev": self.get_vendor_rev(),
+            "slot_id": self.get_slot_id(),
+        })
+        return xcvr_info
+

sonic_platform_base/sonic_xcvr/api/amphenol/backplane.py

@@ -0,0 +1,9 @@
+from ..public.cmis import CmisCodes
+
+class AmphBackplaneCodes(CmisCodes):
+    CARTRDIGE_SLOT_ID = {
+        0x1 : "1",
+        0x2 : "2",
+        0x3 : "3",
+        0x4 : "4"
+    }

sonic_platform_base/sonic_xcvr/codes/amphenol/__init__.py

@@ -38,7 +38,8 @@ class Sff8024(XcvrCodes):
         27: 'DSFP Dual Small Form Factor Pluggable Transceiver',
         28: 'x4 MiniLink/OcuLink',
         29: 'x8 MiniLink',
-        30: 'QSFP+ or later with CMIS'
+        30: 'QSFP+ or later with CMIS',
+       126: 'Backplane Cartridge',
     }
 
     XCVR_IDENTIFIER_ABBRV = {

sonic_platform_base/sonic_xcvr/codes/amphenol/backplane.py

@@ -496,3 +496,4 @@
 #VENDOR SPECIFIC
 VENDOR_CUSTOM = "VendorCustom"
 TARGET_MODE = "TargetMode"
+CARTRDIGE_SLOT_ID = "CartridgeSlotID"

sonic_platform_base/sonic_xcvr/codes/public/sff8024.py

@@ -0,0 +1,18 @@
+"""
+    Custom memory map for Amphenol 800G Backplane connector
+"""
+
+from ...mem_maps.public.cmis import CmisFlatMemMap
+from ...fields.xcvr_field import (
+    NumberRegField,
+    RegGroupField,
+    CodeRegField
+)
+from ...fields import consts
+
+class AmphBackplaneMemMap(CmisFlatMemMap):
+    def __init__(self, codes):
+        super(AmphBackplaneMemMap, self).__init__(codes)
+
+        self.SLOT_ID = CodeRegField(consts.CARTRDIGE_SLOT_ID, self.getaddr(0x0, 203),
+                                                self.codes.CARTRDIGE_SLOT_ID)

sonic_platform_base/sonic_xcvr/fields/consts.py

@@ -18,17 +18,20 @@
 from ...fields import consts
 from ...fields.public.cmis import CableLenField
 
-class CmisMemMap(XcvrMemMap):
+class CmisFlatMemMap(XcvrMemMap):
+    """
+    Memory map for CMIS flat memory (Lower page and Upper page 0h ONLY)
+    """
     def __init__(self, codes):
-        super(CmisMemMap, self).__init__(codes)
+        super(CmisFlatMemMap, self).__init__(codes)
 
         self.MGMT_CHARACTERISTICS = RegGroupField(consts.MGMT_CHAR_FIELD,
             NumberRegField(consts.MGMT_CHAR_MISC_FIELD, self.getaddr(0x0, 2),
                 RegBitField(consts.FLAT_MEM_FIELD, 7)
             )
         )
 
-        # Should contain ONLY Lower page fields
+        # This memmap should contain ONLY Lower page 00h and upper page 00h fields
         self.ADMIN_INFO = RegGroupField(consts.ADMIN_INFO_FIELD,
             CodeRegField(consts.ID_FIELD, self.getaddr(0x0, 0), self.codes.XCVR_IDENTIFIERS),
             CodeRegField(consts.ID_ABBRV_FIELD, self.getaddr(0x0, 128), self.codes.XCVR_IDENTIFIER_ABBRV),
@@ -109,7 +112,14 @@ def __init__(self, codes):
             NumberRegField(consts.ACTIVE_FW_MINOR_REV, self.getaddr(0x0, 40), format="B", size=1),
         )
 
-        # Should contain ONLY upper page fields
+    def getaddr(self, page, offset, page_size=128):
+        return page * page_size + offset
+
+class CmisMemMap(CmisFlatMemMap):
+    def __init__(self, codes):
+        super(CmisMemMap, self).__init__(codes)
+
+        # This memmap should contain ONLY upper page >= 01h fields
         self.ADVERTISING = RegGroupField(consts.ADVERTISING_FIELD,
             NumberRegField(consts.INACTIVE_FW_MAJOR_REV, self.getaddr(0x1, 128), format="B", size=1),
             NumberRegField(consts.INACTIVE_FW_MINOR_REV, self.getaddr(0x1, 129), format="B", size=1),

sonic_platform_base/sonic_xcvr/mem_maps/amphenol/__init__.py

@@ -19,6 +19,10 @@
 
 from .api.innolight.fr_800g import CmisFr800gApi
 
+from .api.amphenol.backplane import AmphBackplaneImpl
+from .mem_maps.amphenol.backplane import AmphBackplaneMemMap
+from .codes.amphenol.backplane import AmphBackplaneCodes
+
 from .codes.public.sff8436 import Sff8436Codes
 from .api.public.sff8436 import Sff8436Api
 from .mem_maps.public.sff8436 import Sff8436MemMap
@@ -70,60 +74,59 @@ def _get_vendor_part_num(self):
            return None
        vendor_pn = part_num.decode()
        return vendor_pn.strip()
-        
-    def create_xcvr_api(self):
-        # TODO: load correct classes from id_mapping file
-        id = self._get_id()
-        # QSFP-DD or OSFP
-        if id == 0x18 or id == 0x19 or id == 0x1e:
-            vendor_name = self._get_vendor_name()
-            vendor_pn = self._get_vendor_part_num()
-            if vendor_name == 'Credo' and vendor_pn in CREDO_800G_AEC_VENDOR_PN_LIST:
-                codes = CmisAec800gCodes
-                mem_map = CmisAec800gMemMap(CmisAec800gCodes)
-                xcvr_eeprom = XcvrEeprom(self.reader, self.writer, mem_map)
-                api = CmisAec800gApi(xcvr_eeprom)
-            elif ('INNOLIGHT' in vendor_name and vendor_pn in INL_800G_VENDOR_PN_LIST) or \
-                 ('EOPTOLINK' in vendor_name and vendor_pn in EOP_800G_VENDOR_PN_LIST):
-                codes = CmisCodes
-                mem_map = CmisMemMap(codes)
-                xcvr_eeprom = XcvrEeprom(self.reader, self.writer, mem_map)
-                api = CmisFr800gApi(xcvr_eeprom)
-            else:
-                codes = CmisCodes
-                mem_map = CmisMemMap(codes)
-                xcvr_eeprom = XcvrEeprom(self.reader, self.writer, mem_map)
-                api = CmisApi(xcvr_eeprom)
 
-            if api.is_coherent_module():
-                mem_map = CCmisMemMap(codes)
-                xcvr_eeprom = XcvrEeprom(self.reader, self.writer, mem_map)
-                api = CCmisApi(xcvr_eeprom)
-
-        # QSFP28
-        elif id == 0x11:
-            codes = Sff8636Codes
-            mem_map = Sff8636MemMap(codes)
-            xcvr_eeprom = XcvrEeprom(self.reader, self.writer, mem_map)
-            api = Sff8636Api(xcvr_eeprom)
-        # QSFP+
-        elif id == 0x0D:
-            revision_compliance = self._get_revision_compliance()
-            if revision_compliance >= 3:
-                codes = Sff8636Codes
-                mem_map = Sff8636MemMap(codes)
-                xcvr_eeprom = XcvrEeprom(self.reader, self.writer, mem_map)
-                api = Sff8636Api(xcvr_eeprom)
-            else:
-                codes = Sff8436Codes
-                mem_map = Sff8436MemMap(codes)
-                xcvr_eeprom = XcvrEeprom(self.reader, self.writer, mem_map)
-                api = Sff8436Api(xcvr_eeprom)
-        elif id == 0x03:
-            codes = Sff8472Codes
-            mem_map = Sff8472MemMap(codes)
-            xcvr_eeprom = XcvrEeprom(self.reader, self.writer, mem_map)
-            api = Sff8472Api(xcvr_eeprom)
+ def _create_cmis_api(self):
+        api = None
+        vendor_name = self._get_vendor_name()
+        vendor_pn = self._get_vendor_part_num()
+
+        if vendor_name == 'Credo' and vendor_pn in CREDO_800G_AEC_VENDOR_PN_LIST:
+            api = self._create_api(CmisAec800gCodes, CmisAec800gMemMap, CmisAec800gApi)
+        elif ('INNOLIGHT' in vendor_name and vendor_pn in INL_800G_VENDOR_PN_LIST) or \
+             ('EOPTOLINK' in vendor_name and vendor_pn in EOP_800G_VENDOR_PN_LIST):
+            api = self._create_api(CmisCodes, CmisMemMap, CmisFr800gApi)
+
         else:
-            api = None
+            api = self._create_api(CmisCodes, CmisMemMap, CmisApi)
+            if api.is_coherent_module():
+                api = self._create_api(CmisCodes, CCmisMemMap, CCmisApi)
         return api
+
+    def _create_qsfp_api(self):
+        """
+        QSFP/QSFP+ API implementation
+        """
+        revision_compliance = self._get_revision_compliance()
+        if revision_compliance >= 3:
+            return self._create_api(Sff8636Codes, Sff8636MemMap, Sff8636Api)
+        else:
+            return self._create_api(Sff8436Codes, Sff8436MemMap, Sff8436Api)
+
+    def _create_api(self, codes_class, mem_map_class, api_class):
+        codes = codes_class
+        mem_map = mem_map_class(codes)
+        xcvr_eeprom = XcvrEeprom(self.reader, self.writer, mem_map)
+        return api_class(xcvr_eeprom) 
+
+    def create_xcvr_api(self):
+        id = self._get_id()
+
+        # Instantiate various Optics implementation based upon their respective ID as per SFF8024
+        id_mapping = {
+            0x03: (self._create_api, (Sff8472Codes, Sff8472MemMap, Sff8472Api)),
+            0x0D: (self._create_qsfp_api, ()),
+            0x11: (self._create_api, (Sff8636Codes, Sff8636MemMap, Sff8636Api)),
+            0x18: (self._create_cmis_api, ()),
+            0x19: (self._create_cmis_api, ()),
+            0x1b: (self._create_cmis_api, ()),
+            0x1e: (self._create_cmis_api, ()),
+            0x7e: (self._create_api, (AmphBackplaneCodes,
+                                     AmphBackplaneMemMap, AmphBackplaneImpl)),
+        }
+
+        # Check if the ID exists in the mapping
+        if id in id_mapping:
+            func, args = id_mapping[id]
+            if isinstance(args, tuple):
+                return func(*args)
+        return None

sonic_platform_base/sonic_xcvr/mem_maps/amphenol/backplane.py

@@ -54,7 +54,6 @@ def read_raw(self, offset, size, return_raw = False):
             data = struct.unpack("%dB" %size, raw_data)
       return data
 
-
    def write(self, field_name, value):
       """
       Write a value to a field in EEPROM
@@ -90,4 +89,4 @@ def write_raw(self, offset, size, bytearray_data):
       Returns:
          Boolean, True if the write is successful and False otherwise
       """
-      return self.writer(offset, size, bytearray_data)
+      return self.writer(offset, size, bytearray_data)