Coverage for lisacattools/monitoring.py: 55%

1# -*- coding: utf-8 -*-

3# jean-christophe.malapert@cnes.fr

5# This file is part of smt_crawler_lib.

7# smt_crawler_lib is a free software; you can redistribute it and/or

8# modify it under the terms of the GNU Lesser General Public

9# License as published by the Free Software Foundation; either

10# version 3.0 of the License, or (at your option) any later version.

11#

12# This library is distributed in the hope that it will be useful,

13# but WITHOUT ANY WARRANTY; without even the implied warranty of

14# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU

15# Lesser General Public License for more details.

16#

17# You should have received a copy of the GNU Lesser General Public

18# License along with this library; if not, write to the Free Software

19# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston,

20# MA 02110-1301 USA

21"""Some Utilities."""

22import logging

23import os

24import time

25import tracemalloc

26from functools import partial

27from functools import wraps

30class UtilsMonitoring(object):

31 """Some Utilities."""

33 # pylint: disable:invalid_name

34 @staticmethod

35 def io(func=None, entry=True, exit=True, level=15): # level=15

36 """Monitor the input/output of a function.

38 NB : Do not use this monitoring method on an __init__ if the class

39 implements __repr__ with attributes

41 Parameters

42 ----------

43 func: func

44 function to monitor (default: {None})

45 entry: bool

46 True when the function must monitor the input (default: {True})

47 exit: bool

48 True when the function must monitor the output (default: {True})

49 level: int

50 Level from which the function must log

51 Returns

52 -------

53 object : the result of the function

54 """

55 if func is None:

56 return partial(

57 UtilsMonitoring.io, entry=entry, exit=exit, level=level

58 )

60 @wraps(func)

61 def wrapped(*args, **kwargs):

62 name = func.__qualname__

63 logger = logging.getLogger(__name__ + "." + name)

65 if entry and logger.getEffectiveLevel() >= level:

66 msg = f"Entering '{name}' (args={args}, kwargs={kwargs})"

67 logger.log(level, msg)

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

71 if exit and logger.getEffectiveLevel() >= level:

72 msg = f"Exiting '{name}' (result={result})"

73 logger.log(level, msg)

75 return result

77 return wrapped

79 @staticmethod

80 def time_spend(func=None, level=logging.DEBUG, threshold_in_ms=1000):

81 """Monitor the performances of a function.

83 Parameters

84 ----------

85 func: func

86 Function to monitor (default: {None})

87 level: int

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

89 threshold_in_ms: int

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

91 threshold_in_ms (default: {1000})

93 Returns

94 -------

95 object : the result of the function

96 """

97 if func is None:

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

100 @wraps(func)

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

102 name = func.__class__.__name__

103 logger = logging.getLogger(__name__ + "." + name)

104 start_time = time.time()

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

106 elapsed_time = time.time() - start_time

107 logger.log(

108 level,

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

110 func.__qualname__, elapsed_time * 1000

111 ),

112 )

113 if float(elapsed_time) * 1000 > threshold_in_ms:

114 logger.warning(

115 "function [{}] is too long to compute : {:.2f} ms".format(

116 func.__qualname__, elapsed_time * 1000

117 )

118 )

119 return result

120

121 return newfunc

122

123 @staticmethod

124 def size(func=None, level=logging.INFO):

125 """Monitor the number of records in a file.

126

127 Parameters

128 ----------

129 func: func

130 Function to monitor (default: {None})

131 level: int

132 Level from which the monitoring starts (default: {logging.INFO})

133

134 Returns

135 -------

136 object : the result of the function

137 """

138 if func is None:

139 return partial(UtilsMonitoring.size, level=level)

140

141 @wraps(func)

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

143 name = func.__name__

144 logger = logging.getLogger(__name__ + "." + name)

145 filename = os.path.basename(args[1])

146 logger.log(level, "Loading file '%s'", filename)

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

148 type_result = type(result)

149 if type_result in [type({}), type([])]:

150 nb_records = len(result)

151 else:

152 try:

153 nb_records = result.shape

154 except Exception:

155 nb_records = None

156

157 if nb_records is not None:

158 logger.info(

159 "File '%s' loaded with %s records",

160 filename,

161 str(nb_records),

162 )

163 else:

164 logger.warning(

165 "Unable to load the number of records in file '%s' - "

166 "type: %s",

167 args[1],

168 type_result,

169 )

170 return result

171

172 return newfunc

173

174 @staticmethod

175 def measure_memory(func=None, level=logging.DEBUG):

176 """Measure the memory of the function

177

178 Args:

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

180 level (int, optional): Level of the log. Defaults to logging.INFO.

181

182 Returns:

183 object : the result of the function

184 """

185 if func is None:

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

187

188 @wraps(func)

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

190 name = func.__class__.__name__

191 logger = logging.getLogger(__name__ + "." + name)

192 tracemalloc.start()

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

194 current, peak = tracemalloc.get_traced_memory()

195 msg = f"""

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

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

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

199 """

200 logger.log(level, msg)

201 tracemalloc.stop()

202 return result

203

204 return newfunc