Coverage for /opt/cloud/slavespace/usr1/096471637100f3de0fcfc01072822a80/ut/src/asys/diagnose/asys_diagnose.py: 61%
228 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-08-19 17:46 +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, 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, LoadSoType
37HBM_MODE = "hbm_detect"
38CPU_MODE = "cpu_detect"
39COMPONENT_MODE = "component"
40AICORE_STL_MODE = "aicore_stl_detect"
41opp_kernels = ["ops_cv", "ops_legacy", "ops_math", "ops_nn", "ops_transformer"]
42SUPPORT_CHIPS = ChipHandler().get_support_chip_regex_list()
45class AsysDiagnose():
46 """"""
48 def __init__(self):
49 self.finish_flag = False
50 self.device_obj = DeviceInfo()
51 self.devices_num = self.device_obj.get_device_count()
53 @staticmethod
54 def __get_hbm_table_data(device_id, ret):
55 link_symbol = ", "
56 devices_ecc = []
58 if device_id is False:
59 devices_ret = [ret[key][0] for key in sorted(ret.keys())]
60 devices_ecc = [ret[key][1] for key in sorted(ret.keys())]
61 if ((ScreenResult.PASS.value in devices_ret and ScreenResult.WARN.value in devices_ret) or
62 len(devices_ret) == 1):
63 devices_str = ", ".join(devices_ret)
64 else:
65 devices_str = f"{devices_ret[0]} - All"
66 else:
67 devices_str = ret[device_id][0]
69 hbm_table = "HBM Detect"
70 if device_id is False and len(devices_ecc) != 1:
71 devices_hbm = ("(" + link_symbol.join(devices_ecc) + ")")
72 ret_data_str = [[hbm_table, devices_str], ["", devices_hbm]]
73 elif len(devices_ecc) == 1:
74 devices_str += ("(" + link_symbol.join(devices_ecc) + ")")
75 ret_data_str = [[hbm_table, devices_str]]
76 else:
77 devices_str += ("(" + ret[device_id][1] + ")")
78 ret_data_str = [[hbm_table, devices_str]]
80 return ret_data_str
82 @staticmethod
83 def __get_other_table_data(device_id, ret):
84 if device_id is False:
85 # without '-d', displays information about all devices.
86 devices_ret = [ret[key] for key in sorted(ret.keys())]
87 if len(set(devices_ret)) > 1 or len(devices_ret) == 1:
88 devices_str = ", ".join(devices_ret)
89 else:
90 devices_str = f"{devices_ret[0]} - All"
91 else:
92 # with '-d', displays information about the input device.
93 devices_str = ret[device_id]
95 return devices_str
97 def print_save(self, device_id, ret, run_mode):
98 """print screen & save file"""
99 if device_id is False:
100 # without '-d', displays information about all devices.
101 table_header = [[f"Group of {len(ret)} Device", "Diagnostic Result"]]
102 else:
103 # with '-d', displays information about the input device.
104 table_header = [[f"Device ID: {device_id}", "Diagnostic Result"]]
106 if run_mode == HBM_MODE:
107 ret_data_str = self.__get_hbm_table_data(device_id, ret)
108 table_data = {" Hardware ": ret_data_str}
109 elif run_mode == CPU_MODE:
110 ret_data_str = self.__get_other_table_data(device_id, ret)
111 table_data = {" Hardware ": [["CPU Detect", ret_data_str]]}
112 elif run_mode == COMPONENT_MODE:
113 ret_data_str = self.__get_other_table_data(device_id, ret)
114 table_data = {" Component ": [["AI Vector", ret_data_str]]}
115 elif run_mode == AICORE_STL_MODE:
116 ret_data_str = self.__get_other_table_data(device_id, ret)
117 table_data = {" Hardware ": [["AICore STL Detect", ret_data_str]]}
118 else:
119 ret_data_str = self.__get_other_table_data(device_id, ret)
120 table_data = {" Performance ": [["Stress Detect", ret_data_str]]}
121 ret_str = generate_report(table_header, table_data)
122 sys.stdout.write(ret_str) # print screen
124 # save result to file
125 output_path = ParamDict().get_arg("output")
126 utc_dt = datetime.now(timezone.utc) # UTC time
127 dir_name = utc_dt.astimezone().strftime('%Y%m%d%H%M%S%f')[:-3]
128 if output_path is not False:
129 try:
130 output_file = os.path.join(ParamDict().get_arg("output"), f"diagnose_result_{dir_name}.txt")
131 with open(output_file, "w", encoding="utf8") as file:
132 file.write(ret_str)
133 log_info(f"output file: {os.path.abspath(output_file)}", force=True)
134 except Exception as e:
135 log_error(f"Failed to save result: {e}.")
137 def _check_support(self, run_mode):
138 # check VMs and docker
139 if not run_linux_cmd("systemd-detect-virt", "none"):
140 log_error("The diagnose command cannot be executed on VMs and docker.")
141 return False
143 # username
144 if os.getuid() != 0: # 0 -> administrator
145 log_error("The diagnose command must be executed as the root user.")
146 return False
148 if run_mode == "stress_detect":
149 # check opp_kernel, ${install_path}/latest/opp_kernel
150 opp_path = EnvVarName().opp_path
151 if not opp_path:
152 log_error("The diagnose command can be executed only after the opp_kernel is installed.")
153 return False
154 for ops in opp_kernels:
155 if not os.path.isfile(os.path.join(opp_path, "..", "share", "info", ops, "version.info")):
156 log_error(f"The diagnose command can be executed only after the {ops} is installed.")
157 return False
159 timeout = ParamDict().get_arg("timeout")
160 if run_mode == HBM_MODE and timeout is not False:
161 if timeout < HBM_MIN_TIMEOUT or timeout > DETECT_MAX_TIMEOUT:
162 log_error(f"The value of timeout must be in the range of [{HBM_MIN_TIMEOUT}, {DETECT_MAX_TIMEOUT}].")
163 return False
165 if run_mode == CPU_MODE and timeout is not False:
166 if timeout < CPU_MIN_TIMEOUT or timeout > DETECT_MAX_TIMEOUT:
167 log_error(f"The value of timeout must be in the range of [{CPU_MIN_TIMEOUT}, {DETECT_MAX_TIMEOUT}].")
168 return False
169 if run_mode == AICORE_STL_MODE and timeout is not False:
170 log_warning(
171 "The --timeout argument is not supported in aicore_stl_detect mode and will be ignored.",
172 force=True
173 )
174 if self.devices_num == 0:
175 return False
176 return True
178 @staticmethod
179 def check_chip_support(_chip_info):
180 if any(re.search(regexp, _chip_info) for regexp in SUPPORT_CHIPS):
181 return True
182 return False
184 def get_diagnose_devices_chip_info(self, device_id):
185 diagnose_devices = []
186 chip_info = UNKNOWN
187 diagnose_supported = AsysDiagnoseSupportedChip()
188 if device_id is False:
189 for i in range(self.devices_num):
190 ret, _chip_info = diagnose_supported.get_supported_chip_info(i)
191 if not ret:
192 log_error(f"The diagnose command does not support on device_{i}: {_chip_info}.")
193 continue
194 diagnose_devices.append(i)
195 chip_info = _chip_info
196 else:
197 ret, _chip_info = diagnose_supported.get_supported_chip_info(device_id)
198 if not ret:
199 log_error(f"The diagnose command does not support {_chip_info}.")
200 else:
201 chip_info = _chip_info
202 diagnose_devices = [device_id]
203 return diagnose_devices, chip_info
205 def hardware_detect(self, run_mode):
206 """detect cmd main"""
207 device_id = ParamDict().get_arg("device_id")
208 diagnose_devices, chip_info = self.get_diagnose_devices_chip_info(device_id)
209 if not diagnose_devices or chip_info == UNKNOWN:
210 return False
212 # aicore_stl_detect is only supported on Ascend950.
213 if run_mode == AICORE_STL_MODE:
214 diagnose_devices = self.__filter_aicore_stl_devices(diagnose_devices)
215 if not diagnose_devices:
216 log_error("The aicore_stl_detect mode is only supported on Ascend950.")
217 return False
219 if not self._check_support(run_mode):
220 return False
222 # load dll: libascend_ml.so (or libaml_aicore_stl.so for AICore STL mode)
223 if run_mode == AICORE_STL_MODE:
224 self.device_obj.aml_aicore_stl = LoadSoType().get_aml_aicore_stl()
225 if self.device_obj.aml_aicore_stl == RetCode.FAILED or self.device_obj.aml_aicore_stl is None:
226 log_error("Failed to load libaml_aicore_stl.so for aicore_stl_detect.")
227 return False
228 elif self.device_obj.ascend_ml == RetCode.FAILED:
229 return False
231 t = threading.Thread(target=self.wait_view, daemon=True)
232 t.start()
233 # Multi-thread parallel execution
234 handler = ChipHandler().get_handler(chip_info)
235 if handler is None:
236 log_error(f"{chip_info} is not supported.")
237 return False
238 ret = handler.run_diagnose(self.device_obj, diagnose_devices, run_mode)
240 # ret add not support device
241 if device_id is False:
242 for i in range(self.devices_num):
243 if i in diagnose_devices:
244 continue
245 ret[i] = [ScreenResult.WARN.value, "0"] if run_mode == HBM_MODE else ScreenResult.WARN.value
247 if ScreenResult.WARN.value in ret.values():
248 log_warning("Diagnosis results have failed, please analyze aml logs")
249 self.finish_flag = True
250 t.join()
251 # screen print & save ret to file
252 self.print_save(device_id, ret, run_mode)
253 return True
255 @staticmethod
256 def run_msaicerr_cmd(msaicerr_path, device_id, res):
257 cmd = f"{sys.executable} {msaicerr_path} --env -dev={device_id}"
258 log_debug(f"Start run: {cmd}")
259 ret, output = run_cmd_output(cmd)
260 res[device_id] = ScreenResult.PASS.value if ret else ScreenResult.FAIL.value
261 return output
263 def env_detect(self, run_mode):
264 device_id = ParamDict().get_arg('device_id')
265 msaicerr_path = ParamDict().tools_path.parents[1].joinpath("msaicerr", "msaicerr.py")
266 log_debug(f"Start load msaicerr tools path: {msaicerr_path}")
267 log_debug(f"Device num is {self.devices_num}")
268 if not self.devices_num:
269 log_error(f"The chip does not have a device for execution.")
270 return False
271 if not os.path.exists(msaicerr_path):
272 log_error("The path of the msaicerr tool cannot be found, please install the whole package.")
273 return False
274 res = {}
275 output = ""
276 if device_id is False:
277 for i in range(self.devices_num):
278 log_debug(f"Start run device {i}")
279 output = self.run_msaicerr_cmd(msaicerr_path, i, res)
280 else:
281 output = self.run_msaicerr_cmd(msaicerr_path, device_id, res)
282 self.print_save(device_id, res, run_mode)
283 if ScreenResult.FAIL.value in res.values():
284 debug_info_path = Path(os.getcwd(), 'debug_info.txt')
285 if not f.check_access(os.getcwd(), os.W_OK) or (debug_info_path.exists() and
286 not f.check_access(debug_info_path, os.W_OK)):
287 log_error("The current directory or debug_info.txt is immutable, Please check.")
289 else:
290 sys.stdout.write(output)
291 return False
292 return True
294 def run(self):
295 """diagnose cmd main"""
296 run_mode = ParamDict().get_arg("run_mode")
297 if run_mode == COMPONENT_MODE:
298 return self.env_detect(run_mode)
299 else:
300 return self.hardware_detect(run_mode)
302 def wait_view(self):
303 while not self.finish_flag:
304 waiting()
305 continue
307 def __filter_aicore_stl_devices(self, diagnose_devices):
308 """aicore_stl_detect 仅支持 Ascend950:过滤掉非 950 的 device 并告警。"""
309 stl_devices = []
310 for dev in diagnose_devices:
311 chip_info = self.device_obj.get_chip_info(dev)
312 if chip_info != UNKNOWN and re.search("950", chip_info):
313 stl_devices.append(dev)
314 else:
315 log_warning(f"device_{dev} ({chip_info}) does not support aicore_stl_detect "
316 "(Ascend950 only), skipped.")
317 return stl_devices