| | """Logger class for IPython's logging facilities. |
| | """ |
| |
|
| | |
| | |
| | |
| | |
| | |
| | |
| | |
| |
|
| | |
| | |
| |
|
| | |
| | import glob |
| | import io |
| | import os |
| | import time |
| |
|
| |
|
| | |
| | |
| | |
| |
|
| | class Logger(object): |
| | """A Logfile class with different policies for file creation""" |
| |
|
| | def __init__(self, home_dir, logfname='Logger.log', loghead=u'', |
| | logmode='over'): |
| |
|
| | |
| | |
| | self.home_dir = home_dir |
| |
|
| | self.logfname = logfname |
| | self.loghead = loghead |
| | self.logmode = logmode |
| | self.logfile = None |
| |
|
| | |
| | self.log_raw_input = False |
| |
|
| | |
| | self.log_output = False |
| |
|
| | |
| | self.timestamp = False |
| |
|
| | |
| | self.log_active = False |
| |
|
| | |
| | def _set_mode(self,mode): |
| | if mode not in ['append','backup','global','over','rotate']: |
| | raise ValueError('invalid log mode %s given' % mode) |
| | self._logmode = mode |
| |
|
| | def _get_mode(self): |
| | return self._logmode |
| |
|
| | logmode = property(_get_mode,_set_mode) |
| |
|
| | def logstart(self, logfname=None, loghead=None, logmode=None, |
| | log_output=False, timestamp=False, log_raw_input=False): |
| | """Generate a new log-file with a default header. |
| | |
| | Raises RuntimeError if the log has already been started""" |
| |
|
| | if self.logfile is not None: |
| | raise RuntimeError('Log file is already active: %s' % |
| | self.logfname) |
| |
|
| | |
| | if logfname is not None: self.logfname = logfname |
| | if loghead is not None: self.loghead = loghead |
| | if logmode is not None: self.logmode = logmode |
| |
|
| | |
| | self.timestamp = timestamp |
| | self.log_output = log_output |
| | self.log_raw_input = log_raw_input |
| |
|
| | |
| | isfile = os.path.isfile |
| | logmode = self.logmode |
| |
|
| | if logmode == 'append': |
| | self.logfile = io.open(self.logfname, 'a', encoding='utf-8') |
| |
|
| | elif logmode == 'backup': |
| | if isfile(self.logfname): |
| | backup_logname = self.logfname+'~' |
| | |
| | |
| | if isfile(backup_logname): |
| | os.remove(backup_logname) |
| | os.rename(self.logfname,backup_logname) |
| | self.logfile = io.open(self.logfname, 'w', encoding='utf-8') |
| |
|
| | elif logmode == 'global': |
| | self.logfname = os.path.join(self.home_dir,self.logfname) |
| | self.logfile = io.open(self.logfname, 'a', encoding='utf-8') |
| |
|
| | elif logmode == 'over': |
| | if isfile(self.logfname): |
| | os.remove(self.logfname) |
| | self.logfile = io.open(self.logfname,'w', encoding='utf-8') |
| |
|
| | elif logmode == 'rotate': |
| | if isfile(self.logfname): |
| | if isfile(self.logfname+'.001~'): |
| | old = glob.glob(self.logfname+'.*~') |
| | old.sort() |
| | old.reverse() |
| | for f in old: |
| | root, ext = os.path.splitext(f) |
| | num = int(ext[1:-1])+1 |
| | os.rename(f, root+'.'+repr(num).zfill(3)+'~') |
| | os.rename(self.logfname, self.logfname+'.001~') |
| | self.logfile = io.open(self.logfname, 'w', encoding='utf-8') |
| |
|
| | if logmode != 'append': |
| | self.logfile.write(self.loghead) |
| |
|
| | self.logfile.flush() |
| | self.log_active = True |
| |
|
| | def switch_log(self,val): |
| | """Switch logging on/off. val should be ONLY a boolean.""" |
| |
|
| | if val not in [False,True,0,1]: |
| | raise ValueError('Call switch_log ONLY with a boolean argument, ' |
| | 'not with: %s' % val) |
| |
|
| | label = {0:'OFF',1:'ON',False:'OFF',True:'ON'} |
| |
|
| | if self.logfile is None: |
| | print(""" |
| | Logging hasn't been started yet (use logstart for that). |
| | |
| | %logon/%logoff are for temporarily starting and stopping logging for a logfile |
| | which already exists. But you must first start the logging process with |
| | %logstart (optionally giving a logfile name).""") |
| |
|
| | else: |
| | if self.log_active == val: |
| | print('Logging is already',label[val]) |
| | else: |
| | print('Switching logging',label[val]) |
| | self.log_active = not self.log_active |
| | self.log_active_out = self.log_active |
| |
|
| | def logstate(self): |
| | """Print a status message about the logger.""" |
| | if self.logfile is None: |
| | print('Logging has not been activated.') |
| | else: |
| | state = self.log_active and 'active' or 'temporarily suspended' |
| | print('Filename :', self.logfname) |
| | print('Mode :', self.logmode) |
| | print('Output logging :', self.log_output) |
| | print('Raw input log :', self.log_raw_input) |
| | print('Timestamping :', self.timestamp) |
| | print('State :', state) |
| |
|
| | def log(self, line_mod, line_ori): |
| | """Write the sources to a log. |
| | |
| | Inputs: |
| | |
| | - line_mod: possibly modified input, such as the transformations made |
| | by input prefilters or input handlers of various kinds. This should |
| | always be valid Python. |
| | |
| | - line_ori: unmodified input line from the user. This is not |
| | necessarily valid Python. |
| | """ |
| |
|
| | |
| | |
| | if self.log_raw_input: |
| | self.log_write(line_ori) |
| | else: |
| | self.log_write(line_mod) |
| |
|
| | def log_write(self, data, kind='input'): |
| | """Write data to the log file, if active""" |
| |
|
| | |
| | if self.log_active and data: |
| | write = self.logfile.write |
| | if kind=='input': |
| | if self.timestamp: |
| | write(time.strftime('# %a, %d %b %Y %H:%M:%S\n', time.localtime())) |
| | write(data) |
| | elif kind=='output' and self.log_output: |
| | odata = u'\n'.join([u'#[Out]# %s' % s |
| | for s in data.splitlines()]) |
| | write(u'%s\n' % odata) |
| | try: |
| | self.logfile.flush() |
| | except OSError: |
| | print("Failed to flush the log file.") |
| | print( |
| | f"Please check that {self.logfname} exists and have the right permissions." |
| | ) |
| | print( |
| | "Also consider turning off the log with `%logstop` to avoid this warning." |
| | ) |
| |
|
| | def logstop(self): |
| | """Fully stop logging and close log file. |
| | |
| | In order to start logging again, a new logstart() call needs to be |
| | made, possibly (though not necessarily) with a new filename, mode and |
| | other options.""" |
| |
|
| | if self.logfile is not None: |
| | self.logfile.close() |
| | self.logfile = None |
| | else: |
| | print("Logging hadn't been started.") |
| | self.log_active = False |
| |
|
| | |
| | close_log = logstop |
| |
|