submit_concat_project Namespace Reference

Functions
def	BuildOutputNameStub (dataset_name, decaf_macro, rename_cafs=True)
def	CalculateNParts (defname, snapshot_id, chunk_size=2, max_singlefile_size=20, requested_num_parts=None)
def	CheckProjects (dataset)
def	CheckProxy (role)
def	MakeOutputDir (dirname)
def	ParseArgs ()
def	ParseParts (parts_string)
def	Submit (args)

Variables
	search_path
	file
	cafana_dir
	CVMFS_RELEASE_DIR
	DECAF_MACROS
	MAX_N_JOBS
	SAM
	args
	testrel
	release_path
	release
	out_name
	decaf_positional
	rename_cafs
	parts
	parts_args
	parts_joined
	experiment
	snapshot_id
	e
	nparts
	dataset
	requested_num_parts
	disk
	memory
	concat

Detailed Description

  submit_concat_project.py:
      Submit jobs for producing concats from a dataset.
    
    Original author: J. Wolcott <jwolcott@fnal.gov>, rewritten from submit_concat_project.sh
       Date: August 2017

Function Documentation

def submit_concat_project.BuildOutputNameStub	(		dataset_name,
			decaf_macro,
			rename_cafs = True
	)

Definition at line 72 of file submit_concat_project.py.

def BuildOutputNameStub(dataset_name, decaf_macro, rename_cafs=True):
    # make the output file name stub.
    # this is a concat script, so we want to ensure "caf" -> "sumcaf",
    # and "decaf" -> "sumdecaf"...

    out_name = dataset_name
    replacements = {"_caf": "_sumcaf", "_decaf": "_sumdecaf", "_restrictedcaf": "_sumrestrictedcaf"}
    for repl in replacements:
        out_name = out_name.replace(repl, replacements[repl])
    
    # ... and if we're actually decaffing, that "caf" -> "decaf"
    if rename_cafs and os.path.basename(decaf_macro) != "concat_dataset.C":
        out_name = out_name.replace("_sumcaf", "_sumdecaf")
        out_name = out_name.replace("_sumrestrictedcaf", "_sumrestricteddecaf")
        if decaf_macro in DECAF_MACROS:
          out_name = out_name + "_" + decaf_macro
        
    return out_name

"""
  Compute number of parts dataset should be split into.
  
    snapshot_id = id of snapshot of dataset
    chunk_size = size of one file (in GB)
    max_singlefile_size = if dataset size is less than this size in GB, only a single file will be made
"""

def submit_concat_project.CalculateNParts	(		defname,
			snapshot_id,
			chunk_size = 2,
			max_singlefile_size = 20,
			requested_num_parts = None
	)

Definition at line 98 of file submit_concat_project.py.

References print.

def CalculateNParts(defname, snapshot_id, chunk_size=2, max_singlefile_size=20, requested_num_parts=None):
    if snapshot_id:
        dimensions = "snapshot_id=%d" % snapshot_id
    else:
        dimensions = "dataset_def_name_newest_snapshot %s" % defname
    summary = SAM.listFilesSummary(dimensions=dimensions)
    size = summary["total_file_size"]
    n_files = summary["file_count"]
    
    if requested_num_parts:
        if requested_num_parts <= n_files:
            return requested_num_parts
        else:
            print("You requested %d parts, but there are only %d files in the dataset.  Can't proceed." % (requested_num_parts, n_files), file=sys.stderr)
            sys.exit(1)
    
    GBs = old_div(size, 1024**3)  # bytes to gigabytes
    if GBs < max_singlefile_size:
        return 1
    else:
        return old_div(GBs,chunk_size)

def submit_concat_project.CheckProjects

(

dataset

)

Definition at line 120 of file submit_concat_project.py.

References print.

def CheckProjects(dataset):
    already_running_projects = SAM.listProjects(name='%%_proj_concat_%s%%' % dataset, state='running')
    if len(already_running_projects) + nparts > MAX_N_JOBS:
        print("Current concat request would require %d jobs, but" % nparts, file=sys.stderr)
        if len(already_running_projects) > 0:
            print("there are already %d SAM projects running for this definition," %  len(already_running_projects), file=sys.stderr)
            print("and ", end=' ', file=sys.stderr)
        print("there is a maximum of %d running concat projects for the same definition." % MAX_N_JOBS, file=sys.stderr)
        print("Please retry using --parts to restrict how many jobs are submitted.", file=sys.stderr)
        sys.exit(1)

def submit_concat_project.CheckProxy

(

role

)

Definition at line 132 of file submit_concat_project.py.

References print.

def CheckProxy(role):
    # now check we have a usable proxy
    arg = "-c"
    if role == "Production":
        arg += "p"
    try:
        subprocess.check_call(["setup_fnal_security", arg])
    except subprocess.CalledProcessError:
        print("Please run 'setup_fnal_security' with the correct options to get a valid proxy and try again.", file=sys.stderr)
        sys.exit(2)

def submit_concat_project.MakeOutputDir

(

dirname

)

Definition at line 143 of file submit_concat_project.py.

References print.

def MakeOutputDir(dirname):
    try:
        if not os.path.isdir(dirname):
            os.makedirs(dirname)
        os.chmod(dirname, 0o775)
    except OSError:
        print("Output directory '%s' cannot be created!" % dirname, file=sys.stderr)
        sys.exit(1)

def submit_concat_project.ParseArgs

(

)

Definition at line 152 of file submit_concat_project.py.

References parse_dependency_file_t.list.

def ParseArgs():    
    parser = argparse.ArgumentParser(description="Submit jobs for concatentation of (de)CAFs.")
    
    parser.add_argument("dataset",
                        metavar="DATASET",
                        help="Dataset containing (de)CAFs to concatenate",
    )
    parser.add_argument("outputdir",
                        metavar="OUTPUTDIR",
                        help="Where to write output",
    )
    parser.add_argument("release",
                        metavar="RELEASE",
                        help="Version of novasoft to use",
    )
    
    # support old (positional) syntax as well as nicer (option-based) one
    gp = parser.add_mutually_exclusive_group()
    gp.add_argument("decaf_positional",
                    metavar="DECAF",
                    nargs="?",
                    default="$NOVAGRIDUTILS_DIR/bin/concat_dataset.C",
                    help="Decaffing macro to run.  Either specify full path here or one of the following shortcut options: %s.  Can also specify as --decaf (below)." % list(DECAF_MACROS.keys()),
    )
    gp.add_argument("--decaf", "--decaf_macro",
                    metavar="DECAF",
                    help="Alternate form of positional DECAF (above).",
    )
    
    # etc.
    gp = parser.add_mutually_exclusive_group()
    gp.add_argument("role_positional",
                    metavar="ROLE",
                    nargs="?",
                    default="Analysis",
                    choices=["Analysis", "Production"],
                    help="VOMS role to submit with.  (You must have Production-eligible proxy to use the Production role.)  Can also specify as --role or --voms_role (below).",
    )
    gp.add_argument("--role", "--voms_role",
                    choices=["Analysis", "Production"],
                    help="Alternate form of positional ROLE (above).",
    )
     
    gp = parser.add_mutually_exclusive_group()
    gp.add_argument("nparts_positional",
                    nargs="?",
                    type=int,
                    metavar="NPARTS",
                    help="Divide concats into this many files.  If PARTS is also specified, NPARTS must correspond exactly to the number of parts.  Can also specify as --nparts (below).",
    )
    gp.add_argument("--nparts",
                    metavar="NPARTS",
                    type=int,
                    help="Alternate form of positional NPARTS (above).",
    )
    
    gp = parser.add_mutually_exclusive_group()
    gp.add_argument("parts_positional",
                    nargs="?",
                    metavar="PARTS",
                    help="The indices of which concat files to process.  Optional form '--parts' (see description below) is preferred. ",
    )
    gp.add_argument("--parts",
                    metavar="PARTS",
                    action="append",
                    help="The indices of which concat files to process.  Can separate multiple items with commas, or specify --parts multiple times." \
                         +"  Ranges can be specified using a hyphen.  So, '--parts 3,5,7-9 --parts 11-15 --parts 27' is legal.",
    )
    
    # allow user to opt out of 'caf' -> 'decaf' renaming in case we don't automatically get it right
    parser.add_argument("--no_caf_rename",
                    action="store_true",
                    default=False,
                    help="Do not substitute 'decaf' for 'caf' in the output name.  Useful if you're only concatting CAFs with some script other than concat_dataset.C.",
    )
    
    parser.add_argument("--use_last_snapshot",
                    action="store_true",
                    default=False,
                    help="Use the last snapshot of the definition rather than taking a new one.",
    )

    parser.add_argument("--flatten",
                    action="store_true",
                    default=True,
                    help="Flatten concat file",
    )
    
    parser.add_argument("--flat_only",
                    action="store_true",
                    default=False,
                    help="Produce only flattened concat file",
    )

    gp = parser.add_mutually_exclusive_group(required=False)
    gp.add_argument('-t', '--testrel', metavar='DIR', 
                            help='Use a test release at location TESTREL. It will be tarred up, and sent to the worker node. (Conflicts with --user_tarball)',
                            default=None)
    
    gp.add_argument("--user_tarball",
                            help="Use existing test release tarball in specified location rather than having jobsub make one for you (conflicts with --testrel, and is redunant with --reuse_tarball)",
                            type=str)

    gp.add_argument('--reuse_tarball',
                            help='Do you want to reuse a tarball that is already in resilient space? If using this option avoid trailing slash in --testrel option.  (redundant with --user_tarball)',
                            action='store_true',default=False)
    
    parser.add_argument("-n", "--dry-run", "--test-only",
                    action="store_true",
                    default=False,
                    help="Test only; don't actually submit any jobs.  Print jobsub command and exit."
    )

    parser.add_argument("--disk",
                    type=int,
                    help="Local disk space requirement for worker node in MB (default is 2400MB without flattening).",) # No default here for flatten resource logic

    parser.add_argument("--memory",
                    type=int,
                    help="Local memory requirement for worker node in MB (default is 3000MB without flattening).") # No default here for flatten resource logic
    
    
    
    return parser.parse_args()

def submit_concat_project.ParseParts

(

parts_string

)

Definition at line 277 of file submit_concat_project.py.

References makeTrainCVSamples.int, and PandAna.Demos.demo1.range.

def ParseParts(parts_string):
    parts = set()
    for tok in parts_string.split(","):
        if "-" in tok:
            start, end = (int(t) for t in tok.split("-"))
            parts |= set(range(start, end+1))
        else:
            parts.add(int(tok))
    
    return parts

def submit_concat_project.Submit

(

args

)

Definition at line 288 of file submit_concat_project.py.

References print, split(), and submit_syst.str.

def Submit(args):
    # can be used for interactive testing if needed
#     cmd = "$NOVAGRIDUTILS_DIR/bin/concat_job_project.sh" \
   

    decafscript = args["decaf_macro"]

    cmd = "jobsub_submit -g -N%(nparts)d --role=%(role)s " 


    # Set resource requirements
    cmd += " --disk=%sMB" % args["disk"]
    cmd += " --memory=%sMB" % args["memory"]
    cmd += " --expected-lifetime=23h"

    # files on blue arc are not accessible directly, we need to copy them to the grid nodes
    if "/nova/" in os.path.expandvars(args["decaf_macro"]):
        print("decaf macro has a blue arc location, copying it to grid nodes")
        # copy it at submission time
        cmd += " -f dropbox://%(decaf_macro)s"
        # remove path string containing /nova/ for concat_job_project.sh
        decafscript = args["decaf_macro"].split('/')[-1]
    
    if args["testrel"]:
        cmd += " --tar_file_name " + args["testrel"]

    cmd += " file://$NOVAGRIDUTILS_DIR/bin/concat_job_project.sh" \
           + " --dataset %(dataset)s --outname %(outname)s --snapid %(snapshot_id)s" \
           + " --nparts %(nparts)d --release %(release)s --outdir %(outputdir)s"
    
    cmd %= args
    cmd += " --decafscript %s " % decafscript
    
    if args["testrel"]:
        cmd += " --testrel"
    if len(args["parts"]):
        cmd += " --parts " + ",".join(str(a) for a in args["parts"])
    if args["flatten"]:
        cmd += " --flatten"
    if args["flat_only"]:
        cmd += " --flat_only"

    cmd = os.path.expandvars(cmd)
    print(cmd)
    if not args["dry_run"]:
        subprocess.check_call(shlex.split(cmd))

Variable Documentation

submit_concat_project.args

Definition at line 338 of file submit_concat_project.py.

submit_concat_project.cafana_dir

Definition at line 30 of file submit_concat_project.py.

submit_concat_project.concat

Definition at line 409 of file submit_concat_project.py.

Referenced by ana.getAllXsecNuTruthSysts_2018(), ana.getAllXsecNuTruthSysts_2018_RPAFix(), MakeCovarSim(), MakeNus18CovarSim(), MakeNus18ExtrapPred(), MakeNus18SidebandPred(), MakeNus18Systs(), MakePPFXRatios(), MakePredictionNoOsc_FHC_FD(), MakePredictionNoOsc_FHC_ND(), MakePredictionNoOsc_RHC_FD(), MakePredictionNoOsc_RHC_ND(), and MakeSysts().

submit_concat_project.CVMFS_RELEASE_DIR

Definition at line 39 of file submit_concat_project.py.

submit_concat_project.dataset

Definition at line 392 of file submit_concat_project.py.

submit_concat_project.DECAF_MACROS

Definition at line 41 of file submit_concat_project.py.

submit_concat_project.decaf_positional

Definition at line 364 of file submit_concat_project.py.

submit_concat_project.disk

Definition at line 404 of file submit_concat_project.py.

Referenced by plot_abstree().

submit_concat_project.e

Definition at line 389 of file submit_concat_project.py.

submit_concat_project.experiment

Definition at line 381 of file submit_concat_project.py.

submit_concat_project.file

Definition at line 28 of file submit_concat_project.py.

submit_concat_project.MAX_N_JOBS

Definition at line 65 of file submit_concat_project.py.

submit_concat_project.memory

Definition at line 406 of file submit_concat_project.py.

submit_concat_project.nparts

Definition at line 392 of file submit_concat_project.py.

Referenced by convert(), evgen::GENIEFilter.filter(), gibuu::RecordLoader.GetRecord(), and gibuu::RecordLoader.RecordLoader().

submit_concat_project.out_name

Definition at line 364 of file submit_concat_project.py.

Referenced by ana.CompareNDDataMC(), ana.CompareNDDataMCFromVector(), ana.CompareNDDataOneMC(), ana.CompareOneShiftPred(), ana.ComparePredictions(), ana.GetNDComponents(), ana.GetNDDecompsFromDecomp(), ana.PlotDataMC(), PlotDataMC(), ana.PlotMCComponentsComparison(), PlotMCComponentsComparison(), PlotMCComponentsErrorBand(), ana.PlotMCComponentsErrorBand(), ana.PlotNDDataTotalMCComparison(), PlotNDDataTotalMCComparison(), Decomp.Write(), and Distribution.Write().

submit_concat_project.parts

Definition at line 367 of file submit_concat_project.py.

Referenced by calib::TestParticleCorrections.analyze(), slicer::SlicerAna.analyze(), genie::masterclass::GNuMcMainFrame.BuildStatusBar(), cheat::CheckBackTracking.CheckTrackIdToParticle(), calib::StopperThreshold.FillHist(), calib::StopperThreshold.FillTree(), trk::CosmicTrackAna.FillTrueInfo(), nuesand::FillNueSandbox.FillTruthVars(), nuesand::FillNueSandbox.GetECF(), numusand::NumuSandFxs.getMissingE(), calib::RockMuonStopperSelection.IsTrulyContainedStopper(), calib::StopperSelection.IsTrulyContainedStopper(), calib::RockMuonStopperSelection.IsTrulyInFiducial(), calib::ThroughgoingSelection.IsTrulyUncontained(), tensorflow::TFFifoHandler.Predict(), calib::PCHitsList.ProcessTrackForBelowThresholdHits(), calib::PCHitsList.ProcessTrackForFLSHits(), util::TruthExpand.produce(), numusand::FillSandbox.produce(), cosrej::CosRej.produce(), lem::MakeLibrary.produce(), g4rwgt::MakeG4WeightTable.produce(), and evgb::CRYHelper.Sample().

submit_concat_project.parts_args

Definition at line 368 of file submit_concat_project.py.

submit_concat_project.parts_joined

Definition at line 370 of file submit_concat_project.py.

submit_concat_project.release

Definition at line 361 of file submit_concat_project.py.

submit_concat_project.release_path

Definition at line 358 of file submit_concat_project.py.

submit_concat_project.rename_cafs

Definition at line 364 of file submit_concat_project.py.

submit_concat_project.requested_num_parts

Definition at line 392 of file submit_concat_project.py.

submit_concat_project.SAM

Definition at line 70 of file submit_concat_project.py.

submit_concat_project.search_path

Definition at line 26 of file submit_concat_project.py.

submit_concat_project.snapshot_id

Definition at line 383 of file submit_concat_project.py.

submit_concat_project.testrel

Definition at line 341 of file submit_concat_project.py.