import sys import os from os import path try: import yaml assert yaml.__version__ > '5.1' except: sys.exit( 'package "PyYAML" (>= 5.1) is required but not installed! Run\n' + 'python -m pip install pyyaml --user \n' + 'to install it.' ) try: from pandas import DataFrame except: sys.exit( 'package "pandas" is required but not installed! Run\n' + 'python -m pip install pandas --user\n' + 'to install it.' ) import glob import re import shutil import datetime import palm import palm.config import palm.system LOG = palm.logger(__name__) def read_io2df(file, doActivate_string, create_remote_batch=False, running_on_remote=False, remote_folder=''): # CHECK FOR MULTIPLE FILES, SET A RESPECTIVE FLAG # REMOVE THE WILDCARD FROM THE ENDING def checkMultiFiles(string): multi=False retstring=string if string.endswith('*'): multi=True retstring=string[:-1] return multi, retstring try: LOG.debug( 'start reading iofiles ...' ) doActivate = doActivate_string.split() with open(file, 'r') as f: res_in=[] res_out=[] line_count = 0 for line in f.readlines(): line_count += 1 if not line.strip(): #ignore empty lines continue if line.strip().startswith('#'): #ignore comment lines continue iolist=line.split() assert (len(iolist) in {5,6}), 'wrong number of columns in .palm.iofiles' assert (len(iolist[1].split(':')) in {1,2}), 'wrong number of file attributes: Two are allowed as a maximum! ' filetype = iolist[1].split(':')[0] action = iolist[1].split(':')[1:] if not action: action = [''] activationList = iolist[2].split(':') # only files whose activation string matches the string given by option -a are processed for activation in activationList: if activation in doActivate or activation == '*': local_pre = iolist[0] #localin_pre/localout_pre path_pre = iolist[3] # pathin_pre/pathout_pre end_pre = iolist[4] if len(iolist) == 5: ext_pre = '' #extin_pre/extout_pre else: ext_pre = iolist[5] # STORE FILE CONNECTION, IF ACTIVATED BY ACTIVATION-STRING FROM # INPUT- OR OUTPUT-LIST. # VARIABLE S3 MAY CONTAIN A LIST OF ACTIVATION STRINGS (FIELD-SEPERATOR ":"). # IF EXECUTION IS SCHEDULED FOR A REMOTE-MACHINE AND THE FILE IS ONLY # LOCALLY REQUIRED ON THAT MACHINE (I.E. s2b != tr), THE FILE CONNECTION # IS NOT CHECKED AND STORED. if (filetype == 'in' or filetype == 'inopt') and not(create_remote_batch == True and action[0] != 'tr'): multi, end_pre = checkMultiFiles(end_pre) # FILES WITH JOB-ATTRIBUTE ARE STORED IN THE SOURCES_FOR_RUN # FOLDER IF THE JOB IS RUNNING ON A REMOTE HOST if running_on_remote == True and action[0] == 'tr': path_pre = remote_folder inopt=False if filetype == 'inopt': inopt=True res_in.append([inopt,multi,local_pre,action[0],path_pre,end_pre,ext_pre]) elif filetype == 'out' and create_remote_batch == False: multi, local_pre = checkMultiFiles(local_pre) if multi: assert action[0] != 'di', '+++ wildcards (*) not allowed with "di" file attribute.' \ + '\n see file "'+file+'", line ' + str(line_count) # CHECK IF WILDCARD IS USED AS ACTIVATION STRING # IN SUCH CASES, NO WARNING WILL LATER BE OUTPUT IF LOCAL FILES DO NOT EXIST warn = True if activation == '*': warn = False res_out.append([multi,local_pre,action[0],path_pre,end_pre,ext_pre,warn]) elif filetype !='in' and filetype !='inopt' and filetype !='out': LOG.error('+++ I/O-attribute in file "'+file+'" has invalid ' \ + '\n value "'+filetype+'". Only "in", "inopt", and "out" are allowed!') raise NameError('wrong argument in .palm.iofiles: "'+file+'"') break df_in = DataFrame(res_in, columns=['inopt','multi','local','action','path','end','ext']) df_out = DataFrame(res_out, columns=['multi','local','action','path','end','ext','warn']) return df_in, df_out except Exception as exception: LOG.error( str(exception) ) LOG.error( '+++ reading iofiles failed...' ) raise def config_inFiles(df_in, run_identifier, expand): def EXP_ENV( message,context=os.environ ): message = expand( message ) return expand( message, keep_unknown=True, context=context ) # GENERATE FULL FILENAMES OF INPUT-FILES, INCLUDING THEIR PATH # CHECK, IF INPUT-FILES EXIST, AND DETERMINE HIGHEST CYCLE NUMBER (IF CYCLES EXIST) try: fin = [] for i in df_in.index: #Loop over all input files if df_in.loc[i,'action'] == 'di': filename = EXP_ENV( path.join(df_in.loc[i,'path'],df_in.loc[i,'end']) ) else: filename = EXP_ENV( path.join(df_in.loc[i,'path'],run_identifier + df_in.loc[i,'end']) ) # CHECK IF FILE EXISTS if not glob.glob(filename+'*'): # FILES WITH ATTRIBUTE opt ARE OPTIONAL. NO ABORT, IF THEY DO NOT EXIST. if df_in.loc[i,'inopt'] == 'False': LOG.error('+++INPUT-file: ') if df_in.loc[i,'ext'] == '' or df_in.loc[i,'ext'] == ' ': LOG.error('\n '+filename) else: LOG.error('\n '+filename+'.'+''+df_in.loc[i,'ext']) LOG.error('\n ....does not exist\n') raise Exception else: fin_inopt = df_in.loc[i,'inopt'] fin_local = df_in.loc[i,'local'] fin_action = 'unavailable' fin_path= EXP_ENV(df_in.loc[i,'path']) fin_end =df_in.loc[i,'end'] fin_ext = df_in.loc[i,'ext'] fin_absname='' fin_frel='' fin_gottmp = False else: # FIRST CHECK FOR MULTIPLE NAMES WITH THE SAME BASENAME # ($run_identifier) AND CREATE A LIST FOR THE DETECTED BASENAME # ENDINGS if df_in.loc[i,'multi'] == True: #find endings -> ending endings = [] filelist = glob.glob(filename+'[_.]*') if len(glob.glob(filename)) == 1: filelist.append(glob.glob(filename)[0]) for file in filelist: # filename without path (i.e. after the last "/") basename = path.basename(file) # check if there is an extension and remove it ext = path.splitext(basename)[1] if ext == df_in.loc[i,'ext']: basename = path.splitext(basename)[0] # check for an existing cycle number and remove it cycle = path.splitext(basename)[1] cycle=cycle.replace('.','') if len(re.findall('^[0-9]+$',cycle)) == 1: basename = path.splitext(basename)[0] # remove the run_identifier from the beginning ending = basename.replace(run_identifier,'') # remove the ending given in the .iofiles from the beginning ending = ending.replace(df_in.loc[i,'end'],'') if ending == '': endings.append('DEFAULT') else: if ending.startswith('_'): endings.append(ending) else: #single name endings = ['DEFAULT'] endings=set(endings) #get a unique list endings=[e.replace('DEFAULT','') for e in endings] # FOR EACH BASENAME ENDING CREATE AN ENTRY IN THE FINAL INPUT FILE LIST for ending in endings: # NEW ENTRY (ENDING IS ALSO ADDED TO LOCAL FILENAME READ BY PALM!) fin_inopt = df_in.loc[i,'inopt'] fin_local = df_in.loc[i,'local']+ending fin_action = df_in.loc[i,'action'] fin_path= EXP_ENV(df_in.loc[i,'path']) fin_end =df_in.loc[i,'end']+ending fin_ext = df_in.loc[i,'ext'] fin_gottmp = False # GENERATE PATH AND FULL FILE NAME (then-BRANCH: FIXED FULL NAME IS GIVEN, I.E. THE #FILE IDENTIFIER IS NOT PART OF THE FILENAME)) if fin_action == 'di': filename = path.join(fin_path, fin_end) else: filename = path.join(fin_path, run_identifier + fin_end) # DETERMINE THE FILE'S CYCLE NUMBER maxcycle = 0 explicit_cycle_number=False filelist = glob.glob(filename+'[.]*') if len(glob.glob(filename)) == 1: filelist.append(glob.glob(filename)[0]) for file in filelist: # filename without path (i.e. after the last "/") basename = path.basename(file) # check if there is an extension and remove it ext = path.splitext(basename)[1] ext=ext.replace('.','') if ext == fin_ext: basename = path.splitext(basename)[0] # check for an existing cycle number and remove it cycle = path.splitext(basename)[1] cycle=cycle.replace('.','') if len(re.findall('^[0-9]+$',cycle)) == 1: icycle = int(re.findall('^-?[0-9]+$',cycle)[0]) explicit_cycle_number = True else: icycle = 0 if icycle > maxcycle: maxcycle = icycle # APPEND CYCLE NUMBER TO FILENAME if explicit_cycle_number == True: if fin_ext != '' and fin_ext !=' ': filename = filename + '.' + str(maxcycle).zfill(3)+'.' + fin_ext else: filename = filename + '.' + str(maxcycle).zfill(3) else: if fin_ext != '' and fin_ext !=' ': filename = filename + '.' + fin_ext # STORE FILENAME WITHOUT PATH BUT WITH CYCLE NUMBER, # IS LATER USED FOR TRANSFERRING FILES WIHIN THE JOB (SEE END OF FILE) fin_absname = filename if explicit_cycle_number == True: if fin_action == 'di': fin_frel = fin_end + '.' + str(maxcycle).zfill(3) else: fin_frel = run_identifier + fin_end + '.' + str(maxcycle).zfill(3) else: if fin_action == 'di': fin_frel = fin_end else: fin_frel = run_identifier + fin_end fin.append([fin_inopt, fin_local, fin_action, fin_path, fin_end, fin_ext, fin_absname, fin_frel, fin_gottmp]) return DataFrame(fin, columns=['inopt','local','action','path','end','ext','absname','frel','gottmp']) except Exception as exception: LOG.error( str(exception) ) LOG.error( '+++ configuration of input files failed...' ) raise def config_outFiles_beforepalm(df_out, expand, remote, run_identifier): def EXP_ENV( message,context=os.environ ): message = expand( message ) return expand( message, keep_unknown=True, context=context ) def palm_mkdir( directory, name='', errmessage=''): if not path.isdir( directory ): try: os.makedirs(directory) except: LOG.error('') LOG.error('+++ can\'t create directory:' ) LOG.error(' ' + directory ) LOG.error(errmessage) raise try: for i in df_out.index: #Loop over all output files if not (remote == True and (df_out.loc[i,'action'] in ['tr','tra','trpe'])): if df_out.loc[i,'action'] == 'tr': df_out.loc[i,'action'] = '' elif df_out.loc[i,'action'] == 'trpe': df_out.loc[i,'action'] = 'pe' elif df_out.loc[i,'action'] == 'tra': df_out.loc[i,'action'] = 'a' filename = EXP_ENV(path.join(df_out.loc[i,'path'], run_identifier + df_out.loc[i,'end']) ) catalogname = EXP_ENV(df_out.loc[i,'path']) # IF OUTPUT-FILE DOES NOT EXIST CHECK, IF IT CAN BE CREATED if not glob.glob(filename+'*'): palm_mkdir(catalogname, name='OUTPUT') try: with open(filename,'w') as f: os.remove(filename) except IOError as err: LOG.error('') LOG.error('+++ OUTPUT-file \n'\ + filename + ' cannot be created.') LOG.error(' '+ err.strerror) raise Exception #BUGFIX: source code moved after calling palm (reason: multiple output files in coupled run, nesting) # -> separate function: ioFiles.config_outFiles_afterpalm() return df_out except Exception as exception: LOG.error( str(exception) ) LOG.error( '+++ configuration of output files failed...' ) raise #BUGFIX: source code moved after calling palm (reason: multiple output files in coupled run, nesting) # -> separate function: ioFiles.config_outFiles_afterpalm() def config_outFiles_afterpalm(df_out, expand, remote): def EXP_ENV( message,context=os.environ ): message = expand( message ) return expand( message, keep_unknown=True, context=context ) try: fout = [] for i in df_out.index: #Loop over all output files # FIRST CHECK FOR MULTIPLE NAMES WITH THE SAME LOCAL NAME AND # CREATE A LIST FOR THE DETECTED ENDINGS if df_out.loc[i,'multi'] == True: filename = df_out.loc[i,'local'] # DETERMINE THE EXISTING EXTENSIONS FROM THE LIST OF FILES endings = ['DEFAULT'] filelist = glob.glob(filename+'_*') if len(glob.glob(filename)) == 1: filelist.append(glob.glob(filename)[0]) for file in filelist: # remove the local name from the beginning ending = file[len(filename):] if ending != '': endings.append(ending) else: # SINGLE NAME endings=['DEFAULT'] # DEFAULT MEANS THAT THE ENDING GIVEN IN .iofiles SHALL BE USED endings=[e.replace('DEFAULT','') for e in endings] # FOR EACH BASENAME ENDING CREATE AN ENTRY IN THE FINAL OUTPUT FILE LIST for ending in endings: # NEW ENTRY (ENDING IS ALSO ADDED TO LOCAL FILENAME READ BY PALM!) fout_multi = df_out.loc[i,'multi'] fout_local = df_out.loc[i,'local']+ending fout_action = df_out.loc[i,'action'] if not remote: fout_path= EXP_ENV(df_out.loc[i,'path']) else: #DON'T RESOLVE ENVIRONMENT VARIABLES IF RUNNING ON REMOTE fout_path= df_out.loc[i,'path'] fout_end =df_out.loc[i,'end']+ending fout_ext = df_out.loc[i,'ext'] fout_warn = df_out.loc[i,'warn'] fout.append([fout_multi, fout_local, fout_action, fout_path, fout_end, fout_ext, fout_warn]) return DataFrame(fout, columns=['multi','local','action','path','end','ext','warn']) except Exception as exception: LOG.error( str(exception) ) LOG.error( '+++ configuration of output files failed...' ) raise def provide_inFiles(df_in, expand, cores, remote): def EXP_ENV( message,context=os.environ ): message = expand( message ) return expand( message, keep_unknown=True, context=context ) def palm_mkdir( directory, name='', errmessage=''): if not path.isdir( directory ): try: os.makedirs(directory) except: LOG.error('') LOG.error('+++ can\'t create directory:' ) LOG.error(' ' + directory ) LOG.error(errmessage) raise # PROVIDE THE INPUT FILES # LOOP OVER ALL ACTIVATED FILES (LISTED IN THE CONFIGURATION FILE) optional_files_missing = False if len(df_in.index) > 0: LOG.info('\n\n*** providing INPUT-files:') LOG.info('---------------------------------------------------------') for i in df_in.index: # SKIP OPTIONAL FILES, IF THEY DO NOT EXIST if df_in.loc[i,'action'] == 'unavailable': optional_files_missing = True continue # CHECK FOR SINGLE FILE (SERIAL RUN) OR DIRECTORY (ONE FILE PER CORE FOR PARELLEL EXECUTION) files_for_cores = False filetype = 'file' if df_in.loc[i,'action'] == 'pe' and cores != None: files_for_cores = True filetype = 'files' df_in.loc[i,'action'] = '' elif df_in.loc[i,'action'] == 'pe' and cores == None: df_in.loc[i,'action'] = '' elif df_in.loc[i,'action'] == 'lnpe' and cores != None: files_for_cores = True filetype = 'files' df_in.loc[i,'action'] = 'ln' # FILE HAS BEEN FOUND INSTEAD OF DIRECTORY if path.isfile(df_in.loc[i,'absname']): files_for_cores = False filetype = 'file' elif df_in.loc[i,'action'] == 'lnpe' and cores == None: df_in.loc[i,'action'] = 'ln' if files_for_cores: LOG.info('>>> INPUT: ' + df_in.loc[i,'absname'] + '/... to ' + df_in.loc[i,'local']) else: LOG.info('>>> INPUT: ' + df_in.loc[i,'absname'] + ' to ' + df_in.loc[i,'local']) if df_in.loc[i,'action'] == 'ln': LOG.info('\n ' + filetype + ' will be linked') if not files_for_cores: if path.isfile(df_in.loc[i,'absname']): os.link( df_in.loc[i,'absname'],df_in.loc[i,'local'] ) df_in.loc[i,'gottmp'] = True else: if path.isdir(df_in.loc[i,'absname']): palm_mkdir(df_in.loc[i,'local'], name = '') try: for file in os.listdir(df_in.loc[i,'absname']): os.link( path.join(df_in.loc[i,'absname'],file), path.join(df_in.loc[i,'local'],file) ) except: LOG.info('\n--- WARNING: ln failed, using cp instead (might be time consuming...') shutil.rmtree(df_in.loc[i,'local'], ignore_errors=True) shutil.copytree(df_in.loc[i,'absname'], df_in.loc[i,'local'], copy_function=shutil.copy) df_in.loc[i,'gottmp'] = True # FILE IS STORED IN THE RESPECTIVE DIRECTORY GIVEN IN THE CONFIGURATION FILE if df_in.loc[i,'action'] == '' or \ df_in.loc[i,'action'] == 'di' or \ df_in.loc[i,'action'] == 'tr': if files_for_cores: # PROVIDE FILES FOR EACH CORE # COPY FILES FROM THE PERMANENT DIRECTORY TO THE LOCAL DIRECTORY LOG.info('\n providing '+cores+' files for the respective cores') #TODO (!?) link_local_input = true: Zeile 2175 in palmrun (Parameter bisher nicht in Konfigurationsdatei vorgesehen...): kein Linken shutil.copytree(df_in.loc[i,'absname'], df_in.loc[i,'local'], copy_function=shutil.copy) else: #TODO (!?) link_local_input == True: link statt copy if remote and df_in.loc[i,'action'] == 'tr': shutil.move(df_in.loc[i,'absname'],df_in.loc[i,'local']) else: shutil.copy(df_in.loc[i,'absname'],df_in.loc[i,'local']) if len(df_in.index) > 0: if optional_files_missing: LOG.info('*** INFORMATIVE: some optional INPUT-files are not present') LOG.info('---------------------------------------------------------') LOG.info('*** all INPUT-files provided') def provide_outFiles(df_out, expand, cores, run_identifier, configuration_identifier, run_id_num, remote, scp_yamlfile): def EXP( message ): return expand( message, keep_unknown=True ) def EXP_ENV( message,context=os.environ ): message = expand( message ) return expand( message, keep_unknown=True, context=context ) def palm_shell( command, capture_output=False, ignore_error=False ): return palm.system.shell( EXP(command), None, capture_output, ignore_error ) def palm_mkdir( directory, name='', errmessage=''): if not path.isdir( directory ): try: os.makedirs(directory) except: LOG.error('') LOG.error('+++ can\'t create directory:' ) LOG.error(' ' + directory ) LOG.error(errmessage) raise transfer_problems = False # remote if remote: scp_list = [] for i in df_out.index: if i == 0: # COPY LOCAL OUTPUT-FILES TO THEIR PERMANENT DESTINATIONS calltime = datetime.datetime.now( datetime.timezone.utc ).strftime('%a %b %d %H:%M:%S UTC %Y') LOG.info('\n\n*** saving OUTPUT-files: local time: '+calltime) # GET RUN NUMBER ASSIGNED BY PALM if path.isfile('RUN_NUMBER'): with open('RUN_NUMBER', 'r') as f: run_number = int(f.read()) LOG.info('\n*** PALM generated run_number = "'+str(run_number)+'" will be used as unified cycle number for all output files') usecycle_option=str(run_number) else: run_number = 0 usecycle_option='' if not (remote and (df_out.loc[i,'action'] == 'tr' or df_out.loc[i,'action'] == 'tra' or df_out.loc[i,'action'] == 'trpe')): filename = EXP_ENV( path.join(df_out.loc[i,'path'], run_identifier+df_out.loc[i,'end']) ) maxcycle = 0 filelist = glob.glob(filename+'[.]*') if len(glob.glob(filename)) == 1: filelist.append(glob.glob(filename)[0]) for file in filelist: # filename without path (i.e. after the last "/") basename = path.basename(file) # check if there is an extension and remove it ext = path.splitext(basename)[1] ext=ext.replace('.','') if ext == df_out.loc[i,'ext']: basename = path.splitext(basename)[0] # check for an existing cycle number cycle = path.splitext(basename)[1] cycle=cycle.replace('.','') if len(re.findall('^[0-9]+$',cycle)) == 1: icycle = int(cycle) + 1 else: icycle = 1 if icycle > maxcycle: maxcycle = icycle # SET THE CYCLE NUMBER # IN CASE OF FILE-APPEND, IT MUST BE THE HIGHEST EXISTING CYCLE NUMBER if df_out.loc[i,'action'] == 'a': if maxcycle > 0: maxcycle = maxcycle -1 cycnum = maxcycle df_out.loc[i,'path'] = filename # ADD CYCLE NUMBER TO FILENAME # IN APPEND MODE, FILES KEEP THEIR CURRENT CYCLE NUMBER if df_out.loc[i,'action'] != 'a': # SET RUN NUMBER AS CYCLE NUMBER, IF THERE IS NOT A CONFLICT # WITH AN EXISTING CYCLE NUMBER if run_number >= cycnum: cycnum = run_number else: if run_number > 0: LOG.info("--- INFORMATIVE: The following file cannot get a unified cycle number") df_out.loc[i,'path'] = df_out.loc[i,'path'] + '.'+str(cycnum).zfill(3) # CHECK FOR SINGLE FILE (SERIAL RUN) OR DIRECTORY (ONE FILE PER CORE FOR PARALLEL EXECUTION) files_for_cores = False #TODO (!?) filetype never used filetype ='file' link_local_output = False if df_out.loc[i,'action'] == 'pe' and cores != None: files_for_cores = True filetype ='directory' df_out.loc[i,'action'] = '' elif df_out.loc[i,'action'] == 'pe' and cores == None: df_out.loc[i,'action'] = '' elif df_out.loc[i,'action'] == 'lnpe'and cores != None: files_for_cores = True filetype ='directory' link_local_output = True df_out.loc[i,'action'] = '' #FILE HAS BEEN FOUND INSTEAD OF DIRECTORY if path.isfile(df_out.loc[i,'local']): files_for_cores = False filetype = file elif df_out.loc[i,'action'] == 'lnpe'and cores == None: link_local_output = True df_out.loc[i,'action'] = '' elif df_out.loc[i,'action'] == 'trpe'and cores != None: files_for_cores = True filetype ='directory' df_out.loc[i,'action'] = 'tr' elif df_out.loc[i,'action'] == 'trpe'and cores == None: df_out.loc[i,'action'] = 'tr' if not path.isfile(df_out.loc[i,'local']) and files_for_cores == False: if df_out.loc[i,'warn']: LOG.info('\n+++ temporary OUTPUT-file "' + df_out.loc[i,'local'] + '" does not exist') elif not path.isdir(df_out.loc[i,'local']) and files_for_cores == True: if df_out.loc[i,'warn']: LOG.info('\n+++ temporary OUTPUT-file "' + df_out.loc[i,'local'] + '/...." does not exist') else: # COPY VIA SCP TO LOCAL HOST # IF TARGET DIRECTORY DOES NOT EXISTS, TRY TO CREATE IT if df_out.loc[i,'action'] == 'tr' or df_out.loc[i,'action'] == 'tra': if remote: # SET OPTIONS FOR TRANSFER VIA batch_scp if df_out.loc[i,'action'] == 'tr': if not files_for_cores: catalog_option = 'file' catalog_string = '' else: catalog_option = 'dir' catalog_string = '/' append_option = False append_string = '' else: append_option = True append_string = 'append' file_local = configuration_identifier + '_'+run_identifier + df_out.loc[i,'end'] + catalog_string path_local = EXP(df_out.loc[i,'path']) LOG.info('\n >>> OUTPUT: ' + df_out.loc[i,'local']+catalog_string +' '+append_string+'SCP to"') LOG.info(' ' + path.join(path_local, file_local)) #SAVE DATA ON REMOTE MACHINE path_remote = EXP_ENV(df_out.loc[i,'path']) file_remote = configuration_identifier + '_'+run_identifier + df_out.loc[i,'end'] + '_' + str(run_id_num) if not path.isdir(path_remote): palm_mkdir(path_remote, name='REMOTE OUTPUT') LOG.info(' +++ Save OUTPUT Data on this host under:') LOG.info(' '+path.join(path_remote,file_remote)) shutil.copy(df_out.loc[i,'local'], path.join(path_remote, file_remote) ) #COLLECT PARAMETERS FOR SCP TRANSFER scp_entry = {'path_remote': path_remote, 'file_remote': file_remote, 'path_local': path_local, 'file_local': file_local, 'ext': df_out.loc[i,'ext'], 'append': append_option, 'cyclenum':usecycle_option, 'catalog': catalog_option } scp_list.append(scp_entry) else: # UNSET actionout. DUE TO THIS SETTING, FILE WILL LATER JUST #BE COPIED OR APPENDED ON THIS MACHINE if df_out.loc[i,'action'] == 'tr': df_out.loc[i,'action'] = '' else: df_out.loc[i,'action'] = 'a' # APPEND ON THIS MACHINE if df_out.loc[i,'action'] == 'a': if df_out.loc[i,'ext'].strip() != '': destin = df_out.loc[i,'path']+'.'+df_out.loc[i,'ext'] LOG.info('\n>>> OUTPUT: '+ df_out.loc[i,'local'] +' append to') LOG.info(' ' + destin) palm_shell('cat ' + df_out.loc[i,'local'] + ' >> ' + destin) else: LOG.info('\n>>> OUTPUT: '+ df_out.loc[i,'local'] +' append to') LOG.info(' '+ df_out.loc[i,'path'] ) palm_shell('cat ' + df_out.loc[i,'local'] + ' >> ' + df_out.loc[i,'path']) # COPY ON THIS MACHINE # COPY HAS TO BE USED, BECAUSE MOVE DOES NOT WORK IF FILE-ORIGIN AND TARGET ARE # ON DIFFERENT FILE-SYSTEMS if df_out.loc[i,'action'] == '' and files_for_cores == False: # COPY IN CASE OF RUNS ON SINGLE CORES if df_out.loc[i,'ext'].strip() != '': destin = df_out.loc[i,'path']+'.'+df_out.loc[i,'ext'] LOG.info('\n>>> OUTPUT: '+ df_out.loc[i,'local'] +' to') LOG.info(' ' + destin) if link_local_output == True: LOG.info(' file will be linked') palm_shell('ln -f ' + df_out.loc[i,'local'] + ' '+ destin) # If "ln -f" fails of if "$link_local_output = false" do a normal "cp" if not path.isfile(destin): if link_local_output == True: LOG.info('--- WARNING: ln failed, using cp instead (might be time consuming...)') shutil.copy(df_out.loc[i,'local'] , destin) else: LOG.info('\n>>> OUTPUT: '+ df_out.loc[i,'local'] +' to') LOG.info(' '+ df_out.loc[i,'path'] ) if link_local_output == True: LOG.info(' file will be linked') palm_shell('ln -f ' + df_out.loc[i,'local'] + ' '+ df_out.loc[i,'path']) # If "ln -f" fails of if "$link_local_output = false" do a normal "cp" if not path.isfile(df_out.loc[i,'path']): if link_local_output == True: LOG.info('--- WARNING: ln failed, using cp instead (might be time consuming...)') shutil.copy(df_out.loc[i,'local'] , df_out.loc[i,'path']) elif df_out.loc[i,'action'] == '' and files_for_cores == True: # FILES FROM THE DIFFERENT CORES ARE MOVED WITH ln-COMMAND TO THE PERMANENT DIRECTORY # AS A FIRST STEP, THE PERMANENT DIRECTORY IS CREATED LOG.info('\n>>> OUTPUT: '+ df_out.loc[i,'local'] +'/_.... to') LOG.info(' ' + df_out.loc[i,'path']) if link_local_output == True: LOG.info(' files will be linked') palm_mkdir(df_out.loc[i,'path'], name='PERMANENT OUTPUT') try: for file in df_out.loc[i,'local']: os.link(path.join(df_out.loc[i,'local'],file), path.join(df_out.loc[i,'path'], file) ) except: LOG.info('\n--- WARNING: ln failed, using cp instead (might be time consuming...') shutil.rmtree(df_out.loc[i,'path'], ignore_errors=True) shutil.copytree(df_out.loc[i,'local'], df_out.loc[i,'path'], copy_function=shutil.copy) else: shutil.copytree(df_out.loc[i,'local'], df_out.loc[i,'path'], copy_function=shutil.copy) if remote: with open(scp_yamlfile, 'w') as yaml_lun: data = yaml.dump(scp_list, yaml_lun) if len(df_out.index) > 0: calltime = datetime.datetime.now( datetime.timezone.utc ).strftime('%a %b %d %H:%M:%S UTC %Y') LOG.info('-------------------------------------------') if transfer_problems == True: LOG.info('*** OUTPUT-files saved local time: ' + calltime) LOG.info('\n+++ WARNING: some data transfers failed!\n') else: LOG.info('*** all OUTPUT-files saved local time: ' + calltime + '\n')