# uncompyle6 version 3.9.3 # Python bytecode version base 3.6 (3379) # Decompiled from: Python 3.6.8 (default, May 21 2025, 11:12:56) # [GCC 8.5.0 20210514 (Anolis 8.5.0-10.0.1)] # Embedded file name: /xz/gyou/nexusbench/toolbox/mgc_tuning_tool.py # Compiled at: 2025-12-25 14:52:30 # Size of source mod 2**32: 78005 bytes from dataclasses import dataclass, field import logging, time, os, json, sys from typing import Dict, List, Any, Optional, Tuple import matplotlib.pyplot as plt from gpu.biren.exp_util import DevWhiteRiverExp from toolbox.opt_reg_access_tool import OptRegAccessTool from toolbox.prbs_tool import PrbsTool, RetimerPrbsResult from parser.topology_parser import TopoMappingParser from parser.transceiver_config_parser import TransceiverConfigParser, RegisterInfo from toolbox.bathtub_curve import BathtubCurve @dataclass class LaneErrInfo: host: str = "" exp: int = -1 route: str = "" slot_id: int = -1 logic_lane: int = -1 rt_phys_id: int = -1 rt_phys_lane: int = -1 peer_slot_id: int = -1 peer_ibias_lane: int = -1 peer_logic_lane: int = -1 peer_rt_phys_id: int = -1 peer_rt_phys_lane: int = -1 preset: int = -1 mgc: int = -1 mgc_err_map: Dict[int, int] = field(default_factory=dict) lowfreq_err_map: Dict[int, int] = field(default_factory=dict) tmp_err_map: Dict[int, int] = field(default_factory=dict) vpeak: int = -1 tia_peak: int = -1 ibias: int = -1 ipcurrent: int = -1 opcurrent: int = -1 lowfreq_eq: int = -1 highfreq_eq: int = -1 vgc_set: int = -1 drv_vpeak: int = -1 rssi1: int = -1 rssi2: int = -1 tmp_raw_value: int = -1 ctle: int = -1 class MgcTuningTool: def __init__(self, host:str, reg_table: TransceiverConfigParser, reg_tool: OptRegAccessTool, remote_reg_tool: OptRegAccessTool, bmc: DevWhiteRiverExp, topo_map: TopoMappingParser, prbs_tool: PrbsTool, route_name: str): self.host = host self.reg_table = reg_table self.bmc = bmc self.target_vpeak = 0 self.local_reg_tool = reg_tool self.remote_reg_tool = remote_reg_tool self.target_vpeaks_file = "target_vpeaks.json" self.mode = route_name self.topo_map = topo_map self.prbs_tool = prbs_tool self.route_name = route_name self.is_onet = self.remote_reg_tool != self.local_reg_tool def debug_calc_target_vpeak_new(self, exp_id: int, slot_id: int) -> Optional[Dict[(int, int)]]: target_vpeaks = {} logging.info(f"-------slot {slot_id}") self.local_reg_tool.enable_tia_agc(exp_id, slot_id) time.sleep(0.05) logging.info("----------step 2: toggle RF off") if not self.toogle_rf(exp_id, slot_id, "off"): logging.error(f"slot {slot_id}: toggle RF off fail") return target_vpeaks else: self.local_reg_tool.write_agc_reg_all_lane(exp_id, slot_id, 0) time.sleep(0.5) logging.info("----------step 3: get base_line_vpeaks, and calculate target vpeaks") base_line_vpeaks = self.local_reg_tool.read_vpeak_reg_all_lane(exp_id, slot_id) logging.info(f"base_line_vpeaks: {base_line_vpeaks}") self.toogle_rf(exp_id, slot_id, "on") self.local_reg_tool.write_agc_reg_all_lane(exp_id, slot_id, 255) time.sleep(0.5) max_vpeaks = self.local_reg_tool.read_vpeak_reg_all_lane(exp_id, slot_id) logging.info(f"max_vpeaks: {max_vpeaks}") percent_target_vpeaks = {} for lane, base_vpeak in base_line_vpeaks.items(): numerator = 19 vpkdelta = round((max_vpeaks[lane] - base_vpeak) * (numerator / 29)) percent_target_vpeaks[lane] = vpkdelta + base_vpeak diff = max_vpeaks[lane] - base_line_vpeaks[lane] if diff < 5: logging.error(f"host: {self.host}, exp: {exp_id}, slot: {slot_id}, lane: {lane}, max_vpeak is error") sys.exit(-1) logging.info(f"percent_target_vpeaks: {percent_target_vpeaks}") diff_target_vpeaks = {} for lane, base_vpeak in base_line_vpeaks.items(): diff = max_vpeaks[lane] - base_line_vpeaks[lane] if diff <= 22: logging.error(f"host: {self.host}, exp: {exp_id}, slot: {slot_id}, lane: {lane}, max_vpeak is error") diff_target_vpeaks[lane] = diff - 2 + base_vpeak else: diff_target_vpeaks[lane] = 21 + base_vpeak target_vpeaks = diff_target_vpeaks logging.info(f"diff_target_vpeaks: {target_vpeaks}") logging.info("\n" + "=" * 90) logging.info(f'{"SLOT":<6} {"LANE":<6} {"BASE":<8} {"MAX":<8} {"RANGE":<8} {"PERCENT_TGT":<12} {"DIFF_TGT":<12} {"PCT_OFF":<10} {"DIFF_OFF":<10}') logging.info("=" * 90) for lane in sorted(base_line_vpeaks.keys()): base = base_line_vpeaks[lane] mx = max_vpeaks[lane] rng = mx - base pct_tgt = percent_target_vpeaks.get(lane, -1) diff_tgt = diff_target_vpeaks.get(lane, -1) pct_off = pct_tgt - base diff_off = diff_tgt - base logging.info(f"{slot_id:<6} {lane:<6} {base:<8} {mx:<8} {rng:<8} {pct_tgt:<12} {diff_tgt:<12} {pct_off:<10} {diff_off:<10}") logging.info("=" * 90 + "\n") self.local_reg_tool.disable_tia_agc(exp_id, slot_id) time.sleep(0.05) try: agc_start, agc_end, agc_step = (80, 180, 2) agc_values = list(range(agc_start, agc_end + 1, agc_step)) all_vpeaks_scan = {lane: [] for lane in base_line_vpeaks.keys()} logging.info(f"Scanning AGC from {agc_start} to {agc_end} step {agc_step}...") for agc in agc_values: self.local_reg_tool.write_tmp_mgc_reg_all_lane(exp_id, slot_id, agc) # self.local_reg_tool.write_agc_reg_all_lane(exp_id, slot_id, agc) time.sleep(0.1) vpeaks = self.local_reg_tool.read_vpeak_reg_all_lane(exp_id, slot_id) for lane in base_line_vpeaks.keys(): all_vpeaks_scan[lane].append(vpeaks.get(lane, 0)) plt.figure(figsize=(20, 8)) for lane in sorted(base_line_vpeaks.keys()): plt.plot(agc_values, (all_vpeaks_scan[lane]), marker="o", label=f"lane {lane}") for lane in base_line_vpeaks.keys(): plt.scatter([80], [base_line_vpeaks[lane]], color="blue") plt.scatter([180], [max_vpeaks[lane]], color="red") plt.title(f"Vpeak vs AGC (exp={exp_id}, slot={slot_id})") plt.xlabel("AGC Value") plt.ylabel("Vpeak") plt.legend() plt.grid(True) plt.tight_layout() plot_dir = "agc_vpeak_plots" os.makedirs(plot_dir, exist_ok=True) plot_path = os.path.join(plot_dir, f"{self.local_reg_tool.host}_agc_vpeak_exp{exp_id}_slot{slot_id}.png") plt.savefig(plot_path) logging.info(f"AGC-Vpeak curve saved to: {plot_path}") plt.close() except Exception as e: logging.warning(f"Failed to generate AGC scan plot: {e}") if target_vpeaks: self.save_target_vpeaks_to_json(exp_id, slot_id, target_vpeaks) return target_vpeaks def auto_tune(self, exp_id: int, slot_id: int, lane_list: List[int]) -> bool: lane_list = [ 7, 6, 5, 4, 3, 2, 1, 0] # self.debug_calc_target_vpeak_new(exp_id, slot_id) target_vpeaks = self.calc_target_vpeak_new(exp_id, slot_id, lane_list) if target_vpeaks is None: return False time.sleep(0.1) logging.info("----------step 7: find optimal MGC") self.match_optimal_mgc_new(exp_id, slot_id, lane_list, target_vpeaks) return True def auto_tia_peak_tune(self, exp_id: int, retimers): tune_value_list = [0, 30, 60, 90, 120, 150, 180, 210, 240] self.general_auto_tune_value(exp_id, retimers, "tia_peak", tune_value_list, self.local_reg_tool) def auto_highfreq_tune(self, exp_id: int, retimers): tune_value_list = [0, 20, 40, 60, 80, 100, 120, 140, 160, 180, 200, 220, 240] self.general_auto_tune_value(exp_id, retimers, "highfreq_eq", tune_value_list, self.remote_reg_tool) def rt_ibias(self, exp_id: int, retimers): tune_value_list = [ 1647, 3647] self.ibias_auto_tune_value_new(exp_id, retimers, "ibias_runtime", tune_value_list, self.remote_reg_tool, False) def auto_ibias_tune2(self, exp_id: int, retimers): tune_value_list = [ 1647, 1847, 2047, 2247, 2447, 2647, 2847, 3047, 3247, 3447, 3647] self.ibias_auto_tune_value(exp_id, retimers, "ibias_runtime", tune_value_list, self.remote_reg_tool, False) def auto_opcurrent_tune(self, exp_id: int, retimers): tune_value_list = [130, 140, 150, 160, 120, 110, 90, 80, 100, 170, 180] self.general_auto_tune_value(exp_id, retimers, "opcurrent", tune_value_list, self.remote_reg_tool) def auto_opcurrent_tune_test(self, exp_id: int, retimers): tune_value_list = [110, 130, 150, 170, 90, 70, 50, 190, 210, 230] self.general_auto_tune_value(exp_id, retimers, "opcurrent", tune_value_list, self.remote_reg_tool) def auto_lowfreq_tune(self, exp_id: int, retimers): tune_value_list = [ 120, 150, 180, 210, 240, 90, 60, 30, 0] self.general_auto_tune_value(exp_id, retimers, "lowfreq_eq", tune_value_list, self.remote_reg_tool) def auto_ctle_tune(self, exp_id: int, retimers): tune_value_list = [ 0, 2, 4, 6, 8, 10, 12, 14, 16, 18] self.general_auto_tune_value(exp_id, retimers, "lowfreq_eq", tune_value_list, self.remote_reg_tool) def auto_tune2(self, exp_id: int, retimers) -> bool: self.prbs_tool.ResetRetimerAndEnablePrbs(exp_id, retimers) logging.info("Starting auto_tune2...") lane_data_list: List[LaneErrInfo] = [] # MGC_OFFSETS = [0, 2, -2, 4, -4, 6, -6, 8, -8, 10, -10, 12, -12, 14, -14] # MGC_OFFSETS = [0, 2] MGC_OFFSETS = [0, 2, -2, 4, -4, 6, -6, 8, -8, 10, -10] for slot_id in [0, 1, 2, 3, 4, 5, 6, 7]: for lane in [0, 1, 2, 3, 4, 5, 6, 7]: current_mgc = self.local_reg_tool.read_mgc_reg(exp_id, slot_id, lane) current_mgc = int(current_mgc) mgc_points = [current_mgc + offset for offset in MGC_OFFSETS] mgc_err_map = {mgc: -1 for mgc in mgc_points} success, rt_phys_id, rt_phys_lane = self.topo_map.get_phys_rtmr_by_slot(slot_id, lane, self.route_name) if success == False: logging.error(f'get_phys_rtmr_by_slot failed, result: {success}') return False lane_info = LaneErrInfo( slot_id=slot_id, logic_lane=lane, rt_phys_id=rt_phys_id, rt_phys_lane=rt_phys_lane, mgc_err_map=mgc_err_map, mgc= current_mgc ) lane_data_list.append(lane_info) # logging.info( # f"Initialized: Slot {slot_id}, Lane {lane}, " # f"Base MGC={current_mgc}, Target MGCs={mgc_points}, ERRs={[-1.0]*5}" # ) # logging.info(lane_data_list) prbs_results = self.prbs_tool.prbs_check(exp_id, retimers, True) all_unlocked = all(not prbs.locked for prbs in prbs_results) if all_unlocked: logging.error(f"PRBS check failed: ALL lanes are UNLOCKED for exp_id={exp_id}. Aborting auto_tune2.") return False time.sleep(2) prbs_results = self.prbs_tool.prbs_check(exp_id, retimers, False) # return True # logging.info(f'-------------- prbs_results: {prbs_results}') for prbs in prbs_results: local_slot = self.topo_map.get_local_slot_by_retimer(prbs.rtmr_id, prbs.rtmr_lane, self.route_name) if not local_slot: logging.error(f'local slot is none!') return False slot_id, lane_id = local_slot # logging.info(f'------------------ slot: {slot_id}, lane_id: {lane_id}') for info in lane_data_list: if info.slot_id == slot_id and info.logic_lane == lane_id: if prbs.locked: info.mgc_err_map[info.mgc] = prbs.err_count else: info.mgc_err_map[info.mgc] = 65535 # logging.info(f'prbs.locked: {prbs.locked}, prbs.err_count: {prbs.err_count}') break # return True for offset in [offset for offset in MGC_OFFSETS if offset != 0]: for info in lane_data_list: new_mgc = info.mgc + offset self.local_reg_tool.write_tmp_mgc_reg(exp_id, info.slot_id, info.logic_lane, new_mgc) time.sleep(0.05) # for slot_id in [0,1,2,3,4,5,6,7]: # self.local_reg_tool.write_confirm_reg(exp_id, slot_id) # time.sleep(0.05) time.sleep(0.5) prbs_results = self.prbs_tool.prbs_check(exp_id, retimers, True) time.sleep(2) # 测 PRBS prbs_results = self.prbs_tool.prbs_check(exp_id, retimers) is_need_prbs_reset = False # 更新结果：当前 MGC 是 info.mgc + offset for prbs in prbs_results: local_slot = self.topo_map.get_local_slot_by_retimer(prbs.rtmr_id, prbs.rtmr_lane, self.route_name) if not local_slot: logging.error(f'local slot is none!') return False slot_id, lane_id = local_slot for info in lane_data_list: if info.slot_id == slot_id and info.logic_lane == lane_id: target_mgc = info.mgc + offset if prbs.locked: info.mgc_err_map[target_mgc] = prbs.err_count else: info.mgc_err_map[target_mgc] = 69999 is_need_prbs_reset = True # logging.info(f'------update mgc err, {target_mgc}: {info.mgc_err_map[target_mgc]}') break if is_need_prbs_reset: self.prbs_tool.ResetRetimerAndEnablePrbs(exp_id, retimers) self.print_err_matrix(lane_data_list, MGC_OFFSETS) # 参数复原 # logging.info(f'------------len: {len(lane_data_list)}') for info in lane_data_list: new_mgc = info.mgc self.local_reg_tool.write_tmp_mgc_reg(exp_id, info.slot_id, info.logic_lane, new_mgc) time.sleep(0.05) # for slot_id in [0,1,2,3,4,5,6,7]: # self.local_reg_tool.write_confirm_reg(exp_id, slot_id) # time.sleep(0.05) # return True # Step 4: 调用简单调优，返回与 lane_data_list 一一对应的推荐 offset 列表 recommended_mgcs, bad_lanes = self.simple_mgc_tuner(lane_data_list, bad_threshold=1) tuning_success = True for info, recommended_mgc in zip(lane_data_list, recommended_mgcs): slot_id = info.slot_id lane_id = info.logic_lane current_mgc = info.mgc target_mgc = recommended_mgc diff_mgc = target_mgc - current_mgc # 日志输出 if diff_mgc != 0: logging.info( f"Slot {slot_id}, Lane {lane_id}: MGC adjusted from {current_mgc} " f"→ {target_mgc} (offset={diff_mgc:.1f})" ) # logging.info(f"Applying MGC reg write: exp_id={exp_id}, slot={slot_id}, lane={lane_id}, mgc={target_mgc}") # self.local_reg_tool.write_mgc_reg(exp_id, slot_id, lane_id, target_mgc) # time.sleep(0.05) # self.local_reg_tool.write_confirm_reg(exp_id, slot_id) # time.sleep(0.05) # else: # logging.info( # f"Slot {slot_id}, Lane {lane_id}: MGC {current_mgc} is optimal. No change." # ) tuning_success = True if bad_lanes: tuning_success = False for slot, lane in bad_lanes: logging.warning(f"Slot {slot}, Lane {lane}: CRITICAL - Too few clean MGC points (<=5). Signal integrity issue?") return True def low_ibias_tune2(self, exp_id: int, retimers, reg_name: str, value_offset_list: Optional[List[int]]=None, value_list: Optional[List[int]]=None) -> bool: logging.info("Starting auto_tune2...") lane_data_list = [] MGC_OFFSETS = [ 0, 2, -2, 4, -4, 6, -6, 8, -8] for slot_id in (0, 1, 2, 3, 4, 5, 6, 7): for lane in (0, 1, 2, 3, 4, 5, 6, 7): if value_offset_list is not None: current_value = self.local_reg_tool.read_opt_reg(exp_id, slot_id, lane, reg_name) current_value = int(current_value) value_points = [current_value + offset for offset in value_offset_list] tmp_err_map = {value: -1 for value in value_points} success, rt_phys_id, rt_phys_lane = self.topo_map.get_phys_rtmr_by_slot(slot_id, lane, self.route_name) if success == False: logging.error(f"get_phys_rtmr_by_slot failed, result: {success}") return False lane_info = LaneErrInfo(slot_id=slot_id, logic_lane=lane, rt_phys_id=rt_phys_id, rt_phys_lane=rt_phys_lane, mgc_err_map=tmp_err_map, mgc=current_value) lane_data_list.append(lane_info) prbs_results = self.prbs_tool.prbs_check(exp_id, retimers, True) all_unlocked = all(not prbs.locked for prbs in prbs_results) if all_unlocked: logging.error(f"PRBS check failed: ALL lanes are UNLOCKED for exp_id={exp_id}. Aborting auto_tune2.") return False else: time.sleep(2) prbs_results = self.prbs_tool.prbs_check(exp_id, retimers, False) for prbs in prbs_results: local_slot = self.topo_map.get_local_slot_by_retimer(prbs.rtmr_id, prbs.rtmr_lane, self.route_name) if not local_slot: logging.error("local slot is none!") return False slot_id, lane_id = local_slot for info in lane_data_list: if info.slot_id == slot_id: if info.logic_lane == lane_id: if prbs.locked: info.tmp_err_map[info.tmp_value] = prbs.err_count else: info.tmp_err_map[info.tmp_value] = 65535 break for offset in [offset for offset in MGC_OFFSETS if offset != 0]: for info in lane_data_list: new_mgc = info.mgc + offset self.local_reg_tool.write_mgc_reg(exp_id, info.slot_id, info.logic_lane, new_mgc) for slot_id in (0, 1, 2, 3, 4, 5, 6, 7): self.local_reg_tool.write_confirm_reg(exp_id, slot_id) time.sleep(0.5) prbs_results = self.prbs_tool.prbs_check(exp_id, retimers, True) time.sleep(2) prbs_results = self.prbs_tool.prbs_check(exp_id, retimers) for prbs in prbs_results: local_slot = self.topo_map.get_local_slot_by_retimer(prbs.rtmr_id, prbs.rtmr_lane, self.route_name) if not local_slot: logging.error("local slot is none!") return False slot_id, lane_id = local_slot for info in lane_data_list: if info.slot_id == slot_id: if info.logic_lane == lane_id: target_mgc = info.mgc + offset if prbs.locked: info.mgc_err_map[target_mgc] = prbs.err_count else: info.mgc_err_map[target_mgc] = 65535 break self.print_err_matrix(lane_data_list, MGC_OFFSETS) for info in lane_data_list: new_mgc = info.mgc self.local_reg_tool.write_mgc_reg(exp_id, info.slot_id, info.logic_lane, new_mgc) recommended_mgcs, bad_lanes = self.simple_mgc_tuner(lane_data_list, bad_threshold=1) tuning_success = True for info, recommended_mgc in zip(lane_data_list, recommended_mgcs): slot_id = info.slot_id lane_id = info.logic_lane current_mgc = info.mgc target_mgc = recommended_mgc diff_mgc = target_mgc - current_mgc if diff_mgc != 0: logging.info(f"Slot {slot_id}, Lane {lane_id}: MGC adjusted from {current_mgc} → {target_mgc} (offset={diff_mgc:.1f})") logging.info(f"Applying MGC reg write: exp_id={exp_id}, slot={slot_id}, lane={lane_id}, mgc={target_mgc}") self.local_reg_tool.write_mgc_reg(exp_id, slot_id, lane_id, target_mgc) self.local_reg_tool.write_confirm_reg(exp_id, slot_id) tuning_success = True if bad_lanes: tuning_success = False for slot, lane in bad_lanes: logging.warning(f"Slot {slot}, Lane {lane}: CRITICAL - Too few clean MGC points (<=5). Signal integrity issue?") return True def general_auto_tune_value_new(self, exp_id: int, retimers, reg_name: str, tune_value_list: List[int], reg_tool: OptRegAccessTool, is_overwrite=False) -> bool: logging.info("Starting auto_tune2...") lane_data_list = [] lane_list = [ 0, 1, 2, 3, 4, 5, 6, 7] slot_list = [0, 1, 2, 3, 4, 5, 6, 7] if reg_name in ('ibias', 'opcurrent'): reg_tool = self.remote_reg_tool else: reg_tool = self.local_reg_tool for rt_phys_id in retimers: for rt_phys_lane in range(0, 16): if reg_tool == self.local_reg_tool: slot = self.topo_map.get_local_slot_by_retimer(rt_phys_id, rt_phys_lane, self.route_name) else: slot = self.topo_map.get_remote_slot_by_retimer(rt_phys_id, rt_phys_lane, self.route_name) if not slot: logging.error("local slot is none!") return False slot_id, lane_id = slot raw_value = reg_tool.read_opt_reg(exp_id, slot_id, lane_id, reg_name) tmp_err_map = {value: -1 for value in tune_value_list} lane_info = LaneErrInfo(host=(reg_tool.host), exp=exp_id, slot_id=slot_id, logic_lane=lane_id, rt_phys_id=rt_phys_id, rt_phys_lane=rt_phys_lane, tmp_err_map=tmp_err_map, tmp_raw_value=raw_value) lane_data_list.append(lane_info) time.sleep(0.05) is_need_reset_prbs = True for value in tune_value_list: for info in lane_data_list: reg_tool.write_opt_reg(exp_id, info.slot_id, info.logic_lane, value, reg_name) time.sleep(0.05) time.sleep(1) if is_need_reset_prbs: logging.info(f"-------- is_need_reset_prbs: {is_need_reset_prbs}") self.prbs_tool.ResetRetimerAndEnablePrbs(exp_id, retimers) prbs_results = self.prbs_tool.prbs_check(exp_id, retimers, True) time.sleep(2) prbs_results = self.prbs_tool.prbs_check(exp_id, retimers) is_need_reset_prbs = False for prbs in prbs_results: if reg_tool == self.local_reg_tool: slot = self.topo_map.get_local_slot_by_retimer(prbs.rtmr_id, prbs.rtmr_lane, self.route_name) else: slot = self.topo_map.get_remote_slot_by_retimer(prbs.rtmr_id, prbs.rtmr_lane, self.route_name) if not slot: logging.error("local slot is none!") return False slot_id, lane_id = slot for info in lane_data_list: if info.slot_id == slot_id: if info.logic_lane == lane_id: if prbs.locked: info.tmp_err_map[value] = prbs.err_count elif value in info.tmp_err_map: info.tmp_err_map[value] = 69999 else: logging.error(f"--------Key {value} not found in tmp_err_map, skipping assignment.") is_need_reset_prbs = True break logging.info(lane_data_list) self.print_general_err_matrix(lane_data_list, tune_value_list) time.sleep(0.05) for info in lane_data_list: reg_tool.write_opt_reg(exp_id, info.slot_id, info.logic_lane, info.tmp_raw_value, reg_name) time.sleep(0.05) recommended_value, bad_lanes = self.simple_value_tuner(lane_data_list) for info, recommended_value in zip(lane_data_list, recommended_value): diff_value = recommended_value - info.tmp_raw_value if diff_value != 0: logging.info(f"Slot {info.slot_id}, Lane {info.logic_lane}: adjusted from {info.tmp_raw_value} → {recommended_value} (offset={diff_value:.1f})") logging.info(f"Applying {reg_name} reg write: exp_id={exp_id}, slot={info.slot_id}, lane={info.logic_lane}, {reg_name}={recommended_value}") if is_overwrite: reg_tool.write_opt_reg(exp_id, info.slot_id, info.logic_lane, recommended_value, reg_name) if bad_lanes: for slot, lane in bad_lanes: logging.error(f"Slot {slot}, Lane {lane}: CRITICAL - Too few clean points (<=5). Bad Modules") return True def ibias_auto_tune_value_new(self, exp_id: int, retimers, reg_name: str, tune_value_list: List[int], reg_tool: OptRegAccessTool, is_overwrite=False) -> bool: logging.info("Starting ibias_auto_tune_value_new...") local_data_list: List[LaneErrInfo] = [] remote_data_list: List[LaneErrInfo] = [] local_slot_list = list(range(8)) local_lane_list = list(range(8)) remote_slot_list = list(range(8)) remote_lane_list = list(range(8)) # 存储结构：{slot_id: ibias_dict} remote_raw_ibias: Dict[int, Dict] = {} local_raw_ibias: Dict[int, Dict] = {} # init remote(ibias) for slot_id in remote_slot_list: self.remote_reg_tool.bmc.SetOpticalModuleRegs(exp_id, slot_id, 0, 180, 136, 1, "01") remote_raw_ibias = self.remote_reg_tool.read_ibias_reg_all_lane(exp_id, slot_id) time.sleep(0.05) if self.is_onet: for slot_id in local_slot_list: self.local_reg_tool.bmc.SetOpticalModuleRegs(exp_id, slot_id, 0, 180, 136, 1, "01") local_raw_ibias = self.local_reg_tool.read_ibias_reg_all_lane(exp_id, slot_id) time.sleep(0.05) # init remote (ibias) for slot_id in local_slot_list: for lane in local_lane_list: peer_info = self.topo_map.get_remote_retimer_and_slot_by_slot(slot_id, lane, self.route_name) if not peer_info: logging.error(f"peer_info is None for slot {slot_id} lane {lane}") return False remote_rt_logic_id, remote_rt_logic_lane, remote_slot_id, remote_lane_id = peer_info tmp_err_map = {value: -1 for value in tune_value_list} result = self.topo_map.get_phys_rtmr_by_slot(slot_id, lane, self.route_name) if result: success, rt_phys_id, rt_phys_lane = result if success == False: logging.error(f"get_phys_rtmr_by_slot failed, result: {success}") return False lane_info = LaneErrInfo(host=(self.local_reg_tool.host), exp=exp_id, route=(self.route_name.split("-")[-1]), slot_id=slot_id, logic_lane=lane, rt_phys_id=rt_phys_id, rt_phys_lane=rt_phys_lane, peer_slot_id=remote_slot_id, peer_logic_lane=remote_lane_id, peer_rt_phys_id=remote_rt_logic_id, peer_rt_phys_lane=remote_rt_logic_lane, tmp_err_map=tmp_err_map) local_data_list.append(lane_info) # init local (ibias) if self.is_onet: for slot_id in remote_slot_list: for lane in remote_lane_list: peer_info = self.topo_map.get_remote_retimer_and_slot_by_slot(slot_id, lane, self.route_name) if not peer_info: logging.error(f"peer_info is None for slot {slot_id} lane {lane}") return False local_rt_logic_id, local_rt_logic_lane, local_slot_id, local_lane_id = peer_info tmp_err_map = {value: -1 for value in tune_value_list} result = self.topo_map.get_phys_rtmr_by_slot(slot_id, lane, self.route_name) if result: success, rt_phys_id, rt_phys_lane = result if success == False: logging.error(f"get_phys_rtmr_by_slot failed, result: {success}") return False lane_info = LaneErrInfo(host=(self.remote_reg_tool.host), exp=exp_id, route=(self.route_name.split("-")[-1]), slot_id=slot_id, logic_lane=lane, rt_phys_id=rt_phys_id, rt_phys_lane=rt_phys_lane, peer_slot_id=local_slot_id, peer_logic_lane=local_lane_id, peer_rt_phys_id=local_rt_logic_id, peer_rt_phys_lane=local_rt_logic_lane, tmp_err_map=tmp_err_map) remote_data_list.append(lane_info) time.sleep(0.05) is_need_reset_prbs = True for value in tune_value_list: for info in local_data_list: self.remote_reg_tool.write_vol_ibias_reg_all_lane(exp_id, info.peer_slot_id, value) time.sleep(0.05) if self.is_onet: for info in remote_data_list: self.local_reg_tool.write_vol_ibias_reg_all_lane(exp_id, info.peer_slot_id, value) time.sleep(0.05) time.sleep(1) if is_need_reset_prbs: logging.info(f"-------- is_need_reset_prbs: {is_need_reset_prbs}") self.prbs_tool.ResetRetimerAndEnablePrbs(exp_id, retimers) prbs_results = self.prbs_tool.prbs_check_bidirection(exp_id, retimers, True) time.sleep(2) prbs_results = self.prbs_tool.prbs_check_bidirection(exp_id, retimers) # logging.info(f'--------- prbs_results: {prbs_results}') is_need_reset_prbs = False for prbs in prbs_results: for info in local_data_list: if info.host == prbs.host and info.rt_phys_id == prbs.rtmr_id \ and info.rt_phys_lane == prbs.rtmr_lane: if prbs.locked: info.tmp_err_map[value] = prbs.err_count elif value in info.tmp_err_map: info.tmp_err_map[value] = 69999 else: logging.error(f"--------Key {value} not found in tmp_err_map, skipping assignment.") is_need_reset_prbs = True break for info in remote_data_list: if info.host == prbs.host and info.rt_phys_id == prbs.rtmr_id \ and info.rt_phys_lane == prbs.rtmr_lane: if prbs.locked: info.tmp_err_map[value] = prbs.err_count elif value in info.tmp_err_map: info.tmp_err_map[value] = 69999 else: logging.error(f"--------Key {value} not found in tmp_err_map, skipping assignment.") is_need_reset_prbs = True break self.print_general_err_matrix(local_data_list, tune_value_list) self.print_general_err_matrix(remote_data_list, tune_value_list) time.sleep(0.05) for slot_id in remote_slot_list: self.remote_reg_tool.write_vol_ibias_reg_all_lane(exp_id, slot_id, wt_value_list=remote_raw_ibias[slot_id]) time.sleep(0.05) self.remote_reg_tool.bmc.SetOpticalModuleRegs(exp_id, slot_id, 0, 180, 136, 1, "00") time.sleep(0.05) if self.is_onet: for slot_id in local_slot_list: self.local_reg_tool.write_vol_ibias_reg_all_lane(exp_id, slot_id, wt_value_list=remote_raw_ibias[slot_id]) time.sleep(0.05) self.local_reg_tool.bmc.SetOpticalModuleRegs(exp_id, slot_id, 0, 180, 136, 1, "00") time.sleep(0.05) recommended_value, bad_lanes = self.simple_value_tuner(local_data_list) reg_name = "ibias" for info, recommended_value in zip(local_data_list, recommended_value): diff_value = recommended_value - info.tmp_raw_value if diff_value != 0: remote_info = self.topo_map.get_remote_retimer_and_slot_by_slot(info.slot_id, info.logic_lane, self.route_name) if not remote_info: logging.error(f"remote_info is None for slot {info.slot_id} lane {info.logic_lane}") return False remote_rt_logic_id, remote_rt_logic_lane, remote_slot_id, remote_lane_id = remote_info logging.info(f"Rx Slot: {info.slot_id},{info.logic_lane}, Tx Slot: {remote_slot_id}, {remote_lane_id}, adjusted from {info.tmp_raw_value} → {recommended_value} (offset={diff_value:.1f})") if is_overwrite: reg_tool.write_ibias_reg(exp_id, remote_slot_id, remote_lane_id, recommended_value) time.sleep(0.05) if bad_lanes: for slot, lane in bad_lanes: logging.error(f"Rx Slot {slot}, Rx Lane {lane}: CRITICAL - Too few clean points (<=5). Bad Modules") return True def ibias_auto_tune_value(self, exp_id: int, retimers, reg_name: str, tune_value_list: List[int], reg_tool: OptRegAccessTool, is_overwrite=False) -> bool: logging.info("Starting auto_tune2...") lane_data_list = [] rx_slot_list = [0, 1, 2, 3, 4, 5, 6, 7] tx_slot_list = [0, 1, 2, 3, 4, 5, 6, 7] for slot_id in rx_slot_list: reg_tool.bmc.SetOpticalModuleRegs(exp_id, slot_id, 0, 180, 136, 1, "01") time.sleep(0.05) rx_lane_list = [0, 1, 2, 3, 4, 5, 6, 7] for rx_slot_id in rx_slot_list: for rx_lane in rx_lane_list: remote_info = self.topo_map.get_remote_retimer_and_slot_by_slot(rx_slot_id, rx_lane, self.route_name) if not remote_info: logging.error(f"remote_info is None for slot {rx_slot_id} lane {rx_lane}") return False remote_rt_logic_id, remote_rt_logic_lane, remote_slot_id, remote_lane_id = remote_info raw_value = reg_tool.read_opt_reg(exp_id, remote_slot_id, remote_lane_id, "ibias") tmp_err_map = {value: -1 for value in tune_value_list} result = self.topo_map.get_phys_rtmr_by_slot(rx_slot_id, rx_lane, self.route_name) if result: success, rt_phys_id, rt_phys_lane = result if success == False: logging.error(f"get_phys_rtmr_by_slot failed, result: {success}") return False lane_info = LaneErrInfo(host=(self.local_reg_tool.host), exp=exp_id, route=(self.route_name.split("-")[-1]), slot_id=rx_slot_id, logic_lane=rx_lane, rt_phys_id=rt_phys_id, rt_phys_lane=rt_phys_lane, tmp_err_map=tmp_err_map, tmp_raw_value=raw_value) lane_data_list.append(lane_info) time.sleep(0.05) is_need_reset_prbs = True for value in tune_value_list: for info in lane_data_list: remote_info = self.topo_map.get_remote_retimer_and_slot_by_slot(info.slot_id, info.logic_lane, self.route_name) if not remote_info: logging.error(f"remote_info is None for slot {info.slot_id} lane {info.logic_lane}") return False remote_rt_logic_id, remote_rt_logic_lane, remote_slot_id, remote_lane_id = remote_info reg_tool.write_opt_reg(exp_id, remote_slot_id, remote_lane_id, value, "ibias_runtime") time.sleep(0.05) time.sleep(1) if is_need_reset_prbs: logging.info(f"-------- is_need_reset_prbs: {is_need_reset_prbs}") self.prbs_tool.ResetRetimerAndEnablePrbs(exp_id, retimers) prbs_results = self.prbs_tool.prbs_check(exp_id, retimers, True) time.sleep(2) prbs_results = self.prbs_tool.prbs_check(exp_id, retimers) # logging.info(f'--------- prbs_results: {prbs_results}') is_need_reset_prbs = False for prbs in prbs_results: for info in lane_data_list: if info.rt_phys_id == prbs.rtmr_id: if info.rt_phys_lane == prbs.rtmr_lane: if prbs.locked: info.tmp_err_map[value] = prbs.err_count elif value in info.tmp_err_map: info.tmp_err_map[value] = 69999 else: logging.error(f"--------Key {value} not found in tmp_err_map, skipping assignment.") is_need_reset_prbs = True break # logging.info(lane_data_list) self.print_general_err_matrix(lane_data_list, tune_value_list) time.sleep(0.05) for info in lane_data_list: reg_tool.write_opt_reg(exp_id, info.slot_id, info.logic_lane, info.tmp_raw_value, "ibias_runtime") time.sleep(0.05) recommended_value, bad_lanes = self.simple_value_tuner(lane_data_list) reg_name = "ibias" for info, recommended_value in zip(lane_data_list, recommended_value): diff_value = recommended_value - info.tmp_raw_value if diff_value != 0: remote_info = self.topo_map.get_remote_retimer_and_slot_by_slot(info.slot_id, info.logic_lane, self.route_name) if not remote_info: logging.error(f"remote_info is None for slot {info.slot_id} lane {info.logic_lane}") return False remote_rt_logic_id, remote_rt_logic_lane, remote_slot_id, remote_lane_id = remote_info logging.info(f"Rx Slot: {info.slot_id},{info.logic_lane}, Tx Slot: {remote_slot_id}, {remote_lane_id}, adjusted from {info.tmp_raw_value} → {recommended_value} (offset={diff_value:.1f})") if is_overwrite: reg_tool.write_ibias_reg(exp_id, remote_slot_id, remote_lane_id, recommended_value) time.sleep(0.05) if bad_lanes: for slot, lane in bad_lanes: logging.error(f"Rx Slot {slot}, Rx Lane {lane}: CRITICAL - Too few clean points (<=5). Bad Modules") return True def general_auto_tune_value(self, exp_id: int, retimers, reg_name: str, tune_value_list: List[int], reg_tool: OptRegAccessTool, is_overwrite=False) -> bool: logging.info("Starting auto_tune2...") lane_data_list = [] lane_list = [ 0, 1, 2, 3, 4, 5, 6, 7] for slot_id in (0, 1, 2, 3, 4, 5, 6, 7): for lane in lane_list: raw_value = reg_tool.read_opt_reg(exp_id, slot_id, lane, reg_name) tmp_err_map = {value: -1 for value in tune_value_list} result = self.topo_map.get_phys_rtmr_by_slot(slot_id, lane, self.route_name) if result: success, rt_phys_id, rt_phys_lane = result if success == False: logging.error(f"get_phys_rtmr_by_slot failed, result: {success}") return False lane_info = LaneErrInfo(host=(self.local_reg_tool.host), exp=exp_id, slot_id=slot_id, logic_lane=lane, rt_phys_id=rt_phys_id, rt_phys_lane=rt_phys_lane, tmp_err_map=tmp_err_map, tmp_raw_value=raw_value) lane_data_list.append(lane_info) time.sleep(0.05) is_need_reset_prbs = True for value in tune_value_list: for info in lane_data_list: reg_tool.write_opt_reg(exp_id, info.slot_id, info.logic_lane, value, reg_name) time.sleep(0.05) time.sleep(1) if is_need_reset_prbs: logging.info(f"-------- is_need_reset_prbs: {is_need_reset_prbs}") self.prbs_tool.ResetRetimerAndEnablePrbs(exp_id, retimers) prbs_results = self.prbs_tool.prbs_check(exp_id, retimers, True) time.sleep(2) prbs_results = self.prbs_tool.prbs_check(exp_id, retimers) is_need_reset_prbs = False for prbs in prbs_results: if reg_tool == self.local_reg_tool: slot = self.topo_map.get_local_slot_by_retimer(prbs.rtmr_id, prbs.rtmr_lane, self.route_name) else: slot = self.topo_map.get_remote_slot_by_retimer(prbs.rtmr_id, prbs.rtmr_lane, self.route_name) if not slot: logging.error("local slot is none!") return False slot_id, lane_id = slot for info in lane_data_list: if info.slot_id == slot_id: if info.logic_lane == lane_id: if prbs.locked: info.tmp_err_map[value] = prbs.err_count elif value in info.tmp_err_map: info.tmp_err_map[value] = 69999 else: logging.error(f"--------Key {value} not found in tmp_err_map, skipping assignment.") is_need_reset_prbs = True break logging.info(lane_data_list) self.print_general_err_matrix(lane_data_list, tune_value_list) time.sleep(0.05) for info in lane_data_list: reg_tool.write_opt_reg(exp_id, info.slot_id, info.logic_lane, info.tmp_raw_value, reg_name) time.sleep(0.05) recommended_value, bad_lanes = self.simple_value_tuner(lane_data_list) for info, recommended_value in zip(lane_data_list, recommended_value): diff_value = recommended_value - info.tmp_raw_value if diff_value != 0: logging.info(f"Slot {info.slot_id}, Lane {info.logic_lane}: adjusted from {info.tmp_raw_value} → {recommended_value} (offset={diff_value:.1f})") logging.info(f"Applying {reg_name} reg write: exp_id={exp_id}, slot={info.slot_id}, lane={info.logic_lane}, {reg_name}={recommended_value}") if is_overwrite: reg_tool.write_opt_reg(exp_id, info.slot_id, info.logic_lane, recommended_value, reg_name) if bad_lanes: for slot, lane in bad_lanes: logging.error(f"Slot {slot}, Lane {lane}: CRITICAL - Too few clean points (<=5). Bad Modules") return True def simple_value_tuner(self, lane_data_list: List[LaneErrInfo], bad_threshold=1, min_good_count=5): recommendations = [] bad_lanes = [] for info in lane_data_list: err_map = info.tmp_err_map good_offsets = [value for value, err in err_map.items() if err < bad_threshold] if not good_offsets: recommended_value = info.tmp_raw_value else: recommended_value = int((min(good_offsets) + max(good_offsets)) / 2) if len(good_offsets) <= min_good_count: bad_lanes.append((info.slot_id, info.logic_lane)) recommendations.append(recommended_value) return (recommendations, bad_lanes) def simple_mgc_tuner(self, lane_data_list, bad_threshold=1, min_good_count=6): recommendations = [] bad_lanes = [] for info in lane_data_list: err_map = info.mgc_err_map good_offsets = [mgc for mgc, err in err_map.items() if err < bad_threshold] if not good_offsets: recommended_mgc = info.mgc else: recommended_mgc = int((min(good_offsets) + max(good_offsets)) / 2) if len(good_offsets) <= 5: bad_lanes.append((info.slot_id, info.logic_lane)) recommendations.append(recommended_mgc) return (recommendations, bad_lanes) def print_general_err_matrix(self, lane_data_list: List[LaneErrInfo], tune_value_list): tune_value_list = sorted(tune_value_list) header_parts = [ 'Host', 'Exp', 'Route', 'Slot', 'Lane', 'RT_Id', 'RT_Lane'] + [str(value) for value in tune_value_list] header_widths = [16, 6, 8, 6, 6, 6, 10] + [6] * len(tune_value_list) header_format = "{:<16} {:<6} {:<8} {:<6} {:<6} {:<6} {:<10} " + " ".join("{:<6}" for _ in tune_value_list) print((header_format.format)(*header_parts)) print("-" * (sum(header_widths) + len(header_widths) * 1)) for info in lane_data_list: try: err_counts = [] for value in tune_value_list: err = info.tmp_err_map.get(value, -2) if err < 0: err_str = "70000" else: err_str = (f"{err}") err_counts.append(err_str) data_format = "{:<16} {:<6} {:<8} {:<6} {:<6} {:<6} {:<10} " + " ".join("{:<6}" for _ in tune_value_list) print((data_format.format)(self.host, info.exp, info.route, info.slot_id, info.logic_lane, info.rt_phys_id, info.rt_phys_lane, *err_counts)) except Exception as e: logging.error(f"Error printing row for Slot {info.slot_id}, Lane {info.logic_lane}: {e}") continue def print_err_matrix(self, lane_data_list, mgc_offsets): mgc_offsets = sorted(mgc_offsets) header_parts = ["Host","Slot", "Lane", "RT_Id", "RT_Lane"] + [str(offset) for offset in mgc_offsets] header_widths = [16, 6, 6] + [6] * len(mgc_offsets) header_format = "{:<16} {:<6} {:<6} {:<6} {:<10} " + " ".join(f"{{:<6}}" for _ in mgc_offsets) print(header_format.format(*header_parts)) print("-" * (sum(header_widths) + len(header_widths) * 1)) for info in lane_data_list: base_mgc = info.mgc try: err_counts = [] for offset in mgc_offsets: target_mgc = base_mgc + offset err = info.mgc_err_map.get(target_mgc, -1.0) # 安全获取，防止 key 不存在 if err < 0: err_str = "N/A" else: err_str = f"{err}" err_counts.append(err_str) # 格式化输出 data_format = "{:<16} {:<6} {:<6} {:<6} {:<10} " + " ".join("{:<6}" for _ in mgc_offsets) print(data_format.format(self.host, info.slot_id, info.logic_lane, info.rt_phys_id, info.rt_phys_lane, *err_counts)) except Exception as e: logging.error(f"Error printing row for Slot {info.slot_id}, Lane {info.logic_lane}: {e}") continue def verify_rt_index_lane_index(self, rt_index: int, lane_index: int) -> Tuple[(bool, int, int)]: if rt_index in (1, 2, 3, 4): if 0 <= lane_index <= 7: new_rt_index = rt_index * 10 + 1 return ( True, new_rt_index, lane_index) else: if 8 <= lane_index <= 15: new_rt_index = rt_index * 10 + 2 new_lane_index = lane_index - 8 return (True, new_rt_index, new_lane_index) logging.error("Invalid lane_index: must be in [0, 15]") return (False, 0, 0) else: if rt_index in (11, 12, 21, 22, 31, 32, 41, 42): logging.info(f"use retimer logic index and lane, rt_index: {rt_index}, rt_lane: {lane_index}") if lane_index > 7: logging.error("For logic rt_index, lane_index must be in [0, 7]") return (False, 0, 0) return (True, rt_index, lane_index) else: logging.error(f"Invalid rt_index: {rt_index}") return (False, 0, 0) def manual_tune(self, exp_id: int, slot_id: int) -> bool: target_vpeaks = self.load_target_vpeaks_from_json(exp_id, slot_id) if target_vpeaks is None: logging.error("从文件加载target_vpeaks失败") return False else: self.match_optimal_mgc_new2(exp_id, slot_id, target_vpeaks) return True def calc_target_vpeak_new(self, exp_id: int, slot_id: int, lane_list: List[int]) -> Optional[Dict[(int, int)]]: target_vpeaks = {} logging.info(f"-------slot {slot_id}") time.sleep(0.05) self.local_reg_tool.enable_tia_agc(exp_id, slot_id) time.sleep(0.05) logging.info("----------step 2: toogle RF off") if not self.toogle_rf(exp_id, slot_id, "off"): logging.error(f"slot {slot_id}: toogle RF off fail") return target_vpeaks else: self.local_reg_tool.write_agc_reg_all_lane(exp_id, slot_id, 0) time.sleep(0.5) logging.info("----------step 3: get base_line_vpeaks, and calculate target vpeaks") base_line_vpeaks = self.local_reg_tool.read_vpeak_reg_all_lane(exp_id, slot_id) logging.info(f"base_line_vpeaks: {base_line_vpeaks}") self.toogle_rf(exp_id, slot_id, "on") self.local_reg_tool.write_agc_reg_all_lane(exp_id, slot_id, 255) time.sleep(0.5) max_vpeaks = self.local_reg_tool.read_vpeak_reg_all_lane(exp_id, slot_id) logging.info(f"max_vpeaks: {max_vpeaks}") for lane, base_vpeak in base_line_vpeaks.items(): numerator = 19 vpkdelta = round((max_vpeaks[lane] - base_vpeak) * (numerator / 29)) target_vpeaks[lane] = vpkdelta + base_vpeak diff = max_vpeaks[lane] - base_line_vpeaks[lane] if diff < 10: logging.error(f"host: {self.host}, exp: {exp_id}, slot: {slot_id}, lane: {lane}, max_vpeak is error") sys.exit(-1) # for lane, base_vpeak in base_line_vpeaks.items(): # diff = max_vpeaks[lane] - base_line_vpeaks[lane] # if diff <= 22: # logging.error(f"host: {self.host}, exp: {exp_id}, slot: {slot_id}, lane: {lane}, max_vpeak is error") # target_vpeaks[lane] = diff - 2 + base_vpeak # else: # target_vpeaks[lane] = 21 + base_vpeak logging.info(f"target_vpeaks: {target_vpeaks}") if target_vpeaks: self.save_target_vpeaks_to_json(exp_id, slot_id, target_vpeaks) self.local_reg_tool.disable_tia_agc(exp_id, slot_id) return target_vpeaks def calc_target_vpeak(self, exp_id: int, lane_list: List[int], slot_id: int) -> Dict[(int, int)]: target_vpeaks = {} logging.info(f"-------slot {slot_id}") logging.info("----------step 1: disable agc") self.disable_agc(exp_id, slot_id) logging.info("----------step 2: toogle RF off") if not self.toogle_rf(exp_id, slot_id, lane_list, "off"): logging.error(f"slot {slot_id}: toogle RF off fail") return target_vpeaks else: logging.info("----------step 3: get base_line_vpeaks, and calculate target vpeaks") base_line_vpeaks = self._read_vpeak_all_lanes(exp_id, slot_id, lane_list) logging.info(f"base_line_vpeaks: {base_line_vpeaks}") target_vpeaks = self.calculate_target_vpeak(base_line_vpeaks) logging.info(f"target_vpeaks: {target_vpeaks}") if target_vpeaks: self.save_target_vpeaks_to_json(exp_id, slot_id, target_vpeaks) logging.info("----------step 4: enable RF") self.toogle_rf(exp_id, slot_id, lane_list, "on") return target_vpeaks def disable_agc(self, exp_id: int, slot_id: int): logging.debug("----------disable_agc, step 1") agc_enable_regs = self.reg_table.get_registers_by_name("tia_agc_en") if not agc_enable_regs: logging.error("not find AGC_ENABLE register") return False else: logging.debug("----------disable_agc, step 2") success = True for reg in agc_enable_regs: logging.debug(f"----------disable_agc, step 2: reg: {reg}") result = self.bmc.SetOpticalModuleRegs(exp_id=exp_id, slot_id=slot_id, bank=(reg.bank), page=(reg.page), reg_offset=(reg.offset), size=(reg.size), hex_str="00") time.sleep(0.05) logging.debug(f"----------disable_agc, step 3: result: {result}") if not result: logging.error(f"设置寄存器失败: bank={reg.bank}, page={reg.page}, offset={reg.offset}") success = False return success def enable_agc(self, exp_id: int, slot_id: int): time.sleep(0.05) logging.debug("----------enable_agc, step 1") agc_enable_regs = self.reg_table.get_registers_by_name("tia_agc_en") if not agc_enable_regs: logging.error("not find AGC_ENABLE register") return False else: logging.debug("----------enable_agc, step 2") success = True for reg in agc_enable_regs: logging.debug(f"----------enable_agc, step 2: reg: {reg}") result = self.bmc.SetOpticalModuleRegs(exp_id=exp_id, slot_id=slot_id, bank=(reg.bank), page=(reg.page), reg_offset=(reg.offset), size=(reg.size), hex_str="FF") time.sleep(0.05) logging.info(f"----------enable_agc, step 3: result: {result}") if not result: logging.error(f"设置寄存器失败: bank={reg.bank}, page={reg.page}, offset={reg.offset}") success = False return success def toogle_rf(self, exp_id, slot_id, switch='off'): ret = False logging.info(f"---------- toogle_rf to {switch}") if switch == "off": ret = self.local_reg_tool.write_opt_reg_all_lane(exp_id, slot_id, 0, "tia_stage2") time.sleep(0.05) if ret == False: logging.error(f"---------- toogle_rf to {switch} failed") else: stage_recommended_value = 144 ret = self.local_reg_tool.write_opt_reg_all_lane(exp_id, slot_id, stage_recommended_value, "tia_stage2") time.sleep(0.05) if ret == False: logging.error(f"---------- toogle_rf to {switch} failed") return ret def calculate_target_vpeak(self, base_line_vpeaks: Dict[(int, int)]) -> Dict[(int, int)]: target_vpeaks = {} for lane, vpeak in base_line_vpeaks.items(): logging.info(f"lane={lane}, vpeak={vpeak}") target_vpeaks[lane] = vpeak + 18 logging.info(f"target_vpeaks={target_vpeaks}") return target_vpeaks def match_optimal_mgc_new2(self, exp_id: int, slot_id: int, target_vpeaks: Optional[Dict[(int, int)]]) -> bool: mgc_regs = self.reg_table.get_registers_by_name("mgc") vpeak_regs = self.reg_table.get_registers_by_name("vpeak") if mgc_regs is None or vpeak_regs is None: logging.error("not find MGC regs") return False else: success = True matched_results = {} unmatched_lanes = [] target_mgcs = {} for reg in mgc_regs: mgc_value = self.bmc.GetOpticalModuleRegs(exp_id=exp_id, slot_id=slot_id, bank=(reg.bank), page=(reg.page), reg_offset=(reg.offset), size=1) mgc_value = int(mgc_value, 16) for vpeak_reg in vpeak_regs: if vpeak_reg.lane == reg.lane: vpeak_value = self.bmc.GetOpticalModuleRegs(exp_id=exp_id, slot_id=slot_id, bank=(vpeak_reg.bank), page=(vpeak_reg.page), reg_offset=(vpeak_reg.offset), size=2) vpeak_value = int(vpeak_value, 16) logging.info(f"mgc_value: {mgc_value}, target_vpeak: {target_vpeaks[vpeak_reg.lane]}, vpeak_value: {vpeak_value}") target_mgc = (target_vpeaks[vpeak_reg.lane] - vpeak_value) * 1.5 + mgc_value logging.info(f"target_mgc: {target_mgc}") target_mgcs[reg.lane] = target_mgc break logging.info(f"----------------Target MGC: {target_mgcs}") for reg in mgc_regs: logging.info(f"mgc regs size: {len(mgc_regs)}") min_val = int(target_mgcs[reg.lane] - 10) if min_val > 255: min_val = 255 max_val = int(target_mgcs[reg.lane] + 10) if max_val > 255: logging.info("max_val is greater than 255, setting to 255") max_val = 255 logging.info(f"lane: {reg.lane}, min mgc : {min_val}, max_mgc: {max_val}, target vpeak: {target_vpeaks[reg.lane]}") step = reg.step search_values = list(range(max_val, min_val - 1, -step)) logging.info(f"Search order for lane {reg.lane}: {search_values}") is_match = False matched_mgc = None matched_vpeak = None for mgc in search_values: result = self.bmc.SetOpticalModuleRegs(exp_id=exp_id, slot_id=slot_id, bank=(reg.bank), page=(reg.page), reg_offset=(reg.offset), size=1, hex_str=f"{mgc:02X}") time.sleep(0.05) result = self.bmc.SetOpticalModuleRegs(exp_id=exp_id, slot_id=slot_id, bank=0, page=208, reg_offset=136, size=1, hex_str="01") time.sleep(0.05) for vpeak_reg in vpeak_regs: if vpeak_reg.lane == reg.lane: value = self.bmc.GetOpticalModuleRegs(exp_id=exp_id, slot_id=slot_id, bank=(vpeak_reg.bank), page=(vpeak_reg.page), reg_offset=(vpeak_reg.offset), size=2) value = int(value, 16) logging.info(f"exp_id:{exp_id}, slot_id:{slot_id}, lane:{reg.lane} -> set mgc {mgc}, Vpeak value: {value}") if value < target_vpeaks[reg.lane] + 1 and value > target_vpeaks[reg.lane] - 2: logging.info(f"----------Vpeak is matched, target vpeak:{target_vpeaks[reg.lane]}, current vpeak: {value}, current mgc:{mgc} ") is_match = True matched_mgc = mgc matched_vpeak = value break if is_match: break time.sleep(0.03) if is_match and matched_mgc is not None: if matched_vpeak is not None: if str(exp_id) not in matched_results: matched_results[str(exp_id)] = {} else: if str(slot_id) not in matched_results[str(exp_id)]: matched_results[str(exp_id)][str(slot_id)] = {} if self.mode not in matched_results[str(exp_id)][str(slot_id)]: matched_results[str(exp_id)][str(slot_id)][self.mode] = {} matched_results[str(exp_id)][str(slot_id)][self.mode][str(reg.lane)] = {'mgc':matched_mgc, 'target_vpeak':target_vpeaks[reg.lane], 'actual_vpeak':matched_vpeak} else: unmatched_lanes.append(reg.lane) if matched_results: self.save_mgc_results_to_json(exp_id, slot_id, matched_results) if unmatched_lanes: logging.error(f"以下lane未匹配到合适的MGC值: {unmatched_lanes}") return success def match_optimal_mgc_new(self, exp_id: int, slot_id: int, lane_list: List[int], target_vpeaks: Dict[(int, int)]) -> bool: matched_results = {} unmatched_lanes = [] for index, lane_id in enumerate(lane_list): is_match = False matched_mgc = None matched_vpeak = None reg = self.reg_table.get_register_by_logic_lane("mgc", lane_id) if reg is None or reg.valid_range is None: logging.error(f"match_optimal_mgc_new error. exp:{exp_id}, slot:{slot_id}, lane: {lane_id}, register name: mgc") return False min_val = reg.valid_range[0] max_val = reg.valid_range[1] step = reg.step if reg.step is not None else 1 wt_mgc = int((min_val + max_val) / 2) target_vpeak = target_vpeaks[lane_id] target_vpeak_range = [target_vpeak, target_vpeak + 1] count = 0 while 1: ret = self.local_reg_tool.write_mgc_reg(exp_id, slot_id, lane_id, wt_mgc) time.sleep(0.05) if ret == False: logging.error(f"match_optimal_mgc_new error. exp:{exp_id}, slot:{slot_id}, lane: {lane_id}, register name: mgc") return False ret = self.local_reg_tool.write_confirm_reg(exp_id, slot_id) if ret == False: logging.error(f"match_optimal_mgc_new error. exp:{exp_id}, slot:{slot_id}, lane: {lane_id}, register name: confirm") return False time.sleep(0.05) current_vpeak = self.local_reg_tool.read_vpeak_reg(exp_id, slot_id, lane_id) logging.info(f"exp_id:{exp_id}, slot_id:{slot_id}, lane:{lane_id} -> set mgc {wt_mgc}, Vpeak value: {current_vpeak}, target: {target_vpeak}") diff_vpeak = abs(target_vpeak - current_vpeak) diff_vpeak_step = { 0: 0, 1: 0, 2: 0, 3: 1, 4: 2, 5: 3, 6: 3, 7: 4, 8: 5} extra_step = diff_vpeak_step.get(diff_vpeak, 5) if current_vpeak > target_vpeak_range[1]: wt_mgc -= step + extra_step if current_vpeak < target_vpeak_range[0]: wt_mgc += step + extra_step if current_vpeak >= target_vpeak_range[0]: if current_vpeak <= target_vpeak_range[1]: is_match = True matched_mgc = wt_mgc matched_vpeak = current_vpeak break if count > 30: logging.error(f"vpeak is not match, target vpeak:{target_vpeak}, current vpeak: {current_vpeak}, current mgc:{wt_mgc} ") sys.exit(-1) break count += 1 if is_match: logging.info(f"------------------ Vpeak is matched, target vpeak:{target_vpeak}, current vpeak: {current_vpeak}, current mgc:{wt_mgc} ") if str(exp_id) not in matched_results: matched_results[str(exp_id)] = {} if str(slot_id) not in matched_results[str(exp_id)]: matched_results[str(exp_id)][str(slot_id)] = {} if self.mode not in matched_results[str(exp_id)][str(slot_id)]: matched_results[str(exp_id)][str(slot_id)][self.mode] = {} matched_results[str(exp_id)][str(slot_id)][self.mode][str(reg.lane)] = {'mgc':matched_mgc, 'target_vpeak':target_vpeaks[lane_id], 'actual_vpeak':matched_vpeak} else: unmatched_lanes.append(lane_id) if matched_results: self.save_mgc_results_to_json(exp_id, slot_id, matched_results) if unmatched_lanes: logging.error(f"以下lane未匹配到合适的MGC值: {unmatched_lanes}") return False else: return True def match_optimal_mgc_old(self, exp_id: int, slot_id: int, lane_list: List[int], target_vpeaks: Dict[(int, int)]) -> bool: mgc_regs = self.reg_table.get_registers_by_name("mgc") vpeak_regs = self.reg_table.get_registers_by_name("vpeak") if mgc_regs is None or vpeak_regs is None: logging.error("not find MGC regs") return False else: success = True matched_results = {} unmatched_lanes = [] for reg in mgc_regs: logging.info(f"mgc regs size: {len(mgc_regs)}") logging.info(f"lane: {reg.lane}, target vpeak:{target_vpeaks[reg.lane]}, mgc valid_range: {reg.valid_range}, step: {reg.step}") min_val = reg.valid_range[0] max_val = reg.valid_range[1] step = reg.step mid_val = (min_val + max_val) // 2 mid_val = mid_val // step * step search_values = [ mid_val] offset = step while len(search_values) < (max_val - min_val) // step + 1: right_val = mid_val + offset if right_val <= max_val: search_values.append(right_val) left_val = mid_val - offset if left_val >= min_val: search_values.append(left_val) offset += step if len(search_values) >= (max_val - min_val) // step + 1: break is_match = False matched_mgc = None matched_vpeak = None for mgc in search_values: result = self.bmc.SetOpticalModuleRegs(exp_id=exp_id, slot_id=slot_id, bank=(reg.bank), page=(reg.page), reg_offset=(reg.offset), size=1, hex_str=f"{mgc:02X}") time.sleep(0.02) result = self.bmc.SetOpticalModuleRegs(exp_id=exp_id, slot_id=slot_id, bank=0, page=208, reg_offset=136, size=1, hex_str="01") time.sleep(0.02) for vpeak_reg in vpeak_regs: if vpeak_reg.lane == reg.lane: value = self.bmc.GetOpticalModuleRegs(exp_id=exp_id, slot_id=slot_id, bank=(vpeak_reg.bank), page=(vpeak_reg.page), reg_offset=(vpeak_reg.offset), size=2) value = int(value, 16) logging.info(f"exp_id:{exp_id}, slot_id:{slot_id}, lane:{reg.lane} -> set mgc {mgc}, Vpeak value: {value}") if value < target_vpeaks[reg.lane] + 2 and value > target_vpeaks[reg.lane] - 2: logging.info(f"----------Vpeak is matched, target vpeak:{target_vpeaks[reg.lane]}, current vpeak: {value}, current mgc:{mgc} ") is_match = True matched_mgc = mgc matched_vpeak = value break if is_match: break time.sleep(0.03) if is_match: if matched_mgc is not None: if matched_vpeak is not None: if str(exp_id) not in matched_results: matched_results[str(exp_id)] = {} else: if str(slot_id) not in matched_results[str(exp_id)]: matched_results[str(exp_id)][str(slot_id)] = {} if self.mode not in matched_results[str(exp_id)][str(slot_id)]: matched_results[str(exp_id)][str(slot_id)][self.mode] = {} matched_results[str(exp_id)][str(slot_id)][self.mode][str(reg.lane)] = {'mgc':matched_mgc, 'target_vpeak':target_vpeaks[reg.lane], 'actual_vpeak':matched_vpeak} unmatched_lanes.append(reg.lane) if matched_results: self.save_mgc_results_to_json(exp_id, slot_id, matched_results) if unmatched_lanes: logging.error(f"以下lane未匹配到合适的MGC值: {unmatched_lanes}") return success def _read_vpeak_all_lanes(self, exp_id: int, slot_id: int, lane_list: List[int]) -> Dict[(int, int)]: vpeak_values = {} for lane in lane_list: vpeak_int = self.local_reg_tool.read_vpeak_reg(exp_id, slot_id, lane) vpeak_values[lane] = vpeak_int return vpeak_values def _write_mgc_all_lanes(self, exp_id: int, slot_id: int, lane_list: List[int], mgc: int) -> bool: for lane_id in lane_list: ret = self.local_reg_tool.write_mgc_reg(exp_id, slot_id, lane_id, mgc) if ret == False: logging.error(f"write mgc reg failed, exp_id:{exp_id}, slot_id:{slot_id}, lane_id:{lane_id}, mgc:{mgc}") return False ret = self.local_reg_tool.write_confirm_reg(exp_id, slot_id) if ret == False: logging.error(f"write confirm reg failed, exp_id:{exp_id}, slot_id:{slot_id}") return False else: return True def _write_agc_all_lanes(self, exp_id: int, slot_id: int, lane_list: List[int], agc: int) -> bool: for lane_id in lane_list: ret = self.local_reg_tool.write_agc_reg(exp_id, slot_id, lane_id, agc) time.sleep(0.05) if ret == False: logging.error(f"write agc reg failed, exp_id:{exp_id}, slot_id:{slot_id}, lane_id:{lane_id}, agc:{agc}") return False ret = self.local_reg_tool.write_confirm_reg(exp_id, slot_id) if ret == False: logging.error(f"write confirm reg failed, exp_id:{exp_id}, slot_id:{slot_id}") return False else: return True def _generate_host_filename(self) -> str: safe_host = self.host.replace(":", "_").replace("/", "_").replace("\\", "_") return f"target_vpeaks_{safe_host}.json" def _generate_mgc_host_filename(self) -> str: safe_host = self.host.replace(":", "_").replace("/", "_").replace("\\", "_") return f"mgc_{safe_host}.json" def save_target_vpeaks_to_json(self, exp_id: int, slot_id: int, target_vpeaks: Dict[(int, int)]) -> bool: try: filename = self._generate_host_filename() os.makedirs((os.path.dirname(filename) if os.path.dirname(filename) else "."), exist_ok=True) if os.path.exists(filename): with open(filename, "r") as f: data = json.load(f) else: data = {} if str(exp_id) not in data: data[str(exp_id)] = {} if str(slot_id) not in data[str(exp_id)]: data[str(exp_id)][str(slot_id)] = {} data[str(exp_id)][str(slot_id)][self.mode] = {str(k): v for k, v in target_vpeaks.items()} with open(filename, "w") as f: json.dump(data, f, indent=4) logging.info(f"target_vpeaks saved to {filename}, path: {exp_id}/{slot_id}/{self.mode}") return True except Exception as e: logging.error(f"Failed to save target_vpeaks to JSON file: {e}") return False def load_target_vpeaks_from_json(self, exp_id: int, slot_id: int) -> Optional[Dict[(int, int)]]: try: filename = self._generate_host_filename() if not os.path.exists(filename): logging.error(f"File {filename} does not exist") return with open(filename, "r") as f: data = json.load(f) if str(exp_id) not in data or str(slot_id) not in data[str(exp_id)] or self.mode not in data[str(exp_id)][str(slot_id)]: logging.error(f"Path {exp_id}/{slot_id}/{self.mode} not found in {filename}") return target_vpeaks_data = data[str(exp_id)][str(slot_id)][self.mode] target_vpeaks = {int(k): v for k, v in target_vpeaks_data.items()} logging.info(f"target_vpeaks loaded successfully from {filename}, path: {exp_id}/{slot_id}/{self.mode}") return target_vpeaks except Exception as e: logging.error(f"Failed to load target_vpeaks from JSON file: {e}") return def save_mgc_results_to_json(self, exp_id: int, slot_id: int, matched_results: Dict[(str, Any)]) -> bool: try: filename = self._generate_mgc_host_filename() os.makedirs((os.path.dirname(filename) if os.path.dirname(filename) else "."), exist_ok=True) if os.path.exists(filename): with open(filename, "r") as f: data = json.load(f) else: data = {} for exp_key, exp_data in matched_results.items(): if exp_key not in data: data[exp_key] = {} for slot_key, slot_data in exp_data.items(): if slot_key not in data[exp_key]: data[exp_key][slot_key] = {} for mode_key, mode_data in slot_data.items(): if mode_key not in data[exp_key][slot_key]: data[exp_key][slot_key][mode_key] = {} data[exp_key][slot_key][mode_key].update(mode_data) with open(filename, "w") as f: json.dump(data, f, indent=4) logging.info(f"MGC matching results saved to {filename}") return True except Exception as e: logging.error(f"Failed to save MGC results to JSON file: {e}") return False