Coverage for /opt/cloud/slavespace/usr1/096471637100f3de0fcfc01072822a80/ut/src/asys/profiling/asys_profiling.py: 22%

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# ---------------------------------------------------------------------------- 

18 

19import os 

20import subprocess 

21from datetime import datetime, timezone 

22 

23from common import log_error, log_info, log_debug 

24from common.const import MAX_PERIOD, UNKNOWN, LP_MODE_NO, LP_MODE_AIC, LP_MODE_LP 

25from params import ParamDict 

26from common import AsysProfilingSupportedChip, ChipHandler 

27 

28support_profiling_list = set(["aicore", "dvpp", "os", "link", "memory", "power"]) 

29 

30 

31class AsysProfiling(): 

32 def __init__(self): 

33 self.device_id = int(ParamDict().get_arg("device_id")) 

34 self.period = ParamDict().get_arg("period", 0) 

35 self.output_path = ParamDict().get_arg("output") 

36 self.run_modes = ParamDict().get_arg("run_mode") 

37 self.aic_metrics = ParamDict().get_arg("aic_metrics", "PipeUtilization") 

38 self.lp_mode = LP_MODE_NO 

39 

40 @staticmethod 

41 def concat_dvpp(cmd): 

42 concat_cmd = "--dvpp-profiling=on" 

43 log_debug("Concat dvpp command.") 

44 out_cmd = cmd + " " + concat_cmd 

45 return out_cmd 

46 

47 @staticmethod 

48 def concat_link(cmd): 

49 concat_cmd = "--sys-interconnection-profiling=on --sys-io-profiling=on" 

50 log_debug("Concat link command.") 

51 out_cmd = cmd + " " + concat_cmd 

52 return out_cmd 

53 

54 @staticmethod 

55 def concat_memory(cmd): 

56 concat_cmd = "--sys-hardware-mem=on --llc-profiling=read" 

57 log_debug("Concat memory command.") 

58 out_cmd = cmd + " " + concat_cmd 

59 return out_cmd 

60 

61 @staticmethod 

62 def concat_os(cmd): 

63 concat_cmd = "--sys-profiling=on" 

64 log_debug("Concat os command.") 

65 out_cmd = cmd + " " + concat_cmd 

66 return out_cmd 

67 

68 def concat_power(self, cmd): 

69 out_cmd = cmd 

70 if self.lp_mode == LP_MODE_AIC: 

71 concat_cmd = "--ai-core=on" 

72 out_cmd += " " + concat_cmd 

73 elif self.lp_mode == LP_MODE_LP: 

74 concat_cmd = "--sys-lp=on" 

75 out_cmd += " " + concat_cmd 

76 return out_cmd 

77 

78 def concat_aicore(self, cmd): 

79 concat_cmd = f"--ai-core=on --aic-mode=sample-based --aic-metrics={self.aic_metrics}" 

80 log_debug("Concat aicore command.") 

81 out_cmd = cmd + " " + concat_cmd 

82 return out_cmd 

83 

84 def run(self): 

85 if not self._check_param(): 

86 return False 

87 log_info(f"Run mode is {self.run_modes}.") 

88 cmd = (f"msprof --output={self.output_path} --sys-period={str(self.period)} " 

89 f"--sys-devices={self.device_id} ") 

90 for run_mode in self.run_modes: 

91 func_name = "concat_" + run_mode 

92 func = getattr(self, func_name, None) 

93 if func: 

94 cmd = func(cmd) 

95 ret = self._run_cmd(cmd) 

96 return ret 

97 

98 def _check_param(self): 

99 if self.period < 1 or self.period > MAX_PERIOD: 

100 log_error("Period is invalid, it should be in [1,2592000].") 

101 return False 

102 

103 profiling_supported = AsysProfilingSupportedChip() 

104 ret, _chip_info = profiling_supported.get_supported_chip_info(self.device_id) 

105 if not ret: 

106 if _chip_info == UNKNOWN: 

107 log_error(f"device id {self.device_id} is invalid, please enter a vaild value.") 

108 return False 

109 else: 

110 log_error(f"The profiling command does not support {_chip_info}.") 

111 return False 

112 

113 self.run_modes = set(self.run_modes.split(',')) 

114 if not self.run_modes.issubset(support_profiling_list): 

115 log_error("Run mode type is unsupported, run 'asys profiling -h' to get help.") 

116 return False 

117 

118 handler = ChipHandler().get_handler(_chip_info) 

119 if handler is None: 

120 log_error(f"{_chip_info} is not supported.") 

121 return False 

122 if handler.need_lp_param(): 

123 self.lp_mode = LP_MODE_LP 

124 elif 'aicore' not in self.run_modes: 

125 self.lp_mode = LP_MODE_AIC 

126 

127 utc_dt = datetime.now(timezone.utc) # UTC time 

128 dir_name = utc_dt.astimezone().strftime('%Y%m%d%H%M%S%f')[:-3] 

129 if not self.output_path: 

130 self.output_path = f"asys_profiling_result_{dir_name}" 

131 else: 

132 self.output_path = os.path.join(self.output_path, f"asys_profiling_result_{dir_name}") 

133 return True 

134 

135 def _run_cmd(self, cmd): 

136 log_info(f"Start run: {cmd}, please wait about {self.period} seconds.") 

137 # Run msprof command 

138 ret = subprocess.run(cmd, shell=True, stdout=subprocess.PIPE, encoding='utf-8', env=os.environ) 

139 if ret.returncode == 0: 

140 log_info(f"Succeeded in running profiling.\n{ret.stdout}") 

141 else: 

142 log_error(f"Failed to run profiling.\n{ret.stdout}") 

143 return False 

144 return True