Coverage for  / home / jenkins / .local / lib / python3.10 / site-packages / hyper_parallel / auto_parallel / sapp_nd / nd / logger.py: 78%

1# Copyright 2025 Huawei Technologies Co., Ltd 

2# 

3# Licensed under the Apache License, Version 2.0 (the "License"); 

4# you may not use this file except in compliance with the License. 

5# You may obtain a copy of the License at 

6# 

7# http://www.apache.org/licenses/LICENSE-2.0 

8# 

9# Unless required by applicable law or agreed to in writing, software 

10# distributed under the License is distributed on an "AS IS" BASIS, 

11# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 

12# See the License for the specific language governing permissions and 

13# limitations under the License. 

14# ============================================================================ 

15"""logger for parall""" 

16import logging 

17from typing import cast 

18 

19DEFAULT_STDOUT_FORMAT = ( 

20 "[%(levelname)s] %(asctime)s [%(filename)s:%(lineno)d] - %(message)s" 

21) 

22PPB_FORMAT = "%(asctime)s - %(name)s - %(levelname)s - %(message)s" 

23PARALL_FORMAT = ( 

24 "%(name)s [%(filename)s:%(lineno)d] - %(levelname)s - %(message)s" 

25) 

26 

27FORMATTER = logging.Formatter(PARALL_FORMAT) 

28OUTPUT_LEVEL_NUM = logging.CRITICAL 

29logging.addLevelName(OUTPUT_LEVEL_NUM, "OUTPUT") 

30 

31 

32class MyLogger(logging.Logger): 

33 """Inherit from standard Logger and add level OUTPUT.""" 

34 

35 def output(self, msg, *args, **kwargs): 

36 """Log 'msg % args' with severity 'OUTPUT'. 

37 

38 Args: 

39 msg (str): The message to log. 

40 *args: Additional arguments. 

41 **kwargs: Additional keyword arguments. 

42 """ 

43 if self.isEnabledFor(OUTPUT_LEVEL_NUM): 

44 self._log(OUTPUT_LEVEL_NUM, msg, args, **kwargs) 

45 

46 

47def setup_logger(name: str, level: int = logging.DEBUG): 

48 """Set up a logger. 

49 

50 Args: 

51 name (str): Logger name. 

52 level (int, optional): Logging level. Default: logging.DEBUG. 

53 

54 Returns: 

55 MyLogger: Configured logger. 

56 """ 

57 ch = logging.StreamHandler() 

58 ch.setLevel(level) 

59 ch.setFormatter(FORMATTER) 

60 

61 previous_logger_cls = logging.getLoggerClass() 

62 logging.setLoggerClass(MyLogger) 

63 try: 

64 nd_logger = cast(MyLogger, logging.getLogger(name)) 

65 finally: 

66 logging.setLoggerClass(previous_logger_cls) 

67 nd_logger.setLevel(level) 

68 nd_logger.addHandler(ch) 

69 

70 return nd_logger 

71 

72 

73logger = setup_logger("ND") 

74perf_logger = setup_logger("Perf. estim.") 

75 

76 

77def set_verbose_level(level): 

78 """Assign level to each logger from a global level""" 

79 memo_logger = logging.getLogger("memory_estimation") 

80 memo_logger.disabled = True 

81 perf_logger.disabled = True 

82 logger.disabled = True 

83 if level >= 1: 

84 logger.disabled = False 

85 logger.setLevel(logging.CRITICAL) 

86 if level >= 2: 

87 logger.setLevel(logging.ERROR) 

88 if level >= 3: 

89 logger.setLevel(logging.INFO) 

90 if level >= 4: 

91 memo_logger.disabled = False 

92 perf_logger.disabled = False 

93 memo_logger.setLevel(logging.CRITICAL) 

94 perf_logger.setLevel(logging.CRITICAL) 

95 if level >= 5: 

96 memo_logger.setLevel(logging.INFO) 

97 perf_logger.setLevel(logging.INFO) 

98 logger.setLevel(logging.DEBUG) 

99 if level >= 6: 

100 memo_logger.setLevel(logging.DEBUG) 

101 perf_logger.setLevel(logging.DEBUG)