Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
directory = dirname(interactive.jobfolder_path)
job = interactive.jobfolder
name = interactive.jobfolder.name[1:]
if job.functional is None:
print("Current jobfolder does not contain any calculation.")
return
# avoid successful jobs.unless specifically requested
if hasattr(job.functional, 'Extract') and not event.force:
p = join(directory, name)
extract = job.functional.Extract(p)
if extract.success:
print("Job {0} completed successfully. " \
"It will not be relaunched.".format(name))
return
# now creates script
pbsargs['n'] = get_mppalloc(shell, event, False)
pbsargs['nnodes'] = (pbsargs['n'] + pbsargs['ppn'] - 1) \
// pbsargs['ppn']
pbsargs['err'] = pbspaths(directory, name, 'err')
pbsargs['out'] = pbspaths(directory, name, 'out')
pbsargs['name'] = name if len(name) \
else "{0}-root".format(basename(path))
pbsargs['directory'] = directory
pbsargs['header'] = ""
pbsargs['header'] += '\nexport PYLADA_JOBARRAY_NAME={0!r}\n'.format(name)
pbsargs['scriptcommand'] = "{0} --nbprocs {n} --ppn {ppn} {1} " \
.format(pyscript, path, **pbsargs)
pbsscript = pbspaths(directory, name, 'script')
local_path(directory).join(name).ensure(dir=True)
with open(pbsscript, "w") as file:
string = pbs_string(**pbsargs) if hasattr(pbs_string, '__call__') \
from .. import logger
if not hasattr(pylada, 'ipython_qstat'):
logger.warning("Missing ipython_qstat function: cannot check for jobs already in queue")
qstat = lambda x: []
else:
qstat = lambda x: self.qstat(x)
logger.info("launch/scattered: event: %s" % event)
shell = get_shell(self)
pbsargs = deepcopy(dict(default_comm))
pbsargs.update(default_pbs)
pbsargs['ppn'] = event.ppn
mppalloc = get_mppalloc(shell, event)
if mppalloc is None:
return
# Set pbsargs['walltime'] to a string like '03:59:59'
if not get_walltime(shell, event, pbsargs):
return
# Set pbsargs['queue'], pbsargs['account']
if not get_queues(shell, event, pbsargs):
return
logger.info("launch/scattered: pbsargs: %s" % pbsargs)
# gets python script to launch in pbs.
pyscript = scattered_script.__file__
logger.info("launch/scattered: pyscript: %s" % pyscript)
if pyscript[-1] == 'c':