#!/usr/bin/env python3
# -*- coding: utf-8 -*-
#
# Script "launch.py": aide au lancement d'un job Metafor
import sys, os, os.path, subprocess
import shutil, socket, platform, glob, fnmatch, re
import datetime, tarfile, signal
from parametricJob import *
from subprocess import * # RB subprocess deja importe + haut??
class LaunchJob(ParametricJob):
def __init__(self, _jobId=''):
self.debug = False
self.jobId=_jobId
cfgfile="launch%s.cfg"%self.jobId
ParametricJob.__init__(self, cfgfile)
# gestion des dépendances entre paramètres
self.applyDependencies()
# liens vers launchGui (lorsque lancé par ce biais pour interaction)
self.launchGui = None
self.outFile = None
def setLaunchGui(self, launchGui):
self.launchGui = launchGui
def setDefaultPars(self):
if len(self.pars)!=0:
return
YesNoPRM(self.pars, 'SEND_MAIL', 'send emails when simulations are over', False)
TextPRM(self.pars, 'MAIL_ADDR', 'e-mail address (reports)', os.getenv('USER'))
TextPRM(self.pars, 'SMTP_SERV', 'SMTP email server', 'smtp.ulg.ac.be')
TextPRM(self.pars, 'PYTHONEXE', 'Python executable', sys.executable)
mtfExe = os.path.abspath(os.path.dirname(__file__))+os.sep+'Metafor'
if (not isUnix()):
mtfExe = mtfExe+'.exe'
TextPRM(self.pars, 'EXEC_NAME', 'exec name', mtfExe)
TextPRM(self.pars, 'TEST_NAME', 'test filename', './mesTests/monTest.py')
TextPRM(self.pars, 'TEST_DIR', 'test dir for multi', './mesTests/')
TextPRM(self.pars, 'OUTFILE', 'logfile (no ext)', 'out')
YesNoPRM(self.pars, 'MULTITEST', 'Run multiple test on dir', False)
MultiPRM(self.pars, 'ALGORITHM', 'algorithm', ["meta", "import", "execfile", "clean", "verif", "restart"], "meta")
TextPRM(self.pars, 'RESTART_STEP', 'restart step', "-1")
TextPRM(self.pars, 'PRIORITY', 'priority [1-5]', "3")
TextPRM(self.pars, 'AFFINITY', 'affinity (cores list)', "")
TextPRM(self.pars, 'NB_TASKS', 'nb of task launched in parallel', "1")
TextPRM(self.pars, 'NB_THREADS', 'nb of threads by task', "1")
# Run Method :
MultiPRM(self.pars, 'RUNMETHOD', 'Run Method', ["interactive", "at", "batch", "slurm", "sge", "pbs"], "interactive")
#AT
TextPRM(self.pars, 'AT_TIME' , 'Delay for at launch (no syntax check, use with care)', "now")
# QUEUING SYSTEMS : SGE / SLURM
TextPRM(self.pars, 'QUEUE', 'Queue name', "defq")
YesNoPRM(self.pars, 'LOCALDISK', 'Metafor run on node local disk', True)
TextPRM(self.pars, 'MEMORY', 'Total Memory (Mb)', "1000")
TextPRM(self.pars, 'TIME', 'Time (slurm : d-hh:mm:ss / PBS : d:hh:mm:ss) ', "0-1:00:00")
# SGE specific
TextPRM(self.pars, 'SGE_PE', 'SGE parallel environment', "snode")
TextPRM(self.pars, 'SGEARGS', 'additional SGE args', "")
# PBS Specific (zenobe)
#TextPRM(self.pars, 'QUEUE', 'PBS Queue name', "main") # zenobe : main ou large
TextPRM(self.pars, 'PROJECT', 'Project Name (Acces group dirs)', "hpcwe")
MultiPRM(self.pars, 'MODEL', 'CPU Model', ["auto", "westemere", "westmere_fat", "westmere_xfat", "ivyBridge", "haswell_xfat"], "auto")
# FTP
YesNoPRM(self.pars, 'ENABLE_FTP', 'ftp transfert', False)
TextPRM(self.pars, 'FTP_HOST', 'ftp host', "")
TextPRM(self.pars, 'FTP_PORT', 'ftp port', "21")
TextPRM(self.pars, 'FTP_USER', 'ftp user', "")
TextPRM(self.pars, 'FTP_PASS', 'ftp passwd', "")
TextPRM(self.pars, 'FTP_DIR', 'ftp directory', "incoming")
# Actions
PRMAction(self.actions, 'a', self.pars['MAIL_ADDR'])
PRMAction(self.actions, 'b', self.pars['PYTHONEXE'])
PRMAction(self.actions, 'c', self.pars['EXEC_NAME'])
PRMAction(self.actions, 'd', self.pars['TEST_NAME'])
PRMAction(self.actions, 'd', self.pars['TEST_DIR'])
PRMAction(self.actions, 'e', self.pars['OUTFILE'])
PRMAction(self.actions, 'f', self.pars['ALGORITHM'])
PRMAction(self.actions, 'g', self.pars['RESTART_STEP'])
PRMAction(self.actions, 'h', self.pars['MULTITEST'])
PRMAction(self.actions, 'i', self.pars['PRIORITY'])
PRMAction(self.actions, 'j', self.pars['AFFINITY'])
PRMAction(self.actions, 'k', self.pars['NB_TASKS'])
PRMAction(self.actions, 'l', self.pars['NB_THREADS'])
PRMAction(self.actions, 'm', self.pars['RUNMETHOD'])
# At parameters
PRMAction(self.actions, 'n', self.pars['AT_TIME'])
# SGE/SLURM/PBS PARAMETERS
PRMAction(self.actions, 'n', self.pars['QUEUE'])
PRMAction(self.actions, 'o', self.pars['LOCALDISK'])
PRMAction(self.actions, 'p', self.pars['MEMORY'])
PRMAction(self.actions, 'q', self.pars['TIME'])
# SGE specific PARAMETERS
#PRMAction(self.actions, 'n', self.pars['SGEQUEUE'])
PRMAction(self.actions, 'r', self.pars['SGE_PE'])
PRMAction(self.actions, 's', self.pars['SGEARGS'])
# PBS (zenobe) specific PARAMETERS
PRMAction(self.actions, 'r', self.pars['PROJECT'])
PRMAction(self.actions, 's', self.pars['MODEL'])
# FTP
PRMAction(self.actions, 'u', self.pars['ENABLE_FTP'])
PRMAction(self.actions, 'v', self.pars['FTP_HOST'])
PRMAction(self.actions, 'w', self.pars['FTP_PORT'])
PRMAction(self.actions, 'x', self.pars['FTP_USER'])
PRMAction(self.actions, 'y', self.pars['FTP_PASS'])
PRMAction(self.actions, 'z', self.pars['FTP_DIR'])
# Actions
NoAction(self.actions)
GoAction(self.actions, 'G')
SaveAction(self.actions, 'S')
QuitAction(self.actions, 'Q')
def applyDependencies(self):
ret = False
if self.debug:
print("applyDependecies: ")
print(" self.pars['ALGORITHM'].val = ", self.pars['ALGORITHM'].val)
#if self.pars['ALGORITHM'].val=='restart':
# self.pars['MULTITEST'].val = False
# ret = True
if self.debug:
print(" self.pars['MULTITEST'].val = ", self.pars['MULTITEST'].val)
return ret
def configActions(self):
#self.pars['MAIL_ADDR'].enable(self.pars['SEND_MAIL'].val==True or
# self.pars['RUNMETHOD'].val=='sge' or
# self.pars['RUNMETHOD'].val=='slurm' or
# self.pars['RUNMETHOD'].val=='pbs')
self.pars['SMTP_SERV'].enable(self.pars['SEND_MAIL'].val==True)
self.pars['TEST_NAME'].enable(self.pars['MULTITEST'].val==False)
self.pars['TEST_DIR'].enable(self.pars['MULTITEST'].val==True)
self.pars['RESTART_STEP'].enable(self.pars['ALGORITHM'].val=='restart' and
self.pars['MULTITEST'].val==False)
self.pars['FTP_HOST'].enable(self.pars['ENABLE_FTP'].val==True)
self.pars['FTP_PORT'].enable(self.pars['ENABLE_FTP'].val==True)
self.pars['FTP_USER'].enable(self.pars['ENABLE_FTP'].val==True)
self.pars['FTP_PASS'].enable(self.pars['ENABLE_FTP'].val==True)
self.pars['FTP_DIR'].enable(self.pars['ENABLE_FTP'].val==True)
self.pars['PRIORITY'].enable(self.pars['RUNMETHOD'].val!='sge' and
self.pars['RUNMETHOD'].val!='slurm')
self.pars['AFFINITY'].enable(self.pars['RUNMETHOD'].val!='sge' and
self.pars['RUNMETHOD'].val!='slurm' and
self.pars['MULTITEST'].val==False)
# At
self.pars['AT_TIME'].enable(self.pars['RUNMETHOD'].val=='at')
# SGE/SLURM/PBS
self.pars['QUEUE'].enable(self.pars['RUNMETHOD'].val=='sge' or
self.pars['RUNMETHOD'].val=='slurm' or
self.pars['RUNMETHOD'].val=='pbs')
self.pars['LOCALDISK'].enable((self.pars['RUNMETHOD'].val=='sge' or
self.pars['RUNMETHOD'].val=='slurm') and # pas zenobe
self.pars['ALGORITHM'].val!='restart' )
self.pars['TIME'].enable(self.pars['RUNMETHOD'].val=='sge' or
self.pars['RUNMETHOD'].val=='slurm' or
self.pars['RUNMETHOD'].val=='pbs')
self.pars['MEMORY'].enable(self.pars['RUNMETHOD'].val=='sge' or
self.pars['RUNMETHOD'].val=='slurm' or
self.pars['RUNMETHOD'].val=='pbs')
# SGE specific
self.pars['SGE_PE'].enable(self.pars['RUNMETHOD'].val=='sge')
self.pars['SGEARGS'].enable(self.pars['RUNMETHOD'].val=='sge')
# PBS specific
self.pars['MODEL'].enable(self.pars['RUNMETHOD'].val=='pbs') # cpu model (to choose subqueue)
self.pars['PROJECT'].enable(self.pars['RUNMETHOD'].val=='pbs') # project (group access & facturation)
def getJobName(self):
if (self.pars['MULTITEST'].val==False):
jobname=os.path.basename(os.getcwd())+"."+self.pars['TEST_NAME'].val
else:
jobname=os.path.basename(os.getcwd())+"."+self.pars['TEST_DIR'].val
jobname=jobname.replace(os.sep,'.')
jobname=jobname.replace('.py','')
jobname=jobname.replace('...','.')
jobname=jobname.replace('..','.')
if jobname.endswith('.'):
jobname=jobname[:-1]
return jobname.encode('ascii','ignore') # convert to ASCII if some strings were unicode
def getOutFileName(self):
outFileName = "%s.%s%s.txt" % (self.pars['OUTFILE'].val, self.pars['ALGORITHM'].val,self.jobId)
if self.debug:
print("outFileName = ", outFileName)
return outFileName
# RUN Functions
def run(self):
# deactivate use of LocalDisk in restart
if (self.pars['ALGORITHM'].val == "restart" ):
self.pars['LOCALDISK'].val = False
# write kill scripts
if isUnix():
if (self.pars['RUNMETHOD'].val == 'interactive' or self.pars['RUNMETHOD'].val == 'at' or
self.pars['RUNMETHOD'].val == 'batch'):
self.killScript(self.jobId, os.getpgrp())
elif ((self.pars['RUNMETHOD'].val == 'sge' or self.pars['RUNMETHOD'].val == 'slurm') and
self.pars['LOCALDISK'].val == True):
self.cpNodeResultsScript(self.jobId)
self.rmNodeResultsScript(self.jobId)
# check exec
if not os.path.isfile(self.pars['EXEC_NAME'].val):
self.error("Metafor executable not found at %s" % self.pars['EXEC_NAME'].val)
# add __init__.py
if not os.path.isfile("__init__.py"):
file = open("__init__.py","w")
file.close()
# starts tests
if (self.pars['MULTITEST'].val==True):
#check
if not os.path.isdir(self.pars['TEST_DIR'].val):
print("Error: 'TEST_DIR' %s non existant directory" % self.pars['TEST_DIR'].val)
return
#asciiname = self.pars['TEST_DIR'].val.encode('ascii','ignore') # convert unicode (from PyQt)
#outRun = self.startMultipleTests(asciiname)
outRun = self.startMultipleTests(self.pars['TEST_DIR'].val)
else:
#check
if not os.path.isfile(self.pars['TEST_NAME'].val):
print("Error: 'TEST_NAME' %s non existant file" % self.pars['TEST_NAME'].val)
return
#run
#asciiname = self.pars['TEST_NAME'].val #.encode('ascii','ignore') # convert unicode (from PyQt)
#outRun = self.startMultipleTests(asciiname)
outRun = self.startMultipleTests(self.pars['TEST_NAME'].val)
'''
if (self.pars['ALGORITHM'].val == "restart" && self.pars['RESTART_STEP'].val > 0):
outRun = self.startSingleTest()
else:
outRun = self.startMultipleTests(self.pars['TEST_NAME'].val)
'''
if self.pars['ENABLE_FTP'].val==True:
#tar facs
cdir=os.path.basename(os.getcwd())
tarname="%s.tar.gz" % cdir
print("creating %s" % tarname)
os.chdir('..')
tar = tarfile.open(tarname,'w:gz')
for path, dirs, files in os.walk(cdir):
for file in files:
tar.add(os.path.join(path,file))
tar.close()
#ftp
print("sending results to %s" % self.pars['FTP_HOST'].val)
import ftplib
ftp = ftplib.FTP()
ftp.connect(self.pars['FTP_HOST'].val, self.pars['FTP_PORT'].val)
ftp.login(self.pars['FTP_USER'].val, self.pars['FTP_PASS'].val)
ftp.cwd(self.pars['FTP_DIR'].val)
file = open(tarname,'r')
ftp.storbinary('STOR %s' %tarname, file)
file.close()
ftp.quit()
#clean
os.remove(tarname)
# remove kill scripts
if isUnix():
fNames = []
if self.pars['RUNMETHOD'].val == 'sge':
fNames.append(self.qDelScriptName(self.jobId))
#fNames.append(self.cfgfile)
elif self.pars['RUNMETHOD'].val == 'slurm':
fNames.append(self.sCancelScriptName(self.jobId))
#fNames.append(self.cfgfile)
elif self.pars['RUNMETHOD'].val == 'at' or self.pars['RUNMETHOD'].val == 'batch':
fNames.append("kill%s.py"%self.jobId)
fNames.append("atrm%s.py"%self.jobId)
fNames.append("runbatch%s.sh"%self.jobId)
#fNames.append(self.cfgfile)
else:
fNames.append("kill%s.py"%self.jobId)
for fil in fNames:
if os.path.isfile(fil):
os.remove(fil)
print("done.")
def startMultipleTests(self, tests):
print("startMultipleTests")
# starting timer
now = datetime.datetime.now()
print("starting Multiple test at %s (come back later)" % now.ctime())
# opening outfile
self.outFile = open(self.getOutFileName(),"w")
mtfdir, mtfexe = os.path.split(self.pars['EXEC_NAME'].val)
#mtfdir, mtfexe = os.path.split(self.pars['EXEC_NAME'].val.encode('ascii','ignore'))
mtfdir = os.path.abspath(mtfdir) # necessaire pour sge sinon, en local disk, soucis pour trouver libGen4.so
# starting python battery
if self.launchGui:
#self.launchGui.outFile=outfile
# starting process (unable to renice it => nice in battery)
prog = self.pars['PYTHONEXE'].val
# python args -u : unbuffered output (to have output synchronized to output)
# -i : force a prompt event even if stdin is not a terminal (else it is not possible to write cmds to python)
arg = ['-u', '-i']
self.launchGui.process.start(prog, arg)
self.launchGui.process.waitForStarted(-1)
# defining the input flux
pin = self.launchGui.process
else:
#cmd = self.getNiceCmd(int(self.pars['NICE_VALUE'].val))
#cmd = cmd + ['python']
cmd = [self.pars['PYTHONEXE'].val]
if isUnix(): # shell=False && sans close_fds = True (ca freeze)
#Add mtfdir to LD_LIBRARY_PATH to allow launch to find mt*.so
if 'LD_LIBRARY_PATH' in os.environ:
os.environ['LD_LIBRARY_PATH'] = mtfdir+':'+os.environ['LD_LIBRARY_PATH']
else:
os.environ['LD_LIBRARY_PATH'] = mtfdir
#self.outFile.write("LD_LIBRARY_PATH = %s"%os.environ['LD_LIBRARY_PATH'])
p = subprocess.Popen(cmd, stdin=subprocess.PIPE, stdout=self.outFile, stderr=self.outFile, env=os.environ, shell=False, close_fds=True)
else: # si nice+Windows => "shell=True" (pour "start")
p = subprocess.Popen(cmd, stdin=subprocess.PIPE, stdout=self.outFile, stderr=self.outFile, env=os.environ, shell=True)
# defining the input flux
pin = p.stdin
# filling command to battery
pin.write(b'import sys, os, os.path\n')
if self.debug:
pin.write(b'print ("sys.path = ", sys.path)\n')
pin.write(b'print ("os.getcwd() = ", os.getcwd())\n')
pin.write(('if os.path.isdir(r"%s"):\n'%mtfdir).encode('utf-8'))
pin.write(('\tsys.path.append(r"%s")\n'%mtfdir).encode('utf-8')) # mtfdir est dorenavant un abspath
pin.write(b'else:\n')
pin.write(('\tprint ("metafor dir %s not found!")\n'%mtfdir).encode('utf-8'))
pin.write(b'\tsys.exit()\n\n')
pin.write(('exec(open(r"%s").read())\n'%os.path.join(mtfdir,'.pythonrc.py')).encode('utf-8'))
pin.write(b'import toolbox.battery as b\n')
pin.write(b'battery = b.Battery() \n')
pin.write(b'battery.keepFacs = True\n')
pin.write(('battery.dirs = [r"%s"]\n'%tests).encode('utf-8'))
pin.write(('battery.jobId = r"%s"\n'%self.jobId).encode('utf-8'))
if (self.pars['ALGORITHM'].val == "execfile"):
reg1=r"(.+)_0*([1-9][0-9]*)\.py"
exp1= re.compile(reg1)
m = exp1.match(os.path.basename(tests))
if m: # chaining tests
pin.write(('battery.addCplxExecPath(r"%s_*")\n'%os.path.join(os.path.dirname(tests), m.group(1))).encode('utf-8'))
print('battery.cplx_exec = [r"%s_*"]\n'%os.path.join(os.path.dirname(tests), m.group(1)))
else:
pin.write(('battery.addCplxExecPath(r"%s")\n'%tests).encode('utf-8'))
print('battery.cplx_exec = [r"%s"]\n'%tests)
elif (self.pars['ALGORITHM'].val == "import" ):
pin.write(('battery.addCplxImportPath(r"%s")\n'%tests).encode('utf-8'))
print('battery.cplx_import = [r"%s"]\n'%tests)
elif (self.pars['ALGORITHM'].val == "restart" ):
pin.write(('battery.addRestartPath(r"%s")\n'%tests).encode('utf-8'))
print('battery.restart = [r"%s"]\n'%tests)
pin.write(b'battery.verifsrc = "verif"\n')
pin.write(b'battery.codes = [ "FAILED", "STP", "ITE", "INW", "EXT", "EXW", "LKS", "CPU", "MEM" ]\n')
if ((self.pars['RUNMETHOD'].val == 'sge' or self.pars['RUNMETHOD'].val == 'slurm') and
self.pars['LOCALDISK'].val == True):
pin.write(('battery.setWDRoot("%s")\n'%self.getLocalDiskDir(self.jobId)).encode('utf-8'))
if self.pars['RUNMETHOD'].val != 'sge' and self.pars['RUNMETHOD'].val != 'slurm':
if self.pars['AFFINITY'].val != '':
pin.write(('battery.setAffinity("%s")\n'%self.pars['AFFINITY'].val).encode('utf-8'))
pin.write(('battery.setPriority(%s)\n'%self.pars['PRIORITY'].val).encode('utf-8'))
pin.write(('battery.setNumTasks(%s)\n'%self.pars['NB_TASKS'].val).encode('utf-8'))
pin.write(('battery.setNumThreads(%s)\n'%self.pars['NB_THREADS'].val).encode('utf-8'))
pin.write(('battery.mtfdir = r"%s"\n'%mtfdir).encode('utf-8'))
if self.pars['ALGORITHM'].val == 'clean':
pin.write(b'battery.start("clean")\n')
elif self.pars['ALGORITHM'].val == 'verif':
pin.write(b'battery.verif()\n')
else:
pin.write(b'battery.start("run")\n')
#pin.write('battery.verif()\n') # pas très utile dans le cadre de launch ou faudrait faire un verif + malin)
# write to exit python at the end of job
pin.write(b'quit()\n')
# wait for process to finish
if self.launchGui:
retcode = self.launchGui.waitQProcessForFinish()
#print ("Qprocess finished with retcode: ",retcode)
else:
#close pin flux
pin.close()
# flush des outfile pour déjà avoir un max d'info de debug
self.outFile.flush()
# waiting execution time
retcode = p.wait()
# closing file
self.outFile.flush()
self.outFile.close()
self.outFile = None
# post pro cmd
if ((self.pars['RUNMETHOD'].val == 'sge' or self.pars['RUNMETHOD'].val == 'slurm') and
self.pars['LOCALDISK'].val == True):
print("Trying to get back local workspace to home")
self.moveLocalDir2Home(self.jobId)
now = datetime.datetime.now()
print("battery completed at %s" % now.ctime())
if self.pars['SEND_MAIL'].val == True:
self.mailmsg("multipleTests complete", file=self.getOutFileName())
return retcode
'''
def startSingleTest(self):
# affinity/numa stuffs
affinitycmd=[]
if self.pars['AFFINITY'].val!='' and self.pars['RUNMETHOD'].val != 'sge':
if self.hasSysCmd('numactl'):
affinitycmd=[ "numactl", "--physcpubind", self.pars['AFFINITY'].val ] #, "--localalloc" ]
elif self.hasSysCmd('taskset'):
affinitycmd=[ "taskset", "-c", self.pars['AFFINITY'].val ]
# set lib path
libpath = os.path.dirname(os.path.abspath(self.pars['EXEC_NAME'].val))
libpath = os.path.join(libpath,"lib")
print "adding extra libpath =",libpath
newenv=dict(os.environ)
#newenv['PATH'] = newenv['PATH']+':'+libpath
if newenv.has_key('LD_LIBRARY_PATH'):
newenv['LD_LIBRARY_PATH'] = newenv['LD_LIBRARY_PATH']+':'+libpath
else:
newenv['LD_LIBRARY_PATH'] = libpath
# spawn algo
outFileName = self.getOutFileName()
self.outFile=open(outFileName,"w")
nicecmd=[]
if platform.uname()[0]!='Windows':
if self.pars['NICE_VALUE'].val!='0':
nicecmd = [ "nice", "-%s" %self.pars['NICE_VALUE'].val]
print "starting \"%s\" on %s" % (self.pars['ALGORITHM'].val, self.pars['TEST_NAME'].val)
#cmd = [ self.pars['EXEC_NAME'].val, "-nogui", "-j", "%s"%self.pars['NB_THREADS'].val]
cmd = [ self.pars['EXEC_NAME'].val, "-nogui"]
p = Popen(nicecmd+affinitycmd+cmd, stdout=self.outFile, stderr=self.outFile, stdin=PIPE, env=newenv)
if ((self.pars['RUNMETHOD'].val == 'sge' or self.pars['RUNMETHOD'].val == 'slurm') and
self.pars['LOCALDISK'].val == True and self.pars['ALGORITHM'].val != "restart" :
# writing recovery scripts and running on local hdd
self.cpNodeResultsScript(self.jobId)
self.rmNodeResultsScript(self.jobId)
p.stdin.write("setTheWDirRoot('%s')\n" % self.getLocalDiskDir(self.jobId))
p.stdin.write("setNumTasks(%s)\n"%self.pars['NB_TASKS'].val)
p.stdin.write("wrap.Blas.setNumThreads(%s)\n"%self.pars['NB_THREADS'].val)
p.stdin.write("wrap.IntelTBB.setNumThreads(%s)\n"%self.pars['NB_THREADS'].val)
if self.pars['ALGORITHM'].val=="meta":
p.stdin.write("load(r'%s')\n" % (self.pars['TEST_NAME'].val))
p.stdin.write("meta()\n")
elif self.pars['ALGORITHM'].val=="execfile":
p.stdin.write("__file__=r'%s'\n" % (self.pars['TEST_NAME'].val)) # needed by .pythonrc.py
p.stdin.write("execfile(r'%s')\n" % (self.pars['TEST_NAME'].val))
elif self.pars['ALGORITHM'].val=="restart":
p.stdin.write("load(r'%s')\n" % self.pars['TEST_NAME'].val)
p.stdin.write("restart(%s)\n" % self.pars['RESTART_STEP'].val)
elif self.pars['ALGORITHM'].val=="import":
p.stdin.write("import os ; print os.environ\n")
p.stdin.write("import %s\n" % self.pars['TEST_NAME'].val)
p.stdin.write("quit()\n")
p.stdin.close()
# waiting execution time
retcode = p.wait()
# recovering results and removing scripts
if (self.pars['RUNMETHOD'].val == 'sge' or self.pars['RUNMETHOD'].val == 'slurm') and
self.pars['LOCALDISK'].val == True and self.pars['ALGORITHM'].val != "restart":
print "Getting back local disk workspace to home disk"
self.moveLocalDir2Home(self.jobId)
if os.path.isdir(self.getLocalDiskDir(self.jobId)): # si la copie a été bien faite => le local dir a été nettoyé => on peut virer les scripts
if os.path.isfile(self.cpNodeResultsScriptName(self.jobId)):
os.remove(self.cpNodeResultsScriptName(self.jobId))
if os.path.isfile(self.rmNodeResultsScriptName(self.jobId)):
os.remove(self.rmNodeResultsScriptName(self.jobId))
self.outFile.close()
# grep results + e-mail
import StringIO
res=StringIO.StringIO()
res.write("cwd=%s\n" % os.getcwd())
for file in glob.glob('%s.*.txt'%self.pars['OUTFILE'].val):
res.write("result file : %s\n"%file)
for line in open(file,'r'):
for txt in ["ERROR", "Error", "error", "TSC-", "Successful", "Problem at time"]:
if line.find(txt)!=-1:
res.write(line)
if self.pars['SEND_MAIL'].val == True:
self.mailmsg("job %s done" % self.pars['TEST_NAME'].val, text=res.getvalue())
res.close()
'''
if __name__ == "__main__":
from argparse import ArgumentParser
parser = ArgumentParser(description="Metafor's Launcher")
# definition of arguments
parser.add_argument('-d', '--directory', dest='rundir',
metavar='DIR', help='specify run directory (batch mode)')
parser.add_argument('-x', '--nogui', action='store_false',
dest='usegui',default=True, help='disable menu')
parser.add_argument('-i', '--jobId', dest='jobId', type=str, default='',
help='job id')
# Parsing arguments
args = parser.parse_args()
#if len(args)!=0:
# parser.error("too many arguments")
# starting launch
if args.rundir:
os.chdir(args.rundir)
job = LaunchJob(args.jobId)
if args.usegui:
job.menu()
else:
job.run()