Coverage for pymarsseason/monitoring.py: 100%

1# skeleton-python-binary - Command-line tool that generates project templates based on predefined Python project template 

2# Copyright (C) 2024-2025 - Centre National d'Etudes Spatiales 

3# SPDX-License-Identifier: Apache-2.0 

4# Auto-generated by skeleton-python-binary 

5"""Some Utilities.""" 

6import time 

7import tracemalloc 

8from functools import partial 

9from functools import wraps 

10 

11from loguru import logger 

12 

13 

14class UtilsMonitoring: # noqa: R0205 

15 """Some Utilities.""" 

16 

17 # pylint: disable:invalid_name 

18 @staticmethod 

19 def io_display( 

20 func=None, input=True, output=True, level="TRACE" 

21 ): # pylint: disable=W0622 

22 """ 

23 Decorator to log the inputs and/or outputs of a function, with a configurable log level. 

24 

25 Args: 

26 func (Callable, optional): Function to decorate. Defaults to `None`, which allows the decorator to be used with parameters. 

27 input (bool, optional): Indicates whether to log the inputs of the function. Defaults to `True`. 

28 output (bool, optional): Indicates whether to log the outputs of the function. Defaults to `True`. 

29 level (str, optional): The log level used for logging messages. Defaults to "TRACE". Can be any of the log levels supported by `loguru` ("TRACE", "DEBUG", "INFO", "SUCCESS", "WARNING", "ERROR", "CRITICAL"). 

30 

31 Returns: 

32 Callable: The decorated function that logs its inputs and/or outputs. 

33 

34 Usage: 

35 This decorator allows you to log the inputs and/or outputs of a function with a configurable log level. 

36 It can be used for debugging or monitoring function execution by tracking the values of arguments and results. 

37 

38 Example usage: 

39 

40 @UtilsMonitoring.io_display(input=True, output=False, level="DEBUG") 

41 def add(a, b): 

42 return a + b 

43 

44 add(3, 4) 

45 

46 In this example, only the inputs will be logged at the `DEBUG` level, and the output will not be logged. 

47 

48 Notes: 

49 - If `func` is `None`, the decorator is configured with the given parameters. 

50 - The `input` and `output` parameters allow fine-grained control over whether the function's arguments or results are logged. 

51 - The `level` parameter must be a valid log level in `loguru` and is converted to uppercase before use. 

52 

53 """ 

54 

55 def decorator(func): 

56 @wraps(func) 

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

58 # Obtenir le logger avec le niveau configuré 

59 log_level = level.upper() 

60 

61 # Logger les entrées si configuré 

62 if input: 

63 logger.log( 

64 log_level, 

65 f"Input in the function '{func.__name__}' with args={args}, kwargs={kwargs}", 

66 ) 

67 

68 # Execute the function 

69 result = func(*args, **kwargs) 

70 

71 # Log the outputs if configured 

72 if output: 

73 logger.log( 

74 log_level, 

75 f"Output in the function '{func.__name__}' with result={result}", 

76 ) 

77 

78 return result 

79 

80 return wrapper 

81 

82 return decorator 

83 

84 @staticmethod 

85 def time_spend(func=None, level="DEBUG", threshold_in_ms=1000): 

86 """Monitor the performances of a function. 

87 

88 Parameters 

89 ---------- 

90 func: func 

91 Function to monitor (default: {None}) 

92 level: int 

93 Level from which the monitoring starts (default: {logging.DEBUG}) 

94 threshold_in_ms: int 

95 an alert is sent at any level when the function duration > 

96 threshold_in_ms (default: {1000}) 

97 

98 Returns 

99 ------- 

100 object : the result of the function 

101 """ 

102 if func is None: 

103 return partial(UtilsMonitoring.time_spend, level=level) 

104 

105 @wraps(func) 

106 def newfunc(*args, **kwargs): 

107 start_time = time.time() 

108 result = func(*args, **kwargs) 

109 elapsed_time = time.time() - start_time 

110 logger.log( 

111 level, 

112 "function [{}] finished in {:.2f} ms".format( 

113 func.__qualname__, elapsed_time * 1000 

114 ), 

115 ) 

116 return result 

117 

118 return newfunc 

119 

120 @staticmethod 

121 def measure_memory(func=None, level="DEBUG"): 

122 """Measure the memory of the function 

123 

124 Args: 

125 func (func, optional): Function to measure. Defaults to None. 

126 level (int, optional): Level of the log. Defaults to DEBUG. 

127 

128 Returns: 

129 object : the result of the function 

130 """ 

131 if func is None: 

132 return partial(UtilsMonitoring.measure_memory, level=level) 

133 

134 @wraps(func) 

135 def newfunc(*args, **kwargs): 

136 tracemalloc.start() 

137 result = func(*args, **kwargs) 

138 current, peak = tracemalloc.get_traced_memory() 

139 msg = f""" 

140 \033[37mFunction Name :\033[35;1m {func.__name__}\033[0m 

141 \033[37mCurrent memory usage:\033[36m {current / 10 ** 6}MB\033[0m 

142 \033[37mPeak :\033[36m {peak / 10 ** 6}MB\033[0m 

143 """ 

144 logger.log(level, msg) 

145 tracemalloc.stop() 

146 return result 

147 

148 return newfunc