"""log_interface handling.For the moment, the log interface is managed using the pyswarms reporter class to avoid interference between loggers."""importloggingimportosimportsysimporttqdmfromicecreamimporticfromtimeimportsleepfrom._MCoreimportMCHfrom._parametersimportparametersfrompyswarms.utilsimportReporter
rep_nrv=init_reporter()# !! ISSUE:# Compatibility between logging and pyswarms when pyswarms is imported (not used necessarily)# all messages log are also printed# !! TEMPORARY SOLUTION# Using a pyswarms reporter to log messages# see old code commented# logger = logging.getLogger()# # logging config# logging.basicConfig(# filename=dir_path + "/log/NRV.log",# level=logging.INFO,# format="%(asctime)s %(message)s",# datefmt="%m/%d/%Y %I:%M:%S %p",# )
[docs]defrise_error(*args,out=1,**kwargs):""" Rises and error to the log and to the prompt for the master process (process ID 0) in case of parallel computing. This function exit the programm. Parameters ---------- *args : anything to pass as error message out : int programm exit value, must be strictly positive. By default equal to 1 """verbose=Trueif"verbose"inkwargs:verbose=kwargs["verbose"]message=""error=Noneforarginargs:print(type(arg),isinstance(arg,Exception))ifisinstance(arg,type)andisinstance(arg(),Exception):error=argelse:message+=str(arg)ifMCH.is_alone():ifparameters.LOG_Status:rep_nrv.log(message,lvl=logging.ERROR)# logger.error(message, lvl=logging.ERROR)ifverboseandparameters.VERBOSITY_LEVEL>=1:print(bcolors.FAIL+"NRV ERROR: "+message+bcolors.ENDC)else:err=("NRV ERROR: "+message+"\n encountered in process "+str(MCH.rank)+" out of "+str(MCH.size))ifparameters.LOG_Status:rep_nrv.log(err,lvl=logging.ERROR)# logger.error(err, lvl=logging.ERROR)ifparameters.VERBOSITY_LEVEL>=1:print(err)sys.stdout.flush()ifout==0:out=1ifnoterrorisNone:raiseerror(message)sys.exit(out)
[docs]defrise_warning(*args,abort=False,**kwargs):""" Rises a warning to the log and to the prompt for the master process (process ID 0) in case of parallel computing. This function can exit the programm. Parameters ---------- *args : anything to pass as warning message abort : boolean if true, the programm exits with the value O (no error) and further computation are avoided, by default set to False. """verbose=Trueif"verbose"inkwargs:verbose=kwargs["verbose"]message=""forarginargs:message+=str(arg)ifMCH.is_alone():ifparameters.LOG_Status:rep_nrv.log("NRV DEBUG: "+message,lvl=logging.DEBUG)# logger.warning("NRV WARNING: " + message)ifverboseandparameters.VERBOSITY_LEVEL>=2:print(bcolors.WARNING+"NRV WARNING: "+message+bcolors.ENDC)else:war=("NRV WARNING: "+message+"\n encountered in process "+str(MCH.rank)+" out of "+str(MCH.size))ifparameters.LOG_Status:rep_nrv.log(war,lvl=logging.DEBUG)# logger.warning(war)ifMCH.do_master_only_work()andparameters.VERBOSITY_LEVEL>=2:print(bcolors.WARNING+war+bcolors.ENDC)sys.stdout.flush()elifparameters.VERBOSITY_LEVEL>=4:print(bcolors.WARNING+war+bcolors.ENDC)sys.stdout.flush()ifabort:sys.exit(0)
[docs]defpass_info(*args,**kwargs):""" Pass an info to the log and to the prompt for the master process (process ID 0) in case of parallel computing. Parameters ---------- *args : anything to pass as info """verbose=Trueif"verbose"inkwargs:verbose=kwargs["verbose"]message=""forarginargs:message+=str(arg)ifMCH.is_alone():ifparameters.LOG_Status:rep_nrv.log("NRV DEBUG: "+message,lvl=logging.INFO)# logger.info("NRV DEBUG: " + message)ifverboseandparameters.VERBOSITY_LEVEL>=3:print("NRV INFO: "+message)else:inf=("NRV INFO: "+message+"\n from process "+str(MCH.rank)+" out of "+str(MCH.size))ifparameters.LOG_Status:rep_nrv.log(inf,lvl=logging.INFO)# logger.info(inf)ifparameters.VERBOSITY_LEVEL>=4:print(inf)sys.stdout.flush()
[docs]defpass_debug_info(*args,**kwargs):""" Pass an info to the log and to the prompt for the master process (process ID 0) in case of parallel computing. Parameters ---------- *args : anything to pass as info """verbose=Trueif"verbose"inkwargs:verbose=kwargs["verbose"]message=""forarginargs:message+=str(arg)ifMCH.is_alone():ifparameters.LOG_Statusandparameters.VERBOSITY_LEVEL>=4:rep_nrv.log("NRV DEBUG: "+message,lvl=logging.DEBUG)# logger.info("NRV DEBUG: " + message)ifverboseandparameters.VERBOSITY_LEVEL>=4:print("NRV DEBUG: "+message)else:inf=("NRV DEBUG: "+message+"\n from process "+str(MCH.rank)+" out of "+str(MCH.size))ifparameters.LOG_Status:rep_nrv.log(inf,lvl=logging.DEBUG)# logger.info(inf)ifparameters.VERBOSITY_LEVEL>=4:print(inf)sys.stdout.flush()
######### pbar #########
[docs]classpbar:""" Progess bar object compatible with MPI """instanciate=True
def__del__(self):delself._pbarifMCH.do_master_only_work():# blanck print for new lineprint()sys.stdout.flush()def__instantiate_delay(self):# Dirty hack to prevent skiping two linessleep(0.05*(MCH.rank))
[docs]defprogression_popup(current,max_iter,begin_message="",end_message="",endl=""):""" Displays the progression on prompt, for single process only, nothing saved to the log Parameters ---------- current : int current iteration max_iter : int maximum iteration number begin_message : str additional message placed at the start of the progression line end_message : str additional message placed after iteration and max at the end of the progression line endl : str line termination, by default empty string """ifMCH.is_alone():print(begin_message+f"{current+1}"+"/"+str(max_iter)+end_message,end=endl,)ifcurrent==max_iter-1:print("\n")
[docs]defprompt_debug(*args):""" outputs for debug, simple redirection to icecream without the installation (see icecream help). Could be interesting to redirect also to the log... Parameters ---------- *args : anything to pass to icecrean """ic(*args)
[docs]defclear_prompt_line(n:int=1)->None:""" Clear lines of the prompt, to overwrite stuffs. Parameters ---------- n : int, optional Number of line to clear, by default 1 """LINE_UP="\033[1A"LINE_CLEAR="\x1b[2K"foriinrange(n):print(LINE_UP,end=LINE_CLEAR)