Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
import subprocess
from copy import deepcopy
from os.path import dirname, join, basename, exists
from IPython import get_ipython
from ...misc import local_path
from ... import pbs_string, default_pbs, qsub_exe, default_comm, \
interactive
from . import get_walltime, get_queues, get_mppalloc
from . import array_script
shell = get_ipython()
pbsargs = deepcopy(dict(default_comm))
pbsargs.update(default_pbs)
pbsargs['ppn'] = event.ppn
if not get_walltime(shell, event, pbsargs):
return
if not get_queues(shell, event, pbsargs):
return
# gets python script to launch in pbs.
pyscript = array_script.__file__
if pyscript[-1] == 'c':
pyscript = pyscript[:-1]
# creates file names.
hasprefix = getattr(event, "prefix", None)
def pbspaths(directory, jobname, suffix):
""" creates filename paths. """
return join(join(directory, jobname),
'{0}-pbs{1}'.format(event.prefix, suffix) if hasprefix
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':
pyscript = pyscript[:-1] # change .pyc to .py
# creates file names.
hasprefix = getattr(event, "prefix", None)