Public Member Functions
def	__init__ (self, stage_element, base_stage, default_input_lists, default_previous_stage, default_num_jobs, default_num_events, default_max_files_per_job, default_merge, default_anamerge, default_cpu, default_disk, default_memory, default_validate_on_worker, default_copy_to_fts, default_cvmfs, default_stash, default_singularity, default_script, default_start_script, default_stop_script, default_site, default_blacklist, check=True)

def	__str__ (self)

def	pubsify_input (self, run, subruns, version)

def	pubsify_output (self, run, subruns, version)

def	checksubmit (self)

def	checkinput (self, checkdef=False)

def	check_output_dirs (self)

def	checkdirs (self)

def	makedirs (self)

Public Attributes
	name

	batchname

	fclname

	outdir

	logdir

	workdir

	bookdir

	dirsize

	dirlevels

	dynamic

	inputfile

	inputlist

	inputmode

	basedef

	inputdef

	inputstream

	previousstage

	mixinputdef

	pubs_input_ok

	pubs_input

	input_run

	input_subruns

	input_version

	pubs_output

	output_run

	output_subruns

	output_version

	ana

	recur

	recurtype

	recurlimit

	singlerun

	filelistdef

	prestart

	activebase

	dropboxwait

	prestagefraction

	maxfluxfilemb

	num_jobs

	num_events

	max_files_per_job

	target_size

	defname

	ana_defname

	data_tier

	data_stream

	ana_data_tier

	ana_data_stream

	submit_script

	init_script

	init_source

	end_script

	mid_source

	mid_script

	project_name

	stage_name

	project_version

	merge

	anamerge

	resource

	lines

	site

	blacklist

	cpu

	disk

	datafiletypes

	memory

	parameters

	output

	TFileName

	jobsub

	jobsub_start

	jobsub_timeout

	exe

	schema

	validate_on_worker

	copy_to_fts

	cvmfs

	stash

	singularity

	script

	start_script

	stop_script

Detailed Description

Definition at line 33 of file stagedef.py.

Constructor & Destructor Documentation

def python.stagedef.StageDef.__init__	(	self,
		stage_element,
		base_stage,
		default_input_lists,
		default_previous_stage,
		default_num_jobs,
		default_num_events,
		default_max_files_per_job,
		default_merge,
		default_anamerge,
		default_cpu,
		default_disk,
		default_memory,
		default_validate_on_worker,
		default_copy_to_fts,
		default_cvmfs,
		default_stash,
		default_singularity,
		default_script,
		default_start_script,
		default_stop_script,
		default_site,
		default_blacklist,
		check = `True`
	)

Definition at line 43 of file stagedef.py.

                  default_site, default_blacklist, check=True):
 
         # Assign default values.
 
         if base_stage != None:
             self.name = base_stage.name
             self.batchname = base_stage.batchname
             self.fclname = base_stage.fclname
             self.outdir = base_stage.outdir
             self.logdir = base_stage.logdir
             self.workdir = base_stage.workdir
             self.bookdir = base_stage.bookdir
             self.dirsize = base_stage.dirsize
             self.dirlevels = base_stage.dirlevels
             self.dynamic = base_stage.dynamic
             self.inputfile = base_stage.inputfile
             self.inputlist = base_stage.inputlist
             self.inputmode = base_stage.inputmode
             self.basedef = base_stage.basedef
             self.inputdef = base_stage.inputdef
             self.inputstream = base_stage.inputstream
             self.previousstage = base_stage.previousstage
             self.mixinputdef = base_stage.mixinputdef
             self.pubs_input_ok = base_stage.pubs_input_ok
             self.pubs_input = base_stage.pubs_input
             self.input_run = base_stage.input_run
             self.input_subruns = base_stage.input_subruns
             self.input_version = base_stage.input_version
             self.pubs_output = base_stage.pubs_output
             self.output_run = base_stage.output_run
             self.output_subruns = base_stage.output_subruns
             self.output_version = base_stage.output_version
             self.ana = base_stage.ana
             self.recur = base_stage.recur
             self.recurtype = base_stage.recurtype
             self.recurlimit = base_stage.recurlimit
             self.singlerun = base_stage.singlerun
             self.filelistdef = base_stage.filelistdef
             self.prestart = base_stage.prestart
             self.activebase = base_stage.activebase
             self.dropboxwait = base_stage.dropboxwait
             self.prestagefraction = base_stage.prestagefraction
             self.maxfluxfilemb = base_stage.maxfluxfilemb
             self.num_jobs = base_stage.num_jobs
             self.num_events = base_stage.num_events
             self.max_files_per_job = base_stage.max_files_per_job
             self.target_size = base_stage.target_size
             self.defname = base_stage.defname
             self.ana_defname = base_stage.ana_defname
             self.data_tier = base_stage.data_tier
             self.data_stream = base_stage.data_stream
             self.ana_data_tier = base_stage.ana_data_tier
             self.ana_data_stream = base_stage.ana_data_stream
             self.submit_script = base_stage.submit_script
             self.init_script = base_stage.init_script
             self.init_source = base_stage.init_source
             self.end_script = base_stage.end_script
             self.mid_source = base_stage.mid_source
             self.mid_script = base_stage.mid_script
             self.project_name = base_stage.project_name
             self.stage_name = base_stage.stage_name
             self.project_version = base_stage.project_version
             self.merge = base_stage.merge
             self.anamerge = base_stage.anamerge
             self.resource = base_stage.resource
             self.lines = base_stage.lines
             self.site = base_stage.site
             self.blacklist = base_stage.blacklist
             self.cpu = base_stage.cpu
             self.disk = base_stage.disk
             self.datafiletypes = base_stage.datafiletypes
             self.memory = base_stage.memory
             self.parameters = base_stage.parameters
             self.output = base_stage.output
             self.TFileName = base_stage.TFileName
             self.jobsub = base_stage.jobsub
             self.jobsub_start = base_stage.jobsub_start
             self.jobsub_timeout = base_stage.jobsub_timeout
             self.exe = base_stage.exe
             self.schema = base_stage.schema
             self.validate_on_worker = base_stage. validate_on_worker
             self.copy_to_fts = base_stage.copy_to_fts
             self.cvmfs = base_stage.cvmfs
             self.stash = base_stage.stash
             self.singularity = base_stage.singularity
             self.script = base_stage.script
             self.start_script = base_stage.start_script
             self.stop_script = base_stage.stop_script
         else:
             self.name = ''         # Stage name.
             self.batchname = ''    # Batch job name
             self.fclname = []
             self.outdir = ''       # Output directory.
             self.logdir = ''       # Log directory.
             self.workdir = ''      # Work directory.
             self.bookdir = ''      # Bookkeeping directory.
             self.dirsize = 0       # Maximum directory size.
             self.dirlevels = 0     # Number of extra directory levels.
             self.dynamic = 0       # Dynamic output/log directory.
             self.inputfile = ''    # Single input file.
             self.inputlist = ''    # Input file list.
             self.inputmode = ''    # Input file type (none or textfile)
             self.basedef = ''      # Base sam dataset definition.
             self.inputdef = ''     # Input sam dataset definition.
             self.inputstream = ''  # Input file stream.
             self.previousstage = '' # Previous stage name.
             self.mixinputdef = ''  # Mix input sam dataset definition.
             self.pubs_input_ok = 1 # Is pubs input allowed?
             self.pubs_input = 0    # Pubs input mode.
             self.input_run = 0     # Pubs input run.
             self.input_subruns = [] # Pubs input subrun number(s).
             self.input_version = 0 # Pubs input version number.
             self.pubs_output = 0   # Pubs output mode.
             self.output_run = 0    # Pubs output run.
             self.output_subruns = [] # Pubs output subrun number.
             self.output_version = 0 # Pubs output version number.
             self.ana = 0           # Analysis flag.
             self.recur = 0         # Recursive flag.
             self.recurtype = ''    # Recursive type.
             self.recurlimit = 0    # Recursive limit.
             self.singlerun=0       # Single run mode.
             self.filelistdef=0     # Convert sam input def to file list.
             self.prestart = 0      # Prestart flag.
             self.activebase = ''   # Active projects base name.
             self.dropboxwait = 0.  # Dropbox waiting interval.
             self.prestagefraction = 0.  # Prestage fraction.
             self.maxfluxfilemb = 0 # MaxFluxFileMB (size of genie flux files to fetch).
             self.num_jobs = default_num_jobs # Number of jobs.
             self.num_events = default_num_events # Number of events.
             self.max_files_per_job = default_max_files_per_job #max num of files per job
             self.target_size = 0   # Target size for output files.
             self.defname = ''      # Sam dataset definition name.
             self.ana_defname = ''  # Sam dataset definition name.
             self.data_tier = ''    # Sam data tier.
             self.data_stream = []  # Sam data stream.
             self.ana_data_tier = '' # Sam data tier.
             self.ana_data_stream = [] # Sam data stream.
             self.submit_script = '' # Submit script.
             self.init_script = []  # Worker initialization script.
             self.init_source = []  # Worker initialization bash source script.
             self.end_script = []   # Worker end-of-job script.
             self.mid_source = {}   # Worker midstage source init scripts.
             self.mid_script = {}   # Worker midstage finalization scripts.
             self.project_name = [] # Project name overrides.
             self.stage_name = []   # Stage name overrides.
             self.project_version = [] # Project version overrides.
             self.merge = default_merge    # Histogram merging program
             self.anamerge = default_anamerge    # Analysis merge flag.
             self.resource = ''     # Jobsub resources.
             self.lines = ''        # Arbitrary condor commands.
             self.site = default_site # Site.
             self.blacklist = default_blacklist # Blacklist site.
             self.cpu = default_cpu # Number of cpus.
             self.disk = default_disk     # Disk space (string value+unit).
             self.datafiletypes = ["root"] # Data file types.
             self.memory = default_memory # Amount of memory (integer MB).
             self.parameters = {}   # Dictionary of metadata parameters.
             self.output = []       # Art output file names.
             self.TFileName = ''    # TFile output file name.
             self.jobsub = ''       # Arbitrary jobsub_submit options.
             self.jobsub_start = '' # Arbitrary jobsub_submit options for sam start/stop jobs.
             self.jobsub_timeout = 0 # Jobsub submit timeout.
             self.exe = []          # Art-like executables.
             self.schema = ''       # Sam schema.
             self.validate_on_worker = default_validate_on_worker # Validate-on-worker flag.
             self.copy_to_fts = default_copy_to_fts   # Upload-on-worker flag.
             self.cvmfs = default_cvmfs               # Default cvmfs flag.
             self.stash = default_stash               # Default stash flag.
             self.singularity = default_singularity   # Default singularity flag.
             self.script = default_script             # Upload-on-worker flag.
             self.start_script = default_start_script # Upload-on-worker flag.
             self.stop_script = default_stop_script   # Upload-on-worker flag.
         
         # Extract values from xml.
 
         # Stage name (attribute).
 
         if 'name' in dict(stage_element.attributes):
             self.name = str(stage_element.attributes['name'].firstChild.data)
         if self.name == '':
             raise XMLError("Stage name not specified.")
 
         # Batch job name (subelement).
 
         batchname_elements = stage_element.getElementsByTagName('batchname')
         if batchname_elements:
             self.batchname = str(batchname_elements[0].firstChild.data)
 
         # Fcl file name (repeatable subelement).
 
         fclname_elements = stage_element.getElementsByTagName('fcl')
         if len(fclname_elements) > 0:
             self.fclname = []
             for fcl in fclname_elements:
                 self.fclname.append(str(fcl.firstChild.data).strip())
         if len(self.fclname) == 0:
             raise XMLError('No Fcl names specified for stage %s.' % self.name)
 
         # Output directory (subelement).
 
         outdir_elements = stage_element.getElementsByTagName('outdir')
         if outdir_elements:
             self.outdir = str(outdir_elements[0].firstChild.data)
         if self.outdir == '':
             raise XMLError('Output directory not specified for stage %s.' % self.name)
 
         # Log directory (subelement).
 
         logdir_elements = stage_element.getElementsByTagName('logdir')
         if logdir_elements:
             self.logdir = str(logdir_elements[0].firstChild.data)
         if self.logdir == '':
             self.logdir = self.outdir
 
         # Work directory (subelement).
 
         workdir_elements = stage_element.getElementsByTagName('workdir')
         if workdir_elements:
             self.workdir = str(workdir_elements[0].firstChild.data)
         if self.workdir == '':
             raise XMLError('Work directory not specified for stage %s.' % self.name)
 
         # Bookkeeping directory (subelement).
 
         bookdir_elements = stage_element.getElementsByTagName('bookdir')
         if bookdir_elements:
             self.bookdir = str(bookdir_elements[0].firstChild.data)
         if self.bookdir == '':
             self.bookdir = self.logdir
 
         # Maximum directory size (subelement).
 
         dirsize_elements = stage_element.getElementsByTagName('dirsize')
         if dirsize_elements:
             self.dirsize = int(dirsize_elements[0].firstChild.data)
 
         # Extra directory levels (subelement).
 
         dirlevels_elements = stage_element.getElementsByTagName('dirlevels')
         if dirlevels_elements:
             self.dirlevels = int(dirlevels_elements[0].firstChild.data)
 
         # Single input file (subelement).
 
         inputfile_elements = stage_element.getElementsByTagName('inputfile')
         if inputfile_elements:
             self.inputfile = str(inputfile_elements[0].firstChild.data)
 
         # Input file list (subelement).
 
         inputlist_elements = stage_element.getElementsByTagName('inputlist')
         if inputlist_elements:
             self.inputlist = str(inputlist_elements[0].firstChild.data)
 
         # Input file type (subelement).
 
         inputmode_elements = stage_element.getElementsByTagName('inputmode')
         if inputmode_elements:
             self.inputmode = str(inputmode_elements[0].firstChild.data)
 
         # Input sam dataset dfeinition (subelement).
 
         inputdef_elements = stage_element.getElementsByTagName('inputdef')
         if inputdef_elements:
             self.inputdef = str(inputdef_elements[0].firstChild.data)
 
         # Analysis flag (subelement).
 
         ana_elements = stage_element.getElementsByTagName('ana')
         if ana_elements:
             self.ana = int(ana_elements[0].firstChild.data)
 
         # Recursive flag (subelement).
 
         recur_elements = stage_element.getElementsByTagName('recur')
         if recur_elements:
             self.recur = int(recur_elements[0].firstChild.data)
 
         # Recursive type (subelement).
 
         recurtype_elements = stage_element.getElementsByTagName('recurtype')
         if recurtype_elements:
             self.recurtype = str(recurtype_elements[0].firstChild.data)
 
         # Recursive limit (subelement).
 
         recurlimit_elements = stage_element.getElementsByTagName('recurlimit')
         if recurlimit_elements:
             self.recurlimit = int(recurlimit_elements[0].firstChild.data)
 
         # Recursive input sam dataset dfeinition (subelement).
 
         recurdef_elements = stage_element.getElementsByTagName('recurdef')
         if recurdef_elements:
             self.basedef = self.inputdef
             self.inputdef = str(recurdef_elements[0].firstChild.data)
             self.recur = 1
 
         # Single run flag (subelement).
 
         singlerun_elements = stage_element.getElementsByTagName('singlerun')
         if singlerun_elements:
             self.singlerun = int(singlerun_elements[0].firstChild.data)
 
         # File list definition flag (subelement).
 
         filelistdef_elements = stage_element.getElementsByTagName('filelistdef')
         if filelistdef_elements:
             self.filelistdef = int(filelistdef_elements[0].firstChild.data)
 
         # Prestart flag.
 
         prestart_elements = stage_element.getElementsByTagName('prestart')
         if prestart_elements:
             self.prestart = int(prestart_elements[0].firstChild.data)
 
         # Active projects basename.
 
         activebase_elements = stage_element.getElementsByTagName('activebase')
         if activebase_elements:
             self.activebase = str(activebase_elements[0].firstChild.data)
 
         # Dropbox wait interval.
 
         dropboxwait_elements = stage_element.getElementsByTagName('dropboxwait')
         if dropboxwait_elements:
             self.dropboxwait = float(dropboxwait_elements[0].firstChild.data)
 
         # Prestage fraction (subelement).
 
         prestagefraction_elements = stage_element.getElementsByTagName('prestagefraction')
         if prestagefraction_elements:
             self.prestagefraction = float(prestagefraction_elements[0].firstChild.data)
 
         # Input stream (subelement).
 
         inputstream_elements = stage_element.getElementsByTagName('inputstream')
         if inputstream_elements:
             self.inputstream = str(inputstream_elements[0].firstChild.data)
 
         # Previous stage name (subelement).
 
         previousstage_elements = stage_element.getElementsByTagName('previousstage')
         if previousstage_elements:
             self.previousstage = str(previousstage_elements[0].firstChild.data)
 
             # If a base stage was specified, nullify any input inherted from base.
 
             if base_stage != None:
                 self.inputfile = ''
                 self.inputlist = ''
                 self.inputdef = ''
 
             # It never makes sense to specify a previous stage with some other input.
 
             if self.inputfile != '' or self.inputlist != '' or self.inputdef != '':
                 raise XMLError('Previous stage and input specified for stage %s.' % self.name)
 
         # Mix input sam dataset (subelement).
 
         mixinputdef_elements = stage_element.getElementsByTagName('mixinputdef')
         if mixinputdef_elements:
             self.mixinputdef = str(mixinputdef_elements[0].firstChild.data)
 
         # It is an error to specify both input file and input list.
 
         if self.inputfile != '' and self.inputlist != '':
             raise XMLError('Input file and input list both specified for stage %s.' % self.name)
 
         # It is an error to specify either input file or input list together
         # with a sam input dataset.
 
         if self.inputdef != '' and (self.inputfile != '' or self.inputlist != ''):
             raise XMLError('Input dataset and input files specified for stage %s.' % self.name)
 
         # It is an error to use textfile inputmode without an inputlist or inputfile
         if self.inputmode == 'textfile' and self.inputlist == '' and self.inputfile == '':
             raise XMLError('Input list (inputlist) or inputfile is needed for textfile model.')
 
         # If none of input definition, input file, nor input list were specified, set
         # the input list to the dafault input list.  If an input stream was specified,
         # insert it in front of the file type.
 
         if self.inputfile == '' and self.inputlist == '' and self.inputdef == '':
 
             # Get the default input list according to the previous stage.
 
             default_input_list = ''
             previous_stage_name = default_previous_stage
             if self.previousstage != '':
                 previous_stage_name = self.previousstage
             if previous_stage_name in default_input_lists:
                 default_input_list = default_input_lists[previous_stage_name]
 
             # Modify default input list according to input stream, if any.
 
             if self.inputstream == '' or default_input_list == '':
                 self.inputlist = default_input_list
             else:
                 n = default_input_list.rfind('.')
                 if n < 0:
                     n = len(default_input_list)
                 self.inputlist = '%s_%s%s' % (default_input_list[:n],
                                                self.inputstream,
                                                default_input_list[n:])
 
         # Pubs input flag.
 
         pubs_input_ok_elements = stage_element.getElementsByTagName('pubsinput')
         if pubs_input_ok_elements:
             self.pubs_input_ok = int(pubs_input_ok_elements[0].firstChild.data)
 
         # MaxFluxFileMB GENIEHelper fcl parameter (subelement).
 
         maxfluxfilemb_elements = stage_element.getElementsByTagName('maxfluxfilemb')
         if maxfluxfilemb_elements:
             self.maxfluxfilemb = int(maxfluxfilemb_elements[0].firstChild.data)
         else:
 
             # If this is a generator job, give maxfluxfilemb parameter a default
             # nonzero value.
 
             if self.inputfile == '' and self.inputlist == '' and self.inputdef == '':
                 self.maxfluxfilemb = 500
 
         # Number of jobs (subelement).
 
         num_jobs_elements = stage_element.getElementsByTagName('numjobs')
         if num_jobs_elements:
             self.num_jobs = int(num_jobs_elements[0].firstChild.data)
 
         # Number of events (subelement).
 
         num_events_elements = stage_element.getElementsByTagName('numevents')
         if num_events_elements:
             self.num_events = int(num_events_elements[0].firstChild.data)
 
         # Max Number of files per jobs.
 
         max_files_per_job_elements = stage_element.getElementsByTagName('maxfilesperjob')
         if max_files_per_job_elements:
             self.max_files_per_job = int(max_files_per_job_elements[0].firstChild.data)
         
         # Run number of events (MC Gen only).
         #overriden by --pubs <run> is running in pubs mode
 
         run_number = stage_element.getElementsByTagName('runnumber')
         if run_number:
             self.output_run = int(run_number[0].firstChild.data)    
 
         # Target size for output files (subelement).
 
         target_size_elements = stage_element.getElementsByTagName('targetsize')
         if target_size_elements:
             self.target_size = int(target_size_elements[0].firstChild.data)
         
 
         # Sam dataset definition name (subelement).
 
         defname_elements = stage_element.getElementsByTagName('defname')
         if defname_elements:
             self.defname = str(defname_elements[0].firstChild.data)
 
         # Sam analysis dataset definition name (subelement).
 
         ana_defname_elements = stage_element.getElementsByTagName('anadefname')
         if ana_defname_elements:
             self.ana_defname = str(ana_defname_elements[0].firstChild.data)
 
         # Sam data tier (subelement).
 
         data_tier_elements = stage_element.getElementsByTagName('datatier')
         if data_tier_elements:
             self.data_tier = str(data_tier_elements[0].firstChild.data)
 
         # Sam data stream (subelement).
 
         data_stream_elements = stage_element.getElementsByTagName('datastream')
         if len(data_stream_elements) > 0:
             self.data_stream = []
             for data_stream in data_stream_elements:
                 self.data_stream.append(str(data_stream.firstChild.data))
 
         # Sam analysis data tier (subelement).
 
         ana_data_tier_elements = stage_element.getElementsByTagName('anadatatier')
         if ana_data_tier_elements:
             self.ana_data_tier = str(ana_data_tier_elements[0].firstChild.data)
 
         # Sam analysis data stream (subelement).
 
         ana_data_stream_elements = stage_element.getElementsByTagName('anadatastream')
         if len(ana_data_stream_elements) > 0:
             self.ana_data_stream = []
             for ana_data_stream in ana_data_stream_elements:
                 self.ana_data_stream.append(str(ana_data_stream.firstChild.data))
 
         # Submit script (subelement).
 
         submit_script_elements = stage_element.getElementsByTagName('submitscript')
         if submit_script_elements:
             self.submit_script = str(submit_script_elements[0].firstChild.data).split()
 
         # Make sure submit script exists, and convert into a full path.
 
         if check:
             if len(self.submit_script) > 0:
                 if larbatch_posix.exists(self.submit_script[0]):
                     self.submit_script[0] = os.path.realpath(self.submit_script[0])
                 else:
 
                     # Look for script on execution path.
 
                     try:
                         jobinfo = subprocess.Popen(['which', self.submit_script[0]],
                                                    stdout=subprocess.PIPE,
                                                    stderr=subprocess.PIPE)
                         jobout, joberr = jobinfo.communicate()
                         jobout = convert_str(jobout)
                         joberr = convert_str(joberr)
                         rc = jobinfo.poll()
                         self.submit_script[0] = jobout.splitlines()[0].strip()
                     except:
                         pass
                 if not larbatch_posix.exists(self.submit_script[0]):
                     raise IOError('Submit script %s not found.' % self.submit_script[0])
 
         # Worker initialization script (repeatable subelement).
 
         init_script_elements = stage_element.getElementsByTagName('initscript')
         if len(init_script_elements) > 0:
             for init_script_element in init_script_elements:
                 init_script = str(init_script_element.firstChild.data)
 
                 # Make sure init script exists, and convert into a full path.
 
                 if check:
                     if init_script != '':
                         if larbatch_posix.exists(init_script):
                             init_script = os.path.realpath(init_script)
                         else:
 
                             # Look for script on execution path.
 
                             try:
                                 jobinfo = subprocess.Popen(['which', init_script],
                                                            stdout=subprocess.PIPE,
                                                            stderr=subprocess.PIPE)
                                 jobout, joberr = jobinfo.communicate()
                                 rc = jobinfo.poll()
                                 init_script = convert_str(jobout.splitlines()[0].strip())
                             except:
                                 pass
 
                         if not larbatch_posix.exists(init_script):
                             raise IOError('Init script %s not found.' % init_script)
 
                     self.init_script.append(init_script)
 
         # Worker initialization source script (repeatable subelement).
 
         init_source_elements = stage_element.getElementsByTagName('initsource')
         if len(init_source_elements) > 0:
             for init_source_element in init_source_elements:
                 init_source = str(init_source_element.firstChild.data)
 
                 # Make sure init source script exists, and convert into a full path.
 
                 if init_source != '':
                     if check:
                         if larbatch_posix.exists(init_source):
                             init_source = os.path.realpath(init_source)
                         else:
 
                             # Look for script on execution path.
 
                             try:
                                 jobinfo = subprocess.Popen(['which', init_source],
                                                            stdout=subprocess.PIPE,
                                                            stderr=subprocess.PIPE)
                                 jobout, joberr = jobinfo.communicate()
                                 rc = jobinfo.poll()
                                 init_source = convert_str(jobout.splitlines()[0].strip())
                             except:
                                 pass
 
                         if not larbatch_posix.exists(init_source):
                             raise IOError('Init source script %s not found.' % init_source)
 
                     # The <initsource> element can occur at the top level of the <stage>
                     # element, or inside a <fcl> element.
                     # Update the StageDef object differently in these two cases.
 
                     parent_element = init_source_element.parentNode
                     if parent_element.nodeName == 'fcl':
 
                         # This <initsource> is located inside a <fcl> element.
                         # Find the index of this fcl file.
                         # Python will raise an exception if the fcl can't be found
                         # (shouldn't happen).
 
                         fcl = str(parent_element.firstChild.data).strip()
                         n = self.fclname.index(fcl)
                         if not n in self.mid_source:
                             self.mid_source[n] = []
                         self.mid_source[n].append(init_source)
 
                     else:
 
                         # This is a <stage> level <initsource> element.
 
                         self.init_source.append(init_source)
 
         # Worker end-of-job script (repeatable subelement).
 
         end_script_elements = stage_element.getElementsByTagName('endscript')
         if len(end_script_elements) > 0:
             for end_script_element in end_script_elements:
                 end_script = str(end_script_element.firstChild.data)
 
                 # Make sure end-of-job scripts exists, and convert into a full path.
 
                 if end_script != '':
                     if check:
                         if larbatch_posix.exists(end_script):
                             end_script = os.path.realpath(end_script)
                         else:
 
                             # Look for script on execution path.
 
                             try:
                                 jobinfo = subprocess.Popen(['which', end_script],
                                                            stdout=subprocess.PIPE,
                                                            stderr=subprocess.PIPE)
                                 jobout, joberr = jobinfo.communicate()
                                 rc = jobinfo.poll()
                                 end_script = convert_str(jobout.splitlines()[0].strip())
                             except:
                                 pass
 
                         if not larbatch_posix.exists(end_script):
                             raise IOError('End-of-job script %s not found.' % end_script)
 
                     # The <endscript> element can occur at the top level of the <stage>
                     # element, or inside a <fcl> element.
                     # Update the StageDef object differently in these two cases.
 
                     parent_element = end_script_element.parentNode
                     if parent_element.nodeName == 'fcl':
 
                         # This <endscript> is located inside a <fcl> element.
                         # Find the index of this fcl file.
                         # Python will raise an exception if the fcl can't be found
                         # (shouldn't happen).
 
                         fcl = str(parent_element.firstChild.data).strip()
                         n = self.fclname.index(fcl)
                         if not n in self.mid_script:
                             self.mid_script[n] = []
                         self.mid_script[n].append(end_script)
 
                     else:
 
                         # This is a <stage> level <endscript> element.
 
                         self.end_script.append(end_script)
 
         # Project name overrides (repeatable subelement).
 
         project_name_elements = stage_element.getElementsByTagName('projectname')
         if len(project_name_elements) > 0:
             for project_name_element in project_name_elements:
 
                 # Match this project name with its parent fcl element.
 
                 fcl_element = project_name_element.parentNode
                 if fcl_element.nodeName != 'fcl':
                     raise XMLError("Found <projectname> element outside <fcl> element.")
                 fcl = str(fcl_element.firstChild.data).strip()
 
                 # Find the index of this fcl file.
                 # Python will raise an exception if the fcl can't be found (shouldn't happen).
 
                 n = self.fclname.index(fcl)
 
                 # Make sure project_name list is long enough.
 
                 while len(self.project_name) < n+1:
                     self.project_name.append('')
 
                 # Extract project name and add it to list.
 
                 project_name = str(project_name_element.firstChild.data)
                 self.project_name[n] = project_name
 
         # Make sure that the size of the project_name list (if present) ia at least as
         # long as the fclname list.
         # If not, extend by adding empty string.
 
         if len(self.project_name) > 0:
             while len(self.project_name) < len(self.fclname):
                 self.project_name.append('')
 
         # Stage name overrides (repeatable subelement).
 
         stage_name_elements = stage_element.getElementsByTagName('stagename')
         if len(stage_name_elements) > 0:
             for stage_name_element in stage_name_elements:
 
                 # Match this project name with its parent fcl element.
 
                 fcl_element = stage_name_element.parentNode
                 if fcl_element.nodeName != 'fcl':
                     raise XMLError("Found <stagename> element outside <fcl> element.")
                 fcl = str(fcl_element.firstChild.data).strip()
 
                 # Find the index of this fcl file.
                 # Python will raise an exception if the fcl can't be found (shouldn't happen).
 
                 n = self.fclname.index(fcl)
 
                 # Make sure stage_name list is long enough.
 
                 while len(self.stage_name) < n+1:
                     self.stage_name.append('')
 
                 # Extract stage name and add it to list.
 
                 stage_name = str(stage_name_element.firstChild.data)
                 self.stage_name[n] = stage_name
 
         # Make sure that the size of the stage_name list (if present) ia at least as
         # long as the fclname list.
         # If not, extend by adding empty string.
 
         if len(self.stage_name) > 0:
             while len(self.stage_name) < len(self.fclname):
                 self.stage_name.append('')
 
         # Project version overrides (repeatable subelement).
 
         project_version_elements = stage_element.getElementsByTagName('version')
         if len(project_version_elements) > 0:
             for project_version_element in project_version_elements:
 
                 # Match this project version with its parent fcl element.
 
                 fcl_element = project_version_element.parentNode
                 if fcl_element.nodeName != 'fcl':
                     raise XMLError("Found stage level <version> element outside <fcl> element.")
                 fcl = str(fcl_element.firstChild.data).strip()
 
                 # Find the index of this fcl file.
                 # Python will raise an exception if the fcl can't be found (shouldn't happen).
 
                 n = self.fclname.index(fcl)
 
                 # Make sure project_version list is long enough.
 
                 while len(self.project_version) < n+1:
                     self.project_version.append('')
 
                 # Extract project version and add it to list.
 
                 project_version = str(project_version_element.firstChild.data)
                 self.project_version[n] = project_version
 
         # Make sure that the size of the project_version list (if present) ia at least as
         # long as the fclname list.
         # If not, extend by adding empty string.
 
         if len(self.project_version) > 0:
             while len(self.project_version) < len(self.fclname):
                 self.project_version.append('')
 
         # Histogram merging program.
 
         merge_elements = stage_element.getElementsByTagName('merge')
         if merge_elements:
             self.merge = str(merge_elements[0].firstChild.data)
         
         # Analysis merge flag.
 
         anamerge_elements = stage_element.getElementsByTagName('anamerge')
         if anamerge_elements:
             self.anamerge = str(anamerge_elements[0].firstChild.data)
         
         # Resource (subelement).
 
         resource_elements = stage_element.getElementsByTagName('resource')
         if resource_elements:
             self.resource = str(resource_elements[0].firstChild.data)
             self.resource = ''.join(self.resource.split())
 
         # Lines (subelement).
 
         lines_elements = stage_element.getElementsByTagName('lines')
         if lines_elements:
             self.lines = str(lines_elements[0].firstChild.data)
 
         # Site (subelement).
 
         site_elements = stage_element.getElementsByTagName('site')
         if site_elements:
             self.site = str(site_elements[0].firstChild.data)
             self.site = ''.join(self.site.split())
 
         # Blacklist (subelement).
 
         blacklist_elements = stage_element.getElementsByTagName('blacklist')
         if blacklist_elements:
             self.blacklist = str(blacklist_elements[0].firstChild.data)
             self.blacklist = ''.join(self.blacklist.split())
 
         # Cpu (subelement).
 
         cpu_elements = stage_element.getElementsByTagName('cpu')
         if cpu_elements:
             self.cpu = int(cpu_elements[0].firstChild.data)
 
         # Disk (subelement).
 
         disk_elements = stage_element.getElementsByTagName('disk')
         if disk_elements:
             self.disk = str(disk_elements[0].firstChild.data)
             self.disk = ''.join(self.disk.split())
 
         # Data file types (subelement).
 
         datafiletypes_elements = stage_element.getElementsByTagName('datafiletypes')
         if datafiletypes_elements:
             data_file_types_str = str(datafiletypes_elements[0].firstChild.data)
             data_file_types_str = ''.join(data_file_types_str.split())
             self.datafiletypes = data_file_types_str.split(',')
 
         # Memory (subelement).
 
         memory_elements = stage_element.getElementsByTagName('memory')
         if memory_elements:
             self.memory = int(memory_elements[0].firstChild.data)
 
         # Dictionary of metadata parameters (repeatable subelement).
 
         param_elements = stage_element.getElementsByTagName('parameter')
         if len(param_elements) > 0:
             self.parameters = {}
             for param_element in param_elements:
                 name = str(param_element.attributes['name'].firstChild.data)
                 value = str(param_element.firstChild.data)
                 self.parameters[name] = value
 
         # Output file name (repeatable subelement).
 
         output_elements = stage_element.getElementsByTagName('output')
         if len(output_elements) > 0:
 
             # The output element can occur once at the top level of the <stage> element, or
             # inside a <fcl> element.  The former applies globally.  The latter applies
             # only to that fcl substage.
 
             # Loop over global output elements.
 
             for output_element in output_elements:
                 parent_element = output_element.parentNode
                 if parent_element.nodeName != 'fcl':
                     output = str(output_element.firstChild.data)
                     self.output = []
                     while len(self.output) < len(self.fclname):
                         self.output.append(output)
 
             # Loop over fcl output elements.
 
             for output_element in output_elements:
                 parent_element = output_element.parentNode
                 if parent_element.nodeName == 'fcl':
 
                     # Match this output name with its parent fcl element.
 
                     fcl = str(parent_element.firstChild.data).strip()
                     n = self.fclname.index(fcl)
 
                     # Make sure project_name list is long enough.
 
                     while len(self.output) < n+1:
                         self.output.append('')
 
                     # Extract output name and add it to list.
 
                     output = str(output_element.firstChild.data)
                     self.output[n] = output
 
         # Make sure that the size of the output list (if present) ia at least as
         # long as the fclname list.
         # If not, extend by adding empty string.
 
         if len(self.output) > 0:
             while len(self.output) < len(self.fclname):
                 self.output.append('')
 
         # TFileName (subelement).
 
         TFileName_elements = stage_element.getElementsByTagName('TFileName')
         if TFileName_elements:
             self.TFileName = str(TFileName_elements[0].firstChild.data)
 
         # Jobsub.
 
         jobsub_elements = stage_element.getElementsByTagName('jobsub')
         if jobsub_elements:
             self.jobsub = str(jobsub_elements[0].firstChild.data)
 
         # Jobsub start/stop.
 
         jobsub_start_elements = stage_element.getElementsByTagName('jobsub_start')
         if jobsub_start_elements:
             self.jobsub_start = str(jobsub_start_elements[0].firstChild.data)
 
         # Jobsub submit timeout.
 
         jobsub_timeout_elements = stage_element.getElementsByTagName('jobsub_timeout')
         if jobsub_timeout_elements:
             self.jobsub_timeout = int(jobsub_timeout_elements[0].firstChild.data)
 
         # Name of art-like executables (repeatable subelement).
 
         exe_elements = stage_element.getElementsByTagName('exe')
         if len(exe_elements) > 0:
 
             # The exe element can occur once at the top level of the <stage> element, or
             # inside a <fcl> element.  The former applies globally.  The latter applies
             # only to that fcl substage.
 
             # Loop over global exe elements.
 
             for exe_element in exe_elements:
                 parent_element = exe_element.parentNode
                 if parent_element.nodeName != 'fcl':
                     exe = str(exe_element.firstChild.data)
                     self.exe = []
                     while len(self.exe) < len(self.fclname):
                         self.exe.append(exe)
 
             # Loop over fcl exe elements.
 
             for exe_element in exe_elements:
                 parent_element = exe_element.parentNode
                 if parent_element.nodeName == 'fcl':
 
                     # Match this exe name with its parent fcl element.
 
                     fcl = str(parent_element.firstChild.data).strip()
                     n = self.fclname.index(fcl)
 
                     # Make sure project_name list is long enough.
 
                     while len(self.exe) < n+1:
                         self.exe.append('')
 
                     # Extract exe name and add it to list.
 
                     exe = str(exe_element.firstChild.data)
                     self.exe[n] = exe
 
         # Make sure that the size of the exe list (if present) ia at least as
         # long as the fclname list.
         # If not, extend by adding empty string.
 
         if len(self.exe) > 0:
             while len(self.exe) < len(self.fclname):
                 self.exe.append('')
 
         # Sam schema.
 
         schema_elements = stage_element.getElementsByTagName('schema')
         if schema_elements:
             self.schema = str(schema_elements[0].firstChild.data)
 
         # Validate-on-worker.
 
         validate_on_worker_elements = stage_element.getElementsByTagName('check')
         if validate_on_worker_elements:
             self.validate_on_worker = int(validate_on_worker_elements[0].firstChild.data)
 
         # Upload-on-worker.
 
         copy_to_fts_elements = stage_element.getElementsByTagName('copy')
         if copy_to_fts_elements:
             self.copy_to_fts = int(copy_to_fts_elements[0].firstChild.data)
 
         # Cvmfs flag.
 
         cvmfs_elements = stage_element.getElementsByTagName('cvmfs')
         if cvmfs_elements:
             self.cvmfs = int(cvmfs_elements[0].firstChild.data)
 
         # Stash flag.
 
         stash_elements = stage_element.getElementsByTagName('stash')
         if stash_elements:
             self.stash = int(stash_elements[0].firstChild.data)
 
         # Singularity flag.
 
         singularity_elements = stage_element.getElementsByTagName('singularity')
         if singularity_elements:
             self.singularity = int(singularity_elements[0].firstChild.data)
 
         # Batch script
 
         script_elements = stage_element.getElementsByTagName('script')
         if script_elements:
             self.script = script_elements[0].firstChild.data
 
         # Make sure batch script exists, and convert into a full path.
 
         if check:
             script_path = ''
             try:
                 jobinfo = subprocess.Popen(['which', self.script],
                                            stdout=subprocess.PIPE,
                                            stderr=subprocess.PIPE)
                 jobout, joberr = jobinfo.communicate()
                 jobout = convert_str(jobout)
                 joberr = convert_str(joberr)
                 rc = jobinfo.poll()
                 script_path = jobout.splitlines()[0].strip()
             except:
                 pass
             if script_path == '' or not larbatch_posix.access(script_path, os.X_OK):
                 raise IOError('Script %s not found.' % self.script)
             self.script = script_path
         
         # Start script
 
         start_script_elements = stage_element.getElementsByTagName('startscript')
         if start_script_elements:
             self.start_script = start_script_elements[0].firstChild.data
 
         # Make sure start project batch script exists, and convert into a full path.
 
         if check:
             script_path = ''
             try:
                 jobinfo = subprocess.Popen(['which', self.start_script],
                                            stdout=subprocess.PIPE,
                                            stderr=subprocess.PIPE)
                 jobout, joberr = jobinfo.communicate()
                 jobout = convert_str(jobout)
                 joberr = convert_str(joberr)
                 rc = jobinfo.poll()
                 script_path = jobout.splitlines()[0].strip()
             except:
                 pass
             self.start_script = script_path
 
         # Stop script
 
         stop_script_elements = stage_element.getElementsByTagName('stopscript')
         if stop_script_elements:
             self.stop_script = stop_script_elements[0].firstChild.data
 
         # Make sure stop project batch script exists, and convert into a full path.
 
         if check:
             script_path = ''
             try:
                 jobinfo = subprocess.Popen(['which', self.stop_script],
                                            stdout=subprocess.PIPE,
                                            stderr=subprocess.PIPE)
                 jobout, joberr = jobinfo.communicate()
                 jobout = convert_str(jobout)
                 joberr = convert_str(joberr)
                 rc = jobinfo.poll()
                 script_path = jobout.splitlines()[0].strip()
             except:
                 pass
             self.stop_script = script_path
 
         # Done.
 
         return
         

Member Function Documentation

def python.stagedef.StageDef.__str__ ( self )

Definition at line 1127 of file stagedef.py.

     def __str__(self):
         result = 'Stage name = %s\n' % self.name
         result = 'Batch job name = %s\n' % self.batchname
         #result += 'Fcl filename = %s\n' % self.fclname
         for fcl in self.fclname:
           result += 'Fcl filename = %s\n' % fcl 
         result += 'Output directory = %s\n' % self.outdir
         result += 'Log directory = %s\n' % self.logdir
         result += 'Work directory = %s\n' % self.workdir
         result += 'Bookkeeping directory = %s\n' % self.bookdir
         result += 'Maximum directory size = %d\n' % self.dirsize
         result += 'Extra directory levels = %d\n' % self.dirlevels
         result += 'Dynamic directories = %d\n' % self.dynamic
         result += 'Input file = %s\n' % self.inputfile
         result += 'Input list = %s\n' % self.inputlist
         result += 'Input mode = %s\n' % self.inputmode
         result += 'Input sam dataset = %s' % self.inputdef
         if self.recur:
             result += ' (recursive)'
         result += '\n'
         result += 'Base sam dataset = %s\n' % self.basedef
         result += 'Analysis flag = %d\n' % self.ana
         result += 'Recursive flag = %d\n' % self.recur
         result += 'Recursive type = %s\n' % self.recurtype
         result += 'Recursive limit = %d\n' % self.recurlimit
         result += 'Single run flag = %d\n' % self.singlerun
         result += 'File list definition flag = %d\n' % self.filelistdef
         result += 'Prestart flag = %d\n' % self.prestart
         result += 'Active projects base name = %s\n' % self.activebase
         result += 'Dropbox waiting interval = %f\n' % self.dropboxwait
         result += 'Prestage fraction = %f\n' % self.prestagefraction
         result += 'Input stream = %s\n' % self.inputstream
         result += 'Previous stage name = %s\n' % self.previousstage
         result += 'Mix input sam dataset = %s\n' % self.mixinputdef
         result += 'Pubs input allowed = %d\n' % self.pubs_input_ok
         result += 'Pubs input mode = %d\n' % self.pubs_input
         result += 'Pubs input run number = %d\n' % self.input_run
         for subrun in self.input_subruns:
             result += 'Pubs input subrun number = %d\n' % subrun
         result += 'Pubs input version number = %d\n' % self.input_version
         result += 'Pubs output mode = %d\n' % self.pubs_output
         result += 'Pubs output run number = %d\n' % self.output_run
         for subrun in self.output_subruns:
             result += 'Pubs output subrun number = %d\n' % subrun
         result += 'Pubs output version number = %d\n' % self.output_version
         result += 'Output file name = %s\n' % self.output
         result += 'TFileName = %s\n' % self.TFileName   
         result += 'Number of jobs = %d\n' % self.num_jobs
         result += 'Number of events = %d\n' % self.num_events
         result += 'Max flux MB = %d\n' % self.maxfluxfilemb
         result += 'Max files per job = %d\n' % self.max_files_per_job
         result += 'Output file target size = %d\n' % self.target_size
         result += 'Dataset definition name = %s\n' % self.defname
         result += 'Analysis dataset definition name = %s\n' % self.ana_defname
         result += 'Data tier = %s\n' % self.data_tier
         result += 'Data stream = %s\n' % self.data_stream
         result += 'Analysis data tier = %s\n' % self.ana_data_tier
         result += 'Analysis data stream = %s\n' % self.ana_data_stream
         result += 'Submit script = %s\n' % self.submit_script
         result += 'Worker initialization script = %s\n' % self.init_script
         result += 'Worker initialization source script = %s\n' % self.init_source
         result += 'Worker end-of-job script = %s\n' % self.end_script
         result += 'Worker midstage source initialization scripts = %s\n' % self.mid_source
         result += 'Worker midstage finalization scripts = %s\n' % self.mid_script
         result += 'Project name overrides = %s\n' % self.project_name
         result += 'Stage name overrides = %s\n' % self.stage_name
         result += 'Project version overrides = %s\n' % self.project_version
         result += 'Special histogram merging program = %s\n' % self.merge
         result += 'Analysis merge flag = %s\n' % self.anamerge
         result += 'Resource = %s\n' % self.resource
         result += 'Lines = %s\n' % self.lines
         result += 'Site = %s\n' % self.site
         result += 'Blacklist = %s\n' % self.blacklist
         result += 'Cpu = %d\n' % self.cpu
         result += 'Disk = %s\n' % self.disk
         result += 'Datafiletypes = %s\n' % self.datafiletypes
         result += 'Memory = %d MB\n' % self.memory
         result += 'Metadata parameters:\n'
         for key in self.parameters:
             result += '%s: %s\n' % (key,self.parameters[key])
         result += 'Output file name = %s\n' % self.output
         result += 'TFile name = %s\n' % self.TFileName
         result += 'Jobsub_submit options = %s\n' % self.jobsub
         result += 'Jobsub_submit start/stop options = %s\n' % self.jobsub_start
         result += 'Jobsub submit timeout = %d\n' % self.jobsub_timeout
         result += 'Executables = %s\n' % self.exe
         result += 'Schema = %s\n' % self.schema
         result += 'Validate-on-worker = %d\n' % self.validate_on_worker
         result += 'Upload-on-worker = %d\n' % self.copy_to_fts
         result += 'Cvmfs flag = %d\n' % self.cvmfs
         result += 'Stash cache flag = %d\n' % self.stash
         result += 'Singularity flag = %d\n' % self.singularity
         result += 'Batch script = %s\n' % self.script
         result += 'Start script = %s\n' % self.start_script
         result += 'Stop script = %s\n' % self.stop_script
         return result
 

def python.stagedef.StageDef.check_output_dirs ( self )

Definition at line 1740 of file stagedef.py.

     def check_output_dirs(self):
         if not larbatch_posix.exists(self.outdir):
             raise IOError('Output directory %s does not exist.' % self.outdir)
         if not larbatch_posix.exists(self.logdir):
             raise IOError('Log directory %s does not exist.' % self.logdir)
         return
     

def python.stagedef.StageDef.checkdirs ( self )

Definition at line 1749 of file stagedef.py.

     def checkdirs(self):
         if not larbatch_posix.exists(self.outdir):
             raise IOError('Output directory %s does not exist.' % self.outdir)
         if self.logdir != self.outdir and not larbatch_posix.exists(self.logdir):
             raise IOError('Log directory %s does not exist.' % self.logdir)
         if not larbatch_posix.exists(self.workdir):
             raise IOError('Work directory %s does not exist.' % self.workdir)
         if self.bookdir != self.logdir and not larbatch_posix.exists(self.bookdir):
             raise IOError('Bookkeeping directory %s does not exist.' % self.bookdir)
         return
     

def python.stagedef.StageDef.checkinput	(	self,
		checkdef = `False`
	)

Definition at line 1548 of file stagedef.py.

     def checkinput(self, checkdef=False):
 
         if self.inputfile != '' and not larbatch_posix.exists(self.inputfile):
             raise IOError('Input file %s does not exist.' % self.inputfile)
         if self.inputlist != '' and not larbatch_posix.exists(self.inputlist):
             raise IOError('Input list %s does not exist.' % self.inputlist)
 
         checkok = False
 
         # Define or update the active projects dataset, if requested.
 
         if self.activebase != '':
             activedef = '%s_active' % self.activebase
             waitdef = '%s_wait' % self.activebase
             project_utilities.make_active_project_dataset(self.activebase,
                                                           self.dropboxwait,
                                                           activedef,
                                                           waitdef)
 
         # If target size is nonzero, and input is from a file list, calculate
         # the ideal number of output jobs and override the current number 
         # of jobs.
 
         if self.target_size != 0 and self.inputlist != '':
             input_filenames = larbatch_posix.readlines(self.inputlist)
             size_tot = 0
             for line in input_filenames:
                 filename = line.split()[0]
                 filesize = larbatch_posix.stat(filename).st_size
                 size_tot = size_tot + filesize
             new_num_jobs = size_tot / self.target_size
             if new_num_jobs < 1:
                 new_num_jobs = 1
             if new_num_jobs > self.num_jobs:
                 new_num_jobs = self.num_jobs
             print("Ideal number of jobs based on target file size is %d." % new_num_jobs)
             if new_num_jobs != self.num_jobs:
                 print("Updating number of jobs from %d to %d." % (self.num_jobs, new_num_jobs))
                 self.num_jobs = new_num_jobs
 
         # If singlerun mode is requested, pick a random file from the input
         # dataset and create (if necessary) a new dataset definition which
         # limits files to be only from that run.  Don't do anything here if 
         # the input dataset is empty.
 
         if self.singlerun and checkdef:
 
             samweb = project_utilities.samweb()
             print("Doing single run processing.")
 
             # First find an input file.
 
             #dim = 'defname: %s with limit 1' % self.inputdef
             dim = 'defname: %s' % self.inputdef
             if self.filelistdef:
                 input_files = list(project_utilities.listFiles(dim))
             else:
                 input_files = samweb.listFiles(dimensions=dim)
             if len(input_files) > 0:
                 random_file = random.choice(input_files)
                 print('Example file: %s' % random_file)
 
                 # Extract run number.
 
                 md = samweb.getMetadata(random_file)
                 run_tuples = md['runs']
                 if len(run_tuples) > 0:
                     run = run_tuples[0][0]
                     print('Input files will be limited to run %d.' % run)
 
                     # Make a new dataset definition.
                     # If this definition already exists, assume it is correct.
 
                     newdef = '%s_run_%d' % (samweb.makeProjectName(self.inputdef), run)
                     def_exists = False
                     try:
                         desc = samweb.descDefinition(defname=newdef)
                         def_exists = True
                     except samweb_cli.exceptions.DefinitionNotFound:
                         pass
                     if not def_exists:
                         print('Creating dataset definition %s' % newdef)
                         newdim = 'defname: %s and run_number %d' % (self.inputdef, run)
                         samweb.createDefinition(defname=newdef, dims=newdim)
                     self.inputdef = newdef
 
                 else:
                     print('Problem extracting run number from example file.')
                     return 1
 
             else:
                 print('Input dataset is empty.')
                 return 1
 
         # If target size is nonzero, and input is from a sam dataset definition,
         # and maxfilesperjob is not one, calculate the ideal number of jobs and
         # maxfilesperjob.
 
         if self.target_size != 0 and self.max_files_per_job != 1 and self.inputdef != '':
 
             # Query sam to determine size and number of files in input
             # dataset.
 
             samweb = project_utilities.samweb()
             dim = 'defname: %s' % self.inputdef
             nfiles = 0
             files = []
             if self.filelistdef:
                 files = project_utilities.listFiles(dim)
                 nfiles = len(files)
             else:
                 sum = samweb.listFilesSummary(dimensions=dim)
                 nfiles = sum['file_count']
             print('Input dataset %s has %d files.' % (self.inputdef, nfiles))
             if nfiles > 0:
                 checkok = True
                 max_files = self.max_files_per_job * self.num_jobs
                 size_tot = 0
                 if max_files > 0 and max_files < nfiles:
                     if self.filelistdef:
                         while len(files) > max_files:
                             files.pop()
                         dim = 'defname: %s' % project_utilities.makeFileListDefinition(files)
                     else:
                         dim += ' with limit %d' % max_files
                     sum = samweb.listFilesSummary(dimensions=dim)
                     size_tot = sum['total_file_size']
                     nfiles = sum['file_count']
                 else:
                     if self.filelistdef:
                         dim = 'defname: %s' % project_utilities.makeFileListDefinition(files)
                     sum = samweb.listFilesSummary(dimensions=dim)
                     size_tot = sum['total_file_size']
                     nfiles = sum['file_count']
 
                 # Calculate updated job parameters.
 
                 new_num_jobs = int(math.ceil(float(size_tot) / float(self.target_size)))
                 if new_num_jobs < 1:
                     new_num_jobs = 1
                 if new_num_jobs > self.num_jobs:
                     new_num_jobs = self.num_jobs
 
                 new_max_files_per_job = int(math.ceil(float(self.target_size) * float(nfiles) / float(size_tot)))
                 if self.max_files_per_job > 0 and new_max_files_per_job > self.max_files_per_job:
                     new_max_files_per_job = self.max_files_per_job
                     new_num_jobs = (nfiles + self.max_files_per_job - 1) / self.max_files_per_job
                     if new_num_jobs < 1:
                         new_num_jobs = 1
                     if new_num_jobs > self.num_jobs:
                         new_num_jobs = self.num_jobs
 
                 print("Ideal number of jobs based on target file size is %d." % new_num_jobs)
                 if new_num_jobs != self.num_jobs:
                     print("Updating number of jobs from %d to %d." % (self.num_jobs, new_num_jobs))
                     self.num_jobs = new_num_jobs
                 print("Ideal number of files per job is %d." % new_max_files_per_job)
                 if new_max_files_per_job != self.max_files_per_job:
                     print("Updating maximum files per job from %d to %d." % (
                         self.max_files_per_job, new_max_files_per_job))
                     self.max_files_per_job = new_max_files_per_job
             else:
                 print('Input dataset is empty.')
                 return 1
 
         # If requested, do a final check in the input dataset.
         # Limit the number of jobs to be not more than the number of files, since
         # it never makes sense to have more jobs than that.
         # If the number of input files is zero, return an error.
 
         if self.inputdef != '' and checkdef and not checkok:
             samweb = project_utilities.samweb()
             n = 0
             if self.filelistdef:
                 files = project_utilities.listFiles('defname: %s' % self.inputdef)
                 n = len(files)
             else:
                 sum = samweb.listFilesSummary(defname=self.inputdef)
                 n = sum['file_count']
             print('Input dataset %s contains %d files.' % (self.inputdef, n))
             if n < self.num_jobs:
                 self.num_jobs = n
             if n == 0:
                 return 1
 
         # Done (all good).
 
         return 0
 
 

def python.stagedef.StageDef.checksubmit ( self )

Definition at line 1510 of file stagedef.py.

     def checksubmit(self):
 
         rc = 0
         if len(self.submit_script) > 0:
             print('Running presubmission check script', end=' ')
             for word in self.submit_script:
                 print(word, end=' ')
             print()
             jobinfo = subprocess.Popen(self.submit_script,
                                        stdout=subprocess.PIPE,
                                        stderr=subprocess.PIPE)
             q = queue.Queue()
             thread = threading.Thread(target=larbatch_utilities.wait_for_subprocess,
                                       args=[jobinfo, q])
             thread.start()
             thread.join(timeout=60)
             if thread.is_alive():
                 print('Submit script timed out, terminating.')
                 jobinfo.terminate()
                 thread.join()
             rc = q.get()
             jobout = convert_str(q.get())
             joberr = convert_str(q.get())
             print('Script exit status = %d' % rc)
             print('Script standard output:')
             print(jobout)
             print('Script diagnostic output:')
             print(joberr)
 
         # Done.
         # Return exit status.
 
         return rc
 
 

def python.stagedef.StageDef.makedirs ( self )

Definition at line 1762 of file stagedef.py.

     def makedirs(self):
         if not larbatch_posix.exists(self.outdir):
             larbatch_posix.makedirs(self.outdir)
         if self.logdir != self.outdir and not larbatch_posix.exists(self.logdir):
             larbatch_posix.makedirs(self.logdir)
         if not larbatch_posix.exists(self.workdir):
             larbatch_posix.makedirs(self.workdir)
         if self.bookdir != self.logdir and not larbatch_posix.exists(self.bookdir):
             larbatch_posix.makedirs(self.bookdir)
 
         # If output is on dcache, make output directory group-writable.
 
         if self.outdir[0:6] == '/pnfs/':
             mode = stat.S_IMODE(larbatch_posix.stat(self.outdir).st_mode)
             if not mode & stat.S_IWGRP:
                 mode = mode | stat.S_IWGRP
                 larbatch_posix.chmod(self.outdir, mode)
         if self.logdir[0:6] == '/pnfs/':
             mode = stat.S_IMODE(larbatch_posix.stat(self.logdir).st_mode)
             if not mode & stat.S_IWGRP:
                 mode = mode | stat.S_IWGRP
                 larbatch_posix.chmod(self.logdir, mode)
 
         self.checkdirs()
         return

def python.stagedef.StageDef.pubsify_input	(	self,
		run,
		subruns,
		version
	)

Definition at line 1268 of file stagedef.py.

     def pubsify_input(self, run, subruns, version):
 
         # Don't do anything if pubs input is disabled.
 
         if not self.pubs_input_ok:
             return
 
         # It never makes sense to specify pubs input mode if there are no 
         # input files (i.e. generation jobs).  This is not considered an error.
 
         if self.inputfile == '' and self.inputlist == '' and self.inputdef == '':
             return
 
         # The case if input from a single file is not supported.  Raise an exception.
 
         if self.inputfile != '':
             raise RuntimeError('Pubs input for single file input is not supported.')
 
         # Set pubs input mode.
 
         self.pubs_input = 1
 
         # Save the run, subrun, and version numbers.
 
         self.input_run = run;
         self.input_subruns = subruns;
         self.input_version = version;
 
         # if input is from a sam dataset, create a restricted dataset that limits
         # input files to selected run and subruns.
 
         if self.inputdef != '':
             newdef = project_utilities.create_limited_dataset(self.inputdef,
                                                               run,
                                                               subruns)
             if not newdef:
                 raise PubsInputError(run, subruns[0], version)                
             self.inputdef = newdef
 
             # Set the number of submitted jobs assuming each worker will get 
             # self.max_files_per_job files.
 
             files_per_job = self.max_files_per_job
             if files_per_job == 0:
                 files_per_job = 1
             self.num_jobs = (len(subruns) + files_per_job - 1) / files_per_job
 
             # Done.
 
             return
 
         # If we get to here, we have input from a file list and a previous stage
         # exists.  This normally indicates a daisy chain.  This is where subcases
         # 3 (a), (b) are handled.
 
         # Case 3(a), single subrun.
 
         if len(subruns) == 1:
 
             # Insert run and subrun into input file list path.
 
             if version == None:
                 pubs_path = '%d/%d' % (run, subruns[0])
             else:
                 pubs_path = '%d/%d/%d' % (version, run, subruns[0])
             dir = os.path.dirname(self.inputlist)
             base = os.path.basename(self.inputlist)
             self.inputlist = os.path.join(dir, pubs_path, base)
 
             # Verify that the input list exists and is not empty.
 
             lines = []
             try:
                 lines = larbatch_posix.readlines(self.inputlist)
             except:
                 lines = []
             if len(lines) == 0:
                 raise PubsInputError(run, subruns[0], version)
 
             # Verify that input files actually exist.
 
             for line in lines:
                 input_file = line.strip()
                 if not larbatch_posix.exists(input_file):
                     raise PubsInputError(run, subruns[0], version)
 
             # Specify that there will be exactly one job submitted.
 
             self.num_jobs = 1
 
             # Everything OK (case 3(a)).
 
             return
 
         # Case 3(b), multiple subruns.
 
         if len(subruns) > 1:
 
             # Generate a new input file list with a unique name and place 
             # it in the same directory as the original input list.  Note that
             # the input list may not actually exist at this point.  If it
             # doesn't exist, just use the original name.  If it already exists,
             # generate a different name.
 
             dir = os.path.dirname(self.inputlist)
             base = os.path.basename(self.inputlist)
             new_inputlist_path = self.inputlist
             if larbatch_posix.exists(new_inputlist_path):
                 new_inputlist_path = '%s/%s_%s.list' % (dir, base, str(uuid.uuid4()))
             self.inputlist = new_inputlist_path
 
             # Defer opening the new input list file until after the original
             # input file is successfully opened.
 
             new_inputlist_file = None
 
             # Loop over subruns.  Read contents of pubs input list for each subrun.
 
             nsubruns = 0
             total_size = 0
             actual_subruns = []
             truncate = False
             for subrun in subruns:
 
                 if truncate:
                     break
 
                 nsubruns += 1
 
                 if version == None:
                     pubs_path = '%d/%d' % (run, subrun)
                 else:
                     pubs_path = '%d/%d/%d' % (version, run, subrun)
 
                 subrun_inputlist = os.path.join(dir, pubs_path, base)
                 lines = []
                 try:
                     lines = larbatch_posix.readlines(subrun_inputlist)
                 except:
                     lines = []
                 if len(lines) == 0:
                     raise PubsInputError(run, subruns[0], version)
                 for line in lines:
                     subrun_inputfile = line.strip()
 
                     # Test size and accessibility of this input file.
 
                     sr_size = -1
                     try:
                         sr = larbatch_posix.stat(subrun_inputfile)
                         sr_size = sr.st_size
                     except:
                         sr_size = -1
 
                     if sr_size > 0:
                         actual_subruns.append(subrun)
                         if new_inputlist_file == None:
                             print('Generating new input list %s\n' % new_inputlist_path)
                             new_inputlist_file = larbatch_posix.open(new_inputlist_path, 'w')
                         new_inputlist_file.write('%s\n' % subrun_inputfile)
                         total_size += sr.st_size
 
                     # If at this point the total size exceeds the target size,
                     # truncate the list of subruns and break out of the loop.
 
                     if self.max_files_per_job > 1 and self.target_size != 0 \
                             and total_size >= self.target_size:
                         truncate = True
                         break
 
             # Done looping over subruns.
 
             new_inputlist_file.close()
 
             # Raise an exception if the actual list of subruns is empty.
 
             if len(actual_subruns) == 0:
                 raise PubsInputError(run, subruns[0], version)
 
             # Update the list of subruns to be the actual list of subruns.
 
             if len(actual_subruns) != len(subruns):
                 print('Truncating subrun list: %s' % str(actual_subruns))
                 del subruns[:]
                 subruns.extend(actual_subruns)
 
             # Set the number of submitted jobs assuming each worker will get 
             # self.max_files_per_job files.
 
             files_per_job = self.max_files_per_job
             if files_per_job == 0:
                 files_per_job = 1
             self.num_jobs = (len(subruns) + files_per_job - 1) / files_per_job
 
             # Everything OK (case 3(b)).
 
             return
 
         # Shouldn't ever fall out of loop.
 
         return
 
 

def python.stagedef.StageDef.pubsify_output	(	self,
		run,
		subruns,
		version
	)

Definition at line 1473 of file stagedef.py.

     def pubsify_output(self, run, subruns, version):
 
         # Set pubs mode.
 
         self.pubs_output = 1
 
         # Save the run, subrun, and version numbers.
 
         self.output_run = run;
         self.output_subruns = subruns;
         self.output_version = version;
 
         # Append run and subrun to workdir, outdir, logdir, and bookdir.
         # In case of multiple subruns, encode the subdir directory as "@s",
         # which informs the batch worker to determine the subrun dynamically.
 
         if len(subruns) == 1:
             if version == None:
                 pubs_path = '%d/%d' % (run, subruns[0])
             else:
                 pubs_path = '%d/%d/%d' % (version, run, subruns[0])
             self.workdir = os.path.join(self.workdir, pubs_path)
         else:
             if version == None:
                 pubs_path = '%d/@s' % run
             else:
                 pubs_path = '%d/%d/@s' % (version, run)
             self.workdir = os.path.join(self.workdir, str(uuid.uuid4()))
             self.dynamic = 1
         self.outdir = os.path.join(self.outdir, pubs_path)
         self.logdir = os.path.join(self.logdir, pubs_path)
         self.bookdir = os.path.join(self.bookdir, pubs_path)
 

Member Data Documentation

python.stagedef.StageDef.activebase

Definition at line 82 of file stagedef.py.

python.stagedef.StageDef.ana

Definition at line 75 of file stagedef.py.

python.stagedef.StageDef.ana_data_stream

Definition at line 95 of file stagedef.py.

python.stagedef.StageDef.ana_data_tier

Definition at line 94 of file stagedef.py.

python.stagedef.StageDef.ana_defname

Definition at line 91 of file stagedef.py.

python.stagedef.StageDef.anamerge

Definition at line 106 of file stagedef.py.

python.stagedef.StageDef.basedef

Definition at line 61 of file stagedef.py.

python.stagedef.StageDef.batchname

Definition at line 49 of file stagedef.py.

python.stagedef.StageDef.blacklist

Definition at line 110 of file stagedef.py.

python.stagedef.StageDef.bookdir

Definition at line 54 of file stagedef.py.

python.stagedef.StageDef.copy_to_fts

Definition at line 124 of file stagedef.py.

python.stagedef.StageDef.cpu

Definition at line 111 of file stagedef.py.

python.stagedef.StageDef.cvmfs

Definition at line 125 of file stagedef.py.

python.stagedef.StageDef.data_stream

Definition at line 93 of file stagedef.py.

python.stagedef.StageDef.data_tier

Definition at line 92 of file stagedef.py.

python.stagedef.StageDef.datafiletypes

Definition at line 113 of file stagedef.py.

python.stagedef.StageDef.defname

Definition at line 90 of file stagedef.py.

python.stagedef.StageDef.dirlevels

Definition at line 56 of file stagedef.py.

python.stagedef.StageDef.dirsize

Definition at line 55 of file stagedef.py.

python.stagedef.StageDef.disk

Definition at line 112 of file stagedef.py.

python.stagedef.StageDef.dropboxwait

Definition at line 83 of file stagedef.py.

python.stagedef.StageDef.dynamic

Definition at line 57 of file stagedef.py.

python.stagedef.StageDef.end_script

Definition at line 99 of file stagedef.py.

python.stagedef.StageDef.exe

Definition at line 121 of file stagedef.py.

python.stagedef.StageDef.fclname

Definition at line 50 of file stagedef.py.

python.stagedef.StageDef.filelistdef

Definition at line 80 of file stagedef.py.

python.stagedef.StageDef.init_script

Definition at line 97 of file stagedef.py.

python.stagedef.StageDef.init_source

Definition at line 98 of file stagedef.py.

python.stagedef.StageDef.input_run

Definition at line 68 of file stagedef.py.

python.stagedef.StageDef.input_subruns

Definition at line 69 of file stagedef.py.

python.stagedef.StageDef.input_version

Definition at line 70 of file stagedef.py.

python.stagedef.StageDef.inputdef

Definition at line 62 of file stagedef.py.

python.stagedef.StageDef.inputfile

Definition at line 58 of file stagedef.py.

python.stagedef.StageDef.inputlist

Definition at line 59 of file stagedef.py.

python.stagedef.StageDef.inputmode

Definition at line 60 of file stagedef.py.

python.stagedef.StageDef.inputstream

Definition at line 63 of file stagedef.py.

python.stagedef.StageDef.jobsub

Definition at line 118 of file stagedef.py.

python.stagedef.StageDef.jobsub_start

Definition at line 119 of file stagedef.py.

python.stagedef.StageDef.jobsub_timeout

Definition at line 120 of file stagedef.py.

python.stagedef.StageDef.lines

Definition at line 108 of file stagedef.py.

python.stagedef.StageDef.logdir

Definition at line 52 of file stagedef.py.

python.stagedef.StageDef.max_files_per_job

Definition at line 88 of file stagedef.py.

python.stagedef.StageDef.maxfluxfilemb

Definition at line 85 of file stagedef.py.

python.stagedef.StageDef.memory

Definition at line 114 of file stagedef.py.

python.stagedef.StageDef.merge

Definition at line 105 of file stagedef.py.

python.stagedef.StageDef.mid_script

Definition at line 101 of file stagedef.py.

python.stagedef.StageDef.mid_source

Definition at line 100 of file stagedef.py.

python.stagedef.StageDef.mixinputdef

Definition at line 65 of file stagedef.py.

python.stagedef.StageDef.name

Definition at line 48 of file stagedef.py.

python.stagedef.StageDef.num_events

Definition at line 87 of file stagedef.py.

python.stagedef.StageDef.num_jobs

Definition at line 86 of file stagedef.py.

python.stagedef.StageDef.outdir

Definition at line 51 of file stagedef.py.

python.stagedef.StageDef.output

Definition at line 116 of file stagedef.py.

python.stagedef.StageDef.output_run

Definition at line 72 of file stagedef.py.

python.stagedef.StageDef.output_subruns

Definition at line 73 of file stagedef.py.

python.stagedef.StageDef.output_version

Definition at line 74 of file stagedef.py.

python.stagedef.StageDef.parameters

Definition at line 115 of file stagedef.py.

python.stagedef.StageDef.prestagefraction

Definition at line 84 of file stagedef.py.

python.stagedef.StageDef.prestart

Definition at line 81 of file stagedef.py.

python.stagedef.StageDef.previousstage

Definition at line 64 of file stagedef.py.

python.stagedef.StageDef.project_name

Definition at line 102 of file stagedef.py.

python.stagedef.StageDef.project_version

Definition at line 104 of file stagedef.py.

python.stagedef.StageDef.pubs_input

Definition at line 67 of file stagedef.py.

python.stagedef.StageDef.pubs_input_ok

Definition at line 66 of file stagedef.py.

python.stagedef.StageDef.pubs_output

Definition at line 71 of file stagedef.py.

python.stagedef.StageDef.recur

Definition at line 76 of file stagedef.py.

python.stagedef.StageDef.recurlimit

Definition at line 78 of file stagedef.py.

python.stagedef.StageDef.recurtype

Definition at line 77 of file stagedef.py.

python.stagedef.StageDef.resource

Definition at line 107 of file stagedef.py.

python.stagedef.StageDef.schema

Definition at line 122 of file stagedef.py.

python.stagedef.StageDef.script

Definition at line 128 of file stagedef.py.

python.stagedef.StageDef.singlerun

Definition at line 79 of file stagedef.py.

python.stagedef.StageDef.singularity

Definition at line 127 of file stagedef.py.

python.stagedef.StageDef.site

Definition at line 109 of file stagedef.py.

python.stagedef.StageDef.stage_name

Definition at line 103 of file stagedef.py.

python.stagedef.StageDef.start_script

Definition at line 129 of file stagedef.py.

python.stagedef.StageDef.stash

Definition at line 126 of file stagedef.py.

python.stagedef.StageDef.stop_script

Definition at line 130 of file stagedef.py.

python.stagedef.StageDef.submit_script

Definition at line 96 of file stagedef.py.

python.stagedef.StageDef.target_size

Definition at line 89 of file stagedef.py.

python.stagedef.StageDef.TFileName

Definition at line 117 of file stagedef.py.

python.stagedef.StageDef.validate_on_worker

Definition at line 123 of file stagedef.py.

python.stagedef.StageDef.workdir

Definition at line 53 of file stagedef.py.

The documentation for this class was generated from the following file:

larbatch/python/stagedef.py

Public Member Functions

Public Attributes

Detailed Description

Constructor & Destructor Documentation

Member Function Documentation

Member Data Documentation