Newer
Older
2001
2002
2003
2004
2005
2006
2007
2008
2009
2010
2011
2012
2013
2014
2015
2016
2017
2018
2019
2020
2021
2022
2023
2024
2025
2026
2027
2028
2029
2030
2031
2032
2033
2034
2035
2036
2037
2038
2039
2040
2041
2042
2043
2044
2045
2046
2047
2048
2049
2050
2051
2052
2053
2054
2055
2056
2057
2058
2059
2060
2061
2062
2063
2064
# the actual script starts empty
self.pbs = ''
# in case you want to redirect stdout to /dev/nul
# self.wine_appendix = '> /dev/null 2>&1'
self.wine_appendix = ''
# /dev/shm should be the RAM of the cluster
# self.node_run_root = '/dev/shm'
self.node_run_root = '/scratch'
self.cases = cases
# location of the output messages .err and .out created by the node
self.pbs_out_dir = 'pbs_out/'
self.pbs_in_dir = 'pbs_in/'
# for the start number, take hour/minute combo
d = datetime.datetime.today()
tmp = int( str(d.hour)+format(d.minute, '02.0f') )*100
self.pbs_start_number = tmp
self.qsub = qsub
# if quemethod == 'time':
# self.que_jobdeps = False
# elif type(quemethod).__name__ == 'int':
# nr_cpus = quemethod
# self.que_jobdeps = True
# nr_jobs = len(cases)
# jobs_per_cpu = int(math.ceil(float(nr_jobs)/float(nr_cpus)))
# # precalculate all the job ids
# self.jobid_deps = []
# self.jobid_list = []
# count2 = self.pbs_start_number
# for cpu in range(nr_cpus):
# self.jobid_list.extend(range(count2, count2+jobs_per_cpu))
# # the first job to start does not have a dependency
# self.jobid_deps.append(None)
# self.jobid_deps.extend(range(count2, count2+jobs_per_cpu-1))
# count2 += jobs_per_cpu
self.copyback_turb = True
self.copyback_fnames = []
self.copyback_fnames_rename = []
self.copyto_generic = []
self.copyto_fname = []
def create(self):
"""
Main loop for creating the pbs scripts, based on cases, which
contains the case name as key and tag dictionairy as value
"""
# dynamically set walltime based on the number of time steps
# for thyra, make a list so we base the walltime on the slowest case
self.nr_time_steps = []
self.duration = []
self.t0 = []
# '[time_stop]' '[dt_sim]'
# REMARK: this i not realy consistent with how the result and log file
# dirs are allowed to change for each individual case...
# first check if the pbs_out_dir exists, this dir is considered to be
# the same for all cases present in cases
# self.tags['[run_dir]']
case0 = list(self.cases.keys())[0]
2066
2067
2068
2069
2070
2071
2072
2073
2074
2075
2076
2077
2078
2079
2080
2081
2082
2083
2084
2085
2086
2087
2088
2089
2090
2091
2092
path = self.cases[case0]['[run_dir]'] + self.pbs_out_dir
if not os.path.exists(path):
os.makedirs(path)
# create pbs_in base dir
path = self.cases[case0]['[run_dir]'] + self.pbs_in_dir
if not os.path.exists(path):
os.makedirs(path)
# number the pbs jobs:
count2 = self.pbs_start_number
# initial cpu count is zero
count1 = 1
# scan through all the cases
i, i_tot = 1, len(self.cases)
ended = True
for case in self.cases:
# get a shorter version for the current cases tag_dict:
tag_dict = self.cases[case]
# group all values loaded from the tag_dict here, to keep overview
# the directories to SAVE the results/logs/turb files
# load all relevant dir settings: the result/logfile/turbulence/zip
# they are now also available for starting() and ending() parts
hawc2_exe = tag_dict['[hawc2_exe]']
self.case = case.replace('.htc', '')
self.sim_id = tag_dict['[sim_id]']
2095
2096
2097
2098
2099
2100
2101
2102
2103
2104
2105
2106
2107
2108
2109
2110
2111
2112
2113
2114
2115
self.results_dir = tag_dict['[res_dir]']
self.eigenfreq_dir = tag_dict['[eigenfreq_dir]']
self.logs_dir = tag_dict['[log_dir]']
self.animation_dir = tag_dict['[animation_dir]']
self.TurbDirName = tag_dict['[turb_dir]']
self.TurbDb = tag_dict['[turb_db_dir]']
self.wakeDb = tag_dict['[wake_db_dir]']
self.meandDb = tag_dict['[meand_db_dir]']
self.WakeDirName = tag_dict['[wake_dir]']
self.MeanderDirName = tag_dict['[meander_dir]']
self.ModelZipFile = tag_dict['[model_zip]']
self.htc_dir = tag_dict['[htc_dir]']
self.hydro_dir = tag_dict['[hydro_dir]']
self.mooring_dir = tag_dict['[mooring_dir]']
self.model_path = tag_dict['[run_dir]']
self.turb_base_name = tag_dict['[turb_base_name]']
self.wake_base_name = tag_dict['[wake_base_name]']
self.meand_base_name = tag_dict['[meand_base_name]']
self.pbs_queue_command = tag_dict['[pbs_queue_command]']
self.walltime = tag_dict['[walltime]']
self.dyn_walltime = tag_dict['[auto_walltime]']
self.case_duration = tag_dict['[duration]']
2117
2118
2119
2120
2121
2122
2123
2124
2125
2126
2127
2128
2129
2130
2131
2132
2133
2134
2135
2136
2137
2138
2139
2140
2141
2142
2143
2144
2145
2146
2147
# create the pbs_out_dir if necesary
try:
path = tag_dict['[run_dir]'] + tag_dict['[pbs_out_dir]']
if not os.path.exists(path):
os.makedirs(path)
self.pbs_out_dir = tag_dict['[pbs_out_dir]']
except:
pass
# create pbs_in subdirectories if necessary
try:
path = tag_dict['[run_dir]'] + tag_dict['[pbs_in_dir]']
if not os.path.exists(path):
os.makedirs(path)
self.pbs_in_dir = tag_dict['[pbs_in_dir]']
except:
pass
try:
self.copyback_files = tag_dict['[copyback_files]']
self.copyback_frename = tag_dict['[copyback_frename]']
except KeyError:
pass
try:
self.copyto_generic = tag_dict['[copyto_generic]']
self.copyto_files = tag_dict['[copyto_files]']
except KeyError:
pass
# one using just one file so it can be used together with the
# DLC spreadsheets
try:
self.copyback_files = [tag_dict['[copyback_f1]']]
self.copyback_frename = [tag_dict['[copyback_f1_rename]']]
except KeyError:
pass
try:
self.copyto_generic = [tag_dict['[copyto_generic_f1]']]
self.copyto_files = [tag_dict['[copyto_f1]']]
except KeyError:
pass
2161
2162
2163
2164
2165
2166
2167
2168
2169
2170
2171
2172
2173
2174
2175
2176
2177
2178
2179
2180
2181
2182
2183
2184
2185
2186
2187
2188
2189
2190
2191
2192
2193
2194
2195
2196
2197
2198
2199
2200
2201
# related to the dynamically setting the walltime
duration = float(tag_dict['[time_stop]'])
dt = float(tag_dict['[dt_sim]'])
self.nr_time_steps.append(duration/dt)
self.duration.append(float(tag_dict['[duration]']))
self.t0.append(float(tag_dict['[t0]']))
if self.verbose:
print('htc_dir in pbs.create:')
print(self.htc_dir)
print(self.model_path)
# we only start a new case, if we have something that ended before
# the very first case has to start with starting
if ended:
count1 = 1
# # when jobs depend on other jobs (constant node loading)
# if self.que_jobdeps:
# jobid = self.pref + str(self.jobid_list[i-1])
# jobid_dep = self.pref + str(self.jobid_deps[i-1])
# else:
# jobid = self.pref + str(count2)
# jobid_dep = None
if self.short_job_names:
jobid = self.pref + str(count2)
else:
jobid = tag_dict['[case_id]']
if self.pbs_fname_appendix and self.short_job_names:
# define the path for the new pbs script
pbs_in_fname = '%s_%s.p' % (tag_dict['[case_id]'], jobid)
else:
pbs_in_fname = '%s.p' % (tag_dict['[case_id]'])
pbs_path = self.model_path + self.pbs_in_dir + pbs_in_fname
# Start a new pbs script, we only need the tag_dict here
self.starting(tag_dict, jobid)
ended = False
# -----------------------------------------------------------------
# WRITING THE ACTUAL JOB PARAMETERS
# browse to the current scratch directory
self.pbs += "\n\n"
# mark start of single PBS mode
self.pbs += '# ' + '='*78 + '\n'
# evaluates to true if LAUNCH_PBS_MODE is NOT set
self.pbs += '# single PBS mode: one case per PBS job\n'
self.pbs += '# evaluates to true if LAUNCH_PBS_MODE is NOT set\n'
self.pbs += "if [ -z ${LAUNCH_PBS_MODE+x} ] ; then\n"
self.pbs += " echo\n"
self.pbs += " echo 'Execute commands on scratch nodes'\n"
self.pbs += " cd %s/$USER/$PBS_JOBID\n" % self.node_run_root
self.pbs += " # create unique dir for each CPU\n"
self.pbs += ' mkdir "%i"; cd "%i"\n' % (count1, count1)
# output the current scratch directory
self.pbs += " pwd\n"
# zip file has been copied to the node before (in start_pbs())
# unzip now in the CPU scratch directory (zip file is one level up)
self.pbs += " /usr/bin/unzip ../" + self.ModelZipFile + '\n'
# create all directories, especially relevant if there are case
# dependent sub directories that are not present in the ZIP file
self.pbs += " mkdir -p " + self.htc_dir + '\n'
self.pbs += " mkdir -p " + self.results_dir + '\n'
self.pbs += " mkdir -p " + self.logs_dir + '\n'
if self.TurbDirName is not None or self.TurbDirName != 'None':
self.pbs += " mkdir -p " + self.TurbDirName + '\n'
if self.WakeDirName and self.WakeDirName != self.TurbDirName:
self.pbs += " mkdir -p " + self.WakeDirName + '\n'
if self.MeanderDirName and self.MeanderDirName != self.TurbDirName:
self.pbs += " mkdir -p " + self.MeanderDirName + '\n'
self.pbs += " mkdir -p " + self.hydro_dir + '\n'
# create the eigen analysis dir just in case that is necessary
if self.eigenfreq_dir:
self.pbs += ' mkdir -p %s\n' % self.eigenfreq_dir
# and copy the htc file to the node
self.pbs += " cp -R $PBS_O_WORKDIR/" + self.htc_dir \
+ case +" ./" + self.htc_dir + '\n'
# if there is a turbulence file data base dir, copy from there
if self.TurbDb:
turb_dir_src = os.path.join('$PBS_O_WORKDIR', self.TurbDb)
turb_dir_src = os.path.join('$PBS_O_WORKDIR', self.TurbDirName)
# the original behaviour makes assumptions on the turbulence box
# names: turb_base_name_xxx_u.bin, turb_base_name_xxx_v.bin
if self.turb_base_name is not None:
turb_src = os.path.join(turb_dir_src, self.turb_base_name)
self.pbs += " cp -R %s*.bin %s\n" % (turb_src, self.TurbDirName)
# more generally, literally define the names of the boxes for u,v,w
# components
elif '[turb_fname_u]' in tag_dict:
turb_u = os.path.join(turb_dir_src, tag_dict['[turb_fname_u]'])
turb_v = os.path.join(turb_dir_src, tag_dict['[turb_fname_v]'])
turb_w = os.path.join(turb_dir_src, tag_dict['[turb_fname_w]'])
self.pbs += " cp %s %s\n" % (turb_u, self.TurbDirName)
self.pbs += " cp %s %s\n" % (turb_v, self.TurbDirName)
self.pbs += " cp %s %s\n" % (turb_w, self.TurbDirName)
# if there is a turbulence file data base dir, copy from there
if self.wakeDb and self.WakeDirName:
wake_dir_src = os.path.join('$PBS_O_WORKDIR', self.wakeDb)
wake_dir_src = os.path.join('$PBS_O_WORKDIR', self.WakeDirName)
if self.wake_base_name is not None:
wake_src = os.path.join(wake_dir_src, self.wake_base_name)
self.pbs += " cp -R %s*.bin %s\n" % (wake_src, self.WakeDirName)
# if there is a turbulence file data base dir, copy from there
if self.meandDb and self.MeanderDirName:
meand_dir_src = os.path.join('$PBS_O_WORKDIR', self.meandDb)
meand_dir_src = os.path.join('$PBS_O_WORKDIR', self.MeanderDirName)
if self.meand_base_name is not None:
meand_src = os.path.join(meand_dir_src, self.meand_base_name)
self.pbs += " cp -R %s*.bin %s\n" % (meand_src, self.MeanderDirName)
# copy and rename input files with given versioned name to the
# required non unique generic version
for fname, fgen in zip(self.copyto_files, self.copyto_generic):
self.pbs += " cp -R $PBS_O_WORKDIR/%s ./%s\n" % (fname, fgen)
# only apply the wine fix in PBS mode
self.pbs += self.winefix
# TODO: activate python env, calculate post-processing
# self.pbs += 'echo `python -c "import wetb; print(wetb.__version__)"`\n'
# mark end of single PBS mode
self.pbs += '# ' + '='*78 + '\n\n'
# end of the file copying in PBS mode
# mark start of find+xargs mode
self.pbs += '# ' + '-'*78 + '\n'
self.pbs += '# find+xargs mode: 1 PBS job, multiple cases\n'
self.pbs += "else\n"
# when in find+xargs mode, browse to the relevant CPU
self.pbs += ' # with find+xargs we first browse to CPU folder\n'
self.pbs += ' cd "$CPU_NR"\n'
self.pbs += "fi\n"
# mark end of find+xargs mode
self.pbs += '# ' + '-'*78 + '\n\n'
self.pbs += 'echo ""\n'
# mark start of single PBS mode
self.pbs += '# ' + '='*78 + '\n'
self.pbs += '# single PBS mode: one case per PBS job\n'
self.pbs += '# evaluates to true if LAUNCH_PBS_MODE is NOT set\n'
self.pbs += "if [ -z ${LAUNCH_PBS_MODE+x} ] ; then\n"
# the hawc2 execution commands via wine, in PBS mode fork and wait
param = (self.wine, hawc2_exe, self.htc_dir+case, self.wine_appendix)
self.pbs += ' echo "execute HAWC2, fork to background"\n'
self.pbs += " %s %s ./%s %s &\n" % param
# FIXME: running post-processing will only work when 1 HAWC2 job
# per PBS file, otherwise you have to wait for each job to finish
# first and then run the post-processing for all those cases
if self.maxcpu == 1:
self.pbs += ' wait\n'
if self.pyenv is not None:
self.pbs += ' echo "POST-PROCESSING"\n'
self.pbs += ' %s %s\n' % (self.pyenv_cmd, self.pyenv)
self.pbs += " "
self.checklogs()
self.pbs += " "
self.postprocessing()
self.pbs += ' source deactivate\n'
# mark end of single PBS mode
self.pbs += '# ' + '='*78 + '\n\n'
# mark start of find+xargs mode
self.pbs += '# ' + '-'*78 + '\n'
self.pbs += '# find+xargs mode: 1 PBS job, multiple cases\n'
self.pbs += "else\n"
# numactl --physcpubind=$CPU_NR
param = (self.winenumactl, hawc2_exe, self.htc_dir+case,
self.wine_appendix)
self.pbs += ' echo "execute HAWC2, do not fork and wait"\n'
self.pbs += " %s %s ./%s %s\n" % param
self.pbs += ' echo "POST-PROCESSING"\n'
self.pbs += " "
self.checklogs()
self.pbs += " "
self.postprocessing()
self.pbs += "fi\n"
# mark end of find+xargs mode
self.pbs += '# ' + '-'*78 + '\n'
#self.pbs += "wine get_mac_adresses" + '\n'
# self.pbs += "cp -R ./*.mac $PBS_O_WORKDIR/." + '\n'
# -----------------------------------------------------------------
# and we end when the cpu's per node are full
if int(count1/self.maxcpu) == 1:
# write the end part of the pbs script
self.ending(pbs_path)
ended = True
# print progress:
replace = ((i/self.maxcpu), (i_tot/self.maxcpu), self.walltime)
if not self.silent:
print('pbs script %3i/%i walltime=%s' % replace)
count2 += 1
i += 1
# the next cpu
count1 += 1
# it could be that the last node was not fully loaded. In that case
# we do not have had a succesfull ending, and we still need to finish
if not ended:
# write the end part of the pbs script
self.ending(pbs_path)
# progress printing
replace = ( (i/self.maxcpu), (i_tot/self.maxcpu), self.walltime )
if not self.silent:
print('pbs script %3i/%i walltime=%s, partially loaded' % replace)
# print 'pbs progress, script '+format(i/self.maxcpu,'2.0f')\
# + '/' + format(i_tot/self.maxcpu, '2.0f') \
# + ' partially loaded...'
def starting(self, tag_dict, jobid):
"""
First part of the pbs script
"""
# a new clean pbs script!
self.pbs = ''
self.pbs += "### Standard Output" + '\n'
case_id = tag_dict['[case_id]']
# PBS job name
self.pbs += "#PBS -N %s\n" % (jobid)
self.pbs += "#PBS -o ./" + self.pbs_out_dir + case_id + ".out" + '\n'
# self.pbs += "#PBS -o ./pbs_out/" + jobid + ".out" + '\n'
self.pbs += "### Standard Error" + '\n'
self.pbs += "#PBS -e ./" + self.pbs_out_dir + case_id + ".err" + '\n'
# self.pbs += "#PBS -e ./pbs_out/" + jobid + ".err" + '\n'
self.pbs += '#PBS -W umask=0003\n'
2399
2400
2401
2402
2403
2404
2405
2406
2407
2408
2409
2410
2411
2412
2413
2414
2415
2416
2417
2418
2419
2420
2421
2422
2423
2424
2425
2426
2427
2428
2429
2430
2431
2432
2433
self.pbs += "### Maximum wallclock time format HOURS:MINUTES:SECONDS\n"
# self.pbs += "#PBS -l walltime=" + self.walltime + '\n'
self.pbs += "#PBS -l walltime=[walltime]\n"
if self.qsub == 'time':
self.pbs += "#PBS -a [start_time]" + '\n'
elif self.qsub == 'depend':
# set job dependencies, job_id refers to PBS job_id, which is only
# assigned to a job at the moment it gets qsubbed into the que
self.pbs += "[nodeps]PBS -W depend=afterany:[job_id]\n"
# if self.que_jobdeps:
# self.pbs += "#PBS -W depend=afterany:%s\n" % jobid_dep
# else:
# self.pbs += "#PBS -a [start_time]" + '\n'
# in case of gorm, we need to make it work correctly. Now each job
# has a different scratch dir. If we set maxcpu to 12 they all have
# the same scratch dir. In that case there should be done something
# differently
# specify the number of nodes and cpu's per node required
if self.maxcpu > 1:
# Number of nodes and cpus per node (ppn)
lnodes = int(math.ceil(len(self.cases)/float(self.maxcpu)))
lnodes = 1
self.pbs += "#PBS -l nodes=%i:ppn=%i\n" % (lnodes, self.maxcpu)
else:
self.pbs += "#PBS -l nodes=1:ppn=1\n"
# Number of nodes and cpus per node (ppn)
self.pbs += "### Queue name" + '\n'
# queue names for Thyra are as follows:
# short walltime queue (shorter than an hour): '#PBS -q xpresq'
# or otherwise for longer jobs: '#PBS -q workq'
self.pbs += self.pbs_queue_command + '\n'
# mark start of single PBS mode
self.pbs += '\n' + '# ' + '='*78 + '\n'
# ignore all the file copying when running in xargs mode:
# when varibale LAUNCH_PBS_MODE is set, file copying is ignored
# and will have to be done elsewhere
# we do this so the same launch script can be used either with the node
# scheduler and the PBS system (for example when re-running cases)
# evaluates to true if LAUNCH_PBS_MODE is NOT set
self.pbs += '# single PBS mode: one case per PBS job\n'
self.pbs += '# evaluates to true if LAUNCH_PBS_MODE is NOT set\n'
self.pbs += "if [ -z ${LAUNCH_PBS_MODE+x} ] ; then\n"
self.pbs += " ### Create scratch directory and copy data to it\n"
# output the current directory
self.pbs += " cd $PBS_O_WORKDIR" + '\n'
self.pbs += ' echo "current working dir (pwd):"\n'
self.pbs += " pwd\n"
# The batch system on Gorm allows more than one job per node.
# Because of this the scratch directory name includes both the
# user name and the job ID, that is /scratch/$USER/$PBS_JOBID
# if not scratch, make the dir
if self.node_run_root != '/scratch':
self.pbs += ' mkdir -p %s/$USER\n' % self.node_run_root
self.pbs += ' mkdir -p %s/$USER/$PBS_JOBID\n' % self.node_run_root
# copy the zip files to the scratch dir on the node
self.pbs += " cp -R ./" + self.ModelZipFile + \
' %s/$USER/$PBS_JOBID\n' % (self.node_run_root)
self.pbs += "fi\n"
# mark end of single PBS mode
self.pbs += '# ' + '='*78 + '\n'
def ending(self, pbs_path):
"""
Last part of the pbs script, including command to write script to disc
COPY BACK: from node to
"""
self.pbs += "\n\n"
self.pbs += "### Epilogue\n"
# mark start of single PBS mode
self.pbs += '# ' + "="*78 + "\n"
# evaluates to true if LAUNCH_PBS_MODE is NOT set
self.pbs += '# single PBS mode: one case per PBS job\n'
self.pbs += '# evaluates to true if LAUNCH_PBS_MODE is NOT set\n'
self.pbs += "if [ -z ${LAUNCH_PBS_MODE+x} ] ; then\n"
self.pbs += " ### wait for jobs to finish\n"
self.pbs += " wait\n"
self.pbs += ' echo ""\n'
self.pbs += ' echo "Copy back from scratch directory"\n'
for i in range(1, self.maxcpu+1, 1):
# navigate to the cpu dir on the node
# The batch system on Gorm allows more than one job per node.
# Because of this the scratch directory name includes both the
# user name and the job ID, that is /scratch/$USER/$PBS_JOBID
self.copyback_all_files("pbs_mode", i)
# find+xargs mode only makes sense when maxcpu==1, cpu handling
# for this mode is handled elsewhere
if self.maxcpu == 1:
# mark start of find+xargs mode
self.pbs += '# ' + "-"*78 + "\n"
self.pbs += '# find+xargs mode: 1 PBS job, multiple cases\n'
self.pbs += 'else\n'
self.copyback_all_files("find+xargs", None)
# mark end of find+xargs mode
# self.pbs += '# ' + "-"*78 + "\n"
# # and delete it all (but that is not allowed)
# self.pbs += 'cd ..\n'
# self.pbs += 'ls -lah\n'
# self.pbs += 'echo $PBS_JOBID\n'
# self.pbs += 'rm -r $PBS_JOBID\n'
# Delete the batch file at the end. However, is this possible since
# the batch file is still open at this point????
# self.pbs += "rm "
self.pbs += 'fi\n'
2513
2514
2515
2516
2517
2518
2519
2520
2521
2522
2523
2524
2525
2526
2527
2528
2529
2530
2531
2532
2533
2534
2535
2536
2537
2538
2539
2540
2541
2542
2543
2544
2545
2546
2547
# base walltime on the longest simulation in the batch
nr_time_steps = max(self.nr_time_steps)
# TODO: take into acccount the difference between time steps with
# and without output. This penelaty also depends on the number of
# channels outputted. So from 0 until t0 we have no penalty,
# from t0 until t0+duration we have the output penalty.
# always a predifined lead time to account for startup losses
tmax = int(nr_time_steps*self.secperiter*self.iterperstep + self.tlead)
if self.dyn_walltime:
dt_seconds = datetime.datetime.fromtimestamp(tmax)
self.walltime = dt_seconds.strftime('%H:%M:%S')
self.pbs = self.pbs.replace('[walltime]', self.walltime)
else:
self.pbs = self.pbs.replace('[walltime]', self.walltime)
# and reset the nr_time_steps list for the next pbs job file
self.nr_time_steps = []
self.t0 = []
self.duration = []
# TODO: add logfile checking support directly here. In that way each
# node will do the logfile checking and statistics calculations right
# after the simulation. Figure out a way how to merge the data from
# all the different cases afterwards
self.pbs += "exit\n"
if self.verbose:
print('writing pbs script to path: ' + pbs_path)
# and write the script to a file:
write_file(pbs_path, self.pbs, 'w')
# make the string empty again, for memory
self.pbs = ''
2548
2549
2550
2551
2552
2553
2554
2555
2556
2557
2558
2559
2560
2561
2562
2563
2564
2565
2566
2567
2568
2569
2570
2571
2572
2573
2574
2575
2576
2577
2578
2579
2580
2581
2582
2583
2584
2585
2586
2587
2588
2589
2590
2591
2592
2593
2594
2595
2596
2597
2598
2599
2600
2601
2602
2603
2604
2605
2606
2607
def copyback_all_files(self, mode, cpu_nr):
"""Copy back all the files from either scratch to run_dir (PBS mode),
or from CPU sub-directory back to main directory in find+xargs mode.
"""
if mode=="find+xargs":
foper = "rsync -a --remove-source-files" # move files instead of copy
dst = os.path.join('..', self.sim_id, '')
dst_db = '../'
cd2model = " cd %s\n" % os.path.join(self.node_run_root, '$USER',
'$PBS_JOBID', '$CPU_NR', '')
pbs_mode = False
else:
foper = "cp -R"
dst = "$PBS_O_WORKDIR/"
dst_db = dst
pbs_mode = True
cd2model = " cd %s\n" % os.path.join(self.node_run_root, '$USER',
'$PBS_JOBID', '%i' % cpu_nr, '')
# navigate to the cpu dir on the node
# The batch system on Gorm/Jess allows more than one job per node.
# Because of this the scratch directory name includes both the
# user name and the job ID, that is /scratch/$USER/$PBS_JOBID/CPU_NR
self.pbs += cd2model
# create the log, res etc dirs in case they do not exist. Only relevant
# for pbs_mode, they are created in advance in find+xargs
if pbs_mode:
mk = ' mkdir -p'
self.pbs += "%s %s\n" % (mk, os.path.join(dst, self.results_dir))
self.pbs += "%s %s\n" % (mk, os.path.join(dst, self.logs_dir))
if self.animation_dir:
self.pbs += "%s %s\n" % (mk, os.path.join(dst, self.animation_dir))
if self.copyback_turb and self.TurbDb:
self.pbs += "%s %s\n" % (mk, os.path.join(dst, self.TurbDb))
elif self.copyback_turb:
self.pbs += "%s %s\n" % (mk, os.path.join(dst, self.TurbDirName))
if self.copyback_turb and self.wakeDb:
self.pbs += "%s %s\n" % (mk, os.path.join(dst, self.wakeDb))
elif self.WakeDirName and self.WakeDirName != self.TurbDirName:
self.pbs += "%s %s\n" % (mk, os.path.join(dst, self.WakeDirName))
if self.copyback_turb and self.meandDb:
self.pbs += "%s %s\n" % (mk, os.path.join(dst, self.meandDb))
elif self.MeanderDirName and self.MeanderDirName != self.TurbDirName:
self.pbs += "%s %s\n" % (mk, os.path.join(dst, self.MeanderDirName))
# and copy the results and log files frome the scratch to dst
res_dst = os.path.join(dst, self.results_dir, ".")
self.pbs += " %s %s. %s\n" % (foper, self.results_dir, res_dst)
log_dst = os.path.join(dst, self.logs_dir, ".")
self.pbs += " %s %s. %s\n" % (foper, self.logs_dir, log_dst)
if self.animation_dir:
ani_dst = os.path.join(dst, self.animation_dir, ".")
self.pbs += " %s %s. %s\n" % (foper, self.animation_dir, ani_dst)
if self.eigenfreq_dir:
# just in case the eig dir has subdirs for the results, only
# select the base path and cp -r will take care of the rest
p1 = self.eigenfreq_dir.split('/')[0]
p2 = os.path.join(dst, p1, ".")
self.pbs += " cp -R %s/. %s\n" % (p1, p2)
# for eigen analysis with floater, modes are in root
eig_dir_sys = os.path.join(dst, self.eigenfreq_dir, 'system/', '.')
self.pbs += ' mkdir -p %s\n' % eig_dir_sys
self.pbs += " cp -R mode* %s\n" % eig_dir_sys
self.pbs += " %s mode* %s\n" % (foper, eig_dir_sys)
2614
2615
2616
2617
2618
2619
2620
2621
2622
2623
2624
2625
2626
2627
2628
2629
2630
2631
2632
2633
2634
2635
2636
2637
2638
2639
2640
2641
2642
2643
2644
2645
2646
2647
2648
2649
2650
2651
2652
2653
2654
2655
2656
2657
2658
2659
2660
2661
2662
2663
2664
2665
2666
2667
2668
2669
2670
2671
# only copy the turbulence files back if they do not exist
# for all *.bin files on the node
cmd = ' for i in `ls *.bin`; do if [ -e %s$i ]; '
cmd += 'then echo "$i exists no copyback"; else echo "$i copyback"; '
cmd += 'cp $i %s; fi; done\n'
# copy back turbulence file?
# browse to the node turb dir
self.pbs += '\n echo ""\n'
self.pbs += ' echo "COPY BACK TURB IF APPLICABLE"\n'
if self.copyback_turb and self.TurbDb:
self.pbs += ' cd %s\n' % self.TurbDirName
tmp = (os.path.join(dst_db, self.TurbDb, ''),)*2
self.pbs += cmd % tmp
# and back to normal model root
self.pbs += cd2model
elif self.copyback_turb:
self.pbs += ' cd %s\n' % self.TurbDirName
tmp = (os.path.join(dst, self.TurbDirName, ''),)*2
self.pbs += cmd % tmp
# and back to normal model root
self.pbs += cd2model
if self.copyback_turb and self.wakeDb:
self.pbs += ' cd %s\n' % self.WakeDirName
tmp = (os.path.join(dst_db, self.wakeDb, ''),)*2
self.pbs += cmd % tmp
# and back to normal model root
self.pbs += cd2model
elif self.copyback_turb and self.WakeDirName:
self.pbs += ' cd %s\n' % self.WakeDirName
tmp = (os.path.join(dst, self.WakeDirName, ''),)*2
self.pbs += cmd % tmp
# and back to normal model root
self.pbs += cd2model
if self.copyback_turb and self.meandDb:
self.pbs += ' cd %s\n' % self.MeanderDirName
tmp = (os.path.join(dst_db, self.meandDb, ''),)*2
self.pbs += cmd % tmp
# and back to normal model root
self.pbs += cd2model
elif self.copyback_turb and self.MeanderDirName:
self.pbs += ' cd %s\n' % self.MeanderDirName
tmp = (os.path.join(dst, self.MeanderDirName, ''),)*2
self.pbs += cmd % tmp
# and back to normal model root
self.pbs += cd2model
self.pbs += ' echo "END COPY BACK TURB"\n'
self.pbs += ' echo ""\n\n'
# copy back any other kind of files, as specified in copyback_files
self.pbs += ' echo "COPYBACK [copyback_files]/[copyback_frename]"\n'
if len(self.copyback_frename) == 0:
self.copyback_frename = self.copyback_files
for fname, fnew in zip(self.copyback_files, self.copyback_frename):
dst_fnew = os.path.join(dst, fnew)
self.pbs += " %s %s %s\n" % (foper, fname, dst_fnew)
self.pbs += ' echo "END COPYBACK"\n'
self.pbs += ' echo ""\n'
if pbs_mode:
# check what is left
self.pbs += ' echo ""\n'
self.pbs += ' echo "following files are on '
self.pbs += 'node/cpu %i (find .):"\n' % cpu_nr
self.pbs += ' find .\n'
self.pbs += '# ' + '='*78 + '\n'
else:
self.pbs += '# ' + '-'*78 + '\n'
def checklogs(self):
"""
"""
self.pbs += 'python -c "from wetb.prepost import statsdel; '
rpl = (os.path.join(self.logs_dir, self.case+'.log'))
self.pbs += 'statsdel.logcheck(\'%s\')"\n' % rpl
def postprocessing(self):
"""Run post-processing just after HAWC2 has ran
"""
self.pbs += 'python -c "from wetb.prepost import statsdel; '
fsrc = os.path.join(self.results_dir, self.case)
rpl = (fsrc, str(self.case_duration), '.csv')
self.pbs += ('statsdel.calc(\'%s\', no_bins=46, m=[3, 4, 6, 8, 10, 12], '
'neq=%s, i0=0, i1=None, ftype=\'%s\')"\n' % rpl)
def check_results(self, cases):
"""
Cross-check if all simulations on the list have returned a simulation.
Combine with ErrorLogs to identify which errors occur where.
"""
cases_fail = {}
if not self.silent:
print('checking if all log and result files are present...', end='')
# check for each case if we have results and a log file
for cname, case in cases.items():
run_dir = case['[run_dir]']
res_dir = case['[res_dir]']
log_dir = case['[log_dir]']

David Verelst
committed
# FIXME: HAWC2 outputs result and logfile always in lower cases
cname_ = cname.replace('.htc', '').lower()
f_log = os.path.join(run_dir, log_dir, cname_)
f_res = os.path.join(run_dir, res_dir, cname_)
if not os.path.exists(f_log + '.log'):
cases_fail[cname] = copy.copy(cases[cname])
continue
try:
size_sel = os.stat(f_res + '.sel').st_size
size_dat = os.stat(f_res + '.dat').st_size
except OSError:
size_sel = 0
size_dat = 0
if size_sel < 5 or size_dat < 5:
cases_fail[cname] = copy.copy(cases[cname])
if not self.silent:
print('done!')
# length will be zero if there are no failures
return cases_fail
# TODO: rewrite the error log analysis to something better. Take different
# approach: start from the case and see if the results are present. Than we
# also have the tags_dict available when log-checking a certain case
class ErrorLogs(windIO.LogFile):
2745
2746
2747
2748
2749
2750
2751
2752
2753
2754
2755
2756
2757
2758
2759
2760
2761
2762
2763
2764
2765
2766
2767
2768
2769
2770
2771
2772
2773
2774
2775
2776
2777
2778
"""
Analyse all HAWC2 log files in any given directory
==================================================
Usage:
logs = ErrorLogs()
logs.MsgList : list with the to be checked messages. Add more if required
logs.ResultFile : name of the result file (default is ErrorLog.csv)
logs.PathToLogs : specify the directory where the logsfile reside,
the ResultFile will be saved in the same directory.
It is also possible to give the path of a specific
file, the logfile will not be saved in this case. Save
when all required messages are analysed with save()
logs.check() to analyse all the logfiles and create the ResultFile
logs.save() to save after single file analysis
logs.MsgListLog : [ [case, line nr, error1, line nr, error2, ....], [], ...]
holding the error messages, empty if no err msg found
will survive as long as the logs object exists. Keep in
mind that when processing many messages with many error types (as defined)
in MsgList might lead to an increase in memory usage.
logs.MsgListLog2 : dict(key=case, value=[found_error, exit_correct]
where found_error and exit_correct are booleans. Found error will just
indicate whether or not any error message has been found
All files in the speficied folder (PathToLogs) will be evaluated.
When Any item present in MsgList occurs, the line number of the first
occurance will be displayed in the ResultFile.
If more messages are required, add them to the MsgList
"""
# TODO: move to the HAWC2 plugin for cases
def __init__(self, silent=False, cases=None, resultfile='ErrorLog.csv'):
# call init from base class
super(ErrorLogs, self).__init__()
self.ResultFile = resultfile
# TODO: save this not a csv text string but a df_dict, and save as excel
# and DataFrame!
def check(self, appendlog=False, save_iter=False):
"""Check all log files that are to be found in the directory
ErrorLogs.PathToLogs, or check the specific log file if
ErrorLogs.PathToLogs points to a specific log file.
"""
# MsgListLog = []
FileList = []
# if a directory, load all files first
if os.path.isdir(self.PathToLogs):
for files in os.walk(self.PathToLogs):
FileList.append(files)
NrFiles = len(FileList[0][2])
else:
# simulate one entry on FileList[0][2], give it the file name
# and save the directory on in self.PathToLogs
NrFiles = 1
FileList.append([ [],[],[os.path.basename(self.PathToLogs)] ])
self.PathToLogs = os.path.dirname(self.PathToLogs)
single_file = True
i=1
# walk trough the files present in the folder path
for fname in FileList[0][2]:
# progress indicator
if NrFiles > 1:
if not self.silent:
print('progress: ' + str(i) + '/' + str(NrFiles))
# open the current log file

David Verelst
committed
f_log = os.path.join(self.PathToLogs, fname)
if self.cases is not None:
case = self.cases[fname.replace('.log', '.htc')]
case = None
self.readlog(f_log, case=case, save_iter=save_iter)
i += 1
# # if no messages are found for the current file, than say so:
# if len(MsgList2) == len(self.MsgList):
# tempLog[-1] = 'NO MESSAGES FOUND'
# if we have only one file, don't save the log file to disk. It is
# expected that if we analyse many different single files, this will
# cause a slower script
if single_file:
# now we make it available over the object to save and let it grow
# over many analysis
# self.MsgListLog = copy.copy(MsgListLog)
pass
else:
self.save(appendlog=appendlog)
def save(self, appendlog=False, suffix=None):
contents = self._header()
contents = self._msglistlog2csv(contents)
# write csv file to disk, append to facilitate more logfile analysis
if isinstance(suffix, str):
tmp = self.ResultFile.replace('.csv', '_%s.csv' % suffix)
fname = os.path.join(self.PathToLogs, tmp)
else:
fname = os.path.join(self.PathToLogs, str(self.ResultFile))
if not self.silent:
print('Error log analysis saved at:')
print(fname)
if appendlog:
mode = 'a'
else:
mode = 'w'
with open(fname, mode) as f:
f.write(contents)
"""
Second generation ModelData function. The HawcPy version is crappy, buggy
and not mutch of use in the optimisation context.
"""
class st_headers(object):
2874
2875
2876
2877
2878
2879
2880
2881
2882
2883
2884
2885
2886
2887
2888
2889
2890
2891
2892
2893
2894
2895
2896
2897
2898
2899
2900
2901
2902
2903
2904
2905
2906
2907
2908
2909
2910
2911
2912
2913
"""
Indices to the respective parameters in the HAWC2 st data file
"""
r = 0
m = 1
x_cg = 2
y_cg = 3
ri_x = 4
ri_y = 5
x_sh = 6
y_sh = 7
E = 8
G = 9
Ixx = 10
Iyy = 11
I_p = 12
k_x = 13
k_y = 14
A = 15
pitch = 16
x_e = 17
y_e = 18
def __init__(self, verbose=False, silent=False):
self.verbose = verbose
self.silent = silent
# define the column width for printing
self.col_width = 13
# formatting and precision
self.float_hi = 9999.9999
self.float_lo = 0.01
self.prec_float = ' 9.05f'
self.prec_exp = ' 8.04e'
self.prec_loss = 0.01
#0 1 2 3 4 5 6 7 8 9 10 11
#r m x_cg y_cg ri_x ri_y x_sh y_sh E G I_x I_y
#12 13 14 15 16 17 18
#I_p/K k_x k_y A pitch x_e y_e
# 19 cols
self.st_column_header_list = ['r', 'm', 'x_cg', 'y_cg', 'ri_x',
'ri_y', 'x_sh', 'y_sh', 'E', 'G', 'I_x', 'I_y', 'J', 'k_x',
'k_y', 'A', 'pitch', 'x_e', 'y_e']
self.st_column_header_list_latex = ['r','m','x_{cg}','y_{cg}','ri_x',
'ri_y', 'x_{sh}','y_{sh}','E', 'G', 'I_x', 'I_y', 'J', 'k_x',
'k_y', 'A', 'pitch', 'x_e', 'y_e']
self.st_fpm_cols = ['r', 'm', 'x_cg', 'y_cg', 'ri_x', 'ri_y', 'pitch',
'x_e', 'y_e', 'E11', 'E12', 'E13', 'E14', 'E15',
'E16', 'E22', 'E23', 'E24', 'E25', 'E26', 'E33',
'E34', 'E35', 'E36', 'E44', 'E45', 'E46', 'E55',
'E56', 'E66']
# set column names/indeices as class attributes
for i, col in enumerate(self.st_fpm_cols):
setattr(self, col, i)
2931
2932
2933
2934
2935
2936
2937
2938
2939
2940
2941
2942
2943
2944
2945
2946
2947
2948
2949
2950
2951
2952
2953
2954
2955
2956
2957
2958
2959
2960
2961
2962
2963
2964
2965
2966
2967
2968
2969
2970
2971
2972
2973
2974
2975
2976
2977
2978
2979
2980
2981
2982
2983
2984
2985
2986
2987
2988
2989
2990
2991
2992
2993
2994
2995
# make the column header
self.column_header_line = 19 * self.col_width * '=' + '\n'
for k in self.st_column_header_list:
self.column_header_line += k.rjust(self.col_width)
self.column_header_line += '\n' + (19 * self.col_width * '=') + '\n'
def fromline(self, line, separator=' '):
# TODO: move this to the global function space (dav-general-module)
"""
split a line, but ignore any blank spaces and return a list with only
the values, not empty places
"""
# remove all tabs, new lines, etc? (\t, \r, \n)
line = line.replace('\t',' ').replace('\n','').replace('\r','')
# trailing and ending spaces
line = line.strip()
line = line.split(separator)
values = []
for k in range(len(line)):
if len(line[k]) > 0: #and k == item_nr:
values.append(line[k])
# break
return values
def load_st(self, file_path, file_name):
"""
Now a better format: st_dict has following key/value pairs
'nset' : total number of sets in the file (int).
This should be autocalculated every time when writing
a new file.
'007-000-0' : set number line in one peace
'007-001-a' : comments for set-subset nr 07-01 (str)
'007-001-b' : subset nr and number of data points, should be
autocalculate every time you generate a file
'007-001-d' : data for set-subset nr 07-01 (ndarray(n,19))
NOW WE ONLY CONSIDER SUBSET COMMENTS, SET COMMENTS, HOW ARE THEY
TREADED NOW??
st_dict is for easy remaking the same file. We need a different format
for easy reading the comments as well. For that we have the st_comments
"""
# TODO: store this in an HDF5 format! This is perfect for that.
# read all the lines of the file into memory
self.st_path, self.st_file = file_path, file_name
FILE = open(os.path.join(file_path, file_name))
lines = FILE.readlines()
FILE.close()
subset = False
st_dict = dict()
st_comments = dict()
for i, line in enumerate(lines):
# convert line to list space seperated list
line_list = self.fromline(line)
# see if the first character is marking something
if i == 0:
# it is possible that the NSET line is not defined
parts = line.split(' ')
try:
parts.remove(' ') # throws error when can't find
except ValueError:
pass
# we don't care what is on the nset line, just capture if