Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def goto(self, cmdl):
""" Moves current dictionary position and working directory (if appropriate). """
from os import chdir
from os.path import exists, join, split as splitpath, isdir
from .explore import explore
from . import get_shell
from pylada import interactive
shell = get_shell(self)
if interactive.jobfolder is None:
print("No current job-folders.")
return
if len(cmdl.split()) == 0:
if interactive.jobfolder_path is None:
print("Current position in job folder:", interactive.jobfolder.name)
else:
print("Current position in job folder:", interactive.jobfolder.name)
print("Filename of job-folder: ", interactive.jobfolder_path)
return
args = cmdl.split()
if len(args) > 1:
print("Invalid argument to goto {0}.".format(cmdl))
return
def launch(self, event, jobfolders):
""" Launch jobs interactively.
This call will block until each job is finished in turn.
"""
from os.path import join, dirname
from copy import deepcopy
from .. import get_shell
from ... import default_comm
try:
kwargs = get_shell(self).ev(event.kwargs)
except:
print("Could not process keyword arguments.")
print(event.kwargs)
return
if event.nbprocs != 0:
comm = deepcopy(default_comm)
comm['n'] = event.nbprocs
comm["ppn"] = event.ppn
kwargs['comm'] = comm
for current, path in jobfolders:
# start computations.
for job in current.values():
if job.is_tagged:
continue
name = str(job.name)
from os import remove
import pylada
from .. import get_shell
from ...misc import local_path, testValidProgram
from ... import pbs_string, default_pbs, qsub_exe, default_comm
from . import get_walltime, get_mppalloc, get_queues, scattered_script
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