/usr/lib/python2.7/dist-packages/cpl/logger.py is in python-cpl 0.7.2-1build2.
This file is owned by root:root, with mode 0o644.
The actual contents of the file can be viewed below.
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 | from __future__ import absolute_import
import datetime
import logging
import os
import re
import tempfile
import threading
class NullHandler(logging.Handler):
def emit(self, record):
pass
logging.getLogger('cpl').addHandler(NullHandler())
level = { "DEBUG":logging.DEBUG, "INFO":logging.INFO, "WARNING":logging.WARN,
"ERROR":logging.ERROR, "OFF":(logging.CRITICAL + 1)}
cpl_verbosity = [ logging.DEBUG, logging.INFO, logging.WARN,
logging.ERROR, logging.CRITICAL + 1 ]
class LogServer(threading.Thread):
def __init__(self, name, level = None):
threading.Thread.__init__(self)
self.name = name
self.logger = logging.getLogger(name)
self.level = cpl_verbosity.index(level) if level is not None else 0
self.entries = LogList()
self.regexp = re.compile('(\\d\\d):(\\d\\d):(\\d\\d)' +
'\\s\\[\\s*(\\w+)\\s*\\]' +
'\\s(\\w+):' +
'(\\s\\[tid=(\\d+)\\])?' +
'\\s(.+)')
tmphdl, self.logfile = tempfile.mkstemp(prefix = 'cpl', suffix='.log')
os.close(tmphdl)
os.remove(self.logfile)
os.mkfifo(self.logfile)
self.start()
def run(self):
try:
with open(self.logfile, 'rb', buffering = 0) as logfile:
line = logfile.readline()
os.remove(self.logfile)
while line:
self.log(str(line.decode('ascii')))
line = logfile.readline()
except:
pass
def log(self, s):
'''Convert CPL log messages into python log records.
A typical CPL log message looks like
10:35:25 [WARNING] rtest: [tid=000] No file tagged with FLAT
'''
try:
m = self.regexp.match(s)
if m is not None:
g = m.groups()
creation_date = datetime.datetime.combine(
datetime.date.today(),
datetime.time(int(g[0]),int(g[1]),int(g[2])))
lvl = level.get(g[3], logging.NOTSET)
func = g[4]
log = logging.getLogger('%s.%s' % (self.logger.name, func))
threadid = int(g[6]) if g[6] else None
msg = g[-1]
record = logging.LogRecord(log.name, lvl, None, None,
msg, None, None, func)
created = float(creation_date.strftime('%s'))
if record.created < created:
created -= 86400
record.relativeCreated -= record.msecs
record.relativeCreated += 1000*(created - record.created + 1)
record.created = created
record.msecs = 0.0
record.threadid = threadid
record.threadName = ('Cpl-%03i' % threadid) if threadid \
else 'CplThread'
elif self.entries:
r0 = self.entries[-1]
msg = s.rstrip()
lvl = r0.levelno
log = logging.getLogger(r0.name)
record = logging.LogRecord(r0.name, lvl, None, None,
msg, None, None, r0.funcName)
record.relativeCreated = r0.relativeCreated
record.created = r0.created
record.msecs = r0.msecs
record.threadid = r0.threadid
record.threadName = r0.threadName
else:
return
self.entries.append(record)
if log.isEnabledFor(lvl) and log.filter(record):
log.handle(record)
except:
pass
class LogList(list):
'''List of log messages.
Accessing this :class:`list` directly will return the
:class:`logging.LogRecord` instances.
Example::
res = muse_bias(bias_frames)
for logrecord in res.log:
print '%s: %s' % (entry.funcname, entry.msg)
To get them formatted as string, use the :attr:`error`, :attr:`warning`,
:attr:`info` or :attr:`debug` attributes::
res = muse_bias(bias_frames)
for line in res.log.info:
print line
'''
def filter(self, level):
return [ '%s: %s' % (entry.funcName, entry.msg) for entry in self
if entry.levelno >= level ]
@property
def error(self):
'''Error messages as list of :class:`str`
'''
return self.filter(logging.ERROR)
@property
def warning(self):
'''Warnings and error messages as list of :class:`str`
'''
return self.filter(logging.WARN)
@property
def info(self):
'''Info, warning and error messages as list of :class:`str`
'''
return self.filter(logging.INFO)
@property
def debug(self):
'''Debug, info, warning, and error messages as list of :class:`str`
'''
return self.filter(logging.DEBUG)
|