[cmis] combine cmis eeprom i2c read (#84)

Change multiple eeprom i2c reads into a single i2c read across multiple addresses, as addresses are continuous in register group.
<!-- Provide a general summary of your changes in the Title above -->

#### Description
<!--
 Describe your changes in detail
-->
1. Update functions get_tx_power_flag, get_tx_bias_flag and get_rx_power_flag.
2. Add dedicated groups: TX_POWER_ALARM_FLAGS_FIELD, TX_BIAS_ALARM_FLAGS_FIELD, RX_POWER_ALARM_FLAGS_FIELD.
3. Move the MEDIA_TYPE_FIELD read and prefix lookup out of the loop.

#### Motivation and Context
CMIS API methods were issuing several separate i2c transactions per call, resulting in high bus overhead, increased latency, and poor scalability. By introducing dedicated RegGroupField groups for TX/RX power and bias alarm flags (e.g. TX_POWER_ALARM_FLAGS, TX_BIAS_ALARM_FLAGS, RX_POWER_ALARM_FLAGS) and refactoring the related methods (get_tx_power_flag, get_tx_bias_flag, get_rx_power_flag, etc.) to use a single block read per group, it resolves API call latency. Also moved constant prefix lookups (media‐type prefix) outside of loops to save time.

#### How Has This Been Tested?
Test with performance cases for one port, if there are 64 physical ports, then it can save 1.3s.
Before:
```bash
Running performance tests (3000 iterations each)...
get_tx_power_flag: total 32.120s, avg 10.707ms per call
get_tx_bias_flag: total 26.635s, avg 8.878ms per call
get_rx_power_flag: total 32.306s, avg 10.769ms per call
```
After:
```bash
Running performance tests (3000 iterations each)...
get_tx_power_flag: total 8.213s, avg 2.738ms per call
get_tx_bias_flag: total 7.831s, avg 2.610ms per call
get_rx_power_flag: total 9.632s, avg 3.211ms per call
```
Test with config reload, works fine.

Author: mssonicbld

sonic_platform_base/sonic_xcvr/api/public/cmis.py

@@ -636,77 +636,45 @@ def get_rx_cdr_lol(self):
             rx_cdr_lol_final.append(bool(rx_cdr_lol[key]))
         return rx_cdr_lol_final
 
+    def get_alarm_flags(self, alarm_flag):
+        '''Generic helper to return alarm and warning flags for given type: TX_POWER, TX_BIAS, RX_POWER.'''
+        flags = self.xcvr_eeprom.read(getattr(consts, f"{alarm_flag}_ALARM_FLAGS_FIELD"))
+        if flags is None:
+            return None
+        high_alarm = flags.get(getattr(consts, f"{alarm_flag}_HIGH_ALARM_FLAG"))
+        low_alarm = flags.get(getattr(consts, f"{alarm_flag}_LOW_ALARM_FLAG"))
+        high_warn = flags.get(getattr(consts, f"{alarm_flag}_HIGH_WARN_FLAG"))
+        low_warn = flags.get(getattr(consts, f"{alarm_flag}_LOW_WARN_FLAG"))
+        if high_alarm is None or low_alarm is None or high_warn is None or low_warn is None:
+            return None
+        for d in (high_alarm, low_alarm, high_warn, low_warn):
+            for key, value in d.items():
+                d[key] = bool(value)
+        prefix = alarm_flag.lower()
+        return {
+            f"{prefix}_high_alarm": high_alarm,
+            f"{prefix}_low_alarm": low_alarm,
+            f"{prefix}_high_warn": high_warn,
+            f"{prefix}_low_warn": low_warn,
+        }
+
     def get_tx_power_flag(self):
         '''
         This function returns TX power out of range flag on TX media lane
         '''
-        tx_power_high_alarm_dict = self.xcvr_eeprom.read(consts.TX_POWER_HIGH_ALARM_FLAG)
-        tx_power_low_alarm_dict = self.xcvr_eeprom.read(consts.TX_POWER_LOW_ALARM_FLAG)
-        tx_power_high_warn_dict = self.xcvr_eeprom.read(consts.TX_POWER_HIGH_WARN_FLAG)
-        tx_power_low_warn_dict = self.xcvr_eeprom.read(consts.TX_POWER_LOW_WARN_FLAG)
-        if tx_power_high_alarm_dict is None or tx_power_low_alarm_dict is None or tx_power_high_warn_dict is None or tx_power_low_warn_dict is None:
-            return None
-        for key, value in tx_power_high_alarm_dict.items():
-            tx_power_high_alarm_dict[key] = bool(value)
-        for key, value in tx_power_low_alarm_dict.items():
-            tx_power_low_alarm_dict[key] = bool(value)
-        for key, value in tx_power_high_warn_dict.items():
-            tx_power_high_warn_dict[key] = bool(value)
-        for key, value in tx_power_low_warn_dict.items():
-            tx_power_low_warn_dict[key] = bool(value)
-        tx_power_flag_dict = {'tx_power_high_alarm': tx_power_high_alarm_dict,
-                              'tx_power_low_alarm': tx_power_low_alarm_dict,
-                              'tx_power_high_warn': tx_power_high_warn_dict,
-                              'tx_power_low_warn': tx_power_low_warn_dict,}
-        return tx_power_flag_dict
+        return self.get_alarm_flags("TX_POWER")
 
     def get_tx_bias_flag(self):
         '''
         This function returns TX bias out of range flag on TX media lane
         '''
-        tx_bias_high_alarm_dict = self.xcvr_eeprom.read(consts.TX_BIAS_HIGH_ALARM_FLAG)
-        tx_bias_low_alarm_dict = self.xcvr_eeprom.read(consts.TX_BIAS_LOW_ALARM_FLAG)
-        tx_bias_high_warn_dict = self.xcvr_eeprom.read(consts.TX_BIAS_HIGH_WARN_FLAG)
-        tx_bias_low_warn_dict = self.xcvr_eeprom.read(consts.TX_BIAS_LOW_WARN_FLAG)
-        if tx_bias_high_alarm_dict is None or tx_bias_low_alarm_dict is None or tx_bias_high_warn_dict is None or tx_bias_low_warn_dict is None:
-            return None
-        for key, value in tx_bias_high_alarm_dict.items():
-            tx_bias_high_alarm_dict[key] = bool(value)
-        for key, value in tx_bias_low_alarm_dict.items():
-            tx_bias_low_alarm_dict[key] = bool(value)
-        for key, value in tx_bias_high_warn_dict.items():
-            tx_bias_high_warn_dict[key] = bool(value)
-        for key, value in tx_bias_low_warn_dict.items():
-            tx_bias_low_warn_dict[key] = bool(value)
-        tx_bias_flag_dict = {'tx_bias_high_alarm': tx_bias_high_alarm_dict,
-                              'tx_bias_low_alarm': tx_bias_low_alarm_dict,
-                              'tx_bias_high_warn': tx_bias_high_warn_dict,
-                              'tx_bias_low_warn': tx_bias_low_warn_dict,}
-        return tx_bias_flag_dict
+        return self.get_alarm_flags("TX_BIAS")
 
     def get_rx_power_flag(self):
         '''
         This function returns RX power out of range flag on RX media lane
         '''
-        rx_power_high_alarm_dict = self.xcvr_eeprom.read(consts.RX_POWER_HIGH_ALARM_FLAG)
-        rx_power_low_alarm_dict = self.xcvr_eeprom.read(consts.RX_POWER_LOW_ALARM_FLAG)
-        rx_power_high_warn_dict = self.xcvr_eeprom.read(consts.RX_POWER_HIGH_WARN_FLAG)
-        rx_power_low_warn_dict = self.xcvr_eeprom.read(consts.RX_POWER_LOW_WARN_FLAG)
-        if rx_power_high_alarm_dict is None or rx_power_low_alarm_dict is None or rx_power_high_warn_dict is None or rx_power_low_warn_dict is None:
-            return None
-        for key, value in rx_power_high_alarm_dict.items():
-            rx_power_high_alarm_dict[key] = bool(value)
-        for key, value in rx_power_low_alarm_dict.items():
-            rx_power_low_alarm_dict[key] = bool(value)
-        for key, value in rx_power_high_warn_dict.items():
-            rx_power_high_warn_dict[key] = bool(value)
-        for key, value in rx_power_low_warn_dict.items():
-            rx_power_low_warn_dict[key] = bool(value)
-        rx_power_flag_dict = {'rx_power_high_alarm': rx_power_high_alarm_dict,
-                              'rx_power_low_alarm': rx_power_low_alarm_dict,
-                              'rx_power_high_warn': rx_power_high_warn_dict,
-                              'rx_power_low_warn': rx_power_low_warn_dict,}
-        return rx_power_flag_dict
+        return self.get_alarm_flags("RX_POWER")
 
     def get_tx_output_status(self):
         '''

sonic_platform_base/sonic_xcvr/fields/consts.py

@@ -291,6 +291,9 @@
 LANE_MON_ADVT_FIELD = "Supported Lane Monitor Advertisement"
 LANE_DATAPATH_CTRL_FIELD = "Lane Control and Data Path Control"
 LANE_DATAPATH_STATUS_FIELD = "Lane Status and Data Path Status"
+TX_POWER_ALARM_FLAGS_FIELD = "TxPowerAlarmFlags"
+TX_BIAS_ALARM_FLAGS_FIELD = "TxBiasAlarmFlags"
+RX_POWER_ALARM_FLAGS_FIELD = "RxPowerAlarmFlags"
 DATAPATH_DEINIT_FIELD = "Data Path Deinit"
 LEN_MULT_FIELD = "LengthMultiplier"
 MAX_POWER_FIELD = "MaxPower"

sonic_platform_base/sonic_xcvr/mem_maps/public/cmis.py

@@ -293,6 +293,78 @@ def __init__(self, codes):
             )
         )
 
+        # Group for TX power alarm and warning flags
+        self.TX_POWER_ALARM_FLAGS = RegGroupField(consts.TX_POWER_ALARM_FLAGS_FIELD,
+            RegGroupField(consts.TX_POWER_HIGH_ALARM_FLAG,
+                *(NumberRegField("%s%d" % (consts.TX_POWER_HIGH_ALARM_FLAG, lane), self.getaddr(0x11, 139),
+                    RegBitField("Bit%d" % (lane-1), (lane-1))
+                ) for lane in range(1, 9))
+            ),
+            RegGroupField(consts.TX_POWER_LOW_ALARM_FLAG,
+                *(NumberRegField("%s%d" % (consts.TX_POWER_LOW_ALARM_FLAG, lane), self.getaddr(0x11, 140),
+                    RegBitField("Bit%d" % (lane-1), (lane-1))
+                ) for lane in range(1, 9))
+            ),
+            RegGroupField(consts.TX_POWER_HIGH_WARN_FLAG,
+                *(NumberRegField("%s%d" % (consts.TX_POWER_HIGH_WARN_FLAG, lane), self.getaddr(0x11, 141),
+                    RegBitField("Bit%d" % (lane-1), (lane-1))
+                ) for lane in range(1, 9))
+            ),
+            RegGroupField(consts.TX_POWER_LOW_WARN_FLAG,
+                *(NumberRegField("%s%d" % (consts.TX_POWER_LOW_WARN_FLAG, lane), self.getaddr(0x11, 142),
+                    RegBitField("Bit%d" % (lane-1), (lane-1))
+                ) for lane in range(1, 9))
+            )
+        )
+
+        # Group for TX bias alarm and warning flags
+        self.TX_BIAS_ALARM_FLAGS = RegGroupField(consts.TX_BIAS_ALARM_FLAGS_FIELD,
+            RegGroupField(consts.TX_BIAS_HIGH_ALARM_FLAG,
+                *(NumberRegField("%s%d" % (consts.TX_BIAS_HIGH_ALARM_FLAG, lane), self.getaddr(0x11, 143),
+                    RegBitField("Bit%d" % (lane-1), (lane-1))
+                ) for lane in range(1, 9))
+            ),
+            RegGroupField(consts.TX_BIAS_LOW_ALARM_FLAG,
+                *(NumberRegField("%s%d" % (consts.TX_BIAS_LOW_ALARM_FLAG, lane), self.getaddr(0x11, 144),
+                    RegBitField("Bit%d" % (lane-1), (lane-1))
+                ) for lane in range(1, 9))
+            ),
+            RegGroupField(consts.TX_BIAS_HIGH_WARN_FLAG,
+                *(NumberRegField("%s%d" % (consts.TX_BIAS_HIGH_WARN_FLAG, lane), self.getaddr(0x11, 145),
+                    RegBitField("Bit%d" % (lane-1), (lane-1))
+                ) for lane in range(1, 9))
+            ),
+            RegGroupField(consts.TX_BIAS_LOW_WARN_FLAG,
+                *(NumberRegField("%s%d" % (consts.TX_BIAS_LOW_WARN_FLAG, lane), self.getaddr(0x11, 146),
+                    RegBitField("Bit%d" % (lane-1), (lane-1))
+                ) for lane in range(1, 9))
+            )
+        )
+
+        # Group for RX power alarm and warning flags
+        self.RX_POWER_ALARM_FLAGS = RegGroupField(consts.RX_POWER_ALARM_FLAGS_FIELD,
+            RegGroupField(consts.RX_POWER_HIGH_ALARM_FLAG,
+                *(NumberRegField("%s%d" % (consts.RX_POWER_HIGH_ALARM_FLAG, lane), self.getaddr(0x11, 149),
+                    RegBitField("Bit%d" % (lane-1), (lane-1))
+                ) for lane in range(1, 9))
+            ),
+            RegGroupField(consts.RX_POWER_LOW_ALARM_FLAG,
+                *(NumberRegField("%s%d" % (consts.RX_POWER_LOW_ALARM_FLAG, lane), self.getaddr(0x11, 150),
+                    RegBitField("Bit%d" % (lane-1), (lane-1))
+                ) for lane in range(1, 9))
+            ),
+            RegGroupField(consts.RX_POWER_HIGH_WARN_FLAG,
+                *(NumberRegField("%s%d" % (consts.RX_POWER_HIGH_WARN_FLAG, lane), self.getaddr(0x11, 151),
+                    RegBitField("Bit%d" % (lane-1), (lane-1))
+                ) for lane in range(1, 9))
+            ),
+            RegGroupField(consts.RX_POWER_LOW_WARN_FLAG,
+                *(NumberRegField("%s%d" % (consts.RX_POWER_LOW_WARN_FLAG, lane), self.getaddr(0x11, 152),
+                    RegBitField("Bit%d" % (lane-1), (lane-1))
+                ) for lane in range(1, 9))
+            )
+        )
+
         self.LANE_DATAPATH_STATUS = RegGroupField(consts.LANE_DATAPATH_STATUS_FIELD,
             RegGroupField(consts.TX_FAULT_FIELD,
                 *(NumberRegField("%s%d" % (consts.TX_FAULT_FIELD, lane), self.getaddr(0x11, 135),

tests/sonic_xcvr/test_cmis.py

@@ -273,74 +273,71 @@ def test_get_rx_cdr_lol(self, mock_response, expected):
         assert result == expected
 
     @pytest.mark.parametrize("mock_response, expected", [
-        ([{'TxPowerHighAlarmFlag1':0}, {'TxPowerLowAlarmFlag1':0}, {'TxPowerHighWarnFlag1':0}, {'TxPowerLowWarnFlag1':0}],
-         {
-            'tx_power_high_alarm':{
-                'TxPowerHighAlarmFlag1': False
-            },
-            'tx_power_low_alarm':{
-                'TxPowerLowAlarmFlag1': False
-            },
-            'tx_power_high_warn':{
-                'TxPowerHighWarnFlag1': False,
+        (
+            {
+                consts.TX_POWER_HIGH_ALARM_FLAG: {'TxPowerHighAlarmFlag1': 0},
+                consts.TX_POWER_LOW_ALARM_FLAG: {'TxPowerLowAlarmFlag1': 0},
+                consts.TX_POWER_HIGH_WARN_FLAG: {'TxPowerHighWarnFlag1': 0},
+                consts.TX_POWER_LOW_WARN_FLAG: {'TxPowerLowWarnFlag1': 0}
             },
-            'tx_power_low_warn':{
-                'TxPowerLowWarnFlag1': False
+            {
+                'tx_power_high_alarm': {'TxPowerHighAlarmFlag1': False},
+                'tx_power_low_alarm': {'TxPowerLowAlarmFlag1': False},
+                'tx_power_high_warn': {'TxPowerHighWarnFlag1': False},
+                'tx_power_low_warn': {'TxPowerLowWarnFlag1': False}
             }
-         }),
-         ([None, None, None, None], None)
+        ),
+        (None, None)
     ])
     def test_get_tx_power_flag(self, mock_response, expected):
         self.api.xcvr_eeprom.read = MagicMock()
-        self.api.xcvr_eeprom.read.side_effect = mock_response
+        self.api.xcvr_eeprom.read.return_value = mock_response
         result = self.api.get_tx_power_flag()
         assert result == expected
 
     @pytest.mark.parametrize("mock_response, expected", [
-        ([{'TxBiasHighAlarmFlag1':0}, {'TxBiasLowAlarmFlag1':0}, {'TxBiasHighWarnFlag1':0}, {'TxBiasLowWarnFlag1':0}],
-         {
-            'tx_bias_high_alarm':{
-                'TxBiasHighAlarmFlag1': False
-            },
-            'tx_bias_low_alarm':{
-                'TxBiasLowAlarmFlag1': False
-            },
-            'tx_bias_high_warn':{
-                'TxBiasHighWarnFlag1': False,
+        (
+            {
+                consts.TX_BIAS_HIGH_ALARM_FLAG: {'TxBiasHighAlarmFlag1': 0},
+                consts.TX_BIAS_LOW_ALARM_FLAG: {'TxBiasLowAlarmFlag1': 0},
+                consts.TX_BIAS_HIGH_WARN_FLAG: {'TxBiasHighWarnFlag1': 0},
+                consts.TX_BIAS_LOW_WARN_FLAG: {'TxBiasLowWarnFlag1': 0}
             },
-            'tx_bias_low_warn':{
-                'TxBiasLowWarnFlag1': False
+            {
+                'tx_bias_high_alarm': {'TxBiasHighAlarmFlag1': False},
+                'tx_bias_low_alarm': {'TxBiasLowAlarmFlag1': False},
+                'tx_bias_high_warn': {'TxBiasHighWarnFlag1': False},
+                'tx_bias_low_warn': {'TxBiasLowWarnFlag1': False}
             }
-         }),
-         ([None, None, None, None], None)
+        ),
+        (None, None)
     ])
     def test_get_tx_bias_flag(self, mock_response, expected):
         self.api.xcvr_eeprom.read = MagicMock()
-        self.api.xcvr_eeprom.read.side_effect = mock_response
+        self.api.xcvr_eeprom.read.return_value = mock_response
         result = self.api.get_tx_bias_flag()
         assert result == expected
 
     @pytest.mark.parametrize("mock_response, expected", [
-        ([{'RxPowerHighAlarmFlag1':0}, {'RxPowerLowAlarmFlag1':0}, {'RxPowerHighWarnFlag1':0}, {'RxPowerLowWarnFlag1':0}],
-         {
-            'rx_power_high_alarm':{
-                'RxPowerHighAlarmFlag1': False
-            },
-            'rx_power_low_alarm':{
-                'RxPowerLowAlarmFlag1': False
-            },
-            'rx_power_high_warn':{
-                'RxPowerHighWarnFlag1': False,
+        (
+            {
+                consts.RX_POWER_HIGH_ALARM_FLAG: {'RxPowerHighAlarmFlag1': 0},
+                consts.RX_POWER_LOW_ALARM_FLAG: {'RxPowerLowAlarmFlag1': 0},
+                consts.RX_POWER_HIGH_WARN_FLAG: {'RxPowerHighWarnFlag1': 0},
+                consts.RX_POWER_LOW_WARN_FLAG: {'RxPowerLowWarnFlag1': 0}
             },
-            'rx_power_low_warn':{
-                'RxPowerLowWarnFlag1': False
+            {
+                'rx_power_high_alarm': {'RxPowerHighAlarmFlag1': False},
+                'rx_power_low_alarm': {'RxPowerLowAlarmFlag1': False},
+                'rx_power_high_warn': {'RxPowerHighWarnFlag1': False},
+                'rx_power_low_warn': {'RxPowerLowWarnFlag1': False}
             }
-         }),
-         ([None, None, None, None], None)
+        ),
+        (None, None)
     ])
     def test_get_rx_power_flag(self, mock_response, expected):
         self.api.xcvr_eeprom.read = MagicMock()
-        self.api.xcvr_eeprom.read.side_effect = mock_response
+        self.api.xcvr_eeprom.read.return_value = mock_response
         result = self.api.get_rx_power_flag()
         assert result == expected