2 # Copyright (C) 2007-2013 EDF R&D
4 # This library is free software; you can redistribute it and/or
5 # modify it under the terms of the GNU Lesser General Public
6 # License as published by the Free Software Foundation; either
7 # version 2.1 of the License.
9 # This library is distributed in the hope that it will be useful,
10 # but WITHOUT ANY WARRANTY; without even the implied warranty of
11 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
12 # Lesser General Public License for more details.
14 # You should have received a copy of the GNU Lesser General Public
15 # License along with this library; if not, write to the Free Software
16 # Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
18 # See http://www.salome-platform.org/ or email : webmaster.salome@opencascade.com
21 """Module to manage information printing : debug, info, error.
22 Should replace 'print' and 'UTMESS' calls at least in the supervisor
24 Only used for debug right now.
31 from functools import partial
32 from subprocess import Popen, PIPE
34 from N_utils import Enum, Singleton
35 from strfunc import convert
38 def default_print(text):
39 """Basic print function."""
50 class Category(object):
52 """Define a category of message for different parts of the code.
53 This allows to store different parameters for each category of message."""
56 self._level = LEVEL.INFO
61 LEVEL.WARN: "WARNING",
65 def set_level(self, level):
66 """Set the current level."""
70 """Return the current level."""
73 def set_header(self, level, header):
74 """Set the header of ``level`` messages."""
75 self._header[level] = header
77 def get_header(self, level):
78 """Return the header at this ``level``."""
79 header = self._header.get(level, "")
81 header = self._fmt % header
84 def active(self, level):
85 """Tell if a message should be print at this ``level``."""
86 return self._level <= level
91 SUPERV.set_header(LEVEL.ERROR, None)
94 REGEXP_ORIG = re.compile('File [\'\"]*(.*?)[\'\"]*, *line ([0-9]+), *in (.*)')
96 # slighty different and very simplier than logger objects
97 # from the logging module.
100 class InfoLevel(Singleton):
102 """Store informations level."""
103 _singleton_id = 'N_info.InfoLevel'
105 def __init__(self, level):
108 for part in self._parts:
110 self.reset_print_function()
111 self._msg_callback = []
112 # self.extend_message(ALL, stack_header_callback)
113 self.extend_message(ALL, insert_header)
115 def add(self, category):
116 """Add a category of message."""
117 self._parts.append(category)
119 def set_level(self, category, level):
120 """Set the current level for ``category``."""
121 assert category in self._parts, "unknown category : %s" % category
122 assert LEVEL.exists(level), "unknown level : %s" % level
123 category.set_level(level)
125 for part in self._parts:
126 part.set_level(level)
129 """Set debug level for all categories."""
130 self.set_level(ALL, LEVEL.DEBUG)
132 def set_header(self, category, level, header):
133 """Set the header of ``level`` messages."""
134 category.set_header(level, header)
136 def register_print_function(self, print_function):
137 """Define the `print_function` to use."""
138 self._print = print_function
140 def reset_print_function(self):
141 """Register the default 'print function'."""
142 self._print = default_print
144 def extend_message(self, category, callback):
145 """Allow to extend the message calling an external function."""
146 self._msg_callback.append((category, callback))
148 def _message(self, category, level, msg, args, kwargs):
149 """Print the message if the level is reached."""
150 if category.active(level):
151 if kwargs.get('utmess'):
152 func = self._message_utmess
154 func = self._message_print
155 func = self._message_print
156 apply(func, (category, level, msg, args, kwargs))
158 def _message_print(self, category, level, msg, args, kwargs):
159 """Print the message if the level is reached."""
160 for cat, cbk in self._msg_callback:
161 if cat in (ALL, category):
162 msg, args = cbk(category, level, msg, args, kwargs)
166 except Exception, err:
167 msg = repr((msg, args, err))
170 def _message_utmess(self, category, level, msg, args, kwargs):
171 """Print the message if the level is reached."""
172 # how to use callbacks ? valk ?
173 from Utilitai.Utmess import MessageLog
180 valk = kwargs.get('valk', ())
181 vali = kwargs.get('vali', ())
182 valr = kwargs.get('valr', ())
183 msg = MessageLog.GetText(code[level], msg, valk, vali, valr)
184 for cat, cbk in self._msg_callback:
185 if cat in (ALL, category):
186 msg, args = cbk(category, level, msg, args, kwargs)
189 def debug(self, category, msg, *args, **kwargs):
190 """Print a debug message."""
191 self._message(category or ALL, LEVEL.DEBUG, msg, args, kwargs)
193 def info(self, category, msg, *args, **kwargs):
194 """Print an information message."""
195 self._message(category or ALL, LEVEL.INFO, msg, args, kwargs)
197 def warn(self, category, msg, *args, **kwargs):
198 """Print a warning message."""
199 self._message(category or ALL, LEVEL.WARN, msg, args, kwargs)
201 def error(self, category, msg, *args, **kwargs):
202 """Print an error message."""
203 self._message(category or ALL, LEVEL.ERROR, msg, args, kwargs)
207 def add_memory_info(self, category):
208 """Shortcut to add memory informations."""
209 self.extend_message(category, mem_msg_callback)
211 def use_aster_print(self):
212 """Shortcut to use aster.affiche function to print the messages."""
214 self.register_print_function(partial(aster.affiche, 'MESSAGE'))
218 def insert_header(category, level, msg, args, kwargs):
219 """Insert the header."""
220 header = category.get_header(level)
226 def stack_header_callback(category, level, msg, args, kwargs):
227 """To insert the origin."""
228 if level <= LEVEL.DEBUG:
229 stack_id = -5 + kwargs.get('stack_id', 0)
230 stack = traceback.format_stack(limit=10)[stack_id]
231 mat = REGEXP_ORIG.search(stack)
232 origin = '[%s:%s in %s] ' % (
233 osp.basename(mat.group(1)), mat.group(2), mat.group(3))
239 message = InfoLevel(LEVEL.INFO)
244 # callback to add memory information
247 RE_VMPEAK = re.compile('VmPeak:\s*([0-9]+)\s*([kMGBo]+)', re.M | re.I)
250 def memory_used(pid):
251 """Return the current VmPeak value."""
252 p = Popen(['cat', '/proc/%s/status' % pid], stdout=PIPE)
253 output = p.communicate()[0]
254 mat = RE_VMPEAK.search(output)
255 mem = mat and int(mat.group(1)) or 0.
258 current_memory_used = partial(memory_used, _pid)
261 def mem_msg_callback(category, level, msg, args, kwargs):
262 """Callback to add memory infos to message."""
263 if level <= LEVEL.DEBUG:
264 msg = msg + " - VmPeak : %.2f Mo"
265 args = tuple(list(args) + [current_memory_used(), ])
269 if __name__ == "__main__":
270 message.set_level(SUPERV, LEVEL.WARN)
271 message.set_level(MISS, LEVEL.DEBUG)
272 # message.debug(None, "debug message")
273 message.info(ALL, "information message")
274 message.warn(None, "warning message")
275 message.error(ALL, "error message")
276 message.add_memory_info()
277 # message.debug(MISS, "debug supervisor message")
278 message.info(SUPERV, "information supervisor message")
279 message.warn(SUPERV, "warning supervisor message")
280 message.error(SUPERV, "error supervisor message")
281 message.critical(MISS, "test the critical alias")