更新 'ocs/note.md'

ocs/note.md

@@ -824,3 +824,339 @@ def process_host(host, reg_table_file, exp_list, slot_list, lane_list, cmd, para
         if cmd in ['eq-auto-tune']:
             slot_list = [0,1,2,3,4,5,6,7]
 
+
+# opt.py
+
+import logging, math, sys
+from typing import List
+from gpu.biren.exp_util import DevWhiteRiverExp
+from toolbox.regs_save_load_tool import LaneRegInfo
+from parser.transceiver_config_parser import TransceiverConfigParser
+
+
+class OptRegAccessTool:
+    def __init__(self, host:str, reg_table_file: str, bmc: DevWhiteRiverExp):
+        self.host = host
+        self.bmc = bmc
+        self.parser = TransceiverConfigParser(reg_table_file)
+        self.reg_table = self.parser.get_all_register_config_groups()
+
+    def read_ibias_reg(self, exp_id: int, slot_id: int, 
+                        lane: int) -> int:
+        ibias_lane = self.get_ibias_by_logic_lane(lane)
+        logging.info(f'exp: {exp_id}, slot: {slot_id}, lane: {lane}, ibias lane: {ibias_lane}')
+        return self.read_opt_reg(exp_id, slot_id, ibias_lane, 'ibias')
+
+    def write_ibias_reg(self, exp_id: int, slot_id: int, 
+                        lane: int, wt_value: int) -> bool:
+        lane = self.get_ibias_by_logic_lane(lane)
+        return self.write_opt_reg(exp_id, slot_id, lane, wt_value, 'ibias')
+    
+    def read_rssi_reg(self, exp_id: int, slot_id: int, 
+                        lane: int) -> int:
+        return self.read_opt_reg(exp_id, slot_id, lane, 'rssi')
+    
+    def read_opcurrent_reg(self, exp_id: int, slot_id: int, 
+                        lane: int) -> int:
+        return self.read_opt_reg(exp_id, slot_id, lane, 'opcurrent')
+
+    def write_opcurrent_reg(self, exp_id: int, slot_id: int, 
+                        lane: int, wt_value: int) -> bool:
+        return self.write_opt_reg(exp_id, slot_id, lane, wt_value, 'opcurrent')
+    
+    def read_mgc_reg(self, exp_id: int, slot_id: int, 
+                        lane: int) -> int:
+        return self.read_opt_reg(exp_id, slot_id, lane, 'mgc')
+
+    def write_mgc_reg(self, exp_id: int, slot_id: int, 
+                        lane: int, wt_value: int) -> bool:
+        return self.write_opt_reg(exp_id, slot_id, lane, wt_value, 'mgc')
+    
+    def read_tia_peak_reg(self, exp_id: int, slot_id: int, 
+                        lane: int) -> int:
+        return self.read_opt_reg(exp_id, slot_id, lane, 'tia_peak')
+
+    def write_tia_peak_reg(self, exp_id: int, slot_id: int, 
+                        lane: int, wt_value: int) -> bool:
+        return self.write_opt_reg(exp_id, slot_id, lane, wt_value, 'tia_peak')
+    
+    def read_high_freq_reg(self, exp_id: int, slot_id: int, 
+                        lane: int) -> int:
+        return self.read_opt_reg(exp_id, slot_id, lane, 'highfreq_eq')
+
+    def write_high_freq_reg(self, exp_id: int, slot_id: int, 
+                        lane: int, wt_value: int) -> bool:
+        return self.write_opt_reg(exp_id, slot_id, lane, wt_value, 'highfreq_eq')
+    
+    def read_tia_stage2_reg(self, exp_id: int, slot_id: int, 
+                        lane: int) -> int:
+        return self.read_opt_reg(exp_id, slot_id, lane, 'tia_stage2')
+
+    def write_tia_stage2_reg(self, exp_id: int, slot_id: int, 
+                        lane: int, wt_value: int) -> bool:
+        return self.write_opt_reg(exp_id, slot_id, lane, wt_value, 'tia_stage2')
+
+    def read_opt_reg(self, exp_id: int, slot_id: int, 
+                        lane: int, reg_name: str):    
+        reg_value = -1
+        reg = self.parser.get_register_by_logic_lane(reg_name, lane)
+        if not reg:
+            logging.error(f'read opt register error. exp:{exp_id}, slot:{slot_id}, lane: {lane}, register name: {reg_name}')
+            return reg_value
+        
+        value = self.bmc.GetOpticalModuleRegs(exp_id=exp_id, slot_id=slot_id,
+                        bank=reg.bank, page=reg.page, 
+                        reg_offset=reg.offset, size=reg.size)
+        reg_value = int(value, 16)
+
+        logging.info(f'{self.host} read_opt_reg exp: {exp_id}, slot: {slot_id}, lane: {lane}, reg: {reg_name}, value: {reg_value}')
+
+        
+        if reg_name == 'rssi':
+            dbm = round(10 * math.log10(reg_value / 10000), 3)
+            logging.info(f'{self.host} exp_id:{exp_id}, slot_id:{slot_id}, lglane:{reg.logic_lane}, lane:{reg.lane}, bank:0x{reg.bank:02X}, page:0x{reg.page:02X}, offset:0x{reg.offset:02X}, value:{reg_value}, dbm: {dbm}')
+        else:
+            logging.info(f'{self.host} exp_id:{exp_id}, slot_id:{slot_id}, lglane:{reg.logic_lane}, lane:{reg.lane}, bank:0x{reg.bank:02X}, page:0x{reg.page:02X}, offset:0x{reg.offset:02X}, value:{reg_value}')
+        return reg_value
+    
+    def write_opt_reg(self, exp_id: int, slot_id: int, 
+                        lane: int, wt_value: int, reg_name: str) -> bool:    
+        ret = False
+        wt_value_str = ''
+
+        reg = self.parser.get_register_by_logic_lane(reg_name, lane)
+        if not reg:
+            logging.error(f'write opt register error. exp:{exp_id}, slot:{slot_id}, lane: {lane}, register name: {reg_name}')
+            return ret
+
+        if reg.size == 1:
+            wt_value_str = f"{wt_value:02X}"
+        elif reg.size == 2:
+            wt_value_str = f"{wt_value:04X}"
+        ret = self.bmc.SetOpticalModuleRegs(exp_id, slot_id=slot_id, 
+                                bank=reg.bank, page=reg.page, 
+                                reg_offset=reg.offset, size=reg.size, 
+                                hex_str=wt_value_str)
+        
+        if ret:
+            logging.info(f'{self.host} write_opt_reg exp: {exp_id}, slot: {slot_id}, lane: {lane}, reg: {reg_name}, value: {wt_value}')
+        else:
+            logging.error(f'{self.host} write_opt_reg exp: {exp_id}, slot: {slot_id}, lane: {lane}, reg: {reg_name}, value: {wt_value}')
+            logging.error(f'{self.host} write_opt_reg ret: {ret}')
+        return ret
+    
+    def get_ibias_by_logic_lane(self, logic_lane) -> int:
+        ibias_lane_map = {
+            0:0,
+            1:0,
+            2:1,
+            3:1,
+            4:2,
+            5:2,
+            6:3,
+            7:3
+        }
+        return ibias_lane_map.get(logic_lane, -1)
+
+    # def ReadAndWriteOptRegs(self, exp_id: int, slot_list: List[int], 
+    #                     lane_list: List[int], regs: List[str],
+    #                     reg_wt_value: str, is_write: bool = False):
+    
+    #     for slot_id in slot_list:
+    #         if is_write:
+    #             # logging.info(f'------------ reg_wt_value:{reg_wt_value}')
+    #             self.SetPrivateLinkRegister(exp_id, slot_id, lane_list, regs, reg_wt_value)
+    #         logging.debug('------------------2')
+    #         self.GetPrivateLinkRegister(exp_id, slot_id, lane_list, regs)
+    #         logging.debug('------------------3')
+
+    def write_opt_regs(self, exp_id: int, slot_list: List[int], 
+                        lane_list: List[int], regs: List[str],
+                        reg_wt_value: str):
+    
+        for slot_id in slot_list:
+            self.write_opt_registers_by_slot(exp_id, slot_id, lane_list, regs, reg_wt_value)
+
+    def read_opt_regs(self, exp_id: int, slot_list: List[int], 
+                        lane_list: List[int], regs: List[str]):
+    
+        for slot_id in slot_list:
+            self.read_opt_registers_by_slot(exp_id, slot_id, lane_list, regs)
+
+    def read_opt_registers_by_slot(self, exp_id: int, slot_id: int, 
+                            lane_list: List[int], regs: List[str]) -> List[LaneRegInfo]: 
+        laneRegValueList: List[LaneRegInfo] = [LaneRegInfo(logic_lane=lane_num) for lane_num in lane_list]
+
+        logging.debug(f'------regs:{regs}')
+        for rd_reg in regs:
+            tmp_lane_list = lane_list
+            if rd_reg == 'ibias' and len(tmp_lane_list) == 8:
+                tmp_lane_list = [0,1,2,3]
+
+            for lane_id in tmp_lane_list:
+                value = self.read_opt_reg(exp_id, slot_id, lane_id, rd_reg)
+                if value == -1:
+                    break
+
+                for langRegValue in laneRegValueList:
+
+                    if langRegValue.logic_lane == lane_id:
+                        if rd_reg == 'mgc':
+                            langRegValue.mgc = value
+                        elif rd_reg == 'vpeak':
+                            langRegValue.vpeak = value
+                        elif rd_reg == 'ibias':
+                            langRegValue.ibias = value
+                        elif rd_reg == 'ipcurrent':
+                            langRegValue.ipcurrent = value
+                        elif rd_reg == 'opcurrent':
+                            langRegValue.opcurrent = value
+                        elif rd_reg == 'lowfreq_eq':
+                            langRegValue.lowfreq_eq = value
+                        elif rd_reg == 'highfreq_eq':
+                            langRegValue.highfreq_eq = value                   
+                        elif rd_reg == 'tia_peak':
+                            langRegValue.tia_peak = value    
+                        elif rd_reg == 'vgc_set':
+                            langRegValue.vgc_set = value                   
+                        elif rd_reg == 'drv_vpeak':
+                            langRegValue.drv_vpeak = value    
+                        
+                        break
+
+        return laneRegValueList   
+
+    def write_opt_registers_by_slot(self, exp_id: int, slot_id: int, 
+                            lane_list: List[int], regs: List[str],
+                            reg_wt_value: str):
+        logging.debug(f'------regs:{regs}')
+        for wt_reg in regs:
+            for lane_id in lane_list:
+                ret = self.write_opt_reg(exp_id, slot_id, lane_id, int(reg_wt_value), wt_reg)
+                if ret == False:
+                    break
+
+    # def GetPrivateLinkRegister(self, exp_id: int, slot_id: int, 
+    #                         lane_list: List[int], regs: List[str]) -> List[LaneRegInfo]: 
+        
+    #     laneRegValueList: List[LaneRegInfo] = [LaneRegInfo(logic_lane=lane_num) for lane_num in lane_list]
+
+    #     logging.debug(f'------regs:{regs}')
+    #     for rd_reg in regs:
+    #         for reg_group in self.reg_table:
+    #             logging.debug(f'reg_group.name:{reg_group.name}, rd_reg:{rd_reg}')
+    #             if reg_group.name == rd_reg:
+    #                 logging.info(f'Read register: {reg_group.name}, description: {reg_group.description}')
+    #                 for reg in reg_group.registers:
+    #                     # logging.info(f'-------------lane_list:{lane_list}, reg.lane: {reg.lane}')
+
+    #                     if reg.logic_lane in lane_list:
+    #                         logging.info(f'-----read logic lane: {reg.logic_lane}, phys lane: {reg.lane}')
+    #                     else:
+    #                         continue
+                
+    #                     size = 1
+    #                     ret = -1
+    #                     if reg.size is not None:
+    #                         size = reg.size
+    #                     #if reg.end_offset is not None:
+    #                     #    size = reg.end_offset - reg.offset + 1
+
+    #                     value = self.bmc.GetOpticalModuleRegs(exp_id, slot_id=slot_id,
+    #                                             bank=reg.bank, page=reg.page, 
+    #                                             reg_offset=reg.offset, size=size)
+                        
+    #                     logging.info(f'----- hex value: {value}')
+    #                     value = int(value, 16)
+
+    #                     for langRegValue in laneRegValueList:
+
+    #                         if langRegValue.logic_lane == reg.logic_lane:
+    #                             if rd_reg == 'mgc':
+    #                                 langRegValue.mgc = value
+    #                             elif rd_reg == 'vpeak':
+    #                                 langRegValue.vpeak = value
+    #                             elif rd_reg == 'ibias':
+    #                                 langRegValue.ibias = value
+    #                             elif rd_reg == 'ipcurrent':
+    #                                 langRegValue.ipcurrent = value
+    #                             elif rd_reg == 'opcurrent':
+    #                                 langRegValue.opcurrent = value
+    #                             elif rd_reg == 'lowfreq_eq':
+    #                                 langRegValue.lowfreq_eq = value
+    #                             elif rd_reg == 'highfreq_eq':
+    #                                 langRegValue.highfreq_eq = value                   
+    #                             elif rd_reg == 'tia_peak':
+    #                                 langRegValue.tia_peak = value    
+    #                             elif rd_reg == 'vgc_set':
+    #                                 langRegValue.vgc_set = value                   
+    #                             elif rd_reg == 'drv_vpeak':
+    #                                 langRegValue.drv_vpeak = value    
+                                
+    #                             break
+
+    #                     if rd_reg == 'rssi':
+    #                         dbm = round(10 * math.log10(value / 10000), 3)
+    #                         logging.info(f'{self.host}: exp_id:{exp_id}, slot_id:{slot_id}, lglane:{reg.logic_lane}, lane:{reg.lane}, bank:0x{reg.bank:02X}, page:0x{reg.page:02X}, offset:0x{reg.offset:02X}, value:{value}, dbm: {dbm}')
+    #                     else:
+    #                         logging.info(f'{self.host}: exp_id:{exp_id}, slot_id:{slot_id}, lglane:{reg.logic_lane}, lane:{reg.lane}, bank:0x{reg.bank:02X}, page:0x{reg.page:02X}, offset:0x{reg.offset:02X}, value:{value}')
+    #                 break
+
+    #     return laneRegValueList    
+
+    # def SetPrivateLinkRegister(self, exp_id: int, slot_id: int, 
+    #                         lane_list: List[int], regs: List[str], reg_wt_value: str):
+
+    #     for wt_reg_name in regs:
+    #         for reg_group in self.reg_table:
+    #             if reg_group.name == wt_reg_name:
+    #                 logging.info(f'Write register: {reg_group.name}, description: {reg_group.description}')
+    #                 for reg in reg_group.registers:
+    #                     if reg.logic_lane in lane_list:
+    #                         logging.info(f'-----read logic lane: {reg.logic_lane}, phys lane: {reg.lane}')
+    #                     else:
+    #                         continue
+
+    #                     size = 1
+    #                     ret = -1
+    #                     if reg.size is not None:
+    #                         size = reg.size
+    #                     if reg.size == 2:    
+    #                         value_hex_str=f"{reg.value:04x}"
+    #                     else:
+    #                         value_hex_str=f"{reg.value:02x}"
+
+    #                     if reg.end_offset is not None:
+    #                         loop = int((reg.end_offset - reg.offset + 1)/size)
+    #                         size = loop * size
+    #                         #logging.info(f"------loop:{loop}")
+    #                         value = ''
+    #                         for i in range(loop):
+    #                             value  += value_hex_str
+    #                         ret = self.bmc.SetOpticalModuleRegs(exp_id, slot_id=slot_id, 
+    #                                                         bank=reg.bank, page=reg.page, 
+    #                                                         reg_offset=reg.offset, size=size, 
+    #                                                         hex_str=value)
+    #                         logging.info(f'exp_id:{exp_id}, slot_id: {slot_id}, bank:0x{reg.bank:02X}, page:0x{reg.page:02X}, offset:0x{reg.offset:02X}, value:{value}, size: {size}, status: {ret}')
+    #                     else:
+    #                         write_value = value_hex_str
+    #                         if reg_wt_value is not '':
+    #                             dec = int(reg_wt_value)
+    #                             write_value = ''
+    #                             if reg.size == 1:   
+    #                                 write_value = format(dec, '02X')
+    #                                 size = 1
+    #                             elif reg.size == 2:   
+    #                                 write_value = format(dec, '04X')
+    #                                 size = 2
+    #                             else:
+    #                                 logging.error(f'error value:{reg_wt_value}')
+                            
+    #                         if write_value is not '': 
+    #                             logging.info(f'reg_wt_value: {reg_wt_value}, value_hex_str: {write_value}')
+    #                             # logging.info(f'-----------------------write hex_str: {write_value}')
+    #                             ret = self.bmc.SetOpticalModuleRegs(exp_id, slot_id=slot_id, 
+    #                                                         bank=reg.bank, page=reg.page, 
+    #                                                         reg_offset=reg.offset, size=size, 
+    #                                                         hex_str=write_value)
+    #                             # logging.info(f'exp_id:{exp_id}, slot_id: {slot_id}, bank:{reg.bank}, page:{reg.page}, offset:{reg.offset}, value:{value_hex_str}, status: {ret}')