Configuration/scripts/submitToCondor.py

#!/usr/bin/env python
import os
import sys
import datetime
import shutil
import re
import subprocess
import signal
import fcntl
from optparse import OptionParser

from OSUT3Analysis.Configuration.configurationOptions import *
from OSUT3Analysis.Configuration.processingUtilities import *

parser = OptionParser()
parser = set_commandline_arguments(parser)

parser.remove_option("-o")
parser.remove_option("-t")
parser.remove_option("-q")
parser.remove_option("-n")
parser.remove_option("-u")
parser.remove_option("-e")
parser.remove_option("-r")
parser.remove_option("-R")
parser.remove_option("-d")
parser.remove_option("-b")
parser.remove_option("--2D")
parser.remove_option("-y")
parser.remove_option("-p")

parser.add_option("-s", "--skimDir", dest="skimDir",
                  help="condor directory containing skim")
parser.add_option("-a", "--skimChannel", dest="skimChannel",
                  help="channel from skim to use")
parser.add_option("-d", "--mergeDaemon", action="store_true", dest="mergeDaemon", default=False,
                  help="launch a daemon to merge output when jobs are done")

(arguments, args) = parser.parse_args()

if (arguments.skimDir and not arguments.skimChannel) or (not arguments.skimDir and arguments.skimChannel):
    print "Both the skim directory (--skimDir) and channel (--skimChannel) must be given."
    exit ()

if arguments.localConfig:
    sys.path.append(os.getcwd())
    exec("from " + arguments.localConfig.rstrip('.py') + " import *")

condor_dir = set_condor_submit_dir(arguments)
short_condor_dir = condor_dir
short_condor_dir = re.sub (r".*/([^/]*)", r"\1", short_condor_dir)
short_condor_dir = list (short_condor_dir)
short_condor_dir[0] = short_condor_dir[0].upper ()
short_condor_dir = "".join (short_condor_dir)

if not os.path.exists (condor_dir):
    os.system("mkdir %s" % (condor_dir))

split_datasets = split_composite_datasets(datasets, composite_dataset_definitions)

clusters = ""
for dataset in split_datasets:
    output_dir = "%s/%s" % (condor_dir, dataset)
    skim_dir = ""
    skim_channel_dir = ""
    command = "osusub -l %s -m %d -p %s %s %s %s %s" % (dataset, maxEvents[dataset], short_condor_dir, dataset_names[dataset], config_file, output_dir, nJobs[dataset])
    if arguments.skimDir:
        skim_dir = "condor/" + arguments.skimDir + "/" + dataset
        skim_channel_dir = "condor/" + arguments.skimDir + "/" + dataset + "/" + arguments.skimChannel
        if os.path.exists (skim_channel_dir):
            command = "osusub -d %s -l %s -m %d -p %s %s %s %s %s" % (dataset_names[dataset], dataset, maxEvents[dataset], short_condor_dir, skim_channel_dir, config_file, output_dir, nJobs[dataset])
        else:
            print dataset + "/" + arguments.skimChannel + " not in skim directory. Skipping."
            continue
    print command
    pid = os.getpid ()
    p0 = subprocess.Popen (command.split (), bufsize=1, stdout=subprocess.PIPE)
    flags = fcntl.fcntl (p0.stdout.fileno (), fcntl.F_GETFL, 0)
    fcntl.fcntl (p0.stdout.fileno (), fcntl.F_SETFL, flags | os.O_NONBLOCK)
    output = ""
    while p0.poll () is None:
        try:
            tmpOutput = p0.stdout.read (1024)
            print tmpOutput,
            output += tmpOutput
        except IOError:
            pass
    tmpOutput = p0.stdout.read (1024)
    print tmpOutput,
    output += tmpOutput
    output = re.sub (r"[\f\n\r]", r"", output)
    output = re.sub (r".*submitted to cluster (.*)\..*$", r"\1", output)
    clusters += " " + output
    if arguments.skimDir and os.path.exists (skim_channel_dir + "/skimNumberOfEvents.txt") and os.path.exists (skim_dir + "/numberOfEvents.txt") and os.path.exists (skim_dir + "/crossSectionInPicobarn.txt"):
        shutil.copy (skim_channel_dir + "/skimNumberOfEvents.txt", output_dir + "/skimNumberOfEvents.txt")
        shutil.copy (skim_dir + "/numberOfEvents.txt", output_dir + "/originalNumberOfEvents.txt")
        f = open (skim_channel_dir + "/skimNumberOfEvents.txt", "r")
        skimNumberOfEvents = float (f.read ().rstrip ())
        f.close ()
        f = open (skim_dir + "/numberOfEvents.txt", "r")
        numberOfEvents = float (f.read ().rstrip ())
        f.close ()
        f = open (skim_dir + "/crossSectionInPicobarn.txt", "r")
        crossSectionInPicobarn = float (f.read ().rstrip ())
        f.close ()
        if numberOfEvents:
            crossSectionInPicobarn *= skimNumberOfEvents / numberOfEvents
        else:
            crossSectionInPicobarn *= skimNumberOfEvents * numberOfEvents
        f = open (output_dir + "/crossSectionInPicobarn.txt", "w")
        f.write (str (crossSectionInPicobarn) + "\n")
        f.close ()

if arguments.mergeDaemon:
    command = "mergeOutput.py"
    if arguments.condorDir:
        command += " -c " + arguments.condorDir
    pid = os.fork ()
    if not pid:
        signal.signal (signal.SIGHUP, signal.SIG_IGN)
        pid = os.getpid ()
        if arguments.localConfig:
            shutil.copy (arguments.localConfig, "mergeDaemonOptions_" + str (pid) + ".py")
            command += " -l mergeDaemonOptions_" + str (pid) + ".py"
        os.execvp ("mergeDaemon.pl", ["mergeDaemon.pl", command] + clusters.split ())
    else:
        print "\nMerging daemon PID: " + str (pid)
Revision:	1.10
Committed:	Tue Jul 9 17:22:41 2013 UTC (11 years, 9 months ago) by ahart
Content type:	text/x-python
Branch:	MAIN
Changes since 1.9:	+25 -3 lines
Log Message:	When submitting jobs to run over a skim, output the product of the cross section and selection efficiency in crossSectionInPicobarn.txt.
#	User	Rev	Content
1	lantonel	1.1	#!/usr/bin/env python
2			import os
3			import sys
4			import datetime
5	ahart	1.7	import shutil
6			import re
7	ahart	1.8	import subprocess
8			import signal
9	ahart	1.9	import fcntl
10	lantonel	1.5	from optparse import OptionParser
11	lantonel	1.1
12	lantonel	1.2	from OSUT3Analysis.Configuration.configurationOptions import *
13			from OSUT3Analysis.Configuration.processingUtilities import *
14	lantonel	1.1
15	lantonel	1.5	parser = OptionParser()
16	lantonel	1.1	parser = set_commandline_arguments(parser)
17	ahart	1.7
18			parser.remove_option("-o")
19			parser.remove_option("-t")
20			parser.remove_option("-q")
21			parser.remove_option("-n")
22			parser.remove_option("-u")
23			parser.remove_option("-e")
24			parser.remove_option("-r")
25			parser.remove_option("-R")
26			parser.remove_option("-d")
27			parser.remove_option("-b")
28			parser.remove_option("--2D")
29			parser.remove_option("-y")
30			parser.remove_option("-p")
31
32			parser.add_option("-s", "--skimDir", dest="skimDir",
33			help="condor directory containing skim")
34			parser.add_option("-a", "--skimChannel", dest="skimChannel",
35			help="channel from skim to use")
36	ahart	1.8	parser.add_option("-d", "--mergeDaemon", action="store_true", dest="mergeDaemon", default=False,
37			help="launch a daemon to merge output when jobs are done")
38	ahart	1.7
39	lantonel	1.5	(arguments, args) = parser.parse_args()
40	lantonel	1.1
41	ahart	1.7	if (arguments.skimDir and not arguments.skimChannel) or (not arguments.skimDir and arguments.skimChannel):
42			print "Both the skim directory (--skimDir) and channel (--skimChannel) must be given."
43			exit ()
44
45	lantonel	1.4	if arguments.localConfig:
46	lantonel	1.1	sys.path.append(os.getcwd())
47	lantonel	1.4	exec("from " + arguments.localConfig.rstrip('.py') + " import *")
48	lantonel	1.1
49	lantonel	1.4	condor_dir = set_condor_submit_dir(arguments)
50	ahart	1.7	short_condor_dir = condor_dir
51			short_condor_dir = re.sub (r"./([^/])", r"\1", short_condor_dir)
52			short_condor_dir = list (short_condor_dir)
53			short_condor_dir[0] = short_condor_dir[0].upper ()
54			short_condor_dir = "".join (short_condor_dir)
55	lantonel	1.1
56	ahart	1.7	if not os.path.exists (condor_dir):
57			os.system("mkdir %s" % (condor_dir))
58	lantonel	1.1
59	ahart	1.6	split_datasets = split_composite_datasets(datasets, composite_dataset_definitions)
60	lantonel	1.1
61	ahart	1.8	clusters = ""
62	lantonel	1.1	for dataset in split_datasets:
63			output_dir = "%s/%s" % (condor_dir, dataset)
64	ahart	1.10	skim_dir = ""
65			skim_channel_dir = ""
66	ahart	1.7	command = "osusub -l %s -m %d -p %s %s %s %s %s" % (dataset, maxEvents[dataset], short_condor_dir, dataset_names[dataset], config_file, output_dir, nJobs[dataset])
67			if arguments.skimDir:
68	ahart	1.10	skim_dir = "condor/" + arguments.skimDir + "/" + dataset
69			skim_channel_dir = "condor/" + arguments.skimDir + "/" + dataset + "/" + arguments.skimChannel
70			if os.path.exists (skim_channel_dir):
71			command = "osusub -d %s -l %s -m %d -p %s %s %s %s %s" % (dataset_names[dataset], dataset, maxEvents[dataset], short_condor_dir, skim_channel_dir, config_file, output_dir, nJobs[dataset])
72	ahart	1.7	else:
73			print dataset + "/" + arguments.skimChannel + " not in skim directory. Skipping."
74			continue
75	lantonel	1.1	print command
76	ahart	1.9	pid = os.getpid ()
77			p0 = subprocess.Popen (command.split (), bufsize=1, stdout=subprocess.PIPE)
78			flags = fcntl.fcntl (p0.stdout.fileno (), fcntl.F_GETFL, 0)
79			fcntl.fcntl (p0.stdout.fileno (), fcntl.F_SETFL, flags \| os.O_NONBLOCK)
80			output = ""
81			while p0.poll () is None:
82			try:
83			tmpOutput = p0.stdout.read (1024)
84			print tmpOutput,
85			output += tmpOutput
86			except IOError:
87			pass
88			tmpOutput = p0.stdout.read (1024)
89			print tmpOutput,
90			output += tmpOutput
91	ahart	1.8	output = re.sub (r"[\f\n\r]", r"", output)
92			output = re.sub (r".submitted to cluster (.)\..*$", r"\1", output)
93			clusters += " " + output
94	ahart	1.10	if arguments.skimDir and os.path.exists (skim_channel_dir + "/skimNumberOfEvents.txt") and os.path.exists (skim_dir + "/numberOfEvents.txt") and os.path.exists (skim_dir + "/crossSectionInPicobarn.txt"):
95			shutil.copy (skim_channel_dir + "/skimNumberOfEvents.txt", output_dir + "/skimNumberOfEvents.txt")
96			shutil.copy (skim_dir + "/numberOfEvents.txt", output_dir + "/originalNumberOfEvents.txt")
97			f = open (skim_channel_dir + "/skimNumberOfEvents.txt", "r")
98			skimNumberOfEvents = float (f.read ().rstrip ())
99			f.close ()
100			f = open (skim_dir + "/numberOfEvents.txt", "r")
101			numberOfEvents = float (f.read ().rstrip ())
102			f.close ()
103			f = open (skim_dir + "/crossSectionInPicobarn.txt", "r")
104			crossSectionInPicobarn = float (f.read ().rstrip ())
105			f.close ()
106			if numberOfEvents:
107			crossSectionInPicobarn *= skimNumberOfEvents / numberOfEvents
108			else:
109			crossSectionInPicobarn = skimNumberOfEvents numberOfEvents
110			f = open (output_dir + "/crossSectionInPicobarn.txt", "w")
111			f.write (str (crossSectionInPicobarn) + "\n")
112			f.close ()
113	ahart	1.8
114			if arguments.mergeDaemon:
115			command = "mergeOutput.py"
116			if arguments.condorDir:
117			command += " -c " + arguments.condorDir
118			pid = os.fork ()
119			if not pid:
120			signal.signal (signal.SIGHUP, signal.SIG_IGN)
121			pid = os.getpid ()
122			if arguments.localConfig:
123			shutil.copy (arguments.localConfig, "mergeDaemonOptions_" + str (pid) + ".py")
124			command += " -l mergeDaemonOptions_" + str (pid) + ".py"
125			os.execvp ("mergeDaemon.pl", ["mergeDaemon.pl", command] + clusters.split ())
126			else:
127			print "\nMerging daemon PID: " + str (pid)