CMS 3D CMS Logo

runTheMatrix.py
Go to the documentation of this file.
1 #!/usr/bin/env python3
2 from __future__ import print_function
3 import sys, os
4 
5 from Configuration.PyReleaseValidation.MatrixReader import MatrixReader
6 from Configuration.PyReleaseValidation.MatrixRunner import MatrixRunner
7 from Configuration.PyReleaseValidation.MatrixInjector import MatrixInjector,performInjectionOptionTest
8 
9 # ================================================================================
10 
11 def showRaw(opt):
12 
13  mrd = MatrixReader(opt)
14  mrd.showRaw(opt.useInput, opt.refRel, opt.fromScratch, opt.raw, opt.step1Only, selected=opt.testList)
15 
16  return 0
17 
18 # ================================================================================
19 
20 def runSelected(opt):
21 
22  mrd = MatrixReader(opt)
23  mrd.prepare(opt.useInput, opt.refRel, opt.fromScratch)
24 
25  # test for wrong input workflows
26  if opt.testList:
27  definedWf = [dwf.numId for dwf in mrd.workFlows]
28  definedSet = set(definedWf)
29  testSet = set(opt.testList)
30  undefSet = testSet - definedSet
31  if len(undefSet)>0: raise ValueError('Undefined workflows: '+', '.join(map(str,list(undefSet))))
32  duplicates = [wf for wf in testSet if definedWf.count(wf)>1 ]
33  if len(duplicates)>0: raise ValueError('Duplicated workflows: '+', '.join(map(str,list(duplicates))))
34 
35  ret = 0
36  if opt.show:
37  mrd.show(opt.testList, opt.extended, opt.cafVeto)
38  if opt.testList : print('selected items:', opt.testList)
39  else:
40  mRunnerHi = MatrixRunner(mrd.workFlows, opt.nProcs, opt.nThreads)
41  ret = mRunnerHi.runTests(opt)
42 
43  if opt.wmcontrol:
44  if ret!=0:
45  print('Cannot go on with wmagent injection with failing workflows')
46  else:
47  wfInjector = MatrixInjector(opt,mode=opt.wmcontrol,options=opt.wmoptions)
48  ret= wfInjector.prepare(mrd,
49  mRunnerHi.runDirs)
50  if ret==0:
51  wfInjector.upload()
52  wfInjector.submit()
53  return ret
54 
55 # ================================================================================
56 
57 if __name__ == '__main__':
58 
59  #this can get out of here
60  predefinedSet={
61  'limited' : [
62  # See README for further details
63 
66  5.1, # TTbar_8TeV_TuneCUETP8M1 FastSim
67  8, # RelValBeamHalo Cosmics
68  9.0, # RelValHiggs200ChargedTaus
69  25, # RelValTTbar
70  101.0, # SingleElectronE120EHCAL + ECALHCAL.customise + fullMixCustomize_cff.setCrossingFrameOn
71 
72  # Run2
73  7.3, # UndergroundCosmicSPLooseMu
74  1306.0, # RelValSingleMuPt1_UP15
75  1330, # RelValZMM_13
76  135.4, # ZEE_13TeV_TuneCUETP8M1
77  25202.0, # RelValTTbar_13 PU = AVE_35_BX_25ns
78  250202.181, # RelValTTbar_13 PREMIX
79 
80  # Run3
81  11634.0, # TTbar_14TeV 2021
82  13234.0, # RelValTTbar_14TeV 2021 FastsSim
83  12434.0, # RelValTTbar_14TeV 2023
84  12834.0, # RelValTTbar_14TeV 2024
85  12846.0, # RelValZEE_13 2024
86  13034.0, # RelValTTbar_14TeV 2024 PU = Run3_Flat55To75_PoissonOOTPU
87  12834.7, # RelValTTbar_14TeV 2024 mkFit
88  14034.0, # RelValTTbar_14TeV Run3_2023_FastSim
89  14234.0, # RelValTTbar_14TeV Run3_2023_FastSim PU = Run3_Flat55To75_PoissonOOTPU
90  2500.201, # RelValTTbar_14TeV NanoAOD from existing MINI
91 
92  # Phase2
93  29634.0, # RelValTTbar_14TeV phase2_realistic_T33 Extended2026D110 (Phase-2 baseline)
94  24834.911, # Previous DD4hep baseline for monitoring the stability of DD4hep workflow
95  29634.911, # TTbar_14TeV_TuneCP5 phase2_realistic_T33 DD4hepExtended2026D110 DD4Hep (HLLHC14TeV BeamSpot)
96  29834.999, # RelValTTbar_14TeV (PREMIX) phase2_realistic_T33 Extended2026D110 AVE_50_BX_25ns_m3p3
97  29696.0, # RelValCloseByPGun_CE_E_Front_120um phase2_realistic_T33 Extended2026D110
98  29700.0, # RelValCloseByPGun_CE_H_Coarse_Scint phase2_realistic_T33 Extended2026D110
99  #23234.0, # Need new workflow with HFNose
100 
101 
102 
104  4.22, # Run2011A Cosmics
105  4.53, # Run2012B Photon miniAODs
106  1000, # Run2011A MinimumBias Prompt RecoTLR.customisePrompt
107  1001, # Run2011A MinimumBias Data+Express
108 
109  136.731, # Run2016B SinglePhoton
110  136.793, # Run2017C DoubleEG
111  136.874, # Run2018C EGamma
112 
113 
115  139.001, # Run2021 MinimumBias Commissioning2021
116 
117  # 2022
118  140.023, # Run2022B ZeroBias
119  140.043, # Run2022C ZeroBias
120  140.063, # Run2022D ZeroBias
121 
122  # 2023
123  141.044, # Run2023D JetMET0
124  141.042, # Run2023D ZeroBias
125  141.046, # Run2023D EGamma0
126 
127 
130  140.56, # HIRun2018A HIHardProbes Run2_2018_pp_on_AA
131 
133  312.0, # Pyquen_ZeemumuJets_pt10_2760GeV 2022 PU : HiMixGEN
134  322.0, # Pyquen_ZeemumuJets_pt10_5362GeV_2023 2023 PU : HiMixGEN
135  332.0, # Pyquen_ZeemumuJets_pt10_5362GeV_2024 2024 PU : HiMixGEN
136  ],
137  'jetmc': [5.1, 13, 15, 25, 38, 39], #MC
138  'metmc' : [5.1, 15, 25, 37, 38, 39], #MC
139  'muonmc' : [5.1, 124.4, 124.5, 20, 21, 22, 23, 25, 30], #MC
140  }
141 
142 
143  import argparse
144  usage = 'usage: runTheMatrix.py --show -s '
145 
146  parser = argparse.ArgumentParser(usage,formatter_class=argparse.ArgumentDefaultsHelpFormatter)
147 
148  parser.add_argument('-b','--batchName',
149  help='relval batch: suffix to be appended to Campaign name',
150  dest='batchName',
151  default='')
152 
153  parser.add_argument('-m','--memoryOffset',
154  help='memory of the wf for single core',
155  dest='memoryOffset',
156  type=int,
157  default=3000)
158 
159  parser.add_argument('--addMemPerCore',
160  help='increase of memory per each n > 1 core: memory(n_core) = memoryOffset + (n_core-1) * memPerCore',
161  dest='memPerCore',
162  type=int,
163  default=1500)
164 
165  parser.add_argument('-j','--nproc',
166  help='number of processes. 0 Will use 4 processes, not execute anything but create the wfs',
167  dest='nProcs',
168  type=int,
169  default=4)
170 
171  parser.add_argument('-t','--nThreads',
172  help='number of threads per process to use in cmsRun.',
173  dest='nThreads',
174  type=int,
175  default=1)
176 
177  parser.add_argument('--nStreams',
178  help='number of streams to use in cmsRun.',
179  dest='nStreams',
180  type=int,
181  default=0)
182 
183  parser.add_argument('--nEvents',
184  help='number of events to process in cmsRun. If 0 will use the standard 10 events.',
185  dest='nEvents',
186  type=int,
187  default=0)
188 
189  parser.add_argument('--numberEventsInLuminosityBlock',
190  help='number of events in a luminosity block',
191  dest='numberEventsInLuminosityBlock',
192  type=int,
193  default=-1)
194 
195  parser.add_argument('-n','--showMatrix',
196  help='Only show the worflows. Use --ext to show more',
197  dest='show',
198  default=False,
199  action='store_true')
200 
201  parser.add_argument('-e','--extended',
202  help='Show details of workflows, used with --show',
203  dest='extended',
204  default=False,
205  action='store_true')
206 
207  parser.add_argument('-s','--selected',
208  help='Run a pre-defined selected matrix of wf. Deprecated, please use -l limited',
209  dest='restricted',
210  default=False,
211  action='store_true')
212 
213  parser.add_argument('-l','--list',
214  help='Comma separated list of workflow to be shown or ran. Possible keys are also '+str(predefinedSet.keys())+'. and wild card like muon, or mc',
215  dest='testList',
216  default=None)
217 
218  parser.add_argument('-f','--failed-from',
219  help='Provide a matrix report to specify the workflows to be run again. Augments the -l option if specified already',
220  dest='failed_from',
221  default=None)
222 
223  parser.add_argument('-r','--raw',
224  help='Temporary dump the .txt needed for prodAgent interface. To be discontinued soon. Argument must be the name of the set (standard, pileup,...)',
225  dest='raw')
226 
227  parser.add_argument('-i','--useInput',
228  help='Use recyling where available. Either all, or a comma separated list of wf number.',
229  dest='useInput',
230  type=lambda x: x.split(','),
231  default=None)
232 
233  parser.add_argument('-w','--what',
234  help='Specify the set to be used. Argument must be the name of a set (standard, pileup,...) or multiple sets separated by commas (--what standard,pileup )',
235  dest='what',
236  default='all')
237 
238  parser.add_argument('--step1',
239  help='Used with --raw. Limit the production to step1',
240  dest='step1Only',
241  default=False)
242 
243  parser.add_argument('--maxSteps',
244  help='Only run maximum on maxSteps. Used when we are only interested in first n steps.',
245  dest='maxSteps',
246  default=9999,
247  type=int)
248 
249  parser.add_argument('--fromScratch',
250  help='Comma separated list of wf to be run without recycling. all is not supported as default.',
251  dest='fromScratch',
252  type=lambda x: x.split(','),
253  default=None)
254 
255  parser.add_argument('--refRelease',
256  help='Allow to modify the recycling dataset version',
257  dest='refRel',
258  default=None)
259 
260  parser.add_argument('--wmcontrol',
261  help='Create the workflows for injection to WMAgent. In the WORKING. -wmcontrol init will create the the workflows, -wmcontrol test will dryRun a test, -wmcontrol submit will submit to wmagent',
262  choices=['init','test','submit','force'],
263  dest='wmcontrol',
264  default=None)
265 
266  parser.add_argument('--revertDqmio',
267  help='When submitting workflows to wmcontrol, force DQM outout to use pool and not DQMIO',
268  choices=['yes','no'],
269  dest='revertDqmio',
270  default='no')
271 
272  parser.add_argument('--optionswm',
273  help='Specify a few things for wm injection',
274  default='',
275  dest='wmoptions')
276 
277  parser.add_argument('--keep',
278  help='allow to specify for which comma separated steps the output is needed',
279  default=None)
280 
281  parser.add_argument('--label',
282  help='allow to give a special label to the output dataset name',
283  default='')
284 
285  parser.add_argument('--command',
286  help='provide a way to add additional command to all of the cmsDriver commands in the matrix',
287  dest='command',
288  action='append',
289  default=None)
290 
291  parser.add_argument('--apply',
292  help='allow to use the --command only for 1 comma separeated',
293  dest='apply',
294  default=None)
295 
296  parser.add_argument('--workflow',
297  help='define a workflow to be created or altered from the matrix',
298  action='append',
299  dest='workflow',
300  default=None)
301 
302  parser.add_argument('--dryRun',
303  help='do not run the wf at all',
304  action='store_true',
305  dest='dryRun',
306  default=False)
307 
308  parser.add_argument('--testbed',
309  help='workflow injection to cmswebtest (you need dedicated rqmgr account)',
310  dest='testbed',
311  default=False,
312  action='store_true')
313 
314  parser.add_argument('--noCafVeto',
315  help='Run from any source, ignoring the CAF label',
316  dest='cafVeto',
317  default=True,
318  action='store_false')
319 
320  parser.add_argument('--overWrite',
321  help='Change the content of a step for another. List of pairs.',
322  dest='overWrite',
323  default=None)
324 
325  parser.add_argument('--noRun',
326  help='Remove all run list selection from wfs',
327  dest='noRun',
328  default=False,
329  action='store_true')
330 
331  parser.add_argument('--das-options',
332  help='Options to be passed to dasgoclient.',
333  dest='dasOptions',
334  default="--limit 0",
335  action='store')
336 
337  parser.add_argument('--job-reports',
338  help='Dump framework job reports',
339  dest='jobReports',
340  default=False,
341  action='store_true')
342 
343  parser.add_argument('--ibeos',
344  help='Use IB EOS site configuration',
345  dest='IBEos',
346  default=False,
347  action='store_true')
348 
349  parser.add_argument('--sites',
350  help='Run DAS query to get data from a specific site. Set it to empty string to search all sites.',
351  dest='dasSites',
352  default='T2_CH_CERN',
353  action='store')
354 
355  parser.add_argument('--interactive',
356  help="Open the Matrix interactive shell",
357  action='store_true',
358  default=False)
359 
360  parser.add_argument('--dbs-url',
361  help='Overwrite DbsUrl value in JSON submitted to ReqMgr2',
362  dest='dbsUrl',
363  default=None,
364  action='store')
365 
366  gpugroup = parser.add_argument_group('GPU-related options','These options are only meaningful when --gpu is used, and is not set to forbidden.')
367 
368  gpugroup.add_argument('--gpu','--requires-gpu',
369  help='Enable GPU workflows. Possible options are "forbidden" (default), "required" (implied if no argument is given), or "optional".',
370  dest='gpu',
371  choices=['forbidden', 'optional', 'required'],
372  nargs='?',
373  const='required',
374  default='forbidden',
375  action='store')
376 
377  gpugroup.add_argument('--gpu-memory',
378  help='Specify the minimum amount of GPU memory required by the job, in MB.',
379  dest='GPUMemoryMB',
380  type=int,
381  default=8000)
382 
383  gpugroup.add_argument('--cuda-capabilities',
384  help='Specify a comma-separated list of CUDA "compute capabilities", or GPU hardware architectures, that the job can use.',
385  dest='CUDACapabilities',
386  type=lambda x: x.split(','),
387  default='6.0,6.1,6.2,7.0,7.2,7.5,8.0,8.6')
388 
389  # read the CUDA runtime version included in CMSSW
390  cudart_version = None
391  libcudart = os.path.realpath(os.path.expandvars('$CMSSW_RELEASE_BASE/external/$SCRAM_ARCH/lib/libcudart.so'))
392  if os.path.isfile(libcudart):
393  cudart_basename = os.path.basename(libcudart)
394  cudart_version = '.'.join(cudart_basename.split('.')[2:4])
395  gpugroup.add_argument('--cuda-runtime',
396  help='Specify major and minor version of the CUDA runtime used to build the application.',
397  dest='CUDARuntime',
398  default=cudart_version)
399 
400  gpugroup.add_argument('--force-gpu-name',
401  help='Request a specific GPU model, e.g. "Tesla T4" or "NVIDIA GeForce RTX 2080". The default behaviour is to accept any supported GPU.',
402  dest='GPUName',
403  default='')
404 
405  gpugroup.add_argument('--force-cuda-driver-version',
406  help='Request a specific CUDA driver version, e.g. 470.57.02. The default behaviour is to accept any supported CUDA driver version.',
407  dest='CUDADriverVersion',
408  default='')
409 
410  gpugroup.add_argument('--force-cuda-runtime-version',
411  help='Request a specific CUDA runtime version, e.g. 11.4. The default behaviour is to accept any supported CUDA runtime version.',
412  dest='CUDARuntimeVersion',
413  default='')
414 
415  opt = parser.parse_args()
416  if opt.command: opt.command = ' '.join(opt.command)
417  os.environ["CMSSW_DAS_QUERY_SITES"]=opt.dasSites
418  if opt.failed_from:
419  rerunthese=[]
420  with open(opt.failed_from,'r') as report:
421  for report_line in report:
422  if 'FAILED' in report_line:
423  to_run,_=report_line.split('_',1)
424  rerunthese.append(to_run)
425  if opt.testList:
426  opt.testList+=','.join(['']+rerunthese)
427  else:
428  opt.testList = ','.join(rerunthese)
429 
430  if opt.IBEos:
431  from subprocess import getstatusoutput as run_cmd
432 
433  ibeos_cache = os.path.join(os.getenv("LOCALRT"), "ibeos_cache.txt")
434  if not os.path.exists(ibeos_cache):
435  err, out = run_cmd("curl -L -s -o %s https://raw.githubusercontent.com/cms-sw/cms-sw.github.io/master/das_queries/ibeos.txt" % ibeos_cache)
436  if err:
437  run_cmd("rm -f %s" % ibeos_cache)
438  print("Error: Unable to download ibeos cache information")
439  print(out)
440  sys.exit(err)
441 
442  for cmssw_env in [ "CMSSW_BASE", "CMSSW_RELEASE_BASE" ]:
443  cmssw_base = os.getenv(cmssw_env,None)
444  if not cmssw_base: continue
445  cmssw_base = os.path.join(cmssw_base,"src/Utilities/General/ibeos")
446  if os.path.exists(cmssw_base):
447  os.environ["PATH"]=cmssw_base+":"+os.getenv("PATH")
448  os.environ["CMS_PATH"]="/cvmfs/cms-ib.cern.ch"
449  os.environ["SITECONFIG_PATH"]="/cvmfs/cms-ib.cern.ch/SITECONF/local"
450  os.environ["CMSSW_USE_IBEOS"]="true"
451  print(">> WARNING: You are using SITECONF from /cvmfs/cms-ib.cern.ch")
452  break
453  if opt.restricted:
454  print('Deprecated, please use -l limited')
455  if opt.testList: opt.testList+=',limited'
456  else: opt.testList='limited'
457 
458  def stepOrIndex(s):
459  if s.isdigit():
460  return int(s)
461  else:
462  return s
463  if opt.apply:
464  opt.apply=map(stepOrIndex,opt.apply.split(','))
465  if opt.keep:
466  opt.keep=map(stepOrIndex,opt.keep.split(','))
467 
468  if opt.testList:
469  testList=[]
470  for entry in opt.testList.split(','):
471  if not entry: continue
472  mapped=False
473  for k in predefinedSet:
474  if k.lower().startswith(entry.lower()) or k.lower().endswith(entry.lower()):
475  testList.extend(predefinedSet[k])
476  mapped=True
477  break
478  if not mapped:
479  try:
480  testList.append(float(entry))
481  except:
482  print(entry,'is not a possible selected entry')
483 
484  opt.testList = list(set(testList))
485 
486  if opt.wmcontrol:
488  if opt.overWrite:
489  opt.overWrite=eval(opt.overWrite)
490  if opt.interactive:
491  import cmd
492  from colorama import Fore, Style
493  from os import isatty
494  import subprocess
495  import time
496 
497  class TheMatrix(cmd.Cmd):
498  intro = "Welcome to the Matrix (? for help)"
499  prompt = "matrix> "
500 
501  def __init__(self, opt):
502  cmd.Cmd.__init__(self)
503  self.opt_ = opt
504  self.matrices_ = {}
505  tmp = MatrixReader(self.opt_)
506  self.processes_ = dict()
507  for what in tmp.files:
508  what = what.replace('relval_','')
509  self.opt_.what = what
510  self.matrices_[what] = MatrixReader(self.opt_)
511  self.matrices_[what].prepare(self.opt_.useInput, self.opt_.refRel,
512  self.opt_.fromScratch)
513  os.system("clear")
514 
515  def do_clear(self, arg):
516  """Clear the screen, put prompt at the top"""
517  os.system("clear")
518 
519  def do_exit(self, arg):
520  print("Leaving the Matrix")
521  return True
522 
523  def default(self, inp):
524  if inp == 'x' or inp == 'q':
525  return self.do_exit(inp)
526  else:
527  is_pipe = not isatty(sys.stdin.fileno())
528  print(Fore.RED + "Error: " + Fore.RESET + "unrecognized command.")
529  # Quit only if given a piped command.
530  if is_pipe:
531  sys.exit(1)
532 
533  def help_predefined(self):
534  print("\n".join(["predefined [predef1 [...]]\n",
535  "Run w/o argument, it will print the list of known predefined workflows.",
536  "Run with space-separated predefined workflows, it will print the workflow-ids registered to them"]))
537 
538  def complete_predefined(self, text, line, start_idx, end_idx):
539  if text and len(text) > 0:
540  return [t for t in predefinedSet.keys() if t.startswith(text)]
541  else:
542  return predefinedSet.keys()
543 
544  def do_predefined(self, arg):
545  """Print the list of predefined workflows"""
546  print("List of predefined workflows")
547  if arg:
548  for w in arg.split():
549  if w in predefinedSet.keys():
550  print("Predefined Set: %s" % w)
551  print(predefinedSet[w])
552  else:
553  print("Unknown Set: %s" % w)
554  else:
555  print("[ " + Fore.RED + ", ".join([str(k) for k in predefinedSet.keys()]) + Fore.RESET + " ]")
556 
557  def help_showWorkflow(self):
558  print("\n".join(["showWorkflow [workflow1 [...]]\n",
559  "Run w/o arguments, it will print the list of registered macro-workflows.",
560  "Run with space-separated workflows, it will print the full list of workflow-ids registered to them"]))
561 
562  def complete_showWorkflow(self, text, line, start_idx, end_idx):
563  if text and len(text) > 0:
564  return [t for t in self.matrices_.keys() if t.startswith(text)]
565  else:
566  return self.matrices_.keys()
567 
568  def do_showWorkflow(self, arg):
569  if arg == '':
570  print("Available workflows:")
571  for k in self.matrices_.keys():
572  print(Fore.RED + Style.BRIGHT + k)
573  print(Style.RESET_ALL)
574  else:
575  selected = arg.split()
576  for k in selected:
577  if k not in self.matrices_.keys():
578  print("Unknown workflow %s: skipping" % k)
579  else:
580  for wfl in self.matrices_[k].workFlows:
581  print("%s %s" % (Fore.BLUE + str(wfl.numId) + Fore.RESET,
582  Fore.GREEN + wfl.nameId + Fore.RESET))
583  print("%s contains %d workflows" % (Fore.RED + k + Fore.RESET, len(self.matrices_[k].workFlows)))
584 
585  def do_runWorkflow(self, arg):
586  # Split the input arguments into a list
587  args = arg.split()
588  if len(args) < 2:
589  print(Fore.RED + Style.BRIGHT + "Wrong number of parameters passed")
590  print(Style.RESET_ALL)
591  return
592  workflow_class = args[0]
593  workflow_id = args[1]
594  passed_down_args = list()
595  if len(args) > 2:
596  passed_down_args = args[2:]
597  print(Fore.YELLOW + Style.BRIGHT + "Running with the following options:\n")
598  print(Fore.GREEN + Style.BRIGHT + "Workflow class: {}".format(workflow_class))
599  print(Fore.GREEN + Style.BRIGHT + "Workflow ID: {}".format(workflow_id))
600  print(Fore.GREEN + Style.BRIGHT + "Additional runTheMatrix options: {}".format(passed_down_args))
601  print(Style.RESET_ALL)
602  if workflow_class not in self.matrices_.keys():
603  print(Fore.RED + Style.BRIGHT + "Unknown workflow selected: {}".format(workflow_class))
604  print("Available workflows:")
605  for k in self.matrices_.keys():
606  print(Fore.RED + Style.BRIGHT + k)
607  print(Style.RESET_ALL)
608  return
609  wflnums = [x.numId for x in self.matrices_[workflow_class].workFlows]
610  if float(workflow_id) not in wflnums:
611  print(Fore.RED + Style.BRIGHT + "Unknown workflow {}".format(workflow_id))
612  print(Fore.GREEN + Style.BRIGHT)
613  print(wflnums)
614  print(Style.RESET_ALL)
615  return
616  if workflow_id in self.processes_.keys():
617  # Check if the process is still active
618  if self.processes_[workflow_id][0].poll() is None:
619  print(Fore.RED + Style.BRIGHT + "Workflow {} already running!".format(workflow_id))
620  print(Style.RESET_ALL)
621  return
622  # If it was there but it's gone, proceeed and update the value for the same key
623  # run a job, redirecting standard output and error to files
624  lognames = ['stdout', 'stderr']
625  logfiles = tuple('%s_%s_%s.log' % (workflow_class, workflow_id, name) for name in lognames)
626  stdout = open(logfiles[0], 'w')
627  stderr = open(logfiles[1], 'w')
628  command = ('runTheMatrix.py', '-w', workflow_class, '-l', workflow_id)
629  if len(passed_down_args) > 0:
630  command += tuple(passed_down_args)
631  print(command)
632  p = subprocess.Popen(command,
633  stdout = stdout,
634  stderr = stderr)
635  self.processes_[workflow_id] = (p, time.time())
636 
637 
638  def complete_runWorkflow(self, text, line, start_idx, end_idx):
639  if text and len(text) > 0:
640  return [t for t in self.matrices_.keys() if t.startswith(text)]
641  else:
642  return self.matrices_.keys()
643 
644  def help_runWorkflow(self):
645  print("\n".join(["runWorkflow workflow_class workflow_id\n",
646  "This command will launch a new and independent process that invokes",
647  "the command:\n",
648  "runTheMatrix.py -w workflow_class -l workflow_id [runTheMatrix.py options]",
649  "\nYou can specify just one workflow_class and workflow_id per invocation.",
650  "The job will continue even after quitting the interactive session.",
651  "stdout and stderr of the new process will be automatically",
652  "redirected to 2 logfiles whose names contain the workflow_class",
653  "and workflow_id. Mutiple command can be issued one after the other.",
654  "The working directory of the new process will be the directory",
655  "from which the interactive session has started.",
656  "Autocompletion is available for workflow_class, but",
657  "not for workflow_id. Supplying a wrong workflow_class or",
658  "a non-existing workflow_id for a valid workflow_class",
659  "will trigger an error and no process will be invoked.",
660  "The interactive shell will keep track of all active processes",
661  "and will prevent the accidental resubmission of an already",
662  "active jobs."]))
663 
664  def do_jobs(self, args):
665  print(Fore.GREEN + Style.BRIGHT + "List of jobs:")
666  for w in self.processes_.keys():
667  if self.processes_[w][0].poll() is None:
668  print(Fore.YELLOW + Style.BRIGHT + "Active job: {} since {:.2f} seconds.".format(w, time.time() - self.processes_[w][1]))
669  else:
670  print(Fore.RED + Style.BRIGHT + "Done job: {}".format(w))
671  print(Style.RESET_ALL)
672 
673  def help_jobs(self):
674  print("\n".join(["Print a full list of active and done jobs submitted",
675  "in the ongoing interactive session"]))
676 
678  print("\n".join(["searchInWorkflow wfl_name search_regexp\n",
679  "This command will search for a match within all workflows registered to wfl_name.",
680  "The search is done on both the workflow name and the names of steps registered to it."]))
681 
682  def complete_searchInWorkflow(self, text, line, start_idx, end_idx):
683  if text and len(text) > 0:
684  return [t for t in self.matrices_.keys() if t.startswith(text)]
685  else:
686  return self.matrices_.keys()
687 
688  def do_searchInWorkflow(self, arg):
689  args = arg.split()
690  if len(args) < 2:
691  print("searchInWorkflow name regexp")
692  return
693  if args[0] not in self.matrices_.keys():
694  print("Unknown workflow")
695  return
696  import re
697  pattern = None
698  try:
699  pattern = re.compile(args[1])
700  except:
701  print("Failed to compile regexp %s" % args[1])
702  return
703  counter = 0
704  for wfl in self.matrices_[args[0]].workFlows:
705  if re.match(pattern, wfl.nameId):
706  print("%s %s" % (Fore.BLUE + str(wfl.numId) + Fore.RESET,
707  Fore.GREEN + wfl.nameId + Fore.RESET))
708  counter +=1
709  print("Found %s compatible workflows inside %s" % (Fore.RED + str(counter) + Fore.RESET,
710  Fore.YELLOW + str(args[0])) + Fore.RESET)
711 
712  def help_search(self):
713  print("\n".join(["search search_regexp\n",
714  "This command will search for a match within all workflows registered.",
715  "The search is done on both the workflow name and the names of steps registered to it."]))
716 
717  def do_search(self, arg):
718  args = arg.split()
719  if len(args) < 1:
720  print("search regexp")
721  return
722  for wfl in self.matrices_.keys():
723  self.do_searchInWorkflow(' '.join([wfl, args[0]]))
724 
726  print("\n".join(["dumpWorkflowId [wfl-id1 [...]]\n",
727  "Dumps the details (cmsDriver commands for all steps) of the space-separated workflow-ids in input."]))
728 
729  def do_dumpWorkflowId(self, arg):
730  wflids = arg.split()
731  if len(wflids) == 0:
732  print("dumpWorkflowId [wfl-id1 [...]]")
733  return
734 
735  fmt = "[%s]: %s\n"
736  maxLen = 100
737  for wflid in wflids:
738  dump = True
739  for key, mrd in self.matrices_.items():
740  for wfl in mrd.workFlows:
741  if wfl.numId == float(wflid):
742  if dump:
743  dump = False
744  print(Fore.GREEN + str(wfl.numId) + Fore.RESET + " " + Fore.YELLOW + wfl.nameId + Fore.RESET)
745  for i,s in enumerate(wfl.cmds):
746  print(fmt % (Fore.RED + str(i+1) + Fore.RESET,
747  (str(s)+' ')))
748  print("\nWorkflow found in %s." % key)
749  else:
750  print("Workflow also found in %s." % key)
751 
752  do_EOF = do_exit
753 
754  TheMatrix(opt).cmdloop()
755  sys.exit(0)
756 
757  if opt.raw and opt.show:
758  ret = showRaw(opt)
759  else:
760  ret = runSelected(opt)
761 
762 
763  sys.exit(ret)
def performInjectionOptionTest(opt)
def runSelected(opt)
Definition: runTheMatrix.py:20
def do_predefined(self, arg)
def do_showWorkflow(self, arg)
def complete_showWorkflow(self, text, line, start_idx, end_idx)
def do_search(self, arg)
def showRaw(opt)
Definition: runTheMatrix.py:11
def do_clear(self, arg)
def help_dumpWorkflowId(self)
def complete_predefined(self, text, line, start_idx, end_idx)
def help_searchInWorkflow(self)
def do_dumpWorkflowId(self, arg)
void print(TMatrixD &m, const char *label=nullptr, bool mathematicaFormat=false)
Definition: Utilities.cc:47
def __init__(self, opt)
static std::string join(char **cmd)
Definition: RemoteFile.cc:21
def do_exit(self, arg)
def default(self, inp)
def complete_searchInWorkflow(self, text, line, start_idx, end_idx)
def stepOrIndex(s)
def complete_runWorkflow(self, text, line, start_idx, end_idx)
def do_runWorkflow(self, arg)
#define str(s)
def do_jobs(self, args)
def do_searchInWorkflow(self, arg)