import os import math import tempfile from io import StringIO import sys # --- update the path of Python in case to test this file alone py_path = os.sys.path py_pwd = os.path.normpath(os.getcwd() + "/../../../..") if (py_pwd not in py_path): (os.sys.path).append(py_pwd) # --- import PyROS celestial mechanics from the pyros root directory import vendor.guitastro as guitastro ''' logger = None def log_d(msg:str, *args, **kwargs): logger.debug(msg, *args, **kwargs) def log_i(msg:str, *args, **kwargs): logger.info(msg, *args, **kwargs) def log_w(msg:str, *args, **kwargs): logger.warning(msg, *args, **kwargs) def log_e(msg:str, *args, **kwargs): logger.error(msg, *args, **kwargs) def log_c(msg:str, *args, **kwargs): logger.critical(msg, *args, **kwargs) ''' class LogPyros: """ Manage PyROS logs in display, file and database. First, create an instance: log = LogPyros("test",None) Second, use the print methods to log: log.print("Something to log") """ # CLASS attributes _CHRONO_FILENAME = "pyros" _CHRONO_FILENAME_WARN = "pyros_warn" _GLOBAL_PATH_DATA = '' # === Level of log ''' LOG_LEVEL_INFO = 0 LOG_LEVEL_WARNING = -1 LOG_LEVEL_ERROR = -2 LOG_LEVEL_DEBUG = 1 LOG_LEVEL_DEBUG2 = 2 ''' # Python standard logging module values: NOTSET=0, DEBUG=10, INFO=20, WARN=30, ERROR=40, CRITICAL=50 LOG_LEVEL_NOTSET = 0 LOG_LEVEL_DEBUG = 10 LOG_LEVEL_DEBUG2 = 15 LOG_LEVEL_INFO = 20 LOG_LEVEL_WARNING = 30 LOG_LEVEL_ERROR = 40 LOG_LEVEL_CRITICAL = 50 # Default LOG level is NOTSET (no filter) #_GLOBAL_LOG_LEVEL = LOG_LEVEL_NOTSET _GLOBAL_LOG_LEVEL = LOG_LEVEL_INFO # Start Log_level (threshold) for logging in file _FILE_LOG_LEVEL = LOG_LEVEL_DEBUG # === Constants NO_ERROR = 0 ERR_ALIAS_NAME_NOT_DEFINED = 10 ERR_FILE_NOT_EXISTS = 101 ERR_PATH_NOT_EXISTS = 102 ERR_EMPTY_FILENAME = 103 ERR_EMPTY_PATHNAME = 104 # === Private variables _last_errno = NO_ERROR ##_path_data = '' _caller_alias = '' _path_data_log = '' _path_wwwpyros = '' _date = None _home = None _noon_hour = 12 _last_lines = {'chrono':[], 'agent':[]} #_last_lines = [] #_last_lines_agent = [] _nbmax_last_lines = 30 # ===================================================================== # ===================================================================== # Private methods # ===================================================================== # ===================================================================== ''' def _mkdir_path_log(self): # --- if self._caller_alias == "": self._last_errno = self.ERR_EMPTY_FILENAME else: self._path_www_log = self._path_wwwpyros + "/logs/" + self._caller_alias os.mkdir(self._path_www_log) # --- if self._path_data == "": self._last_errno = self.ERR_EMPTY_PATHNAME else: self._path_data_log = self._path_data + "/logs/" + self._caller_alias os.mkdir(self._path_data_log) ''' def _jd2datedigit(self, jd): """ Convert a julian day into a string of YYYYMMDD """ datetmp = self._date.date(jd); datedigit = (datetmp.digits(0))[0:8] return datedigit def _date2night(self, date): # night is the truncated part of the date of the previous noon self._date.date(date) jd = self._date.jd() jd0 = math.floor(jd) - self._noon_hour/24 d = guitastro.Date(jd0) djd = jd-jd0 # print(f"jd0 = {d.iso()} djd={djd}") if djd>=1: jd0 += 1 djd = jd-jd0 if djd>=1: jd0 += 1 d = guitastro.Date(jd0) night = d.digits(0)[0:8] return night def _night2dates(self, night): """ jd_noon0 = previous noon jd_midnight = midnight instant jd_noon1 = next noon """ d = guitastro.Date(night) d_noon0 = d + self._noon_hour/24 jd_noon0 = d_noon0.jd() jd_midnight = jd_noon0 + 0.5 jd_noon1 = jd_noon0 + 1.0 return jd_noon0, jd_midnight, jd_noon1 def _convert_args2str(self, *args, **kwargs): """ return a string as the result of a print method """ bak = sys.stdout # on sauvegarde l'ancien stdout result = StringIO() sys.stdout = result print (*args, **kwargs,end='') sys.stdout = bak # on restore stdout return result.getvalue() # ===================================================================== # ===================================================================== # Private methods getter/setter # ===================================================================== # ===================================================================== def set_global_log_level(self, log_level:int): LogPyros._GLOBAL_LOG_LEVEL = log_level def get_global_log_level(self): return LogPyros._GLOBAL_LOG_LEVEL ''' def _set_debug_level(self, level:str): if type(level).__name__=="bool": if level==True: level = 1 else: level = 0 self._debug_level = level def _get_debug_level(self): return self._debug_level ''' def _set_caller_alias(self, caller_alias:str): if caller_alias=="": self._last_errno = self.ERR_ALIAS_NAME_NOT_DEFINED raise Exception(f"Agent alias must not be empty string.") self._caller_alias = caller_alias return self.NO_ERROR def _get_caller_alias(self): return self._caller_alias def _set_path_data(self, path_data:str): if not os.path.exists(path_data): self._last_errno = self.ERR_PATH_NOT_EXISTS raise Exception(f"Path '{path_data}' for data does not exists. Create it first manually.") self._path_data = path_data return self.NO_ERROR def _get_path_data(self): return self._path_data def set_global_path_data(self, path_data:str): if not os.path.exists(path_data): self._last_errno = self.ERR_PATH_NOT_EXISTS raise Exception(f"Path '{path_data}' for data does not exists. Create it first manually.") LogPyros._GLOBAL_PATH_DATA = path_data return self.NO_ERROR def get_global_path_data(self): return LogPyros._GLOBAL_PATH_DATA def get_full_GLOBAL_PATH_DATA(self): return LogPyros._GLOBAL_PATH_DATA + os.sep + "logs" def _set_path_wwwpyros(self, path_wwwpyros:str): if not os.path.exists(path_wwwpyros): self._path_wwwpyros = self.ERR_PATH_NOT_EXISTS raise Exception(f"Path '{path_wwwpyros}' for wwwpyros does not exists. Create it first manually.") self._path_wwwpyros = path_wwwpyros return self.NO_ERROR def _get_path_wwwpyros(self): return self._path_wwwpyros def _set_home(self, home:str): self._home.home(home) longitude = self._home.longitude self._noon_hour = (180 - longitude)/15.0 # - place noon_hour in the range [0 24[ if self._noon_hour < 0: self._noon_hour += 24 return self.NO_ERROR def _get_home(self): return self._home.gps def _set_logtable(self, logtable): self._logtable = logtable return self.NO_ERROR def _get_logtable(self): return self._logtable def _set_nbmax_last_lines(self, nbmax_last_lines): self._nbmax_last_lines = nbmax_last_lines return self.NO_ERROR def _get_nbmax_last_lines(self): return self._nbmax_last_lines # ===================================================================== # ===================================================================== # Property methods # ===================================================================== # ===================================================================== caller_alias = property(_get_caller_alias, _set_caller_alias) ##path_data = property(_get_path_data, _set_path_data) path_wwwpyros = property(_get_path_wwwpyros, _set_path_wwwpyros) home = property(_get_home, _set_home) logtable = property(_get_logtable, _set_logtable) # ===================================================================== # ===================================================================== # Public Property methods # ===================================================================== # ===================================================================== ##debug_level = property(_get_debug_level, _set_debug_level) nbmax_last_lines = property(_get_nbmax_last_lines, _set_nbmax_last_lines) # ===================================================================== # ===================================================================== # Methods for users # ===================================================================== # ===================================================================== def get_level_name(self, log_level:int): if log_level == self.LOG_LEVEL_DEBUG: return "DEBUG" if log_level == self.LOG_LEVEL_INFO: return "INFO" if log_level == self.LOG_LEVEL_WARNING: return "WARN" if log_level == self.LOG_LEVEL_ERROR: return "ERROR" if log_level == self.LOG_LEVEL_CRITICAL: return "CRITIC" raise Exception("Invalid log level number") def log_if_level_lower_than(self, log_level:int, *args, **kwargs): """ This is the method to print in the console display and in a log file. In the console display, the message is presented as: (Agent_name) message In the log file the message is presented as: Date-ISO message """ # FILTER : log only if global log level is <= log_level if self.get_global_log_level() > log_level: return msg = self._convert_args2str(*args, **kwargs) # --- prepare the super_msg formatted_msg = f"({self._caller_alias}) ({self.get_level_name(log_level)}) {msg}" if msg else "" # --- classical print print(formatted_msg) # --- no more if the agent name is not defined if not self._caller_alias: return # --- # LOG into a file if Agent or if level is enough ##self.file(msg) ##if self._caller_alias.startswith('Agent') or log_level >= self.LOG_LEVEL_WARNING: ##if self._caller_alias.startswith('Agent') or log_level >= self.LOG_LEVEL_INFO: #if self._caller_alias.startswith('Agent') or log_level >= self.LOG_LEVEL_DEBUG: if self._caller_alias.startswith('Agent') or log_level >= self._FILE_LOG_LEVEL: self.file(log_level, msg) if log_level >= self.LOG_LEVEL_WARNING: self.filewarn(log_level, msg) if log_level >= self.LOG_LEVEL_ERROR: self.sendmail_error(msg) ##if self._debug_level > 0: def printd(self, *args, **kwargs): self.log_d(*args, **kwargs) def log_d(self, *args, **kwargs): self.log_if_level_lower_than(self.LOG_LEVEL_DEBUG, *args, **kwargs) def log_i(self, *args, **kwargs): self.log_if_level_lower_than(self.LOG_LEVEL_INFO, *args, **kwargs) def log_w(self, *args, **kwargs): self.log_if_level_lower_than(self.LOG_LEVEL_WARNING, *args, **kwargs) def log_e(self, *args, **kwargs): self.log_if_level_lower_than(self.LOG_LEVEL_ERROR, *args, **kwargs) def log_c(self, *args, **kwargs): self.log_if_level_lower_than(self.LOG_LEVEL_CRITICAL, *args, **kwargs) # DEFAULT LOG methods that log at level INFO def log(self, *args, **kwargs): self.log_i(*args, **kwargs) def print(self, *args, **kwargs): self.log(*args, **kwargs) def log_msg_to_file(self, log_msg, path, file_name, night=None): # 1) Create path if not exists if not os.path.exists(path): try: os.makedirs(path) except: p = f"Cannot create log path {path}" raise Exception(p) # 2) APPEND msg to FULL DAY file file_abs_name = file_prefix = os.path.normpath(path + os.sep + file_name) if file_name != self._CHRONO_FILENAME_WARN: file_abs_name += '_' + night with open(file_abs_name+'.log', 'a') as fic: fic.write(log_msg+"\n") if file_name == self._CHRONO_FILENAME_WARN: return # 3) UPDATE (REWRITE) LAST LINES file (truncated to nbmax_last_lines) filetype = "chrono" if file_name==self._CHRONO_FILENAME else "agent" ll = self._last_lines[filetype] #ll = self._last_lines if file_name == self._CHRONO_FILENAME else self._last_lines_agent ll.append(log_msg) n = len(ll) if n > self.nbmax_last_lines: ll = ll[n-self.nbmax_last_lines:] #print("ll is", len(ll)) #print("self._last_lines is", len(self._last_lines)) with open(file_prefix+'_'+'last'+'.log','w') as fic: for line in ll: fic.write(line+"\n") ''' if file_name == self._CHRONO_FILENAME: self._last_lines = ll else: self._last_lines_agent = ll ''' self._last_lines[filetype] = ll ##def file(self, *args, **kwargs): def file(self, log_level:int, msg:str): """ This is the method to print in a log file. In the log file the message is presented as: Date-ISO message The last file is also appended with the message """ ##msg = self._convert_args2str(*args, **kwargs) # --- no more if the agent name is not defined if self._caller_alias=="": return # --- 1) Compute the current night digital date night = self._date2night("now") self._date.date("NOW") base_path = os.path.normpath(self.get_full_GLOBAL_PATH_DATA()) # --- 2) LOG message to the CHRONO daily unique log file name (pyros.log) path = base_path + os.sep + self._CHRONO_FILENAME msg_prefix = f"{self._date.iso()} ({self._caller_alias}) ({self.get_level_name(log_level)}) " self.log_msg_to_file(msg_prefix + msg, path, self._CHRONO_FILENAME, night) # --- 3) (only) If AGENT, DO THE SAME but to the AGENT daily file (in its own folder) if self._caller_alias.startswith("Agent"): path = base_path + os.sep + self._caller_alias msg_prefix = f"{self._date.iso()} ({self.get_level_name(log_level)}) " self.log_msg_to_file(msg_prefix + msg, path, self._caller_alias, night) def filewarn(self, log_level:int, msg:str): path = os.path.normpath(self.get_full_GLOBAL_PATH_DATA()) + os.sep + self._CHRONO_FILENAME_WARN msg_prefix = f"{self._date.iso()} ({self._caller_alias}) ({self.get_level_name(log_level)}) " self.log_msg_to_file(msg_prefix + msg, path, self._CHRONO_FILENAME_WARN) ''' # 1) Create path if not exists if not os.path.exists(path): try: os.makedirs(path) except: p = f"Cannot create log path {path}" raise Exception(p) log_filename = os.path.normpath(path + os.sep + self._CHRONO_FILENAME_WARN) log_msg = f"{self._date.iso()} ({self._caller_alias}) ({self.get_level_name(log_level)}) {msg}" with open(log_filename+'.log', 'a') as fic: fic.write(log_msg+"\n") ''' # TODO: def sendmail_error(self, msg:str): pass def db(self, *args, **kwargs): """ This is the method to print in the agent_log table of the database. """ if self.logtable==None: return msg = self._convert_args2str(*args, **kwargs) self.logtable.objects.create(name=self._caller_alias, message=msg) # ===================================================================== # ===================================================================== # Special methods # ===================================================================== # ===================================================================== def __init__(self, caller_name:str, logtable=None, home:str="GPS 0 E 43 150", path_data:str="", path_wwwpyros:str="" ): self._last_errno = self.NO_ERROR # --- self.caller_alias = caller_name if caller_name else "Unknown_agent" #print("CALLER is set to ", self.caller_alias) # --- if path_data: ##self.path_data = path_data self.set_global_path_data(path_data) elif not self.get_global_path_data(): ##self.path_data = tempfile.gettempdir() ##print("log self.path_data is", self.path_data) self.set_global_path_data(tempfile.gettempdir()) #print("log self.global_path_data is", self.get_global_path_data()) # --- if path_wwwpyros != "": self.path_wwwpyros = path_wwwpyros self._date = guitastro.Date() self._home = guitastro.Home(home) self._noon_hour = 12 ; # local hour corresponding to the date change self.logtable = None self.nbmax_last_lines = 30 # ===================================================================== # ===================================================================== # Test if main # ===================================================================== # ===================================================================== if __name__ == "__main__": # === Instance parameters # --- caller_alias = string added at the begining of each line of display log # --- caller_alias = is the name of log files # --- caller_alias = is the field name of database logs caller_alias = "test" # --- caller_alias = directory where are written log files path_data = "c:/srv/work/pyros" # --- home = define the hour of noon when log files are archived home = "GPS 2 E 43 148" # === Instanciation # --- The second parameter is the Django object for database log = LogPyros(caller_alias,None,home,path_data) # === Use of logs. Parameters are the same as a Python print a = 2 b = 'tutu' log.print(f"a={a} b={b}") log.print(a,b)