Coverage for /opt/cloud/slavespace/usr1/096471637100f3de0fcfc01072822a80/ut/src/asys/info/asys_info.py: 94%
254 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-08-14 17:42 +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
20import re
21import sys
23from common import run_command
24from common import get_project_conf, get_ascend_home
25from common import get_device
26from common import FileOperate as f
27from common import timeout_decorator
28from common import log_error
29from common.const import UNKNOWN, NONE, MEMORY_FREQUENCY, HBM_FREQUENCY, CONTROL_CPU_FREQUENCY, CannPkg
30from common.const import AI_CORE_USE, AI_CPU_USE, CONTROL_CPU_USE, MEM_BANDWIDTH_USE, NOT_SUPPORT, MAX_CHAR_LINE
31from common.const import ConfigOperateType, ConfigOptionName, ALL_NOT_SUPPORTED_CHIP_TYPE, ALL_SUPPORTED_CHIP_TYPE
32from common.const import GET_DEVICES_INFO_TIMEOUT
33from params import ParamDict
34from view import generate_report
36CPU_TABLE_TITLE = ' CPU Information '
37AIC_TABLE_TITLE = ' AI Core Information '
38BUS_TABLE_TITLE = ' Bus Information '
39MEM_TABLE_TITLE = ' Memory Information '
41HOST_VERSION = ' Host Version '
42DEVICE_VERSION = ' Device Version '
44PCIE_INFO = ' PCIe Info '
46LSPCI_GREP_VERSION = "lspci | grep -E 'd100|d500|d801|d802|d803|d806'"
47GET_COUNT = 'wc -l'
50class AsysInfo:
51 def __init__(self):
52 self.device_info = get_device()
53 self.device_num = self.device_info.get_device_count()
54 self.output_root_path = ParamDict().asys_output_timestamp_dir
55 self.config_table = f().read_config()
56 self._chip_info = ""
58 @staticmethod
59 def __get_pcie_info(table_data):
60 pcie_info_query_cmds = {
61 "PCIe Dev Count": f"{LSPCI_GREP_VERSION} | {GET_COUNT}",
62 "PCIe Dev Count(normal)": f"{LSPCI_GREP_VERSION} | {GET_COUNT}",
63 "PCIe Dev Count(abnormal)": f"{LSPCI_GREP_VERSION} | grep 'rev ff' | {GET_COUNT}",
64 }
65 for query_name, query_cmds in pcie_info_query_cmds.items():
66 count = run_command(query_cmds)
67 if not count.isdigit() or (query_name == "PCIe Dev Count" and count == "0"):
68 table_data[PCIE_INFO] = []
69 break
70 table_data[PCIE_INFO].append([query_name, count])
71 if table_data[PCIE_INFO]:
72 table_data[PCIE_INFO][1][1] = int(table_data[PCIE_INFO][0][1]) \
73 - int(table_data[PCIE_INFO][2][1])
74 else:
75 del table_data[PCIE_INFO]
77 @staticmethod
78 def __get_host_info(table_data):
79 host_info_query_cmds = {
80 "Cpu Info": "lscpu | grep -oP '^\\s*(BIOS Model name|Model name):\\s+\\K.*' | grep -v '^-$' | head -n1",
81 "Cpu Physical Count": "lscpu | grep 'Socket(s):' | cut -f2 -d: | uniq", # arm cpuinfo no 'physical id'
82 "Cpu Logical Count": "cat /proc/cpuinfo| grep 'processor' | wc -l",
83 "Memory Total Size": "cat /proc/meminfo | sed -n '1p' | awk 'NR=2{print $2, $3}'",
84 "Disk Total Size": """df -k / |sed -n '2p' | awk 'NR=2{printf $2}END{print " kB"}'""",
85 }
86 for query_name, query_cmds in host_info_query_cmds.items():
87 host_info = run_command(query_cmds)
88 if not host_info:
89 continue
90 table_data[" Host Info "].append([query_name, host_info])
92 def __get_device_info(self, table_data):
93 ccpu_count = self.device_info.get_device_info_loop(
94 self.device_num, self.device_info.get_ccpu_count, NOT_SUPPORT
95 )
96 aicpu_count = self.device_info.get_device_info_loop(
97 self.device_num, self.device_info.get_aicpu_count, NOT_SUPPORT
98 )
99 aicore_count = self.device_info.get_device_info_loop(
100 self.device_num, self.device_info.get_aicore_count, NOT_SUPPORT
101 )
102 vector_count = self.device_info.get_device_info_loop(
103 self.device_num, self.device_info.get_veccore_count, NOT_SUPPORT
104 )
105 device_info_query_cmds = {
106 "NPU Count": self.device_num,
107 "Chip Info": self.device_info.get_device_info_loop(
108 self.device_num, self.device_info.get_chip_info, UNKNOWN
109 ),
110 "Arch Info": self.device_info.get_device_info_loop(
111 self.device_num, self.device_info.get_npu_arch, UNKNOWN
112 ),
113 "Control CPU Count": str(ccpu_count * self.device_num) + f" ({ccpu_count} * {self.device_num})",
114 "AI CPU Count": str(aicpu_count * self.device_num) + f" ({aicpu_count} * {self.device_num})",
115 "AI Core Count": str(aicore_count * self.device_num) + f" ({aicore_count} * {self.device_num})",
116 "AI Vector Count": str(vector_count * self.device_num) + f" ({vector_count} * {self.device_num})",
117 }
118 for query_name, query_cmds in device_info_query_cmds.items():
119 table_data[" Device Info "].append([query_name, query_cmds])
121 def get_hardware_info(self, write_file=False):
122 """
123 return hardware info report
124 """
125 table_data = {
126 " Host Info ": [],
127 " Device Info ": [],
128 PCIE_INFO: []
129 }
130 # Host Info
131 self.__get_host_info(table_data)
132 # Device Info
133 self.__get_device_info(table_data)
134 # PCIe Info
135 self.__get_pcie_info(table_data)
136 table_header = [[f"Group of {self.device_num} Device", "INFORMATION"]]
137 table_string = generate_report(table_header, table_data)
138 if write_file:
139 hardware_file = os.path.join(self.output_root_path, "hardware_info.txt")
140 f.write_file(hardware_file, table_string)
141 else:
142 sys.stdout.write(table_string)
144 @staticmethod
145 def __software_set_env(table_data):
146 env_info = []
147 envs = os.environ
148 for env_name, env_value in envs.items():
149 if env_name == "LS_COLORS" or env_value == "":
150 continue
151 if len(env_value) > MAX_CHAR_LINE:
152 env_info.append([env_name, env_value[:MAX_CHAR_LINE]])
153 for i in range(MAX_CHAR_LINE, len(env_value), MAX_CHAR_LINE):
154 env_info.append(["", env_value[i:i + MAX_CHAR_LINE]])
155 else:
156 env_info.append([env_name, env_value])
157 if env_info:
158 table_data[" Env Information "] = env_info
160 @staticmethod
161 def __software_set_dep(table_data):
162 dependent_packet = []
163 dep_info = f.read_file(os.path.join(get_project_conf(), "dependent_package.csv"))
164 for item in dep_info:
165 info = run_command(item[1])
166 if info == "NONE":
167 continue
168 dependent_packet.append([item[0], info])
169 if dependent_packet:
170 table_data[" Dependent Packet "] = dependent_packet
172 def __check_support_read_option(self, option):
173 if option not in self.config_table or self._chip_info == UNKNOWN:
174 return True
175 supported_chips = self.config_table[option][ConfigOperateType.GET.value]
176 if ALL_NOT_SUPPORTED_CHIP_TYPE in supported_chips:
177 return False
178 if (
179 ALL_SUPPORTED_CHIP_TYPE in supported_chips or
180 any(re.search(rf"{i}", self._chip_info) for i in supported_chips)
181 ):
182 return True
183 return False
185 def __table_data_append(self, table, value, option):
186 if value[1] != NOT_SUPPORT and self.__check_support_read_option(option):
187 table.append(value)
189 @staticmethod
190 def __software_set_pkg(table_data):
191 grep_version = "| grep Version | awk -v FS='=' '{print $2}'"
192 if ParamDict().get_env_type() == "EP":
193 install_path = get_ascend_home()
194 for pag_name in CannPkg.get_all_pkg_list():
195 if pag_name in [CannPkg.firmware, CannPkg.driver]:
196 version = run_command('cat {}/{}/version.info {}'.format(install_path, pag_name, grep_version))
197 else:
198 version = run_command('cat {}/cann/share/info/{}/version.info {}'.format(install_path, pag_name,
199 grep_version))
200 if version == "":
201 version = "None"
202 table_data[DEVICE_VERSION].append([pag_name, version])
203 else:
204 driver_version = run_command(f"cat /var/davinci/driver/version.info {grep_version}")
205 table_data[DEVICE_VERSION].append([CannPkg.driver, driver_version])
206 firmware_version = run_command(f"cat /fw/version.info {grep_version}")
207 table_data[DEVICE_VERSION].append([CannPkg.firmware, firmware_version])
208 runtime_version = run_command(f"cat /usr/local/Ascend/latest/runtime/version.info {grep_version}")
209 if not runtime_version or runtime_version == "NONE":
210 runtime_version = run_command(f"cat /usr/local/Ascend/runtime/version.info {grep_version}")
211 table_data[DEVICE_VERSION].append(["runtime", runtime_version])
213 def get_software_info(self, write_file=False):
214 """
215 return software info report
216 """
217 table_data = {
218 HOST_VERSION: [],
219 DEVICE_VERSION: []
220 }
221 os_version_path = os.sep + "etc/*release"
222 table_data[HOST_VERSION].append(["Kernel", run_command('uname -r')])
223 os_version = run_command("cat " + os_version_path + """ | grep PRETTY_NAME | awk -v FS='"' '{print $2}'""")
224 table_data[HOST_VERSION].append(["OS", os_version])
226 self.__software_set_pkg(table_data)
227 table_header = [[f"Group of {self.device_num} Device", "INFORMATION"]]
228 if write_file:
229 self.__software_set_dep(table_data)
230 self.__software_set_env(table_data)
231 table_string = generate_report(table_header, table_data)
232 software_file = os.path.join(self.output_root_path, "software_info.txt")
233 f.write_file(software_file, table_string)
234 else:
235 table_string = generate_report(table_header, table_data)
236 sys.stdout.write(table_string)
238 def __add_status_cpu_info(self, table_data, device_id):
239 cpu_info = self.device_info.get_device_cpu_info(device_id)
240 if NOT_SUPPORT in cpu_info:
241 ai_cpu_c = self.device_info.get_aicpu_count(device_id)
242 c_cpu_c = self.device_info.get_ccpu_count(device_id)
243 c_cpu_v = NOT_SUPPORT
244 c_cpu_f = self.device_info.get_device_frequency(device_id, CONTROL_CPU_FREQUENCY)
245 else:
246 ai_cpu_c, c_cpu_c, c_cpu_v, c_cpu_f = cpu_info
248 cpu_info_list = []
249 self.__table_data_append(cpu_info_list, ["AI CPU Count", ai_cpu_c], ConfigOptionName.ACPU_CNT.value)
250 self.__table_data_append(
251 cpu_info_list,
252 ["AI CPU Usage (%)", self.device_info.get_device_utilization_rate(device_id, AI_CPU_USE)],
253 ConfigOptionName.ACPU_USAGE.value,
254 )
255 self.__table_data_append(cpu_info_list, ["Control CPU Count", c_cpu_c], ConfigOptionName.CCPU_CNT.value)
256 self.__table_data_append(
257 cpu_info_list,
258 ["Control CPU Usage (%)", self.device_info.get_device_utilization_rate(device_id, CONTROL_CPU_USE)],
259 ConfigOptionName.CCPU_USAGE.value,
260 )
261 self.__table_data_append(
262 cpu_info_list, ["Control CPU Frequency (MHZ)", c_cpu_f], ConfigOptionName.CCPU_FREQUENCY.value
263 )
264 self.__table_data_append(
265 cpu_info_list, ["Control CPU Voltage (MV)", c_cpu_v], ConfigOptionName.CCPU_VOLTAGE.value
266 )
267 if cpu_info_list:
268 table_data[CPU_TABLE_TITLE] = cpu_info_list
269 else:
270 table_data.pop(CPU_TABLE_TITLE)
272 def __add_status_aic_info(self, table_data, device_id):
273 accuracy_device = get_device(device_id)
274 aic_info = accuracy_device.get_device_aic_info(device_id)
275 if any(i in aic_info for i in accuracy_device.UNSUPPORTED_KEY_WORDS):
276 aic_c = self.device_info.get_aicore_count(device_id)
277 aic_v = accuracy_device.get_device_voltage(device_id)
278 aic_f = accuracy_device.get_device_aicore_frequency(device_id)
279 else:
280 aic_c, aic_v, aic_f = aic_info
281 aic_info_list = []
282 self.__table_data_append(aic_info_list, ["AI Core Count", aic_c], ConfigOptionName.AIC_CNT.value)
283 self.__table_data_append(
284 aic_info_list,
285 ["AI Core Usage (%)", self.device_info.get_device_utilization_rate(device_id, AI_CORE_USE)],
286 ConfigOptionName.AIC_USAGE.value,
287 )
288 self.__table_data_append(
289 aic_info_list, ["AI Core Frequency (MHZ)", aic_f], ConfigOptionName.AIC_FREQUENCY.value
290 )
291 self.__table_data_append(aic_info_list, ["AI Core Voltage (MV)", aic_v], ConfigOptionName.AIC_VOLTAGE.value)
292 if aic_info_list:
293 table_data[AIC_TABLE_TITLE] = aic_info_list
294 else:
295 table_data.pop(AIC_TABLE_TITLE)
297 def __add_status_bus_info(self, table_data, device_id):
298 accuracy_device = get_device(device_id)
299 bus_v, ring_f, cpu_f, mate_f, l2_buf_f = accuracy_device.get_device_bus_info(device_id)
300 bus_info_list = []
301 self.__table_data_append(bus_info_list, ["Bus Voltage (MV)", bus_v], ConfigOptionName.BUS_VOLTAGE.value)
302 self.__table_data_append(bus_info_list, ["Ring Frequency (MHZ)", ring_f], ConfigOptionName.RING_FREQUENCY.value)
303 self.__table_data_append(bus_info_list, ["CPU Frequency (MHZ)", cpu_f], ConfigOptionName.CPU_FREQUENCY.value)
304 self.__table_data_append(bus_info_list, ["Mata Frequency (MHZ)", mate_f], ConfigOptionName.MATA_FREQUENCY.value)
305 self.__table_data_append(
306 bus_info_list, ["L2buffer Frequency (MHZ)", l2_buf_f], ConfigOptionName.L2BUFFER_FREQUENCY.value
307 )
308 if bus_info_list:
309 table_data[BUS_TABLE_TITLE] = bus_info_list
310 else:
311 table_data.pop(BUS_TABLE_TITLE)
313 def __add_status_memory_info(self, table_data, device_id):
314 ddr_total, ddr_use = self.device_info.get_device_memory_info(device_id)
315 hbm_total, hbm_use, _, hbm_bandwidth = get_device(device_id).get_device_hbm_info(device_id)
317 memory_info_list = []
318 if ddr_total != NOT_SUPPORT:
319 ddr_bandwidth = self.device_info.get_device_utilization_rate(device_id, MEM_BANDWIDTH_USE)
320 ddr_frequency = self.device_info.get_device_frequency(device_id, MEMORY_FREQUENCY)
321 self.__table_data_append(memory_info_list, ["DDR Total (MB)", ddr_total], ConfigOptionName.DDR_TOTAL.value)
322 self.__table_data_append(memory_info_list, ["DDR Used (MB)", ddr_use], ConfigOptionName.DDR_USED.value)
323 self.__table_data_append(
324 memory_info_list, ["DDR Bandwidth Usage (%)", ddr_bandwidth], ConfigOptionName.DDR_BANDWIDTH.value
325 )
326 self.__table_data_append(
327 memory_info_list, ["DDR Frequency (MHZ)", ddr_frequency], ConfigOptionName.DDR_FREQUENCY.value
328 )
330 if hbm_total != NOT_SUPPORT:
331 hbm_v, hbm_f = self.device_info.get_device_hbm_volt_freq(device_id)
332 if hbm_f == NOT_SUPPORT:
333 hbm_f = self.device_info.get_device_frequency(device_id, HBM_FREQUENCY)
334 self.__table_data_append(memory_info_list, ["HBM Total (MB)", hbm_total], ConfigOptionName.HBM_TOTAL.value)
335 self.__table_data_append(memory_info_list, ["HBM Used (MB)", hbm_use], ConfigOptionName.HBM_USED.value)
336 self.__table_data_append(
337 memory_info_list, ["HBM Bandwidth Usage (%)", hbm_bandwidth], ConfigOptionName.HBM_BANDWIDTH_USE.value
338 )
339 self.__table_data_append(
340 memory_info_list, ["HBM Frequency (MHZ)", hbm_f], ConfigOptionName.HBM_FREQUENCY.value
341 )
342 self.__table_data_append(memory_info_list, ["HBM Voltage (MV)", hbm_v], ConfigOptionName.HBM_VOLTAGE.value)
344 if memory_info_list:
345 table_data[MEM_TABLE_TITLE] = memory_info_list
346 else:
347 table_data.pop(MEM_TABLE_TITLE)
349 def get_status_info(self, device_id, write_file=False):
350 """
351 return status info report
352 """
353 table_data = {
354 NONE: [],
355 CPU_TABLE_TITLE: [],
356 AIC_TABLE_TITLE: [],
357 BUS_TABLE_TITLE: [],
358 MEM_TABLE_TITLE: []
359 }
360 # add public information
361 self._chip_info = self.device_info.get_chip_info(device_id)
362 self.__table_data_append(table_data[NONE], ["Chip Name", self._chip_info], ConfigOptionName.CHIP_NAME.value)
363 self.__table_data_append(
364 table_data[NONE], ["Power (W)", self.device_info.get_device_power(device_id)], ConfigOptionName.POWER.value
365 )
366 self.__table_data_append(
367 table_data[NONE],
368 ["Temperature (C)", get_device(device_id).get_device_temperature(device_id)],
369 ConfigOptionName.TEMPERATURE.value
370 )
371 self.__table_data_append(
372 table_data[NONE],
373 ["health", self.device_info.get_device_health(device_id)],
374 ConfigOptionName.HEALTH.value
375 )
377 # add cpu information
378 self.__add_status_cpu_info(table_data=table_data, device_id=device_id)
380 # add aicore information
381 self.__add_status_aic_info(table_data=table_data, device_id=device_id)
383 # add bus information
384 self.__add_status_bus_info(table_data=table_data, device_id=device_id)
386 # add memory information
387 self.__add_status_memory_info(table_data=table_data, device_id=device_id)
389 # Delete the key whose value is empty.
390 for key, value in table_data.items():
391 if not value:
392 table_data.pop(key)
394 table_header = [[f"Device ID: {device_id}", "INFORMATION"]]
395 table_string = generate_report(table_header, table_data)
396 if write_file:
397 status_file = os.path.join(self.output_root_path, "status_info.txt")
398 f.append_write_file(status_file, table_string)
399 else:
400 sys.stdout.write(table_string)
402 @timeout_decorator(GET_DEVICES_INFO_TIMEOUT)
403 def run_info(self, run_mode, device_id):
404 if run_mode == "hardware":
405 self.get_hardware_info()
406 elif run_mode == "software":
407 self.get_software_info()
408 elif run_mode == "status":
409 self.get_status_info(device_id)
411 def run(self):
412 run_mode = ParamDict().get_arg('run_mode')
413 device_id = ParamDict().get_arg('device_id') if ParamDict().get_arg('device_id') else 0
414 try:
415 self.run_info(run_mode, device_id)
416 except TimeoutError:
417 log_error(f"Timeout in retrieving the {device_id} chip status, Please check for malfunctions.")
418 return False
419 return True
421 def write_info(self):
422 self.get_hardware_info(write_file=True)
423 self.get_software_info(write_file=True)
424 for device_id in range(self.device_num):
425 self.get_status_info(device_id, write_file=True)