Coverage for /opt/cloud/slavespace/usr1/096471637100f3de0fcfc01072822a80/ut/utils/show_kernel_debug_data/show_kernel_debug_data/dump_logger.py: 93%

1#!/usr/bin/python 

2# -*- coding: utf-8 -*- 

3# ---------------------------------------------------------------------------------------------------------- 

4# Copyright (c) 2025 Huawei Technologies Co., Ltd. 

5# This program is free software, you can redistribute it and/or modify it under the terms and conditions of 

6# CANN Open Software License Agreement Version 2.0 (the "License"). 

7# Please refer to the License for details. You may not use this file except in compliance with the License. 

8# THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND, EITHER EXPRESS OR IMPLIED, 

9# INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT, MERCHANTABILITY, OR FITNESS FOR A PARTICULAR PURPOSE. 

10# See LICENSE in the root of the software repository for the full text of the License. 

11# ---------------------------------------------------------------------------------------------------------- 

12 

13import logging 

14import os 

15 

16LOG_LEVEL_DICT = { 

17 '0': logging.DEBUG, 

18 '1': logging.INFO, 

19 '2': logging.WARNING, 

20 '3': logging.ERROR 

21} 

22 

23OUT_LOG_FORMAT = logging.Formatter('%(asctime)s %(message)s') 

24FILE_LOG_FORMAT = logging.Formatter('[%(levelname)s] show_kernel_debug_data [%(process)d] %(asctime)s %(message)s') 

25 

26DEFAULT_LOG_FILE = 'dump_parser.log' 

27 

28 

29class DumpParserLog: 

30 def __init__(self) -> None: 

31 self.log_level = logging.INFO 

32 self.logger = logging.getLogger("show_kernel_debug_data_log") 

33 self.logger.setLevel(self.log_level) 

34 

35 # 设置打屏日志， WARNING以上, 不能改变 

36 self.out_handler = logging.StreamHandler() 

37 self.out_handler.setFormatter(OUT_LOG_FORMAT) 

38 self.out_handler.setLevel(logging.WARNING) 

39 self.logger.addHandler(self.out_handler) 

40 

41 # 文件日志, 通过 _set_log_file 接口设置 

42 self.log_file = '' 

43 self.file_handler = None 

44 self._set_log_file(os.path.join(os.getcwd(), DEFAULT_LOG_FILE)) 

45 

46 def get_log_file(self) -> str: 

47 return self.log_file 

48 

49 def _set_log_file(self, log_file): 

50 if self.file_handler: 

51 self.logger.removeHandler(self.file_handler) 

52 

53 log_path = os.path.dirname(log_file) 

54 if log_path and not os.path.isdir(log_path): 

55 os.makedirs(log_path) 

56 

57 if os.path.exists(log_file): 

58 os.remove(log_file) 

59 self.log_file = log_file 

60 self.file_handler = logging.FileHandler(self.log_file, encoding='utf-8', delay=True) 

61 self.file_handler.setFormatter(FILE_LOG_FORMAT) 

62 self.file_handler.setLevel(self.log_level) 

63 self.logger.addHandler(self.file_handler) 

64 

65 def set_log_level(self, log_level: str = '3') -> None: 

66 self.log_level = LOG_LEVEL_DICT.get(log_level, logging.ERROR) 

67 self.logger.setLevel(self.log_level) 

68 if self.file_handler: 

69 self.file_handler.setLevel(self.log_level) 

70 

71 def set_log_file(self, log_file=''): 

72 file_name = os.path.basename(log_file) 

73 if not file_name: 

74 file_name = DEFAULT_LOG_FILE 

75 

76 log_path = os.path.dirname(log_file) 

77 if not log_path: 

78 log_path = os.getcwd() 

79 

80 log_file = os.path.join(log_path, file_name) 

81 print("log file saves to ", log_file) 

82 self._set_log_file(log_file) 

83 

84 

85 def info(self, msg: object, *args: object, **kwargs: object) -> None: 

86 """ 

87 Auto tune pregress info 

88 :param msg: 

89 :param args: 

90 :param kwargs: 

91 :return: 

92 """ 

93 self.logger.info(f"[INFO]: {msg}", *args, **kwargs) 

94 

95 def debug(self, msg: object, *args: object, **kwargs: object) -> None: 

96 """ 

97 debug msg 

98 :param msg: 

99 :param args: 

100 :param kwargs: 

101 :return: 

102 """ 

103 self.logger.debug(f"[DEBUG]: {msg}", *args, **kwargs) 

104 

105 def warning(self, msg: object, *args: object, **kwargs: object) -> None: 

106 """ 

107 warning msg 

108 :param msg: 

109 :param args: 

110 :param kwargs: 

111 :return: 

112 """ 

113 self.logger.warning(f"[WARNING]: {msg}", *args, **kwargs) 

114 

115 def error(self, msg: object, *args: object, **kwargs: object) -> None: 

116 """ 

117 error msg 

118 :param msg: 

119 :param args: 

120 :param kwargs: 

121 :return: 

122 """ 

123 self.logger.error(f"[ERROR]: {msg}", *args, **kwargs) 

124 

125 

126DUMP_PARSER_LOG = DumpParserLog()