Coverage for /opt/cloud/slavespace/usr1/096471637100f3de0fcfc01072822a80/ut/src/asys/diagnose/asys_diagnose.py: 52%
208 statements
« prev ^ index » next coverage.py v7.14.1, created at 2026-08-21 15:37 +0800
1#!/usr/bin/env python3
2# -*- coding: utf-8 -*-
3# ----------------------------------------------------------------------------
4# Copyright (c) 2025 Huawei Technologies Co., Ltd.
5#
6# Licensed under the Apache License, Version 2.0 (the "License");
7# you may not use this file except in compliance with the License.
8# You may obtain a copy of the License at
9#
10# http://www.apache.org/licenses/LICENSE-2.0
11#
12# Unless required by applicable law or agreed to in writing, software
13# distributed under the License is distributed on an "AS IS" BASIS,
14# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
15# See the License for the specific language governing permissions and
16# limitations under the License.
17# ----------------------------------------------------------------------------
19import os
20from pathlib import Path
21import re
22import sys
23import threading
24from datetime import datetime, timezone
26from common import log_error, log_warning, log_info, open_log, close_log, log_debug
27from common.const import RetCode, UNKNOWN, ScreenResult
28from common.const import HBM_MIN_TIMEOUT, CPU_MIN_TIMEOUT, DETECT_MAX_TIMEOUT
29from common.cmd_run import run_linux_cmd, run_cmd_output
30from common import AsysDiagnoseSupportedChip, DeviceInfo, ChipHandler
31from common.file_operate import FileOperate as f
32from params.param_dict import ParamDict
33from view.table import generate_report
34from view.progress_display import waiting
35from drv import EnvVarName
37HBM_MODE = "hbm_detect"
38CPU_MODE = "cpu_detect"
39COMPONENT_MODE = "component"
40opp_kernels = ["ops_cv", "ops_legacy", "ops_math", "ops_nn", "ops_transformer"]
41SUPPORT_CHIPS = ChipHandler().get_support_chip_regex_list()
44class AsysDiagnose():
45 """"""
47 def __init__(self):
48 self.finish_flag = False
49 self.device_obj = DeviceInfo()
50 self.devices_num = self.device_obj.get_device_count()
52 @staticmethod
53 def __get_hbm_table_data(device_id, ret):
54 link_symbol = ", "
55 devices_ecc = []
57 if device_id is False:
58 devices_ret = [ret[key][0] for key in sorted(ret.keys())]
59 devices_ecc = [ret[key][1] for key in sorted(ret.keys())]
60 if ((ScreenResult.PASS.value in devices_ret and ScreenResult.WARN.value in devices_ret) or
61 len(devices_ret) == 1):
62 devices_str = ", ".join(devices_ret)
63 else:
64 devices_str = f"{devices_ret[0]} - All"
65 else:
66 devices_str = ret[device_id][0]
68 hbm_table = "HBM Detect"
69 if device_id is False and len(devices_ecc) != 1:
70 devices_hbm = ("(" + link_symbol.join(devices_ecc) + ")")
71 ret_data_str = [[hbm_table, devices_str], ["", devices_hbm]]
72 elif len(devices_ecc) == 1:
73 devices_str += ("(" + link_symbol.join(devices_ecc) + ")")
74 ret_data_str = [[hbm_table, devices_str]]
75 else:
76 devices_str += ("(" + ret[device_id][1] + ")")
77 ret_data_str = [[hbm_table, devices_str]]
79 return ret_data_str
81 @staticmethod
82 def __get_other_table_data(device_id, ret):
83 if device_id is False:
84 # without '-d', displays information about all devices.
85 devices_ret = [ret[key] for key in sorted(ret.keys())]
86 if len(set(devices_ret)) > 1 or len(devices_ret) == 1:
87 devices_str = ", ".join(devices_ret)
88 else:
89 devices_str = f"{devices_ret[0]} - All"
90 else:
91 # with '-d', displays information about the input device.
92 devices_str = ret[device_id]
94 return devices_str
96 def print_save(self, device_id, ret, run_mode):
97 """print screen & save file"""
98 if device_id is False:
99 # without '-d', displays information about all devices.
100 table_header = [[f"Group of {len(ret)} Device", "Diagnostic Result"]]
101 else:
102 # with '-d', displays information about the input device.
103 table_header = [[f"Device ID: {device_id}", "Diagnostic Result"]]
105 if run_mode == HBM_MODE:
106 ret_data_str = self.__get_hbm_table_data(device_id, ret)
107 table_data = {" Hardware ": ret_data_str}
108 elif run_mode == CPU_MODE:
109 ret_data_str = self.__get_other_table_data(device_id, ret)
110 table_data = {" Hardware ": [["CPU Detect", ret_data_str]]}
111 elif run_mode == COMPONENT_MODE:
112 ret_data_str = self.__get_other_table_data(device_id, ret)
113 table_data = {" Component ": [["AI Vector", ret_data_str]]}
114 else:
115 ret_data_str = self.__get_other_table_data(device_id, ret)
116 table_data = {" Performance ": [["Stress Detect", ret_data_str]]}
117 ret_str = generate_report(table_header, table_data)
118 sys.stdout.write(ret_str) # print screen
120 # save result to file
121 output_path = ParamDict().get_arg("output")
122 utc_dt = datetime.now(timezone.utc) # UTC time
123 dir_name = utc_dt.astimezone().strftime('%Y%m%d%H%M%S%f')[:-3]
124 if output_path is not False:
125 try:
126 output_file = os.path.join(ParamDict().get_arg("output"), f"diagnose_result_{dir_name}.txt")
127 with open(output_file, "w", encoding="utf8") as file:
128 file.write(ret_str)
129 open_log()
130 log_info(f"output file: {os.path.abspath(output_file)}")
131 close_log()
132 except Exception as e:
133 log_error(f"Failed to save result: {e}.")
135 def _check_support(self, run_mode):
136 # check VMs and docker
137 if not run_linux_cmd("systemd-detect-virt", "none"):
138 log_error("The diagnose command cannot be executed on VMs and docker.")
139 return False
141 # username
142 if os.getuid() != 0: # 0 -> administrator
143 log_error("The diagnose command must be executed as the root user.")
144 return False
146 if run_mode == "stress_detect":
147 # check opp_kernel, ${install_path}/latest/opp_kernel
148 opp_path = EnvVarName().opp_path
149 if not opp_path:
150 log_error("The diagnose command can be executed only after the opp_kernel is installed.")
151 return False
152 for ops in opp_kernels:
153 if not os.path.isfile(os.path.join(opp_path, "..", "share", "info", ops, "version.info")):
154 log_error(f"The diagnose command can be executed only after the {ops} is installed.")
155 return False
157 timeout = ParamDict().get_arg("timeout")
158 if run_mode == HBM_MODE and timeout is not False:
159 if timeout < HBM_MIN_TIMEOUT or timeout > DETECT_MAX_TIMEOUT:
160 log_error(f"The value of timeout must be in the range of [{HBM_MIN_TIMEOUT}, {DETECT_MAX_TIMEOUT}].")
161 return False
163 if run_mode == CPU_MODE and timeout is not False:
164 if timeout < CPU_MIN_TIMEOUT or timeout > DETECT_MAX_TIMEOUT:
165 log_error(f"The value of timeout must be in the range of [{CPU_MIN_TIMEOUT}, {DETECT_MAX_TIMEOUT}].")
166 return False
167 if self.devices_num == 0:
168 return False
169 return True
171 @staticmethod
172 def check_chip_support(_chip_info):
173 if any(re.search(regexp, _chip_info) for regexp in SUPPORT_CHIPS):
174 return True
175 return False
177 def get_diagnose_devices_chip_info(self, device_id):
178 diagnose_devices = []
179 chip_info = UNKNOWN
180 diagnose_supported = AsysDiagnoseSupportedChip()
181 if device_id is False:
182 for i in range(self.devices_num):
183 ret, _chip_info = diagnose_supported.get_supported_chip_info(i)
184 if not ret:
185 log_error(f"The diagnose command does not support on device_{i}: {_chip_info}.")
186 continue
187 diagnose_devices.append(i)
188 chip_info = _chip_info
189 else:
190 ret, _chip_info = diagnose_supported.get_supported_chip_info(device_id)
191 if not ret:
192 log_error(f"The diagnose command does not support {_chip_info}.")
193 else:
194 chip_info = _chip_info
195 diagnose_devices = [device_id]
196 return diagnose_devices, chip_info
198 def hardware_detect(self, run_mode):
199 """detect cmd main"""
200 device_id = ParamDict().get_arg("device_id")
201 diagnose_devices, chip_info = self.get_diagnose_devices_chip_info(device_id)
202 if not diagnose_devices or chip_info == UNKNOWN:
203 return False
205 if not self._check_support(run_mode):
206 return False
208 # load dll: libascend_ml.so
209 if self.device_obj.ascend_ml == RetCode.FAILED:
210 return False
212 t = threading.Thread(target=self.wait_view, daemon=True)
213 t.start()
214 # Multi-thread parallel execution
215 handler = ChipHandler().get_handler(chip_info)
216 if handler is None:
217 log_error(f"{chip_info} is not supported.")
218 return False
219 ret = handler.run_diagnose(self.device_obj, diagnose_devices, run_mode)
221 # ret add not support device
222 if device_id is False:
223 for i in range(self.devices_num):
224 if i in diagnose_devices:
225 continue
226 ret[i] = [ScreenResult.WARN.value, "0"] if run_mode == HBM_MODE else ScreenResult.WARN.value
228 if ScreenResult.WARN.value in ret.values():
229 log_warning("Diagnosis results have failed, please analyze aml logs")
230 self.finish_flag = True
231 t.join()
232 # screen print & save ret to file
233 self.print_save(device_id, ret, run_mode)
234 return True
236 @staticmethod
237 def run_msaicerr_cmd(msaicerr_path, device_id, res):
238 cmd = f"{sys.executable} {msaicerr_path} --env -dev={device_id}"
239 log_debug(f"Start run: {cmd}")
240 ret, output = run_cmd_output(cmd)
241 res[device_id] = ScreenResult.PASS.value if ret else ScreenResult.FAIL.value
242 return output
244 def env_detect(self, run_mode):
245 device_id = ParamDict().get_arg('device_id')
246 msaicerr_path = ParamDict().tools_path.parents[1].joinpath("msaicerr", "msaicerr.py")
247 log_debug(f"Start load msaicerr tools path: {msaicerr_path}")
248 log_debug(f"Device num is {self.devices_num}")
249 if not self.devices_num:
250 log_error(f"The chip does not have a device for execution.")
251 return False
252 if not os.path.exists(msaicerr_path):
253 log_error("The path of the msaicerr tool cannot be found, please install the whole package.")
254 return False
255 res = {}
256 output = ""
257 if device_id is False:
258 for i in range(self.devices_num):
259 log_debug(f"Start run device {i}")
260 output = self.run_msaicerr_cmd(msaicerr_path, i, res)
261 else:
262 output = self.run_msaicerr_cmd(msaicerr_path, device_id, res)
263 self.print_save(device_id, res, run_mode)
264 if ScreenResult.FAIL.value in res.values():
265 debug_info_path = Path(os.getcwd(), 'debug_info.txt')
266 if not f.check_access(os.getcwd(), os.W_OK) or (debug_info_path.exists() and
267 not f.check_access(debug_info_path, os.W_OK)):
268 open_log()
269 log_error("The current directory or debug_info.txt is immutable, Please check.")
270 close_log()
272 else:
273 sys.stdout.write(output)
274 return False
275 return True
277 def run(self):
278 """diagnose cmd main"""
279 run_mode = ParamDict().get_arg("run_mode")
280 if run_mode == COMPONENT_MODE:
281 return self.env_detect(run_mode)
282 else:
283 return self.hardware_detect(run_mode)
285 def wait_view(self):
286 while not self.finish_flag:
287 waiting()
288 continue