o rri2 @sddlZddlZddlZddlZddlZddlZddlZddlZddlZddl Z ddl Z ddl m Z ddlZejde jde jdfddZGdddeZGd d d eZGd d d eZGd ddZGdddZejde je efde je fddZ d+de je jee fde je jee fde fddZde jee fde jee fde fddZde jee fde jfddZ de jde jee fddfd d!Z! "d,d#ej"d$ej#d%ej#d&e$fd'd(Z%Gd)d*d*Z&dS)-N)Pathkwargsreturnc ksvdd|D}tjdd|DzdVW|D]}tj|=qtj|dS|D]}tj|=q-tj|w)NcSs"i|] }|tjvr|tj|qS)osenviron.0xrrG/home/ubuntu/.local/lib/python3.10/site-packages/submitit/core/utils.py s"z)environment_variables..cSsi|] \}}|t|qSrstrr r yrrr r s)rrupdateitems)rbackupr rrr environment_variabless  rc@eZdZdZdS)UncompletedJobErrorz.Job is uncomplete: either unfinished or failedN__name__ __module__ __qualname____doc__rrrr r#rc@r)FailedJobErrorzJob failed during processingNrrrrr r'rrc@r)FailedSubmissionErrorzJob Submission failedNrrrrr r+rrc @s"eZdZdZ d"dejeefdejedeje ddfddZ e defd d Z e defd d Z e defd dZe defddZe defddZe defddZdejeefdefddZ d#dejeefdededdfddZedejeefdefddZdefd d!ZdS)$JobPathsz9Creates paths related to the slurm job and its submissionNfolderjob_idtask_idrcCs&t||_||_|pd|_dS)Nr)r expanduserabsolute_folderr!r")selfr r!r"rrr __init__2szJobPaths.__init__cCs ||jSN) _format_idr%r&rrr r 9s zJobPaths.foldercCs0|jrd|jvr||jdS||jdS)N_z%A_submission.shz%j_submission.sh)r!r)r r*rrr submission_file=szJobPaths.submission_filecC||jdS)Nz%j_submitted.pklr)r r*rrr submitted_pickleDzJobPaths.submitted_picklecCr-)Nz%j_%t_result.pklr.r*rrr result_pickleHr0zJobPaths.result_picklecCr-)Nz %j_%t_log.errr.r*rrr stderrLr0zJobPaths.stderrcCr-)Nz %j_%t_log.outr.r*rrr stdoutPr0zJobPaths.stdoutpathcCs|jdur t|St|dt|jdt|j}t|jdd^}}d|vrmszMJobPaths.get_first_id_independent_folder....)anyrMtagsrMr msz:JobPaths.get_first_id_independent_folder..)rr#r$parts itertools takewhile)r rS indep_partsrrPr get_first_id_independent_folderhsz(JobPaths.get_first_id_independent_foldercCs|jjd|jdS)N()) __class__rr r*rrr __repr__pszJobPaths.__repr__)NNF)rrrrtpUnionrrOptionalintr'propertyr r,r/r1r2r3r)boolrK staticmethodrWr[rrrr r/sL     rc@seZdZdZdejdejfdejdejddfdd Zdejfd d Zde fd d Z dej e e fddfddZdededdfddZedejddej e e fddfddZdejdfddZdS)DelayedSubmissiona]Object for specifying the function/callable call to submit and process later. This is only syntactic sugar to make sure everything is well formatted: If what you want to compute later is func(*args, **kwargs), just instanciate: DelayedSubmission(func, *args, **kwargs). It also provides convenient tools for dumping and loading. function.argsrrNcOs.||_||_||_d|_d|_d|_d|_dS)NFr)rerfr_result_done _timeout_min_timeout_countdown)r&rerfrrrr r'|s zDelayedSubmission.__init__cCs.|jr|jS|j|ji|j|_d|_|jS)NT)rhrgrerfrr*rrr results zDelayedSubmission.resultcCs|jSr()rhr*rrr doneszDelayedSubmission.donefilepathcCst||dSr()cloudpickle_dump)r&rmrrr dumpszDelayedSubmission.dump timeout_minmax_num_timeoutcCs||_||_dSr()rirj)r&rprqrrr set_timeouts zDelayedSubmission.set_timeoutclscCs4t|}|jj|jksJdt|d|d|S)NzLoaded object is z but should be .) pickle_loadrZrtype)rsrmobjrrr loads(zDelayedSubmission.loadcCsBt|jdd}|durt|jdd}|durdS||ji|jS)N__submitit_checkpoint__ checkpoint)rIrerfr)r&rzrrr _checkpoint_functions z&DelayedSubmission._checkpoint_function)rrrrr]CallableAnyr'rkrbrlr^rrror`rr classmethodTyperxr_r{rrrr rdts* (rdrmccsbt|}||jd}|rJd|V|s td|r)t|t||dS)aTYields a path where to save a file and moves it afterward to the provided location (and replaces any existing file) This is useful to avoid processes monitoring the filepath to break if trying to read when the file is being written. Note ---- The temporary path is the provided path appended with .save_tmp z .save_tmpz&A temporary saved file already exists.z(No file was saved at the temporary path.N)r with_suffixsuffixexistsFileNotFoundErrorrremoverH)rmtmppathrrr temporary_save_paths  rfoldersoutfilecCst|ttfs Jd|durd}t|}t|ds Jdtj|dd}|D]}|jt|t|j dq*Wd|S1sDwY|S) z/Creates a tar.gz file with all provided foldersz,Only lists and tuples of folders are allowedNz_dev_folders_.tar.gzz.tar.gzz(Archive file must have extension .tar.gzw)mode)arcname) isinstancelisttuplerrendswithtarfileTarFileaddrB)rrtfr rrr archive_dev_folderss rpar_filer cCsHt|}t|}|jddd||j}t|||S)aCopy the par (or xar) file in the folder Parameter --------- par_file: str/Path Par file generated by buck folder: str/Path folder where the par file must be copied Returns ------- Path Path of the copied .par file TrD)rr#r$rGrBshutilcopy2)rr dst_namerrr copy_par_files   rfilenamecCs8t|d }t|WdS1swYdS)Nrb)openpicklerx)rifilerrr rus $rurwcCs@t|d}t||tjWddS1swYdS)Nwb)r cloudpicklerorHIGHEST_PROTOCOL)rwrofilerrr rns "rnFprocessr3r2verbosecCsdtjtjtdtjtfdd}||j||j}}|||tjf|||tjfi}t| }t } |D] } | | t j t jBq;|r| } | D]5\} } || \} }}| d}|sm|| | | qP|}||||r|||qP|sJdSdS)z Reads the given process stdout/stderr and write them to StringIO objects. Make sure that there is no deadlock because of pipe congestion. If `verbose` the process stdout/stderr are also copying to the interpreter stdout/stderr. streamrcSs&|durtdt|tjr|j}|S)NzStream should not be None) RuntimeErrorrioBufferedIOBaseraw)rrrr rs  z!copy_process_streams..rawiN)r]r_IObytesr3r2filenosysrkeysselectpollregisterPOLLINPOLLPRIreadr unregisterdecodewriteflush)rr3r2rrp_stdoutp_stderr stream_by_fdfdspollerfdreadyr+p_streamstringstdraw_bufbufrrr copy_process_streamss4$         rc @sveZdZdZ   ddejededejej ee fdejej eefddf d d Z d ej d ej defd dZdS)CommandFunctionaWraps a command as a function in order to make sure it goes through the pipeline and notify when it is finished. The output is a string containing everything that has been sent to stdout. WARNING: use CommandFunction only if you know the output won't be too big ! Otherwise use subprocess.run() that also streams the outputto stdout/stderr. Parameters ---------- command: list command to run, as a list verbose: bool prints the command and stdout at runtime cwd: Path/str path to the location where the command must run from Returns ------- str Everything that has been sent to stdout TNcommandrcwdenvrcCs>t|ts td||_||_|durdnt||_||_dS)Nz&The command must be provided as a list)rr TypeErrorrrrrr)r&rrrrrrr r'3s  zCommandFunction.__init__rfrc Os<|jdd|Ddd|D}|jr!tdd|dtj|tjtjd|j|j dd}t }t }z t ||||jWnt y\}z ||td |d }~ww|}|} |} | r{| r{|js{t| tjd | rtj| |j|| d } t| | Wd |S1swY|S) aCall the cammand line with addidional arguments The keyword arguments will be sent as --{key}={val} The logs bufferized. They will be printed if the job fails, or sent as output of the function Errors are provided with the internal stderr. cSsg|]}t|qSrr rrrr Hsz,CommandFunction.__call__..cSs g|] \}}d|d|qS)z--=rrrrr rHs z The following command is sent: " "F)r3r2shellrrz%Job got killed for an unknown reason.N)file)outputr2)rrrprintjoin subprocessPopenPIPErrrStringIOr Exceptionkillwaitrgetvaluestriprr2CalledProcessErrorrf) r&rfr full_commandr stdout_buffer stderr_bufferer3r2retcodesubprocess_errorrrr __call__AsL$      zCommandFunction.__call__)TNN)rrrrr]Listrrbr_r^rDictr'r}rrrrr rs" rr(r\)' contextlibrrTrrrrrrrtypingr]pathlibrrcontextmanagerr}Iteratorrrrrrrrdr^rrrr_rrrurnrrrbrrrrrr sX  E4$ *" ,