add cpo bailly xcvr cmis

setup.py

@@ -43,14 +43,18 @@
         'sonic_platform_base.sonic_xcvr',
         'sonic_platform_base.sonic_xcvr.fields',
         'sonic_platform_base.sonic_xcvr.fields.public',
+        'sonic_platform_base.sonic_xcvr.fields.bailly',
         'sonic_platform_base.sonic_xcvr.mem_maps',
         'sonic_platform_base.sonic_xcvr.mem_maps.public',
+        'sonic_platform_base.sonic_xcvr.mem_maps.bailly',
         'sonic_platform_base.sonic_xcvr.mem_maps.public.cmis',
         'sonic_platform_base.sonic_xcvr.mem_maps.public.cmis.pages',
         'sonic_platform_base.sonic_xcvr.api',
         'sonic_platform_base.sonic_xcvr.api.public',
+        'sonic_platform_base.sonic_xcvr.api.bailly',
         'sonic_platform_base.sonic_xcvr.codes',
         '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.credo',
         'sonic_platform_base.sonic_xcvr.mem_maps.credo',

sonic_platform_base/sonic_xcvr/api/bailly/bailly_api.py

@@ -0,0 +1,105 @@
+"""
+    bailly_api.py
+
+    Implementation of Micas Bailly CPO specific in addition to the CMIS specification.
+"""
+from ..public.cmis import CmisApi
+from ...fields.bailly import bailly_consts
+
+class BaillyApi(CmisApi):
+    def __init__(self, xcvr_eeprom):
+        super(BaillyApi, self).__init__(xcvr_eeprom)
+
+    def get_dpinit_pending(self):
+        '''
+        Bailly not supported, return fake value.
+        '''
+        dpinit_pending_dict = {}
+        for lane in range(self.NUM_CHANNELS):
+            key = "DPInitPending{}".format(lane + 1)
+            dpinit_pending_dict[key] = True
+        return dpinit_pending_dict
+
+    def get_active_apsel_hostlane(self):
+        '''
+        Bailly not supported Deinit, if it is deinit return fake value.
+        '''
+        has_zero  = False
+        current_map = {}
+        for lane in range(self.NUM_CHANNELS):
+            lane_key = 'ActiveAppSelLane{}'.format(lane + 1)
+            app_lane = self.get_application(lane)
+            current_map[lane_key] = app_lane
+            if app_lane == 0:
+                has_zero = True
+
+        if has_zero:
+            return current_map
+        else:
+            normal =  super().get_active_apsel_hostlane()
+            return normal
+    def _format_revision(self, revision):
+        if revision is None:
+            return None
+        return "{}.{}".format((revision >> 4) & 0xf, revision & 0xf)
+
+    def get_transceiver_info(self):
+        info = super().get_transceiver_info()
+        if info is None:
+            return None
+
+        els_info = self.get_els_info()
+        cpo_info = els_info.get("cpo_info")
+        vendor_info = els_info.get("vendor_info")
+        laser_power_mode = els_info.get("laser_power_mode")
+        if cpo_info is None and vendor_info is None and laser_power_mode is None:
+            return info
+
+        if cpo_info is not None:
+            info.update({
+                "els_identifier": cpo_info.get(bailly_consts.CPO_IDENTIFIER),
+                "els_revision": self._format_revision(cpo_info.get(bailly_consts.CPO_REVISION)),
+                "els_laser_grid_and_count": cpo_info.get(bailly_consts.LASER_GRID_AND_COUNT),
+                "els_laser_wavelength_grid": cpo_info.get(bailly_consts.LASER_WAVELENGTH_GRID),
+                "els_laser_count": cpo_info.get(bailly_consts.LASER_COUNT),
+            })
+
+        if vendor_info is not None:
+            info.update({
+                "els_vendor_name": self._strip_str(
+                    vendor_info.get(bailly_consts.VENDOR_NAME_ASCII_FIELD)
+                ),
+                "els_vendor_oui": vendor_info.get(bailly_consts.VENDOR_OUI_HEX_FIELD),
+                "els_vendor_pn": self._strip_str(
+                    vendor_info.get(bailly_consts.VENDOR_PART_NUMBER_ASCII_FIELD)
+                ),
+                "els_vendor_rev": self._strip_str(
+                    vendor_info.get(bailly_consts.VENDOR_REVISION_ASCII_FIELD)
+                ),
+                "els_vendor_sn": self._strip_str(
+                    vendor_info.get(bailly_consts.VENDOR_SERIAL_NUMBER_ASCII_FIELD)
+                ),
+                "els_date_code": self._strip_str(
+                    vendor_info.get(bailly_consts.DATE_CODE_FIELD)
+                ),
+                "els_max_power": vendor_info.get(bailly_consts.MAX_POWER_CONSUMPTION_FIELD),
+            })
+
+        if laser_power_mode is not None:
+            info.update({
+                "els_laser_power_mode_control": laser_power_mode.get(
+                    bailly_consts.LASER_POWER_MODE_CONTROL_BITS_FIELD
+                ),
+            })
+
+        return info
+
+    def get_els_vendor_info(self):
+        return self.xcvr_eeprom.read(bailly_consts.CPO_VENDOR_INFO_FIELD)
+
+    def get_els_info(self):
+        return {
+            "cpo_info": self.xcvr_eeprom.read(bailly_consts.CPO_INFO_FIELD),
+            "vendor_info": self.get_els_vendor_info(),
+            "laser_power_mode": self.xcvr_eeprom.read(bailly_consts.LASER_POWER_MODE_CONTROL_FIELD),
+        }

sonic_platform_base/sonic_xcvr/bailly_optoe_base.py

@@ -0,0 +1,176 @@
+"""
+    cpo_optoe_base.py
+
+    Platform-independent class with which to interact with a cpo module
+    in SONiC
+"""
+import abc
+import os
+import json
+from .sfp_optoe_base import SfpOptoeBase
+from sonic_py_common.device_info import  get_platform,  get_path_to_platform_dir
+
+CPO_JSON_FILE = "cpo.json"
+def get_cpo_json_data():
+    """
+    Retrieve the data from cpo.json file
+
+    Returns:
+        A dictionary containing the key/value pairs as found in the cpo.json file
+    """
+    platform = get_platform()
+    if not platform:
+        return None
+
+    platform_path = get_path_to_platform_dir()
+    if not platform_path:
+        return None
+
+    platform_json = os.path.join(platform_path, CPO_JSON_FILE)
+    if not os.path.isfile(platform_json):
+        return None
+
+    try:
+        with open(platform_json, 'r') as f:
+            cpo_data = json.loads(f.read())
+            return cpo_data
+    except (json.JSONDecodeError, IOError, TypeError, ValueError):
+        return None
+
+class CpoOptoeBase(SfpOptoeBase):
+    def __init__(self):
+        super().__init__()
+        self._port_id = -1
+        self._oe_bank_id = -1
+        self._oe_id = -1
+        self._els_id = -1
+        self._els_bank_id = -1
+
+    def get_oe_eeprom_path(self):
+        oes_cfg = self.get_oes_config() or {}
+        cpo_bus = oes_cfg.get("oe_cmis_path")
+        return (cpo_bus + "eeprom") if cpo_bus else None
+
+    def get_eeprom_path(self):
+        return self.get_oe_eeprom_path()
+
+    def get_oes_config(self):
+        key = f"oe{self._oe_id}"
+        cpo_data = get_cpo_json_data() or {}
+        return (cpo_data.get("oes") or {}).get(key)
+
+    def get_elss_config(self):
+        key = f"els{self._els_id}"
+        cpo_data = get_cpo_json_data() or {}
+        return (cpo_data.get("elss") or {}).get(key)
+
+    def read_eeprom(self, offset, num_bytes):
+        sys_path = self.get_eeprom_path()
+        if not sys_path:
+            return None
+        try:
+            with open(sys_path, mode='rb', buffering=0) as f:
+                f.seek(offset)
+                return bytearray(f.read(num_bytes))
+        except (OSError, IOError, TypeError):
+            return None
+
+    def write_eeprom(self, offset, num_bytes, write_buffer):
+        sys_path = self.get_eeprom_path()
+        if not sys_path:
+            return False
+        try:
+            with open(sys_path, mode='r+b', buffering=0) as f:
+                f.seek(offset)
+                f.write(write_buffer[0:num_bytes])
+        except (OSError, IOError, TypeError):
+            return False
+        return True
+
+    def get_els_presence(self):
+        try:
+            elss_cfg = self.get_elss_config() or {}
+            els_presence = elss_cfg.get("els_presence") or {}
+
+            els_presence_file = els_presence.get("presence_file")
+            presence_offset = int(els_presence.get("presence_offset", "0"), 16)
+            presence_len = int(els_presence.get("presence_len", 0))
+            presence_bit = int(els_presence.get("presence_bit", 0))
+            presence_value = int(els_presence.get("presence_value", 1))
+
+            if not els_presence_file or presence_len <= 0:
+                return False
+
+            with open(els_presence_file, mode='rb', buffering=0) as f:
+                f.seek(presence_offset)
+                raw = bytearray(f.read(presence_len))
+                int_value = int.from_bytes(raw, byteorder='little')
+                return ((int_value >> presence_bit) & 1) == presence_value
+        except (OSError, IOError, TypeError, ValueError, AttributeError):
+            return False
+
+    def get_presence(self):
+        return self.get_els_presence()
+
+    def get_els_base_page(self):
+        elss_cfg = self.get_elss_config() or {}
+        return int(elss_cfg.get("base_page", 0))
+
+    def get_oe_bank_id(self):
+        return self._oe_bank_id
+    def get_oe_id(self):
+        return self._oe_id
+    def get_els_bank_id(self):
+        return self._els_bank_id
+    def get_els_id(self):
+        return self._els_id
+
+    @abc.abstractmethod
+    def check_fiber_dirty(self):
+        """
+        Check whether the fiber is dirty. True:check ok; False: check failed
+        """
+        raise NotImplementedError
+
+    @abc.abstractmethod
+    def check_calibration(self):
+        """
+        Check whether the calibration such as oe power sufficient. True:check ok; False: check failed
+        """
+        raise NotImplementedError
+
+    @abc.abstractmethod
+    def is_els_power_sufficient(self):
+        """
+        Check whether els power sufficient.
+        """
+        raise NotImplementedError
+
+    @abc.abstractmethod
+    def is_calibration_checked(self):
+        """
+        Check whether the calibration such as oe power sufficient detection​ has been completed.
+        """
+        raise NotImplementedError
+
+    @abc.abstractmethod
+    def is_fiber_checked(self):
+        """
+        Check whether the fiber detection​ has been completed.
+        """
+        raise NotImplementedError
+
+    @abc.abstractmethod
+    def is_els_tx_on(self):
+        """
+        Check the ELS TX​ status to see if it is emitting light normally.
+        """
+        raise NotImplementedError
+
+    @abc.abstractmethod
+    def is_els_tx_enabled(self):
+        """
+        Check whether the ELS TX enable​ has been set.
+        """
+        raise NotImplementedError
+

sonic_platform_base/sonic_xcvr/codes/bailly/bailly_codes.py

@@ -0,0 +1,66 @@
+from ..public.cmis import CmisCodes
+
+class BaillyCodes(CmisCodes):
+    # Vendor specific implementation to be added here
+    XCVR_IDENTIFIERS = {
+        **CmisCodes.XCVR_IDENTIFIERS,
+        128: 'CPO Bailly',
+    }
+
+    XCVR_IDENTIFIER_ABBRV = {
+        **CmisCodes.XCVR_IDENTIFIER_ABBRV,
+        128: 'QSFP-DD',
+    }
+
+    HOST_ELECTRICAL_INTERFACE = {
+        **CmisCodes.HOST_ELECTRICAL_INTERFACE,
+        253: 'Bailly-Reserved-1',
+        254: 'Bailly-Reserved-2',
+    }
+
+    SM_MEDIA_INTERFACE = {
+        **CmisCodes.SM_MEDIA_INTERFACE,
+        193: 'Bailly-800G-2xFR4',
+        253: 'Bailly-Reserved-LC-1',
+        254: 'Bailly-Reserved-LC-2',
+    }
+
+    LASER_WAVELENGTH_GRID = {
+        0: "CWDM4",
+        1: "DR4",
+    }
+
+    LASER_COUNT = {
+        code: code + 1 for code in range(16)
+    }
+
+    POWER_MODE = {
+        0: "High power mode",
+        1: "Low power mode",
+    }
+
+    INTERRUPT_STATUS = {
+        0: "Interrupt event occurred",
+        1: "Interrupt event cleared",
+    }
+
+    LASER_DISABLE_CONTROL = {
+        0: "Enable",
+        1: "Disable",
+    }
+
+    LASER_ACTIVE_STATUS = {
+        0: "Inactive",
+        1: "Active",
+    }
+
+    LASER_POWER_MODE_ENABLE = {
+        0: "Disable",
+        1: "Enable",
+    }
+
+    MAX_BANKS_SUPPORTED = {
+        **CmisCodes.MAX_BANKS_SUPPORTED,
+        3: 8,
+    }
+