python.EvgenHelpers Namespace Reference

Functions
	_mk_symlink (srcfile, dstfile)
	_count_lhe_events (lhe_file)
	_find_unique_file (pattern)
	_merge_lhe_files (listOfFiles, outputFile)
	_handle_input_files (generators, flags)
	_validate_sample_properties (sample)
	_is_txt_only_run (flags)

Variables
	evgenLog = logging.getLogger("Gen_tf")

Function Documentation

_count_lhe_events()

python.EvgenHelpers._count_lhe_events ( lhe_file )

protected

Helper function to count LHE events in a file.
Support for plain text, gz, tar.gz, tgz files.
Use chunked reading to avoid memory issues with large files.

Definition at line 26 of file EvgenHelpers.py.

def _count_lhe_events(lhe_file):
    """Helper function to count LHE events in a file.
    Support for plain text, gz, tar.gz, tgz files.
    Use chunked reading to avoid memory issues with large files.
    """
    def _count_in_stream(stream):
        count_ev = 0
        for chunk in iter(lambda: stream.read(1024 * 1024), b""):
            count_ev += chunk.count(b"/event")
        return count_ev
 
    if lhe_file.endswith((".tar.gz", ".tgz", ".tar")):
        count_ev = 0
        with tarfile.open(lhe_file, "r:*") as tar:
            for member in tar:
                if not member.isfile():
                    continue
                extracted = tar.extractfile(member)
                if extracted is None:
                    continue
                with extracted:
                    count_ev += _count_in_stream(extracted)
        return count_ev
 
    if lhe_file.endswith(".gz"):
        with gzip.open(lhe_file, "rb") as f:
            return _count_in_stream(f)
 
    with open(lhe_file, "rb") as f:
        return _count_in_stream(f)
 
 

_find_unique_file()

python.EvgenHelpers._find_unique_file ( pattern )

protected

Helper functions for finding input file

Definition at line 58 of file EvgenHelpers.py.

def _find_unique_file(pattern):
    """Helper functions for finding input file"""
    import glob
    files = glob.glob(pattern)
    # Check that there is exactly 1 match
    if not files:
        raise RuntimeError(f"No {pattern} file found")
    elif len(files) > 1:
        raise RuntimeError(f"More than one {pattern} file found")
    return files[0]
 
 

_handle_input_files()

python.EvgenHelpers._handle_input_files ( generators, flags )

protected

Helper for handling input files

Definition at line 120 of file EvgenHelpers.py.

def _handle_input_files(generators, flags):
    """Helper for handling input files"""
    from GeneratorConfig.GenConfigHelpers import gens_lhef
    is_lhe_input = gens_lhef(generators)
 
    # Name of event files produced by various generators.
    events_file_map = {
        "Alpgen": "alpgen.unw_events",
        "Protos": "protos.events",
        "ProtosLHEF": "protoslhef.events",
        "BeamHaloGenerator": "beamhalogen.events",
        "HepMCAscii": "events.hepmc",
        "ReadMcAscii": "events.hepmc",
    }
    eventsFile = None
    for gen_name, out_file in events_file_map.items():
        if gen_name in generators:
            eventsFile = out_file
            break
    if eventsFile is None:
        if is_lhe_input:
            eventsFile = "events.lhe"
        else:
            raise RuntimeError(f"Unknown type of ME generator: {generators}")
 
    genInputFiles = [f.strip() for f in flags.Generator.inputGeneratorFile.split(",") if f.strip()]
    if not genInputFiles:
        raise RuntimeError("Generator.inputGeneratorFile is empty while input handling is requested")
 
    def _input_root(path, keep_suffix_after_underscore=False):
        fname = os.path.basename(path)
        if any(ext in fname for ext in (".tar.", ".tgz", ".gz")):
            return re.split(r"\.tar\.|\.tgz|\.gz", fname, maxsplit=1)[0]
        parts = fname.split("._", 1)
        if keep_suffix_after_underscore and len(parts) > 1:
            return parts[0] + "._" + parts[1].split(".", 1)[0]
        return parts[0]
 
    # If there is a single file, make a symlink. If multiple files, merge them into one output eventsFile.
    if len(genInputFiles) == 1:
        inputroot = _input_root(genInputFiles[0], keep_suffix_after_underscore=False)
        if inputroot.endswith(".events"):
            inputroot = inputroot[:-7]
        realEventsFile = _find_unique_file(f"*{inputroot}.*ev*ts")
        _mk_symlink(realEventsFile, eventsFile)
        if is_lhe_input:
            return _count_lhe_events(eventsFile)
        return None
 
    allFiles = []
    for file in genInputFiles:
        # Since we can have multiple files from the same task, include more of the filename
        # to make the lookup unique in the plain-file case.
        inputroot = _input_root(file, keep_suffix_after_underscore=True)
        evgenLog.info("inputroot = %s", inputroot)
        realEventsFile = _find_unique_file(f"*{inputroot}.*ev*ts")
        # The only input format where merging is permitted is LHE.
        with open(realEventsFile, "r") as f:
            first_line = f.readline()
            if "LesHouche" not in first_line:
                raise RuntimeError(f"{realEventsFile} is NOT a LesHouche file")
        allFiles.append(realEventsFile)
    _merge_lhe_files(allFiles, eventsFile)
 
    return _count_lhe_events(eventsFile)
 
 

_is_txt_only_run()

python.EvgenHelpers._is_txt_only_run ( flags )

protected

Helper function to determine if this is LHE-only generation with no showering)

Definition at line 211 of file EvgenHelpers.py.

def _is_txt_only_run(flags):
    """Helper function to determine if this is LHE-only generation with no showering)"""
    has_txt = bool(flags.Output.TXTFileName)
    has_evnt = bool(flags.Output.EVNTFileName)
    has_yoda =bool(flags.Generator.outputYODAFile)
    
    return has_txt and not has_evnt and not has_yoda

_merge_lhe_files()

python.EvgenHelpers._merge_lhe_files ( listOfFiles, outputFile )

protected

This function merges a list of input LHE files into one output file.
The header is taken from the first file, but the number of events is
updated to equal the total number of events in all input files.

Definition at line 70 of file EvgenHelpers.py.

def _merge_lhe_files(listOfFiles, outputFile):
    """
    This function merges a list of input LHE files into one output file.
    The header is taken from the first file, but the number of events is
    updated to equal the total number of events in all input files.
    """
    if os.path.exists(outputFile):
        print("outputFile", outputFile, "already exists. Will rename to", outputFile + ".OLD")
        os.rename(outputFile, outputFile + ".OLD")
 
    total_events = 0
    for file in listOfFiles:
        total_events += _count_lhe_events(file)
 
    wrote_header = False
    with open(outputFile, "w") as output:
        for file in listOfFiles:
            inHeader = True
            header = ""
            print("*** Starting file", file)
            with open(file, "r") as infile:
                for line in infile:
                    # Reading first event signals that we are done with all header information.
                    if "<event" in line and inHeader:
                        inHeader = False
                        if not wrote_header:
                            wrote_header = True
                            output.write(header)
                        output.write(line)
                    # Each input file ends with "</LesHouchesEvents>". We only write it once at the end.
                    elif not inHeader and "</LesHouchesEvents>" not in line:
                        output.write(line)
 
                    if inHeader:
                        # Format for storing number of events differs in MG and Powheg.
                        if "nevents" in line:
                            # MG5 format is "n = nevents".
                            parts = line.split("=")
                            if parts:
                                line = line.replace(parts[0], str(total_events), 1)
                        elif "numevts" in line:
                            # Powheg format is "numevts n".
                            parts = line.split()
                            if len(parts) > 1:
                                line = line.replace(parts[1], str(total_events), 1)
                        header += line
 
        output.write("</LesHouchesEvents>\n")
 
 

_mk_symlink()

python.EvgenHelpers._mk_symlink ( srcfile, dstfile )

protected

Helper function to make symlinks.

Definition at line 13 of file EvgenHelpers.py.

def _mk_symlink(srcfile, dstfile):
    """Helper function to make symlinks."""
    if dstfile:
        if os.path.exists(dstfile) and not os.path.samefile(dstfile, srcfile):
            os.remove(dstfile)
        if not os.path.exists(dstfile):
            evgenLog.info(f"Symlinking {srcfile} to {dstfile}")
            print (f"Symlinking {srcfile} to {dstfile}")
            os.symlink(srcfile, dstfile)
        else:
            evgenLog.debug(f"Symlinking: {dstfile} is already the same as {srcfile}")
 
 

_validate_sample_properties()

python.EvgenHelpers._validate_sample_properties ( sample )

protected

Helper function to validate and set sample properties

Definition at line 187 of file EvgenHelpers.py.

def _validate_sample_properties(sample):
    """Helper function to validate and set sample properties"""
    # Required fields with lightweight, explicit validators.
    required_rules = {
        "keywords": lambda v: isinstance(v, list) and len(v) > 0,
        "contact": lambda v: isinstance(v, list) and len(v) > 0,
        "nEventsPerJob": lambda v: v is not None,
    }
    for field, validator in required_rules.items():
        value = getattr(sample, field, None)
        if not validator(value):
            raise RuntimeError(f"self.{field} should be set in Sample(EvgenConfig)")
 
    input_files_per_job = getattr(sample, "inputFilesPerJob", 0)
    me_generator = getattr(sample, "MEgenerator", None)
 
    if input_files_per_job < 0:
        raise RuntimeError("self.inputFilesPerJob should be >= 0 in Sample(EvgenConfig)")
    if input_files_per_job > 0 and not me_generator:
        raise RuntimeError("self.MEgenerator should be set when self.inputFilesPerJob > 0 in Sample(EvgenConfig)")
    if input_files_per_job == 0 and me_generator:
        raise RuntimeError("self.MEgenerator should be empty when self.inputFilesPerJob == 0 in Sample(EvgenConfig)")
 
 

Variable Documentation

evgenLog

python.EvgenHelpers.evgenLog = logging.getLogger("Gen_tf")

Definition at line 10 of file EvgenHelpers.py.