10 print 'Not injecting to wmagent in --show mode. Need to run the worklfows.' 12 if opt.wmcontrol==
'init':
15 if opt.wmcontrol==
'test':
18 if opt.wmcontrol==
'submit' and opt.nProcs==0:
19 print 'Not injecting to wmagent in -j 0 mode. Need to run the worklfows.' 21 if opt.wmcontrol==
'force':
22 print "This is an expert setting, you'd better know what you're doing" 26 from modules.wma
import upload_to_couch
27 (filePath,labelInCouch,user,group,where) = arguments
28 cacheId=upload_to_couch(filePath,
44 for k
in options.split(
','):
45 if k.startswith(
'dqm:'):
46 self.
dqmgui=k.split(
':',1)[-1]
47 elif k.startswith(
'wma:'):
50 self.
testMode=((mode!=
'submit')
and (mode!=
'force'))
62 self.
wmagent=os.getenv(
'WMAGENT_REQMGR')
68 self.
wmagent =
'cmsweb-testbed.cern.ch' 69 self.
DbsUrl =
"https://"+self.
wmagent+
"/dbs/int/global/DBSReader" 72 self.
dqmgui=
"https://cmsweb.cern.ch/dqm/relval" 77 self.
user = os.getenv(
'USER')
85 if not os.getenv(
'WMCORE_ROOT'):
86 print '\n\twmclient is not setup properly. Will not be able to upload or submit requests.\n' 91 print '\n\tFound wmclient\n' 94 "RequestType" :
"TaskChain",
95 "SubRequestType" :
"RelVal",
96 "RequestPriority": 500000,
97 "Requestor": self.
user,
99 "CMSSWVersion": os.getenv(
'CMSSW_VERSION'),
100 "Campaign": os.getenv(
'CMSSW_VERSION'),
101 "ScramArch": os.getenv(
'SCRAM_ARCH'),
102 "ProcessingVersion": self.
version,
104 "ConfigCacheUrl": self.
couch,
112 "SizePerEvent" : 1234,
113 "TimePerEvent" : 0.1,
114 "PrepID": os.getenv(
'CMSSW_VERSION')
118 "EnableHarvesting" :
"True",
119 "DQMUploadUrl" : self.
dqmgui,
120 "DQMConfigCacheID" :
None,
126 "ConfigCacheID" :
None,
128 "SplittingAlgo" :
"EventBased",
129 "EventsPerJob" :
None,
130 "RequestNumEvents" :
None,
131 "Seeding" :
"AutomaticSeeding",
132 "PrimaryDataset" :
None,
134 "Multicore" : opt.nThreads,
138 "TaskName" :
"DigiHLT",
139 "ConfigCacheID" :
None,
141 "InputDataset" :
None,
142 "SplittingAlgo" :
"LumiBased",
145 "Multicore" : opt.nThreads,
151 "InputFromOutputModule" :
None,
152 "ConfigCacheID" :
None,
154 "SplittingAlgo" :
"LumiBased",
157 "Multicore" : opt.nThreads,
164 def prepare(self,mReader, directories, mode='init'):
171 wmsplit[
'RECODreHLT']=2
175 wmsplit[
'DIGIUP15_PU50']=1
176 wmsplit[
'RECOUP15_PU50']=1
177 wmsplit[
'DIGIUP15_PU25']=1
178 wmsplit[
'RECOUP15_PU25']=1
179 wmsplit[
'DIGIUP15_PU25HS']=1
180 wmsplit[
'RECOUP15_PU25HS']=1
181 wmsplit[
'DIGIHIMIX']=5
182 wmsplit[
'RECOHIMIX']=5
183 wmsplit[
'RECODSplit']=1
184 wmsplit[
'SingleMuPt10_UP15_ID']=1
185 wmsplit[
'DIGIUP15_ID']=1
186 wmsplit[
'RECOUP15_ID']=1
187 wmsplit[
'TTbar_13_ID']=1
188 wmsplit[
'SingleMuPt10FS_ID']=1
189 wmsplit[
'TTbarFS_ID']=1
190 wmsplit[
'RECODR2_50nsreHLT']=5
191 wmsplit[
'RECODR2_25nsreHLT']=5
192 wmsplit[
'RECODR2_2016reHLT']=5
193 wmsplit[
'RECODR2_50nsreHLT_HIPM']=5
194 wmsplit[
'RECODR2_25nsreHLT_HIPM']=5
195 wmsplit[
'RECODR2_2016reHLT_HIPM']=5
196 wmsplit[
'RECODR2_2016reHLT_skimSingleMu']=5
197 wmsplit[
'RECODR2_2016reHLT_skimDoubleEG']=5
198 wmsplit[
'RECODR2_2016reHLT_skimMuonEG']=5
199 wmsplit[
'RECODR2_2016reHLT_skimJetHT']=5
200 wmsplit[
'RECODR2_2016reHLT_skimMET']=5
201 wmsplit[
'RECODR2_2016reHLT_skimSinglePh']=5
202 wmsplit[
'RECODR2_2016reHLT_skimMuOnia']=5
203 wmsplit[
'RECODR2_2016reHLT_skimSingleMu_HIPM']=5
204 wmsplit[
'RECODR2_2016reHLT_skimDoubleEG_HIPM']=5
205 wmsplit[
'RECODR2_2016reHLT_skimMuonEG_HIPM']=5
206 wmsplit[
'RECODR2_2016reHLT_skimJetHT_HIPM']=5
207 wmsplit[
'RECODR2_2016reHLT_skimMET_HIPM']=5
208 wmsplit[
'RECODR2_2016reHLT_skimSinglePh_HIPM']=5
209 wmsplit[
'RECODR2_2016reHLT_skimMuOnia_HIPM']=5
210 wmsplit[
'HLTDR2_50ns']=1
211 wmsplit[
'HLTDR2_25ns']=1
212 wmsplit[
'HLTDR2_2016']=1
213 wmsplit[
'Hadronizer']=1
214 wmsplit[
'DIGIUP15']=1
215 wmsplit[
'RECOUP15']=1
216 wmsplit[
'RECOAODUP15']=5
217 wmsplit[
'DBLMINIAODMCUP15NODQM']=5
218 wmsplit[
'DigiFull']=5
219 wmsplit[
'RecoFull']=5
220 wmsplit[
'DigiFullPU']=1
221 wmsplit[
'RecoFullPU']=1
227 print "Not set up for step splitting" 231 for (n,dir)
in directories.items():
233 print "inspecting",dir
235 for (x,s)
in mReader.workFlowSteps.items():
245 if len( [step
for step
in s[3]
if "HARVESTGEN" in step] )>0:
246 chainDict[
'TimePerEvent']=0.01
247 thisLabel=thisLabel+
"_gen" 249 if len( [step
for step
in s[3]
if "DBLMINIAODMCUP15NODQM" in step] )>0:
250 thisLabel=thisLabel+
"_dblMiniAOD" 255 if 'INPUT' in step
or (
not isinstance(s[2][index],str)):
256 nextHasDSInput=s[2][index]
264 chainDict[
'nowmTasklist'][-1][
'nowmIO']=json.loads(open(
'%s/%s.io'%(dir,step)).read())
266 print "Failed to find",
'%s/%s.io'%(dir,step),
".The workflows were probably not run on cfg not created" 269 chainDict[
'nowmTasklist'][-1][
'PrimaryDataset']=
'RelVal'+s[1].
split(
'+')[0]
270 if not '--relval' in s[2][index]:
271 print 'Impossible to create task from scratch without splitting information with --relval' 274 arg=s[2][index].
split()
275 ns=
map(int,arg[arg.index(
'--relval')+1].
split(
','))
276 chainDict[
'nowmTasklist'][-1][
'RequestNumEvents'] = ns[0]
277 chainDict[
'nowmTasklist'][-1][
'EventsPerJob'] = ns[1]
278 if 'FASTSIM' in s[2][index]
or '--fast' in s[2][index]:
279 thisLabel+=
'_FastSim' 280 if 'lhe' in s[2][index]
in s[2][index]:
281 chainDict[
'nowmTasklist'][-1][
'LheInputFiles'] =
True 286 chainDict[
'nowmTasklist'][-1][
'nowmIO']=json.loads(open(
'%s/%s.io'%(dir,step)).read())
288 print "Failed to find",
'%s/%s.io'%(dir,step),
".The workflows were probably not run on cfg not created" 290 chainDict[
'nowmTasklist'][-1][
'InputDataset']=nextHasDSInput.dataSet
291 splitForThisWf=nextHasDSInput.split
292 chainDict[
'nowmTasklist'][-1][
'LumisPerJob']=splitForThisWf
294 chainDict[
'nowmTasklist'][-1][
'LumisPerJob']=wmsplit[step]
296 if len(nextHasDSInput.run):
297 chainDict[
'nowmTasklist'][-1][
'RunWhitelist']=nextHasDSInput.run
298 if len(nextHasDSInput.ls):
299 chainDict[
'nowmTasklist'][-1][
'LumiList']=nextHasDSInput.ls
301 if '--data' in s[2][index]
and nextHasDSInput.label:
302 thisLabel+=
'_RelVal_%s'%nextHasDSInput.label
303 if 'filter' in chainDict[
'nowmTasklist'][-1][
'nowmIO']:
304 print "This has an input DS and a filter sequence: very likely to be the PyQuen sample" 305 processStrPrefix=
'PU_' 306 setPrimaryDs =
'RelVal'+s[1].
split(
'+')[0]
308 chainDict[
'nowmTasklist'][-1][
'PrimaryDataset']=setPrimaryDs
314 chainDict[
'nowmTasklist'][-1][
'nowmIO']=json.loads(open(
'%s/%s.io'%(dir,step)).read())
316 print "Failed to find",
'%s/%s.io'%(dir,step),
".The workflows were probably not run on cfg not created" 319 chainDict[
'nowmTasklist'][-1][
'LumisPerJob']=splitForThisWf
321 chainDict[
'nowmTasklist'][-1][
'LumisPerJob']=wmsplit[step]
324 if 'Hadronizer' in step:
325 chainDict[
'nowmTasklist'][-1][
'LumisPerJob']=wmsplit[
'Hadronizer']
328 chainDict[
'nowmTasklist'][-1][
'TaskName']=step
330 chainDict[
'nowmTasklist'][-1][
'PrimaryDataset']=setPrimaryDs
331 chainDict[
'nowmTasklist'][-1][
'ConfigCacheID']=
'%s/%s.py'%(dir,step)
332 chainDict[
'nowmTasklist'][-1][
'GlobalTag']=chainDict[
'nowmTasklist'][-1][
'nowmIO'][
'GT']
333 chainDict[
'GlobalTag']=chainDict[
'nowmTasklist'][-1][
'nowmIO'][
'GT']
334 if 'pileup' in chainDict[
'nowmTasklist'][-1][
'nowmIO']:
335 chainDict[
'nowmTasklist'][-1][
'MCPileup']=chainDict[
'nowmTasklist'][-1][
'nowmIO'][
'pileup']
336 if '--pileup ' in s[2][index]:
337 processStrPrefix=
'PU_' 339 processStrPrefix=
'PU25ns_' 341 processStrPrefix=
'PU50ns_' 342 if 'DIGIPREMIX_S2' in s[2][index] :
344 processStrPrefix=
'PUpmx25ns_' 345 elif s[2][index].
split()[ s[2][index].
split().
index(
'--pileup_input')+1 ].
find(
'50ns') > 0 :
346 processStrPrefix=
'PUpmx50ns_' 350 chainDict[
'AcquisitionEra'][step]=chainDict[
'CMSSWVersion']
351 chainDict[
'ProcessingString'][step]=processStrPrefix+chainDict[
'nowmTasklist'][-1][
'GlobalTag'].
replace(
'::All',
'')+thisLabel
354 chainDict[
'nowmTasklist'][-1][
'AcquisitionEra']=chainDict[
'CMSSWVersion']
355 chainDict[
'nowmTasklist'][-1][
'ProcessingString']=processStrPrefix+chainDict[
'nowmTasklist'][-1][
'GlobalTag'].
replace(
'::All',
'')+thisLabel
358 chainDict[
'nowmTasklist'][-1][
'Campaign'] = chainDict[
'nowmTasklist'][-1][
'AcquisitionEra']+self.
batchName 361 if (
'DBLMINIAODMCUP15NODQM' in step):
362 chainDict[
'nowmTasklist'][-1][
'ProcessingString']=chainDict[
'nowmTasklist'][-1][
'ProcessingString']+
'_miniAOD' 364 if( chainDict[
'nowmTasklist'][-1][
'Multicore'] ):
371 chainDict[
'nowmTasklist'][-1][
'Memory'] = self.
memoryOffset +
int( chainDict[
'nowmTasklist'][-1][
'Multicore'] -1 ) * self.
memPerCore 375 chainDict[
'RequestString']=
'RV'+chainDict[
'CMSSWVersion']+s[1].
split(
'+')[0]
376 if processStrPrefix
or thisLabel:
377 chainDict[
'RequestString']+=
'_'+processStrPrefix+thisLabel
380 chainDict[
'PrepID'] = chainDict[
'CMSSWVersion']+
'__'+self.
batchTime+
'-'+s[1].
split(
'+')[0]
382 chainDict[
'PrepID'] = chainDict[
'CMSSWVersion']+self.
batchName+
'-'+s[1].
split(
'+')[0]
384 chainDict[
'SubRequestType'] =
"HIRelVal" 391 for i_second
in reversed(range(len(chainDict[
'nowmTasklist']))):
392 t_second=chainDict[
'nowmTasklist'][i_second]
394 if 'primary' in t_second[
'nowmIO']:
396 primary=t_second[
'nowmIO'][
'primary'][0].
replace(
'file:',
'')
397 for i_input
in reversed(range(0,i_second)):
398 t_input=chainDict[
'nowmTasklist'][i_input]
399 for (om,o)
in t_input[
'nowmIO'].
items():
402 t_second[
'InputTask'] = t_input[
'TaskName']
403 t_second[
'InputFromOutputModule'] = om
405 if t_second[
'TaskName'].startswith(
'HARVEST'):
407 chainDict[
'DQMConfigCacheID']=t_second[
'ConfigCacheID']
427 chainDict[
'AcquisitionEra'] = chainDict[
'AcquisitionEra'].
values()[0]
428 chainDict[
'ProcessingString'] = chainDict[
'ProcessingString'].
values()[0]
430 chainDict[
'AcquisitionEra'] = chainDict[
'nowmTasklist'][0][
'AcquisitionEra']
431 chainDict[
'ProcessingString'] = chainDict[
'nowmTasklist'][0][
'ProcessingString']
435 chainDict[
'Campaign'] = chainDict[
'AcquisitionEra']+self.
batchName 441 if type(i)==int
and i < len(chainDict[
'nowmTasklist']):
442 chainDict[
'nowmTasklist'][i][
'KeepOutput']=
True 443 for (i,t)
in enumerate(chainDict[
'nowmTasklist']):
444 if t[
'TaskName'].startswith(
'HARVEST'):
448 elif t[
'TaskName']
in self.
keep:
452 chainDict[
'Task%d'%(itask)]=t
459 chainDict[
'TaskChain']=itask
461 chainDict.pop(
'nowmTasklist')
468 labelInCouch=self.
label+
'_'+label
469 cacheName=filePath.split(
'/')[-1]
472 print '\tFake upload of',filePath,
'to couch with label',labelInCouch
476 from modules.wma
import upload_to_couch,DATABASE_NAME
478 print '\n\tUnable to find wmcontrol modules. Please include it in your python path\n' 483 print "Not re-uploading",filePath,
"to",where,
"for",label
486 print "Loading",filePath,
"to",where,
"for",label
488 pool = multiprocessing.Pool(1)
489 cacheIds = pool.map( upload_to_couch_oneArg, [(filePath,labelInCouch,self.
user,self.
group,where)] )
490 cacheId = cacheIds[0]
495 for (n,d)
in self.chainDicts.items():
497 if it.startswith(
"Task")
and it!=
'TaskChain':
499 couchID=self.
uploadConf(d[it][
'ConfigCacheID'],
500 str(n)+d[it][
'TaskName'],
503 print d[it][
'ConfigCacheID'],
" uploaded to couchDB for",
str(n),
"with ID",couchID
504 d[it][
'ConfigCacheID']=couchID
505 if it ==
'DQMConfigCacheID':
506 couchID=self.
uploadConf(d[
'DQMConfigCacheID'],
510 print d[
'DQMConfigCacheID'],
"uploaded to couchDB for",
str(n),
"with ID",couchID
511 d[
'DQMConfigCacheID']=couchID
516 from modules.wma
import makeRequest,approveRequest
517 from wmcontrol
import random_sleep
518 print '\n\tFound wmcontrol\n' 520 print '\n\tUnable to find wmcontrol modules. Please include it in your python path\n' 526 for (n,d)
in self.chainDicts.items():
528 print "Only viewing request",n
529 print pprint.pprint(d)
532 print "For eyes before submitting",n
533 print pprint.pprint(d)
534 print "Submitting",n,
"..........." 535 workFlow=makeRequest(self.
wmagent,d,encodeDict=
True)
536 approveRequest(self.
wmagent,workFlow)
537 print "...........",n,
"submitted"
def prepare(self, mReader, directories, mode='init')
def performInjectionOptionTest(opt)
def uploadConf(self, filePath, label, where)
the info are not in the task specific dict but in the general dict t_input.update(copy.deepcopy(self.defaultHarvest)) t_input['DQMConfigCacheID']=t_second['ConfigCacheID']
def replace(string, replacements)
void find(edm::Handle< EcalRecHitCollection > &hits, DetId thisDet, std::vector< EcalRecHitCollection::const_iterator > &hit, bool debug=false)
def __init__(self, opt, mode='init', options='')
def upload_to_couch_oneArg(arguments)