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

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 concurrent.futures 

20import multiprocessing 

21import os 

22from datetime import datetime, timezone 

23from functools import wraps 

24import sys 

25 

26from params import ParamDict 

27from common.const import RetCode, CANN_LOG_NAME, consts, STACKTRACE 

28from common.log import log_error, log_info 

29from common.file_operate import FileOperate as f 

30from drv import EnvVarName 

31 

32__all__ = [ 

33 "create_out_timestamp_dir", "get_asys_output_path", "get_target_cnt", "out_progress_bar", "is_hexadecimal", 

34 "str_to_hex", "int_to_hex", "get_cann_log_path", "timeout_decorator" 

35] 

36 

37 

38_asys_output_path = None 

39 

40 

41def get_asys_output_path(): 

42 return _asys_output_path 

43 

44 

45def create_out_timestamp_dir(): 

46 def init_output_dir_parent(): 

47 output_arg = ParamDict().get_arg("output") 

48 return EnvVarName().current_path if not output_arg else output_arg 

49 

50 if ParamDict().get_command() not in [consts.collect_cmd, consts.launch_cmd, consts.analyze_cmd]: 

51 return RetCode.SUCCESS 

52 

53 output_dir = init_output_dir_parent() 

54 if not os.access(output_dir, os.W_OK): 

55 log_error("No write permission to asys output root directory: {}.".format(output_dir)) 

56 return RetCode.PERMISSION_FAILED 

57 

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

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

60 

61 global _asys_output_path 

62 _asys_output_path = os.path.abspath(os.path.join(output_dir, dir_name)) 

63 if not f.create_dir(_asys_output_path): 

64 return RetCode.ARG_CREATE_DIR_FAILED 

65 ParamDict().asys_output_timestamp_dir = _asys_output_path 

66 if not (ParamDict().get_command() == consts.collect_cmd and ParamDict().get_arg("run_mode") == STACKTRACE): 

67 log_info("asys output directory: {0}".format(_asys_output_path)) 

68 return RetCode.SUCCESS 

69 

70 

71def is_hexadecimal(value): 

72 try: 

73 int(value, 16) 

74 return True 

75 except ValueError: 

76 return False 

77 

78 

79def str_to_hex(str_number): 

80 return int(str_number, 16) 

81 

82 

83def int_to_hex(value): 

84 return hex(int(value, 16)) 

85 

86 

87def get_target_cnt(dir_path): 

88 """ 

89 Counts the number of bin files in the trace folder. 

90 """ 

91 count = 0 

92 atrace_dirs = f.walk_dir(dir_path) 

93 if not atrace_dirs: 

94 return count 

95 for *_, files in atrace_dirs: 

96 for _ in files: 

97 count += 1 

98 return count 

99 

100 

101def out_progress_bar(count, num): 

102 """ 

103 Show progress bar 

104 """ 

105 if count == 0: 

106 return 

107 sys.stdout.write("\r") 

108 sys.stdout.write("Parse progress: {:.2f}%: ".format(num/count * 100)) 

109 sys.stdout.write("\r") 

110 sys.stdout.flush() 

111 

112 

113def get_cann_log_path(log_type): 

114 """ 

115 get trace or cann log path from env 

116 """ 

117 env_var = EnvVarName() 

118 if log_type == CANN_LOG_NAME: 

119 if env_var.process_log_path: 

120 return env_var.process_log_path, "${ASCEND_PROCESS_LOG_PATH}" 

121 

122 if env_var.work_path: 

123 return os.path.join(env_var.work_path, log_type), f"${{ASCEND_WORK_PATH}}/{log_type}" 

124 return os.path.join(env_var.home_path, "ascend", log_type), f"${{HOME}}/ascend/{log_type}" 

125 

126 

127def timeout_decorator(timeout): 

128 def decorator(func): 

129 @wraps(func) 

130 def wrapper(*args, **kwargs): 

131 # Use fork context explicitly: forkserver (Python 3.14+ default on Linux) 

132 # requires pickling bound methods, which fails when the method is decorated. 

133 ctx = multiprocessing.get_context('fork') 

134 p = ctx.Process(target=func, args=args, kwargs=kwargs) 

135 p.daemon = True 

136 p.start() 

137 p.join(timeout) 

138 if p.is_alive(): 

139 p.terminate() 

140 p.join() 

141 raise TimeoutError() 

142 return wrapper 

143 return decorator 

144