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

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 shutil 

21import signal 

22import sys 

23 

24from common import consts, log_info, log_error, close_log 

25from common import RetCode 

26from common.task_common import create_out_timestamp_dir 

27from common import compress_output_dir_tar 

28from params import ParamDict 

29from cmdline import CommandLineParser 

30from config import AsysConfigParser 

31from collect import AsysCollect 

32from launch import AsysLaunch 

33from info import AsysInfo 

34from diagnose import AsysDiagnose 

35from health import AsysHealth 

36from analyze import AsysAnalyze 

37from config_cmd import AsysConfig 

38from profiling import AsysProfiling 

39 

40__all__ = ['main'] 

41 

42 

43def _check_args_duplicate(): 

44 input_args = [arg.split('=')[0] for arg in sys.argv[1:] if '-' in arg.split('=')[0]] 

45 # remove args duplicate 

46 args_no_duplicate = set(input_args) 

47 if len(input_args) > len(args_no_duplicate): 

48 log_error(f'Only one of the {list(args_no_duplicate)} args can be specified.') 

49 return False 

50 return True 

51 

52 

53def clean_pycache(): 

54 """clean __pycache__""" 

55 current_file = os.path.dirname(os.path.abspath(__file__)) 

56 for root, dirs, _ in os.walk(current_file): 

57 if '__pycache__' in dirs and os.path.exists(os.path.join(root, '__pycache__')): 

58 shutil.rmtree(os.path.join(root, '__pycache__'), ignore_errors=True) 

59 

60 

61EXECUTE_CMD_FUNC = { 

62 consts.collect_cmd: AsysCollect, 

63 consts.launch_cmd: AsysLaunch, 

64 consts.info_cmd: AsysInfo, 

65 consts.diagnose_cmd: AsysDiagnose, 

66 consts.health_cmd: AsysHealth, 

67 consts.analyze_cmd: AsysAnalyze, 

68 consts.config_cmd: AsysConfig, 

69 consts.profiling_cmd: AsysProfiling 

70} 

71 

72 

73def main(): 

74 """entrance of Ascend system advisor""" 

75 # check args duplicate 

76 if not _check_args_duplicate(): 

77 return False 

78 

79 # error stack when ctrl c is ignored 

80 signal.signal(signal.SIGINT, signal.SIG_DFL) 

81 

82 # 1. parse the command line and check args 

83 asys_parser = CommandLineParser() 

84 try: 

85 parse_ret = asys_parser.parse() 

86 except SystemExit: 

87 return False 

88 param_dict = ParamDict() 

89 command = param_dict.get_command() 

90 if command is None: 

91 if parse_ret == RetCode.SUCCESS: # -h, --help, and only asys 

92 asys_parser.print_help() 

93 return True 

94 else: 

95 log_error('Arguments parse failed, asys exit.') 

96 return False 

97 

98 # info/diagnose/health/config(get)/analyze(aicore_error) close info & warning level log 

99 if any([ 

100 command in [consts.info_cmd, consts.diagnose_cmd, consts.health_cmd], 

101 command == consts.config_cmd and param_dict.get_arg('get'), 

102 command == consts.analyze_cmd and param_dict.get_arg('run_mode') == 'aicore_error' 

103 ]): 

104 close_log() 

105 

106 # 2. check the environment type 

107 env_ret = param_dict.get_env_type() 

108 if not env_ret: 

109 log_error('Failed to obtain the execution environment type.') 

110 return False 

111 if env_ret == 'RC': 

112 if any([ 

113 command not in [consts.collect_cmd, consts.launch_cmd], 

114 command == consts.collect_cmd and param_dict.get_arg("run_mode") 

115 ]): 

116 log_error('The RC supports the launch command and the collect command without the -r parameter.') 

117 return False 

118 

119 # 2. read the config file and load configs 

120 conf_parser = AsysConfigParser() 

121 conf_res = conf_parser.parse() 

122 if not conf_res: 

123 log_error('Configs parse failed, asys exit.') 

124 return False 

125 

126 log_info('asys start.') 

127 

128 if create_out_timestamp_dir() != RetCode.SUCCESS: 

129 log_error('Create asys output directory failed.') 

130 return False 

131 

132 # 3. execute the command 

133 obj = EXECUTE_CMD_FUNC.get(command) 

134 task_res = obj().run() if obj else False 

135 

136 log_info(f'{command} task execute finish.') 

137 

138 # 4. Compress the output dir using tar. 

139 if param_dict.get_arg('tar') in ['T', 'TRUE']: 

140 compress_output_dir_tar() 

141 

142 log_info('asys finish.') 

143 return task_res 

144 

145 

146if __name__ == '__main__': 

147 main() 

148 # clean __pycache__ file 

149 clean_pycache()