""" Custom log tool. This module contains some interesting tools for get a pretty and easy terminal log """ import datetime from multiprocessing import Pipe from collections import deque import os import time CONST_ERROR_KEY = 'ERROR' CONST_WARNING_KEY = 'WARNING' CONST_SUCCESS_KEY = 'SUCCESS' CONST_INFO_KEY = 'INFO' class EntryLog: """ Class defining the entries of the log Parameters ---------- date: datetime.datetime Timestamp of message msg_type: str Message type message: str Message Attributes ---------- date: datetime.datetime Timestamp of message msg_type: str Message type message: str Message str: str Formatted string entry """ MESSAGE_TYPE_LENGTH = 8 COLOR = "" def __init__(self, date, msg_type, message): self.datetime = date self.type = msg_type self.message = message self.str = self.build() def build(self): """ Format the entry Returns ------- """ date = self.datetime message_type = self.type unformatted_time_pre_log = '[{}-{}-{} | {}:{}:{}{}] '.format(date.year, date.month, date.day, date.hour, date.minute, '0' * (2 - len(str(date.second))), date.second) unformatted_type_pre_log = '{}{}' \ .format(message_type, ' ' * (self.MESSAGE_TYPE_LENGTH - len(message_type))) unformatted_pre_log = unformatted_time_pre_log + unformatted_type_pre_log space_string = "\n" + ' ' * (len(unformatted_pre_log)) message_log = space_string.join(self.message.split('\n')) log_entry = "\u001b[33;1m{}\u001b[1m{}{}\u001b[0m{}".format(unformatted_time_pre_log, self.COLOR, unformatted_type_pre_log, message_log) return log_entry def __str__(self): return self.str class EntryLogError(EntryLog): """ Subclassing EntryLog with other color """ COLOR = '\u001b[31m' def __init__(self, date, message): super().__init__(date, 'ERROR', message) class EntryLogInfo(EntryLog): """ Subclassing EntryLog with other color """ COLOR = '\u001b[36m' def __init__(self, date, message): super().__init__(date, 'INFO', message) class EntryLogSuccess(EntryLog): """ Subclassing EntryLog with other color """ COLOR = '\u001b[32;1m' def __init__(self, date, message): super().__init__(date, 'SUCCESS', message) class EntryLogWarning(EntryLog): """ Subclassing EntryLog with other color """ COLOR = '\u001b[31;1m' def __init__(self, date, message): super().__init__(date, 'WARNING', message) class Log: """ Main class to manage and work with the log. Attributes ---------- output_log: bool Show log? config_log: str Where to save exported log (deprecated) warning_silent: bool Silent warnings messages? max_entries: int Max entries in log entries: deque of EntryLog """ ERROR_KEY = CONST_ERROR_KEY WARNING_KEY = CONST_INFO_KEY SUCCESS_KEY = CONST_SUCCESS_KEY INFO_KEY = CONST_WARNING_KEY ENTRY_TYPES = { CONST_ERROR_KEY: EntryLogError, CONST_INFO_KEY: EntryLogInfo, CONST_SUCCESS_KEY: EntryLogSuccess, CONST_WARNING_KEY: EntryLogWarning } MESSAGE_TYPE_LENGTH = 8 def __init__(self, output_log=True, warning_silent=True, max_entries=1000000): """ Parameters ---------- output_log: bool Show log? warning_silent: bool Silent warnings messages? max_entries: int Max entries in log """ self.output_log = output_log self.config_log = "" self.warning_silent = warning_silent self.max_entries = max_entries self.entries = deque([], maxlen=max_entries) def add_entry(self, text, message_type='INFO', overwrite=False, silent=False): """ Adding a message to the log Parameters ---------- text: str Message to be added message_type: str Type of message overwrite: bool Overwrite last message? silent: bool Not verbose? Returns ------- success: EntryLog Created entry """ if not isinstance(text, str): text = str(text) entry = self.ENTRY_TYPES[message_type](datetime.datetime.now(), text) if len(self.entries) > self.max_entries: self.entries.pop() self.entries.appendleft(entry) if self.output_log: if silent: return entry if message_type == 'WARNING' and self.warning_silent: return entry if overwrite: print('\r{}'.format(entry), end='') else: print(entry) return entry def formatted_input(self, text): """ Get an formatted input terminal Parameters ---------- text: str Text to be shown when input prompted Returns ------- success: str Text input by user """ self.add_entry(text) resp = input() return resp def status_bar(self, percentage, status_width=25, verbose=False): """ Method to show a status bar with the percentage Parameters ---------- percentage: float Percentage to show status_width: int With of the bar verbose: bool Verbose? Returns ------- success: None """ percentage_real = percentage percentage = status_width*percentage_real/100 total_first = status_width/2 total_second = status_width/2 first_percentage = min(percentage, total_first) second_percentage = max(percentage, total_second) self.add_entry('{} {:.4}/{} {}' .format('/'*int(first_percentage)+'.'*int(total_first-first_percentage), percentage_real, 100, '/' * int(second_percentage-total_second) + '.'*int(status_width-second_percentage)), overwrite=True, silent=not verbose) if percentage_real >= 100: print('') def time_it(self, func, *args, **kwargs): """ Measure execution time of a function `func` Parameters ---------- func: function Function to be executed args: list Arguments of the function kwargs: dict Kwargs dict of the function Returns ------- success: object Return the returned value from function execution. """ start = time.time() res = func(*args, **kwargs) self.add_entry('{}: {} s'.format(func.__name__, time.time()-start)) return res def multithread(self, threads): """ Create a multi thread log. Useful to manage log inside different threads. Parameters ---------- threads: int Num of threads Returns ------- pipe_list: list of Pipe List containing the pipe needed to communicate with the log inside the thread log_copy_list: list List containing the logs. """ pipe_list = [] log_copy_list = [] for _ in range(threads): recv, send = Pipe(False) pipe_list.append(recv) log_m = MultiLog(send, self.output_log, self.warning_silent) log_copy_list.append(log_m) return pipe_list, log_copy_list def catch_multi(self, pipe_list): """ Print the multiple thread log Parameters ---------- pipe_list: list of Pipe List containing the pipe needed to communicate with the log inside the thread Returns ------- success: EntryLog Entry with all threads log fused. """ text = "" for index_pipe, pipe in enumerate(pipe_list): if pipe.poll(1): entry = pipe.recv() else: continue text += "Thread {}: {} {}|".format(index_pipe, entry.type, entry.message) self.add_entry(text, overwrite=True) def squared(self, *texts, silent=True): """ Method tu get a squared print. A normal print rounded by #. Parameters ---------- texts: list of str List of strings. Each element of the list will be a new line silent: bool Silent? Returns ------- success: EntryLog Created entry """ groups = [] lines = [] for text in texts: groups.append(text.split('\n')) lines.extend(text.split('\n')) max_length = len(max(lines, key=len)) del lines squared_string = "#" * (max_length+2+2) + "\n" group_string = "" for group in groups[:-1]: for line in group: group_string += "# {}{} #\n".format(line, ' '*(max_length-len(line))) group_string += "#{}#\n".format('-'*(max_length+2)) for line in groups[-1]: group_string += "# {}{} #\n".format(line, ' ' * (max_length - len(line))) squared_string += group_string squared_string += "#" * (max_length+2+2) + "\n" self.add_entry(squared_string, silent=silent) return squared_string def get_by_type(self, msg_type, verbose=True): """ Get entries of the log by message type Parameters ---------- msg_type: str Message type verbose: bool Verbose? Returns ------- selected_entries: list of EntryLog Selected entries. """ selected_entries = [entry for entry in self.entries if entry.type == msg_type] if verbose: for entry in selected_entries: print(entry) return selected_entries def __call__(self, text, message_type='INFO', overwrite=False, silent=False): self.add_entry(text, message_type=message_type, overwrite=overwrite, silent=silent) class MultiLog(Log): """ Subclass of Log. Multilog has the utility to run in multiple thread monitorizing each thread and then showing this information in the main thread. Attributes ---------- pipe: Pipe Pipe to connect with. """ def __init__(self, pipe, output_log=True, warning_silent=True): super().__init__(output_log=output_log, warning_silent=warning_silent) self.pipe = pipe def add_entry(self, text, message_type='INFO', overwrite=False, silent=False): """ Adding a message to the log Parameters ---------- text: str Message to be added message_type: str Type of message overwrite: bool Overwrite last message? silent: bool Not verbose? Returns ------- success: EntryLog Created entry """ entry = super().add_entry(text, message_type=message_type, overwrite=overwrite, silent=True) self.pipe.send(entry) def status_bar(self, percentage, status_width=10, verbose=False): """ Method to show a status bar with the percentage Parameters ---------- percentage: float Percentage to show status_width: int With of the bar verbose: bool Verbose? Returns ------- success: None """ super().status_bar(percentage, status_width, verbose=verbose) def close(self): """ Close multilog Returns ------- """ os.close(self.pipe.fileno())