Inheritance diagram for python.CPBaseRunner.CPBaseRunner:

Collaboration diagram for python.CPBaseRunner.CPBaseRunner:

Public Member Functions
	__init__ (self)
	args (self)
	inputList (self)
	outputName (self)
	printFlags (self)
	addCustomArguments (self)
	makeAlgSequence (self)
	run (self)
	setup (self)
	printAvailableArguments (self)
	modifyParserArguments (self)
	modifyYamlConfig (self)
	modifyAlgSequence (self)

Static Public Member Functions
	findLocalPathYamlConfig (textConfigPath)
	findRepoPathYamlConfig (textConfigPath)

Public Attributes
	logger = logging.getLogger("CPBaseRunner")
	parser = self._defaultParseArguments()
	rawConfig = None
	flags = self._defaultFlagsInitialization()
	config = self._readYamlConfig()

Protected Member Functions
	_defaultFlagsInitialization (self)
	_defaultParseArguments (self)
	_mergeYamlconfig (self, Path yaml_path, list[Path] yaml_paths)
	_readYamlConfig (self)
	_findYamlConfig (self, local=True)
	_parseInputFileList (Path path)

Protected Attributes
	_args = None
list	_inputList = None

Detailed Description

Definition at line 10 of file CPBaseRunner.py.

Constructor & Destructor Documentation

◆ init()

python.CPBaseRunner.CPBaseRunner.__init__ ( self )

Definition at line 11 of file CPBaseRunner.py.

    def __init__(self):
        self.logger = logging.getLogger("CPBaseRunner")
        self._args = None
        self._inputList = None
        self.parser = self._defaultParseArguments()
        self.rawConfig = None
        # parse the arguments here is a bad idea
 

Member Function Documentation

◆ _defaultFlagsInitialization()

python.CPBaseRunner.CPBaseRunner._defaultFlagsInitialization ( self )

protected

Definition at line 74 of file CPBaseRunner.py.

    def _defaultFlagsInitialization(self):
        from AthenaConfiguration.AllConfigFlags import initConfigFlags
        flags = initConfigFlags()
        flags.Input.Files = self.inputList
        flags.Exec.MaxEvents = self.args.max_events
        flags.Exec.SkipEvents = self.args.skip_n_events
        flags.Exec.EventPrintoutInterval = 10000
        return flags
 

◆ _defaultParseArguments()

python.CPBaseRunner.CPBaseRunner._defaultParseArguments ( self )

protected

Definition at line 83 of file CPBaseRunner.py.

    def _defaultParseArguments(self):
        parser = argparse.ArgumentParser(
            description='Runscript for CP Algorithm unit tests')
        baseGroup = parser.add_argument_group('Base Script Options')
        baseGroup.add_argument('-i', '--input-list', dest='input_list',
                               help='path to text file containing list of input files, or a single root file')
        baseGroup.add_argument('-o', '--output-name', dest='output_name', default='output',
                               help='output name of the analysis root file')
        baseGroup.add_argument('-e', '--max-events', dest='max_events', type=int, default=-1,
                               help='Number of events to run')
        baseGroup.add_argument('-t', '--text-config', dest='text_config',
                               help='path to the YAML configuration file. Tips: use atlas_install_data(path/to/*.yaml) in CMakeLists.txt can help locating the config just by the config file name.')
        baseGroup.add_argument('--no-systematics', dest='no_systematics',
                               action='store_true', help='Disable systematics')
        baseGroup.add_argument('--skip-n-events', dest='skip_n_events', type=int, default=0,
                               help='Skip the first N events in the run, not first N events for each file. This is meant for debugging only. \nIn Eventloop, this option disable the cutbookkeeper algorithms due to technical reasons, and can only be ran in direct-driver.')
        baseGroup.add_argument('--merge-output-files', dest='merge_output_files', action='store_true', help='Merge the output histogram and n-tuple files into a single file.')
        return parser
 

◆ _findYamlConfig()

python.CPBaseRunner.CPBaseRunner._findYamlConfig	(		self,
			local = True )

protected

Definition at line 128 of file CPBaseRunner.py.

    def _findYamlConfig(self, local=True):
        # Find local and abs path first
        if local and ((yamlConfig := CPBaseRunner.findLocalPathYamlConfig(self.args.text_config)) is not None):
            return yamlConfig, [yamlConfig.parent]
        # Then search in the analysis repository and warn for duplicates
        else:
            yamlConfig, yamlBasePath = CPBaseRunner.findRepoPathYamlConfig(self.args.text_config)
            # Try the slowest method using AthenaCommon if nothing found
            if not yamlConfig:
                from AthenaCommon.Utils.unixtools import find_datafile
                return find_datafile(self.args.text_config)
    
            if len(yamlConfig) > 1:
                raise FileExistsError(
                    f'Multiple files named \"{self.args.text_config}\" found in the analysis repository. Please provide a more specific path to the config file.\nMatches found:\n' + '\n'.join(yamlConfig))
            else:
                return yamlConfig[0], [yamlBasePath[0]]
 

◆ _mergeYamlconfig()

python.CPBaseRunner.CPBaseRunner._mergeYamlconfig	(		self,
		Path	yaml_path,
		list[Path]	yaml_paths )

protected

Definition at line 102 of file CPBaseRunner.py.

    def _mergeYamlconfig(self, yaml_path: Path, yaml_paths: list[Path]):
        with yaml_path.open("r", encoding="utf-8") as cfg_file:
            import yaml
            config_data = yaml.safe_load(cfg_file)
            from AnalysisAlgorithmsConfig.ConfigText import combineConfigFiles
            combined = combineConfigFiles(config_data, yaml_paths, fragment_key="include")
            if combined:
                with open("merged_config.yaml", "w") as cfg:
                    cfg.write(yaml.dump(config_data))
                self.logger.info("Merged included fragments into main config.")
            return config_data, combined
 

◆ _parseInputFileList()

python.CPBaseRunner.CPBaseRunner._parseInputFileList ( Path path )

protected

Definition at line 182 of file CPBaseRunner.py.

    def _parseInputFileList(path: Path):
        files = []
        with path.open('r') as inputText:
            for line in inputText.readlines():
                # Strip the line and skip comments and empty lines
                line = line.strip()
                if line.startswith('#') or not line:
                    continue
                line_path = Path(line)
                if line_path.is_dir():
                    if not any(line_path.iterdir()):
                        raise FileNotFoundError(
                            f"The directory \"{path}\" is empty. Please provide a directory with .root files.")
                    for root_file in line_path.iterdir():
                        if root_file.suffix == '.root':
                            files.append(str(root_file))
                else:
                    files += line.split(',')
            # Remove leading/trailing whitespaces from file names
            files = [file.strip() for file in files]
        return files
 

◆ _readYamlConfig()

python.CPBaseRunner.CPBaseRunner._readYamlConfig ( self )

protected

Definition at line 114 of file CPBaseRunner.py.

    def _readYamlConfig(self):
        yamlConfig, yamlPaths = self._findYamlConfig(local=True)
        if yamlConfig is None:
            raise FileNotFoundError(f'Failed to locate \"{self.args.text_config}\" config file!'
                                    'Check if you have a typo in -t/--text-config argument or missing file in the analysis configuration sub-directory.')
        self.logger.info(f"Found YAML config at: {yamlConfig}")
        self.logger.info("Setting up configuration based on YAML config:")
 
        from AnalysisAlgorithmsConfig.ConfigText import TextConfig
        self.rawConfig, merged = self._mergeYamlconfig(yamlConfig, yamlPaths)
        self.modifyYamlConfig()
        config = TextConfig(config=self.rawConfig)
        return config
 

◆ addCustomArguments()

python.CPBaseRunner.CPBaseRunner.addCustomArguments ( self )

Definition at line 62 of file CPBaseRunner.py.

    def addCustomArguments(self):
        pass
 

◆ args()

python.CPBaseRunner.CPBaseRunner.args ( self )

Definition at line 20 of file CPBaseRunner.py.

    def args(self):
        if self._args is None:
            self._args = self.parser.parse_args()
        return self._args
 

◆ findLocalPathYamlConfig()

python.CPBaseRunner.CPBaseRunner.findLocalPathYamlConfig ( textConfigPath )

static

Definition at line 147 of file CPBaseRunner.py.

    def findLocalPathYamlConfig(textConfigPath):
        configPath = Path(textConfigPath).expanduser()        
        if configPath.is_absolute() and configPath.is_file():
            return configPath
        cwdPath = Path.cwd() / configPath
        if cwdPath.is_file():
            return cwdPath
        return None
 

◆ findRepoPathYamlConfig()

python.CPBaseRunner.CPBaseRunner.findRepoPathYamlConfig ( textConfigPath )

static

Search for the file up to two levels deep within the first DATAPATH entry.
First, check directly under the analysis repository (depth 0).
Then, check immediate subdirectories (depth 1), looking for the file inside each.
Returns a list of all matches found and a list of packages/base paths.

Definition at line 157 of file CPBaseRunner.py.

    def findRepoPathYamlConfig(textConfigPath):
        """
        Search for the file up to two levels deep within the first DATAPATH entry.
        First, check directly under the analysis repository (depth 0).
        Then, check immediate subdirectories (depth 1), looking for the file inside each.
        Returns a list of all matches found and a list of packages/base paths.
        """
        matches = []
        basePaths = []
        analysisRepoPath = Path(environ.get("DATAPATH", "").split(pathsep)[0])
        # Depth 0: Directly under analysisRepoPath
        searchPath = analysisRepoPath / textConfigPath
        if searchPath.is_file():
            matches.append(searchPath)
            basePaths.append(analysisRepoPath if searchPath.parent == analysisRepoPath else analysisRepoPath / Path(textConfigPath).parts[0])
        # Depth 1: Inside immediate subdirectories
        for subdir in analysisRepoPath.iterdir():
            if not subdir.is_dir():
                continue
            candidate = analysisRepoPath / subdir / textConfigPath
            if candidate.is_file():
                matches.append(candidate)
                basePaths.append(analysisRepoPath / subdir)
        return matches, basePaths
 

◆ inputList()

python.CPBaseRunner.CPBaseRunner.inputList ( self )

Definition at line 26 of file CPBaseRunner.py.

    def inputList(self):
        if self._inputList is None:
            if self.args.input_list.endswith('.txt'):
                self._inputList = CPBaseRunner._parseInputFileList(
                    Path(self.args.input_list))
            elif ".root" in self.args.input_list:
                self._inputList = [self.args.input_list]
            else:
                raise FileNotFoundError(f'Input file list \"{self.args.input_list}\" is not supported!'
                                        'Please provide a text file with a list of input files or a single root file.')
            self.logger.info("Initialized input files: %s", self._inputList)
        return self._inputList
 

◆ makeAlgSequence()

python.CPBaseRunner.CPBaseRunner.makeAlgSequence ( self )

Definition at line 66 of file CPBaseRunner.py.

    def makeAlgSequence(self):
        pass
 

◆ modifyAlgSequence()

python.CPBaseRunner.CPBaseRunner.modifyAlgSequence ( self )

Definition at line 229 of file CPBaseRunner.py.

    def modifyAlgSequence(self):  # noqa: B027
        # For AthAnalysis: self.configSeq.some_attribute = some_value
        # For EventLoop: self.algSeq.some_attribute = some_value
        # The seemingly trivial log is to prevent CI from complaining about empty hook functions
        pass

◆ modifyParserArguments()

python.CPBaseRunner.CPBaseRunner.modifyParserArguments ( self )

Definition at line 219 of file CPBaseRunner.py.

    def modifyParserArguments(self):  # noqa: B027
        # Example: self.parser.add_argument('--no-filter', dest='no_filter', action='store_true', help='Disable filtering')
        # The seemingly trivial log is to prevent CI from complaining about empty hook functions
        pass
 

◆ modifyYamlConfig()

python.CPBaseRunner.CPBaseRunner.modifyYamlConfig ( self )

Definition at line 224 of file CPBaseRunner.py.

    def modifyYamlConfig(self):  # noqa: B027
        # Example: self.rawConfig['SomeSection']['SomeOption'] = some_value
        # The seemingly trivial log is to prevent CI from complaining about empty hook functions
        pass
 

◆ outputName()

python.CPBaseRunner.CPBaseRunner.outputName ( self )

Definition at line 40 of file CPBaseRunner.py.

    def outputName(self):
        if self.args.output_name.endswith('.root'):
            return self.args.output_name[:-5]
        else:
            return self.args.output_name
 

◆ printAvailableArguments()

python.CPBaseRunner.CPBaseRunner.printAvailableArguments ( self )

Definition at line 210 of file CPBaseRunner.py.

    def printAvailableArguments(self):
        self.parser.description = 'CPRunScript available arguments'
        self.parser.usage = argparse.SUPPRESS
        self.parser.print_help()
 

◆ printFlags()

python.CPBaseRunner.CPBaseRunner.printFlags ( self )

Definition at line 46 of file CPBaseRunner.py.

    def printFlags(self):
        self.logger.info("="*73)
        self.logger.info("="*20 + "FLAG CONFIGURATION" + "="*20)
        self.logger.info("="*73)
        self.logger.info("    Input files:     %s", self.flags.Input.isMC)
        self.logger.info("    RunNumber:       %s",
                         self.flags.Input.RunNumbers)
        self.logger.info("    MCCampaign:      %s",
                         self.flags.Input.MCCampaign)
        self.logger.info("    GeneratorInfo:   %s",
                         self.flags.Input.GeneratorsInfo)
        self.logger.info("    MaxEvents:       %s", self.flags.Exec.MaxEvents)
        self.logger.info("    SkipEvents:      %s", self.flags.Exec.SkipEvents)
        self.logger.info("="*73)
 

◆ run()

python.CPBaseRunner.CPBaseRunner.run ( self )

Definition at line 70 of file CPBaseRunner.py.

    def run(self):
        pass
 

◆ setup()

python.CPBaseRunner.CPBaseRunner.setup ( self )

Definition at line 204 of file CPBaseRunner.py.

    def setup(self):
        self.flags = self._defaultFlagsInitialization()
        self.modifyParserArguments()
        self.parser.parse_args()
        self.config = self._readYamlConfig()
 

Member Data Documentation

◆ _args

python.CPBaseRunner.CPBaseRunner._args = None

protected

Definition at line 13 of file CPBaseRunner.py.

◆ _inputList

python.CPBaseRunner.CPBaseRunner._inputList = None

protected

Definition at line 14 of file CPBaseRunner.py.

◆ config

python.CPBaseRunner.CPBaseRunner.config = self._readYamlConfig()

Definition at line 208 of file CPBaseRunner.py.

◆ flags

python.CPBaseRunner.CPBaseRunner.flags = self._defaultFlagsInitialization()

Definition at line 205 of file CPBaseRunner.py.

◆ logger

python.CPBaseRunner.CPBaseRunner.logger = logging.getLogger("CPBaseRunner")

Definition at line 12 of file CPBaseRunner.py.

◆ parser

python.CPBaseRunner.CPBaseRunner.parser = self._defaultParseArguments()

Definition at line 15 of file CPBaseRunner.py.

◆ rawConfig

python.CPBaseRunner.CPBaseRunner.rawConfig = None

Definition at line 16 of file CPBaseRunner.py.

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

CPBaseRunner.py

Public Member Functions

Static Public Member Functions

Public Attributes

Protected Member Functions

Protected Attributes

Detailed Description

Constructor & Destructor Documentation

◆ __init__()

Member Function Documentation

◆ _defaultFlagsInitialization()

◆ _defaultParseArguments()

◆ _findYamlConfig()

◆ _mergeYamlconfig()

◆ _parseInputFileList()

◆ _readYamlConfig()

◆ addCustomArguments()

◆ args()

◆ findLocalPathYamlConfig()

◆ findRepoPathYamlConfig()

◆ inputList()

◆ makeAlgSequence()

◆ modifyAlgSequence()

◆ modifyParserArguments()

◆ modifyYamlConfig()

◆ outputName()

◆ printAvailableArguments()

◆ printFlags()

◆ run()

◆ setup()

Member Data Documentation

◆ _args

◆ _inputList

◆ config

◆ flags

◆ logger

◆ parser

◆ rawConfig

◆ init()