LaunchOnCondor.py

Go to the documentation of this file.

#!/usr/bin/env python

from __future__ import print_function
import urllib
import string
import os
import sys
import glob
import fnmatch
import commands

CopyRights  = '####################################\n'
CopyRights += '#        LaunchOnFarm Script       #\n'
CopyRights += '#     Loic.quertenmont@cern.ch     #\n'
CopyRights += '#            April 2010            #\n'
CopyRights += '####################################\n'

Farm_Directories  = []
Path_Cmd          = ''
Path_Shell        = ''
Path_Log          = ''
Path_Cfg          = ''
Jobs_Count        = 0
Jobs_Name         = ''
Jobs_Index        = ''
Jobs_Seed     = 0
Jobs_NEvent   =-1
Jobs_Skip         = 0
Jobs_Queue        = '8nh'
Jobs_Inputs   = []
Jobs_FinalCmds    = []
Jobs_RunHere      = 0

useLSF = True
LSFlog = True
runInteractively = False

def CreateTheConfigFile(argv):
    global Jobs_Name
    global Jobs_Index
    global Jobs_Count
    global Jobs_Seed
    global Jobs_Skip
    global Jobs_NEvent
    global Jobs_Inputs
    global Jobs_FinalCmds
    global Path_Cfg
    global CopyRights
    Path_Cfg   = Farm_Directories[1]+Jobs_Index+Jobs_Name+'_cfg.py'

    config_file=open(argv[1],'r')
    config_txt   = '\n\n' + CopyRights + '\n\n'
    config_txt  += config_file.read()
    config_file.close()
    i = 2
    while i < len(argv)-1:
        config_txt = config_txt.replace(argv[i],argv[i+1])
        i+=2

    #Default Replacements
    config_txt = config_txt.replace("XXX_I_XXX"         ,"%04i"%Jobs_Count)
    config_txt = config_txt.replace("XXX_PATH_XXX"      ,os.getcwd())
    config_txt = config_txt.replace("XXX_OUTPUT_XXX"        ,Jobs_Name)
    config_txt = config_txt.replace("XXX_NAME_XXX"      ,Jobs_Index+Jobs_Name)
    config_txt = config_txt.replace("XXX_SEED_XXX"      ,str(Jobs_Seed+Jobs_Count))
    config_txt = config_txt.replace("XXX_NEVENTS_XXX"   ,str(Jobs_NEvent))
    config_txt = config_txt.replace("XXX_SKIP_XXX"      ,str(Jobs_Skip))
    if Jobs_Count < len(Jobs_Inputs):
        config_txt = config_txt.replace("XXX_INPUT_XXX"         ,Jobs_Inputs[Jobs_Count])

    config_file=open(Path_Cfg,'w')
    config_file.write(config_txt)
    config_file.close()

def CreateTheShellFile(argv):
    global Path_Shell
    global Path_Cfg
    global CopyRights   
    global Jobs_RunHere
    global Jobs_FinalCmds
    Path_Shell = Farm_Directories[1]+Jobs_Index+Jobs_Name+'.sh'

    function_argument='('
    for i in range(2,len(argv)):
        function_argument+="%s" % argv[i]
        if i != len(argv)-1:
            function_argument+=', '
    function_argument+=')'

    shell_file=open(Path_Shell,'w')
    shell_file.write('#! /bin/sh\n')
    shell_file.write(CopyRights + '\n')
    shell_file.write('export SCRAM_ARCH='+os.getenv("SCRAM_ARCH","slc5_amd64_gcc462")+'\n')
    shell_file.write('export BUILD_ARCH='+os.getenv("BUILD_ARCH","slc5_amd64_gcc462")+'\n')
    shell_file.write('export VO_CMS_SW_DIR='+os.getenv("VO_CMS_SW_DIR","/nfs/soft/cms")+'\n')
    #shell_file.write('source /nfs/soft/cms/cmsset_default.sh\n')
    shell_file.write('cd ' + os.getcwd() + '\n')
    shell_file.write('eval `scramv1 runtime -sh`\n')

    if   argv[0]=='BASH':
        if Jobs_RunHere==0:
            shell_file.write('cd -\n')
        shell_file.write(argv[1] + " %s\n" % function_argument)
    elif argv[0]=='ROOT':
        if Jobs_RunHere==0:
            shell_file.write('cd -\n')
        shell_file.write('root -l -b << EOF\n')
        shell_file.write('   TString makeshared(gSystem->GetMakeSharedLib());\n')
        shell_file.write('   TString dummy = makeshared.ReplaceAll("-W ", "-Wno-deprecated-declarations -Wno-deprecated -Wno-unused-local-typedefs ");\n')
        shell_file.write('   TString dummy = makeshared.ReplaceAll("-Wshadow ", " -std=c++0x -D__USE_XOPEN2K8 ");\n')
        shell_file.write('   cout << "Compilling with the following arguments: " << makeshared << endl;\n')
        shell_file.write('   gSystem->SetMakeSharedLib(makeshared);\n')
        shell_file.write('   gSystem->SetIncludePath( "-I$ROOFITSYS/include" );\n')  
        shell_file.write('   .x %s+' % argv[1] + function_argument + '\n')
        shell_file.write('   .q\n')
        shell_file.write('EOF\n\n')
    elif argv[0]=='FWLITE':                 
        if Jobs_RunHere==0:
            shell_file.write('cd -\n')
        shell_file.write('root -l -b << EOF\n')
        shell_file.write('   TString makeshared(gSystem->GetMakeSharedLib());\n')
        shell_file.write('   TString dummy = makeshared.ReplaceAll("-W ", "-Wno-deprecated-declarations -Wno-deprecated ");\n')
        shell_file.write('   TString dummy = makeshared.ReplaceAll("-Wshadow ", " -std=c++0x -D__USE_XOPEN2K8 ");\n')
        shell_file.write('   cout << "Compilling with the following arguments: " << makeshared << endl;\n')
        shell_file.write('   gSystem->SetMakeSharedLib(makeshared);\n')
        shell_file.write('   gSystem->SetIncludePath("-I$ROOFITSYS/include");\n')
        shell_file.write('   gSystem->Load("libFWCoreFWLite");\n')
        shell_file.write('   FWLiteEnabler::enable();\n')
        shell_file.write('   gSystem->Load("libDataFormatsFWLite.so");\n')
        shell_file.write('   gSystem->Load("libAnalysisDataFormatsSUSYBSMObjects.so");\n')
        shell_file.write('   gSystem->Load("libDataFormatsVertexReco.so");\n')
        shell_file.write('   gSystem->Load("libDataFormatsHepMCCandidate.so");\n')
        shell_file.write('   gSystem->Load("libPhysicsToolsUtilities.so");\n')
        shell_file.write('   gSystem->Load("libdcap.so");\n')
        shell_file.write('   .x %s+' % argv[1] + function_argument + '\n')
        shell_file.write('   .q\n')
        shell_file.write('EOF\n\n')
    elif argv[0]=='CMSSW':
        CreateTheConfigFile(argv);
        if Jobs_RunHere==0:
            shell_file.write('cd -\n')
        shell_file.write('cmsRun ' + os.getcwd() + '/'+Path_Cfg + '\n')
    else:
        print() #Program to use is not specified... Guess it is bash command        
        shell_file.write('#Program to use is not specified... Guess it is bash command\n')
        shell_file.write(argv[1] + " %s\n" % function_argument)

    for i in range(len(Jobs_FinalCmds)):
        shell_file.write(Jobs_FinalCmds[i]+'\n')
    shell_file.write('mv '+ Jobs_Name+'* '+os.getcwd()+'/'+Farm_Directories[3]+'\n')
    shell_file.close()
    os.system("chmod 777 "+Path_Shell)


def CreateTheCmdFile():
    global useLSF
    global Path_Cmd
    global CopyRights
    Path_Cmd   = Farm_Directories[1]+Jobs_Name+'.cmd'
    cmd_file=open(Path_Cmd,'w')

    if useLSF:
        cmd_file.write(CopyRights + '\n')
    else:
        cmd_file.write('Universe                = vanilla\n')
        cmd_file.write('Environment             = CONDORJOBID=$(Process)\n')
        cmd_file.write('notification            = Error\n')
        #code specific for louvain
        if(commands.getstatusoutput("uname -n")[1].find("ucl.ac.be")!=-1):
            cmd_file.write('requirements            = (CMSFARM=?=True)&&(Memory > 200)\n')
        else:
            cmd_file.write('requirements            = (Memory > 200)\n')
        cmd_file.write('should_transfer_files   = YES\n')
        cmd_file.write('when_to_transfer_output = ON_EXIT\n')
    cmd_file.close()

def AddJobToCmdFile():
    global runInteractively
    global useLSF
    global Path_Shell
    global Path_Cmd
    global Path_Log
    Path_Log   = Farm_Directories[2]+Jobs_Index+Jobs_Name
    cmd_file=open(Path_Cmd,'a')

    if runInteractively:
        cmd_file.write("sh "+ os.getcwd() + "/"+Path_Shell + "\n")
    elif useLSF:
        if LSFlog :
            cmd_file.write("bsub -q " + Jobs_Queue + " -J " + Jobs_Name+Jobs_Index + " '" + os.getcwd() + "/"+Path_Shell + " 0 ele'\n")
        else :
            cmd_file.write("bsub -o /dev/null -q " + Jobs_Queue + " -J " + Jobs_Name+Jobs_Index + " '" + os.getcwd() + "/"+Path_Shell + " 0 ele'\n")
    else:
        cmd_file.write('\n')
        cmd_file.write('Executable              = %s\n'     % Path_Shell)
        cmd_file.write('output                  = %s.out\n' % Path_Log)
        cmd_file.write('error                   = %s.err\n' % Path_Log)
        cmd_file.write('log                     = %s.log\n' % Path_Log)
        cmd_file.write('Queue 1\n')
    cmd_file.close()

def CreateDirectoryStructure(FarmDirectory):
    global Jobs_Name
    global Farm_Directories
    Farm_Directories = [FarmDirectory+'/', FarmDirectory+'/inputs/', FarmDirectory+'/logs/', FarmDirectory+'/outputs/']
    for i in range(0,len(Farm_Directories)):
        if os.path.isdir(Farm_Directories[i]) == False:
            os.system('mkdir ' + Farm_Directories[i])

def SendCluster_LoadInputFiles(path, NJobs):
    global Jobs_Inputs
    input_file  = open(path,'r')
    input_lines = input_file.readlines()
    input_file.close()
    #input_lines.sort()

    BlockSize = (len(input_lines)/NJobs)
    LineIndex  = 0
    JobIndex   = 0
    BlockIndex = 0  
    Jobs_Inputs = [""]
    while LineIndex < len(input_lines):
        Jobs_Inputs[JobIndex] += input_lines[LineIndex]
        LineIndex +=1
        BlockIndex+=1
        if BlockIndex>BlockSize:
            BlockIndex = 0
            JobIndex  += 1
            Jobs_Inputs.append("")
    return JobIndex+1

def SendCluster_Create(FarmDirectory, JobName):
    global useLSF
    global Jobs_Name
    global Jobs_Count
    global Farm_Directories
    global runInteractively

    if(runInteractively): useLSF=True

    #determine if the submission system is LSF batch or condor
    command_out = commands.getstatusoutput("bjobs")[1]
    if(command_out.find("command not found")<0): useLSF = True
    else:                        useLSF = False;

    Jobs_Name  = JobName
    Jobs_Count = 0
    CreateDirectoryStructure(FarmDirectory)
    CreateTheCmdFile()

def SendCluster_Push(Argv):
    global Farm_Directories
    global Jobs_Count
    global Jobs_Index
    global Path_Shell
    global Path_Log

    Jobs_Index = "%04i_" % Jobs_Count
    if Jobs_Count==0 and (Argv[0]=="ROOT" or Argv[0]=="FWLITE"):                
        #First Need to Compile the macro --> Create a temporary shell path with no arguments
        print("Compiling the Macro...")
        CreateTheShellFile([Argv[0],Argv[1]])
        os.system('sh '+Path_Shell)
        os.system('rm '+Path_Shell)
        print("Getting the jobs...")
    print(Argv)
    CreateTheShellFile(Argv)
    AddJobToCmdFile()
    Jobs_Count = Jobs_Count+1

def SendCluster_Submit():
    global useLSF
    global CopyRights
    global Jobs_Count
    global Path_Cmd

    if useLSF:
        os.system("sh " + Path_Cmd)
    else:
        os.system("condor_submit " + Path_Cmd)  

    print('\n'+CopyRights)
    print('%i Job(s) has/have been submitted on the Computing Cluster' % Jobs_Count)

def SendSingleJob(FarmDirectory, JobName, Argv):
    SendCluster_Create(FarmDirectory, JobName, Argv)
    SendCluster_Push(FarmDirectory, JobName, Argv)
    SendCluster_Submit(FarmDirectory, JobName,Argv)

def SendCMSJobs(FarmDirectory, JobName, ConfigFile, InputFiles, NJobs, Argv):
    SendCluster_Create(FarmDirectory, JobName)
    NJobs = SendCluster_LoadInputFiles(InputFiles, NJobs)
    for i in range(NJobs):
        LaunchOnCondor.SendCluster_Push  (["CMSSW", ConfigFile])
    LaunchOnCondor.SendCluster_Submit()



def GetListOfFiles(Prefix, InputPattern, Suffix):
    List = []

    if(InputPattern.find('/store/cmst3')==0) :
        index = InputPattern.rfind('/')
        Listtmp = commands.getstatusoutput('cmsLs ' + InputPattern[0:index] + ' | awk \'{print $5}\'')[1].split('\n')
        pattern = InputPattern[index+1:len(InputPattern)]
        for file in Listtmp:
            if fnmatch.fnmatch(file, pattern): List.append(InputPattern[0:index]+'/'+file)
    elif(InputPattern.find('/castor/')==0):
        index = InputPattern.rfind('/')
        Listtmp = commands.getstatusoutput('rfdir ' + InputPattern[0:index] + ' | awk \'{print $9}\'')[1].split('\n')
        pattern = InputPattern[index+1:len(InputPattern)]
        for file in Listtmp:
            if fnmatch.fnmatch(file, pattern): List.append(InputPattern[0:index]+'/'+file)
    else :
        List = glob.glob(InputPattern)

    List = sorted(List)
    for i in range(len(List)):
        List[i] = Prefix + List[i] + Suffix
    return List


def ListToString(InputList):
    outString = ""
    for i in range(len(InputList)):
        outString += InputList[i]
    return outString

def ListToFile(InputList, outputFile):
    out_file=open(outputFile,'w')
    for i in range(len(InputList)):
        out_file.write('     ' + InputList[i] + '\n')
    out_file.close()

def FileToList(path):
    input_file  = open(path,'r')
    input_lines = input_file.readlines()
    input_file.close()
    input_lines.sort()
    return input_lines


def SendCMSMergeJob(FarmDirectory, JobName, InputFiles, OutputFile, KeepStatement):
    SendCluster_Create(FarmDirectory, JobName)
    Temp_Cfg   = Farm_Directories[1]+Jobs_Index+Jobs_Name+'_TEMP_cfg.py'

    if len(InputFiles)==0:
        print('Empty InputFile List for Job named "%s", Job will not be submitted' % JobName)
        return

    InputFilesString = ""
    for i in range(len(InputFiles)):
        InputFilesString += "     " + InputFiles[i] + '\n'

    cfg_file=open(Temp_Cfg,'w')
    cfg_file.write('import FWCore.ParameterSet.Config as cms\n')
    cfg_file.write('process = cms.Process("Merge")\n')
    cfg_file.write('\n')
    cfg_file.write('process.maxEvents = cms.untracked.PSet( input = cms.untracked.int32(-1) )\n')
    cfg_file.write('process.load("FWCore.MessageService.MessageLogger_cfi")\n')
    cfg_file.write('\n')
    cfg_file.write('process.MessageLogger.cerr.FwkReport.reportEvery = 50000\n')
    cfg_file.write('process.source = cms.Source("PoolSource",\n')
    cfg_file.write('   fileNames = cms.untracked.vstring(\n')
    cfg_file.write('%s' % InputFilesString)
    cfg_file.write('   )\n')
    cfg_file.write(')\n')
    cfg_file.write('\n')
    cfg_file.write('process.OUT = cms.OutputModule("PoolOutputModule",\n')
    cfg_file.write('    outputCommands = cms.untracked.vstring(%s),\n' % KeepStatement)
    cfg_file.write('    fileName = cms.untracked.string(%s)\n' % OutputFile)
    cfg_file.write(')\n')
    cfg_file.write('\n')
    cfg_file.write('process.endPath = cms.EndPath(process.OUT)\n')
    cfg_file.close()
    SendCluster_Push  (["CMSSW", Temp_Cfg])
    SendCluster_Submit()
    os.system('rm '+ Temp_Cfg)