CDB Refactor (#584)

Description
Refactor existing CDB support for firmware management to make it more modular and fungible to add newer CDB support beyond firmware management support

Motivation and Context
The Scope of CDB support for CMIS optics is going beyond just supporting optics firmware download. Support for VCS control via CDB is newly added support with many more features in future to be added via CDB protocol. This needs a refactor of existing CDB code to make it more modular and fungible to add newer CDB support beyond firmware management support

Author: prgeor

sonic_platform_base/sonic_xcvr/api/public/c_cmis.py

@@ -55,8 +55,8 @@
 })
 
 class CCmisApi(CmisApi):
-    def __init__(self, xcvr_eeprom):
-        super(CCmisApi, self).__init__(xcvr_eeprom)
+    def __init__(self, xcvr_eeprom, cdb):
+        super(CCmisApi, self).__init__(xcvr_eeprom, cdb)
 
     def _get_vdm_key_to_db_prefix_map(self):
         combined_map = {**CMIS_VDM_KEY_TO_DB_PREFIX_KEY_MAP, **C_CMIS_DELTA_VDM_KEY_TO_DB_PREFIX_KEY_MAP}

sonic_platform_base/sonic_xcvr/api/public/cmis.py

@@ -121,10 +121,14 @@ def set_cache_enabled(cls, enabled: bool):
         """
         cls.cache_enabled = bool(enabled)
 
-    def __init__(self, xcvr_eeprom):
+    def __init__(self, xcvr_eeprom, cdb_fw_hdlr=None):
         super(CmisApi, self).__init__(xcvr_eeprom)
         self.vdm = CmisVdmApi(xcvr_eeprom) if not self.is_flat_memory() else None
-        self.cdb = CmisCdbApi(xcvr_eeprom) if not self.is_flat_memory() else None
+        self.cdb = CmisCdbApi(xcvr_eeprom) if self.is_cdb_supported() else None
+        self.cdb_fw_hdlr = cdb_fw_hdlr if self.is_cdb_supported() else None
+
+    def get_cdb_fw_handler(self):
+        return self.cdb_fw_hdlr
 
     def _get_vdm_key_to_db_prefix_map(self):
         return CMIS_VDM_KEY_TO_DB_PREFIX_KEY_MAP
@@ -161,8 +165,8 @@ def _update_vdm_dict(self, dict_to_update, new_key, vdm_raw_dict, vdm_observable
     def freeze_vdm_stats(self):
         '''
         This function freeze all the vdm statistics reporting registers.
-        When raised by the host, causes the module to freeze and hold all 
-        reported statistics reporting registers (minimum, maximum and 
+        When raised by the host, causes the module to freeze and hold all
+        reported statistics reporting registers (minimum, maximum and
         average values)in Pages 24h-27h.
 
         Returns True if the provision succeeds and False incase of failure.
@@ -180,9 +184,9 @@ def get_vdm_freeze_status(self):
     def unfreeze_vdm_stats(self):
         '''
         This function unfreeze all the vdm statistics reporting registers.
-        When freeze is ceased by the host, releases the freeze request, allowing the 
+        When freeze is ceased by the host, releases the freeze request, allowing the
         reported minimum, maximum and average values to update again.
-        
+
         Returns True if the provision succeeds and False incase of failure.
         '''
         return self.xcvr_eeprom.write(consts.VDM_CONTROL, VDM_UNFREEZE)
@@ -370,7 +374,7 @@ def get_transceiver_info_firmware_versions(self):
             ( _, _, _, _, _, _, _, _, ActiveFirmware, InactiveFirmware) = result['result']
         except (ValueError, TypeError):
             return return_dict
-        
+
         return_dict["active_firmware"] = ActiveFirmware
         return_dict["inactive_firmware"] = InactiveFirmware
         return return_dict
@@ -1067,10 +1071,10 @@ def get_media_lane_count(self, appl=1):
         '''
         if self.is_flat_memory():
             return 0
-        
+
         if (appl <= 0):
             return 0
-        
+
         appl_advt = self.get_application_advertisement()
         return appl_advt[appl]['media_lane_count'] if len(appl_advt) >= appl else 0
 
@@ -1103,10 +1107,10 @@ def get_media_lane_assignment_option(self, appl=1):
         '''
         if self.is_flat_memory():
             return 'N/A'
-        
+
         if (appl <= 0):
             return 0
-        
+
         appl_advt = self.get_application_advertisement()
         return appl_advt[appl]['media_lane_assignment_options'] if len(appl_advt) >= appl else 0
 
@@ -1605,6 +1609,22 @@ def set_loopback_mode(self, loopback_mode, lane_mask = 0xff, enable = False):
         logger.error('Invalid loopback mode:%s, lane_mask:%#x', loopback_mode, lane_mask)
         return False
 
+    def is_cdb_supported(self):
+        '''
+        This function returns whether CDB is supported
+        '''
+        if self.is_flat_memory():
+            return False
+
+        cdb_inst = self.xcvr_eeprom.read(consts.CDB_SUPPORT)
+        if cdb_inst is None:
+            return False
+
+        if cdb_inst == 1 or cdb_inst == 2:
+            return True
+
+        return False
+
     def is_transceiver_vdm_supported(self):
         '''
         This function returns whether VDM is supported
@@ -2346,7 +2366,7 @@ def get_transceiver_vdm_real_value(self):
         biasyq{lane_num}                               = FLOAT                  ; modulator bias yq in percentage
         biasyp{lane_num}                               = FLOAT                  ; modulator bias yq in percentage
         cdshort{lane_num}                              = FLOAT                  ; chromatic dispersion, high granularity, short link in ps/nm
-        cdlong{lane_num}                               = FLOAT                  ; chromatic dispersion, high granularity, long link in ps/nm  
+        cdlong{lane_num}                               = FLOAT                  ; chromatic dispersion, high granularity, long link in ps/nm
         dgd{lane_num}                                  = FLOAT                  ; differential group delay in ps
         sopmd{lane_num}                                = FLOAT                  ; second order polarization mode dispersion in ps^2
         soproc{lane_num}                               = FLOAT                  ; state of polarization rate of change in krad/s
@@ -2375,7 +2395,7 @@ def get_transceiver_vdm_thresholds(self):
         Returns:
             A dict containing the following keys/values :
         ========================================================================
-        xxx refers to HALARM/LALARM/HWARN/LWARN threshold 
+        xxx refers to HALARM/LALARM/HWARN/LWARN threshold
         ;Defines Transceiver VDM high/low alarm/warning threshold for a port
         key                                            = TRANSCEIVER_VDM_XXX_THRESHOLD|ifname    ; information module VDM high/low alarm/warning threshold on port
         ; field                                        = value
@@ -2698,7 +2718,7 @@ def decommission_all_datapaths(self):
             name = "DP{}State".format(lane + 1)
             if dp_state[name] != 'DataPathDeactivated':
                 return False
-            
+
             name = "ConfigStatusLane{}".format(lane + 1)
             if config_state[name] != 'ConfigSuccess':
                 return False

sonic_platform_base/sonic_xcvr/cdb/cdb.py

@@ -0,0 +1,160 @@
+"""
+   cdb.py
+
+   CDB Command handler
+"""
+
+import time
+from ..fields import cdb_consts
+from ..xcvr_eeprom import XcvrEeprom
+
+class CdbCmdHandler(XcvrEeprom):
+    def __init__(self, reader, writer, mem_map):
+        super(CdbCmdHandler, self).__init__(reader, writer, mem_map)
+
+    def read_reply(self, cdb_cmd_id):
+        """
+        Read a reply from the CDB
+        """
+        cdb_cmd = self.mem_map.get_cdb_cmd(cdb_cmd_id)
+        reply_field = cdb_cmd.get_reply_field()
+        if reply_field is not None:
+            return self.read(reply_field)
+        return None
+
+    def write_cmd(self, cdb_cmd_id, payload=None):
+        """
+        Write CDB command
+        """
+        cdb_cmd = self.mem_map.get_cdb_cmd(cdb_cmd_id)
+        if payload is not None:
+            bytes = cdb_cmd.encode(payload)
+        else:
+            bytes = cdb_cmd.encode()
+        # TODO Check the module capability CdbCommandTriggerMethod to write in single I2C transaction
+        # Write the bytes starting from the 3rd byte(0x9F:130)
+        self.writer(cdb_cmd.getaddr() + 2, len(bytes) - 2, bytes[2:])
+        # Finally write the first two CMD bytes to trigger CDB processing
+        return self.writer(cdb_cmd.getaddr(), 2, bytes[:2])
+    
+    def write_epl_page(self, page, data):
+        """
+        Write a page of data to the EPL page
+        """
+        # Write the data to the specified page and offset
+        assert len(data) <= cdb_consts.PAGE_SIZE, \
+                                    "Data length exceeds page size"
+        assert page >= cdb_consts.EPL_PAGE, \
+                    "Page number must be greater than or equal to 0xA0"
+        return self.write_raw((page * cdb_consts.PAGE_SIZE) + 128, len(data), data)
+
+
+    def wait_for_cdb_status(self, timeout=None):
+        """
+        Wait for CDB status to be ready
+        Returns False if failed to get the status
+        True otherwise
+        """
+        delay = 0
+        if timeout is None:
+            timeout = cdb_consts.CDB_MAX_ACCESS_HOLD_OFF_PERIOD  + 5000  # 5 sec safety margin
+        status = None
+
+        assert timeout > delay, "Timeout must be greater than delay"
+
+        while (delay < timeout):
+            time.sleep(cdb_consts.CDB_MAX_CAPTURE_TIME / 1000)
+            delay += cdb_consts.CDB_MAX_CAPTURE_TIME
+
+            status = self.read(cdb_consts.CDB1_CMD_STATUS)
+            if (status is None) or \
+                    (True == status[cdb_consts.CDB1_IS_BUSY]):
+                continue
+
+            if (True == status[cdb_consts.CDB1_HAS_FAILED]):
+                break
+
+            if (False == status[cdb_consts.CDB1_IS_BUSY]) and \
+                    (False == status[cdb_consts.CDB1_HAS_FAILED]):
+                break
+
+        if delay >= timeout or status is None:
+            return [False, status]
+
+        return [True, status]
+
+    def send_cmd(self, cdb_cmd_id, payload=None, timeout=None):
+        """
+        Send CDB command, wait for completion and check status
+        """
+        # Write the command to the CDB
+        if True != self.write_cmd(cdb_cmd_id, payload):
+            print(f"Failed to write CDB command: {cdb_cmd_id}")
+            return None
+
+        # Wait for the command to complete
+        ret, status = self.wait_for_cdb_status(timeout)
+        if not ret:
+            print(f"CDB command: {cdb_cmd_id} failed to complete or read status")
+            return None
+
+        is_busy = status[cdb_consts.CDB1_IS_BUSY]
+        if True == is_busy:
+            print(f"CDB command: {cdb_cmd_id} is busy with status: {status[cdb_consts.CDB1_STATUS]}")
+            return False
+
+        is_failed = status[cdb_consts.CDB1_HAS_FAILED]
+        if True == is_failed:
+            print(f"CDB command: {cdb_cmd_id} failed with status: {status[cdb_consts.CDB1_STATUS]}")
+            return False
+
+        return status[cdb_consts.CDB1_STATUS] == 0x1
+
+    def get_last_cmd_status(self):
+        """
+        Get the status of the last CDB command
+        Returns None if Module failed to reply to I2C command
+        """
+        status = self.read(cdb_consts.CDB1_COMMAND_RESULT)
+        return status
+    
+    def write_lpl_block(self, blkaddr, blkdata):
+        """
+        Write LPL block
+        """
+        payload = {
+            "blkaddr" : blkaddr,
+            "blkdata" : blkdata
+        }
+        # Send the CDB write firmware LPL command
+        if True != self.write_cmd(cdb_consts.CDB_WRITE_FIRMWARE_LPL_CMD, payload):
+            status = self.get_last_cmd_status()
+            print(f"Write LPL block status: {status}")
+
+    def write_epl_pages(self, blkdata):
+        """
+        Write EPL pages starting from page 0xA0
+        """
+        pages = len(blkdata) // cdb_consts.PAGE_SIZE
+        assert pages <= cdb_consts.EPL_MAX_PAGES, "Data exceeds maximum number of EPL pages"
+
+        for page in range(pages):
+            page_data = blkdata[page * cdb_consts.PAGE_SIZE : (page + 1) * cdb_consts.PAGE_SIZE]
+            assert True == self.write_epl_page(page + cdb_consts.EPL_PAGE, page_data)
+
+        # Handle any remaining data that doesn't fit into a full page
+        if len(blkdata) % cdb_consts.PAGE_SIZE != 0:
+            remaining_data = blkdata[pages * cdb_consts.PAGE_SIZE:]
+            assert True == self.write_epl_page(pages + cdb_consts.EPL_PAGE, remaining_data)
+
+    def write_epl_block(self, blkaddr, blkdata):
+        """
+        Write EPL block
+        """
+        payload = {
+            "blkaddr" : blkaddr,
+            "blkdata" : blkdata
+        }
+
+        # Send the CDB write firmware EPL command
+        return self.send_cmd(cdb_consts.CDB_WRITE_FIRMWARE_EPL_CMD, payload)